feat: 信号权威层级+推荐转场推送统一+🔔分级修正

1. 权威层级: 新鲜(<20h)12维动作级信号, 技术路径(regenerate_all/ price_monitor)无权降级为关注/信号不充分/持有——只有LLM路径或风险信号(卖出/止盈)可覆盖。根治推荐区'出现-消失'抖动 2. 推荐转场推送统一收拢到 mofin_db.push_recommend_alert: tag新转为current_recommend时推送(全路径), 含质量门禁; 移除batch_reassess内联推送(防双重告警) 3. price_monitor 🔔分级: 只有持仓股的卖出/止盈才ACTION直推, 买入/加仓机会一律进INFO摘要(老爸: 这种消息不要再发了)
2026-07-22 09:48:58 +08:00
parent a42e3a16e5
commit 8c2226e2b4
3 changed files with 86 additions and 49 deletions
@@ -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}")
@@ -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}")
@@ -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}"