Add batch backtest validation and doctrine kernel reset

SESSION.md

@@ -2,13 +2,14 @@
 
 ## Current Goal
 
-Automation execution-layer PXX remains closed and its known scope stays frozen. Under explicit user direction, the active product phase has shifted from broad Decision Assistant expansion to mainline correction: strictly follow `docs/ICT交易全书_出版终校版.md`, `docs/AI_ICT_硬性主线约束_必须遵循ICT交易全书_2026-05-21_v1.md`, and `docs/AI_ICT_主线纠偏与功能裁剪方案_2026-05-20_v1.md`; keep only Trader Cockpit / Strategy Backtest / Trading Plan / Journal / System Ops in the trader-facing shell; productize existing ICT backtest evidence before adding any new feature surface.
+Automation execution-layer PXX remains closed and its known scope stays frozen. Under explicit user direction, the active product phase has shifted from broad Decision Assistant expansion and surface-shape candidates to `ICT Doctrine Kernel / 全书规则内核重建`: strictly follow `docs/ICT交易全书_出版终校版.md`, `docs/AI_ICT_硬性主线约束_必须遵循ICT交易全书_2026-05-21_v1.md`, `docs/AI_ICT_主线纠偏与功能裁剪方案_2026-05-20_v1.md`, and `docs/AI_ICT_DoctrineKernel全书规则内核重建_D0_2026-05-21_v1.md`; keep only Trader Cockpit / Strategy Backtest / Trading Plan / Journal / System Ops in the trader-facing shell; treat old sweep/MSS/FVG/retrace candidates as research-only material until they are remapped through Doctrine setup playbooks and pass historical evidence gates.
 
 ## Hard Mainline Gate
 
 - Highest-priority rule: every trading-analysis slice must map to `docs/ICT交易全书_出版终校版.md`.
 - Required chain: Range / Draw / Location -> Liquidity / Sweep / Inducement / Path -> Displacement / MSS / FVG / Retest / Session -> backtest evidence -> pre-market plan -> intraday unique focus or no-trade reason -> journal / study-notes feedback.
 - If a slice does not serve this chain, it must not enter the trader-facing product; necessary diagnostics belong under `System Ops`.
+- Existing surface-shape candidates are evidence components only. They are not a complete setup and must not enter `Review Now` without Doctrine setup mapping plus backtest evidence.
 - Do not restart external-context/archive/digest/workflow/handoff long-tail work unless it directly supports the ICT book mainline.
 - This gate overrides older backlog items, old phase plans, and ambiguous prompts.
 
@@ -842,11 +843,13 @@ Automation execution-layer PXX remains closed and its known scope stays frozen.
 - S13 read-only runtime blocker recovery and fifth real feedback sample are complete: remote `.env` had empty `OKX_API_KEY` / `OKX_API_SECRET` / `OKX_API_PASSPHRASE`, so the existing local read-only OKX values were restored into the remote runtime env only; `LIVE_TRADING_ENABLED` and all execution/readiness gates stayed unchanged, and temporary credential fragments/backups were removed. Remote runtime now has `current_blockers=[]`, `readonly_account_context_status=available`, `live_analysis_assistant_ready=true`, `decision_support_ready=false`, `execution_assist_ready=false`, `real_live_ready=false`, and `must_not_trade=true`. Then the existing shadow-observation feedback path recorded a distinct `ALL` / `no_candidate_monitor` observation `7949` as `operator_agreement=agree`, moving `operator_feedback_count` from `4/5` to `5/5` without creating paper tickets, paper executions, or weekly review entries. Current R2 state: `next_validation_gate=paper_ticket_sample_gate`, `operator_feedback_count=5/5`, `operator_observation_feedback_count=5`, `operator_feedback_day_count=3`, `paper_ticket_count=0`, `paper_execution_count=0`, and `weekly_review_entry_count=0`. Validation passed documentation consistency tests, `git diff --check`, remote ops smoke, readonly live analysis smoke, direct shadow-paper field smoke, and gstack console check.
 - S14 R2 readonly fallback candidate-refresh closure is complete: R2 shadow-paper validation and sample collection now use the same read-only runtime blocker policy as Maintenance / Unattended Harness, so `market_stream_degraded` with fresh REST candles and clear data gaps is surfaced as a suppressed warning instead of blocking `runtime_clear_for_shadow`. Runtime `watch` also now continues read-only candidate scanning when REST market data is fresh and gap-clear even if the stream is still connecting, and passes `latest_market_state_by_symbol` into shadow-paper observation so fallback blockers can be suppressed consistently. Local validation passed focused runtime/readiness/maintenance/unattended tests, full unit discovery (`811` tests), script static checks (`78` scripts), py_compile, documentation consistency, and `git diff --check` with only LF/CRLF warnings. Remote deploy passed 11 endpoint deploy smoke, `run_remote_ops_smoke.py`, and `run_readonly_live_analysis_smoke.py`; field smoke showed `watch_market_input_mode=rest_fallback`, no `watch_skipped_market_degraded`, and new Candidate `4612` generated. Candidate `4612` is still correctly `blocked` / `executable=false` because `fvg_retrace_confirmed` is missing and `fvg_direction_mismatch` is rejected, so no paper ticket or paper execution was created.
 - S15 R2 paper gate live polling is complete with a real no-trade blocker: after S14, remote runtime was triggered again and then polled through three short-window attempts. Runtime stayed usable with `current_blockers=[]` and `watch_market_input_mode=rest_fallback`; R2 validation stayed correctly at `status=collecting_paper_samples`, `next_validation_gate=paper_ticket_sample_gate`, `operator_feedback_count=5`, `paper_ticket_count=0`, `paper_execution_count=0`, and `weekly_review_entry_count=0`. The newest real candidate advanced to Candidate `4615`, but it remained `blocked` / `executable=false` because `fvg_retrace_confirmed` is still missing and `fvg_direction_mismatch` is rejected. No paper ticket, paper execution, or weekly review evidence was created, because blocked candidates must remain observation/no-trade only.
