每日计划：记录消费顺序并核验提议依据（台账 018）

一、让候选顺序传到建仓。记录计划版本、有效参数、过滤与信号优先原因，保留实际消费历史。

二、采纳前检查依据。材料或组合依据变化时重走现有研判与数量流程，旧确认返回更新内容；条件更新防止重复采纳。新增控制默认关闭，保留所有现有参数。

三、验证与上线。隔离离线全量测试及三道页面守卫通过，新批次 15 例；先模拟部署，生产观察另记。

DEVLOG.md

@@ -1803,3 +1803,17 @@ docs/交接_2026-09-20.md（新）、docs/dev_preview/mock_preview.py 与 sample
 ## YYYY-MM-DD · 一句话标题
 **做了什么** / **动了哪些文件** / **部署方式** / **真机判收** / **还欠着什么**
 -->
+
+## 2026-10-15 · 每日计划版本贯通、消费顺序和提议重新核验
+
+**做了什么。** 按用户批准的跨仓方案落本地四批实现；旧阈值与运行开关未改。候选消费记录分三类，执行保留消费顺序与信号调整原因；提议核验与人工再次确认通过条件更新保护并发；报告链接和两端研究展示贯通。
+
+**动了哪些文件。** plan_feed、action_engine、plan_revision、plan_recheck、proposal_service、command_service、planner、executor、pms_repo、company_report、logic_state_service、Web 入口及页面；新控制 PMS_PLAN_VERSION_RECHECK 默认 False；新增第 38 批 15 例并登记总入口。
+
+**本地验证。** 配置隔离且禁止联网的全套入口 ALL SUITES PASS，包含新增第 38 批 15 例；三个页面守卫通过。浏览器只连接 127.0.0.1 假数据页面，看到研究信息与过期内容重新确认提示，没有连接交易通道。
+
+**部署方式。** 先兼容读与留痕，再桥发布，最后模拟实例启用修订与核验。实际机器、表前缀和容器需部署前确认，容器操作需用户同意。
+
+**真机判收。** 未判收；本轮未连接生产机器，未改运行参数，未生成任何真实指令。
+
+**还欠着什么。** 获准后的部署、真实参数及实例身份核对、十二分钟传播时延和连续五个交易日模拟观察。详见四仓同步规格书及交接。

UPSTREAM_PLAN_API.md

@@ -1,5 +1,7 @@
 # 上游选股计划接口 · 接入记录与待确认口径
 
+当前开发契约见本文末尾“2026-10-15 补充：完整决策版本契约”；下文早期接入实测保留历史语义。
+
 > 2026-07-30 建。上游「选股系统」提供 HTTP 计划接口, PMS 的升仓/建仓候选池自此改吃它,
 > 不再读 `trading_buy_plan`。本文是接口现状 + PMS 侧接法 + **需要上游回答的问题**。
 > 代码入口: `app/services/plan_feed.py`; 单测: `scripts/test_batch7_units.py` (23 例);
@@ -672,3 +674,17 @@ make changes                      # 页面同款结论
 **候选按上游名次排 (自门控)。** 计划顶层 `rank_axis` 为 quality 时, `select_candidates` 按上游名次升序排、关掉 PMS 自己的基本面立场分桶 (参数 `PMS_PLAN_RANK_BY_UPSTREAM`, 默认开); 上游没带这个键时无作用, 逐字回到按立场分桶排。这样选股序、入池、PMS 候选三处同一个序。本次用哪种序记在候选返回与名册快照元数据的 `rank_mode` 里 (upstream / quality_bucket / score)。
 
 **前缀效应 —— 强传导票变少不是故障。** 质地序把好公司排到前面, 质地差但强传导的票会往后掉, 可能落到我们要的三百只主榜前缀之外, 于是这批票在候选里变少甚至消失。这与"质地差降关注"同向, 是预期结果, 不要当成上游漏发或接口出错。要看全量强传导票, 去上游的逐票对账接口 `/plan/verdict` (它的"主榜第 N 名"仍按原始分序, 应答带 `rank_axis: transmission` 说明), 或把 `PMS_PLAN_RANK_BY_UPSTREAM` 关掉临时回到按分桶排。
+# 2026-10-15 补充：完整决策版本契约
+
+本节为当前开发实现；下文历史实测里“实时重装配、没有版本戳”的描述不再用于新版。
+部署状态见 `docs/交接_每日选股计划完善_2026-10-15.md`，本地实现不等于已经上线。
+
+- 顶层新增 schema_version=2、decision、traceability；plan_version 仍为代码版本。
+- decision 包括 id、parent_id、cutoff_at、input_hash、params_hash、reason；时间带时区。
+- 计划行新增全量 research_rank、research、review_ref。展示裁剪不改变全量名次。
+- GET /plan、/plan/verdict、/logic_state 可指定 decision_id；指定日期或版本只读发布内容。缺档不现算。
+- 最新查询带 refresh_status；pending_codes 为已知待修订或来源核验失败的股票，暂停新增自主提议。
+- 评析报告 report_version_url 引用不可变对象；旧数据缺引用明确标示，不能用最新报告替代历史。
+- PMS meta.kind 区分 received、consumed、rechecked；消费记录包括真实顺序、有效参数和过滤原因。
+- 采纳核验由 PMS_PLAN_VERSION_RECHECK 控制，默认关闭；内容更新时接口返回 ok=false、原因及更新后的 proposal，页面必须重新确认其 proposal_revision。
+- 旧消费者可忽略新增字段，新消费者接受旧字段为空；无版本旧数据不能通过已启用的自主采纳核验。

app/core/action_engine.py

@@ -858,6 +858,8 @@ def eval_open(c: dict, params: dict, caps: dict, room_amt: float):
         "tier": c.get("tier"), "upside": c.get("upside"), "heat": c.get("heat"),
         "plan_rank": c.get("rank"), "plan_bucket": c.get("bucket"),
         "plan_src": c.get("src"), "sector": c.get("sector"),
+        **{k: c.get(k) for k in ('decision', 'consumption_id', 'consumption_error',
+            'plan_basis', 'selection_rank', 'research_rank', 'research', 'review_ref')},
         # 决策系统今天盘中判过这只票转多没有。这是**定性材料**不是仓位数字, 所以送研判时
         # 也带过去 (见 judge.OPEN_JUDGE_KEYS) —— 等于当面告诉它「你自己今天判过这只票转多」,
         # 它可以拿自己的结论去对照; 没有信号时这两项是 None, 不误导。
@@ -972,13 +974,18 @@ def scan_open(*, candidates: list, params: dict, caps: dict, room_amt: float,
     # 候选池的产物), 候选层那一整套过滤一道都不绕。
     # PMS_OPEN_SIGNAL_PRIORITY 关掉即退回纯分数排序, 页面可改, 不用改码。
     sig_first = bool(params.get("open_signal_priority", True))
-    for c in sorted(candidates or [],
+    for execution_rank, c in enumerate(sorted(candidates or [],
                     key=lambda x: (0 if (sig_first and x.get("sig_buy")) else 1,
-                                   -_f(x.get("score")), _f(x.get("rank"), 10 ** 9),
+                                   (0, x['selection_rank']) if x.get('selection_rank') is not None
-                                   str(x.get("ts_code") or ""))):
+                                   else (1, -_f(x.get('score'))),
+                                   _f(x.get("rank"), 10 ** 9),
+                                   str(x.get("ts_code") or ""))), 1):
         code = c.get("ts_code")
         if not code:
             continue
+        if code in ((c.get('refresh_status') or {}).get('pending_codes') or []) or c.get('consumption_error'):
+            skipped.append({'ts_code': code, 'action': A_OPEN, 'why': '计划修订或消费留痕尚未完成，暂停新的自主建仓'})
+            continue
         if route and c.get("verdict") == VERDICT_DISPLAY:
             # 仅展示的行不占名额、不占金额, 也不进后面任何一道闸; 放在名额判断之前,
             # 是为了让它的跳过原因永远是「仅展示」而不是碰巧的「名额用完」。
@@ -1043,6 +1050,9 @@ def scan_open(*, candidates: list, params: dict, caps: dict, room_amt: float,
             continue
         try:
             cand.setdefault("hard_numbers", {})["advice"] = adv
+            cand['hard_numbers']['execution_rank'] = execution_rank
+            cand['hard_numbers']['priority_reason'] = ('盘中转多信号优先' if sig_first and c.get('sig_buy')
+                                                       else '继承候选消费顺序')
             cand["hard_numbers"]["advice_text"] = adv.get("text") if isinstance(adv, dict) else None
             cand["hard_numbers"]["plan_by_advice"] = bool(params.get("plan_by_advice"))
             if isinstance(adv, dict) and adv.get("tier") == "试探仓":

app/core/plan_revision.py

@@ -0,0 +1,45 @@
+"""计划消费与提议版本核验的纯函数；不改变准入、仓位或自然语言判断。"""
+import hashlib
+import json
+
+
+def fingerprint(value):
+    return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True,
+        separators=(',', ':'), default=str).encode()).hexdigest()
+
+
+def basis(row):
+    # 排名与审计时间不是资格/风险/数量依据。完整评析引用变化需要重新研判。
+    return {k: row.get(k) for k in ('score', 'tier', 'upside', 'heat', 'theme',
+        'verdict', 'risk', 'missing', 'reasons', 'logic_state', 'company_gate',
+        'company_review', 'company_review_text', 'events', 'pricing_state', 'news')}
+
+
+def portfolio_context(view, commitments, stock=None):
+    """持仓/资金/在途承诺变化使旧数量依据失效；价格漂移仍由执行终检管理。"""
+    totals = view.get('totals') or {}
+    return fingerprint({'held': sorted((r['ts_code'], r.get('total_qty'), r.get('sector'))
+                                      for r in view.get('held') or []),
+                        'cash_source': totals.get('cash_source'), 'cash_avail': totals.get('cash_avail'),
+                        'sector_ready': view.get('sector_ready'), 'stock': stock or {},
+                        'inflight': {'names': sorted(commitments.get('names') or []),
+                                     'amount': commitments.get('amount')}})
+
+
+def compare(old, current, code):
+    """返回 unavailable/material/unchanged。资格失效由调用方通过缺候选表达。"""
+    if not current or not (current.get('decision') or {}).get('id'):
+        return 'unavailable'
+    state = current.get('refresh_status') or {}
+    if code in (state.get('pending_codes') or []) or current.get('consumption_error'):
+        return 'unavailable'
+    if old is None or fingerprint(old) != fingerprint(basis(current)):
+        return 'material'
+    return 'unchanged'
+
+
+def needs_new_judgement(hard, previous):
+    """决策系统缓存按股票/动作，缺同依据的既有研判就必须要求重新回答。"""
+    keys = ('plan_basis', 'scan_params_hash', 'portfolio_context_hash')
+    return (not previous or any(previous.get(k) is None for k in keys)
+            or any(fingerprint(hard.get(k)) != fingerprint(previous.get(k)) for k in keys))

app/core/planner.py

