feat: 资格评估数据层(strategy_qualify)——标准A(长期10y/近期2y/当下1y适应温区年化>大盘)+手动可用性json管理

2026-08-16 07:32:07 +08:00
parent cda582f046
commit 7e372a7655
1 changed files with 151 additions and 0 deletions
@@ -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]