diff --git a/deploy/profile-scripts/price_monitor.py b/deploy/profile-scripts/price_monitor.py index 529ba145..dc7fd2c9 100644 --- a/deploy/profile-scripts/price_monitor.py +++ b/deploy/profile-scripts/price_monitor.py @@ -172,6 +172,116 @@ try: except ImportError: HAS_REASSESS = False + +# ── 2026-08-24 并发重评：单只"重评+结果分发+推送"（ThreadPoolExecutor 6 路调用）── +_can_push_lock = __import__("threading").Lock() + +def _reassess_one(ctx): + """并发执行单只 12 维重评并按结论推送。返回值: dict(outputs/zone_entries/reassessed) 或 None。 + 取自原 zones 循环内两个调用点(止损/区间分支)的重评+推送逻辑,代价是共享 _can_push 需加锁。""" + code = ctx["code"]; name = ctx["name"]; price = ctx["price"]; d = ctx["d"] + key = ctx["key"]; hi = ctx["hi"]; lo = ctx["lo"] + out, ze = [], [] + try: + cost = d.get("cost", 0) or 0 + shares = d.get("shares", 0) or 0 + current_action = d.get("action", "") + result = _do_llm_reassess(code, name, price, cost, shares, current_action) + rea = [] + if key != "stop_loss" and result: + rea.append(code) + elif key == "stop_loss": + rea.append(code) + if result: + timing_signal = result.get("timing_signal", "") + action = result.get("action", "") + rr = result.get("rr_ratio", 0) + sl_new = result.get("stop_loss", 0) + tp_new = result.get("take_profit", 0) + if key == "stop_loss": + # ── 止损分支推送 ── + reason_extra = "" + for k in ("action_note", "signal_factors"): + v = result.get(k) + if v: + if isinstance(v, list): + v = "、".join(str(x) for x in v) + reason_extra += f" [{k}={v}]" + if timing_signal in ("卖出", "止盈") and (d.get("shares") or 0) > 0: + with _can_push_lock: + _cp = _can_push(code, "stop_loss") + if _cp: + msg = (f"🔔 {name}({code}) 价{price} → 跌破止损，重评结论【{timing_signal}】| RR={rr}" + f" | 止损{sl_new}/止盈{tp_new}" + f" | 操作: {str(action)[:250]}{reason_extra}") + _push_action("操作信号", msg) + elif "持有" in timing_signal or "关注" in timing_signal or "观望" in timing_signal: + try: + import sqlite3 as _sq + _c = _sq.connect("/home/hmo/MoFin/data/mofin.db", timeout=10) + if sl_new: + _c.execute( + "UPDATE holding_strategies SET stop_loss=?, updated_at=datetime('now','localtime') " + "WHERE code=? AND status='active'", (sl_new, code)) + _c.commit() + _c.close() + except Exception: + pass + ze.append(f"{name}({code}) {price}→破止损但重评{timing_signal}，止损已更新至{sl_new}") + out.append(f" 📨 破止损→重评{timing_signal}（不卖出，止损更新{sl_new}）: {str(action)[:120]}") + else: + ze.append(f"{name}({code}) {price}→入区+重评{timing_signal}|RR={rr}") + out.append(f" 📨 止损重评→{timing_signal}: {str(action)[:120]}") + else: + # ── 区间分支推送 ── + 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 ("卖出", "止盈"): + if timing_signal in ("卖出", "止盈") and (d.get("shares") or 0) > 0: + with _can_push_lock: + _cp = _can_push(code, key) + if _cp: + msg = f"🔔 {name}({code}) 价{price}→触发{zone_desc}，已触发重评|RR={rr}" + _push_action("操作信号", msg) + out.append(f" 📨 区间触发重评→已推送Dad: {action}") + try: + _record_signal(code, name, f"风控:{timing_signal} {zone_desc}", strategy=d.get("tag") or d.get("version") or "") + except Exception: + pass + else: + if _is_strategy_active(d): + ze.append(f"{name}({code}) {price}→{zone_desc}+重评{timing_signal}|RR={rr}") + out.append(f" 📋 机会记入摘要: {timing_signal} RR={rr}") + try: + _record_signal(code, name, f"买入机会:{zone_desc} {timing_signal}", strategy=d.get("tag") or d.get("version") or "") + except Exception: + pass + else: + out.append(f" 🧊 温区未激活跳过: {name}({code}) {timing_signal} RR={rr}") + else: + reason = f"重评结果:{timing_signal}，不构成操作建议" + out.append(f" 📋 本地日志(不推): {reason}") + else: + # 重评未生效（冷却/子进程失败）→ 原始告警+人工核查标注(不拿旧策略冒充结论) + with _can_push_lock: + _cp = _can_push(code, "stop_loss" if key == "stop_loss" else key) + if _cp: + if key == "stop_loss": + _push_action("止损告警", f"⚠️ {name}({code}) {price} → 跌破止损{hi}！（12维重评未生效，暂无最新结论，请人工核查）") + else: + _push_action("止损告警", f"⚠️ {name}({code}) {price} → 触发区间{lo}~{hi}但重评未生效，请人工核查") + out.append(f" ⚠️ {code} 重评未生效，已发原始告警") + return {"outputs": out, "zone_entries": ze, "reassessed": rea} + except Exception as e: + out.append(f" ⚠️ {code} 重评失败: {e}") + try: + record_alert(level="error", source="price_monitor", title="重评失败", detail=str(e)[:200], code=code) + except Exception: + pass + return {"outputs": out, "zone_entries": ze, "reassessed": []} + UA = "Mozilla/5.0" # ── XMPP推送 ────────────────────────────────────────────────────────── @@ -654,6 +764,7 @@ def run_once(round_label=""): active = [d for d in dec.get("decisions", []) if d.get("status") == "active"] state = load_state() outputs = [] + _reassess_pending = [] # 2026-08-24 并发重评队列表(循环内只快判收集,循环后统一并发) state_updated = False # 时间冷却：同股同区间30分钟内不重复推 _push_cooldown = {} @@ -751,65 +862,11 @@ def run_once(round_label=""): if _can_push(code, "stop_loss"): _push_action("止损告警", 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 = _do_llm_reassess(code, name, price, cost, shares, current_action) - if result: - timing_signal = result.get("timing_signal", "") - action = result.get("action", "") - # ── 2026-08-17 老莫：进操作区间→重评→按重评发推荐，且带详细原因 ── - # 推送内容 = 重评的完整操作结论（timing_signal + action 全文），不自拼价格/RR - buy_lo = d.get("entry_low", 0) - buy_hi = d.get("entry_high", 0) - rr = result.get("rr_ratio", 0) - sl_new = result.get("stop_loss", 0) - tp_new = result.get("take_profit", 0) - # 汇总重评原因（信号因子/备注） - reason_extra = "" - for k in ("action_note", "signal_factors"): - v = result.get(k) - if v: - if isinstance(v, list): - v = "、".join(str(x) for x in v) - reason_extra += f" [{k}={v}]" - if timing_signal in ("卖出", "止盈") and (d.get("shares") or 0) > 0: - if _can_push(code, "stop_loss"): - msg = (f"🔔 {name}({code}) 价{price} → 跌破止损，重评结论【{timing_signal}】| RR={rr}" - f" | 止损{sl_new}/止盈{tp_new}" - f" | 操作: {str(action)[:250]}{reason_extra}") - _push_action("操作信号", msg) - elif "持有" in timing_signal or "关注" in timing_signal or "观望" in timing_signal: - # 破止损但重评确认持有 → 不推卖出（逻辑自洽），更新止损位到重评值 - try: - import sqlite3 as _sq - _c = _sq.connect("/home/hmo/MoFin/data/mofin.db", timeout=10) - if sl_new: - _c.execute( - "UPDATE holding_strategies SET stop_loss=?, updated_at=datetime('now','localtime') " - "WHERE code=? AND status='active'", (sl_new, code)) - _c.commit() - _c.close() - except Exception: - pass - _zone_entries.append(f"{name}({code}) {price}→破止损但重评{timing_signal}，止损已更新至{sl_new}") - outputs.append(f" 📨 破止损→重评{timing_signal}（不卖出，止损更新{sl_new}）: {str(action)[:120]}") - else: - _zone_entries.append(f"{name}({code}) {price}→入区+重评{timing_signal}|RR={rr}") - outputs.append(f" 📨 止损重评→{timing_signal}: {str(action)[:120]}") - else: - # 2026-08-24：重评未生效（冷却/子进程失败）→发原始告警+人工核查标注， - # 绝不拿旧策略冒充"重评结论"（02202卖出/深套持有鬼消息教训） - if _can_push(code, "stop_loss"): - _push_action("止损告警", f"⚠️ {name}({code}) {price} → 跌破止损{hi}！（12维重评未生效，暂无最新结论，请人工核查）") - outputs.append(f" ⚠️ {code} 重评未生效，已发原始止损告警") - except Exception as e: - outputs.append(f" ⚠️ 止损重评失败: {e}") - try: - record_alert(level="error", source="price_monitor", title="止损重评失败", detail=str(e)[:200], code=code) - except Exception: - pass + # 2026-08-24 并发化：不串行等重评,收集到pending,循环后6路并发(OCG router 6 key) + _reassess_pending.append({ + "code": code, "name": name, "price": price, "d": d, + "key": key, "label": label, "lo": lo, "hi": hi, "trig": trig}) + outputs.append(f" 📨 {code} 破止损→重评入队(并发)") else: extra = "" if "_price" in key: @@ -829,47 +886,11 @@ def run_once(round_label=""): _zone_entries.append(f"{name}({code}) {price}→{label}{lo}~{hi}") outputs.append(f" 📨 区间触发(超时)→记入摘要") else: - try: - cost = d.get("cost", 0) or 0 - shares = d.get("shares", 0) or 0 - current_action = d.get("action", "") - result = _do_llm_reassess(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) - # 分级（老爸规则：只有持仓风控动作才ACTION直推；买入/加仓机会进摘要） - if timing_signal in ("卖出","止盈") and (d.get("shares") or 0) > 0: - if _can_push(code, key): - msg = f"🔔 {name}({code}) 价{price}→触发{zone_desc}，已触发重评|RR={rr}" - _push_action("操作信号", msg) - outputs.append(f" 📨 区间触发重评→已推送Dad: {action}") - # 2026-08-13 信号溯源：风控动作记录 - _record_signal(code, name, f"风控:{timing_signal} {zone_desc}", strategy=d.get("tag") or d.get("version") or "") - else: - # 2026-08-13 温区感知：非激活策略的买入/加仓机会抑制（不进摘要） - if _is_strategy_active(d): - _zone_entries.append(f"{name}({code}) {price}→{zone_desc}+重评{timing_signal}|RR={rr}") - outputs.append(f" 📋 机会记入摘要: {timing_signal} RR={rr}") - # 2026-08-13 信号溯源：买入机会记录 - _record_signal(code, name, f"买入机会:{zone_desc} {timing_signal}", strategy=d.get("tag") or d.get("version") or "") - else: - outputs.append(f" 🧊 温区未激活跳过: {name}({code}) {timing_signal} RR={rr}") - else: - reason = f"重评结果:{timing_signal}，不构成操作建议" - outputs.append(f" 📋 本地日志(不推): {reason}") - except Exception as e: - outputs.append(f" ⚠️ 区间重评失败: {e}") - try: - record_alert(level="error", source="price_monitor", title="区间重评失败", detail=str(e)[:200], code=code) - except Exception: - pass + # 2026-08-24 并发化(同止损分支) + _reassess_pending.append({ + "code": code, "name": name, "price": price, "d": d, + "key": key, "label": label, "lo": lo, "hi": hi, "trig": trig}) + outputs.append(f" 📨 {code} 区间触发→重评入队(并发)") state[code][key] = True state_updated = True @@ -884,6 +905,21 @@ def run_once(round_label=""): if _zone_entries: outputs.append(f"📋 进区事件{len(_zone_entries)}只（仅记日志不推送）: {'; '.join(_zone_entries[:5])}") + # ── 2026-08-24 并发重评（老莫：OCG router 6 key，为什么不并发！）── + # 触发已全部快判完成,这里6路并发跑LLM重评+结果推送 + if _reassess_pending: + import concurrent.futures as _cf2 + outputs.append(f" ⏭ {len(_reassess_pending)} 只重评入队,6路并发(ocg_router 6key)") + _ex = _cf2.ThreadPoolExecutor(max_workers=6) + try: + for _sub in _ex.map(_reassess_one, _reassess_pending): + if _sub: + outputs.extend(_sub.get("outputs", [])) + _zone_entries.extend(_sub.get("zone_entries", [])) + reassesed_codes.extend(_sub.get("reassessed", [])) + finally: + _ex.shutdown(wait=True) + # === 第三步：买入区偏离检测 + 自动重评 === # 2026-08-11 修复：告警/重评只在连续竞价时段执行（避开集合竞价虚拟撮合价误报）。 # 2026-08-14 港股接入：时段判断按市场——A股 9:30-11:30/13:00-15:00，港股 9:30-12:00/13:00-16:00。