@@ -303,7 +303,7 @@ def plan_increase_exposure(*, add_amount: float, positions: list, candidates=Non
         ctx = _ctx_after(ctx, amt, is_new_name=False, sector=p.get("sector"))
 
     # ---- ② 候选池新票建仓 ----
-    for c in sorted(candidates or [], key=lambda x: _sortable(x, "score", reverse=True)):
+    for execution_rank, c in enumerate(sorted(candidates or [], key=lambda x: _sortable(x, "score", reverse=True)), 1):
         if acc >= target:
             break
         price = float(c.get("price") or 0)
@@ -328,8 +328,11 @@ def plan_increase_exposure(*, add_amount: float, positions: list, candidates=Non
         if bad:
             rejects.append({"ts_code": c.get("ts_code"), "reasons": bad})
             continue
-        items.extend(_open_batch_items(c.get("ts_code"), sp, price,
+        batches = _open_batch_items(c.get("ts_code"), sp, price, reason_prefix="升仓建新仓")
-                                       reason_prefix="升仓建新仓"))
+        for item in batches:
+            item['plan_provenance'] = dict(c.get('plan_provenance') or {}, execution_rank=execution_rank,
+                                           priority_reason='用户升仓命令，沿用命令调配次序')
+        items.extend(batches)
         intended += want_amt
         acc += actual
         ctx = _ctx_after(ctx, actual, is_new_name=True, sector=c.get("sector"))

app/repo/pms_repo.py

@@ -513,10 +513,20 @@ def get_proposal(proposal_id: str):
     return r
 
 
-def decide_proposal(proposal_id: str, status: str) -> int:
+def revise_pending_proposal(old, candidate, verdict):
+    return execute('UPDATE pms_proposal SET qty=:qty, hard_numbers_json=:hn, '
+                   'judge_verdict=:jv, judge_reason=:jr WHERE proposal_id=:pid '
+                   "AND status='WAIT_USER' AND hard_numbers_json=:old",
+                   {'qty': candidate['qty'], 'hn': _dumps(candidate['hard_numbers']),
+                    'jv': verdict.get('verdict'), 'jr': (verdict.get('reason') or '')[:500],
+                    'pid': old['proposal_id'], 'old': old['hard_numbers_json']})
+
+
+def decide_proposal(proposal_id: str, status: str, expected_hard=None) -> int:
     return execute("UPDATE pms_proposal SET status = :st, decided_at = :ts "
-                   "WHERE proposal_id = :pid AND status = 'WAIT_USER'",
+                   "WHERE proposal_id = :pid AND status = 'WAIT_USER'" +
-                   {"st": status, "pid": proposal_id, "ts": _NOW()})
+                   (' AND hard_numbers_json = :expected' if expected_hard is not None else ''),
+                   {"st": status, "pid": proposal_id, "ts": _NOW(), 'expected': expected_hard})
 
 
 def expire_proposals(now=None) -> int:
@@ -811,9 +821,14 @@ def upsert_industry(rows: list) -> int:
 
 
 # ================================================================ pms_plan_snapshot
-def latest_plan_digest():
+def latest_plan_digest(kind=None):
     """最新一份快照的 (plan_date, digest); 没有则 None。判"这份榜变没变"只该拿它比。"""
-    r = fetch_one("SELECT plan_date, digest FROM pms_plan_snapshot ORDER BY id DESC LIMIT 1")
+    if kind:
+        r = fetch_one("SELECT plan_date, digest FROM pms_plan_snapshot "
+                      "WHERE JSON_UNQUOTE(JSON_EXTRACT(meta_json, '$.kind')) = :kind "
+                      "ORDER BY id DESC LIMIT 1", {'kind': kind})
+    else:
+        r = fetch_one("SELECT plan_date, digest FROM pms_plan_snapshot ORDER BY id DESC LIMIT 1")
     return (r["plan_date"], r["digest"]) if r else None
 
 
@@ -837,7 +852,8 @@ def insert_plan_snapshot(*, plan_date, digest, roster, meta=None, n_main=0, n_ob
     不产生任何假信号), 换掉的是一整类方向错误, 这笔交换是划算的。
     """
     now = fetched_at or _NOW()
-    cur = latest_plan_digest()
+    kind = (meta or {}).get('kind')
+    cur = latest_plan_digest(kind) if kind else latest_plan_digest()
     if cur and cur[0] == str(plan_date) and cur[1] == str(digest):
         return 0
     execute(
@@ -857,7 +873,8 @@ def latest_plan_snapshots(limit: int = 2) -> list:
     rows = fetch_all(
         "SELECT id, plan_date, digest, fetched_ymd, fetched_at, n_main, n_observe, "
         "       capped_main, capped_obs, roster_json, meta_json "
-        "FROM pms_plan_snapshot ORDER BY id DESC LIMIT :n", {"n": max(1, int(limit))})
+        "FROM pms_plan_snapshot WHERE COALESCE(JSON_UNQUOTE(JSON_EXTRACT(meta_json, '$.kind')), '') "
+        "NOT IN ('consumed', 'rechecked') ORDER BY id DESC LIMIT :n", {"n": max(1, int(limit))})
     return [_snapshot_row(r) for r in rows]
 
 
@@ -881,20 +898,12 @@ def list_plan_snapshots(*, plan_date=None, limit: int = 50) -> list:
 
 
 def prune_plan_snapshots(keep: int = 200) -> int:
-    """只留最近 keep 份。
+    """停止按行数删除。保留全部记录，满足至少 120 交易日与未结束引用保留。
 
-    单表守卫不许子查询跨表, 所以先查出保留水位的 id 再按 id 删 —— 两条单表语句,
+    未来若加清理器，必须同时核验交易日水位、未结束提议及持仓引用；不能恢复行数上限。
-    比一条 `DELETE ... WHERE id NOT IN (SELECT ...)` 干净, 也不会被守卫拦下。
+    keep 仅为旧调用兼容，当前不删除任何历史。
-
-    下限是 **2 份不是 1 份**: 比对天生要两份, 留一份等于把榜单变化永久锁死在"首次"上,
-    而且页面上看起来一切正常。参数填了 0 或 1 也按 2 算。
     """
-    keep = max(2, int(keep or 0))
+    return 0
-    r = fetch_one("SELECT id FROM pms_plan_snapshot ORDER BY id DESC LIMIT 1 OFFSET :k",
-                  {"k": keep})
-    if not r:
-        return 0
-    return execute("DELETE FROM pms_plan_snapshot WHERE id <= :id", {"id": int(r["id"])})
 
 
 def _snapshot_row(r: dict) -> dict:

app/services/command_service.py

@@ -349,6 +349,9 @@ def plan_command(cmd: dict) -> dict:
         notes.insert(0, f"⚠ 停买入侧有 {len(stop_r['errors'])} 处未成: "
                         + "; ".join(str(x) for x in stop_r["errors"][:5]))
     progress = {"target_amount": result.get("target_amount", 0.0),
+                'plan_provenance': [{'ts_code': it['ts_code'], 'plan_id': cs.make_plan_id(command_id, i),
+                                     **it['plan_provenance']}
+                                    for i, it in enumerate(items, 1) if it.get('plan_provenance')],
                 "planned_amount": result.get("planned_amount", 0.0),
                 "done_amount": 0.0, "gap": result.get("gap", 0.0),
                 "plan_count": len(rows), "deadline": str(deadline),
@@ -768,7 +771,7 @@ def _candidates(view: dict) -> list:
            or plan_feed.SRC_PLAN_API).strip()
     out, seen, noprice, fallback = [], set(), [], []
 
-    def _push(c, score, tag, *, price=None, theme=None):
+    def _push(c, score, tag, *, price=None, theme=None, plan_provenance=None):
         if not c or c in held or c in black or c in seen:
             return
         px, psrc = float(price or 0), "given"
@@ -785,13 +788,16 @@ def _candidates(view: dict) -> list:
         # 不参与约束 —— theme 是"传导主题"(事件驱动、天天变、覆盖率约三成), 拿它当行业标签
         # 会让集中度约束跟着漂, 已于 2026-07-31 明确不再灌行业表 (见 UPSTREAM_PLAN_API.md §6)。
         out.append({"ts_code": c, "price": px, "price_source": psrc, "score": float(score or 0),
-                    "sector": industry.get(c), "theme": theme, "src": tag})
+                    "sector": industry.get(c), "theme": theme, "src": tag,
+                    'plan_provenance': plan_provenance})
 
     if src in (plan_feed.SRC_PLAN_API, plan_feed.SRC_BOTH):
         try:
             sel = plan_feed.candidates(held=held, black=black)
             for r in sel["items"]:
-                _push(r["ts_code"], r["score"], "plan_api", theme=r.get("theme"))
+                _push(r["ts_code"], r["score"], "plan_api", theme=r.get("theme"),
+                      plan_provenance={k: r.get(k) for k in ('decision', 'consumption_id',
+                          'selection_rank', 'research_rank', 'review_ref', 'research')})
             logger.info("[候选池] 上游计划 %s: 排序池 %s → 合格 %s → 取 %d 只 (丢弃 %s)",
                         sel.get("date"), sel.get("considered"), sel.get("eligible"),
                         len(sel["items"]), sel.get("dropped"))

app/services/company_report.py

@@ -34,7 +34,7 @@ _cache: dict = {}        # url -> {"at": 时刻, "text": 正文}
 def _url_of(cr) -> str | None:
     if not isinstance(cr, dict):
         return None
-    u = cr.get("report_url")
+    u = cr.get('report_version_url') or cr.get("report_url")
     u = str(u).strip() if u else ""
     return u if u.startswith(("http://", "https://")) else None
 
@@ -106,9 +106,26 @@ def _fail_why(e: Exception, timeout: int) -> str:
     return f"{type(e).__name__}: {e}"
 
 
-def get(code: str, *, now=None, timeout: int = TIMEOUT_SEC, getter=None, state_map=None, plan=None) -> dict:
+def get(code: str, *, now=None, timeout: int = TIMEOUT_SEC, getter=None, state_map=None, plan=None, ref=None) -> dict:
     """页面用的一站式取法。返回固定键: ok / ts_code / url / source / markdown / cached / fetched_at / error。"""
     url, src = report_url_for(code, state_map=state_map, plan=plan)
+    if ref:
+        import re
+        from urllib.parse import urlsplit, urlunsplit, urlencode
+        if not re.fullmatch('[0-9a-f]{64}', ref):
+            return {'ok': False, 'error': '报告版本标识无效'}
+        if not url:
+            from app.repo import pms_repo
+            for row in pms_repo.list_ledger(ts_code=code, limit=200):
+                cr = (row.get('hard_numbers') or {}).get('company_review') or {}
+                if cr.get('review_ref') == ref:
+                    url = _url_of(cr)
+                    if url:
+                        src = 'ledger'
+                        break
+        if url:
+            parts = urlsplit(url)
+            url = urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode({'ref': ref}), ''))
     out = {"ok": False, "ts_code": code, "url": url, "source": src, "markdown": "", "cached": False,
            "fetched_at": None, "error": None}
     if not url:

app/services/executor.py

