diff --git a/README.md b/README.md
index 2b19126..a1bfbf5 100644
--- a/README.md
+++ b/README.md
@@ -141,6 +141,65 @@ test "$(docker inspect -f '{{.State.Health.Status}}' ai-ict)" = "healthy"
 .venv/bin/python scripts/check_local_demo_readiness.py
 ```
 
+## Trading Delivery Checklist
+
+Use this locked trading delivery checklist before every market validation or deploy, so the trading module can be promoted without relying on chat memory.
+
+must-run before every market validation or deploy:
+
+```bash
+.venv/bin/python -m unittest discover -s tests/unit -p "test_*.py"
+.venv/bin/python scripts/run_offline_checks.py
+.venv/bin/python scripts/check_local_demo_readiness.py
+.venv/bin/python scripts/check_trading_delivery_readiness.py
+```
+
+Database gate:
+
+```bash
+READINESS_OUTPUT_JSON=1 .venv/bin/python scripts/run_database_readiness.py
+HISTORICAL_REPAIR_OUTPUT_JSON=1 .venv/bin/python scripts/run_historical_compatibility_repair.py
+.venv/bin/python scripts/run_historical_compatibility_repair.py --apply-safe-repairs
+HISTORICAL_REPAIR_OUTPUT_JSON=1 .venv/bin/python scripts/run_historical_compatibility_repair.py
+PIPELINE_OUTPUT_JSON=1 BACKTEST_REQUIRE_QUALITY_GATE=1 .venv/bin/python scripts/run_pipeline.py
+```
+
+Use `--apply-safe-repairs` only when the dry-run output shows safe candidates. After the second dry-run, manual review is still required for `needs_sweep_reference_migration` and `keep_version_history`, and `backtest_quality_gate.passed` must be `true` before promotion.
+
+Deploy and health check:
+
+```bash
+docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml down --remove-orphans
+docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml up -d --build
+docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml ps
+test "$(docker ps --filter label=com.docker.compose.project=ai-ict --format '{{.Names}}' | wc -l | tr -d ' ')" = "1"
+test "$(docker ps --filter name=^/ai-ict$ --format '{{.Names}}' | wc -l | tr -d ' ')" = "1"
+test "$(docker inspect -f '{{.State.Health.Status}}' ai-ict)" = "healthy"
+curl -fsS http://127.0.0.1:8000/health
+curl -fsS "http://127.0.0.1:8000/signals?instrument_id=1&limit=1"
+```
+
+Post-deploy observe:
+
+```bash
+READINESS_OUTPUT_JSON=1 .venv/bin/python scripts/run_database_readiness.py
+REPORTS_OUTPUT_JSON=1 .venv/bin/python scripts/run_reports.py
+```
+
+The paper execution boundary remains in force here: `run_pipeline.py` and the execution path are still paper-only, and any API `5xx`, Docker unhealthy state, or non-empty readiness reasons should stop promotion immediately.
+
+Rollback trigger:
+
+```bash
+git rev-parse HEAD
+docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml logs --tail=100 ai-ict
+git worktree add /tmp/ai-ict-rollback <last-known-good-sha>
+cd /tmp/ai-ict-rollback && ./scripts/deploy_server.exp
+rm -rf /tmp/ai-ict-rollback
+```
+
+Record the current commit as the last-known-good SHA before deploy. If any gate above fails, capture logs first, then redeploy from the clean rollback worktree rather than mutating the active development checkout.
+
 ## Runtime utility scripts
 
 - `python3 scripts/run_dev.py` - start the local development API server and minimal dashboard
@@ -161,13 +220,14 @@ test "$(docker inspect -f '{{.State.Health.Status}}' ai-ict)" = "healthy"
 - `python3 scripts/run_backtest.py` - run backtest and write results
 - `python3 scripts/run_autonomous_cycle.py` - run the unattended cycle: offline checks, database probe, readiness checks, then guarded pipeline when the database is available; prints `Next actions` for the next unattended step
 - `python3 scripts/run_database_readiness.py` - run read-only database readiness checks for connection, schema, seed data, and candle availability; prints `next_actions` and `action_plan` in JSON mode
-- `.venv/bin/python scripts/run_offline_checks.py` - run the broader offline check suite, including the M35 acceptance posture through the Training Case acceptance manifest for focused milestone test suites, the Training Case package smoke registered as `training_case_package`, dashboard syntax validation as `dashboard_js_syntax`, and dashboard and public docs privacy smoke coverage; this public validation runs as offline checks without external services
+- `.venv/bin/python scripts/run_offline_checks.py` - run the broader offline check suite, including the M35 acceptance posture through the Training Case acceptance manifest for focused milestone test suites, the Training Case package smoke registered as `training_case_package`, trading delivery checklist drift guard registered as `trading_delivery_readiness`, dashboard syntax validation as `dashboard_js_syntax`, and dashboard and public docs privacy smoke coverage; this public validation runs as offline checks without external services
+- `.venv/bin/python scripts/check_trading_delivery_readiness.py` - verify `README.md` and `RUNBOOK.md` still expose the locked trading delivery checklist for preflight, deploy, observe, and rollback without relying on chat memory
 - `.venv/bin/python scripts/run_training_case_package.py --help` - show the offline package smoke help for public package import/export behavior, public export/public output semantics, and stable public error codes for invalid input paths and invalid output targets
 - `.venv/bin/python scripts/run_training_case_package.py --input tests/fixtures/training_case_package_m6.json --output /tmp/training_case_public_export.json` - run the Training Case offline package smoke, an import/export smoke with local files, and write public output as the public export in the public package shape; exported public JSON omits internal solution details
 - `.venv/bin/python scripts/run_training_case_generator.py --help` - show the offline prototype generator help for read-only Training case draft generation from trading-derived seed events
 - `.venv/bin/python scripts/run_training_case_generator.py --input tests/fixtures/training_case_generator_seed_v1.json --output /tmp/generated_training_case_drafts.json` - run the offline Training case generator prototype and write a public draft package without touching Training database tables
 - `.venv/bin/python scripts/run_historical_compatibility_repair.py` - scan for older stored signal/FVG/MSS/liquidity compatibility issues in the connected database without applying changes
-- `.venv/bin/python scripts/run_historical_compatibility_repair.py --apply-safe-repairs` - apply only the safe historical compatibility repairs for signal invalidations plus MSS/FVG meta backfills; liquidity duplicate groups remain manual-review only
+- `.venv/bin/python scripts/run_historical_compatibility_repair.py --apply-safe-repairs` - apply the safe historical compatibility repairs for signal invalidations, MSS/FVG meta backfills, and liquidity safe merges limited to unreferenced supersets plus exact duplicate rows with identical sweep events; `needs_sweep_reference_migration` and `keep_version_history` groups still require manual review
 - `python3 scripts/run_migration_static_checks.py` - check migration order and required core schema without connecting to the database
 - `python3 scripts/run_script_static_checks.py` - check automation scripts for unsafe ambient environment fallback patterns
 - `python3 scripts/run_replay_file.py <candles.json|candles.jsonl|candles.csv>` - run offline no-lookahead candle replay without database access
diff --git a/RUNBOOK.md b/RUNBOOK.md
index cee33a3..929bb12 100644
--- a/RUNBOOK.md
+++ b/RUNBOOK.md
@@ -38,6 +38,7 @@ cd /home/yuqei/project/ai-exchange
 .venv/bin/python -m unittest discover -s tests/unit -p "test_*.py"
 .venv/bin/python scripts/run_offline_checks.py
 .venv/bin/python scripts/check_local_demo_readiness.py
+.venv/bin/python scripts/check_trading_delivery_readiness.py
 python3 scripts/run_migration_static_checks.py
 python3 scripts/run_script_static_checks.py
 ```
@@ -74,8 +75,8 @@ Prototype intent:
 The current historical compatibility repair path is intentionally safe by default:
 
 - default mode is read-only scan
-- safe apply mode repairs only signal invalidations plus MSS/FVG meta compatibility fields
-- liquidity duplicate groups are reported for manual review and are not auto-merged
+- safe apply mode repairs signal invalidations, MSS/FVG meta compatibility fields, and liquidity safe merges limited to unreferenced supersets plus exact duplicate rows with identical sweep events
+- manual review still owns `needs_sweep_reference_migration` and `keep_version_history` groups
 
 Dry-run scan:
 
