feat: regime_perf v2——每策略x温区跑组合模拟(portfolio_sim 100万/10槽), 温区级total_return/cagr/max_dd/capital_final/sharpe/profit_factor

deploy/profile-scripts/regime_perf.py

@@ -1,18 +1,19 @@
-#!/usr/bin/env python3
+#!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-"""regime_perf.py — 策略-温区表现常态化记录（2026-08-13）
+"""regime_perf.py v2 — 策略-温区表现常态化记录（2026-08-13 温区级组合模拟）
 
-记录策略在不同温区（trend_up/choppy/trend_down）的表现，供"适用温度"动态评估。
-- 数据来源：strategy_research 回测 trades（按入场日归入温区）+ 实盘 strategy_tracking
-- 表：strategy_regime_perf（strategy, regime, trades, win_rate, avg_pnl, updated_at）
-- 原则：策略全温区发信号（去门控后），记录各温区真实表现；适用温区是动态的，随数据更新
+记录策略在不同温区（trend_up/choppy/trend_down）的表现，含【温区级组合模拟】——
+每个策略×温区，把该温区 trades 跑 portfolio_sim（100万本金/10槽/含费），
+得到温区级 total_return/cagr/max_dd/capital_final/positions_taken/sharpe/profit_factor。
 
-用法：
-  python3 regime_perf.py            # 全量更新（从回测+实盘重算）
-  from regime_perf import get_regime_perf
+表 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
@@ -24,28 +25,24 @@ sys.path.insert(0, "/home/hmo/MoFin")
 
 DB = "/home/hmo/MoFin/data/mofin.db"
 
+
 def load_all_strategies():
-    """从 strategy_research 读取所有策略版本（含历史/表现不佳的——可能在特定温区能打）"""
     conn = sqlite3.connect(DB, timeout=30)
     conn.execute("PRAGMA busy_timeout=30000")
     rows = conn.execute("SELECT DISTINCT version FROM strategy_research ORDER BY version").fetchall()
     conn.close()
     return [r[0] for r in rows if r[0]]
 
-# 关注的策略（动态：全部版本）
-STRATEGIES = load_all_strategies()
 
 def load_regime_map():
-    """date -> regime（用平滑 regime_tracker 的周期反查更合理，这里用 market_regime 原始 + 手动按 K=5 平滑）
-    简化：直接用 market_regime 的 regime（与平滑 K=5 差异主要在边界几天，评估可接受）"""
     conn = sqlite3.connect(DB, timeout=30)
     conn.execute("PRAGMA busy_timeout=30000")
     rows = conn.execute("SELECT date, regime FROM market_regime").fetchall()
     conn.close()
     return dict(rows)
 
+
 def get_trades_from_research(version):
-    """从 strategy_research 取最新回测 trades"""
     conn = sqlite3.connect(DB, timeout=30)
     conn.execute("PRAGMA busy_timeout=30000")
     rows = conn.execute(
@@ -60,58 +57,51 @@ def get_trades_from_research(version):
     except Exception:
         return []
 
-def get_trades_from_tracking():
-    """从实盘 strategy_tracking 取已平仓交易"""
-    conn = sqlite3.connect(DB, timeout=30)
-    conn.execute("PRAGMA busy_timeout=30000")
-    rows = conn.execute(
-        "SELECT version_seq, tracked_at, theoretical_pnl FROM strategy_tracking WHERE status='closed'"
-    ).fetchall()
-    conn.close()
-    result = []
-    for version_seq, tracked_at, pnl in rows:
-        if version_seq and tracked_at:
-            result.append({"version": version_seq, "entry_date": tracked_at[:10], "profit_pct": pnl})
-    return result
 
-def compute(use_tracking=True):
-    """计算所有策略各温区表现"""
+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():
     regime_map = load_regime_map()
-    stats = defaultdict(lambda: defaultdict(lambda: {"n": 0, "win": 0, "pnl": 0}))
+    strategies = load_all_strategies()
+    print(f"策略数: {len(strategies)}")
 
-    for v in STRATEGIES:
-        trades = get_trades_from_research(v)
-        for t in trades:
-            ed = t.get("entry_date", "")
-            if ed not in regime_map:
-                continue
-            reg = regime_map[ed]
-            pnl = t.get("profit_pct", 0) or 0
-            stats[v][reg]["n"] += 1
-            stats[v][reg]["pnl"] += pnl
-            if pnl > 0:
-                stats[v][reg]["win"] += 1
-
-    if use_tracking:
-        for t in get_trades_from_tracking():
-            v = t["version"]
-            if v not in stats:
-                stats[v] = defaultdict(lambda: {"n": 0, "win": 0, "pnl": 0})
-            ed = t["entry_date"]
-            if ed in regime_map:
-                reg = regime_map[ed]
-                pnl = t["profit_pct"] or 0
-                stats[v][reg]["n"] += 1
-                stats[v][reg]["pnl"] += pnl
-                if pnl > 0:
-                    stats[v][reg]["win"] += 1
-
-    return stats
-
-def save(stats):
-    """写入 strategy_regime_perf 表（清空重建，保持与最新数据同步）"""
-    conn = sqlite3.connect(DB, timeout=30)
-    conn.execute("PRAGMA busy_timeout=30000")
+    conn = sqlite3.connect(DB, timeout=60)
+    conn.execute("PRAGMA busy_timeout=60000")
     conn.execute("""
         CREATE TABLE IF NOT EXISTS strategy_regime_perf (
             strategy TEXT,
@@ -119,56 +109,66 @@ def save(stats):
             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)
         )
     """)
     conn.execute("DELETE FROM strategy_regime_perf")
+
+    written = 0
     now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-    for v, regs in stats.items():
-        for reg, s in regs.items():
-            if s["n"] < 2:
-                continue  # 样本太少不记录（>=2 给观察机会）
-            wr = s["win"] / s["n"] * 100
-            avg = s["pnl"] / s["n"]
+    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, regime, trades, win_rate, avg_pnl, updated_at) VALUES (?,?,?,?,?,?)",
-                (v, reg, s["n"], round(wr, 1), round(avg, 2), now)
+                """INSERT OR REPLACE INTO 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)
+                   VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
+                (v, 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} 条（含温区级组合模拟）")
 
