diff --git a/deploy/profile-scripts/agent_spiral_watchdog.py b/deploy/profile-scripts/agent_spiral_watchdog.py
index 8032a3f6..5451d4c1 100644
--- a/deploy/profile-scripts/agent_spiral_watchdog.py
+++ b/deploy/profile-scripts/agent_spiral_watchdog.py
@@ -32,12 +32,10 @@ def log(msg):
 
 def xmpp(msg):
     try:
-        import urllib.request
-        req = urllib.request.Request(
-            "http://127.0.0.1:5805/",
-            data=json.dumps({"body": msg, "to": "hmo@yoin.fun", "type": "chat"}).encode(),
-            headers={"Content-Type": "application/json"})
-        urllib.request.urlopen(req, timeout=5)
+        import sys
+        sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+        from alert_helper import notify, INFO
+        notify("螺旋监控", msg, INFO)
     except Exception as e:
         log(f"XMPP发送失败: {e}")
 
@@ -105,15 +103,13 @@ def main():
                 continue
             found += 1
             alerted.add(sid)
-            msg = (f"📟【MoFin系统·螺旋监控】（非知微本人）\n"
-                   f"agent 螺旋嫌疑\n"
+            msg = (f"agent 螺旋嫌疑\n"
                    f"profile: {profile}\n"
                    f"session: {sid}\n"
                    f"已运行: {age_sec/60:.0f} 分钟\n"
                    f"消息数: {r['message_count']} | 工具调用: {r['tool_call_count']} | "
                    f"输入token: {r['input_tokens']}\n"
-                   f"特征类似 603288 事件（gateway agent 运行时螺旋）。"
-                   f"如是正常长任务可忽略；否则需人工检查 gateway。")
+                   f"特征类似 603288 事件。正常长任务可忽略；否则需人工检查。")
             log(f"SPIRAL: [{profile}] {sid} age={age_sec/60:.0f}min msgs={r['message_count']} tools={r['tool_call_count']}")
             xmpp(msg)
 
diff --git a/deploy/profile-scripts/alert_helper.py b/deploy/profile-scripts/alert_helper.py
new file mode 100644
index 00000000..08605dbd
--- /dev/null
+++ b/deploy/profile-scripts/alert_helper.py
@@ -0,0 +1,122 @@
+#!/usr/bin/env python3
+"""alert_helper.py — MoFin 统一告警网关（信噪比控制）
+
+原则（老爸 2026-07-21 定）：
+- 真正有意义的信息（重点推荐操作/买入信号/需人工处理）绝不能被淹没
+- 纯通知型信息（系统报备/日常状态）必须控制频率和篇幅
+
+两级通道：
+- ACTION（行动级）：买入信号、重点推荐、需人工核查的故障
+  → 直通，不限速，🚨 醒目前缀，独立成条
+- INFO（通知级）：部署报备、卫生审计、修复报备、螺旋嫌疑
+  → 同类 30 分钟内最多 1 条；篇幅 ≤8 行；24h 内容去重（同一问题不重复报）
+
+所有系统消息统一 📟【MoFin系统·类别】前缀，与知微本人消息一眼区分。
+
+用法:
+    from alert_helper import notify, ACTION, INFO
+    notify("信号", "📈 300308 买入信号...", level=ACTION)   # 直通
+    notify("部署守卫", "自动部署完成...", level=INFO)         # 限速
+"""
+import json, os, time, hashlib
+from datetime import datetime
+
+ACTION = "action"
+INFO = "info"
+
+STATE_FILE = "/home/hmo/MoFin/gateway/logs/alert_state.json"
+LOG = "/home/hmo/MoFin/gateway/logs/alert_helper.log"
+
+INFO_MIN_INTERVAL = 1800   # 同类 info 30 分钟最多 1 条
+INFO_MAX_LINES = 8         # info 篇幅上限
+ACTION_MAX_LINES = 30      # action 篇幅上限（宽松但不失控）
+DEDUP_SEC = 24 * 3600      # 相同内容 24h 不重复
+
+
+def _log(msg):
+    line = f"[{datetime.now().isoformat(timespec='seconds')}] {msg}"
+    print(line, flush=True)
+    try:
+        os.makedirs(os.path.dirname(LOG), exist_ok=True)
+        with open(LOG, "a", encoding="utf-8") as f:
+            f.write(line + "\n")
+    except Exception:
+        pass
+
+
+def _load_state():
+    try:
+        with open(STATE_FILE, encoding="utf-8") as f:
+            return json.load(f)
+    except Exception:
+        return {}
+
+
+def _save_state(st):
+    try:
+        # 只保留最近 100 个类别条目
+        if len(st) > 100:
+            st = dict(sorted(st.items(), key=lambda kv: kv[1].get("last_ts", 0))[-100:])
+        with open(STATE_FILE, "w", encoding="utf-8") as f:
+            json.dump(st, f)
+    except Exception:
+        pass
+
+
+def _send(body):
+    import urllib.request
+    req = urllib.request.Request(
+        "http://127.0.0.1:5805/",
+        data=json.dumps({"body": body, "to": "hmo@yoin.fun", "type": "chat"}).encode(),
+        headers={"Content-Type": "application/json"})
+    urllib.request.urlopen(req, timeout=5)
+
+
+def notify(category, body, level=INFO):
+    """统一告警入口。返回 True=已发送, False=被限速/去重静默。"""
+    now = time.time()
+    st = _load_state()
+    entry = st.get(category, {"last_ts": 0, "last_hash": "", "suppressed": 0})
+
+    body_hash = hashlib.md5(body.encode()).hexdigest()
+
+    if level == INFO:
+        # 24h 内容去重：同一问题不重复报
+        if body_hash == entry.get("last_hash") and now - entry.get("last_ts", 0) < DEDUP_SEC:
+            _log(f"[{category}] 内容重复(24h内已报)，静默")
+            return False
+        # 频率限制：同类 30min 最多 1 条
+        if now - entry.get("last_ts", 0) < INFO_MIN_INTERVAL:
+            entry["suppressed"] = entry.get("suppressed", 0) + 1
+            st[category] = entry
+            _save_state(st)
+            _log(f"[{category}] 30min 频率限制，静默(累计压制{entry['suppressed']}条)")
+            return False
+        # 篇幅截断
+        lines = body.splitlines()
+        if len(lines) > INFO_MAX_LINES:
+            body = "\n".join(lines[:INFO_MAX_LINES]) + \
+                   f"\n…(共{len(lines)}行,详见 gateway/logs/)"
+        # 告知压制历史
+        if entry.get("suppressed", 0) > 0:
+            body += f"\n(注: 上次以来另有 {entry['suppressed']} 条同类通知已按频率策略静默)"
+            entry["suppressed"] = 0
+        prefix = f"📟【MoFin系统·{category}】（非知微本人）"
+    else:
+        lines = body.splitlines()
+        if len(lines) > ACTION_MAX_LINES:
+            body = "\n".join(lines[:ACTION_MAX_LINES]) + f"\n…(共{len(lines)}行)"
+        prefix = f"🚨【MoFin·{category}】"
+
+    try:
+        _send(f"{prefix}\n{body}")
+    except Exception as e:
+        _log(f"[{category}] XMPP发送失败: {e}")
+        return False
+
+    entry["last_ts"] = now
+    entry["last_hash"] = body_hash
+    st[category] = entry
+    _save_state(st)
+    _log(f"[{category}] 已发送({level})")
+    return True
diff --git a/deploy/profile-scripts/batch_reassess.py b/deploy/profile-scripts/batch_reassess.py
index dcff8e7a..5a464459 100644
--- a/deploy/profile-scripts/batch_reassess.py
+++ b/deploy/profile-scripts/batch_reassess.py
@@ -371,12 +371,9 @@ def save_result(code, full_text, parsed):
             _sl = parsed.get("stop_loss", 0)
             _tp = parsed.get("take_profit", 0)
             _pos = parsed.get("position", "")
