盘前昨收兜底(阻断); theme不做行业源改gp_stock_category; ST剔除; 吃满top改tier_complete判据

README.md

@@ -60,8 +60,8 @@ scripts/
   test_batch4_units.py 动作引擎 四类自主动作触发与数量口径          11 例
   test_batch5_units.py 决策系统信号流解析与消化口径                  8 例
   test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 65 例
-  test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫   30 例
+  test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫   32 例
-  test_wiring.py       装配自检: 服务层→核心→落表 全链路 (内存桩)   51 例
+  test_wiring.py       装配自检: 服务层→核心→落表 全链路 (内存桩)   52 例
   init_db.py           建表 (应用 ddl_pms_v1.sql, 幂等, 默认演练; 含 DDL 体检)
   check_db.py          实机连通性与表结构自检 (需真实 .env)
   gen_keys.py          ws 通道密钥: 生成 / 只取公钥(--pubkey) / PEM(--pem) / 自检(--check)

UPSTREAM_PLAN_API.md

@@ -269,3 +269,81 @@ PMS_PLAN_TOP_N=30                                                 # 摊开之后
 
 > 注意: 服务器上如果还留着 `PMS_PLAN_QUERY_EXTRA=top=30`, 清掉它。同名键以显式参数为准,
 > 留着不影响功能, 但两个地方写同一件事迟早看走眼。
+
+
+---
+
+## 6. 拿到上游《选股说明·下游对接》之后的修正 (2026-07-31)
+
+对方给了系统说明 + `api.py`, 加上当天盘前的实机跑批, 推翻了两条我们先前的判断, 也补上了
+几条只有实盘才看得见的坑。
+
+### 6.1 `theme` 不能当行业标签 (推翻 07-30 的决定)
+
+07-30 定的是「`theme` 灌进 `pms_industry_map`, 行业源走 `custom_table`」。当天的样例里每条
+都有 theme, 看着挺好。07-31 拉 300 条真实数据, 结论反了:
+
+```
+[5] 主榜主题分布 (共 10 个主题): (无)×205, 整机制造×51, 储能×23, 整机×9, 数据中心×4,
+    整车×3, 锂电×2, 消费类锂电池×1, 动力电池×1, 储能电池×1
+```
+
+三条硬伤:
+
+1. **覆盖率约三成。** 300 条里 205 条没有 theme —— 上游文档写得很清楚:「无板块传导的日子
+   主榜靠"冷+便宜"排序」, 没有传导就没有主题。持仓票不在传导链上就永远没有标签,
+   它的行业硬拦截会**悄悄失效**。
+2. **词表不规范。** `整机制造×51` 与 `整机×9` 并存; `锂电 / 动力电池 / 储能电池 /
+   消费类锂电池` 分成四个。当行业名做集中度累计, 同一门生意会被算成好几个行业。
+3. **语义就不是行业。** 它是**传导主题** —— 事件驱动, 昨天哪个板块异动就叫什么。同一只票
+   今天算「储能」明天算「整机制造」, 集中度约束跟着一起漂。
+
+**改法**: `PMS_PLAN_THEME_SYNC` 默认关; 行业源用 `PMS_SECTOR_SOURCE=gp_stock_category`
+(决策系统生态已有的行业表, 语义正确、覆盖全市场、不随事件漂)。theme 保留在候选阶段的
+`PMS_PLAN_THEME_CAP_LOCAL` 上 —— 防单一传导主题刷屏, 那才是它该干的活。两个维度各干各的。
+
+### 6.2 盘前没有实时价 (阻断性, 已修)
+
+07-31 盘前跑批, 主榜前 30 只**全部**「无价」。原因: `market.get_price` 读 Redis db13 的
+**当日**分钟线, 盘前那张 key 根本不存在。而候选池的规矩是「取不到价就不进池」——
+盘前下的升仓命令会得到一个空候选池, 而且日志上只是一行 warning。
+
+已加 `market.plan_price()`: 实时价优先, 取不到回落**昨收** (日线 `close_qfq`, 按日缓存),
+两者都没有才剔除。候选带 `price_source` 字段, 日志把「用昨收定量」和「彻底无价剔除」分开报。
+规划价本来就只用来算批次数量, 真正的委托价由择时环节按实时行情现定, 用昨收完全够用。
+
+### 6.3 ST 要 PMS 自己剔 (新增)
+
+上游文档: 「名单当前**不剔除 ST / \*ST**; 需要过滤请在下游做」。已加
+`PMS_PLAN_EXCLUDE_ST` (默认开), 判据覆盖 `ST` / `*ST` / `SST` / `S*ST` 前缀、`退市` 前缀、
+`退` 后缀。名称缺失时**不误杀也不静默**: 保留该行, 但计进 `st_unknown` 由页面和探活脚本报出来。
+
+### 6.4 「吃满 top」不等于漏了票 (假警报已消)
+
+主榜分 = 200 + 传导档位×20 + 组内分, **档位是分层的**。我们只要「强传导」, 而强传导的分
+最高、全在榜首。所以吃满 `top=300` 并不意味着漏了强传导。
+
+判据换成 `tier_complete`: 返回的池子里只要出现过一条白名单之外的档位 (比如「弱传导」),
+就说明白名单那几档已经全拿到了。实测 07-31: 300 条里 226 条被 tier 白名单挡掉 → 已经看见
+大量弱传导 → 强传导取全。页面横幅和探活脚本都改看这个, 不再拿「吃满 top」报警。
+
+### 6.5 文档回答掉的口径 (Q2/Q4/Q7/Q8/Q9 结案)
+
+| 问题 | 上游答复 |
+|---|---|
+| Q2 `date` 语义与出单时点 | **date = 数据日 (上一交易日)**; 每交易日早晨约 **07:15 前**就绪; 周二至周六晨跑, 周一盘前读周六那期。→ PMS 的 08:40 调度位与 `STALE_TDAYS=1` 都对得上 |
+| Q4 `changes` | 升降档: 覆盖新增、估值闸翻转、传导链进出, 每日几十只。「首次覆盖 / 新进传导链」本身有信息量 —— 可以做成页面提示, 列进待办 |
+| Q7 `tier` 枚举 | 强传导 > 弱传导 > 无传导 |
+| Q8 查不到 | 返回 **404**; 局域网内部服务, v1 无鉴权 |
+| Q9 观察档无 upside | 观察档=无券商覆盖, 没有估值锚, 所以 `upside` 恒 null。置信度本就低 |
+| upside 用法 | 「一致预期取 90 天内全部研报的简单平均, 且研报报喜不报忧, **绝对值偏高, 请当相对排序用**」→ 印证 PMS 的做法: 永不参与排序 (上游已把"还便宜"折进 score 的组内分, 再排一次是重复计分), 只留 `PMS_PLAN_MIN_UPSIDE` 做下限 |
+
+### 6.6 还要盯着的两件事
+
+1. **强传导规模随市场结构波动**: 「典型 1~5 个主题、合计几只到几十只」, 无板块传导的日子
+   可能一个都没有。所以候选池空**不一定是故障**。实测 07-31 合格只有 10 只
+   (强传导 74 条 ∩ PMS 侧每主题 5 只 × 储能/整机制造 2 个主题)。要不要在没有强传导的日子
+   放宽到「弱传导」, 是个策略选择, 目前保持不放宽。
+2. **赛道硬门槛启用后主榜会收窄到约三分之一** (十五五前沿产业链: 核聚变、商业航天、通信、
+   算力、人工智能、低空经济)。上游说「已建成、暂未启用, 启用后另行通知」。到时候候选池
+   规模会跳变一次, `PMS_PLAN_TOP` 与 `THEME_CAP_LOCAL` 都要重新看。

