hmo/MoFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(alerts): alert_helper 统一告警网关——信噪比控制

Browse Source 
两级通道:
- ACTION(买入信号/重点推荐/部署验证失败): 直通不限速, 🚨醒目前缀
- INFO(部署/卫生/修复/螺旋报备): 同类30min限1条 + ≤8行截断 +
  24h内容去重(同一问题不重复轰炸) + 压制计数透明披露

6个调用点全部迁移: deploy_guard/hygiene/spiral/self_repair(INFO)
+ batch_reassess/per_stock_reassess 买入信号(ACTION)
解决: 真正有意义的信息(重点推荐操作)不被纯通知淹没
This commit is contained in:
hmo
2026-07-21 02:43:57 +08:00
parent 54652bff12
commit 1e71a2d852
7 changed files with 161 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/profile-scripts/agent_spiral_watchdog.py
+6 -10
View File
@@ -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)
deploy/profile-scripts/alert_helper.py
+122
View File
@@ -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
deploy/profile-scripts/batch_reassess.py
+3 -6
View File
@@ -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}")
deploy/profile-scripts/deploy_guard.py
+16 -12
View File
@@ -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
deploy/profile-scripts/per_stock_reassess.py
+2 -4
View File
@@ -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
deploy/profile-scripts/self_repair.py
+4 -5
View File
@@ -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}')
deploy/profile-scripts/system_hygiene_audit.py
+8 -8
View File
@@ -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: