314 lines
14 KiB
Python
314 lines
14 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_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()
|