@@ -98,6 +99,65 @@ cd /home/yuqei/project/ai-exchange
 .venv/bin/python scripts/run_historical_compatibility_repair.py --apply-safe-repairs
 ```
 
+## Trading Delivery Checklist
+
+Use this locked trading delivery checklist before every market validation or deploy, so the trading module can be promoted without relying on chat memory.
+
+must-run before every market validation or deploy:
+
+```bash
+.venv/bin/python -m unittest discover -s tests/unit -p "test_*.py"
+.venv/bin/python scripts/run_offline_checks.py
+.venv/bin/python scripts/check_local_demo_readiness.py
+.venv/bin/python scripts/check_trading_delivery_readiness.py
+```
+
+Database gate:
+
+```bash
+READINESS_OUTPUT_JSON=1 .venv/bin/python scripts/run_database_readiness.py
+HISTORICAL_REPAIR_OUTPUT_JSON=1 .venv/bin/python scripts/run_historical_compatibility_repair.py
+.venv/bin/python scripts/run_historical_compatibility_repair.py --apply-safe-repairs
+HISTORICAL_REPAIR_OUTPUT_JSON=1 .venv/bin/python scripts/run_historical_compatibility_repair.py
+PIPELINE_OUTPUT_JSON=1 BACKTEST_REQUIRE_QUALITY_GATE=1 .venv/bin/python scripts/run_pipeline.py
+```
+
+Use `--apply-safe-repairs` only when the dry-run output shows safe candidates. After the second dry-run, manual review is still required for `needs_sweep_reference_migration` and `keep_version_history`, and `backtest_quality_gate.passed` must be `true` before promotion.
+
+Deploy and health check:
+
+```bash
+docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml down --remove-orphans
+docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml up -d --build
+docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml ps
+test "$(docker ps --filter label=com.docker.compose.project=ai-ict --format '{{.Names}}' | wc -l | tr -d ' ')" = "1"
+test "$(docker ps --filter name=^/ai-ict$ --format '{{.Names}}' | wc -l | tr -d ' ')" = "1"
+test "$(docker inspect -f '{{.State.Health.Status}}' ai-ict)" = "healthy"
+curl -fsS http://127.0.0.1:8000/health
+curl -fsS "http://127.0.0.1:8000/signals?instrument_id=1&limit=1"
+```
+
+Post-deploy observe:
+
+```bash
+READINESS_OUTPUT_JSON=1 .venv/bin/python scripts/run_database_readiness.py
+REPORTS_OUTPUT_JSON=1 .venv/bin/python scripts/run_reports.py
+```
+
+The paper execution boundary remains in force here: `run_pipeline.py` and the execution path are still paper-only, and any API `5xx`, Docker unhealthy state, or non-empty readiness reasons should stop promotion immediately.
+
+Rollback trigger:
+
+```bash
+git rev-parse HEAD
+docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml logs --tail=100 ai-ict
+git worktree add /tmp/ai-ict-rollback <last-known-good-sha>
+cd /tmp/ai-ict-rollback && ./scripts/deploy_server.exp
+rm -rf /tmp/ai-ict-rollback
+```
+
+Record the current commit as the last-known-good SHA before deploy. If any gate above fails, capture logs first, then redeploy from the clean rollback worktree rather than mutating the active development checkout.
+
 ## Git Remote
 
 This repo is currently configured to use SSH for `origin`:
diff --git a/SESSION.md b/SESSION.md
index f61b8dd..a5c3188 100644
--- a/SESSION.md
+++ b/SESSION.md
@@ -2,7 +2,7 @@
 
 ## Current Goal
 
-Current roadmap items are implemented. The next practical step is to run the new historical compatibility repair path in dry-run mode against a real database and decide whether to apply the safe repairs, or define the next roadmap increment.
+从现在开始只推进交易模块交付，优先把运行清单、部署纪律和市场验证前准备补齐，不再扩展 Training 新能力。
 
 ## Current Branch
 
@@ -14,12 +14,14 @@ Current roadmap items are implemented. The next practical step is to run the new
 
 ## Completed In Last Session
 
-- Added a historical compatibility repair core module in [src/services/maintenance/historical_compatibility.py](/home/yuqei/project/ai-exchange/src/services/maintenance/historical_compatibility.py) covering three safe auto-repair classes: signal invalidation normalization, MSS meta backfill, and FVG meta compatibility backfill.
-- Added [scripts/run_historical_compatibility_repair.py](/home/yuqei/project/ai-exchange/scripts/run_historical_compatibility_repair.py) with default dry-run behavior, explicit safe-apply flags, JSON output support, and manual-review reporting for liquidity duplicate groups.
-- Kept liquidity duplicate handling scan-only by default; the script reports duplicate groups for review instead of auto-merging or mutating them.
-- Added focused unit coverage for repair planning logic, script dry-run/apply behavior, database-unavailable JSON behavior, and runtime script registry compatibility.
-- Documented the new historical compatibility repair script in `README.md` and `RUNBOOK.md`.
-- Revalidated the repository with `.venv/bin/python3 -m unittest discover -s tests/unit -p "test_*.py"`, `.venv/bin/python3 scripts/run_offline_checks.py`, and `.venv/bin/python3 scripts/check_local_demo_readiness.py`.
+- 已经把交易模块交付前 checklist 锁成可执行单一来源：新增 `scripts/check_trading_delivery_readiness.py`，并把 preflight / database gate / deploy / observe / rollback 的固定命令顺序同步到 `README.md` 与 `RUNBOOK.md`。
+- 已经用 focused tests 把这套 checklist 固化：新增 `tests/unit/test_trading_delivery_readiness.py`，扩展 `tests/unit/test_documentation_consistency.py` 和 `scripts/run_offline_checks.py`，避免运行清单再次依赖聊天记忆漂移。
+- 已把历史兼容 safe repair 的文档语义纠正到当前实现：`--apply-safe-repairs` 现在包含 signal invalidations、MSS/FVG meta，以及 liquidity safe merges；`needs_sweep_reference_migration` / `keep_version_history` 仍保留人工决策边界。
+- 已经完成交易主线的主要规则与验证增强，包括：MSS/FVG richer state、execution/backtest/report lifecycle contract、backtest `by_symbol`、FVG candidate priority convergence，以及历史兼容安全修复脚本。
+- 已经在真实数据库上执行了历史兼容安全修复：`signal_invalidations=11`、`mss_meta=4`、`fvg_meta=1189`，再次 dry-run 后这三类候选修复已全部清零。
+- 已把 `liquidity` duplicate groups 程序化分类成四类：`safe_auto_merge_unreferenced_superset`、`safe_exact_duplicate_with_duplicate_sweeps`、`needs_sweep_reference_migration`、`keep_version_history`。
+- 已在真实数据库上执行 liquidity safe merges：共处理 `8` 组，其中 `1` 组未引用超集重复、`7` 组 exact duplicate 且 duplicate sweep 完全同构；再次 dry-run 后仅剩 `5` 组 `keep_version_history`。
+- 训练侧的 quality / selector / generator 原型已完成，但从本轮开始这些不再是当前开发主线。
 
 ## Current Working State
 
@@ -43,9 +45,10 @@ Current roadmap items are implemented. The next practical step is to run the new
 - Training case list/export service/API surfaces now also support stable selector inputs for `quality_rank`, `min_quality_score`, and `sort_by`.
 - The generator prototype now exists as an offline-only flow: trading-derived seed JSON -> public Training draft package JSON, with empty `answer` content and no database writes.
 - Signal-side FVG selection is now explicit and deterministic: exact MSS displacement match wins, otherwise newest same-direction wins, otherwise newest overall wins.
-- Historical compatibility repair now exists as a script-backed path: default dry-run scan, optional safe apply for signal invalidations plus MSS/FVG meta, and manual-review-only reporting for liquidity duplicate groups.
-- `TASKS.md` now reflects the intended development sequence more explicitly:
-  there are no active roadmap items left after the current sequence.
+- Historical compatibility repair now exists as a script-backed path: default dry-run scan, optional safe apply for signal invalidations plus MSS/FVG meta plus liquidity safe merges, and explicit manual-review boundaries for `needs_sweep_reference_migration` / `keep_version_history`.
+- 交易模块交付前 checklist 现在已经被 `README.md` / `RUNBOOK.md` / `scripts/check_trading_delivery_readiness.py` / offline checks 四处对齐，真实部署步骤不再只存在于聊天上下文里。
+- 当前真正阻碍交易模块进入市场验证的，已经不再是历史兼容 safe repair 或运行清单缺失，而是是否要继续进入高风险的 `sweep_event pool_id` 引用迁移执行器，以及何时在目标环境执行首轮 checklist rehearsal。
+- `TASKS.md` 已重新切回交易模块交付优先，不再把 Training 继续当作当前 in-progress 主线。
 - FVG state metadata in the dirty worktree still includes `last_touch_ts`, `touch_count`, `status_reason`, and gap warnings.
 - Replay-backed no-lookahead backtest mode remains present and validation is green.
 - `scripts/run_dev.py` still starts the persistent local API/dashboard dev server using `AI_ICT_API_HOST` / `AI_ICT_API_PORT`.
@@ -54,10 +57,10 @@ Current roadmap items are implemented. The next practical step is to run the new
 
 ## Exact Next Step
 
-If continuing from the current implementation, do this next:
-1. run `.venv/bin/python scripts/run_historical_compatibility_repair.py` against the target database in dry-run mode and inspect candidate repair counts plus liquidity duplicate groups;
-2. if the dry-run output is acceptable, execute `.venv/bin/python scripts/run_historical_compatibility_repair.py --apply-safe-repairs` in a controlled environment;
-3. after the real repair run, decide whether the roadmap should branch into a new migration-backed cleanup phase or shift to a new feature roadmap.
+直接在目标环境做首轮 checklist rehearsal：
+1. 先跑 `.venv/bin/python scripts/check_trading_delivery_readiness.py`，确认本地文档/脚本未漂移；
+2. 按 README/RUNBOOK 中的 Trading Delivery Checklist 顺序执行 database gate、deploy、health-check、post-deploy observe；
+3. 根据 rehearsal 结果决定是进入真实市场验证，还是开启高风险的 `sweep_event pool_id` 引用迁移执行器。
 
 ## Validation Commands
 
@@ -65,14 +68,11 @@ If continuing from the current implementation, do this next:
 .venv/bin/python3 -m unittest discover -s tests/unit -p "test_*.py"
 .venv/bin/python3 scripts/run_offline_checks.py
 .venv/bin/python3 scripts/check_local_demo_readiness.py
+.venv/bin/python3 scripts/check_trading_delivery_readiness.py
 ```
 
 ## Risks Or Open Questions
 
-- The dashboard now surfaces the new structure-state fields directly, but signal detail is still rendered as structured JSON rather than a dedicated custom detail card.
-- Legacy `trade_signal.invalidations` strings are now normalized for API/report/dashboard consumers, but historical rows still physically store mixed old/new string forms in the database until a later migration or cleanup pass.
-- The new FVG selector rule is explicit, but it still reflects one specific interpretation of “recent plus context”; if trading review later wants recency to dominate exact context in some setups, that will require another deliberate rule change rather than a repository ordering tweak.
-- The generator prototype is intentionally offline and read-only; it is not yet wired into repository-backed Training flows or any public generation route.
-- The historical repair path is script-based rather than migration-based. That keeps it safer and reversible, but it also means real data repair still depends on an explicit operator run rather than automatic deploy-time migration.
-- The current worktree is no longer fully clean because this session added both code and documentation changes, and there is unrelated untracked legal/planning content in the repo; decide later whether to commit or separate that unrelated content.
-- Training remains a maintenance-first module for now; upstream trading semantics should stabilize before adding larger Training features.
+- 剩余 `5` 组 `keep_version_history` 类型 liquidity 记录目前被有意保留；如果后面仍要继续清理，就不再是 safe repair，而是更高风险的 sweep 引用迁移工程。
+- 交易模块虽然已经有了稳定的 checklist 与回滚边界，但真正的目标环境 rehearsal 还没执行，部署纪律是否足够仍需要一次实跑验证。
+- 当前本地工作区还有未跟踪的法务内容草稿，不属于交易交付主线，后续应避免把它们混入交易模块发布。
diff --git a/TASKS.md b/TASKS.md
index a985781..7153abe 100644
--- a/TASKS.md
+++ b/TASKS.md
@@ -2,11 +2,17 @@
 
 ## In Progress
 
-- [ ] No active in-progress items beyond the current roadmap.
+- [ ] 交易模块交付优先：目标环境首轮 checklist rehearsal
+Files: `README.md`, `RUNBOOK.md`, `SESSION.md`, `scripts/check_trading_delivery_readiness.py`
+Acceptance: 按锁定后的 Trading Delivery Checklist 在目标环境跑完一次 database gate、deploy、health-check、post-deploy observe，并记录人工复核结论。
+Risks: 如果 rehearsal 没有在真实环境执行，当前 checklist 仍然只是在本地文档和测试层面成立。
 
 ## Next Up
 