-            _msg = f"\U0001f4c8 {_name}({code}) 价{_p}\u219212维分析生成买入信号！区间{_el}~{_eh} 损{_sl} 盈{_tp} 仓位{_pos}"
-            import urllib.request, json as _jj
-            _req = urllib.request.Request("http://127.0.0.1:5805/",
-                data=_jj.dumps({"body": _msg, "to": "hmo@yoin.fun", "type": "chat"}).encode(),
-                headers={"Content-Type": "application/json"})
-            urllib.request.urlopen(_req, timeout=5)
+            _msg = f"📈 {_name}({code}) 价{_p}→12维分析生成买入信号！区间{_el}~{_eh} 损{_sl} 盈{_tp} 仓位{_pos}"
+            from alert_helper import notify as _notify, ACTION as _ACT
+            _notify("买入信号", _msg, _ACT)
             print(f"  \U0001f4e8 XMPP推送成功: {_msg[:60]}")
         except Exception as _e:
             print(f"  \u26a0\ufe0f XMPP推送失败: {_e}")
diff --git a/deploy/profile-scripts/deploy_guard.py b/deploy/profile-scripts/deploy_guard.py
index be2be02b..1b5b17d9 100644
--- a/deploy/profile-scripts/deploy_guard.py
+++ b/deploy/profile-scripts/deploy_guard.py
@@ -50,14 +50,11 @@ def log(msg):
         pass
 
 
-def xmpp(msg):
+def xmpp(msg, level=None):
     try:
-        import urllib.request
-        req = urllib.request.Request(
-            "http://127.0.0.1:5805/",
-            data=json.dumps({"body": msg, "to": "hmo@yoin.fun", "type": "chat"}).encode(),
-            headers={"Content-Type": "application/json"})
-        urllib.request.urlopen(req, timeout=5)
+        sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+        from alert_helper import notify, INFO
+        notify("部署守卫", msg, level or INFO)
     except Exception as e:
         log(f"XMPP发送失败: {e}")
 
@@ -206,12 +203,19 @@ def main():
         pass
 
     if actions or problems:
