From b61598fd3ddd1fdcabe9018f27606f00e37e72e2 Mon Sep 17 00:00:00 2001 From: zlt Date: Mon, 3 Aug 2026 14:02:13 +0800 Subject: [PATCH] =?UTF-8?q?=E9=80=89=E8=82=A1=E8=AE=A1=E5=88=92=E5=85=A5?= =?UTF-8?q?=E6=B1=A0=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 36 ++++ README.md | 8 + config.py | 49 ++++++ db.py | 2 +- docs/选股计划入池_对接说明.md | 100 +++++++++++ pool.py | 319 ++++++++++++++++++++++++++++++++++ requirements.txt | 1 + run.py | 11 ++ test_pool_logic.py | 158 +++++++++++++++++ 9 files changed, 683 insertions(+), 1 deletion(-) create mode 100644 docs/选股计划入池_对接说明.md create mode 100644 pool.py create mode 100644 test_pool_logic.py diff --git a/.env.example b/.env.example index 8b49206..874122d 100644 --- a/.env.example +++ b/.env.example @@ -68,3 +68,39 @@ PRICE_CODE_COL=symbol # 命中即说明该行 moved_ratio 建立在任意抽样上、覆盖被展示逻辑锁住 —— 只告警不拦截。 # UPSTREAM_MEMBER_CAP=30 # UPSTREAM_QUIET_CAP=12 + +# ============================================================================ +# 选股计划入池（2026-08-03 定稿；规则与流程见 docs/选股计划入池_对接说明.md） +# 把每日计划写进 Mongo 的股票池分组，决策系统每晚认知扫描按分组并集覆盖， +# 候选票自动获得夜间推理（支撑/压力/定性）→ PMS 的参考位、择时区间、研判由此可用。 +# ============================================================================ + +# --- Mongo（决策系统夜扫读的同一处；填决策系统 .env 里的同名值）--- +MONGO_HOST= +MONGO_PORT=27017 +MONGO_USERNAME= +MONGO_PASSWORD= +MONGO_DB=stock_predictions + +# --- 持仓与决策系统结论（153 代理，与热度同一台；默认复用 HEAT_MYSQL_*， +# 代理路由不通时才需要单独填）--- +# PMS_MYSQL_HOST= +# PMS_MYSQL_PORT=3306 +# PMS_MYSQL_USER= +# PMS_MYSQL_PASSWORD= +# PMS_MYSQL_DB= + +# --- 入池口径（默认与 PMS 候选一致：强传导主榜前 20；改动记得与 PMS 页面的 +# PMS_PLAN_TOP_N / PMS_PLAN_TIERS 保持一致）--- +# POOL_GROUP_CODE=AKG_PLAN +# POOL_GROUP_NAME=AKG每日选股计划池 +# POOL_ORG_ID=489281497140 +# POOL_TOP=20 +# POOL_TIERS=强传导 +# POOL_THEME_CAP=5 +# POOL_MAX=60 + +# --- 写完池子后触发决策系统的增量补扫（只补当天没分析过的票；留空=不触发， +# 当晚 22:30 全量扫兜底）。key 填决策系统 .env 的 XXL_TRIGGER_KEY --- +# BIONIC_SCAN_URL=http://192.168.16.188:38000/api/v1/xxl/daily-scan +# BIONIC_SCAN_KEY= diff --git a/README.md b/README.md index c046192..8a9604a 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,14 @@ akg-factor-bridge：读视图+热度 → 四路日截面变换 → 【漏斗合 | `akg_event` | t_factor_akg_event | Σ 事件极性×时间衰减（**仅公告来源、单文档封顶**） | 不出行 | | `akg_transmission` | t_factor_akg_transmission | **distinct 源数**×(1−已动比例) | 不出行（合成侧填 0） | +## 选股计划入池（2026-08-03） + +`python run.py push-pool` 把每日计划写进 Mongo 股票池分组（group_code=AKG_PLAN）， +决策系统每晚认知扫描按分组并集覆盖 → 候选票自动获得夜间推理（支撑/压力/定性）， +PMS 的参考位、择时执行区间、研判上下文由此可用。入池=当日计划(强传导前20)∪持仓； +掉榜未恶化留池观察；无持仓、不在计划且形态恶化 → 移入回收站 stock_recycle_bin； +持仓永不出池。规则、时间线、部署与判收见 `docs/选股计划入池_对接说明.md`。 + ## 用法（全程 Docker，不在宿主机直跑） ```bash diff --git a/config.py b/config.py index 77d28e4..eec2419 100644 --- a/config.py +++ b/config.py @@ -60,6 +60,33 @@ def price_mysql() -> Conn: _opt("PRICE_MYSQL_DB", _req("FACTOR_MYSQL_DB"))) +def pms_mysql() -> Conn: + """读持仓（trading_position）与决策系统结论（strategy_daily_results），入池用。 + 这两张表都在 153 代理后面——与热度是同一台，默认直接复用 HEAT_MYSQL_*， + 只有代理路由不通时才需要单独配 PMS_MYSQL_*。""" + return Conn(_opt("PMS_MYSQL_HOST", _req("HEAT_MYSQL_HOST")), + int(_opt("PMS_MYSQL_PORT", os.environ.get("HEAT_MYSQL_PORT", "3306"))), + _opt("PMS_MYSQL_USER", _req("HEAT_MYSQL_USER")), + _opt("PMS_MYSQL_PASSWORD", os.environ.get("HEAT_MYSQL_PASSWORD", "")), + _opt("PMS_MYSQL_DB", _req("HEAT_MYSQL_DB"))) + + +@dataclass(frozen=True) +class MongoConn: + host: str + port: int + user: str + password: str + db: str + + +def mongo() -> MongoConn: + """选股计划入池的落点（决策系统每晚扫描读同一处）。""" + return MongoConn(_req("MONGO_HOST"), int(os.environ.get("MONGO_PORT", 27017)), + _req("MONGO_USERNAME"), os.environ.get("MONGO_PASSWORD", ""), + _opt("MONGO_DB", "stock_predictions")) + + # gp_day_data 股票代码列名（平台实测 = symbol；代码仍会自动兜底逐个试） PRICE_CODE_COL = os.environ.get("PRICE_CODE_COL", "ts_code") FACTOR_API_BASE = os.environ.get("FACTOR_API_BASE", "").rstrip("/") @@ -119,3 +146,25 @@ ENABLE_TRACK_GATE = os.environ.get("ENABLE_TRACK_GATE", "0") == "1" # ST / *ST / S(S)T / 退市族按证券简称识别，默认挡在档位之外——gate 记 0（不采纳）， # 因此不进主榜、观察档、计划与升降档。研究口径想看全貌时置 0 关闭。 EXCLUDE_RISK_NAMES = os.environ.get("EXCLUDE_RISK_NAMES", "1") == "1" + +# --- 选股计划入池（08-03 定稿；规则与流程见 docs/选股计划入池_对接说明.md）------ +# 写 Mongo stock_groups 的独立分组，决策系统每晚扫描按分组并集覆盖 → 候选票自动 +# 获得夜间推理。入池范围与 PMS 候选同口径：强传导主榜前 POOL_TOP 只 + 当前持仓。 +# 注意：POOL_TOP / POOL_TIERS 若调整，记得与 PMS 页面的 PMS_PLAN_TOP_N / +# PMS_PLAN_TIERS 保持一致，两边看到的候选才是同一批。 +POOL_GROUP_CODE = os.environ.get("POOL_GROUP_CODE", "AKG_PLAN") +POOL_GROUP_NAME = os.environ.get("POOL_GROUP_NAME", "AKG每日选股计划池") +POOL_ORG_ID = os.environ.get("POOL_ORG_ID", "489281497140") +POOL_COLLECTION = os.environ.get("POOL_COLLECTION", "stock_groups") +POOL_RECYCLE_COLLECTION = os.environ.get("POOL_RECYCLE_COLLECTION", "stock_recycle_bin") +POOL_TOP = int(os.environ.get("POOL_TOP", "20")) +POOL_THEME_CAP = int(os.environ.get("POOL_THEME_CAP", "5")) +# 档位白名单（逗号分隔；空串=主榜全部）。默认只收强传导——候选宁缺毋滥。 +POOL_TIERS = {s.strip() for s in os.environ.get("POOL_TIERS", "强传导").split(",") if s.strip()} +# 池子上限：计划+持仓+留池观察合计超过它时，从留池观察里清最久没上榜的。 +# 上限直接决定决策系统每晚的推理量（每只全量分析约两三分钟）。 +POOL_MAX = int(os.environ.get("POOL_MAX", "60")) +# 写完池子后触发决策系统的增量补扫（只补当天没分析过的票）。留空=不触发， +# 当晚 22:30 全量扫兜底。例：http://192.168.16.188:38000/api/v1/xxl/daily-scan +BIONIC_SCAN_URL = os.environ.get("BIONIC_SCAN_URL", "").rstrip("/") +BIONIC_SCAN_KEY = os.environ.get("BIONIC_SCAN_KEY", "") diff --git a/db.py b/db.py index 83e922e..b0b4ea8 100644 --- a/db.py +++ b/db.py @@ -27,7 +27,7 @@ def _mysql(c: "config.Conn"): @contextmanager def _mysql_cm(which: str): c = {"heat": config.heat_mysql, "factor": config.factor_mysql, - "price": config.price_mysql}[which]() + "price": config.price_mysql, "pms": config.pms_mysql}[which]() conn = _mysql(c) try: yield conn diff --git a/docs/选股计划入池_对接说明.md b/docs/选股计划入池_对接说明.md new file mode 100644 index 0000000..04dde45 --- /dev/null +++ b/docs/选股计划入池_对接说明.md @@ -0,0 +1,100 @@ +# 选股计划入池 —— 与决策系统夜间推理的对接说明 + +> 2026-08-03 与用户定稿并落码。代码：`pool.py`（入口 `run.py push-pool`）， +> 纯逻辑单测 `test_pool_logic.py`。本文写清楚三件事：为什么这么接、规则是什么、怎么部署与验证。 + +## 1. 为什么走股票池，而不是新造一条推理链 + +决策系统（bionic_trader）每晚 22:30 的认知扫描，扫描范围就是 Mongo `stock_groups` +集合里**所有分组的股票代码并集**（`daily_scan_v2.get_mongo_stock_pool()` 对整个集合取 +并集，不看分组归属）。扫到的每只票会产出支撑位、压力位、定性结论，落在 +`strategy_daily_results`——持仓系统（PMS）的参考位、择时执行区间、研判上下文全部读它。 + +所以把每日计划写成集合里的一个独立分组，候选票当晚就自动获得夜间推理，不需要在任何 +系统里新增推理步骤。这正是「提前计算为主、盘中监控为辅」的落法：大模型的活儿全部发生 +在凌晨，盘中各接口只读现成结论。 + +配套的收尾机制决策系统也已经有：掉出池子并集的票，下一次扫描会被 +`prune_stale_strategies` 把策略标成 `DROPPED` 并生成离场报告——所以出池动作只需要在 +Mongo 侧记账（回收站），分析侧的清理是自动的。 + +## 2. 入池、留池、出池的规则（用户拍板，`pool.decide()` 逐条对应） + +| 规则 | 内容 | +|---|---| +| 入池 | 当日计划（强传导主榜前 `POOL_TOP`，与 PMS 候选同口径）∪ 当前持仓（`trading_position` 数量>0） | +| 留池 | 旧成员既不在计划也无持仓、但形态未恶化的，留下继续接受每晚分析——榜单是按条数截断过的，掉榜不等于变坏 | +| 出池 | 无持仓、不在计划、且形态已恶化 → 移入回收站 `stock_recycle_bin`。恶化判据用决策系统自己的结论：`strategy_daily_results` 最新定性为 SELL / AVOID / DROPPED | +| 上限 | 池子超过 `POOL_MAX`（默认 60）时，从「留池观察」里清最久没上过榜的；这类清退不进回收站（它们没有恶化记录），出池后由决策系统的 DROPPED 机制收尾 | +| 底线 | **持仓永不出池**，即使形态恶化——恶化持仓在池子备注里给警示，处置是 PMS 风控与体检的事，池子只保证它每晚有结论可用 | + +拿不到就不动的三条安全边界：计划数据缺失（因子表没跑出来）→ 整轮中止，池子保持原样； +持仓读不到 → 同样中止（否则可能把持仓票错清出池）；决策系统结论读不到 → 本轮不判恶化、 +一只都不回收，其余照常并在备注里说明。 + +## 3. 写进 Mongo 的东西 + +分组文档（`stock_groups`，按 `group_code=AKG_PLAN` + `org_id` 覆盖式更新）： +`stock_codes`（前缀码）、`remark`（summary 一句人话 + factor_details + retained_positions + +update_time_str，格式沿用现有池子）、`strategy_context`（每只计划票的分数/档位/主题/预期 +空间，供人查）、`member_meta`（每只票的入池日与最近上榜日，上限清退的排序依据）。 + +回收站文档（`stock_recycle_bin`）字段照抄现有格式（group_id / group_name / org_id / +removal_batch / removed_at / stock_code），**另加了一个 `reason` 字段**说明移入原因—— +Mongo 加字段对老读者无影响，事后能分清是形态恶化还是别的原因。 + +老模拟系统目前处于停用状态（2026-08-03 用户确认），所以新分组不会引发任何自动交易； +将来若重启那套系统，需要先确认它只认自己的分组。 + +## 4. 每天的时间线 + +``` +07:10 桥机 cron（UTC 23:10）：build + plan（既有盘前链，07-30 D 案） +07:1x 接着跑 push-pool：计划写入池子，恶化票移入回收站 +07:1x push-pool 顺手触发决策系统增量补扫（/api/v1/xxl/daily-scan?mode=incremental， + 只补当天还没结论的票——通常就是几只新进榜的） +08:30 决策系统的例行查漏补缺照跑（另一层兜底） +08:40 PMS 拉 /plan 刷新候选池（原有流程，一字未动） +09:30 开盘：候选票的支撑/压力/定性已就绪 → 参考位、择时区间、研判全部可用 +22:30 决策系统全量夜扫：覆盖整个池子并清理掉出池的票（终极兜底） +``` + +增量补扫失败只提示不报错——当晚全量扫是兜底，最坏情形是新进票当天白天没有结论 +（择时对它们回「不可用」，PMS 自动退内置择时，行为与接入前相同）。 + +## 5. 部署（桥机 factorevaluation，akg-factor-bridge 根目录） + +```bash +git pull +# 新增 pymongo 依赖，需要重建一次镜像（代码本身是挂载卷，之后改码不用再 build） +docker compose build && docker compose up -d +# .env 增补（对照 .env.example 底部「选股计划入池」一节）： +# MONGO_* 五项（填决策系统 .env 里的同名值） +# BIONIC_SCAN_URL / BIONIC_SCAN_KEY（可选，触发增量补扫用） +# 纯逻辑单测（不连库，秒级）： +docker compose exec -T akg-factor-bridge python test_pool_logic.py +# 首次试跑（只看不写，核对入池/出池明细）： +docker compose exec -T akg-factor-bridge python run.py push-pool --dry-run +# 确认无误后真写（写完会打印分组 _id 与回收站回执）： +docker compose exec -T akg-factor-bridge python run.py push-pool +``` + +宿主 crontab：把 push-pool **串在既有盘前链命令末尾**（用 && 衔接，链子没跑成就不动 +池子），不要单独定时——单独定时撞上链子超时会拿昨天的因子重复入池： + +``` +# 既有（UTC 23:10 = CST 07:10，示意）： +10 23 * * 1-5 docker exec akg_factor_bridge python run.py build all --mode daily && \ + docker exec akg_factor_bridge python run.py plan && \ + docker exec akg_factor_bridge python run.py push-pool +``` + +## 6. 验证与判收 + +部署当天：`--dry-run` 的明细与预期一致；真写后在 Mongo 里能查到 `AKG_PLAN` 分组且 +`stock_codes` = 计划 ∪ 持仓。当晚 22:30 后：`strategy_daily_results` 里新进票有当日行。 +次日盘中：PMS 侧 `probe_bionic.py` 对候选票探活，择时接口应回 FIRE/WAIT 并带出区间 +（不再是「没有昨夜结论」的不可用）。连续跑几天后：掉榜未恶化的票留在池里继续有结论； +出现 SELL/AVOID 的闲置票进回收站，且决策系统随后把它们标 DROPPED。 + +判收纪律照旧：代码就绪 ≠ 接通 ≠ 判收。判收 = 上面这四条在实机各看到一次。 diff --git a/pool.py b/pool.py new file mode 100644 index 0000000..66a6e5b --- /dev/null +++ b/pool.py @@ -0,0 +1,319 @@ +"""选股计划入池：把每日计划写进 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: + """当前持仓（trading_position，数量>0）。列名在下游换过多次，按候选名单挑； + 读不到就抛——持仓是出池判断的底线输入，读不到宁可整轮不动池子。""" + 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 diff --git a/requirements.txt b/requirements.txt index 33659cb..7403f26 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,3 +6,4 @@ python-dotenv>=1.0 PyYAML>=6.0 fastapi>=0.110 uvicorn>=0.29 +pymongo>=4.6 diff --git a/run.py b/run.py index 5132f5b..149bd99 100644 --- a/run.py +++ b/run.py @@ -6,6 +6,8 @@ python run.py freeze [--date D] # 输入冻结（G0.5，详见 freeze.py） python run.py tracks # 赛道覆盖体检 + 成员表快照（G2） python run.py plan [--date D] [--top N] # 每日选股计划（R4，读已落库因子表） + python run.py push-pool [--date D] [--dry-run] # 计划入池：写 Mongo 股票池分组， + # 供决策系统每晚推理覆盖（详见 pool.py） python run.py register # 注册全部因子到 factor_metadata python run.py build all --mode history --start 2024-01-01 --end 2025-12-31 python run.py build akg_heat --mode daily --date 2026-07-24 @@ -165,6 +167,12 @@ def main(): pl.add_argument("--obs-top", type=int, default=10, help="观察档条数") pl.add_argument("--theme-cap", type=int, default=5, help="每个传导主题最多几条（防单板块刷屏；0=不设限）") + pp = sub.add_parser("push-pool") # 计划入池（08-03，写 Mongo 股票池分组） + pp.add_argument("--date", help="默认取 score 表最新日（与 plan 同口径）") + pp.add_argument("--top", type=int, help="计划取主榜前几只（默认读 POOL_TOP=20）") + pp.add_argument("--dry-run", action="store_true", help="只打印入池/出池明细，不写库") + pp.add_argument("--no-kick", action="store_true", + help="写完不触发决策系统增量补扫（当晚全量扫兜底）") p = sub.add_parser("probe") p.add_argument("--section", choices=["all", "pools", "price", "upside", "corr"], default="all", @@ -199,6 +207,9 @@ def main(): elif a.cmd == "plan": import plan plan.generate(a.date, a.top, a.obs_top, a.theme_cap) + elif a.cmd == "push-pool": + import pool + pool.push(a.date, a.top, dry_run=a.dry_run, kick=not a.no_kick) elif a.cmd == "tracks": import tracks tracks.coverage_report() diff --git a/test_pool_logic.py b/test_pool_logic.py new file mode 100644 index 0000000..aacab1c --- /dev/null +++ b/test_pool_logic.py @@ -0,0 +1,158 @@ +# -*- coding: utf-8 -*- +"""入池/留池/出池纯逻辑的单测（不连任何库）。 + +运行: docker compose exec -T akg-factor-bridge python test_pool_logic.py +全过输出 "ALL PASS (n cases)"，任一失败退出码 1。 +被测函数: pool.decide / pool.build_remark / pool.build_recycle_docs。 +""" +import datetime as dt +import sys +import traceback + +import pool + +RESULTS = [] + + +def case(name): + def deco(fn): + RESULTS.append((name, fn)) + return fn + return deco + + +TODAY = "2026-08-04" + + +@case("首次建池: 空池 + 计划20只 + 持仓3只(1只重叠) → 池=计划∪持仓, 无出池") +def _(): + plan = [f"SH60{i:04d}" for i in range(20)] + hold = {"SH600000", "SZ000001", "SZ000002"} # SH600000 与计划重叠 + d = pool.decide(set(), {}, plan, hold, set(), 60, TODAY) + assert len(d["pool"]) == 22, d["pool"] + assert set(plan) <= set(d["pool"]) and hold <= set(d["pool"]) + assert d["retained_holdings"] == ["SZ000001", "SZ000002"] + assert d["recycled"] == [] and d["cap_evicted"] == [] and d["observers"] == [] + assert d["new_entrants"] == sorted(plan) + # 成员记录: 计划票有 last_plan, 纯持仓票只有 added + assert d["meta"]["SH600000"]["last_plan"] == TODAY + assert "last_plan" not in d["meta"]["SZ000001"] + + +@case("掉榜不等于变坏: 掉出计划且无持仓、形态未恶化 → 留池观察") +def _(): + old = {"SH600001", "SH600002", "SH600003"} + meta = {c: {"added": "2026-08-01", "last_plan": "2026-08-01"} for c in old} + d = pool.decide(old, meta, ["SH600001"], set(), set(), 60, TODAY) + assert d["observers"] == ["SH600002", "SH600003"], d + assert set(d["pool"]) == old # 都还在池里 + assert d["recycled"] == [] + + +@case("出池三条件缺一不可: 无持仓 + 不在计划 + 形态恶化 → 回收站") +def _(): + old = {"SH600001", "SH600002", "SH600003", "SH600004"} + meta = {c: {"added": "2026-08-01"} for c in old} + bad = {"SH600002", "SH600003", "SH600004"} + # 600002 恶化但仍在计划 → 留; 600003 恶化但有持仓 → 留(警示); 600004 三条全中 → 回收 + d = pool.decide(old, meta, ["SH600001", "SH600002"], {"SH600003"}, bad, 60, TODAY) + assert d["recycled"] == ["SH600004"], d + assert "SH600002" in d["pool"] and "SH600003" in d["pool"] + assert d["held_bad"] == ["SH600003"] + + +@case("池满出清: 只清留池观察、按最久没上榜的先清, 计划与持仓永不清") +def _(): + plan = [f"SH61{i:04d}" for i in range(5)] + hold = {"SZ000001"} + old = set(plan) | hold | {"SH620001", "SH620002", "SH620003"} + meta = {"SH620001": {"added": "2026-07-01", "last_plan": "2026-07-10"}, + "SH620002": {"added": "2026-07-01", "last_plan": "2026-07-20"}, + "SH620003": {"added": "2026-07-01", "last_plan": "2026-07-30"}} + d = pool.decide(old, meta, plan, hold, set(), 8, TODAY) # 5+1+3=9 > 8, 清 1 只 + assert d["cap_evicted"] == ["SH620001"], d # 最久没上榜的 + assert len(d["pool"]) == 8 + assert d["recycled"] == [] # 池满出清不进回收站 + assert "SH620001" not in d["meta"] # 出池即清成员记录 + + +@case("计划+持仓本身超上限时不硬砍 (允许超, 只把观察清空)") +def _(): + plan = [f"SH63{i:04d}" for i in range(10)] + hold = {f"SZ00{i:04d}" for i in range(5)} + old = set(plan) | hold | {"SH640001"} + d = pool.decide(old, {"SH640001": {"added": "2026-07-01"}}, plan, hold, set(), 12, TODAY) + assert d["cap_evicted"] == ["SH640001"] + assert len(d["pool"]) == 15 # 10+5, 超 12 但不砍计划/持仓 + + +@case("恶化名单为空 (结论读不到的降级) → 一只都不回收") +def _(): + old = {"SH600001", "SH600002"} + d = pool.decide(old, {}, [], set(), set(), 60, TODAY) + assert d["recycled"] == [] and set(d["observers"]) == old + + +@case("幂等: 同一天跑两遍, 第二遍无新进无回收, 池子不变") +def _(): + plan = ["SH600001", "SH600002"] + d1 = pool.decide(set(), {}, plan, {"SZ000001"}, set(), 60, TODAY) + d2 = pool.decide(set(d1["pool"]), d1["meta"], plan, {"SZ000001"}, set(), 60, TODAY) + assert d2["pool"] == d1["pool"] + assert d2["new_entrants"] == [] and d2["recycled"] == [] + + +@case("remark 汇总句: 数字对得上, 持仓恶化有警示, 降级有说明") +def _(): + d = pool.decide({"SH600009"}, {"SH600009": {"added": "2026-08-01"}}, + ["SH600001"], {"SH600003"}, {"SH600003"}, 60, TODAY) + r = pool.build_remark(d, "2026-08-04", "2026-08-04 07:20:00", degraded="测试降级说明") + # 池 = SH600001(计划) + SH600003(持仓, 恶化但保留) + SH600009(未恶化留池) = 3 只 + assert "共入池 3 只" in r["summary"], r["summary"] + assert "当日计划 1 只" in r["summary"] and "持仓保留 1 只" in r["summary"] + assert "留池观察 1 只" in r["summary"], r["summary"] + assert "警示" in r["summary"] and "SH600003" in r["summary"] + assert "测试降级说明" in r["summary"] + assert r["retained_positions"] == ["SH600003"] + assert r["factor_details"][0]["selected_codes"] == ["SH600001"] + assert r["factor_details"][0]["trade_date"] == "2026-08-04" + + +@case("回收站文档: 字段与现有格式一致 (group_id/removal_batch/removed_at/stock_code)") +def _(): + d = pool.decide({"SH600004"}, {}, [], set(), {"SH600004"}, 60, TODAY) + now = dt.datetime(2026, 8, 4, 7, 20, 0) + docs = pool.build_recycle_docs(d, "6865e8eb97623ef14325c8d3", "AKG每日选股计划池", + "489281497140", now) + assert len(docs) == 1 + doc = docs[0] + for k in ("group_id", "group_name", "org_id", "removal_batch", "removed_at", + "stock_code", "reason"): + assert k in doc, k + assert doc["stock_code"] == "SH600004" + assert doc["removal_batch"] == now.isoformat() + assert doc["removed_at"] == now + assert "形态恶化" in doc["reason"] + + +# ---------------------------------------------------------------- runner +def main(): + passed, failed = 0, 0 + for name, fn in RESULTS: + try: + fn() + print(f" PASS {name}") + passed += 1 + except Exception: + print(f" FAIL {name}") + traceback.print_exc() + failed += 1 + print("-" * 60) + if failed: + print(f"FAILED: {failed} / {passed + failed}") + sys.exit(1) + print(f"ALL PASS ({passed} cases)") + + +if __name__ == "__main__": + main()