-def main():
-    stats = compute(use_tracking=True)
-    save(stats)
-    # 打印
-    print("=== 策略-温区表现（strategy_regime_perf）===")
+    # 打印样例
     conn = sqlite3.connect(DB, timeout=30)
-    rows = conn.execute("SELECT strategy, regime, trades, win_rate, avg_pnl FROM strategy_regime_perf ORDER BY strategy, regime").fetchall()
+    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') ORDER BY strategy, regime"
+    ).fetchall()
     conn.close()
     for r in rows:
-        print(f"  {r[0]:<12} {r[1]:<12} {r[2]:>4}笔 胜率{r[3]:.0f}% 均盈{r[4]:+.2f}%")
+        print(f"  {r[0]:<12} {r[1]:<12} {r[2]:>4}笔 胜率{r[3]:>5.1f}% 年化{r[4]:>6.1f}% 资产{r[5]:>12.0f}")
 
-def get_regime_perf(strategy=None):
-    """读取策略-温区表现（供 router 动态适用温区）"""
-    conn = sqlite3.connect(DB, timeout=30)
-    conn.execute("PRAGMA busy_timeout=30000")
-    if strategy:
-        rows = conn.execute(
-            "SELECT regime, trades, win_rate, avg_pnl FROM strategy_regime_perf WHERE strategy=?",
-            (strategy,)
-        ).fetchall()
-    else:
-        rows = conn.execute(
-            "SELECT strategy, regime, trades, win_rate, avg_pnl FROM strategy_regime_perf"
-        ).fetchall()
-    conn.close()
-    if strategy:
-        return {r[0]: {"trades": r[1], "win_rate": r[2], "avg_pnl": r[3]} for r in rows}
-    result = defaultdict(dict)
-    for r in rows:
-        result[r[0]][r[1]] = {"trades": r[2], "win_rate": r[3], "avg_pnl": r[4]}
-    return dict(result)
 
 if __name__ == "__main__":
     main()