210 lines
8.6 KiB
Python
210 lines
8.6 KiB
Python
#!/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_score 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
|
||
# 普适：该温区 trades 的 entry_date 去重月份数 / 该温区总月份数（2026-08-16 修复：
|
||
# 原 server 端信号数÷3估算 → s2_panic 2255信号估算751月=100分,实际只2个月）
|
||
_uniq_months = len({t.get("entry_date", "")[:7] for t in reg_trades if t.get("entry_date")})
|
||
_regime_months_all = {d[:7] for d, r in rmap.items() if r == reg}
|
||
_regime_total_months = len(_regime_months_all)
|
||
_univ_score = round(min(_uniq_months / max(_regime_total_months, 1) * 100, 100)) if _uniq_months else 0
|
||
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_score, 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_score, 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()
|