上游主榜默认只回20条: truncated标记+QUERY_EXTRA免改码取全量+--try-limit探参数名

README.md

@@ -60,7 +60,7 @@ scripts/
   test_batch4_units.py 动作引擎 四类自主动作触发与数量口径          11 例
   test_batch5_units.py 决策系统信号流解析与消化口径                  8 例
   test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 65 例
-  test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫   25 例
+  test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫   27 例
   test_wiring.py       装配自检: 服务层→核心→落表 全链路 (内存桩)   51 例
   init_db.py           建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练; 含 DDL 体检)
   check_db.py          实机连通性与表结构自检 (需真实 .env)
@@ -199,7 +199,7 @@ QMT ──trade/order_update──▶ pms-ws ──落 pms_qmt_inbox──▶
 
 ## 已实现 / 待开发
 
-**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器九个调度位；**上游选股计划接口接入**（`/plan` 取候选池、交易日龄硬校验、`theme` 灌行业映射表、页面预览抽屉与不可用横幅）；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 230 例**。
+**已实现**：建表 DDL 与建表脚本；配置与运行参数中心；仓位规划器与安全垫账；命令系统（27 类命令全目录 + 双状态机 + 冲突识别）；方案生成器（降仓凑额四档、升仓、建仓分批、清仓/减至、行业清仓与限额、暂停买入撤单）；账本回放与对账引擎（成交认领、外部成交并入 BASE 告警、以下游为准修正、除权检测、T+1 可用量、连续不一致升级）；规则闸终检；择时执行器实现 B（分日配额、分笔、VWAP/回踩/不追高、14:45 兜底、停牌一字板顺延、窗口耗尽收口、挂单有效期）；动作引擎四类自主动作 + 研判闸客户端 + 提议分流；决策系统信号消化（两条流独立消费组订阅、置信度分档转清仓指令或提议）；管理页面四块 + 运维/日报抽屉；调度器九个调度位；**上游选股计划接口接入**（`/plan` 取候选池、交易日龄硬校验、`theme` 灌行业映射表、页面预览抽屉与不可用横幅）；**ws 直连通道的连接层**（常驻进程 + 出口队列 + 签名 + seq 水位与累积确认，见下）；**单测 232 例**。
 
 ### 下一步（按可动工顺序）
 
@@ -209,7 +209,8 @@ QMT ──trade/order_update──▶ pms-ws ──落 pms_qmt_inbox──▶
 | 2 | **ws 通道联调**（协议 §9 的 S1/S2/S3） | 密钥已交换，可开工 |
 | 2.5 | 部署便利性：`make deploy` 一句话完成 build + up + 各 profile | 待办（每次改动要敲 4 条命令太烦） |
 | 2.6 | **行业硬拦截开闸**：跑一次「强刷 + 灌行业映射」把 `theme` 灌进 `pms_industry_map`，再把 `PMS_SECTOR_SOURCE` 改成 `custom_table` | 可做（数据源已就位，只差一次刷新 + 一个参数） |
-| 2.7 | 上游计划的剩余待确认口径（7 条） | 阻塞：等上游答复，见 `UPSTREAM_PLAN_API.md` §4。`upside`/`gate_covered`/`theme_cap` 已于 07-30 由 `format=md` 输出反推确认 |
+| 2.7 | **主榜取全量**：上游默认只回 20 条（`counts.main=961`），候选池实际只在这 20 条里选 | 先跑 `probe_plan_api.py --try-limit` 探参数名（探到即填 `PMS_PLAN_QUERY_EXTRA`，无需改码）；探不到则等上游给分页方式 |
+| 2.8 | 上游计划的剩余待确认口径（6 条） | 阻塞：等上游答复，见 `UPSTREAM_PLAN_API.md` §4 |
 | 3 | T0 做T（二期） | 可做，设计已有，无外部依赖 |
 | 4 | 择时实现 A（委托决策系统盘中择时） | 阻塞：等 bionic 侧接口 |
 | 5 | 研判闸接通 | 阻塞：等 bionic 侧 `process_intraday_audit` 新增 PMS 请求 direction。客户端已就位，接口好了在页面填 `PMS_JUDGE_API_BASE` 即通 |

UPSTREAM_PLAN_API.md

