PMS 接通选股系统判决：计划解析增读判决五键、按判决分流新建仓、研判键放行与必答题、人工裁决理由必填、信号来源区分、会话密钥补进密钥名单

按判决分流（PMS_PLAN_ROUTE_BY_VERDICT，默认开）：候选走自动流程，关注强制人工确认，仅展示不出提议；上游无判决或开关关闭时与旧行为一致。自动执行开关（PMS_OPEN_AUTO_EXEC_ON_VERDICT，默认关）留给一致性检查通过后再开。
测试：新增第二十批 15 例，第七批 32 到 35 例，全套 626 例 ALL SUITES PASS；例数表按实跑校正（原记 591 已过期）。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

app/core/action_engine.py

@@ -42,6 +42,32 @@ A_FILL, A_ADD, A_DCA, A_TRIM = "FILL", "ADD", "DCA", "TRIM"
 A_OPEN = "OPEN"                # 新建仓 (与 planner.A_OPEN、executor.BUY_ACTIONS 同名同义)
 BUY, SELL = "buy", "sell"
 
+# 上游选股系统给每张候选卡的判决 (2026-09-02 起随计划接口每行下发; 旧版计划没有这个字段)。
+#   候选   门槛全过、无硬风险、确认线在 —— 走现有的自动流程 (规则闸 → 研判闸 → 档位分流)
+#   关注   门槛全过、无硬风险、但确认线缺失或陈旧 —— 这是上游说「无法判断」, 强制交人裁决
+#   仅展示 只差覆盖或潜在吸筹 —— 不出提议, 只在候选处置里记一句原因
+# 三个值是中文原文, 与上游接口逐字对应, 这里不做任何映射。
+VERDICT_CANDIDATE, VERDICT_WATCH, VERDICT_DISPLAY = "候选", "关注", "仅展示"
+WHY_WATCH_CONFIRM = "选股系统判为关注（无法判断），交人裁决"
+WHY_DISPLAY_ONLY = "选股系统判为仅展示"
+
+
+def verdict_confirm_why(verdict):
+    """按判决要不要强制人工确认: 要则回原因, 不要回 None。
+
+    候选与没有判决 (旧版计划) 都回 None —— 走档位。关注回固定文案。仅展示不该走到这里
+    (scan_open 在更前面就跳过了), 真走到也按关注处理。判决值认不出来 (上游改了词表)
+    一律强制交人: 宁可多问一次人, 也不拿一个看不懂的判决自动建仓。
+    """
+    v = (str(verdict).strip() if verdict is not None else "")
+    if not v or v == VERDICT_CANDIDATE:
+        return None
+    if v == VERDICT_WATCH:
+        return WHY_WATCH_CONFIRM
+    if v == VERDICT_DISPLAY:
+        return WHY_DISPLAY_ONLY + "，交人裁决"
+    return f"选股系统判决值无法识别（{v}），交人裁决"
+
 # **有资格**送研判闸的动作 (设计 §7: 自主提议的补足/加仓/补仓/调仓; 2026-08-06 加入新建仓)。
 # 注意只是"有资格"——真正送不送由页面参数 PMS_JUDGE_ACTIONS 决定 (judge.request 第一行就按它
 # 过滤)。两道门分开是有用的: 决策系统那侧的 OPEN 判据万一要退回去, 页面上摘掉一个词就行,
@@ -354,6 +380,11 @@ def eval_open(c: dict, params: dict, caps: dict, room_amt: float):
         # 它可以拿自己的结论去对照; 没有信号时这两项是 None, 不误导。
         "intraday_buy_signal": bool(c.get("sig_buy")),
         "intraday_buy_reason": (c.get("sig_buy") or {}).get("reason"),
+        # 上游候选卡的判决与理由 (2026-09-02 起下发; 旧版计划这五项都是 None)。同样是定性
+        # 材料: 送研判时一并带过去 (judge.OPEN_JUDGE_KEYS), 让决策系统逐条核对「这几条理由
+        # 到今天还成不成立」; 进评审账本与提议卡, 让人看得见系统为什么挑了它。
+        "verdict": c.get("verdict"), "reasons": c.get("reasons"),
+        "missing": c.get("missing"), "risk": c.get("risk"), "card_rank": c.get("card_rank"),
         # ---- 仓位口径: 只进评审账本做判分锚。judge.py 送研判时会把这几项过滤掉,
         #      理由见那边的 OPEN_JUDGE_KEYS —— 决策系统本来就不管仓位, 别送过去带偏它。
         "target_pct": target_pct, "target_amount": round(full_amt, 2),
@@ -394,11 +425,22 @@ def scan_open(*, candidates: list, params: dict, caps: dict, room_amt: float,
     room_amt: 还有多少钱可投, 已取「总仓上限×总规模 − 组合市值」与「真实可用资金」的小者
     skip:     不再评估的 (代码, 动作) —— 字典时值是跳过原因 (在途提议 / 在途指令 /
               今日被规则闸拒过 / 今日被研判闸驳回过, 这四种处境完全不同, 必须分得开)
+
+    按上游判决分流 (2026-09-03, params["open_route_by_verdict"] 为真时生效):
+      仅展示 → 不产出候选, 写一条跳过原因 (正常路径上 plan_feed 已在候选阶段剔掉, 这里是
+               兜底 —— 直接调本函数的人也得到同一口径);
+      关注   → 照常产出候选, 但打上 needs_user_confirm 并带上原因 (confirm_why), 由
+               proposal_service._route_one 强制入人工队列;
+      候选 / 没有判决 → 一个字不改。
+    落点选在这里而不是 eval_open: 判决决定的是「这条候选往哪条路走」, 与「建不建、建多少」
+    是两个问题; 放在产出候选的循环里, 仅展示的行连名额与金额都不扣, 而且开关关着时
+    这段代码一行都不执行 —— 旧行为逐字保留。
     """
     skip = skip or {}
     out, skipped = [], []
     slots = int(slots or 0)
     left = _f(room_amt)
+    route = bool(params.get("open_route_by_verdict"))
     if slots <= 0:
         return {"candidates": [], "skipped": [
             {"ts_code": "*", "action": A_OPEN,
@@ -427,6 +469,12 @@ def scan_open(*, candidates: list, params: dict, caps: dict, room_amt: float,
         code = c.get("ts_code")
         if not code:
             continue
+        if route and c.get("verdict") == VERDICT_DISPLAY:
+            # 仅展示的行不占名额、不占金额, 也不进后面任何一道闸; 放在名额判断之前,
+            # 是为了让它的跳过原因永远是「仅展示」而不是碰巧的「名额用完」。
+            skipped.append({"ts_code": code, "action": A_OPEN,
+                            "why": WHY_DISPLAY_ONLY + "，不出提议"})
+            continue
         if slots <= 0:
             skipped.append({"ts_code": code, "action": A_OPEN,
                             "why": "本轮新仓名额已用完 (下一跳按最新持仓数重算)"})
@@ -448,6 +496,13 @@ def scan_open(*, candidates: list, params: dict, caps: dict, room_amt: float,
         if not cand:
             skipped.append({"ts_code": code, "action": A_OPEN, "why": why or "未产出候选"})
             continue
+        if route:
+            confirm_why = verdict_confirm_why(c.get("verdict"))
+            if confirm_why:
+                # 关注 (或认不出的判决): 候选照常产出、照常占名额、照常过规则闸与研判闸,
+                # 只是最后一步不许自动落指令 —— 与深档补仓强制确认走的是同一个字段。
+                cand["needs_user_confirm"] = True
+                cand["confirm_why"] = confirm_why
         out.append(cand)
         used = _f(cand.get("target_amount"))
         left -= used

app/core/signal_rules.py

@@ -46,6 +46,28 @@ ACT_EXIT, ACT_PROPOSE, ACT_RECORD, ACT_IGNORE = "EXIT", "PROPOSE", "RECORD", "IG
 # 买入信号留痕: 不产生任何买入动作, 但**持没持仓都要写账本** (与 ACT_RECORD 的差别就在这)
 ACT_NOTE_BUY = "NOTE_BUY"
 
+# 信号来源 (2026-09-03 起区分)。db2 那条盘中流**两家都写**: 择时决策系统的大脑广播
+# (producer_id 形如 bionic_brain_intraday_v2.0) 与盘中择时程序的入场信号 (producer_id 形如
+# intraday_timing_v0.1.0), 格式相近而语义不同 —— 前者是知情二审的结论, 后者是量化程序的
+# 触发。此前 PMS 一律记成「决策系统盘中判该股转多」, 于是新建仓的插队排序把择时程序的
+# 触发也当成了决策系统的结论。现在按 producer_id 前缀分开: 只有以 bionic 开头的才算
+# 择时决策系统; 缺失一律记作 unknown, 不猜。
+PRODUCER_UNKNOWN = "unknown"
+BIONIC_PRODUCER_PREFIX = "bionic"
+# 留痕 reason 的固定开头。动作引擎那条路 (proposal_service._buy_signals_today) 靠这个开头
+# 认出「哪几条是择时决策系统判的」—— 账本查询只回 reason 不回硬数字, 用开头判最省一次改表。
+BUY_NOTE_PREFIX_BIONIC = "决策系统盘中判该股转多"
+BUY_NOTE_PREFIX_OTHER = "盘中择时程序买入信号"
+
+
+def is_bionic_producer(producer_id) -> bool:
+    return str(producer_id or "").strip().lower().startswith(BIONIC_PRODUCER_PREFIX)
+
+
+def is_bionic_buy_note(reason) -> bool:
+    """账本里一条 SIGNAL_BUY 留痕是不是择时决策系统判的 (看 reason 开头)。"""
+    return str(reason or "").startswith(BUY_NOTE_PREFIX_BIONIC)
+
 
 def _num(v, d=0.0):
     try:
@@ -92,7 +114,9 @@ def parse_intraday(fields: dict, *, msg_id: str = None) -> dict:
             "minute_qrs": _num(scores.get("minute_qrs")),
             "suggested_price": _num(f.get("suggested_price")) or None,
             "reason": f.get("reason") or f.get("llm_reason") or "",
-            "dominant_signal": f.get("dominant_signal") or ""}
+            "dominant_signal": f.get("dominant_signal") or "",
+            # 发送方标识 (2026-09-03 起保留): 缺就是 unknown, 不猜是谁发的
+            "producer_id": str(f.get("producer_id") or "").strip() or PRODUCER_UNKNOWN}
 
 
 def parse_risk_sell(fields: dict, *, msg_id: str = None) -> dict:
@@ -136,10 +160,17 @@ def digest(signal: dict, position: dict, params: dict) -> dict:
     if act == "BUY":
         # 买入信号**仍然不产生买入动作** —— 买什么买多少归动作引擎。
         # 但无论持没持仓都要留痕: 未持仓的票正是新建仓关心的那一批, 从前它们连账本都没有。
+        # 留痕文案按来源分写 (2026-09-03): 择时决策系统的保持原句「决策系统盘中判该股转多」;
+        # 其他发送方 (盘中择时程序、未知) 写「盘中择时程序买入信号（来源 xxx）」——
+        # 动作引擎的插队排序只认前者, 所以这个开头就是两类的分界线, 别改。
         held_now = int((position or {}).get("total_qty") or 0)
         hard["held"] = held_now
+        producer = str(signal.get("producer_id") or "").strip() or PRODUCER_UNKNOWN
+        hard["producer_id"] = producer
+        head = (BUY_NOTE_PREFIX_BIONIC if is_bionic_producer(producer)
+                else f"{BUY_NOTE_PREFIX_OTHER}（来源 {producer}）")
         return _r(ACT_NOTE_BUY, 0,
-                  f"决策系统盘中判该股转多 (置信度 {conf:.0%}"
+                  f"{head} (置信度 {conf:.0%}"
                   + (f", 建议价 {signal.get('suggested_price')}" if signal.get("suggested_price")
                      else "")
                   + f"){'; 该股当前有持仓' if held_now > 0 else '; 该股当前无持仓'} —— "

app/services/judge.py

@@ -39,7 +39,15 @@ OPEN_JUDGE_KEYS = ("price", "score", "theme", "tier", "upside", "heat",
                    "plan_rank", "plan_bucket", "plan_src", "sector",
                    # 决策系统今天盘中判没判过这只票转多。这是它自己的结论, 不是仓位数字,
                    # 送回去等于当面提醒它「你今天判过」, 让它拿自己的结论对照一次。
-                   "intraday_buy_signal", "intraday_buy_reason")
+                   "intraday_buy_signal", "intraday_buy_reason",
+                   # 上游候选卡的判决与理由 (2026-09-02 起随计划下发)。决策系统那侧是逐键
+                   # 渲染进提示词的, 所以这五项放行之后不用改它的提示词就能看见; 配合下面
+                   # must_answer 里那条「每条理由到今天是否仍成立」, 研判就有了可逐条核对的靶子。
+                   "verdict", "reasons", "missing", "risk", "card_rank")
+
+# 新建仓送研判时的必答题 (2026-09-03)。候选卡的理由随硬数字送过去了, 研判要逐条回答
+# 「这条理由到今天还成不成立」—— 这是把上游的静态判决与决策系统的当日观点接起来的那一句。
+OPEN_MUST_ANSWER = "上游候选卡的每条理由到今天是否仍成立"
 
 
 def _judge_hard_numbers(action: str, hard: dict) -> dict:
@@ -112,8 +120,18 @@ def status() -> dict:
     return {"available": True, "base": base_url(), "actions": sorted(judged_actions())}
 
 
+def _conf_or_none(v):
+    """应答里的 confidence (契约 §2.3: 0~100) 转成数字; 缺失或不是数字一律 None。"""
+    try:
+        if v is None or (isinstance(v, str) and not v.strip()):
+            return None
+        return float(v)
+    except (TypeError, ValueError):
+        return None
+
+
 def _map_verdict(data: dict) -> dict:
-    """把决策系统应答的 verdict 归一到 PASS / REJECT / UNAVAILABLE, 并带出原因。
+    """把决策系统应答的 verdict 归一到 PASS / REJECT / UNAVAILABLE, 并带出原因与置信度。
 
     契约见 BIONIC_PMS_INTERFACE §2.3: verdict 合法取值是 PASS / REJECT / UNAVAILABLE,
     UNAVAILABLE 还会带一句原因。UNAVAILABLE 是设计内的正常降级 (该股无昨夜结论、裁决
@@ -122,21 +140,38 @@ def _map_verdict(data: dict) -> dict:
     UNAVAILABLE 会落进最后那道兜底, 页面显示成「研判答复无法识别: UNAVAILABLE」——
     读起来像协议 bug, 其实只是决策系统说「这只票我没有昨夜结论」。
     只有既不是三种合法值、又解析不出的乱码, 才真算「无法识别」。
+
+    confidence (2026-09-03 起保留): 应答里的置信度原样带出 (0~100 制, 缺就是 None), 随
+    提议的硬数字进队列与账本, 让人裁决时看得见决策系统有多确定。
     """
     data = data or {}
     verdict = str(data.get("verdict") or data.get("decision") or "").upper()
     reason = data.get("reason") or data.get("rationale") or ""
+    conf = _conf_or_none(data.get("confidence"))
     if verdict in ("PASS", "APPROVE", "APPROVED", "ALLOW", "通过"):
-        return {"verdict": PASS, "reason": reason, "degraded": False, "raw": data}
+        return {"verdict": PASS, "reason": reason, "degraded": False, "raw": data,
+                "confidence": conf}
     if verdict in ("REJECT", "DENY", "DENIED", "BLOCK", "驳回"):
-        return {"verdict": REJECT, "reason": reason, "degraded": False, "raw": data}
+        return {"verdict": REJECT, "reason": reason, "degraded": False, "raw": data,
+                "confidence": conf}
     if verdict in ("UNAVAILABLE", "PMS_UNAVAILABLE", "NA", "N/A", "不可用"):
         logger.warning("[研判闸] 决策系统回不可用, 降级人工确认: %s", reason or "(无原因)")
         return {"verdict": UNAVAILABLE, "reason": reason or "决策系统研判不可用",
-                "degraded": True, "raw": data}
+                "degraded": True, "raw": data, "confidence": conf}
     logger.error("[研判闸] 答复无法识别 (%s), 按不可用降级", verdict or data)
     return {"verdict": UNAVAILABLE, "reason": f"研判答复无法识别: {verdict or data}",
-            "degraded": True, "raw": data}
+            "degraded": True, "raw": data, "confidence": conf}
+
+
+def must_answer_for(action: str) -> list:
+    """按动作给研判的必答题。补仓类问「杀逻辑还是杀情绪」(设计 §7 原文); 新建仓问
+    「候选卡的每条理由到今天是否仍成立」(2026-09-03 加); 其余不问。"""
+    a = str(action or "").upper()
+    if a == "DCA":
+        return ["下跌是杀逻辑还是杀情绪"]
+    if a == "OPEN":
+        return [OPEN_MUST_ANSWER]
+    return []
 
 
 def request(candidate: dict, context: dict = None, *, timeout: int = None) -> dict:
@@ -144,11 +179,12 @@ def request(candidate: dict, context: dict = None, *, timeout: int = None) -> di
     action = str(candidate.get("action") or "").upper()
     if action not in judged_actions():
         return {"verdict": PASS, "reason": f"{action} 不在研判范围, 规则闸通过即可",
-                "degraded": False, "raw": None}
+                "degraded": False, "raw": None, "confidence": None}
     if not available():
         st = status()
         logger.error("[研判闸] 不可用, 降级人工确认: %s", st["reason"])
-        return {"verdict": UNAVAILABLE, "reason": st["reason"], "degraded": True, "raw": None}
+        return {"verdict": UNAVAILABLE, "reason": st["reason"], "degraded": True, "raw": None,
+                "confidence": None}
 
     payload = {
         "direction": "PMS_JUDGE", "action": action, "ts_code": candidate.get("ts_code"),
@@ -157,9 +193,10 @@ def request(candidate: dict, context: dict = None, *, timeout: int = None) -> di
         "hard_numbers": _judge_hard_numbers(action, candidate.get("hard_numbers")),
         "context": context or {},
         # 设计要求: 补仓类研判必须回答「下跌是杀逻辑还是杀情绪」。
-        # 新建仓的必答 (「把这只票挑出来的驱动到今天还成不成立」) **只写在决策系统的判据里**,
+        # 新建仓 (2026-09-03 起) 也带一条: 「上游候选卡的每条理由到今天是否仍成立」。
-        # 这边不传 —— 「建仓该问什么」属于仲裁哲学, 是它的知识, PMS 不需要懂。
+        # 早先这条只写在决策系统的判据里、这边不传, 理由是「建仓该问什么」属于它的知识;
-        "must_answer": (["下跌是杀逻辑还是杀情绪"] if action == "DCA" else []),
+        # 现在候选卡的理由随硬数字送了过去, 问题就有了具体的靶子, 由 PMS 点名要它逐条回答。
+        "must_answer": must_answer_for(action),
     }
     to = int(timeout or param_store.get_int("PMS_JUDGE_TIMEOUT", 90))
     url = base_url() + (param_store.get("PMS_JUDGE_PATH", "/api/intraday/pms_judge") or "")
@@ -172,6 +209,6 @@ def request(candidate: dict, context: dict = None, *, timeout: int = None) -> di
     except Exception as e:
         logger.error("[研判闸] 请求失败, 降级人工确认: %s: %s", type(e).__name__, e)
         return {"verdict": UNAVAILABLE, "reason": f"研判请求失败: {type(e).__name__}: {e}",
-                "degraded": True, "raw": None}
+                "degraded": True, "raw": None, "confidence": None}
 
     return _map_verdict(data)

app/services/param_store.py