-- Current assisted-analysis next gate: stop collecting duplicate operator feedback; wait for a real executable candidate before creating shadow paper ticket / paper execution samples, then let real execution-derived Journal / weekly review evidence appear. Do not fabricate paper samples from blocked candidates or no-candidate observations. Do not add new product features, do not restart external-context/archive/digest long-tail work, do not physically delete S1 deprecated surfaces yet, and keep `execution_allowed`, `decision_support_ready`, `real_live_ready`, risk governor, trust bucket, and operator confirmation protocol unchanged.
+- S16 batch historical backtest validation is complete: Strategy Backtest now has `scripts/run_backtest_batch.py` for matrix runs across instrument / subject / timeframe / session / candidate model, and `BacktestService.run_backtest` persists `run_metadata.batch_id` while keeping the read-only boundary `read_only_backtest_no_order_submission`. Local focused backtest tests, `py_compile`, script static checks (`79` scripts), documentation consistency, full unit discovery (`814` tests), and `git diff --check` passed. Local batch smoke correctly blocked on the known local DB privilege issue for `trading_plans`, so the decisive validation was remote: deploy smoke passed 11 endpoints, remote ops smoke and readonly live analysis smoke passed, remote smoke batch `S16-remote-20260521-a` wrote run ids `490-495`, and the larger remote matrix `S16-remote-20260521-b` wrote run ids `496-513`. The larger batch ran `18` historical no-lookahead runs across 3 timeframe sets and 5 session groups with `211` total results and `156` executed signals; `0/18` passed the quality gate. Main blockers were `expectancy_below_threshold=18`, `profit_factor_below_threshold=18`, `executed_sample_too_small=16`, and `sample_size_too_small=15`; failure reasons were dominated by `stopped_out=124`, `outside_session_window=126`, `no_future_candles=54`, and `same_candle_entry_exit_ambiguous=30`. `/trading/backtests?instrument_id=1&limit=18` returned latest id `513` and batch metadata `S16-remote-20260521-b` for ids `496-513`, so the existing Strategy Backtest page can read the batch output. This proves the batch backtest capability is usable, but the current strategy evidence is not good enough to promote into Trader Cockpit `Review Now`.
+- D0 Doctrine Kernel stage reset is complete locally: added `docs/AI_ICT_DoctrineKernel全书规则内核重建_D0_2026-05-21_v1.md` as the new phase anchor. D0 explicitly downgrades existing surface-shape candidates (`sweep / MSS / FVG / retrace`, including `OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE`) to `research_only` / `study_only` evidence components unless they are remapped through a full Doctrine setup playbook and pass historical evidence gates. The reset keeps the current infrastructure but changes product priority: do not continue expanding old candidates; first extract Doctrine Rules from the full book, define setup playbooks, then backtest by playbook. No execution/readiness/risk/trust/operator-confirmation gate was changed.
+- Current assisted-analysis next gate: keep R2 executable-candidate waiting in the background; do not collect duplicate operator feedback or fabricate paper samples. The active product work is now D1 `Market Language Kernel`: extract the first Doctrine Rule matrix from 《ICT交易全书》第 1-6 章, build the first `Range / Draw / Location / Time / Invalidation` registry, and mark old candidates research-only whenever those fields are missing. Do not add order-submission features, do not restart external-context/archive/digest long-tail work, do not physically delete S1 deprecated surfaces yet, and keep `execution_allowed`, `decision_support_ready`, `real_live_ready`, risk governor, trust bucket, and operator confirmation protocol unchanged.
 
 ## Exact Next Step
 
-Use `docs/AI_ICT_实盘辅助分析决策助手ICT主线开发方案_2026-05-21_v1.md`, `docs/AI_ICT_ICT主线概念字典_S2_2026-05-21_v1.md`, `docs/AI_ICT_ReviewNow主线Gate_S3S4_2026-05-21_v1.md`, `docs/AI_ICT_Cockpit链路句与反馈错误分类_S5S6_2026-05-21_v1.md`, `docs/AI_ICT_只读实盘可靠性与交易员SOP_S7S8_2026-05-21_v1.md`, `docs/AI_ICT_TraderCockpit状态话术压缩_S9_2026-05-21_v1.md`, and `docs/AI_ICT_TraderCockpit真实反馈验证_S10_2026-05-21_v1.md` as the active anchors. Continue with the R2 paper gate only: wait for a real executable candidate after the S15 polling confirmed Candidate `4615` is still blocked, then create paper ticket / paper execution evidence through the existing shadow-paper path and weekly review flow. Do not keep clicking feedback now that `operator_feedback_count=5/5`; do not restart external-context/archive/digest long-tail work, do not create order-submission features, do not physically delete S1 deprecated surfaces yet, and do not fabricate R2 paper samples from blocked candidates. Keep `execution_allowed`, `decision_support_ready`, `real_live_ready`, risk governor, trust bucket, and operator confirmation protocol unchanged.
+Use `docs/AI_ICT_DoctrineKernel全书规则内核重建_D0_2026-05-21_v1.md`, `docs/ICT交易全书_出版终校版.md`, `docs/AI_ICT_实盘辅助分析决策助手ICT主线开发方案_2026-05-21_v1.md`, `docs/AI_ICT_批量历史回测验证_S16_2026-05-21_v1.md`, `SESSION.md`, and `TASKS.md` as the active anchors. Start D1 `Market Language Kernel`: extract Doctrine Rules from chapters 1-6, produce the first `Range / Draw / Location / Time / Invalidation` rule registry, and update the existing candidate/backtest/cockpit language so old sweep/MSS/FVG/retrace candidates are clearly `research_only` unless they map to a full setup playbook and pass historical gates. Keep R2 executable-candidate waiting in the background; do not keep clicking feedback now that `operator_feedback_count=5/5`; do not restart external-context/archive/digest long-tail work, do not create order-submission features, do not physically delete S1 deprecated surfaces yet, and do not fabricate R2 paper samples from blocked candidates. Keep `execution_allowed`, `decision_support_ready`, `real_live_ready`, risk governor, trust bucket, and operator confirmation protocol unchanged.
 
 ## Validation Commands Used Recently
 

TASKS.md

@@ -7,6 +7,7 @@
 - [x] Each stage must close the loop through implementation, validation, deploy/smoke when possible, and state writeback
 - [x] Hard mainline gate active: every trading-analysis task must strictly follow `docs/ICT交易全书_出版终校版.md` and `docs/AI_ICT_硬性主线约束_必须遵循ICT交易全书_2026-05-21_v1.md`
 - [x] Any task that cannot map to Range/Draw/Location -> Liquidity/Sweep/Path -> Displacement/MSS/FVG/Retest/Session -> backtest -> plan -> intraday focus/no-trade -> journal feedback must not enter the trader-facing product
+- [x] Doctrine Kernel reset active: old sweep/MSS/FVG/retrace candidates are research-only evidence components unless remapped through full book-derived setup playbooks and historical quality gates
 - [x] External-context/archive/digest/workflow/handoff long-tail work stays frozen unless it directly supports the ICT book mainline
 - [x] P0-P6 Research Council phase expansion is frozen
 - [x] P6 `no_op_healthy` must not create Research Council P7/P8
@@ -946,10 +947,12 @@
 - [x] S13 read-only runtime blocker and fifth feedback sample closure: restored blank remote OKX read-only credential keys from local runtime values without changing live trading flags, confirmed `runtime_status.current_blockers=[]`, recorded distinct `ALL/no_candidate_monitor` observation `7949` as `operator_agreement=agree`, and moved `operator_feedback_count` from `4/5` to `5/5`
 - [x] S14 R2 readonly fallback candidate-refresh closure: shadow-paper validation/sample collection now share the read-only blocker policy, runtime watch continues scanning on fresh REST candles with clear gaps while stream is connecting, remote smoke passed, and new Candidate `4612` was generated but remains blocked (`fvg_retrace_confirmed` missing, `fvg_direction_mismatch` rejected)
 - [x] S15 R2 paper gate live polling closure: remote runtime stayed usable, short-window polling generated latest Candidate `4615`, but executable count remained `0`; no paper ticket/execution was created because the candidate is still blocked by missing `fvg_retrace_confirmed` and rejected `fvg_direction_mismatch`
+- [x] S16 batch historical backtest validation closure: added `scripts/run_backtest_batch.py`, persisted `run_metadata.batch_id`, documented `docs/AI_ICT_批量历史回测验证_S16_2026-05-21_v1.md`, passed focused/static/docs/full-unit validation, deployed to remote, ran smoke batch `S16-remote-20260521-a` and larger matrix `S16-remote-20260521-b`, wrote run ids `490-513`, and confirmed `/trading/backtests` can read the latest batch metadata; current matrix evidence is `0/18` quality-gate passes, so existing setup evidence must stay study-only / no promotion
+- [x] D0 ICT Doctrine Kernel stage reset: created `docs/AI_ICT_DoctrineKernel全书规则内核重建_D0_2026-05-21_v1.md`, made it an active doc anchor, downgraded surface-shape candidates to research/study-only in the product roadmap, and defined the new Doctrine Rule -> Setup Playbook -> Backtest Cohort -> Trading Plan -> Cockpit -> Journal chain
 - [ ] R2 paper-ticket sample gate is waiting on a real executable candidate; do not fabricate paper tickets from blocked candidates
 - [x] R2 operator feedback sample gate satisfied with distinct real samples (`operator_feedback_count=5/5`); do not keep clicking feedback to pad counts
 - [ ] R2 weekly review visible gate still needs a real execution-derived journal/weekly review entry (`weekly_review_entry_count=0`)
-- [ ] Next R2 gate: wait for real executable candidate, then collect paper ticket / paper execution / execution-derived weekly review evidence through existing read-only shadow-paper flow; no live order submission and no fabricated samples
+- [ ] D1 next product gate: extract 《ICT交易全书》第 1-6 章 into a Doctrine Rule matrix, build the first `Range / Draw / Location / Time / Invalidation` registry, and make old candidates expose `research_only_reason` when any required book-field is missing; no live order submission and no fabricated samples
 - [x] R3 OKX sandbox isolated smoke tooling: guarded script and service can call OKX demo submit/query/cancel/query/fills only when `OKX_SANDBOX_SMOKE_EXECUTE=1`; local validation, deploy smoke, and remote ops smoke passed
 - [ ] R3 sandbox credential permission gate deferred: current phase is assisted analysis / decision support only, so do not request trade-permission credentials or make submit/cancel smoke the next gate
 - [ ] Long-term target: complete `Real-time Assisted Trading V1` without enabling default live auto-trading

docs/AI_ICT_DoctrineKernel全书规则内核重建_D0_2026-05-21_v1.md

