From 3395b7711d528f9d4a69b85cc08f8865f8295ff2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=9F=A5=E5=BE=AE?= Date: Thu, 9 Jul 2026 16:54:40 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=80=99=E9=80=89=E8=82=A1=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E6=8E=A8=E5=B9=BF=E8=84=9A=E6=9C=AC+=E7=9B=98?= =?UTF-8?q?=E4=B8=ADcron?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/promote_candidates.py | 207 ++++++++++++++++++++++++++++++++++ 1 file changed, 207 insertions(+) create mode 100644 scripts/promote_candidates.py diff --git a/scripts/promote_candidates.py b/scripts/promote_candidates.py new file mode 100644 index 00000000..266dd75b --- /dev/null +++ b/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()