fix(P1): review_needed_watchdog对接DB——strategy_lifecycle写holding_strategies.status=review_needed但此watchdog读废弃decisions.json(永远空跑)。改读DB+4h冷却防重复重评+走alert_helper统一网关(ACTINO级)

deploy/profile-scripts/review_needed_watchdog.py

@@ -1,106 +1,108 @@
 #!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""review_needed_watchdog.py 修复（2026-08-19 老莫：P1 架构收敛）
+原 bug：strategy_lifecycle 把 status=review_needed 写入 DB（holding_strategies.status，
+经 per_stock_reassess result.status 写回），但本 watchdog 却读废弃的 decisions.json →
+永远空跑，DB 里 review_needed 没人消费（写入方与消费方脱节）。
+修复：改读 DB（holding_strategies.status='review_needed'），对接 strategy_lifecycle 写入；
+XMPP 走 alert_helper 统一网关（分级 ACTION）。
 """
-review_needed_watchdog.py — review_needed 策略自动跟进
+import sys, json, os, datetime, sqlite3
 
-每30分钟扫描DB中 status=review_needed 的策略：
-1. 对每只策略调用 per_stock_reassess 重评
-2. 重评后 status 变 active → 通过，写入 changelog
-3. 还是 review_needed → retry_count+=1
-4. retry_count>=3 → 推 Dad 人工介入
-"""
-
-import sys, json, os, datetime
-from mo_data import read_decisions
-
-sys.path.insert(0, "/home/hmo/web-dashboard")
+sys.path.insert(0, "/home/hmo/MoFin")
 os.chdir("/home/hmo/MoFin")
-
-DEC_PATH = "/home/hmo/web-dashboard/data/decisions.json"
-RETRY_FILE = "/home/hmo/web-dashboard/data/review_needed_retry.json"
-XMPP_USER = "hmo@yoin.fun"
-XMPP_BRIDGE = "http://192.168.1.246:5805/xmpp/send"
+DB = "/home/hmo/MoFin/data/mofin.db"
+RETRY_FILE = "/home/hmo/MoFin/data/review_needed_retry.json"
+MAX_RETRY = 3
+RETRY_INTERVAL_HOURS = 4  # 同股 4 小时内不重复跟进（配合 per_stock_reassess 冷却）
 
 def load_retry():
     try:
         return json.load(open(RETRY_FILE))
-    except:
+    except Exception:
         return {}
 
 def save_retry(data):
     json.dump(data, open(RETRY_FILE, "w"), indent=2)
 
-def push_xmpp(text):
+def push_alert(body):
+    """统一网关推送（ACTION 级直通，含分级/去重）"""
     try:
-        from urllib.request import Request, urlopen
-        payload = json.dumps({"to": XMPP_USER, "body": text.strip()}).encode()
-        req = Request(XMPP_BRIDGE, data=payload, headers={"Content-Type": "application/json"})
-        urlopen(req, timeout=5)
-        print(f"  [XMPP] 已推送")
+        from alert_helper import notify, ACTION
+        notify("复习跟进", body, level=ACTION)
+        return True
     except Exception as e:
         print(f"  [XMPP推送失败] {e}")
+        return False
 
 def main():
-    dec = read_decisions()
-    review_list = [d for d in dec.get("decisions", []) if d.get("status") == "review_needed"]
+    conn = sqlite3.connect(DB, timeout=30)
+    conn.execute("PRAGMA busy_timeout=30000")
+    # 对接 strategy_lifecycle：读 DB status=review_needed（写入方经 per_stock_reassess 落库）
+    try:
+        rows = conn.execute(
+            "SELECT code, name, status, reassessed_at FROM holding_strategies "
+            "WHERE status='review_needed'",
+        ).fetchall()
+    except Exception as e:
+        print(f"[ERR] review_needed 查询失败: {e}")
+        conn.close()
+        return 0
+    conn.close()
+    review_list = [r for r in rows if r[0]]
     retry_data = load_retry()
     today = datetime.date.today().isoformat()
-
     if not review_list:
-        print("[SILENT] 无待处理策略")
-        return
-
+        print("[SILENT] 无 review_needed 策略")
+        return 0
     print(f"发现 {len(review_list)} 只 review_needed 策略")