@@ -0,0 +1,263 @@
+# AI_ICT Doctrine Kernel 全书规则内核重建 D0 2026-05-21 v1
+
+## 0. 阶段判定
+
+本阶段正式承认：旧系统只吸收了《ICT交易全书》中的部分表面形态，例如 sweep、MSS、FVG、retrace。它不足以支撑“帮助交易员稳定盈利”的目标。
+
+S16 批量历史回测已经给出证据：远端矩阵 `S16-remote-20260521-b` 共 `18` 个 no-lookahead run、`211` 个历史结果、`156` 个 executed signals，质量门 `0/18` 通过。这个结果不是单纯参数问题，而是 setup 抽象不够完整：旧候选器没有完整表达书中的 Range、Draw、Location、Time、Invalidation、IRL/ERL、no-trade 条件与复盘错误闭环。
+
+因此从 D0 开始，开发主线切换为：
+
+```text
+ICT Doctrine Kernel / 全书规则内核重建
+```
+
+目标不是继续美化旧候选器，而是从《ICT交易全书》中重建可验证、可回测、可复盘的 setup playbook。
+
+## 1. 立即降级
+
+现有表面形态候选器从 D0 开始降级为研究材料：
+
+- `sweep / MSS / FVG / retrace` 只能作为 evidence component，不再等同于 setup。
+- `OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE` 等旧模型不得直接进入 Trader Cockpit `Review Now`。
+- 旧候选可以继续被记录、回测和复盘，但只能标记为 `research_only` / `study_only`。
+- 任何未经过 Doctrine setup mapping 和历史质量门的候选，默认只能输出 `Observe Only` 或 `Do Not Use`。
+
+这个降级不删除历史数据、不破坏现有页面、不停止只读 runtime。它只改变产品解释权：旧候选器不再被视为主交易逻辑。
+
+## 2. 不变边界
+
+本阶段仍然严格只读：
+
+- 不下单
+- 不提交 OKX trade order
+- 不改变 `execution_allowed`
+- 不改变 `decision_support_ready`
+- 不改变 `real_live_ready`
+- 不改变 risk governor
+- 不改变 trust bucket
+- 不改变 operator confirmation protocol
+- 不伪造 R2 paper ticket / paper execution / weekly review 样本
+
+R2 executable candidate watch 可以继续后台等待，但它不再主导产品开发节奏。
+
+## 3. 新内核定义
+
+Doctrine Kernel 必须把书本内容拆成三层。
+
+### 3.1 Doctrine Rule
+
+每条规则来自《ICT交易全书》的明确章节或执行模板，必须包含：
+
+- `book_anchor`: 章节 / 小节 / 关键公式。
+- `rule_statement`: 交易员可理解的一句话规则。
+- `required_inputs`: 系统需要哪些数据才能判断。
+- `measurable_fields`: 可被代码计算或标注的字段。
+- `invalid_when`: 何时该规则失效。
+- `no_trade_when`: 何时必须不交易。
+- `journal_error_class`: 违反该规则时的复盘错误类型。
+- `backtest_observable`: 回测中如何验证这条规则是否有效。
+
+示例：
+
+```text
+Rule: Range 未定义不得交易
+Anchor: 第1章 / 第2章
+Statement: 未定义当前 Dealing Range 时，任何 FVG/MSS 都只是研究材料。
+No-trade: 缺 Key High / Key Low / Equilibrium / Premium-Discount 位置。
+Journal error: traded_without_range
+Backtest observable: setup_without_range_definition
+```
+
+### 3.2 Setup Playbook
+
+真正可交易的 setup 必须是完整 playbook，而不是一个形态标签。每个 playbook 必须包含：
+
+- `setup_id`
+- `setup_name`
+- `book_anchors`
+- `market_narrative`
+- `range_definition`
+- `draw_rule`
+- `location_rule`
+- `liquidity_event_rule`
+- `displacement_structure_rule`
+- `pd_array_rule`
+- `session_timing_rule`
+- `confirmation_rule`
+- `invalidation_rule`
+- `target_rule`: IRL / ERL / partial / full target
+- `no_trade_conditions`
+- `backtest_cohort`
+- `promotion_gate`
+- `cockpit_sentence`
+- `journal_error_classes`
+
+未满足以上字段的 setup 不能进入 Trader Cockpit。
+
+### 3.3 Evidence Chain
+
+所有 Trader Cockpit 输出必须来自如下链路：
+
+```text
+Doctrine Rule
+-> Setup Playbook
+-> Historical Backtest Cohort
+-> Trading Plan
+-> Intraday Evidence
+-> Cockpit Conclusion
+-> Journal Feedback
+```
+
+如果链路断在任何一环，系统必须输出：
+
+```text
+Observe Only / Do Not Use
+```
+
+不能输出交易倾向、买卖建议或伪装成信号的状态。
+
+## 4. 全书吸收顺序
+
+D 阶段按书本主线吸收，不按功能菜单吸收。
+
+### D1 Market Language Kernel
+
+吸收第 1-6 章：
+
+- Price Delivery / IPDA
+- Dealing Range
+- Context / Framework / Reference Point
+- Draw on Liquidity
+- Premium / Discount / Equilibrium / OTE
+- IRL / ERL
+
+交付：
+
+- `Range + Draw + Path + Time + Invalidation` 规则矩阵。
+- Dealing Range / Premium-Discount / IRL-ERL 数据字段定义。
+- 旧候选器缺失这些字段时的 `research_only` 原因。
+
+### D2 Liquidity Kernel
+
+吸收第 7-12 章：
+
+- BSL / SSL
+- Liquidity Pool
+- Liquidity Void
+- Stop Run / Sweep / Raid
+- Judas Swing / Manipulation
+- 多目标 Draw 优先级
+
+交付：
+
+- 流动性目标排序规则。
+- sweep 不是 setup 的硬规则。
+- manipulation 与 session timing 的 no-trade / wait 规则。
+
+### D3 PD Array / Reaction Kernel
+
+吸收全书 PD Arrays 与反应位章节：
+
+- FVG
+- Order Block
+- Breaker
+- Mitigation
+- Rebalance / Void
+- PD array 质量排序
+
+交付：
+
+- FVG/MSS 从“入场触发”降级为“完整链路中的确认组件”。
+- PD array 选择必须服从 Range、Draw、Location。
+
+### D4 Time / Session Kernel
+
+吸收 session / kill zone / day model / week model：
+
+- Asia / London / NY AM / NY PM
+- Judas / manipulation window
+- opening range
+- weekly profile
+- news / no-trade time
+
+交付：
+
+- setup 必须绑定可验证时段。
+- session 之外的候选默认 study-only。
+
+### D5 Model Playbook Kernel
+
+把书中模型整理为可回测 playbook：
+
+- continuation model
+- reversal model
+- sweep-to-displacement model
+- HTF draw pullback model
+- London / NY specific model
+
+交付：
+
+- 每个模型形成完整 setup playbook。
+- 每个 playbook 都能生成 batch backtest cohort。
+
+### D6 Trader Product Rebuild
+
+只把通过回测与人工复核的 playbook 放回交易员产品：
+
+- Trader Cockpit 只显示合格 playbook。
+- Strategy Backtest 以 playbook 为核心，而不是以裸信号为核心。
+- Trading Plan 必须引用 playbook。
+- Journal 必须按 playbook 错误类型复盘。
+
+## 5. 保留 / 冻结 / 重做
+
+### 保留
+
+- 行情与 K 线数据管道。
+- 只读 runtime。
+- Strategy Backtest 批量 runner。
+- Trader Cockpit 外壳。
+- Trading Plan / Journal / System Ops。
+- R2 后台 observation 机制。
+
+### 冻结
+
+- 外部 context 长尾。
+- archive / workflow / digest 扩展。
+- 旧 Decision Assistant 大面板。
+- 任何未映射书本规则的候选扩展。
+
+### 重做
+
+- setup 定义。
+- candidate promotion gate。
+- backtest cohort grouping。
+- Trader Cockpit 的 `Historical Edge` 来源。
+- Journal error taxonomy。
+
+## 6. D0 闭环标准
+
+D0 完成时必须满足：
+
+- 新阶段文档成为活锚点。
+- `SESSION.md` / `TASKS.md` 明确下一步从 Doctrine Kernel 开始。
+- 旧候选器被文字上降级为 `research_only` / `study_only`。
+- S16 `0/18` 质量门结果被明确用作重建理由。
+- 不做任何交易权限、下单、执行 readiness 改动。
+
+## 7. 下一步最小闭环
+
+下一步不是继续扩大旧模型回测，而是做 D1：
+
+```text
+D1 Market Language Kernel
+```
+
+最小交付：
+
+1. 从《ICT交易全书》第 1-6 章抽取 Doctrine Rule matrix。
+2. 生成第一版 `Range / Draw / Location / Time / Invalidation` 规则注册表。
+3. 给旧候选器补 `research_only_reason`：缺 Range、缺 Draw、缺 Location、缺 Invalidation、缺 session/time 任一项都不能 promoted。
+4. 让 Strategy Backtest 按 Doctrine Rule 缺失项聚合失败原因。
+5. 更新 Trader Cockpit：没有完整 playbook 时，直接说明“当前只有研究材料，没有可交易 setup”。

docs/AI_ICT_批量历史回测验证_S16_2026-05-21_v1.md