-        msg = "📟【MoFin系统·部署守卫】（非知微本人）\n"
-        for a in actions:
-            msg += f"✅ {a}\n"
+        # 分级：部署后验证失败=需人工核查(ACTION直通)；其余=INFO(限速聚合)
+        info_parts = list(actions)
+        action_parts = []
         for p in problems:
-            msg += f"⚠️ {p}\n"
-        xmpp(msg.strip())
+            if "部署后验证发现" in p:
+                action_parts.append(p)
+            else:
+                info_parts.append(p)
+        if info_parts:
+            xmpp("\n".join(f"✅ {a}" if a in actions else f"⚠️ {a}" for a in info_parts))
+        for p in action_parts:
+            from alert_helper import notify as _notify, ACTION as _ACT
+            _notify("部署验证", f"⚠️ {p}", _ACT)
     log(f"── 结束: actions={len(actions)} problems={len(problems)} ──")
     return 0 if not problems else 1
 
diff --git a/deploy/profile-scripts/per_stock_reassess.py b/deploy/profile-scripts/per_stock_reassess.py
index e138a370..f02b5be4 100644
--- a/deploy/profile-scripts/per_stock_reassess.py
+++ b/deploy/profile-scripts/per_stock_reassess.py
@@ -567,10 +567,8 @@ def main():
                                                 "SELECT name, price, entry_low, entry_high, stop_loss, take_profit, position_advice FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
                                             if _nr2:
                                                 _xm = f"📈 {_nr2[0] or code}({code}) 价{_nr2[1]}→12维买入信号！区间{_nr2[2]}~{_nr2[3]} 损{_nr2[4]} 盈{_nr2[5]} 仓位{_nr2[6] or '-'}"
-                                                _xr = __import__('urllib.request').Request("http://127.0.0.1:5805/",
-                                                    data=__import__('json').dumps({"body": _xm, "to": "hmo@yoin.fun", "type": "chat"}).encode(),
-                                                    headers={"Content-Type": "application/json"})
-                                                __import__('urllib.request').urlopen(_xr, timeout=5)
+                                                from alert_helper import notify as _notify2, ACTION as _ACT2
+                                                _notify2("买入信号", _xm, _ACT2)
                                                 print(f"  📨 XMPP推送买入信号")
                                         except: pass
                         except: pass
diff --git a/deploy/profile-scripts/self_repair.py b/deploy/profile-scripts/self_repair.py
index f8c8d6fb..9bebf942 100644
--- a/deploy/profile-scripts/self_repair.py
+++ b/deploy/profile-scripts/self_repair.py
@@ -198,11 +198,10 @@ def llm_diagnose(failure):
 
 def xmpp_report(lines):
     try:
-        import urllib.request
-        payload = json.dumps({'to': 'hmo@yoin.fun', 'body': '\n'.join(lines), 'type': 'chat'}).encode()
-        req = urllib.request.Request('http://127.0.0.1:5805/', data=payload,
-                                     headers={'Content-Type': 'application/json'})
-        urllib.request.urlopen(req, timeout=5)
+        import sys as _s, os as _o
+        _s.path.insert(0, _o.path.dirname(_o.path.abspath(__file__)))
+        from alert_helper import notify, INFO
+        notify("L3自愈", '\n'.join(lines), INFO)
     except Exception as e:
         print(f'XMPP 失败: {e}')
 
diff --git a/deploy/profile-scripts/system_hygiene_audit.py b/deploy/profile-scripts/system_hygiene_audit.py
index ae26d1e5..7b5f8407 100644
--- a/deploy/profile-scripts/system_hygiene_audit.py
+++ b/deploy/profile-scripts/system_hygiene_audit.py
@@ -297,19 +297,19 @@ def main():
         json.dump(report, f, ensure_ascii=False, indent=2)
 
     if all_issues:
-        # 推 XMPP
+        # 推 XMPP（经 alert_helper：30min限速+8行截断+24h内容去重——同一批问题不会每天重复轰炸）
         try:
-            import urllib.request
-            lines = [f"📟【MoFin系统·卫生审计】（非知微本人）发现 {len(all_issues)} 个问题："]
+            sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+            from alert_helper import notify, INFO
+            lines = [f"发现 {len(all_issues)} 个问题："]
             for i in all_issues[:8]:
                 lines.append(f"• [{i['type']}] {i.get('file') or i.get('job') or i.get('table') or i.get('pid')}: {i.get('action','')[:60]}")
             if len(all_issues) > 8:
                 lines.append(f'… 共 {len(all_issues)} 个，详见 hygiene_report.json')
-            payload = json.dumps({'to': 'hmo@yoin.fun', 'body': '\n'.join(lines), 'type': 'chat'}).encode()
-            req = urllib.request.Request('http://127.0.0.1:5805/', data=payload,
-                                         headers={'Content-Type': 'application/json'})
-            urllib.request.urlopen(req, timeout=5)
-            print('  📨 已推 XMPP')
+            if notify("卫生审计", '\n'.join(lines), INFO):
+                print('  📨 已推 XMPP')
+            else:
+                print('  📨 XMPP 已按频率/去重策略静默')
         except Exception as e:
             print(f'  XMPP 推送失败: {e}')
     else: