"""选股计划入池：把每日计划写进 Mongo 的股票池分组，供决策系统每晚推理覆盖。 背景（2026-08-03 与用户定稿）：决策系统 (bionic_trader) 每晚 22:30 的认知扫描，扫描 范围就是 Mongo `stock_groups` 集合里**所有分组的股票代码并集**。把计划写成一个独立 分组（默认 group_code=AKG_PLAN），候选票当晚就会被夜间推理覆盖，产出支撑位、压力位、 定性结论——持仓系统 (PMS) 的参考位、择时执行区间、研判上下文全部由此而来。 这就是「提前计算为主」：不新造任何推理步骤，把票放进池子，已有的夜间推理自然完成预计算。 入池、留池、出池的规则（用户拍板，decide() 的注释里逐条对应）： 入池 = 当日计划（强传导主榜前 N，与 PMS 候选同口径） ∪ 当前持仓 留池 = 旧成员既不在计划也无持仓、但形态未恶化的，留下继续接受每晚分析 ——榜单是按条数截断过的，掉榜不等于变坏 出池 = 无持仓、不在计划、且形态已恶化（决策系统最新定性 SELL / AVOID / DROPPED） → 移入回收站集合 stock_recycle_bin（字段沿用现有格式，另加 reason 说明原因） 上限 = 池子超过 POOL_MAX 时，从「留池观察」里清最久没上过榜的（不进回收站； 出池后决策系统下次扫描会自动把其策略标 DROPPED 并出离场报告） 底线 = 持仓永不出池，即使形态恶化——那是 PMS 风控与体检的事，池子只保证它每晚有结论。 安全边界（都是「拿不到就不动」）： * 计划数据缺失（因子表没跑出来）→ 整个入池动作中止，池子保持昨日原样。 * 持仓读不到 → 同样中止——读不到持仓就可能把持仓票错清出池，宁可不动。 * 决策系统结论读不到 → 本轮不判恶化、一只都不回收（缺数据不算恶化），其余照常。 用法： python run.py push-pool --dry-run # 只打印入池/出池明细，不写库 python run.py push-pool # 真写（写完顺手触发决策系统的增量补扫） python run.py push-pool --no-kick # 写库但不触发补扫（比如夜间已近 22:30 全量扫） """ import datetime as dt import urllib.parse import urllib.request import common import config import db import plan # 决策系统结论里算「形态恶化」的定性（DROPPED 是它对掉出池子票的收尾标记） BAD_SIGNALS = {"SELL", "AVOID", "DROPPED"} # ============================================================================ # 纯逻辑：入池/留池/出池的决定（不碰任何库，test_pool_logic.py 直接测它） # ============================================================================ def decide(old_members, member_meta, plan_codes, holdings, bad_codes, max_size, today: str) -> dict: """按定稿规则算出新池子与各类进出明细。 old_members 上一版池子的代码集合 member_meta 上一版的成员记录 {code: {"added": 日期, "last_plan": 最近上榜日}} plan_codes 当日计划代码（有序，榜单名次序） holdings 当前持仓代码集合 bad_codes 形态已恶化的代码集合（判据 BAD_SIGNALS，缺数据时传空集=不回收） """ old = set(old_members or set()) plan_set = set(plan_codes or []) hold = set(holdings or set()) meta = {k: dict(v) for k, v in (member_meta or {}).items()} # 旧成员里既不在计划也无持仓的，按形态分流：恶化 → 回收站；未恶化 → 留池观察 idle = old - plan_set - hold recycled = sorted(idle & set(bad_codes or set())) observers = sorted(idle - set(recycled)) pool = list(dict.fromkeys(list(plan_codes or []) + sorted(hold) + observers)) # 上限：只清「留池观察」，按最久没上过榜的先清；计划与持仓永不清 cap_evicted = [] if max_size and len(pool) > max_size: def _last_seen(c): m = meta.get(c) or {} return m.get("last_plan") or m.get("added") or "" for c in sorted(observers, key=_last_seen): if len(pool) <= max_size: break pool.remove(c) cap_evicted.append(c) observers = [c for c in observers if c not in cap_evicted] # 成员记录：新进的记 added，今天在计划里的刷 last_plan，出池的删掉 for c in pool: meta.setdefault(c, {"added": today}) if c in plan_set: meta[c]["last_plan"] = today for c in list(meta): if c not in pool: meta.pop(c) return { "pool": pool, "new_entrants": sorted(plan_set - old), # 计划带来的新面孔 "retained_holdings": sorted(hold - plan_set), # 因持仓保留（不在当日计划里） "observers": observers, # 留池观察 "recycled": recycled, # 移入回收站（形态恶化） "cap_evicted": cap_evicted, # 池满出清（不进回收站） "held_bad": sorted(hold & set(bad_codes or set())), # 持仓且形态恶化——只警示不出池 "meta": meta, } def build_remark(d: dict, plan_date: str, now_str: str, degraded: str = "") -> dict: """分组文档的 remark 字段，格式沿用现有池子的写法（summary / factor_details / retained_positions / update_time_str），人读为主。""" n_plan = len(d["pool"]) - len(d["retained_holdings"]) - len(d["observers"]) summary = (f"共入池 {len(d['pool'])} 只。当日计划 {n_plan} 只" f"（其中新进 {len(d['new_entrants'])} 只），因持仓保留 " f"{len(d['retained_holdings'])} 只，留池观察 {len(d['observers'])} 只；" f"移入回收站 {len(d['recycled'])} 只（形态恶化），" f"池满出清 {len(d['cap_evicted'])} 只。") if d["held_bad"]: summary += f" 警示：持仓中 {', '.join(d['held_bad'])} 形态已恶化（持仓不出池，请在 PMS 侧关注）。" if degraded: summary += f" 注意：{degraded}。" plan_codes = [c for c in d["pool"] if c not in set(d["retained_holdings"]) and c not in set(d["observers"])] return { "summary": summary, "factor_details": [{ "factor_code": "akg_score", "trade_date": plan_date, "selected_count": len(plan_codes), "selected_codes": plan_codes, }], "retained_positions": d["retained_holdings"], "update_time_str": now_str, } def build_recycle_docs(d: dict, group_id: str, group_name: str, org_id: str, now: dt.datetime) -> list: """回收站文档，字段照抄现有格式（group_id/group_name/org_id/removal_batch/ removed_at/stock_code），另加一个 reason 说明为什么移入——Mongo 加字段对 老读者无影响，但事后能分清「形态恶化」与其他原因。""" batch = now.isoformat() return [{"group_id": group_id, "group_name": group_name, "org_id": org_id, "removal_batch": batch, "removed_at": now, "stock_code": c, "reason": "形态恶化（决策系统最新定性 SELL/AVOID/DROPPED），且无持仓、不在当日计划"} for c in d["recycled"]] # ============================================================================ # 取数（每一路的失败语义见模块头「安全边界」） # ============================================================================ def _read_holdings() -> set: """当前持仓。首选 PMS 账本 pms_position（新架构下它才是有人维护的持仓事实源： 下游 trading_position 表已没有写入方，2026-07-30 实测账户有仓时该表也是空的）； 账本表读不到再退回 trading_position 兜底。两条路都读不到就抛——持仓是出池判断 的底线输入，读不到宁可整轮不动池子。""" try: rows = db.read_mysql( "pms", "SELECT ts_code, total_qty, status FROM pms_position") out = set() for r in rows.itertuples(): try: qty = float(r.total_qty) except (TypeError, ValueError): qty = 0.0 if qty > 0 and str(r.status or "").upper() != "CLOSED": out.add(common.to_prefix(str(r.ts_code).strip().upper())) return out except Exception as e: # noqa: BLE001 —— 账本表读不到才走下游表兜底 print(f" （PMS 账本 pms_position 读取失败，退回 trading_position 兜底: {e!r}）") rows = db.read_mysql("pms", "SELECT * FROM trading_position") if rows.empty: return set() cols = {c.lower(): c for c in rows.columns} code_col = next((cols[c] for c in ("stock_code", "ts_code", "code") if c in cols), None) qty_col = next((cols[c] for c in ("total_quantity", "current_qty", "total_qty", "volume", "quantity") if c in cols), None) if not code_col: raise RuntimeError(f"trading_position 找不到代码列（现有列: {list(rows.columns)}）") out = set() for _, r in rows.iterrows(): code = str(r[code_col] or "").strip().upper() if not code: continue try: qty = float(r[qty_col]) if qty_col else 1.0 except (TypeError, ValueError): qty = 1.0 if qty > 0: out.add(common.to_prefix(code)) return out def _read_bad_signals(codes: set): """这批票在决策系统结论表里的最新定性，恶化的挑出来。 读失败返回 (空集, 原因)——缺数据不算恶化，本轮不回收任何票。""" if not codes: return set(), "" try: marks = ",".join(["%s"] * len(codes)) rows = db.read_mysql( "pms", f"SELECT stock_code, signal_type, trade_date FROM strategy_daily_results " f"WHERE stock_code IN ({marks})", tuple(codes)) except Exception as e: # noqa: BLE001 return set(), f"决策系统结论表读取失败（{e!r}），本轮不判恶化、不回收" if rows.empty: return set(), "" rows = rows.sort_values("trade_date").drop_duplicates("stock_code", keep="last") bad = {str(r.stock_code).strip().upper() for r in rows.itertuples() if str(r.signal_type or "").strip().upper() in BAD_SIGNALS} return bad, "" def _mongo(): from pymongo import MongoClient c = config.mongo() uri = (f"mongodb://{urllib.parse.quote_plus(c.user)}:{urllib.parse.quote_plus(c.password)}" f"@{c.host}:{c.port}/{c.db}?authSource=admin") return MongoClient(uri, serverSelectionTimeoutMS=8000) def _kick_bionic_scan() -> str: """写完池子后触发决策系统的增量补扫（只补当天还没分析过的票）。 失败只提示不报错——当晚 22:30 的全量扫描是兜底。""" if not config.BIONIC_SCAN_URL or not config.BIONIC_SCAN_KEY: return "未配置 BIONIC_SCAN_URL / BIONIC_SCAN_KEY，跳过补扫触发（当晚全量扫兜底）" url = (f"{config.BIONIC_SCAN_URL}?key={urllib.parse.quote_plus(config.BIONIC_SCAN_KEY)}" f"&mode=incremental") try: with urllib.request.urlopen(url, timeout=15) as resp: body = resp.read().decode("utf-8", "replace")[:200] return f"已触发决策系统增量补扫: {body}" except Exception as e: # noqa: BLE001 return f"补扫触发失败（当晚 22:30 全量扫兜底）: {e!r}" # ============================================================================ # 主流程 # ============================================================================ def push(date: str | None = None, top: int | None = None, dry_run: bool = False, kick: bool = True) -> dict: # 1. 当日计划（与 /plan 同一段装配代码，口径天然一致），按档位白名单过滤 top = top or config.POOL_TOP data = plan.collect(date, top=top, obs_top=0, theme_cap=config.POOL_THEME_CAP) tiers = config.POOL_TIERS plan_rows = [r for r in data["main"] if not tiers or r.get("tier") in tiers] plan_codes = [r["code"] for r in plan_rows] ds = data["date"] # 2. 持仓（读不到直接抛，整轮不动池子）与旧池子 holdings = _read_holdings() col_name = config.POOL_COLLECTION client = None if dry_run and not config_mongo_ready() else _mongo() old_doc, old_members, member_meta = None, set(), {} if client is not None: col = client[config.mongo().db][col_name] old_doc = col.find_one({"group_code": config.POOL_GROUP_CODE, "org_id": config.POOL_ORG_ID}) if old_doc: old_members = {str(c).strip().upper() for c in old_doc.get("stock_codes") or []} member_meta = old_doc.get("member_meta") or {} # 3. 形态恶化名单（只查可能出池的那批；读失败=不回收） idle = old_members - set(plan_codes) - holdings bad, degraded = _read_bad_signals(idle | (holdings & old_members)) now = dt.datetime.now() today = now.date().isoformat() d = decide(old_members, member_meta, plan_codes, holdings, bad, config.POOL_MAX, today) # 4. 打印明细（干跑到此为止） print(f"计划日 {ds}，档位白名单 {sorted(tiers) if tiers else '（不过滤）'}，" f"计划入选 {len(plan_codes)} 只；持仓 {len(holdings)} 只；" f"旧池 {len(old_members)} 只 → 新池 {len(d['pool'])} 只（上限 {config.POOL_MAX}）") for label, items in (("计划新进", d["new_entrants"]), ("持仓保留", d["retained_holdings"]), ("留池观察", d["observers"]), ("移入回收站(形态恶化)", d["recycled"]), ("池满出清", d["cap_evicted"]), ("警示: 持仓且形态恶化(不出池)", d["held_bad"])): if items: print(f" {label} {len(items)} 只: {', '.join(items)}") if degraded: print(f" ⚠️ {degraded}") if dry_run: print("（--dry-run：只看不写）") if client is not None: client.close() return d # 5. 写分组文档（按 group_code+org_id 覆盖式更新）+ 回收站 + 触发补扫 remark = build_remark(d, ds, now.strftime("%Y-%m-%d %H:%M:%S"), degraded) ctx = dict((old_doc or {}).get("strategy_context") or {}) ctx = {k: v for k, v in ctx.items() if k in set(d["pool"])} for r in plan_rows: ctx[r["code"]] = {"factor_code": "akg_score", "score": r.get("score"), "tier": r.get("tier"), "upside": r.get("upside"), "theme": (r.get("evidence") or {}).get("theme"), "plan_date": ds} for c in d["retained_holdings"]: ctx.setdefault(c, {"factor_code": "akg_score", "note": "持仓保留"}) col = client[config.mongo().db][col_name] col.update_one( {"group_code": config.POOL_GROUP_CODE, "org_id": config.POOL_ORG_ID}, {"$set": {"group_name": config.POOL_GROUP_NAME, "pool_type": "core", "is_public": False, "description": "akg-factor-bridge 每日选股计划池：当日计划(强传导主榜) + " "持仓保留 + 留池观察。决策系统每晚认知扫描按本组产出结论，" "供 PMS 参考位/择时/研判使用。规则见 akg-factor-bridge/" "docs/选股计划入池_对接说明.md", "stock_codes": d["pool"], "member_meta": d["meta"], "strategy_context": ctx, "remark": remark, "updated_at": now}, "$setOnInsert": {"created_at": now}}, upsert=True) doc = col.find_one({"group_code": config.POOL_GROUP_CODE, "org_id": config.POOL_ORG_ID}, {"_id": 1}) print(f"✅ 分组已写入 {col_name}（group_code={config.POOL_GROUP_CODE}，" f"_id={doc['_id']}，{len(d['pool'])} 只）") if d["recycled"]: bin_col = client[config.mongo().db][config.POOL_RECYCLE_COLLECTION] docs = build_recycle_docs(d, str(doc["_id"]), config.POOL_GROUP_NAME, config.POOL_ORG_ID, now) bin_col.insert_many(docs) print(f"✅ 回收站已记 {len(docs)} 只: {', '.join(d['recycled'])}") client.close() if kick: print(_kick_bionic_scan()) return d def config_mongo_ready() -> bool: """干跑时若 Mongo 还没配置（比如首次在开发机看效果），允许把旧池当空集。""" try: config.mongo() return True except Exception: # noqa: BLE001 print(" （Mongo 未配置，按空池试算）") return False