From b4a5020cf563c86033502b37aa9b176e27d95fb9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=9F=A5=E5=BE=AE?= Date: Tue, 14 Jul 2026 15:33:35 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=9A=B4=E8=B7=8C=E5=91=8A=E8=AD=A6?= =?UTF-8?q?=E4=BB=85=E6=8C=81=E4=BB=93=E8=A7=A6=E5=8F=91=20+=20=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E7=AB=8B=E5=8D=B3=E6=8C=81=E4=B9=85=E5=8C=96=E9=98=B2?= =?UTF-8?q?=E9=87=8D=E5=A4=8D=E6=8E=A8=E9=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/price_monitor.py | 125 +++++++++++++++++++++++++-------------- 1 file changed, 79 insertions(+), 46 deletions(-) diff --git a/scripts/price_monitor.py b/scripts/price_monitor.py index 9074632c..5c1a3641 100644 --- a/scripts/price_monitor.py +++ b/scripts/price_monitor.py @@ -385,9 +385,22 @@ def get_trigger_zones(trigger): return zones +def _cleanup_lock(): + """清理进程锁文件""" + try: + os.remove("/tmp/price_monitor.lock") + except Exception: + pass + +def _handle_sigterm(signum, frame): + """收到SIGTERM时清理锁文件后退出""" + _cleanup_lock() + sys.exit(0) + def run_once(round_label=""): """执行一轮完整的监控流程""" - import os # 必须在开头import，否则os变量会被后面的局部import绑定覆盖 + import os, signal # 必须在开头import，否则os变量会被后面的局部import绑定覆盖 + signal.signal(signal.SIGTERM, _handle_sigterm) os.nice(10) # 降低优先级，避免与DB其他写操作抢占 # ── 进程锁：同一时间只跑一个实例 ── _lk = "/tmp/price_monitor.lock" @@ -405,6 +418,7 @@ def run_once(round_label=""): label = f" [{round_label}]" if round_label else "" start = time.time() + TIME_BUDGET = 90 # 预留30s给输出和清理，90s内必须完成核心逻辑 # === 第一步：一次性刷新所有价格 === refreshed = refresh_data_prices() @@ -477,6 +491,9 @@ def run_once(round_label=""): if code not in state: state[code] = {} + # 时间预算检查：如果超时，跳过重评只做状态记录 + _budget_low = (time.time() - start) > TIME_BUDGET + for key, label, lo, hi in zones: in_zone = lo <= price <= hi prev_in_zone = state[code].get(key, None) @@ -485,25 +502,30 @@ def run_once(round_label=""): if key == "stop_loss": outputs.append(f"⚠️ {name}({code}) {price} → 跌破止损{hi}！") record_event(code, name, "stop_loss", price, str(hi)) - # 止损触发 → 立即重评并推送给Dad - try: - cost = d.get("cost", 0) or 0 - shares = d.get("shares", 0) or 0 - current_action = d.get("action", "") - result = reassess_with_context(code, name, price, cost, shares, current_action) - if result: - timing_signal = result.get("timing_signal", "") - action = result.get("action", "") - if "买入" in timing_signal or "加仓" in timing_signal or timing_signal in ("卖出","止盈"): - buy_lo = d.get("entry_low", 0) - buy_hi = d.get("entry_high", 0) - 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) - outputs.append(f" 📨 止损重评→已推送Dad: {action}") - except Exception as e: - outputs.append(f" ⚠️ 止损重评失败: {e}") + # 止损触发 → 立即重评并推送给Dad（时间不够则直接推原始告警） + if _budget_low: + outputs.append(f" 📨 止损触发(超时跳过重评)→已推送Dad") + if _can_push(code, "stop_loss"): + push_to_xmpp(f"⚠️ {name}({code}) {price} → 跌破止损{hi}！") + else: + try: + cost = d.get("cost", 0) or 0 + shares = d.get("shares", 0) or 0 + current_action = d.get("action", "") + result = reassess_with_context(code, name, price, cost, shares, current_action) + if result: + timing_signal = result.get("timing_signal", "") + action = result.get("action", "") + if "买入" in timing_signal or "加仓" in timing_signal or timing_signal in ("卖出","止盈"): + buy_lo = d.get("entry_low", 0) + buy_hi = d.get("entry_high", 0) + 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) + outputs.append(f" 📨 止损重评→已推送Dad: {action}") + except Exception as e: + outputs.append(f" ⚠️ 止损重评失败: {e}") else: extra = "" if "_price" in key: @@ -517,31 +539,36 @@ 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 - try: - cost = d.get("cost", 0) or 0 - shares = d.get("shares", 0) or 0 - current_action = d.get("action", "") - result = reassess_with_context(code, name, price, cost, shares, current_action) - if result: - timing_signal = result.get("timing_signal", "") - action = result.get("action", "") - # 格式化区间描述（止盈区lo=0时美化显示） - if key == "take_profit_zone" and lo == 0: - zone_desc = f"止盈监控(目标{hi:.0f})" - else: - zone_desc = f"操作区间{lo}~{hi}" - if "买入" in timing_signal or "加仓" in timing_signal or timing_signal in ("卖出","止盈"): - rr = result.get("rr_ratio", 0) - if _can_push(code, key): - msg = f"🔔 {name}({code}) 价{price}→触发{zone_desc}，已触发重评|RR={rr}" - push_to_xmpp(msg) - outputs.append(f" 📨 区间触发重评→已推送Dad: {action}") - else: - reason = f"重评结果:{timing_signal}，不构成操作建议" - outputs.append(f" 📋 本地日志(不推): {reason}") - except Exception as e: - outputs.append(f" ⚠️ 区间重评失败: {e}") + # 进入区间 → 立即重评并推送给Dad（时间不够则跳过重评直接推原始告警） + if _budget_low: + if _can_push(code, key): + push_to_xmpp(f"⚡ {name}({code}) {price} → 进入{label}{lo}~{hi}") + outputs.append(f" 📨 区间触发(超时跳过重评)→已推送Dad") + else: + try: + cost = d.get("cost", 0) or 0 + shares = d.get("shares", 0) or 0 + current_action = d.get("action", "") + result = reassess_with_context(code, name, price, cost, shares, current_action) + if result: + timing_signal = result.get("timing_signal", "") + action = result.get("action", "") + # 格式化区间描述（止盈区lo=0时美化显示） + if key == "take_profit_zone" and lo == 0: + zone_desc = f"止盈监控(目标{hi:.0f})" + else: + zone_desc = f"操作区间{lo}~{hi}" + if "买入" in timing_signal or "加仓" in timing_signal or timing_signal in ("卖出","止盈"): + rr = result.get("rr_ratio", 0) + if _can_push(code, key): + msg = f"🔔 {name}({code}) 价{price}→触发{zone_desc}，已触发重评|RR={rr}" + push_to_xmpp(msg) + outputs.append(f" 📨 区间触发重评→已推送Dad: {action}") + else: + reason = f"重评结果:{timing_signal}，不构成操作建议" + outputs.append(f" 📋 本地日志(不推): {reason}") + except Exception as e: + outputs.append(f" ⚠️ 区间重评失败: {e}") state[code][key] = True state_updated = True @@ -553,9 +580,13 @@ def run_once(round_label=""): # === 第三步：买入区偏离检测 + 自动重评 === reassesed_codes = [] - # 先做急跌检测（所有持仓，不依赖买入区） + # 先做急跌检测（仅持仓，自选股不推送暴跌告警） + holdings_codes = {d["code"] for d in active if d.get("shares", 0) > 0} for d in active: code = d["code"] + # 非持仓跳过 + if code not in holdings_codes: + continue name = d.get("name", code) price_info = prices.get(code) if not price_info: @@ -578,6 +609,8 @@ def run_once(round_label=""): outputs.append(msg) state.setdefault(code, {})["__sharp_decline_triggered"] = True state_updated = True + # 立即持久化，防止后续超时导致状态丢失而重复推送 + save_state(state) elif cp > -5: # 反弹后清除告警标记，下次再跌还能报 state.setdefault(code, {}).pop("__sharp_decline_triggered", None)