""" evolution/evolution_api.py — 进化模块 API 接口 供 dashboard 查询健康度、教训、迭代历史 """ import sys, os, json, sqlite3 sys.path.insert(0, '/home/hmo/MoFin') sys.path.insert(0, '/home/hmo/MoFin/deploy/profile-scripts') DB = os.environ.get('MOFIN_DB', '/home/hmo/MoFin/data/mofin.db') def get_evolution_dashboard(): """进化模块 Dashboard 数据""" conn = sqlite3.connect(DB) conn.row_factory = sqlite3.Row # 最近健康度（近30天） health = [] for r in conn.execute(""" SELECT strategy_version, date, live_trades, live_wins, live_return_pct, backtest_wr, backtest_avg_ret, deviation, health_score FROM strategy_health ORDER BY date DESC LIMIT 30 """).fetchall(): health.append(dict(r)) # 最近教训（近20条） lessons = [] for r in conn.execute(""" SELECT strategy_version, lesson_type, lesson_text, confidence, applied, created_at FROM strategy_lessons ORDER BY id DESC LIMIT 20 """).fetchall(): lessons.append(dict(r)) # 迭代历史 evolution = [] for r in conn.execute(""" SELECT parent_version, child_version, change_description, promoted, created_at FROM strategy_evolution ORDER BY id DESC LIMIT 20 """).fetchall(): evolution.append(dict(r)) # 当前策略基线（2026-08-11: 改为当前实盘 v_weak + 新策略 v_oversold） baseline = {} for v in ['v_weak', 'v_oversold']: r = conn.execute(""" SELECT results_json FROM strategy_research WHERE version=? AND period_tag='5y' ORDER BY id DESC LIMIT 1 """, (v,)).fetchone() if r: res = json.loads(r[0]) s = res.get('summary', {}) pf = s.get('portfolio_full', {}) baseline[v] = { 'win_rate': s.get('win_rate', 0), 'total_return': pf.get('total_return_pct', 0), 'cagr': pf.get('cagr_pct', 0), 'max_dd': pf.get('portfolio_max_dd_pct', 0), } conn.close() return { 'health': health, 'lessons': lessons, 'evolution': evolution, 'baseline': baseline, } def get_combo_dashboard(): """组合方案 Dashboard 数据 (2026-08-02 新增) 返回: 当前组合方案(v_next4+v_mr按regime分工) + 组合回测版本(v_combo) + 市场阶段 """ conn = sqlite3.connect(DB) conn.row_factory = sqlite3.Row # 1. 当前市场阶段 (market_regime) regime = None r = conn.execute("SELECT * FROM market_regime ORDER BY date DESC LIMIT 1").fetchone() if r: regime = dict(r) # 2. 组合回测版本 (v_combo 家族) combos = [] rows = conn.execute( "SELECT id, version, market, period_tag, created_at, results_json" " FROM strategy_research WHERE version LIKE '%combo%' OR version LIKE 'v_combo%'" " ORDER BY id DESC" ).fetchall() for r in rows: d = dict(r) res = json.loads(d.pop("results_json") or "{}") s = res.get("summary", {}) pf = s.get("portfolio_full", {}) p5 = s.get("portfolio", {}) d["summary_stats"] = { "total_trades": s.get("total_trades"), "win_rate": s.get("win_rate"), "avg_profit_pct": s.get("avg_profit_pct"), "avg_hold_days": s.get("avg_hold_days"), "sharpe_ratio": s.get("sharpe_ratio"), "profit_factor": s.get("profit_factor"), "universality": s.get("universality", {}), "portfolio": p5, "portfolio_full": pf, } combos.append(d) # 3. 组合成员策略的独立指标 # 2026-08-11 更新：组合成员 = v_weak（实盘）+ p_oversold（新策略），替代旧的 v_next4+v_mr # v_next4 移除（池内卫星仓，全市场失效；现有池子票不是它选的） members = {} for v in ["v_weak", "p_oversold"]: sel_v = "v_weak" if v == "v_mr" else None candidates = [sel_v, "v_mr_sel", v] if sel_v else [v] r = None used_sel = False for cv in candidates: r = conn.execute( "SELECT results_json FROM strategy_research" " WHERE version=? AND period_tag='10y' ORDER BY id DESC LIMIT 1", (cv,), ).fetchone() if r: used_sel = (cv == "v_weak") break if r: res = json.loads(r[0]) s = res.get("summary", {}) pf = s.get("portfolio_full", {}) p5 = s.get("portfolio", {}) members[v] = { "role": "弱市超跌确认（实盘）" if v == "v_weak" else "预测超跌反弹（新策略）", "version": ("v_weak" if used_sel else "v_mr") if v == "v_mr" else v, "is_sel": used_sel, "trades": s.get("total_trades"), "win_rate": s.get("win_rate"), "avg_profit_pct": s.get("avg_profit_pct"), "avg_hold_days": s.get("avg_hold_days"), "cagr_pct": p5.get("cagr_pct"), "return_pct": p5.get("total_return_pct"), "max_dd_pct": p5.get("portfolio_max_dd_pct"), "slots": p5.get("slots") or 6, "universality": s.get("universality", {}), # 组合模拟实际执行笔数(扣费后) + 年均(手工可行性参考) "positions_taken_5slot": p5.get("positions_taken"), "positions_taken_full": pf.get("positions_taken"), } # 2026-08-11：p_oversold 无回测数据时给兜底卡片（新策略待回测） if "p_oversold" not in members: members["p_oversold"] = { "role": "预测超跌反弹（新策略）", "version": "p_oversold", "is_sel": False, "trades": None, "win_rate": None, "avg_profit_pct": None, "avg_hold_days": None, "cagr_pct": None, "return_pct": None, "max_dd_pct": None, "slots": 10, "universality": {}, "positions_taken_5slot": None, "positions_taken_full": None, "note": "新策略，待回测/实盘验证", } conn.close() return { "regime": regime, "combos": combos, "members": members, "routing": [ {"regime": "trend_up", "active": "p_oversold", "action": "预测超跌反弹", "desc": "趋势市/反弹期, p_oversold 预测超跌反弹"}, {"regime": "choppy", "active": "v_weak", "action": "弱市超跌确认", "desc": "震荡/下跌市, v_weak 均值回复主战场"}, {"regime": "trend_down", "active": "v_weak", "action": "深超跌管理", "desc": "下跌市, v_weak 管理超跌持仓"}, ], } def get_health_trend(version='v_weak', days=30): """健康度趋势""" conn = sqlite3.connect(DB) conn.row_factory = sqlite3.Row rows = conn.execute(""" SELECT date, health_score, deviation, live_trades FROM strategy_health WHERE strategy_version=? ORDER BY date DESC LIMIT ? """, (version, days)).fetchall() conn.close() return [dict(r) for r in rows] def record_evolution(parent, child, description, backtest_result=None, promoted=0): """记录一次策略迭代""" conn = sqlite3.connect(DB) conn.execute(""" INSERT INTO strategy_evolution (parent_version, child_version, change_description, backtest_result, promoted) VALUES (?, ?, ?, ?, ?) """, (parent, child, description, json.dumps(backtest_result) if backtest_result else None, promoted)) conn.commit() conn.close() if __name__ == '__main__': d = get_evolution_dashboard() print(f"健康度: {len(d['health'])}条, 教训: {len(d['lessons'])}条, 迭代: {len(d['evolution'])}条") print(f"基线: {list(d['baseline'].keys())}")