#!/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, universality_months INTEGER, universality_years INTEGER, universality_valid_years INTEGER, universality_score REAL, universality_leave1 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 # ── 2026-08-18 切窗周期(1m/6m/1y)：从最长周期 trades 切窗→温区聚合 ── # 老莫：1m/6m/1y温区数据须有真实差异 + 每天定时刷新（server只读表,快且新） SLICE_DAYS_2 = {'1m': 30, '6m': 185, '1y': 365} def process_slice_period(conn, market, period_tag): """对切窗周期：所有策略从最长周期trades切出窗口算温区表现，写表""" days = SLICE_DAYS_2.get(period_tag) if not days: return 0 from datetime import timedelta rmap = dict(conn.execute( "SELECT date, regime FROM market_regime WHERE market=?", (market,)).fetchall()) if not rmap: return 0 ver_rows = conn.execute( "SELECT DISTINCT version FROM strategy_research WHERE COALESCE(market,'a')=?", (market,)).fetchall() written = 0 now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") for (v,) in ver_rows: # 最长周期 trades rr = conn.execute( "SELECT results_json FROM strategy_research WHERE COALESCE(market,'a')=? AND version=? " "ORDER BY CASE COALESCE(period_tag,'2y') WHEN '10y' THEN 3 WHEN '5y' THEN 2 ELSE 1 END DESC, id DESC LIMIT 1", (market, v)).fetchone() if not rr or not rr[0]: continue try: trades = json.loads(rr[0]).get("trades", []) except Exception: continue if not trades: continue max_date = max(t.get('entry_date', '') for t in trades) cutoff = (datetime.strptime(max_date, '%Y-%m-%d') - timedelta(days=days)).strftime('%Y-%m-%d') sliced = [t for t in trades if t.get('entry_date', '') >= cutoff] if not sliced: continue by_regime = defaultdict(list) for t in sliced: ed = t.get("entry_date", "") if ed in rmap: by_regime[rmap[ed]].append(t) for reg, reg_trades in by_regime.items(): if not reg_trades: continue wins = sum(1 for t in reg_trades if (t.get('profit_pct') or 0) > 0) pnl = sum(t.get('profit_pct') or 0 for t in reg_trades) hold = sum(t.get('hold_days') or 0 for t in reg_trades) n = len(reg_trades) # 2026-08-18 修复：1m/6m/1y 切窗周期 cagr 缺失 → 激活矩阵三项永不齐 → 全灭。 # 切窗数据短，用线性年化（total_return × 365/窗口实际跨度），与脚本头"线性放大"一致。 _cagr = None try: _eds = [t.get('entry_date') for t in reg_trades if t.get('entry_date')] if _eds and len(_eds) >= 2: from datetime import datetime as _dts _d0 = _dts.strptime(min(_eds), '%Y-%m-%d') _d1 = _dts.strptime(max(_eds), '%Y-%m-%d') _span = max((_d1 - _d0).days, 30) _cagr = round(pnl * (365.0 / _span), 1) except Exception: _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, universality_months, universality_years, " "universality_valid_years, universality_score, universality_leave1, updated_at) " "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (v, market, reg, period_tag, n, round(wins / n * 100, 1), round(pnl / n, 2), round(hold / n, 1), round(pnl, 1), _cagr, None, 0, n, None, None, 0, 0, 0, 0, 0, now)) written += 1 return written 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 # 普适（2026-08-17 重构，老莫）：跨独立年份的有效性，而非覆盖广度 # 核心：策略不是靠某次历史特例（如某次大反弹）才成立，而是多个独立时段都有效 _uniq_months = len({t.get("entry_date", "")[:7] for t in reg_trades if t.get("entry_date")}) _univ_score = _univ_years = _univ_valid = _univ_leave1 = 0 _ed_list = [t.get("entry_date", "")[:4] for t in reg_trades if t.get("entry_date")] if len(_ed_list) >= 5: _yr = defaultdict(list) for _y, _t in zip(_ed_list, reg_trades): _yr[_y].append(_t.get("profit_pct", 0)) _yearly = {_y: {"n": len(_v), "wr": sum(1 for p in _v if p > 0) / len(_v) * 100, "avg": sum(_v) / len(_v)} for _y, _v in _yr.items()} # 只统计有足够样本的年（>=5笔） _stat_years = {_y: _d for _y, _d in _yearly.items() if _d["n"] >= 5} if _stat_years: _univ_years = len(_stat_years) _univ_valid = sum(1 for _d in _stat_years.values() if _d["wr"] > 50) # 覆盖因子：>=3年给满覆盖分，<3年按比例 _coverage = min(_univ_years / 3.0, 1.0) # 稳定性：有效年占比 _stability = _univ_valid / _univ_years _univ_score = round(100 * _coverage * _stability) # 特例剔除：去掉最好一年的平均收益，剩余平均是否仍 > 0 _avgs = sorted((_d["avg"] for _d in _stat_years.values()), reverse=True) if len(_avgs) >= 2: _univ_leave1 = round(sum(_avgs[1:]) / (len(_avgs) - 1), 2) 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, universality_months, universality_years, universality_valid_years, universality_score, universality_leave1, 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"), _uniq_months, _univ_years, _univ_valid, _univ_score, _univ_leave1, 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: if pt in SLICE_DAYS_2: n = process_slice_period(conn, market, pt) else: 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()