@@ -103,6 +103,10 @@ def materialize_plans(limit: int = 100) -> dict:
                      or param_store.get_int("PMS_EXEC_WINDOW_TDAYS", 3))
         prog0 = {"deadline": str(p.get("deadline") or ""), "command_id": cid,
                  "is_command": True, "children": []}
+        provenance = next((x for x in (cmd.get('progress') or {}).get('plan_provenance', [])
+                           if x.get('plan_id') == p['plan_id']), None)
+        if provenance:
+            prog0['plan'] = provenance
         if cmd.get("cmd_type") == "LIQUIDATE_ALL":
             # 一键清仓是 danger 级紧急命令, 方案注记写着"紧急, 不做择时优化" ——
             # 这个语义从前没传到执行层, 紧急清仓照样等开盘半小时、等均价。
@@ -234,7 +238,7 @@ def run_tick(*, now=None, dry_run: bool = False) -> dict:
                         pms_repo.insert_ledger(
                             ts_code=code, action=ins.get("action"), arbiter="rule",
                             verdict="WARN", price_at=day_ctx.get("price") or 0,
-                            hard_numbers={"ref_lock": prog.get("ref_lock"),
+                            hard_numbers={**(prog.get('plan') or {}), "ref_lock": prog.get("ref_lock"),
                                           "advice": (prog.get("exec_advice") or {})},
                             ref_id=ins["instruction_id"], reason=d["ref_drift"])
                     except Exception as e:
@@ -304,7 +308,7 @@ def run_tick(*, now=None, dry_run: bool = False) -> dict:
                         pms_repo.insert_ledger(
                             ts_code=code, action=ins.get("action"), arbiter="rule",
                             verdict="REJECT", price_at=day_ctx.get("price") or 0,
-                            hard_numbers=gate["hard_numbers"],
+                            hard_numbers={**gate["hard_numbers"], **(prog.get('plan') or {})},
                             failed_checks=gate["failed"], ref_id=ins["instruction_id"],
                             reason="规则闸终检未通过 (宁可不动; 同因当日只记一条)")
                         prog["gate_reject_logged"] = {"ymd": ymd_today, "key": rej_key}
@@ -361,7 +365,7 @@ def run_tick(*, now=None, dry_run: bool = False) -> dict:
                     f"出口表与本端不一致")
             pms_repo.insert_ledger(ts_code=code, action=ins.get("action"), arbiter="rule",
                                    verdict="PASS", price_at=day_ctx.get("price") or 0,
-                                   hard_numbers={**gate["hard_numbers"], "limit": d["limit_price"],
+                                   hard_numbers={**gate["hard_numbers"], **(prog.get('plan') or {}), "limit": d["limit_price"],
                                                  "mode": res["mode"]},
                                    ref_id=ins["instruction_id"], reason=d["reason"])
             out["fired"].append({"instruction_id": ins["instruction_id"], "code": code,
@@ -450,7 +454,8 @@ def cancel_instruction(instruction_id: str, reason: str = "页面人工撤销")
             logger.warning("撤销指令进度回写失败 %s: %s", instruction_id, e)
     pms_repo.update_instruction(instruction_id, status=ST_CANCELLED)
     pms_repo.insert_ledger(ts_code=ins["ts_code"], action=ins.get("action"), arbiter="user",
-                           verdict="REJECT", price_at=0, ref_id=instruction_id, reason=reason)
+                           verdict="REJECT", price_at=0, ref_id=instruction_id, reason=reason,
+                           hard_numbers=(ins.get('progress') or {}).get('plan') or {})
     return {"ok": True, "downstream": r, "message": f"指令 {instruction_id} 已撤销"}
 
 

app/services/logic_state_service.py

@@ -186,7 +186,9 @@ def company_view(st):
             "line": f"质地{cr.get('overall')}（回报{g.get('回报与护城河') or '缺'}、盈余{g.get('盈余质量与财务安全') or '缺'}、成长{g.get('成长与含金量') or '缺'}）"
                     f"，估值{cr.get('valuation') or '无'}" + (f"，置信{cr['confidence']}" if cr.get("confidence") else ""),
             "invalidation": cr.get("invalidation"), "doubt_hard": bool(cr.get("doubt_hard")),
-            "report_url": cr.get("report_url"), "text": (st or {}).get("company_review_text")}
+            "report_url": cr.get("report_version_url") or cr.get("report_url"),
+            "text": (st or {}).get("company_review_text"), 'company_review': cr,
+            **{k: st.get(k) for k in ('decision', 'research', 'review_ref', 'observation_at', 'observation_scope')}}
 
 
 def ref_target_view(st):
@@ -237,6 +239,8 @@ def entry_view(ts_code: str) -> dict:
     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],
+            **{k: hn.get(k) for k in ('decision', 'research', 'review_ref', 'company_review',
+                 'consumption_id', 'execution_rank', 'priority_reason')},
             "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],

app/services/param_store.py

@@ -40,6 +40,7 @@ SECRET_KEYS = ("PMS_QMT_SIGN_SEED_HEX", "PMS_QMT_PEER_PUBKEY_B64", "PMS_SESSION_
 
 # 运行态开关: key -> (默认值, 类型, 说明)
 RUNTIME_EXTRA = {
+    'PMS_PLAN_VERSION_RECHECK': (False, bool, '自主建仓采纳前核验计划版本；新版发布与留痕验收后开启，当前不可核验时暂停采纳'),
     "PMS_GLOBAL_BUY_HALT": (False, bool, "全局暂停买入 (HALT_BUY 命令置位)"),
     "PMS_GLOBAL_EXEC_HALT": (False, bool, "全局暂停执行"),
     "PMS_BRAKE_UNTIL": (0, int, "组合刹车解除日 YYYYMMDD, 0=未刹车"),

app/services/plan_feed.py

@@ -204,7 +204,10 @@ def _company_review_or_none(v):
             "invalidation": _text_or_none(v.get("invalidation")), "antithesis": _text_or_none(v.get("antithesis")),
             "catalyst": _text_or_none(v.get("catalyst")), "brain_status": _text_or_none(v.get("brain_status")),
             "ran_at": _text_or_none(v.get("ran_at")), "age_days": _int_or_none(v.get("age_days")),
-            "report_url": _text_or_none(v.get("report_url"))}
+            "report_url": _text_or_none(v.get("report_url")),
+            **{k: v.get(k) for k in ('review_ref', 'report_version_url', 'report_ran_at',
+                 'period_fi', 'period_rf', 'qual', 'n_materials', 'unknown_n',
+                 'numbers_unmatched', 'expectation', 'cycle', 'thesis_lens_refs')}}
 
 
 def _industry_catalyst_or_none(v, limit=3):
@@ -337,6 +340,9 @@ def _rows(raw, bucket: str) -> list:
             "pointed": (bool(it.get("pointed")) if it.get("pointed") is not None else None),
             "rank_axis": _text_or_none(it.get("rank_axis")),
             "rank_old": _int_or_none(it.get("rank_old")),
+            "research_rank": _int_or_none(it.get('research_rank')),
+            "research": it.get('research') if isinstance(it.get('research'), dict) else None,
+            "review_ref": _text_or_none(it.get('review_ref')),
         })
     return out
 
@@ -356,6 +362,8 @@ def parse_plan(payload, *, requested=None) -> dict:
     observe = _rows(payload.get("observe"), BUCKET_OBSERVE)
     if not main and not observe:
         raise PlanFeedError(f"计划 {date} 主榜与观察档都是空的")
+    state = dict(payload.get('refresh_status') or {})
+    state['pending_codes'] = [normalize_code(c) for c in state.get('pending_codes') or []]
     counts = payload.get("counts") if isinstance(payload.get("counts"), dict) else {}
     req = dict(requested or {})
     req_top = _int_or_none(req.get("top"))
@@ -368,6 +376,8 @@ def parse_plan(payload, *, requested=None) -> dict:
             themes[r["ts_code"]] = r["theme"]
     return {
         "date": date,
+        **{k: payload.get(k) for k in ('schema_version', 'decision', 'traceability')},
+        'refresh_status': state,
         # 计划版本与生成时刻 (2026-09-02 起上游下发)。同一个 date 的计划会被重算多版,
         # 这两项是上游自己的版本戳; 旧版应答没有, 缺就是 None (名册快照那套指纹照旧管用)。
         "plan_version": _text_or_none(payload.get("plan_version")),
@@ -505,6 +515,10 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N
     dropped = {"held": 0, "black": 0, "tier": 0, "score": 0, "sources": 0, "upside": 0,
                "st": 0, "theme": 0, "dup": 0, "capped": 0, "display": 0}
     passed, seen, per_theme, st_unknown, display_only = [], set(), {}, [], []
+    filtered = []
+    def drop(reason, code):
+        dropped[reason] += 1
+        filtered.append({'ts_code': code, 'reason': reason})
     # 排序: 默认按分数降序、同分按 rank 升序 (与 2026-09-11 之前逐字相同)。开了按质地排序后, 先按
     # 基本面立场分桶 (质地看多的桶排在前) 再按分数, 在主题限额与截断之前生效; 技术面不参与排序。
     # 2026-09-17 工作包乙: 上游是质地轴且开了按上游名次排时, 直接按上游名次升序 (同名次原始分降序),
@@ -513,28 +527,28 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N
     mode = _rank_mode(plan, rank_by_upstream, rank_by_quality)
     if mode == "upstream":
         def _sort_key(x):
-            return (x.get("rank") or 10 ** 9, -(x.get("score") or 0.0))
+            return (x.get('research_rank') or x.get("rank") or 10 ** 9, -(x.get("score") or 0.0), x['ts_code'])
     elif rank_by_quality:
         def _sort_key(x):
             return (0 if _fund_stance_of(x) == "看多" else 1,
-                    -(x.get("score") or 0.0), x.get("rank") or 10 ** 9)
+                    -(x.get("score") or 0.0), x.get("rank") or 10 ** 9, x['ts_code'])
     else:
         def _sort_key(x):
-            return (-(x.get("score") or 0.0), x.get("rank") or 10 ** 9)
+            return (-(x.get("score") or 0.0), x.get("rank") or 10 ** 9, x['ts_code'])
     for r in sorted(pool, key=_sort_key):
         c = r["ts_code"]
         if c in seen:
-            dropped["dup"] += 1
+            drop('dup', c)
             continue
         seen.add(c)
         if c in held:
-            dropped["held"] += 1
+            drop('held', c)
             continue
         if c in black:
-            dropped["black"] += 1
+            drop('black', c)
             continue
         if route_by_verdict and r.get("verdict") == VERDICT_DISPLAY:
-            dropped["display"] += 1           # 上游判为仅展示: 不进池, 单独列出让人看得见
+            drop('display', c)
             # 带上判决依据: 仅展示至少有三种来路 (差一个当日涨幅、吸筹没到确认线、
             # 被昨夜坏信号一票否决), 三者处置完全不同。只记代码的话, 下游只能对所有
             # 仅展示的票写同一句话, 人看不出哪只该继续盯。
