MoFin/deploy/profile-scripts/promote_candidates.py

#!/usr/bin/env python3
"""promote_candidates.py — 自动提拔候选股入自选

从 candidates 表读未提拔的候选，评估后自动加入 holding_strategies。
"""
import sys, json, sqlite3
from pathlib import Path
from datetime import datetime

DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")

def main():
    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
        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 >= 4
        ORDER BY c.score_final DESC
    """).fetchall()
    
    if not rows:
        print("[PROMOTE] 无待提拔候选")
        conn.close()
        return
    
    promoted = 0
    for r in rows:
        code = str(r[0])
        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
        
        # 查是否已在 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
        
        # 验证实时价格：无有效价格的候选股不入自选（防假数据污染）
        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)
            if float(_q.get("price", 0)) <= 0:
                print(f"  ⏭ {code} {name} 无实时价格，跳过")
                continue
        except Exception as _e:
            print(f"  ⏭ {code} {name} 价格获取失败({_e})，跳过")
            continue
        
        # 构建策略
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        timing_signal = "买入" if score >= 7 else "关注"
        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})"
        
        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, created_at, updated_at,
             sector_context, quality_check)
            VALUES (?,?,?,?,?,?,?,?,?,'自选策略','scan',
                    'active',0,'关注',?,?,'', 'pending')
        """, (code, name, 0, el, eh, sl, tp, timing_signal, action, now, now))
        newly_added = cur.rowcount > 0
        
        conn.execute("UPDATE candidates SET promoted=1 WHERE code=?", (code,))
        if newly_added:
            promoted += 1
            print(f"  ✅ {code} {name} 评分{score} → 已加入自选({timing_signal})", flush=True)
        else:
            print(f"  ⏭ {code} {name} 已在自选策略中，标记promoted", flush=True)
        
        # 触发全量重评（生成完整9维策略）——仅新插入的股票需要
        if newly_added:
            try:
                import subprocess as _sp
                r = _sp.run(["python3", "/home/hmo/MoFin/scripts/per_stock_reassess.py", code],
                           capture_output=True, text=True, timeout=480)
                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] 本次提拔{promoted}只", flush=True)
    
    # 推XMPP
    if promoted > 0:
        try:
            import urllib.request
            msg = f"📈 自动提拔{promoted}只候选入自选"
            payload = json.dumps({"to": "hmo@yoin.fun", "body": msg, "type": "chat"}).encode()
            req = urllib.request.Request("http://127.0.0.1:5805/", data=payload,
                                        headers={"Content-Type": "application/json"})
            urllib.request.urlopen(req, timeout=5)
        except Exception:
            pass
    
    conn.close()

if __name__ == "__main__":
    main()