diff --git a/deploy/profile-scripts/review_needed_watchdog.py b/deploy/profile-scripts/review_needed_watchdog.py index 13751ed6..88d3511c 100644 --- a/deploy/profile-scripts/review_needed_watchdog.py +++ b/deploy/profile-scripts/review_needed_watchdog.py @@ -1,106 +1,108 @@ -#!/usr/bin/env python3 -""" -review_needed_watchdog.py — review_needed 策略自动跟进 - -每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") -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" - -def load_retry(): - try: - return json.load(open(RETRY_FILE)) - except: - return {} - -def save_retry(data): - json.dump(data, open(RETRY_FILE, "w"), indent=2) - -def push_xmpp(text): - 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] 已推送") - except Exception as e: - print(f" [XMPP推送失败] {e}") - -def main(): - dec = read_decisions() - review_list = [d for d in dec.get("decisions", []) if d.get("status") == "review_needed"] - retry_data = load_retry() - today = datetime.date.today().isoformat() - - if not review_list: - print("[SILENT] 无待处理策略") - return - - print(f"发现 {len(review_list)} 只 review_needed 策略") - changes = False - for d in review_list: - code = d["code"] - name = d.get("name", 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}次,跳过") - continue - - print(f" 🔄 {name}({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 - ) - out = (r.stdout or "") + (r.stderr or "") - print(f" {out[:200]}") - - # 重读决策 - 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] - if dead: - names = [] - 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) - - save_retry(retry_data) - if not changes: - print("[SILENT] 状态无变化") - -if __name__ == "__main__": - main() +#!/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)。 +""" +import sys, json, os, datetime, sqlite3 + +sys.path.insert(0, "/home/hmo/MoFin") +os.chdir("/home/hmo/MoFin") +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 Exception: + return {} + +def save_retry(data): + json.dump(data, open(RETRY_FILE, "w"), indent=2) + +def push_alert(body): + """统一网关推送(ACTION 级直通,含分级/去重)""" + try: + from alert_helper import notify, ACTION + notify("复习跟进", body, level=ACTION) + return True + except Exception as e: + print(f" [XMPP推送失败] {e}") + return False + +def main(): + 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] 无 review_needed 策略") + return 0 + print(f"发现 {len(review_list)} 只 review_needed 策略") + + for code, name, status, reassessed in review_list: + nm = name or code + retries = retry_data.get(code, {}).get("count", 0) + 1 + # 同股 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 + 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=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 '无记录'}") + + # 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: + 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) + return 0 + +if __name__ == "__main__": + main()