Add doctrine playbook backtest cohorts

2026-05-21 20:19:31 +08:00
parent ebb1b12b0a
commit dfef7b951c
22 changed files with 868 additions and 58 deletions
@@ -846,10 +846,11 @@ Automation execution-layer PXX remains closed and its known scope stays frozen.
 - 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.
 - D1 Doctrine Market Language Kernel is complete locally: added `src/services/ict_mainline/doctrine.py`, `GET /trading/doctrine-rules`, Strategy Backtest `doctrine_research_status` / `doctrine_rule_gap_summary`, and Trader Cockpit watchlist/readout `research_only` gating. Old sweep/MSS/FVG/retrace candidates now expose `missing_doctrine_setup_playbook` or the specific missing Doctrine rule reason unless mapped to a full setup playbook. No execution/readiness/risk/trust/operator-confirmation gate was changed.
+- D2 Doctrine Playbook / Backtest Cohort is complete locally: added the first registered setup playbook `ICT-CH1-6-LONDON-DISCOUNT-REVERSAL`, `GET /trading/doctrine-playbooks`, `doctrine_backtest_cohort` / `doctrine_cohort_summary`, `BACKTEST_SETUP_PLAYBOOK_ID`, `BACKTEST_BATCH_SETUP_PLAYBOOK_IDS`, and a Review Now gate that requires Doctrine playbook, active Trading Plan, and Strategy Backtest quality evidence to all pass. No execution/readiness/risk/trust/operator-confirmation gate was changed.

 ## Exact Next Step

-Use `docs/AI_ICT_DoctrineKernel全书规则内核重建_D0_2026-05-21_v1.md`, `docs/AI_ICT_DoctrineKernel市场语言规则内核_D1_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 D2: define the first real Doctrine setup playbook schema, make Strategy Backtest cohorts use `setup_playbook_id + doctrine completeness` rather than old `model_code` alone, and keep Trader Cockpit `Review Now` blocked unless playbook mapping, historical quality gate, and active Trading Plan all pass. Do not restart external-context/archive/digest long-tail work, do not add 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/AI_ICT_DoctrineKernel市场语言规则内核_D1_2026-05-21_v1.md`, `docs/AI_ICT_DoctrineKernel_PlaybookBacktestCohort_D2_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 D3 Liquidity Kernel: extract BSL / SSL / liquidity pool / stop run / sweep / raid from the book into target-ranking and no-trade/wait rules; keep sweep as evidence component only, not a setup; wire the playbook `liquidity_event_rule` toward measurable backtest fields. Do not restart external-context/archive/digest long-tail work, do not add 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

@@ -953,7 +953,8 @@
 - [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`)
 - [x] D1 Doctrine Market Language Kernel: built the first `Range / Draw / Location / Context / Time / Invalidation / IRL-ERL` registry, added `/trading/doctrine-rules`, Strategy Backtest Doctrine gap aggregation, and Trader Cockpit `research_only_reason` gating for old surface-shape candidates; no live order submission and no fabricated samples
