第三件 PMS 侧：按逻辑状态分流，入场论点随持仓走

新模块 logic_state_service：早上拉完计划后对在持票查一次逻辑状态（桥的 /logic_state），
写进运行参数 PMS_LOGIC_STATE_MAP，取不到留空绝不折成存疑；按结果暂停或恢复策略买入腿
（来源 logic，不动风控与定性停的）；把当日态挂到持仓行；持仓视图两栏（当初为什么买 /
现在证据还在不在），三种情形都说得出话。
plan_feed：解析层收 logic_state（出处最多三条），候选 items 带它；加按代码查询与应答解析。
action_engine：硬数字带 logic_state（研判白名单不收）；判决候选而逻辑存疑的新建仓强制交人；
持仓存疑停增持侧；新求值器 eval_weak_research（默认关，开了必交人，按可卖量夹紧）排在
保垫减仓前；同轮减持优先级改为到价清仓 > 研究走弱 > 保垫减仓。
proposal_service：扫描参数加三个旋钮，扫描前挂逻辑状态。scheduler.plan_pull 顺带取回。
param_store：四个运行参数（分流开关默认开、减持提议默认关、比例三分之一、映射只读）。
pms_repo.ledger_by_ref 与 FakeRepo 同签名；api_positions 加两块；页面加两列。
测试：新建 test_batch22_units.py 十四例并登记；第十二批求值器名单钉子同步。

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

app/core/action_engine.py

@@ -53,6 +53,16 @@ BUY, SELL = "buy", "sell"
 # 只在同一只票的同一轮里比较, 不影响不同票, 也不影响下一轮。
 _SELL_PRIORITY = {A_EXIT: 0, A_TRIM: 1}
 
+
+def _sell_priority(c: dict) -> int:
+    """同一轮多条减持留哪一条 (数字小的优先; 2026-09-07 第三件起按动作加来源):
+    到价清仓 (人的意思) → 研究证据走弱的减持 (必交人) → 保垫减仓 (规则自动)。"""
+    if c.get("action") == A_EXIT:
+        return 0
+    if c.get("source") == SRC_RESEARCH_WEAK:
+        return 1
+    return 2
+
 # 候选来源 (2026-09-03)。每条候选都带一个来源, 提议分流那一步拿它决定「这条减持能不能不等人」。
 #
 # 为什么需要这一项: 分流从前写的是「卖出方向一律自动执行」, 只看方向、不看来源。那条规矩
@@ -96,6 +106,38 @@ VERDICT_CANDIDATE, VERDICT_WATCH, VERDICT_DISPLAY = "候选", "关注", "仅展
 WHY_WATCH_CONFIRM = "选股系统判为关注（无法判断），交人裁决"
 WHY_DISPLAY_ONLY = "选股系统判为仅展示"
 
+# 逻辑状态四态里唯一会触发动作的值 (2026-09-07 第三件)。四态是选股系统按「支撑这只票的研究证据
+# 还在不在」算出来的跨日跟踪, 与候选卡判决是正交的两维; 只有「逻辑存疑」会改分流通道 —— 持仓停
+# 增持侧自主动作、按开关出必交人的减持; 判决候选而逻辑存疑的新建仓强制人工确认。其余三个值
+# (逻辑强化 / 逻辑成立 / 无法判断) 不产生任何动作。中文原文与上游逐字对应, 不做映射。
+LOGIC_DOUBT = "逻辑存疑"
+WHY_LOGIC_DOUBT_CONFIRM = "可交易口径成立，但研究证据在走弱（逻辑存疑），交人裁决"
+WHY_LOGIC_DOUBT_PAUSE = "研究证据在走弱（逻辑存疑），停掉增持侧的自主动作"
+
+
+def is_logic_doubt(logic_state) -> bool:
+    return isinstance(logic_state, dict) and str(logic_state.get("state") or "").strip() == LOGIC_DOUBT
+
+
+def logic_doubt_text(logic_state, head: str) -> str:
+    """把存疑的出处接在一句话后面: 第一条说明与证据截止日, 让人看得出是哪一路在走弱、证据是哪天的。"""
+    st = logic_state if isinstance(logic_state, dict) else {}
+    parts = [head]
+    reasons = [str(x) for x in (st.get("reasons") or []) if x]
+    if reasons:
+        parts.append(reasons[0][:120])
+    if st.get("as_of"):
+        parts.append(f"证据截止 {st.get('as_of')}")
+    return "；".join(parts)
+
+
+def logic_confirm_why(logic_state):
+    """按逻辑状态要不要强制人工确认: 逻辑存疑回原因 (带出处), 其余一律 None。
+    与 verdict_confirm_why 并列: 判决管「今天要不要买」, 逻辑状态管「研究证据还在不在」。"""
+    if not is_logic_doubt(logic_state):
+        return None
+    return logic_doubt_text(logic_state, WHY_LOGIC_DOUBT_CONFIRM)
+
 
 # 冻结的三种来路，对交易员是三件不同的事：自己下的命令、自己手设的、系统刹车期间。
 # 原先一句「该股 COMMAND_HALT, 禁增持」三种都不分，还把枚举值印在了页面上。
@@ -367,6 +409,36 @@ def eval_target(p: dict, params: dict = None, mkt: dict = None):
     return c
 
 
