feat: 候选股自动推广脚本+盘中cron

scripts/promote_candidates.py

@@ -0,0 +1,207 @@
+#!/usr/bin/env python3
+"""promote_candidates.py — 将候选股自动提拔入自选
+
+管道位置：
+  candidates 表（候选池）→ 本脚本（价格验证+写入策略）→ holding_strategies（自选股）
+
+运行时机：
+  每30分钟，交易日 9:30~15:00
+
+no_agent模式：有提拔→输出，无→静默
+
+流程：
+  1. 从 candidates 读 promoted=0 AND dropped=0
+  2. 用 stock_quote.py 验证实时价
+  3. 检查是否已在持仓/自选（避免重复）
+  4. 写入 holding_strategies (decision_type='自选策略', status='active')
+  5. 标记 candidates.promoted=1
+"""
+
+import json, os, sqlite3, subprocess, sys, time
+from pathlib import Path
+from datetime import datetime
+
+BASE = Path("/home/hmo/MoFin")
+DATA = BASE / "data"
+DB_PATH = DATA / "mofin.db"
+
+sys.path.insert(0, str(BASE / "scripts"))
+from mofin_db import get_conn, write_holding_strategy
+
+
+def get_quote(code):
+    """用 stock_quote.py 获取实时行情"""
+    try:
+        r = subprocess.run(
+            [sys.executable, str(BASE / "scripts/stock_quote.py"), str(code)],
+            capture_output=True, text=True, timeout=15
+        )
+        if r.returncode != 0:
+            return None
+        for line in r.stdout.strip().split("\n"):
+            if line.startswith("{"):
+                return json.loads(line)
+    except Exception:
+        return None
+    return None
+
+
+def already_in_system(conn, code):
+    """检查是否已在持仓或自选中"""
+    # holdings
+    cur = conn.execute("SELECT COUNT(*) FROM holdings WHERE code=? AND is_active=1", (code,))
+    if cur.fetchone()[0] > 0:
+        return "持仓"
+    # holding_strategies active
+    cur = conn.execute(
+        "SELECT COUNT(*) FROM holding_strategies WHERE code=? AND status='active'",
+        (code,)
+    )
+    if cur.fetchone()[0] > 0:
+        return "自选"
+    return None
+
+
+def promote_one(conn, cand):
+    """将一只候选股提拔为自选股（带策略）"""
+    code = cand["code"]
+    name = cand.get("name", "")
+
+    # 1. 检查是否已在系统内
+    exist = already_in_system(conn, code)
+    if exist:
+        mark_promoted(conn, code, f"已在{exist}")
+        return None
+
+    # 2. 获取实时行情
+    quote = get_quote(code)
+    if not quote or not quote.get("price"):
+        return (code, name, f"取价失败，跳过")
+
+    price = quote["price"]
+    change_pct = quote.get("change_pct", 0)
+
+    # 3. 解析 entry_range
+    entry_str = cand.get("entry_range", "")
+    entry_low = 0
+    entry_high = 0
+    if entry_str and "~" in entry_str:
+        try:
+            parts = entry_str.split("~")
+            entry_low = float(parts[0].strip())
+            entry_high = float(parts[1].strip())
+        except (ValueError, IndexError):
+            pass
+
+    # 如果没entry_range，用价格±3%作为默认区间
+    if entry_low <= 0 or entry_high <= 0:
+        entry_low = round(price * 0.97, 2)
+        entry_high = round(price * 1.03, 2)
+
+    stop_loss = cand.get("stop_loss") or round(entry_low * 0.93, 2)
+    target = cand.get("target") or round(entry_high * 1.15, 2)
+    reason = cand.get("reason", "系统挖掘候选")
+
+    # 4. 检测市场（A股/港股）
+    code_str = str(code).strip()
+    currency = "HKD" if (len(code_str) <= 5 or code_str.startswith("0")) and len(code_str) < 6 else "CNY"
+    # 更精确的检测：港股5位
+    if len(code_str) <= 5:
+        currency = "HKD"
+    elif code_str.startswith("0") and len(code_str) <= 5:
+        currency = "HKD"
+    else:
+        currency = "CNY"
+
+    sector = cand.get("sector", "")
+
+    # 5. 写入 holding_strategies
+    strategy_data = {
+        "version": 1,
+        "price": price,
+        "cost": price,
+        "entry_low": entry_low,
+        "entry_high": entry_high,
+        "stop_loss": stop_loss,
+        "take_profit": target,
+        "currency": currency,
+        "strategy_type": "watch",
+        "action": "观望",
+        "timing_signal": "中性",
+        "rr_ratio": round((target - entry_low) / (entry_low - stop_loss), 2) if stop_loss and (entry_low - stop_loss) > 0 else 0,
+        "stock_category": sector or "未分类",
+        "sector_context": sector or "",
+        "status": "active",
+        "source": "candidates",
+        "reason": reason,
+        "type": "自选策略",
+        "decision_type": "自选策略",
+        "time_horizon": "中线",
+        "note": f"候选股自动推广 {datetime.now().strftime('%Y-%m-%d %H:%M')}",
+    }
+
+    ok, msg = write_holding_strategy(conn, code, name, strategy_data)
+    if ok:
+        mark_promoted(conn, code)
+        return (code, name, f"已加自选 {entry_low}~{entry_high} 止损{stop_loss}")
+    else:
+        return (code, name, f"写入失败: {msg}")
+
+
+def mark_promoted(conn, code, reason=None):
+    """标记候选为已推广"""
+    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    if reason:
+        conn.execute(
+            "UPDATE candidates SET promoted=1, promoted_at=?, dropped=1, drop_reason=? WHERE code=?",
+            (now, reason, code)
+        )
+    else:
+        conn.execute(
+            "UPDATE candidates SET promoted=1, promoted_at=? WHERE code=?",
+            (now, code)
+        )
+    conn.commit()
+
+
+def main():
+    start = time.time()
+    today = datetime.now().strftime("%Y-%m-%d %H:%M")
+
+    conn = get_conn()
+    if not conn:
+        print("[SILENT] 无法连接DB", flush=True)
+        return
+
+    try:
+        # 读待推广候选股
+        rows = conn.execute(
+            "SELECT * FROM candidates WHERE promoted=0 AND dropped=0 ORDER BY created_at ASC"
+        ).fetchall()
+
+        if not rows:
+            print("[SILENT] 无待推广候选", flush=True)
+            return
+
+        results = []
+        for r in rows:
+            res = promote_one(conn, dict(r))
+            if res:
+                results.append(res)
+
+        conn.commit()
+        elapsed = time.time() - start
+
+        if results:
+            print(f"候选股自动推广 | {today} | {len(results)}只处理 ({elapsed:.0f}s)")
+            for code, name, msg in results:
+                print(f"  {code} {name}: {msg}")
+        else:
+            print("[SILENT] 候选推广结束（无结果）", flush=True)
+
+    finally:
+        conn.close()
+
+
+if __name__ == "__main__":
+    main()