@@ -545,30 +559,31 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N
             if not nm:
                 st_unknown.append(c)          # 没名字判不了, 留着但要让人知道
             elif is_st(nm):
-                dropped["st"] += 1
+                drop('st', c)
                 continue
         if tiers and r.get("tier") is not None and r["tier"] not in tiers:
-            dropped["tier"] += 1
+            drop('tier', c)
             continue
         if min_score is not None and (r.get("score") or 0.0) < min_score:
-            dropped["score"] += 1
+            drop('score', c)
             continue
         if min_sources and (r.get("n_sources") or 0) < min_sources:
-            dropped["sources"] += 1
+            drop('sources', c)
             continue
         if min_upside is not None and (r.get("upside") or 0.0) < min_upside:
-            dropped["upside"] += 1
+            drop('upside', c)
             continue
-        if theme_cap > 0:
+        if theme_cap > 0 and r.get('theme'):
-            t = r.get("theme") or "(无主题)"
+            t = r['theme']
             if per_theme.get(t, 0) >= theme_cap:
-                dropped["theme"] += 1
+                drop('theme', c)
                 continue
             per_theme[t] = per_theme.get(t, 0) + 1
         passed.append(r)
 
     n = max(0, int(top_n or 0)) or len(passed)
     dropped["capped"] = max(0, len(passed) - n)
+    filtered.extend({'ts_code': r['ts_code'], 'reason': 'capped'} for r in passed[n:])
     items = [{"ts_code": r["ts_code"], "name": r["name"], "score": r.get("score") or 0.0,
               "sector": r.get("theme"), "theme": r.get("theme"), "tier": r.get("tier"),
               "heat": r.get("heat"), "upside": r.get("upside"), "rank": r.get("rank"),
@@ -593,9 +608,14 @@ def select_candidates(plan: dict, *, held=(), black=(), top_n: int = 30, tiers=N
               "company_gate": r.get("company_gate"),
               # 排序轴四键 (2026-09-17 工作包乙): 原样透传给动作引擎与提议卡, 缺就是 None。
               "quality_band": r.get("quality_band"), "pointed": r.get("pointed"),
-              "rank_axis": r.get("rank_axis"), "rank_old": r.get("rank_old")}
+              "rank_axis": r.get("rank_axis"), "rank_old": r.get("rank_old"),
+              **{k: r.get(k) for k in ('research_rank', 'research', 'review_ref')}}
              for r in passed[:n]]
+    for i, item in enumerate(items, 1):
+        item.update(selection_rank=i, decision=plan.get('decision'),
+                    refresh_status=plan.get('refresh_status'))
     return {"date": plan.get("date"), "considered": len(pool), "eligible": len(passed),
+            'filtered': filtered, 'decision': plan.get('decision'),
             "items": items, "dropped": dropped, "st_unknown": st_unknown,
             "display_only": display_only, "rank_mode": mode,
             "tier_complete": _tier_complete(pool, tiers)}
@@ -806,6 +826,11 @@ def parse_logic_states(payload) -> dict:
         # 公司深度两键随状态回 (2026-09-09 接入方案): 持仓页"证据还在不在"旁显示质地与失效条件。
         st["company_review"] = _company_review_or_none(it.get("company_review"))
         st["company_review_text"] = _text_or_none(it.get("company_review_text"))
+        st['decision'] = it.get('decision_version') or (payload or {}).get('decision_version')
+        st['observation_at'] = it.get('observation_at')
+        st['observation_scope'] = it.get('observation_scope')
+        st['research'] = it.get('research')
+        st['review_ref'] = it.get('review_ref')
         out[code] = st
     return out
 
@@ -844,7 +869,7 @@ def get_plan(*, force: bool = False, date=None) -> dict:
     now = time.time()
     with _lock:
         fresh_hit = (not force and _cache["key"] == key and _cache["plan"] is not None
-                     and now - _cache["at"] < max(1, p["cache_sec"]))
+                     and now - _cache["at"] < min(300, max(1, p["cache_sec"])))
         if fresh_hit:
             return _cache["plan"]
         if (not force and _cache["key"] == key and _cache["error"]
@@ -914,10 +939,15 @@ def snapshot(plan: dict, *, keep: int = 200, rank_mode=None) -> dict:
     from app.repo import pms_repo
     roster = pdf.roster_of(plan)
     digest = pdf.digest_of(roster)
+    if plan.get('decision'):
+        from app.core.plan_revision import fingerprint
+        digest = fingerprint({'received': plan['decision'], 'requested': plan.get('requested')})[:40]
     meta = {k: plan.get(k) for k in ("counts", "returned", "requested", "funnel",
                                      "theme_cap", "heat_date", "market_snapshot_days",
                                      "age_tdays", "url",
-                                     "plan_source", "snapshot_generated_at", "gate_on")}
+                                     "plan_source", "snapshot_generated_at", "gate_on",
+                                     "decision", "traceability")}
+    meta['kind'] = 'received'
     # 排序模式 (2026-09-17 工作包乙): 让选股系统周报按它还原当天交付名单的序。旧调用没传即不写。
     if rank_mode is not None:
         meta["rank_mode"] = rank_mode
@@ -929,7 +959,8 @@ def snapshot(plan: dict, *, keep: int = 200, rank_mode=None) -> dict:
         capped_main=bool(truncated.get("main")), capped_obs=bool(truncated.get("observe")))
     out = {"digest": digest, "rows": len(roster), "stored": bool(n)}
     if n:
-        out["pruned"] = pms_repo.prune_plan_snapshots(keep=keep)
+        # 停用按行数清理：盘中修订会把 200 行很快用尽。归档前保留全部历史及引用。
+        out["pruned"] = 0
         logger.info("[上游计划] 名册快照已落: %s digest=%s (%d 只)",
                     plan.get("date"), digest[:10], len(roster))
     return out
@@ -1109,7 +1140,8 @@ def snapshot_log(*, plan_date=None, limit: int = 50) -> dict:
 
 
 # ================================================================ 对外: 候选与状态
-def candidates(*, held=(), black=(), route_by_verdict: bool = False) -> dict:
+def candidates(*, held=(), black=(), route_by_verdict: bool = False, force=False,
+               record=True, recheck_code=None) -> dict:
     """参数驱动的候选清单 (不含价格 —— 价格由调用方用 market.get_price 现取)。
 
     route_by_verdict 默认 False: 命令驱动建仓 (command_service._candidates) 不传, 于是用户
@@ -1119,14 +1151,14 @@ def candidates(*, held=(), black=(), route_by_verdict: bool = False) -> dict:
     显式传 True。要让命令驱动也按判决过滤, 改这里的调用方, 不改默认值。
     """
     p = _params()
-    plan = get_plan()
+    plan = get_plan(force=force) if force else get_plan()
     route = bool(route_by_verdict)