+def eval_weak_research(p: dict, params: dict, mkt: dict = None):
+    """研究证据走弱的减持 (2026-09-07 第三件): 持仓行上的逻辑状态是「逻辑存疑」时, 提议减掉一部分。
+
+    这是 reduce_on_weak_research 那条预留通道的第一个上游。三条口径一次定死:
+      开关   params["logic_doubt_trim"] (PMS_LOGIC_DOUBT_TRIM), **默认关** —— 先跑停增持一个月看误伤;
+      比例   params["logic_doubt_trim_ratio"] (PMS_LOGIC_DOUBT_TRIM_RATIO), 默认与保垫减仓相同的三分之一;
+      数量   按 T+1 可卖量夹紧, 不足最小申报数量不产出 (与保垫减仓、到价清仓同口径)。
+    产出的候选带 research_weak 来源与强制确认标记, 分流那一步一票否决入人工队列, 绝不自动卖。
+    没有逻辑状态 (早上没查到) 不产出 —— 「没有读数」不是「证据走弱」。
+    """
+    if not params.get("logic_doubt_trim"):
+        return None
+    st = p.get("logic_state")
+    if not is_logic_doubt(st):
+        return None
+    total = int(p.get("total_qty") or 0)
+    ratio = _f(params.get("logic_doubt_trim_ratio"), 1.0 / 3)
+    if ratio <= 0 or ratio > 1:
+        ratio = 1.0 / 3
+    qty = clamp_sell_qty(int(total * ratio // LOT) * LOT, p)
+    if qty < lot_of(p.get("ts_code")):
+        return None
+    why = logic_doubt_text(st, f"研究证据走弱: 拟减 {qty} 股 (持仓 {total} 股的 {ratio:.0%}), 卖不卖由你拍板")
+    return reduce_on_weak_research(
+        p["ts_code"], qty=qty, reason=why,
+        hard_numbers={"logic_state": st, "price": p.get("price"),
+                      "cushion_pct": p.get("cushion_pct"), "total_qty": total,
+                      "avail_qty": p.get("avail_qty"), "trim_ratio": round(ratio, 4)})
+
+
 def clamp_sell_qty(qty: int, pos: dict) -> int:
     """卖出数量按 T+1 可卖量夹紧。可卖量缺失 (None) 时按总持仓, 不当成零。
 
@@ -383,7 +455,9 @@ def clamp_sell_qty(qty: int, pos: dict) -> int:
     return max(0, min(q, int(avail or 0)))
 
 
-EVALUATORS = ((A_TRIM, eval_trim), (A_EXIT, eval_target),
+# 研究走弱的减持排在保垫减仓前面 (2026-09-07 第三件): 两者动作名都是 TRIM, 同轮同现时按
+# _sell_priority 留研究走弱那条 (它必交人), 保垫减仓让路 —— 证据在走弱的票, 不该由规则自动先卖一刀。
+EVALUATORS = ((A_TRIM, eval_weak_research), (A_TRIM, eval_trim), (A_EXIT, eval_target),
               (A_ADD, eval_add), (A_FILL, eval_fill), (A_DCA, eval_dca))
 
 # 减持方向的动作 —— 冻结只禁增持, 这几类照评 (与规则闸 _check_sell 同一口径)。
@@ -461,6 +535,9 @@ def scan(*, positions: list, params: dict, market: dict, skip=None,
         if sp:
             p = {**p, "target_price": sp.get("target_price"), "stop_price": sp.get("stop_price")}
         frozen = (p.get("frozen_reason") or "NONE") != "NONE"
+        # 逻辑存疑停增持侧 (2026-09-07 第三件): 持仓行上挂着早上查回的逻辑状态 (logic_state_service.attach),
+        # 存疑的票补足 / 加仓 / 补仓一律不评, 减持侧照评。没挂这个键 = 没有读数, 不拦。
+        doubt = bool(params.get("logic_state_route")) and is_logic_doubt(p.get("logic_state"))
         # 跨轮的减持让路 (2026-09-07 审查修)。同轮只发一条减持那条规矩 (见下) 只管一次扫描;
         # 下一分钟再扫时, 等人拍板的到价清仓已成在途、按 (代码, EXIT) 被跳过, 而保垫减仓
         # 单独产出、不需确认、卖出方向又不走强制入队 —— 系统刚说「卖不卖由你定」, 一分钟后
@@ -486,6 +563,10 @@ def scan(*, positions: list, params: dict, market: dict, skip=None,
                 skipped.append({"ts_code": code, "action": action,
                                 "why": frozen_why(p.get("frozen_reason"))})
                 continue
+            if doubt and action not in SELL_SIDE_ACTIONS:
+                skipped.append({"ts_code": code, "action": action,
+                                "why": logic_doubt_text(p.get("logic_state"), WHY_LOGIC_DOUBT_PAUSE)})
+                continue
             try:
                 c = fn(p, params, mkt)
             except Exception as e:      # 单票异常不能拖垮整轮扫描
@@ -516,7 +597,7 @@ def scan(*, positions: list, params: dict, market: dict, skip=None,
         # 算出来的减仓前面。被让路的那条记进跳过原因, 下一轮条件仍成立时自然会再来。
         sells_now = [c for c in cands_this if c["side"] == SELL]
         if len(sells_now) > 1:
-            keep = min(sells_now, key=lambda c: _SELL_PRIORITY.get(c["action"], 99))
+            keep = min(sells_now, key=_sell_priority)
             for c in sells_now:
                 if c is not keep:
                     skipped.append({"ts_code": code, "action": c["action"],
@@ -603,6 +684,9 @@ def eval_open(c: dict, params: dict, caps: dict, room_amt: float):
         # 研报里说这家公司好在哪、出处是哪份研报哪一天。两者只进评审账本与提议卡给人看,
         # 不参与任何判定, 也不送研判 —— 产业逻辑归数据基座与选股系统, 不归择时决策系统。
         "basis": c.get("basis"), "logic": c.get("logic"),
+        # 逻辑状态四态 (2026-09-07 第三件): 入场那一刻「研究证据还在不在」的读数, 进账本与提议卡,
+        # 持仓视图的「当初为什么买」从这里回溯; 不送研判 (judge.OPEN_JUDGE_KEYS 不收它)。
+        "logic_state": c.get("logic_state"),
         # ---- 仓位口径: 只进评审账本做判分锚。judge.py 送研判时会把这几项过滤掉,
         #      理由见那边的 OPEN_JUDGE_KEYS —— 决策系统本来就不管仓位, 别送过去带偏它。
         "target_pct": target_pct, "target_amount": round(full_amt, 2),
@@ -721,6 +805,14 @@ def scan_open(*, candidates: list, params: dict, caps: dict, room_amt: float,
                 # 只是最后一步不许自动落指令 —— 与深档补仓强制确认走的是同一个字段。
                 cand["needs_user_confirm"] = True
                 cand["confirm_why"] = confirm_why
+        # 判决候选而逻辑存疑 (2026-09-07 第三件): 可交易口径成立而研究证据在走弱, 这正是杀逻辑与
+        # 杀估值那条边界, 强制交人; 复用同一个强制确认字段, 与关注判决走同一条队列。已经要确认的
+        # (关注 / 认不出的判决) 保留原因, 不覆盖。开关 PMS_LOGIC_STATE_ROUTE 关着时一行都不执行。
+        if params.get("open_route_by_logic") and not cand.get("needs_user_confirm"):
+            lw = logic_confirm_why(c.get("logic_state"))
+            if lw:
+                cand["needs_user_confirm"] = True
+                cand["confirm_why"] = lw
         out.append(cand)
         used = _f(cand.get("target_amount"))
         left -= used

app/repo/pms_repo.py

@@ -673,6 +673,23 @@ def list_ledger(*, ts_code=None, limit: int = 200) -> list:
     return rows
 
 
+def ledger_by_ref(ref_ids, limit: int = 50) -> list:
+    """按 ref_id (指令号或提议号) 取账本行, 按写入顺序 (2026-09-07 第三件: 持仓视图回溯入场论点用)。"""
+    ids = [str(x) for x in (ref_ids or []) if x]
+    if not ids:
+        return []
+    keys, p = [], {"n": int(limit)}
+    for i, x in enumerate(ids):
+        keys.append(f":r{i}")
+        p[f"r{i}"] = x
+    rows = fetch_all(f"SELECT * FROM pms_action_ledger WHERE ref_id IN ({', '.join(keys)}) "
+                     f"ORDER BY id ASC LIMIT :n", p)
+    for r in rows:
+        r["hard_numbers"] = _loads(r.get("hard_numbers_json"), {})
+        r["failed_checks"] = _loads(r.get("failed_checks_json"), [])
+    return rows
+
+
 # ================================================================ pms_daily_report
 def upsert_report(ymd: int, report: dict) -> int:
     return execute(

app/scheduler.py

@@ -159,8 +159,17 @@ def plan_pull():
     from app.services import plan_feed
     plan_feed.invalidate()
     plan = plan_feed.get_plan(force=True)
-    return {"ok": True, "date": plan["date"], "age_tdays": plan.get("age_tdays"),
+    out = {"ok": True, "date": plan["date"], "age_tdays": plan.get("age_tdays"),
            "returned": plan["returned"], "theme_sync": plan.get("theme_sync")}
+    # 在持票的逻辑状态 (2026-09-07 第三件): 拉完计划顺带查一次, 写进运行参数并按结果停或恢复策略
+    # 买入腿。失败只记录, 不影响拉计划 —— 取不到就留空, 扫描层按没有读数处理。
+    try:
+        from app.services import logic_state_service
+        out["logic_state"] = logic_state_service.pull_for_held()
+    except Exception as e:  # noqa: BLE001
+        logger.error("[plan_pull] 逻辑状态取回失败: %s", e)
+        out["logic_state"] = {"ok": False, "error": f"{type(e).__name__}: {e}"}
+    return out
 
 
 @celery_app.task(name="pms.premarket")

app/services/logic_state_service.py

@@ -0,0 +1,239 @@
+# -*- coding: utf-8 -*-
+"""逻辑状态四态在 PMS 的落点 (2026-09-07 下一阶段方案第三件 PMS 侧)。
+
+选股系统每天早上给每只票算一个「支撑它的研究证据还在不在」的状态 —— 逻辑强化 / 逻辑成立 /
+无法判断 / 逻辑存疑 (它那边的 logic_state.py), 随计划每行下发, 另有一个按代码查询的接口给持仓票用
+(持仓票在候选筛选第一步就被整行剔掉, 计划里读不到它)。PMS 只读它, 不算它, 也不送研判 —— 产业逻辑
+不归择时决策系统判, judge.OPEN_JUDGE_KEYS 不收它, 那是一致性检查表第九行要继续成立的东西。
+
+四态里只有「逻辑存疑」会改分流通道, 其余三个值不产生任何动作 (收敛规则是单调的: 强化不升判决,
+存疑不把仅展示变成可执行)。本模块做四件事:
+
+  一, 早上拉完计划后 (scheduler.plan_pull), 对在持的票查一次状态, 写进运行参数 PMS_LOGIC_STATE_MAP。
+      取不到就留空 (写一份空映射并记原因), **绝不折成存疑**。
+  二, 按状态暂停或恢复策略腿的买入: 存疑 → strategy_service.pause_buy(source="logic");
+      已明确不存疑 → 只清本来源的暂停, 不动风控 (signal) 与定性 (accum) 停的; 没查到的票不动。
+  三, 把当日态挂到持仓行上 (attach), 动作引擎据此停增持侧的自主动作、按开关出研究走弱的减持
+      (那条减持带 research_weak 来源, 分流那一步一票否决强制交人, 绝不自动卖)。
+  四, 持仓视图的两栏: 「当初为什么买」(账本里首批入场那条记录的判决依据与论断) 与
+      「现在证据还在不在」(今天的状态与出处)。
+
+安全方向: 读不到状态 = 没有读数, 什么都不额外拦, 也不解除已有的暂停。总开关
+PMS_LOGIC_STATE_ROUTE 关掉即回到只按判决分流; 减持提议另有 PMS_LOGIC_DOUBT_TRIM, 默认关。
+"""
+from __future__ import annotations
+
+import json
+import logging
+from collections import Counter
+from datetime import date, datetime
+
+from app.core import action_engine as ae
+from app.repo import pms_repo
+from app.services import param_store, plan_feed, strategy_service
+
+logger = logging.getLogger("pms.logic")
+
+MAP_KEY = "PMS_LOGIC_STATE_MAP"
+PAUSE_SOURCE = "logic"            # 策略买入暂停表里本模块用的来源名, 与 signal / accum 并列
+FRESH_DAYS = 3                    # 映射超过这么多个自然日没刷新就当没有读数 (调度断了不能拿旧态拦人)
+
+# 入场那条账本记录里, 只要带了这几个键之一就算「有研究理由」
+_ENTRY_KEYS = ("basis", "logic", "verdict", "logic_state")
+
+
+# ================================================================ 运行参数里的映射
+def load_map() -> dict:
+    """整份映射 (含时刻与错误), 页面与运维看。读不到或坏了返回空字典。"""
+    raw = param_store.get(MAP_KEY, "") or ""
+    if not raw:
+        return {}
+    try:
+        m = json.loads(raw)
+    except (TypeError, ValueError):
+        return {}
+    return m if isinstance(m, dict) else {}
+
+
+def state_map(m: dict | None = None) -> dict:
+    """{ts_code: 逻辑状态} —— 只在映射新鲜时给; 超过 FRESH_DAYS 个自然日没刷新一律按没有读数。"""
+    m = load_map() if m is None else (m or {})
+    at = str(m.get("at") or "")[:10]
+    try:
+        age = (date.today() - date.fromisoformat(at)).days
+    except ValueError:
+        return {}
+    if age > FRESH_DAYS:
+        return {}
+    states = m.get("states")
+    return dict(states) if isinstance(states, dict) else {}
+
+
+def _save_map(m: dict) -> dict:
+    r = param_store.set_param(MAP_KEY, json.dumps(m, ensure_ascii=False), "system")
+    if not r.get("ok"):
+        # 写不进要吭声: 映射停在旧值, 明天早上以前扫描层看到的都是昨天的态
+        logger.error("[逻辑状态] 映射写入失败: %s —— 扫描层沿用上一份 (最多 %d 天后自动失效)",
+                     r.get("error"), FRESH_DAYS)
+    return r
+
+
+def route_enabled() -> bool:
+    return param_store.get_bool("PMS_LOGIC_STATE_ROUTE", True)
+
+
+# ================================================================ 早上那一步
+def pull_for_held(now=None, fetch=None) -> dict:
+    """在持票的逻辑状态取回一次并落映射, 再按结果暂停或恢复策略买入腿。
+
+    fetch 可注入 (单测)。任何一步失败都不抛: 取数失败写空映射带原因 (扫描层按没有读数处理),
+    暂停那一步失败只记进 errors。返回一份读数给调度任务写日志。
+    """
+    now = now or datetime.now()
+    stamp = now.strftime("%Y-%m-%d %H:%M:%S")
+    out = {"ok": True, "at": stamp, "held": 0, "got": 0, "by_state": {}, "paused": [],
+           "resumed": [], "errors": []}
+    try:
+        held = [r["ts_code"] for r in pms_repo.list_positions(only_open=True)]
+    except Exception as e:  # noqa: BLE001
+        held = []
+        out["errors"].append(f"读持仓失败: {type(e).__name__}: {e}")
+    out["held"] = len(held)
+    if not held:
+        _save_map({"at": stamp, "date": None, "states": {}, "note": "没有在持的票"})
+        return out
+    try:
+        states = (fetch or plan_feed.fetch_logic_states)(held)
+    except Exception as e:  # noqa: BLE001
+        # 取不到就留空, 绝不折成存疑; 也不动已有的暂停 (没有新证据不解除)
+        _save_map({"at": stamp, "date": None, "states": {}, "error": str(e)[:300]})
+        out.update(ok=False)
+        out["errors"].append(f"查逻辑状态失败: {e}")
+        return out
+    day = next((s.get("date") for s in states.values() if s.get("date")), None)
+    r = _save_map({"at": stamp, "date": day, "states": states})
+    if not r.get("ok"):
+        out["errors"].append(f"映射写入失败: {r.get('error')}")
+    out["got"] = len(states)
+    out["by_state"] = dict(Counter(s.get("state") or "空" for s in states.values()))
+    missing = [c for c in held if c not in states]
+    if missing:
+        out["missing"] = missing
+    p = apply_pauses(states, held)
+    out["paused"], out["resumed"] = p["paused"], p["resumed"]
+    out["errors"].extend(p["errors"])
+    if p.get("skipped"):
+        out["pause_skipped"] = p["skipped"]
+    return out
+
+
+def apply_pauses(states: dict, held) -> dict:
+    """存疑 → 暂停策略买入腿 (来源 logic); 明确不存疑 → 只清本来源的暂停; 没查到的票不动。"""
+    out = {"paused": [], "resumed": [], "errors": []}
+    if not route_enabled():
+        out["skipped"] = "PMS_LOGIC_STATE_ROUTE 关着, 不动策略买入腿"
+        return out
+    for code in held or []:
+        st = (states or {}).get(code)
+        if not st:
+            continue                                   # 没有读数: 不拦也不放
+        try:
+            if ae.is_logic_doubt(st):
+                strategy_service.pause_buy(
+                    code, reason=ae.logic_doubt_text(st, ae.WHY_LOGIC_DOUBT_PAUSE),
+                    source=PAUSE_SOURCE)
+                out["paused"].append(code)
+            else:
+                r = strategy_service.clear_buypause(code, only_source=PAUSE_SOURCE) or {}
+                if r.get("cleared"):
+                    out["resumed"].append(code)
+        except Exception as e:  # noqa: BLE001
+            out["errors"].append(f"{code} 策略买入腿处理失败: {type(e).__name__}: {e}")
+    return out
+
+
+# ================================================================ 扫描层
+def attach(rows: list, states: dict | None = None) -> list:
+    """把当日逻辑状态挂到持仓行上 (原地改)。没有读数的票不挂键, 动作引擎按「没有读数」处理。"""
+    states = state_map() if states is None else (states or {})
+    for r in rows or []:
+        st = states.get(r.get("ts_code"))
+        if st:
+            r["logic_state"] = st
+        else:
+            r.pop("logic_state", None)
+    return rows
+
+
+# ================================================================ 持仓视图两栏
+def now_view(st) -> dict:
+    """「现在证据还在不在」: 今天的状态、截止日、落定说明与第一条出处; 没有读数就说没有。"""
+    if not isinstance(st, dict) or not st.get("state"):
+        return {"state": None, "text": "今天没有读数（早上拉计划时没查到这只票）"}
+    reasons = [str(x) for x in (st.get("reasons") or []) if x]
+    return {"state": st.get("state"), "raw_state": st.get("raw_state"),
+            "as_of": st.get("as_of"), "settle_note": st.get("settle_note"),
+            "text": (reasons[0] if reasons else (st.get("why") or ""))[:160]}
+
+
+def entry_view(ts_code: str) -> dict:
+    """「当初为什么买」: 首批未平批次 → 它的指令 → 该指令或它来自的提议在账本上的放行记录 → 硬数字里的
+    判决依据、论断、判决与入场时的逻辑状态。三种情形都要说得出话: 有指令链、外部成交并入、账本无行。"""
+    try:
+        lots = pms_repo.list_lots(ts_code, status="OPEN", limit=50)
+    except Exception as e:  # noqa: BLE001
+        return {"why": f"批次读取失败: {type(e).__name__}"}
+    if not lots:
+        return {"why": "没有未平的批次记录"}
+    first = lots[0]
+    open_date = str(first.get("open_date") or "")[:10]
+    iid = first.get("instruction_id")
+    if not iid:
+        return {"why": "外部成交并入的持仓，没有系统的入场记录", "open_date": open_date}
+    refs = [str(iid)]
+    try:
+        ins = pms_repo.get_instruction(str(iid)) or {}
+    except Exception:  # noqa: BLE001
+        ins = {}
+    for k in (ins.get("origin_id"), (ins.get("progress") or {}).get("from_proposal")):
+        if k and str(k) not in refs:
+            refs.append(str(k))
+    try:
+        rows = pms_repo.ledger_by_ref(refs)
+    except Exception as e:  # noqa: BLE001
+        return {"why": f"账本读取失败: {type(e).__name__}", "open_date": open_date}
+    hit = None
+    for r in rows:
+        hn = r.get("hard_numbers") or {}
+        if r.get("verdict") == "PASS" and any(hn.get(k) for k in _ENTRY_KEYS):
+            hit = r
+            break
+    if not hit:
+        return {"why": "账本里没有这次入场的研究理由（命令建仓或策略买入不带候选卡）",
+                "open_date": open_date, "instruction_id": str(iid)}
+    hn = hit.get("hard_numbers") or {}
+    st = hn.get("logic_state") if isinstance(hn.get("logic_state"), dict) else {}
+    return {"basis": hn.get("basis"), "logic": list(hn.get("logic") or [])[:3],
+            "verdict": hn.get("verdict"), "logic_state_at_entry": st.get("state"),
+            "decided_at": str(hit.get("decided_at") or "")[:16],
+            "arbiter": hit.get("arbiter"), "reason": str(hit.get("reason") or "")[:200],
+            "open_date": open_date, "instruction_id": str(iid)}
+
+
+def decorate_positions(rows: list, states: dict | None = None) -> list:
+    """给持仓接口的每一行加 entry 与 logic_now 两块 (原地改)。任何一行失败只写原因, 页面不能塌。"""
+    states = state_map() if states is None else (states or {})
+    for r in rows or []:
+        code = r.get("ts_code")
+        try:
+            r["logic_now"] = now_view(states.get(code))
+        except Exception as e:  # noqa: BLE001
+            r["logic_now"] = {"state": None, "text": f"读数处理失败: {type(e).__name__}"}
+        if int(r.get("total_qty") or 0) <= 0:
+            r["entry"] = None
+            continue
+        try:
+            r["entry"] = entry_view(code)
+        except Exception as e:  # noqa: BLE001
+            r["entry"] = {"why": f"入场记录处理失败: {type(e).__name__}"}
+    return rows

app/services/param_store.py

@@ -67,6 +67,12 @@ RUNTIME_EXTRA = {
     "PMS_PUBLISH_ACCOUNT": ("天盟实业", str, "公示表·开单账户列的账户名"),
     "PMS_PUBLISH_STRUCTURE": ("二级（平层）", str, "公示表·结构列的产品结构标签"),
     "PMS_PUBLISH_NAV_SCALE": (0.0, float, "公示净值规模 (元): 净值=1+累计盈亏/此数; 0=用 PMS_TOTAL_SCALE"),
+    # 2026-09-07 第三件: 选股系统的逻辑状态四态在 PMS 的三个开关与一份运行态映射 (模块见
+    # logic_state_service.py)。映射与 PMS_MACRO_GATE_STATE 同一手法: 早上写一次, 扫描层与页面只读。
+    "PMS_LOGIC_STATE_ROUTE": (True, bool, "按选股系统的逻辑状态分流: 逻辑存疑的持仓停增持侧自主动作并暂停策略买入腿, 判决候选而逻辑存疑的新建仓强制人工确认 (默认开; 关掉即回到只按判决)"),
+    "PMS_LOGIC_DOUBT_TRIM": (False, bool, "逻辑存疑的持仓出减持提议 (必定交人裁决, 绝不自动卖; 默认关, 先跑停增持一个月看误伤再开)"),
+    "PMS_LOGIC_DOUBT_TRIM_RATIO": (0.3333, float, "逻辑存疑减持提议的比例 (占总持仓; 默认与保垫减仓相同的三分之一; 一次定死, 不按复盘读数回调)"),
+    "PMS_LOGIC_STATE_MAP": ("", str, "在持票的逻辑状态映射 (plan_pull 每早写入的 JSON: 时刻/日期/逐票状态), 扫描层与页面只读, 勿手改"),
 }
 
 # **读不到时必须按"已暂停"处理的键 (fail-closed)。**

app/services/plan_feed.py

@@ -145,6 +145,27 @@ REASONS_KEEP = 4
 #     第一条, 其余折叠, 3 条正好是「折开也还看得完」的量。
 LOGIC_KEEP = 3
 
+# 逻辑状态四态 (2026-09-07 第三件): 上游每行带一个字典 —— state 是落定态 (下游按它分流), raw_state
+# 是当天原始态, why 是无法判断的子因 (证据不足 / 证据矛盾), settle_note 是抗抖动的说明, as_of 是
+# 证据截止日, usable / missing 是四路里哪几路在、哪几路缺, reasons 是每路的一句话出处。
+# 只收这几个键; 出处最多带三条 (与因果论断同一个量级, 提议卡与账本放得下)。
+LOGIC_STATE_REASONS_KEEP = 3
+
+
+def _logic_state_or_none(v):
+    """逻辑状态归一: 不是字典或没有 state 一律 None (旧版计划没有这个字段)。"""
+    if not isinstance(v, dict):
+        return None
+    state = _text_or_none(v.get("state"))
+    if not state:
+        return None
+    return {"state": state, "raw_state": _text_or_none(v.get("raw_state")) or state,
+            "why": _text_or_none(v.get("why")), "settle_note": _text_or_none(v.get("settle_note")),
+            "as_of": _text_or_none(v.get("as_of")),
+            "usable": _list_or_none(v.get("usable")) or [],
+            "missing": _list_or_none(v.get("missing")) or [],
+            "reasons": _list_or_none(v.get("reasons"), limit=LOGIC_STATE_REASONS_KEEP) or []}
+
 
 def _list_or_none(v, limit: int = 0):
     """字符串列表归一: 列表逐项去空白、丢空项, 单个字符串当一项; 缺失或类型不对一律 None。
@@ -199,6 +220,8 @@ def _rows(raw, bucket: str) -> list:
             # 研究理由 (2026-09-03): 只给人看, 不参与任何判定, 也不送研判。见模块说明。
             "basis": _text_or_none(it.get("basis")),
             "logic": _list_or_none(it.get("logic"), limit=LOGIC_KEEP),
+            # 逻辑状态四态 (2026-09-07 第三件): 跟着候选进硬数字, 不送研判 (judge.OPEN_JUDGE_KEYS 不收)。
+            "logic_state": _logic_state_or_none(it.get("logic_state")),
         })
     return out
 
