diff --git a/deploy/profile-scripts/batch_reassess.py b/deploy/profile-scripts/batch_reassess.py index 450eee54..b040e5e5 100644 --- a/deploy/profile-scripts/batch_reassess.py +++ b/deploy/profile-scripts/batch_reassess.py @@ -376,49 +376,11 @@ def save_result(code, full_text, parsed): # ── 推荐操作 tag 同步(与 XMPP 动作级信号同源)── sync_recommend_tag(conn, code, parsed.get("signal", "")) - # 买入信号→推XMPP通知(在conn close前执行)——推送质量门禁: - # 价格必须>0(live_prices实时价)、区间有效(下沿<上沿<下沿x3)、现价不超过上沿5%、 - # 损<下沿、盈>上沿、损在(0.5x~1.0x)现价内。任何一项不过 → 不推,只记日志。 - if parsed.get("signal") == "买入": - try: - _nr = conn.execute("SELECT name FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() - _lp = conn.execute("SELECT price FROM live_prices WHERE code=?", (code,)).fetchone() - _name = _nr[0] if _nr else code - _p = _lp[0] if _lp and _lp[0] else 0 - _el = parsed.get("entry_low", 0) - _eh = parsed.get("entry_high", 0) - _sl = parsed.get("stop_loss", 0) - _tp = parsed.get("take_profit", 0) - _pos = parsed.get("position", "") - _ok, _why = _validate_buy_alert(_p, _el, _eh, _sl, _tp) - if _ok: - _msg = f"📈 {_name}({code}) 价{_p}→12维分析生成买入信号!区间{_el}~{_eh} 损{_sl} 盈{_tp} 仓位{_pos}" - from alert_helper import notify as _notify, ACTION as _ACT - _notify("买入信号", _msg, _ACT) - print(f" \U0001f4e8 XMPP推送成功: {_msg[:60]}") - else: - print(f" ⚠️ 买入信号未过推送门禁({_why}),仅记日志不推送", flush=True) - except Exception as _e: - print(f" \u26a0\ufe0f XMPP推送失败: {_e}") + # 买入信号推送已统一收拢到 sync_recommend_tag 的转场推送(防双重告警)。 + # 本路径只负责写库+tag,推送由 mofin_db.push_recommend_alert 在 tag 转场时触发。 conn.close() - -def _validate_buy_alert(price, el, eh, sl, tp): - """买入信号推送门禁(垃圾信号不发)。 - 返回 (ok, reason)""" - if not price or price <= 0: - return False, f"无实时价格({price})" - if not (el > 0 and eh > el and eh < el * 3): - return False, f"区间无效({el}~{eh})" - if price > eh * 1.05: - return False, f"现价{price}高于区间上沿{eh}超5%(追高信号不推)" - if not (sl > 0 and sl < el and price * 0.5 <= sl <= price): - return False, f"止损{sl}不合理(需0.5x~1.0x现价且<下沿{el})" - if not (tp > eh and tp > sl): - return False, f"止盈{tp}需>上沿{eh}且>止损{sl}" - return True, "" - def process_stock(code, force_today=False): """处理单只股票""" print(f"\n{'='*50}") diff --git a/deploy/profile-scripts/price_monitor.py b/deploy/profile-scripts/price_monitor.py index 25f5cd92..0ca7ed6e 100644 --- a/deploy/profile-scripts/price_monitor.py +++ b/deploy/profile-scripts/price_monitor.py @@ -541,9 +541,13 @@ def run_once(round_label=""): 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_action("操作信号", msg) + # 分级(老爸规则:只有持仓风控动作才ACTION直推;买入/加仓机会进摘要) + if timing_signal in ("卖出","止盈") and (d.get("shares") or 0) > 0: + if _can_push(code, "stop_loss"): + msg = f"🔔 {name}({code}) 价{price}→触发操作区间{max(buy_lo,0):.2f}~{buy_hi:.2f},已触发重评|RR={rr}" + _push_action("操作信号", msg) + else: + _zone_entries.append(f"{name}({code}) {price}→入区+重评{timing_signal}|RR={rr}") outputs.append(f" 📨 止损重评→已推送Dad: {action}") except Exception as e: outputs.append(f" ⚠️ 止损重评失败: {e}") @@ -581,10 +585,15 @@ def run_once(round_label=""): 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_action("操作信号", msg) + # 分级(老爸规则:只有持仓风控动作才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}") + else: + _zone_entries.append(f"{name}({code}) {price}→{zone_desc}+重评{timing_signal}|RR={rr}") + outputs.append(f" 📋 机会记入摘要: {timing_signal} RR={rr}") else: reason = f"重评结果:{timing_signal},不构成操作建议" outputs.append(f" 📋 本地日志(不推): {reason}") diff --git a/mofin_db.py b/mofin_db.py index 605e5b24..f9da99bf 100644 --- a/mofin_db.py +++ b/mofin_db.py @@ -1114,12 +1114,17 @@ def sync_recommend_tag(conn, code: str, timing_signal: str): active_manual(人工标记)永不动。与 XMPP 动作级告警同源(红线#12)。""" _ACTION_SIGNALS = ("买入", "可买入", "可加仓", "卖出", "止盈") try: + _old_tag_row = conn.execute( + "SELECT tag FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() + _old_tag = (_old_tag_row[0] or '') if _old_tag_row else '' if timing_signal in _ACTION_SIGNALS: conn.execute( "UPDATE holding_strategies SET tag='current_recommend' " "WHERE code=? AND status='active' AND (tag IS NULL OR tag != 'active_manual')", (code,)) conn.commit() + if _old_tag != 'current_recommend': + push_recommend_alert(conn, code) # 新推荐 → 推送 elif timing_signal: conn.execute( "UPDATE holding_strategies SET tag='' " @@ -1130,6 +1135,43 @@ def sync_recommend_tag(conn, code: str, timing_signal: str): print(f" [TAG SYNC] {code} 失败: {e}", flush=True) +def push_recommend_alert(conn, code: str): + """推荐操作 XMPP 推送(tag 转为 current_recommend 时调用,全路径统一)。 + 质量门禁:实时价>0、区间有效(下沿<上沿<下沿x3)、现价不超上沿5%、 + 损<下沿且在(0.5x~1.0x)现价内、盈>上沿>损。不过不推。""" + try: + row = conn.execute( + "SELECT name, timing_signal, entry_low, entry_high, stop_loss, take_profit, " + "rr_ratio, position_advice FROM holding_strategies WHERE code=? AND status='active'", + (code,)).fetchone() + if not row: + return False + name, sig, el, eh, sl, tp, rr, pos = row + lp = conn.execute("SELECT price FROM live_prices WHERE code=?", (code,)).fetchone() + price = lp[0] if lp and lp[0] else 0 + el, eh, sl, tp = el or 0, eh or 0, sl or 0, tp or 0 + # ── 门禁 ── + if price <= 0: + print(f" [ALERT] {code} 无实时价,不推", flush=True); return False + if not (el > 0 and eh > el and eh < el * 3): + print(f" [ALERT] {code} 区间无效({el}~{eh}),不推", flush=True); return False + if price > eh * 1.05: + print(f" [ALERT] {code} 现价{price}高超上沿{eh}5%,不推", flush=True); return False + if not (sl > 0 and sl < el and price * 0.5 <= sl <= price): + print(f" [ALERT] {code} 止损{sl}不合理,不推", flush=True); return False + if not (tp > eh and tp > sl): + print(f" [ALERT] {code} 止盈{tp}不合理,不推", flush=True); return False + import sys as _s, os as _o + _s.path.insert(0, _o.path.dirname(_o.path.abspath(__file__))) + from alert_helper import notify, ACTION + msg = (f"📈 {name or code}({code}) 价{price}→12维{sig}!" + f"区间{el}~{eh} 损{sl} 盈{tp} RR={rr or 0} 仓位{pos or '-'}") + return notify("买入信号", msg, ACTION) + except Exception as e: + print(f" [ALERT] {code} 推送异常: {e}", flush=True) + return False + + def snapshot_strategy_history(conn, code: str, source_trigger: str = "write_holding_strategy"): """在修改前快照当前策略到 strategy_history 表。永不抛异常。""" try: @@ -1189,28 +1231,49 @@ def write_holding_strategy(conn, code: str, name: str, data: dict, # 信号降级 → 清除 current_recommend(区域同步消失) # active_manual(人工标记)永远不被自动流程覆盖或清除 _ACTION_SIGNALS = ("买入", "可买入", "可加仓", "卖出", "止盈") + _RISK_SIGNALS = ("卖出", "止盈") _existing_fa = data.get('full_analysis', '') _existing_ra = data.get('reassessed_at', '') _tag_absent = 'tag' not in data _new_sig = data.get('timing_signal', '') or '' _explicit_tag = data.get('tag', None) _old_tag = '' - if not _existing_fa or _tag_absent: + _old_sig = '' + _old_ra = '' + if True: try: - _old = conn.execute("SELECT full_analysis, reassessed_at, tag FROM holding_strategies WHERE code=? ORDER BY id DESC LIMIT 1", (code,)).fetchone() + _old = conn.execute("SELECT full_analysis, reassessed_at, tag, timing_signal FROM holding_strategies WHERE code=? ORDER BY id DESC LIMIT 1", (code,)).fetchone() if _old: if not _existing_fa: if _old[0]: _existing_fa = _old[0] if _old[1]: _existing_ra = _old[1] _old_tag = _old[2] or '' + _old_sig = _old[3] or '' + _old_ra = _old[1] or '' except: pass + # ── 信号权威层级(2026-07-22):新鲜(<20h)12维动作级信号, + # 技术路径(regenerate_all/price_monitor)无权降级为 关注/信号不充分/持有。 + # 只有 LLM 路径(batch_12d/per_stock_12d)或风险信号(卖出/止盈)可以覆盖。── + _TECHNICAL_PATHS = ('write_holding_strategy',) + if source_trigger in _TECHNICAL_PATHS and _old_sig in ("买入", "可买入", "可加仓") \ + and _new_sig not in _ACTION_SIGNALS and _old_ra: + try: + from datetime import datetime as _ddt, timedelta as _dtd + _ra_dt = _ddt.fromisoformat(str(_old_ra)[:19]) + if (_ddt.now() - _ra_dt) < _dtd(hours=20): + print(f" [AUTHORITY] {code} 保留新鲜12维信号'{_old_sig}'({_old_ra[:16]})," + f"拒绝技术路径降级为'{_new_sig}'", flush=True) + _new_sig = _old_sig + data['timing_signal'] = _old_sig + except Exception: + pass if _old_tag == 'active_manual': _existing_tag = 'active_manual' # 人工标记不可动 elif _explicit_tag is not None: _existing_tag = _explicit_tag # 显式传入优先(含''清除) elif _new_sig in _ACTION_SIGNALS: - _existing_tag = 'current_recommend' # 动作级信号 → 自动推荐 + _existing_tag = 'current_recommend' # 动作级信号 → 自动推荐 elif _new_sig and _old_tag == 'current_recommend': _existing_tag = '' # 信号降级 → 清除自动推荐 else: @@ -1266,6 +1329,9 @@ def write_holding_strategy(conn, code: str, name: str, data: dict, _existing_tag, )) conn.commit() + # ── 推荐转场推送:tag 新转为 current_recommend 时全路径统一告警 ── + if _existing_tag == 'current_recommend' and _old_tag != 'current_recommend': + push_recommend_alert(conn, code) return True, f"策略 {code} 已写入" except sqlite3.IntegrityError as e: return False, f"币种约束: {e}"