知微
7c255d5234
mo_data.py DB_PATH: /home/hmo/web-dashboard/data/mofin.db → /home/hmo/MoFin/data/mofin.db 新增 verify_reassess_pipeline.py：管道审计，静默时无声、异常时报警
119 lines
4.2 KiB
Python
119 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""verify_reassess_pipeline.py — 重评推送管道审计
|
|
|
|
检查三个代码级约束是否在正常运转：
|
|
1. price_monitor 每2分触发 → 突破检测 → reassess调用
|
|
2. stale_detector/stale_push_wlin 自选买入区检测
|
|
3. mofin_collect LLM前强制重评
|
|
|
|
输出：JSON格式管道状态（无异常时静默[SILENT]）
|
|
"""
|
|
import json, os, sys, subprocess
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
|
|
BASE = Path(__file__).parent.parent
|
|
sys.path.insert(0, str(BASE))
|
|
sys.path.insert(0, "/home/hmo/MoFin")
|
|
|
|
try:
|
|
from mofin_db import get_conn
|
|
conn = get_conn()
|
|
cur = conn.cursor()
|
|
except Exception:
|
|
conn = None
|
|
|
|
now = datetime.now()
|
|
ok = True
|
|
checks = []
|
|
|
|
# 1. price_monitor 最近运行时间
|
|
try:
|
|
if conn:
|
|
lp = cur.execute("SELECT MAX(updated_at) FROM live_prices").fetchone()[0]
|
|
if lp:
|
|
lp_dt = datetime.fromisoformat(lp) if isinstance(lp, str) else lp
|
|
mins_ago = (now - lp_dt).total_seconds() / 60
|
|
status = "ok" if mins_ago < 10 else "warn"
|
|
if mins_ago > 15:
|
|
status = "fail"
|
|
ok = False
|
|
checks.append({"check":"price_monitor","status":status,"detail":f"最后更新{mins_ago:.0f}分前 @ {lp}"})
|
|
else:
|
|
checks.append({"check":"price_monitor","status":"warn","detail":"live_prices无数据"})
|
|
except Exception as e:
|
|
checks.append({"check":"price_monitor","status":"fail","detail":str(e)})
|
|
ok = False
|
|
|
|
# 2. 今日价格事件(有无zone breach触发)
|
|
try:
|
|
if conn:
|
|
pe = cur.execute("SELECT COUNT(*) FROM price_events WHERE date=date('now')").fetchone()[0]
|
|
checks.append({"check":"zone_breaches_today","status":"ok","detail":f"{pe}次突破事件"})
|
|
except Exception as e:
|
|
checks.append({"check":"zone_breaches","status":"fail","detail":str(e)})
|
|
|
|
# 3. holding_strategies 是否有策略数据
|
|
try:
|
|
if conn:
|
|
hs = cur.execute("SELECT COUNT(*) FROM holding_strategies").fetchone()[0]
|
|
valid = cur.execute("SELECT COUNT(*) FROM holding_strategies WHERE stop_loss IS NOT NULL").fetchone()[0]
|
|
status = "ok" if hs > 0 else "warn"
|
|
if hs == 0:
|
|
ok = False
|
|
checks.append({"check":"strategies_in_db","status":status,"detail":f"{hs}条策略（含止损{valid}条）"})
|
|
except Exception as e:
|
|
checks.append({"check":"strategies","status":"fail","detail":str(e)})
|
|
|
|
# 4. mofin_collect 今日是否跑过（检查策略新鲜度日志）
|
|
try:
|
|
if conn:
|
|
fresh = cur.execute("""
|
|
SELECT COUNT(*) FROM holding_strategies
|
|
WHERE date(created_at)=date('now')
|
|
""").fetchone()[0]
|
|
checks.append({"check":"today_reassessed","status":"ok","detail":f"今日{('reassess刷新'+str(fresh)+'条') if fresh else '暂无新重评记录'}"})
|
|
except Exception as e:
|
|
checks.append({"check":"today_reassess","status":"warn","detail":str(e)})
|
|
|
|
# 5. XMPP bridge是否在线
|
|
try:
|
|
import urllib.request
|
|
req = urllib.request.Request("http://127.0.0.1:5805/",
|
|
data=b'{"to":"hmo@yoin.fun","body":"ping","type":"chat"}',
|
|
headers={"Content-Type":"application/json"})
|
|
resp = urllib.request.urlopen(req, timeout=3)
|
|
result = json.loads(resp.read())
|
|
bridge_ok = result.get("ok") == True
|
|
checks.append({"check":"xmpp_bridge_5805","status":"ok" if bridge_ok else "fail","detail":"在线" if bridge_ok else "异常"})
|
|
if not bridge_ok:
|
|
ok = False
|
|
except Exception as e:
|
|
checks.append({"check":"xmpp_bridge","status":"fail","detail":str(e)})
|
|
ok = False
|
|
|
|
# 6. 验证reassess_with_context可导入
|
|
try:
|
|
sys.path.insert(0, "/home/hmo/MoFin")
|
|
from strategy_lifecycle import reassess_with_context
|
|
checks.append({"check":"reassess_module","status":"ok","detail":"reassess_with_context可导入"})
|
|
except Exception as e:
|
|
checks.append({"check":"reassess_module","status":"fail","detail":str(e)})
|
|
ok = False
|
|
|
|
if conn:
|
|
conn.close()
|
|
|
|
# 输出：无异常静默，有异常报警
|
|
result = {
|
|
"pipeline": "ok" if ok else "degraded",
|
|
"checked_at": now.isoformat(),
|
|
"checks": checks
|
|
}
|
|
|
|
if ok:
|
|
# 只有异常时才出声
|
|
print("[SILENT]")
|
|
else:
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|