From 16be2a067bcc49304ff12d60dd90c50fdf2f1884 Mon Sep 17 00:00:00 2001 From: xxm Date: Mon, 17 Aug 2026 13:05:56 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20mofin=5Fguard=5Fmonitor=E2=80=94?= =?UTF-8?q?=E2=80=94=E5=90=8E=E5=8F=B0=E7=9B=91=E6=8E=A7=E7=9F=A5=E5=BE=AE?= =?UTF-8?q?=E6=B6=88=E6=81=AF+=E5=86=85=E9=83=A8=E6=B5=81=E7=A8=8B?= =?UTF-8?q?=E5=90=88=E8=A7=84(=E6=8E=A8=E8=8D=90=E4=BE=9D=E6=8D=AE/?= =?UTF-8?q?=E4=BF=A1=E5=8F=B7=E4=B8=8D=E5=85=85=E5=88=86/executor/?= =?UTF-8?q?=E5=AE=88=E5=8D=AB/LLM),detached=E4=B8=8D=E9=98=BB=E5=A1=9ETUI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deploy/profile-scripts/mofin_guard_monitor.py | 145 ++++++++++++++++++ 1 file changed, 145 insertions(+) create mode 100644 deploy/profile-scripts/mofin_guard_monitor.py diff --git a/deploy/profile-scripts/mofin_guard_monitor.py b/deploy/profile-scripts/mofin_guard_monitor.py new file mode 100644 index 00000000..e44b14ce --- /dev/null +++ b/deploy/profile-scripts/mofin_guard_monitor.py @@ -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()