fix: promote_candidates 超时修复——限批10+单例守卫+重评超时120s+时长护栏500s

deploy/profile-scripts/promote_candidates.py

@@ -1,21 +1,39 @@
 #!/usr/bin/env python3
-"""promote_candidates.py — 自动提拔候选股入自选
+"""promote_candidates.py — 自动提拔候选股入自选（修复版 2026-08-10）
-
+修复：每次运行限处理 N 个新候选（按评分降序优中选优），
-从 candidates 表读未提拔的候选，评估后自动加入 holding_strategies。
+加单例守卫防重叠，缩短重评子进程超时，加总时长护栏。
 """
-import sys, json, sqlite3
+import sys, json, sqlite3, time
 from pathlib import Path
 from datetime import datetime
 
 DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
 
+# 2026-08-10 修复参数：控制单次运行时长，防 hermes 600s 超时杀进程
+MAX_NEW_PER_RUN = 10        # 每次最多提拔 10 个新候选（评分降序优中选优）
+REASSESS_TIMEOUT = 120       # 重评子进程超时（原480s→120s，重评是辅助不阻塞主流程）
+TOTAL_TIME_BUDGET = 500      # 总时长护栏（hermes child_timeout=600，留100s余量）
+START_TIME = time.time()
+
+def over_budget():
+    return (time.time() - START_TIME) > TOTAL_TIME_BUDGET
+
 def main():
+    # ── 单例守卫（2026-05-31 铁律：常驻/定时脚本必须防重复）──
+    import fcntl
+    lock_path = "/tmp/promote_candidates.lock"
+    lock_f = open(lock_path, "w")
+    try:
+        fcntl.flock(lock_f, fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except OSError:
+        print("[PROMOTE] 已有实例在运行，退出")
+        return
+
     conn = sqlite3.connect(str(DB_PATH), timeout=30)
     conn.execute("PRAGMA busy_timeout=30000")
     conn.row_factory = sqlite3.Row
     
     # 读未提拔候选（按评分降序）
-    # 2026-07-24 老爸"优中选优"：score>=7 才可入候选评估（原 4 = 91%通过率等于没门槛）
     rows = conn.execute("""
         SELECT c.code, c.name, c.score_final, c.entry_range, c.stop_loss, c.target
         FROM candidates c
@@ -30,9 +48,29 @@ def main():
         conn.close()
         return
     
+    print(f"[PROMOTE] 待处理 {len(rows)} 个候选（本次最多 {MAX_NEW_PER_RUN} 个新提拔）", flush=True)
     promoted = 0
+    processed = 0
     for r in rows:
+        if over_budget():
+            print(f"[PROMOTE] 已达时长护栏 {TOTAL_TIME_BUDGET}s，停止本批（已处理{processed}）", flush=True)
+            break
+        # 已有"在自选"的候选（上次已插但被标记或重复）快速跳过，不计入 new 配额
         code = str(r[0])
+        exists = conn.execute(
+            "SELECT id FROM holding_strategies WHERE code=? AND status='active'",
+            (code,)
+        ).fetchone()
+        if exists:
+            conn.execute("UPDATE candidates SET promoted=1 WHERE code=?", (code,))
+            processed += 1
+            continue
+        
+        # 已达本批新提拔配额 → 停止（剩余的留到下批，按评分降序保证优中选优）
+        if promoted >= MAX_NEW_PER_RUN:
+            print(f"[PROMOTE] 本批已达 {MAX_NEW_PER_RUN} 个新提拔配额，停止（剩余留待下批）", flush=True)
+            break
+        
         name = r[1] or code
         score = r[2] or 0
         entry_range = r[3] or ""
@@ -48,17 +86,7 @@ def main():
                 eh = float(parts[1])
             except: pass
         
-        # 查是否已在 holding_strategies
+        # 验证实时价格
-        exists = conn.execute(
-            "SELECT id FROM holding_strategies WHERE code=? AND status='active'",
-            (code,)
-        ).fetchone()
-        if exists:
-            conn.execute("UPDATE candidates SET promoted=1 WHERE code=?", (code,))
-            print(f"  ⏭ {code} {name} 已在自选中，标记promoted")
-            continue
-        
-        # 验证实时价格：无有效价格的候选股不入自选（防假数据污染）
         _price = 0.0
         try:
             import subprocess, json as _jj
@@ -68,16 +96,18 @@ def main():
             _price = float(_q.get("price", 0))
             if _price <= 0:
                 print(f"  ⏭ {code} {name} 无实时价格，跳过")
+                processed += 1
                 continue
         except Exception as _e:
             print(f"  ⏭ {code} {name} 价格获取失败({_e})，跳过")
+            processed += 1
             continue
 
-        # ── 优中选优闸（2026-07-24 老爸）：ST排除 + 技术位锚定参数 + RR>=2.0 ──
+        # 优中选优闸：ST排除 + 技术位锚定 + RR>=2.0
         if "ST" in (name or "").upper():
             print(f"  ⏭ {code} {name} ST股，不入自选")
+            processed += 1
             continue
-        # 技术位锚定：不信扫描器拍的 entry/sl/tp，用 ta.full_analysis 的确定性技术位重定
         try:
             import sys as _s
             if '/home/hmo/MoFin/deploy/profile-scripts' not in _s.path:
@@ -88,10 +118,10 @@ def main():
             _ws, _ss = _sr.get("weak_support"), _sr.get("strong_support")
             _wr, _sr2 = _sr.get("weak_resist"), _sr.get("strong_resist")
             if _ws and _wr and _price > 0:
-                el = round(_ws * 0.995, 2)               # 区下沿贴弱撑
+                el = round(_ws * 0.995, 2)
-                eh = round(min(_wr, _price * 1.05), 2)   # 区上沿取弱压（且不超现价5%）
+                eh = round(min(_wr, _price * 1.05), 2)
-                sl = round((_ss or _ws) * 0.985, 2)      # 止损=强撑下1.5%（无强撑用弱撑）
+                sl = round((_ss or _ws) * 0.985, 2)
-                tp = round(_sr2 or _wr * 1.15, 2)        # 止盈=强压（无强压则弱压+15%）
+                tp = round(_sr2 or _wr * 1.15, 2)
         except Exception as _te:
             print(f"  ⚠️ {code} 技术位锚定失败({_te})，用扫描器参数", flush=True)
         if el > 0 and eh > el and sl > 0 and tp > 0:
@@ -99,12 +129,14 @@ def main():
             _rr = (tp - _mid) / (_mid - sl) if (_mid - sl) > 0 else 0
             if _rr < 2.0:
                 print(f"  ⏭ {code} {name} RR={_rr:.2f}<2.0，不入自选")
+                processed += 1
                 continue
         else:
             print(f"  ⏭ {code} {name} 锚定后参数无效(区{el}~{eh} 损{sl} 盈{tp})，跳过")
+            processed += 1
             continue
         
-        # 构建策略（2026-07-24 老爸：提拔不直接给"买入"——先入观察，12维确认后再升）
+        # 构建策略
         now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
         timing_signal = "关注"
         price_est = (el + eh) / 2 if el > 0 and eh > 0 else 0
@@ -118,7 +150,7 @@ def main():
         reason_text.append(f"评分{score}")
         action = " | ".join(reason_text) if reason_text else f"市场扫描发现(评分{score})"
         
-        # ── 容量闸：自选上限60只，超出时删综合分最弱的（RR低优先）──
+        # 容量闸：自选上限60只
         MAX_WATCH = 60
         wl_count = conn.execute(
             "SELECT COUNT(*) FROM holding_strategies WHERE status='active' AND decision_type='自选策略'").fetchone()[0]
@@ -138,6 +170,7 @@ def main():
                 print(f"  🔴 容量淘汰: {weakest[0]} {weakest[1]} (RR={weakest[2]})", flush=True)
             else:
                 print(f"  ⏭ 自选已满{MAX_WATCH}且现有标的均RR>=2.0，{code}暂缓提拔", flush=True)
+                processed += 1
                 continue
 
         cur = conn.execute("""
@@ -154,16 +187,18 @@ def main():
         conn.execute("UPDATE candidates SET promoted=1 WHERE code=?", (code,))
         if newly_added:
             promoted += 1
+            processed += 1
             print(f"  ✅ {code} {name} 评分{score} → 已加入自选({timing_signal})", flush=True)
         else:
             print(f"  ⏭ {code} {name} 已在自选策略中，标记promoted", flush=True)
+            processed += 1
         
-        # 触发全量重评（生成完整9维策略）——仅新插入的股票需要
+        # 触发全量重评（仅新插入）
         if newly_added:
             try:
                 import subprocess as _sp
                 r = _sp.run(["python3", "/home/hmo/MoFin/deploy/profile-scripts/per_stock_reassess.py", code],
-                           capture_output=True, text=True, timeout=480)
+                           capture_output=True, text=True, timeout=REASSESS_TIMEOUT)
                 if r.returncode == 0:
                     print(f"    重评完成", flush=True)
                 else:
@@ -172,19 +207,20 @@ def main():
                 print(f"    重评异常: {e}", flush=True)
     
     conn.commit()
-    print(f"\n[PROMOTE] 本次提拔{promoted}只", flush=True)
+    print(f"\n[PROMOTE] 本次处理{processed}个，提拔{promoted}只（剩余留待下批）", flush=True)
     
     # 推XMPP
     if promoted > 0:
         try:
             import urllib.request
-            msg = f"📈 自动提拔{promoted}只候选入自选"
+            body = f"🤖 候选自动提拔: {promoted}只新入自选（评分降序优中选优，每批≤{MAX_NEW_PER_RUN}）"
-            payload = json.dumps({"to": "hmo@yoin.fun", "body": msg, "type": "chat"}).encode()
+            req = urllib.request.Request("http://127.0.0.1:5805/",
-            req = urllib.request.Request("http://127.0.0.1:5805/", data=payload,
+                data=json.dumps({"to": "hmo@yoin.fun", "body": body, "type": "chat"}).encode(),
-                                        headers={"Content-Type": "application/json"})
+                headers={"Content-Type": "application/json"})
             urllib.request.urlopen(req, timeout=5)
-        except Exception:
+            print("[PROMOTE] XMPP 通知已发送", flush=True)
-            pass
+        except Exception as e:
+            print(f"[PROMOTE] XMPP 通知失败: {e}", flush=True)
     
     conn.close()