zlt
/
tradingSystem
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

接上游选股计划接口: 候选池改吃/plan, theme灌行业映射, 单测228例

This commit is contained in:
zlt 2026-07-30 16:21:16 +08:00
parent 0ab611b9e3
commit dcd05fd949
14 changed files with 1382 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
README.md
View File

@ -13,6 +13,7 @@
| `POSITION_MGMT_DESIGN.md` | 总体设计 **V0.4（定稿，开发启动）**：命令系统与管理页面/账本/仓位框架/动作引擎/两道关口/择时执行/下游通道。功能一次性开发，上线按依赖分三步切换 | | `POSITION_MGMT_DESIGN.md` | 总体设计 **V0.4（定稿，开发启动）**：命令系统与管理页面/账本/仓位框架/动作引擎/两道关口/择时执行/下游通道。功能一次性开发，上线按依赖分三步切换 |
| `QMT_WS_PROTOCOL.md` | **PMS ↔ QMT WebSocket 指令与回报协议 V1.0（定稿）**：传输与重连、Ed25519 签名与幂等、消息集、状态机、断线补发与对账兜底、部署前检查清单。**这是下发通道的唯一实现依据** | | `QMT_WS_PROTOCOL.md` | **PMS ↔ QMT WebSocket 指令与回报协议 V1.0（定稿）**：传输与重连、Ed25519 签名与幂等、消息集、状态机、断线补发与对账兜底、部署前检查清单。**这是下发通道的唯一实现依据** |
| `QMT_INTERFACE_REQUIREMENTS.md` | 与 QMT 侧的数据与接口需求清单 **V2.0**：A 部分（只读数据）与 C 部分（切换约定）有效；**B 部分的表通道已废止**，改由上面的 ws 协议承担 | | `QMT_INTERFACE_REQUIREMENTS.md` | 与 QMT 侧的数据与接口需求清单 **V2.0**：A 部分（只读数据）与 C 部分（切换约定）有效；**B 部分的表通道已废止**，改由上面的 ws 协议承担 |
| `UPSTREAM_PLAN_API.md` | **上游选股计划接口 (`/plan`) 的接入记录**：应答结构、PMS 侧五条口径、参数表，以及**待上游确认的 10 个口径问题**（`upside` 单位、`date` 语义、分页、`changes` 结构、`theme` 稳定性…）|
| `ddl_pms_v1.sql` | PMS 全部自有表建表语句（153 代理侧，**14 张**：设计 §11 的 10 张 + ws 通道 3 张 + 现金流水 1 张） | | `ddl_pms_v1.sql` | PMS 全部自有表建表语句（153 代理侧，**14 张**：设计 §11 的 10 张 + ws 通道 3 张 + 现金流水 1 张） |
| `config/settings.py` | 配置（基础设施键名对齐 bionic；业务参数为初值，页面调参持久化到 `pms_runtime_param` 后优先） | | `config/settings.py` | 配置（基础设施键名对齐 bionic；业务参数为初值，页面调参持久化到 `pms_runtime_param` 后优先） |
@ -46,23 +47,26 @@ app/
signal_service.py 盘中信号订阅 (db2 广播 + db3 风控卖出) → 卖出指令或提议 signal_service.py 盘中信号订阅 (db2 广播 + db3 风控卖出) → 卖出指令或提议
ledger_service.py 成交回放 / 对账 / 除权 / 盘前 / 日终结算 / 运营日报 ledger_service.py 成交回放 / 对账 / 除权 / 盘前 / 日终结算 / 运营日报
market.py 行情 (Redis db13) 与参考位 (决策系统主口径 + 兜底自算) market.py 行情 (Redis db13) 与参考位 (决策系统主口径 + 兜底自算)
plan_feed.py **上游选股计划 /plan 接入**: 解析 / 交易日龄校验 / 候选筛选 / theme 灌行业映射
industry.py 行业划分可插拔适配器 (custom_table / gp_stock_category / 停用) industry.py 行业划分可插拔适配器 (custom_table / gp_stock_category / 停用)
ws/runner.py **常驻连接进程 (pms-ws)**: 握手/心跳/重连/补发 + 出口出栈 + 上行落库确认 ws/runner.py **常驻连接进程 (pms-ws)**: 握手/心跳/重连/补发 + 出口出栈 + 上行落库确认
web/ FastAPI + 单页 (Vue3 + ElementPlus)，页面四块 + 运维/日报抽屉 web/ FastAPI + 单页 (Vue3 + ElementPlus)，页面四块 + 运维/日报抽屉
scheduler.py Celery beat 调度总表 (设计 §10 八个调度位 + 三条守卫) scheduler.py Celery beat 调度总表 (九个调度位 + 三条守卫)
scripts/ scripts/
run_tests.py 一次跑完全部单测 (见下方「Docker 部署」) run_tests.py 一次跑完全部单测 (见下方「Docker 部署」)
test_core_units.py 仓位与安全垫核心逻辑 14 例 test_core_units.py 仓位与安全垫核心逻辑 14 例
test_batch2_units.py 命令 / 方案 / 回放对账 纯逻辑 35 例 test_batch2_units.py 命令 / 方案 / 回放对账 纯逻辑 35 例
test_batch3_units.py 规则闸 / 择时执行器实现B 纯逻辑 19 例 test_batch3_units.py 规则闸 / 择时执行器实现B 纯逻辑 21 例
test_batch4_units.py 动作引擎 四类自主动作触发与数量口径 11 例 test_batch4_units.py 动作引擎 四类自主动作触发与数量口径 11 例
test_batch5_units.py 决策系统信号流解析与消化口径 8 例 test_batch5_units.py 决策系统信号流解析与消化口径 8 例
test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 58 例 test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 65 例
test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 35 例 test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫 23 例
test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 51 例
init_db.py 建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练; 含 DDL 体检) init_db.py 建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练; 含 DDL 体检)
check_db.py 实机连通性与表结构自检 (需真实 .env) check_db.py 实机连通性与表结构自检 (需真实 .env)
gen_keys.py ws 通道密钥: 生成 / 只取公钥(--pubkey) / PEM(--pem) / 自检(--check) gen_keys.py ws 通道密钥: 生成 / 只取公钥(--pubkey) / PEM(--pem) / 自检(--check)
reset_ledger.py 清空账本并把回放游标对齐到当前 (影子运行期重来一次; 不碰下游表) reset_ledger.py 清空账本并把回放游标对齐到当前 (影子运行期重来一次; 不碰下游表)
probe_plan_api.py 上游计划实机探活: 通不通 / 字段口径 / 候选筛选结果 / 有没有价
ws_smoke.py ws 联调工具: status/watch/place/cancel/inbox (绕开 dispatch_mode) ws_smoke.py ws 联调工具: status/watch/place/cancel/inbox (绕开 dispatch_mode)
``` ```
@ -116,9 +120,10 @@ git pull && docker compose build && docker compose up -d
| 调度 | 时间 | 任务 | 本批状态 | | 调度 | 时间 | 任务 | 本批状态 |
|---|---|---|---| |---|---|---|---|
| 拉上游计划 | 交易日 08:40 | `/plan` 拉当日选股计划 → 刷候选池缓存 + `theme` 灌 `pms_industry_map` | ✅ |
| 盘前准备 | 交易日 08:50 | T+1 可卖重置 / 参考位取数 / 刹车结算 | ✅ | | 盘前准备 | 交易日 08:50 | T+1 可卖重置 / 参考位取数 / 刹车结算 | ✅ |
| 命令轮询 | 每 1 分钟（全天） | 新命令解析 → 方案生成 → 状态机推进 | ✅ | | 命令轮询 | 每 1 分钟（全天） | 新命令解析 → 方案生成 → 状态机推进 | ✅ |
| 成交回放 | 交易时段每 5 分钟 | `trading_order` 增量回放 + 盘中轻对账 | ✅ | | 成交回放 | 交易时段每 1 分钟 | ws 逐笔入账 + `trading_order` 增量回放 + 盘中轻对账 | ✅ |
| 盘中执行 | 交易时段每 1 分钟 | 方案转指令 → 自主提议扫描 → 择时出手（规则闸终检 → 下发 → 记子单） | ✅ | | 盘中执行 | 交易时段每 1 分钟 | 方案转指令 → 自主提议扫描 → 择时出手（规则闸终检 → 下发 → 记子单） | ✅ |
| 信号消化 | 交易时段每 1 分钟 | 订阅 db2 盘中广播 + db3 风控卖出 → 卖出指令或提议 | ✅ | | 信号消化 | 交易时段每 1 分钟 | 订阅 db2 盘中广播 + db3 风控卖出 → 卖出指令或提议 | ✅ |
| T 仓平回 | 14:50 | 做T强制平回 | 🔜 二期（现只自证 T 仓为 0） | | T 仓平回 | 14:50 | 做T强制平回 | 🔜 二期（现只自证 T 仓为 0） |
@ -194,7 +199,7 @@ QMT ──trade/order_update──▶ pms-ws ──落 pms_qmt_inbox──▶
## 已实现 / 待开发 ## 已实现 / 待开发
**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器八个调度位；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 163 例**。 **已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器九个调度位；**上游选股计划接口接入**（`/plan` 取候选池、交易日龄硬校验、`theme` 灌行业映射表、页面预览抽屉与不可用横幅）；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 228 例**。
### 下一步（按可动工顺序） ### 下一步（按可动工顺序）
@ -203,10 +208,13 @@ QMT ──trade/order_update──▶ pms-ws ──落 pms_qmt_inbox──▶
| 1 | ~~ws 通道的账本侧改造（清单 4~6）~~ | ✅ 2026-07-29 完成 | | 1 | ~~ws 通道的账本侧改造（清单 4~6）~~ | ✅ 2026-07-29 完成 |
| 2 | **ws 通道联调**（协议 §9 的 S1/S2/S3） | 密钥已交换，可开工 | | 2 | **ws 通道联调**（协议 §9 的 S1/S2/S3） | 密钥已交换，可开工 |
| 2.5 | 部署便利性：`make deploy` 一句话完成 build + up + 各 profile | 待办（每次改动要敲 4 条命令太烦） | | 2.5 | 部署便利性：`make deploy` 一句话完成 build + up + 各 profile | 待办（每次改动要敲 4 条命令太烦） |
| 2.6 | **行业硬拦截开闸**：跑一次「强刷 + 灌行业映射」把 `theme` 灌进 `pms_industry_map`，再把 `PMS_SECTOR_SOURCE` 改成 `custom_table` | 可做（数据源已就位，只差一次刷新 + 一个参数） |
| 2.7 | 上游计划的 10 个待确认口径 | 阻塞：等上游答复，见 `UPSTREAM_PLAN_API.md` §3。`upside` 口径定了才敢接进排序 |
| 3 | T0 做T（二期） | 可做，设计已有，无外部依赖 | | 3 | T0 做T（二期） | 可做，设计已有，无外部依赖 |
| 4 | 择时实现 A（委托决策系统盘中择时） | 阻塞：等 bionic 侧接口 | | 4 | 择时实现 A（委托决策系统盘中择时） | 阻塞：等 bionic 侧接口 |
| 5 | 研判闸接通 | 阻塞：等 bionic 侧 `process_intraday_audit` 新增 PMS 请求 direction。客户端已就位，接口好了在页面填 `PMS_JUDGE_API_BASE` 即通 | | 5 | 研判闸接通 | 阻塞：等 bionic 侧 `process_intraday_audit` 新增 PMS 请求 direction。客户端已就位，接口好了在页面填 `PMS_JUDGE_API_BASE` 即通 |
| 6 | bionic 侧配套改造（出口改道 + PMS direction） | 另一仓库 | | 6 | bionic 侧配套改造（出口改道 + PMS direction） | 另一仓库 |
| 7 | `trading_buy_plan` 退场：PMS 已不读它（`PMS_CANDIDATE_SOURCE=plan_api`），确认无其他消费方后上游可停写 | 待上游确认（`UPSTREAM_PLAN_API.md` Q10） |
### ws 通道实现清单 ### ws 通道实现清单

134
UPSTREAM_PLAN_API.md Normal file
View File

