异常浮窗+统一告警通道+冷却1小时: alert_logger统一异常写入(alerts.json+history.log+XMPP), 前端右上角浮动异常区(自动展开5秒折叠,只显示active), price_monitor接入异常钩子(重评失败/拉取失败/DB同步), 冷却30分->1小时+15分5%突发例外

alert_logger.py

@@ -0,0 +1,104 @@
+# -*- coding: utf-8 -*-
+"""alert_logger.py — MoFin 统一异常告警通道（2026-08-12 落地）
+
+任何脚本/任务出错时调用 record_alert()：
+  - 写入 data/alerts.json（active 异常列表，前端浮窗读取 /api/overview 的 alerts）
+  - 追加 data/alerts_history.log（历史异常实时日志，永久保留）
+  - 同时通过 xmpp_logger 发送 XMPP 告警（老莫铁律：任何错误/异常 → xmpp告警 + MoFin监控）
+
+用法：
+  from alert_logger import record_alert
+  record_alert(level="error", source="price_monitor", title="重评失败",
+               detail="688002 LLM 调用超时", code="688002")
+
+level: info / warning / error / critical
+"""
+import json
+import time
+from pathlib import Path
+
+DATA_DIR = Path("/home/hmo/MoFin/data")
+ALERTS_FILE = DATA_DIR / "alerts.json"
+HISTORY_FILE = DATA_DIR / "alerts_history.log"
+
+# 保留的 active 异常数（浮窗只显示 active，最多保留 20 条）
+MAX_ACTIVE = 20
+
+
+def _now_str():
+    from datetime import datetime
+    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+
+def record_alert(level="error", source="", title="", detail="", code="", clear_after=0):
+    """记录一条异常。clear_after>0 表示自动清除的秒数（可空）"""
+    ts = time.time()
+    ts_str = _now_str()
+    entry = {
+        "ts": ts,
+        "ts_str": ts_str,
+        "level": level,
+        "source": source,
+        "title": title,
+        "detail": detail,
+        "code": code,
+        "clear_after": clear_after,
+    }
+
+    # 1. active 列表（alerts.json）
+    try:
+        alerts = json.loads(ALERTS_FILE.read_text(encoding="utf-8")) if ALERTS_FILE.exists() else []
+        if not isinstance(alerts, list):
+            alerts = []
+        # 同 source+code 的旧告警先移除（避免重复堆积），再插到头部
+        alerts = [a for a in alerts if not (a.get("source") == source and a.get("code") == code)]
+        alerts.insert(0, entry)
+        # 清除超时的
+        now = time.time()
+        alerts = [a for a in alerts if not a.get("clear_after") or now - a.get("ts", 0) < a["clear_after"]]
+        alerts = alerts[:MAX_ACTIVE]
+        ALERTS_FILE.write_text(json.dumps(alerts, ensure_ascii=False, indent=1), encoding="utf-8")
+    except Exception:
+        pass
+
+    # 2. 历史日志（永久追加）
+    try:
+        with HISTORY_FILE.open("a", encoding="utf-8") as f:
+            f.write(f"[{ts_str}] [{level.upper()}] {source} | {title} | {detail} | {code}\n")
+    except Exception:
+        pass
+
+    # 3. XMPP 告警（非 info 级别才发，避免轰炸）→ 本地 XMPP 发送服务 :5805
+    if level != "info":
+        try:
+            import urllib.request as _ur
+            body = f"⚠️[{level.upper()}] {source}: {title} {detail} {code}".strip()
+            req = _ur.Request(
+                "http://127.0.0.1:5805/",
+                data=body.encode("utf-8"),
+                headers={"Content-Type": "text/plain"},
+                method="POST",
+            )
+            _ur.urlopen(req, timeout=5).read()
+        except Exception:
+            pass
+
+    return entry
+
+
+def clear_alert(source="", code=""):
+    """清除指定 source+code 的 active 告警（恢复后调用）"""
+    try:
+        alerts = json.loads(ALERTS_FILE.read_text(encoding="utf-8")) if ALERTS_FILE.exists() else []
+        if not isinstance(alerts, list):
+            return
+        alerts = [a for a in alerts if not (a.get("source") == source and (not code or a.get("code") == code))]
+        ALERTS_FILE.write_text(json.dumps(alerts, ensure_ascii=False, indent=1), encoding="utf-8")
+    except Exception:
+        pass
+
+
+if __name__ == "__main__":
+    # 自测
+    record_alert(level="info", source="alert_logger", title="自测", detail="告警通道正常", code="test")
+    print("alerts.json 写入 OK:", ALERTS_FILE.exists())

