138 lines
4.6 KiB
Python
138 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""strategy_router.py v2 — MoFin 策略动态路由（三态 + 温度维度，2026-08-13 重写）
|
||
|
||
结论（数据验证）：三态骨架 + 温度(rsi)连续维度。
|
||
- 三态（trend_up/choppy/trend_down）→ 选策略类型（market_regime.py 判定）
|
||
- 温度（rsi 档位）→ 仓位乘数（temp_band.py 判定，连续量）
|
||
|
||
策略-适用温度映射（数据来源：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 由 状态×温度 决定）
|
||
"""
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
|
||
_SCRIPT_DIR = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(_SCRIPT_DIR))
|
||
sys.path.insert(0, "/home/hmo/MoFin")
|
||
|
||
OUT = Path("/home/hmo/MoFin/data/strategy_weights.json")
|
||
|
||
|
||
def load_regime():
|
||
"""读取最新 market_regime（三态）"""
|
||
try:
|
||
from market_regime import load_market_regime
|
||
return load_market_regime()
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def get_temp():
|
||
"""读取市场温度（rsi 档位）"""
|
||
try:
|
||
from temp_band import get_market_temp, temp_multiplier
|
||
return get_market_temp()
|
||
except Exception as e:
|
||
print(f" [router] 温度获取失败: {e}", file=sys.stderr)
|
||
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 route(regime, temp):
|
||
"""三态 + 温度 → 各策略权重"""
|
||
current_regime = regime.get("regime", "unknown") if regime else "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"]
|
||
base = 1.0 if matched else 0.3
|
||
# 温度乘数（按策略家族）
|
||
try:
|
||
from temp_band import temp_multiplier
|
||
mult = temp_multiplier(band, cfg["family"])
|
||
except Exception:
|
||
mult = 0.8
|
||
weights[name] = {
|
||
"weight": round(base * mult, 2),
|
||
"regime": current_regime,
|
||
"temp_band": band,
|
||
"family": cfg["family"],
|
||
"matched": matched,
|
||
"desc": cfg["desc"],
|
||
}
|
||
return weights
|
||
|
||
|
||
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)
|
||
out = {
|
||
"state": regime.get("regime", "unknown"),
|
||
"temp_band": temp.get("band", "unknown"),
|
||
"temp_rsi": temp.get("rsi"),
|
||
"weights": weights,
|
||
"updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"note": "三态(trend_up/choppy/trend_down)选策略类型 + 温度(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 '观察'}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|