@@ -33,7 +33,10 @@ INFRA_PREFIX = ("PROXY_DB", "SOURCE_DB", "DB_MYSQL", "SIGNAL_REDIS", "PMS_REDIS"
 # 协议 QMT_WS_PROTOCOL.md §10.1.1 明确要求「只走 .env, 不入库、不进 ParamStore、
 # 不写进代码」。它们同样以 PMS_ 开头, 不单列的话会被 _editable_keys() 当成普通业务参数:
 # 页面能改 (于是私钥 seed 落进 pms_runtime_param 表), snapshot() 还会把它原文显示出来。
-SECRET_KEYS = ("PMS_QMT_SIGN_SEED_HEX", "PMS_QMT_PEER_PUBKEY_B64")
+# 2026-09-03 补进会话票签名密钥 PMS_SESSION_SECRET: 它 2026-08 随登录一起加进 settings,
+# 当时漏了这份名单, 于是参数设置页把它当普通业务参数原文显示出来 (登录中间件读的是
+# settings, 功能没受影响, 但密钥不该出现在页面上)。
+SECRET_KEYS = ("PMS_QMT_SIGN_SEED_HEX", "PMS_QMT_PEER_PUBKEY_B64", "PMS_SESSION_SECRET")
 
 # 运行态开关: key -> (默认值, 类型, 说明)
 RUNTIME_EXTRA = {
@@ -153,6 +156,10 @@ DESC = {
     "PMS_OPEN_REQUIRE_WS_CASH": "拿不到 ws 资金快照时不自动新建仓 (只影响新建仓, 其余动作照旧)",
     "PMS_OPEN_SIGNAL_PRIORITY": "今天被决策系统判过盘中转多的候选票排最前 (只影响先后, 不影响资格)",
     "PMS_OPEN_REF_DRIFT_MAX": "参考位盘中被改写超此幅度 → 该票当日暂停新建仓",
+    "PMS_PLAN_ROUTE_BY_VERDICT": "按选股系统的判决分流自主新建仓: 候选走自动流程 / 关注强制人工确认 / "
+                                 "仅展示不出提议 (上游行无 verdict 或本开关关着时按档位, 与旧行为一致)",
+    "PMS_OPEN_AUTO_EXEC_ON_VERDICT": "propose_only 档位下, 判决为候选 + 决策系统研判通过 + 规则闸通过 + "
+                                     "上游风险列表为空 的新建仓提议直接落指令 (默认关; 其余仍入队等人裁决)",
     "PMS_JUDGE_TICK_BUDGET_SEC": "单轮提议扫描用于研判的时间预算 (秒), 用尽则剩下的候选留到下一跳",
     "PMS_JUDGE_ENABLED": "研判闸开关", "PMS_JUDGE_ACTIONS": "需过研判闸的动作",
     "PMS_JUDGE_TIMEOUT": "研判超时 (秒) → 降级 propose_only",

app/services/plan_feed.py

@@ -43,9 +43,19 @@
 `app.core.plan_diff` 纯逻辑算。顺带补上 §7.3 那个洞: `/plan` 是实时算的, 同一个 `date` 能
 对应多个版本而应答里没有版本戳 —— 落了快照, 同一 `plan_date` 底下有几行就是重算过几版。
 
-模块级只依赖 stdlib + `app.core.command_spec` (纯逻辑), 其余 (requests / param_store /
+**候选卡的判决与理由 (2026-09-02 起上游随每行下发, 向下兼容)**: 每行多了五个键 ——
-pms_repo / plan_diff / tradedays) 一律函数内懒加载 —— 让解析与筛选这两段纯逻辑可以零依赖
+`verdict` (候选 / 关注 / 仅展示 三个中文值)、`reasons` (理由列表, 这里只留前四条)、
-单测。
+`missing` (缺失项列表)、`risk` (风险列表)、`card_rank` (卡内序, 越小越靠前); 顶层多了
+`plan_version` 与 `generated_at`。解析层只做「原样带出来, 缺就是 None」, 不在这里做任何
+判断; 唯一的例外是 `select_candidates(route_by_verdict=True)` 会把判为「仅展示」的行在
+top_n 截断**之前**剔掉 —— 那是为了不让它们白占候选池的名额, 剔掉的代码单独列在
+`display_only` 里, 由自主建仓那条路写成跳过原因 (页面「系统在盯的候选」能看到)。
+「关注」强制人工确认与「候选」自动流程的分流不在本模块, 见 action_engine.scan_open 与
+proposal_service._route_one。旧版计划没有这些键时, 一切行为与 2026-09-02 之前一字不差。
+
+模块级只依赖 stdlib + `app.core.command_spec` / `app.core.action_engine` (都是纯逻辑),
+其余 (requests / param_store / pms_repo / plan_diff / tradedays) 一律函数内懒加载 ——
+让解析与筛选这两段纯逻辑可以零依赖单测。
 """
 from __future__ import annotations
 
@@ -53,6 +63,7 @@ import logging
 import threading
 import time
 
+from app.core.action_engine import VERDICT_DISPLAY
 from app.core.command_spec import normalize_code
 
 logger = logging.getLogger("pms.plan")
@@ -101,6 +112,31 @@ def _text_or_none(v):
     return s or None
 
 
+# 候选卡理由最多带几条进 PMS。上游一张卡的理由可以很长, 而这几条要进硬数字、进评审账本、
+# 进研判请求体、显示在提议卡上 —— 四个去处都只需要前几条最要紧的。
+REASONS_KEEP = 4
+
+
+def _list_or_none(v, limit: int = 0):
+    """字符串列表归一: 列表逐项去空白、丢空项, 单个字符串当一项; 缺失或类型不对一律 None。
+
+    空列表**保留为空列表**而不是折成 None —— 「上游说没有风险」(risk=[]) 与「上游没给这个
+    字段」(缺键) 是两件事, 自动执行开关那条「风险列表为空」判据要分得开。
+    """
+    if v is None:
+        return None
+    if isinstance(v, str):
+        s = v.strip()
+        if not s:
+            return None                       # 空串与缺键同义 (与 _text_or_none 一个口径)
+        items = [s]
+    elif isinstance(v, (list, tuple)):
+        items = [str(x).strip() for x in v if x is not None and str(x).strip()]
+    else:
+        return None
+    return items[:limit] if limit and limit > 0 else items
+
+
 def _rows(raw, bucket: str) -> list:
     """一档榜单归一。坏行跳过而不是整体抛错 —— 单条变形不该让整张计划失效。"""
     out, seen = [], set()
@@ -125,6 +161,12 @@ def _rows(raw, bucket: str) -> list:
             "n_sources": _int_or_none(ev.get("n_sources")),
             "moved_ratio": _float_or_none(ev.get("moved_ratio")),
             "bucket": bucket,
+            # 候选卡的判决与理由 (2026-09-02 起上游下发; 旧版计划没有, 缺就是 None)
+            "verdict": _text_or_none(it.get("verdict")),
+            "reasons": _list_or_none(it.get("reasons"), limit=REASONS_KEEP),
+            "missing": _list_or_none(it.get("missing")),
+            "risk": _list_or_none(it.get("risk")),
+            "card_rank": _int_or_none(it.get("card_rank")),
         })
     return out
 
@@ -156,6 +198,10 @@ def parse_plan(payload, *, requested=None) -> dict:
             themes[r["ts_code"]] = r["theme"]
     return {
         "date": date,
+        # 计划版本与生成时刻 (2026-09-02 起上游下发)。同一个 date 的计划会被重算多版,
+        # 这两项是上游自己的版本戳; 旧版应答没有, 缺就是 None (名册快照那套指纹照旧管用)。
+        "plan_version": _text_or_none(payload.get("plan_version")),
+        "generated_at": _text_or_none(payload.get("generated_at")),
         "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")),
@@ -211,9 +257,16 @@ 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,
-                      exclude_st: bool = True) -> dict:
+                      exclude_st: bool = True, route_by_verdict: bool = False) -> dict:
     """排序池 → 候选清单。
 
+    route_by_verdict: 按上游判决分流时, 判为「仅展示」的行在 top_n 截断**之前**剔掉并列进
+    `display_only` (代码列表), 让自主建仓那条路写成跳过原因。放在这一层而不是动作引擎, 是
+    因为 top_n 那一刀按纯 score 切 —— 仅展示的票分数未必低, 不先剔掉的话前 30 名可能大半
+    是仅展示, 真正的候选与关注排在第 31 名之后永远轮不上。默认 False = 与 2026-09-02
+    之前一字不差 (命令驱动建仓与探活脚本各自显式决定要不要开)。没有 verdict 的行永远
+    不受这条影响。
+
     排序: score 降序, 同分按 rank 升序 (上游 rank 已是它自己的最终次序, 拿来当稳定次序)。
     tier 白名单**只对带 tier 的行生效** —— 观察档没有 tier, 它的闸门是 include_observe。
     min_upside 相反, **对所有行生效**: upside 缺失按 0 算一起挡掉 (观察档 upside 恒为
@@ -242,8 +295,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,
-               "st": 0, "theme": 0, "dup": 0, "capped": 0}
+               "st": 0, "theme": 0, "dup": 0, "capped": 0, "display": 0}
-    passed, seen, per_theme, st_unknown = [], set(), {}, []
+    passed, seen, per_theme, st_unknown, display_only = [], 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:
@@ -256,6 +309,10 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N
         if c in black:
             dropped["black"] += 1
             continue
+        if route_by_verdict and r.get("verdict") == VERDICT_DISPLAY:
+            dropped["display"] += 1           # 上游判为仅展示: 不进池, 单独列出让人看得见
+            display_only.append(c)
+            continue
         if exclude_st:
             nm = (r.get("name") or "").strip()
             if not nm:
@@ -288,10 +345,15 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=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"}
+              "bucket": r["bucket"], "src": "plan_api",
+              # 候选卡五键原样透传 (缺就是 None), 动作引擎把它们放进硬数字
+              "verdict": r.get("verdict"), "reasons": r.get("reasons"),
+              "missing": r.get("missing"), "risk": r.get("risk"),
+              "card_rank": r.get("card_rank")}
              for r in passed[:n]]
     return {"date": plan.get("date"), "considered": len(pool), "eligible": len(passed),
             "items": items, "dropped": dropped, "st_unknown": st_unknown,
+            "display_only": display_only,
             "tier_complete": _tier_complete(pool, tiers)}
 
 
@@ -353,6 +415,7 @@ def _params() -> dict:
         "snapshot_keep": ps.get_int("PMS_PLAN_SNAPSHOT_KEEP", 200),
         "diff_rank_jump": ps.get_int("PMS_PLAN_DIFF_RANK_JUMP", 50),
         "diff_tail_guard": ps.get_float("PMS_PLAN_DIFF_TAIL_GUARD", 0.5),
+        "route_by_verdict": ps.get_bool("PMS_PLAN_ROUTE_BY_VERDICT", True),
     }
 
 
@@ -705,18 +768,28 @@ def snapshot_log(*, plan_date=None, limit: int = 50) -> dict:
 
 
 # ================================================================ 对外: 候选与状态
-def candidates(*, held=(), black=()) -> dict:
+def candidates(*, held=(), black=(), route_by_verdict: bool = False) -> dict:
-    """参数驱动的候选清单 (不含价格 —— 价格由调用方用 market.get_price 现取)。"""
+    """参数驱动的候选清单 (不含价格 —— 价格由调用方用 market.get_price 现取)。
+
+    route_by_verdict 默认 False: 命令驱动建仓 (command_service._candidates) 不传, 于是用户
+    下的升仓命令照旧看全池, 与 2026-09-02 之前一字不差 —— 判决分流只管自主提议这条路
+    (方案 3.2 的措辞是「候选自动进方案、关注强制确认、仅展示不出**提议**」), 命令是用户
+    自己的决定, 不替他过滤。自主建仓 (proposal_service._scan_open) 按 PMS_PLAN_ROUTE_BY_VERDICT
+    显式传 True。要让命令驱动也按判决过滤, 改这里的调用方, 不改默认值。
+    """
     p = _params()
     plan = get_plan()
+    route = bool(route_by_verdict)
     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"],
                             min_upside=(p["min_upside"] or None),
-                            theme_cap=p["theme_cap_local"], exclude_st=p["exclude_st"])
+                            theme_cap=p["theme_cap_local"], exclude_st=p["exclude_st"],
+                            route_by_verdict=route)
     out["age_tdays"] = plan.get("age_tdays")
     out["theme_cap"] = plan.get("theme_cap")
+    out["route_by_verdict"] = route
     return out
 
 
@@ -742,6 +815,9 @@ def status() -> dict:
         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"),
+               "plan_version": plan.get("plan_version"),
+               "generated_at": plan.get("generated_at"),
+               "route_by_verdict": p["route_by_verdict"],
                "heat_date": plan.get("heat_date"),
                "market_snapshot_days": plan.get("market_snapshot_days"),
                "counts": plan["counts"], "returned": plan["returned"],

app/services/proposal_service.py

@@ -32,6 +32,7 @@ from datetime import datetime, timedelta
 from app.core import action_engine as ae
 from app.core import command_spec as cs
 from app.core import rule_gate
+from app.core import signal_rules as sr
 from app.core import tradedays as td
 from app.repo import pms_repo
 from app.services import (command_service, executor, industry, judge, market, param_store,
@@ -180,12 +181,20 @@ def _scan_open(view, params, stock_params, skip, mkt, out) -> list:
     black = {c for c, d in (stock_params or {}).items() if d.get("black")}
     held = [x["ts_code"] for x in view["held"]]
     try:
-        sel = plan_feed.candidates(held=held, black=black)
+        # 按判决分流开着时, 上游判为「仅展示」的票在候选阶段就剔掉 (理由见
+        # plan_feed.select_candidates 的说明: 不剔的话它们会白占 top_n 的名额)。
+        sel = plan_feed.candidates(held=held, black=black,
+                                   route_by_verdict=bool(params.get("open_route_by_verdict")))
     except plan_feed.PlanFeedError as e:
         # 拿不到 ≠ 今天没票可买。显式留痕, 本轮不产新建仓候选, 已有持仓的四类照常。
         logger.error("[新建仓] 候选池取数失败, 本轮不建仓: %s", e)
         out["skipped"].append({"action": ae.A_OPEN, "why": f"候选池取不到, 本轮不建仓: {e}"})
         return []
+    # 仅展示的票逐只写跳过原因: disposition_snapshot 复用本函数, 页面「系统在盯的候选」
+    # 由此显示「选股系统判为仅展示」而不是一个说不清的空白。
+    for code in (sel.get("display_only") or []):
+        out["skipped"].append({"ts_code": code, "action": ae.A_OPEN,
+                               "why": ae.WHY_DISPLAY_ONLY + "，不出提议"})
     items = list(sel.get("items") or [])
     if not items:
         out["skipped"].append({"action": ae.A_OPEN,
@@ -470,31 +479,68 @@ def _route_one(c, view, params, stock_params, brake_active, now, dry_run, out,
     autonomy = (out.get("open_autonomy") or AUTONOMY_FULL) if is_open else out["autonomy"]
     force_queue = bool(c.get("needs_user_confirm")) or verdict.get("degraded")
     auto_exec = (side == "sell") or (autonomy == AUTONOMY_FULL and not force_queue)
+    # 自动执行开关 (2026-09-03, PMS_OPEN_AUTO_EXEC_ON_VERDICT): 新建仓档位是 propose_only 时,
+    # 「判决候选 + 决策系统研判真回了通过 + 规则闸通过 (走到这里就是通过了) + 上游风险列表
+    # 为空」四条齐, 这一条按 full 处理。强制入队的 (关注 / 深档补仓 / 研判不可用) 永远不走。
+    auto_why = None
+    if is_open and not auto_exec and not force_queue and autonomy == AUTONOMY_PROPOSE:
+        auto_why = _verdict_auto_exec_why(c, verdict)
+        if auto_why:
+            auto_exec = True
 
     if dry_run:
         (out["executed"] if auto_exec else out["queued"]).append(
             {**_brief(c), "route": "auto" if auto_exec else "queue",
-             "judge": verdict["verdict"], "dry_run": True})
+             "judge": verdict["verdict"], "dry_run": True,
+             **({"why": auto_why} if auto_why else {})})
         return
 
     if auto_exec:
         iid = _make_instruction(c, price, now)
+        reason = verdict.get("reason") or c["reason"]
+        if auto_why:
+            reason = f"判决候选自动执行: {reason}"
         pms_repo.insert_ledger(ts_code=code, action=action,
                                arbiter="judge" if c.get("judge_required") else "rule",
                                verdict="PASS", price_at=price, hard_numbers=c["hard_numbers"],
-                               ref_id=iid,
+                               ref_id=iid, reason=reason[:500])
-                               reason=(verdict.get("reason") or c["reason"])[:500])
         out["executed"].append({**_brief(c), "instruction_id": iid,
                                 "why": ("减持方向自动执行" if side == "sell"
-                                        else ("新建仓档位 full" if is_open else "档位 full"))})
+                                        else (auto_why or ("新建仓档位 full" if is_open
+                                                           else "档位 full")))})
     else:
         pid = _make_proposal(c, price, verdict)
