diff --git a/scripts/verify_reassess_pipeline.py b/scripts/verify_reassess_pipeline.py index 21bddd51..b35fed9a 100644 --- a/scripts/verify_reassess_pipeline.py +++ b/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()