fix: 管道审计+全局cron失败监控, 异常时主动推XMPP

scripts/verify_reassess_pipeline.py

@@ -1,118 +1,185 @@
 #!/usr/bin/env python3
-"""verify_reassess_pipeline.py — 重评推送管道审计
+"""verify_reassess_pipeline.py — 重评推送管道审计 + 全局cron失败监控
 
-检查三个代码级约束是否在正常运转：
-1. price_monitor 每2分触发 → 突破检测 → reassess调用
-2. stale_detector/stale_push_wlin 自选买入区检测
-3. mofin_collect LLM前强制重评
+检查：
+1. price_monitor 每2分正常跑
+2. zone breach检测正常
+3. holding_strategies有数据
+4. XMPP bridge在线
+5. reassess模块可导入
+6. 【新增】所有关键cron job状态（是否有failed）
 
-输出：JSON格式管道状态（无异常时静默[SILENT]）
+输出：正常时 [SILENT]，有异常时推XMPP
 """
-import json, os, sys, subprocess
+import json, os, sys, subprocess, sqlite3
 from pathlib import Path
 from datetime import datetime, timedelta
+from urllib.request import Request, urlopen
 
 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
+XMPP_BRIDGE = "http://127.0.0.1:5805/"
+XMPP_USER = "hmo@yoin.fun"
 
-now = datetime.now()
-ok = True
-checks = []
+def xmpp_push(text):
+    try:
+        payload = json.dumps({"to": XMPP_USER, "body": text, "type": "chat"}).encode()
+        req = Request(XMPP_BRIDGE, data=payload, headers={"Content-Type": "application/json"})
+        urlopen(req, timeout=5)
+    except Exception as e:
+        print(f"[XMPP推送失败] {e}", file=sys.stderr)
 
-# 1. price_monitor 最近运行时间
-try:
-    if conn:
-        lp = cur.execute("SELECT MAX(updated_at) FROM live_prices").fetchone()[0]
+def scan_cron_failures():
+    """扫描两个cron jobs.json看是否有failed状态的关键job"""
+    failures = []
+    jobs_files = [
+        "/home/hmo/.hermes/cron/jobs.json",
+        "/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json",
+    ]
+    for jf in jobs_files:
+        try:
+            data = json.load(open(jf))
+            for job in data.get("jobs", []):
+                jid = job.get("id", "?")
+                name = job.get("name", "") or jid[:12]
+                status = job.get("last_status", "")
+                enabled = job.get("enabled", True)
+                if not enabled:
+                    continue
+                # 关键job：价格监控、重评、盘前中监控
+                key_job = any(kw in name.lower() for kw in [
+                    "price_monitor", "monitor", "盘前中", "reassess",
+                    "重评", "自选买入", "stale_push", "管道审计",
+                    "宏观风险", "策略时效"
+                ])
+                if not key_job:
+                    continue
+                if status == "failed":
+                    last_run = job.get("last_run_at", "?")
+                    failures.append(f"  ❌ {name} ({jid[:8]}) last_run={last_run}")
+        except Exception:
+            pass
+    return failures
+
+def check_cron_jobs():
+    """另法：直接查cron数据库"""
+    issues = []
+    for db_path in [
+        BASE / "cron" / "cron.db",
+        Path("/home/hmo/.hermes/cron/cron.db"),
+    ]:
+        if not db_path.exists():
+            continue
+        try:
+            c = sqlite3.connect(str(db_path))
+            for row in c.execute("""
+                SELECT id, name, last_status, last_run_at, enabled
+                FROM cron_jobs WHERE enabled=1
+                ORDER BY last_run_at DESC
+            """).fetchall():
+                jid, name, status, last_run, enabled = row
+                if status == "failed":
+                    issues.append(f"  ❌ {name}({jid[:8]}) last_run={last_run}")
+            c.close()
+        except Exception:
+            pass
+    return issues
+
+def run():
+    ok = True
+    alerts = []
+    checks = []
+
+    # 1. price_monitor 最近运行时间
+    try:
+        conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db")
+        lp = conn.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
+            if hasattr(lp_dt, 'tzinfo') and lp_dt.tzinfo is None:
+                if isinstance(lp, str) and '+' not in lp:
+                    lp_dt = lp_dt.replace(tzinfo=None)
+            mins_ago = (datetime.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}"})
+                alerts.append(f"price_monitor {mins_ago:.0f}分未更新")
+            checks.append({"check":"price_monitor","status":status,"detail":f"最后更新{mins_ago:.0f}分前"})
         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
+    except Exception as e:
+        checks.append({"check":"price_monitor","status":"fail","detail":str(e)})
+        ok = False
+        alerts.append(f"price_monitor异常: {e}")
 
-# 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"
+    # 2. holding_strategies 是否有数据
+    try:
+        hs = conn.execute("SELECT COUNT(*) FROM holding_strategies").fetchone()[0]
+        valid = conn.execute("SELECT COUNT(*) FROM holding_strategies WHERE stop_loss IS NOT NULL").fetchone()[0]
         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:
+            alerts.append("holding_strategies为0条")
+        checks.append({"check":"strategies_in_db","status":"ok" if hs > 0 else "warn","detail":f"{hs}条策略（含止损{valid}条）"})
+    except Exception as e:
+        checks.append({"check":"strategies","status":"fail","detail":str(e)})
         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
+    # 3. XMPP bridge 是否在线
+    try:
+        req = Request(XMPP_BRIDGE, data=b'{"to":"hmo@yoin.fun","body":"ping","type":"chat"}',
+                      headers={"Content-Type":"application/json"})
+        resp = urlopen(req, timeout=3)
+        bridge_ok = json.loads(resp.read()).get("ok") == True
+        if not bridge_ok:
+            ok = False
+            alerts.append("XMPP bridge异常")
+        checks.append({"check":"xmpp_bridge","status":"ok" if bridge_ok else "fail","detail":"在线" if bridge_ok else "异常"})
+    except Exception as e:
+        checks.append({"check":"xmpp_bridge","status":"fail","detail":str(e)})
+        ok = False
+        alerts.append(f"XMPP bridge不可达: {e}")
+
+    # 4. reassess模块可导入
+    try:
+        from strategy_lifecycle import reassess_with_context
+        checks.append({"check":"reassess_module","status":"ok","detail":"可导入"})
+    except Exception as e:
+        checks.append({"check":"reassess_module","status":"fail","detail":str(e)})
+        ok = False
+        alerts.append(f"reassess模块导入失败: {e}")
+
+    # 5. cron job失败检测
+    cron_issues = scan_cron_failures() + check_cron_jobs()
+    if cron_issues:
+        ok = False
+        alerts.append(f"{len(cron_issues)}个cron job失败")
+        for issue in cron_issues[:5]:
+            alerts.append(issue)
+        checks.append({"check":"cron_jobs","status":"fail","detail":"; ".join(cron_issues[:3])})
+    else:
+        checks.append({"check":"cron_jobs","status":"ok","detail":"所有关键job正常"})
 
-if conn:
     conn.close()
 
-# 输出：无异常静默，有异常报警
-result = {
-    "pipeline": "ok" if ok else "degraded",
-    "checked_at": now.isoformat(),
-    "checks": checks
-}
+    # 输出
+    result = {
+        "pipeline": "ok" if ok else "degraded",
+        "checked_at": datetime.now().isoformat(),
+        "checks": checks,
+        "alerts": alerts
+    }
 
-if ok:
-    # 只有异常时才出声
-    print("[SILENT]")
-else:
-    print(json.dumps(result, ensure_ascii=False, indent=2))
+    if ok:
+        print("[SILENT]")
+    else:
+        msg = "🔴 重评管道异常:\n" + "\n".join(alerts)
+        print(json.dumps(result, ensure_ascii=False, indent=2))
+        # 有异常时主动推XMPP（取代静默）
+        xmpp_push(msg)
+        print(f"\n已推送XMPP: {len(alerts)}条告警", file=sys.stderr)
+
+if __name__ == "__main__":
+    run()