From a782a839532ed147ba3f8d52cf35bfcf54576cd4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=9F=A5=E5=BE=AE?= Date: Tue, 21 Jul 2026 11:31:44 +0800 Subject: [PATCH] sync: price_monitor SIGALRM fix to scripts/ (legacy location) --- scripts/price_monitor.py | 56 ++++++++++++++++++++++++++++++++++------ 1 file changed, 48 insertions(+), 8 deletions(-) diff --git a/scripts/price_monitor.py b/scripts/price_monitor.py index f28e7896..b5159d11 100644 --- a/scripts/price_monitor.py +++ b/scripts/price_monitor.py @@ -37,7 +37,8 @@ XMPP_USER = "hmo@yoin.fun" XMPP_BRIDGE = "http://127.0.0.1:5805/" def push_to_xmpp(text): - """通过知微 HTTP bridge 推送到Dad私信""" + """原始直推（已废弃直用）——保留给极少数必须原样的场景。 + 新代码请用 _push_action/_push_digest。""" if not text.strip(): return try: @@ -51,6 +52,25 @@ def push_to_xmpp(text): except Exception as e: print(f"[XMPP推送失败] {e}", file=sys.stderr) + +# ── 分级推送（2026-07-21 信噪比纪律，红线#12）── +# ACTION: 破止损/重评确认的操作信号 — 直通不限速 +# INFO: 未确认的进区提示 — 聚合成摘要，30min 限 1 条 +def _push_action(category, text): + try: + from alert_helper import notify, ACTION + notify(category, text, ACTION) + except Exception as e: + print(f"[ACTION推送失败] {e}", file=sys.stderr) + + +def _push_digest(category, text): + try: + from alert_helper import notify, INFO + notify(category, text, INFO) + except Exception as e: + print(f"[INFO推送失败] {e}", file=sys.stderr) + # ── 批量拉取价格 ────────────────────────────────────────────────────────── def fetch_all_prices(codes): @@ -387,10 +407,17 @@ def _handle_sigterm(signum, frame): _cleanup_lock() sys.exit(0) +def _handle_sigalrm(signum, frame): + """收到SIGALRM强制超时时清理锁文件后退出""" + _cleanup_lock() + print(f"[TIMEOUT] 本轮执行超时({signum}s)，已清理锁文件", file=sys.stderr, flush=True) + sys.exit(0) + def run_once(round_label=""): """执行一轮完整的监控流程""" import os, signal # 必须在开头import，否则os变量会被后面的局部import绑定覆盖 signal.signal(signal.SIGTERM, _handle_sigterm) + signal.signal(signal.SIGALRM, _handle_sigalrm) os.nice(10) # 降低优先级，避免与DB其他写操作抢占 # ── 进程锁：同一时间只跑一个实例 ── _lk = "/tmp/price_monitor.lock" @@ -405,6 +432,7 @@ def run_once(round_label=""): pass with open(_lk, "w") as _f: _f.write(str(os.getpid())) + signal.alarm(120) # 硬上限120s，超时自动清理锁退出 label = f" [{round_label}]" if round_label else "" start = time.time() @@ -460,6 +488,9 @@ def run_once(round_label=""): # 批量拉取这些股票的价格 prices = fetch_all_prices(list(check_codes)) + # 本轮进区事件收集（聚合成一条摘要推送，替代逐条轰炸） + _zone_entries = [] + for d in active: code = d["code"] trig = d.get("trigger", {}) @@ -496,7 +527,7 @@ def run_once(round_label=""): if _budget_low: outputs.append(f" 📨 止损触发(超时跳过重评)→已推送Dad") if _can_push(code, "stop_loss"): - push_to_xmpp(f"⚠️ {name}({code}) {price} → 跌破止损{hi}！") + _push_action("止损告警", f"⚠️ {name}({code}) {price} → 跌破止损{hi}！") else: try: cost = d.get("cost", 0) or 0 @@ -512,7 +543,7 @@ def run_once(round_label=""): rr = result.get("rr_ratio", 0) if _can_push(code, "stop_loss"): msg = f"🔔 {name}({code}) 价{price}→触发操作区间{max(buy_lo,0):.2f}~{buy_hi:.2f}，已触发重评|RR={rr}" - push_to_xmpp(msg) + _push_action("操作信号", msg) outputs.append(f" 📨 止损重评→已推送Dad: {action}") except Exception as e: outputs.append(f" ⚠️ 止损重评失败: {e}") @@ -529,11 +560,11 @@ def run_once(round_label=""): extra = f"（{act}）" outputs.append(f"⚡ {name}({code}) {price} → 进入{label}{lo}~{hi}{extra}") record_event(code, name, "entry_zone", price, f"{lo}~{hi}", label) - # 进入区间 → 立即重评并推送给Dad（时间不够则跳过重评直接推原始告警） + # 进入区间 → 立即重评并推送给Dad（时间不够则记入摘要，不逐条轰炸） if _budget_low: if _can_push(code, key): - push_to_xmpp(f"⚡ {name}({code}) {price} → 进入{label}{lo}~{hi}") - outputs.append(f" 📨 区间触发(超时跳过重评)→已推送Dad") + _zone_entries.append(f"{name}({code}) {price}→{label}{lo}~{hi}") + outputs.append(f" 📨 区间触发(超时)→记入摘要") else: try: cost = d.get("cost", 0) or 0 @@ -552,7 +583,7 @@ def run_once(round_label=""): rr = result.get("rr_ratio", 0) if _can_push(code, key): msg = f"🔔 {name}({code}) 价{price}→触发{zone_desc}，已触发重评|RR={rr}" - push_to_xmpp(msg) + _push_action("操作信号", msg) outputs.append(f" 📨 区间触发重评→已推送Dad: {action}") else: reason = f"重评结果:{timing_signal}，不构成操作建议" @@ -568,6 +599,12 @@ def run_once(round_label=""): state[code][key] = False state_updated = True + # === 第二步收尾：进区事件聚合成一条摘要推送（INFO级，30min限1条+截断）=== + if _zone_entries: + _digest = f"📋 {len(_zone_entries)}只进入操作区：\n" + "\n".join(f"• {e}" for e in _zone_entries) + _push_digest("盘中触发", _digest) + outputs.append(f"📨 进区摘要({len(_zone_entries)}只)→已按INFO策略推送") + # === 第三步：买入区偏离检测 + 自动重评 === reassesed_codes = [] # 先做急跌检测（仅持仓，自选股不推送暴跌告警） @@ -595,7 +632,7 @@ def run_once(round_label=""): stop_loss = d.get("stop_loss", 0) sl_note = f" 止损{stop_loss}" if stop_loss else "" msg = f"🔻 {name}({code}) {price} 暴跌{cp:.1f}%！{sl_note}" - push_to_xmpp(msg) + _push_action("急跌告警", msg) outputs.append(msg) state.setdefault(code, {})["__sharp_decline_triggered"] = True state_updated = True @@ -716,6 +753,9 @@ def run_once(round_label=""): # 输出耗时 print(f"⏱{label} {elapsed:.1f}s", flush=True) + # 取消超时定时器（正常完成） + signal.alarm(0) + # 清理进程锁 try: os.remove("/tmp/price_monitor.lock")