101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""regime_gate.py — 温区门控共享工具（2026-08-13）
|
||
|
||
scanner/重评统一调用：读取当前温区（优先平滑 K=5，回退原始 market_regime），
|
||
判断某策略是否在当前温区激活。
|
||
|
||
用法：
|
||
from regime_gate import get_current_regime, is_strategy_active, strategy_enabled
|
||
rg = get_current_regime() # {"regime": "trend_down", "date": "..."}
|
||
ok = is_strategy_active("v_weak") # 该策略是否当前温区激活（读 strategy_weights.json）
|
||
"""
|
||
import json
|
||
import sqlite3
|
||
from pathlib import Path
|
||
|
||
MOFIN_DATA = "/home/hmo/MoFin/data"
|
||
WEIGHTS_FILE = Path(MOFIN_DATA) / "strategy_weights.json"
|
||
SMOOTHED_FILE = Path(MOFIN_DATA) / "market_regime_smoothed.json"
|
||
DB = Path(MOFIN_DATA) / "mofin.db"
|
||
|
||
_cache_regime = None
|
||
_cache_weights = None
|
||
|
||
|
||
def get_current_regime(use_smoothed=True):
|
||
"""读取当前温区。优先平滑(K=5)，回退原始 market_regime。返回 {"regime","date"}"""
|
||
global _cache_regime
|
||
if _cache_regime:
|
||
return _cache_regime
|
||
# 1. 平滑温区（regime_tracker K=5）
|
||
if use_smoothed:
|
||
try:
|
||
if SMOOTHED_FILE.exists():
|
||
d = json.loads(SMOOTHED_FILE.read_text(encoding="utf-8"))
|
||
_cache_regime = {
|
||
"regime": d.get("current_regime", "unknown"),
|
||
"date": d.get("current_date", ""),
|
||
}
|
||
return _cache_regime
|
||
except Exception:
|
||
pass
|
||
# 2. 原始 market_regime 表
|
||
try:
|
||
conn = sqlite3.connect(str(DB), timeout=30)
|
||
conn.execute("PRAGMA busy_timeout=30000")
|
||
row = conn.execute(
|
||
"SELECT date, regime FROM market_regime ORDER BY date DESC LIMIT 1"
|
||
).fetchone()
|
||
conn.close()
|
||
if row:
|
||
_cache_regime = {"regime": row[1], "date": row[0]}
|
||
return _cache_regime
|
||
except Exception:
|
||
pass
|
||
return {"regime": "unknown", "date": ""}
|
||
|
||
|
||
def _load_weights():
|
||
"""读取 strategy_weights.json（缓存）"""
|
||
global _cache_weights
|
||
if _cache_weights is not None:
|
||
return _cache_weights
|
||
try:
|
||
if WEIGHTS_FILE.exists():
|
||
_cache_weights = json.loads(WEIGHTS_FILE.read_text(encoding="utf-8"))
|
||
return _cache_weights
|
||
except Exception:
|
||
pass
|
||
_cache_weights = {}
|
||
return _cache_weights
|
||
|
||
|
||
def is_strategy_active(strategy_name):
|
||
"""该策略是否在当前温区激活（matched）。无数据默认激活"""
|
||
w = _load_weights()
|
||
if not w or not w.get("weights"):
|
||
return True
|
||
entry = w["weights"].get(strategy_name)
|
||
if entry is None:
|
||
return True # 不在权重表 → 默认激活（持仓管理类）
|
||
return entry.get("matched", True)
|
||
|
||
|
||
def strategy_enabled(strategy_name):
|
||
"""该策略权重>0（激活 + 非0乘数）"""
|
||
w = _load_weights()
|
||
if not w or not w.get("weights"):
|
||
return True
|
||
entry = w["weights"].get(strategy_name)
|
||
if entry is None:
|
||
return True
|
||
return entry.get("weight", 0) > 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
rg = get_current_regime()
|
||
print(f"当前温区: {rg['regime']} ({rg['date']})")
|
||
for s in ["v_weak", "v_oversold", "v_next4", "s2_panic", "v_mr_sel"]:
|
||
print(f" {s}: active={is_strategy_active(s)} enabled={strategy_enabled(s)}")
|