-        why = ("深档补仓强制确认" if c.get("needs_user_confirm")
+        # 强制确认的原因: 候选自带的 (关注判决等, 见 action_engine.verdict_confirm_why)
+        # 优先, 没有就是深档补仓那条老规矩。
+        why = ((c.get("confirm_why") or "深档补仓强制确认") if c.get("needs_user_confirm")
                else ("研判不可用, 降级人工确认" if verdict.get("degraded")
                      else ("新建仓档位 propose_only" if is_open else "档位 propose_only")))
         out["queued"].append({**_brief(c), "proposal_id": pid, "why": why})
 
 
+def _verdict_auto_exec_why(c, verdict):
+    """自动执行开关的四条件判定: 全部成立回一句原因 (写进 executed 与账本), 否则 None。
+
+    四条: ① 开关 PMS_OPEN_AUTO_EXEC_ON_VERDICT 为真; ② 上游判决是「候选」; ③ 决策系统的
+    研判**真的回了通过** —— 只认带应答体 (raw) 的 PASS, 「动作不在研判范围即放行」那种
+    没有问过决策系统的 PASS 不算, 研判不可用更不算; ④ 上游风险列表为空 (None 与 [] 都算
+    空: 判为候选本身就意味着上游没标硬风险, 缺键是旧字段布局)。规则闸通过是调用方保证的
+    (未过早就 return 了)。任一条不满足就回 None, 调用方照旧入队 —— 这条路只放宽不收紧。
+    """
+    if not param_store.get_bool("PMS_OPEN_AUTO_EXEC_ON_VERDICT", False):
+        return None
+    hn = c.get("hard_numbers") or {}
+    if hn.get("verdict") != ae.VERDICT_CANDIDATE:
+        return None
+    if (not c.get("judge_required") or verdict.get("verdict") != judge.PASS
+            or verdict.get("degraded") or not isinstance(verdict.get("raw"), dict)):
+        return None
+    if hn.get("risk"):
+        return None
+    return ("判决候选自动执行 (判决候选 + 研判通过 + 规则闸通过 + 风险列表为空; "
+            "PMS_OPEN_AUTO_EXEC_ON_VERDICT=True)")
+
+
 def _judge_budget_left(deadline) -> bool:
     """这一轮还够不够再送一次研判。
 
@@ -600,7 +646,10 @@ def _make_proposal(c, price, verdict) -> str:
     ttl = param_store.get_int("PMS_PROPOSAL_TTL_HOURS", 24)
     pid = f"PRP_{td.ymd()}_{c['ts_code'].replace('.', '')}_{c['action']}"
     hn = {**(c.get("hard_numbers") or {}), "price": price, "reason": c["reason"],
-          "needs_user_confirm": c.get("needs_user_confirm", False)}
+          "needs_user_confirm": c.get("needs_user_confirm", False),
+          # 研判应答的结论与置信度 (2026-09-03): 人裁决时要看得见决策系统怎么说、有多确定。
+          # judge_reason 另有一列, 这两项进硬数字是为了随账本走 (采纳/驳回时原样落账)。
+          "judge_verdict": verdict.get("verdict"), "judge_conf": verdict.get("confidence")}
     try:
         pms_repo.insert_proposal(
             proposal_id=pid, ts_code=c["ts_code"], action=c["action"], qty=c["qty"],
@@ -635,6 +684,7 @@ def _scan_params(view: dict) -> dict:
         "build_window_tdays": param_store.get_int("PMS_BUILD_WINDOW_TDAYS", 10),
         "fill_max_loss": param_store.get_float("PMS_FILL_MAX_LOSS", -0.03),
         "open_signal_priority": param_store.get_bool("PMS_OPEN_SIGNAL_PRIORITY", True),
+        "open_route_by_verdict": param_store.get_bool("PMS_PLAN_ROUTE_BY_VERDICT", True),
     })
     return p
 
@@ -686,14 +736,23 @@ def _rejected_today_keys() -> dict:
 
 
 def _buy_signals_today() -> dict:
-    """今天的盘中转多留痕。读不到就返回空 —— 它只影响候选的先后, 不影响资格,
+    """今天**择时决策系统**判过盘中转多的留痕。读不到就返回空 —— 它只影响候选的先后,
-    所以读失败时降级成「按分数排」就够了, 不该因此让整轮新建仓停摆。"""
+    不影响资格, 所以读失败时降级成「按分数排」就够了, 不该因此让整轮新建仓停摆。
+
+    只认发送方以 bionic 开头的那些 (2026-09-03): db2 那条流盘中择时程序也在写, 它的入场
+    触发不是决策系统的结论, 不该拿来插队。区分靠留痕 reason 的固定开头 (signal_rules 里
+    两边共用同一个常量), 账本查询本身一个字不改。同一只票同一天两家都发过时, 查询按
+    最新一条取 reason —— 最新那条若是择时程序的, 决策系统更早的那条会被盖掉, 这只票就
+    不插队; 方向是少插一次队, 不是多插。
+    """
     try:
-        return pms_repo.buy_signals_today(datetime.now().replace(
+        rows = pms_repo.buy_signals_today(datetime.now().replace(
             hour=0, minute=0, second=0, microsecond=0))
     except Exception as e:
         logger.warning("[新建仓] 读当日转多留痕失败 (本轮按纯分数排序): %s", e)
         return {}
+    return {code: v for code, v in (rows or {}).items()
+            if sr.is_bionic_buy_note((v or {}).get("reason"))}
 
 
 def _judge_rejected_open_keys() -> dict:

app/services/signal_service.py

@@ -189,6 +189,9 @@ def _handle(sig, view, prm, seen, ymd, dry_run, out, strat_codes=frozenset()):
         #      直接免疫不派单, 所以 REVERSAL_BUY 是几十条的量级, 不是全市场广播。
         # verdict 用 NOTE 不用 PASS —— 这是「记下来」不是「放行」, 账本里必须分得开
         # (watch.py 对不认识的 verdict 有兜底符号, 不会显示异常)。
+        # 留痕 reason 按来源分写 (2026-09-03, 文案在 signal_rules.digest): 择时决策系统的
+        # 保持「决策系统盘中判该股转多」开头, 其他发送方写「盘中择时程序买入信号（来源 xxx）」;
+        # 硬数字里另存 producer_id。新建仓的插队排序只认前者 (proposal_service._buy_signals_today)。
         key = sr.dedup_key(sig, ymd)
         if key in seen:
             out["ignored"] += 1

app/web/static/index.html

@@ -1011,7 +1011,8 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
                     <span class="muted mono" style="font-size:11px">{{ p.ts_code }}</span>
                     <el-tag size="small" effect="dark" type="primary">{{ tx('action', p.action) }}</el-tag>
                     <span class="spacer"></span><span class="muted" style="font-size:11px">{{ p.expire_at }} 前有效</span></div>
-                  <div class="st" v-if="p.judge_reason"><span class="muted">决策系统：</span>{{ p.judge_reason }}</div>
+                  <div class="st" v-if="p.judge_reason"><span class="muted">决策系统：</span>{{ p.judge_reason }}<span class="muted" v-if="(p.hard_numbers||{}).judge_conf!=null">（置信度 {{ (p.hard_numbers||{}).judge_conf }}）</span></div>
+                  <div class="st" v-if="(p.hard_numbers||{}).verdict"><span class="muted">选股系统：</span>判为「{{ (p.hard_numbers||{}).verdict }}」<span v-if="propReasons(p).length">，{{ propReasons(p).join('；') }}</span></div>
                   <div class="st muted">{{ propWhy(p) }}</div>
                   <div class="st">约 {{ p.qty }} 股 · 参考价 {{ (p.hard_numbers||{}).price==null ? '—' : (p.hard_numbers||{}).price }}
                     · 约需 {{ money(((p.hard_numbers||{}).price||0)*(p.qty||0)) }}</div>
@@ -1653,6 +1654,12 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
             <template #default="s"><span class="mono muted">{{ JSON.stringify(s.row.hard_numbers) }}</span></template>
           </el-table-column>
           <el-table-column prop="judge_verdict" label="研判" width="100"></el-table-column>
+          <el-table-column label="选股判决" min-width="220">
+            <template #default="s">
+              <span v-if="(s.row.hard_numbers||{}).verdict">「{{ s.row.hard_numbers.verdict }}」 {{ propReasons(s.row).join('；') }}</span>
+              <span v-else class="muted">—</span>
+            </template>
+          </el-table-column>
           <el-table-column prop="expire_at" label="失效时间" width="160"></el-table-column>
           <el-table-column label="裁决" width="170">
             <template #default="s">
@@ -2316,13 +2323,20 @@ createApp({
     function scrollTo(id) { const e = document.getElementById(id); if (e) e.scrollIntoView({ behavior: 'smooth', block: 'start' }); }
     function propWhy(p) {
       const h = p.hard_numbers || {}, bits = [];
-      if (h.rank != null) bits.push('候选榜第 ' + h.rank);
+      // 键名是 plan_rank (动作引擎的硬数字里从来没有 rank 这个键, 2026-09-03 修); 更早的旧提议兜底看 rank
+      const rk = (h.plan_rank != null) ? h.plan_rank : h.rank;
+      if (rk != null) bits.push('候选榜第 ' + rk);
       if (h.theme) bits.push('主题「' + h.theme + '」' + (h.tier ? (' · ' + h.tier) : ''));
       if (h.upside != null) bits.push('预期空间 ' + (h.upside > 0 ? '+' : '') + (h.upside * 100).toFixed(0) + '%');
       if (h.score != null) bits.push('评分 ' + (typeof h.score === 'number' ? h.score.toFixed(0) : h.score));
       if (h.heat != null) bits.push('热度 ' + (typeof h.heat === 'number' ? h.heat.toFixed(2) : h.heat));
       return bits.join(' · ') || '—';
     }
+    // 选股系统候选卡的理由 (硬数字 reasons, 后端已截到四条), 提议卡上只显示前两条
+    function propReasons(p) {
+      const r = (p.hard_numbers || {}).reasons;
+      return Array.isArray(r) ? r.slice(0, 2) : [];
+    }
     function nextStep(r) {
       if (r.frozen_reason && r.frozen_reason !== 'NONE') return '这只票现在只卖不买。';
       if (r.cushion_state === 'SOLID') return '浮盈够厚，条件符合就会自动加仓。';
@@ -2933,7 +2947,21 @@ createApp({
       await loadCommands(); if (plansOf.value === cid) await loadPlans(cid);
     }
     async function decide(row, decision) {
-      const d = await call('post', '/api/proposals/' + row.proposal_id + '/decide', { decision });
+      // 采纳与驳回都必须写理由 (2026-09-03): 理由随请求体进评审账本, 复盘时「人为什么这么判」
+      // 才查得到。空理由不提交; 点取消就什么都不发。后端 /decide 早已接受 reason, 不必改。
+      const accepted = decision === 'ACCEPTED';
+      let reason = '';
+      try {
+        const r = await ElementPlus.ElMessageBox.prompt(
+          (accepted ? '采纳' : '驳回') + ' ' + nm(row.ts_code) + ' 的' + tx('action', row.action) + '提议。请写下理由（必填，会记进评审账本）',
+          accepted ? '采纳理由' : '驳回理由',
+          { inputType: 'textarea', inputPlaceholder: '例如：候选卡理由仍成立、今天量能配合 / 上游确认线缺失、先观望',
+            inputValidator: v => (v && v.trim()) ? true : '理由不能为空',
+            confirmButtonText: accepted ? '确认采纳' : '确认驳回', cancelButtonText: '取消' });
+        reason = ((r && r.value) || '').trim();
+      } catch (e) { return; }
+      if (!reason) { ElementPlus.ElMessage.warning('理由不能为空，未提交'); return; }
+      const d = await call('post', '/api/proposals/' + row.proposal_id + '/decide', { decision, reason });
       if (d.ok) {
         ElementPlus.ElMessage.success(decision === 'ACCEPTED'
           ? ('已采纳, 指令 ' + d.instruction_id) : '已驳回');
@@ -3339,7 +3367,7 @@ createApp({
       chgRaw, chgSt, chgLoading, chgTab, chg, chgHeld, chgLog, chgRevised, chgWatch,
       loadChanges,
       mode, tx, nm, fullName, nameMap, attention, heldPositions, traderParams,
-      propWhy, nextStep, scrollTo, actExitStock, actReduceStock, rowMore,
+      propWhy, propReasons, nextStep, scrollTo, actExitStock, actReduceStock, rowMore,
       pHaltBuy, pResumeBuy, pHaltAll, pResumeAll, pReduce, pIncrease, pLiquidate, pSectorExit,
       pctOf, tgtPos, posMoveValid, posMovePreview, doPosMove, heldSectors, secSel, doSectorExit,
       pmap, pval, dcaOn, sumPosition, sumDca, sumAutonomy,

config/settings.py

@@ -131,6 +131,18 @@ class Settings(BaseSettings):
     #   完全由支撑压力推出来, 区间一平移, 上午「不追」的票下午可能变成「出手」, 而无人在看。
     #   处置: 当日首答把 observed 的支撑/压力/区间锁进指令 progress.ref_lock, 之后每答比对。
 
+    # --- 按选股系统的判决分流 (2026-09-03; 上游计划接口 09-02 起每行带 verdict) ---
+    #     上游给每张候选卡一个判决: 候选 / 关注 / 仅展示。开着时: 候选走现有的自动方案
+    #     流程 (规则闸、研判闸、档位分流一个都不少); 关注强制入人工队列等人裁决 (这是上游
+    #     说「无法判断」的出口); 仅展示不出提议, 只在候选处置里记一句原因。上游行没有
+    #     verdict (旧版计划) 或本开关关着, 行为与 2026-09-02 之前逐字一致 (按档位)。
+    PMS_PLAN_ROUTE_BY_VERDICT: bool = True
+    #     自动执行开关: 档位是 propose_only 时, 对「判决为候选、决策系统研判真正回了通过、
+    #     规则闸通过、上游风险列表为空」四条同时成立的新建仓提议按 full 档位处理 —— 直接落
+    #     指令并在评审账本写明「判决候选自动执行」; 其余照旧入队。默认关, 由人在一致性
+    #     检查全部一致之后打开。这是放大自动化的一步, 关掉即回到全部入队。
+    PMS_OPEN_AUTO_EXEC_ON_VERDICT: bool = False
+
     # --- 上游选股计划接口 (候选池的事实源; 口径与待确认项见 UPSTREAM_PLAN_API.md) ---
     #     上游只回答「买什么、排第几」, 不给价格金额 —— 买多少/什么价是 PMS 自己算。
     #     接口拿不到一律显式失败 (候选池为空 + ERROR), 绝不静默回退旧表。

scripts/probe_plan_api.py

@@ -247,6 +247,9 @@ def main():
     print(f"[5] 主榜主题分布 (共 {len(themes)} 个主题): " +
           ", ".join(f"{k}×{v}" for k, v in top_themes))
 
+    # 按判决分流 (2026-09-03): 与自主建仓那条路同一口径 —— 开关开着时仅展示的票不进池。
+    # 命令驱动建仓不走这条 (plan_feed.candidates 默认不传), 所以这里印的是**自主建仓看到的池子**。
+    route = ps.get_bool("PMS_PLAN_ROUTE_BY_VERDICT", True)
     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),
@@ -254,12 +257,22 @@ def main():
         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),
-        exclude_st=ps.get_bool("PMS_PLAN_EXCLUDE_ST", True))
+        exclude_st=ps.get_bool("PMS_PLAN_EXCLUDE_ST", True),
+        route_by_verdict=route)
     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"剔ST={ps.get_bool('PMS_PLAN_EXCLUDE_ST', True)})")
+          f"剔ST={ps.get_bool('PMS_PLAN_EXCLUDE_ST', True)} "
+          f"按判决分流={route})")
+    vd = {}
+    for r in plan["main"]:
+        k = r.get("verdict") or "(无判决)"
+        vd[k] = vd.get(k, 0) + 1
+    print(f"    计划版本 {plan.get('plan_version') or '-'}, 生成于 {plan.get('generated_at') or '-'}; "
+          "主榜判决分布: " + ", ".join(f"{k}×{v}" for k, v in sorted(vd.items()))
+          + ("  → 仅展示不进池 (合格数 = 候选 + 关注 + 无判决, 再过其余各闸)" if route
+             else "  → 分流开关关着, 判决不影响候选池"))
     tc = sel.get("tier_complete")
     tcs = ("是 —— 已看到白名单之外的档位, 后面再多也不影响候选池" if tc is True else
            ("**否** —— 返回的全是白名单档位, 说明被 top 卡住了, 调大 PMS_PLAN_TOP" if tc is False
@@ -269,6 +282,9 @@ def main():
     if sel.get("st_unknown"):
         print(f"    ! {len(sel['st_unknown'])} 只没有名称, ST 判不了已放行: {sel['st_unknown'][:8]}")
     print(f"    丢弃明细 {sel['dropped']}   (此处未扣持仓/黑名单, 下命令时还会再扣)")
+    if sel.get("display_only"):
+        print(f"    仅展示已剔 {len(sel['display_only'])} 只: {sel['display_only'][:10]}"
+              + (" …" if len(sel["display_only"]) > 10 else ""))
     st = {}
     for x in sel["items"]:
         st[x.get("theme") or "(无)"] = st.get(x.get("theme") or "(无)", 0) + 1

scripts/run_tests.py

@@ -4,17 +4,18 @@
 ======================================
 运行:  docker compose run --rm pms-web python scripts/run_tests.py
 
-包含:
+包含 (例数按 2026-09-03 实跑校正; 此前写的 591 是过期数字, 当时实跑已是 608):
   test_core_units.py    仓位规划器 / 安全垫与成本账                     (14 例)
   test_batch2_units.py  命令状态机 / 方案生成器 / 回放对账纯逻辑         (35 例)
-  test_batch3_units.py  规则闸 / 择时执行器实现B 纯逻辑                 (21 例)
+  test_batch3_units.py  规则闸 / 择时执行器实现B 纯逻辑                 (25 例)
   test_batch4_units.py  动作引擎 四类自主动作触发与数量口径             (11 例)
   test_batch5_units.py  决策系统信号流解析与消化口径                    (8 例)
   test_batch6_units.py  ws 通道: 测试向量/签名/公钥/水位/弃洞/DDL/逐笔入账 (68 例)
-  test_batch7_units.py  上游选股计划: 解析/新鲜度/候选筛选/取数守卫      (32 例)
+  test_batch7_units.py  上游选股计划: 解析/新鲜度/候选筛选/取数守卫/
+                        候选卡五键透传与截断/仅展示在候选阶段剔除        (35 例)
   test_batch8_units.py  榜单变化: 名册指纹/三种语义/尾部闸/落库往返      (60 例)
   test_batch9_units.py  成本价体检 / 对账按日推进 / 行业闸 / 取整记账      (49 例)
-  test_batch10_units.py 静默失败专项: 关键路径不许丢返回值 + 八条实例      (48 例)
+  test_batch10_units.py 静默失败专项: 关键路径不许丢返回值 + 八条实例      (45 例)
   test_batch11_units.py 择时实现A: 本地检查等价/应答折算/缓存冷却/退B      (24 例)
   test_batch12_units.py 自主新建仓: 选票与滚动扣减/硬数字裁剪/漂移/预算/
                         转多信号插队与留痕/窗口末日只对命令/跳过原因分得开/
@@ -38,9 +39,31 @@
                         取整与部分卖/同轮买卖互斥/信号百分制契约/除权核销
                         缩放/日历按年降级/网格中枢与止盈闩锁/买入暂停
                         按来源分记/宏观失败路径保留留痕/双实例表名前缀   (19 例)
+  test_batch20_units.py 选股系统候选卡判决接入 (2026-09-03): 按判决分流三态/
+                        无判决回退/判决认不出交人/仅展示不占名额/自动执行
+                        开关四条件与任一缺失仍入队/信号来源区分/密钥名单
+                        含会话密钥/研判键放行与必答题/置信度进硬数字/
+                        页面静态守卫                                    (15 例)
   test_wiring.py        装配自检: 服务层→核心→落表 全链路 (内存桩)      (67 例)
-                                                                  共 591 例
+                                                                  共 626 例
 任一子集失败即整体失败 (退出码 1)。
+
+哨兵位置清单 (2026-09-03 抄录; 改了对应的东西就得来这些地方改断言, 断言不动就是漏了):
+  DDL 表数      scripts/test_batch6_units.py 用例「DDL 文件本身体检通过」(约第 580 行),
+                断言 `eq(len([... k == "table"]), 19)` (约第 593 行) —— 加表就 +1
+  调度位        scripts/test_wiring.py 用例「装配·调度表覆盖设计 §10 全部调度位」(约第 654 行),
+                beat_schedule 名字集合与 pms.* 任务名两处
+  路由清单      scripts/test_wiring.py 用例「装配·API 路由齐全 (四块页面 + 运维)」(约第 634 行)
+                的 need 列表 —— 加接口就登记
+  批次登记      本文件 SUITES 列表 + 上面的例数表 —— 加测试文件必须两处同改, 否则「缺失」
+                会被判失败 (「没跑」不许算「通过」)
+  软失败函数    scripts/test_batch10_units.py 顶部 SOFT_FAIL 名单 (约第 51 行) 与用例 [A1]
+                (约第 95 行, AST 扫 app/ 全库) —— 新加「失败不抛异常只回 ok=False」的函数就登记
+  桩签名比对    scripts/test_batch10_units.py 用例 [M1] (约第 836 行): 从源码读 pms_repo 真实
+                签名与 test_wiring.FakeRepo 逐个比 —— 改 repo 函数签名, 桩必须同改
+  密钥名单      scripts/test_batch6_units.py 用例「密钥不进 ParamStore (协议 §10.1.1)」(约第
+                642 行) 与 scripts/test_batch20_units.py 用例「密钥名单·会话密钥」——
+                新增密钥类配置要同时进 param_store.SECRET_KEYS 与这两处断言
 """
 import os
 import subprocess
@@ -54,7 +77,8 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py",
           "test_batch10_units.py", "test_batch11_units.py", "test_batch12_units.py",
           "test_batch13_units.py", "test_batch14_units.py", "test_batch15_units.py",
           "test_batch16_units.py", "test_batch17_units.py",
-          "test_batch18_units.py", "test_batch19_units.py", "test_wiring.py"]
+          "test_batch18_units.py", "test_batch19_units.py", "test_batch20_units.py",
+          "test_wiring.py"]
 
 
 def main():

