116 lines
4.9 KiB
Python
116 lines
4.9 KiB
Python
#!/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
|
||
|
||
# ── 消息通道统一路由(broadcast/xmpp by delivery) ──
|
||
try:
|
||
from messenger import install_stdio_hook as _msh
|
||
_msh()
|
||
except Exception:
|
||
pass
|
||
|
||
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=180 # 2026-08-25 120→180(积压卡t_de0d60f2: LLM重评60-100s,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()
|