stale_push_wlin: 按Dad确认流程重构推送逻辑

1. 加重评冷却（4小时）— 查holding_strategies.updated_at
2. 修zone_notes早退bug — 有区间说明时不静默退出
3. zone_notes冷却30分→4小时（匹配重评冷却）
4. 标题自适应：有推荐→自选买入提醒 | 仅区间提醒→操作区间提醒
5. 区间说明格式：'进入操作区间，但重评结果:x'
6. 操作建议标题只在有可操作项时显示
7. has_actionable判定改用实际lines内容

scripts/intraday_health_check.py

@@ -167,7 +167,8 @@ def check_signal_pipeline():
     unproc = 0
     try:
         conn = sqlite3.connect(str(DB_PATH))
-        r = conn.execute("SELECT COUNT(*) FROM signal_news WHERE source LIKE 'xiaoguo%' AND (processed=0 OR processed IS NULL)").fetchone()
+        # 只检查时效内的信号堆积（4小时以内），过期信号不被消费者处理但仍会计入堆积
+        r = conn.execute("SELECT COUNT(*) FROM signal_news WHERE source LIKE 'xiaoguo%' AND (processed=0 OR processed IS NULL) AND created_at > datetime('now', '-4 hours')").fetchone()
         unproc = r[0]
         conn.close()
     except:

