118 lines
5.1 KiB
Python
118 lines
5.1 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""择优激活规则 v2：修正上限逻辑——按质量排序选最优N个,家族去重,软上限
|
||
输出：每个温区应激活的策略（写入 strategy_weights.json 的 active）"""
|
||
import sys, json, sqlite3
|
||
sys.path.insert(0, "/home/hmo/MoFin")
|
||
sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
|
||
import strategy_qualify as sq
|
||
|
||
MIN_YEARLY = 15 # 年化成交下限（低于此=机会不足）
|
||
MAX_YEARLY = 300 # 年化成交上限（软上限，质量优先）
|
||
|
||
FAMILIES = {
|
||
"s2_panic": "s2", "s2_panic_v2": "s2", "s2_panic_v3": "s2",
|
||
"v_lurk_v1": "vlurk", "v_lurk_v2": "vlurk", "v_lurk_v3": "vlurk",
|
||
"v_mr": "vmr", "v_mr2": "vmr", "v_mr3": "vmr", "v_mr4": "vmr",
|
||
"v_oversold": "vover", "v_weak": "vweak",
|
||
"b_td1": "b_td", "b_td1_v2": "b_td", "b_td1_v3": "b_td",
|
||
"v1.0": "v1", "v2.0": "v2", "v3.0": "v3", "v_next": "vnext",
|
||
"hk_pe_mom": "hkpe", "hk_pe_oversold": "hkpe", "hk_mr1": "hkmr", "hk_mr2": "hkmr",
|
||
}
|
||
|
||
|
||
def efficiency_factor(sig, pos):
|
||
ratio = sig / pos if pos else 99
|
||
if ratio <= 2: return 1.0
|
||
if ratio <= 5: return 0.9
|
||
if ratio <= 10: return 0.75
|
||
return 0.5
|
||
|
||
|
||
def get_strategy_info(version, market, regime):
|
||
conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db", timeout=10)
|
||
r = conn.execute(
|
||
"SELECT trades, positions_taken, win_rate, sharpe_ratio, profit_factor, total_return_pct, "
|
||
"portfolio_max_dd_pct, universality_score, universality_years, universality_valid_years "
|
||
"FROM strategy_regime_perf_by_period WHERE strategy=? AND market=? AND regime=? AND period_tag='2y'",
|
||
(version, market, regime)).fetchone()
|
||
dep = conn.execute("SELECT MAX(deprecated) FROM strategy_research WHERE version=? AND deprecated IS NOT NULL AND deprecated!=''", (version,)).fetchone()[0]
|
||
conn.close()
|
||
if not r:
|
||
return None
|
||
sig, pos, wr, sh, pf, ret, dd, univ, uyears, uvalid = r
|
||
pos = pos or 0
|
||
ret_c = min(ret or 0, 100) / 100 * 30
|
||
wr_c = (wr or 0) / 100 * 20
|
||
sh_c = min(max(sh or 0, 0), 20) / 20 * 20
|
||
pf_c = min(pf or 0, 5) / 5 * 15
|
||
dd_c = (1 - min(dd or 0, 50) / 50) * 15
|
||
n = sig or 0
|
||
conf = min(1, n / 40)
|
||
eff = efficiency_factor(sig, pos)
|
||
comp = round((ret_c + wr_c + sh_c + pf_c + dd_c) * conf * eff)
|
||
return {"sig": sig, "pos": pos, "comp": comp, "univ": univ or 0,
|
||
"uyears": uyears or 0, "uvalid": uvalid or 0, "dep": dep}
|
||
|
||
|
||
def select_best(regime, market='a'):
|
||
conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db", timeout=10)
|
||
rows = conn.execute("SELECT DISTINCT strategy FROM strategy_regime_perf_by_period WHERE market=? AND regime=?", (market, regime)).fetchall()
|
||
conn.close()
|
||
avail = sq.load_availability()
|
||
candidates = []
|
||
for (version,) in rows:
|
||
info = get_strategy_info(version, market, regime)
|
||
if not info or info["dep"]:
|
||
continue
|
||
if not avail.get(version, {}).get("available", False):
|
||
continue
|
||
try:
|
||
q = sq.evaluate_all_regimes(version, market=market)
|
||
qr = q.get(regime, {})
|
||
if not (qr.get("long_ok") and qr.get("mid_ok") and qr.get("short_ok")):
|
||
continue
|
||
except Exception:
|
||
continue
|
||
# 质量分 = 综合分 × 普适（有效年占比越高越好）
|
||
univ_ratio = (info["uvalid"] / info["uyears"]) if info["uyears"] else 0
|
||
quality = info["comp"] * (0.5 + 0.5 * univ_ratio)
|
||
candidates.append({"version": version, "info": info, "quality": round(quality, 1),
|
||
"family": FAMILIES.get(version, version)})
|
||
candidates.sort(key=lambda x: -x["quality"])
|
||
|
||
# 择优：家族去重 + 软上限
|
||
selected = []
|
||
used_fam = set()
|
||
total_yearly = 0
|
||
for c in candidates:
|
||
if c["family"] in used_fam:
|
||
continue
|
||
# 出手次数太少的不选（年化 < MIN_YEARLY/2）
|
||
yearly = c["info"]["pos"] / 2 # 2y→年化
|
||
if yearly < 5:
|
||
continue
|
||
if total_yearly + yearly > MAX_YEARLY and selected:
|
||
break # 超上限停止
|
||
used_fam.add(c["family"])
|
||
selected.append(c)
|
||
total_yearly += yearly
|
||
return selected, total_yearly
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print("# 择优激活建议（质量分=综合分×普适有效年占比）")
|
||
all_sel = {}
|
||
for market in ["a", "hk"]:
|
||
for regime in ["trend_down", "choppy", "trend_up"]:
|
||
sel, total = select_best(regime, market)
|
||
if sel:
|
||
print(f"\n## {market} {regime}: 年化总出手≈{total:.0f}")
|
||
for c in sel:
|
||
i = c["info"]
|
||
print(f" {c['version']:18} 质量{c['quality']:5.1f} 综合{i['comp']:4} 信号{i['sig']:5} 成交{i['pos']:4} 普适{i['univ']:4.0f}({i['uvalid']}/{i['uyears']}年)")
|
||
all_sel[(market, regime)] = [c["version"] for c in sel]
|
||
# 保存建议
|
||
with open("/tmp/activation_suggestion.json", "w") as f:
|
||
json.dump({f"{m}:{r}": v for (m, r), v in all_sel.items()}, f, ensure_ascii=False, indent=1)
|
||
print("\n# 建议已存 /tmp/activation_suggestion.json")
|