scripts/test_batch20_units.py

@@ -0,0 +1,509 @@
+# -*- coding: utf-8 -*-
+"""
+第二十批模块单测 (零外部依赖, 不连库不触网)
+============================================
+运行: 在 tradingSystem 仓库根目录执行  python scripts/test_batch20_units.py
+覆盖: 选股系统候选卡判决接入 (2026-09-03, 方案《主观量化系统方案_2026-09-03》第三节 PMS 表):
+  * 按判决分流三态 (候选 / 关注 / 仅展示) 与没有判决时的回退, 判决值认不出一律交人;
+  * 仅展示在候选阶段就剔掉 (不白占 top_n 名额) 并单列;
+  * 自动执行开关的四条件 (判决候选 + 研判真通过 + 规则闸通过 + 风险列表为空), 任一缺失仍入队;
+  * 信号来源区分: producer_id 缺省 unknown、留痕文案按来源分写、新建仓插队只认择时决策系统;
+  * 密钥名单含会话密钥; 研判键放行五键、新建仓必答题、置信度保留、提议硬数字带研判结论与置信度;
+  * 页面静态守卫: 提议卡按 plan_rank 取名次、裁决理由必填并随请求体发。
+约定同前: 全过输出 "ALL PASS (n cases)" 退出码 0。
+"""
+import os
+import sys
+import traceback
+from datetime import datetime
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from app.core import action_engine as ae                                       # noqa: E402
+from app.core import signal_rules as sr                                        # noqa: E402
+from app.services import judge as jd                                           # noqa: E402
+from app.services import param_store as pstore                                 # noqa: E402
+from app.services import plan_feed as pf                                       # noqa: E402
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+RESULTS = []
+
+
+def case(name):
+    def deco(fn):
+        RESULTS.append((name, fn))
+        return fn
+    return deco
+
+
+# ================================================================ 夹具 (口径同第十二批)
+def params(**kw):
+    p = {"scale": 2_000_000.0, "stock_target_default": 0.06,
+         "batch_split": (0.5, 0.25, 0.25), "min_lot_merge": True}
+    p.update(kw)
+    return p
+
+
+def caps(**kw):
+    c = {"scale": 2_000_000.0, "portfolio_cap": 0.70, "stock_cap": 0.08,
+         "max_names": 20, "portfolio_mv": 0.0, "names_count": 0,
+         "stock_mv": 0.0, "is_new_name": False, "sector": None,
+         "sector_names": 0, "sector_mv": 0.0,
+         "sector_names_map": {}, "sector_mv_map": {},
+         "sector_max_names": 4, "sector_max_ratio": 0.40,
+         "cash_reserve": 0.0, "sector_source_ready": True,
+         "cash_avail": None, "cash_source": "estimate"}
+    c.update(kw)
+    return c
+
+
+def cand(code, price=10.0, score=100.0, sector=None, rank=1, **kw):
+    """候选池一行 (plan_feed.select_candidates 的 items 形状 + 现价)。不传 verdict 就是旧版计划。"""
+    d = {"ts_code": code, "price": price, "score": score, "sector": sector,
+         "theme": kw.pop("theme", None), "tier": kw.pop("tier", "强传导"),
+         "upside": kw.pop("upside", 0.3), "heat": kw.pop("heat", 0.5),
+         "rank": rank, "bucket": "main", "src": "plan_api"}
+    d.update(kw)
+    return d
+
+
+FIVE = ("verdict", "reasons", "missing", "risk", "card_rank")
+
+
+def _scan(rows, *, route=True, slots=20, names_count=0, room=1_400_000, **pkw):
+    return ae.scan_open(candidates=rows, params=params(open_route_by_verdict=route, **pkw),
+                        caps=caps(names_count=names_count), room_amt=room, slots=slots)
+
+
+class _Patch:
+    """临时替换若干模块属性, 退出时原样还回去 —— 单测不连库不触网。"""
+
+    def __init__(self):
+        self.saved = []
+
+    def __call__(self, mod, name, value):
+        self.saved.append((mod, name, getattr(mod, name)))
+        setattr(mod, name, value)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *a):
+        for mod, name, old in reversed(self.saved):
+            setattr(mod, name, old)
+        return False
+
+
+def _patch_params(p: _Patch, mod, values: dict):
+    """把某模块引用的 param_store 取值函数换成一份字典。"""
+    v = values
+
+    def _get_list(k, d=None, sep=","):
+        x = v.get(k)
+        if x in (None, ""):
+            return list(d or [])
+        return [s.strip() for s in str(x).split(sep) if s.strip()]
+    p(mod, "get", lambda k, d=None: v.get(k, d))
+    p(mod, "get_int", lambda k, d=0: int(v.get(k, d)))
+    p(mod, "get_float", lambda k, d=0.0: float(v.get(k, d)))
+    p(mod, "get_bool", lambda k, d=False: bool(v.get(k, d)))
+    p(mod, "get_list", _get_list)
+
+
+# ================================================================ 判决分流 (动作引擎)
+@case("判决分流·候选: 走现有自动流程, 不强制人工确认, 硬数字带候选卡五键")
+def _():
+    row = cand("600000.SH", verdict="候选", reasons=["强传导链首环", "券商覆盖 7 家"],
+               missing=[], risk=[], card_rank=1)
+    r = _scan([row])
+    assert len(r["candidates"]) == 1 and not r["skipped"], r
+    c = r["candidates"][0]
+    assert c["needs_user_confirm"] is False and "confirm_why" not in c, c
+    assert c["judge_required"] is True, "候选照样送研判闸, 分流不绕闸"
+    hn = c["hard_numbers"]
+    assert hn["verdict"] == "候选" and hn["reasons"] == ["强传导链首环", "券商覆盖 7 家"], hn
+    assert hn["missing"] == [] and hn["risk"] == [] and hn["card_rank"] == 1, hn
+
+
+@case("判决分流·关注: 照常产出候选、照常占名额, 但强制人工确认并写明「无法判断, 交人裁决」")
+def _():
+    rows = [cand("600000.SH", score=300, verdict="关注", reasons=["强传导"],
+                 missing=["确认线缺失"], risk=[], card_rank=3),
+            cand("600001.SH", score=200, verdict="候选", reasons=[], missing=[], risk=[])]
+    r = _scan(rows, slots=1, names_count=19)
+    assert [c["ts_code"] for c in r["candidates"]] == ["600000.SH"], r["candidates"]
+    c = r["candidates"][0]
+    assert c["needs_user_confirm"] is True, c
+    assert c["confirm_why"] == ae.WHY_WATCH_CONFIRM, c["confirm_why"]
+    assert "关注" in c["confirm_why"] and "交人裁决" in c["confirm_why"], c["confirm_why"]
+    assert c["hard_numbers"]["verdict"] == "关注" and c["hard_numbers"]["missing"] == ["确认线缺失"]
+    # 关注占掉了唯一的名额, 第二只按名额用完跳过 (关注不是免费的)
+    assert any(s["ts_code"] == "600001.SH" and "名额" in s["why"] for s in r["skipped"]), r["skipped"]
+
+
+@case("判决分流·仅展示: 不产出、不占名额、不进任何闸, 跳过原因写明「选股系统判为仅展示」")
+def _():
+    rows = [cand("600000.SH", score=300, verdict="仅展示", reasons=["只差覆盖"], missing=["券商覆盖"],
+                 risk=[]),
+            cand("600001.SH", score=200, verdict="候选", reasons=[], missing=[], risk=[])]
+    r = _scan(rows, slots=1, names_count=19)
+    assert [c["ts_code"] for c in r["candidates"]] == ["600001.SH"], r["candidates"]
+    sk = {s["ts_code"]: s["why"] for s in r["skipped"]}
+    assert "仅展示" in sk["600000.SH"] and "名额" not in sk["600000.SH"], sk
+    assert sk["600000.SH"].startswith(ae.WHY_DISPLAY_ONLY), sk
+    # 名额没被仅展示占掉: 只有一个名额, 候选那只拿到了
+    assert r["candidates"][0]["needs_user_confirm"] is False
+
+
+@case("判决分流·没有判决 (旧版计划) 时开关开着与关着产出逐字一致, 五键全 None")
+def _():
+    rows = [cand("600000.SH", score=300), cand("600001.SH", score=200, rank=2)]
+    on, off = _scan(rows, route=True), _scan(rows, route=False)
+    assert on == off, (on, off)
+    assert [c["ts_code"] for c in on["candidates"]] == ["600000.SH", "600001.SH"]
+    for c in on["candidates"]:
+        assert c["needs_user_confirm"] is False and "confirm_why" not in c, c
+        for k in FIVE:
+            assert k in c["hard_numbers"] and c["hard_numbers"][k] is None, (k, c["hard_numbers"])
+    # 计划层同样: 没有 verdict 的行永远不会被 display 那一刀碰到
+    p = pf.parse_plan({"date": "2026-09-02", "main": [
+        {"rank": 1, "code": "SH600000", "name": "甲", "score": 300},
+        {"rank": 2, "code": "SH600001", "name": "乙", "score": 200}]})
+    s = pf.select_candidates(p, top_n=5, route_by_verdict=True)
+    assert len(s["items"]) == 2 and s["dropped"]["display"] == 0 and s["display_only"] == []
+
+
+@case("判决分流·开关关着时判决不起作用: 关注不强制确认、仅展示照常产出 (按档位, 即旧行为)")
+def _():
+    rows = [cand("600000.SH", score=300, verdict="关注", reasons=["a"], missing=["b"], risk=[]),
+            cand("600001.SH", score=200, rank=2, verdict="仅展示", reasons=[], missing=[], risk=[])]
+    r = _scan(rows, route=False)
+    assert [c["ts_code"] for c in r["candidates"]] == ["600000.SH", "600001.SH"], r
+    assert not r["skipped"], r["skipped"]
+    for c in r["candidates"]:
+        assert c["needs_user_confirm"] is False and "confirm_why" not in c, c
+    # 判决仍然原样进硬数字 (开关只管分流, 不管记录)
+    assert r["candidates"][0]["hard_numbers"]["verdict"] == "关注"
+    assert r["candidates"][1]["hard_numbers"]["verdict"] == "仅展示"
+
+
+@case("判决分流·判决值认不出一律交人 (上游改了词表不许被静默当成候选)")
+def _():
+    r = _scan([cand("600000.SH", verdict="推荐", reasons=[], missing=[], risk=[])])
+    c = r["candidates"][0]
+    assert c["needs_user_confirm"] is True and "无法识别" in c["confirm_why"], c
+    assert "推荐" in c["confirm_why"] and "交人裁决" in c["confirm_why"], c["confirm_why"]
+    # 三个合法值与空值的判定钉死
+    assert ae.verdict_confirm_why(None) is None and ae.verdict_confirm_why("") is None
+    assert ae.verdict_confirm_why(ae.VERDICT_CANDIDATE) is None
+    assert ae.verdict_confirm_why(ae.VERDICT_WATCH) == ae.WHY_WATCH_CONFIRM
+    assert ae.verdict_confirm_why(ae.VERDICT_DISPLAY).startswith(ae.WHY_DISPLAY_ONLY)
+    assert (ae.VERDICT_CANDIDATE, ae.VERDICT_WATCH, ae.VERDICT_DISPLAY) == ("候选", "关注", "仅展示")
+
+
+@case("候选筛选·仅展示在 top_n 截断之前剔掉并单列 (否则前 N 名被仅展示占满, 候选永远轮不上)")
+def _():
+    def row(i, code, score, verdict=None):
+        d = {"rank": i, "code": code, "name": f"票{i}", "score": score, "tier": "强传导",
+             "evidence": {"theme": "整车"}}
+        if verdict:
+            d["verdict"] = verdict
+        return d
+    p = pf.parse_plan({"date": "2026-09-02", "main": [
+        row(1, "SH600001", 300, "仅展示"), row(2, "SH600002", 299, "仅展示"),
+        row(3, "SH600003", 298, "仅展示"), row(4, "SH600004", 250, "候选"),
+        row(5, "SH600005", 240, "关注"), row(6, "SH600006", 230),      # 没有判决 (旧行)
+        row(7, "SH600007", 220, "候选")]})
+    on = pf.select_candidates(p, top_n=3, route_by_verdict=True)
+    assert [x["ts_code"] for x in on["items"]] == ["600004.SH", "600005.SH", "600006.SH"], on["items"]
+    assert on["dropped"]["display"] == 3 and on["dropped"]["capped"] == 1, on["dropped"]
+    assert on["display_only"] == ["600001.SH", "600002.SH", "600003.SH"], on["display_only"]
+    assert on["eligible"] == 4 and on["considered"] == 7
+    # 不分流: 前三名全是仅展示 —— 这正是要在候选阶段剔掉的理由
+    off = pf.select_candidates(p, top_n=3)
+    assert {x["verdict"] for x in off["items"]} == {"仅展示"}, off["items"]
+    assert off["dropped"]["display"] == 0 and off["display_only"] == []
+
+
+# ================================================================ 自动执行开关 (提议分流)
+def _route(c, *, values, judge_resp, open_autonomy="propose_only"):
+    """把一条新建仓候选送进 proposal_service._route_one, 规则闸/研判/落表全部换成桩。"""
+    from app.core import rule_gate
+    from app.repo import pms_repo
+    from app.services import judge, param_store, portfolio, proposal_service as psvc
+    got = {"ledger": [], "instructions": [], "proposals": []}
+    with _Patch() as p:
+        p(rule_gate, "check", lambda **kw: {"passed": True, "failed": [], "warnings": []})
+        p(judge, "request", lambda cand_, context=None, **kw: dict(judge_resp))
+        p(portfolio, "caps_ctx", lambda *a, **kw: {})
+        p(pms_repo, "insert_ledger", lambda **kw: got["ledger"].append(kw) or 1)
+        p(pms_repo, "insert_instruction", lambda **kw: got["instructions"].append(kw) or 1)
+        p(pms_repo, "insert_proposal", lambda **kw: got["proposals"].append(kw) or 1)
+        p(pms_repo, "list_ledger", lambda **kw: [])
+        _patch_params(p, param_store, {"PMS_EXEC_WINDOW_TDAYS": 3, "PMS_PROPOSAL_TTL_HOURS": 24,
+                                       "PMS_JUDGE_TICK_BUDGET_SEC": 150, **values})
+        out = {"autonomy": "propose_only", "open_autonomy": open_autonomy, "executed": [],
+               "queued": [], "rejected": [], "skipped": [], "errors": [], "degraded": False}
+        view = {"positions": [], "held": [], "sector_ready": False, "params": {}, "totals": {}}
+        psvc._route_one(c, view, {"_mkt": {}}, {}, False, datetime(2026, 9, 3, 10, 0), False, out)
+    return out, got
+
+
+PASS_REAL = {"verdict": "PASS", "reason": "理由仍成立", "degraded": False,
+             "raw": {"verdict": "PASS", "confidence": 72}, "confidence": 72.0}
+
+
+def _open_cand(verdict="候选", risk=None):
+    r = _scan([cand("600000.SH", verdict=verdict, reasons=["强传导"], missing=[],
+                    risk=([] if risk is None else risk), card_rank=1)])
+    assert len(r["candidates"]) == 1, r
+    return r["candidates"][0]
+
+
+@case("自动执行开关·四条件齐: propose_only 档位下判决候选直接落指令, 账本写「判决候选自动执行」")
+def _():
+    out, got = _route(_open_cand(), values={"PMS_OPEN_AUTO_EXEC_ON_VERDICT": True},
+                      judge_resp=PASS_REAL)
+    assert len(out["executed"]) == 1 and not out["queued"], out
+    assert "判决候选自动执行" in out["executed"][0]["why"], out["executed"]
+    assert len(got["instructions"]) == 1 and not got["proposals"], got
+    ins = got["instructions"][0]
+    assert ins["action"] == "OPEN" and ins["side"] == "buy" and ins["qty"] == 6000, ins
+    assert ins["progress"]["auto"] is True and ins["progress"]["is_command"] is False
+    led = got["ledger"][0]
+    assert led["arbiter"] == "judge" and led["verdict"] == "PASS", led
+    assert led["reason"].startswith("判决候选自动执行"), led["reason"]
+    assert led["hard_numbers"]["verdict"] == "候选" and led["ref_id"] == ins["instruction_id"]
+
+
+@case("自动执行开关·任一条件不满足仍入队 (开关关 / 关注 / 研判不可用 / 没问过决策系统的放行 / 有风险)")
+def _():
+    # ① 开关关着 (默认): 四条其余全满足也入队, 硬数字带研判结论与置信度
+    out, got = _route(_open_cand(), values={}, judge_resp=PASS_REAL)
+    assert not out["executed"] and len(out["queued"]) == 1, out
+    assert out["queued"][0]["why"] == "新建仓档位 propose_only", out["queued"]
+    hn = got["proposals"][0]["hard_numbers"]
+    assert hn["judge_verdict"] == "PASS" and hn["judge_conf"] == 72.0, hn
+    assert hn["verdict"] == "候选" and hn["reasons"] == ["强传导"], hn
+    on = {"PMS_OPEN_AUTO_EXEC_ON_VERDICT": True}
+    # ② 判为关注: 强制入队, 原因是关注那句 (开关开着也不许自动)
+    out, got = _route(_open_cand("关注"), values=on, judge_resp=PASS_REAL)
+    assert not out["executed"] and out["queued"][0]["why"] == ae.WHY_WATCH_CONFIRM, out
+    assert got["proposals"][0]["hard_numbers"]["needs_user_confirm"] is True
+    # ②b 关注在 full 档位下同样入队 (needs_user_confirm 压过档位)
+    out, _ = _route(_open_cand("关注"), values=on, judge_resp=PASS_REAL, open_autonomy="full")
+    assert not out["executed"] and out["queued"][0]["why"] == ae.WHY_WATCH_CONFIRM, out
+    # ③ 研判不可用: 降级入队
+    out, _ = _route(_open_cand(), values=on, judge_resp={"verdict": "UNAVAILABLE", "reason": "未接通",
+                                                         "degraded": True, "raw": None,
+                                                         "confidence": None})
+    assert not out["executed"] and "研判不可用" in out["queued"][0]["why"], out
+    # ④ PASS 但没有应答体 (动作不在研判范围那种放行) 不算研判通过
+    out, _ = _route(_open_cand(), values=on, judge_resp={"verdict": "PASS", "reason": "不在研判范围",
+                                                         "degraded": False, "raw": None,
+                                                         "confidence": None})
+    assert not out["executed"] and len(out["queued"]) == 1, out
+    # ⑤ 上游标了风险: 入队
+    out, _ = _route(_open_cand(risk=["昨夜信号陈旧"]), values=on, judge_resp=PASS_REAL)
+    assert not out["executed"] and len(out["queued"]) == 1, out
+    # ⑥ 没有判决 (旧版计划): 入队 —— 自动执行只认明确的「候选」
+    out, _ = _route(_open_cand(None), values=on, judge_resp=PASS_REAL)
+    assert not out["executed"] and len(out["queued"]) == 1, out
+    # ⑦ 研判驳回: 既不执行也不入队, 留驳回痕 (分流不改这条老规矩)
+    out, got = _route(_open_cand(), values=on, judge_resp={"verdict": "REJECT", "reason": "形态走坏",
+                                                           "degraded": False, "raw": {"verdict": "REJECT"},
+                                                           "confidence": 80.0})
+    assert not out["executed"] and not out["queued"] and out["rejected"][0]["by"] == "judge", out
+    assert got["ledger"][0]["verdict"] == "REJECT" and not got["proposals"]
+
+
+# ================================================================ 信号来源区分
+@case("信号来源·producer_id 缺省 unknown; 留痕文案按来源分写, 硬数字带发送方")
+def _():
+    base = {"ts_code": "600000.SH", "action": "BUY", "confidence": "0.9", "reason": "放量突破"}
+    s = sr.parse_intraday(dict(base))
+    assert s["producer_id"] == "unknown", s
+    d = sr.digest(s, None, {})
+    assert d["action"] == sr.ACT_NOTE_BUY and d["hard_numbers"]["producer_id"] == "unknown", d
+    assert d["reason"].startswith("盘中择时程序买入信号（来源 unknown）"), d["reason"]
+    # 择时决策系统 (大脑广播) 的写法: 以 bionic 开头 → 老文案原句
+    sb = sr.parse_intraday(dict(base, producer_id="bionic_brain_intraday_v2.0"))
+    db = sr.digest(sb, {"total_qty": 1000}, {})
+    assert db["reason"].startswith("决策系统盘中判该股转多"), db["reason"]
+    assert db["hard_numbers"]["producer_id"] == "bionic_brain_intraday_v2.0"
+    assert sr.is_bionic_buy_note(db["reason"]) and not sr.is_bionic_buy_note(d["reason"])
+    # 盘中择时程序的写法
+    st = sr.parse_intraday(dict(base, producer_id="intraday_timing_v0.1.0"))
+    dt = sr.digest(st, None, {})
+    assert dt["reason"].startswith("盘中择时程序买入信号（来源 intraday_timing_v0.1.0）"), dt["reason"]
+    assert not sr.is_bionic_producer("intraday_timing_v0.1.0") and sr.is_bionic_producer("Bionic_x")
+    # 两类文案的尾巴一个字没变: 只留痕、买不买归动作引擎、持仓状态
+    for x in (d, db, dt):
+        assert "只留痕" in x["reason"] and "买不买由动作引擎" in x["reason"], x["reason"]
+    assert "无持仓" in d["reason"] and "有持仓" in db["reason"]
+    # 卖出与 HOLD 的口径不受影响
+    assert sr.digest(sr.parse_intraday({"ts_code": "600000.SH", "action": "HOLD"}), None, {})["action"] == sr.ACT_RECORD
+
+
+@case("信号来源·新建仓插队只认择时决策系统的留痕 (盘中择时程序的触发不插队; 读失败按空)")
+def _():
+    from app.repo import pms_repo
+    from app.services import proposal_service as psvc
+    rows = {"600000.SH": {"reason": "决策系统盘中判该股转多 (置信度 90%) —— 只留痕", "at": "", "price": 10.0},
+            "600001.SH": {"reason": "盘中择时程序买入信号（来源 intraday_timing_v0.1.0） (置信度 80%)",
+                          "at": "", "price": 11.0},
+            "600002.SH": {"reason": None, "at": "", "price": 0.0}}
+    with _Patch() as p:
+        p(pms_repo, "buy_signals_today", lambda since: dict(rows))
+        got = psvc._buy_signals_today()
+    assert set(got) == {"600000.SH"}, got
+    assert got["600000.SH"]["price"] == 10.0
+
+    def _boom(since):
+        raise RuntimeError("库挂了")
+    with _Patch() as p:
+        p(pms_repo, "buy_signals_today", _boom)
+        assert psvc._buy_signals_today() == {}
+
+
+# ================================================================ 密钥名单
+@case("密钥名单·会话密钥 PMS_SESSION_SECRET 不进参数中心 (页面读不到、改不了、快照里不出现)")
+def _():
+    assert "PMS_SESSION_SECRET" in pstore.SECRET_KEYS, pstore.SECRET_KEYS
+    assert {"PMS_QMT_SIGN_SEED_HEX", "PMS_QMT_PEER_PUBKEY_B64"} <= set(pstore.SECRET_KEYS)
+    assert "PMS_SESSION_SECRET" not in pstore._editable_keys()
+    assert pstore.get("PMS_SESSION_SECRET") == ""
+    r = pstore.set_param("PMS_SESSION_SECRET", "deadbeef")
+    assert r["ok"] is False and "不可修改" in r["error"], r
+    # 两个新开关是普通业务参数: 可调、有说明、类型是布尔
+    ek = pstore._editable_keys()
+    for k in ("PMS_PLAN_ROUTE_BY_VERDICT", "PMS_OPEN_AUTO_EXEC_ON_VERDICT"):
+        assert k in ek and ek[k].annotation is bool and k in pstore.DESC, k
+    from config.settings import Settings
+    assert Settings.model_fields["PMS_PLAN_ROUTE_BY_VERDICT"].default is True
+    assert Settings.model_fields["PMS_OPEN_AUTO_EXEC_ON_VERDICT"].default is False
+
+
+# ================================================================ 研判闸
+@case("研判键放行·候选卡五键随硬数字送研判, 仓位数字照旧不送; 新建仓带必答「每条理由是否仍成立」")
+def _():
+    hard = {"price": 10.0, "score": 300.0, "verdict": "候选", "reasons": ["强传导", "券商覆盖"],
+            "missing": [], "risk": [], "card_rank": 2,
+            "target_amount": 120000.0, "names_before": 3, "room_amt_before": 1_400_000.0}
+    got = jd._judge_hard_numbers("OPEN", hard)
+    for k in FIVE:
+        assert k in got and got[k] == hard[k], (k, got)
+    for k in ("target_amount", "names_before", "room_amt_before"):
+        assert k not in got, k
+    assert set(FIVE) <= set(jd.OPEN_JUDGE_KEYS)
+    assert jd.must_answer_for("OPEN") == ["上游候选卡的每条理由到今天是否仍成立"]
+    assert jd.must_answer_for("DCA") == ["下跌是杀逻辑还是杀情绪"]
+    assert jd.must_answer_for("ADD") == [] and jd.must_answer_for("TRIM") == []
+    # 请求体真的带上了 (假 requests 抓请求)
+    import types
+    sent = []
+    fake = types.ModuleType("requests")
+
+    class _Resp:
+        status_code = 200
+
+        def raise_for_status(self):
+            pass
+
+        def json(self):
+            return {"verdict": "PASS", "reason": "都成立", "confidence": 66}
+    fake.post = lambda url, json=None, timeout=None: sent.append({"url": url, "json": json}) or _Resp()
+    prev = sys.modules.get("requests")
+    sys.modules["requests"] = fake
+    try:
+        with _Patch() as p:
+            _patch_params(p, jd.param_store, {"PMS_JUDGE_ENABLED": True,
+                                              "PMS_JUDGE_API_BASE": "http://bionic:38000",
+                                              "PMS_JUDGE_ACTIONS": "FILL,ADD,DCA,SWITCH,OPEN",
+                                              "PMS_JUDGE_TIMEOUT": 5,
+                                              "PMS_JUDGE_PATH": "/api/intraday/pms_judge"})
+            r = jd.request({"action": "OPEN", "ts_code": "600000.SH", "qty": 6000,
+                            "reason": "新建仓", "hard_numbers": hard})
+    finally:
+        if prev is None:
+            sys.modules.pop("requests", None)
+        else:
+            sys.modules["requests"] = prev
+    body = sent[0]["json"]
+    assert body["must_answer"] == ["上游候选卡的每条理由到今天是否仍成立"], body
+    assert body["hard_numbers"]["verdict"] == "候选" and body["hard_numbers"]["reasons"] == ["强传导", "券商覆盖"]
+    assert "target_amount" not in body["hard_numbers"]
+    assert r["verdict"] == jd.PASS and r["confidence"] == 66.0 and r["raw"]["verdict"] == "PASS", r
+
+
+@case("研判应答·置信度保留 (缺失/非数字为 None), 提议硬数字带 judge_verdict 与 judge_conf")
+def _():
+    r = jd._map_verdict({"verdict": "PASS", "reason": "ok", "confidence": 72})
+    assert r["verdict"] == jd.PASS and r["confidence"] == 72.0, r
+    assert jd._map_verdict({"verdict": "REJECT", "confidence": "abc"})["confidence"] is None
+    assert jd._map_verdict({"verdict": "REJECT", "confidence": "55.5"})["confidence"] == 55.5
+    u = jd._map_verdict({"verdict": "UNAVAILABLE", "reason": "无昨夜结论"})
+    assert u["degraded"] is True and u["confidence"] is None and u["reason"] == "无昨夜结论", u
+    assert jd._map_verdict({"verdict": "???"})["confidence"] is None
+    # 不在研判范围 / 未接通两条早退路径也带 confidence 键 (下游按键取)
+    with _Patch() as p:
+        _patch_params(p, jd.param_store, {"PMS_JUDGE_ENABLED": False, "PMS_JUDGE_ACTIONS": "OPEN"})
+        assert jd.request({"action": "TRIM"})["confidence"] is None
+        assert jd.request({"action": "OPEN"})["confidence"] is None
+    # _make_proposal 把结论与置信度写进硬数字
+    from app.repo import pms_repo
+    from app.services import param_store, proposal_service as psvc
+    got = []
+    c = _open_cand()
+    with _Patch() as p:
+        p(pms_repo, "insert_proposal", lambda **kw: got.append(kw) or 1)
+        _patch_params(p, param_store, {"PMS_PROPOSAL_TTL_HOURS": 24})
+        pid = psvc._make_proposal(c, 10.0, {"verdict": "PASS", "reason": "都成立", "degraded": False,
+                                            "raw": {}, "confidence": 72.0})
+    assert pid.startswith("PRP_") and pid.endswith("_600000SH_OPEN"), pid
+    hn = got[0]["hard_numbers"]
+    assert hn["judge_verdict"] == "PASS" and hn["judge_conf"] == 72.0, hn
+    assert hn["price"] == 10.0 and hn["verdict"] == "候选" and hn["plan_rank"] == 1, hn
+    assert got[0]["judge_verdict"] == "PASS" and got[0]["judge_reason"] == "都成立"
+
+
+# ================================================================ 页面静态守卫
+@case("页面守卫·提议卡按 plan_rank 取候选榜名次并显示判决与理由; 采纳/驳回都弹必填理由并随请求体发")
+def _():
+    path = os.path.join(ROOT, "app", "web", "static", "index.html")
+    with open(path, encoding="utf-8") as f:
+        html = f.read()
+    i = html.index("function propWhy(")
+    seg = html[i:i + 800]
+    assert "h.plan_rank" in seg, "提议卡的候选榜名次要读 plan_rank (硬数字里从来没有 rank 这个键)"
+    assert "function propReasons(" in html and "propReasons," in html, "理由函数要定义并注册到模板"
+    assert "判为「{{ (p.hard_numbers||{}).verdict }}」" in html, "提议卡要显示选股系统的判决"
+    j = html.index("async function decide(")
+    dseg = html[j:html.index("async function loadStrategies(")]
+    assert "ElMessageBox.prompt" in dseg and "inputValidator" in dseg, "采纳与驳回都要弹必填理由"
+    assert "{ decision, reason }" in dseg, "理由要随请求体发 (后端 /decide 已读 payload.reason)"
+    assert "if (!reason)" in dseg, "空理由不许提交"
+
+
+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()