- [ ] D2 next product gate: define the first real Doctrine setup playbook schema, make Strategy Backtest cohorts use `setup_playbook_id + doctrine completeness`, and keep `Review Now` blocked unless playbook mapping, historical quality gate, and active Trading Plan all pass
+- [x] D2 Doctrine Playbook / Backtest Cohort: defined first registered setup playbook `ICT-CH1-6-LONDON-DISCOUNT-REVERSAL`, added `/trading/doctrine-playbooks`, `doctrine_backtest_cohort` / `doctrine_cohort_summary`, setup-playbook batch envs, Strategy Backtest UI cohort display, and Review Now gating that requires playbook mapping + historical quality gate + active Trading Plan
+- [ ] D3 next product gate: rebuild Liquidity Kernel from the book by turning BSL / SSL / liquidity pool / stop run / sweep / raid into target-ranking and no-trade/wait rules; sweep remains evidence component only, not a setup
 - [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
@@ -0,0 +1,106 @@
+# AI_ICT Doctrine Kernel Playbook / Backtest Cohort D2 2026-05-21 v1
+
+## 1. 本阶段结论
+
+D2 把 Strategy Backtest 和 Trader Cockpit 从旧 `model_code` 主导，推进到 Doctrine setup playbook 主导。
+
+旧 `SWEEP_MSS_FVG` / `OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE` 仍可作为历史研究标签，但不能单独代表可交易 setup。后续判断必须先看：
+
+```text
+setup_playbook_id + doctrine_completeness + historical quality gate + active Trading Plan
+```
+
+## 2. 第一版 Playbook Schema
+
+新增 `src/services/ict_mainline/doctrine.py` 中的 `DoctrineSetupPlaybook`，第一版注册 playbook：
+
+```text
+ICT-CH1-6-LONDON-DISCOUNT-REVERSAL
+```
+
+该 playbook 必须具备：
+
+- range_definition
+- draw_rule
+- location_rule
+- liquidity_event_rule
+- displacement_structure_rule
+- pd_array_rule
+- session_timing_rule
+- confirmation_rule
+- invalidation_rule
+- target_rule
+- no_trade_conditions
+- backtest_cohort
+- promotion_gate
+- cockpit_sentence
+- journal_error_classes
+
+缺失 playbook 的候选、回测、计划，一律保留为 `research_only` / `Observe Only`。
+
+## 3. Backtest Cohort 变化
+
+Strategy Backtest 不再只按旧 `model_code` 解释结果。
+
+新增字段：
+
+- `doctrine_backtest_cohort`
+- `doctrine_cohort_summary`
+- `cohort_key`
+- `setup_playbook_id`
+- `doctrine_completeness`
+- `promotion_gate`
+
+cohort key 规则：
+
+```text
+{setup_playbook_id or missing_playbook}::{complete or incomplete}
+```
+
+例如：
+
+```text
+ICT-CH1-6-LONDON-DISCOUNT-REVERSAL::complete
+missing_playbook::incomplete
+```
+
+## 4. Review Now Gate 变化
+
+Trader Cockpit 的 `Review Now` 现在必须同时满足：
+
+1. Doctrine setup playbook gate passed
+2. Active Trading Plan gate passed
+3. Strategy Backtest quality gate passed
+4. Candidate 自身没有 Doctrine / confirmation / plan mismatch blocker
+
+任一项不满足，首屏只能是：
+
+```text
+Observe Only
+```
+
+本阶段仍不改变：
+
+- `execution_allowed`
+- `decision_support_ready`
+- `real_live_ready`
+- risk governor
+- trust bucket
+- operator confirmation protocol
+- live order submission
+
+## 5. 接口与脚本
+
+- `GET /trading/doctrine-playbooks`
+- `GET /trading/backtests` 增加 playbook cohort 字段
+- `BACKTEST_SETUP_PLAYBOOK_ID`
+- `BACKTEST_BATCH_SETUP_PLAYBOOK_IDS`
+
+## 6. 下一步
+
+D3 应该继续吸收《ICT交易全书》的 Liquidity Kernel：
+
+1. 把 BSL / SSL / liquidity pool / stop run / sweep / raid 从“形态标签”重建成流动性目标排序规则。
+2. 明确 sweep 只是 evidence component，不是 setup。
+3. 把 manipulation / Judas swing / session timing 纳入 no-trade / wait 规则。
+4. 让 playbook 的 liquidity_event_rule 产生可回测字段，而不是只写自然语言。
@@ -78,6 +78,12 @@ def build_backtest_candidate_model_code_from_env(env=None) -> Optional[str]:
    return value or None


+def build_backtest_setup_playbook_id_from_env(env=None) -> Optional[str]:
+    env = os.environ if env is None else env
+    value = str(env.get("BACKTEST_SETUP_PLAYBOOK_ID") or "").strip()
+    return value or None
+
+
 def build_replay_config_from_env(env=None) -> BacktestReplayConfig:
    env = os.environ if env is None else env
    defaults = BacktestReplayConfig()
@@ -119,6 +125,7 @@ def run_backtest_once(
    subject_type: str = BACKTEST_SUBJECT_PLAN_REPLAY,
    trading_plan_id: Optional[int] = None,
    candidate_model_code: Optional[str] = None,
+    setup_playbook_id: Optional[str] = None,
    replay_config: Optional[BacktestReplayConfig] = None,
    timeframe_set: Optional[dict] = None,
    session_codes: Optional[list[str]] = None,
@@ -135,6 +142,7 @@ def run_backtest_once(
        replay_config=replay_config,
        timeframe_set=timeframe_set,
        session_codes=session_codes,
+        run_metadata={"setup_playbook_id": setup_playbook_id} if setup_playbook_id else None,
    )
    summary = summarize_backtest_outcomes(outcomes, subject_type=subject_type)
    quality_gate_result = evaluate_backtest_quality(
@@ -149,6 +157,7 @@ def run_backtest_once(
        "subject_type": subject_type,
        "trading_plan_id": trading_plan_id,
        "candidate_model_code": candidate_model_code,
+        "setup_playbook_id": setup_playbook_id,
        "timeframe_set": timeframe_set,
        "session_codes": session_codes,
        "summary": summary,
@@ -183,6 +192,7 @@ def main(env=None) -> int:
            subject_type=build_backtest_subject_type_from_env(env=env),
            trading_plan_id=build_backtest_trading_plan_id_from_env(env=env),
            candidate_model_code=build_backtest_candidate_model_code_from_env(env=env),
+            setup_playbook_id=build_backtest_setup_playbook_id_from_env(env=env),
            replay_config=build_replay_config_from_env(env=env),
            timeframe_set=build_backtest_timeframe_set_from_env(env=env),
            session_codes=build_backtest_session_codes_from_env(env=env),
@@ -18,6 +18,7 @@ from scripts.run_backtest import (
    build_quality_gate_from_env,
    build_replay_config_from_env,
 )
+from src.services.ict_mainline import build_doctrine_backtest_cohort
 from src.services.backtest import (
    BACKTEST_MODE_REPLAY_NO_LOOKAHEAD,
    BACKTEST_SUBJECT_PLAN_REPLAY,
@@ -47,6 +48,7 @@ def main(env=None) -> int:
            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),
+            setup_playbook_ids=build_setup_playbook_ids_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),
@@ -91,6 +93,7 @@ def run_backtest_batch(
    timeframe_sets: list[dict],
    session_groups: list[list[str]],
    candidate_model_codes: list[Optional[str]],
+    setup_playbook_ids: list[Optional[str]],
    require_quality_gate: bool,
    quality_gate,
    cost_config,
@@ -101,6 +104,7 @@ def run_backtest_batch(
    failure_reason_counts: dict[str, int] = {}
    aggregate_by_subject: dict[str, dict] = {}
    aggregate_by_session_group: dict[str, dict] = {}
+    aggregate_by_doctrine_cohort: dict[str, dict] = {}
    total_result_count = 0
    total_executed_signals = 0

@@ -111,56 +115,81 @@ def run_backtest_batch(
                    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),
+                        for setup_playbook_id in setup_playbook_ids:
+                            for session_codes in sessions_to_run:
+                                run_metadata = {
+                                    "batch_id": batch_id,
+                                    "batch_mode": "bulk_historical_validation",
+                                    "setup_playbook_id": setup_playbook_id,
+                                    "execution_boundary": "read_only_backtest_no_order_submission",
                                }
-                            )
+                                doctrine_cohort = build_doctrine_backtest_cohort(
+                                    {
+                                        "setup_playbook_id": setup_playbook_id,
+                                        "candidate_model_code": candidate_model_code,
+                                        "timeframe_set": timeframe_set,
+                                        "run_metadata": run_metadata,
+                                    }
+                                )
+                                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)
+                                doctrine_cohort = build_doctrine_backtest_cohort(
+                                    {
+                                        "setup_playbook_id": setup_playbook_id,
+                                        "candidate_model_code": candidate_model_code,
+                                        "timeframe_set": timeframe_set,
+                                        "summary": {"quality_gate": gate_result},
+                                        "run_metadata": run_metadata,
+                                    },
+                                    quality_gate=gate_result,
+                                )
+                                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)
+                                _accumulate_doctrine_cohort_group(aggregate_by_doctrine_cohort, doctrine_cohort, 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,
+                                        "setup_playbook_id": setup_playbook_id,
+                                        "doctrine_cohort_key": doctrine_cohort.get("cohort_key"),
+                                        "doctrine_completeness": doctrine_cohort.get("doctrine_completeness"),
+                                        "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,
+                                        "doctrine_backtest_cohort": doctrine_cohort,
+                                        "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
@@ -179,6 +208,7 @@ def run_backtest_batch(
        "failure_reason_counts": failure_reason_counts,
        "aggregate_by_subject": aggregate_by_subject,
        "aggregate_by_session_group": aggregate_by_session_group,
+        "aggregate_by_doctrine_cohort": aggregate_by_doctrine_cohort,
        "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)),
@@ -212,6 +242,12 @@ def build_candidate_model_codes_from_env(env) -> list[Optional[str]]:
    return [None] if not values else [None if item.upper() == "ALL" else item for item in values]


+def build_setup_playbook_ids_from_env(env) -> list[Optional[str]]:
+    raw = str(env.get("BACKTEST_BATCH_SETUP_PLAYBOOK_IDS") or env.get("BACKTEST_SETUP_PLAYBOOK_ID") 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:
@@ -261,6 +297,36 @@ def _accumulate_group(container: dict[str, dict], key: str, summary: dict, quali
    bucket["total_r"] += float(summary.get("total_r") or 0.0)


+def _accumulate_doctrine_cohort_group(container: dict[str, dict], cohort: dict, summary: dict, quality_gate: dict) -> None:
+    key = str(cohort.get("cohort_key") or "missing_playbook::incomplete")
+    bucket = container.setdefault(
+        key,
+        {
+            "cohort_key": key,
+            "setup_playbook_id": cohort.get("setup_playbook_id"),
+            "doctrine_completeness": cohort.get("doctrine_completeness"),
+            "run_count": 0,
+            "passed_quality_gate_count": 0,
+            "promotable_count": 0,
+            "total_result_count": 0,
+            "total_executed_signals": 0,
+            "total_r": 0.0,
+            "promotion_blocker_counts": {},
+        },
+    )
+    bucket["run_count"] += 1
+    if quality_gate.get("passed"):
+        bucket["passed_quality_gate_count"] += 1
+    if (cohort.get("promotion_gate") or {}).get("passed"):
+        bucket["promotable_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)
+    for reason in list((cohort.get("promotion_gate") or {}).get("reasons") or []):
+        reason_key = str(reason)
+        bucket["promotion_blocker_counts"][reason_key] = int(bucket["promotion_blocker_counts"].get(reason_key) or 0) + 1
+
+
 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."
@@ -6,9 +6,11 @@ from sqlalchemy import text
 from src.db.session import SessionLocal
 from src.services.backtest import evaluate_backtest_quality
 from src.services.ict_mainline import (
+    build_doctrine_backtest_cohort,
    build_backtest_quality_gate,
    build_ict_rulebook_mapping,
    evaluate_doctrine_research_status,
+    summarize_doctrine_backtest_cohorts,
    summarize_doctrine_rule_gaps,
 )

@@ -30,6 +32,7 @@ def get_strategy_backtest_surface(instrument_id: Optional[int] = None, limit: in
        },
    )
    doctrine_rule_gap_summary = summarize_doctrine_rule_gaps(items)
+    doctrine_cohort_summary = summarize_doctrine_backtest_cohorts(items)
    return {
        "instrument_id": instrument_id,
        "item_count": len(items),
@@ -38,6 +41,7 @@ def get_strategy_backtest_surface(instrument_id: Optional[int] = None, limit: in
        "operator_summary": _build_operator_summary(latest),
        "cockpit_promotion_gate": build_backtest_quality_gate(quality_gate),
        "doctrine_rule_gap_summary": doctrine_rule_gap_summary,
+        "doctrine_cohort_summary": doctrine_cohort_summary,
        "ict_rulebook_mapping": ict_rulebook_mapping,
        "items": items,
    }
@@ -101,6 +105,17 @@ def normalize_backtest_summary_row(row: dict) -> dict:
        },
        require_setup_playbook=True,
    )
+    doctrine_backtest_cohort = build_doctrine_backtest_cohort(
+        {
+            "subject_type": row.get("subject_type"),
+            "model_code": row.get("model_code"),
+            "candidate_model_code": row.get("candidate_model_code"),
+            "timeframe_set": timeframe_set,
+            "config": config,
+            "summary": summary,
+        },
+        quality_gate=dict(summary.get("quality_gate") or {}),
+    )
    return {
        **row,
        "timeframe_set": timeframe_set,
@@ -108,6 +123,7 @@ def normalize_backtest_summary_row(row: dict) -> dict:
        "summary": summary,
        "ict_rulebook_mapping": ict_rulebook_mapping,
        "doctrine_research_status": doctrine_research_status,
+        "doctrine_backtest_cohort": doctrine_backtest_cohort,
    }


@@ -132,10 +148,17 @@ def _build_operator_summary(latest: Optional[dict]) -> dict:
        reasons = quality_gate.get("reasons") or []
        reason = str(reasons[0]) if reasons else "quality_gate_not_passed"
    doctrine_status = dict(latest.get("doctrine_research_status") or {})
+    doctrine_cohort = dict(latest.get("doctrine_backtest_cohort") or {})
    if doctrine_status.get("research_only"):
        status = "doctrine_research_only"
        decision = "do_not_promote_strategy"
        reason = str(doctrine_status.get("research_only_reason") or reason)
+    if not (doctrine_cohort.get("promotion_gate") or {}).get("passed"):
+        cohort_reasons = list((doctrine_cohort.get("promotion_gate") or {}).get("reasons") or [])
+        if cohort_reasons:
+            status = "doctrine_cohort_blocked"
+            decision = "do_not_promote_strategy"
+            reason = str(cohort_reasons[0])
    return {
        "status": status,
        "decision": decision,
@@ -122,7 +122,7 @@ from src.api.training import (
    parse_quality_sort_query,
    submit_training_attempt,
 )
-from src.services.ict_mainline import build_doctrine_rule_matrix
+from src.services.ict_mainline import build_doctrine_playbook_registry, build_doctrine_rule_matrix
 from src.services.execution import ExecutionService
 from src.services.market_state import start_market_truth_runtime
 from src.services.runtime import start_trading_runtime_service
@@ -223,6 +223,7 @@ def build_api_dependencies() -> dict:
        "trading_plans": get_recent_trading_plans,
        "trading_backtests": get_strategy_backtest_surface,
        "trading_doctrine_rules": build_doctrine_rule_matrix,
+        "trading_doctrine_playbooks": build_doctrine_playbook_registry,
        "training_cases": get_training_cases,
        "training_case": get_training_case,
        "training_export_cases": export_training_case_package,
@@ -846,6 +847,8 @@ def route_request(
        return 200, deps["trading_backtests"](instrument_id=instrument_id, limit=limit)
    if method == "GET" and path == "/trading/doctrine-rules":
        return 200, deps["trading_doctrine_rules"]()
+    if method == "GET" and path == "/trading/doctrine-playbooks":
+        return 200, deps["trading_doctrine_playbooks"]()
    if method == "GET" and len(path_parts) == 3 and path_parts[:2] == ["trading", "candidates"]:
        candidate = deps["trading_candidate_by_id"](_parse_path_int(path_parts[2], "candidate_id"))
        if candidate is None:
@@ -164,6 +164,13 @@ class BacktestRepository:
                tpc.tp2,
                tpc.rr_tp1,
                tpc.rr_tp2,
+                coalesce(
+                    tpc.evidence->>'setup_playbook_id',
+                    tpc.entry->>'setup_playbook_id',
+                    tp.inputs->>'setup_playbook_id',
+                    tp.inputs->>'doctrine_setup_id',
+                    tp.inputs->>'playbook_id'
+                ) as setup_playbook_id,
                tpc.invalidations,
                tpc.management_rules,
                tpc.ai_explanation,
@@ -175,6 +182,8 @@ class BacktestRepository:
                tp.primary_draw,
                tp.secondary_draw,
                tp.bias,
+                tp.inputs as trading_plan_inputs,
+                tp.quality as trading_plan_quality,
                tp.activated_ts,
                tp.created_ts as trading_plan_created_ts
            from trade_plan_candidates tpc
@@ -4,6 +4,7 @@ from typing import Optional

 from src.repositories.backtest_repository import BacktestRepository
 from src.services.backtest.replay import ReplaySwing, iter_historical_replay
+from src.services.ict_mainline import build_doctrine_backtest_cohort, extract_setup_playbook_id


 SAME_CANDLE_PRIORITY = "stop_first"
@@ -116,9 +117,25 @@ class BacktestService:
            raise ValueError(f"invalid_backtest_subject_type:{subject_type}")
        if mode not in SUPPORTED_BACKTEST_MODES:
            raise ValueError(f"invalid_backtest_mode:{mode}")
+        active_trading_plan = None
        if subject_type == BACKTEST_SUBJECT_PLAN_REPLAY and trading_plan_id is None:
            active_trading_plan = self.repository.fetch_active_trading_plan(instrument_id=instrument_id)
            trading_plan_id = active_trading_plan.get('id') if active_trading_plan else None
+        run_metadata = dict(run_metadata or {})
+        setup_playbook_id = extract_setup_playbook_id(run_metadata)
+        if not setup_playbook_id and active_trading_plan:
+            setup_playbook_id = extract_setup_playbook_id(active_trading_plan)
+            if setup_playbook_id:
+                run_metadata['setup_playbook_id'] = setup_playbook_id
+        doctrine_backtest_cohort = build_doctrine_backtest_cohort(
+            {
+                'subject_type': subject_type,
+                'model_code': 'SWEEP_MSS_FVG',
+                'candidate_model_code': candidate_model_code,
+                'timeframe_set': timeframe_set,
+                'run_metadata': run_metadata,
+            }
+        )
        run_id = self.repository.insert_backtest_run(
            model_code='SWEEP_MSS_FVG',
            instrument_id=instrument_id,
@@ -138,7 +155,9 @@ 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 {}),
+                'setup_playbook_id': setup_playbook_id,
+                'doctrine_backtest_cohort': doctrine_backtest_cohort,
+                'run_metadata': run_metadata,
            },
        )
        if subject_type == BACKTEST_SUBJECT_PLAN_REPLAY:
@@ -155,6 +174,18 @@ class BacktestService:
            )
            summary = summarize_backtest_outcomes(outcomes, subject_type=subject_type)
            summary['quality_gate'] = evaluate_backtest_quality(summary, gate=quality_gate, sample_basis='executed')
+            outcome_setup_playbook_id = setup_playbook_id or _first_outcome_meta_value(outcomes, 'setup_playbook_id')
+            summary['doctrine_backtest_cohort'] = build_doctrine_backtest_cohort(
+                {
+                    'subject_type': subject_type,
+                    'model_code': 'SWEEP_MSS_FVG',
+                    'candidate_model_code': candidate_model_code,
+                    'timeframe_set': timeframe_set,
+                    'run_metadata': {**run_metadata, 'setup_playbook_id': outcome_setup_playbook_id},
+                    'summary': summary,
+                },
+                quality_gate=summary['quality_gate'],
+            )
            self.repository.update_backtest_run_summary(run_id, summary)
            return run_id, outcomes

@@ -315,6 +346,17 @@ class BacktestService:
            )
        summary = summarize_backtest_outcomes(outcomes, subject_type=subject_type)
        summary['quality_gate'] = evaluate_backtest_quality(summary, gate=quality_gate)
+        summary['doctrine_backtest_cohort'] = build_doctrine_backtest_cohort(
+            {
+                'subject_type': subject_type,
+                'model_code': 'SWEEP_MSS_FVG',
+                'candidate_model_code': candidate_model_code,
+                'timeframe_set': timeframe_set,
+                'run_metadata': run_metadata,
+                'summary': summary,
+            },
+            quality_gate=summary['quality_gate'],
+        )
        self.repository.update_backtest_run_summary(run_id, summary)
        return run_id, outcomes

@@ -852,10 +894,27 @@ def build_plan_replay_units(candidate_rows: list[dict]) -> list[dict]:
        entry_payload = dict((confirmed_row or latest).get('entry') or {})
        target_plan = dict(entry_payload.get('target_plan') or {})
        management_rules = list((confirmed_row or latest).get('management_rules') or target_plan.get('management_rules') or [])
+        setup_playbook_id = (
+            extract_setup_playbook_id(first)
+            or extract_setup_playbook_id(latest)
+            or extract_setup_playbook_id(confirmed_row or {})
+        )
+        doctrine_backtest_cohort = build_doctrine_backtest_cohort(
+            {
+                'setup_playbook_id': setup_playbook_id,
+                'candidate_model_code': first.get('candidate_model_code'),
+                'timeframe_set': {
+                    'setup_tf': first.get('setup_timeframe'),
+                    'trigger_tf': first.get('trigger_timeframe'),
+                },
+            }
+        )
        units.append(
            {
                'trading_plan_id': first.get('trading_plan_id'),
                'candidate_id': (confirmed_row or first).get('candidate_id'),
+                'setup_playbook_id': setup_playbook_id,
+                'doctrine_backtest_cohort': doctrine_backtest_cohort,
                'candidate_model_code': first.get('candidate_model_code'),
                'side': first.get('side'),
                'session_code': first.get('session_code'),
@@ -901,6 +960,14 @@ def build_plan_replay_units(candidate_rows: list[dict]) -> list[dict]:
    return units


+def _first_outcome_meta_value(outcomes: list[BacktestOutcome], key: str):
+    for outcome in outcomes:
+        value = dict(outcome.meta or {}).get(key)
+        if value:
+            return value
+    return None
+
+
 def candidate_row_is_executable(row: dict) -> bool:
    confirmation_state = dict(row.get('confirmation_state') or {})
    return bool(confirmation_state.get('executable')) or row.get('candidate_status') == 'executable'
@@ -935,6 +1002,10 @@ def build_plan_replay_meta(
    return {
        'subject_type': BACKTEST_SUBJECT_PLAN_REPLAY,
        'trading_plan_id': unit.get('trading_plan_id'),
+        'setup_playbook_id': unit.get('setup_playbook_id'),
+        'doctrine_backtest_cohort': dict(unit.get('doctrine_backtest_cohort') or {}),
+        'doctrine_cohort_key': (unit.get('doctrine_backtest_cohort') or {}).get('cohort_key'),
+        'doctrine_completeness': (unit.get('doctrine_backtest_cohort') or {}).get('doctrine_completeness'),
        'plan_activated_ts': _normalize_meta_ts(unit.get('plan_activated_ts')),
        'plan_created_ts': _normalize_meta_ts(unit.get('plan_created_ts')),
        'candidate_id': unit.get('candidate_id'),
@@ -1120,6 +1191,7 @@ def summarize_backtest_outcomes(outcomes: list[BacktestOutcome], subject_type: s
    grouped_by_weekday: dict[str, dict] = {}
    grouped_by_symbol: dict[str, dict] = {}
    grouped_by_timeframe: dict[str, dict] = {}
+    grouped_by_doctrine_cohort: dict[str, dict] = {}
    r_multiples: list[float] = []
    cost_values: list[float] = []
    mfe_r_values: list[float] = []
@@ -1164,6 +1236,7 @@ def summarize_backtest_outcomes(outcomes: list[BacktestOutcome], subject_type: s
        weekday = str(outcome.meta.get('weekday') or 'unknown')
        symbol = _extract_symbol_from_outcome(outcome)
        timeframe_bucket = str(outcome.meta.get('signal_timeframe_bucket') or 'unknown')
+        doctrine_cohort_key = str(outcome.meta.get('doctrine_cohort_key') or 'missing_playbook::incomplete')
        r_multiples.append(float(outcome.meta.get('r_multiple') or 0.0))
        cost_values.append(float(outcome.meta.get('cost_r') or 0.0))
        mfe_r_values.append(float(outcome.meta.get('mfe_r') or 0.0))
@@ -1173,6 +1246,7 @@ def summarize_backtest_outcomes(outcomes: list[BacktestOutcome], subject_type: s
        _accumulate_bucket(grouped_by_weekday, weekday, outcome)
        _accumulate_bucket(grouped_by_symbol, symbol, outcome)
        _accumulate_bucket(grouped_by_timeframe, timeframe_bucket, outcome)
+        _accumulate_bucket(grouped_by_doctrine_cohort, doctrine_cohort_key, outcome)
        for session_code in session_codes:
            _accumulate_bucket(grouped_by_session, str(session_code), outcome)

@@ -1220,6 +1294,7 @@ def summarize_backtest_outcomes(outcomes: list[BacktestOutcome], subject_type: s
        'by_weekday': grouped_by_weekday,
        'by_symbol': grouped_by_symbol,
        'by_timeframe': grouped_by_timeframe,
+        'by_doctrine_cohort': grouped_by_doctrine_cohort,
    }
    result['session_performance'] = result['by_session']
    return result
@@ -19,16 +19,28 @@ from src.services.ict_mainline.feedback import (
 )
 from src.services.ict_mainline.doctrine import (
    DOCTRINE_KERNEL_PHASE,
+    DOCTRINE_PLAYBOOK_PHASE,
+    build_doctrine_backtest_cohort,
    build_doctrine_rule_matrix,
+    build_doctrine_playbook_gate,
+    build_doctrine_playbook_registry,
    evaluate_doctrine_research_status,
+    extract_setup_playbook_id,
+    get_doctrine_setup_playbook,
+    list_doctrine_setup_playbooks,
    list_doctrine_rules,
+    summarize_doctrine_backtest_cohorts,
    summarize_doctrine_rule_gaps,
 )

 __all__ = [
    "DOCTRINE_KERNEL_PHASE",
+    "DOCTRINE_PLAYBOOK_PHASE",
    "ICT_MAINLINE_CHAIN",
+    "build_doctrine_backtest_cohort",
    "build_doctrine_rule_matrix",
+    "build_doctrine_playbook_gate",
+    "build_doctrine_playbook_registry",
    "build_ict_rulebook_mapping",
    "build_ict_rulebook_summary",
    "build_backtest_quality_gate",
@@ -39,9 +51,13 @@ __all__ = [
    "classify_book_rule_error",
    "classify_ict_concepts",
    "evaluate_doctrine_research_status",
+    "extract_setup_playbook_id",
    "get_ict_concept",
+    "get_doctrine_setup_playbook",
+    "list_doctrine_setup_playbooks",
    "list_doctrine_rules",
    "list_ict_concepts",
+    "summarize_doctrine_backtest_cohorts",
    "summarize_doctrine_rule_gaps",
    "summarize_book_rule_errors",
 ]
@@ -7,6 +7,7 @@ from typing import Iterable


 DOCTRINE_KERNEL_PHASE = "D1_MARKET_LANGUAGE_KERNEL"
+DOCTRINE_PLAYBOOK_PHASE = "D2_PLAYBOOK_BACKTEST_COHORT"


@dataclass(frozen=True)
@@ -37,6 +38,44 @@ class DoctrineRule:
        return payload


+@dataclass(frozen=True)
+class DoctrineSetupPlaybook:
+    setup_id: str
+    setup_name: str
+    book_anchors: tuple[str, ...]
+    market_narrative: str
+    range_definition: dict
+    draw_rule: dict
+    location_rule: dict
+    liquidity_event_rule: dict
+    displacement_structure_rule: dict
+    pd_array_rule: dict
+    session_timing_rule: dict
+    confirmation_rule: dict
+    invalidation_rule: dict
+    target_rule: dict
+    no_trade_conditions: tuple[str, ...]
+    backtest_cohort: dict
+    promotion_gate: dict
+    cockpit_sentence: str
+    journal_error_classes: tuple[str, ...]
+    required_rule_ids: tuple[str, ...]
+    aliases: tuple[str, ...] = ()
+
+    def to_public_dict(self) -> dict:
+        payload = asdict(self)
+        for key in (
+            "book_anchors",
+            "no_trade_conditions",
+            "journal_error_classes",
+            "required_rule_ids",
+            "aliases",
+        ):
+            payload[key] = list(payload[key])
+        payload["setup_playbook_id"] = self.setup_id
+        return payload
+
+
 D1_DOCTRINE_RULES: tuple[DoctrineRule, ...] = (
    DoctrineRule(
        rule_id="range_defined",
@@ -92,7 +131,7 @@ D1_DOCTRINE_RULES: tuple[DoctrineRule, ...] = (
        journal_error_class="tool_first_setup",
        backtest_observable="setup_without_context_reference_pair",
        research_only_reason="doctrine_missing_context_reference",
-        aliases=("context", "framework", "reference_point", "setup_type", "model_code", "setup_timeframe", "trigger_timeframe"),
+        aliases=("context", "framework", "reference_point", "setup_type", "market_narrative", "model_code", "setup_timeframe", "trigger_timeframe"),
    ),
    DoctrineRule(
        rule_id="path_time_defined",
@@ -139,6 +178,115 @@ D1_DOCTRINE_RULES: tuple[DoctrineRule, ...] = (
 )


+D2_DOCTRINE_SETUP_PLAYBOOKS: tuple[DoctrineSetupPlaybook, ...] = (
+    DoctrineSetupPlaybook(
+        setup_id="ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
+        setup_name="London discount reversal after range sweep",
+        book_anchors=(
+            "ICT交易全书 chapters 1-2: dealing range, equilibrium, market structure context",
+            "ICT交易全书 chapters 2-4: premium/discount, OTE, and PD array location",
+            "ICT交易全书 chapters 4-6: IRL/ERL target architecture and invalidation",
+        ),
+        market_narrative=(
+            "A London-session candidate is study-only until the active plan defines the dealing range, "
+            "primary draw, discount location, sequence path, invalidation, and IRL/ERL target plan."
+        ),
+        range_definition={
+            "required_fields": ("range_id", "range_high", "range_low", "equilibrium", "range_timeframe"),
+            "operator_rule": "Trade idea must be located relative to the active dealing range and equilibrium.",
+        },
+        draw_rule={
+            "required_fields": ("primary_draw", "draw_side", "objective_price"),
+            "operator_rule": "Bias follows the declared draw on liquidity, not the local FVG/MSS shape alone.",
+        },
+        location_rule={
+            "required_fields": ("premium_discount_zone", "location_advantage"),
+            "operator_rule": "Long ideas need discount or an explicit exception; short ideas need premium or an explicit exception.",
+        },
+        liquidity_event_rule={
+            "required_fields": ("liquidity_event", "swept_level", "sweep_confirmed"),
+            "operator_rule": "Sweep/raid is an evidence component, not the setup by itself.",
+        },
+        displacement_structure_rule={
+            "required_fields": ("mss_state", "displacement_state", "structure_shift_ts"),
+            "operator_rule": "Displacement and MSS must confirm the post-sweep direction before review.",
+        },
+        pd_array_rule={
+            "required_fields": ("pd_array_type", "pd_array_quality", "fvg_retrace_confirmed"),
+            "operator_rule": "FVG is usable only when it serves the range/draw/location chain.",
+        },
+        session_timing_rule={
+            "required_fields": ("session_code", "allowed_sessions", "path_context", "sequence_validity"),
+            "operator_rule": "Only London/NY timing defined by the active plan can produce a review candidate.",
+        },
+        confirmation_rule={
+            "required_fields": ("confirmation_state", "missing_confirmations", "reason_codes"),
+            "operator_rule": "Review is blocked while required confirmations are missing or rejected.",
+        },
+        invalidation_rule={
+            "required_fields": ("invalidations", "stop_loss", "invalid_level"),
+            "operator_rule": "No explicit invalidation means no Review Now.",
+        },
+        target_rule={
+            "required_fields": ("target_plan", "tp1", "tp2", "irl", "erl", "management_rules"),
+            "operator_rule": "Targets must separate internal and external range objectives before promotion.",
+        },
+        no_trade_conditions=(
+            "missing_doctrine_setup_playbook",
+            "doctrine_missing_range",
+            "doctrine_missing_draw",
+            "doctrine_missing_location",
+            "doctrine_missing_time_path",
+            "doctrine_missing_invalidation",
+            "doctrine_missing_exit_architecture",
+            "backtest_quality_gate_failed",
+            "missing_active_trading_plan",
+        ),
+        backtest_cohort={
+            "cohort_family": "london_discount_reversal",
+            "cohort_key_fields": (
+                "setup_playbook_id",
+                "doctrine_completeness",
+                "bias_tf",
+                "setup_tf",
+                "trigger_tf",
+                "session_code",
+            ),
+            "sample_basis": "executed",
+            "execution_boundary": "read_only_backtest_no_order_submission",
+        },
+        promotion_gate={
+            "requires_playbook_schema_complete": True,
+            "requires_backtest_quality_gate": True,
+            "requires_active_trading_plan": True,
+            "requires_operator_confirmation": True,
+            "execution_boundary": "read_only_analysis_no_order_submission",
+        },
+        cockpit_sentence=(
+            "Observe Only until the London discount reversal playbook, historical cohort, and active plan all pass."
+        ),
+        journal_error_classes=(
+            "traded_without_range",
+            "traded_without_draw",
+            "poor_trade_location",
+            "traded_without_time_or_path",
+            "traded_without_invalidation",
+            "no_exit_architecture",
+        ),
+        required_rule_ids=(
+            "range_defined",
+            "draw_defined",
+            "location_quality_defined",
+            "context_reference_match",
+            "path_time_defined",
+            "invalidation_defined",
+            "exit_architecture_defined",
+        ),
+        aliases=("london_discount_reversal", "range_sweep_mss_fvg_retrace", "osok_daily_range_sweep_mss_fvg_retrace"),
+    ),
+)
+
+
 def list_doctrine_rules() -> list[dict]:
    return [rule.to_public_dict() for rule in D1_DOCTRINE_RULES]

@@ -162,6 +310,183 @@ def build_doctrine_rule_matrix() -> dict:
    }


+def list_doctrine_setup_playbooks() -> list[dict]:
+    return [playbook.to_public_dict() for playbook in D2_DOCTRINE_SETUP_PLAYBOOKS]
+
+
+def build_doctrine_playbook_registry() -> dict:
+    return {
+        "phase": DOCTRINE_PLAYBOOK_PHASE,
+        "source": "docs/AI_ICT_DoctrineKernel市场语言规则内核_D1_2026-05-21_v1.md",
+        "status": "active",
+        "playbook_count": len(D2_DOCTRINE_SETUP_PLAYBOOKS),
+        "playbooks": list_doctrine_setup_playbooks(),
+        "execution_boundary": "read_only_analysis_no_order_submission",
+    }
+
+
+def get_doctrine_setup_playbook(setup_playbook_id: str | None) -> dict | None:
+    key = str(setup_playbook_id or "").strip()
+    if not key:
+        return None
+    key_lower = key.lower()
+    for playbook in D2_DOCTRINE_SETUP_PLAYBOOKS:
+        if playbook.setup_id.lower() == key_lower:
+            return playbook.to_public_dict()
+        if key_lower in {alias.lower() for alias in playbook.aliases}:
+            return playbook.to_public_dict()
+    return None
+
+
+def extract_setup_playbook_id(payload: object):
+    return _extract_first_value(
+        payload,
+        keys=("setup_playbook_id", "doctrine_setup_id", "playbook_id"),
+    )
+
+
+def build_doctrine_playbook_gate(payload: object) -> dict:
+    setup_playbook_id = extract_setup_playbook_id(payload)
+    if not setup_playbook_id:
+        return {
+            "phase": DOCTRINE_PLAYBOOK_PHASE,
+            "status": "blocked",
+            "passed": False,
+            "reason": "missing_doctrine_setup_playbook",
+            "reasons": ["missing_doctrine_setup_playbook"],
+            "setup_playbook_id": None,
+            "setup_playbook_known": False,
+            "schema_completeness": "missing",
+            "operator_summary": "Review Now is blocked until the active plan/candidate maps to a Doctrine setup playbook.",
+            "execution_boundary": "read_only_analysis_no_order_submission",
+        }
+
+    playbook = get_doctrine_setup_playbook(str(setup_playbook_id))
+    if not playbook:
+        return {
+            "phase": DOCTRINE_PLAYBOOK_PHASE,
+            "status": "blocked",
+            "passed": False,
+            "reason": "unknown_doctrine_setup_playbook",
+            "reasons": ["unknown_doctrine_setup_playbook"],
+            "setup_playbook_id": setup_playbook_id,
+            "setup_playbook_known": False,
+            "schema_completeness": "unknown",
+            "operator_summary": "Review Now is blocked because the referenced Doctrine setup playbook is not registered.",
+            "execution_boundary": "read_only_analysis_no_order_submission",
+        }
+
+    missing_sections = _missing_playbook_sections(playbook)
+    doctrine_status = evaluate_doctrine_research_status(playbook, require_setup_playbook=True)
+    reasons = [*missing_sections]
+    if doctrine_status.get("research_only"):
+        reasons.extend([str(item) for item in list(doctrine_status.get("research_only_reasons") or [])])
+    reasons = _dedupe(reasons)
+    passed = not reasons
+    return {
+        "phase": DOCTRINE_PLAYBOOK_PHASE,
+        "status": "passed" if passed else "blocked",
+        "passed": passed,
+        "reason": "doctrine_setup_playbook_ready" if passed else reasons[0],
+        "reasons": [] if passed else reasons,
+        "setup_playbook_id": playbook.get("setup_id"),
+        "setup_playbook_known": True,
+        "setup_name": playbook.get("setup_name"),
+        "schema_completeness": "complete" if passed else "incomplete",
+        "doctrine_research_status": doctrine_status,
+        "promotion_gate": dict(playbook.get("promotion_gate") or {}),
+        "backtest_cohort": dict(playbook.get("backtest_cohort") or {}),
+        "operator_summary": (
+            "Doctrine setup playbook schema is complete; historical quality gate and active Trading Plan still decide promotion."
+            if passed
+            else "Review Now is blocked until the Doctrine setup playbook schema is complete."
+        ),
+        "execution_boundary": "read_only_analysis_no_order_submission",
+    }
+
+
+def build_doctrine_backtest_cohort(payload: object, *, quality_gate: dict | None = None) -> dict:
+    setup_playbook_id = extract_setup_playbook_id(payload)
+    playbook_gate = build_doctrine_playbook_gate(payload)
+    quality = dict(quality_gate or _extract_quality_gate(payload) or {})
+    quality_reasons = [str(item) for item in list(quality.get("reasons") or []) if item]
+    quality_passed = bool(quality.get("passed"))
+    if not quality:
+        quality_reasons = ["backtest_quality_gate_missing"]
+    elif not quality_passed and not quality_reasons:
+        quality_reasons = ["backtest_quality_gate_failed"]
+    doctrine_completeness = "complete" if playbook_gate.get("passed") else "incomplete"
+    resolved_playbook_id = str(playbook_gate.get("setup_playbook_id") or setup_playbook_id or "missing_playbook")
+    cohort_key = f"{resolved_playbook_id}::{doctrine_completeness}"
+    promotion_reasons = [
+        *[str(item) for item in list(playbook_gate.get("reasons") or []) if item],
+        *([] if quality_passed else quality_reasons),
+    ]
+    promotion_reasons = _dedupe(promotion_reasons)
+    return {
+        "phase": DOCTRINE_PLAYBOOK_PHASE,
+        "setup_playbook_id": playbook_gate.get("setup_playbook_id") or setup_playbook_id,
+        "setup_playbook_known": bool(playbook_gate.get("setup_playbook_known")),
+        "setup_name": playbook_gate.get("setup_name") or "",
+        "doctrine_completeness": doctrine_completeness,
+        "cohort_key": cohort_key,
+        "cohort_key_fields": ["setup_playbook_id", "doctrine_completeness"],
+        "quality_gate_passed": quality_passed,
+        "playbook_gate": playbook_gate,
+        "promotion_gate": {
+            "passed": not promotion_reasons,
+            "reasons": promotion_reasons,
+            "requires_playbook_schema_complete": True,
+            "requires_backtest_quality_gate": True,
+            "requires_active_trading_plan": True,
+            "execution_boundary": "read_only_analysis_no_order_submission",
+        },
+        "operator_summary": (
+            "Backtest cohort is promotable only after setup playbook and historical quality gate both pass."
+            if promotion_reasons
+            else "Doctrine setup playbook cohort passed the stored historical quality gate."
+        ),
+        "execution_boundary": "read_only_backtest_no_order_submission",
+    }
+
+
+def summarize_doctrine_backtest_cohorts(items: Iterable[dict]) -> dict:
+    cohorts: dict[str, dict] = {}
+    for item in items:
+        cohort = dict(item.get("doctrine_backtest_cohort") or build_doctrine_backtest_cohort(item))
+        key = str(cohort.get("cohort_key") or "missing_playbook::incomplete")
+        bucket = cohorts.setdefault(
+            key,
+            {
+                "cohort_key": key,
+                "setup_playbook_id": cohort.get("setup_playbook_id"),
+                "doctrine_completeness": cohort.get("doctrine_completeness"),
+                "run_count": 0,
+                "quality_gate_passed_count": 0,
+                "promotable_count": 0,
+                "promotion_blocker_counts": {},
+            },
+        )
+        bucket["run_count"] += 1
+        if cohort.get("quality_gate_passed"):
+            bucket["quality_gate_passed_count"] += 1
+        if (cohort.get("promotion_gate") or {}).get("passed"):
+            bucket["promotable_count"] += 1
+        for reason in list((cohort.get("promotion_gate") or {}).get("reasons") or []):
+            reason_key = str(reason)
+            bucket["promotion_blocker_counts"][reason_key] = int(bucket["promotion_blocker_counts"].get(reason_key) or 0) + 1
+    return {
+        "phase": DOCTRINE_PLAYBOOK_PHASE,
+        "cohort_count": len(cohorts),
+        "cohorts": sorted(cohorts.values(), key=lambda item: (-int(item["run_count"]), str(item["cohort_key"]))),
+        "operator_summary": (
+            "No Doctrine backtest cohorts yet."
+            if not cohorts
+            else "Review playbook-level cohorts; legacy model_code alone is no longer enough for promotion."
+        ),
+    }
+
+
 def evaluate_doctrine_research_status(
    payload: object,
    *,
@@ -293,6 +618,50 @@ def _extract_first_value(payload: object, *, keys: tuple[str, ...]):
    return None


+def _missing_playbook_sections(playbook: dict) -> list[str]:
+    required_sections = (
+        "range_definition",
+        "draw_rule",
+        "location_rule",
+        "liquidity_event_rule",
+        "displacement_structure_rule",
+        "pd_array_rule",
+        "session_timing_rule",
+        "confirmation_rule",
+        "invalidation_rule",
+        "target_rule",
+        "backtest_cohort",
+        "promotion_gate",
+    )
+    missing = []
+    for section in required_sections:
+        if not dict(playbook.get(section) or {}):
+            missing.append(f"playbook_missing_{section}")
+    if not list(playbook.get("required_rule_ids") or []):
+        missing.append("playbook_missing_required_rule_ids")
+    return missing
+
+
+def _extract_quality_gate(payload: object) -> dict | None:
+    if isinstance(payload, dict):
+        quality_gate = payload.get("quality_gate")
+        if isinstance(quality_gate, dict):
+            return quality_gate
+        summary = payload.get("summary")
+        if isinstance(summary, dict) and isinstance(summary.get("quality_gate"), dict):
+            return dict(summary.get("quality_gate") or {})
+        for value in payload.values():
+            found = _extract_quality_gate(value)
+            if found:
+                return found
+    if isinstance(payload, list):
+        for value in payload:
+            found = _extract_quality_gate(value)
+            if found:
+                return found
+    return None
+
+
 def _dedupe(values: Iterable[str]) -> list[str]:
    result = []
    seen = set()
@@ -1,5 +1,7 @@
 from __future__ import annotations

+from src.services.ict_mainline.doctrine import build_doctrine_playbook_gate
+

 PLAN_BLOCKING_CONDITIONS = {
    "missing_primary_draw",
@@ -91,25 +93,29 @@ def build_review_now_mainline_gate(
    *,
    active_trading_plan: dict | None,
    backtest_quality_gate: dict | None,
+    doctrine_playbook_gate: dict | None = None,
 ) -> dict:
    plan_gate = build_trading_plan_gate(active_trading_plan)
    backtest_gate = build_backtest_quality_gate(backtest_quality_gate)
+    doctrine_gate = doctrine_playbook_gate or build_doctrine_playbook_gate(active_trading_plan or {})
    blockers = [
        *[str(item) for item in list(plan_gate.get("reasons") or []) if item],
        *[str(item) for item in list(backtest_gate.get("reasons") or []) if item],
+        *[str(item) for item in list(doctrine_gate.get("reasons") or []) if item],
    ]
-    review_now_allowed = bool(plan_gate.get("passed") and backtest_gate.get("passed"))
+    review_now_allowed = bool(plan_gate.get("passed") and backtest_gate.get("passed") and doctrine_gate.get("passed"))
    return {
        "status": "passed" if review_now_allowed else "blocked",
        "review_now_allowed": review_now_allowed,
        "blockers": _dedupe(blockers),
        "trading_plan_gate": plan_gate,
        "backtest_quality_gate": backtest_gate,
+        "doctrine_playbook_gate": doctrine_gate,
        "execution_boundary": "read_only_analysis_no_order_submission",
        "operator_summary": (
-            "Review Now may be shown only after the active ICT Trading Plan and Strategy Backtest quality gate both pass."
+            "Review Now may be shown only after the Doctrine playbook, active ICT Trading Plan, and Strategy Backtest quality gate all pass."
            if review_now_allowed
-            else "Review Now is downgraded to Observe Only until Trading Plan and Strategy Backtest evidence both pass."
+            else "Review Now is downgraded to Observe Only until Doctrine playbook, Trading Plan, and Strategy Backtest evidence all pass."
        ),
    }

@@ -5526,6 +5526,8 @@ function renderStrategyBacktests(payload) {
  const qualityGate = payload?.quality_gate || summary.quality_gate || {};
  const passed = Boolean(qualityGate.passed);
  const operatorSummary = payload?.operator_summary || {};
+  const doctrineCohort = latest.doctrine_backtest_cohort || summary.doctrine_backtest_cohort || {};
+  const doctrinePromotionGate = doctrineCohort.promotion_gate || {};

  strategyBacktestPanel.appendChild(
    buildInfoGrid([
@@ -5533,6 +5535,10 @@ function renderStrategyBacktests(payload) {
      ['Quality Gate', passed ? 'Passed' : 'Blocked'],
      ['Decision', formatBacktestDecision(operatorSummary.decision)],
      ['Reason', formatBacktestDecision(operatorSummary.reason)],
+      ['Setup Playbook', doctrineCohort.setup_playbook_id || 'missing_playbook'],
+      ['Doctrine Cohort', doctrineCohort.cohort_key || 'missing_playbook::incomplete'],
+      ['Doctrine Completeness', doctrineCohort.doctrine_completeness || 'incomplete'],
+      ['Promotion Gate', doctrinePromotionGate.passed ? 'Passed' : 'Blocked'],
      ['Instrument ID', latest.instrument_id ?? 'n/a'],
      ['Timeframe', formatJsonInline(latest.timeframe_set || {})],
      ['Started', formatTimestamp(latest.started_ts)],
@@ -5582,13 +5588,23 @@ function renderStrategyBacktests(payload) {
      topCountEntries(summary.failure_reasons || {}).map((item) => [formatBacktestDecision(item.key), String(item.value)])
    )
  );
+  detailGrid.appendChild(
+    buildItemCard(
+      'Doctrine Cohorts',
+      `${payload?.doctrine_cohort_summary?.cohort_count ?? 0} cohorts`,
+      (payload?.doctrine_cohort_summary?.cohorts || []).slice(0, 5).map((item) => [
+        item.cohort_key || 'missing_playbook::incomplete',
+        `${item.run_count ?? 0} runs / ${item.quality_gate_passed_count ?? 0} passed / ${item.promotable_count ?? 0} promotable`,
+      ])
+    )
+  );
  detailGrid.appendChild(
    buildItemCard(
      'Recent Backtest Runs',
      `${payload?.item_count ?? (payload?.items || []).length} runs`,
      (payload?.items || []).slice(0, 5).map((item) => [
        `#${item.id}`,
-        `${item.model_code || 'model'} / ${formatTimestamp(item.started_ts)} / ${item.summary?.quality_gate?.passed ? 'Passed' : 'Blocked'}`,
+        `${item.doctrine_backtest_cohort?.cohort_key || item.model_code || 'model'} / ${formatTimestamp(item.started_ts)} / ${item.summary?.quality_gate?.passed ? 'Passed' : 'Blocked'}`,
      ])
    )
  );
@@ -738,6 +738,18 @@
    ['backtest_quality_gate_passed', '回测质量门已通过'],
    ['backtest_quality_gate_missing', '缺少回测质量门'],
    ['backtest_quality_gate_failed', '回测质量门未通过'],
+    ['missing_doctrine_setup_playbook', '缺少 Doctrine setup playbook'],
+    ['unknown_doctrine_setup_playbook', '未知 Doctrine setup playbook'],
+    ['doctrine_setup_playbook_ready', 'Doctrine setup playbook 已就绪'],
+    ['missing_playbook', '缺少 playbook'],
+    ['Doctrine Cohort', 'Doctrine 回测分组'],
+    ['Doctrine Cohorts', 'Doctrine 回测分组'],
+    ['Setup Playbook', 'Setup Playbook'],
+    ['Doctrine Completeness', 'Doctrine 完整度'],
+    ['Promotion Gate', '晋级门'],
+    ['complete', '完整'],
+    ['incomplete', '不完整'],
+    ['promotable', '可晋级'],
    ['sample_size_too_small', '回测样本太少'],
    ['executed_sample_too_small', '执行样本太少'],
    ['expectancy_below_threshold', '期望值低于阈值'],
@@ -749,6 +761,7 @@
    ['candidate_plan_mismatch', '候选与当前盘前计划不匹配'],
    ['active_plan_ready', '盘前计划已就绪'],
    ['Review Now is downgraded to Observe Only until Trading Plan and Strategy Backtest evidence both pass.', '盘前计划和回测证据同时通过前，Review Now 降级为仅观察。'],
+    ['Review Now is downgraded to Observe Only until Doctrine playbook, Trading Plan, and Strategy Backtest evidence all pass.', 'Doctrine playbook、交易计划和回测证据全部通过前，Review Now 降级为仅观察。'],
    ['positive_expectancy', '正期望'],
    ['operator_feedback_sample_gate', '操作员反馈样本门'],
    ['framework_execution_rate_gate', '框架执行率门'],
@@ -134,6 +134,22 @@ class ApiTests(unittest.TestCase):
        self.assertEqual(payload["phase"], "D1_MARKET_LANGUAGE_KERNEL")
        self.assertEqual(payload["rules"][0]["rule_id"], "range_defined")

+    def test_route_request_trading_doctrine_playbooks_returns_d2_registry(self) -> None:
+        status, payload = route_request(
+            "/trading/doctrine-playbooks",
+            {},
+            dependencies={
+                "trading_doctrine_playbooks": lambda: {
+                    "phase": "D2_PLAYBOOK_BACKTEST_COHORT",
+                    "playbooks": [{"setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL"}],
+                }
+            },
+        )
+
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["phase"], "D2_PLAYBOOK_BACKTEST_COHORT")
+        self.assertEqual(payload["playbooks"][0]["setup_playbook_id"], "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL")
+
    def test_route_request_trading_candidates_returns_items_payload(self) -> None:
        status, payload = route_request(
            "/trading/candidates",
@@ -102,6 +102,22 @@ class ApiServerRoutingTests(unittest.TestCase):
        self.assertEqual(payload["latest"]["id"], 7)
        self.assertTrue(payload["quality_gate"]["passed"])

+    def test_routes_trading_doctrine_playbooks(self) -> None:
+        status, payload = route_request(
+            "/trading/doctrine-playbooks",
+            {},
+            dependencies={
+                "trading_doctrine_playbooks": lambda: {
+                    "phase": "D2_PLAYBOOK_BACKTEST_COHORT",
+                    "playbook_count": 1,
+                }
+            },
+        )
+
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["phase"], "D2_PLAYBOOK_BACKTEST_COHORT")
+        self.assertEqual(payload["playbook_count"], 1)
+
    def test_routes_trading_alert_quality_summary(self) -> None:
        status, payload = route_request(
            "/trading/alerts/quality-summary",
@@ -7,6 +7,7 @@ from scripts.run_backtest_batch import (
    build_candidate_model_codes_from_env,
    build_instrument_ids_from_env,
    build_session_groups_from_env,
+    build_setup_playbook_ids_from_env,
    build_subject_types_from_env,
    build_timeframe_sets_from_env,
    run_backtest_batch as run_batch,
@@ -22,6 +23,7 @@ class RunBacktestBatchScriptTests(unittest.TestCase):
            "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",
+            "BACKTEST_BATCH_SETUP_PLAYBOOK_IDS": "ALL,ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
        }

        self.assertEqual(build_instrument_ids_from_env(env), [1, 2])
@@ -35,6 +37,7 @@ class RunBacktestBatchScriptTests(unittest.TestCase):
        )
        self.assertEqual(build_session_groups_from_env(env), [[], ["LONDON", "NY_AM"], ["ASIA"]])
        self.assertEqual(build_candidate_model_codes_from_env(env), [None, "OSOK"])
+        self.assertEqual(build_setup_playbook_ids_from_env(env), [None, "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL"])

    def test_run_batch_aggregates_quality_gate_and_metadata(self) -> None:
        service = FakeBatchBacktestService(
@@ -54,6 +57,7 @@ class RunBacktestBatchScriptTests(unittest.TestCase):
            timeframe_sets=[{"bias_tf": "1h", "setup_tf": "15m", "trigger_tf": "5m"}],
            session_groups=[[], ["LONDON"]],
            candidate_model_codes=[None],
+            setup_playbook_ids=["ICT-CH1-6-LONDON-DISCOUNT-REVERSAL"],
            require_quality_gate=False,
            quality_gate=BacktestQualityGate(min_total_signals=1, min_executed_signals=1, min_profit_factor_r=0.0),
            cost_config=None,
@@ -65,7 +69,9 @@ class RunBacktestBatchScriptTests(unittest.TestCase):
        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.assertIn("ICT-CH1-6-LONDON-DISCOUNT-REVERSAL::complete", payload["aggregate_by_doctrine_cohort"])
        self.assertEqual(service.calls[0]["run_metadata"]["batch_id"], "batch-test")
+        self.assertEqual(service.calls[0]["run_metadata"]["setup_playbook_id"], "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL")
        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")

@@ -698,6 +698,7 @@ class BacktestServiceRunTests(unittest.TestCase):
                        "candidate_id": 11,
                        "trading_plan_id": 77,
                        "candidate_model_code": "OSOK",
+                        "setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
                        "candidate_status": "awaiting_confirmation",
                        "side": "buy",
                        "setup_timeframe": "15m",
@@ -716,6 +717,7 @@ class BacktestServiceRunTests(unittest.TestCase):
                        "candidate_id": 12,
                        "trading_plan_id": 77,
                        "candidate_model_code": "OSOK",
+                        "setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
                        "candidate_status": "executable",
                        "side": "buy",
                        "setup_timeframe": "15m",
@@ -787,6 +789,14 @@ class BacktestServiceRunTests(unittest.TestCase):
        self.assertEqual(repository.summary["confirmed_candidates"], 1)
        self.assertAlmostEqual(repository.summary["candidate_confirmation_rate"], 1.0)
        self.assertIn("draw_target_hit_rate", repository.summary)
+        self.assertEqual(
+            repository.summary["doctrine_backtest_cohort"]["cohort_key"],
+            "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL::complete",
+        )
+        self.assertEqual(
+            repository.summary["by_doctrine_cohort"]["ICT-CH1-6-LONDON-DISCOUNT-REVERSAL::complete"]["count"],
+            1,
+        )

    def test_normalize_timeframe_set_and_bucket_helpers(self) -> None:
        normalized = normalize_timeframe_set({"bias_tf": "1h", "setup_tf": "15m"})
@@ -19,6 +19,7 @@ NONEXISTENT_VISUAL_PLAN_NAME = "AI_ICT_" + "视觉盯盘与主动告警增强方

 ACTIVE_TOP_LEVEL_DOCS = {
    "AI_ICT_DoctrineKernel市场语言规则内核_D1_2026-05-21_v1.md",
+    "AI_ICT_DoctrineKernel_PlaybookBacktestCohort_D2_2026-05-21_v1.md",
    "AI_ICT_硬性主线约束_必须遵循ICT交易全书_2026-05-21_v1.md",
    "AI_ICT_实盘辅助分析决策助手ICT主线开发方案_2026-05-21_v1.md",
    "AI_ICT_主线裁剪审计_S1_2026-05-21_v1.md",
@@ -2,6 +2,10 @@ import unittest

 from src.services.ict_mainline import (
    DOCTRINE_KERNEL_PHASE,
+    DOCTRINE_PLAYBOOK_PHASE,
+    build_doctrine_backtest_cohort,
+    build_doctrine_playbook_gate,
+    build_doctrine_playbook_registry,
    build_doctrine_rule_matrix,
    evaluate_doctrine_research_status,
    summarize_doctrine_rule_gaps,
@@ -21,6 +25,30 @@ class IctDoctrineKernelTests(unittest.TestCase):
        self.assertGreaterEqual(len(matrix["rules"]), 7)
        self.assertEqual(matrix["rules"][0]["rule_id"], "range_defined")

+    def test_d2_playbook_registry_exposes_first_real_setup_schema(self) -> None:
+        registry = build_doctrine_playbook_registry()
+
+        self.assertEqual(registry["phase"], DOCTRINE_PLAYBOOK_PHASE)
+        self.assertEqual(registry["playbook_count"], 1)
+        playbook = registry["playbooks"][0]
+        self.assertEqual(playbook["setup_playbook_id"], "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL")
+        self.assertIn("range_definition", playbook)
+        self.assertIn("target_rule", playbook)
+        self.assertTrue(playbook["promotion_gate"]["requires_backtest_quality_gate"])
+
+    def test_playbook_gate_blocks_missing_and_passes_registered_schema(self) -> None:
+        missing = build_doctrine_playbook_gate({"model_code": "SWEEP_MSS_FVG"})
+
+        self.assertFalse(missing["passed"])
+        self.assertEqual(missing["reason"], "missing_doctrine_setup_playbook")
+
+        passed = build_doctrine_playbook_gate(
+            {"setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL"}
+        )
+
+        self.assertTrue(passed["passed"])
+        self.assertEqual(passed["schema_completeness"], "complete")
+
    def test_surface_shape_candidate_without_doctrine_playbook_is_research_only(self) -> None:
        status = evaluate_doctrine_research_status(
            {
@@ -83,6 +111,23 @@ class IctDoctrineKernelTests(unittest.TestCase):
        self.assertEqual(summary["research_only_reason_counts"]["missing_doctrine_setup_playbook"], 2)
        self.assertGreater(summary["missing_rule_counts"]["draw_defined"], 0)

+    def test_doctrine_backtest_cohort_uses_playbook_and_completeness(self) -> None:
+        cohort = build_doctrine_backtest_cohort(
+            {
+                "setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
+                "summary": {"quality_gate": {"passed": True, "reasons": []}},
+            }
+        )
+
+        self.assertEqual(cohort["cohort_key"], "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL::complete")
+        self.assertEqual(cohort["doctrine_completeness"], "complete")
+        self.assertTrue(cohort["promotion_gate"]["passed"])
+
+        missing = build_doctrine_backtest_cohort({"model_code": "SWEEP_MSS_FVG"})
+
+        self.assertEqual(missing["cohort_key"], "missing_playbook::incomplete")
+        self.assertIn("missing_doctrine_setup_playbook", missing["promotion_gate"]["reasons"])
+

 if __name__ == "__main__":
    unittest.main()
@@ -132,6 +132,7 @@ class IctMainlineTaxonomyTests(unittest.TestCase):
            active_trading_plan={
                "id": 10,
                "status": "active",
+                "setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
                "primary_draw": {"side": "buy_side"},
                "invalidations": [{"code": "range_low_lost"}],
                "no_trade_conditions": [],
@@ -1458,6 +1458,7 @@ class _CandidateWatchlistRepo:
            "instrument_id": instrument_id,
            "plan_date": "2026-05-12",
            "status": "active",
+            "setup_playbook_id": "ICT-CH1-6-LONDON-DISCOUNT-REVERSAL",
            "primary_draw": {"side": "buy_side", "source": "liquidity_pool"},
            "invalidations": [{"code": "range_low_lost"}],
            "no_trade_conditions": [],