-- [ ] No additional next-up items beyond the current roadmap.
+- [ ] 交易模块交付优先：sweep_event 引用迁移与 liquidity 合并执行器
+Files: `src/services/maintenance/`, `scripts/`, `tests/unit/`, `migrations/`
+Acceptance: 在 exact duplicate 且引用可安全迁移的前提下，提供 `sweep_events.pool_id` 重绑与冗余 pool 清理执行器；默认可 dry-run，显式 apply，且有冲突保护。
+Risks: 这项工作会直接修改历史引用链，如果唯一约束冲突处理不完整，会破坏 sweep 历史语义。
 
 ## Backlog
 
@@ -18,7 +24,9 @@
 
 ## Done
 
-- [x] 2026-05-04 完成历史数据修复与兼容迁移：新增历史兼容修复核心模块与 `scripts/run_historical_compatibility_repair.py`，默认 dry-run 扫描旧 `invalidations` 文案、MSS/FVG 粗元数据与 liquidity duplicate group，并支持仅对 signal invalidations 与 MSS/FVG meta 执行 safe repairs；补充 focused tests、README/RUNBOOK 入口说明，并通过单测、offline checks 和 local demo readiness。
+- [x] 2026-05-04 完成交易模块交付前运行/观测/回滚清单固化：新增 `scripts/check_trading_delivery_readiness.py` 作为 checklist 单一来源，把 preflight / database gate / deploy / observe / rollback 命令顺序同步到 `README.md` 与 `RUNBOOK.md`，扩展 `tests/unit/test_trading_delivery_readiness.py`、`tests/unit/test_documentation_consistency.py` 与 offline checks；同时把 `run_historical_compatibility_repair.py --apply-safe-repairs` 文档语义更新为当前真实实现（signal invalidations、MSS/FVG meta、liquidity safe merges），避免交易交付继续依赖聊天记忆。
+- [x] 2026-05-04 完成交易模块交付优先的 liquidity 历史重复组分类与安全收敛路径：脚本已能把历史 pool 分成 `safe_auto_merge_unreferenced_superset`、`safe_exact_duplicate_with_duplicate_sweeps`、`needs_sweep_reference_migration`、`keep_version_history` 四类；并已在真实数据库上自动处理 `8` 组安全合并（`1` 组未引用超集 + `7` 组 exact duplicate 且 duplicate sweep 完全同构），剩余 `5` 组保留为结构版本历史；已通过 focused tests、全量单测、offline checks 和 local demo readiness。
+- [x] 2026-05-04 完成历史数据修复与兼容迁移：新增历史兼容修复核心模块与 `scripts/run_historical_compatibility_repair.py`，默认 dry-run 扫描旧 `invalidations` 文案、MSS/FVG 粗元数据与 liquidity duplicate group，并支持对 signal invalidations、MSS/FVG meta 与 liquidity safe merges 执行受控 safe repairs；补充 focused tests、README/RUNBOOK 入口说明，并通过单测、offline checks 和 local demo readiness。
 - [x] 2026-05-04 完成 FVG 候选优先级规则再收敛：signal 层不再被 repository 的隐式 SQL 偏好控制，而是显式使用“先取最近候选，再优先与当前 MSS displacement 精确匹配，其次同方向，最后回退到最新 FVG”的规则；已通过 focused FVG/signal tests、全量单测、offline checks 和 local demo readiness。
 - [x] 2026-05-04 完成自动 Training case generator 预研：新增只读离线原型 `src/services/training/case_generator.py` 与 `scripts/run_training_case_generator.py`，输入 trading-derived seed package，输出公开可消费的 Training case 草案 package；新增 fixture、focused tests、README/RUNBOOK 入口说明；已通过单测、offline checks 和 local demo readiness。
 - [x] 2026-05-04 完成 Training case quality scorer / selector：在现有公开只读 `quality` 对象上新增稳定的 quality filter/sort 能力，支持 `quality_rank`、`min_quality_score`、`sort_by` 的服务/API 选择器入口，且默认行为不变；已通过单测、offline checks 和 local demo readiness。
diff --git a/scripts/check_trading_delivery_readiness.py b/scripts/check_trading_delivery_readiness.py
new file mode 100644
index 0000000..9814f6d
--- /dev/null
+++ b/scripts/check_trading_delivery_readiness.py
@@ -0,0 +1,197 @@
+from dataclasses import dataclass
+from pathlib import Path
+import re
+
+
+ROOT = Path(__file__).resolve().parents[1]
+TRADING_DELIVERY_DOCS = (
+    ROOT / "README.md",
+    ROOT / "RUNBOOK.md",
+)
+TRADING_DELIVERY_SECTION_HEADING = "Trading Delivery Checklist"
+TRADING_DELIVERY_COMPOSE_PREFIX = (
+    "docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml"
+)
+TRADING_DELIVERY_COMMAND_GROUPS = (
+    (
+        "pre_deploy_offline",
+        (
+            '.venv/bin/python -m unittest discover -s tests/unit -p "test_*.py"',
+            ".venv/bin/python scripts/run_offline_checks.py",
+            ".venv/bin/python scripts/check_local_demo_readiness.py",
+            ".venv/bin/python scripts/check_trading_delivery_readiness.py",
+        ),
+    ),
+    (
+        "database_gate",
+        (
+            "READINESS_OUTPUT_JSON=1 .venv/bin/python scripts/run_database_readiness.py",
+            "HISTORICAL_REPAIR_OUTPUT_JSON=1 .venv/bin/python scripts/run_historical_compatibility_repair.py",
+            ".venv/bin/python scripts/run_historical_compatibility_repair.py --apply-safe-repairs",
+            "HISTORICAL_REPAIR_OUTPUT_JSON=1 .venv/bin/python scripts/run_historical_compatibility_repair.py",
+            "PIPELINE_OUTPUT_JSON=1 BACKTEST_REQUIRE_QUALITY_GATE=1 .venv/bin/python scripts/run_pipeline.py",
+        ),
+    ),
+    (
+        "deploy_and_health",
+        (
+            f"{TRADING_DELIVERY_COMPOSE_PREFIX} down --remove-orphans",
+            f"{TRADING_DELIVERY_COMPOSE_PREFIX} up -d --build",
+            f"{TRADING_DELIVERY_COMPOSE_PREFIX} ps",
+            'test "$(docker ps --filter label=com.docker.compose.project=ai-ict --format \'{{.Names}}\' | wc -l | tr -d \' \')" = "1"',
+            'test "$(docker ps --filter name=^/ai-ict$ --format \'{{.Names}}\' | wc -l | tr -d \' \')" = "1"',
+            'test "$(docker inspect -f \'{{.State.Health.Status}}\' ai-ict)" = "healthy"',
+            "curl -fsS http://127.0.0.1:8000/health",
+            'curl -fsS "http://127.0.0.1:8000/signals?instrument_id=1&limit=1"',
+        ),
+    ),
+    (
+        "post_deploy_observe",
+        (
+            "READINESS_OUTPUT_JSON=1 .venv/bin/python scripts/run_database_readiness.py",
+            "REPORTS_OUTPUT_JSON=1 .venv/bin/python scripts/run_reports.py",
+        ),
+    ),
+    (
+        "rollback_boundary",
+        (
+            "git rev-parse HEAD",
+            f"{TRADING_DELIVERY_COMPOSE_PREFIX} logs --tail=100 ai-ict",
+            "git worktree add /tmp/ai-ict-rollback <last-known-good-sha>",
+            "cd /tmp/ai-ict-rollback && ./scripts/deploy_server.exp",
+            "rm -rf /tmp/ai-ict-rollback",
+        ),
+    ),
+)
+TRADING_DELIVERY_COMMANDS = tuple(
+    command
+    for _group_name, commands in TRADING_DELIVERY_COMMAND_GROUPS
+    for command in commands
+)
+TRADING_DELIVERY_REQUIRED_SNIPPETS = (
+    "must-run before every market validation or deploy",
+    "only when the dry-run output shows safe candidates",
+    "manual review is still required for `needs_sweep_reference_migration` and `keep_version_history`",
+    "paper execution boundary",
+    "Rollback trigger",
+    "last-known-good SHA",
+    "without relying on chat memory",
+)
+TRADING_DELIVERY_REQUIRED_LOCAL_PATHS = (
+    ROOT / "scripts" / "check_local_demo_readiness.py",
+    ROOT / "scripts" / "deploy_server.exp",
+    ROOT / "scripts" / "run_database_readiness.py",
+    ROOT / "scripts" / "run_historical_compatibility_repair.py",
+    ROOT / "scripts" / "run_offline_checks.py",
+    ROOT / "scripts" / "run_pipeline.py",
+    ROOT / "scripts" / "run_reports.py",
+)
+TRADING_DELIVERY_COMMAND_PREFIXES = (
+    ".venv/bin/python",
+    "READINESS_OUTPUT_JSON=1",
+    "HISTORICAL_REPAIR_OUTPUT_JSON=1",
+    "PIPELINE_OUTPUT_JSON=1",
+    "REPORTS_OUTPUT_JSON=1",
+    "docker compose",
+    "test ",
+    "curl ",
+    "git ",
+    "cd /tmp/ai-ict-rollback &&",
+    "rm -rf /tmp/ai-ict-rollback",
+)
+
+
+@dataclass(frozen=True)
+class TradingDeliveryReadinessReport:
+    checked_docs: tuple[Path, ...]
+    command_lines_by_doc: tuple[tuple[Path, tuple[str, ...]], ...]
+    errors: tuple[str, ...]
+
+
+def check_trading_delivery_readiness(
+    docs: tuple[Path, ...] = TRADING_DELIVERY_DOCS,
+) -> TradingDeliveryReadinessReport:
+    command_lines_by_doc = []
+    errors = []
+
+    for path in TRADING_DELIVERY_REQUIRED_LOCAL_PATHS:
+        if not path.is_file():
+            errors.append(f"{_display_path(path)} is missing")
+
+    for path in docs:
+        if not path.is_file():
+            command_lines_by_doc.append((path, ()))
+            errors.append(f"{_display_path(path)} is missing")
+            continue
+
+        text = path.read_text(encoding="utf-8")
+        try:
+            section = _trading_delivery_section(text)
+        except ValueError as exc:
+            command_lines_by_doc.append((path, ()))
+            errors.append(f"{_display_path(path)} {exc}")
+            continue
+
+        command_lines = _trading_delivery_command_lines(section)
+        command_lines_by_doc.append((path, command_lines))
+
+        for snippet in TRADING_DELIVERY_REQUIRED_SNIPPETS:
+            if snippet not in section:
+                errors.append(f"{_display_path(path)} missing trading delivery snippet: {snippet}")
+        if command_lines != TRADING_DELIVERY_COMMANDS:
+            errors.append(
+                f"{_display_path(path)} trading delivery commands drifted from the locked checklist shape"
+            )
+
+    return TradingDeliveryReadinessReport(
+        checked_docs=tuple(docs),
+        command_lines_by_doc=tuple(command_lines_by_doc),
+        errors=tuple(errors),
+    )
+
+
+def main() -> int:
+    report = check_trading_delivery_readiness()
+    if report.errors:
+        print("Trading delivery readiness check failed")
+        for error in report.errors:
+            print(f"- {error}")
+        return 1
+
+    print(
+        "Trading delivery readiness check passed: "
+        f"{len(report.checked_docs)} docs, "
+        f"{len(TRADING_DELIVERY_COMMAND_GROUPS)} phases, "
+        f"{len(TRADING_DELIVERY_COMMANDS)} locked commands."
+    )
+    return 0
+
+
+def _trading_delivery_section(text: str) -> str:
+    match = re.search(
+        rf"^## {re.escape(TRADING_DELIVERY_SECTION_HEADING)}\n(?P<section>.*?)(?:\n## |\Z)",
+        text,
+        re.DOTALL | re.MULTILINE,
+    )
+    if match is None:
+        raise ValueError(f"is missing the {TRADING_DELIVERY_SECTION_HEADING} section")
+    return match.group("section")
+
+
+def _trading_delivery_command_lines(text: str) -> tuple[str, ...]:
+    return tuple(
+        line.strip()
+        for line in text.splitlines()
+        if line.strip().startswith(TRADING_DELIVERY_COMMAND_PREFIXES)
+    )
+
+
+def _display_path(path: Path) -> str:
+    try:
+        return str(path.relative_to(ROOT))
+    except ValueError:
+        return str(path)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/run_historical_compatibility_repair.py b/scripts/run_historical_compatibility_repair.py
index 76b6833..8627157 100644
--- a/scripts/run_historical_compatibility_repair.py
+++ b/scripts/run_historical_compatibility_repair.py
@@ -14,10 +14,10 @@ if str(ROOT) not in sys.path:
 from src.db.engine import engine
 from src.services.maintenance.historical_compatibility import (
     build_historical_compatibility_summary,
+    classify_liquidity_duplicate_groups,
     plan_fvg_meta_repairs,
     plan_mss_meta_repairs,
     plan_trade_signal_invalidation_repairs,
-    summarize_liquidity_duplicate_groups,
 )
 
 
