feat: 温区表现按周期预计算脚本(regime_perf_by_period.py)——数据加工层落地,1y/2y/5y/10y各算一份,线性年化
This commit is contained in:
xxm
@@ -0,0 +1,192 @@
|
||||
#!/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")
|
||||
# 线性年化：收益 × (窗口总天数 / 该温区窗口内天数)
|
||||
rd = reg_days.get(reg, 0)
|
||||
cagr = round(ret * (total_days / rd), 1) if ret is not None and rd > 0 and total_days > 0 else 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()
|
||||
Reference in New Issue
Block a user