@@ -92,6 +92,7 @@ PMS_PLAN_MIN_SOURCES      0        evidence.n_sources 下限, 0=不设
 PMS_PLAN_MIN_UPSIDE       0        预期空间下限 (0.5=+50%), 0=不设; 只过滤不排序
 PMS_PLAN_STALE_TDAYS      1        日龄上限 (交易日)
 PMS_PLAN_THEME_SYNC       true     刷新时把 theme 灌进 pms_industry_map
+PMS_PLAN_QUERY_EXTRA      (空)     附加查询串, 如 limit=1000 (取全量用, 见 §4 Q3)
 ```
 
 行业硬拦截的开法: 先跑一次「强刷 + 灌行业映射」(或等 08:40 调度位), 再把
@@ -101,7 +102,7 @@ PMS_PLAN_THEME_SYNC       true     刷新时把 theme 灌进 pms_industry_map
 
 ---
 
-## 3. 容器 → 宿主机 :8300 走不通 (2026-07-30 实测)
+## 3. 容器 → 宿主机 :8300 (2026-07-30 已打通)
 
 宿主机上两个地址都通, 但 `pms-web` 容器里直连超时:
 
@@ -137,6 +138,33 @@ docker compose run --rm --no-deps pms-web python scripts/probe_plan_api.py --try
 
 最后一行是最干净的走法 —— 少一跳 NAT, 也不用碰防火墙。
 
+### 实际结果 (2026-07-30)
+
+宿主机上放行网桥网段后四条路全通:
+
+```bash
+sudo iptables -I INPUT -s 172.16.0.0/12 -p tcp --dport 8300 -j ACCEPT
+```
+
+```
+  通    http://host.docker.internal:8300  当前配置   → 计划 2026-07-29, 主榜 20 条
+  通    http://172.20.0.1:8300      (compose 网络网关)
+  通    http://172.17.0.1:8300      (docker0)
+  通    http://192.168.16.155:8300  (宿主机 LAN IP)
+  不通  http://127.0.0.1:8300       ← 正确, 容器自己的 loopback 里当然没有服务
+```
+
+判断得到证实: 服务一直监听在 `0.0.0.0`, 挡住的是宿主机 INPUT 链。配置保持
+`host.docker.internal` —— 它不依赖具体网段, 换 compose 网络也不用改。
+
+> ⚠️ **`iptables -I` 重启就没了。** 要持久化, 二选一:
+> ```bash
+> sudo apt install iptables-persistent && sudo netfilter-persistent save   # Debian/Ubuntu
+> # 或 firewalld:
+> sudo firewall-cmd --permanent --zone=trusted --add-source=172.16.0.0/12 && sudo firewall-cmd --reload
+> ```
+> 不持久化的话, 服务器重启后候选池会静默变空 —— 页面顶部会挂红横幅, 但没人看页面就发现不了。
+
 ---
 
 ## 4. 口径确认进度
@@ -149,6 +177,7 @@ docker compose run --rm --no-deps pms-web python scripts/probe_plan_api.py --try
 | Q5 | `counts.gate_covered` | md 里写作「全池**档位覆盖** 2386 只」——即被档位模型覆盖到的全池股票数, 与 main+observe(1068) 是包含关系。PMS 只透传展示 |
 | Q6a | `theme_cap=5` 的含义 | md 标题写明「主榜 Top 20（有券商预期、目标价不低于现价, **每主题限额 5**）」——上游**已自行执行**同主题限额。PMS 侧 `PMS_SECTOR_MAX_NAMES=4` 更严, 不冲突 |
 | — | 主榜的隐含前置 | 同一行标题揭示主榜已过两道筛: **有券商预期** + **目标价不低于现价**。所以主榜里不会出现 `upside < 0` |
+| Q3a | 默认返回条数 | **主榜 20 / 观察档 10** (实测), 与 md 版「主榜 Top 20」一致。取全量的方式仍未知 —— 见下 Q3 |
 | — | 快照滞后是已知设计 | md 注: 「本日传导用的行情快照 = 2026-07-28（与计划日不同——**历史降级日口径**）」 |
 
 ### 仍需上游回答
@@ -157,9 +186,25 @@ docker compose run --rm --no-deps pms-web python scripts/probe_plan_api.py --try
 而快照是 07-28。若 date 是生成日, 07-30 开盘该用哪一份? PMS 现在按「至多比今天旧 1 个
 交易日」放行, 盘前 08:40 拉 —— 若上游出计划晚于 08:40, 这个调度位要往后挪。
 
-**Q3 `main` 会不会分页或截断?** `counts.main=961`, 一次应答会给全 961 条吗? 有
+**Q3 (最重要) 怎么取主榜全量? 实测默认只回 20 条。**
-`limit`/`offset` 参数吗? PMS 已把 `counts` (上游全量) 与实际返回条数分开记, 不一致会打
+
-warning, 但需要知道正确的取全量方式。
+```
+counts(上游全量)={'main': 961, 'observe': 107}   returned(本次应答)={'main': 20, 'observe': 10}
+```
+
+`format=md` 那版的标题也印证了: 「主榜 **Top 20**（有券商预期、目标价不低于现价, 每主题限额 5）」。
+两个后果得说清:
+
+1. **候选池实际只在这 20 条里选**, `PMS_PLAN_TOP_N=30` 根本吃不满 —— 排序池比以为的小 48 倍。
+2. **这 20 条已被上游按「每主题限额 5」裁过** (实测主题分布: 储能×5 / 传感器×5 / 整机×5 /
+   整车×3 / 集成电路设计×2)。PMS 侧的行业集中度约束因此是在一个**已经被裁过**的池子上再裁
+   一次 —— 约束还成立, 但它看不到全貌, 也就选不出"上游主题限额之外但更合适"的票。
+
+**请给取全量 (或分页) 的方式**: 参数名是什么? 有上限吗? 还是另有端点?
+在此之前 PMS 侧已备好: 探参数名 `probe_plan_api.py --try-limit` (自动试 limit/top/size/
+page_size/… 十来个常见名), 探到就填 `PMS_PLAN_QUERY_EXTRA=limit=1000`, **不用改代码**。
+`parse_plan` 也已经把 `truncated` 标记算出来, 日志 warning + 页面抽屉橙色横幅都会显式提示,
+不会静默拿 20 条当全量用。
 
 **Q4 `changes` 的结构?** 样例是 `null`。若是「与上一份计划的差异」, 给个非空示例 ——
 PMS 想在页面上标「新进榜 / 掉榜」, 那是上游观点变化最直接的信号。

app/services/param_store.py

@@ -74,6 +74,7 @@ DESC = {
     "PMS_PLAN_MIN_UPSIDE": "候选预期空间下限 (相对现价, 0.5=+50%), 0=不设。只过滤不参与排序; 设了就会把 upside 缺失的行(含整个观察档)一起挡掉",
     "PMS_PLAN_STALE_TDAYS": "计划日龄超此交易日数即判过期拒用 (防上游停更时拿旧榜当今天)",
     "PMS_PLAN_THEME_SYNC": "刷新计划时把 evidence.theme 灌进 pms_industry_map (行业源 custom_table 的数据来源)",
+    "PMS_PLAN_QUERY_EXTRA": "计划接口附加查询串, 如 limit=1000。上游默认只回主榜 20 条(counts 却是 961), 用 probe_plan_api.py --try-limit 探出参数名后填这里",
     "PMS_SECTOR_SOURCE": "行业划分数据源: 空=约束停用 / custom_table (推荐, 由上游计划的 theme 灌数) / gp_stock_category",
     "PMS_SECTOR_MAX_NAMES": "同行业最大持仓只数 (硬拦截)",
     "PMS_SECTOR_MAX_RATIO": "同行业最大占总仓比例 (硬拦截)",

app/services/plan_feed.py

@@ -140,10 +140,20 @@ def parse_plan(payload) -> dict:
                    "observe": _int_or_none(counts.get("observe")),
                    "gate_covered": _int_or_none(counts.get("gate_covered"))},
         "returned": {"main": len(main), "observe": len(observe)},
+        # counts 是上游全量, returned 是这次真给了几条 —— 两者不等就是被截断了。
+        # 2026-07-30 实测: counts.main=961 而只回 20 条 (上游默认 Top20), 排序池远小于
+        # 以为的规模, 且那 20 条已经被上游按"每主题限额 5"裁过。必须显式暴露, 不能当没事。
+        "truncated": {"main": _truncated(counts.get("main"), len(main)),
+                      "observe": _truncated(counts.get("observe"), len(observe))},
         "main": main, "observe": observe, "themes": themes,
     }
 
 
+def _truncated(total, got) -> bool:
+    t = _int_or_none(total)
+    return bool(t is not None and got < t)
+
+
 # ================================================================ 纯逻辑: 新鲜度
 def plan_age_tdays(plan_date, today=None) -> int:
     """计划日期距今的交易日龄。当天=0, 上一个交易日=1; 日期在未来 (为下一交易日出的计划)=0。"""
@@ -243,6 +253,7 @@ def _params() -> dict:
         "min_upside": ps.get_float("PMS_PLAN_MIN_UPSIDE", 0.0),
         "stale_tdays": ps.get_int("PMS_PLAN_STALE_TDAYS", 1),
         "theme_sync": ps.get_bool("PMS_PLAN_THEME_SYNC", True),
+        "query_extra": parse_query_extra(ps.get("PMS_PLAN_QUERY_EXTRA", "")),
         "source": (ps.get("PMS_CANDIDATE_SOURCE", SRC_PLAN_API) or SRC_PLAN_API).strip(),
     }
 
@@ -251,7 +262,27 @@ def enabled() -> bool:
     return bool(_params()["base"])
 
 
-def fetch(*, date=None, base=None, path=None, timeout=None) -> dict:
+def parse_query_extra(text) -> dict:
+    """`"limit=1000&offset=0"` → dict。解析不了就返回空 dict (不炸, 记 warning)。
+
+    存在的理由: 上游默认只回主榜 20 条 (见 parse_plan 的 truncated 注释)。取全量要带哪个
+    参数名还没确认 (limit? top? size?), 探出来以后**只改这个参数就能生效**, 不用改代码。
+    """
+    text = (text or "").strip().lstrip("?")
+    if not text:
+        return {}
+    try:
+        from urllib.parse import parse_qsl
+        out = {k: v for k, v in parse_qsl(text, keep_blank_values=False) if k}
+        if not out:
+            raise ValueError("解析结果为空")
+        return out
+    except Exception as e:
+        logger.warning("[上游计划] PMS_PLAN_QUERY_EXTRA 解析失败, 已忽略 (%r): %s", text, e)
+        return {}
+
+
+def fetch(*, date=None, base=None, path=None, timeout=None, extra_params=None) -> dict:
     """拉一次并解析 (不走缓存、不校新鲜度)。任何失败抛 PlanFeedError。
 
     三个参数全给齐时**不读参数中心** —— 纯取数函数不该依赖 ParamStore, 否则连
@@ -270,9 +301,12 @@ def fetch(*, date=None, base=None, path=None, timeout=None) -> dict:
         path = "/" + path
     url = base + path
     to = int(timeout or 10)
+    q = dict(extra_params or {})
+    if date:
+        q["date"] = date          # date 是我们自己的语义, 不许被 extra 覆盖
     try:
         import requests
-        r = requests.get(url, params=({"date": date} if date else None), timeout=to)
+        r = requests.get(url, params=(q or None), timeout=to)
         r.raise_for_status()
         payload = r.json()
     except PlanFeedError:
@@ -289,7 +323,7 @@ def fetch(*, date=None, base=None, path=None, timeout=None) -> dict:
 def get_plan(*, force: bool = False, date=None) -> dict:
     """带缓存的当前计划。失败同样缓存 FAIL_CACHE_SEC, 但每次调用都照样抛。"""
     p = _params()
-    key = (p["base"], p["path"], date or "")
+    key = (p["base"], p["path"], date or "", tuple(sorted(p["query_extra"].items())))
     now = time.time()
     with _lock:
         fresh_hit = (not force and _cache["key"] == key and _cache["plan"] is not None
@@ -300,7 +334,7 @@ def get_plan(*, force: bool = False, date=None) -> dict:
                 and now - _cache["at"] < FAIL_CACHE_SEC):
             raise PlanFeedError(_cache["error"])
     try:
-        plan = fetch(date=date)
+        plan = fetch(date=date, extra_params=p["query_extra"])
         assert_fresh(plan, max_stale_tdays=p["stale_tdays"])
     except PlanFeedError as e:
         with _lock:
@@ -314,6 +348,12 @@ def get_plan(*, force: bool = False, date=None) -> dict:
     logger.info("[上游计划] %s 主榜 %d / 观察 %d (日龄 %d 交易日) ← %s",
                 plan["date"], plan["returned"]["main"], plan["returned"]["observe"],
                 plan["age_tdays"], plan["url"])
+    if plan["truncated"]["main"]:
+        logger.warning("[上游计划] 主榜被截断: 上游 counts=%s 但只回了 %d 条 —— 候选池实际是从"
+                       " 这 %d 条里选, 且上游已按自己的主题限额裁过。取全量的参数名探出来后"
+                       " 填进 PMS_PLAN_QUERY_EXTRA (如 limit=1000)",
+                       plan["counts"]["main"], plan["returned"]["main"],
+                       plan["returned"]["main"])
     return plan
 
 
@@ -382,6 +422,7 @@ def status() -> dict:
                "heat_date": plan.get("heat_date"),
                "market_snapshot_days": plan.get("market_snapshot_days"),
                "counts": plan["counts"], "returned": plan["returned"],
+               "truncated": plan.get("truncated"),
                "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"),

app/web/static/index.html

@@ -560,6 +560,13 @@
       上游全量 {{ JSON.stringify(planStatus.counts||{}) }} · 本次应答 {{ JSON.stringify(planStatus.returned||{}) }} ·
       theme 灌映射 {{ JSON.stringify(planStatus.theme_sync||{}) }}
     </div>
+    <el-alert v-if="(planStatus.truncated||{}).main" type="warning" effect="dark" show-icon
+              :closable="false" style="margin-bottom:8px"
+              :title="'主榜被上游截断: counts=' + ((planStatus.counts||{}).main) + ' 但只回了 '
+                + ((planStatus.returned||{}).main) + ' 条 —— 候选池实际只在这些里选, 且它们已被'
+                + '上游按「每主题限额 ' + planStatus.theme_cap + '」裁过。'
+                + '取全量的参数名用 probe_plan_api.py --try-limit 探, 探到后填 PMS_PLAN_QUERY_EXTRA'">
+    </el-alert>
     <div class="muted" v-if="planCand" style="margin-bottom:8px">
       按当前参数筛选: 排序池 {{ planCand.considered }} → 合格 {{ planCand.eligible }} →
       取 {{ (planCand.items||[]).length }} 只 · 丢弃 {{ JSON.stringify(planCand.dropped||{}) }}
@@ -585,7 +592,8 @@
       「预期空间」= 券商目标价相对现价的空间 (上游 upside 字段, 2.12 → +212%)。噪音大,
       所以<b>永不参与排序</b> —— 排序始终按 score; 只在 PMS_PLAN_MIN_UPSIDE 上做下限过滤。
       筛选参数在「参数设置」里: PMS_PLAN_TOP_N / PMS_PLAN_TIERS / PMS_PLAN_INCLUDE_OBSERVE /
-      PMS_PLAN_MIN_SCORE / PMS_PLAN_MIN_SOURCES / PMS_PLAN_MIN_UPSIDE / PMS_PLAN_STALE_TDAYS。
+      PMS_PLAN_MIN_SCORE / PMS_PLAN_MIN_SOURCES / PMS_PLAN_MIN_UPSIDE / PMS_PLAN_STALE_TDAYS /
+      PMS_PLAN_QUERY_EXTRA (附加查询串, 如 limit=1000)。
     </div>
     <pre class="json" v-if="planResult">{{ planResult }}</pre>
   </el-drawer>

config/settings.py

@@ -91,6 +91,8 @@ class Settings(BaseSettings):
     #   口径, 噪音大 (榜首能到 +212%), 只做下限过滤, **不参与排序** —— 排序始终是 score
     PMS_PLAN_STALE_TDAYS: int = 1          # 计划日龄超此交易日数即判过期并拒用 (防上游停更)
     PMS_PLAN_THEME_SYNC: bool = True       # 刷新时把 evidence.theme 灌进 pms_industry_map
+    PMS_PLAN_QUERY_EXTRA: str = ""         # 附加查询串, 如 "limit=1000"。上游默认只回主榜
+    #   20 条 (counts 却是 961), 取全量的参数名待确认 —— 探出来填这里即可, 不用改代码
 
     # --- 行业约束 (硬拦截; 数据源接口化) ---
     PMS_SECTOR_SOURCE: str = ""            # "" = 停用并页面提示 / custom_table / gp_stock_category

scripts/probe_plan_api.py

@@ -96,6 +96,62 @@ def _try_bases(ps, args) -> int:
     return 0
 
 
+LIMIT_KEYS = ["limit", "top", "topn", "top_n", "n", "size", "page_size", "per_page",
+              "count", "max", "rows", "num", "full"]
+
+
+def _try_limit(ps, args) -> int:
+    """探"取全量"的参数名。
+
+    上游默认只回主榜 20 条, 而 counts.main=961 —— 候选池实际只在这 20 条里选, 且那 20 条
+    已经被上游按自己的「每主题限额 5」裁过。参数名探出来填进 PMS_PLAN_QUERY_EXTRA 即可,
+    不用改代码。全都不认就只能等上游给分页方式 (UPSTREAM_PLAN_API.md Q3)。
+    """
+    base = (args.base or ps.get("PMS_PLAN_API_BASE", "") or "").strip().rstrip("/")
+    path = ps.get("PMS_PLAN_API_PATH", "/plan")
+    if not base:
+        print("PMS_PLAN_API_BASE 为空, 先把地址配上 (或用 --base)")
+        return 2
+    try:
+        base_plan = pf.fetch(base=base, path=path, timeout=15, date=args.date)
+    except pf.PlanFeedError as e:
+        print(f"基准请求就失败了: {e}")
+        return 1
+    n0, c0 = base_plan["returned"]["main"], base_plan["counts"]["main"]
+    print(f"[基准] 不带任何参数: 主榜回 {n0} 条 (上游 counts={c0})")
+    if not base_plan["truncated"]["main"]:
+        print("      没有截断 —— 已经是全量, 不需要分页参数。")
+        return 0
+    print(f"[探参] 依次试 {len(LIMIT_KEYS)} 个常见参数名, 取值 {args.limit_value}")
+    hits = []
+    for k in LIMIT_KEYS:
+        try:
+            p = pf.fetch(base=base, path=path, timeout=20, date=args.date,
+                         extra_params={k: args.limit_value})
+            n = p["returned"]["main"]
+            flag = "  ← 生效" if n > n0 else ""
+            print(f"  {k:<10}= {args.limit_value:<6} → 主榜 {n} 条{flag}")
+            if n > n0:
+                hits.append((k, n))
+        except pf.PlanFeedError as e:
+            print(f"  {k:<10}= {args.limit_value:<6} → 报错: {str(e).split(': ', 1)[-1][:90]}")
+    print("-" * 62)
+    if not hits:
+        print(f"没有一个参数名生效, 主榜始终 {n0} 条。")
+        print("→ 这条只能问上游: 取全量 961 条的正确方式是什么 (分页? 另一个端点?)")
+        print("   见 UPSTREAM_PLAN_API.md Q3。在此之前候选池就是从这 20 条里选,")
+        print("   并且要知道: 这 20 条已被上游按「每主题限额 5」裁过, PMS 侧的行业约束")
+        print("   是在一个已经被裁过的池子上再裁一次。")
+        return 1
+    hits.sort(key=lambda x: -x[1])
+    k, n = hits[0]
+    print(f"生效: {k}={args.limit_value} → 主榜 {n} 条 (原 {n0})")
+    print("→ 页面「参数设置」把 PMS_PLAN_QUERY_EXTRA 填成下面这串, 或直接:")
+    print(f"   curl -s -X POST http://127.0.0.1:38100/api/params -H 'Content-Type: application/json' \\")
+    print(f"     -d '{{\"key\":\"PMS_PLAN_QUERY_EXTRA\",\"value\":\"{k}={args.limit_value}\"}}'")
+    return 0
+
+
 def main():
     ap = argparse.ArgumentParser()
     ap.add_argument("--date", default=None, help="指定计划日期 YYYY-MM-DD (缺省=上游最新)")
@@ -107,12 +163,17 @@ def main():
                     help="临时覆盖 PMS_PLAN_API_BASE (只影响本次, 不落库)")
     ap.add_argument("--try-bases", action="store_true",
                     help="逐个试候选地址, 报告哪个能通 —— 容器连不上宿主机时先跑这个")
+    ap.add_argument("--try-limit", action="store_true",
+                    help="探取全量的参数名 (上游默认只回主榜 20 条, counts 却是 961)")
+    ap.add_argument("--limit-value", default="1000", help="--try-limit 用的取值 (默认 1000)")
     ap.add_argument("--json", action="store_true", help="原样打印解析后的结构")
     args = ap.parse_args()
 
     from app.services import param_store as ps
     if args.try_bases:
         return _try_bases(ps, args)
+    if args.try_limit:
+        return _try_limit(ps, args)
     base = (args.base or ps.get("PMS_PLAN_API_BASE", "") or "").strip()
     print(f"[1] 接口配置  base={base or '(空 —— 候选池将恒为空)'} "
           f"path={ps.get('PMS_PLAN_API_PATH', '/plan')} "
@@ -130,9 +191,12 @@ def main():
     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"]:
+    if plan["truncated"]["main"]:
-        print(f"    ! 主榜 counts={plan['counts']['main']} 但只回了 "
+        print(f"    ! 主榜被截断: counts={plan['counts']['main']} 但只回了 "
-              f"{plan['returned']['main']} 条 —— 上游做了截断, 确认是否分页")
+              f"{plan['returned']['main']} 条")
+        print(f"      候选池实际只在这 {plan['returned']['main']} 条里选, 且它们已被上游按"
+              f"「每主题限额 {plan['theme_cap']}」裁过。")
+        print(f"      探取全量的参数名: probe_plan_api.py --try-limit")
 
     age = pf.plan_age_tdays(plan["date"])
     limit = ps.get_int("PMS_PLAN_STALE_TDAYS", 1)

scripts/run_tests.py

@@ -11,9 +11,9 @@
   test_batch4_units.py  动作引擎 四类自主动作触发与数量口径             (11 例)
   test_batch5_units.py  决策系统信号流解析与消化口径                    (8 例)
   test_batch6_units.py  ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账    (65 例)
-  test_batch7_units.py  上游选股计划: 解析/新鲜度/候选筛选/取数守卫      (25 例)
+  test_batch7_units.py  上游选股计划: 解析/新鲜度/候选筛选/取数守卫      (27 例)
   test_wiring.py        装配自检: 服务层→核心→落表 全链路 (内存桩)      (51 例)
-                                                                  共 230 例
+                                                                  共 232 例
 任一子集失败即整体失败 (退出码 1)。
 """
 import os

scripts/test_batch7_units.py

@@ -155,6 +155,35 @@ def _():
     assert r["n_sources"] == 7 and r["theme"] == "整车"          # theme 两头空白要去掉
 
 
+@case("解析·截断标记: counts 大于实际返回条数就是被上游截断了 (2026-07-30 实测 961→20)")
+def _():
+    p = pf.parse_plan(SAMPLE)                 # counts.main=961, 实际 6 条
+    assert p["truncated"] == {"main": True, "observe": True}, p["truncated"]
+    full = {"date": "2026-07-29", "counts": {"main": 6, "observe": 3},
+            "main": SAMPLE["main"], "observe": SAMPLE["observe"]}
+    assert pf.parse_plan(full)["truncated"] == {"main": False, "observe": False}
+    # counts 缺失时不能瞎报截断 (无从判断)
+    nc = {"date": "2026-07-29", "main": SAMPLE["main"]}
+    assert pf.parse_plan(nc)["truncated"] == {"main": False, "observe": False}
+
+
+@case("取数·附加查询串: 解析 / 合并进请求 / date 不可被覆盖 / 坏串忽略不炸")
+def _():
+    assert pf.parse_query_extra("limit=1000") == {"limit": "1000"}
+    assert pf.parse_query_extra("?limit=1000&offset=20") == {"limit": "1000", "offset": "20"}
+    assert pf.parse_query_extra("") == {} and pf.parse_query_extra(None) == {}
+    assert pf.parse_query_extra("   ") == {}
+    assert pf.parse_query_extra("这不是查询串") == {}      # 解析不出 k=v → 空, 不抛
+    calls, restore = _with_fake_requests(lambda *a: _FakeResp(SAMPLE))
+    try:
+        pf.fetch(base="http://x:8300", path="/plan", timeout=3,
+                 extra_params={"limit": "1000", "date": "被覆盖的"}, date="2026-07-29")
+    finally:
+        restore()
+    # date 是 PMS 自己的语义, extra 里同名键不许赢
+    assert calls[0]["params"] == {"limit": "1000", "date": "2026-07-29"}, calls[0]["params"]
+
+
 # ================================================================ 新鲜度
 @case("新鲜度·日龄: 当天 0 / 上一交易日 1 / 未来日期按 0 (为下一交易日出的计划)")
 def _():