From 0859c6b13371cfc0387b3ddee38025d44742b9ee Mon Sep 17 00:00:00 2001 From: hmo Date: Thu, 13 Aug 2026 12:13:41 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20regime=5Fperf=20v2=E2=80=94=E2=80=94?= =?UTF-8?q?=E6=AF=8F=E7=AD=96=E7=95=A5x=E6=B8=A9=E5=8C=BA=E8=B7=91?= =?UTF-8?q?=E7=BB=84=E5=90=88=E6=A8=A1=E6=8B=9F(portfolio=5Fsim=20100?= =?UTF-8?q?=E4=B8=87/10=E6=A7=BD),=20=E6=B8=A9=E5=8C=BA=E7=BA=A7total=5Fre?= =?UTF-8?q?turn/cagr/max=5Fdd/capital=5Ffinal/sharpe/profit=5Ffactor?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deploy/profile-scripts/regime_perf.py | 198 +++++++++++++------------- 1 file changed, 99 insertions(+), 99 deletions(-) diff --git a/deploy/profile-scripts/regime_perf.py b/deploy/profile-scripts/regime_perf.py index 339a6cc8..d5c33be8 100644 --- a/deploy/profile-scripts/regime_perf.py +++ b/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()