#!/usr/bin/env python3 """promote_candidates.py — 自动提拔候选股入自选（修复版 2026-08-10） 修复：每次运行限处理 N 个新候选（按评分降序优中选优）， 加单例守卫防重叠，缩短重评子进程超时，加总时长护栏。 """ 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 = 300 # 重评子进程超时（2026-08-18 老莫：120太短,LLM重评90-120s不够,至少300） 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 # 读未提拔候选（按评分降序） rows = conn.execute(""" SELECT c.code, c.name, c.score_final, c.entry_range, c.stop_loss, c.target, c.sector, c.rr, c.source_strategy FROM candidates c WHERE (c.promoted IS NULL OR c.promoted = 0) AND (c.dropped IS NULL OR c.dropped = 0) AND c.score_final >= 7 ORDER BY c.score_final DESC """).fetchall() if not rows: print("[PROMOTE] 无待提拔候选") conn.close() return print(f"[PROMOTE] 待处理 {len(rows)} 个候选（本次最多 {MAX_NEW_PER_RUN} 个新提拔）", flush=True) promoted = 0 processed = 0 newly_promoted_codes = [] for r in rows: if over_budget(): print(f"[PROMOTE] 已达时长护栏 {TOTAL_TIME_BUDGET}s，停止本批（已处理{processed}）", flush=True) break # 已有"在自选"的候选（上次已插但被标记或重复）快速跳过，不计入 new 配额 code = str(r[0]) # 2026-08-11：读取候选 sector（用于 p_oversold RR 例外） cand_sector = str(r[6] or "").strip() if len(r) > 6 else "" source_strategy = str(r[8] or "").strip() if len(r) > 8 else "" # 2026-08-18 fallback：source_strategy 为 unknown/空时用 candidates.sector 作来源 # （自选退出回归候选 sector 是正确策略标签，但 source_strategy 可能 inherited unknown） if not source_strategy or source_strategy == "unknown": source_strategy = cand_sector if cand_sector and cand_sector != "unknown" else "unknown" "" 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 "" sl = r[4] or 0 tp = r[5] or 0 # 解析 entry_range el, eh = 0, 0 if "~" in entry_range: parts = entry_range.split("~") try: el = float(parts[0]) eh = float(parts[1]) except: pass # 验证实时价格 _price = 0.0 try: import subprocess, json as _jj _r = subprocess.run(["python3", "/home/hmo/.hermes/profiles/position-analyst/scripts/stock_quote.py", code], capture_output=True, text=True, timeout=10) _q = _jj.loads(_r.stdout) _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-08-17 老莫规范：RR 由策略算，promote 只卡阈值，不重锚定 ── # 原逻辑用 technical_analysis 近20日低点当止损（暴跌插针）→ RR 从策略的2.3崩到0.82， # promote 管道自 08-05 死12天。现改为读候选自带 c.rr（策略负责计算）。 if "ST" in (name or "").upper(): print(f" ⏭ {code} {name} ST股，不入自选") processed += 1 continue _rr = r[7] # c.rr（策略自带） if _rr is None: print(f" ⏭ {code} {name} 策略未提供RR（scanner需在INSERT时写candidates.rr），跳过") processed += 1 continue # 超跌类策略豁免 RR 门槛（超跌反弹 RR 天然 <2，用策略自带参数即可，同 p_oversold 先例） if _rr < 1.0 and cand_sector not in ("p_oversold", "b_td1_v3", "b_td1"): print(f" ⏭ {code} {name} RR={_rr:.2f}<1.0（保底门槛），不入自选") processed += 1 continue # 用候选自带参数（不再重锚定） _parts = (r[3] or "").split("~") if len(_parts) >= 2: try: el = float(_parts[0]); eh = float(_parts[1]) except Exception: el = eh = 0 else: el = eh = 0 sl = r[4] or 0 tp = r[5] or 0 if not (el > 0 and eh > el and sl > 0 and tp > 0): print(f" ⏭ {code} {name} 候选参数无效(区{el}~{eh} 损{sl} 盈{tp})，跳过") processed += 1 continue # ── B. entry_range 偏离现价>20% → 数据异常，先重评校准（LLM算正确区间）→ 用正确区间入库 ── _dev = 0 try: import subprocess, json as _jj _r2 = subprocess.run(["python3", "/home/hmo/.hermes/profiles/position-analyst/scripts/stock_quote.py", code], capture_output=True, text=True, timeout=10) _q = _jj.loads(_r2.stdout) _cur_price = float(_q.get("price", 0)) if _cur_price > 0: _dev = abs((_cur_price - (el + eh) / 2) / _cur_price) * 100 except Exception: pass if _dev > 20.0: # 数据异常：先重评（LLM算正确区间），成功则用正确区间，失败则跳过 print(f" ⚠️ {code} {name} 买入区{el}~{eh} 偏离现价{_dev:.0f}%（数据异常），先重评校准...", flush=True) try: _r = subprocess.run(["/home/hmo/MoFin/venv/bin/python", "/home/hmo/MoFin/deploy/profile-scripts/per_stock_reassess.py", code], capture_output=True, text=True, timeout=REASSESS_TIMEOUT) if _r.returncode == 0: # 重评后读 holding 里 LLM 校准的 entry_low/high（batch_reassess A 层已覆盖） _cur2 = conn.execute( "SELECT entry_low, entry_high FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() if _cur2 and _cur2[0] and _cur2[1]: el, eh = float(_cur2[0]), float(_cur2[1]) print(f" ✅ 重评校准后买入区 {el}~{eh}", flush=True) else: print(f" ⏭ 重评后无有效买入区，跳过", flush=True) processed += 1 continue else: print(f" ⏭ 重评失败（区间异常且无法校准），跳过", flush=True) processed += 1 continue except Exception as _e: print(f" ⏭ 重评异常({_e})，跳过", flush=True) processed += 1 continue # ── C. 可执行性检查：止损距离≥2%、止盈空间≥3%、修正风报比(含成本)≥2 ── _price_est = (el + eh) / 2 _stop_dist = (_price_est - sl) / _price_est * 100 if sl > 0 else 0 _tp_space = (tp - _price_est) / _price_est * 100 if tp > 0 else 0 # 修正风报比：考虑交易成本（双边~0.2%） _COST = 0.002 _rr_exec = ((_tp_space / 100 - _COST) / (_stop_dist / 100 + _COST)) if _stop_dist > 0 else 0 if _stop_dist < 2.0: print(f" ⏭ {code} {name} 止损距离{_stop_dist:.1f}%<2%（不可执行，贴死技术位），跳过") processed += 1 continue if _tp_space < 3.0: print(f" ⏭ {code} {name} 止盈空间{_tp_space:.1f}%<3%（不可执行，覆盖不了成本），跳过") processed += 1 continue if _rr_exec < 2.0: print(f" ⏭ {code} {name} 修正风报比{_rr_exec:.1f}<2.0（含交易成本后不达标的阿猫阿狗），跳过") processed += 1 continue # 构建策略 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 reason_text = [] if el > 0: reason_text.append(f"买{el}~{eh}") if sl > 0: reason_text.append(f"损{sl}") if tp > 0: reason_text.append(f"盈{tp}") if sl > 0 and tp > 0 and price_est > 0: rr = (tp - price_est) / (price_est - sl) if (price_est - sl) > 0 else 0 reason_text.append(f"RR{rr:.1f}") reason_text.append(f"评分{score}") action = " | ".join(reason_text) if reason_text else f"市场扫描发现(评分{score})" # 容量闸：自选上限60只 MAX_WATCH = 60 wl_count = conn.execute( "SELECT COUNT(*) FROM holding_strategies WHERE status='active' AND decision_type='自选策略'").fetchone()[0] if wl_count >= MAX_WATCH: weakest = conn.execute(""" SELECT code, name, COALESCE(rr_ratio,0) as rr FROM holding_strategies WHERE status='active' AND decision_type='自选策略' AND code != ? ORDER BY COALESCE(rr_ratio,0) ASC, updated_at ASC LIMIT 1""", (code,)).fetchone() if weakest and (weakest[2] or 0) < 1.0: conn.execute( "INSERT INTO watchlist_log (code, name, event, reason, old_signal, new_signal, price) " "VALUES (?,?,?,?,?,?,?)", (weakest[0], weakest[1] or "", "exit", f"容量{MAX_WATCH}淘汰为新标的{code}腾位", "", "已删除", 0)) conn.execute( "DELETE FROM holding_strategies WHERE code=? AND status='active' AND decision_type='自选策略'", (weakest[0],)) 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(""" INSERT OR IGNORE INTO holding_strategies (code, name, price, entry_low, entry_high, stop_loss, take_profit, timing_signal, action, decision_type, strategy_type, status, rr_ratio, stock_category, strategy_name, created_at, updated_at, sector_context, quality_check) VALUES (?,?,?,?,?,?,?,?,?,'自选策略','scan', 'active',?,'关注',?,?,?,'', 'pending') """, (code, name, _price, el, eh, sl, tp, timing_signal, action, _rr, source_strategy, now, now)) newly_added = cur.rowcount > 0 conn.execute("UPDATE candidates SET promoted=1 WHERE code=?", (code,)) if newly_added: promoted += 1 processed += 1 newly_promoted_codes.append(code) print(f" ✅ {code} {name} 评分{score} → 已加入自选({timing_signal})", flush=True) else: print(f" ⏭ {code} {name} 已在自选策略中，标记promoted", flush=True) processed += 1 # 触发全量重评（仅新插入） 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=REASSESS_TIMEOUT) if r.returncode == 0: print(f" 重评完成", flush=True) else: print(f" 重评失败: {r.stderr.strip()[:100]}", flush=True) except Exception as e: print(f" 重评异常: {e}", flush=True) conn.commit() print(f"\n[PROMOTE] 本次处理{processed}个，提拔{promoted}只（剩余留待下批）", flush=True) # 推XMPP（老莫：进自选必须附完整 12 维重评全文） if promoted > 0: try: import urllib.request codes = ",".join("?" * len(newly_promoted_codes)) promoted_rows = conn.execute( f"SELECT code, name, strategy_name, rr_ratio, full_analysis " f"FROM holding_strategies WHERE status='active' AND code IN ({codes})", tuple(newly_promoted_codes) ).fetchall() if newly_promoted_codes else [] body = f"🤖 候选自动提拔: {promoted}只新入自选（评分降序优中选优，每批≤{MAX_NEW_PER_RUN}）\n" for i, (code, name, strat, rr, fa) in enumerate(promoted_rows, 1): nm = name or code body += f"\n━━━ [{i}] {code} {nm} ━━━\n" body += f"策略: {strat or 'unknown'} | RR: {rr or 'N/A'}\n" if fa and len(str(fa)) > 50: body += f"【12维重评】\n{fa}\n" else: body += "（重评报告生成中/未就绪，稍后补推）\n" if len(body) > 6000: body = body[:6000] + "\n…（内容过长已截断）" req = urllib.request.Request("http://127.0.0.1:5805/", data=json.dumps({"to": "hmo@yoin.fun", "body": body, "type": "chat"}).encode(), headers={"Content-Type": "application/json"}) urllib.request.urlopen(req, timeout=10) print(f"[PROMOTE] XMPP 通知已发送（{promoted}只含完整重评）", flush=True) except Exception as e: print(f"[PROMOTE] XMPP 通知失败: {e}", flush=True) # 2026-08-18 老莫：重评未就绪的必须补推——后台 detach 补跑重评+补推 if promoted > 0: try: _bf = subprocess.Popen( ["/home/hmo/MoFin/venv/bin/python", "/home/hmo/MoFin/deploy/profile-scripts/promote_reassess_backfill.py", "90"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True) print(f"[PROMOTE] 后台补推已启动（PID={_bf.pid}, 90s后补跑未就绪重评并补推）", flush=True) except Exception as e: print(f"[PROMOTE] 后台补推启动失败: {e}", flush=True) conn.close() if __name__ == "__main__": main()