hmo/MoFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 普适指标重构——跨独立年份有效性(有效年占比×覆盖因子)+特例剔除检验(leave1),体现非特例宗旨

Browse Source
This commit is contained in:
xxm
2026-08-17 01:22:17 +08:00
parent a42a80827a
commit 4a78664f72
2 changed files with 43 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/profile-scripts/regime_perf_by_period.py
+32 -8
View File
@@ -57,7 +57,10 @@ def create_table(conn):
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)
)
@@ -160,24 +163,45 @@ def process_period(conn, market, period_tag):
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个月）
# 普适（2026-08-17 重构，老莫）：跨独立年份的有效性，而非覆盖广度
# 核心：策略不是靠某次历史特例（如某次大反弹）才成立，而是多个独立时段都有效
_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
_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:
from collections import defaultdict
_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_score, updated_at)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
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_score, now))
_uniq_months, _univ_years, _univ_valid, _univ_score, _univ_leave1, now))
written += 1
return written
server.py
+11 -5
View File
@@ -60,7 +60,8 @@ def _compute_regime_winrates_cached(pt, _approx_univ):
for _r in _c.execute(
"SELECT 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 "
"positions_taken, sharpe_ratio, profit_factor, universality_months, universality_years, "
"universality_valid_years, universality_score, universality_leave1 "
"FROM strategy_regime_perf_by_period "
"WHERE period_tag=? ORDER BY strategy, market, regime",
(_pt_use,)).fetchall():
@@ -73,7 +74,10 @@ def _compute_regime_winrates_cached(pt, _approx_univ):
_sh_v = _r[13]
_pf_v = _r[14]
_umon = _r[15]
_uscore = _r[16]
_uyears = _r[16]
_uvalid = _r[17]
_uscore = _r[18]
_uleave1 = _r[19]
_regime_winrates.setdefault(_ver, {})[_reg] = {
"trades": _r[4], "win_rate": _r[5], "avg_pnl": _r[6],
"avg_hold_days": _r[7],
@@ -86,7 +90,7 @@ def _compute_regime_winrates_cached(pt, _approx_univ):
"portfolio_max_dd_pct": _dd_v, "capital_final": _cf_v,
"positions_taken": _pt_v, "sharpe_ratio": _sh_v,
"profit_factor": _pf_v},
"universality": _approx_univ(_ver, _reg, _r[4], _umon, _uscore),
"universality": _approx_univ(_ver, _reg, _r[4], _umon, _uscore, _uyears, _uvalid, _uleave1),
}
_c.close()
except Exception:
@@ -628,11 +632,13 @@ def get_tracking():
@app.route("/api/research/strategies")
def api_research_strategies():
"""策略版本列表（含回测结果摘要，支持 period_tag 区间过滤）"""
def _approx_regime_universality(strategy, regime, trades, umon=None, uscore=None):
def _approx_regime_universality(strategy, regime, trades, umon=None, uscore=None, uyears=None, uvalid=None, uleave1=None):
"""温区级普适：优先用预计算真实值（trades entry_date 去重月份/温区总月份），
缺失时退回旧近似（信号数÷3估算，2026-08-16 修复——原估算对集中信号虚高）"""
if umon is not None and uscore is not None:
return {"months": umon, "score": uscore, "regime_total_months": 0}
return {"months": umon, "score": uscore, "years": uyears or 0,
"valid_years": uvalid or 0, "leave1_avg": uleave1,
"regime_total_months": 0}
try:
import sqlite3 as _sq6
_c6 = _sq6.connect(str(DATA_DIR / "mofin.db"), timeout=10)