fix: tag只许LLM路径写（防技术路径抖动幽灵化） + 推荐通知改批量摘要（防09:55式14连发轰炸）

deploy/profile-scripts/batch_reassess.py

@@ -511,6 +511,12 @@ def main():
     print(f"\n{'='*50}")
     print(f"完成: {ok}成功, {fail}失败, {skip}跳过")
     print(f"{'='*50}")
+    # ── 推荐摘要：本轮新增推荐聚成一条推送（防逐只轰炸）──
+    try:
+        from mofin_db import flush_rec_digest
+        flush_rec_digest()
+    except Exception as _e:
+        print(f"  ⚠️ 推荐摘要发送失败: {_e}")
 
 if __name__ == "__main__":
     main()

mofin_db.py

@@ -1124,7 +1124,7 @@ def sync_recommend_tag(conn, code: str, timing_signal: str):
                 (code,))
             conn.commit()
             if _old_tag != 'current_recommend':
-                push_recommend_alert(conn, code)  # 新推荐 → 推送
+                enqueue_recommend(conn, code)  # 新推荐 → 摘要队列（batch 结束统一发）
         elif timing_signal:
             conn.execute(
                 "UPDATE holding_strategies SET tag='' "
@@ -1135,6 +1135,65 @@ def sync_recommend_tag(conn, code: str, timing_signal: str):
         print(f"  [TAG SYNC] {code} 失败: {e}", flush=True)
 
 
+def enqueue_recommend(conn, code: str):
+    """新推荐入摘要队列（防逐只轰炸）。batch_reassess 跑完后 flush_rec_digest 统一发一条。"""
+    try:
+        import json as _j
+        from datetime import datetime as _dt
+        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
+        name, sig, el, eh, sl, tp, rr, pos = row
+        qf = '/home/hmo/MoFin/gateway/logs/rec_digest_queue.jsonl'
+        import os as _os
+        _os.makedirs(_os.path.dirname(qf), exist_ok=True)
+        with open(qf, 'a', encoding='utf-8') as f:
+            f.write(_j.dumps({"code": code, "name": name, "signal": sig,
+                              "entry_low": el, "entry_high": eh, "stop_loss": sl,
+                              "take_profit": tp, "rr": rr, "position": pos,
+                              "ts": _dt.now().isoformat()}, ensure_ascii=False) + "\n")
+        print(f"  [REC] {code} 已入推荐摘要队列", flush=True)
+    except Exception as e:
+        print(f"  [REC] {code} 入队失败: {e}", flush=True)
+
+
+def flush_rec_digest(max_items=5):
+    """把队列里的新推荐聚成一条 XMPP 摘要（RR 降序，最多 max_items 只）。"""
+    import json as _j, os as _os
+    qf = '/home/hmo/MoFin/gateway/logs/rec_digest_queue.jsonl'
+    if not _os.path.exists(qf):
+        return 0
+    try:
+        with open(qf, encoding='utf-8') as f:
+            items = [_j.loads(l) for l in f if l.strip()]
+    except Exception:
+        return 0
+    if not items:
+        return 0
+    _os.remove(qf)
+    items.sort(key=lambda x: x.get('rr') or 0, reverse=True)
+    top = items[:max_items]
+    lines = [f"📈 新增推荐 {len(items)} 只（按RR排序，精选前{len(top)}）:"]
+    for it in top:
+        lines.append(f"• {it.get('name') or it['code']}({it['code']}) {it['signal']}"
+                     f" 区{it.get('entry_low') or '—'}~{it.get('entry_high') or '—'}"
+                     f" 损{it.get('stop_loss') or '—'} 盈{it.get('take_profit') or '—'}"
+                     f" RR={it.get('rr') or 0}")
+    if len(items) > max_items:
+        lines.append(f"…另有 {len(items) - max_items} 只详见盯盘推荐操作区")
+    try:
+        import sys as _s, os as _o2
+        _s.path.insert(0, '/home/hmo/MoFin/deploy/profile-scripts')
+        from alert_helper import notify, ACTION
+        return notify("推荐操作", "\n".join(lines), ACTION)
+    except Exception as e:
+        print(f"  [REC] 摘要推送失败: {e}", flush=True)
+        return False
+
+
 def push_recommend_alert(conn, code: str):
     """推荐操作 XMPP 推送（tag 转为 current_recommend 时调用，全路径统一）。
     质量门禁：实时价>0、区间有效(下沿<上沿<下沿x3)、现价不超上沿5%、
@@ -1272,12 +1331,12 @@ def write_holding_strategy(conn, code: str, name: str, data: dict,
             _existing_tag = 'active_manual'      # 人工标记不可动
         elif _explicit_tag is not None:
             _existing_tag = _explicit_tag         # 显式传入优先（含''清除）
-        elif _new_sig in _ACTION_SIGNALS:
-            _existing_tag = 'current_recommend'   # 动作级信号 → 自动推荐
-        elif _new_sig and _old_tag == 'current_recommend':
-            _existing_tag = ''                    # 信号降级 → 清除自动推荐
+        elif _new_sig in _ACTION_SIGNALS and source_trigger in ('batch_12d', 'per_stock_12d'):
+            _existing_tag = 'current_recommend'   # 仅 LLM 路径可创建推荐（防技术路径抖动）
+        elif _new_sig and _old_tag == 'current_recommend' and source_trigger in ('batch_12d', 'per_stock_12d'):
+            _existing_tag = ''                    # 仅 LLM 路径可撤销推荐
         else:
-            _existing_tag = _old_tag              # 其余保留
+            _existing_tag = _old_tag              # 技术路径一律不动 tag
         
         # ── 类型守卫：shares 必须是数值，防止字符串写入导致下游崩溃 ──
         _shares = data.get('shares', 0)
@@ -1329,9 +1388,10 @@ 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)
+        # ── 推荐转场：LLM路径新转为 current_recommend → 记入摘要队列（不逐只推送）──
+        if _existing_tag == 'current_recommend' and _old_tag != 'current_recommend' \
+           and source_trigger in ('batch_12d', 'per_stock_12d'):
+            enqueue_recommend(conn, code)
         return True, f"策略 {code} 已写入"
     except sqlite3.IntegrityError as e:
         return False, f"币种约束: {e}"