@ -0,0 +1,134 @@
# 上游选股计划接口 · 接入记录与待确认口径
> 2026-07-30 建。上游「选股系统」提供 HTTP 计划接口, PMS 的升仓/建仓候选池自此改吃它,
> 不再读 `trading_buy_plan`。本文是接口现状 + PMS 侧接法 + **需要上游回答的问题**。
> 代码入口: `app/services/plan_feed.py`; 单测: `scripts/test_batch7_units.py` (23 例);
> 实机探活: `scripts/probe_plan_api.py`。
---
## 1. 接口现状 (上游给的)
```
GET http://192.168.16.155:8300/plan
?date=2026-07-29 可选, 缺省=最新一份计划
&format=md 可选, 输出格式 (PMS 只用默认的 JSON)
```
应答骨架:
```json
{
"date": "2026-07-29",
"counts": { "main": 961, "observe": 107, "gate_covered": 2386 },
"market_snapshot_days": ["2026-07-28"],
"heat_date": "2026-07-27",
"theme_cap": 5,
"main": [
{ "rank": 1, "code": "SH600418", "name": "江淮汽车", "score": 242.24,
"evidence": { "theme": "整车", "n_sources": 7, "moved_ratio": 0.0 },
"heat": 0.2249, "upside": 2.1203202525935945, "tier": "强传导" }
],
"observe": [
{ "rank": 1, "code": "SH600877", "name": "电科芯片", "score": 101.18,
"evidence": { "theme": "集成电路设计", "n_sources": 9, "moved_ratio": 0.0 },
"heat": 0.4743, "upside": null }
],
"changes": null,
"encoding": "主榜分=200+传导档位×20+组内分（还没热、还便宜）；观察档分=100+0.6z(传导)+0.4z(−热度)"
}
```
字段差异: 观察档没有 `tier`, `upside` 恒为 `null`。代码是前缀式 (`SH600418`),
PMS 内部统一点式 (`600418.SH`), 由 `command_spec.normalize_code` 转。
---
## 2. PMS 侧接法 (已实现)
| 环节 | 落点 |
|---|---|
| 取数与解析 | `plan_feed.fetch()` / `parse_plan()` —— 变形/超时/过期一律抛 `PlanFeedError` |
| 新鲜度 | `assert_fresh()` —— 日龄按**交易日**算, 超 `PMS_PLAN_STALE_TDAYS` 拒用 |
| 筛选 | `select_candidates()` —— score 降序、档位白名单、top-N、观察档闸门 |
| 候选池 | `command_service._candidates()` —— 按 `PMS_CANDIDATE_SOURCE` 取源 |
| 行业源 | 刷新时把 `evidence.theme` upsert 进 `pms_industry_map`, 行业源仍走 `custom_table` |
| 调度 | `pms.plan_pull` 交易日 08:40 (在 premarket 08:50 之前, 盘前算集中度要用映射) |
| 页面 | 头部「上游计划」抽屉 + 不可用时顶部红色横幅; 运维按钮「强刷 + 灌行业映射」 |
| 接口 | `GET /api/upstream/plan` 预览 · `POST /api/ops/plan-refresh` 强刷 |
五条钉死的口径:
1. **计划里没有价格、没有金额。** 上游只回答「买什么、排第几」; 买多少、什么价位是 PMS
自己的活 (sizer/planner)。价格一律现取 `market.get_price`, **取不到就不进池**并记 warning
—— 不进池比拿一个假价格进去安全。
2. **拿不到 ≠ 今天没票可买。** 接口失败时候选池为空 + ERROR 告警, **绝不静默回退**到
`trading_buy_plan`。那张表在目标架构下没有明确写入方, 拿它当事实源比没有候选更危险。
页面顶部会挂红色横幅, 不会安静地什么都不买。
3. **961 只主榜不是买入清单, 是排序池。** 默认按 score 降序取前 30 (`PMS_PLAN_TOP_N`),
档位再按 `PMS_PLAN_TIERS` 白名单过滤 (默认只要「强传导」)。
4. **日龄按交易日算并硬校验。** 默认允许 1 个交易日 (上游是收盘后出下一日的计划),
超了直接拒用。防的是上游停更/长假时拿上周的榜当今天用 —— 这种错在盘中完全静默。
5. **`upside` 不参与任何过滤与排序** (单位未确认, 见下 Q1), 只在页面和探活脚本里展示。
参数 (页面「参数设置」可改, 表值优先):
```
PMS_CANDIDATE_SOURCE plan_api(默认) / buy_plan / both
PMS_PLAN_API_BASE http://192.168.16.155:8300 空=停用(候选池恒为空并告警)
PMS_PLAN_API_PATH /plan
PMS_PLAN_TIMEOUT 10 秒
PMS_PLAN_CACHE_SEC 300 秒 (失败另有 60 秒负缓存, 防每分钟调度位叠超时)
PMS_PLAN_TOP_N 30 主榜取前 N 进候选池
PMS_PLAN_TIERS 强传导 传导档白名单, 空=不按档过滤
PMS_PLAN_INCLUDE_OBSERVE false 观察档是否进候选池
PMS_PLAN_MIN_SCORE 0 score 下限, 0=不设
PMS_PLAN_MIN_SOURCES 0 evidence.n_sources 下限, 0=不设
PMS_PLAN_STALE_TDAYS 1 日龄上限 (交易日)
PMS_PLAN_THEME_SYNC true 刷新时把 theme 灌进 pms_industry_map
```
行业硬拦截的开法: 先跑一次「强刷 + 灌行业映射」(或等 08:40 调度位), 再把
`PMS_SECTOR_SOURCE` 改成 `custom_table`。映射表页面可见可手改, 覆盖面随每天刷新累积
—— 这就是不用「即时查当日计划」的原因: 今天没上榜的持仓票也得有行业标签, 否则它的
行业约束会悄悄失效。
---
## 3. 待上游确认 (按重要性排)
**Q1 `upside` 的单位和口径是什么?** 样例里 0.2617 ~ 2.1203。若是百分比, 那 rank 1 的
「上涨空间 2.12%」小得不足以建仓; 若是倍数, 则是 212%。两种解读会导出完全相反的用法。
分母是什么 —— 压力位 / 目标价 / 近期高点? PMS 现在**完全不用它**, 口径定了才敢接。
**Q2 `date` 是「计划生成日」还是「适用交易日」?** 样例 `date=2026-07-29`,
`market_snapshot_days=["2026-07-28"]`, `heat_date=2026-07-27`。若 date 是生成日, 那么
07-30 开盘该用哪一份? PMS 目前按「date 至多比今天旧 1 个交易日」放行, 口径确认后调
`PMS_PLAN_STALE_TDAYS`。**另: 上游出计划的时点是每天几点?** 08:40 的拉取会不会太早。
**Q3 `main` 会不会分页或截断?** `counts.main=961`, 但一次应答真的会回 961 条吗?
有没有 `limit` / `offset` / `top` 参数? PMS 已把 `counts` (上游全量) 与实际返回条数分开
记录, 两者不一致时会打 warning, 但需要知道正确的取全量方式。
**Q4 `changes` 字段的结构?** 样例是 `null`。若是「与上一份计划的差异」, 请给一个非空
示例 —— PMS 想用它在页面上标「新进榜 / 掉榜」, 那是判断上游观点变化最直接的信号。
**Q5 `counts.gate_covered = 2386` 是什么口径?** 闸门覆盖的股票总数? 与 main+observe
(=1068) 的关系是什么? 目前只做透传展示。
**Q6 `theme` 词表稳定吗? 同一只票的 theme 会日间变动吗?** PMS 把它当行业标签落库,
用来做行业集中度**硬拦截**; 词表漂移会直接影响拦不拦。另: `theme_cap=5` 是上游自己的
同主题限额吗? 若是, PMS 侧的 `PMS_SECTOR_MAX_NAMES` (默认 4) 应否与它对齐。
**Q7 `tier` 的完整枚举?** 样例只见「强传导」。PMS 用它做白名单过滤, 需要知道全集
(以及档位之间的强弱次序)。
**Q8 没有当日计划时的失败语义?** 返回 404 / 空 `main` / 上一日的计划? PMS 目前把
「两档全空」按变形处理并拒用。**接口需要鉴权或有限流吗?** (PMS 默认缓存 300 秒,
每交易日至少一次盘前拉取 + 页面手动强刷。)
**Q9 观察档 `upside` 恒为 `null` 是设计如此吗?** 若观察档本就没有空间测算, PMS 就把它
只当备选池 (当前默认 `PMS_PLAN_INCLUDE_OBSERVE=false`)。
**Q10 (给决策系统侧) PMS 已不再读 `trading_buy_plan`。** 那张表还有别的消费方吗?
若没有, 上游可以停写 —— 少一处没有明确写入方的"事实源"。

20
app/scheduler.py
View File

