diff --git a/deploy/profile-scripts/strategy_qualify.py b/deploy/profile-scripts/strategy_qualify.py new file mode 100644 index 00000000..edd87b2b --- /dev/null +++ b/deploy/profile-scripts/strategy_qualify.py @@ -0,0 +1,151 @@ +# -*- coding: utf-8 -*- +"""2026-08-16 资格评估与可用性管理(数据层) +标准 A(老莫确认): + 长期合格 = 适应温区 10y 年化 > 大盘10y + 近期不错 = 适应温区 2y 年化 > 大盘2y + 当下能打 = 适应温区 1y 年化 > 大盘1y + 三项全过 → 激活资格;长期+近期 → 可用;长期不过 → 历史 +可用性(manual_available)由老莫手动把关:激活只能是可用的策略。 +新策略按默认标准(A)初始化可用性;定期评估自动标记「可用性下降/回升」警告。 + +存储:data/strategy_availability.json(git 可追踪,同 strategy_weights.json 模式) +""" +import json +import os +import sqlite3 +from datetime import datetime, timedelta +from pathlib import Path + +DATA_DIR = Path("/home/hmo/MoFin/data") +AVAIL_JSON = DATA_DIR / "strategy_availability.json" + +# 大盘基准(market_regime 指数年化,按需现算) +BENCH_DAYS = {"1y": 365, "2y": 730, "10y": 3650} + + +def get_benchmarks(market="a"): + """大盘年化基准:1y/2y/10y""" + out = {} + try: + conn = sqlite3.connect(str(DATA_DIR / "mofin.db"), timeout=5) + for label, days in BENCH_DAYS.items(): + cutoff = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d") + last = conn.execute( + "SELECT close FROM market_regime WHERE market=? AND close IS NOT NULL ORDER BY date DESC LIMIT 1", + (market,)).fetchone() + base = conn.execute( + "SELECT close FROM market_regime WHERE market=? AND close IS NOT NULL AND date>=? ORDER BY date LIMIT 1", + (market, cutoff)).fetchone() + if last and base and base[0]: + pct = (last[0] / base[0] - 1) * 100 + yrs = days / 365.0 + if pct > -100: + out[label] = round(((1 + pct / 100) ** (1 / yrs) - 1) * 100, 1) + conn.close() + except Exception: + pass + return out + + +def evaluate_strategy(version, market="a", best_regime=None, bench=None): + """评估单策略资格:长期/近期/当下 三项达标 + 返回 {long_ok, mid_ok, short_ok, cagr_10y, cagr_2y, cagr_1y, regime} + """ + bench = bench or get_benchmarks(market) + if not bench: + return None + try: + conn = sqlite3.connect(str(DATA_DIR / "mofin.db"), timeout=5) + conn.row_factory = sqlite3.Row + # 适应温区:best_regime 传入或用 10y 最高胜率温区 + if not best_regime: + rows = conn.execute( + "SELECT regime, win_rate FROM strategy_regime_perf_by_period " + "WHERE strategy=? AND market=? AND period_tag='10y'", + (version, market)).fetchall() + if not rows: + conn.close() + return None + best_regime = max(rows, key=lambda x: x[1] or 0)[0] + # 各周期适应温区年化 + cagr = {} + for pt in ["1y", "2y", "10y"]: + r = conn.execute( + "SELECT cagr_pct, win_rate, trades FROM strategy_regime_perf_by_period " + "WHERE strategy=? AND market=? AND regime=? AND period_tag=?", + (version, market, best_regime, pt)).fetchone() + cagr[pt] = (r["cagr_pct"], r["win_rate"], r["trades"]) if r else (None, None, None) + conn.close() + except Exception: + return None + c10, w10, t10 = cagr.get("10y", (None, None, None)) + c2, w2, t2 = cagr.get("2y", (None, None, None)) + c1, w1, t1 = cagr.get("1y", (None, None, None)) + return { + "regime": best_regime, + "cagr_10y": c10, "cagr_2y": c2, "cagr_1y": c1, + "trades_10y": t10, "trades_2y": t2, "trades_1y": t1, + "long_ok": c10 is not None and c10 > bench.get("10y", 0), + "mid_ok": c2 is not None and c2 > bench.get("2y", 0), + "short_ok": c1 is not None and c1 > bench.get("1y", 0), + } + + +def load_availability(): + """读手动可用性状态 {version: {available: bool, note, updated_at}}""" + try: + if AVAIL_JSON.exists(): + return json.loads(AVAIL_JSON.read_text(encoding="utf-8")) + except Exception: + pass + return {} + + +def save_availability(data): + """写手动可用性状态""" + AVAIL_JSON.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8") + + +def auto_init_availability(versions, bench=None): + """新策略按标准 A 自动初始化可用性(一般默认可用;长期不合格则不可用) + 已存在的手动状态不动。返回新增的默认可用策略。 + """ + bench = bench or get_benchmarks() + avail = load_availability() + changed = [] + for v in versions: + if v in avail: + continue + ev = evaluate_strategy(v, "a", bench=bench) + # 默认:长期合格即可用(A 标准);长期不合格(10y跑不赢大盘)默认不可用 + default_avail = bool(ev and ev["long_ok"]) + avail[v] = { + "available": default_avail, + "note": "自动初始化(标准A)" if default_avail else "自动初始化:长期不合格(10y<大盘)", + "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + } + changed.append((v, default_avail)) + if changed: + save_availability(avail) + return changed + + +def is_available(version): + """查询手动可用性(默认 true——新策略默认可用,除非明确标不可用)""" + avail = load_availability() + entry = avail.get(version) + if entry is None: + return True # 未标记 = 默认可用 + return bool(entry.get("available", True)) + + +def set_available(version, available, note=""): + """手动设置可用性""" + avail = load_availability() + avail[version] = { + "available": bool(available), + "note": note or ("手动可用" if available else "手动不可用"), + "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + } + save_availability(avail) + return avail[version]