218 lines
8.3 KiB
Python
218 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
||
"""verify_reassess_pipeline.py — 重评推送管道审计 + 全局cron失败监控
|
||
|
||
检查：
|
||
1. price_monitor 每2分正常跑
|
||
2. zone breach检测正常
|
||
3. holding_strategies有数据
|
||
4. XMPP bridge在线
|
||
5. reassess模块可导入
|
||
6. 【新增】所有关键cron job状态（是否有failed）
|
||
|
||
输出：正常时 [SILENT]，有异常时推XMPP
|
||
"""
|
||
import json, os, sys, subprocess, sqlite3
|
||
from pathlib import Path
|
||
from datetime import datetime, timedelta
|
||
from urllib.request import Request, urlopen
|
||
|
||
# ── 消息通道统一路由(broadcast/xmpp by delivery) ──
|
||
try:
|
||
from messenger import install_stdio_hook as _msh
|
||
_msh()
|
||
except Exception:
|
||
pass
|
||
|
||
BASE = Path(__file__).parent.parent
|
||
sys.path.insert(0, str(BASE))
|
||
sys.path.insert(0, "/home/hmo/MoFin")
|
||
|
||
XMPP_BRIDGE = "http://127.0.0.1:5805/"
|
||
XMPP_USER = "hmo@yoin.fun"
|
||
|
||
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)
|
||
|
||
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:
|
||
# 开盘 grace：price_monitor 09:00 才启动，09:15 前隔夜数据属正常，不误报
|
||
from datetime import datetime as _dt
|
||
_now = _dt.now()
|
||
_morning_grace = _now.weekday() < 5 and _now.hour == 9 and _now.minute < 15
|
||
conn = None
|
||
last_err = None
|
||
# malformed 可能是 I/O 风暴下的瞬态 WAL 损坏（2026-07-21 事件）：
|
||
# checkpoint 后自愈。重试一次再告警，避免误报轰炸
|
||
for _attempt in range(2):
|
||
try:
|
||
conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db")
|
||
conn.execute("SELECT 1 FROM live_prices LIMIT 1").fetchone()
|
||
break
|
||
except Exception as e:
|
||
last_err = e
|
||
import time as _t
|
||
_t.sleep(3)
|
||
if conn is None:
|
||
raise last_err
|
||
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
|
||
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 and not _morning_grace:
|
||
status = "fail"
|
||
ok = False
|
||
alerts.append(f"price_monitor {mins_ago:.0f}分未更新")
|
||
checks.append({"check":"price_monitor","status":status,"detail":f"最后更新{mins_ago:.0f}分前" + ("(开盘grace)" if _morning_grace and mins_ago > 15 else "")})
|
||
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
|
||
alerts.append(f"price_monitor异常: {e}")
|
||
|
||
# 2. 策略评估活动（reassess_with_context写strategy_evaluations，不是holding_strategies）
|
||
try:
|
||
today_se = conn.execute("SELECT COUNT(*) FROM strategy_evaluations WHERE date(created_at)=date('now')").fetchone()[0]
|
||
total_se = conn.execute("SELECT COUNT(*) FROM strategy_evaluations").fetchone()[0]
|
||
# 也尝试查holding_strategies（如果存在并有数据）
|
||
hs_exists = conn.execute("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='holding_strategies'").fetchone()[0]
|
||
hs = 0
|
||
if hs_exists:
|
||
hs = conn.execute("SELECT COUNT(*) FROM holding_strategies").fetchone()[0]
|
||
detail = f"今日{today_se}次评估, 累计{total_se}条"
|
||
if hs > 0:
|
||
detail += f", holding_strategies{hs}条"
|
||
checks.append({"check":"strategy_activity","status":"ok","detail":detail})
|
||
except Exception as e:
|
||
checks.append({"check":"strategies","status":"fail","detail":str(e)})
|
||
ok = False
|
||
|
||
# 3. XMPP bridge 是否在线（TCP端口检测，不发消息到Dad）
|
||
try:
|
||
import socket
|
||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
sock.settimeout(3)
|
||
result = sock.connect_ex(("127.0.0.1", 5805))
|
||
sock.close()
|
||
bridge_ok = (result == 0)
|
||
if not bridge_ok:
|
||
ok = False
|
||
alerts.append("XMPP bridge(5805)端口无响应")
|
||
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正常"})
|
||
|
||
conn.close()
|
||
|
||
# 输出
|
||
result = {
|
||
"pipeline": "ok" if ok else "degraded",
|
||
"checked_at": datetime.now().isoformat(),
|
||
"checks": checks,
|
||
"alerts": alerts
|
||
}
|
||
|
||
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()
|