@ -4,6 +4,7 @@
==================================================== ====================================================
| 调度 | 时间 | 任务 | | 调度 | 时间 | 任务 |
|-----------|----------------------|---------------------------------------------| |-----------|----------------------|---------------------------------------------|
| 拉上游计划| 交易日 08:40 | /plan 拉当日选股计划 + theme 灌行业映射表 |
| 盘前准备 | 交易日 08:50 | T+1 可卖重置 / 参考位取数 / 刹车结算 | | 盘前准备 | 交易日 08:50 | T+1 可卖重置 / 参考位取数 / 刹车结算 |
| 命令轮询 | 每 1 分钟 (全天) | 新命令解析 → 方案生成 → 任务状态机推进 | | 命令轮询 | 每 1 分钟 (全天) | 新命令解析 → 方案生成 → 任务状态机推进 |
| 盘中执行 | 交易时段每 1 分钟 | 择时出手 + 自主提议扫描 (执行器下一批交付) | | 盘中执行 | 交易时段每 1 分钟 | 择时出手 + 自主提议扫描 (执行器下一批交付) |
@ -92,7 +93,7 @@ def _brief(r):
return str(r)[:200] return str(r)[:200]
keep = {k: v for k, v in r.items() keep = {k: v for k, v in r.items()
if k in ("ok", "fills", "actions", "planned", "failed", "diffs", "errors", if k in ("ok", "fills", "actions", "planned", "failed", "diffs", "errors",
"skipped", "avail_reset", "refs", "cursor")} "skipped", "avail_reset", "refs", "cursor", "date", "returned")}
if isinstance(keep.get("diffs"), list): if isinstance(keep.get("diffs"), list):
keep["diffs"] = len(keep["diffs"]) keep["diffs"] = len(keep["diffs"])
if isinstance(keep.get("errors"), list): if isinstance(keep.get("errors"), list):
@ -101,6 +102,22 @@ def _brief(r):
# ================================================================ 任务 # ================================================================ 任务
@celery_app.task(name="pms.plan_pull")
@guard(trade_day=True, respect_exec_halt=False) # 取数动作, 休假模式照跑 (只更新映射不产生指令)
def plan_pull():
"""盘前拉上游选股计划: 刷候选池缓存 + evidence.theme 灌进 pms_industry_map。
放在 premarket (08:50) 之前, 是因为盘前那一跳要用行业映射算集中度。
拉失败照旧只记 ERROR —— 候选池当天就是空的, 升仓类命令会明确报"无票可选",
比拿昨天的榜静默买进去好。
"""
from app.services import plan_feed
plan_feed.invalidate()
plan = plan_feed.get_plan(force=True)
return {"ok": True, "date": plan["date"], "age_tdays": plan.get("age_tdays"),
"returned": plan["returned"], "theme_sync": plan.get("theme_sync")}
@celery_app.task(name="pms.premarket") @celery_app.task(name="pms.premarket")
@guard(trade_day=True) @guard(trade_day=True)
def premarket(): def premarket():
@ -190,6 +207,7 @@ def daily_report():
# ================================================================ beat 调度表 # ================================================================ beat 调度表
celery_app.conf.beat_schedule = { celery_app.conf.beat_schedule = {
"plan_pull": {"task": "pms.plan_pull", "schedule": crontab(hour=8, minute=40)},
"premarket": {"task": "pms.premarket", "schedule": crontab(hour=8, minute=50)}, "premarket": {"task": "pms.premarket", "schedule": crontab(hour=8, minute=50)},
"command_poll": {"task": "pms.command_poll", "schedule": crontab(minute="*")}, "command_poll": {"task": "pms.command_poll", "schedule": crontab(minute="*")},
# ws 通道的成交是推过来的, 落 inbox 后没必要再等 5 分钟才入账 —— 改成每分钟。 # ws 通道的成交是推过来的, 落 inbox 后没必要再等 5 分钟才入账 —— 改成每分钟。

78
app/services/command_service.py
View File

@ -453,34 +453,68 @@ def _price_of(view: dict, ts_code: str):
def _candidates(view: dict) -> list: def _candidates(view: dict) -> list:
"""升仓候选池 = 上游买入计划 ∪ 白名单, 剔除黑名单/已持有/冻结。""" """升仓候选池 = 上游选股计划 (∪ 旧买入计划表) ∪ 白名单, 剔除黑名单/已持有。
from app.repo import downstream_repo
来源由 `PMS_CANDIDATE_SOURCE` 决定: plan_api (默认, 上游 /plan 接口) / buy_plan
(旧 trading_buy_plan 表) / both (并集)。三条口径:
1. **计划不带价格** —— /plan 只回答「买什么、排第几」, 价格一律 market.get_price
现取, 取不到就不进池并记一条 warning。旧表的 target_price 若有则直接用。
2. **上游拿不到不回退旧表** —— 候选池宁可为空。那张表在目标架构下没有明确写入方,
拿它当事实源比没有候选更危险 (沿用「拿不到 ≠ 通过」的纪律)。
3. **白名单必须压过计划票** —— planner 只认 score 一把尺子, 而计划的 score 是 200+
量级 (旧表的 prob_thresh 是 0~1)。白名单是用户点名的票, 故给 max(池内分)+1,
写死 1.0 会让点名票沉到池底。
"""
from app.services import market, plan_feed
held = {x["ts_code"] for x in view["held"]} held = {x["ts_code"] for x in view["held"]}
sp = effective_stock_params() sp = effective_stock_params()
black = {c for c, d in sp.items() if d.get("black")} black = {c for c, d in sp.items() if d.get("black")}
out, seen = [], set() src = (param_store.get("PMS_CANDIDATE_SOURCE", plan_feed.SRC_PLAN_API)
try: or plan_feed.SRC_PLAN_API).strip()
plans = downstream_repo.fetch_buy_plans(is_active=7, limit=100) out, seen, noprice = [], set(), []
except Exception as e:
logger.warning("读上游买入计划失败: %s", e) def _push(c, score, tag, *, price=None, theme=None):
plans = []
for p in plans:
c = p["ts_code"]
if not c or c in held or c in black or c in seen: if not c or c in held or c in black or c in seen:
continue return
px = float(price or 0) or (market.get_price(c) or 0)
if px <= 0:
noprice.append(c)
return
seen.add(c) seen.add(c)
out.append({"ts_code": c, "price": p.get("price") or 0, "score": p.get("score") or 0, # sector 取已配置的行业源 (与持仓侧的 sector_*_map 同一套词表); theme 只作展示,
"sector": industry.get(c), "src": "upstream"}) # 不参与约束 —— 行业源若是 gp_stock_category, 混入上游主题会把累计口径搞乱。
from app.services import market out.append({"ts_code": c, "price": px, "score": float(score or 0),
"sector": industry.get(c), "theme": theme, "src": tag})
if src in (plan_feed.SRC_PLAN_API, plan_feed.SRC_BOTH):
try:
sel = plan_feed.candidates(held=held, black=black)
for r in sel["items"]:
_push(r["ts_code"], r["score"], "plan_api", theme=r.get("theme"))
logger.info("[候选池] 上游计划 %s: 排序池 %s → 合格 %s → 取 %d 只 (丢弃 %s)",
sel.get("date"), sel.get("considered"), sel.get("eligible"),
len(sel["items"]), sel.get("dropped"))
except plan_feed.PlanFeedError as e:
logger.error("[候选池] 上游计划不可用, 计划票一只不进池 (按纪律不回退旧表): %s", e)
except Exception as e:
logger.exception("[候选池] 上游计划处理异常: %s", e)
if src in (plan_feed.SRC_BUY_PLAN, plan_feed.SRC_BOTH):
from app.repo import downstream_repo
try:
for p in downstream_repo.fetch_buy_plans(is_active=7, limit=100):
_push(p["ts_code"], p.get("score"), "buy_plan", price=p.get("price"))
except Exception as e:
logger.warning("[候选池] 读旧买入计划表失败: %s", e)
white_score = max([c["score"] for c in out], default=0.0) + 1.0
for c, d in sp.items(): for c, d in sp.items():
if not d.get("white") or c in held or c in black or c in seen: if d.get("white"):
continue _push(c, white_score, "whitelist")
px = market.get_price(c)
if not px: if noprice:
continue logger.warning("[候选池] %d 只候选取不到价格已剔除 (前 10): %s", len(noprice), noprice[:10])
seen.add(c)
out.append({"ts_code": c, "price": px, "score": 1.0, "sector": industry.get(c),
"src": "whitelist"})
return out return out

5
app/services/industry.py
View File

@ -8,6 +8,11 @@
custom_table 自定义映射表 pms_industry_map (默认建议; 用户后续灌何种划分都只是灌数) custom_table 自定义映射表 pms_industry_map (默认建议; 用户后续灌何种划分都只是灌数)
gp_stock_category 决策系统生态已有的行业表 (映射表就绪前的临时数据源) gp_stock_category 决策系统生态已有的行业表 (映射表就绪前的临时数据源)
""(空) 未配置 → 行业约束自动停用, 页面显著提示 (不静默放行也不误拦) ""(空) 未配置 → 行业约束自动停用, 页面显著提示 (不静默放行也不误拦)
映射表现在有稳定的灌数来源: 上游选股计划 `/plan` 的 `evidence.theme`, 由
`plan_feed` 每次刷新时 upsert 进 pms_industry_map (见 UPSTREAM_PLAN_API.md §2)。
所以本模块**不需要**新增 plan_api 数据源 —— 即时查当日计划有个洞: 今天没上榜的持仓票
查不到 theme, 它的行业硬拦截就悄悄失效了。落库以后覆盖面随每天刷新累积, 且页面可见可手改。
""" """
from __future__ import annotations from __future__ import annotations

19
app/services/param_store.py
View File

@ -61,7 +61,19 @@ DESC = {
"PMS_FILL_MAX_LOSS": "浮亏深于此不走回踩补足", "PMS_FILL_MAX_LOSS": "浮亏深于此不走回踩补足",
"PMS_WEAK_NEG_DAYS": "降仓清弱票: 安全垫连续为负天数", "PMS_WEAK_NEG_DAYS": "降仓清弱票: 安全垫连续为负天数",
"PMS_PROPOSAL_TTL_HOURS": "自主提议待确认有效期 (小时)", "PMS_PROPOSAL_TTL_HOURS": "自主提议待确认有效期 (小时)",
"PMS_SECTOR_SOURCE": "行业划分数据源: 空=约束停用 / custom_table / gp_stock_category", "PMS_CANDIDATE_SOURCE": "升仓候选池来源: plan_api=上游选股计划接口 / buy_plan=旧 trading_buy_plan 表 / both=并集",
"PMS_PLAN_API_BASE": "上游选股计划接口根地址; 留空=停用, 此时候选池恒为空并告警",
"PMS_PLAN_API_PATH": "计划接口路径 (默认 /plan)",
"PMS_PLAN_TIMEOUT": "计划接口超时 (秒)",
"PMS_PLAN_CACHE_SEC": "计划缓存秒数 (上游日频产出)",
"PMS_PLAN_TOP_N": "主榜按 score 降序取前 N 只进候选池",
"PMS_PLAN_TIERS": "传导档白名单 (如 强传导), 留空=不按档过滤",
"PMS_PLAN_INCLUDE_OBSERVE": "观察档是否进候选池",
"PMS_PLAN_MIN_SCORE": "候选 score 下限, 0=不设",
"PMS_PLAN_MIN_SOURCES": "候选 evidence.n_sources 下限, 0=不设",
"PMS_PLAN_STALE_TDAYS": "计划日龄超此交易日数即判过期拒用 (防上游停更时拿旧榜当今天)",
"PMS_PLAN_THEME_SYNC": "刷新计划时把 evidence.theme 灌进 pms_industry_map (行业源 custom_table 的数据来源)",
"PMS_SECTOR_SOURCE": "行业划分数据源: 空=约束停用 / custom_table (推荐, 由上游计划的 theme 灌数) / gp_stock_category",
"PMS_SECTOR_MAX_NAMES": "同行业最大持仓只数 (硬拦截)", "PMS_SECTOR_MAX_NAMES": "同行业最大持仓只数 (硬拦截)",
"PMS_SECTOR_MAX_RATIO": "同行业最大占总仓比例 (硬拦截)", "PMS_SECTOR_MAX_RATIO": "同行业最大占总仓比例 (硬拦截)",
"PMS_EXEC_WINDOW_TDAYS": "任务命令默认执行窗口 (交易日)", "PMS_EXEC_WINDOW_TDAYS": "任务命令默认执行窗口 (交易日)",
@ -248,6 +260,9 @@ _RANGES = {
"PMS_DCA_MAX_RATIO": (0, 1), "PMS_SECTOR_MAX_RATIO": (0, 1), "PMS_T0_RATIO_MAX": (0, 0.3334), "PMS_DCA_MAX_RATIO": (0, 1), "PMS_SECTOR_MAX_RATIO": (0, 1), "PMS_T0_RATIO_MAX": (0, 0.3334),
"PMS_MAX_NAMES": (1, 200), "PMS_TOTAL_SCALE": (0, 10 ** 12), "PMS_MAX_NAMES": (1, 200), "PMS_TOTAL_SCALE": (0, 10 ** 12),
"PMS_EXEC_WINDOW_TDAYS": (1, 20), "PMS_BRAKE_DAYS": (0, 30), "PMS_EXEC_WINDOW_TDAYS": (1, 20), "PMS_BRAKE_DAYS": (0, 30),
"PMS_PLAN_TOP_N": (1, 1000), "PMS_PLAN_TIMEOUT": (1, 120),
"PMS_PLAN_CACHE_SEC": (5, 86400), "PMS_PLAN_STALE_TDAYS": (0, 20),
"PMS_PLAN_MIN_SOURCES": (0, 100),
} }
@ -258,6 +273,8 @@ def _range_check(key, v):
return "PMS_SECTOR_SOURCE 只能是 空 / custom_table / gp_stock_category" return "PMS_SECTOR_SOURCE 只能是 空 / custom_table / gp_stock_category"
if key == "PMS_DISPATCH_MODE" and v not in ("shadow", "ws"): if key == "PMS_DISPATCH_MODE" and v not in ("shadow", "ws"):
return "PMS_DISPATCH_MODE 只能是 shadow / ws" return "PMS_DISPATCH_MODE 只能是 shadow / ws"
if key == "PMS_CANDIDATE_SOURCE" and v not in ("plan_api", "buy_plan", "both"):
return "PMS_CANDIDATE_SOURCE 只能是 plan_api / buy_plan / both"
lo_hi = _RANGES.get(key) lo_hi = _RANGES.get(key)
if lo_hi and isinstance(v, (int, float)) and not isinstance(v, bool): if lo_hi and isinstance(v, (int, float)) and not isinstance(v, bool):
lo, hi = lo_hi lo, hi = lo_hi

378
app/services/plan_feed.py Normal file
View File

@ -0,0 +1,378 @@
# -*- coding: utf-8 -*-
"""
上游选股计划接入 (HTTP /plan) —— 候选池的事实源
================================================
上游「选股系统」2026-07-30 起提供计划接口, 取代原先从 `trading_buy_plan` 捞
`is_active=7` 的路子 (那张表是 trading_service 时代的产物, 目标架构下没有明确写入方)。
GET {PMS_PLAN_API_BASE}{PMS_PLAN_API_PATH}?date=YYYY-MM-DD
→ {date, counts{main,observe,gate_covered}, theme_cap, heat_date,
market_snapshot_days[], main:[...], observe:[...], changes, encoding}
主榜 main 每项: {rank, code("SH600418" 前缀式), name, score, heat, upside, tier,
evidence{theme, n_sources, moved_ratio}}; 观察档 observe 同形但
无 tier、upside 为 null。
五条口径在此钉死 (口径本身的疑问见 UPSTREAM_PLAN_API.md 的待确认项):
1. **计划里没有价格、没有金额。** 上游只回答「买什么、排第几」, 买多少、什么价位是
PMS 自己的活 (sizer/planner)。价格一律走 `market.get_price`, 取不到就不进池 ——
绝不拿 score/upside 当价格用。
2. **拿不到 ≠ 今天没票可买。** 超时、报错、JSON 变形、日期过期一律抛 `PlanFeedError`,
由调用方记 ERROR 且让候选池为空, 绝不静默回退到旧表。沿用研判闸那条
「UNAVAILABLE ≠ PASS」的纪律: 事实源缺失要显式失败, 不能被当成"没有候选"。
3. **961 只主榜不是买入清单, 是排序池。** 按 score 降序取前 `PMS_PLAN_TOP_N`
(默认 30) 进候选, 档位再由 `PMS_PLAN_TIERS` 白名单过滤。
4. **日期新鲜度硬校验。** 计划日期比今天旧超过 `PMS_PLAN_STALE_TDAYS` 个交易日即
过期 (抛错)。防的是节假日/上游停更时拿上周的榜当今天用 —— 这种错在盘中是静默的。
5. **upside 单位未确认** (样例 0.26~2.12, 既可能是 % 也可能是倍数), 因此**默认不参与
任何过滤和排序**, 只透传给页面看。见 UPSTREAM_PLAN_API.md Q1。
行业约束的数据源也从这里来: `evidence.theme` 随每次刷新 upsert 进 `pms_industry_map`,
行业源仍是 `custom_table` (`PMS_SECTOR_SOURCE=custom_table`)。落库而不是即时查, 是因为
即时查有个洞 —— 今天没上榜的持仓票查不到 theme, 行业硬拦截就对它悄悄失效了; 落库以后
覆盖面随时间累积, 且页面可见可手改。
模块级只依赖 stdlib + `app.core.command_spec` (纯逻辑), 其余 (requests / param_store /
pms_repo / tradedays) 一律函数内懒加载 —— 让解析与筛选这两段纯逻辑可以零依赖单测。
"""
from __future__ import annotations
import logging
import threading
import time
from app.core.command_spec import normalize_code
logger = logging.getLogger("pms.plan")
BUCKET_MAIN, BUCKET_OBSERVE = "main", "observe"
# 候选池来源 (PMS_CANDIDATE_SOURCE)
SRC_PLAN_API, SRC_BUY_PLAN, SRC_BOTH = "plan_api", "buy_plan", "both"
FAIL_CACHE_SEC = 60.0 # 失败也缓存一会儿, 免得每分钟的调度位把 10 秒超时叠成雪崩
class PlanFeedError(RuntimeError):
"""上游计划取不到 / 变形 / 过期。**必须**冒泡, 不得吞成空计划 (见头部口径 2)。"""
# ================================================================ 纯逻辑: 解析
def _int_or_none(v):
try:
if v is None or (isinstance(v, str) and not v.strip()):
return None
return int(float(v))
except (TypeError, ValueError):
return None
def _float_or_none(v):
try:
if v is None or (isinstance(v, str) and not v.strip()):
return None
return float(v)
except (TypeError, ValueError):
return None
def _text_or_none(v):
if v is None:
return None
s = str(v).strip()
return s or None
def _rows(raw, bucket: str) -> list:
"""一档榜单归一。坏行跳过而不是整体抛错 —— 单条变形不该让整张计划失效。"""
out, seen = [], set()
for i, it in enumerate(raw or []):
if not isinstance(it, dict):
continue
code = normalize_code(str(it.get("code") or it.get("ts_code") or ""))
if not code or code in seen:
continue
seen.add(code)
ev = it.get("evidence") if isinstance(it.get("evidence"), dict) else {}
rank = _int_or_none(it.get("rank"))
out.append({
"ts_code": code,
"name": _text_or_none(it.get("name")),
"rank": rank if rank is not None else i + 1,
"score": _float_or_none(it.get("score")),
"heat": _float_or_none(it.get("heat")),
"upside": _float_or_none(it.get("upside")), # 单位待确认, 不参与判断
"tier": _text_or_none(it.get("tier")),
"theme": _text_or_none(ev.get("theme")),
"n_sources": _int_or_none(ev.get("n_sources")),
"moved_ratio": _float_or_none(ev.get("moved_ratio")),
"bucket": bucket,
})
return out
def parse_plan(payload) -> dict:
"""应答 → 内部结构。缺 date 或两档全空都算变形 (抛 PlanFeedError)。"""
if not isinstance(payload, dict):
raise PlanFeedError(f"应答不是 JSON 对象: {type(payload).__name__}")
date = _text_or_none(payload.get("date"))
if not date:
raise PlanFeedError("应答缺 date —— 判不了新鲜度, 按取不到处理")
main = _rows(payload.get("main"), BUCKET_MAIN)
observe = _rows(payload.get("observe"), BUCKET_OBSERVE)
if not main and not observe:
raise PlanFeedError(f"计划 {date} 主榜与观察档都是空的")
counts = payload.get("counts") if isinstance(payload.get("counts"), dict) else {}
themes = {}
for r in main + observe: # 主榜在前, 同码以主榜的 theme 为准
if r["theme"] and r["ts_code"] not in themes:
themes[r["ts_code"]] = r["theme"]
return {
"date": date,
"heat_date": _text_or_none(payload.get("heat_date")),
"market_snapshot_days": [str(x) for x in (payload.get("market_snapshot_days") or [])],
"theme_cap": _int_or_none(payload.get("theme_cap")),
"encoding": _text_or_none(payload.get("encoding")),
"counts": {"main": _int_or_none(counts.get("main")),
"observe": _int_or_none(counts.get("observe")),
"gate_covered": _int_or_none(counts.get("gate_covered"))},
"returned": {"main": len(main), "observe": len(observe)},
"main": main, "observe": observe, "themes": themes,
}
# ================================================================ 纯逻辑: 新鲜度
def plan_age_tdays(plan_date, today=None) -> int:
"""计划日期距今的交易日龄。当天=0, 上一个交易日=1; 日期在未来 (为下一交易日出的计划)=0。"""
from app.core import tradedays as td
left = td.trade_days_left(today, plan_date) # 含首尾; plan_date 晚于 today 时为 0
return max(0, left - 1)
def assert_fresh(plan: dict, *, max_stale_tdays: int = 1, today=None) -> int:
age = plan_age_tdays(plan["date"], today)
if age > int(max_stale_tdays or 0):
raise PlanFeedError(
f"上游计划已过期: 日期 {plan['date']} 距今 {age} 个交易日 "
f"(上限 {max_stale_tdays})。上游可能停更 —— 拿旧榜当今天用比没有候选更危险")
return age
# ================================================================ 纯逻辑: 筛选
def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=None,
include_observe: bool = False, min_score=None,
min_sources: int = 0) -> dict:
"""排序池 → 候选清单。
排序: score 降序, 同分按 rank 升序 (上游 rank 已是它自己的最终次序, 拿来当稳定次序)。
tier 白名单**只对带 tier 的行生效** —— 观察档没有 tier, 它的闸门是 include_observe。
"""
held = {normalize_code(c) for c in (held or []) if c}
black = {normalize_code(c) for c in (black or []) if c}
tiers = {str(t).strip() for t in (tiers or []) if str(t).strip()}
min_score = _float_or_none(min_score)
min_sources = int(min_sources or 0)
pool = list(plan.get("main") or [])
if include_observe:
pool += list(plan.get("observe") or [])
dropped = {"held": 0, "black": 0, "tier": 0, "score": 0, "sources": 0, "dup": 0, "capped": 0}
passed, seen = [], set()
for r in sorted(pool, key=lambda x: (-(x.get("score") or 0.0), x.get("rank") or 10 ** 9)):
c = r["ts_code"]
if c in seen:
dropped["dup"] += 1
continue
seen.add(c)
if c in held:
dropped["held"] += 1
continue
if c in black:
dropped["black"] += 1
continue
if tiers and r.get("tier") is not None and r["tier"] not in tiers:
dropped["tier"] += 1
continue
if min_score is not None and (r.get("score") or 0.0) < min_score:
dropped["score"] += 1
continue
if min_sources and (r.get("n_sources") or 0) < min_sources:
dropped["sources"] += 1
continue
passed.append(r)
n = max(0, int(top_n or 0)) or len(passed)
dropped["capped"] = max(0, len(passed) - n)
items = [{"ts_code": r["ts_code"], "name": r["name"], "score": r.get("score") or 0.0,
"sector": r.get("theme"), "theme": r.get("theme"), "tier": r.get("tier"),
"heat": r.get("heat"), "upside": r.get("upside"), "rank": r.get("rank"),
"bucket": r["bucket"], "src": "plan_api"}
for r in passed[:n]]
return {"date": plan.get("date"), "considered": len(pool), "eligible": len(passed),
"items": items, "dropped": dropped}
# ================================================================ 取数 (带缓存)
_cache = {"at": 0.0, "key": None, "plan": None, "error": None}
_lock = threading.Lock()
def _params() -> dict:
from app.services import param_store as ps
return {
"base": (ps.get("PMS_PLAN_API_BASE", "") or "").strip().rstrip("/"),
"path": (ps.get("PMS_PLAN_API_PATH", "/plan") or "/plan").strip(),
"timeout": ps.get_int("PMS_PLAN_TIMEOUT", 10),
"cache_sec": ps.get_int("PMS_PLAN_CACHE_SEC", 300),
"top_n": ps.get_int("PMS_PLAN_TOP_N", 30),
"tiers": ps.get_list("PMS_PLAN_TIERS", []),
"include_observe": ps.get_bool("PMS_PLAN_INCLUDE_OBSERVE", False),
"min_score": ps.get_float("PMS_PLAN_MIN_SCORE", 0.0),
"min_sources": ps.get_int("PMS_PLAN_MIN_SOURCES", 0),
"stale_tdays": ps.get_int("PMS_PLAN_STALE_TDAYS", 1),
"theme_sync": ps.get_bool("PMS_PLAN_THEME_SYNC", True),
"source": (ps.get("PMS_CANDIDATE_SOURCE", SRC_PLAN_API) or SRC_PLAN_API).strip(),
}
def enabled() -> bool:
return bool(_params()["base"])
def fetch(*, date=None, base=None, path=None, timeout=None) -> dict:
"""拉一次并解析 (不走缓存、不校新鲜度)。任何失败抛 PlanFeedError。
三个参数全给齐时**不读参数中心** —— 纯取数函数不该依赖 ParamStore, 否则连
「base 为空立即失败」这条守卫都没法脱库单测。
"""
if base is None or path is None or timeout is None:
p = _params()
base = p["base"] if base is None else base
path = p["path"] if path is None else path
timeout = p["timeout"] if timeout is None else timeout
base = (base or "").strip().rstrip("/")
if not base:
raise PlanFeedError("上游计划接口未配置 (PMS_PLAN_API_BASE 为空)")
path = (path or "/plan").strip() or "/plan"
if not path.startswith("/"):
path = "/" + path
url = base + path
to = int(timeout or 10)
try:
import requests
r = requests.get(url, params=({"date": date} if date else None), timeout=to)
r.raise_for_status()
payload = r.json()
except PlanFeedError:
raise
except Exception as e:
raise PlanFeedError(f"拉取上游计划失败 {url}: {type(e).__name__}: {e}") from e
plan = parse_plan(payload)
plan["url"] = url
plan["fetched_at"] = time.time()
plan["requested_date"] = date
return plan
def get_plan(*, force: bool = False, date=None) -> dict:
"""带缓存的当前计划。失败同样缓存 FAIL_CACHE_SEC, 但每次调用都照样抛。"""
p = _params()
key = (p["base"], p["path"], date or "")
now = time.time()
with _lock:
fresh_hit = (not force and _cache["key"] == key and _cache["plan"] is not None
and now - _cache["at"] < max(1, p["cache_sec"]))
if fresh_hit:
return _cache["plan"]
if (not force and _cache["key"] == key and _cache["error"]
and now - _cache["at"] < FAIL_CACHE_SEC):
raise PlanFeedError(_cache["error"])
try:
plan = fetch(date=date)
assert_fresh(plan, max_stale_tdays=p["stale_tdays"])
except PlanFeedError as e:
with _lock:
_cache.update({"at": time.time(), "key": key, "plan": None, "error": str(e)})
raise
plan["age_tdays"] = plan_age_tdays(plan["date"])
if p["theme_sync"]:
plan["theme_sync"] = _sync_themes_quiet(plan)
with _lock:
_cache.update({"at": time.time(), "key": key, "plan": plan, "error": None})
logger.info("[上游计划] %s 主榜 %d / 观察 %d (日龄 %d 交易日) ← %s",
plan["date"], plan["returned"]["main"], plan["returned"]["observe"],
plan["age_tdays"], plan["url"])
return plan
def invalidate():
with _lock:
_cache.update({"at": 0.0, "key": None, "plan": None, "error": None})
# ================================================================ theme → 行业映射
def sync_themes(plan: dict) -> dict:
"""把 code→theme upsert 进 pms_industry_map (行业源 custom_table 的灌数口径)。"""
from app.repo import pms_repo
from app.services import industry
rows = [{"ts_code": c, "industry": t} for c, t in (plan.get("themes") or {}).items()]
if not rows:
return {"rows": 0, "affected": 0}
n = pms_repo.upsert_industry(rows)
industry.invalidate()
return {"rows": len(rows), "affected": n}
def _sync_themes_quiet(plan: dict) -> dict:
"""theme 落库失败不许阻断候选池 —— 行业约束停用是可接受的降级, 没候选不是。"""
try:
return sync_themes(plan)
except Exception as e:
logger.warning("[上游计划] theme 落 pms_industry_map 失败 (行业约束按未配置降级): %s", e)
return {"rows": 0, "affected": 0, "error": f"{type(e).__name__}: {e}"}
# ================================================================ 对外: 候选与状态
def candidates(*, held=(), black=()) -> dict:
"""参数驱动的候选清单 (不含价格 —— 价格由调用方用 market.get_price 现取)。"""
p = _params()
plan = get_plan()
out = select_candidates(plan, held=held, black=black, top_n=p["top_n"], tiers=p["tiers"],
include_observe=p["include_observe"],
min_score=(p["min_score"] or None),
min_sources=p["min_sources"])
out["age_tdays"] = plan.get("age_tdays")
out["theme_cap"] = plan.get("theme_cap")
return out
def status() -> dict:
"""页面/运维用: 绝不抛错, 失败也要能显示出来。"""
p = _params()
st = {"source": p["source"], "base": p["base"], "path": p["path"],
"top_n": p["top_n"], "tiers": p["tiers"],
"include_observe": p["include_observe"], "stale_tdays": p["stale_tdays"],
"theme_sync": p["theme_sync"], "enabled": bool(p["base"])}
if not p["base"]:
st.update({"ok": False, "hint": "上游计划接口未配置 (PMS_PLAN_API_BASE 为空) —— "
"候选池将为空, 升仓/建仓类命令无票可选"})
return st
try:
plan = get_plan()
except PlanFeedError as e:
st.update({"ok": False, "hint": str(e)})
return st
except Exception as e: # 兜底: status 不许抛
st.update({"ok": False, "hint": f"{type(e).__name__}: {e}"})
return st
st.update({"ok": True, "date": plan["date"], "age_tdays": plan.get("age_tdays"),
"heat_date": plan.get("heat_date"),
"market_snapshot_days": plan.get("market_snapshot_days"),
"counts": plan["counts"], "returned": plan["returned"],
"theme_cap": plan.get("theme_cap"), "encoding": plan.get("encoding"),
"theme_sync": plan.get("theme_sync"),
"fetched_at": plan.get("fetched_at"), "url": plan.get("url"),
"hint": f"计划 {plan['date']} 已就绪 (日龄 {plan.get('age_tdays')} 交易日)"})
return st

39
app/web/main.py
View File

@ -404,6 +404,45 @@ def api_clear_resync():
return ok(_clear) return ok(_clear)
# ================================================================ 上游选股计划
@app.get("/api/upstream/plan")
def api_upstream_plan(limit: int = Query(30), date: str = Query(None),
bucket: str = Query("main")):
"""上游 /plan 预览。绝不抛错 —— 接口不通时 status.ok=false + hint, 页面照常渲染。
candidates_raw 是**不扣持仓/黑名单**的原始筛选结果 (纯看参数选出来什么),
真正下命令时的候选池还会再剔除已持有、黑名单与取不到价的票。
"""
def _plan():
from app.services import plan_feed
st = plan_feed.status()
out = {"ok": True, "status": st, "rows": [], "candidates_raw": None}
if not st.get("ok"):
return out
try:
plan = plan_feed.get_plan(date=(date or None))
rows = plan.get("observe" if bucket == "observe" else "main") or []
out["rows"] = rows[:max(1, int(limit or 30))]
out["candidates_raw"] = plan_feed.candidates()
except Exception as e:
out["status"] = {**st, "ok": False, "hint": f"{type(e).__name__}: {e}"}
return out
return ok(_plan)
@app.post("/api/ops/plan-refresh")
def api_plan_refresh(date: str = Query(None)):
"""强刷上游计划, 并把 evidence.theme 灌进 pms_industry_map (行业源 custom_table 的数据来源)。"""
def _refresh():
from app.services import plan_feed
plan_feed.invalidate()
plan = plan_feed.get_plan(force=True, date=(date or None))
return {"ok": True, "date": plan["date"], "age_tdays": plan.get("age_tdays"),
"returned": plan["returned"], "theme_sync": plan.get("theme_sync"),
"industry": industry.status()}
return ok(_refresh)
@app.get("/api/ops/downstream-schema") @app.get("/api/ops/downstream-schema")
def api_downstream_schema(): def api_downstream_schema():
"""导出下游三表的实际列定义 —— 用于回填 QMT_INTERFACE_REQUIREMENTS D1。""" """导出下游三表的实际列定义 —— 用于回填 QMT_INTERFACE_REQUIREMENTS D1。"""

79
app/web/static/index.html
View File

@ -52,6 +52,9 @@
<el-tag type="warning" effect="dark" size="small">自主档位 {{ ov.autonomy || '-' }}</el-tag> <el-tag type="warning" effect="dark" size="small">自主档位 {{ ov.autonomy || '-' }}</el-tag>
<el-tag :type="dm.shadow ? 'info':'success'" effect="dark" size="small"> <el-tag :type="dm.shadow ? 'info':'success'" effect="dark" size="small">
下发 {{ dm.mode || '-' }}</el-tag> 下发 {{ dm.mode || '-' }}</el-tag>
<el-tag :type="planStatus.ok===false ? 'danger':'success'" effect="dark" size="small">
计划 {{ planStatus.ok===false ? '不可用' : (planStatus.date || '-') }}</el-tag>
<el-button size="small" @click="openPlan">上游计划</el-button>
<el-button size="small" @click="openReport">日报</el-button> <el-button size="small" @click="openReport">日报</el-button>
<el-button size="small" @click="opsDrawer=true">运维</el-button> <el-button size="small" @click="opsDrawer=true">运维</el-button>
<el-button size="small" type="primary" :loading="loading" @click="loadAll">刷新</el-button> <el-button size="small" type="primary" :loading="loading" @click="loadAll">刷新</el-button>
@ -59,6 +62,10 @@
<el-alert v-if="err" class="banner" :title="err" type="error" effect="dark" show-icon <el-alert v-if="err" class="banner" :title="err" type="error" effect="dark" show-icon
@close="err=''"></el-alert> @close="err=''"></el-alert>
<el-alert v-if="planStatus.ok===false" class="banner" type="error" effect="dark" show-icon
:closable="false"
:title="'上游选股计划不可用 —— 升仓/建仓类命令将无票可选: ' + (planStatus.hint||'')">
</el-alert>
<el-alert v-if="ov.sector_ready===false" class="banner" type="warning" effect="dark" show-icon <el-alert v-if="ov.sector_ready===false" class="banner" type="warning" effect="dark" show-icon
:closable="false" :closable="false"
title="行业划分数据源未配置 —— 行业集中度硬拦截已停用, 行业类命令置灰 (设计 §5)"></el-alert> title="行业划分数据源未配置 —— 行业集中度硬拦截已停用, 行业类命令置灰 (设计 §5)"></el-alert>
@ -534,6 +541,51 @@
<pre class="json" v-if="opsResult">{{ opsResult }}</pre> <pre class="json" v-if="opsResult">{{ opsResult }}</pre>
</el-drawer> </el-drawer>
<!-- 上游选股计划抽屉 -->
<el-drawer v-model="planDrawer" title="上游选股计划 (候选池的事实源)" size="56%">
<div class="row" style="margin-bottom:8px">
<el-button type="primary" :loading="planLoading" @click="refreshPlan">
强刷 + 灌行业映射</el-button>
<el-button :loading="planLoading" @click="loadPlan">只重读</el-button>
<span class="muted" style="margin-left:8px">
计划不带价格与金额 —— 买多少/什么价一律 PMS 自己算; 行情取不到价的票会被剔除。
</span>
</div>
<el-alert v-if="planStatus.ok===false" type="error" effect="dark" show-icon :closable="false"
:title="planStatus.hint || '不可用'" style="margin-bottom:8px"></el-alert>
<div class="muted" v-else style="margin-bottom:8px">
计划日 <b>{{ planStatus.date }}</b> (日龄 {{ planStatus.age_tdays }} 交易日) ·
热度日 {{ planStatus.heat_date || '-' }} · 快照 {{ (planStatus.market_snapshot_days||[]).join(',') }} ·
上游主题上限 {{ planStatus.theme_cap }} ·
上游全量 {{ JSON.stringify(planStatus.counts||{}) }} · 本次应答 {{ JSON.stringify(planStatus.returned||{}) }} ·
theme 灌映射 {{ JSON.stringify(planStatus.theme_sync||{}) }}
</div>
<div class="muted" v-if="planCand" style="margin-bottom:8px">
按当前参数筛选: 排序池 {{ planCand.considered }} → 合格 {{ planCand.eligible }} →
取 {{ (planCand.items||[]).length }} 只 · 丢弃 {{ JSON.stringify(planCand.dropped||{}) }}
<span class="mono">{{ (planCand.items||[]).map(x=>x.ts_code).join(' ') }}</span>
</div>
<el-table :data="planRows" size="small" border height="420">
<el-table-column prop="rank" label="rank" width="62"></el-table-column>
<el-table-column prop="ts_code" label="代码" width="104"></el-table-column>
<el-table-column prop="name" label="名称" width="92"></el-table-column>
<el-table-column prop="score" label="score" width="82"></el-table-column>
<el-table-column prop="tier" label="传导档" width="82"></el-table-column>
<el-table-column prop="theme" label="主题" width="120"></el-table-column>
<el-table-column prop="heat" label="heat" width="78"></el-table-column>
<el-table-column label="upside" width="90">
<template #default="s"><span class="muted">{{ s.row.upside }}</span></template>
</el-table-column>
<el-table-column prop="n_sources" label="源数" width="66"></el-table-column>
</el-table>
<div class="muted" style="margin-top:6px">
upside 单位尚未与上游确认 (样例 0.26~2.12), 因此**不参与任何过滤和排序**, 只在此展示。
筛选参数在「参数设置」里: PMS_PLAN_TOP_N / PMS_PLAN_TIERS / PMS_PLAN_INCLUDE_OBSERVE /
PMS_PLAN_MIN_SCORE / PMS_PLAN_MIN_SOURCES / PMS_PLAN_STALE_TDAYS。
</div>
<pre class="json" v-if="planResult">{{ planResult }}</pre>
</el-drawer>
<!-- 日报抽屉 --> <!-- 日报抽屉 -->
<el-drawer v-model="reportDrawer" :title="'运营日报 ' + (report.ymd||'')" size="52%"> <el-drawer v-model="reportDrawer" :title="'运营日报 ' + (report.ymd||'')" size="52%">
<div v-if="(report.report||{}).attention"> <div v-if="(report.report||{}).attention">
@ -562,6 +614,13 @@ createApp({
const instructions = ref([]), ledger = ref([]), proposals = ref([]); const instructions = ref([]), ledger = ref([]), proposals = ref([]);
const report = ref({}), reportDrawer = ref(false), dm = ref({}); const report = ref({}), reportDrawer = ref(false), dm = ref({});
const opsDrawer = ref(false), opsResult = ref(''), opsLoading = ref(false); const opsDrawer = ref(false), opsResult = ref(''), opsLoading = ref(false);
// 上游选股计划: 单独取数, 不塞进 /api/overview —— overview 是轮询的, 上游不通时
// 会把 10 秒超时拖进每一次页面刷新。
const planDrawer = ref(false), planLoading = ref(false), planRaw = ref({}),
planResult = ref('');
const planStatus = computed(() => planRaw.value.status || {});
const planRows = computed(() => planRaw.value.rows || []);
const planCand = computed(() => planRaw.value.candidates_raw || null);
const issuing = ref(false); const issuing = ref(false);
const form = reactive({ cmd_type: '', params: {}, note: '' }); const form = reactive({ cmd_type: '', params: {}, note: '' });
@ -814,12 +873,26 @@ createApp({
const d = await call('get', '/api/ops/downstream-schema'); const d = await call('get', '/api/ops/downstream-schema');
opsResult.value = JSON.stringify(d, null, 2); opsLoading.value = false; opsResult.value = JSON.stringify(d, null, 2); opsLoading.value = false;
} }
async function loadPlan() {
planLoading.value = true;
const d = await call('get', '/api/upstream/plan?limit=50');
planRaw.value = (d && d.ok) ? d : { status: { ok: false, hint: (d||{}).error || '取数失败' } };
planLoading.value = false;
}
async function openPlan() { planDrawer.value = true; await loadPlan(); }
async function refreshPlan() {
planLoading.value = true;
const d = await call('post', '/api/ops/plan-refresh');
planResult.value = JSON.stringify(d, null, 2);
planLoading.value = false;
await Promise.all([loadPlan(), loadOverview()]);
}
async function openReport() { async function openReport() {
const d = await call('get', '/api/report'); report.value = d.data || d || {}; const d = await call('get', '/api/report'); report.value = d.data || d || {};
reportDrawer.value = true; reportDrawer.value = true;
} }
onMounted(loadAll); onMounted(() => { loadAll(); loadPlan(); }); // 计划状态并行拉, 不阻塞首屏
return { tab, loading, err, health, ov, params, catalog, commands, plans, plansOf, return { tab, loading, err, health, ov, params, catalog, commands, plans, plansOf,
positions, lots, lotsOf, instructions, ledger, proposals, report, reportDrawer, positions, lots, lotsOf, instructions, ledger, proposals, report, reportDrawer,
opsDrawer, opsResult, opsLoading, issuing, form, curSpec, dirtyCount, dm, opsDrawer, opsResult, opsLoading, issuing, form, curSpec, dirtyCount, dm,
@ -827,7 +900,9 @@ createApp({
scaleGap, scaleGap,
wsRaw, wsAuto, wsHidePong, wsc, wst, wsMode, wsOrders, wsInbox, wsWarnings, loadWs, wsRaw, wsAuto, wsHidePong, wsc, wst, wsMode, wsOrders, wsInbox, wsWarnings, loadWs,
loadAll, loadParams, saveParams, loadPlans, loadLots, onCmdChange, issue, loadAll, loadParams, saveParams, loadPlans, loadLots, onCmdChange, issue,
cancelCmd, replan, decide, ops, loadSchema, openReport, cancelIns }; cancelCmd, replan, decide, ops, loadSchema, openReport, cancelIns,
planDrawer, planLoading, planStatus, planRows, planCand, planResult,
loadPlan, openPlan, refreshPlan };
} }
}).use(ElementPlus).mount('#app'); }).use(ElementPlus).mount('#app');
</script> </script>