-    changes = False
-    for d in review_list:
-        code = d["code"]
-        name = d.get("name", code)
+
+    for code, name, status, reassessed in review_list:
+        nm = name or code
         retries = retry_data.get(code, {}).get("count", 0) + 1
-        retry_data[code] = {"count": retries, "last_attempt": today}
-
-        if retries >= 3:
-            print(f"  ⛔ {name}({code}) 已重试{retries}次，跳过")
+        # 同股 4 小时内已跟进过 → 跳过（防重复重评烧 token）
+        last = retry_data.get(code, {}).get("last_ts", 0)
+        now_ts = datetime.datetime.now().timestamp()
+        if last and now_ts - last < RETRY_INTERVAL_HOURS * 3600:
+            print(f"  ⏭ {nm}({code}) 4小时内已跟进，跳过")
             continue
-
-        print(f"  🔄 {name}({code}) 第{retries}次重试...")
+        retry_data[code] = {"count": retries, "last_attempt": today, "last_ts": now_ts}
+        if retries >= MAX_RETRY:
+            print(f"  ⛔ {nm}({code}) 已重试{retries}次，跳过")
+            continue
+        print(f"  🔄 {nm}({code}) 第{retries}次重试...")
         import subprocess
         r = subprocess.run(
             [sys.executable, "/home/hmo/MoFin/deploy/profile-scripts/per_stock_reassess.py", code],
-            capture_output=True, text=True, timeout=30
+            capture_output=True, text=True, timeout=120
         )
         out = (r.stdout or "") + (r.stderr or "")
         print(f"    {out[:200]}")
+        # 重读 DB
+        conn = sqlite3.connect(DB, timeout=30)
+        new_status = conn.execute("SELECT status FROM holding_strategies WHERE code=? ORDER BY id DESC LIMIT 1", (code,)).fetchone()
+        conn.close()
+        if new_status and new_status[0] == "active":
+            print(f"  ✅ {nm}({code}) 重评通过（status 恢复正常）")
+            retry_data[code] = {"count": 0, "last_attempt": today, "last_ts": now_ts}
+        elif new_status and new_status[0] == "review_needed":
+            print(f"  ❌ {nm}({code}) 仍 review_needed")
+        else:
+            print(f"  ⚠️ {nm}({code}) 状态: {new_status[0] if new_status else '无记录'}")
 
-        # 重读决策
-        dec2 = read_decisions()
-        for d2 in dec2.get("decisions", []):
-            if d2["code"] == code:
-                if d2.get("status") == "active":
-                    print(f"  ✅ {name}({code}) 重评通过！")
-                    retry_data[code] = {"count": 0, "last_attempt": today}
-                    changes = True
-                elif d2.get("status") == "review_needed":
-                    issues = d2.get("quality_issues", {}).get("critical", [])
-                    print(f"  ❌ {name}({code}) 仍 review_needed ({issues})")
-                break
-
-    # 3次以上失败 → 推 Dad
-    dead = [code for code, v in retry_data.items() if v.get("count", 0) >= 3]
+    # 3 次以上失败 → 推 Dad 人工介入（统一网关 ACTION）
+    dead = [code for code, v in retry_data.items() if v.get("count", 0) >= MAX_RETRY]
     if dead:
         names = []
+        conn = sqlite3.connect(DB, timeout=30)
         for code in dead:
-            for d in dec.get("decisions", []):
-                if d["code"] == code:
-                    names.append(f"{d.get('name', code)}({code})")
-                    break
-        msg = f"【知微】策略质量审核 {today}\n以下策略3次自动重评均失败，需人工介入：\n"
-        for n in names:
-            msg += f"  - {n}\n"
-        msg += "\n原因可能是：缺少技术面数据 / 行业信息不完整 / 利润保护目标无法确定。"
-        push_xmpp(msg)
-
+            nm = conn.execute("SELECT name FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
+            names.append(f"{nm[0] if nm else code}({code})")
+        conn.close()
+        msg = f"【MoFin·策略复习】{today}\n以下策略 {MAX_RETRY} 次自动重评均失败：\n" + "\n".join(f"  - {n}" for n in names) + "\n\n原因可能是：缺少技术面数据 / 行业信息不完整 / 参数连续被拒。"
+        push_alert(msg)
     save_retry(retry_data)
-    if not changes:
-        print("[SILENT] 状态无变化")
+    return 0
 
 if __name__ == "__main__":
     main()