app/services/command_service.py

@@ -458,8 +458,10 @@ def _candidates(view: dict) -> list:
     来源由 `PMS_CANDIDATE_SOURCE` 决定: plan_api (默认, 上游 /plan 接口) / buy_plan
     (旧 trading_buy_plan 表) / both (并集)。三条口径:
 
-      1. **计划不带价格** —— /plan 只回答「买什么、排第几」, 价格一律 market.get_price
+      1. **计划不带价格** —— /plan 只回答「买什么、排第几」, 价格由 market.plan_price
-         现取, 取不到就不进池并记一条 warning。旧表的 target_price 若有则直接用。
+         现取: 实时价优先, 盘前回落昨收 (db13 存的是当日分钟线, 盘前那张 key 不存在,
+         2026-07-31 盘前实测前 30 只全部无实时价)。两者都没有才剔除并记 warning。
+         旧表的 target_price 若有则直接用。
       2. **上游拿不到不回退旧表** —— 候选池宁可为空。那张表在目标架构下没有明确写入方,
          拿它当事实源比没有候选更危险 (沿用「拿不到 ≠ 通过」的纪律)。
       3. **白名单必须压过计划票** —— planner 只认 score 一把尺子, 而计划的 score 是 200+
@@ -472,19 +474,25 @@ def _candidates(view: dict) -> list:
     black = {c for c, d in sp.items() if d.get("black")}
     src = (param_store.get("PMS_CANDIDATE_SOURCE", plan_feed.SRC_PLAN_API)
            or plan_feed.SRC_PLAN_API).strip()
-    out, seen, noprice = [], set(), []
+    out, seen, noprice, fallback = [], set(), [], []
 
     def _push(c, score, tag, *, price=None, theme=None):
         if not c or c in held or c in black or c in seen:
             return
