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

fix: 暴跌告警仅持仓触发 + 状态立即持久化防重复推送

Browse Source
This commit is contained in:
知微
2026-07-14 15:33:35 +08:00
parent 2c0d6ff587
commit b4a5020cf5
1 changed files with 79 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/price_monitor.py
+79 -46
View File
@@ -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)