@@ -387,7 +410,10 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N
               "card_rank": r.get("card_rank"),
               # 研究理由两键: 跟着候选一路带到提议卡与评审账本, 让人看得到研报说了什么。
               # 它们不进送研判的那份名单 (judge.OPEN_JUDGE_KEYS), 产业逻辑不归择时决策系统判。
-              "basis": r.get("basis"), "logic": r.get("logic")}
+              "basis": r.get("basis"), "logic": r.get("logic"),
+              # 逻辑状态 (2026-09-07 第三件): 动作引擎放进硬数字; 判决候选而逻辑存疑的在 scan_open
+              # 里强制交人。它不改判决、不改排序 —— 收敛规则是单调的 (选股系统 logic_state.apply_to_card)。
+              "logic_state": r.get("logic_state")}
              for r in passed[:n]]
     return {"date": plan.get("date"), "considered": len(pool), "eligible": len(passed),
             "items": items, "dropped": dropped, "st_unknown": st_unknown,
@@ -567,6 +593,52 @@ def fetch(*, date=None, base=None, path=None, timeout=None, extra_params=None) -
     return plan
 
 
+def parse_logic_states(payload) -> dict:
+    """按代码查询接口的应答 → {ts_code: 逻辑状态}。带 error 的行跳过 (代码形态认不出), 查不到的不在结果里。
+
+    上游按我们送去的原样 (input) 回, 代码用它映射回 PMS 的形态; 每条另带 source (daily = 早上落定的
+    当日行, computed = 此刻现算) 与日期, 页面要区分「早上的态」与「现算的态」。"""
+    out = {}
+    day = _text_or_none((payload or {}).get("date")) if isinstance(payload, dict) else None
+    for it in ((payload or {}).get("states") or []) if isinstance(payload, dict) else []:
+        if not isinstance(it, dict) or it.get("error"):
+            continue
+        code = normalize_code(str(it.get("input") or it.get("code") or ""))
+        st = _logic_state_or_none(it)
+        if not code or not st:
+            continue
+        st["source"] = _text_or_none(it.get("source"))
+        st["date"] = day
+        out[code] = st
+    return out
+
+
+def fetch_logic_states(codes, *, base=None, timeout=None, path: str = "/logic_state") -> dict:
+    """按代码向选股系统查逐票逻辑状态 (2026-09-07 第三件, 给在持的票用)。任何失败抛 PlanFeedError,
+    由调用方决定留空 —— 这里不吞错, 因为「查不到」与「判为成立」在扫描层是两件事。"""
+    codes = [normalize_code(str(c)) for c in (codes or []) if c]
+    if not codes:
+        return {}
+    if base is None or timeout is None:
+        p = _params()
+        base = p["base"] if base is None else base
+        timeout = p["timeout"] if timeout is None else timeout
+    base = (base or "").strip().rstrip("/")
+    if not base:
+        raise PlanFeedError(_NO_BASE_WHY)
+    url = base + (path if path.startswith("/") else "/" + path)
+    to = int(timeout or 10)
+    try:
+        import requests
+        r = requests.get(url, params={"codes": ",".join(codes)}, timeout=to)
+        r.raise_for_status()
+        payload = r.json()
+    except Exception as e:
+        logger.warning("查逻辑状态失败 %s: %s: %s", url, type(e).__name__, e)
+        raise PlanFeedError(_fetch_fail_why(e, to)) from e
+    return parse_logic_states(payload)
+
+
 def get_plan(*, force: bool = False, date=None) -> dict:
     """带缓存的当前计划。失败同样缓存 FAIL_CACHE_SEC, 但每次调用都照样抛。"""
     p = _params()

app/services/proposal_service.py

@@ -104,6 +104,13 @@ def scan_and_route(*, now=None, dry_run: bool = False) -> dict:
             strategy_codes = pms_repo.active_strategy_codes()
         except Exception:
             strategy_codes = set()
+        # 逻辑状态挂到持仓行上 (2026-09-07 第三件): 早上拉计划时查回的当日态。读不到就没有这个键,
+        # 动作引擎按「没有读数」处理, 绝不折成存疑; 挂不上只记日志, 不拖垮扫描。
+        try:
+            from app.services import logic_state_service
+            logic_state_service.attach(view["held"])
+        except Exception as e:  # noqa: BLE001
+            logger.warning("[逻辑状态] 挂到持仓行失败, 本轮按没有读数: %s", e)
         scanned = ae.scan(positions=view["held"], params=params, market=mkt, skip=skip,
                           strategy_codes=strategy_codes, stock_params=stock_params)
     except Exception as e:
@@ -766,6 +773,12 @@ def _scan_params(view: dict) -> dict:
         "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),
+        # 逻辑状态四态的三个旋钮 (2026-09-07 第三件): 分流开关 (持仓停增持 + 新建仓强制确认)、
+        # 减持提议开关 (默认关)、减持比例 (默认与保垫减仓相同的三分之一)。
+        "logic_state_route": param_store.get_bool("PMS_LOGIC_STATE_ROUTE", True),
+        "open_route_by_logic": param_store.get_bool("PMS_LOGIC_STATE_ROUTE", True),
+        "logic_doubt_trim": param_store.get_bool("PMS_LOGIC_DOUBT_TRIM", False),
+        "logic_doubt_trim_ratio": param_store.get_float("PMS_LOGIC_DOUBT_TRIM_RATIO", 1.0 / 3),
     })
     return p
 