-    out = select_candidates(plan, held=held, black=black, top_n=p["top_n"], tiers=p["tiers"],
+    out = select_candidates(plan, held=held, black=black, top_n=0 if recheck_code else 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=0 if recheck_code else p["theme_cap_local"], exclude_st=p["exclude_st"],
                             route_by_verdict=route,
                             rank_by_quality=p["rank_by_quality"],
                             rank_by_upstream=p["rank_by_upstream"],
@@ -1134,6 +1166,36 @@ def candidates(*, held=(), black=(), route_by_verdict: bool = False) -> dict:
     out["age_tdays"] = plan.get("age_tdays")
     out["theme_cap"] = plan.get("theme_cap")
     out["route_by_verdict"] = route
+    from app.core.plan_revision import fingerprint, basis
+    # 实际参数与组合上下文参与消费指纹；来源地址等接入信息不进入业务参数。
+    effective = {k: p[k] for k in ('top_n', 'tiers', 'include_observe', 'min_score',
+        'min_sources', 'min_upside', 'theme_cap_local', 'exclude_st', 'rank_by_quality',
+        'rank_by_upstream', 'observe_if_fund_bull')}
+    context = {'held': sorted(held), 'black': sorted(black), 'route_by_verdict': route,
+               'recheck_code': recheck_code}
+    if recheck_code:
+        out['items'] = [r for r in out['items'] if r['ts_code'] == recheck_code]
+    consumption = {'kind': 'rechecked' if recheck_code else 'consumed', 'decision': plan.get('decision'),
+                   'params': effective, 'context': context, 'rank_mode': out['rank_mode'],
+                   'filtered': out['filtered'], 'order': [r['ts_code'] for r in out['items']]}
+    cid = fingerprint(consumption)
+    for row in out['items']:
+        row['consumption_id'] = cid
+        row['plan_basis'] = basis(row)
+    out['consumption'] = dict(consumption, id=cid)
+    if record:
+        from app.repo import pms_repo
+        try:
+            pms_repo.insert_plan_snapshot(plan_date=plan['date'], digest=cid[:40],
+                roster=out['items'], meta=dict(consumption, id=cid),
+                n_main=sum(r.get('bucket') == BUCKET_MAIN for r in out['items']),
+                n_observe=sum(r.get('bucket') != BUCKET_MAIN for r in out['items']),
+                capped_main=False, capped_obs=False)
+        except Exception as e:
+            # 记录失败可展示，不能把无凭据候选变成自主指令。
+            out['consumption']['error'] = type(e).__name__
+            for row in out['items']:
+                row['consumption_error'] = type(e).__name__
     return out
 
 

app/services/plan_recheck.py

@@ -0,0 +1,78 @@
+"""采纳前核验；复用原扫描、合议、研判与规则闸，不解释自然语言。"""
+from datetime import datetime
+from app.core.plan_revision import compare, fingerprint, portfolio_context
+
+
+def current_basis(hard, code, proposal_id=None):
+    from app.services import plan_feed, portfolio, command_service, proposal_service as ps
+    view = portfolio.positions_view()
+    stock = command_service.effective_stock_params()
+    params = ps._scan_params(view)
+    selected = plan_feed.candidates(held=[x['ts_code'] for x in view['held']],
+        black=[c for c, v in stock.items() if v.get('black')], force=True,
+        route_by_verdict=bool(params.get('open_route_by_verdict')), recheck_code=code)
+    current = next((r for r in selected['items'] if r['ts_code'] == code), None)
+    if not (selected.get('decision') or {}).get('id'):
+        return {'ok': False, 'error': '当前计划没有可核验版本'}
+    if current is None:
+        return {'ok': False, 'error': '该票已不满足当前候选资格', 'material': True}
+    result = compare(hard.get('plan_basis'), current, code)
+    if result == 'unavailable':
+        return {'ok': False, 'error': '当前计划修订或消费留痕尚未完成'}
+    ph = fingerprint({k: v for k, v in params.items() if not k.startswith('_')})
+    commitments = ps._inflight_open_commitments(view, exclude_proposal=proposal_id)
+    context = portfolio_context(view, commitments, stock.get(code))
+    material = (result == 'material' or hard.get('scan_params_hash') != ph
+                or hard.get('portfolio_context_hash') != context)
+    return {'ok': not material, 'material': material,
+            'error': '研究、资格或仓位参数已变化，须重新评估' if material else None,
+            'view': view, 'stock': stock, 'params': params,
+            'provenance': {'decision': current['decision'], 'consumption_id': current['consumption_id'],
+                           'version_check': {'status': 'passed', 'at': datetime.now().astimezone().isoformat(),
+                                             'reason': '依据相同，名次变化不撤销'}}}
+
+
+def before_accept(p, expected_revision):
+    from app.services import proposal_service as ps
+    from app.repo import pms_repo
+    hard = p.get('hard_numbers') or {}
+    if p.get('action') != 'OPEN' or hard.get('kind') == 'strategy':
+        return {'ok': True, 'proposal': p}
+    try:
+        check = current_basis(hard, p['ts_code'], proposal_id=p.get('proposal_id'))
+        if check.get('ok'):
+            if hard.get('proposal_revision') and expected_revision != hard['proposal_revision']:
+                return {'ok': False, 'error': '提议内容已更新，请刷新后再次确认', 'proposal': p}
+            return {'ok': True, 'proposal': dict(p, hard_numbers=dict(hard, **check['provenance']))}
+        if not check.get('material') or not check.get('view'):
+            return check
+        now = datetime.now()
+        out = {'skipped': [], 'rejected': [], 'queued': [], 'executed': []}
+        params, view = check['params'], check['view']
+        mkt = {}
+        params['_mkt'] = mkt
+        fresh = ps._scan_open(view, params, check['stock'], {}, mkt, out, now=now,
+                              review_code=p['ts_code'], exclude_proposal=p['proposal_id'])
+        if not fresh:
+            return {'ok': False, 'error': '重新评估未产出可采纳提议', 'details': out}
+        c = fresh[0]
+        c['_plan_recheck'] = True
+        from app.services import param_store
+        from app.core import tradedays as td
+        brake = td.ymd() < param_store.get_int('PMS_BRAKE_UNTIL', 0)
+        result = ps._route_one(c, view, params, check['stock'], brake, now, True, out, review_only=True)
+        if not result:
+            return {'ok': False, 'error': '新版重新评估未通过', 'details': out}
+        hn = c['hard_numbers']
+        hn['version_check'] = {'status': 'reconfirmed_required', 'at': now.astimezone().isoformat(),
+                               'reason': '依据变化，已重跑合议、研判、数量及规则闸'}
+        hn['proposal_revision'] = fingerprint({'qty': c['qty'], 'hard_numbers': hn})
+        if not pms_repo.revise_pending_proposal(p, c, result['verdict']):
+            return {'ok': False, 'error': '提议被并发更新，请刷新'}
+        pms_repo.insert_ledger(ts_code=p['ts_code'], action=p['action'], arbiter='rule',
+            verdict='WARN', price_at=hn.get('price') or 0, hard_numbers=hn,
+            ref_id=p['proposal_id'], reason='新版依据变化，更新提议并要求重新确认')
+        return {'ok': False, 'error': '提议已按新版重新评估，请查看更新内容后再次确认',
+                'proposal': dict(p, qty=c['qty'], hard_numbers=hn)}
+    except Exception as e:
+        return {'ok': False, 'error': f'版本核验暂不可用，保留提议：{type(e).__name__}'}

app/services/proposal_service.py

@@ -212,7 +212,8 @@ def scan_and_route(*, now=None, dry_run: bool = False) -> dict:
         out["skipped"].append({"action": ae.A_OPEN, "why": "新建仓档位 off, 不扫描候选池"})
     else:
         try:
-            open_cands = _scan_open(view, params, stock_params, skip, mkt, out, now=now)
+            open_cands = _scan_open(view, params, stock_params, skip, mkt, out, now=now,
+                                    record_consumption=not dry_run)
         except Exception as e:
             logger.exception("新建仓扫描失败")
             out["errors"].append(f"新建仓扫描失败: {type(e).__name__}: {e}")
@@ -307,7 +308,8 @@ def _save_tech_exit_done(codes) -> None:
 
 
 # ================================================================ 新建仓的取数与筛选
-def _scan_open(view, params, stock_params, skip, mkt, out, now=None) -> list:
+def _scan_open(view, params, stock_params, skip, mkt, out, now=None,
+               review_code=None, exclude_proposal=None, record_consumption=True) -> list:
     """候选池 → 新建仓候选。产出的候选数天生不超过剩余名额 (名额与金额在纯逻辑里边走边扣)。
 
     与命令驱动那条路 (command_service._candidates) 有意不同的两点:
@@ -339,7 +341,9 @@ def _scan_open(view, params, stock_params, skip, mkt, out, now=None) -> list:
         # 按判决分流开着时, 上游判为「仅展示」的票在候选阶段就剔掉 (理由见
         # plan_feed.select_candidates 的说明: 不剔的话它们会白占 top_n 的名额)。
         sel = plan_feed.candidates(held=held, black=black,
-                                   route_by_verdict=bool(params.get("open_route_by_verdict")))
+                                   route_by_verdict=bool(params.get("open_route_by_verdict")),
+                                   **({'recheck_code': review_code} if review_code else {}),
+                                   **({'record': False} if not record_consumption else {}))
     except plan_feed.PlanFeedError as e:
         # 拿不到 ≠ 今天没票可买。显式留痕, 本轮不产新建仓候选, 已有持仓的四类照常。
         logger.error("[新建仓] 候选池取数失败, 本轮不建仓: %s", e)
@@ -353,6 +357,8 @@ def _scan_open(view, params, stock_params, skip, mkt, out, now=None) -> list:
         out["skipped"].append({"ts_code": row.get("ts_code"), "action": ae.A_OPEN,
                                "why": ae.display_only_why(row)})
     items = list(sel.get("items") or [])
+    if review_code:
+        items = [x for x in items if x['ts_code'] == review_code]
     if not items:
         out["skipped"].append({"action": ae.A_OPEN, "why": _no_candidate_why(sel)})
         return []
@@ -381,7 +387,8 @@ def _scan_open(view, params, stock_params, skip, mkt, out, now=None) -> list:
     # 名额与金额都要扣掉**在途的建仓承诺** (2026-08-28 审查修): names_count 只数已成交
     # 持仓, 在途 OPEN 指令与等拍板的 OPEN 提议两头都不占 —— 候选没进买入区间时指令滞留
     # 在途, 下一跳名额照旧又放两只, 几分钟就把持仓数上限击穿。同票去重挡不住换一只票。
-    inflight = _inflight_open_commitments(view)
+    inflight = (_inflight_open_commitments(view, exclude_proposal=exclude_proposal)
+                if exclude_proposal else _inflight_open_commitments(view))
     slots = int(p["max_names"] or 0) - int(t["names_count"] or 0) - len(inflight["names"])
     room = float(p["portfolio_cap"] or 0) * float(p["scale"] or 0) \
         - float(t["portfolio_mv"] or 0) - inflight["amount"]
@@ -416,7 +423,11 @@ def _scan_open(view, params, stock_params, skip, mkt, out, now=None) -> list:
     # 当日行情快照只对**真的产出了候选**的票取 —— 候选数已被名额与金额扣到很小,
     # 不必为整池几十只票各拉一遍分钟线。规则闸的「不追高」靠的就是这一份。
     for c in res["candidates"]:
+        from app.core.plan_revision import fingerprint, portfolio_context
+        c.setdefault('hard_numbers', {})['scan_params_hash'] = fingerprint(
+            {k: v for k, v in _scan_params(view).items() if not k.startswith('_')})
         code = c["ts_code"]
+        c['hard_numbers']['portfolio_context_hash'] = portfolio_context(view, inflight, stock_params.get(code))
         d = {"ma5": market.get_ma5(code)}
         try:
             d["day"] = market.day_snapshot(code) or {}
@@ -585,7 +596,7 @@ def _no_candidate_why(sel) -> str:
     return head + "：" + "，".join(parts) + "。"
 
 
-def _inflight_open_commitments(view) -> dict:
+def _inflight_open_commitments(view, exclude_proposal=None) -> dict:
     """在途的新建仓承诺: {names: {代码}, amount: 预估占用金额}。
 
     口径: 在途 (LIVE) 的 OPEN 指令 + 等拍板 (WAIT_USER) 的 OPEN 提议, 只算**尚未持有**
@@ -603,6 +614,8 @@ def _inflight_open_commitments(view) -> dict:
         logger.warning("读在途 OPEN 指令失败 (按无在途): %s", e)
     try:
         for pr in pms_repo.list_proposals(statuses=("WAIT_USER",), limit=200):
+            if exclude_proposal and pr.get('proposal_id') == exclude_proposal:
+                continue
             if pr.get("action") == ae.A_OPEN and pr["ts_code"] not in held:
                 hn = pr.get("hard_numbers") or {}
                 rows.append((pr["ts_code"], int(pr.get("qty") or 0),
@@ -719,7 +732,7 @@ def disposition_snapshot(now=None) -> dict:
         for _k in _inflight_keys():   # 在途; 前端多半已按指令/提议标成「已下单/待你确认」
             skip[_k] = "系统正在处理这只票（见『待确认提议』或『当日指令』）"
         sink = {"skipped": []}      # 丢弃用: _scan_open 只往里 append/extend, 不读它
-        would = _scan_open(view, params, stock_params, skip, {}, sink, now=now)
+        would = _scan_open(view, params, stock_params, skip, {}, sink, now=now, record_consumption=False)
     except Exception as e:
         logger.exception("候选处置快照失败")
         return {"ok": False, "ready": False, "error": f"{type(e).__name__}: {e}",
@@ -754,7 +767,7 @@ def disposition_snapshot(now=None) -> dict:
 
 
 def _route_one(c, view, params, stock_params, brake_active, now, dry_run, out,
-               *, deadline=None):
+               *, deadline=None, review_only=False):
     code, action, side = c["ts_code"], c["action"], c["side"]
     is_open = (action == ae.A_OPEN)
     pos = _pos_of(view, code)
@@ -829,9 +842,18 @@ def _route_one(c, view, params, stock_params, brake_active, now, dry_run, out,
                 _rtext = reask_service.reask_text(_rk)
             else:
                 c.pop("reask", None)        # 名额没落上就不算重问, 这一跳按普通候选走
+        if is_open and param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False):
+            from app.core.plan_revision import needs_new_judgement
+            try:
+                previous = next((r.get('hard_numbers') for r in pms_repo.list_ledger(ts_code=code, limit=20)
+                                 if r.get('action') == ae.A_OPEN and r.get('arbiter') in ('judge', 'system')), None)
+            except Exception:
+                previous = None
+            # 即使上一跳因版本变化放弃自动采纳，下一跳也不能复用该票旧批准的缓存。
+            _force = _force or needs_new_judgement(c.get('hard_numbers') or {}, previous)
         verdict = judge.request(c, context={"position": _judge_ctx(pos),
                                             "recent_ledger": _recent_ledger(code)},
-                                force=_force, reask_text=_rtext)
+                                force=_force or bool(c.get('_plan_recheck')), reask_text=_rtext)
         # 新建仓的低把握驳回改交人 (2026-09-07, 第二道保险)。择时决策系统的提示词已改成
         # 「证据不足以判断 → 不可用」, 但模型未必每次守得住; 它给的把握度是现成的读数,
         # 低于阈值的驳回按「不可用」处理 —— 进人的待确认队列, 不记驳回、不杀提议。
@@ -967,6 +989,14 @@ def _route_one(c, view, params, stock_params, brake_active, now, dry_run, out,
         if auto_why:
             auto_exec = True
 
+    if review_only:
+        _apply_advice_tier(c, c.setdefault('hard_numbers', {}), verdict,
+                           force_trial=(sd.get('tier') == 'trial'))
+        c['hard_numbers'].update(judge_verdict=verdict.get('verdict'),
+                                  judge_conf=verdict.get('confidence'),
+                                  price=price, reason=c['reason'])
+        return {'candidate': c, 'verdict': verdict}
+
     if dry_run:
         (out["executed"] if auto_exec else out["queued"]).append(
             {**_brief(c), "route": "auto" if auto_exec else "queue",
@@ -975,6 +1005,13 @@ def _route_one(c, view, params, stock_params, brake_active, now, dry_run, out,
         return
 
     if auto_exec:
+        if is_open and param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False):
+            check = verify_open_basis(c.get('hard_numbers') or {}, code)
+            if not check.get('ok'):
+                out['skipped'].append({'ts_code': code, 'action': action,
+                    'why': check['error'] + '；下一跳按新输入重新评估'})
+                return
+            c['hard_numbers'].update(check['provenance'])
         sd_accept = sd_decision == "accept"
         strat_cancelled, buys_cancelled = [], []
         if sd_accept and side == "sell":
@@ -1089,6 +1126,14 @@ def _judge_budget_left(deadline) -> bool:
 
 
 # ================================================================ 落表
+def verify_open_basis(hard, code):
+    from app.services.plan_recheck import current_basis
+    try:
+        return current_basis(hard, code)
+    except Exception as e:
+        return {'ok': False, 'error': f'当前计划版本无法核验：{type(e).__name__}'}
+
+
 def _make_instruction(c, price, now) -> str:
     ymd = td.ymd(now)
     # 序号用完整时分秒 (2026-08-28 修): 原来 %1000 只留「分钟个位+秒」, 同日同票同动作
@@ -1107,7 +1152,9 @@ def _make_instruction(c, price, now) -> str:
                 window_tdays=window, status=executor.ST_PROPOSED,
                 progress={"deadline": str(td.window_deadline(now.date(), window)),
                           "is_command": False, "children": [], "auto": True,
-                          "reason": c["reason"]})
+                          "reason": c["reason"],
+                          'plan': {k: (c.get('hard_numbers') or {}).get(k) for k in
+                              ('decision', 'consumption_id', 'review_ref', 'execution_rank', 'priority_reason')}})
             iid = try_iid
             break
         except Exception as e:
@@ -1376,6 +1423,8 @@ def _make_proposal(c, price, verdict) -> str:
     # 采纳按钮那边直接抄提议里的股数原样落单 (只有卖出侧会按可卖量重算), 所以不在这里
     # 改, 后面就没有第二次机会了。
     _apply_advice_tier(c, hn, verdict)
+    from app.core.plan_revision import fingerprint
+    hn['proposal_revision'] = fingerprint({'qty': c['qty'], 'hard_numbers': hn})
     try:
         pms_repo.insert_proposal(
             proposal_id=pid, ts_code=c["ts_code"], action=c["action"], qty=c["qty"],

app/web/main.py

@@ -517,7 +517,18 @@ def api_decide(request: Request, proposal_id: str, payload: dict = Body(default=
             except Exception as e:  # noqa: BLE001
                 logger.warning("[提议] 过期清扫失败: %s", e)
             return {"ok": False, "error": "提议已过失效时刻, 不能再裁决; 该票由系统重新评估"}
-        if not pms_repo.decide_proposal(proposal_id, decision):
+        expected_hard = None
+        if (decision == 'ACCEPTED' and p['action'] == ae.A_OPEN
+                and param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False)):
+            from app.services.plan_recheck import before_accept
+            checked = before_accept(p, payload.get('proposal_revision'))
+            if not checked.get('ok'):
+                return checked
+            expected_hard = p.get('hard_numbers_json')
+            p = checked['proposal']
+        changed = (pms_repo.decide_proposal(proposal_id, decision, expected_hard=expected_hard)
+                   if expected_hard is not None else pms_repo.decide_proposal(proposal_id, decision))
+        if not changed:
             return {"ok": False, "error": "提议状态已变更, 请刷新"}
         hn = p.get("hard_numbers") or {}
         # 用到的预设理由码进硬数字备查 (复盘要能按码聚合"人常用哪条理由驳回")。
@@ -556,6 +567,8 @@ def api_decide(request: Request, proposal_id: str, payload: dict = Body(default=
                 window_tdays=_win,
                 status="PROPOSED",
                 progress={"from_proposal": proposal_id, "is_command": False, "children": [],
+                          'plan': {k: hn.get(k) for k in ('decision', 'consumption_id', 'review_ref',
+                                                         'execution_rank', 'priority_reason', 'version_check')},
                           "deadline": str(td.window_deadline(None, _win))})
             # 与自主执行同一口径: 采纳即算「做过一次」, 计数器要跟着走
             # (否则页面采纳的那条动作绕开了 §6 的一次性约束)
@@ -805,12 +818,12 @@ def _research_st(code):
 
 
 @app.get("/api/research/{ts_code}/report")
-def api_research_report(ts_code: str):
+def api_research_report(ts_code: str, ref: str = None):
     """个股深度评析全文 (2026-09-18): 上游给的 markdown 由后端代取, 页面自己排版成同款样式。
     点击时调, 不进轮询; 按链接缓存十分钟。链接只认上游给的, 不收页面传入的地址。"""
     from app.services import company_report
     try:
-        return company_report.get(cs.normalize_code(ts_code))
+        return company_report.get(cs.normalize_code(ts_code), **({'ref': ref} if ref else {}))
     except Exception as e:  # noqa: BLE001
         return {"ok": False, "error": f"评析报告读取失败: {type(e).__name__}: {e}"}
 
@@ -907,7 +920,8 @@ def api_upstream_plan(limit: int = Query(30), date: str = Query(None),
             plan = plan_feed.get_plan(date=(date or None))
             rows = plan.get("observe" if bucket == "observe" else "main") or []
             out["rows"] = rows[:max(1, int(limit or 30))]
-            out["candidates_raw"] = plan_feed.candidates()
+            out['rows'] = [dict(r, decision=plan.get('decision')) for r in out['rows']]
+            out["candidates_raw"] = plan_feed.candidates(record=False)
         except Exception as e:
             out["status"] = {**st, "ok": False, "hint": f"{type(e).__name__}: {e}"}
         return out

app/web/static/index.html

@@ -1262,6 +1262,7 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
                     </div>
                   </div>
                   <div style="margin-top:6px">
+                    <details><summary>计划版本与研究变化</summary><div v-for="line in researchInfo(p.hard_numbers)" :key="line">{{ line }}</div></details>
                     <el-button type="success" size="small" @click="decide(p,'ACCEPTED')">采纳</el-button>
                     <el-button type="danger" size="small" @click="decide(p,'DECLINED')">驳回</el-button>
                   </div>
@@ -1838,6 +1839,7 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
                查回的逻辑状态、第一条出处与证据截止日。状态值是上游给的中文原文, 不是枚举码。 -->
           <el-table-column label="当初为什么买" min-width="220">
             <template #default="s">
+              <details><summary>入场计划版本与研究依据</summary><div v-for="line in researchInfo(s.row.entry)" :key="line">{{ line }}</div></details>
               <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>
@@ -1850,6 +1852,7 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
           </el-table-column>
           <el-table-column label="证据还在不在" min-width="200">
             <template #default="s">
+              <details v-if="s.row.company"><summary>当前研究版本与观察时间</summary><div v-for="line in researchInfo(s.row.company)" :key="line">{{ line }}</div></details>
               <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>
@@ -1859,7 +1862,7 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
               <!-- 公司质地 · 失效条件 (2026-09-09 接入方案): 数据基座代码合成的质地档与评议给的失效条件, 只显示。 -->
               <div class="muted" v-if="s.row.company && s.row.company.line" :title="s.row.company.text || ''" style="font-size:11px">
                 {{ s.row.company.line }}<template v-if="s.row.company.invalidation">；失效条件：{{ s.row.company.invalidation }}</template><span
-                  v-if="s.row.company.report_url" class="lnk" style="margin-left:6px" @click.stop="openCompanyReport(s.row.ts_code)">评析报告</span></div>
+                  v-if="s.row.company.report_url" class="lnk" style="margin-left:6px" @click.stop="openCompanyReport(s.row.ts_code, s.row.company.review_ref)">评析报告</span></div>
               <!-- 兜底只在既无逻辑状态又无公司评析时出 (2026-09-11 修): 原来这条是公司质地那行的 v-else,
                    一旦没有公司评析就把 logic_now.text 再印一遍 —— 而它上面已印过一次, 于是证据印两遍。 -->
               <span v-if="!(s.row.logic_now && s.row.logic_now.state) && !(s.row.company && s.row.company.line)"
@@ -2146,7 +2149,8 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
             <div class="cs-sect" v-if="cr(propDetailP) && cr(propDetailP).invalidation"><div class="h" style="color:var(--warn)">失效条件</div><div class="st">{{ cr(propDetailP).invalidation }}</div></div>
             <div class="cs-sect" v-if="cr(propDetailP) && cr(propDetailP).antithesis"><div class="h" style="color:var(--up)">反方</div><div class="st">{{ cr(propDetailP).antithesis }}</div></div>
             <div class="cs-sect" v-if="cr(propDetailP) && cr(propDetailP).catalyst"><div class="h">催化</div><div class="st muted">{{ cr(propDetailP).catalyst }}</div></div>
-            <div class="st" v-if="cr(propDetailP) && cr(propDetailP).report_url" style="margin-top:8px"><span class="lnk" @click="openCompanyReport(propDetailP.ts_code)">查看完整评析报告</span></div>
+            <div class="st" v-if="cr(propDetailP) && cr(propDetailP).report_version_url" style="margin-top:8px"><span class="lnk" @click="openCompanyReport(propDetailP.ts_code, cr(propDetailP).review_ref)">查看本版完整评析报告</span></div>
+            <div class="muted" v-else>未保留历史报告版本；摘要不代表完整原文。</div>
             <div class="st muted" v-if="!cr(propDetailP)">这只票没有公司深度评析。</div>
           </template>
 
@@ -2565,14 +2569,16 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
           <el-table-column prop="digest" label="指纹" min-width="140"></el-table-column>
         </el-table>
         <div class="muted" style="margin-top:6px">
-          <b>同一「计划日」出现多行 = 上游那天重算过多版</b>(UPSTREAM_PLAN_API.md §7.3:
+          同一计划日可包含收到计划、实际消费及提议核验记录。计划版本、消费参数或组合上下文变化时追加记录；
-          `/plan` 是实时算的, 应答里没有 generated_at, 下游本来没办法判断手上这份是哪一版)。
+          旧记录缺少版本时标明追溯不完整，不能据此重构历史决策。
-          指纹不含 score —— 分数抖动但次序没变不算新版本, 否则表白涨而信息量为零。
         </div>
       </el-tab-pane>
     </el-tabs>
     <el-divider content-position="left">主榜明细</el-divider>
     <el-table :data="planRows" size="small" border height="420">
+      <el-table-column label="版本与研究" min-width="190"><template #default="s">
+        <details><summary>研究依据与变化</summary><div v-for="line in researchInfo(s.row)" :key="line">{{ line }}</div></details>
+      </template></el-table-column>
       <el-table-column prop="rank" label="rank" width="62"></el-table-column>
       <el-table-column prop="ts_code" label="代码" width="104"></el-table-column>
       <el-table-column prop="name" label="名称" width="92"></el-table-column>
@@ -3237,6 +3243,27 @@ createApp({
     // 标签返回 {text, cls}; cls 走 .ptag 取色 —— 技术面看多看空按页面口径红涨绿跌 (up/down),
     // 其余评价类用中性色 (accent 蓝 / warn 黄 / neutral 灰), 不掺进涨跌红绿。
     const cr = (p) => (p && p.hard_numbers && p.hard_numbers.company_review) || null;
+    function researchInfo(value) {
+      const h = value || {}, d = h.decision || {}, r = h.research || h.company_review || {};
+      const fields = {overall:'质地', valuation:'估值', doubt_hard:'财务风险', expectation:'预期差',
+        cycle:'周期', period_fi:'财务材料期', period_rf:'研报材料期', n_materials:'材料数量',
+        unknown_n:'未知问题数', numbers_unmatched:'数字核验缺项', thesis:'投资论点', antithesis:'反方论点',
+        text:'说明', gap_pts:'预期增速差（百分点）', implied_pe_vs_cons:'隐含市盈率与一致预期比',
+        grade:'评价', stance:'立场'};
+      const show = x => x == null ? '未知' : Array.isArray(x) ? x.map(show).join('；') :
+        typeof x === 'object' ? Object.entries(x).map(([k,v]) => (fields[k] || k) + '：' + show(v)).join('；') : String(x);
+      const when = x => x ? new Date(x).toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' }) + '（北京时间）' : '未知';
+      return ['计划版本：' + (d.id || '旧数据，追溯不完整'), '变更原因：' + (d.reason || '未记录'),
+        '决策截止：' + when(d.cutoff_at), '独立研究观察：' + when(h.observation_at), '报告生成：' + when(r.report_ran_at),
+        ...[['投资论点','thesis'], ['反方论点','antithesis'], ['预期差','expectation'], ['周期位置','cycle'],
+          ['财务材料期','period_fi'], ['研报材料期','period_rf'], ['材料数量','n_materials'],
+          ['未知问题数','unknown_n'], ['数字核验缺项','numbers_unmatched']].map(([label,key]) => label + '：' + show(r[key])),
+        '催化与验证事项：' + show((h.company_review || {}).catalyst),
+        '失效条件：' + show((h.company_review || {}).invalidation),
+        '研究变化：' + ((r.changes || []).length ? r.changes.map(x =>
+          (fields[x.field] || x.field) + '：' + show(x.before) + ' → ' + show(x.after)).join('；') : '无已记录变化'),
+        '执行次序：' + show(h.execution_rank) + '；' + (h.priority_reason || '未记录')];
+    }
     const propPlanTag = (p) => {
       const v = String((p.hard_numbers || {}).verdict || '').trim();
       if (!v) return null;
@@ -4261,7 +4288,8 @@ createApp({
         ElementPlus.ElMessage.warning('选「其他」时必须填补充说明'); return; }
       dd.busy = true;
       const d = await call('post', '/api/proposals/' + dd.row.proposal_id + '/decide',
-        { decision: dd.decision, reason_code: dd.code || null, reason_note: dd.note.trim() || null });
+        { decision: dd.decision, reason_code: dd.code || null, reason_note: dd.note.trim() || null,
+          proposal_revision: (dd.row.hard_numbers || {}).proposal_revision || null });
       dd.busy = false;
       if (d.ok) {
         ElementPlus.ElMessage.success(dd.decision === 'ACCEPTED'
@@ -4269,7 +4297,7 @@ createApp({
         if (d.warning) ElementPlus.ElMessage.warning(d.warning);
         dd.visible = false;
         await Promise.all([loadProposals(), loadInstructions(), loadLedger()]);
-      } else { ElementPlus.ElMessage.error(d.error); }
+      } else { ElementPlus.ElMessage.error(d.error); if (d.proposal) { dd.visible = false; await loadProposals(); } }
     }
     async function loadStrategies() {
       const q = '/api/strategies' + (showAllStrategies.value ? '?include_archived=1' : '');
@@ -4450,12 +4478,13 @@ createApp({
       if (!_mdReady) return '<pre style="white-space:pre-wrap">' + _escHtml(md) + '</pre>';
       return window.marked.parse(md);
     }
-    async function openCompanyReport(code) {
+    async function openCompanyReport(code, reviewRef = null) {
       if (!code) return;
       Object.assign(rp, { open: true, code, url: '', html: '', meta: '', loading: true, error: '' });
       // 不走 call(): 取不到报告只在抽屉里说, 不再往顶部横幅重复报一次
       let d;
-      try { d = (await api.get('/api/research/' + encodeURIComponent(code) + '/report')).data; }
+      try { d = (await api.get('/api/research/' + encodeURIComponent(code) + '/report' +
+        (reviewRef ? '?ref=' + encodeURIComponent(reviewRef) : ''))).data; }
       catch (e) { d = { ok: false, error: '评析报告读取失败: ' + ((e && e.message) || '请求失败') }; }
       if (!rp.open || rp.code !== code) return;
       rp.loading = false;
@@ -4743,7 +4772,7 @@ createApp({
       scaleGap,
       wsRaw, wsAuto, wsHidePong, wsc, wst, wsMode, wsOrders, wsInbox, wsWarnings, loadWs,
       loadAll, loadParams, saveParams, loadPlans, loadLots, onCmdChange, issue,
-      cancelCmd, replan, decide, decideDlg, decideCodes, submitDecide, ops, loadSchema, openReport, cancelIns,
+      cancelCmd, replan, decide, decideDlg, decideCodes, submitDecide, ops, loadSchema, openReport, cancelIns, researchInfo,
       planDrawer, planLoading, planStatus, planFetchedText, planRows, planCand, planResult,
       loadPlan, openPlan, refreshPlan,
       techStatus, techChipText, techChipCls,

config/settings.py

@@ -181,6 +181,7 @@ class Settings(BaseSettings):
 
     # --- 榜单变化提示 (PMS 自算; 上游 changes 字段恒为 null, 见 UPSTREAM_PLAN_API.md §9) ---
     PMS_PLAN_SNAPSHOT: bool = True         # 每次拉计划落一份名册快照 (变化提示的底本)。
+    PMS_PLAN_VERSION_RECHECK: bool = False  # 分批上线：新版计划稳定后显式开启采纳前重核。
     #   关掉只是没有变化提示, 不影响候选池; 但同时也失去"手上这份是哪一版"的自证能力
     PMS_PLAN_SNAPSHOT_KEEP: int = 200      # 快照保留份数 (一份 300 只约 20KB)
     PMS_PLAN_DIFF_RANK_JUMP: int = 50      # 名次跳变多少名才报

docs/复盘决定台账.md

@@ -249,3 +249,15 @@
 **上线。** 09-19 交付、我复审后 09-23 收盘后重启 PMS 进程（重建镜像加 force-recreate，经用户同意）。不与 09-21 选股系统拉代码同一天，避免两处同时变化说不清。
 
 **复核日期。** 上线次日 09-24 核候选序等于上游名次序；四周（10-16 前后）随选股系统排序轴一起复核。
+
+## 018 · 2026-10-15 · 每日计划消费留痕与提议采纳前核验
+
+**日期。** 2026-10-15，北京时间；用户已批准三仓联动，本轮仅本地实现，未调整线上参数。
+
+**改动。** 记录 received、consumed、rechecked，保留有效参数、顺序、过滤原因和版本指纹；自主扫描继承消费顺序，盘中信号优先留原因。采纳前仅名次变化保留提议，资格、风险或仓位依据变化重走现有判断与数量流程，人工基于更新内容重新确认；未知版本不生成新的自主买入指令。条件更新防并发重复采纳。命令仍优先并补来源到最终指令与账本。停止按行数清理快照，当前保留全部历史。
+
+**依据。** 用户批准的《每日选股计划完善整体方案》；单名额优先级、排名变化、资格失效、数量缩减、旧确认、版本不可用及并发采纳已有离线回归。桥台账 064 记录上游发布约定。
+
+**预期。** 候选到建仓次序可解释，历史消费不受今天参数影响。新发布控制 PMS_PLAN_VERSION_RECHECK 默认关闭，兼容读取和桥版本验收后才在模拟实例启用；不改变现有排序开关、档位和仓位阈值。既有指令及持仓沿原执行风控处理。
+
+**复核日期。** 获准部署后连续五个交易日记录顺序、版本、传播时延、重核结果和重复指令；具体首日待部署确定。本地测试不能替代真机判收。

scripts/run_tests.py

@@ -5,6 +5,7 @@
 运行:  docker compose run --rm pms-web python scripts/run_tests.py
 
 包含 (例数按 2026-10-12 隔离回归实跑校正):
+  test_batch38_units.py 每日计划顺序、版本核验、重新确认、并发采纳、历史保留 (15 例)
   test_core_units.py    仓位规划器 / 安全垫与成本账                     (14 例)
   test_batch2_units.py  命令状态机 / 方案生成器 / 回放对账纯逻辑         (35 例)
   test_batch3_units.py  规则闸 / 择时执行器实现B 纯逻辑 / 用户止损价只披露不拦截 (26 例)
@@ -187,6 +188,7 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py",
           "test_batch35_units.py",
           "test_batch36_units.py",
           "test_batch37_units.py",
+          "test_batch38_units.py",
           "test_page_copy_guard.py",
           "test_page_enum_guard.py",
           "test_page_wiring_guard.py",

scripts/test_batch20_units.py

@@ -284,6 +284,7 @@ def _route(c, *, values, judge_resp, open_autonomy="propose_only"):
     got = {"ledger": [], "instructions": [], "proposals": []}
     with _Patch() as p:
         p(rule_gate, "check", lambda **kw: {"passed": True, "failed": [], "warnings": []})
+        p(psvc, 'verify_open_basis', lambda *a: {'ok': True, 'provenance': {}})
         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)

scripts/test_batch38_units.py

@@ -0,0 +1,165 @@
+"""本地：python scripts/unit_test_guard.py scripts/test_batch38_units.py。禁止联网。"""
+import unit_test_guard  # noqa: F401
+import copy
+import os
+import sys
+import unittest
+from unittest.mock import patch
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from app.services import plan_feed as pf
+from app.core import plan_revision as pr
+from app.core import action_engine as ae
+from test_batch20_units import params, caps
+
+
+def selected():
+    plan = pf.parse_plan({'date': '2026-10-14', 'rank_axis': 'quality', 'decision': {'id': 'v1'},
+        'main': [{'code': 'SH600001', 'name': '甲', 'rank': 1, 'score': 210, 'verdict': '候选'},
+                 {'code': 'SH600002', 'name': '乙', 'rank': 2, 'score': 250, 'verdict': '候选'}]})
+    return pf.select_candidates(plan, rank_by_upstream=True, theme_cap=1)['items']
+
+
+class DailyPlanTest(unittest.TestCase):
+    def test_judge_cache_requires_same_material_basis(self):
+        old = {'plan_basis': {'risk': []}, 'scan_params_hash': 'p', 'portfolio_context_hash': 'c'}
+        ranked = dict(old, decision={'id': 'new'}, research_rank=99)
+        self.assertFalse(pr.needs_new_judgement(ranked, old))
+        self.assertTrue(pr.needs_new_judgement(dict(old, plan_basis={'risk': ['变化']}), old))
+        self.assertTrue(pr.needs_new_judgement(old, None))
+
+    def test_portfolio_quantity_basis_detects_new_commitments(self):
+        view = {'held': [{'ts_code': '600001.SH', 'total_qty': 100, 'sector': '电子', 'price': 10}],
+                'totals': {'cash_source': 'ws', 'cash_avail': 10000}, 'sector_ready': True}
+        commitments = {'names': [], 'amount': 0}
+        original = pr.portfolio_context(view, commitments)
+        view['held'][0]['price'] = 11
+        self.assertEqual(original, pr.portfolio_context(view, commitments))
+        commitments.update(names=['600002.SH'], amount=1000)
+        self.assertNotEqual(original, pr.portfolio_context(view, commitments))
+
+    def test_recheck_ignores_rank_cutoffs_but_keeps_qualification(self):
+        raw = {'date': '2026-10-14', 'rank_axis': 'quality', 'decision': {'id': 'v2'},
+               'main': [{'code': 'SH600001', 'rank': 1, 'score': 230, 'tier': '强传导', 'evidence': {'theme': '电子'}},
+                        {'code': 'SH600002', 'rank': 2, 'score': 210, 'tier': '强传导', 'evidence': {'theme': '电子'}}]}
+        p = dict(top_n=1, tiers=['强传导'], include_observe=False, min_score=0,
+                 min_sources=0, min_upside=0, theme_cap_local=1, exclude_st=False,
+                 rank_by_quality=True, rank_by_upstream=True, observe_if_fund_bull=False)
+        with patch.object(pf, '_params', return_value=p), patch.object(pf, 'get_plan', return_value=pf.parse_plan(raw)):
+            self.assertEqual([x['ts_code'] for x in pf.candidates(record=False)['items']], ['600001.SH'])
+            fresh = pf.candidates(record=False, recheck_code='600002.SH')
+            self.assertEqual([x['ts_code'] for x in fresh['items']], ['600002.SH'])
+            self.assertEqual(fresh['consumption']['kind'], 'rechecked')
+            self.assertEqual(pf.candidates(record=False, recheck_code='600002.SH', black=['600002.SH'])['items'], [])
+
+    def test_snapshot_retention_never_deletes_by_row_count(self):
+        from app.repo import pms_repo
+        with patch.object(pms_repo, 'execute', side_effect=AssertionError('不得按行数删除历史')):
+            self.assertEqual(pms_repo.prune_plan_snapshots(2), 0)
+
+    def test_material_change_replaces_qty_and_never_accepts_old_confirmation(self):
+        from app.services import plan_recheck as rc, proposal_service as ps, param_store
+        from app.repo import pms_repo
+        p = {'proposal_id': 'p1', 'action': 'OPEN', 'ts_code': '600001.SH', 'qty': 300,
+             'hard_numbers_json': '{}', 'hard_numbers': {'proposal_revision': 'old'}}
+        c = {'qty': 100, 'ts_code': p['ts_code'], 'hard_numbers': {'price': 10, 'plan_basis': {}}}
+        check = {'material': True, 'view': {'held': []}, 'params': {}, 'stock': {}}
+        with patch.object(rc, 'current_basis', return_value=check), \
+             patch.object(ps, '_scan_open', return_value=[c]) as scan, \
+             patch.object(ps, '_route_one', return_value={'verdict': {'verdict': 'PASS'}}) as route, \
+             patch.object(param_store, 'get_int', return_value=0), \
+             patch.object(pms_repo, 'revise_pending_proposal', return_value=1) as revise, \
+             patch.object(pms_repo, 'insert_ledger', return_value=1):
+            out = rc.before_accept(p, 'old')
+        self.assertFalse(out['ok'])
+        self.assertEqual(out['proposal']['qty'], 100)
+        self.assertNotEqual(out['proposal']['hard_numbers']['proposal_revision'], 'old')
+        self.assertEqual(scan.call_args.kwargs['exclude_proposal'], 'p1')
+        self.assertTrue(route.call_args.kwargs['review_only'])
+        self.assertTrue(c['_plan_recheck'])
+
+    def test_version_unavailable_and_qualification_loss_preserve_proposal(self):
+        from app.services import plan_recheck as rc
+        p = {'action': 'OPEN', 'ts_code': '600001.SH', 'hard_numbers': {}}
+        for check in ({'ok': False, 'error': '版本不可用'},
+                      {'ok': False, 'material': True, 'error': '资格失效'}):
+            with patch.object(rc, 'current_basis', return_value=check):
+                self.assertFalse(rc.before_accept(p, None)['ok'])
+
+    def test_compare_and_set_rejects_stale_and_duplicate_acceptance(self):
+        import sqlite3
+        import threading
+        from concurrent.futures import ThreadPoolExecutor
+        from app.repo import pms_repo
+        db = sqlite3.connect(':memory:', check_same_thread=False)
+        self.addCleanup(db.close)
+        db.execute('CREATE TABLE pms_proposal(proposal_id TEXT, status TEXT, decided_at TEXT, hard_numbers_json TEXT)')
+        db.execute("INSERT INTO pms_proposal VALUES ('p1', 'WAIT_USER', NULL, 'new')")
+        lock = threading.Lock()
+        def execute(sql, params):
+            with lock:
+                n = db.execute(sql, params).rowcount
+                db.commit()
+                return n
+        with patch.object(pms_repo, 'execute', execute):
+            self.assertEqual(pms_repo.decide_proposal('p1', 'ACCEPTED', expected_hard='old'), 0)
+            with ThreadPoolExecutor(2) as pool:
+                results = list(pool.map(lambda _: pms_repo.decide_proposal('p1', 'ACCEPTED', expected_hard='new'), range(2)))
+        self.assertEqual(sum(results), 1)
+
+    def test_rank_survives_to_single_slot(self):
+        rows = [dict(r, price=10) for r in selected()]
+        out = ae.scan_open(candidates=rows, params=params(open_signal_priority=False),
+            caps=caps(), room_amt=1_000_000, slots=1)
+        self.assertEqual(out['candidates'][0]['ts_code'], '600001.SH')
+        self.assertEqual(out['candidates'][0]['hard_numbers']['execution_rank'], 1)
+
+    def test_signal_priority_has_reason(self):
+        rows = [dict(r, price=10) for r in selected()]
+        rows[1]['sig_buy'] = {'reason': '盘中转多'}
+        out = ae.scan_open(candidates=rows, params=params(open_signal_priority=True),
+            caps=caps(), room_amt=1_000_000, slots=1)
+        self.assertEqual(out['candidates'][0]['ts_code'], '600002.SH')
+        self.assertEqual(out['candidates'][0]['hard_numbers']['priority_reason'], '盘中转多信号优先')
+
+    def test_unknown_themes_do_not_form_sector(self):
+        self.assertEqual(len(selected()), 2)
+
+    def test_rank_only_change_keeps_proposal(self):
+        row = selected()[0]
+        old = pr.basis(row)
+        row.update(selection_rank=20, research_rank=100, decision={'id': 'v2'})
+        self.assertEqual(pr.compare(old, row, row['ts_code']), 'unchanged')
+        row['risk'] = ['材料存在矛盾']
+        self.assertEqual(pr.compare(old, row, row['ts_code']), 'material')
+
+    def test_pending_or_unverifiable_does_not_pass(self):
+        row = selected()[0]
+        old = pr.basis(row)
+        row['refresh_status'] = {'pending_codes': [row['ts_code']]}
+        self.assertEqual(pr.compare(old, row, row['ts_code']), 'unavailable')
+        row.pop('refresh_status')
+        row['decision'] = None
+        self.assertEqual(pr.compare(old, row, row['ts_code']), 'unavailable')
+
+    def test_stale_human_confirmation_rejected(self):
+        from app.services import plan_recheck as rc
+        p = {'action': 'OPEN', 'ts_code': '600001.SH', 'hard_numbers': {'proposal_revision': 'new'}}
+        with patch.object(rc, 'current_basis', return_value={'ok': True, 'provenance': {}}):
+            self.assertFalse(rc.before_accept(p, 'old')['ok'])
+            self.assertTrue(rc.before_accept(p, 'new')['ok'])
+
+    def test_received_and_consumed_fingerprint_include_context(self):
+        old = {'decision': {'id': 'v1'}, 'params': {'top_n': 2}, 'order': ['A', 'B']}
+        new = copy.deepcopy(old)
+        new['params']['top_n'] = 1
+        self.assertNotEqual(pr.fingerprint(old), pr.fingerprint(new))
+        self.assertEqual(old['params']['top_n'], 2)
+
+    def test_pending_prefix_codes_normalize(self):
+        p = pf.parse_plan({'date': '2026-10-14', 'main': [{'code': 'SH600001', 'score': 1}],
+                         'refresh_status': {'pending_codes': ['SH600001']}})
+        self.assertEqual(p['refresh_status']['pending_codes'], ['600001.SH'])
+
+
+if __name__ == '__main__':
+    unittest.main()

scripts/test_batch7_units.py

@@ -500,10 +500,10 @@ def _():
     got = [(x["theme"], x["ts_code"]) for x in capped["items"]]
     assert [t for t, _ in got] == ["储能", "储能", "传感器", "传感器", "整车"], got
     assert capped["dropped"]["theme"] == 5      # 储能多 4 只 + 传感器多 1 只
-    # 主题为空的行归到「(无主题)」一档, 不跟着别的主题挤
+    # 没有主题不是同一行业，不能共享限额。
     p2 = pf.parse_plan({"date": "2026-07-29", "main": [
         {"rank": 1, "code": "SH600001", "score": 9}, {"rank": 2, "code": "SH600002", "score": 8}]})
-    assert len(pf.select_candidates(p2, top_n=5, theme_cap=1)["items"]) == 1
+    assert len(pf.select_candidates(p2, top_n=5, theme_cap=1)["items"]) == 2
 
 
 @case("取数·build_query: 显式 top/obs_top/theme_cap 覆盖 QUERY_EXTRA 同名键; 0=不传")