@@ -0,0 +1,156 @@
+# AI_ICT 批量历史回测验证 S16 2026-05-21 v1
+
+## 目标
+
+把 Strategy Backtest 从“能查看最近回测结果”推进到“可以批量跑历史验证矩阵”。R2 继续由 runtime 后台等待真实 executable candidate；开发主线转向用历史数据验证 ICT setup 是否有优势。
+
+## 当前完成度判断
+
+在 S16 之前，回测能力不能算“大量历史回测完成”。当时已经完成的是：
+
+- Strategy Backtest 只读页面 `/trading/backtests`。
+- 单次回测脚本 `scripts/run_backtest.py`。
+- 最近回测、质量门、样本、R 指标、失败原因、最近 run 展示。
+
+S16 补齐的是批量历史验证入口：一次性按 instrument / subject / timeframe / session / candidate model 组合跑多个回测 run，并把所有 run 写回现有 `backtest_runs` / `backtest_results`，用于快速判断哪些 ICT setup 不值得盘中关注。
+
+仍未把它夸大成完整量化研究平台：本阶段不做参数自动寻优、不做外部数据商接入、不做交易执行、不把未过质量门的历史统计提升为实盘信号。
+
+本阶段仍然只读：
+- 不下单
+- 不提交 OKX
+- 不改变 `execution_allowed`
+- 不改变 `decision_support_ready`
+- 不改变 `real_live_ready`
+- 不伪造 paper / execution / weekly review 样本
+
+## 新增入口
+
+```bash
+python scripts/run_backtest_batch.py
+```
+
+默认行为：
+- instrument: `1`
+- subject: `plan_replay,signal`
+- mode: `replay_no_lookahead`
+- limit: `1000`
+- session groups: `ALL;LONDON;NY_AM;NY_PM;ASIA`
+- 每个 run 写入现有 `backtest_runs` / `backtest_results`
+- 输出批量质量门汇总
+
+## 常用参数
+
+通过环境变量控制：
+
+```bash
+BACKTEST_BATCH_OUTPUT_JSON=1
+BACKTEST_BATCH_ID=bt-ict-mainline-001
+BACKTEST_BATCH_INSTRUMENT_IDS=1,2
+BACKTEST_BATCH_SUBJECT_TYPES=plan_replay,signal
+BACKTEST_BATCH_MODES=replay_no_lookahead
+BACKTEST_BATCH_LIMIT=2000
+BACKTEST_BATCH_SESSION_GROUPS=ALL;LONDON;NY_AM;NY_PM;ASIA
+BACKTEST_BATCH_TIMEFRAME_SETS=1m/1m/1m;15m/5m/1m
+python scripts/run_backtest_batch.py
+```
+
+PowerShell 示例：
+
+```powershell
+$env:BACKTEST_BATCH_OUTPUT_JSON='1'
+$env:BACKTEST_BATCH_ID='bt-ict-mainline-001'
+$env:BACKTEST_BATCH_LIMIT='2000'
+$env:BACKTEST_BATCH_SUBJECT_TYPES='plan_replay,signal'
+$env:BACKTEST_BATCH_SESSION_GROUPS='ALL;LONDON;NY_AM;NY_PM;ASIA'
+$env:BACKTEST_BATCH_TIMEFRAME_SETS='1m/1m/1m;15m/5m/1m'
+.venv\Scripts\python.exe scripts\run_backtest_batch.py
+```
+
+质量门参数沿用单次回测：
+
+```bash
+BACKTEST_MIN_TOTAL_SIGNALS=100
+BACKTEST_MIN_EXECUTED_SIGNALS=30
+BACKTEST_MIN_EXPECTANCY_R=0
+BACKTEST_MIN_PROFIT_FACTOR_R=1.2
+BACKTEST_MAX_DRAWDOWN_R=10
+BACKTEST_MAX_CONSECUTIVE_LOSSES=5
+```
+
+## 输出如何读
+
+关键字段：
+- `run_count`: 本批次实际跑了多少个回测 run。
+- `passed_quality_gate_count`: 通过质量门的 run 数。
+- `blocked_quality_gate_count`: 被质量门阻断的 run 数。
+- `total_result_count`: 总历史样本数。
+- `total_executed_signals`: 总执行样本数。
+- `quality_gate_reason_counts`: 质量门失败原因聚合。
+- `failure_reason_counts`: setup / replay 失败原因聚合。
+- `aggregate_by_subject`: plan replay 与 signal 回测分层。
+- `aggregate_by_session_group`: session 分层。
+- `top_runs_by_expectancy`: 按 executed expectancy 排序的候选 run。
+
+建议先看三件事：
+
+1. `passed_quality_gate_count` 是否大于 0。
+2. `quality_gate_reason_counts` 中最大阻断项是什么。
+3. `aggregate_by_session_group` 哪些 session 的样本和执行结果明显更差。
+
+## 与 Trader Cockpit 的关系
+
+批量回测不是买卖信号。它只回答：
+
+```text
+这个 ICT setup 在历史样本里是否有足够证据，值得进入盘中观察？
+```
+
+如果样本太小、expectancy 不足、profit factor 不足、回撤过大、连续亏损过高，Trader Cockpit 必须继续降级为 Observe Only 或 Do Not Use。
+
+## S16 闭环标准
+
+- 批量 runner 可本地测试。
+- 远端可执行真实批量回测。
+- 新 run 写入 `backtest_runs`，可被 `/trading/backtests` 读取。
+- 输出清楚显示哪些 setup/session 不可信。
+- 不影响 R2 实盘监控后台等待。
+
+## 首批远端验证结果
+
+2026-05-21 首批远端 smoke 验证：
+
+- batch id: `S16-remote-20260521-a`
+- run ids: `490-495`
+- run count: `6`
+- total result count: `79`
+- total executed signals: `66`
+- quality gate: `0/6` passed
+- main quality blockers: `expectancy_below_threshold=6`, `profit_factor_below_threshold=6`, `sample_size_too_small=5`, `executed_sample_too_small=5`
+- main failure reasons: `stopped_out=52`, `outside_session_window=42`, `no_future_candles=12`, `same_candle_entry_exit_ambiguous=10`
+
+结论：批量回测能力已经跑通并落库，但当前策略样本不支持进入 Trader Cockpit `Review Now`。这正是回测应发挥的作用：先把历史上没有优势的 setup 拦住，而不是等盘中再靠主观判断。
+
+同日扩大一轮 timeframe/session 矩阵：
+
+- batch id: `S16-remote-20260521-b`
+- run ids: `496-513`
+- run count: `18`
+- total result count: `211`
+- total executed signals: `156`
+- quality gate: `0/18` passed
+- timeframe sets: `1m/1m/1m`, `15m/5m/1m`, `1h/15m/5m`
+- session groups: `ALL`, `LONDON`, `NY_AM`, `NY_PM`, `ASIA`
+- main quality blockers: `expectancy_below_threshold=18`, `profit_factor_below_threshold=18`, `executed_sample_too_small=16`, `sample_size_too_small=15`
+- main failure reasons: `stopped_out=124`, `outside_session_window=126`, `no_future_candles=54`, `same_candle_entry_exit_ambiguous=30`
+
+扩大矩阵后的结论更明确：当前抽象出来的 setup 在现有历史样本里没有优势，必须继续停留在 study / observe-only，不能进入实盘 Review Now。
+
+## 下一步
+
+S16 之后，主线不应该继续等待 R2。应优先用批量回测积累历史证据：
+
+- 先跑 BTC/ETH/SOL 的同一套 session/timeframe 矩阵。
+- 再对通过质量门或接近质量门的 setup 做人工复核。
+- 未通过质量门的 setup 不允许进入 Trader Cockpit `Review Now`。
+- 如果长期没有任何 run 通过质量门，下一步不是实盘试，而是回到《ICT交易全书》的 setup 定义，检查策略抽象是否偏离书本。

scripts/run_backtest_batch.py