scripts/test_batch7_units.py

@@ -106,6 +106,12 @@ def _():
     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
+    # 2026-09-02 起的候选卡五键: 这份旧样例没有, 每行都必须是 None (不是 0 / 空串 / 空列表),
+    # 顶层的版本与生成时刻同理 —— 「旧版计划」与「新版但判为空」在下游必须分得开
+    for k in ("verdict", "reasons", "missing", "risk", "card_rank"):
+        assert k in r and r[k] is None, (k, r.get(k))
+        assert p["observe"][0][k] is None, (k, p["observe"][0][k])
+    assert p["plan_version"] is None and p["generated_at"] is None
 
 
 @case("解析·themes 映射: 主榜优先于观察档, 同码不被观察档覆盖")
@@ -142,6 +148,69 @@ def _():
     # rank 缺失用出现序号补 (第 7 个元素 → 7), 免得排序键出现 None
     assert p["main"][1]["rank"] == 7, p["main"][1]["rank"]
     assert p["counts"] == {"main": None, "observe": None, "gate_covered": None}
+    # 脏行里也没有候选卡五键 → 全 None, 不许因 evidence 变形连带把这五项算成别的东西
+    for k in ("verdict", "reasons", "missing", "risk", "card_rank"):
+        assert p["main"][0][k] is None and p["main"][1][k] is None, k
+
+
+@case("解析·候选卡五键透传: 判决/理由/缺失/风险/卡内序原样带到候选, 顶层带版本与生成时刻")
+def _():
+    row = dict(_m(1, "SH600418", "江淮汽车", 242.24, "整车", 0.2249, 1.2),
+               verdict="候选", reasons=["强传导链首环", "券商覆盖 7 家", "目标价高于现价"],
+               missing=[], risk=[], card_rank=2)
+    watch = dict(_m(2, "SH688717", "艾罗能源", 242.05, "储能", 0.4, 1.1),
+                 verdict="关注", reasons=["强传导"], missing=["确认线缺失"], risk=["昨夜信号陈旧"],
+                 card_rank=5)
+    d = {"date": "2026-09-02", "plan_version": "2026-09-02.3",
+         "generated_at": "2026-09-02T06:31:07", "main": [row, watch]}
+    p = pf.parse_plan(d)
+    assert p["plan_version"] == "2026-09-02.3" and p["generated_at"] == "2026-09-02T06:31:07"
+    r = p["main"][0]
+    assert r["verdict"] == "候选" and r["card_rank"] == 2, r
+    assert r["reasons"] == ["强传导链首环", "券商覆盖 7 家", "目标价高于现价"], r["reasons"]
+    assert r["missing"] == [] and r["risk"] == [], r          # 空列表保留为空列表, 不折成 None
+    w = p["main"][1]
+    assert w["verdict"] == "关注" and w["missing"] == ["确认线缺失"] and w["risk"] == ["昨夜信号陈旧"]
+    # 筛选后的候选原样带着这五项 (动作引擎靠它们进硬数字)
+    items = {x["ts_code"]: x for x in pf.select_candidates(p, top_n=5)["items"]}
+    assert items["600418.SH"]["verdict"] == "候选" and items["600418.SH"]["card_rank"] == 2
+    assert items["600418.SH"]["reasons"] == r["reasons"] and items["600418.SH"]["risk"] == []
+    assert items["688717.SH"]["verdict"] == "关注" and items["688717.SH"]["missing"] == ["确认线缺失"]
+    # 默认不按判决分流: 关注照样进池, 也没有 display_only
+    assert set(items) == {"600418.SH", "688717.SH"} and pf.select_candidates(p)["display_only"] == []
+
+
+@case("解析·候选卡键缺失一律 None (旧版计划一个键都没有); 空串同缺失, 空列表要保留")
+def _():
+    d = {"date": "2026-09-02", "main": [
+        _m(1, "SH600418", "甲", 100.0, "整车", 0.1, 1.0),                    # 一个键都没有
+        dict(_m(2, "SH600419", "乙", 99.0, "整车", 0.1, 1.0),
+             verdict="  ", reasons=None, missing="", risk=[], card_rank=""),   # 键在但是空的
+        dict(_m(3, "SH600420", "丙", 98.0, "整车", 0.1, 1.0),
+             verdict="候选", reasons={"不是": "列表"}, missing=42, risk="单条风险", card_rank="7")]}
+    p = pf.parse_plan(d)
+    a, b, c = p["main"]
+    for k in ("verdict", "reasons", "missing", "risk", "card_rank"):
+        assert a[k] is None, (k, a[k])
+    assert b["verdict"] is None and b["reasons"] is None and b["missing"] is None, b
+    assert b["risk"] == [] and b["card_rank"] is None, b       # [] 是「上游说没有风险」, 要留
+    assert c["verdict"] == "候选" and c["reasons"] is None and c["missing"] is None, c
+    assert c["risk"] == ["单条风险"] and c["card_rank"] == 7, c   # 单个字符串当一项, 数字串可用
+
+
+@case("解析·reasons 只留前四条 (硬数字/账本/研判/提议卡都只要最要紧的几条); missing 与 risk 不截")
+def _():
+    assert pf.REASONS_KEEP == 4
+    many = ["理由一", "", "理由二", None, "  理由三  ", "理由四", "理由五", "理由六"]
+    d = {"date": "2026-09-02", "main": [
+        dict(_m(1, "SH600418", "甲", 100.0, "整车", 0.1, 1.0),
+             verdict="候选", reasons=many, missing=list(many), risk=[f"风险{i}" for i in range(6)])]}
+    r = pf.parse_plan(d)["main"][0]
+    assert r["reasons"] == ["理由一", "理由二", "理由三", "理由四"], r["reasons"]   # 空项去掉、空白去掉、只留四条
+    assert r["missing"] == ["理由一", "理由二", "理由三", "理由四", "理由五", "理由六"], r["missing"]
+    assert len(r["risk"]) == 6, r["risk"]
+    x = pf.select_candidates(pf.parse_plan(d), top_n=1)["items"][0]
+    assert x["reasons"] == r["reasons"] and len(x["risk"]) == 6
 
 
 @case("解析·数值容错: 字符串数字可用, 空串/None/非数字一律 None 而不是 0")