app/web/main.py

@@ -25,7 +25,8 @@ from app.core import command_spec as cs
 from app.core import tradedays as td
 from app.db import session as dbs
 from app.repo import downstream_repo, pms_repo
-from app.services import command_service, industry, ledger_service, param_store, portfolio
+from app.services import (command_service, industry, ledger_service, logic_state_service,
+                          param_store, portfolio)
 from app.web import auth as authmod
 
 logging.basicConfig(level=logging.INFO,
@@ -374,6 +375,12 @@ def api_positions():
         # 把两个数分列显示 —— 只是显示, 判断那一侧照旧各读各的事实源, 见
         # command_service.attach_user_prices 的说明。
         command_service.attach_user_prices(v["positions"], sp)
+        # 入场论点随持仓走 (2026-09-07 第三件): 每行加「当初为什么买」(entry) 与「现在证据还在不在」
+        # (logic_now)。只是显示; 任何一行取不到都写原因, 页面不能因此塌掉。
+        try:
+            logic_state_service.decorate_positions(v["positions"])
+        except Exception as e:  # noqa: BLE001
+            logger.warning("[持仓] 入场论点与逻辑状态两栏取不到: %s", e)
         return {"ok": True, **v, "stock_params": sp}
     return ok(_view)
 

app/web/static/index.html

@@ -1567,6 +1567,32 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
               <span v-else class="muted">未设</span>
             </template>
           </el-table-column>
+          <!-- 入场论点随持仓走 (2026-09-07 第三件)。左栏回答「当初为什么买」: 账本里这只票首批入场那条
+               记录的判决依据与第一条论断 (悬停看全部论断); 右栏回答「现在证据还在不在」: 选股系统每早
+               查回的逻辑状态、第一条出处与证据截止日。状态值是上游给的中文原文, 不是枚举码。 -->
+          <el-table-column label="当初为什么买" min-width="220">
+            <template #default="s">
+              <template v-if="s.row.entry && s.row.entry.basis">
+                <div :title="(s.row.entry.logic||[]).join('\n')">{{ s.row.entry.basis }}</div>
+                <div class="muted" v-if="(s.row.entry.logic||[]).length">{{ s.row.entry.logic[0] }}</div>
+                <div class="muted mono" style="font-size:11px">{{ s.row.entry.decided_at || '' }}
+                  <template v-if="s.row.entry.verdict"> · 判决 {{ s.row.entry.verdict }}</template>
+                  <template v-if="s.row.entry.logic_state_at_entry"> · 入场时 {{ s.row.entry.logic_state_at_entry }}</template></div>
+              </template>
+              <span v-else class="muted">{{ (s.row.entry && s.row.entry.why) || '—' }}</span>
+            </template>
+          </el-table-column>
+          <el-table-column label="证据还在不在" min-width="200">
+            <template #default="s">
+              <template v-if="s.row.logic_now && s.row.logic_now.state">
+                <el-tag size="small" effect="dark" :type="logicTag(s.row.logic_now.state)">{{ s.row.logic_now.state }}</el-tag>
+                <div class="muted" :title="s.row.logic_now.text">{{ s.row.logic_now.text }}</div>
+                <div class="muted mono" style="font-size:11px">证据截止 {{ s.row.logic_now.as_of || '—' }}<template
+                  v-if="s.row.logic_now.settle_note && s.row.logic_now.settle_note!=='维持'">；{{ s.row.logic_now.settle_note }}</template></div>
+              </template>
+              <span v-else class="muted">{{ (s.row.logic_now && s.row.logic_now.text) || '今天没有读数' }}</span>
+            </template>
+          </el-table-column>
           <el-table-column label="做T" width="80">
             <template #default="s">
               <el-tag v-if="s.row.t0_enabled" size="small" type="warning" effect="dark">
@@ -2194,6 +2220,8 @@ createApp({
     const stTag = s => ({ EXECUTING: 'primary', DONE: 'success', PARTIAL: 'warning',
       CANCELLED: 'info', EFFECTIVE: 'success', SUPERSEDED: 'info' }[s] || 'warning');
     const cuTag = s => ({ SOLID: 'info', THIN: 'warning', NONE: 'warning' }[s] || 'info');
+    // 逻辑状态四态的标签色 (上游给的中文原文, 直接当键): 存疑红、强化与成立绿、无法判断灰。
+    const logicTag = s => ({ '逻辑存疑': 'danger', '逻辑强化': 'success', '逻辑成立': 'success', '无法判断': 'info' }[s] || 'info');
     const canCancel = r => r.cmd_class === 'task' &&
       ['PENDING', 'PLANNING', 'EXECUTING', 'PARTIAL'].includes(r.status);
     const progPct = p => {
@@ -3577,7 +3605,7 @@ createApp({
     return { tab, loading, err, health, ov, params, catalog, commands, plans, plansOf,
       positions, lots, lotsOf, instructions, ledger, ledgerToday, proposals, report, reportDrawer,
       opsDrawer, opsResult, opsLoading, issuing, form, curSpec, dirtyCount, dm,
-      money, pct, groupLabel, fieldLabel, stTag, cuTag, canCancel, progPct, insPct,
+      money, pct, groupLabel, fieldLabel, stTag, cuTag, logicTag, canCancel, progPct, insPct,
       scaleGap,
       wsRaw, wsAuto, wsHidePong, wsc, wst, wsMode, wsOrders, wsInbox, wsWarnings, loadWs,
       loadAll, loadParams, saveParams, loadPlans, loadLots, onCmdChange, issue,

scripts/run_tests.py

@@ -56,15 +56,19 @@
                         强制入队对卖出同样有效/研判不可用的减持入队/研究证据走弱
                         来源必定交人裁决/风控高置信卖出与命令清仓不经提议分流/
                         到价提议挂着时止损照落并作废该提议                    (17 例)
+  test_batch22_units.py 逻辑状态四态接入 (2026-09-07 第三件): 解析层收逻辑状态/硬数字带它而研判
+                        白名单不收/判决候选而逻辑存疑强制交人/持仓存疑停增持侧/研究走弱
+                        减持默认关且开了必交人/同轮减持优先级/早上取回与映射新鲜度/策略
+                        买入腿按来源暂停恢复/持仓视图两栏三情形/假仓库签名            (14 例)
   test_page_enum_guard.py 页面文案守卫 (静态扫描, 不连库不起浏览器): 枚举字段不许
                         直接印到页面上 / 判据码显示前必须剥前缀 / 不许把整个对象
                         打给交易员看 / 翻译兜底不许让英文码单独当句子   (1 例)
   test_wiring.py        装配自检: 服务层→核心→落表 全链路 (内存桩) +
                         目标价到价必定入队 (档位 full 也不自动卖) +
                         用户设的止损价与目标价单独成列显示                (70 例)
-                                                                  共 661 例
+                                                                  共 675 例
   (总数按实跑逐批相加校正过两次: 曾写 649 是笔误, 实为 650; 09-03 先后加了同轮只发一条
-   减持与研究理由两键各一例, 到 652; 09-04 加了仅展示跳过原因与空候选说明各一例, 到 654; 又加了页面文案守卫一例, 到 655; 09-07 审查修复加了跨轮减持等五例, 到 660; 第二件低把握驳回交人一例, 现为 661)
+   减持与研究理由两键各一例, 到 652; 09-04 加了仅展示跳过原因与空候选说明各一例, 到 654; 又加了页面文案守卫一例, 到 655; 09-07 审查修复加了跨轮减持等五例, 到 660; 第二件低把握驳回交人一例, 到 661; 第三件逻辑状态接入第二十二批十四例, 现为 675)
 任一子集失败即整体失败 (退出码 1)。
 
 哨兵位置清单 (2026-09-03 抄录; 改了对应的东西就得来这些地方改断言, 断言不动就是漏了):
@@ -106,7 +110,7 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_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_batch20_units.py",
-          "test_batch21_units.py",
+          "test_batch21_units.py", "test_batch22_units.py",
           "test_page_enum_guard.py",
           "test_wiring.py"]
 

scripts/test_batch12_units.py

@@ -367,7 +367,11 @@ def _():
     # 2026-09-03 接通「设定某股目标价」后多了一条 EXIT (到价产出清仓提议)。它排在 TRIM 之后、
     # 三类买入之前 —— 顺序有意义: 同轮买卖互斥要先看到减持才让买入让路。四类老动作的相对
     # 次序与判据一个字没动, 减持照旧不送研判。
-    assert [a for a, _ in ae.EVALUATORS] == ["TRIM", "EXIT", "ADD", "FILL", "DCA"]
+    # 2026-09-07 第三件: 研究走弱的减持 (eval_weak_research, 动作名同为 TRIM) 排在保垫减仓前面,
+    # 其余五个的名单与次序一个字不动。
+    assert [a for a, _ in ae.EVALUATORS] == ["TRIM", "TRIM", "EXIT", "ADD", "FILL", "DCA"]
+    assert [f.__name__ for _, f in ae.EVALUATORS] == [
+        "eval_weak_research", "eval_trim", "eval_target", "eval_add", "eval_fill", "eval_dca"]
     assert "EXIT" not in ae.JUDGE_ACTIONS, "减持永远不送研判 (目标价到价也是减持)"
     # 持仓那条扫描的输入与产出一个字没改: 空持仓进去, 空候选出来, 不会去碰候选池
     r = ae.scan(positions=[], params=params(), market={})

scripts/test_batch22_units.py

@@ -0,0 +1,455 @@
+# -*- coding: utf-8 -*-
+"""
+第二十二批模块单测 (逻辑状态四态接入 PMS, 零外部依赖, 不连库不触网)
+==================================================================
+运行: 在 tradingSystem 仓库根目录执行  python scripts/test_batch22_units.py
+
+背景 (2026-09-07 下一阶段方案第三件): 选股系统每早给每只票算一个「支撑它的研究证据还在不在」的
+状态 (逻辑强化 / 逻辑成立 / 无法判断 / 逻辑存疑), 随计划每行下发, 另有按代码查询的接口给持仓票用。
+此前 PMS 收到了却没人读; 持仓不绑入场论点; 「逻辑不证伪不退出」没有判据。本批钉住:
+
+  * 解析层收逻辑状态, 硬数字带它, 但送研判的白名单不收 (一致性检查表第九行: 择时层不判产业逻辑);
+  * 判决候选而逻辑存疑的新建仓强制人工确认 (复用强制确认字段, 与关注判决同一条队列);
+  * 逻辑存疑的持仓停增持侧自主动作, 减持侧照评; 开关关着一行都不执行; 没有读数不拦;
+  * 研究走弱的减持: 默认关, 开了也必定交人 (research_weak 来源一票否决), 数量按可卖量夹紧;
+  * 同轮只发一条减持的优先级: 到价清仓 > 研究走弱 > 保垫减仓;
+  * 早上取回: 取不到写空映射带原因、绝不折成存疑、不动已有暂停; 映射超龄按没有读数;
+  * 策略买入腿按来源暂停与恢复, 不动风控与定性停的;
+  * 持仓视图两栏: 有指令链 / 外部成交并入 / 账本无行 三种情形都说得出话, 仓库读失败页面不塌。
+约定同前: 全过输出 "ALL PASS (n cases)" 退出码 0。
+"""
+import os
+import sys
+import traceback
+from datetime import date, datetime, timedelta
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.dirname(_HERE))
+sys.path.insert(0, _HERE)
+
+from app.core import action_engine as ae                                       # noqa: E402
+from app.services import judge as jd                                           # noqa: E402
+from app.services import plan_feed as pf                                       # noqa: E402
+from app.services import logic_state_service as lss                            # noqa: E402
+
+RESULTS = []
+
+
+def case(name):
+    def deco(fn):
+        RESULTS.append((name, fn))
+        return fn
+    return deco
+
+
+NOW = datetime(2026, 9, 8, 10, 30)
+CODE = "600000.SH"
+TODAY = date.today().isoformat()
+
+
+# ================================================================ 夹具
+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, mod, values):
+    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)
+
+
+def doubt(as_of="2026-09-04"):
+    return {"state": "逻辑存疑", "raw_state": "逻辑存疑", "why": None, "settle_note": "进入逻辑存疑即刻成立",
+            "as_of": as_of, "usable": ["券商行动"], "missing": ["研报论断", "产业研判", "公司事件"],
+            "reasons": ["券商行动的转弱达到进入条件：每股收益预测下修且覆盖机构收缩",
+                        "券商行动（2026-09-04）：每股收益预测中位数下修 40%，同时覆盖机构从 5 家收缩到 3 家"]}
+
+
+def hold():
+    return {"state": "逻辑成立", "raw_state": "逻辑成立", "why": None, "settle_note": "维持",
+            "as_of": "2026-09-04", "usable": ["研报论断"], "missing": ["产业研判", "券商行动", "公司事件"],
+            "reasons": ["研报论断（2026-08-25）：最近一条利好"]}
+
+
+def pos(code=CODE, **kw):
+    p = {"ts_code": code, "total_qty": 3000, "avail_qty": 3000, "price": 10.0, "price_ok": True,
+         "frozen_reason": "NONE", "cushion_pct": 0.02, "cushion_peak": 0.03, "avg_cost": 9.8}
+    p.update(kw)
+    return p
+
+
+SCAN_PARAMS = {"trim_peak": 0.06, "trim_giveback": 0.5, "logic_state_route": True,
+               "logic_doubt_trim": False, "logic_doubt_trim_ratio": 1.0 / 3, "scale": 1_000_000}
+
+PARAMS_BASE = {"PMS_EXEC_WINDOW_TDAYS": 3, "PMS_PROPOSAL_TTL_HOURS": 24,
+               "PMS_JUDGE_TICK_BUDGET_SEC": 150}
+JUDGE_PASS = {"verdict": "PASS", "reason": "理由仍成立", "degraded": False,
+              "raw": {"verdict": "PASS"}, "confidence": 70.0}
+
+
+def _route(c, *, autonomy="propose_only", judge_resp=JUDGE_PASS, dry_run=False, values=None,
+           price=10.0):
+    """把一条候选送进 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": [], "judge_calls": []}
+    with _Patch() as p:
+        p(rule_gate, "check", lambda **kw: {"passed": True, "failed": [], "warnings": []})
+        p(judge, "request", lambda cand_, context=None, **kw: (
+            got["judge_calls"].append(cand_.get("action")) or 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: [])
+        p(pms_repo, "update_position", lambda code, **kw: 1)
+        _patch_params(p, param_store, {**PARAMS_BASE, **(values or {})})
+        out = {"autonomy": autonomy, "open_autonomy": autonomy, "executed": [], "queued": [],
+               "rejected": [], "skipped": [], "errors": [], "degraded": False}
+        held = [{"ts_code": c["ts_code"], "total_qty": 3000, "avail_qty": 3000,
+                 "price": price, "frozen_reason": "NONE"}]
+        view = {"positions": held, "held": held, "sector_ready": False,
+                "params": {}, "totals": {}}
+        psvc._route_one(c, view, {"_mkt": {}}, {}, False, NOW, dry_run, out)
+    return out, got
+
+
+# ================================================================ 一, 解析层与硬数字
+@case("解析层·逻辑状态按约定的键归一, 出处最多三条; 缺键 / 非字典 / 没有 state 一律 None")
+def _():
+    raw = {"state": "逻辑存疑", "raw_state": "逻辑存疑", "why": None, "settle_note": "进入逻辑存疑即刻成立",
+           "as_of": "2026-09-04", "usable": ["券商行动"], "missing": ["研报论断", "公司事件"],
+           "reasons": ["a", "b", "c", "d"], "paths": [{"path": "x"}], "prev_state": "逻辑成立"}
+    st = pf._logic_state_or_none(raw)
+    assert st["state"] == "逻辑存疑" and st["as_of"] == "2026-09-04", st
+    assert st["reasons"] == ["a", "b", "c"] and st["usable"] == ["券商行动"], st
+    assert "paths" not in st and "prev_state" not in st, st          # 只收约定的键
+    assert pf._logic_state_or_none(None) is None and pf._logic_state_or_none("逻辑存疑") is None
+    assert pf._logic_state_or_none({"why": "x"}) is None
+    # 没有原始态的旧形状: 原始态退回落定态
+    assert pf._logic_state_or_none({"state": "逻辑成立"})["raw_state"] == "逻辑成立"
+    row = pf._rows([{"code": "600000.SH", "rank": 1, "score": 210, "logic_state": raw}], "main")[0]
+    assert row["logic_state"]["state"] == "逻辑存疑", row
+    row2 = pf._rows([{"code": "600000.SH", "rank": 1, "score": 210}], "main")[0]
+    assert row2["logic_state"] is None, row2
+
+
+@case("解析层·按代码查询接口的应答映射回 PMS 的代码形态, 带 error 的行跳过, source 与日期带上")
+def _():
+    payload = {"date": "2026-09-04", "count": 3, "states": [
+        {"input": "600000.SH", "code": "SH600000", "source": "daily", **doubt()},
+        {"input": "300750", "code": "SZ300750", "source": "computed", **hold()},
+        {"input": "abc", "error": "认不出的代码形态"}]}
+    m = pf.parse_logic_states(payload)
+    assert set(m) == {"600000.SH", "300750.SZ"}, m
+    assert m["600000.SH"]["state"] == "逻辑存疑" and m["600000.SH"]["source"] == "daily"
+    assert m["300750.SZ"]["source"] == "computed" and m["300750.SZ"]["date"] == "2026-09-04"
+    assert pf.parse_logic_states(None) == {} and pf.parse_logic_states({"states": "x"}) == {}
+    assert pf.fetch_logic_states([]) == {}                              # 没票不触网
+
+
+@case("候选筛选·items 带 logic_state (缺就是 None), 不改资格也不改次序")
+def _():
+    plan = {"date": "2026-09-04", "main": pf._rows([
+        {"code": "600000.SH", "rank": 1, "score": 220, "verdict": "候选", "logic_state": doubt()},
+        {"code": "300750.SZ", "rank": 2, "score": 210, "verdict": "候选"}], "main"), "observe": []}
+    sel = pf.select_candidates(plan, top_n=10, route_by_verdict=True)
+    items = sel["items"]
+    assert [x["ts_code"] for x in items] == ["600000.SH", "300750.SZ"], items
+    assert items[0]["logic_state"]["state"] == "逻辑存疑" and items[1]["logic_state"] is None
+
+
+@case("硬数字带逻辑状态, 而送研判的白名单不收它 (择时层不判产业逻辑, 检查表第九行)")
+def _():
+    c = {"ts_code": CODE, "price": 10.0, "score": 220, "rank": 1, "bucket": "main", "src": "plan_api",
+         "verdict": "候选", "basis": "三门槛全过", "logic": ["研报说好"], "logic_state": doubt()}
+    params = {"scale": 1_000_000, "stock_target_default": 0.06, "batch_split": (0.5, 0.25, 0.25)}
+    with _Patch() as p:
+        p(ae, "check_all_caps", lambda **kw: [])
+        p(ae, "_new_name_ctx", lambda caps, c: caps)
+        cand, why = ae.eval_open(c, params, {"names_count": 0, "max_names": 10}, 500_000)
+    assert cand and not why, why
+    hn = cand["hard_numbers"]
+    assert hn["logic_state"]["state"] == "逻辑存疑" and hn["basis"] == "三门槛全过", hn
+    sent = jd._judge_hard_numbers("OPEN", hn)
+    assert "logic_state" not in sent and "basis" not in sent and "verdict" in sent, sent
+    assert "logic_state" not in jd.OPEN_JUDGE_KEYS
+
+
+# ================================================================ 二, 新建仓与持仓的分流
+def _scan_open(cands, *, route_logic=True):
+    params = {"scale": 1_000_000, "stock_target_default": 0.06, "batch_split": (0.5, 0.25, 0.25),
+              "open_route_by_verdict": True, "open_route_by_logic": route_logic,
+              "open_signal_priority": False}
+    with _Patch() as p:
+        p(ae, "check_all_caps", lambda **kw: [])
+        p(ae, "_new_name_ctx", lambda caps, c: caps)
+        p(ae, "_ctx_after", lambda ctx, *a, **kw: ctx)
+        return ae.scan_open(candidates=cands, params=params,
+                            caps={"names_count": 0, "max_names": 10}, room_amt=900_000, slots=5)
+
+
+@case("新建仓·判决候选而逻辑存疑 → 强制人工确认并写明出处; 关注保留自己的原因; 成立不强制; 开关关不强制")
+def _():
+    base = {"price": 10.0, "score": 220, "rank": 1, "bucket": "main", "src": "plan_api"}
+    cands = [{**base, "ts_code": "600000.SH", "verdict": "候选", "logic_state": doubt()},
+             {**base, "ts_code": "600001.SH", "verdict": "关注", "logic_state": doubt()},
+             {**base, "ts_code": "600002.SH", "verdict": "候选", "logic_state": hold()},
+             {**base, "ts_code": "600003.SH", "verdict": "候选"}]
+    r = _scan_open(cands)
+    by = {c["ts_code"]: c for c in r["candidates"]}
+    assert set(by) == {"600000.SH", "600001.SH", "600002.SH", "600003.SH"}, r["skipped"]
+    a = by["600000.SH"]
+    assert a["needs_user_confirm"] and "逻辑存疑" in a["confirm_why"] and "证据截止 2026-09-04" in a["confirm_why"], a
+    assert "每股收益预测下修" in a["confirm_why"], a["confirm_why"]
+    b = by["600001.SH"]
+    assert b["needs_user_confirm"] and b["confirm_why"] == ae.WHY_WATCH_CONFIRM, b
+    assert not by["600002.SH"].get("needs_user_confirm") and not by["600003.SH"].get("needs_user_confirm")
+    r2 = _scan_open(cands, route_logic=False)
+    by2 = {c["ts_code"]: c for c in r2["candidates"]}
+    assert not by2["600000.SH"].get("needs_user_confirm"), by2["600000.SH"]
+    assert ae.logic_confirm_why(hold()) is None and ae.logic_confirm_why(None) is None
+
+
+@case("持仓·逻辑存疑停增持侧 (补足/加仓/补仓写明原因), 减持侧照评; 开关关不停; 没有读数不停")
+def _():
+    p = pos(logic_state=doubt(), cushion_peak=0.10, cushion_pct=0.04)   # 保垫回吐过半 → TRIM 照评
+    r = ae.scan(positions=[p], params=SCAN_PARAMS, market={CODE: {}}, skip={})
+    acts = {c["action"] for c in r["candidates"]}
+    assert acts == {"TRIM"}, r["candidates"]
+    stopped = {s["action"]: s["why"] for s in r["skipped"] if "逻辑存疑" in s["why"]}
+    assert set(stopped) == {"ADD", "FILL", "DCA"}, r["skipped"]
+    assert all("停掉增持侧" in w and "证据截止 2026-09-04" in w for w in stopped.values()), stopped
+    r2 = ae.scan(positions=[pos(logic_state=doubt())], params={**SCAN_PARAMS, "logic_state_route": False},
+                 market={CODE: {}}, skip={})
+    assert not [s for s in r2["skipped"] if "逻辑存疑" in s["why"]], r2["skipped"]
+    r3 = ae.scan(positions=[pos()], params=SCAN_PARAMS, market={CODE: {}}, skip={})
+    assert not [s for s in r3["skipped"] if "逻辑存疑" in s["why"]], r3["skipped"]
+
+
+@case("研究走弱的减持·默认关不产出; 开了按三分之一取整并按可卖量夹紧, 来源 research_weak 且强制确认; 不足一手不产出")
+def _():
+    p = pos(logic_state=doubt())
+    assert ae.eval_weak_research(p, SCAN_PARAMS) is None                       # 开关关
+    on = {**SCAN_PARAMS, "logic_doubt_trim": True}
+    c = ae.eval_weak_research(p, on)
+    assert c and c["action"] == "TRIM" and c["side"] == "sell" and c["qty"] == 1000, c
+    assert c["source"] == ae.SRC_RESEARCH_WEAK and c["needs_user_confirm"] and c["confirm_why"], c
+    assert "拟减 1000 股" in c["reason"] and "证据截止 2026-09-04" in c["reason"], c["reason"]
+    assert c["hard_numbers"]["logic_state"]["state"] == "逻辑存疑" and c["hard_numbers"]["trim_ratio"] == 0.3333
+    c2 = ae.eval_weak_research(pos(logic_state=doubt(), avail_qty=400), on)
+    assert c2["qty"] == 400, c2                                                # 夹到可卖量
+    assert ae.eval_weak_research(pos(logic_state=doubt(), total_qty=200, avail_qty=200), on) is None
+    assert ae.eval_weak_research(pos(logic_state=hold()), on) is None
+    assert ae.eval_weak_research(pos(), on) is None                            # 没有读数
+    c3 = ae.eval_weak_research(p, {**on, "logic_doubt_trim_ratio": 0.5})
+    assert c3["qty"] == 1500, c3
+
+
+@case("同轮只发一条减持·优先级 到价清仓 > 研究走弱 > 保垫减仓, 让路的记进跳过原因")
+def _():
+    on = {**SCAN_PARAMS, "logic_doubt_trim": True}
+    p = pos(logic_state=doubt(), cushion_peak=0.10, cushion_pct=0.04)
+    r = ae.scan(positions=[p], params=on, market={CODE: {}}, skip={},
+                stock_params={CODE: {"target_price": 9.5}})
+    assert [c["action"] for c in r["candidates"]] == ["EXIT"], r["candidates"]
+    yielded = [s for s in r["skipped"] if "一轮只发一条减持" in s["why"]]
+    assert len(yielded) == 2 and all(s["action"] == "TRIM" for s in yielded), yielded
+    r2 = ae.scan(positions=[p], params=on, market={CODE: {}}, skip={})
+    assert len(r2["candidates"]) == 1 and r2["candidates"][0]["source"] == ae.SRC_RESEARCH_WEAK, r2
+    assert ae._sell_priority({"action": "EXIT"}) < ae._sell_priority({"action": "TRIM", "source": "research_weak"}) \
+        < ae._sell_priority({"action": "TRIM", "source": "engine"})
+
+
+@case("分流·研究走弱的减持在 full 档也入人工队列, 不落指令, 提议带来源与逻辑状态")
+def _():
+    c = ae.eval_weak_research(pos(logic_state=doubt()), {**SCAN_PARAMS, "logic_doubt_trim": True})
+    for autonomy in ("propose_only", "full"):
+        out, got = _route(c, autonomy=autonomy)
+        assert not out["executed"] and len(out["queued"]) == 1, (autonomy, out)
+        assert "研究证据走弱" in out["queued"][0]["why"], out["queued"]
+        assert not got["instructions"] and len(got["proposals"]) == 1, (autonomy, got)
+        hn = got["proposals"][0]["hard_numbers"]
+        assert hn["source"] == ae.SRC_RESEARCH_WEAK and hn["logic_state"]["state"] == "逻辑存疑", hn
+
+
+# ================================================================ 三, 早上取回与策略买入腿
+@case("映射·超过三个自然日没刷新按没有读数; 时刻缺失或坏了也按没有读数; attach 只给有读数的行挂键")
+def _():
+    fresh = {"at": f"{TODAY} 08:41:00", "states": {CODE: doubt()}}
+    assert lss.state_map(fresh) == {CODE: doubt()}
+    old = {"at": (date.today() - timedelta(days=5)).isoformat() + " 08:41:00", "states": {CODE: doubt()}}
+    assert lss.state_map(old) == {}
+    assert lss.state_map({"states": {CODE: doubt()}}) == {} and lss.state_map({}) == {}
+    rows = [pos(), pos("300750.SZ", logic_state="陈旧的")]
+    lss.attach(rows, {CODE: doubt()})
+    assert rows[0]["logic_state"]["state"] == "逻辑存疑" and "logic_state" not in rows[1], rows
+
+
+@case("策略买入腿·存疑按来源 logic 暂停; 明确不存疑只清本来源; 没读数不动; 开关关整段跳过")
+def _():
+    from app.services import param_store, strategy_service
+    calls = {"pause": [], "clear": []}
+    with _Patch() as p:
+        p(strategy_service, "pause_buy", lambda code, *, reason="", source="signal": (
+            calls["pause"].append((code, source, reason)) or ["S1"]))
+        p(strategy_service, "clear_buypause", lambda code, only_source=None: (
+            calls["clear"].append((code, only_source)) or {"ok": True, "cleared": code == "600001.SH"}))
+        _patch_params(p, param_store, {"PMS_LOGIC_STATE_ROUTE": True})
+        r = lss.apply_pauses({CODE: doubt(), "600001.SH": hold(), "600002.SH": hold()},
+                             [CODE, "600001.SH", "600002.SH", "600003.SH"])
+    assert r["paused"] == [CODE] and r["resumed"] == ["600001.SH"] and not r["errors"], r
+    assert calls["pause"] == [(CODE, "logic", calls["pause"][0][2])] and "停掉增持侧" in calls["pause"][0][2]
+    assert [c for c, _ in calls["clear"]] == ["600001.SH", "600002.SH"], calls   # 没读数的 600003 不动
+    assert all(s == "logic" for _, s in calls["clear"]), calls
+    with _Patch() as p:
+        p(strategy_service, "pause_buy", lambda *a, **k: (_ for _ in ()).throw(AssertionError("不该调")))
+        p(strategy_service, "clear_buypause", lambda *a, **k: (_ for _ in ()).throw(AssertionError("不该调")))
+        _patch_params(p, param_store, {"PMS_LOGIC_STATE_ROUTE": False})
+        r2 = lss.apply_pauses({CODE: doubt()}, [CODE])
+    assert r2["paused"] == [] and r2.get("skipped"), r2
+
+
+@case("早上取回·查不到写空映射带原因且不动暂停 (绝不折成存疑); 查到写映射并按结果暂停")
+def _():
+    from app.repo import pms_repo
+    from app.services import param_store, strategy_service
+    saved, calls = {}, {"pause": [], "clear": []}
+    with _Patch() as p:
+        p(pms_repo, "list_positions", lambda *, only_open=False: [{"ts_code": CODE}, {"ts_code": "600001.SH"}])
+        p(param_store, "set_param", lambda k, v, by="user": saved.__setitem__(k, v) or {"ok": True})
+        _patch_params(p, param_store, {"PMS_LOGIC_STATE_ROUTE": True})
+        p(strategy_service, "pause_buy", lambda code, *, reason="", source="signal": calls["pause"].append(code) or [])
+        p(strategy_service, "clear_buypause", lambda code, only_source=None: calls["clear"].append(code) or {"ok": True, "cleared": False})
+
+        def _boom(codes):
+            raise pf.PlanFeedError("选股系统连不上")
+        r = lss.pull_for_held(now=NOW, fetch=_boom)
+        import json
+        m = json.loads(saved[lss.MAP_KEY])
+        assert r["ok"] is False and m["states"] == {} and "连不上" in m["error"], (r, m)
+        assert not calls["pause"] and not calls["clear"], calls
+
+        r2 = lss.pull_for_held(now=NOW, fetch=lambda codes: {CODE: {**doubt(), "date": "2026-09-04"}})
+        m2 = json.loads(saved[lss.MAP_KEY])
+        assert r2["ok"] and r2["got"] == 1 and r2["by_state"] == {"逻辑存疑": 1}, r2
+        assert m2["date"] == "2026-09-04" and m2["states"][CODE]["state"] == "逻辑存疑", m2
+        assert r2["paused"] == [CODE] and r2["missing"] == ["600001.SH"], r2   # 没查到的票记明, 不动
+
+
+# ================================================================ 四, 持仓视图两栏
+def _fake_repo():
+    from test_wiring import FakeRepo
+    f = FakeRepo()
+    f.insert_instruction(instruction_id="I1", origin_type="proposal", origin_id="P1", ts_code=CODE,
+                         action="OPEN", side="buy", qty=1000)
+    f.insert_ledger(ts_code=CODE, action="OPEN", arbiter="user", verdict="PASS", price_at=10.0,
+                    hard_numbers={"basis": "三门槛全过、无硬风险", "logic": ["研报说好 —— 出处 A", "B", "C", "D"],
+                                  "verdict": "候选", "logic_state": hold()},
+                    ref_id="P1", reason="人工采纳: 看好")
+    f.insert_lot(ts_code=CODE, lot_type="BASE", qty=1000, open_price=10.0, open_date="2026-09-01",
+                 instruction_id="I1")
+    f.insert_lot(ts_code="600001.SH", lot_type="BASE", qty=1000, open_price=10.0, open_date="2026-09-01")
+    f.insert_instruction(instruction_id="I2", origin_type="command", origin_id="C1", ts_code="600002.SH",
+                         action="OPEN", side="buy", qty=1000)
+    f.insert_lot(ts_code="600002.SH", lot_type="BASE", qty=1000, open_price=10.0, open_date="2026-09-02",
+                 instruction_id="I2")
+    return f
+
+
+@case("入场论点·有指令链 (批次→指令→提议号→账本) / 外部成交并入 / 账本无行 三种情形都说得出话")
+def _():
+    from app.repo import pms_repo
+    f = _fake_repo()
+    with _Patch() as p:
+        p(pms_repo, "list_lots", f.list_lots)
+        p(pms_repo, "get_instruction", f.get_instruction)
+        p(pms_repo, "ledger_by_ref", f.ledger_by_ref)
+        a = lss.entry_view(CODE)
+        b = lss.entry_view("600001.SH")
+        c = lss.entry_view("600002.SH")
+        d = lss.entry_view("600009.SH")
+    assert a["basis"] == "三门槛全过、无硬风险" and a["logic"] == ["研报说好 —— 出处 A", "B", "C"], a
+    assert a["verdict"] == "候选" and a["logic_state_at_entry"] == "逻辑成立" and a["open_date"] == "2026-09-01", a
+    assert a["instruction_id"] == "I1" and a["arbiter"] == "user", a
+    assert "外部成交并入" in b["why"] and b["open_date"] == "2026-09-01", b
+    assert "账本里没有" in c["why"] and c["instruction_id"] == "I2", c
+    assert "没有未平的批次" in d["why"], d
+    assert f.ledger_by_ref(["P1", "X"]) and not f.ledger_by_ref([]), "假仓库的按引用取账本"
+
+
+@case("持仓接口两栏·仓库读失败也不抛, 每行都有 entry 与 logic_now; 没有读数写明; 空仓行 entry 为空")
+def _():
+    from app.repo import pms_repo
+    rows = [pos(), pos("600001.SH"), pos("600002.SH", total_qty=0)]
+    with _Patch() as p:
+        p(pms_repo, "list_lots", lambda *a, **k: (_ for _ in ()).throw(OSError("db down")))
+        lss.decorate_positions(rows, {CODE: doubt()})
+    assert rows[0]["logic_now"]["state"] == "逻辑存疑" and "每股收益预测下修" in rows[0]["logic_now"]["text"]
+    assert rows[0]["logic_now"]["as_of"] == "2026-09-04" and "批次读取失败" in rows[0]["entry"]["why"], rows[0]
+    assert rows[1]["logic_now"]["state"] is None and "没有读数" in rows[1]["logic_now"]["text"], rows[1]
+    assert rows[2]["entry"] is None, rows[2]
+    nv = lss.now_view(hold())
+    assert nv["state"] == "逻辑成立" and nv["settle_note"] == "维持" and "研报论断" in nv["text"], nv
+
+
+@case("假仓库·ledger_by_ref 与真 repo 同名同签名 (第十批 [M1] 也会扫, 这里先钉一次)")
+def _():
+    import ast
+    import inspect
+    from test_wiring import FakeRepo
+    path = os.path.join(os.path.dirname(_HERE), "app", "repo", "pms_repo.py")
+    with open(path, encoding="utf-8") as fh:
+        names = {n.name: [a.arg for a in n.args.args] for n in ast.parse(fh.read()).body
+                 if isinstance(n, ast.FunctionDef)}
+    assert names["ledger_by_ref"] == ["ref_ids", "limit"], names["ledger_by_ref"]
+    fp = list(inspect.signature(FakeRepo.ledger_by_ref).parameters)
+    assert fp == ["self", "ref_ids", "limit"], fp
+
+
+# ================================================================ 入口
+def main():
+    ok = 0
+    for name, fn in RESULTS:
+        try:
+            fn()
+            ok += 1
+            print(f"  ok  {name}")
+        except Exception:
+            print(f"  FAIL {name}")
+            traceback.print_exc()
+    print("-" * 60)
+    if ok == len(RESULTS):
+        print(f"ALL PASS ({ok} cases)")
+        return 0
+    print(f"FAILED {len(RESULTS) - ok}/{len(RESULTS)}")
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/test_wiring.py

@@ -314,6 +314,10 @@ class FakeRepo:
     def list_ledger(self, *, ts_code=None, limit=200):
         return self.ledger[-limit:]
 
+    def ledger_by_ref(self, ref_ids, limit=50):
+        ids = {str(x) for x in (ref_ids or []) if x}
+        return [r for r in self.ledger if str(r.get("ref_id")) in ids][:limit]
+
     def rule_rejected_today(self, since):
         # 内存桩里所有留痕都算"今天"; 只认规则闸的 REJECT (研判结论会变, 不参与当日去重)
         return {(r["ts_code"], r["action"]) for r in self.ledger