189 lines
7.0 KiB
Python
189 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""regime_perf.py v2 — 策略-温区表现常态化记录（2026-08-13 温区级组合模拟）
|
||
|
||
记录策略在不同温区（trend_up/choppy/trend_down）的表现，含【温区级组合模拟】——
|
||
每个策略×温区，把该温区 trades 跑 portfolio_sim（100万本金/10槽/含费），
|
||
得到温区级 total_return/cagr/max_dd/capital_final/positions_taken/sharpe/profit_factor。
|
||
|
||
表 strategy_regime_perf 扩展列（温区级组合指标）：
|
||
strategy, regime, trades, win_rate, avg_pnl, avg_hold_days,
|
||
total_return_pct, cagr_pct, portfolio_max_dd_pct, capital_final,
|
||
positions_taken, sharpe_ratio, profit_factor, updated_at
|
||
"""
|
||
import sys
|
||
import json
|
||
import math
|
||
import sqlite3
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
from collections import defaultdict
|
||
|
||
_SCRIPT_DIR = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(_SCRIPT_DIR))
|
||
sys.path.insert(0, "/home/hmo/MoFin")
|
||
|
||
DB = "/home/hmo/MoFin/data/mofin.db"
|
||
|
||
|
||
def load_all_strategies(market='a'):
|
||
"""读取策略版本列表。market='a'(默认,A股,行为不变) / 'hk'(港股策略)"""
|
||
conn = sqlite3.connect(DB, timeout=30)
|
||
conn.execute("PRAGMA busy_timeout=30000")
|
||
rows = conn.execute(
|
||
"SELECT DISTINCT version FROM strategy_research WHERE COALESCE(market,'a')=? ORDER BY version",
|
||
(market,)).fetchall()
|
||
conn.close()
|
||
return [r[0] for r in rows if r[0]]
|
||
|
||
|
||
def load_regime_map(market='a'):
|
||
"""读取指定市场的 date→regime 映射（温区归因用）"""
|
||
conn = sqlite3.connect(DB, timeout=30)
|
||
conn.execute("PRAGMA busy_timeout=30000")
|
||
rows = conn.execute(
|
||
"SELECT date, regime FROM market_regime WHERE market=?", (market,)
|
||
).fetchall()
|
||
conn.close()
|
||
return dict(rows)
|
||
|
||
|
||
def get_trades_from_research(version):
|
||
conn = sqlite3.connect(DB, timeout=30)
|
||
conn.execute("PRAGMA busy_timeout=30000")
|
||
rows = conn.execute(
|
||
"SELECT results_json FROM strategy_research WHERE version=? ORDER BY period_tag DESC, created_at DESC LIMIT 1",
|
||
(version,)
|
||
).fetchall()
|
||
conn.close()
|
||
if not rows:
|
||
return []
|
||
try:
|
||
return json.loads(rows[0][0]).get("trades", [])
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def portfolio_sim_wrap(trades, capital=1000000, max_positions=10):
|
||
"""温区 trades → 组合模拟（复用 strategy_lab.portfolio_sim）"""
|
||
if not trades:
|
||
return {}
|
||
try:
|
||
from strategy_lab import portfolio_sim
|
||
return portfolio_sim(trades, capital=capital, max_positions=max_positions, cost=True)
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def calc_extra(trades):
|
||
"""从 trades 算温区级 win_rate/avg_pnl/avg_hold/sharpe/profit_factor"""
|
||
if not trades:
|
||
return {}
|
||
profits = [t.get("profit_pct", 0) for t in trades]
|
||
wins = [p for p in profits if p > 0]
|
||
losses = [p for p in profits if p <= 0]
|
||
win_rate = len(wins) / len(profits) * 100 if profits else 0
|
||
avg_p = sum(profits) / len(profits) if profits else 0
|
||
avg_w = sum(wins) / len(wins) if wins else 0
|
||
avg_l = abs(sum(losses) / len(losses)) if losses else 1
|
||
pf = avg_w / avg_l if avg_l > 0 else 0
|
||
mean_r = avg_p / 100
|
||
std_r = math.sqrt(sum((p / 100 - mean_r) ** 2 for p in profits) / (len(profits) - 1)) if len(profits) > 1 else 0
|
||
sharpe = mean_r / std_r * math.sqrt(252) if std_r > 0 else 0
|
||
holds = [t.get("hold_days", 0) for t in trades if t.get("hold_days")]
|
||
avg_hold = sum(holds) / len(holds) if holds else 0
|
||
return {
|
||
"win_rate": round(win_rate, 1),
|
||
"avg_pnl": round(avg_p, 2),
|
||
"avg_hold_days": round(avg_hold, 1),
|
||
"sharpe_ratio": round(sharpe, 2),
|
||
"profit_factor": round(pf, 2),
|
||
}
|
||
|
||
|
||
def main():
|
||
# --market=a|hk：策略温区归因市场。当前所有策略为 A 股（market='a' 默认），
|
||
# HK 维度留给阶段4（港股策略落地后按 strategy→market 映射归因）
|
||
market = "a"
|
||
for a in sys.argv[1:]:
|
||
if a.startswith("--market="):
|
||
market = a.split("=", 1)[1]
|
||
regime_map = load_regime_map(market)
|
||
strategies = load_all_strategies(market)
|
||
print(f"市场: {market} | 策略数: {len(strategies)}")
|
||
|
||
conn = sqlite3.connect(DB, timeout=60)
|
||
conn.execute("PRAGMA busy_timeout=60000")
|
||
conn.execute("""
|
||
CREATE TABLE IF NOT EXISTS strategy_regime_perf (
|
||
strategy TEXT,
|
||
market TEXT NOT NULL DEFAULT 'a',
|
||
regime TEXT,
|
||
trades INTEGER,
|
||
win_rate REAL,
|
||
avg_pnl REAL,
|
||
avg_hold_days REAL,
|
||
total_return_pct REAL,
|
||
cagr_pct REAL,
|
||
portfolio_max_dd_pct REAL,
|
||
capital_final REAL,
|
||
positions_taken INTEGER,
|
||
sharpe_ratio REAL,
|
||
profit_factor REAL,
|
||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
PRIMARY KEY (strategy, regime, market)
|
||
)
|
||
""")
|
||
conn.execute("DELETE FROM strategy_regime_perf WHERE market=?", (market,))
|
||
|
||
written = 0
|
||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
for v in strategies:
|
||
trades = get_trades_from_research(v)
|
||
if not trades:
|
||
continue
|
||
# 按温区分组
|
||
by_regime = defaultdict(list)
|
||
for t in trades:
|
||
ed = t.get("entry_date", "")
|
||
if ed in regime_map:
|
||
by_regime[regime_map[ed]].append(t)
|
||
for reg, reg_trades in by_regime.items():
|
||
if len(reg_trades) < 2:
|
||
continue
|
||
extra = calc_extra(reg_trades)
|
||
sim = portfolio_sim_wrap(reg_trades)
|
||
if not sim:
|
||
continue
|
||
conn.execute(
|
||
"""INSERT OR REPLACE INTO strategy_regime_perf
|
||
(strategy, market, regime, trades, win_rate, avg_pnl, avg_hold_days,
|
||
total_return_pct, cagr_pct, portfolio_max_dd_pct, capital_final,
|
||
positions_taken, sharpe_ratio, profit_factor, updated_at)
|
||
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
|
||
(v, market, reg, len(reg_trades),
|
||
extra.get("win_rate"), extra.get("avg_pnl"), extra.get("avg_hold_days"),
|
||
sim.get("total_return_pct"), sim.get("cagr_pct"), sim.get("portfolio_max_dd_pct"),
|
||
sim.get("capital_final"), sim.get("positions_taken"),
|
||
extra.get("sharpe_ratio"), extra.get("profit_factor"), now)
|
||
)
|
||
written += 1
|
||
conn.commit()
|
||
conn.close()
|
||
print(f"写入 strategy_regime_perf {written} 条（market={market}，含温区级组合模拟）")
|
||
|
||
# 打印样例
|
||
conn = sqlite3.connect(DB, timeout=30)
|
||
rows = conn.execute(
|
||
"SELECT strategy, regime, trades, win_rate, cagr_pct, capital_final FROM strategy_regime_perf "
|
||
"WHERE strategy IN ('v_oversold','v_mr_sel','s2_panic') AND market=? ORDER BY strategy, regime",
|
||
(market,)
|
||
).fetchall()
|
||
conn.close()
|
||
for r in rows:
|
||
print(f" {r[0]:<12} {r[1]:<12} {r[2]:>4}笔 胜率{r[3]:>5.1f}% 年化{r[4]:>6.1f}% 资产{r[5]:>12.0f}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|