16
config/settings.py
View File

@ -71,6 +71,22 @@ class Settings(BaseSettings):
PMS_WEAK_NEG_DAYS: int = 5 # 降仓「清弱票」判定: 安全垫连续为负 N 日 (设计 §3.2) PMS_WEAK_NEG_DAYS: int = 5 # 降仓「清弱票」判定: 安全垫连续为负 N 日 (设计 §3.2)
PMS_PROPOSAL_TTL_HOURS: int = 24 # 自主提议待确认有效期 (超时置 EXPIRED) PMS_PROPOSAL_TTL_HOURS: int = 24 # 自主提议待确认有效期 (超时置 EXPIRED)
# --- 上游选股计划接口 (候选池的事实源; 口径与待确认项见 UPSTREAM_PLAN_API.md) ---
# 上游只回答「买什么、排第几」, 不给价格金额 —— 买多少/什么价是 PMS 自己算。
# 接口拿不到一律显式失败 (候选池为空 + ERROR), 绝不静默回退旧表。
PMS_CANDIDATE_SOURCE: str = "plan_api" # plan_api(默认) / buy_plan(旧表) / both(并集)
PMS_PLAN_API_BASE: str = "http://192.168.16.155:8300" # 空=停用, 此时候选池恒为空
PMS_PLAN_API_PATH: str = "/plan"
PMS_PLAN_TIMEOUT: int = 10 # 单次请求超时 (秒)
PMS_PLAN_CACHE_SEC: int = 300 # 计划缓存秒数 (上游日频产出, 没必要每跳都拉)
PMS_PLAN_TOP_N: int = 30 # 主榜按 score 降序取前 N 进池 (900+ 只是排序池不是清单)
PMS_PLAN_TIERS: str = "强传导" # 传导档白名单; 留空=不按档过滤
PMS_PLAN_INCLUDE_OBSERVE: bool = False # 观察档是否进候选池 (上游把它定位为备选, 无 upside)
PMS_PLAN_MIN_SCORE: float = 0.0 # score 下限, 0=不设
PMS_PLAN_MIN_SOURCES: int = 0 # evidence.n_sources 下限, 0=不设
PMS_PLAN_STALE_TDAYS: int = 1 # 计划日龄超此交易日数即判过期并拒用 (防上游停更)
PMS_PLAN_THEME_SYNC: bool = True # 刷新时把 evidence.theme 灌进 pms_industry_map
# --- 行业约束 (硬拦截; 数据源接口化) --- # --- 行业约束 (硬拦截; 数据源接口化) ---
PMS_SECTOR_SOURCE: str = "" # "" = 停用并页面提示 / custom_table / gp_stock_category PMS_SECTOR_SOURCE: str = "" # "" = 停用并页面提示 / custom_table / gp_stock_category
PMS_SECTOR_MAX_NAMES: int = 4 PMS_SECTOR_MAX_NAMES: int = 4