@@ -0,0 +1,312 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Optional
+import json
+import os
+import sys
+
+from sqlalchemy.exc import SQLAlchemyError
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from scripts.run_backtest import (
+    build_cost_config_from_env,
+    build_quality_gate_from_env,
+    build_replay_config_from_env,
+)
+from src.services.backtest import (
+    BACKTEST_MODE_REPLAY_NO_LOOKAHEAD,
+    BACKTEST_SUBJECT_PLAN_REPLAY,
+    BACKTEST_SUBJECT_SIGNAL,
+    BacktestService,
+    evaluate_backtest_quality,
+    summarize_backtest_outcomes,
+)
+
+
+DEFAULT_SESSION_GROUPS = "ALL;LONDON;NY_AM;NY_PM;ASIA"
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    output_json = _get_bool(env, "BACKTEST_BATCH_OUTPUT_JSON", _get_bool(env, "BACKTEST_OUTPUT_JSON", False))
+    require_quality_gate = _get_bool(env, "BACKTEST_BATCH_REQUIRE_QUALITY_GATE", False)
+    batch_id = str(env.get("BACKTEST_BATCH_ID") or _default_batch_id()).strip()
+    try:
+        payload = run_backtest_batch(
+            service=BacktestService(),
+            batch_id=batch_id,
+            instrument_ids=build_instrument_ids_from_env(env),
+            limit=_get_int(env, "BACKTEST_BATCH_LIMIT", _get_int(env, "BACKTEST_LIMIT", 1000)),
+            subject_types=build_subject_types_from_env(env),
+            modes=build_modes_from_env(env),
+            timeframe_sets=build_timeframe_sets_from_env(env),
+            session_groups=build_session_groups_from_env(env),
+            candidate_model_codes=build_candidate_model_codes_from_env(env),
+            require_quality_gate=require_quality_gate,
+            quality_gate=build_quality_gate_from_env(env),
+            cost_config=build_cost_config_from_env(env),
+            replay_config=build_replay_config_from_env(env),
+        )
+    except SQLAlchemyError as exc:
+        payload = {
+            "status": "blocked",
+            "mode": "batch_historical_backtest",
+            "batch_id": batch_id,
+            "error_type": exc.__class__.__name__,
+            "error": str(exc),
+            "execution_boundary": "read_only_backtest_no_order_submission",
+        }
+        if output_json:
+            print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True))
+        else:
+            print(f"Database unavailable for batch backtest: {exc.__class__.__name__}")
+        return 2
+
+    if output_json:
+        print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True))
+    else:
+        print(f"Batch backtest id: {payload['batch_id']}")
+        print(f"Runs: {payload['run_count']} / quality gate passed: {payload['passed_quality_gate_count']}")
+        print(f"Blocked quality gates: {payload['blocked_quality_gate_count']}")
+        print(f"Total results: {payload['total_result_count']}; executed: {payload['total_executed_signals']}")
+        if payload["quality_gate_reason_counts"]:
+            print(f"Top quality gate reasons: {_format_counts(payload['quality_gate_reason_counts'])}")
+        print(payload["operator_summary"])
+    return 2 if require_quality_gate and payload["blocked_quality_gate_count"] else 0
+
+
+def run_backtest_batch(
+    *,
+    service: BacktestService,
+    batch_id: str,
+    instrument_ids: list[int],
+    limit: int,
+    subject_types: list[str],
+    modes: list[str],
+    timeframe_sets: list[dict],
+    session_groups: list[list[str]],
+    candidate_model_codes: list[Optional[str]],
+    require_quality_gate: bool,
+    quality_gate,
+    cost_config,
+    replay_config,
+) -> dict:
+    results = []
+    quality_gate_reason_counts: dict[str, int] = {}
+    failure_reason_counts: dict[str, int] = {}
+    aggregate_by_subject: dict[str, dict] = {}
+    aggregate_by_session_group: dict[str, dict] = {}
+    total_result_count = 0
+    total_executed_signals = 0
+
+    for instrument_id in instrument_ids:
+        for subject_type in subject_types:
+            for mode in modes:
+                for timeframe_set in timeframe_sets:
+                    model_codes = candidate_model_codes if subject_type == BACKTEST_SUBJECT_PLAN_REPLAY else [None]
+                    sessions_to_run = session_groups if subject_type == BACKTEST_SUBJECT_SIGNAL else [[]]
+                    for candidate_model_code in model_codes:
+                        for session_codes in sessions_to_run:
+                            run_metadata = {
+                                "batch_id": batch_id,
+                                "batch_mode": "bulk_historical_validation",
+                                "execution_boundary": "read_only_backtest_no_order_submission",
+                            }
+                            run_id, outcomes = service.run_backtest(
+                                instrument_id=instrument_id,
+                                limit=limit,
+                                cost_config=cost_config,
+                                quality_gate=quality_gate,
+                                subject_type=subject_type,
+                                candidate_model_code=candidate_model_code,
+                                mode=mode,
+                                replay_config=replay_config,
+                                timeframe_set=timeframe_set,
+                                session_codes=session_codes,
+                                run_metadata=run_metadata,
+                            )
+                            summary = summarize_backtest_outcomes(outcomes, subject_type=subject_type)
+                            sample_basis = "executed" if subject_type == BACKTEST_SUBJECT_PLAN_REPLAY else "total_plus_executed"
+                            gate_result = evaluate_backtest_quality(summary, gate=quality_gate, sample_basis=sample_basis)
+                            total_result_count += len(outcomes)
+                            total_executed_signals += int(summary.get("executed_signals") or 0)
+                            for reason in list(gate_result.get("reasons") or []):
+                                quality_gate_reason_counts[str(reason)] = quality_gate_reason_counts.get(str(reason), 0) + 1
+                            for reason, count in dict(summary.get("failure_reasons") or {}).items():
+                                failure_reason_counts[str(reason)] = failure_reason_counts.get(str(reason), 0) + int(count or 0)
+                            _accumulate_group(aggregate_by_subject, subject_type, summary, gate_result)
+                            session_key = _session_group_key(session_codes)
+                            _accumulate_group(aggregate_by_session_group, session_key, summary, gate_result)
+                            results.append(
+                                {
+                                    "backtest_run_id": run_id,
+                                    "instrument_id": instrument_id,
+                                    "subject_type": subject_type,
+                                    "mode": mode,
+                                    "timeframe_set": dict(timeframe_set),
+                                    "session_codes": list(session_codes),
+                                    "candidate_model_code": candidate_model_code,
+                                    "result_count": len(outcomes),
+                                    "executed_signals": int(summary.get("executed_signals") or 0),
+                                    "expectancy_r": float(summary.get("expectancy_r") or 0.0),
+                                    "executed_expectancy_r": float(summary.get("executed_expectancy_r") or 0.0),
+                                    "profit_factor_r": summary.get("profit_factor_r"),
+                                    "max_drawdown_r": float(summary.get("max_drawdown_r") or 0.0),
+                                    "quality_gate": gate_result,
+                                    "top_failure_reasons": _top_counts(dict(summary.get("failure_reasons") or {}), limit=5),
+                                }
+                            )
+
+    passed_count = sum(1 for item in results if item["quality_gate"].get("passed"))
+    blocked_count = len(results) - passed_count
+    payload = {
+        "status": "blocked" if require_quality_gate and blocked_count else "completed",
+        "mode": "batch_historical_backtest",
+        "batch_id": batch_id,
+        "generated_ts": datetime.now(timezone.utc).isoformat(),
+        "run_count": len(results),
+        "passed_quality_gate_count": passed_count,
+        "blocked_quality_gate_count": blocked_count,
+        "total_result_count": total_result_count,
+        "total_executed_signals": total_executed_signals,
+        "quality_gate_required": require_quality_gate,
+        "quality_gate_reason_counts": quality_gate_reason_counts,
+        "failure_reason_counts": failure_reason_counts,
+        "aggregate_by_subject": aggregate_by_subject,
+        "aggregate_by_session_group": aggregate_by_session_group,
+        "top_runs_by_expectancy": sorted(
+            results,
+            key=lambda item: (float(item.get("executed_expectancy_r") or 0.0), int(item.get("executed_signals") or 0)),
+            reverse=True,
+        )[:5],
+        "results": results,
+        "operator_summary": _build_operator_summary(results, quality_gate_reason_counts),
+        "execution_boundary": "read_only_backtest_no_order_submission",
+    }
+    return payload
+
+
+def build_instrument_ids_from_env(env) -> list[int]:
+    raw = str(env.get("BACKTEST_BATCH_INSTRUMENT_IDS") or env.get("BACKTEST_INSTRUMENT_IDS") or "1").strip()
+    return [int(item.strip()) for item in raw.split(",") if item.strip()]
+
+
+def build_subject_types_from_env(env) -> list[str]:
+    raw = str(env.get("BACKTEST_BATCH_SUBJECT_TYPES") or "plan_replay,signal").strip()
+    return _csv_values(raw)
+
+
+def build_modes_from_env(env) -> list[str]:
+    raw = str(env.get("BACKTEST_BATCH_MODES") or BACKTEST_MODE_REPLAY_NO_LOOKAHEAD).strip()
+    return _csv_values(raw)
+
+
+def build_candidate_model_codes_from_env(env) -> list[Optional[str]]:
+    raw = str(env.get("BACKTEST_BATCH_CANDIDATE_MODEL_CODES") or "").strip()
+    values = _csv_values(raw)
+    return [None] if not values else [None if item.upper() == "ALL" else item for item in values]
+
+
+def build_timeframe_sets_from_env(env) -> list[dict]:
+    raw = str(env.get("BACKTEST_BATCH_TIMEFRAME_SETS") or "").strip()
+    if not raw:
+        return [
+            {
+                "bias_tf": str(env.get("AI_ICT_BIAS_TIMEFRAME") or env.get("BACKTEST_BIAS_TIMEFRAME") or "1m"),
+                "setup_tf": str(env.get("AI_ICT_SETUP_TIMEFRAME") or env.get("BACKTEST_SETUP_TIMEFRAME") or "1m"),
+                "trigger_tf": str(env.get("AI_ICT_TRIGGER_TIMEFRAME") or env.get("BACKTEST_TRIGGER_TIMEFRAME") or "1m"),
+            }
+        ]
+    result = []
+    for group in raw.split(";"):
+        values = [item.strip() for item in group.split("/") if item.strip()]
+        if len(values) == 1:
+            values = [values[0], values[0], values[0]]
+        if len(values) != 3:
+            raise ValueError(f"invalid_timeframe_set:{group}")
+        result.append({"bias_tf": values[0], "setup_tf": values[1], "trigger_tf": values[2]})
+    return result
+
+
+def build_session_groups_from_env(env) -> list[list[str]]:
+    raw = str(env.get("BACKTEST_BATCH_SESSION_GROUPS") or DEFAULT_SESSION_GROUPS).strip()
+    groups = []
+    for group in raw.split(";"):
+        values = _csv_values(group)
+        groups.append([] if not values or values == ["ALL"] else values)
+    return groups or [[]]
+
+
+def _accumulate_group(container: dict[str, dict], key: str, summary: dict, quality_gate: dict) -> None:
+    bucket = container.setdefault(
+        key,
+        {
+            "run_count": 0,
+            "passed_quality_gate_count": 0,
+            "total_result_count": 0,
+            "total_executed_signals": 0,
+            "total_r": 0.0,
+        },
+    )
+    bucket["run_count"] += 1
+    if quality_gate.get("passed"):
+        bucket["passed_quality_gate_count"] += 1
+    bucket["total_result_count"] += int(summary.get("total_signals") or 0)
+    bucket["total_executed_signals"] += int(summary.get("executed_signals") or 0)
+    bucket["total_r"] += float(summary.get("total_r") or 0.0)
+
+
+def _build_operator_summary(results: list[dict], reason_counts: dict[str, int]) -> str:
+    if not results:
+        return "No backtest runs were executed; check batch dimensions."
+    passed = sum(1 for item in results if item["quality_gate"].get("passed"))
+    if passed:
+        return f"Batch backtest completed: {passed}/{len(results)} runs passed the quality gate. Review top expectancy runs before promoting any setup."
+    top_reason = next(iter(_top_counts(reason_counts, limit=1)), {}).get("code", "quality_gate_not_passed")
+    return f"Batch backtest completed: 0/{len(results)} runs passed the quality gate. Main blocker: {top_reason}."
+
+
+def _top_counts(values: dict[str, int], *, limit: int) -> list[dict]:
+    return [
+        {"code": key, "count": int(value)}
+        for key, value in sorted(values.items(), key=lambda item: (-int(item[1]), str(item[0])))[:limit]
+    ]
+
+
+def _format_counts(values: dict[str, int]) -> str:
+    return ", ".join(f"{item['code']}={item['count']}" for item in _top_counts(values, limit=5))
+
+
+def _session_group_key(session_codes: list[str]) -> str:
+    return "ALL" if not session_codes else ",".join(session_codes)
+
+
+def _csv_values(raw: str) -> list[str]:
+    return [item.strip() for item in str(raw or "").split(",") if item.strip()]
+
+
+def _get_int(env, key: str, default: int) -> int:
+    value = env.get(key)
+    if value is None or value == "":
+        return default
+    return int(value)
+
+
+def _get_bool(env, key: str, default: bool) -> bool:
+    value = env.get(key)
+    if value is None or value == "":
+        return default
+    return str(value).strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _default_batch_id() -> str:
+    return "bt-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/services/backtest/backtest_service.py

