#!/usr/bin/env python3 # -*- coding: utf-8 -*- """regime_perf_by_period.py — 策略-温区表现按周期预计算（2026-08-15 数据加工层落地） 背景（老莫指正）：温区数据是数据加工层，必须预计算存储，不能每次请求现场重算。 本脚本把每个策略×市场×周期(1y/2y/5y/10y)×温区的表现预先算好落库， server 直接读表返回（毫秒级，无需内存缓存/现场计算）。 与 regime_perf.py 的区别： - regime_perf.py 只算最长窗口（全量），写 strategy_regime_perf - 本脚本按周期分窗，写 strategy_regime_perf_by_period（新增 period_tag 维度） - 年化用【线性放大】：温区组合总收益 × (窗口交易日数 / 该温区窗口内天数) （替代 portfolio_sim 的复利年化——复利在温区 trades 跨度短时会爆炸） 用法： python3 regime_perf_by_period.py --market=a --periods="1y 2y 5y 10y" python3 regime_perf_by_period.py --market=hk --periods="2y" """ import sys import json import math import sqlite3 from datetime import datetime from collections import defaultdict sys.path.insert(0, "/home/hmo/MoFin") sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts") DB = "/home/hmo/MoFin/data/mofin.db" # 周期 → 目标 period_tag（strategy_research 里的记录标签） PERIOD_TAGS = ["1y", "2y", "5y", "10y"] def get_conn(): conn = sqlite3.connect(DB, timeout=60) conn.execute("PRAGMA busy_timeout=60000") return conn def create_table(conn): conn.execute(""" CREATE TABLE IF NOT EXISTS strategy_regime_perf_by_period ( strategy TEXT, market TEXT NOT NULL DEFAULT 'a', regime TEXT, period_tag TEXT NOT NULL DEFAULT '2y', 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, market, regime, period_tag) ) """) 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 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 regime_days_in_window(conn, market, d_min, d_max): """窗口内各温区天数 + 总天数（线性年化用）""" rows = conn.execute( "SELECT regime, COUNT(*) n FROM market_regime WHERE market=? AND date>=? AND date<=? GROUP BY regime", (market, d_min, d_max)).fetchall() total = sum(r[1] for r in rows) return {r[0]: r[1] for r in rows}, total def process_period(conn, market, period_tag): """处理单个周期：所有策略的温区表现，写入 strategy_regime_perf_by_period""" # 温区映射 rmap = dict(conn.execute( "SELECT date, regime FROM market_regime WHERE market=?", (market,)).fetchall()) # 该周期所有策略记录 rows = conn.execute( "SELECT version, results_json FROM strategy_research " "WHERE COALESCE(market,'a')=? AND period_tag=? ORDER BY version", (market, period_tag)).fetchall() written = 0 now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") for v, results_json in rows: if not results_json: continue try: trades = json.loads(results_json).get("trades", []) except Exception: continue if not trades: continue # 温区归因 by_regime = defaultdict(list) for t in trades: ed = t.get("entry_date", "") if ed in rmap: by_regime[rmap[ed]].append(t) if not by_regime: continue # 窗口天数（该策略 trades 的 entry_date 范围） eds = [t.get("entry_date") for t in trades if t.get("entry_date") and t.get("entry_date") in rmap] if not eds: continue reg_days, total_days = regime_days_in_window(conn, market, min(eds), max(eds)) 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 ret = sim.get("total_return_pct") # 复利年化（2026-08-16 修正）：按该温区 trades 实际时间跨度 # 原线性放大 ret×(total_days/rd) 对高频复利策略失真（b_td1 388%×4=1547%） _eds = [t.get("entry_date") for t in reg_trades if t.get("entry_date")] if ret is not None and _eds: from datetime import datetime as _dt _d0 = _dt.strptime(min(_eds), "%Y-%m-%d") _d1 = _dt.strptime(max(_eds), "%Y-%m-%d") span_days = max((_d1 - _d0).days, 30) cagr = round(((1 + ret / 100) ** (365 / span_days) - 1) * 100, 1) else: cagr = None conn.execute( """INSERT OR REPLACE INTO strategy_regime_perf_by_period (strategy, market, regime, period_tag, 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, period_tag, len(reg_trades), extra.get("win_rate"), extra.get("avg_pnl"), extra.get("avg_hold_days"), ret, cagr, 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 return written def main(): market = "a" periods = PERIOD_TAGS for a in sys.argv[1:]: if a.startswith("--market="): market = a.split("=", 1)[1] elif a.startswith("--periods="): periods = a.split("=", 1)[1].split() conn = get_conn() create_table(conn) # 先清该市场旧数据 conn.execute("DELETE FROM strategy_regime_perf_by_period WHERE market=?", (market,)) total = 0 for pt in periods: n = process_period(conn, market, pt) print(f"[{market}][{pt}] 写入 {n} 条", flush=True) total += n conn.commit() conn.close() print(f"完成: market={market} periods={periods} 共 {total} 条") if __name__ == "__main__": main()