@@ -26,23 +26,25 @@ def run_historical_compatibility_repair(
     apply_signal_invalidations: bool = False,
     apply_mss_meta: bool = False,
     apply_fvg_meta: bool = False,
+    apply_liquidity_safe_merges: bool = False,
 ) -> dict:
-    dry_run = not any((apply_signal_invalidations, apply_mss_meta, apply_fvg_meta))
+    dry_run = not any((apply_signal_invalidations, apply_mss_meta, apply_fvg_meta, apply_liquidity_safe_merges))
     with engine.begin() as connection:
         signal_rows = _fetch_trade_signal_rows(connection)
         mss_rows = _fetch_mss_rows(connection)
         fvg_rows = _fetch_fvg_rows(connection)
-        liquidity_duplicate_rows = _fetch_liquidity_duplicate_groups(connection)
+        liquidity_duplicate_rows = _fetch_liquidity_duplicate_rows(connection)
 
         signal_repairs = plan_trade_signal_invalidation_repairs(signal_rows)
         mss_repairs = plan_mss_meta_repairs(mss_rows)
         fvg_repairs = plan_fvg_meta_repairs(fvg_rows)
-        liquidity_duplicates = summarize_liquidity_duplicate_groups(liquidity_duplicate_rows)
+        liquidity_duplicates = classify_liquidity_duplicate_groups(liquidity_duplicate_rows)
 
         applied_repairs = {
             "signal_invalidations": 0,
             "mss_meta": 0,
             "fvg_meta": 0,
+            "liquidity_safe_merges": 0,
         }
 
         if apply_signal_invalidations:
@@ -57,6 +59,8 @@ def run_historical_compatibility_repair(
             for repair in fvg_repairs:
                 _update_fvg_meta(connection, repair)
             applied_repairs["fvg_meta"] = len(fvg_repairs)
+        if apply_liquidity_safe_merges:
+            applied_repairs["liquidity_safe_merges"] = _apply_liquidity_safe_merges(connection, liquidity_duplicates)
 
     return build_historical_compatibility_summary(
         signal_repairs=signal_repairs,
@@ -102,27 +106,96 @@ def _fetch_fvg_rows(connection) -> list[dict]:
     return [dict(row) for row in connection.execute(sql).mappings().all()]
 
 
-def _fetch_liquidity_duplicate_groups(connection) -> list[dict]:
+def _fetch_liquidity_duplicate_rows(connection) -> list[dict]:
     sql = text(
         """
+        with dupes as (
+            select instrument_id, timeframe, side, price_low, price_high
+            from liquidity_pools
+            where status = 'untouched'
+            group by instrument_id, timeframe, side, price_low, price_high
+            having count(*) > 1
+        )
         select
-            instrument_id,
-            timeframe,
-            side,
-            price_low,
-            price_high,
-            count(*) as duplicate_count,
-            array_agg(id order by created_ts desc, id desc) as ids
-        from liquidity_pools
-        where status = 'untouched'
-        group by instrument_id, timeframe, side, price_low, price_high
-        having count(*) > 1
-        order by count(*) desc, instrument_id asc, timeframe asc, side asc
+            lp.instrument_id,
+            lp.timeframe,
+            lp.side,
+            lp.price_low,
+            lp.price_high,
+            lp.id,
+            lp.pool_type,
+            lp.reference_ids,
+            lp.created_ts,
+            lp.meta,
+            coalesce(
+                json_agg(
+                    json_build_object(
+                        'id', se.id,
+                        'sweep_ts', se.sweep_ts,
+                        'side_swept', se.side_swept,
+                        'sweep_high', se.sweep_high,
+                        'sweep_low', se.sweep_low,
+                        'close_back_inside', se.close_back_inside,
+                        'validation_level', se.validation_level,
+                        'status', se.status,
+                        'meta', se.meta
+                    )
+                    order by se.sweep_ts, se.id
+                ) filter (where se.id is not null),
+                '[]'::json
+            ) as sweep_events,
+            coalesce(array_agg(se.sweep_ts order by se.sweep_ts) filter (where se.id is not null), '{}') as sweep_ts_list,
+            count(se.id) as sweep_count
+        from liquidity_pools lp
+        left join sweep_events se on se.pool_id = lp.id
+        join dupes d
+          on d.instrument_id = lp.instrument_id
+         and d.timeframe = lp.timeframe
+         and d.side = lp.side
+         and d.price_low = lp.price_low
+         and d.price_high = lp.price_high
+        where lp.status = 'untouched'
+        group by
+            lp.instrument_id,
+            lp.timeframe,
+            lp.side,
+            lp.price_low,
+            lp.price_high,
+            lp.id,
+            lp.pool_type,
+            lp.reference_ids,
+            lp.created_ts,
+            lp.meta
+        order by
+            lp.instrument_id asc,
+            lp.timeframe asc,
+            lp.side asc,
+            lp.price_low asc,
+            lp.created_ts desc,
+            lp.id desc
         """
     )
     return [dict(row) for row in connection.execute(sql).mappings().all()]
 
 
+def _apply_liquidity_safe_merges(connection, liquidity_duplicates: dict) -> int:
+    applied = 0
+    for group in liquidity_duplicates.get("groups") or []:
+        classification = group.get("classification")
+        if classification not in {"safe_auto_merge_unreferenced_superset", "safe_exact_duplicate_with_duplicate_sweeps"}:
+            continue
+        remove_pool_ids = list(group.get("remove_pool_ids") or [])
+        if not remove_pool_ids:
+            continue
+        if classification == "safe_exact_duplicate_with_duplicate_sweeps":
+            for pool_id in remove_pool_ids:
+                _delete_sweep_events_for_pool(connection, pool_id)
+        for pool_id in remove_pool_ids:
+            _delete_liquidity_pool(connection, pool_id)
+            applied += 1
+    return applied
+
+
 def _update_trade_signal_invalidations(connection, repair: dict) -> None:
     sql = text(
         """
@@ -174,6 +247,27 @@ def _update_fvg_meta(connection, repair: dict) -> None:
     )
 
 
+def _delete_sweep_events_for_pool(connection, pool_id: int) -> None:
+    sql = text(
+        """
+        delete from sweep_events
+        where pool_id = :pool_id
+        """
+    )
+    connection.execute(sql, {"pool_id": int(pool_id)})
+
+
+def _delete_liquidity_pool(connection, pool_id: int) -> None:
+    sql = text(
+        """
+        delete from liquidity_pools
+        where id = :pool_id
+          and status = 'untouched'
+        """
+    )
+    connection.execute(sql, {"pool_id": int(pool_id)})
+
+
 def main(argv=None, env=None) -> int:
     env = os.environ if env is None else env
     parser = _build_parser()
@@ -182,12 +276,14 @@ def main(argv=None, env=None) -> int:
     apply_signal_invalidations = bool(args.apply_signal_invalidations or apply_safe_repairs)
     apply_mss_meta = bool(args.apply_mss_meta or apply_safe_repairs)
     apply_fvg_meta = bool(args.apply_fvg_meta or apply_safe_repairs)
+    apply_liquidity_safe_merges = bool(args.apply_liquidity_safe_merges or apply_safe_repairs)
 
     try:
         summary = run_historical_compatibility_repair(
             apply_signal_invalidations=apply_signal_invalidations,
             apply_mss_meta=apply_mss_meta,
             apply_fvg_meta=apply_fvg_meta,
+            apply_liquidity_safe_merges=apply_liquidity_safe_merges,
         )
     except SQLAlchemyError as exc:
         summary = {
@@ -212,6 +308,7 @@ def _build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--apply-signal-invalidations", action="store_true", help="Apply only trade_signals invalidation normalization.")
     parser.add_argument("--apply-mss-meta", action="store_true", help="Apply only MSS meta backfill for state/status_reason.")
     parser.add_argument("--apply-fvg-meta", action="store_true", help="Apply only FVG meta backfill for status_reason/default compatibility fields.")
+    parser.add_argument("--apply-liquidity-safe-merges", action="store_true", help="Apply only liquidity safe merges for unreferenced superset rows and exact duplicate rows with identical sweep events.")
     return parser
 
 
@@ -229,6 +326,8 @@ def _print_summary(summary: dict, json_output: bool) -> None:
     print(f"MSS meta repairs: {candidate_repairs.get('mss_meta', 0)}")
     print(f"FVG meta repairs: {candidate_repairs.get('fvg_meta', 0)}")
     print(f"Liquidity duplicate groups: {candidate_repairs.get('liquidity_duplicate_groups', 0)}")
+    print(f"Liquidity safe auto-merge groups: {candidate_repairs.get('liquidity_safe_auto_merge_groups', 0)}")
+    print(f"Liquidity safe duplicate-sweep merge groups: {candidate_repairs.get('liquidity_safe_duplicate_sweep_merge_groups', 0)}")
 
 
 def _get_bool(env, key: str, default: bool) -> bool:
diff --git a/scripts/run_offline_checks.py b/scripts/run_offline_checks.py
index 67970b0..53c206d 100644
--- a/scripts/run_offline_checks.py
+++ b/scripts/run_offline_checks.py
@@ -13,6 +13,7 @@ DEFAULT_CHECKS = [
     ("training_case_package", [sys.executable, "scripts/run_training_case_package.py"]),
     ("migration_static_checks", [sys.executable, "scripts/run_migration_static_checks.py"]),
     ("script_static_checks", [sys.executable, "scripts/run_script_static_checks.py"]),
+    ("trading_delivery_readiness", [sys.executable, "scripts/check_trading_delivery_readiness.py"]),
     ("dashboard_js_syntax", ["node", "--check", "src/web/app.js"]),
 ]
 
diff --git a/src/services/maintenance/__init__.py b/src/services/maintenance/__init__.py
index a808f06..29e079e 100644
--- a/src/services/maintenance/__init__.py
+++ b/src/services/maintenance/__init__.py
@@ -1,5 +1,6 @@
 from src.services.maintenance.historical_compatibility import (
     build_historical_compatibility_summary,
+    classify_liquidity_duplicate_groups,
     derive_historical_fvg_meta,
     derive_historical_mss_meta,
     plan_fvg_meta_repairs,
@@ -10,6 +11,7 @@ from src.services.maintenance.historical_compatibility import (
 
 __all__ = [
     "build_historical_compatibility_summary",
+    "classify_liquidity_duplicate_groups",
     "derive_historical_fvg_meta",
     "derive_historical_mss_meta",
     "plan_fvg_meta_repairs",
diff --git a/src/services/maintenance/historical_compatibility.py b/src/services/maintenance/historical_compatibility.py
index 1abba8e..bb0262a 100644
--- a/src/services/maintenance/historical_compatibility.py
+++ b/src/services/maintenance/historical_compatibility.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 from typing import Optional
 
 from src.services.signals import normalize_signal_invalidations
@@ -107,6 +108,174 @@ def summarize_liquidity_duplicate_groups(rows: list[dict]) -> dict:
     }
 
 
+def classify_liquidity_duplicate_groups(rows: list[dict]) -> dict:
+    grouped_rows: dict[tuple, list[dict]] = {}
+    for row in rows:
+        key = (
+            row.get("instrument_id"),
+            row.get("timeframe"),
+            row.get("side"),
+            str(row.get("price_low")),
+            str(row.get("price_high")),
+        )
+        grouped_rows.setdefault(key, []).append(_normalize_liquidity_duplicate_row(row))
+
+    groups = []
+    class_counts = {
+        "safe_auto_merge_unreferenced_superset": 0,
+        "safe_exact_duplicate_with_duplicate_sweeps": 0,
+        "needs_sweep_reference_migration": 0,
+        "keep_version_history": 0,
+    }
+    row_count = 0
+
+    for key, items in grouped_rows.items():
+        items.sort(key=lambda row: (row["created_ts"], row["id"]))
+        classification = _classify_liquidity_group(items)
+        groups.append(
+            {
+                "instrument_id": key[0],
+                "timeframe": key[1],
+                "side": key[2],
+                "price_low": key[3],
+                "price_high": key[4],
+                "duplicate_count": len(items),
+                **classification,
+            }
+        )
+        class_counts[classification["classification"]] += 1
+        row_count += len(items)
+
+    groups.sort(key=lambda row: (-int(row["duplicate_count"]), row["instrument_id"], row["timeframe"], row["side"], row["price_low"]))
+    return {
+        "group_count": len(groups),
+        "row_count": row_count,
+        "classification_counts": class_counts,
+        "groups": groups,
+    }
+
+
+def _normalize_liquidity_duplicate_row(row: dict) -> dict:
+    return {
+        "id": int(row["id"]),
+        "pool_type": row.get("pool_type"),
+        "reference_ids": list(row.get("reference_ids") or []),
+        "created_ts": str(row.get("created_ts")),
+        "meta": dict(row.get("meta") or {}),
+        "sweep_ts_list": [str(value) for value in (row.get("sweep_ts_list") or [])],
+        "sweep_events": [_normalize_sweep_event_payload(item) for item in (row.get("sweep_events") or [])],
+        "sweep_count": int(row.get("sweep_count") or 0),
+    }
+
+
+def _classify_liquidity_group(items: list[dict]) -> dict:
+    if _is_unreferenced_superset_pair(items):
+        canonical = max(items, key=lambda row: (len(row["reference_ids"]), row["id"]))
+        remove = min(items, key=lambda row: (len(row["reference_ids"]), row["id"]))
+        return {
+            "classification": "safe_auto_merge_unreferenced_superset",
+            "reason": "unreferenced_same_type_superset_chain",
+            "canonical_pool_id": canonical["id"],
+            "remove_pool_ids": [remove["id"]],
+            "rows": items,
+        }
+
+    exact_duplicate_buckets = _exact_duplicate_buckets(items)
+    if exact_duplicate_buckets and len(exact_duplicate_buckets) == 1 and len(exact_duplicate_buckets[0]) == len(items):
+        bucket = exact_duplicate_buckets[0]
+        canonical = max(bucket, key=lambda row: row["id"])
+        remove_ids = sorted(row["id"] for row in bucket if row["id"] != canonical["id"])
+        if _rows_have_identical_sweep_event_signatures(bucket):
+            return {
+                "classification": "safe_exact_duplicate_with_duplicate_sweeps",
+                "reason": "exact_duplicate_pool_rows_with_identical_sweep_events",
+                "canonical_pool_id": canonical["id"],
+                "remove_pool_ids": remove_ids,
+                "rows": items,
+            }
+        return {
+            "classification": "needs_sweep_reference_migration",
+            "reason": "exact_duplicate_pool_rows_with_nonidentical_sweep_events",
+            "canonical_pool_id": canonical["id"],
+            "remove_pool_ids": remove_ids,
+            "rows": items,
+        }
+
+    return {
+        "classification": "keep_version_history",
+        "reason": "mixed_structure_versions_or_reference_collisions",
+        "canonical_pool_id": None,
+        "remove_pool_ids": [],
+        "rows": items,
+    }
+
+
+def _is_unreferenced_superset_pair(items: list[dict]) -> bool:
+    if len(items) != 2:
+        return False
+    left, right = items
+    if left["pool_type"] != right["pool_type"]:
+        return False
+    if left["sweep_count"] != 0 or right["sweep_count"] != 0:
+        return False
+    left_refs = set(left["reference_ids"])
+    right_refs = set(right["reference_ids"])
+    return left_refs < right_refs or right_refs < left_refs
+
+
+def _exact_duplicate_buckets(items: list[dict]) -> list[list[dict]]:
+    buckets: dict[tuple, list[dict]] = {}
+    for row in items:
+        key = (
+            row["pool_type"],
+            tuple(row["reference_ids"]),
+            row["created_ts"],
+            json.dumps(row["meta"], sort_keys=True, default=str),
+        )
+        buckets.setdefault(key, []).append(row)
+    return [bucket for bucket in buckets.values() if len(bucket) > 1]
+
+
+def _normalize_sweep_event_payload(item: dict) -> dict:
+    return {
+        "id": int(item.get("id") or 0),
+        "sweep_ts": str(item.get("sweep_ts")),
+        "side_swept": item.get("side_swept"),
+        "sweep_high": str(item.get("sweep_high")),
+        "sweep_low": str(item.get("sweep_low")),
+        "close_back_inside": bool(item.get("close_back_inside")),
+        "validation_level": item.get("validation_level"),
+        "status": item.get("status"),
+        "meta": dict(item.get("meta") or {}),
+    }
+
+
+def _rows_have_identical_sweep_event_signatures(rows: list[dict]) -> bool:
+    if not rows:
+        return False
+    signatures = [_sweep_event_signatures(row) for row in rows]
+    return all(signature == signatures[0] for signature in signatures[1:])
+
+
+def _sweep_event_signatures(row: dict) -> tuple:
+    events = list(row.get("sweep_events") or [])
+    normalized = []
+    for event in events:
+        normalized.append(
+            (
+                event.get("sweep_ts"),
+                event.get("side_swept"),
+                event.get("sweep_high"),
+                event.get("sweep_low"),
+                event.get("close_back_inside"),
+                event.get("validation_level"),
+                event.get("status"),
+                json.dumps(event.get("meta") or {}, sort_keys=True, default=str),
+            )
+        )
+    return tuple(normalized)
+
+
 def build_historical_compatibility_summary(
     *,
     signal_repairs: list[dict],
@@ -126,11 +295,16 @@ def build_historical_compatibility_summary(
             "fvg_meta": len(fvg_repairs),
             "liquidity_duplicate_groups": int(liquidity_duplicates.get("group_count") or 0),
             "liquidity_duplicate_rows": int(liquidity_duplicates.get("row_count") or 0),
+            "liquidity_safe_auto_merge_groups": int((liquidity_duplicates.get("classification_counts") or {}).get("safe_auto_merge_unreferenced_superset", 0)),
+            "liquidity_safe_duplicate_sweep_merge_groups": int((liquidity_duplicates.get("classification_counts") or {}).get("safe_exact_duplicate_with_duplicate_sweeps", 0)),
+            "liquidity_reference_migration_groups": int((liquidity_duplicates.get("classification_counts") or {}).get("needs_sweep_reference_migration", 0)),
+            "liquidity_keep_history_groups": int((liquidity_duplicates.get("classification_counts") or {}).get("keep_version_history", 0)),
         },
         "applied_repairs": {
             "signal_invalidations": int(applied_repairs.get("signal_invalidations") or 0),
             "mss_meta": int(applied_repairs.get("mss_meta") or 0),
             "fvg_meta": int(applied_repairs.get("fvg_meta") or 0),
+            "liquidity_safe_merges": int(applied_repairs.get("liquidity_safe_merges") or 0),
         },
         "manual_review": {
             "liquidity_duplicates": liquidity_duplicates.get("groups") or [],
@@ -139,5 +313,6 @@ def build_historical_compatibility_summary(
             "signal_invalidations",
             "mss_meta",
             "fvg_meta",
+            "liquidity_safe_merges",
         ],
     }
diff --git a/tests/unit/test_documentation_consistency.py b/tests/unit/test_documentation_consistency.py
index 46eb8aa..421adfd 100644
--- a/tests/unit/test_documentation_consistency.py
+++ b/tests/unit/test_documentation_consistency.py
@@ -7,6 +7,7 @@ from pathlib import Path
 
 ROOT = Path(__file__).resolve().parents[2]
 PROJECT_README = ROOT / "README.md"
+PROJECT_RUNBOOK = ROOT / "RUNBOOK.md"
 TRAINING_PRODUCT_DOC = ROOT / "docs" / "training_case_engine_m1_m6_product_readme.md"
 TRAINING_PACKAGE_SMOKE_DOC = ROOT / "docs" / "examples" / "training_case_package_smoke_m6.md"
 TRAINING_DOCS_CLI_TEST = ROOT / "tests" / "unit" / "test_training_case_docs_cli_m10.py"
@@ -456,6 +457,98 @@ LOCAL_DOCKER_POLICY_FORBIDDEN_ENCOURAGEMENT = (
     "start an extra ai-ict container",
     "duplicate ai-ict containers",
 )
+TRADING_DELIVERY_DOCS = (PROJECT_README, PROJECT_RUNBOOK)
+TRADING_DELIVERY_SECTION_HEADING = "Trading Delivery Checklist"
+TRADING_DELIVERY_DOCUMENTATION_GUARD_TEST_INVENTORY = (
+    "test_docs_describe_trading_delivery_checklist",
+    "test_docs_lock_trading_delivery_command_shape",
+)
+TRADING_DELIVERY_DOCUMENTATION_GUARD_SOURCE_CONSTANTS = (
+    "TRADING_DELIVERY_DOCS",
+    "TRADING_DELIVERY_COMMAND_GROUPS",
+    "TRADING_DELIVERY_COMMANDS",
+)
+TRADING_DELIVERY_COMPOSE_PREFIX = (
+    "docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml"
+)
+TRADING_DELIVERY_COMMAND_GROUPS = (
+    (
+        "pre_deploy_offline",
+        (
+            '.venv/bin/python -m unittest discover -s tests/unit -p "test_*.py"',
+            ".venv/bin/python scripts/run_offline_checks.py",
+            ".venv/bin/python scripts/check_local_demo_readiness.py",
+            ".venv/bin/python scripts/check_trading_delivery_readiness.py",
+        ),
+    ),
+    (
+        "database_gate",
+        (
+            "READINESS_OUTPUT_JSON=1 .venv/bin/python scripts/run_database_readiness.py",
+            "HISTORICAL_REPAIR_OUTPUT_JSON=1 .venv/bin/python scripts/run_historical_compatibility_repair.py",
+            ".venv/bin/python scripts/run_historical_compatibility_repair.py --apply-safe-repairs",
+            "HISTORICAL_REPAIR_OUTPUT_JSON=1 .venv/bin/python scripts/run_historical_compatibility_repair.py",
+            "PIPELINE_OUTPUT_JSON=1 BACKTEST_REQUIRE_QUALITY_GATE=1 .venv/bin/python scripts/run_pipeline.py",
+        ),
+    ),
+    (
+        "deploy_and_health",
+        (
+            f"{TRADING_DELIVERY_COMPOSE_PREFIX} down --remove-orphans",
+            f"{TRADING_DELIVERY_COMPOSE_PREFIX} up -d --build",
+            f"{TRADING_DELIVERY_COMPOSE_PREFIX} ps",
+            'test "$(docker ps --filter label=com.docker.compose.project=ai-ict --format \'{{.Names}}\' | wc -l | tr -d \' \')" = "1"',
+            'test "$(docker ps --filter name=^/ai-ict$ --format \'{{.Names}}\' | wc -l | tr -d \' \')" = "1"',
+            'test "$(docker inspect -f \'{{.State.Health.Status}}\' ai-ict)" = "healthy"',
+            "curl -fsS http://127.0.0.1:8000/health",
+            'curl -fsS "http://127.0.0.1:8000/signals?instrument_id=1&limit=1"',
+        ),
+    ),
+    (
+        "post_deploy_observe",
+        (
+            "READINESS_OUTPUT_JSON=1 .venv/bin/python scripts/run_database_readiness.py",
+            "REPORTS_OUTPUT_JSON=1 .venv/bin/python scripts/run_reports.py",
+        ),
+    ),
+    (
+        "rollback_boundary",
+        (
+            "git rev-parse HEAD",
+            f"{TRADING_DELIVERY_COMPOSE_PREFIX} logs --tail=100 ai-ict",
+            "git worktree add /tmp/ai-ict-rollback <last-known-good-sha>",
+            "cd /tmp/ai-ict-rollback && ./scripts/deploy_server.exp",
+            "rm -rf /tmp/ai-ict-rollback",
+        ),
+    ),
+)
+TRADING_DELIVERY_COMMANDS = tuple(
+    command
+    for _group_name, commands in TRADING_DELIVERY_COMMAND_GROUPS
+    for command in commands
+)
+TRADING_DELIVERY_REQUIRED_SNIPPETS = (
+    "must-run before every market validation or deploy",
+    "only when the dry-run output shows safe candidates",
+    "manual review is still required for `needs_sweep_reference_migration` and `keep_version_history`",
+    "paper execution boundary",
+    "Rollback trigger",
+    "last-known-good SHA",
+    "without relying on chat memory",
+)
+TRADING_DELIVERY_COMMAND_PREFIXES = (
+    ".venv/bin/python",
+    "READINESS_OUTPUT_JSON=1",
+    "HISTORICAL_REPAIR_OUTPUT_JSON=1",
+    "PIPELINE_OUTPUT_JSON=1",
+    "REPORTS_OUTPUT_JSON=1",
+    "docker compose",
+    "test ",
+    "curl ",
+    "git ",
+    "cd /tmp/ai-ict-rollback &&",
+    "rm -rf /tmp/ai-ict-rollback",
+)
 TRAINING_PUBLIC_PRIVACY_EXCLUSION_WORDS = (
     "scrub",
     "omit",
@@ -608,6 +701,26 @@ def _local_docker_command_lines(text: str) -> tuple[str, ...]:
     )
 
 
+def _trading_delivery_section(path: Path) -> str:
+    text = path.read_text(encoding="utf-8")
+    match = re.search(
+        rf"^## {re.escape(TRADING_DELIVERY_SECTION_HEADING)}\n(?P<section>.*?)(?:\n## |\Z)",
+        text,
+        re.DOTALL | re.MULTILINE,
+    )
+    if match is None:
+        raise AssertionError(f"{path} is missing the {TRADING_DELIVERY_SECTION_HEADING} section")
+    return match.group("section")
+
+
+def _trading_delivery_command_lines(text: str) -> tuple[str, ...]:
+    return tuple(
+        line.strip()
+        for line in text.splitlines()
+        if line.strip().startswith(TRADING_DELIVERY_COMMAND_PREFIXES)
+    )
+
+
 def _training_public_privacy_lines(text: str) -> tuple[str, ...]:
     lines = []
     for line in text.splitlines():
@@ -4395,6 +4508,53 @@ TRAINING_M41_RELEASE_INDEX_REQUIRED_SNIPPET_INVENTORY = (
                 self.assertIn(constant_name, top_level_names)
                 self.assertIn(f"{constant_name} =", module_source)
 
+    def test_trading_delivery_documentation_guard_inventory_is_current(self) -> None:
+        module_path = Path(__file__)
+        module_source = module_path.read_text(encoding="utf-8")
+        module_tree = ast.parse(module_source)
+        class_nodes = [
+            node
+            for node in module_tree.body
+            if isinstance(node, ast.ClassDef) and node.name == "DocumentationConsistencyTests"
+        ]
+        top_level_names = {
+            target.id
+            for node in module_tree.body
+            if isinstance(node, ast.Assign)
+            for target in node.targets
+            if isinstance(target, ast.Name)
+        }
+        required_guard_tests = {
+            "test_docs_describe_trading_delivery_checklist",
+            "test_docs_lock_trading_delivery_command_shape",
+        }
+        inventory_names = tuple(TRADING_DELIVERY_DOCUMENTATION_GUARD_TEST_INVENTORY)
+
+        self.assertEqual(len(class_nodes), 1)
+        self.assertEqual(len(inventory_names), len(set(inventory_names)))
+        self.assertLessEqual(required_guard_tests, set(inventory_names))
+
+        test_names = {
+            item.name
+            for item in class_nodes[0].body
+            if isinstance(item, ast.FunctionDef) and item.name.startswith("test_")
+        }
+        for name in inventory_names:
+            with self.subTest(name=name):
+                loader = unittest.TestLoader()
+                suite = loader.loadTestsFromName(
+                    f"{type(self).__module__}.DocumentationConsistencyTests.{name}"
+                )
+
+                self.assertIn(name, test_names)
+                self.assertEqual(loader.errors, [])
+                self.assertGreater(suite.countTestCases(), 0)
+
+        for constant_name in TRADING_DELIVERY_DOCUMENTATION_GUARD_SOURCE_CONSTANTS:
+            with self.subTest(constant_name=constant_name):
+                self.assertIn(constant_name, top_level_names)
+                self.assertIn(f"{constant_name} =", module_source)
+
     def test_training_case_acceptance_manifest_modules_are_discoverable(self) -> None:
         manifest_modules = tuple(TRAINING_CASE_ACCEPTANCE_TEST_MANIFEST)
         missing_modules = [
@@ -4501,6 +4661,33 @@ TRAINING_M41_RELEASE_INDEX_REQUIRED_SNIPPET_INVENTORY = (
                 self.assertNotIn("docker compose -p ai-ict down --remove-orphans", command_lines)
                 self.assertNotIn("docker compose -p ai-ict up -d --build", command_lines)
 
+    def test_docs_describe_trading_delivery_checklist(self) -> None:
+        for path in TRADING_DELIVERY_DOCS:
+            section = _trading_delivery_section(path)
+
+            with self.subTest(path=path.name):
+                for snippet in TRADING_DELIVERY_REQUIRED_SNIPPETS:
+                    self.assertIn(snippet, section)
+                self.assertLess(
+                    section.index("must-run before every market validation or deploy"),
+                    section.index("Rollback trigger"),
+                )
+                self.assertIn("backtest_quality_gate.passed", section)
+
+    def test_docs_lock_trading_delivery_command_shape(self) -> None:
+        command_lines_by_path = {
+            path: _trading_delivery_command_lines(_trading_delivery_section(path))
+            for path in TRADING_DELIVERY_DOCS
+        }
+
+        self.assertEqual(
+            command_lines_by_path[PROJECT_README],
+            command_lines_by_path[PROJECT_RUNBOOK],
+        )
+        for path, command_lines in command_lines_by_path.items():
+            with self.subTest(path=path.name):
+                self.assertEqual(command_lines, TRADING_DELIVERY_COMMANDS)
+
     def test_training_case_product_doc_names_public_response_privacy_contracts(self) -> None:
         text = TRAINING_PRODUCT_DOC.read_text(encoding="utf-8")
 
diff --git a/tests/unit/test_historical_compatibility_repair_script.py b/tests/unit/test_historical_compatibility_repair_script.py
index 1f113df..82c7f0f 100644
--- a/tests/unit/test_historical_compatibility_repair_script.py
+++ b/tests/unit/test_historical_compatibility_repair_script.py
@@ -35,7 +35,7 @@ class _ConnectionStub:
             return _MappingsResult([{"id": 2, "accepted": True, "meta": {}}])
         if "from fvg_zones" in sql_text and "select" in sql_text:
             return _MappingsResult([{"id": 3, "status": "open", "meta": {}}])
-        if "from liquidity_pools" in sql_text:
+        if "select" in sql_text and "from liquidity_pools" in sql_text:
             return _MappingsResult(
                 [
                     {
@@ -44,8 +44,29 @@ class _ConnectionStub:
                         "side": "buy_side",
                         "price_low": 100.0,
                         "price_high": 100.0,
-                        "duplicate_count": 2,
-                        "ids": [10, 9],
+                        "id": 9,
+                        "pool_type": "equal_highs",
+                        "reference_ids": [1, 2],
+                        "created_ts": "2026-04-29T10:36:00+00:00",
+                        "meta": {"merged_reference_count": 2},
+                        "sweep_events": [],
+                        "sweep_ts_list": [],
+                        "sweep_count": 0,
+                    },
+                    {
+                        "instrument_id": 1,
+                        "timeframe": "1m",
+                        "side": "buy_side",
+                        "price_low": 100.0,
+                        "price_high": 100.0,
+                        "id": 10,
+                        "pool_type": "equal_highs",
+                        "reference_ids": [1, 2, 3],
+                        "created_ts": "2026-04-29T10:36:00+00:00",
+                        "meta": {"merged_reference_count": 3},
+                        "sweep_events": [],
+                        "sweep_ts_list": [],
+                        "sweep_count": 0,
                     }
                 ]
             )
@@ -77,6 +98,8 @@ class HistoricalCompatibilityRepairScriptTests(unittest.TestCase):
         self.assertEqual(summary["candidate_repairs"]["mss_meta"], 1)
         self.assertEqual(summary["candidate_repairs"]["fvg_meta"], 1)
         self.assertEqual(summary["candidate_repairs"]["liquidity_duplicate_groups"], 1)
+        self.assertEqual(summary["candidate_repairs"]["liquidity_safe_auto_merge_groups"], 1)
+        self.assertEqual(summary["candidate_repairs"]["liquidity_safe_duplicate_sweep_merge_groups"], 0)
         self.assertEqual(connection.updates, [])
 
     def test_apply_safe_repairs_updates_supported_categories(self) -> None:
@@ -87,13 +110,28 @@ class HistoricalCompatibilityRepairScriptTests(unittest.TestCase):
                 apply_signal_invalidations=True,
                 apply_mss_meta=True,
                 apply_fvg_meta=True,
+                apply_liquidity_safe_merges=True,
             )
 
         self.assertFalse(summary["dry_run"])
         self.assertEqual(summary["applied_repairs"]["signal_invalidations"], 1)
         self.assertEqual(summary["applied_repairs"]["mss_meta"], 1)
         self.assertEqual(summary["applied_repairs"]["fvg_meta"], 1)
-        self.assertEqual(len(connection.updates), 3)
+        self.assertEqual(summary["applied_repairs"]["liquidity_safe_merges"], 1)
+        self.assertEqual(len(connection.updates), 4)
+
+    def test_apply_liquidity_safe_merges_can_run_independently(self) -> None:
+        connection = _ConnectionStub()
+
+        with patch("scripts.run_historical_compatibility_repair.engine.begin", return_value=_BeginStub(connection)):
+            summary = run_historical_compatibility_repair.run_historical_compatibility_repair(
+                apply_liquidity_safe_merges=True,
+            )
+
+        self.assertFalse(summary["dry_run"])
+        self.assertEqual(summary["applied_repairs"]["signal_invalidations"], 0)
+        self.assertEqual(summary["applied_repairs"]["liquidity_safe_merges"], 1)
+        self.assertEqual(len(connection.updates), 1)
 
     def test_main_reports_database_unavailable_json(self) -> None:
         output = io.StringIO()
diff --git a/tests/unit/test_historical_compatibility_service.py b/tests/unit/test_historical_compatibility_service.py
index 87e6007..2b0f065 100644
--- a/tests/unit/test_historical_compatibility_service.py
+++ b/tests/unit/test_historical_compatibility_service.py
@@ -5,6 +5,7 @@ os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:
 
 from src.services.maintenance import (
     build_historical_compatibility_summary,
+    classify_liquidity_duplicate_groups,
     derive_historical_fvg_meta,
     derive_historical_mss_meta,
     plan_fvg_meta_repairs,
@@ -63,12 +64,223 @@ class HistoricalCompatibilityServiceTests(unittest.TestCase):
         self.assertEqual(summary["row_count"], 2)
         self.assertEqual(summary["groups"][0]["ids"], [10, 9])
 
+    def test_classifies_unreferenced_superset_pair_as_safe_auto_merge(self) -> None:
+        classified = classify_liquidity_duplicate_groups(
+            [
+                {
+                    "instrument_id": 1,
+                    "timeframe": "1m",
+                    "side": "buy_side",
+                    "price_low": 100.0,
+                    "price_high": 100.0,
+                    "id": 9,
+                    "pool_type": "equal_highs",
+                    "reference_ids": [1, 2],
+                    "created_ts": "2026-04-29T10:36:00+00:00",
+                    "meta": {"merged_reference_count": 2},
+                    "sweep_ts_list": [],
+                    "sweep_count": 0,
+                },
+                {
+                    "instrument_id": 1,
+                    "timeframe": "1m",
+                    "side": "buy_side",
+                    "price_low": 100.0,
+                    "price_high": 100.0,
+                    "id": 10,
+                    "pool_type": "equal_highs",
+                    "reference_ids": [1, 2, 3],
+                    "created_ts": "2026-04-29T10:36:00+00:00",
+                    "meta": {"merged_reference_count": 3},
+                    "sweep_ts_list": [],
+                    "sweep_count": 0,
+                },
+            ]
+        )
+
+        self.assertEqual(classified["classification_counts"]["safe_auto_merge_unreferenced_superset"], 1)
+        self.assertEqual(classified["groups"][0]["classification"], "safe_auto_merge_unreferenced_superset")
+        self.assertEqual(classified["groups"][0]["canonical_pool_id"], 10)
+        self.assertEqual(classified["groups"][0]["remove_pool_ids"], [9])
+
+    def test_classifies_exact_duplicate_referenced_pair_as_reference_migration(self) -> None:
+        classified = classify_liquidity_duplicate_groups(
+            [
+                {
+                    "instrument_id": 1,
+                    "timeframe": "1m",
+                    "side": "sell_side",
+                    "price_low": 100.0,
+                    "price_high": 100.0,
+                    "id": 11,
+                    "pool_type": "prior_low",
+                    "reference_ids": [7],
+                    "created_ts": "2026-04-23T05:24:00+00:00",
+                    "meta": {},
+                    "sweep_events": [
+                        {
+                            "id": 100,
+                            "sweep_ts": "2026-04-23T05:24:00+00:00",
+                            "side_swept": "sell_side",
+                            "sweep_high": "101.0",
+                            "sweep_low": "99.0",
+                            "close_back_inside": False,
+                            "validation_level": "weak",
+                            "status": "active",
+                            "meta": {"pool_type": "prior_low"},
+                        }
+                    ],
+                    "sweep_ts_list": ["2026-04-23T05:24:00+00:00"],
+                    "sweep_count": 1,
+                },
+                {
+                    "instrument_id": 1,
+                    "timeframe": "1m",
+                    "side": "sell_side",
+                    "price_low": 100.0,
+                    "price_high": 100.0,
+                    "id": 12,
+                    "pool_type": "prior_low",
+                    "reference_ids": [7],
+                    "created_ts": "2026-04-23T05:24:00+00:00",
+                    "meta": {},
+                    "sweep_events": [
+                        {
+                            "id": 101,
+                            "sweep_ts": "2026-04-23T05:24:00+00:00",
+                            "side_swept": "sell_side",
+                            "sweep_high": "101.0",
+                            "sweep_low": "99.0",
+                            "close_back_inside": False,
+                            "validation_level": "weak",
+                            "status": "active",
+                            "meta": {"pool_type": "prior_low"},
+                        }
+                    ],
+                    "sweep_ts_list": ["2026-04-23T05:24:00+00:00"],
+                    "sweep_count": 1,
+                },
+            ]
+        )
+
+        self.assertEqual(classified["classification_counts"]["safe_exact_duplicate_with_duplicate_sweeps"], 1)
+        self.assertEqual(classified["groups"][0]["classification"], "safe_exact_duplicate_with_duplicate_sweeps")
+
+    def test_classifies_nonidentical_referenced_pair_as_reference_migration(self) -> None:
+        classified = classify_liquidity_duplicate_groups(
+            [
+                {
+                    "instrument_id": 1,
+                    "timeframe": "1m",
+                    "side": "sell_side",
+                    "price_low": 100.0,
+                    "price_high": 100.0,
+                    "id": 21,
+                    "pool_type": "prior_low",
+                    "reference_ids": [9],
+                    "created_ts": "2026-04-23T05:24:00+00:00",
+                    "meta": {},
+                    "sweep_events": [
+                        {
+                            "id": 200,
+                            "sweep_ts": "2026-04-23T05:24:00+00:00",
+                            "side_swept": "sell_side",
+                            "sweep_high": "101.0",
+                            "sweep_low": "99.0",
+                            "close_back_inside": False,
+                            "validation_level": "weak",
+                            "status": "active",
+                            "meta": {"pool_type": "prior_low"},
+                        }
+                    ],
+                    "sweep_ts_list": ["2026-04-23T05:24:00+00:00"],
+                    "sweep_count": 1,
+                },
+                {
+                    "instrument_id": 1,
+                    "timeframe": "1m",
+                    "side": "sell_side",
+                    "price_low": 100.0,
+                    "price_high": 100.0,
+                    "id": 22,
+                    "pool_type": "prior_low",
+                    "reference_ids": [9],
+                    "created_ts": "2026-04-23T05:24:00+00:00",
+                    "meta": {},
+                    "sweep_events": [
+                        {
+                            "id": 201,
+                            "sweep_ts": "2026-04-23T05:24:00+00:00",
+                            "side_swept": "sell_side",
+                            "sweep_high": "102.0",
+                            "sweep_low": "98.0",
+                            "close_back_inside": False,
+                            "validation_level": "weak",
+                            "status": "active",
+                            "meta": {"pool_type": "prior_low"},
+                        }
+                    ],
+                    "sweep_ts_list": ["2026-04-23T05:24:00+00:00"],
+                    "sweep_count": 1,
+                },
+            ]
+        )
+
+        self.assertEqual(classified["classification_counts"]["needs_sweep_reference_migration"], 1)
+        self.assertEqual(classified["groups"][0]["classification"], "needs_sweep_reference_migration")
+
+    def test_classifies_mixed_version_group_as_keep_history(self) -> None:
+        classified = classify_liquidity_duplicate_groups(
+            [
+                {
+                    "instrument_id": 1,
+                    "timeframe": "1m",
+                    "side": "buy_side",
+                    "price_low": 100.0,
+                    "price_high": 100.0,
+                    "id": 13,
+                    "pool_type": "prior_high",
+                    "reference_ids": [1],
+                    "created_ts": "2026-04-23T05:24:00+00:00",
+                    "meta": {},
+                    "sweep_ts_list": ["2026-04-23T05:24:00+00:00"],
+                    "sweep_count": 1,
+                },
+                {
+                    "instrument_id": 1,
+                    "timeframe": "1m",
+                    "side": "buy_side",
+                    "price_low": 100.0,
+                    "price_high": 100.0,
+                    "id": 14,
+                    "pool_type": "equal_highs",
+                    "reference_ids": [1, 2],
+                    "created_ts": "2026-04-23T05:24:00+00:00",
+                    "meta": {"merged_reference_count": 2},
+                    "sweep_ts_list": ["2026-04-23T05:24:00+00:00"],
+                    "sweep_count": 1,
+                },
+            ]
+        )
+
+        self.assertEqual(classified["classification_counts"]["keep_version_history"], 1)
+        self.assertEqual(classified["groups"][0]["classification"], "keep_version_history")
+
     def test_builds_historical_compatibility_summary(self) -> None:
         summary = build_historical_compatibility_summary(
             signal_repairs=[{"id": 1}],
             mss_repairs=[{"id": 2}],
             fvg_repairs=[],
-            liquidity_duplicates={"group_count": 1, "row_count": 3, "groups": [{"ids": [9, 8, 7]}]},
+            liquidity_duplicates={
+                "group_count": 1,
+                "row_count": 3,
+                "classification_counts": {
+                    "safe_auto_merge_unreferenced_superset": 1,
+                    "needs_sweep_reference_migration": 0,
+                    "keep_version_history": 0,
+                },
+                "groups": [{"ids": [9, 8, 7]}],
+            },
             applied_repairs={"signal_invalidations": 1, "mss_meta": 1, "fvg_meta": 0},
             dry_run=False,
         )
@@ -76,6 +288,7 @@ class HistoricalCompatibilityServiceTests(unittest.TestCase):
         self.assertTrue(summary["passed"])
         self.assertFalse(summary["dry_run"])
         self.assertEqual(summary["candidate_repairs"]["liquidity_duplicate_rows"], 3)
+        self.assertEqual(summary["candidate_repairs"]["liquidity_safe_auto_merge_groups"], 1)
         self.assertEqual(summary["applied_repairs"]["signal_invalidations"], 1)
         self.assertEqual(summary["manual_review"]["liquidity_duplicates"][0]["ids"], [9, 8, 7])
 
diff --git a/tests/unit/test_trading_delivery_readiness.py b/tests/unit/test_trading_delivery_readiness.py
new file mode 100644
index 0000000..25f3032
--- /dev/null
+++ b/tests/unit/test_trading_delivery_readiness.py
@@ -0,0 +1,134 @@
+import ast
+import subprocess
+import sys
+import unittest
+from pathlib import Path
+
+from scripts import check_trading_delivery_readiness as readiness
+from scripts import run_offline_checks
+
+
+ROOT = Path(__file__).resolve().parents[2]
+CHECK_COMMAND = ".venv/bin/python scripts/check_trading_delivery_readiness.py"
+EXPECTED_DOC_PATHS = (
+    ROOT / "README.md",
+    ROOT / "RUNBOOK.md",
+)
+EXPECTED_COMMAND_GROUPS = (
+    (
+        "pre_deploy_offline",
+        (
+            '.venv/bin/python -m unittest discover -s tests/unit -p "test_*.py"',
+            ".venv/bin/python scripts/run_offline_checks.py",
+            ".venv/bin/python scripts/check_local_demo_readiness.py",
+            ".venv/bin/python scripts/check_trading_delivery_readiness.py",
+        ),
+    ),
+    (
+        "database_gate",
+        (
+            "READINESS_OUTPUT_JSON=1 .venv/bin/python scripts/run_database_readiness.py",
+            "HISTORICAL_REPAIR_OUTPUT_JSON=1 .venv/bin/python scripts/run_historical_compatibility_repair.py",
+            ".venv/bin/python scripts/run_historical_compatibility_repair.py --apply-safe-repairs",
+            "HISTORICAL_REPAIR_OUTPUT_JSON=1 .venv/bin/python scripts/run_historical_compatibility_repair.py",
+            "PIPELINE_OUTPUT_JSON=1 BACKTEST_REQUIRE_QUALITY_GATE=1 .venv/bin/python scripts/run_pipeline.py",
+        ),
+    ),
+    (
+        "deploy_and_health",
+        (
+            "docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml down --remove-orphans",
+            "docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml up -d --build",
+            "docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml ps",
+            'test "$(docker ps --filter label=com.docker.compose.project=ai-ict --format \'{{.Names}}\' | wc -l | tr -d \' \')" = "1"',
+            'test "$(docker ps --filter name=^/ai-ict$ --format \'{{.Names}}\' | wc -l | tr -d \' \')" = "1"',
+            'test "$(docker inspect -f \'{{.State.Health.Status}}\' ai-ict)" = "healthy"',
+            "curl -fsS http://127.0.0.1:8000/health",
+            'curl -fsS "http://127.0.0.1:8000/signals?instrument_id=1&limit=1"',
+        ),
+    ),
+    (
+        "post_deploy_observe",
+        (
+            "READINESS_OUTPUT_JSON=1 .venv/bin/python scripts/run_database_readiness.py",
+            "REPORTS_OUTPUT_JSON=1 .venv/bin/python scripts/run_reports.py",
+        ),
+    ),
+    (
+        "rollback_boundary",
+        (
+            "git rev-parse HEAD",
+            "docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml logs --tail=100 ai-ict",
+            "git worktree add /tmp/ai-ict-rollback <last-known-good-sha>",
+            "cd /tmp/ai-ict-rollback && ./scripts/deploy_server.exp",
+            "rm -rf /tmp/ai-ict-rollback",
+        ),
+    ),
+)
+EXPECTED_COMMANDS = tuple(
+    command
+    for _group_name, commands in EXPECTED_COMMAND_GROUPS
+    for command in commands
+)
+
+
+class TradingDeliveryReadinessTests(unittest.TestCase):
+    def test_checker_locks_trading_delivery_constants(self) -> None:
+        self.assertEqual(readiness.TRADING_DELIVERY_DOCS, EXPECTED_DOC_PATHS)
+        self.assertEqual(readiness.TRADING_DELIVERY_COMMAND_GROUPS, EXPECTED_COMMAND_GROUPS)
+        self.assertEqual(readiness.TRADING_DELIVERY_COMMANDS, EXPECTED_COMMANDS)
+
+    def test_checker_verifies_documented_trading_delivery_commands(self) -> None:
+        report = readiness.check_trading_delivery_readiness()
+
+        self.assertEqual(report.errors, ())
+        self.assertEqual(report.checked_docs, EXPECTED_DOC_PATHS)
+        for path, command_lines in report.command_lines_by_doc:
+            with self.subTest(path=path.name):
+                self.assertEqual(command_lines, EXPECTED_COMMANDS)
+
+    def test_checker_command_is_documented_and_offline(self) -> None:
+        for path in EXPECTED_DOC_PATHS:
+            text = path.read_text(encoding="utf-8")
+            with self.subTest(path=path.name):
+                self.assertIn(CHECK_COMMAND, text)
+                self.assertIn("without relying on chat memory", text)
+
+        process = subprocess.run(
+            [sys.executable, "scripts/check_trading_delivery_readiness.py"],
+            cwd=ROOT,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+
+        self.assertEqual(process.returncode, 0, process.stderr)
+        self.assertIn("Trading delivery readiness check passed", process.stdout)
+
+    def test_checker_source_has_no_process_launches(self) -> None:
+        source = Path(readiness.__file__).read_text(encoding="utf-8")
+        tree = ast.parse(source)
+        imported_modules = {
+            alias.name
+            for node in ast.walk(tree)
+            if isinstance(node, ast.Import)
+            for alias in node.names
+        }
+        called_attributes = {
+            node.attr for node in ast.walk(tree) if isinstance(node, ast.Attribute)
+        }
+
+        self.assertNotIn("subprocess", imported_modules)
+        self.assertNotIn("system", called_attributes)
+
+    def test_offline_checks_include_trading_delivery_readiness(self) -> None:
+        commands = dict(run_offline_checks.build_check_commands(env={}))
+
+        self.assertEqual(
+            commands["trading_delivery_readiness"],
+            [sys.executable, "scripts/check_trading_delivery_readiness.py"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/unit/test_training_case_docs_cli_m10.py b/tests/unit/test_training_case_docs_cli_m10.py
index cf9d026..f34299a 100644
--- a/tests/unit/test_training_case_docs_cli_m10.py
+++ b/tests/unit/test_training_case_docs_cli_m10.py
@@ -36,6 +36,7 @@ OFFLINE_CHECK_REGISTRY_NAMES = (
     OFFLINE_CHECK_NAME,
     "migration_static_checks",
     "script_static_checks",
+    "trading_delivery_readiness",
     DASHBOARD_SYNTAX_CHECK_NAME,
 )
 CURRENT_TRAINING_DOC_MATURITY_END = 28