119
scripts/probe_plan_api.py Normal file
View File

@ -0,0 +1,119 @@
# -*- coding: utf-8 -*-
"""
上游选股计划接口实机探活 (只读, 不写任何表)
============================================
docker compose run --rm --no-deps pms-web python scripts/probe_plan_api.py
docker compose run --rm --no-deps pms-web python scripts/probe_plan_api.py --date 2026-07-29
docker compose run --rm --no-deps pms-web python scripts/probe_plan_api.py --top 30 --with-price
干什么: 拿真实应答验证四件事 —— ① 接口通不通、② 字段口径与单测 fixture 是否一致、
③ 按当前参数筛出来的候选池长什么样、④ (--with-price) 这些票行情里到底有没有价。
第 ④ 项最容易翻车: 计划不带价格, 价格取不到的票会在候选池里被静默剔除。
只读: 不会 upsert pms_industry_map (那由 /api/ops/plan-refresh 或盘前调度做), 也不下单。
"""
from __future__ import annotations
import argparse
import json
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.services import plan_feed as pf # noqa: E402
def _fmt(v, n=4):
return "-" if v is None else (f"{v:.{n}f}" if isinstance(v, float) else str(v))
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--date", default=None, help="指定计划日期 YYYY-MM-DD (缺省=上游最新)")
ap.add_argument("--top", type=int, default=20, help="打印主榜前 N 行")
ap.add_argument("--observe", action="store_true", help="连观察档一起打印")
ap.add_argument("--with-price", action="store_true",
help="逐只查行情价 (需 Redis 通; 用来看候选池会不会因无价掉票)")
ap.add_argument("--json", action="store_true", help="原样打印解析后的结构")
args = ap.parse_args()
from app.services import param_store as ps
base = (ps.get("PMS_PLAN_API_BASE", "") or "").strip()
print(f"[1] 接口配置 base={base or '(空 —— 候选池将恒为空)'} "
f"path={ps.get('PMS_PLAN_API_PATH', '/plan')} "
f"timeout={ps.get_int('PMS_PLAN_TIMEOUT', 10)}s")
if not base:
print(" → PMS_PLAN_API_BASE 为空。页面「参数设置」填上再跑。")
return 2
try:
plan = pf.fetch(date=args.date)
except pf.PlanFeedError as e:
print(f"[2] 取数失败: {e}")
return 1
print(f"[2] 取数成功 {plan['url']}")
print(f" date={plan['date']} heat_date={plan['heat_date']} "
f"snapshot={plan['market_snapshot_days']} theme_cap={plan['theme_cap']}")
print(f" counts(上游全量)={plan['counts']} returned(本次应答)={plan['returned']}")
if plan["counts"]["main"] is not None and plan["returned"]["main"] != plan["counts"]["main"]:
print(f" ! 主榜 counts={plan['counts']['main']} 但只回了 "
f"{plan['returned']['main']} 条 —— 上游做了截断, 确认是否分页")
age = pf.plan_age_tdays(plan["date"])
limit = ps.get_int("PMS_PLAN_STALE_TDAYS", 1)
verdict = "新鲜" if age <= limit else f"**过期** (上限 {limit}, 候选池会拒用)"
print(f"[3] 新鲜度 日龄 {age} 个交易日 → {verdict}")
print(f"[4] 主榜前 {args.top} (score 降序即上游 rank 序)")
hdr = f" {'rank':>4} {'代码':<11} {'名称':<8} {'score':>8} {'档位':<6} {'主题':<10} " \
f"{'heat':>7} {'upside':>8}"
print(hdr + (" " + "行情价" if args.with_price else ""))
rows = plan["main"][:max(1, args.top)]
if args.observe:
rows += plan["observe"][:max(1, args.top)]
prices = {}
if args.with_price:
from app.services import market
for r in rows:
try:
prices[r["ts_code"]] = market.get_price(r["ts_code"])
except Exception as e:
prices[r["ts_code"]] = f"ERR {type(e).__name__}"
for r in rows:
line = (f" {str(r['rank']):>4} {r['ts_code']:<11} {(r['name'] or '-'):<8} "
f"{_fmt(r['score'], 2):>8} {(r['tier'] or '观察'):<6} "
f"{(r['theme'] or '-'):<10} {_fmt(r['heat']):>7} {_fmt(r['upside']):>8}")
if args.with_price:
p = prices.get(r["ts_code"])
line += f" {p if p else '**无价(会被剔除)**'}"
print(line)
themes = {}
for r in plan["main"]:
themes[r["theme"] or "(无)"] = themes.get(r["theme"] or "(无)", 0) + 1
top_themes = sorted(themes.items(), key=lambda x: -x[1])[:12]
print(f"[5] 主榜主题分布 (共 {len(themes)} 个主题): " +
", ".join(f"{k}×{v}" for k, v in top_themes))
sel = pf.select_candidates(
plan, top_n=ps.get_int("PMS_PLAN_TOP_N", 30), tiers=ps.get_list("PMS_PLAN_TIERS", []),
include_observe=ps.get_bool("PMS_PLAN_INCLUDE_OBSERVE", False),
min_score=(ps.get_float("PMS_PLAN_MIN_SCORE", 0.0) or None),
min_sources=ps.get_int("PMS_PLAN_MIN_SOURCES", 0))
print(f"[6] 按当前参数筛选 (top_n={ps.get_int('PMS_PLAN_TOP_N', 30)} "
f"tiers={ps.get_list('PMS_PLAN_TIERS', [])} "
f"observe={ps.get_bool('PMS_PLAN_INCLUDE_OBSERVE', False)})")
print(f" 排序池 {sel['considered']} → 合格 {sel['eligible']} → 取 {len(sel['items'])} 只")
print(f" 丢弃明细 {sel['dropped']} (此处未扣持仓/黑名单, 下命令时还会再扣)")
print(" " + ", ".join(x["ts_code"] for x in sel["items"]))
if args.json:
print("[7] 解析结构")
print(json.dumps({k: v for k, v in plan.items() if k not in ("main", "observe")},
ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
sys.exit(main())

18
scripts/run_tests.py
View File

@ -5,13 +5,15 @@
运行: docker compose run --rm pms-web python scripts/run_tests.py 运行: docker compose run --rm pms-web python scripts/run_tests.py
包含: 包含:
test_core_units.py 仓位规划器 / 安全垫与成本账 (14 例) test_core_units.py 仓位规划器 / 安全垫与成本账 (14 例)
test_batch2_units.py 命令状态机 / 方案生成器 / 回放对账纯逻辑 (35 例) test_batch2_units.py 命令状态机 / 方案生成器 / 回放对账纯逻辑 (35 例)
test_batch3_units.py 规则闸 / 择时执行器实现B 纯逻辑 (19 例) test_batch3_units.py 规则闸 / 择时执行器实现B 纯逻辑 (21 例)
test_batch4_units.py 动作引擎 四类自主动作触发与数量口径 (11 例) test_batch4_units.py 动作引擎 四类自主动作触发与数量口径 (11 例)
test_batch5_units.py 决策系统信号流解析与消化口径 (8 例) test_batch5_units.py 决策系统信号流解析与消化口径 (8 例)
test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 (58 例) test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 (65 例)
test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (32 例) test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫 (23 例)
test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (51 例)
共 228 例
任一子集失败即整体失败 (退出码 1)。 任一子集失败即整体失败 (退出码 1)。
""" """
import os import os
@ -22,7 +24,7 @@ HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE) ROOT = os.path.dirname(HERE)
SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py", SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py",
"test_batch4_units.py", "test_batch5_units.py", "test_batch6_units.py", "test_batch4_units.py", "test_batch5_units.py", "test_batch6_units.py",
"test_wiring.py"] "test_batch7_units.py", "test_wiring.py"]
def main(): def main():