deploy/profile-scripts/price_monitor.py

@@ -9,6 +9,7 @@ import sqlite3
 from datetime import datetime
 
 from mo_data import read_decisions
+from alert_logger import record_alert, clear_alert
 
 BREACH_PATH = "/home/hmo/.hermes/zone_breach.json"
 STATE_PATH = "/home/hmo/.hermes/price_trigger_state.json"
@@ -105,6 +106,10 @@ def fetch_all_prices(codes):
             text = r.read().decode("gbk")
     except Exception as e:
         print(f"⚠️ 批量拉取失败: {e}", file=sys.stderr)
+        try:
+            record_alert(level="warning", source="price_monitor", title="批量拉取失败", detail=str(e)[:200])
+        except Exception:
+            pass
         return {}
 
     results = {}
@@ -149,6 +154,10 @@ def refresh_data_prices():
         conn.close()
     except Exception as e:
         print(f"⚠️ 从DB读代码失败: {e}", file=sys.stderr)
+        try:
+            record_alert(level="warning", source="price_monitor", title="读取持仓失败", detail=str(e)[:200])
+        except Exception:
+            pass
         return 0
 
     if not all_codes:
@@ -295,6 +304,10 @@ def refresh_data_prices():
                     except Exception: pass
                     conn = None
                 print(f"⚠️ DB同步异常: {e}", file=sys.stderr)
+                try:
+                    record_alert(level="error", source="price_monitor", title="DB同步异常", detail=str(e)[:200])
+                except Exception:
+                    pass
                 break
         else:
             # for-else: loop exhausted without break
@@ -537,7 +550,10 @@ def run_once(round_label=""):
                             record_event(code, name, "swing_re", price, str(_ma10))
                             _push_action("波段再进", f"📈 {name}({code}) {price} 收回MA10({_ma10:.2f})且突破前一日高点 → 波段再进，可接回")
             except Exception as _e:
-                pass
+                try:
+                    record_alert(level="warning", source="price_monitor", title="波段检测异常", detail=str(_e)[:200], code=code)
+                except Exception:
+                    pass
 
         # 时间预算检查：如果超时，跳过重评只做状态记录
         _budget_low = (time.time() - start) > TIME_BUDGET
@@ -578,6 +594,10 @@ def run_once(round_label=""):
                                     outputs.append(f"  📨 止损重评→已推送Dad: {action}")
                         except Exception as e:
                             outputs.append(f"  ⚠️ 止损重评失败: {e}")
+                            try:
+                                record_alert(level="error", source="price_monitor", title="止损重评失败", detail=str(e)[:200], code=code)
+                            except Exception:
+                                pass
                 else:
                     extra = ""
                     if "_price" in key:
@@ -626,6 +646,10 @@ def run_once(round_label=""):
                                     outputs.append(f"  📋 本地日志(不推): {reason}")
                         except Exception as e:
                             outputs.append(f"  ⚠️ 区间重评失败: {e}")
+                            try:
+                                record_alert(level="error", source="price_monitor", title="区间重评失败", detail=str(e)[:200], code=code)
+                            except Exception:
+                                pass
                 state[code][key] = True
                 state_updated = True