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

feat: price_monitor重评并发化——zones循环触发快判收集pending,循环后ThreadPoolExecutor(6=OCG router 6key)并发跑12维重评+结果推送(止损/区间两分支逻辑提取为_reassess_one,共享_can_push加锁), 老莫:6个key在搞什么串行

Browse Source
This commit is contained in:
xxm
2026-08-24 16:04:15 +08:00
parent b2f0c72977
commit 20abd77e09
1 changed files with 136 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/profile-scripts/price_monitor.py
+136 -100
View File
@@ -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。