@@ -405,21 +474,34 @@ def _():
     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, "计划不带价格, 这里绝不能凭空造一个价出来"
+    # 候选卡五键: 键必须在 (下游按键取, 不按 in 判), 旧版计划里值是 None
+    for k in ("verdict", "reasons", "missing", "risk", "card_rank"):
+        assert k in x and x[k] is None, (k, x.get(k))
 
 
-@case("筛选·计数自洽: considered = eligible + 各项丢弃; items = eligible − capped")
+@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="弱传导")]
+        _m(500, "SH600519", "弱票", 220.0, "白酒", 0.1, 0.5, tier="弱传导"),
+        # 2026-09-02 起的判决: 仅展示的行按判决分流时在候选阶段剔掉, 单独计数
+        dict(_m(600, "SH600600", "展示票", 242.50, "整车", 0.1, 0.9), verdict="仅展示")]   # 分最高
     p = pf.parse_plan(d)
     r = pf.select_candidates(p, held=["SH600418"], black=["SZ300952"], tiers=["强传导"],
-                             min_sources=7, top_n=2, include_observe=True, theme_cap=2)
+                             min_sources=7, top_n=2, include_observe=True, theme_cap=2,
+                             route_by_verdict=True)
     dr = r["dropped"]
-    assert r["considered"] == 10, r["considered"]
+    assert r["considered"] == 11, r["considered"]
     assert r["considered"] == r["eligible"] + dr["held"] + dr["black"] + dr["tier"] \
-        + dr["score"] + dr["sources"] + dr["upside"] + dr["theme"] + dr["dup"], (r, dr)
+        + dr["score"] + dr["sources"] + dr["upside"] + dr["theme"] + dr["dup"] \
+        + dr["display"], (r, dr)
+    assert dr["display"] == 1 and r["display_only"] == ["600600.SH"], (dr, r["display_only"])
     assert len(r["items"]) == r["eligible"] - dr["capped"] == 2
+    # 分流开关关着: 那只仅展示的票照常进池 (分数高, 会排到第一), 计数里 display 是 0
+    r0 = pf.select_candidates(p, held=["SH600418"], black=["SZ300952"], tiers=["强传导"],
+                              min_sources=7, top_n=2, include_observe=True, theme_cap=2)
+    assert r0["dropped"]["display"] == 0 and r0["display_only"] == []
+    assert r0["items"][0]["ts_code"] == "600600.SH", r0["items"]
 
 
 @case("筛选·空计划不炸 (main/observe 缺键) 且 date 透传")