@@ -106,6 +106,7 @@ class BacktestService:
         replay_config: Optional[BacktestReplayConfig] = None,
         timeframe_set: Optional[dict] = None,
         session_codes: Optional[list[str]] = None,
+        run_metadata: Optional[dict] = None,
     ) -> tuple[int, list[BacktestOutcome]]:
         cost_config = cost_config or BacktestCostConfig()
         quality_gate = quality_gate or BacktestQualityGate()
@@ -137,6 +138,7 @@ class BacktestService:
                 'quality_gate': quality_gate.__dict__,
                 'replay': replay_config.__dict__ if mode == BACKTEST_MODE_REPLAY_NO_LOOKAHEAD else None,
                 'session_codes': list(session_codes or []),
+                'run_metadata': dict(run_metadata or {}),
             },
         )
         if subject_type == BACKTEST_SUBJECT_PLAN_REPLAY:

tests/unit/test_backtest_batch_script.py

@@ -0,0 +1,173 @@
+import json
+import unittest
+from datetime import datetime, timezone
+
+from scripts import run_backtest_batch
+from scripts.run_backtest_batch import (
+    build_candidate_model_codes_from_env,
+    build_instrument_ids_from_env,
+    build_session_groups_from_env,
+    build_subject_types_from_env,
+    build_timeframe_sets_from_env,
+    run_backtest_batch as run_batch,
+)
+from src.services.backtest import BacktestOutcome, BacktestQualityGate
+
+
+class RunBacktestBatchScriptTests(unittest.TestCase):
+    def test_build_batch_dimensions_from_env(self) -> None:
+        env = {
+            "BACKTEST_BATCH_INSTRUMENT_IDS": "1,2",
+            "BACKTEST_BATCH_SUBJECT_TYPES": "plan_replay,signal",
+            "BACKTEST_BATCH_TIMEFRAME_SETS": "1h/15m/5m;15m/5m/1m",
+            "BACKTEST_BATCH_SESSION_GROUPS": "ALL;LONDON,NY_AM;ASIA",
+            "BACKTEST_BATCH_CANDIDATE_MODEL_CODES": "ALL,OSOK",
+        }
+
+        self.assertEqual(build_instrument_ids_from_env(env), [1, 2])
+        self.assertEqual(build_subject_types_from_env(env), ["plan_replay", "signal"])
+        self.assertEqual(
+            build_timeframe_sets_from_env(env),
+            [
+                {"bias_tf": "1h", "setup_tf": "15m", "trigger_tf": "5m"},
+                {"bias_tf": "15m", "setup_tf": "5m", "trigger_tf": "1m"},
+            ],
+        )
+        self.assertEqual(build_session_groups_from_env(env), [[], ["LONDON", "NY_AM"], ["ASIA"]])
+        self.assertEqual(build_candidate_model_codes_from_env(env), [None, "OSOK"])
+
+    def test_run_batch_aggregates_quality_gate_and_metadata(self) -> None:
+        service = FakeBatchBacktestService(
+            {
+                "plan_replay": [_win_outcome(signal_id=None, trading_plan_id=10)],
+                "signal": [_loss_outcome(signal_id=1)],
+            }
+        )
+
+        payload = run_batch(
+            service=service,
+            batch_id="batch-test",
+            instrument_ids=[1],
+            limit=25,
+            subject_types=["plan_replay", "signal"],
+            modes=["replay_no_lookahead"],
+            timeframe_sets=[{"bias_tf": "1h", "setup_tf": "15m", "trigger_tf": "5m"}],
+            session_groups=[[], ["LONDON"]],
+            candidate_model_codes=[None],
+            require_quality_gate=False,
+            quality_gate=BacktestQualityGate(min_total_signals=1, min_executed_signals=1, min_profit_factor_r=0.0),
+            cost_config=None,
+            replay_config=None,
+        )
+
+        self.assertEqual(payload["status"], "completed")
+        self.assertEqual(payload["run_count"], 3)
+        self.assertEqual(payload["total_result_count"], 3)
+        self.assertEqual(payload["aggregate_by_subject"]["plan_replay"]["run_count"], 1)
+        self.assertEqual(payload["aggregate_by_subject"]["signal"]["run_count"], 2)
+        self.assertEqual(service.calls[0]["run_metadata"]["batch_id"], "batch-test")
+        self.assertEqual(service.calls[0]["run_metadata"]["execution_boundary"], "read_only_backtest_no_order_submission")
+        self.assertEqual(payload["execution_boundary"], "read_only_backtest_no_order_submission")
+
+    def test_main_outputs_json_payload(self) -> None:
+        service = FakeBatchBacktestService({"plan_replay": [_win_outcome(signal_id=None, trading_plan_id=10)]})
+        output = _Capture()
+
+        original_service = run_backtest_batch.BacktestService
+        run_backtest_batch.BacktestService = lambda: service
+        try:
+            with output:
+                exit_code = run_backtest_batch.main(
+                    env={
+                        "BACKTEST_BATCH_OUTPUT_JSON": "1",
+                        "BACKTEST_BATCH_ID": "batch-json",
+                        "BACKTEST_BATCH_SUBJECT_TYPES": "plan_replay",
+                        "BACKTEST_BATCH_SESSION_GROUPS": "ALL",
+                        "BACKTEST_MIN_TOTAL_SIGNALS": "1",
+                        "BACKTEST_MIN_EXECUTED_SIGNALS": "1",
+                        "BACKTEST_MIN_PROFIT_FACTOR_R": "0",
+                    }
+                )
+        finally:
+            run_backtest_batch.BacktestService = original_service
+
+        payload = json.loads(output.value)
+        self.assertEqual(exit_code, 0)
+        self.assertEqual(payload["batch_id"], "batch-json")
+        self.assertEqual(payload["run_count"], 1)
+
+
+class FakeBatchBacktestService:
+    def __init__(self, outcomes_by_subject: dict[str, list[BacktestOutcome]]) -> None:
+        self.outcomes_by_subject = outcomes_by_subject
+        self.calls = []
+
+    def run_backtest(self, **kwargs):
+        self.calls.append(kwargs)
+        return 100 + len(self.calls), list(self.outcomes_by_subject.get(kwargs["subject_type"]) or [])
+
+
+class _Capture:
+    def __enter__(self):
+        import contextlib
+        import io
+
+        self._stream = io.StringIO()
+        self._context = contextlib.redirect_stdout(self._stream)
+        self._context.__enter__()
+        return self
+
+    def __exit__(self, *args):
+        self._context.__exit__(*args)
+        self.value = self._stream.getvalue()
+
+
+def _win_outcome(*, signal_id, trading_plan_id=None) -> BacktestOutcome:
+    return BacktestOutcome(
+        signal_id=signal_id,
+        outcome="win",
+        mfe=2.0,
+        mae=0.0,
+        tp1_hit=True,
+        tp2_hit=True,
+        invalidated_before_entry=False,
+        error_tags=[],
+        trading_plan_id=trading_plan_id,
+        meta={
+            "signal_side": "buy",
+            "signal_symbol": "BTC-USDT",
+            "session_codes": ["LONDON"],
+            "weekday": "Thursday",
+            "mfe_r": 2.0,
+            "mae_r": 0.0,
+            "r_multiple": 2.0,
+            "cost_r": 0.0,
+        },
+    )
+
+
+def _loss_outcome(*, signal_id) -> BacktestOutcome:
+    return BacktestOutcome(
+        signal_id=signal_id,
+        outcome="loss",
+        mfe=0.2,
+        mae=1.0,
+        tp1_hit=False,
+        tp2_hit=False,
+        invalidated_before_entry=False,
+        error_tags=["stopped_out"],
+        meta={
+            "signal_side": "sell",
+            "signal_symbol": "BTC-USDT",
+            "session_codes": ["NY_AM"],
+            "weekday": "Thursday",
+            "mfe_r": 0.2,
+            "mae_r": 1.0,
+            "r_multiple": -1.0,
+            "cost_r": 0.0,
+        },
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_documentation_consistency.py

@@ -24,9 +24,11 @@ ACTIVE_TOP_LEVEL_DOCS = {
     "AI_ICT_ICT主线概念字典_S2_2026-05-21_v1.md",
     "AI_ICT_ReviewNow主线Gate_S3S4_2026-05-21_v1.md",
     "AI_ICT_Cockpit链路句与反馈错误分类_S5S6_2026-05-21_v1.md",
+    "AI_ICT_DoctrineKernel全书规则内核重建_D0_2026-05-21_v1.md",
     "AI_ICT_只读实盘可靠性与交易员SOP_S7S8_2026-05-21_v1.md",
     "AI_ICT_TraderCockpit状态话术压缩_S9_2026-05-21_v1.md",
     "AI_ICT_TraderCockpit真实反馈验证_S10_2026-05-21_v1.md",
+    "AI_ICT_批量历史回测验证_S16_2026-05-21_v1.md",
     "AI_ICT_主线纠偏与功能裁剪方案_2026-05-20_v1.md",
     "AI_ICT_实盘辅助分析决策助手上线必做清单_2026-05-18_v1.md",
     "AI_ICT_实盘辅助分析决策助手使用说明_2026-05-18_v1.md",