63 lines
2.7 KiB
Python
63 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
||
"""health_monitor_daily.py — 策略健康度每日监控（v_weak + v_oversold）
|
||
|
||
背景（2026-08-12 事项四：健康监控注册 p_oversold）：
|
||
evolution/health_monitor.py 能算策略健康度（实盘近7天浮盈 vs 回测基线5y 偏差 + 写 strategy_health 表 + 告警），
|
||
但**不在 cron**——策略健康没有自动写。本脚本每天定时跑，对当前策略（v_weak 实盘 + v_oversold 新策略）
|
||
分别做健康检查，写 strategy_health 表（供进化模块 dashboard 展示健康度趋势）。
|
||
|
||
输出：strategy_health 表（strategy_version/date/live_trades/live_wins/live_return_pct/backtest_wr/backtest_avg_ret/deviation/health_score）
|
||
调度：每日收盘后（45 16 * * 1-5，实盘当日浮盈定型后）
|
||
规范：单例守卫(5.3) + 各策略独立健康分（backtest 基线用各自回测 5y）
|
||
"""
|
||
import sys, os, sqlite3, fcntl
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
|
||
sys.path.insert(0, "/home/hmo/MoFin")
|
||
sys.path.insert(0, "/home/hmo/MoFin/evolution")
|
||
sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
|
||
|
||
|
||
def _singleton_guard(tag="health_monitor_daily.py"):
|
||
lock_dir = Path("/tmp/mofin_locks")
|
||
lock_dir.mkdir(exist_ok=True)
|
||
try:
|
||
fd = os.open(str(lock_dir / f"{tag}.lock"), os.O_CREAT | os.O_RDWR)
|
||
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
return fd
|
||
except OSError:
|
||
print(f"[{tag}] 已有实例在运行，退出", flush=True)
|
||
sys.exit(0)
|
||
|
||
|
||
def main():
|
||
_fd = _singleton_guard()
|
||
print(f"[health_monitor_daily] {datetime.now().strftime('%H:%M:%S')} 策略健康度监控开始", flush=True)
|
||
try:
|
||
from evolution.health_monitor import run_health_check
|
||
except Exception:
|
||
# evolution 包导入兜底（直接按路径加载）
|
||
import importlib.util as _ilu
|
||
_spec = _ilu.spec_from_file_location("health_monitor", "/home/hmo/MoFin/evolution/health_monitor.py")
|
||
_m = _ilu.module_from_spec(_spec)
|
||
_spec.loader.exec_module(_m)
|
||
run_health_check = _m.run_health_check
|
||
|
||
# 当前策略：v_weak（实盘在跑）+ v_oversold（新策略上线）
|
||
results = {}
|
||
for version in ("v_weak", "v_oversold"):
|
||
try:
|
||
r = run_health_check(strategy_version=version)
|
||
results[version] = r
|
||
if r and r.get("alert"):
|
||
print(f" ⚠️ {version}: {r['alert']}", flush=True)
|
||
except Exception as e:
|
||
print(f" ❌ {version} 健康检查失败: {e}", flush=True)
|
||
ok = sum(1 for r in results.values() if r)
|
||
print(f"[health_monitor_daily] 完成: {ok}/{len(results)} 策略健康分已写入 strategy_health", flush=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|