-        px = float(price or 0) or (market.get_price(c) or 0)
+        px, psrc = float(price or 0), "given"
+        if px <= 0:
+            pp = market.plan_price(c)
+            px, psrc = float(pp["price"] or 0), pp["source"]
         if px <= 0:
             noprice.append(c)
             return
+        if psrc == "prev_close":
+            fallback.append(c)
         seen.add(c)
         # sector 取已配置的行业源 (与持仓侧的 sector_*_map 同一套词表); theme 只作展示,
-        # 不参与约束 —— 行业源若是 gp_stock_category, 混入上游主题会把累计口径搞乱。
+        # 不参与约束 —— theme 是"传导主题"(事件驱动、天天变、覆盖率约三成), 拿它当行业标签
-        out.append({"ts_code": c, "price": px, "score": float(score or 0),
+        # 会让集中度约束跟着漂, 已于 2026-07-31 明确不再灌行业表 (见 UPSTREAM_PLAN_API.md §6)。
+        out.append({"ts_code": c, "price": px, "price_source": psrc, "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):
@@ -513,8 +521,12 @@ def _candidates(view: dict) -> list:
         if d.get("white"):
             _push(c, white_score, "whitelist")
 
+    if fallback:
+        logger.info("[候选池] %d 只候选无实时价, 已用昨收定量 (盘前正常): %s",
+                    len(fallback), fallback[:10])
     if noprice:
-        logger.warning("[候选池] %d 只候选取不到价格已剔除 (前 10): %s", len(noprice), noprice[:10])
+        logger.warning("[候选池] %d 只候选实时价与昨收都取不到, 已剔除 (前 10): %s",
+                       len(noprice), noprice[:10])
     return out
 
 

app/services/market.py

@@ -170,6 +170,46 @@ def self_calc_refs(ts_code: str, base_cost=None) -> dict:
 _ma_cache = {"day": None, "data": {}}
 
 
+def get_last_close(ts_code: str):
+    """上一交易日收盘 (日线 close_qfq)。**盘前定价的兜底** —— 按日缓存。
+
+    db13 里是**当日**分钟线, 盘前那张 key 根本不存在, get_price 必然返回 None。
+    2026-07-31 盘前实测: 上游计划前 30 只全部"无价", 候选池会整个空掉。计划价只用来
+    算批次数量 (真正的委托价由择时环节按实时行情现定), 用昨收完全够用。
+    """
+    today = datetime.now().strftime("%Y%m%d")
+    key = f"PC:{ts_code}"
+    if _ma_cache["day"] != today:
+        _ma_cache.update({"day": today, "data": {}})
+    if key in _ma_cache["data"]:
+        return _ma_cache["data"][key]
+    val = None
+    try:
+        rows = _factor_rows(ts_code, days=3)
+        closes = [float(r["close_qfq"]) for r in rows if r.get("close_qfq")]
+        if closes:
+            val = round(closes[-1], 3)
+    except Exception as e:
+        logger.warning("昨收取数失败 [%s]: %s", ts_code, e)
+    _ma_cache["data"][key] = val
+    return val
+
+
+def plan_price(ts_code: str) -> dict:
+    """规划用价: 实时价优先, 盘前/停更回落昨收。返回 {price, source}。
+
+    source: realtime / prev_close / none —— **一定要带出来**。拿昨收当现价去做"不追高"
+    这类判断会出错, 所以只给规划期定量用, 判断类的检查仍走 day_snapshot。
+    """
+    px = get_price(ts_code)
+    if px:
+        return {"price": px, "source": "realtime"}
+    px = get_last_close(ts_code)
+    if px:
+        return {"price": px, "source": "prev_close"}
+    return {"price": None, "source": "none"}
+
+
 def get_ma5(ts_code: str):
     """MA5 (规则闸「不追高」用)。按日缓存 —— 因子分表是日频数据, 盘中重复查没意义。"""
     today = datetime.now().strftime("%Y%m%d")

app/services/param_store.py

@@ -77,7 +77,8 @@ DESC = {
     "PMS_PLAN_MIN_SOURCES": "候选 evidence.n_sources 下限, 0=不设",
     "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_EXCLUDE_ST": "剔除 ST/*ST/退市整理 (上游名单明确不剔除, 这道闸在 PMS 侧)",
+    "PMS_PLAN_THEME_SYNC": "把 theme 灌进 pms_industry_map。默认关 —— theme 是事件驱动的传导主题(覆盖约三成、词表不规范), 当行业标签会让集中度约束天天漂; 行业源请用 gp_stock_category",
     "PMS_PLAN_QUERY_EXTRA": "计划接口附加查询串逃生口 (上游加了新参数时不用改代码); top/obs_top/theme_cap 请用各自的显式参数, 同名键以显式参数为准",
     "PMS_SECTOR_SOURCE": "行业划分数据源: 空=约束停用 / custom_table (推荐, 由上游计划的 theme 灌数) / gp_stock_category",
     "PMS_SECTOR_MAX_NAMES": "同行业最大持仓只数 (硬拦截)",

app/services/plan_feed.py

@@ -30,10 +30,12 @@
      到 +212% —— 所以**永不参与排序** (排序始终是 score), 只提供下限过滤
      `PMS_PLAN_MIN_UPSIDE`。
 
-行业约束的数据源也从这里来: `evidence.theme` 随每次刷新 upsert 进 `pms_industry_map`,
+**`theme` 不是行业, 别拿它做行业约束** (2026-07-31 实测后定的, 见 UPSTREAM_PLAN_API.md §6):
-行业源仍是 `custom_table` (`PMS_SECTOR_SOURCE=custom_table`)。落库而不是即时查, 是因为
+主榜 300 条里 205 条根本没有 theme (没有传导的票靠"冷+便宜"排序), 词表也不规范
-即时查有个洞 —— 今天没上榜的持仓票查不到 theme, 行业硬拦截就对它悄悄失效了; 落库以后
+(整机制造×51 与 整机×9 并存; 锂电/动力电池/储能电池/消费类锂电池 四个分开), 而且它按定义
-覆盖面随时间累积, 且页面可见可手改。
+就是**事件驱动的传导主题**, 天天变。拿它当行业标签, 集中度硬拦截会跟着一起漂。
+所以 `PMS_PLAN_THEME_SYNC` 默认关, 行业源走 `gp_stock_category`; theme 只用在候选阶段的
+`PMS_PLAN_THEME_CAP_LOCAL` 上 —— 防单一传导主题刷屏, 那才是它擅长的事。
 
 模块级只依赖 stdlib + `app.core.command_spec` (纯逻辑), 其余 (requests / param_store /
 pms_repo / tradedays) 一律函数内懒加载 —— 让解析与筛选这两段纯逻辑可以零依赖单测。
@@ -201,7 +203,8 @@ def assert_fresh(plan: dict, *, max_stale_tdays: int = 1, today=None) -> int:
 # ================================================================ 纯逻辑: 筛选
 def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=None,
                       include_observe: bool = False, min_score=None,
-                      min_sources: int = 0, min_upside=None, theme_cap: int = 0) -> dict:
+                      min_sources: int = 0, min_upside=None, theme_cap: int = 0,
+                      exclude_st: bool = True) -> dict:
     """排序池 → 候选清单。
 
     排序: score 降序, 同分按 rank 升序 (上游 rank 已是它自己的最终次序, 拿来当稳定次序)。
@@ -209,6 +212,10 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N
     min_upside 相反, **对所有行生效**: upside 缺失按 0 算一起挡掉 (观察档 upside 恒为
     null, 所以设了下限等于把观察档全挡了)。方向选保守那边 —— 宁可少票。
 
+    exclude_st: 剔除 ST / *ST / S* / 退市整理。上游文档写明「名单当前**不剔除 ST/*ST**;
+    需要过滤请在下游做」—— 那这道闸就是 PMS 的责任, 默认开。判据只有 name 一个字段,
+    所以 name 缺失时**不敢放行**也不敢误杀: 保留但计入 dropped 之外的 st_unknown。
+
     theme_cap: 同主题最多取几只, **在 top_n 截断之前**按 score 序生效 (0=不限)。
     这一层存在的理由: 上游的 theme_cap 是请求参数, 我们可以让它别裁 (要个宽池子), 但
     top_n 那一刀是按纯 score 切的 —— 宽池子里前 30 名可能全是储能, 切完再交给规则闸,
@@ -228,8 +235,8 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N
 
     theme_cap = int(theme_cap or 0)
     dropped = {"held": 0, "black": 0, "tier": 0, "score": 0, "sources": 0, "upside": 0,
-               "theme": 0, "dup": 0, "capped": 0}
+               "st": 0, "theme": 0, "dup": 0, "capped": 0}
-    passed, seen, per_theme = [], set(), {}
+    passed, seen, per_theme, st_unknown = [], 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:
@@ -242,6 +249,13 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N
         if c in black:
             dropped["black"] += 1
             continue
+        if exclude_st:
+            nm = (r.get("name") or "").strip()
+            if not nm:
+                st_unknown.append(c)          # 没名字判不了, 留着但要让人知道
+            elif is_st(nm):
+                dropped["st"] += 1
+                continue
         if tiers and r.get("tier") is not None and r["tier"] not in tiers:
             dropped["tier"] += 1
             continue
@@ -270,7 +284,35 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N
               "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}
+            "items": items, "dropped": dropped, "st_unknown": st_unknown,
+            "tier_complete": _tier_complete(pool, tiers)}
+
+
+def is_st(name: str) -> bool:
+    """ST / *ST / 退市整理 判定 (只看名字 —— 上游只给了这一个可判据)。
+
+    覆盖四种真实写法: 前缀 `ST` / `*ST` / `SST` / `S*ST`, 前缀 `退市`(退市整理期,
+    如「退市海润」), 后缀 `退`(如「海润退」)。A 股名称都是中文, 所以前缀按拉丁字母
+    `ST` 匹配不会误伤 —— 「圣农发展」「深赛格」这类 S 开头的中文名不受影响。
+    """
+    n = (name or "").replace(" ", "").upper()
+    if not n:
+        return False
+    return (n.startswith(("*ST", "S*ST", "SST", "ST", "退市"))
+            or n.endswith(("退", "退市")))
+
+
+def _tier_complete(pool, tiers) -> bool:
+    """白名单档位是不是已经取全了。
+
+    主榜分 = 200 + 传导档位×20 + 组内分, 所以**档位是分层的**: 按 score 降序排, 只要看见
+    了一条白名单之外的档位, 就说明白名单那几档已经全在前面拿到手了。
+    这条判据用来消掉"吃满 top 就报警"的假警报 —— 只要白名单齐了, 后面还有多少弱传导
+    根本不影响候选池。tiers 为空 (不按档过滤) 时无从判断, 返回 None。
+    """
+    if not tiers:
+        return None
+    return any((r.get("tier") is not None and r["tier"] not in tiers) for r in (pool or []))
 
 
 # ================================================================ 取数 (带缓存)
@@ -292,11 +334,12 @@ def _params() -> dict:
         "min_sources": ps.get_int("PMS_PLAN_MIN_SOURCES", 0),
         "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),
+        "theme_sync": ps.get_bool("PMS_PLAN_THEME_SYNC", False),
         "top": ps.get_int("PMS_PLAN_TOP", 300),
         "obs_top": ps.get_int("PMS_PLAN_OBS_TOP", 100),
         "theme_cap": ps.get_int("PMS_PLAN_THEME_CAP", 999),
         "theme_cap_local": ps.get_int("PMS_PLAN_THEME_CAP_LOCAL", 5),
+        "exclude_st": ps.get_bool("PMS_PLAN_EXCLUDE_ST", 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(),
     }
@@ -457,7 +500,7 @@ def candidates(*, held=(), black=()) -> dict:
                             min_score=(p["min_score"] or None),
                             min_sources=p["min_sources"],
                             min_upside=(p["min_upside"] or None),
-                            theme_cap=p["theme_cap_local"])
+                            theme_cap=p["theme_cap_local"], exclude_st=p["exclude_st"])
     out["age_tdays"] = plan.get("age_tdays")
     out["theme_cap"] = plan.get("theme_cap")
     return out

app/web/static/index.html

@@ -562,15 +562,19 @@
       →(券商预期 + 目标价≥现价 + 主题限额 + top)→ 实收
       主榜 {{ (planStatus.funnel||{}).returned_main }} / 观察 {{ (planStatus.funnel||{}).returned_observe }}
     </div>
-    <el-alert v-if="(planStatus.truncated||{}).main" type="warning" effect="dark" show-icon
+    <el-alert v-if="planCand && planCand.tier_complete===false" type="warning" effect="dark"
-              :closable="false" style="margin-bottom:8px"
+              show-icon :closable="false" style="margin-bottom:8px"
-              :title="'主榜正好吃满 top=' + ((planStatus.requested||{}).top)
+              :title="'返回的全是白名单档位 (top=' + ((planStatus.requested||{}).top)
-                + ' —— 可能还有更多没拿到, 调大 PMS_PLAN_TOP 再看'">
+                + ' 吃满了) —— 白名单档位可能没取全, 调大 PMS_PLAN_TOP'">
     </el-alert>
     <div class="muted" v-if="planCand" style="margin-bottom:8px">
       按当前参数筛选: 排序池 {{ planCand.considered }} → 合格 {{ planCand.eligible }} →
-      取 {{ (planCand.items||[]).length }} 只 · 丢弃 {{ JSON.stringify(planCand.dropped||{}) }}
+      取 {{ (planCand.items||[]).length }} 只 ·
-      <span class="mono">{{ (planCand.items||[]).map(x=>x.ts_code).join(' ') }}</span>
+      白名单档位取全 {{ planCand.tier_complete===null ? '不适用' : (planCand.tier_complete ? '是':'否') }} ·
+      丢弃 {{ JSON.stringify(planCand.dropped||{}) }}
+      <span v-if="(planCand.st_unknown||[]).length" style="color:#E6A23C">·
+        {{ planCand.st_unknown.length }} 只无名称, ST 判不了已放行</span>
+      <br><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>
@@ -596,6 +600,9 @@
       <br>发给上游的三个: PMS_PLAN_TOP / PMS_PLAN_OBS_TOP / PMS_PLAN_THEME_CAP (0=不传, 用上游默认
       20/10/5)。<b>PMS_PLAN_THEME_CAP_LOCAL</b> 是 PMS 侧的同主题限额, 在 TOP_N 截断之前生效 ——
       让上游别裁 (THEME_CAP 设大) + PMS 自己裁, 前 N 名才不会被单一主题垄断。
+      <br><b>PMS_PLAN_EXCLUDE_ST</b> 默认开: 上游名单明确不剔除 ST/*ST, 这道闸在 PMS 侧。
+      「主题」是<b>传导主题</b>(事件驱动、天天变、覆盖约三成), <b>不是行业</b> —— 行业硬拦截
+      请把 PMS_SECTOR_SOURCE 设成 gp_stock_category, 别用 PMS_PLAN_THEME_SYNC。
     </div>
     <pre class="json" v-if="planResult">{{ planResult }}</pre>
   </el-drawer>

config/settings.py

@@ -97,7 +97,10 @@ class Settings(BaseSettings):
     PMS_PLAN_MIN_UPSIDE: float = 0.0       # 预期空间下限 (0.5=+50%), 0=不设。券商目标价
     #   口径, 噪音大 (榜首能到 +212%), 只做下限过滤, **不参与排序** —— 排序始终是 score
     PMS_PLAN_STALE_TDAYS: int = 1          # 计划日龄超此交易日数即判过期并拒用 (防上游停更)
-    PMS_PLAN_THEME_SYNC: bool = True       # 刷新时把 evidence.theme 灌进 pms_industry_map
+    PMS_PLAN_EXCLUDE_ST: bool = True       # 剔除 ST/*ST/退市 (上游明确不剔除, 这道闸归 PMS)
+    PMS_PLAN_THEME_SYNC: bool = False      # 把 theme 灌进 pms_industry_map。**默认关**:
+    #   theme 是事件驱动的传导主题, 覆盖率约三成且词表不规范, 当行业标签会让集中度约束
+    #   天天漂 —— 行业源请用 PMS_SECTOR_SOURCE=gp_stock_category
     PMS_PLAN_QUERY_EXTRA: str = ""         # 附加查询串逃生口, 如 "foo=1"。上游哪天加了新参数
     #   不用改代码即可透传; top/obs_top/theme_cap 已有显式参数, 同名键以显式参数为准
 

scripts/probe_plan_api.py

@@ -198,8 +198,8 @@ def main():
           f"  →(券商预期 + 目标价≥现价 + 主题限额 + top)→"
           f"  实收 主榜 {f['returned_main']} / 观察 {f['returned_observe']}")
     if plan["truncated"]["main"]:
-        print(f"    ! 主榜正好吃满 top={plan['requested']['top']} —— 可能还有更多, "
+        print(f"    · 主榜正好吃满 top={plan['requested']['top']} —— 要不要紧看 [6] 的"
-              f"调大 PMS_PLAN_TOP 再看")
+              f"「白名单档位取全」那一行")
     else:
         print(f"    主榜没吃满 top={plan['requested']['top']}, 说明这就是上游能给的全部"
               f" (受主题限额 {plan['theme_cap']} 与价格筛限制)")
@@ -221,7 +221,10 @@ def main():
         from app.services import market
         for r in rows:
             try:
-                prices[r["ts_code"]] = market.get_price(r["ts_code"])
+                pp = market.plan_price(r["ts_code"])
+                prices[r["ts_code"]] = ((f"{pp['price']}" if pp["source"] == "realtime"
+                                         else f"{pp['price']} (昨收)") if pp["price"]
+                                        else "**无价(会被剔除)**")
             except Exception as e:
                 prices[r["ts_code"]] = f"ERR {type(e).__name__}"
     for r in rows:
@@ -229,8 +232,7 @@ def main():
                 f"{_fmt(r['score'], 2):>8} {(r['tier'] or '观察'):<6} "
                 f"{(r['theme'] or '-'):<10} {_fmt(r['heat']):>7} {_pct(r['upside']):>9}")
         if args.with_price:
-            p = prices.get(r["ts_code"])
+            line += f"   {prices.get(r['ts_code']) or '-'}"
-            line += f"   {p if p else '**无价(会被剔除)**'}"
         print(line)
 
     themes = {}
@@ -246,12 +248,21 @@ def main():
         min_score=(ps.get_float("PMS_PLAN_MIN_SCORE", 0.0) or None),
         min_sources=ps.get_int("PMS_PLAN_MIN_SOURCES", 0),
         min_upside=(ps.get_float("PMS_PLAN_MIN_UPSIDE", 0.0) or None),
-        theme_cap=ps.get_int("PMS_PLAN_THEME_CAP_LOCAL", 5))
+        theme_cap=ps.get_int("PMS_PLAN_THEME_CAP_LOCAL", 5),
+        exclude_st=ps.get_bool("PMS_PLAN_EXCLUDE_ST", True))
     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)} "
-          f"PMS侧主题限额={ps.get_int('PMS_PLAN_THEME_CAP_LOCAL', 5)})")
+          f"PMS侧主题限额={ps.get_int('PMS_PLAN_THEME_CAP_LOCAL', 5)} "
+          f"剔ST={ps.get_bool('PMS_PLAN_EXCLUDE_ST', True)})")
+    tc = sel.get("tier_complete")
+    tcs = ("是 —— 已看到白名单之外的档位, 后面再多也不影响候选池" if tc is True else
+           ("**否** —— 返回的全是白名单档位, 说明被 top 卡住了, 调大 PMS_PLAN_TOP" if tc is False
+            else "不适用 (没设档位白名单)"))
+    print(f"    白名单档位取全: {tcs}")
     print(f"    排序池 {sel['considered']} → 合格 {sel['eligible']} → 取 {len(sel['items'])} 只")
+    if sel.get("st_unknown"):
+        print(f"    ! {len(sel['st_unknown'])} 只没有名称, ST 判不了已放行: {sel['st_unknown'][:8]}")
     print(f"    丢弃明细 {sel['dropped']}   (此处未扣持仓/黑名单, 下命令时还会再扣)")
     st = {}
     for x in sel["items"]:

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  上游选股计划: 解析/新鲜度/候选筛选/取数守卫      (30 例)
+  test_batch7_units.py  上游选股计划: 解析/新鲜度/候选筛选/取数守卫      (32 例)
-  test_wiring.py        装配自检: 服务层→核心→落表 全链路 (内存桩)      (51 例)
+  test_wiring.py        装配自检: 服务层→核心→落表 全链路 (内存桩)      (52 例)
-                                                                  共 235 例
+                                                                  共 238 例
 任一子集失败即整体失败 (退出码 1)。
 """
 import os

scripts/test_batch7_units.py

@@ -323,6 +323,45 @@ def _():
     assert [x["ts_code"] for x in r["items"]] == ["600001.SH", "600002.SH"], r["items"]
 
 
+@case("筛选·ST 剔除 (上游明确不剔, 这道闸归 PMS): ST/*ST/退市 都要挡, 名字缺失不误杀")
+def _():
+    assert pf.is_st("ST喜临门") and pf.is_st("*ST辉丰") and pf.is_st("S*ST前锋")
+    assert pf.is_st("SST华塑") and pf.is_st("*st 辉丰")            # 大小写与空格都要吃掉
+    assert pf.is_st("退市海润") and pf.is_st("海润退")              # 退市整理期两种写法
+    assert not pf.is_st("江淮汽车") and not pf.is_st("") and not pf.is_st(None)
+    # 「圣农发展」「深赛格」这类 S 开头的中文名不许误杀 (A 股名称都是中文, 拉丁 ST 前缀安全)
+    assert not pf.is_st("圣农发展") and not pf.is_st("深赛格") and not pf.is_st("三安光电")
+    d = {"date": "2026-07-29", "main": [
+        _m(1, "SH600001", "ST喜临门", 240.0, "整机", 0.1, 1.0),
+        _m(2, "SH600002", "*ST辉丰", 239.0, "整机", 0.1, 1.0),
+        _m(3, "SH600003", "江淮汽车", 238.0, "整车", 0.1, 1.0),
+        {"rank": 4, "code": "SH600004", "score": 237.0, "tier": "强传导",
+         "evidence": {"theme": "整车"}}]}                      # 没有 name
+    p = pf.parse_plan(d)
+    r = pf.select_candidates(p, top_n=10, exclude_st=True)
+    assert [x["ts_code"] for x in r["items"]] == ["600003.SH", "600004.SH"], r["items"]
+    assert r["dropped"]["st"] == 2
+    assert r["st_unknown"] == ["600004.SH"], r["st_unknown"]   # 判不了要报出来, 不静默
+    assert len(pf.select_candidates(p, top_n=10, exclude_st=False)["items"]) == 4
+
+
+@case("筛选·白名单档位取全判据: 看见白名单之外的档位才算齐 (吃满 top 不等于漏了)")
+def _():
+    strong = [_m(i, f"SH60{i:04d}", f"强{i}", 240 - i, "整车", 0.1, 1.0) for i in range(1, 4)]
+    weak = [_m(9, "SH609999", "弱", 220.0, "整车", 0.1, 1.0, tier="弱传导")]
+    # 只有强传导 → 判不出后面还有没有, tier_complete=False (被 top 卡住了)
+    only_strong = pf.select_candidates(pf.parse_plan({"date": "2026-07-29", "main": strong}),
+                                       tiers=["强传导"], top_n=10)
+    assert only_strong["tier_complete"] is False
+    # 出现弱传导 → 强传导已经全在前面了
+    both = pf.select_candidates(pf.parse_plan({"date": "2026-07-29", "main": strong + weak}),
+                                tiers=["强传导"], top_n=10)
+    assert both["tier_complete"] is True
+    # 没设白名单时无从判断
+    assert pf.select_candidates(pf.parse_plan({"date": "2026-07-29", "main": strong}),
+                                top_n=10)["tier_complete"] is None
+
+
 @case("筛选·候选级主题限额在 top_n 截断之前生效 (否则前 N 名被单一主题垄断)")
 def _():
     # 造一个"宽池子": 储能 6 只分最高, 传感器 3 只, 整车 2 只

scripts/test_wiring.py

@@ -419,7 +419,7 @@ class FakeQmtRepo:
         return None
 
 
-def install_fakes(prices=None, positions=None, params=None, high5=None):
+def install_fakes(prices=None, positions=None, params=None, high5=None, prev_close=None):
     """把内存桩装到各模块上, 返回 FakeRepo 实例 (ws 通道桩挂在 .qmt 上)。"""
     from app.repo import downstream_repo, pms_repo, qmt_repo
     from app.services import industry, market, param_store, portfolio
@@ -449,6 +449,13 @@ def install_fakes(prices=None, positions=None, params=None, high5=None):
     market.get_refs = lambda c, **kw: {"support": None, "pressure": None, "stop": None,
                                        "source": "none"}
     market.get_ma5 = lambda c: (prices or {}).get(c)
+    # plan_price 是规划期定价 (实时价 → 昨收兜底)。桩不接库, 一律按"有实时价"处理;
+    # 昨收那条路单独给 prev_close 参数的用例覆盖。
+    market.get_last_close = lambda c: (prev_close or {}).get(c)
+    market.plan_price = lambda c: (
+        {"price": (prices or {}).get(c), "source": "realtime"} if (prices or {}).get(c)
+        else ({"price": (prev_close or {}).get(c), "source": "prev_close"}
+              if (prev_close or {}).get(c) else {"price": None, "source": "none"}))
     market.get_high5 = lambda c: (high5 or prices or {}).get(c)
     market.day_snapshot = lambda c: ({} if not (prices or {}).get(c) else {
         "price": prices[c], "vwap": prices[c], "open": prices[c], "high": prices[c] * 1.02,
@@ -1710,6 +1717,31 @@ def _():
     assert by["600000.SH"]["score"] > plan_max, (by["600000.SH"]["score"], plan_max)
 
 
+@case("候选池·盘前无实时价回落昨收 (db13 存的是当日分钟线, 盘前那张 key 不存在)")
+def _():
+    from app.services import command_service as csvc, plan_feed
+    install_fakes(prices={}, prev_close={"600418.SH": 12.5, "300952.SZ": 30.0},
+                  params={"PMS_CANDIDATE_SOURCE": "plan_api"})
+    restore_sp = _stub_sp(csvc, {})
+    orig = plan_feed.candidates
+    try:
+        plan_feed.candidates = lambda **kw: {
+            "date": "2026-07-30", "considered": 3, "eligible": 3, "dropped": {},
+            "items": [{"ts_code": "600418.SH", "score": 242.24, "theme": "整车"},
+                      {"ts_code": "300952.SZ", "score": 241.91, "theme": "传感器"},
+                      {"ts_code": "688717.SH", "score": 242.05, "theme": "储能"}]}
+        pool = csvc._candidates({"held": []})
+    finally:
+        plan_feed.candidates = orig
+        restore_sp()
+    by = {c["ts_code"]: c for c in pool}
+    # 有昨收的进池并标明价格来源; 实时价与昨收都没有的才剔除
+    assert set(by) == {"600418.SH", "300952.SZ"}, by
+    assert by["600418.SH"]["price"] == 12.5
+    assert by["600418.SH"]["price_source"] == "prev_close", by["600418.SH"]
+    assert "688717.SH" not in by
+
+
 @case("候选池·上游计划不可用: 池为空, 绝不静默回退旧 trading_buy_plan 表")
 def _():
     from app.repo import downstream_repo