#!/usr/bin/env python3 # -*- coding: utf-8 -*- """promote_reassess_backfill.py — 新提拔自选的重评补推(老莫:未就绪的要补推) promote 结束后 detach 启动: 1. 读 promote 记录的新提拔候选(holding_strategies 里 full_analysis 为空/短的 active 自选) 2. 对未就绪的补跑 per_stock_reassess(300s 超时) 3. 收集就绪的完整重评,通过 XMPP 补推给老莫 """ import sys, os, json, time, subprocess, sqlite3, urllib.request from datetime import datetime DB = "/home/hmo/MoFin/data/mofin.db" PY = "/home/hmo/MoFin/venv/bin/python" REASSESS = "/home/hmo/MoFin/deploy/profile-scripts/per_stock_reassess.py" XMPP = "http://127.0.0.1:5805/" TIMEOUT = 300 MAX_RETRY = 2 def xmpp_push(body): try: req = urllib.request.Request( XMPP, data=json.dumps({"to": "hmo@yoin.fun", "body": body, "type": "chat"}).encode(), headers={"Content-Type": "application/json"}) urllib.request.urlopen(req, timeout=10) return True except Exception as e: print(f" ⚠️ XMPP 补推失败: {e}", flush=True) return False def main(wait_sec=90): # 给首次重评完成时间(promote 触发的 REASSESS 可能还在跑) print(f"[BACKFILL] 等待 {wait_sec}s 后检查未就绪重评...", flush=True) time.sleep(wait_sec) conn = sqlite3.connect(DB, timeout=30) conn.execute("PRAGMA busy_timeout=30000") # 最近 30 分钟新提拔的自选,full_analysis 为空/短(未就绪) rows = conn.execute(""" SELECT code, name, strategy_name, rr_ratio, LENGTH(full_analysis) as fa_len, created_at FROM holding_strategies WHERE status='active' AND decision_type='自选策略' AND created_at >= datetime('now', 'localtime', '-30 minutes') AND (full_analysis IS NULL OR LENGTH(full_analysis) < 50) ORDER BY created_at DESC """).fetchall() conn.close() if not rows: print("[BACKFILL] 无未就绪重评", flush=True) return print(f"[BACKFILL] 发现 {len(rows)} 只未就绪重评,逐一补跑...", flush=True) backfilled = [] for code, name, strat, rr, fa_len, created in rows: ok = False for attempt in range(MAX_RETRY): try: r = subprocess.run([PY, REASSESS, code], capture_output=True, text=True, timeout=TIMEOUT) if r.returncode == 0: ok = True break else: print(f" ⚠️ {code} 重评失败(尝试{attempt+1}): {r.stderr.strip()[:100]}", flush=True) except Exception as e: print(f" ⚠️ {code} 重评异常(尝试{attempt+1}): {e}", flush=True) time.sleep(5) if ok: backfilled.append((code, name, strat, rr)) if not backfilled: print("[BACKFILL] 补跑后仍无就绪报告,跳过补推", flush=True) return # 收集就绪的完整报告 conn = sqlite3.connect(DB, timeout=30) codes = ",".join("?" * len(backfilled)) ready = conn.execute( f"SELECT code, name, strategy_name, rr_ratio, full_analysis FROM holding_strategies " f"WHERE status='active' AND code IN ({codes})", tuple(c[0] for c in backfilled)).fetchall() conn.close() # 2026-08-25 老莫新规则:只推有操作建议信号的,单只标准通道(门禁+12维全文), # 观望/关注不推XMPP只归档broadcast _ACTION_SIG = ("买入", "可买入", "可加仓") pushed = 0 conn2 = sqlite3.connect(DB, timeout=30) for code, name, strat, rr, fa in ready: tsig = conn2.execute( "SELECT timing_signal FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() if tsig and any(s in str(tsig[0] or "") for s in _ACTION_SIG): try: from mofin_db import push_recommend_alert if push_recommend_alert(conn2, code): pushed += 1 except Exception as e: print(f" ⚠️ {code} 补推失败: {e}", flush=True) conn2.close() print(f"[BACKFILL] 补推完成: {pushed}只有操作建议已推XMPP,其余仅归档broadcast", flush=True) if __name__ == "__main__": wait = int(sys.argv[1]) if len(sys.argv) > 1 else 90 main(wait)