scripts/stale_detector.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 """stale_detector.py — 检查所有策略，标记价格偏离/过期的策略
 
-读取 decisions.json 的扁平列表。自选策略和持仓策略分开判断。
+读取 holding_strategies + watchlist_stocks 的DB双源数据。
 可被 cron no_agent 模式调用：stdout 注入到后续 LLM 分析。
 
 输出格式：
@@ -162,6 +162,49 @@ def main():
     except Exception as e:
         print(f"[AUTO_REASSESS FAIL] {e}")
     # ----- 结束 自选股重评 -----
+    # 🔁 重评后重新从DB读取策略数据，刷新to_check
+    try:
+        decisions_list = read_decisions()
+        if not isinstance(decisions_list, list):
+            decisions_list = decisions_list.get("decisions", []) if isinstance(decisions_list, dict) else []
+        to_check = [d for d in decisions_list if (d.get("entry_low") is not None or d.get("entry_high") is not None) and d.get("status") not in EXCLUDED_STATUSES]
+        # 重新合并watchlist_stocks
+        db2 = sqlite3.connect('/home/hmo/MoFin/data/mofin.db')
+        db2.row_factory = sqlite3.Row
+        wl_rows2 = db2.execute(
+            "SELECT code, name, price, entry_low, entry_high, stop_loss, analysis_json "
+            "FROM watchlist_stocks WHERE is_active=1 AND entry_low IS NOT NULL AND entry_high IS NOT NULL"
+        ).fetchall()
+        db2.close()
+        existing_codes2 = {d["code"] for d in to_check}
+        for row in wl_rows2:
+            code = str(row["code"])
+            if code in existing_codes2:
+                continue
+            entry_low = row["entry_low"]
+            entry_high = row["entry_high"]
+            if not entry_low or not entry_high or entry_low <= 0:
+                continue
+            analysis = {}
+            try:
+                analysis = json.loads(row["analysis_json"]) if row["analysis_json"] and row["analysis_json"] != "null" else {}
+            except (json.JSONDecodeError, TypeError):
+                pass
+            action = analysis.get("action", "") if isinstance(analysis, dict) else ""
+            timing_signal = analysis.get("timing_signal", "买入") if isinstance(analysis, dict) else "买入"
+            wl_entry = {
+                "code": code,
+                "name": row["name"] or code,
+                "entry_low": entry_low,
+                "entry_high": entry_high,
+                "stop_loss": row["stop_loss"],
+                "type": "自选策略",
+                "action": action,
+                "timing_signal": timing_signal,
+            }
+            to_check.append(wl_entry)
+    except Exception as e:
+        print(f"[RELOAD FAIL] {e}", file=sys.stderr)
 
     # ----- 组合级监测：读取总仓位 + 弱势比例 -----
     position_pct = 0

scripts/stale_push_wlin.py

@@ -806,7 +806,8 @@ def main():
         return text, gap
 
     # 标准格式：每个可操作标的 — 大盘/行业/个股三面 + 仓位
-    lines.append(f"【💡 操作建议】（当前{n}只自选可操作 | 总资产{total_assets:,.0f}元 现金{available_cash:,.0f}元）")
+    if actionable:
+        lines.append(f"【💡 操作建议】（当前{len(actionable)}只自选可操作 | 总资产{total_assets:,.0f}元 现金{available_cash:,.0f}元）")
     for s in actionable:
         name, code, price, buy_low, buy_high, lot, ratio = s
         d = code_data.get(code, {})
@@ -935,16 +936,26 @@ def main():
     save_cooldown(cooldown)
 
     # 修正可操作数量（剔除冷却跳过后的实际数量）
-    actual_n = len(lines) - (1 if macro_line else 0) - 1  # 减去市场背景 + 操作建议标题
-    if actual_n != n:
-        # 更新操作建议行
-        for i, ln in enumerate(lines):
-            if "【💡 操作建议】" in ln:
-                lines[i] = f"【💡 操作建议】（当前{actual_n}只自选可操作 | 总资产{total_assets:,.0f}元 现金{available_cash:,.0f}元）"
-                break
+    if actionable:
+        actual_n = sum(
+            1 for ln in lines
+            if ln.startswith("  🛒") or ln.startswith("  ⚠️")
+        )
+        if actual_n != len(actionable):
+            for i, ln in enumerate(lines):
+                if "【💡 操作建议】" in ln:
+                    if actual_n > 0:
+                        lines[i] = f"【💡 操作建议】（当前{actual_n}只自选可操作 | 总资产{total_assets:,.0f}元 现金{available_cash:,.0f}元）"
+                    else:
+                        lines.pop(i)  # 全部冷却，移除空标题
+                    break
 
-    if actual_n <= 0:
-        return 0  # 全部冷却中 → 静默，不推
+    # 检查最终是否还有内容要推
+    has_actionable = any(
+        ln.startswith("  🛒") or ln.startswith("  ⚠️") for ln in lines
+    )
+    if not has_actionable and not zone_notes:
+        return 0  # 全部冷却+无区间说明 → 静默
 
     # ── T+2前瞻：扫描近期可能入买区的A股，提前准备现金 ──
     t2_lines = []
@@ -1004,10 +1015,14 @@ def main():
         for name, code, price, buy_low, buy_high, reason, sig in zone_notes:
             lines.append(
                 f"  {name}({code}) 价{price:.2f} 区间{buy_low}~{buy_high} "
-                f"→ 重评结果: {reason}"
+                f"→ 进入操作区间，但重评结果: {reason}"
             )
 
-    lines.insert(0, f"【知微】自选买入提醒 {now} | 总资产{total_assets:,.0f}元")
+    # 标题：有推荐操作→"自选买入提醒"，仅有区间说明→"操作区间提醒"
+    if has_actionable:
+        lines.insert(0, f"【知微】自选买入提醒 {now} | 总资产{total_assets:,.0f}元")
+    else:
+        lines.insert(0, f"【知微】操作区间提醒 {now} | 总资产{total_assets:,.0f}元")
     out = "\n".join(lines)
     print(out)
     push_to_xmpp(out)

scripts/system_audit.py

@@ -41,7 +41,7 @@ def log_ok(area, desc):
 def audit_signals(conn):
     try:
         total = conn.execute("SELECT COUNT(*) FROM signal_news").fetchone()[0]
-        unproc = conn.execute("SELECT COUNT(*) FROM signal_news WHERE source LIKE 'xiaoguo%' AND (processed=0 OR processed IS NULL)").fetchone()[0]
+        unproc = conn.execute("SELECT COUNT(*) FROM signal_news WHERE source LIKE 'xiaoguo%' AND (processed=0 OR processed IS NULL) AND created_at > datetime('now', '-4 hours')").fetchone()[0]
         today = conn.execute("SELECT COUNT(*) FROM signal_news WHERE created_at > datetime('now','-1 day')").fetchone()[0]
         log_ok("信号管道", f"信号库{total}条，今日{today}条，未处理{unproc}条")
         if unproc > 30:

scripts/xiaoguo_signal_consumer.py

@@ -17,7 +17,7 @@ from datetime import datetime
 # 确保 MoFin 根目录在模块搜索路径中（兼容 cron 环境）
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from mo_data import read_watchlist, get_price
+from mo_data import read_watchlist
 from mofin_db import write_watchlist_stock
 
 BASE = Path("/home/hmo/MoFin")
@@ -33,7 +33,7 @@ def clean_proxy():
 
 
 def fetch_quote(code):
-    """拉行情。DB 优先，腾讯 fallback"""
+    """拉行情。DB 优先，腾讯 API fallback"""
     # DB 优先
     try:
         from mofin_db import get_price_from_db
@@ -42,11 +42,18 @@ def fetch_quote(code):
             return {"name":"", "code":code, "price":p, "change_pct":chg or 0}
     except:
         pass
-    # Fallback: mo_data.get_price
+    # Fallback: 腾讯实时行情 API
     try:
-        price, chg = get_price(code)
-        if price is not None:
-            return {"name": "", "code": code, "price": price, "change_pct": chg or 0, "pe": 0, "turnover": 0}
+        url = f"http://qt.gtimg.cn/q={code}"
+        import urllib.request
+        resp = urllib.request.urlopen(url, timeout=10).read().decode("gbk")
+        parts = resp.split("~")
+        if len(parts) > 32:
+            name = parts[1]
+            price = float(parts[3]) if parts[3] else None
+            chg_pct = float(parts[32]) if parts[32] else 0
+            if price:
+                return {"name": name, "code": code, "price": price, "change_pct": chg_pct, "pe": 0, "turnover": 0}
         return {"code": code, "error": "取价失败"}
     except Exception as e:
         return {"code": code, "error": str(e)[:60]}
@@ -186,22 +193,28 @@ def main():
     clean_proxy()
     start = time.time()
     today = datetime.now().strftime("%Y-%m-%d")
-    
+
     conn = get_conn()
     
-    # 读未处理 xiaoguo 信号（今日）
+    # 读未处理 xiaoguo 信号（SIGNAL_MAX_AGE_HOURS 以内）
     rows = conn.execute(
         "SELECT id, sector, overall_sentiment, summary, key_articles, searched_stocks, source "
         "FROM signal_news "
         "WHERE source LIKE 'xiaoguo%' AND (processed=0 OR processed IS NULL) "
-        "AND date(created_at) = ? "
-        "ORDER BY created_at DESC LIMIT 20",
-        (today,)
+        f"AND created_at > datetime('now', '-{SIGNAL_MAX_AGE_HOURS} hours') "
+        "ORDER BY created_at DESC LIMIT 20"
     ).fetchall()
     
     if not rows:
+        # 标记过期信号为已处理（超出时效边界）
+        old = conn.execute(f"SELECT COUNT(*) FROM signal_news WHERE source LIKE 'xiaoguo%' AND (processed=0 OR processed IS NULL) AND created_at <= datetime('now', '-{SIGNAL_MAX_AGE_HOURS} hours')").fetchone()[0]
+        if old:
+            conn.execute(f"UPDATE signal_news SET processed=1 WHERE source LIKE 'xiaoguo%' AND (processed=0 OR processed IS NULL) AND created_at <= datetime('now', '-{SIGNAL_MAX_AGE_HOURS} hours')")
+            conn.commit()
+            print(f"[SILENT] 清理 {old} 条过期信号（>{SIGNAL_MAX_AGE_HOURS}h）")
+        else:
+            print("[SILENT] 今日无未处理小果信号")
         conn.close()
-        print("[SILENT] 今日无未处理小果信号")
         return
     
     # 尝试从 searched_stocks 提取股票代码