95 lines
4.3 KiB
Python
95 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""strategy_alert.py — MoFin 策略失效预警（三振出局，2026-08-13 落地）
|
|
|
|
核心理念（机器学习策略攻防体系文）：
|
|
- 三振出局：黄牌（减半）→ 橙牌（1/4）→ 红牌（清仓淘汰）
|
|
- 五类失效预警：胜率持续下降 / 盈亏比恶化 / 波动新高 / 信号质量突变 / 风格漂移
|
|
- 原则：不是 IC 掉到负就删，是降权到观察模式（权重设 0 保留，恢复可启用）
|
|
|
|
数据源：strategy_health（已有）+ strategy_research（回测）+ 实盘持仓表现
|
|
"""
|
|
import sqlite3
|
|
import json
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
|
|
# ── 消息通道统一路由(broadcast/xmpp by delivery) ──
|
|
try:
|
|
from messenger import install_stdio_hook as _msh
|
|
_msh()
|
|
except Exception:
|
|
pass
|
|
|
|
DB = "/home/hmo/MoFin/data/mofin.db"
|
|
OUT = Path("/home/hmo/MoFin/data/strategy_alerts.json")
|
|
|
|
# 三振阈值（机器学习策略文）：
|
|
# 黄牌：滚动20日风险调整收益连续3日 < -0.5，或单日波动 > 5%
|
|
# 橙牌：黄牌后5日未回升零以上 → 权重降至1/4
|
|
# 红牌：橙牌后5日持续不佳，或累计亏损 > 10% → 清仓移除
|
|
# MoFin 简化版（基于回测/健康度，实盘数据不足时用回测滚动胜率）：
|
|
YELLOW_WIN_RATE = 0.40 # 滚动胜率 < 40% → 黄牌（v_weak 近1年 39.3% 已触发）
|
|
ORANGE_WIN_RATE = 0.35 # 滚动胜率 < 35% → 橙牌
|
|
RED_WIN_RATE = 0.30 # 滚动胜率 < 30% → 红牌（淘汰）
|
|
ORANGE_PNL_RATIO = 0.5 # 盈亏比 < 0.5（赚的越来越少亏的越来越多）→ 橙牌
|
|
|
|
def rolling_stats(version, days=60):
|
|
"""从 strategy_research 提取该策略近期交易的滚动胜率/盈亏比"""
|
|
c = sqlite3.connect(DB)
|
|
rows = c.execute(
|
|
"SELECT results_json FROM strategy_research WHERE version=? ORDER BY period_tag DESC LIMIT 1",
|
|
(version,)
|
|
).fetchall()
|
|
c.close()
|
|
if not rows:
|
|
return None
|
|
try:
|
|
res = json.loads(rows[0][0])
|
|
trades = res.get("trades", [])
|
|
cutoff = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
|
|
recent = [t for t in trades if t.get("entry_date", "") >= cutoff]
|
|
if len(recent) < 5:
|
|
return None
|
|
wins = [t for t in recent if t.get("profit_pct", 0) > 0]
|
|
losses = [t for t in recent if t.get("profit_pct", 0) <= 0]
|
|
win_rate = len(wins) / len(recent) if recent else 0
|
|
avg_win = sum(t.get("profit_pct", 0) for t in wins) / len(wins) if wins else 0
|
|
avg_loss = abs(sum(t.get("profit_pct", 0) for t in losses) / len(losses)) if losses else 1
|
|
pnl_ratio = avg_win / avg_loss if avg_loss > 0 else 0
|
|
return {
|
|
"n": len(recent), "win_rate": win_rate,
|
|
"avg_win": avg_win, "avg_loss": avg_loss, "pnl_ratio": pnl_ratio,
|
|
}
|
|
except Exception:
|
|
return None
|
|
|
|
def assess(version, stats):
|
|
"""三振评估"""
|
|
if not stats:
|
|
return {"level": "ok", "action": "数据不足，无法评估", "weight": 1.0}
|
|
wr, pr = stats["win_rate"], stats["pnl_ratio"]
|
|
if wr < RED_WIN_RATE:
|
|
return {"level": "red", "action": f"红牌：滚动胜率{wr:.1%}<30%，淘汰（权重0，观察模式）", "weight": 0.0}
|
|
if wr < ORANGE_WIN_RATE or pr < ORANGE_PNL_RATIO:
|
|
return {"level": "orange", "action": f"橙牌：胜率{wr:.1%}或盈亏比{pr:.2f}恶化，权重降1/4", "weight": 0.25}
|
|
if wr < YELLOW_WIN_RATE:
|
|
return {"level": "yellow", "action": f"黄牌：滚动胜率{wr:.1%}<40%，权重减半", "weight": 0.5}
|
|
return {"level": "ok", "action": f"正常：胜率{wr:.1%}，盈亏比{pr:.2f}", "weight": 1.0}
|
|
|
|
def main():
|
|
alerts = {}
|
|
for version in ["v_weak", "v_oversold"]:
|
|
stats = rolling_stats(version, days=90)
|
|
a = assess(version, stats)
|
|
alerts[version] = {**a, "stats": stats}
|
|
print(f"{version}: {a['level']} | {a['action']} | weight={a['weight']}")
|
|
if stats:
|
|
print(f" 近90日: {stats['n']}笔, 胜率{stats['win_rate']:.1%}, 盈亏比{stats['pnl_ratio']:.2f}")
|
|
alerts["updated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
OUT.write_text(json.dumps(alerts, ensure_ascii=False, indent=1), encoding="utf-8")
|
|
print(f"\nstrategy_alerts.json 写入")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|