feat: mofin_guard_monitor——后台监控知微消息+内部流程合规(推荐依据/信号不充分/executor/守卫/LLM),detached不阻塞TUI

2026-08-17 13:05:56 +08:00
parent cedd0a557c
commit 16be2a067b
1 changed files with 145 additions and 0 deletions
@@ -0,0 +1,145 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""mofin_guard_monitor.py — 知微消息监控（2026-08-17 下午盘，老莫要求）
+持续监控知微发给老莫的消息 + 内部流程，判断：
+  1. 推荐是否符合策略推荐规范（docs/strategy-recommendation-spec.md）
+     - 推荐必须带策略依据（版本/重评/止损/止盈/信号）
+     - 信号不充分/观望 不得推操作
+     - 候选股（非持仓）不推
+  2. 内部流程是否合规（executor/scanner/守卫/LLM）
+后台 nohup 运行，写状态到 /tmp/mofin_monitor_status.json，不阻塞 TUI。
+"""
+import json, os, time, glob
+from datetime import datetime, timedelta
+
+STATUS = "/tmp/mofin_monitor_status.json"
+XMPP_LOG = "/home/hmo/MoFin/gateway/logs/xmpp_messages.jsonl"
+EXEC_LOG = "/home/hmo/MoFin/gateway/logs/strategy_executor.log"
+GUARD_STATUS = "/home/hmo/MoFin/gateway/logs/deploy_guard_status.json"
+GUARD_LOG = "/home/hmo/MoFin/gateway/logs/deploy_guard.log"
+ERRORS_LOG = "/home/hmo/.hermes/profiles/position-analyst/logs/errors.log"
+POOL_OUT = "/home/hmo/.hermes/profiles/position-analyst/cron/output/pool_news_collector_30min"
+
+INTERVAL = 120  # 每2分钟一轮
+
+
+def load_status():
+    try:
+        return json.load(open(STATUS, encoding="utf-8"))
+    except Exception:
+        return {"started": datetime.now().isoformat(), "checks": [], "alerts": []}
+
+
+def save_status(s):
+    json.dump(s, open(STATUS, "w", encoding="utf-8"), ensure_ascii=False, indent=1)
+
+
+def check_messages(s, last_check):
+    """检查知微新消息：是否含操作推荐，是否带策略依据"""
+    alerts = []
+    try:
+        with open(XMPP_LOG, encoding="utf-8") as f:
+            lines = f.readlines()
+        new = []
+        for l in lines:
+            try:
+                d = json.loads(l)
+                ts = str(d.get("timestamp", ""))
+                if ts > last_check and d.get("direction") == "out" and "zhiwei" in str(d.get("from", "")):
+                    new.append(d)
+            except Exception:
+                pass
+        for d in new:
+            body = str(d.get("body_preview", ""))
+            # 判断：是否含操作建议（买入/卖出/止损/建仓/清仓）
+            ops = ["买入", "卖出", "建仓", "清仓", "止损", "止盈", "加仓", "减仓"]
+            has_op = any(k in body for k in ops)
+            # 是否带策略依据（版本/重评/止损值/信号）
+            has_basis = any(k in body for k in ["依据", "重评", "止损", "信号", "策略", "RR"])
+            # 是否可能是"信号不充分/观望"却推了
+            weak_signal = any(k in body for k in ["信号不充分", "暂不买入", "不追", "等信号转买入"])
+            issue = None
+            if has_op and not has_basis:
+                issue = f"推荐无策略依据: {body[:100]}"
+            elif weak_signal and has_op:
+                issue = f"信号不充分却含操作建议: {body[:100]}"
+            if issue:
+                alerts.append({"ts": d.get("timestamp"), "issue": issue, "body": body[:150]})
+    except Exception as e:
+        alerts.append({"ts": datetime.now().isoformat(), "issue": f"消息读取失败: {e}"})
+    return alerts
+
+
+def check_executor(s):
+    """executor 是否正常（崩溃/超时）"""
+    alerts = []
+    try:
+        with open(EXEC_LOG, encoding="utf-8") as f:
+            lines = f.readlines()
+        for l in lines[-20:]:
+            if any(k in l for k in ["✗", "崩溃", "超时", "stderr:", "⚠️"]):
+                if "重评" not in l and "stderr(rc=0)" not in l:
+                    alerts.append({"ts": l[:25], "issue": l.strip()[:120]})
+    except Exception:
+        pass
+    return alerts
+
+
+def check_guard(s):
+    """部署守卫是否有问题"""
+    alerts = []
+    try:
+        d = json.load(open(GUARD_STATUS, encoding="utf-8"))
+        probs = d.get("problems", []) or []
+        for p in probs:
+            alerts.append({"ts": datetime.now().isoformat(), "issue": f"守卫: {str(p)[:120]}"})
+    except Exception:
+        pass
+    return alerts
+
+
+def check_llm(s, last_check):
+    """LLM 是否连续失败"""
+    alerts = []
+    try:
+        with open(ERRORS_LOG, encoding="utf-8") as f:
+            for l in f.readlines()[-30:]:
+                if "API call failed" in l or "Stream stale" in l or "empty stream" in l:
+                    ts = l[:19]
+                    if ts > last_check:
+                        alerts.append({"ts": ts, "issue": l.strip()[60:150]})
+    except Exception:
+        pass
+    return alerts
+
+
+def main():
+    s = load_status()
+    last_msg = s.get("last_msg_check", "")
+    last_llm = s.get("last_llm_check", "")
+    while True:
+        now = datetime.now()
+        ts = now.strftime("%Y-%m-%d %H:%M:%S")
+        all_alerts = []
+        all_alerts += check_messages(s, last_msg)
+        all_alerts += check_executor(s)
+        all_alerts += check_guard(s)
+        all_alerts += check_llm(s, last_llm)
+        if all_alerts:
+            s["alerts"] = (s.get("alerts") or []) + [{"check_ts": ts, "alerts": all_alerts}]
+            s["alerts"] = s["alerts"][-30:]
+        s["last_check"] = ts
+        s["last_msg_check"] = now.strftime("%Y-%m-%d %H:%M:%S")
+        s["last_llm_check"] = now.strftime("%Y-%m-%d %H:%M:%S")
+        save_status(s)
+        # 简明写一行日志（不刷屏）
+        try:
+            with open("/tmp/mofin_monitor.log", "a") as f:
+                f.write(f"[{ts}] checks=ok alerts={len(all_alerts)}\n")
+        except Exception:
+            pass
+        time.sleep(INTERVAL)
+
+
+if __name__ == "__main__":
+    main()