400
scripts/test_batch7_units.py Normal file
View File

@ -0,0 +1,400 @@
# -*- coding: utf-8 -*-
"""
第七批模块单测 (零外部依赖, 不连库不触网)
==========================================
运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch7_units.py
覆盖: 上游选股计划接口 (plan_feed) 的解析、新鲜度校验、候选筛选三段纯逻辑。
fixture 用的是 2026-07-30 上游给的真实应答样例 (截取前若干条, 结构一字未改),
坏数据用例另造 —— 「上游改了字段/漏了字段」这类事故必须由单测先炸, 而不是盘中静默少票。
"""
import os
import sys
import traceback
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.services import plan_feed as pf # noqa: E402
RESULTS = []
def case(name):
def deco(fn):
RESULTS.append((name, fn))
return fn
return deco
def raises(exc, fn, *a, **kw):
try:
fn(*a, **kw)
except exc:
return True
except Exception as e:
raise AssertionError(f"期望 {exc.__name__}, 实际 {type(e).__name__}: {e}")
raise AssertionError(f"期望抛 {exc.__name__}, 但没抛")
# ================================================================ fixture
def _m(rank, code, name, score, theme, heat, upside, tier="强传导", n_sources=7, moved=0.0):
return {"rank": rank, "code": code, "name": name, "score": score,
"evidence": {"theme": theme, "n_sources": n_sources, "moved_ratio": moved},
"heat": heat, "upside": upside, "tier": tier}
def _o(rank, code, name, score, theme, heat, n_sources=9, moved=0.075):
return {"rank": rank, "code": code, "name": name, "score": score,
"evidence": {"theme": theme, "n_sources": n_sources, "moved_ratio": moved},
"heat": heat, "upside": None}
SAMPLE = {
"date": "2026-07-29",
"counts": {"main": 961, "observe": 107, "gate_covered": 2386},
"market_snapshot_days": ["2026-07-28"],
"heat_date": "2026-07-27",
"theme_cap": 5,
"main": [
_m(1, "SH600418", "江淮汽车", 242.24, "整车", 0.2249, 2.1203202525935945),
_m(2, "SH688717", "艾罗能源", 242.05, "储能", 0.4165, 1.5817543859649121),
_m(3, "SZ300952", "恒辉安防", 241.91, "传感器", 0.3391, 1.4061712846347607),
_m(4, "SH605598", "上海港湾", 241.72, "储能", 0.214, 1.292594091460947),
_m(6, "SH600875", "东方电气", 241.34, "整机", 0.4074, 1.2340787236824915,
n_sources=9, moved=0.075),
_m(10, "SH600104", "上汽集团", 241.18, "整车", 0.1239, 0.8065614360879017),
],
"observe": [
_o(1, "SH600877", "电科芯片", 101.18, "集成电路设计", 0.4743, moved=0.0),
_o(2, "SH603611", "诺力股份", 101.12, "整机", 0.5868),
_o(3, "SZ000768", "中航西飞", 101.06, "整机", 0.121),
],
"changes": None,
"encoding": "主榜分=200+传导档位×20+组内分（还没热、还便宜）；观察档分=100+0.6z(传导)+0.4z(−热度)",
}
# ================================================================ 解析
@case("解析·真实样例: 计数/主题上限/热度日全部落位, 返回条数与 counts 分开记")
def _():
p = pf.parse_plan(SAMPLE)
assert p["date"] == "2026-07-29" and p["heat_date"] == "2026-07-27"
assert p["theme_cap"] == 5
assert p["counts"] == {"main": 961, "observe": 107, "gate_covered": 2386}
# counts 是上游全量, returned 是这次应答里真的给了几条 —— 混用会看不出上游截断了
assert p["returned"] == {"main": 6, "observe": 3}
assert p["market_snapshot_days"] == ["2026-07-28"]
assert p["encoding"].startswith("主榜分=200")
@case("解析·代码从前缀式归一为点式 (沪/深/科创/创业板四种都要对)")
def _():
p = pf.parse_plan(SAMPLE)
codes = [r["ts_code"] for r in p["main"]]
assert codes[:4] == ["600418.SH", "688717.SH", "300952.SZ", "605598.SH"], codes
assert [r["ts_code"] for r in p["observe"]][2] == "000768.SZ"
@case("解析·字段口径: score/heat/upside 转 float, evidence 拆平, bucket 打标")
def _():
p = pf.parse_plan(SAMPLE)
r = p["main"][0]
assert r["name"] == "江淮汽车" and r["rank"] == 1 and r["tier"] == "强传导"
assert abs(r["score"] - 242.24) < 1e-9 and abs(r["heat"] - 0.2249) < 1e-9
assert abs(r["upside"] - 2.1203202525935945) < 1e-12
assert r["theme"] == "整车" and r["n_sources"] == 7 and r["moved_ratio"] == 0.0
assert r["bucket"] == pf.BUCKET_MAIN and p["observe"][0]["bucket"] == pf.BUCKET_OBSERVE
# 观察档没有 tier、upside 是 null —— 都得是 None 而不是 0
assert p["observe"][0]["tier"] is None and p["observe"][0]["upside"] is None
@case("解析·themes 映射: 主榜优先于观察档, 同码不被观察档覆盖")
def _():
d = dict(SAMPLE)
d["observe"] = list(SAMPLE["observe"]) + [_o(9, "SH600418", "江淮汽车", 100.1, "汽车零部件", 0.2)]
p = pf.parse_plan(d)
assert p["themes"]["600418.SH"] == "整车", p["themes"]["600418.SH"]
assert p["themes"]["600877.SH"] == "集成电路设计"
@case("解析·变形应答一律抛 PlanFeedError (不是 JSON 对象 / 缺 date / 两档全空)")
def _():
raises(pf.PlanFeedError, pf.parse_plan, ["不是对象"])
raises(pf.PlanFeedError, pf.parse_plan, None)
raises(pf.PlanFeedError, pf.parse_plan, {"main": SAMPLE["main"]}) # 缺 date
raises(pf.PlanFeedError, pf.parse_plan, {"date": "2026-07-29", "main": [], "observe": []})
raises(pf.PlanFeedError, pf.parse_plan, {"date": " ", "main": SAMPLE["main"]})
@case("解析·脏行跳过而不整体失效: 非 dict / 缺 code / 同码重复 / evidence 非 dict")
def _():
d = {"date": "2026-07-29", "main": [
"我是脏行", None, 42,
{"rank": 1, "code": "SH600418", "score": 1.0, "evidence": "不是对象"},
{"rank": 2, "code": "", "score": 2.0},
{"rank": 3, "code": "600418", "score": 3.0}, # 与第一条同码 (归一后相同)
{"code": "SZ300952", "score": 4.0}, # 缺 rank
]}
p = pf.parse_plan(d)
codes = [r["ts_code"] for r in p["main"]]
assert codes == ["600418.SH", "300952.SZ"], codes
assert p["main"][0]["theme"] is None and p["main"][0]["n_sources"] is None
# rank 缺失用出现序号补 (第 7 个元素 → 7), 免得排序键出现 None
assert p["main"][1]["rank"] == 7, p["main"][1]["rank"]
assert p["counts"] == {"main": None, "observe": None, "gate_covered": None}
@case("解析·数值容错: 字符串数字可用, 空串/None/非数字一律 None 而不是 0")
def _():
d = {"date": "2026-07-29", "main": [
{"rank": "5", "code": "SH600418", "score": "242.24", "heat": "", "upside": None,
"evidence": {"theme": " 整车 ", "n_sources": "7", "moved_ratio": "abc"}}]}
r = pf.parse_plan(d)["main"][0]
assert r["rank"] == 5 and abs(r["score"] - 242.24) < 1e-9
assert r["heat"] is None and r["upside"] is None and r["moved_ratio"] is None
assert r["n_sources"] == 7 and r["theme"] == "整车" # theme 两头空白要去掉
# ================================================================ 新鲜度
@case("新鲜度·日龄: 当天 0 / 上一交易日 1 / 未来日期按 0 (为下一交易日出的计划)")
def _():
assert pf.plan_age_tdays("2026-07-30", today="2026-07-30") == 0
assert pf.plan_age_tdays("2026-07-29", today="2026-07-30") == 1
assert pf.plan_age_tdays("2026-07-27", today="2026-07-30") == 3
assert pf.plan_age_tdays("2026-07-31", today="2026-07-30") == 0
@case("新鲜度·跨周末只算交易日: 周五的计划到周一仍是 1 个交易日龄")
def _():
# 2026-07-24 周五, 2026-07-27 周一 —— 中间两天不是交易日, 不该被算进日龄
assert pf.plan_age_tdays("2026-07-24", today="2026-07-27") == 1
assert pf.plan_age_tdays("2026-07-24", today="2026-07-28") == 2
@case("新鲜度·超期抛错并说清日龄 (上游停更时拿旧榜当今天比没候选更危险)")
def _():
p = pf.parse_plan(SAMPLE) # date = 2026-07-29
assert pf.assert_fresh(p, max_stale_tdays=1, today="2026-07-30") == 1
assert pf.assert_fresh(p, max_stale_tdays=0, today="2026-07-29") == 0
raises(pf.PlanFeedError, pf.assert_fresh, p, max_stale_tdays=0, today="2026-07-30")
try:
pf.assert_fresh(p, max_stale_tdays=1, today="2026-08-05")
except pf.PlanFeedError as e:
assert "2026-07-29" in str(e) and "交易日" in str(e), str(e)
# ================================================================ 筛选
@case("筛选·按 score 降序取前 N, 同分按上游 rank 稳定次序")
def _():
p = pf.parse_plan(SAMPLE)
r = pf.select_candidates(p, top_n=3)
assert [x["ts_code"] for x in r["items"]] == ["600418.SH", "688717.SH", "300952.SZ"]
assert r["considered"] == 6 and r["eligible"] == 6 and r["dropped"]["capped"] == 3
# 同分: rank 小的在前
d = {"date": "2026-07-29", "main": [_m(9, "SH600001", "甲", 100.0, "整车", 0.1, 1.0),
_m(2, "SH600002", "乙", 100.0, "整车", 0.1, 1.0)]}
rr = pf.select_candidates(pf.parse_plan(d), top_n=2)
assert [x["ts_code"] for x in rr["items"]] == ["600002.SH", "600001.SH"]
@case("筛选·top_n=0 视为不截断 (别把'不限'写成'一只都不要')")
def _():
p = pf.parse_plan(SAMPLE)
assert len(pf.select_candidates(p, top_n=0)["items"]) == 6
assert len(pf.select_candidates(p, top_n=999)["items"]) == 6
@case("筛选·已持有与黑名单剔除, 且两者分开计数 (输入接受前缀式)")
def _():
p = pf.parse_plan(SAMPLE)
r = pf.select_candidates(p, held=["SH600418", "688717.SH"], black=["SZ300952"], top_n=10)
codes = [x["ts_code"] for x in r["items"]]
assert "600418.SH" not in codes and "688717.SH" not in codes and "300952.SZ" not in codes
assert r["dropped"]["held"] == 2 and r["dropped"]["black"] == 1
assert r["eligible"] == 3
@case("筛选·tier 白名单只对带 tier 的行生效; 观察档的闸门是 include_observe")
def _():
d = dict(SAMPLE)
d["main"] = list(SAMPLE["main"]) + [
_m(500, "SH600519", "弱票", 220.0, "白酒", 0.1, 0.5, tier="弱传导")]
p = pf.parse_plan(d)
r = pf.select_candidates(p, tiers=["强传导"], top_n=50)
assert "600519.SH" not in [x["ts_code"] for x in r["items"]]
assert r["dropped"]["tier"] == 1
# 观察档没有 tier —— tiers 非空时也不该被 tier 规则误杀, 但默认根本不进池
assert all(x["bucket"] == "main" for x in r["items"])
r2 = pf.select_candidates(p, tiers=["强传导"], include_observe=True, top_n=50)
obs = [x["ts_code"] for x in r2["items"] if x["bucket"] == "observe"]
assert obs == ["600877.SH", "603611.SH", "000768.SZ"], obs
assert r2["dropped"]["tier"] == 1
@case("筛选·跨档同码只留一次, 且留主榜那份 (主榜分高, 排序天然优先)")
def _():
d = dict(SAMPLE)
d["observe"] = list(SAMPLE["observe"]) + [_o(9, "SH600418", "江淮汽车", 100.1, "汽车零部件", 0.2)]
r = pf.select_candidates(pf.parse_plan(d), include_observe=True, top_n=50)
hit = [x for x in r["items"] if x["ts_code"] == "600418.SH"]
assert len(hit) == 1 and hit[0]["bucket"] == "main" and hit[0]["theme"] == "整车"
assert r["dropped"]["dup"] == 1
@case("筛选·score 下限与 n_sources 下限各自独立计数")
def _():
p = pf.parse_plan(SAMPLE)
r = pf.select_candidates(p, min_score=241.7, top_n=50)
assert [x["ts_code"] for x in r["items"]] == ["600418.SH", "688717.SH", "300952.SZ",
"605598.SH"]
assert r["dropped"]["score"] == 2
r2 = pf.select_candidates(p, min_sources=9, top_n=50)
assert [x["ts_code"] for x in r2["items"]] == ["600875.SH"], r2["items"]
assert r2["dropped"]["sources"] == 5
@case("筛选·输出字段: sector 用 theme 灌 (planner 吃这个), score 缺失兜 0.0 不留 None")
def _():
d = {"date": "2026-07-29", "main": [{"rank": 1, "code": "SH600418",
"evidence": {"theme": "整车"}}]}
x = pf.select_candidates(pf.parse_plan(d), top_n=5)["items"][0]
assert x["score"] == 0.0 and x["sector"] == "整车" and x["theme"] == "整车"
assert x["src"] == "plan_api" and x["ts_code"] == "600418.SH"
assert "price" not in x, "计划不带价格, 这里绝不能凭空造一个价出来"
@case("筛选·计数自洽: considered = eligible + 各项丢弃; items = eligible − capped")
def _():
d = dict(SAMPLE)
d["main"] = list(SAMPLE["main"]) + [
_m(500, "SH600519", "弱票", 220.0, "白酒", 0.1, 0.5, tier="弱传导")]
p = pf.parse_plan(d)
r = pf.select_candidates(p, held=["SH600418"], black=["SZ300952"], tiers=["强传导"],
min_sources=7, top_n=2, include_observe=True)
dr = r["dropped"]
assert r["considered"] == 10, r["considered"]
assert r["considered"] == r["eligible"] + dr["held"] + dr["black"] + dr["tier"] \
+ dr["score"] + dr["sources"] + dr["dup"], (r, dr)
assert len(r["items"]) == r["eligible"] - dr["capped"] == 2
@case("筛选·空计划不炸 (main/observe 缺键) 且 date 透传")
def _():
r = pf.select_candidates({"date": "2026-07-29"}, top_n=5)
assert r["items"] == [] and r["considered"] == 0 and r["eligible"] == 0
assert r["date"] == "2026-07-29"
# ================================================================ 取数守卫
class _FakeResp:
def __init__(self, payload, status=200):
self._p, self.status_code = payload, status
def raise_for_status(self):
if self.status_code >= 400:
raise RuntimeError(f"HTTP {self.status_code}")
def json(self):
if isinstance(self._p, Exception):
raise self._p
return self._p
def _with_fake_requests(handler):
"""把 requests 换成假模块 (装在 sys.modules 上, plan_feed 是函数内 import)。"""
import types
calls = []
fake = types.ModuleType("requests")
def _get(url, params=None, timeout=None):
calls.append({"url": url, "params": params, "timeout": timeout})
return handler(url, params, timeout)
fake.get = _get
prev = sys.modules.get("requests")
sys.modules["requests"] = fake
def restore():
if prev is None:
sys.modules.pop("requests", None)
else:
sys.modules["requests"] = prev
return calls, restore
@case("取数·base 为空立即抛 PlanFeedError, 一个 HTTP 请求都不发")
def _():
calls, restore = _with_fake_requests(
lambda *a: (_ for _ in ()).throw(AssertionError("不该发请求")))
try:
raises(pf.PlanFeedError, pf.fetch, base="", path="/plan", timeout=1)
raises(pf.PlanFeedError, pf.fetch, base=" ", path="/plan", timeout=1)
finally:
restore()
assert calls == [], calls
@case("取数·成功路径: URL 拼接/date 传参/超时透传, 并盖上 url 与 fetched_at")
def _():
calls, restore = _with_fake_requests(lambda *a: _FakeResp(SAMPLE))
try:
p = pf.fetch(base="http://192.168.16.155:8300/", path="plan", timeout=7,
date="2026-07-29")
finally:
restore()
assert calls[0]["url"] == "http://192.168.16.155:8300/plan", calls
assert calls[0]["params"] == {"date": "2026-07-29"} and calls[0]["timeout"] == 7
assert p["date"] == "2026-07-29" and p["requested_date"] == "2026-07-29"
assert p["url"].endswith("/plan") and p["fetched_at"] > 0
# 不传 date 时不能带一个 date=None 的空参上去
calls2, restore2 = _with_fake_requests(lambda *a: _FakeResp(SAMPLE))
try:
pf.fetch(base="http://x:8300", path="/plan", timeout=3)
finally:
restore2()
assert calls2[0]["params"] is None, calls2
@case("取数·HTTP 报错/JSON 坏掉一律包装成 PlanFeedError 并带上 URL")
def _():
for handler in (lambda *a: _FakeResp(SAMPLE, status=500),
lambda *a: _FakeResp(ValueError("Expecting value")),
lambda *a: (_ for _ in ()).throw(OSError("Connection refused"))):
calls, restore = _with_fake_requests(handler)
try:
try:
pf.fetch(base="http://192.168.16.155:8300", path="/plan", timeout=1)
raise AssertionError("应当抛 PlanFeedError")
except pf.PlanFeedError as e:
assert "192.168.16.155:8300/plan" in str(e), str(e)
finally:
restore()
@case("取数·来源常量与旧表来源都在: plan_api / buy_plan / both")
def _():
assert (pf.SRC_PLAN_API, pf.SRC_BUY_PLAN, pf.SRC_BOTH) == ("plan_api", "buy_plan", "both")
assert pf.FAIL_CACHE_SEC > 0, "失败也要缓存一会儿, 否则每分钟的调度位会把超时叠成雪崩"
def main():
import logging
logging.disable(logging.CRITICAL)
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()

