feat: 温区自适应前端横幅+router v5+overview API——概览页置顶显示温区/温度/激活策略, server.py /api/overview 返回 regime_weights, strategy_router v5 动态策略库

deploy/profile-scripts/strategy_router.py

@@ -1,20 +1,18 @@
-#!/usr/bin/env python3
+#!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-"""strategy_router.py v2 — MoFin 策略动态路由（三态 + 温度维度，2026-08-13 重写）
+"""strategy_router.py v5 — MoFin 策略动态路由（动态策略库，2026-08-13）
 
-结论（数据验证）：三态骨架 + 温度(rsi)连续维度。
-- 三态（trend_up/choppy/trend_down）→ 选策略类型（market_regime.py 判定）
-- 温度（rsi 档位）→ 仓位乘数（temp_band.py 判定，连续量）
+核心（老莫原则）：
+- 策略全温区发信号（不留温区门控在策略内）
+- 策略-温区表现常态化记录（regime_perf.py / strategy_regime_perf），适用温区动态化
+- 组合按当前温区自动激活/休眠策略：当前温区实测胜率高的策略优先
+- 不漏历史策略（v_mr_sel trend_down 93%、v7.3 trend_up 94% 等被埋没的强者）
 
-策略-适用温度映射（数据来源：2026-08-13 分市场状态检验，见 docs/decisions/2026-08-13-事项五）：
-  v_weak     → choppy × [fear, neutral]    （震荡市+恐慌最佳 71%）
-  v_oversold → trend_down × [panic, fear]  （下跌市+恐慌最佳 64-86%）
-  v_next4    → trend_up × [greed, euphoria]（趋势市 63%）
-  v8.1       → trend_up                    （趋势市 57%）
-  s2_panic   → trend_down × [panic]        （极端恐慌 86%）
-  v_lurk_v3  → [choppy, trend_down] × [fear, neutral]（77%/71%）
-
-输出：strategy_weights.json（各策略 weight 由 状态×温度 决定）
+数据流：
+  market_regime(三态) + regime_tracker(平滑K=5) → 当前温区
+  temp_band(rsi) → 温度/仓位乘数
+  regime_perf(实测) → 各策略温区表现
+  → strategy_weights.json（各策略权重/激活状态）
 """
 import json
 import sys
@@ -27,121 +25,124 @@ sys.path.insert(0, "/home/hmo/MoFin")
 
 OUT = Path("/home/hmo/MoFin/data/strategy_weights.json")
 
+DEFAULT_FAMILY = "mr"
+FAMILY_MAP = {
+    "v_weak": "mr", "v_oversold": "mr", "v_mr": "mr", "v_mr2": "mr", "v_mr3": "mr",
+    "v_mr4": "mr", "v_mr_sel": "mr", "v_lurk_v1": "mr", "v_lurk_v2": "mr", "v_lurk_v3": "mr",
+    "s2_panic": "mr", "v_osc": "mr",
+    "v_next": "trend", "v_next3": "trend", "v_next4": "trend", "v_next5": "trend",
+    "v8.0": "trend", "v8.1": "trend", "v8.2": "trend", "v8.3": "trend",
+    "v7.0": "trend", "v7.1": "trend", "v7.1b": "trend", "v7.2": "trend", "v7.3": "trend",
+    "v6.0": "trend", "v6.1": "trend", "v6.2": "trend",
+    "v9.0": "trend", "v9.1": "trend", "v9.2": "trend",
+    "v_combo": "trend", "v1.0": "trend", "v2.0": "trend", "v3.0": "trend",
+    "v4.0": "trend", "v4.0a": "trend", "v4.0b": "trend", "v4.0c": "trend",
+    "v4.0d": "trend", "v4.0e": "trend", "v4.0f": "trend", "v4.0g": "trend",
+    "v4.0h": "trend", "v4.1": "trend", "v5.0": "trend",
+    "v11.0": "trend", "v11.1": "trend",
+}
+
 
 def load_regime():
-    """读取平滑温区（regime_tracker 输出，K=5 滞回确认）。
-    优先用平滑结果（避免单日噪音误判）；regime_tracker 不可用时回退原始 market_regime。"""
     try:
         p = Path("/home/hmo/MoFin/data/market_regime_smoothed.json")
         if p.exists():
             d = json.loads(p.read_text(encoding="utf-8"))
             return {"regime": d.get("current_regime", "unknown"),
-                    "date": d.get("current_date", ""),
-                    "temp_band": (d.get("temp") or {}).get("band"),
-                    "temp_rsi": (d.get("temp") or {}).get("rsi")}
+                    "date": d.get("current_date", "")}
     except Exception:
         pass
     try:
         from market_regime import load_market_regime
         return load_market_regime()
     except Exception:
-        return None
+        return {"regime": "unknown"}
 
 
-def get_temp():
-    """读取市场温度（rsi 档位）。regime_tracker 已含温度则直接用，否则实时算。"""
+def load_temp():
     try:
         from temp_band import get_market_temp
         return get_market_temp()
-    except Exception as e:
-        print(f"  [router] 温度获取失败: {e}", file=sys.stderr)
+    except Exception:
         return {"band": "unknown", "rsi": None}
 
 
