# -*- 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 evaluate_all_regimes(version, market="a", bench=None): """评估策略在【每个有数据温区】的资格（2026-08-16 老莫：策略可适应多个温区） 返回 {regime: {long_ok, mid_ok, short_ok, cagr_10y, cagr_2y, cagr_1y, trades_10y, trades_2y, trades_1y}} 只包含温区数据存在（有 10y 记录）的温区 """ bench = bench or get_benchmarks(market) if not bench: return {} out = {} try: conn = sqlite3.connect(str(DATA_DIR / "mofin.db"), timeout=5) conn.row_factory = sqlite3.Row # 该策略所有温区（有 10y 记录才算数） regimes = [r[0] for r in conn.execute( "SELECT DISTINCT regime FROM strategy_regime_perf_by_period " "WHERE strategy=? AND market=? AND period_tag='10y'", (version, market)).fetchall()] for rg in regimes: 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, rg, pt)).fetchone() cagr[pt] = (r["cagr_pct"], r["win_rate"], r["trades"]) if r else (None, None, 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)) out[rg] = { "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), } conn.close() except Exception: pass return out 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 _strategy_market(version): """探测策略市场：strategy_research.market（'hk'=港股，否则 A股）""" try: conn = sqlite3.connect(str(DATA_DIR / "mofin.db"), timeout=5) r = conn.execute( "SELECT market FROM strategy_research WHERE version=? AND COALESCE(market,'')!='' " "ORDER BY id DESC LIMIT 1", (version,)).fetchone() conn.close() if r and r[0] == "hk": return "hk" except Exception: pass return "a" def auto_init_availability(versions, bench=None, market_by_version=None): """新策略按标准 A 自动初始化可用性（一般默认可用；长期不合格则不可用） 已存在的手动状态不动。返回新增的默认可用策略。 market_by_version: dict {version: 'a'|'hk'}，缺省按 strategy_research.market 探测 """ bench = bench or get_benchmarks() avail = load_availability() changed = [] for v in versions: if v in avail: continue mkt = (market_by_version or {}).get(v) or _strategy_market(v) ev = evaluate_strategy(v, mkt, bench=bench) # 默认：长期合格即可用（A 标准）；长期不合格(10y跑不赢大盘)默认不可用 # 无评估数据（如港股策略在 hk 也无温区数据）→ 默认可用（不误杀新策略） default_avail = True if ev is None else bool(ev["long_ok"]) if ev is None: note = "自动初始化:无温区数据,默认可用" elif default_avail: note = "自动初始化(标准A)" else: note = "自动初始化:长期不合格(10y<大盘)" avail[v] = { "available": default_avail, "note": note, "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]