102
scripts/test_wiring.py
View File

@ -469,6 +469,9 @@ def install_fakes(prices=None, positions=None, params=None, high5=None):
# 用 "0" 而不是随便一个字符串: next_cursor 只进不退, 且非数字 id 会退化成字典序比较, # 用 "0" 而不是随便一个字符串: next_cursor 只进不退, 且非数字 id 会退化成字典序比较,
# 占位值若比真实 order_id 大 (比如 "SEED"), 游标就永远推不动了。 # 占位值若比真实 order_id 大 (比如 "SEED"), 游标就永远推不动了。
fake.params.setdefault("PMS_REPLAY_CURSOR", "0") fake.params.setdefault("PMS_REPLAY_CURSOR", "0")
# 上游选股计划接口在单测里一律停用 (base 为空 → plan_feed 立即抛 PlanFeedError,
# 不会发出任何 HTTP 请求)。要测候选池的用例自己 stub plan_feed.candidates。
fake.params.setdefault("PMS_PLAN_API_BASE", "")
param_store._cache.update({"at": 0.0, "data": {}, "error": None, "loaded": False}) param_store._cache.update({"at": 0.0, "data": {}, "error": None, "loaded": False})
portfolio.save_neg_streak({}) portfolio.save_neg_streak({})
return fake return fake
@ -481,7 +484,8 @@ def _():
for m in ["app.core.sizer", "app.core.cushion", "app.core.command_spec", "app.core.planner", for m in ["app.core.sizer", "app.core.cushion", "app.core.command_spec", "app.core.planner",
"app.core.recon", "app.core.tradedays", "app.db.session", "app.repo.pms_repo", "app.core.recon", "app.core.tradedays", "app.db.session", "app.repo.pms_repo",
"app.repo.downstream_repo", "app.services.param_store", "app.services.industry", "app.repo.downstream_repo", "app.services.param_store", "app.services.industry",
"app.services.market", "app.services.portfolio", "app.services.command_service", "app.services.market", "app.services.plan_feed",
"app.services.portfolio", "app.services.command_service",
"app.services.ledger_service", "app.web.main", "app.scheduler"]: "app.services.ledger_service", "app.web.main", "app.scheduler"]:
importlib.import_module(m) importlib.import_module(m)
@ -496,7 +500,8 @@ def _():
"/api/positions/{ts_code}/lots", "/api/instructions", "/api/ledger", "/api/report", "/api/positions/{ts_code}/lots", "/api/instructions", "/api/ledger", "/api/report",
"/api/proposals", "/api/proposals/{proposal_id}/decide", "/api/ops/replay", "/api/proposals", "/api/proposals/{proposal_id}/decide", "/api/ops/replay",
"/api/ops/reconcile", "/api/ops/daily-settle", "/api/ops/report", "/api/ops/reconcile", "/api/ops/daily-settle", "/api/ops/report",
"/api/ops/downstream-schema", "/api/industry"] "/api/ops/downstream-schema", "/api/industry",
"/api/upstream/plan", "/api/ops/plan-refresh"]
missing = [p for p in need if p not in paths] missing = [p for p in need if p not in paths]
assert not missing, missing assert not missing, missing
import os as _os import os as _os
@ -508,11 +513,13 @@ def _():
def _(): def _():
from app import scheduler as sch from app import scheduler as sch
names = set(sch.celery_app.conf.beat_schedule) names = set(sch.celery_app.conf.beat_schedule)
assert names == {"premarket", "command_poll", "replay_fills", "intraday_exec", assert names == {"plan_pull", "premarket", "command_poll", "replay_fills",
"signal_digest", "t0_close", "daily_settle", "daily_report"}, names "intraday_exec", "signal_digest", "t0_close", "daily_settle",
"daily_report"}, names
tasks = set(sch.celery_app.tasks) tasks = set(sch.celery_app.tasks)
for t in ("pms.premarket", "pms.command_poll", "pms.replay_fills", "pms.daily_settle", for t in ("pms.premarket", "pms.command_poll", "pms.replay_fills", "pms.daily_settle",
"pms.daily_report", "pms.t0_close", "pms.intraday_exec", "pms.signal_digest"): "pms.daily_report", "pms.t0_close", "pms.intraday_exec", "pms.signal_digest",
"pms.plan_pull"):
assert t in tasks, t assert t in tasks, t
assert sch.in_session(datetime(2026, 7, 27, 10, 0)) is True assert sch.in_session(datetime(2026, 7, 27, 10, 0)) is True
assert sch.in_session(datetime(2026, 7, 27, 12, 0)) is False assert sch.in_session(datetime(2026, 7, 27, 12, 0)) is False
@ -1664,6 +1671,91 @@ def _():
# ---------------------------------------------------------------- runner # ---------------------------------------------------------------- runner
# ================================================================ 候选池 (上游 /plan)
def _stub_sp(csvc, mapping):
"""替换 effective_stock_params, 返回 restore 闭包。"""
orig = csvc.effective_stock_params
csvc.effective_stock_params = lambda: dict(mapping)
return lambda: setattr(csvc, "effective_stock_params", orig)
@case("候选池·plan_api 独占: 价格现取 / 无价剔除 / 黑名单与持仓剔除 / 白名单压过计划票")
def _():
from app.services import command_service as csvc, plan_feed
install_fakes(prices={"600418.SH": 12.5, "300952.SZ": 30.0, "600000.SH": 10.0,
"600104.SH": 18.0},
params={"PMS_TOTAL_SCALE": "980000", "PMS_CANDIDATE_SOURCE": "plan_api"})
restore_sp = _stub_sp(csvc, {"600000.SH": {"white": True}, "600104.SH": {"black": True}})
orig = plan_feed.candidates
try:
plan_feed.candidates = lambda **kw: {
"date": "2026-07-29", "considered": 4, "eligible": 4, "dropped": {},
"items": [{"ts_code": "600418.SH", "score": 242.24, "theme": "整车"},
{"ts_code": "688717.SH", "score": 242.05, "theme": "储能"},
{"ts_code": "300952.SZ", "score": 241.91, "theme": "传感器"},
{"ts_code": "600104.SH", "score": 241.18, "theme": "整车"}]}
pool = csvc._candidates({"held": [{"ts_code": "605598.SH"}]})
finally:
plan_feed.candidates = orig
restore_sp()
by = {c["ts_code"]: c for c in pool}
# 688717.SH 行情里没有价 → 必须剔除 (planner 对 price<=0 只是静默跳过, 那样少票看不出来)
assert "688717.SH" not in by, by
assert "600104.SH" not in by, "黑名单票不得进池"
assert by["600418.SH"]["price"] == 12.5 and by["600418.SH"]["src"] == "plan_api"
assert by["600418.SH"]["theme"] == "整车" and by["600418.SH"]["sector"] is None
# 白名单是用户点名的票, 分必须压过任何计划票 —— planner 只认 score 一把尺子
assert by["600000.SH"]["src"] == "whitelist"
plan_max = max(c["score"] for c in pool if c["src"] == "plan_api")
assert by["600000.SH"]["score"] > plan_max, (by["600000.SH"]["score"], plan_max)
@case("候选池·上游计划不可用: 池为空, 绝不静默回退旧 trading_buy_plan 表")
def _():
from app.repo import downstream_repo
from app.services import command_service as csvc, plan_feed
install_fakes(prices={"601111.SH": 9.0}, params={"PMS_CANDIDATE_SOURCE": "plan_api"})
restore_sp = _stub_sp(csvc, {})
orig_c, orig_bp = plan_feed.candidates, downstream_repo.fetch_buy_plans
def _boom(**kw):
raise plan_feed.PlanFeedError("接口超时")
try:
plan_feed.candidates = _boom
downstream_repo.fetch_buy_plans = lambda **kw: [
{"ts_code": "601111.SH", "price": 9.0, "score": 0.8}]
pool = csvc._candidates({"held": []})
finally:
plan_feed.candidates, downstream_repo.fetch_buy_plans = orig_c, orig_bp
restore_sp()
assert pool == [], f"上游拿不到时必须让候选池为空, 实际拿到 {pool}"
@case("候选池·both: 计划票与旧表票取并集, 同码只留计划那份")
def _():
from app.repo import downstream_repo
from app.services import command_service as csvc, plan_feed
install_fakes(prices={"600418.SH": 12.5, "601111.SH": 9.0},
params={"PMS_CANDIDATE_SOURCE": "both"})
restore_sp = _stub_sp(csvc, {})
orig_c, orig_bp = plan_feed.candidates, downstream_repo.fetch_buy_plans
try:
plan_feed.candidates = lambda **kw: {
"date": "2026-07-29", "considered": 1, "eligible": 1, "dropped": {},
"items": [{"ts_code": "600418.SH", "score": 242.24, "theme": "整车"}]}
downstream_repo.fetch_buy_plans = lambda **kw: [
{"ts_code": "600418.SH", "price": 11.0, "score": 0.9}, # 同码, 计划优先
{"ts_code": "601111.SH", "price": 9.0, "score": 0.8}]
pool = csvc._candidates({"held": []})
finally:
plan_feed.candidates, downstream_repo.fetch_buy_plans = orig_c, orig_bp
restore_sp()
by = {c["ts_code"]: c for c in pool}
assert set(by) == {"600418.SH", "601111.SH"}, by
assert by["600418.SH"]["src"] == "plan_api" and by["600418.SH"]["price"] == 12.5
assert by["601111.SH"]["src"] == "buy_plan" and by["601111.SH"]["price"] == 9.0
def main(): def main():
# 静音日志: 本套里有好几条用例**故意**触发异常与告警来验证「守成」行为 # 静音日志: 本套里有好几条用例**故意**触发异常与告警来验证「守成」行为
# (调度守卫吞异常、外部成交告警、连续对账升级 ERROR、窗口耗尽告警), # (调度守卫吞异常、外部成交告警、连续对账升级 ERROR、窗口耗尽告警),