-# 策略-适用温度映射（基础数据，数据驱动固化）
-# weight = 状态匹配度 × 温度乘数
-STRATEGY_TEMP = {
-    "v_weak": {
-        "family": "mr",
-        "regimes": ["choppy"],
-        "desc": "震荡市超跌反弹（choppy+恐慌71%最佳）",
-    },
-    "v_oversold": {
-        "family": "mr",
-        "regimes": ["trend_down"],
-        "desc": "下跌市超跌（trend_down+恐慌64-86%）",
-    },
-    "v_next4": {
-        "family": "trend",
-        "regimes": ["trend_up"],
-        "desc": "趋势市追涨（trend_up 63%）",
-    },
-    "v8.1": {
-        "family": "trend",
-        "regimes": ["trend_up"],
-        "desc": "波段先出再进（trend_up 57%）",
-    },
-    "s2_panic": {
-        "family": "mr",
-        "regimes": ["trend_down"],
-        "desc": "极端恐慌买超跌（panic 86%）",
-    },
-    "v_lurk_v3": {
-        "family": "mr",
-        "regimes": ["choppy", "trend_down"],
-        "desc": "潜伏型（choppy/trend_down 71-77%）",
-    },
-}
+def load_regime_perf():
+    conn = None
+    try:
+        import sqlite3
+        conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db", timeout=30)
+        conn.execute("PRAGMA busy_timeout=30000")
+        rows = conn.execute(
+            "SELECT strategy, regime, trades, win_rate, avg_pnl FROM strategy_regime_perf"
+        ).fetchall()
+        result = {}
+        for strat, reg, n, wr, pnl in rows:
+            result.setdefault(strat, {})[reg] = {"trades": n, "win_rate": wr, "avg_pnl": pnl}
+        return result
+    except Exception:
+        return {}
+    finally:
+        if conn:
+            conn.close()
 
 
-def route(regime, temp):
-    """三态 + 温度 → 各策略权重"""
-    current_regime = regime.get("regime", "unknown") if regime else "unknown"
+def route(regime, temp, perf):
+    current_regime = regime.get("regime", "unknown")
     band = temp.get("band", "unknown")
-    family = temp.get("family", "mr")
-
     weights = {}
-    for name, cfg in STRATEGY_TEMP.items():
-        # 状态匹配度：适用温区包含当前状态 → 1.0，否则 0.3（保留观察）
-        matched = current_regime in cfg["regimes"]
+    for name, regs in perf.items():
+        if not regs:
+            continue
+        cur = regs.get(current_regime)
+        best = max(regs.items(), key=lambda x: x[1].get("win_rate", 0))
+        best_wr = best[1].get("win_rate", 0)
+        cur_wr = cur.get("win_rate") if cur else None
+        matched = (best_wr >= 50) and (cur_wr is not None and cur_wr >= 50)
+        family = FAMILY_MAP.get(name, DEFAULT_FAMILY)
         base = 1.0 if matched else 0.3
-        # 温度乘数（按策略家族）
         try:
             from temp_band import temp_multiplier
-            mult = temp_multiplier(band, cfg["family"])
+            mult = temp_multiplier(band, family)
         except Exception:
             mult = 0.8
         weights[name] = {
             "weight": round(base * mult, 2),
             "regime": current_regime,
             "temp_band": band,
-            "family": cfg["family"],
+            "family": family,
             "matched": matched,
-            "desc": cfg["desc"],
+            "best_regime": best[0],
+            "best_win_rate": best[1].get("win_rate", 0),
+            "cur_win_rate": cur.get("win_rate") if cur else None,
+            "cur_avg_pnl": cur.get("avg_pnl") if cur else None,
+            "trades": cur.get("trades") if cur else 0,
         }
-    return weights
+    return dict(sorted(weights.items(), key=lambda x: -x[1]["weight"]))
 
 
 def main():
     regime = load_regime()
-    temp = get_temp()
-    if not regime:
-        print("  [router] market_regime 无数据", file=sys.stderr)
-        regime = {"regime": "unknown"}
-
-    weights = route(regime, temp)
+    temp = load_temp()
+    perf = load_regime_perf()
+    weights = route(regime, temp, perf)
     out = {
         "state": regime.get("regime", "unknown"),
+        "state_date": regime.get("date", ""),
         "temp_band": temp.get("band", "unknown"),
         "temp_rsi": temp.get("rsi"),
         "weights": weights,
+        "active": [k for k, v in weights.items() if v["matched"]],
         "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
-        "note": "三态(trend_up/choppy/trend_down)选策略类型 + 温度(rsi)乘数决定仓位",
+        "note": "温区(平滑K=5)选策略类型 + 温度(rsi)乘数；策略全温区发信号，适用温区由实测动态决定",
     }
     OUT.write_text(json.dumps(out, ensure_ascii=False, indent=1), encoding="utf-8")
     print(f"strategy_weights.json: 状态={out['state']} 温度={out['temp_band']}(rsi={out['temp_rsi']})")
-    for name, w in weights.items():
-        print(f"  {name:<12} weight={w['weight']:<5} regime={w['regime']:<10} band={w['temp_band']:<8} {'✓' if w['matched'] else '观察'}")
+    print(f"激活策略: {out['active']}")
+    print("权重排名（前15）:")
+    for name, w in list(weights.items())[:15]:
+        mark = "✓" if w["matched"] else "观察"
+        print(f"  {name:<12} w={w['weight']:<5} {w['best_regime']:<11} 最佳{w['best_win_rate']:>3.0f}% {mark}")
 
 
 if __name__ == "__main__":