152 lines
5.8 KiB
Python
152 lines
5.8 KiB
Python
# -*- 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]
|