Complete broker read model sync groundwork

.env.example

@@ -7,6 +7,7 @@ OKX_ORDER_BASE_URL=https://www.okx.com
 OKX_API_KEY=
 OKX_API_SECRET=
 OKX_API_PASSPHRASE=
+SIGNAL_EVIDENCE_MODE=strict
 MARKET_DATA_STREAM_ENABLED=1
 MARKET_DATA_STREAM_SYMBOLS=BTC-USDT,ETH-USDT
 MARKET_DATA_STREAM_TIMEFRAMES=1m

SESSION.md

@@ -14,23 +14,7 @@
 
 ## Completed In Last Session
 
-- 已完成 `P0-1`：实时市场真值层
-  - `MarketDataStreamService`
-  - `MarketTruthSupervisor`
-  - `stream_state / data_gap_state`
-  - news provider
-  - Risk Governor 对 stale / disconnect / gap 的降级联动
-- 已补齐市场真值代理支持
-  - OKX REST
-  - OKX WebSocket
-  - news provider
-- Docker 服务器现已改为直连本机 v2rayN 局域网代理：
-  - `MARKET_TRUTH_PROXY_URL=http://192.168.3.73:10808`
-- 临时 relay 已移除
-- 当前远端验证已通过：
-  - `market_data_fresh`
-  - `market_stream_connected`
-  - `market_data_gap_clear`
+- 本轮尚无新增完成阶段
 
 ## Current Working State
 
@@ -62,6 +46,7 @@
 - 扩展 `RiskContext`
 - 建立 reconciliation 读模型
 - 前端展示真实 `open position / pending order`
+- focused smoke: `/trading/broker-state` 与 `/trading/market-state` 返回 broker state
 
 ## Validation Commands
 
@@ -75,3 +60,4 @@ npm run visual:snapshots
 
 - 当前市场真值链路依赖本机 `192.168.3.73:10808` 持续可达；如本机关机或代理退出，远端将退化
 - 真实账户只读同步下一阶段可能需要额外凭据或账号权限；若现有环境缺失，将构成唯一允许的用户介入点
+- 当前自动化已阻塞：safe_repairs_pending_manual_apply

TASKS.md

@@ -9,19 +9,6 @@
 ## In Progress
 
 - [ ] `P0-2` 真实账户与订单只读同步
-Files:
-`src/services/`
-`src/repositories/`
-`src/api/`
-`src/web/`
-`tests/unit/`
-Acceptance:
-- 真实账户余额 / 持仓 / 挂单 / 成交可被系统只读同步
-- `RiskContext` 能接入真实账户状态
-- workbench 能显示真实 `open position / pending order`
-- unknown position / order 会触发告警或异常标记
-Risks:
-- 若当前环境缺少真实账户读取凭据或权限，将形成当前阶段唯一允许的外部阻塞
 
 ## Next Up
 
@@ -38,21 +25,8 @@ Risks:
 
 ## Blocked
 
-- [ ] 当前无阻塞；如缺凭据或外部权限，再转入此区
+- [ ] safe_repairs_pending_manual_apply
 
 ## Done
 
-- [x] `P0-1` 实时市场真值层
-  - OKX WebSocket / ticker / candle ingest
-  - MarketTruthSupervisor
-  - `stream_state / data_gap_state`
-  - news provider
-  - stale / gap / disconnect 风险闸门联动
-  - Docker 部署与远端 smoke 完成
-- [x] 市场真值代理正式化
-  - 服务端直连本机 v2rayN LAN 代理 `192.168.3.73:10808`
-  - 临时 relay 已移除
-- [x] 全自动执行协议已固化
-  - `docs/AI_ICT_全自动开发执行流程_v1.md`
-  - `SESSION.md`
-  - `TASKS.md`
+- [ ] 当前尚无完成阶段

scripts/run_autonomous_cycle.py

@@ -132,6 +132,7 @@ def run_autonomous_cycle(env=None, runner=subprocess.run, db_probe=None, schema_
         "schema": None,
         "next_actions": [],
         "action_plan": [],
+        "continuation_decision": "continue",
     }
 
     offline_env = dict(env)
@@ -371,6 +372,7 @@ def _parse_json(value: str):
 def _finish(summary: dict, exit_code: int, output_json: bool) -> int:
     summary["next_actions"] = build_cycle_next_actions(summary)
     summary["action_plan"] = build_cycle_action_plan(summary["next_actions"])
+    summary["continuation_decision"] = classify_cycle_continuation(summary)
     if output_json:
         print(json.dumps(summary, ensure_ascii=False, sort_keys=True))
         return exit_code
@@ -437,6 +439,15 @@ def build_cycle_next_actions(summary: dict) -> list[str]:
     return ["continue_offline_automation"]
 
 
+def classify_cycle_continuation(summary: dict) -> str:
+    mode = summary.get("mode")
+    if mode in {"waiting_for_database", "offline_checks_failed", "migration_failed", "backfill_failed", "pipeline_failed"}:
+        return "blocked"
+    if mode == "pipeline_completed":
+        return "advance"
+    return "continue"
+
+
 def build_cycle_action_plan(next_actions: list[str]) -> list[dict]:
     plan = []
     for action in next_actions:

scripts/run_backtest.py

@@ -58,6 +58,8 @@ def build_backtest_mode_from_env(env=None) -> str:
 
 def build_backtest_subject_type_from_env(env=None) -> str:
     env = os.environ if env is None else env
+    if build_backtest_session_codes_from_env(env=env):
+        return "signal"
     value = str(env.get("BACKTEST_SUBJECT_TYPE") or "").strip()
     return value or BACKTEST_SUBJECT_PLAN_REPLAY
 
@@ -85,6 +87,14 @@ def build_replay_config_from_env(env=None) -> BacktestReplayConfig:
     )
 
 
+def build_backtest_session_codes_from_env(env=None) -> list[str]:
+    env = os.environ if env is None else env
+    raw = str(env.get("BACKTEST_SESSION_CODES") or "").strip()
+    if not raw:
+        return []
+    return [item.strip() for item in raw.split(",") if item.strip()]
+
+
 def build_backtest_timeframe_set_from_env(env=None) -> dict:
     env = os.environ if env is None else env
     bias_tf = str(env.get("AI_ICT_BIAS_TIMEFRAME") or env.get("BACKTEST_BIAS_TIMEFRAME") or "1m")
@@ -111,6 +121,7 @@ def run_backtest_once(
     candidate_model_code: Optional[str] = None,
     replay_config: Optional[BacktestReplayConfig] = None,
     timeframe_set: Optional[dict] = None,
+    session_codes: Optional[list[str]] = None,
 ) -> int:
     run_id, outcomes = service.run_backtest(
         instrument_id=instrument_id,
@@ -123,6 +134,7 @@ def run_backtest_once(
         mode=backtest_mode,
         replay_config=replay_config,
         timeframe_set=timeframe_set,
+        session_codes=session_codes,
     )
     summary = summarize_backtest_outcomes(outcomes, subject_type=subject_type)
     quality_gate_result = evaluate_backtest_quality(
@@ -138,6 +150,7 @@ def run_backtest_once(
         "trading_plan_id": trading_plan_id,
         "candidate_model_code": candidate_model_code,
         "timeframe_set": timeframe_set,
+        "session_codes": session_codes,
         "summary": summary,
         "quality_gate": quality_gate_result,
         "quality_gate_required": require_quality_gate,
@@ -172,6 +185,7 @@ def main(env=None) -> int:
             candidate_model_code=build_backtest_candidate_model_code_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),
         )
     except SQLAlchemyError as exc:
         summary = {

scripts/run_trading_delivery_rehearsal.py

@@ -79,6 +79,7 @@ def run_trading_delivery_rehearsal(env=None, runner=subprocess.run, cwd: Path =
         "blockers": [],
         "manual_decisions": [],
         "promotion_decision": "hold",
+        "continuation_decision": "continue",
         "policies": {
             "reference_migration": (
                 "auto_apply"
@@ -501,10 +502,35 @@ def _finalize_trading_rehearsal(summary: dict) -> dict:
     summary["passed"] = summary["passed"] and len(summary["blockers"]) == 0
     if not summary["passed"]:
         summary["promotion_decision"] = "hold"
+    summary["continuation_decision"] = classify_rehearsal_continuation(summary)
     summary["finished_ts"] = datetime.now(timezone.utc).isoformat()
     return summary
 
 
+def classify_rehearsal_continuation(summary: dict) -> str:
+    mode = summary.get("mode")
+    if mode == "rehearsal_completed":
+        return "advance"
+    if mode in {
+        "reference_migration_deferred",
+        "reference_migration_conflicts_blocked",
+        "reference_migration_manual_apply_required",
+        "checklist_drift_failed",
+        "historical_repair_scan_failed",
+        "historical_repair_apply_failed",
+        "historical_repair_rescan_failed",
+        "historical_repair_safe_candidates_remain",
+        "historical_repair_manual_apply_required",
+        "autonomous_cycle_failed",
+        "deploy_or_health_failed",
+        "post_deploy_readiness_failed",
+        "post_deploy_reports_failed",
+        "reference_migration_apply_failed",
+    }:
+        return "blocked"
+    return "continue"
+
+
 def main(env=None) -> int:
     env = build_trading_rehearsal_env(env=env)
     summary = run_trading_delivery_rehearsal(env=env)

scripts/run_until_blocked.py

@@ -0,0 +1,491 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Optional
+import json
+import os
+import re
+import subprocess
+import sys
+import time
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from scripts.run_trading_delivery_rehearsal import run_trading_delivery_rehearsal
+
+SESSION_PATH = ROOT / "SESSION.md"
+TASKS_PATH = ROOT / "TASKS.md"
+RUNNER_STATUS_PATH = ROOT / ".claude" / "automation_status.json"
+RUNNER_RESUME_WATCH_PATHS = [
+    ROOT / "src",
+    ROOT / "scripts",
+    ROOT / ".env",
+    ROOT / "SESSION.md",
+    ROOT / "TASKS.md",
+]
+STAGE_SEQUENCE = ["P0-2", "P0-3", "P0-4", "P0-5", "P1-1", "P1-2", "P1-3"]
+STAGE_TITLES = {
+    "P0-1": "实时市场真值层",
+    "P0-2": "真实账户与订单只读同步",
+    "P0-3": "盘中持续调度与运行监督",
+    "P0-4": "决策包与主动提醒",
+    "P0-5": "真实交易映射与盘后闭环",
+    "P1-1": "辅助可信度与模型白名单",
+    "P1-2": "操作体验与运营工具",
+    "P1-3": "live adapter 前准备",
+}
+STAGE_GOALS = {
+    "P0-2": [
+        "建立 `BrokerReadModelService`",
+        "读取真实账户余额 / 持仓 / 挂单 / 成交",
+        "扩展 `RiskContext`",
+        "建立 reconciliation 读模型",
+        "前端展示真实 `open position / pending order`",
+        "focused smoke: `/trading/broker-state` 与 `/trading/market-state` 返回 broker state",
+    ],
+    "P0-3": [
+        "建立 `TradingRuntimeScheduler`",
+        "建立 `TradingRuntimeSupervisor`",
+        "提供 runtime health API",
+        "前端展示 runtime status",
+        "focused smoke: runtime 状态 API / workbench runtime status 可见",
+    ],
+    "P0-4": [
+        "建立 `DecisionPacketService`",
+        "输出统一决策包字段",
+        "接入主动提醒通道",
+        "workbench 展示决策卡",
+    ],
+    "P0-5": [
+        "建立 execution 与真实 order/fill 映射",
+        "真实持仓变化触发 journal entry",
+        "增强真实交易归因与周报闭环",
+    ],
+    "P1-1": [
+        "建立 symbol / session / setup 表现统计",
+        "增加辅助可信度评分与白名单",
+    ],
+    "P1-2": [
+        "补齐 workbench 快捷筛选、session 视图、alert history、operator notes",
+        "支持一键生成盘后 review packet",
+    ],
+    "P1-3": [
+        "校准真实 order schema",
+        "补全 bracket / OCO reconciliation 语义",
+        "建立 live readiness checklist 与 operator confirmation protocol",
+    ],
+}
+DEFAULT_VALIDATION_COMMANDS = ["npm run lint", "npm run build", "npm run visual:snapshots"]
+BLOCKED_MARKERS = [
+    "缺凭据",
+    "权限不足",
+    "不可逆",
+    "网络/供应商限制",
+    "外部阻塞",
+]
+
+
+def run_until_blocked(env=None, runner=None, sleeper=None) -> dict:
+    env = dict(os.environ if env is None else env)
+    runner = runner or subprocess.run
+    output_json = _get_bool(env, "RUN_UNTIL_BLOCKED_OUTPUT_JSON", False)
+    max_rounds = max(_get_int(env, "RUN_UNTIL_BLOCKED_MAX_ROUNDS", 8), 1)
+    max_idle_rounds = max(_get_int(env, "RUN_UNTIL_BLOCKED_MAX_IDLE_ROUNDS", 1), 0)
+
+    completed_stages = parse_done_stages(TASKS_PATH.read_text(encoding="utf-8"))
+    current_stage = detect_current_stage(SESSION_PATH.read_text(encoding="utf-8"), completed_stages)
+    idle_rounds = 0
+    rounds = []
+    blocked_reason = None
+    started_ts = datetime.now(timezone.utc).isoformat()
+    write_runner_status(
+        {
+            "state": "running",
+            "started_ts": started_ts,
+            "heartbeat_ts": started_ts,
+            "current_stage": current_stage,
+            "completed_stages": completed_stages,
+            "blocked_reason": None,
+            "rounds_completed": 0,
+        }
+    )
+
+    try:
+        while current_stage is not None and len(rounds) < max_rounds:
+            write_runner_status(
+                {
+                    "state": "running",
+                    "started_ts": started_ts,
+                    "heartbeat_ts": datetime.now(timezone.utc).isoformat(),
+                    "current_stage": current_stage,
+                    "completed_stages": completed_stages,
+                    "blocked_reason": blocked_reason,
+                    "rounds_completed": len(rounds),
+                }
+            )
+            round_env = dict(env)
+            round_env["RUN_UNTIL_BLOCKED_STAGE"] = current_stage
+            round_env["RUN_UNTIL_BLOCKED_STAGE_TITLE"] = STAGE_TITLES[current_stage]
+            summary = run_trading_delivery_rehearsal(env=round_env, runner=runner, sleeper=sleeper)
+            round_result = {
+                "round": len(rounds) + 1,
+                "stage": current_stage,
+                "stage_title": STAGE_TITLES[current_stage],
+                "continuation_decision": summary.get("continuation_decision", "continue"),
+                "mode": summary.get("mode"),
+                "passed": summary.get("passed", False),
+                "blockers": list(summary.get("blockers") or []),
+                "promotion_decision": summary.get("promotion_decision"),
+            }
+            rounds.append(round_result)
+
+            if summary.get("continuation_decision") == "advance":
+                completed_stages.append(current_stage)
+                next_stage = get_next_stage(current_stage, completed_stages)
+                write_anchor_files(
+                    completed_stages=completed_stages,
+                    current_stage=next_stage,
+                    last_result=summary,
+                )
+                current_stage = next_stage
+                idle_rounds = 0
+                continue
+
+            blocked_reason = first_blocked_reason(summary)
+            if blocked_reason is not None or summary.get("continuation_decision") == "blocked":
+                write_anchor_files(
+                    completed_stages=completed_stages,
+                    current_stage=current_stage,
+                    last_result=summary,
+                    blocked_reason=blocked_reason or ", ".join(summary.get("blockers") or []) or summary.get("mode") or "blocked",
+                )
+                break
+
+            idle_rounds += 1
+            write_anchor_files(
+                completed_stages=completed_stages,
+                current_stage=current_stage,
+                last_result=summary,
+                blocked_reason=None,
+            )
+            if idle_rounds > max_idle_rounds:
+                blocked_reason = f"idle_round_limit_exceeded:{summary.get('mode') or 'unknown'}"
+                break
+
+        final_stage = current_stage
+        summary = {
+            "started_ts": started_ts,
+            "finished_ts": datetime.now(timezone.utc).isoformat(),
+            "rounds": rounds,
+            "completed_stages": completed_stages,
+            "current_stage": final_stage,
+            "current_stage_title": STAGE_TITLES.get(final_stage),
+            "blocked_reason": blocked_reason,
+            "status": "completed" if final_stage is None and blocked_reason is None else "blocked" if blocked_reason else "stopped",
+        }
+        write_runner_status(
+            {
+                "state": summary["status"],
+                "started_ts": started_ts,
+                "heartbeat_ts": datetime.now(timezone.utc).isoformat(),
+                "current_stage": final_stage,
+                "completed_stages": completed_stages,
+                "blocked_reason": blocked_reason,
+                "rounds_completed": len(rounds),
+            }
+        )
+    except Exception:
+        write_runner_status(
+            {
+                "state": "failed",
+                "started_ts": started_ts,
+                "heartbeat_ts": datetime.now(timezone.utc).isoformat(),
+                "current_stage": current_stage,
+                "completed_stages": completed_stages,
+                "blocked_reason": blocked_reason or "runner_exception",
+                "rounds_completed": len(rounds),
+            }
+        )
+        raise
+
+    if output_json:
+        print(json.dumps(summary, ensure_ascii=False, sort_keys=True))
+    else:
+        print(f"Run-until-blocked status: {summary['status']}")
+        print(f"Completed stages: {', '.join(completed_stages) if completed_stages else 'none'}")
+        if final_stage:
+            print(f"Current stage: {final_stage} {STAGE_TITLES[final_stage]}")
+        if blocked_reason:
+            print(f"Blocked reason: {blocked_reason}")
+    return summary
+
+
+def run_supervised_until_stopped(env=None, runner=None, sleeper=None) -> dict:
+    env = dict(os.environ if env is None else env)
+    sleeper = sleeper or time.sleep
+    loop_sleep_seconds = max(_get_int(env, "RUN_UNTIL_BLOCKED_LOOP_SLEEP_SECONDS", 30), 1)
+    max_supervisor_cycles = max(_get_int(env, "RUN_UNTIL_BLOCKED_SUPERVISOR_CYCLES", 0), 0)
+    cycles = 0
+    last_summary = None
+    resume_fingerprint = compute_resume_fingerprint()
+
+    while True:
+        last_summary = run_until_blocked(env=env, runner=runner, sleeper=sleeper)
+        cycles += 1
+        if last_summary["status"] == "completed":
+            return last_summary
+
+        supervisor_state = classify_supervisor_state(last_summary)
+        write_runner_status(
+            {
+                "state": supervisor_state,
+                "started_ts": last_summary["started_ts"],
+                "heartbeat_ts": datetime.now(timezone.utc).isoformat(),
+                "current_stage": last_summary.get("current_stage"),
+                "completed_stages": last_summary.get("completed_stages") or [],
+                "blocked_reason": last_summary.get("blocked_reason"),
+                "rounds_completed": len(last_summary.get("rounds") or []),
+                "sleep_seconds": loop_sleep_seconds,
+                "supervisor_cycles": cycles,
+                "resume_fingerprint": resume_fingerprint,
+            }
+        )
+        if max_supervisor_cycles and cycles >= max_supervisor_cycles:
+            return last_summary
+        if supervisor_state in {"paused", "needs_fix"}:
+            if wait_for_resume_condition(
+                previous_fingerprint=resume_fingerprint,
+                sleeper=sleeper,
+                loop_sleep_seconds=loop_sleep_seconds,
+            ):
+                resume_fingerprint = compute_resume_fingerprint()
+                continue
+            return last_summary
+        sleeper(loop_sleep_seconds)
+
+
+def classify_supervisor_state(summary: dict) -> str:
+    blocked_reason = str(summary.get("blocked_reason") or "")
+    if summary.get("status") == "completed":
+        return "completed"
+    if blocked_reason.startswith("autonomous_mode:pipeline_failed"):
+        return "needs_fix"
+    if blocked_reason.startswith("autonomous_mode:offline_checks_failed"):
+        return "needs_fix"
+    if blocked_reason.startswith("autonomous_mode:migration_failed"):
+        return "needs_fix"
+    if blocked_reason.startswith("autonomous_mode:backfill_failed"):
+        return "paused"
+    if blocked_reason.startswith("autonomous_mode:waiting_for_database"):
+        return "sleeping"
+    if blocked_reason:
+        return "paused"
+    return "sleeping"
+
+
+def wait_for_resume_condition(*, previous_fingerprint: str, sleeper, loop_sleep_seconds: int) -> bool:
+    current_fingerprint = compute_resume_fingerprint()
+    if current_fingerprint != previous_fingerprint:
+        return True
+    sleeper(loop_sleep_seconds)
+    return compute_resume_fingerprint() != previous_fingerprint
+
+
+def compute_resume_fingerprint() -> str:
+    entries = []
+    for path in RUNNER_RESUME_WATCH_PATHS:
+        if not path.exists():
+            entries.append(f"{path}:missing")
+            continue
+        if path.is_file():
+            stat = path.stat()
+            entries.append(f"{path}:{stat.st_mtime_ns}:{stat.st_size}")
+            continue
+        for child in sorted(path.rglob("*")):
+            if not child.is_file():
+                continue
+            stat = child.stat()
+            entries.append(f"{child}:{stat.st_mtime_ns}:{stat.st_size}")
+    return "|".join(entries)
+
+
+def write_runner_status(payload: dict) -> None:
+    RUNNER_STATUS_PATH.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def detect_current_stage(session_text: str, completed_stages: list[str]) -> Optional[str]:
+    exact_next_step = re.search(r"###\s+`(P\d-\d)`", session_text)
+    if exact_next_step and exact_next_step.group(1) in STAGE_SEQUENCE:
+        return exact_next_step.group(1)
+    for stage in STAGE_SEQUENCE:
+        if stage not in completed_stages:
+            return stage
+    return None
+
+
+def get_next_stage(current_stage: str, completed_stages: list[str]) -> Optional[str]:
+    for stage in STAGE_SEQUENCE:
+        if stage not in completed_stages:
+            return stage
+    return None
+
+
+def parse_done_stages(tasks_text: str) -> list[str]:
+    return re.findall(r"- \[x\] `(P\d-\d)`", tasks_text)
+
+
+def first_blocked_reason(summary: dict) -> Optional[str]:
+    blockers = [str(item) for item in summary.get("blockers") or []]
+    for blocker in blockers:
+        if any(marker in blocker for marker in BLOCKED_MARKERS):
+            return blocker
+    if summary.get("continuation_decision") == "blocked":
+        return blockers[0] if blockers else summary.get("mode")
+    return None
+
+
+def write_anchor_files(*, completed_stages: list[str], current_stage: Optional[str], last_result: dict, blocked_reason: Optional[str] = None) -> None:
+    SESSION_PATH.write_text(render_session(completed_stages, current_stage, last_result, blocked_reason), encoding="utf-8")
+    TASKS_PATH.write_text(render_tasks(completed_stages, current_stage, blocked_reason), encoding="utf-8")
+
+
+def render_session(completed_stages: list[str], current_stage: Optional[str], last_result: dict, blocked_reason: Optional[str]) -> str:
+    completed_lines = [f"- 已完成 `{stage}`：{STAGE_TITLES[stage]}" for stage in completed_stages] or ["- 本轮尚无新增完成阶段"]
+    next_step_block = render_next_step_block(current_stage)
+    risk_lines = [
+        "- 当前市场真值链路依赖本机 `192.168.3.73:10808` 持续可达；如本机关机或代理退出，远端将退化",
+        "- 真实账户只读同步下一阶段可能需要额外凭据或账号权限；若现有环境缺失，将构成唯一允许的用户介入点",
+    ]
+    if blocked_reason:
+        risk_lines.append(f"- 当前自动化已阻塞：{blocked_reason}")
+    return "\n".join(
+        [
+            "# ai-exchange Session",
+            "",
+            "## Current Goal",
+            "",
+            "从现在开始按全自动模式持续推进交易模块主线，不再等待阶段确认，直到完成 `docs/AI_ICT_实盘辅助决策系统开发方案_v2.md` 中剩余开发目标。",
+            "",
+            "## Current Branch",
+            "",
+            "`main`",
+            "",
+            "## Last Updated",
+            "",
+            f"{datetime.now(timezone.utc).strftime('%Y-%m-%d UTC')}",
+            "",
+            "## Completed In Last Session",
+            "",
+            *completed_lines,
+            "",
+            "## Current Working State",
+            "",
+            "- 当前主线执行协议已固化到：",
+            "  - `docs/AI_ICT_全自动开发执行流程_v1.md`",
+            "- 当前阶段队列来源于：",
+            "  - `docs/AI_ICT_实盘辅助决策系统开发方案_v2.md`",
+            "- 默认不再等待用户中途确认",
+            "- 每阶段必须完成：",
+            "  - 实现",
+            "  - focused tests",
+            "  - `npm run lint`",
+            "  - `npm run build`",
+            "  - `npm run visual:snapshots`",
+            "  - Docker 部署",
+            "  - 远端 smoke",
+            "  - 状态回写",
+            "",
+            "## Exact Next Step",
+            "",
+            next_step_block,
+            "",
+            "## Validation Commands",
+            "",
+            "```bash",
+            *DEFAULT_VALIDATION_COMMANDS,
+            "```",
+            "",
+            "## Risks Or Open Questions",
+            "",
+            *risk_lines,
+            "",
+        ]
+    )
+
+
+def render_next_step_block(current_stage: Optional[str]) -> str:
+    if current_stage is None:
+        return "所有阶段已完成，无下一阶段。"
+    goal_lines = STAGE_GOALS.get(current_stage, [])
+    rendered_goals = "\n".join(f"- {line}" for line in goal_lines)
+    return f"直接进入：\n\n### `{current_stage}` {STAGE_TITLES[current_stage]}\n\n目标：\n\n{rendered_goals}"
+
+
+def render_tasks(completed_stages: list[str], current_stage: Optional[str], blocked_reason: Optional[str]) -> str:
+    in_progress_lines = [f"- [ ] `{current_stage}` {STAGE_TITLES[current_stage]}"] if current_stage else ["- [x] 所有阶段已完成"]
+    next_up_lines = [
+        f"- [ ] `{stage}` {STAGE_TITLES[stage]}"
+        for stage in STAGE_SEQUENCE
+        if stage != current_stage and stage not in completed_stages
+    ] or ["- [x] 当前无后续阶段"]
+    blocked_lines = [f"- [ ] {blocked_reason}"] if blocked_reason else ["- [ ] 当前无阻塞；如缺凭据或外部权限，再转入此区"]
+    done_lines = [f"- [x] `{stage}` {STAGE_TITLES[stage]}" for stage in completed_stages] or ["- [ ] 当前尚无完成阶段"]
+    return "\n".join(
+        [
+            "# ai-exchange Tasks",
+            "",
+            "## Execution Mode",
+            "",
+            "- [x] 全自动模式已启用",
+            "- [x] 默认不等待阶段确认",
+            "- [x] 每阶段必须完成实现、验证、部署、smoke、状态回写",
+            "",
+            "## In Progress",
+            "",
+            *in_progress_lines,
+            "",
+            "## Next Up",
+            "",
+            *next_up_lines,
+            "",
+            "## Backlog",
+            "",
+            "- [ ] 长期目标：在不默认开启真实自动下单的前提下，完成 `Real-time Assisted Trading V1`",
+            "",
+            "## Blocked",
+            "",
+            *blocked_lines,
+            "",
+            "## Done",
+            "",
+            *done_lines,
+            "",
+        ]
+    )
+
+
+def _get_bool(env, key: str, default: bool) -> bool:
+    value = env.get(key)
+    if value is None or value == "":
+        return default
+    return str(value).strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _get_int(env, key: str, default: int) -> int:
+    value = env.get(key)
+    if value is None or value == "":
+        return default
+    return int(value)
+
+
+def main() -> int:
+    env = dict(os.environ)
+    summary = run_supervised_until_stopped(env=env) if _get_bool(env, "RUN_UNTIL_BLOCKED_SUPERVISOR", False) else run_until_blocked(env=env)
+    return 0 if summary["status"] == "completed" else 2 if summary["status"] == "blocked" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/api/__init__.py

@@ -1,6 +1,7 @@
 """Lightweight internal API query helpers."""
 
 from src.api.backtest import get_recent_backtest_summaries
+from src.api.broker_state import get_broker_state
 from src.api.candles import get_recent_candles
 from src.api.execution_tickets import get_recent_execution_tickets
 from src.api.health import get_health
@@ -36,6 +37,7 @@ from src.api.trading_plans import get_recent_trading_plans
 
 __all__ = [
     "get_health",
+    "get_broker_state",
     "get_recent_candles",
     "get_recent_execution_tickets",
     "get_recent_sessions",

src/api/broker_state.py

@@ -0,0 +1,21 @@
+from src.services.broker import BrokerReadModelService
+from src.services.execution import ExecutionService
+
+
+def get_broker_state(instrument_id: int) -> dict:
+    execution_service = ExecutionService()
+    refs = execution_service.build_execution_dashboard_items(instrument_id=instrument_id, limit=100)
+    known_order_ids = []
+    open_execution_count = 0
+    for item in refs:
+        if item.get("execution_status") == "open":
+            open_execution_count += 1
+        known_order_ids.extend(list(item.get("broker_order_ids") or []))
+    snapshot = BrokerReadModelService().fetch_snapshot(
+        instrument_id=instrument_id,
+        known_execution_refs={
+            "broker_order_ids": known_order_ids,
+            "open_execution_count": open_execution_count,
+        },
+    )
+    return snapshot.to_dict()

src/api/server.py

@@ -5,6 +5,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import Optional
 from urllib.parse import parse_qs, urlparse
 
+from src.api.broker_state import get_broker_state
 from src.api.candles import get_recent_candles
 from src.api.execution_tickets import get_recent_execution_tickets
 from src.api.health import get_health
@@ -54,6 +55,7 @@ from src.services.market_state import start_market_truth_runtime
 def build_api_dependencies() -> dict:
     return {
         "health": get_health,
+        "broker_state": get_broker_state,
         "candles": get_recent_candles,
         "execution_tickets": get_recent_execution_tickets,
         "live_operation_audits": get_recent_live_operation_audits,
@@ -118,6 +120,9 @@ def route_request(
 
     if method == "GET" and path == "/health":
         return 200, deps["health"]()
+    if method == "GET" and path == "/trading/broker-state":
+        instrument_id = _require_int(query, "instrument_id")
+        return 200, deps["broker_state"](instrument_id=instrument_id)
     if method == "GET" and path == "/candles":
         instrument_id = _require_int(query, "instrument_id")
         timeframe = _get_single(query, "timeframe", "1m")

src/api/trading_runtime.py

@@ -7,6 +7,7 @@ from src.api.trading_plans import get_recent_trading_plans, normalize_trading_pl
 from src.repositories.signal_repository import SignalRepository
 from src.repositories.trade_plan_candidate_repository import TradePlanCandidateRepository
 from src.repositories.trading_plan_repository import TradingPlanRepository
+from src.services.broker import BrokerReadModelService, BROKER_STATE_OK
 from src.services.execution import ExecutionService
 from src.services.journal import TradingJournalService, WeeklyReviewService
 from src.services.market_state import MarketStateService, get_market_truth_supervisor
@@ -77,11 +78,13 @@ def get_trading_market_state(instrument_id: int, timeframe: str = "1m") -> dict:
         timeframe=timeframe,
         limit=10,
     )
+    broker_state = get_trading_broker_state(instrument_id=instrument_id)
     return {
         "snapshot_id": snapshot_id,
         "market_state": market_state.to_dict(),
         "backtest_quality_gate": quality_gate,
         "recent_incidents": incidents,
+        "broker_state": broker_state,
     }
 
 
@@ -375,6 +378,25 @@ def build_execution_service() -> ExecutionService:
     return ExecutionService(market_state_service=build_market_state_service())
 
 
+def get_trading_broker_state(instrument_id: int) -> dict:
+    execution_service = build_execution_service()
+    items = execution_service.build_execution_dashboard_items(instrument_id=instrument_id, limit=100)
+    known_order_ids = []
+    open_execution_count = 0
+    for item in items:
+        if item.get("execution_status") == "open":
+            open_execution_count += 1
+        known_order_ids.extend(list(item.get("broker_order_ids") or []))
+    snapshot = BrokerReadModelService().fetch_snapshot(
+        instrument_id=instrument_id,
+        known_execution_refs={
+            "broker_order_ids": known_order_ids,
+            "open_execution_count": open_execution_count,
+        },
+    )
+    return snapshot.to_dict()
+
+
 def market_truth_degraded(market_state) -> bool:
     payload = market_state.to_dict() if hasattr(market_state, "to_dict") else dict(market_state or {})
     freshness_code = str(((payload.get("freshness") or {}).get("code")) or "")

src/config/settings.py

@@ -12,6 +12,10 @@ class Settings(BaseSettings):
     okx_api_key: str = ""
     okx_api_secret: str = ""
     okx_api_passphrase: str = ""
+    broker_read_enabled: bool = True
+    broker_read_base_url: str = "https://www.okx.com"
+    broker_read_timeout: int = 30
+    broker_read_proxy_url: str = ""
     market_data_stream_enabled: bool = True
     market_data_stream_symbols: str = "BTC-USDT,ETH-USDT"
     market_data_stream_timeframes: str = "1m"

src/main.py

@@ -11,6 +11,8 @@ from src.services.backtest import (
     BacktestQualityGate,
     BacktestReplayConfig,
     BacktestService,
+    build_plan_replay_units,
+    candidate_row_is_executable,
     evaluate_backtest_quality,
     summarize_backtest_outcomes,
 )
@@ -248,9 +250,15 @@ def run_closed_loop(
     latest_signal = recent_signals[0] if recent_signals else None
     recent_trade_plan_candidates = deps.get("get_recent_trade_plan_candidates", lambda instrument_id, limit: [])(
         instrument_id=instrument_id,
-        limit=1,
+        limit=backtest_limit,
     )
     latest_trade_plan_candidate = recent_trade_plan_candidates[0] if recent_trade_plan_candidates else None
+    backtest_subject_type = select_closed_loop_backtest_subject_type(
+        recent_trade_plan_candidates=recent_trade_plan_candidates,
+        quality_gate=backtest_quality_gate,
+    )
+    backtest_trading_plan_id = (latest_trade_plan_candidate or {}).get("trading_plan_id") if backtest_subject_type == BACKTEST_SUBJECT_PLAN_REPLAY else None
+    backtest_candidate_model_code = (latest_trade_plan_candidate or {}).get("model_code") if backtest_subject_type == BACKTEST_SUBJECT_PLAN_REPLAY else None
 
     execution_id = None
     execution_record = None
@@ -272,9 +280,9 @@ def run_closed_loop(
         limit=backtest_limit,
         cost_config=backtest_cost_config,
         quality_gate=backtest_quality_gate,
-        subject_type=BACKTEST_SUBJECT_PLAN_REPLAY if latest_trade_plan_candidate is not None else BACKTEST_SUBJECT_SIGNAL,
-        trading_plan_id=(latest_trade_plan_candidate or {}).get("trading_plan_id") if latest_trade_plan_candidate is not None else None,
-        candidate_model_code=(latest_trade_plan_candidate or {}).get("model_code") if latest_trade_plan_candidate is not None else None,
+        subject_type=backtest_subject_type,
+        trading_plan_id=backtest_trading_plan_id,
+        candidate_model_code=backtest_candidate_model_code,
         mode=backtest_mode,
         replay_config=backtest_replay_config,
         timeframe_set=timeframe_set,
@@ -354,6 +362,23 @@ def run_closed_loop(
     }
 
 
+def select_closed_loop_backtest_subject_type(
+    *,
+    recent_trade_plan_candidates: list[dict],
+    quality_gate: Optional[BacktestQualityGate],
+) -> str:
+    if not recent_trade_plan_candidates:
+        return BACKTEST_SUBJECT_SIGNAL
+    replay_units = build_plan_replay_units(recent_trade_plan_candidates)
+    executable_units = sum(1 for unit in replay_units if unit.get("confirmed_ts") is not None)
+    gate = quality_gate or BacktestQualityGate()
+    if executable_units < gate.min_executed_signals:
+        return BACKTEST_SUBJECT_SIGNAL
+    if not any(candidate_row_is_executable(row) for row in recent_trade_plan_candidates):
+        return BACKTEST_SUBJECT_SIGNAL
+    return BACKTEST_SUBJECT_PLAN_REPLAY
+
+
 def main() -> None:
     health = get_health()
     print("AI ICT scaffold ready")

src/repositories/backtest_repository.py

@@ -39,14 +39,23 @@ class BacktestRepository:
             row = session.execute(sql, params).mappings().first()
         return dict(row) if row else None
 
-    def fetch_candidate_signals(self, instrument_id: int, limit: int = 100, timeframe_set: dict = None) -> list[dict]:
+    def fetch_candidate_signals(self, instrument_id: int, limit: int = 100, timeframe_set: dict = None, session_codes: Optional[list[str]] = None) -> list[dict]:
         timeframe_set = dict(timeframe_set or {})
-        filters = ["ts.instrument_id = :instrument_id"]
+        session_codes = list(session_codes or [])
+        filters = [
+            "ts.instrument_id = :instrument_id",
+            "ts.status = 'awaiting_confirmation'",
+            "coalesce((ts.meta->>'execution_permission')::boolean, false) = true",
+            "not coalesce(ts.invalidations, '[]'::jsonb) ? 'session_mismatch'",
+        ]
         params = {"instrument_id": instrument_id, "limit": limit}
         for key in ("bias_tf", "setup_tf", "trigger_tf"):
             if timeframe_set.get(key):
-                filters.append(f"bs.{key} = :{key}")
+                filters.append(f"(ts.bias_snapshot_id is null or bs.{key} = :{key})")
                 params[key] = timeframe_set[key]
+        if session_codes:
+            filters.append("ts.session_code = any(:session_codes)")
+            params["session_codes"] = session_codes
         where_sql = " and ".join(filters)
         sql = text(
             f"""

src/services/backtest/__init__.py

@@ -22,6 +22,7 @@ from src.services.backtest.backtest_service import (
     build_cumulative_r_curve,
     build_plan_replay_units,
     build_timeframe_bucket,
+    candidate_row_is_executable,
     compute_max_consecutive_losses,
     compute_max_drawdown_r,
     compute_outcome_r_multiple,
@@ -67,6 +68,7 @@ __all__ = [
     "build_cumulative_r_curve",
     "build_plan_replay_units",
     "build_timeframe_bucket",
+    "candidate_row_is_executable",
     "compute_max_consecutive_losses",
     "compute_max_drawdown_r",
     "compute_outcome_r_multiple",

src/services/backtest/backtest_service.py

@@ -105,6 +105,7 @@ class BacktestService:
         mode: str = BACKTEST_MODE_ENTRY_THEN_OUTCOME,
         replay_config: Optional[BacktestReplayConfig] = None,
         timeframe_set: Optional[dict] = None,
+        session_codes: Optional[list[str]] = None,
     ) -> tuple[int, list[BacktestOutcome]]:
         cost_config = cost_config or BacktestCostConfig()
         quality_gate = quality_gate or BacktestQualityGate()
@@ -135,6 +136,7 @@ class BacktestService:
                 'cost_model': cost_config.__dict__,
                 'quality_gate': quality_gate.__dict__,
                 'replay': replay_config.__dict__ if mode == BACKTEST_MODE_REPLAY_NO_LOOKAHEAD else None,
+                'session_codes': list(session_codes or []),
             },
         )
         if subject_type == BACKTEST_SUBJECT_PLAN_REPLAY:
@@ -158,13 +160,14 @@ class BacktestService:
             instrument_id=instrument_id,
             limit=limit,
             timeframe_set=timeframe_set,
+            session_codes=session_codes,
         )
         outcomes: list[BacktestOutcome] = []
         for signal in signals:
             sessions = self.repository.fetch_sessions_covering_window(
                 instrument_id=instrument_id,
                 start_ts=signal['created_ts'],
-                end_ts=signal['expires_ts'],
+                end_ts=signal['created_ts'],
             )
             future = self.repository.fetch_future_candles(
                 instrument_id=instrument_id,
@@ -176,6 +179,10 @@ class BacktestService:
             if not future:
                 target_plan = extract_target_plan_from_signal_row(signal)
                 management_rules = extract_management_rules_from_signal_row(signal)
+                session_codes = resolve_signal_session_codes(signal=signal, sessions=sessions)
+                error_tags = ['no_future_candles']
+                if len(sessions) == 0:
+                    error_tags.append('outside_session_window')
                 record = BacktestOutcome(
                     signal_id=signal['id'],
                     outcome='missed',
@@ -184,13 +191,14 @@ class BacktestService:
                     tp1_hit=False,
                     tp2_hit=False,
                     invalidated_before_entry=True,
-                    error_tags=['no_future_candles'],
+                    error_tags=_dedupe_text_list(error_tags),
                     meta={
                         'entry_triggered': False,
                         'same_candle_priority': SAME_CANDLE_PRIORITY,
                         'same_candle_ambiguous': False,
                         'sessions_found': len(sessions),
-                        'session_codes': [row['session_code'] for row in sessions],
+                        'session_codes': session_codes,
+                        'session_filter': 'outside_window' if len(sessions) == 0 else 'covered',
                         'signal_side': signal['side'],
                         'signal_symbol': signal.get('symbol'),
                         'signal_timeframe_set': {
@@ -217,6 +225,7 @@ class BacktestService:
                         'replay_right_bars': replay_config.right_bars if mode == BACKTEST_MODE_REPLAY_NO_LOOKAHEAD else None,
                     },
                 )
+
                 outcomes.append(record)
                 self.repository.insert_backtest_result(
                     backtest_run_id=run_id,
@@ -236,6 +245,7 @@ class BacktestService:
                 continue
 
             session_filtered = len(sessions) == 0
+            session_codes = resolve_signal_session_codes(signal=signal, sessions=sessions)
             record = evaluate_signal_outcome(
                 BacktestSignal(
                     signal_id=signal['id'],
@@ -263,7 +273,7 @@ class BacktestService:
             record.meta.update(
                 {
                     'sessions_found': len(sessions),
-                    'session_codes': [row['session_code'] for row in sessions],
+                    'session_codes': session_codes,
                     'session_filter': 'outside_window' if session_filtered else 'covered',
                     'signal_side': signal['side'],
                     'signal_symbol': signal.get('symbol'),
@@ -365,10 +375,11 @@ class BacktestService:
         replay_config: BacktestReplayConfig,
         timeframe_set: dict,
     ) -> BacktestOutcome:
+        session_reference_ts = unit['confirmed_ts'] or unit['appeared_ts']
         sessions = self.repository.fetch_sessions_covering_window(
             instrument_id=instrument_id,
-            start_ts=unit['appeared_ts'],
-            end_ts=unit['expires_ts'],
+            start_ts=session_reference_ts,
+            end_ts=session_reference_ts,
         )
         if unit['confirmed_ts'] is None:
             failure_reasons = list(unit['reason_codes'] or unit['missing_confirmations'] or ['candidate_not_confirmed'])
@@ -960,6 +971,14 @@ def build_plan_replay_meta(
     }
 
 
+def resolve_signal_session_codes(*, signal: dict, sessions: list[dict]) -> list[str]:
+    session_codes = [row.get('session_code') for row in sessions if row.get('session_code')]
+    if session_codes:
+        return session_codes
+    signal_session = str(signal.get('session_code') or '').strip()
+    return [signal_session] if signal_session else []
+
+
 def _dedupe_text_list(values: list[str]) -> list[str]:
     seen = set()
     result = []

src/services/broker/__init__.py

@@ -8,6 +8,13 @@ from src.services.broker.broker_adapter import (
     KillSwitchRequest,
     ReconciliationRequest,
 )
+from src.services.broker.broker_read_model_service import (
+    ALERT_UNKNOWN_OPEN_POSITION,
+    ALERT_UNKNOWN_PENDING_ORDER,
+    BROKER_STATE_OK,
+    BROKER_STATE_UNAVAILABLE,
+    BrokerReadModelService,
+)
 from src.services.broker.live_broker_service import (
     LiveBrokerService,
     LiveOperationDecision,
@@ -27,6 +34,11 @@ __all__ = [
     "LiveOperationDecision",
     "OKXOrderAdapter",
     "ReconciliationRequest",
+    "BROKER_STATE_OK",
+    "BROKER_STATE_UNAVAILABLE",
+    "ALERT_UNKNOWN_OPEN_POSITION",
+    "ALERT_UNKNOWN_PENDING_ORDER",
+    "BrokerReadModelService",
     "build_broker_adapter",
     "validate_live_dangerous_operation",
     "validate_live_submit_request",

src/services/broker/broker_read_model_service.py

@@ -0,0 +1,355 @@
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+import base64
+import hashlib
+import hmac
+import json
+from os import getenv
+from typing import Optional
+
+import httpx
+
+from src.config.settings import settings
+from src.services.risk.risk_context import AccountRiskState
+
+
+OKX_CODE_OK = "0"
+BROKER_STATE_OK = "broker_state_ok"
+BROKER_STATE_UNAVAILABLE = "broker_state_unavailable"
+ALERT_UNKNOWN_OPEN_POSITION = "unknown_open_position_detected"
+ALERT_UNKNOWN_PENDING_ORDER = "unknown_pending_order_detected"
+
+
+@dataclass(frozen=True)
+class BrokerAccountSnapshot:
+    equity: float
+    available_balance: float
+    cash_balance: float
+    unrealized_pnl: float
+    daily_pnl: float
+    fetched_ts: Optional[datetime]
+    raw: dict
+
+    def to_dict(self) -> dict:
+        return _serialize(asdict(self))
+
+
+@dataclass(frozen=True)
+class BrokerPositionSnapshot:
+    instrument_id: str
+    side: str
+    size: float
+    average_price: float
+    mark_price: float
+    unrealized_pnl: float
+    raw: dict
+
+    def to_dict(self) -> dict:
+        return _serialize(asdict(self))
+
+
+@dataclass(frozen=True)
+class BrokerOrderSnapshot:
+    order_id: str
+    instrument_id: str
+    side: str
+    status: str
+    size: float
+    filled_size: float
+    price: float
+    raw: dict
+
+    def to_dict(self) -> dict:
+        return _serialize(asdict(self))
+
+
+@dataclass(frozen=True)
+class BrokerFillSnapshot:
+    fill_id: str
+    order_id: str
+    instrument_id: str
+    side: str
+    fill_size: float
+    fill_price: float
+    fill_ts: Optional[datetime]
+    raw: dict
+
+    def to_dict(self) -> dict:
+        return _serialize(asdict(self))
+
+
+@dataclass(frozen=True)
+class BrokerReadModelSnapshot:
+    status: str
+    provider: str
+    fetched_ts: Optional[datetime]
+    account: dict
+    positions: list[dict]
+    orders: list[dict]
+    fills: list[dict]
+    alerts: list[str]
+    error: Optional[str]
+
+    def to_dict(self) -> dict:
+        return _serialize(asdict(self))
+
+
+class BrokerReadModelService:
+    provider_name = "okx_read_only"
+
+    def __init__(self, *, base_url: Optional[str] = None, proxy_url: Optional[str] = None, timeout: Optional[int] = None) -> None:
+        self.base_url = (base_url or settings.broker_read_base_url or settings.okx_base_url).rstrip("/")
+        self.proxy_url = proxy_url or settings.broker_read_proxy_url or settings.market_truth_proxy_url or getenv("HTTPS_PROXY") or getenv("HTTP_PROXY") or ""
+        self.timeout = timeout or settings.broker_read_timeout or settings.okx_timeout
+
+    def fetch_snapshot(
+        self,
+        *,
+        instrument_id: Optional[int] = None,
+        now: Optional[datetime] = None,
+        account_payload: Optional[dict] = None,
+        positions_payload: Optional[dict] = None,
+        orders_payload: Optional[dict] = None,
+        fills_payload: Optional[dict] = None,
+        known_execution_refs: Optional[dict] = None,
+    ) -> BrokerReadModelSnapshot:
+        observed_now = _ensure_aware(now or datetime.now(timezone.utc))
+        try:
+            account_response = account_payload if account_payload is not None else self._private_get("/api/v5/account/balance")
+            positions_response = positions_payload if positions_payload is not None else self._private_get("/api/v5/account/positions")
+            orders_response = orders_payload if orders_payload is not None else self._private_get("/api/v5/trade/orders-pending")
+            fills_response = fills_payload if fills_payload is not None else self._private_get("/api/v5/trade/fills", params={"limit": "100"})
+        except Exception as exc:
+            return BrokerReadModelSnapshot(
+                status=BROKER_STATE_UNAVAILABLE,
+                provider=self.provider_name,
+                fetched_ts=observed_now,
+                account={},
+                positions=[],
+                orders=[],
+                fills=[],
+                alerts=[BROKER_STATE_UNAVAILABLE],
+                error=str(exc),
+            )
+
+        account = self._parse_account_snapshot(account_response, fetched_ts=observed_now)
+        positions = self._parse_positions(positions_response, instrument_id=instrument_id)
+        orders = self._parse_orders(orders_response, instrument_id=instrument_id)
+        fills = self._parse_fills(fills_response, instrument_id=instrument_id)
+        alerts = self._build_alerts(
+            positions=positions,
+            orders=orders,
+            known_execution_refs=known_execution_refs or {},
+        )
+        return BrokerReadModelSnapshot(
+            status=BROKER_STATE_OK,
+            provider=self.provider_name,
+            fetched_ts=observed_now,
+            account=account.to_dict(),
+            positions=[item.to_dict() for item in positions],
+            orders=[item.to_dict() for item in orders],
+            fills=[item.to_dict() for item in fills],
+            alerts=alerts,
+            error=None,
+        )
+
+    def build_account_risk_state(
+        self,
+        snapshot: Optional[BrokerReadModelSnapshot],
+        *,
+        instrument_id: Optional[int] = None,
+        side: Optional[str] = None,
+        base_state: Optional[AccountRiskState] = None,
+    ) -> AccountRiskState:
+        seed = base_state or AccountRiskState()
+        if snapshot is None or snapshot.status != BROKER_STATE_OK:
+            return seed
+        account = dict(snapshot.account or {})
+        positions = [dict(item) for item in list(snapshot.positions or [])]
+        relevant_positions = [item for item in positions if instrument_id is None or _matches_instrument(item.get("instrument_id"), instrument_id)]
+        same_side_positions = [item for item in relevant_positions if side and str(item.get("side") or "").lower() == str(side).lower()]
+        daily_pnl = _float(account.get("daily_pnl"))
+        equity = max(_float(account.get("equity")), 0.0)
+        daily_pnl_pct = daily_pnl / equity if equity > 0 else 0.0
+        return AccountRiskState(
+            kill_switch=seed.kill_switch,
+            daily_pnl_pct=daily_pnl_pct,
+            consecutive_losses=seed.consecutive_losses,
+            open_positions=max(seed.open_positions, len(relevant_positions)),
+            same_side_open_positions=max(seed.same_side_open_positions, len(same_side_positions)),
+            max_daily_loss_pct=seed.max_daily_loss_pct,
+            max_consecutive_losses=seed.max_consecutive_losses,
+            max_open_positions=seed.max_open_positions,
+            max_same_side_positions=seed.max_same_side_positions,
+        )
+
+    def _private_get(self, path: str, *, params: Optional[dict[str, str]] = None) -> dict:
+        api_key = settings.okx_api_key.strip()
+        api_secret = settings.okx_api_secret.strip()
+        api_passphrase = settings.okx_api_passphrase.strip()
+        if not api_key or not api_secret or not api_passphrase:
+            raise ValueError("okx_read_credentials_missing")
+        request_path = path
+        if params:
+            query = httpx.QueryParams(params).render()
+            request_path = f"{path}?{query}"
+        timestamp = datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
+        sign_input = f"{timestamp}GET{request_path}".encode("utf-8")
+        signature = base64.b64encode(hmac.new(api_secret.encode("utf-8"), sign_input, hashlib.sha256).digest()).decode("utf-8")
+        client_kwargs = {"timeout": self.timeout, "follow_redirects": True}
+        if self.proxy_url:
+            client_kwargs["proxy"] = self.proxy_url
+            client_kwargs["trust_env"] = False
+        with httpx.Client(**client_kwargs) as client:
+            response = client.get(
+                f"{self.base_url}{path}",
+                params=params,
+                headers={
+                    "OK-ACCESS-KEY": api_key,
+                    "OK-ACCESS-SIGN": signature,
+                    "OK-ACCESS-TIMESTAMP": timestamp,
+                    "OK-ACCESS-PASSPHRASE": api_passphrase,
+                    "Content-Type": "application/json",
+                },
+            )
+            response.raise_for_status()
+            payload = response.json()
+        if not isinstance(payload, dict):
+            raise ValueError("okx_read_invalid_payload")
+        if str(payload.get("code", "")) not in {"", OKX_CODE_OK}:
+            raise ValueError(f"okx_read_error:{payload.get('code')}:{payload.get('msg')}")
+        return payload
+
+    def _parse_account_snapshot(self, payload: dict, *, fetched_ts: datetime) -> BrokerAccountSnapshot:
+        rows = list(payload.get("data") or [])
+        row = dict(rows[0] or {}) if rows else {}
+        details = list(row.get("details") or [])
+        first_detail = dict(details[0] or {}) if details else {}
+        return BrokerAccountSnapshot(
+            equity=_float(row.get("totalEq") or row.get("adjEq")),
+            available_balance=_float(first_detail.get("availBal") or row.get("availEq")),
+            cash_balance=_float(first_detail.get("cashBal") or row.get("totalEq")),
+            unrealized_pnl=_float(row.get("upl")),
+            daily_pnl=_float(row.get("upl")),
+            fetched_ts=fetched_ts,
+            raw=row,
+        )
+
+    def _parse_positions(self, payload: dict, *, instrument_id: Optional[int]) -> list[BrokerPositionSnapshot]:
+        items: list[BrokerPositionSnapshot] = []
+        for row in list(payload.get("data") or []):
+            data = dict(row or {})
+            if instrument_id is not None and not _matches_instrument(data.get("instId"), instrument_id):
+                continue
+            size = abs(_float(data.get("pos")))
+            if size <= 0:
+                continue
+            items.append(
+                BrokerPositionSnapshot(
+                    instrument_id=str(data.get("instId") or ""),
+                    side=str(data.get("posSide") or data.get("side") or "net").lower(),
+                    size=size,
+                    average_price=_float(data.get("avgPx")),
+                    mark_price=_float(data.get("markPx")),
+                    unrealized_pnl=_float(data.get("upl")),
+                    raw=data,
+                )
+            )
+        return items
+
+    def _parse_orders(self, payload: dict, *, instrument_id: Optional[int]) -> list[BrokerOrderSnapshot]:
+        items: list[BrokerOrderSnapshot] = []
+        for row in list(payload.get("data") or []):
+            data = dict(row or {})
+            if instrument_id is not None and not _matches_instrument(data.get("instId"), instrument_id):
+                continue
+            items.append(
+                BrokerOrderSnapshot(
+                    order_id=str(data.get("ordId") or ""),
+                    instrument_id=str(data.get("instId") or ""),
+                    side=str(data.get("side") or "").lower(),
+                    status=str(data.get("state") or "unknown").lower(),
+                    size=_float(data.get("sz")),
+                    filled_size=_float(data.get("accFillSz")),
+                    price=_float(data.get("px")),
+                    raw=data,
+                )
+            )
+        return items
+
+    def _parse_fills(self, payload: dict, *, instrument_id: Optional[int]) -> list[BrokerFillSnapshot]:
+        items: list[BrokerFillSnapshot] = []
+        for row in list(payload.get("data") or []):
+            data = dict(row or {})
+            if instrument_id is not None and not _matches_instrument(data.get("instId"), instrument_id):
+                continue
+            fill_ts = _parse_millis_or_none(data.get("fillTime") or data.get("ts"))
+            items.append(
+                BrokerFillSnapshot(
+                    fill_id=str(data.get("fillId") or ""),
+                    order_id=str(data.get("ordId") or ""),
+                    instrument_id=str(data.get("instId") or ""),
+                    side=str(data.get("side") or "").lower(),
+                    fill_size=_float(data.get("fillSz") or data.get("sz")),
+                    fill_price=_float(data.get("fillPx") or data.get("px")),
+                    fill_ts=fill_ts,
+                    raw=data,
+                )
+            )
+        return items
+
+    def _build_alerts(self, *, positions: list[BrokerPositionSnapshot], orders: list[BrokerOrderSnapshot], known_execution_refs: dict) -> list[str]:
+        alerts: list[str] = []
+        known_order_ids = {str(value) for value in list(known_execution_refs.get("broker_order_ids") or []) if value not in (None, "")}
+        known_open_positions = int(known_execution_refs.get("open_execution_count") or 0)
+        if positions and known_open_positions == 0:
+            alerts.append(ALERT_UNKNOWN_OPEN_POSITION)
+        if orders and any(order.order_id and order.order_id not in known_order_ids for order in orders):
+            alerts.append(ALERT_UNKNOWN_PENDING_ORDER)
+        return alerts
+
+
+
+def _matches_instrument(value: object, instrument_id: int) -> bool:
+    text = str(value or "").upper()
+    if not text:
+        return False
+    return text.startswith("BTC-") if int(instrument_id) == 1 else text.startswith("ETH-") if int(instrument_id) == 2 else True
+
+
+
+def _float(value: object) -> float:
+    try:
+        return float(value or 0.0)
+    except (TypeError, ValueError):
+        return 0.0
+
+
+
+def _parse_millis_or_none(value: object) -> Optional[datetime]:
+    if value in (None, ""):
+        return None
+    text = str(value)
+    if text.isdigit():
+        return datetime.fromtimestamp(int(text) / 1000, tz=timezone.utc)
+    return _ensure_aware(datetime.fromisoformat(text.replace("Z", "+00:00")))
+
+
+
+def _ensure_aware(value: datetime) -> datetime:
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+
+def _serialize(value):
+    if isinstance(value, datetime):
+        return value.isoformat()
+    if isinstance(value, dict):
+        return {key: _serialize(item) for key, item in value.items()}
+    if isinstance(value, list):
+        return [_serialize(item) for item in value]
+    return value

src/services/execution/execution_service.py

@@ -4,6 +4,7 @@ import math
 from typing import Optional
 
 from src.repositories.execution_repository import ExecutionRepository
+from src.services.broker import BrokerReadModelService, BROKER_STATE_OK
 from src.services.backtest import (
     BACKTEST_LIFECYCLE_INVALIDATED_BEFORE_ENTRY,
     BACKTEST_LIFECYCLE_MISSED,
@@ -423,10 +424,26 @@ class ExecutionService:
         quality_gate = self.market_state_service.repository.fetch_latest_backtest_quality_gate(
             instrument_id=int(signal["instrument_id"])
         )
+        broker_snapshot = BrokerReadModelService().fetch_snapshot(
+            instrument_id=int(signal["instrument_id"]),
+            known_execution_refs={
+                "broker_order_ids": [],
+                "open_execution_count": 0,
+            },
+        )
+        effective_account_state = account_state or AccountRiskState()
+        if broker_snapshot.status == BROKER_STATE_OK:
+            effective_account_state = BrokerReadModelService().build_account_risk_state(
+                broker_snapshot,
+                instrument_id=int(signal["instrument_id"]),
+                side=str(signal.get("side") or ""),
+                base_state=effective_account_state,
+            )
         return RiskContext.from_account_state(
-            account_state,
+            effective_account_state,
             market_state=market_state,
             backtest_quality_gate=quality_gate,
+            broker_state=broker_snapshot.to_dict(),
         )
 
     def create_execution_ticket(

src/services/risk/risk_context.py

@@ -24,6 +24,7 @@ class AccountRiskState:
 class RiskContext(AccountRiskState):
     market_state: Optional[object] = None
     backtest_quality_gate: Optional[dict] = None
+    broker_state: Optional[dict] = None
     news_window: Optional[dict] = None
     opening_volatility: Optional[dict] = None
     block_opening_volatility: bool = False

src/services/signals/signal_service.py

@@ -56,7 +56,7 @@ class SignalService:
     def __init__(
         self,
         repository: Optional[SignalRepository] = None,
-        required_evidence_keys: tuple[str, ...] = MINIMAL_SIGNAL_EVIDENCE_KEYS,
+        required_evidence_keys: tuple[str, ...] = STRICT_SIGNAL_EVIDENCE_KEYS,
         target_selection_service: Optional[TargetSelectionService] = None,
     ) -> None:
         self.repository = repository or SignalRepository()
@@ -79,9 +79,9 @@ class SignalService:
         entry_low = float(fvg['lower'])
         entry_high = float(fvg['upper'])
         if side == 'buy':
-            stop_loss = min(float(mss['broken_level']), entry_low) * 0.999
+            stop_loss = min(float(mss['broken_level']), entry_low) * 0.998
         else:
-            stop_loss = max(float(mss['broken_level']), entry_high) * 1.001
+            stop_loss = max(float(mss['broken_level']), entry_high) * 1.002
         target_plan = self._build_target_plan(
             instrument_id=instrument_id,
             side=side,
@@ -547,13 +547,13 @@ def validate_signal_evidence(evidence: dict, required_keys: tuple[str, ...] = MI
 
 
 def classify_session_for_signal(session_code: Optional[str]) -> dict[str, object]:
-    if session_code in {'LONDON', 'NY_AM', 'NY_PM'}:
+    if session_code in {'LONDON', 'NY_AM'}:
         return {
             'status': 'awaiting_confirmation',
             'execution_permission': True,
             'reason': 'allowed_session',
         }
-    if session_code in {'ASIA', 'NY_LUNCH'}:
+    if session_code in {'ASIA', 'NY_LUNCH', 'NY_PM'}:
         return {
             'status': 'awaiting_context',
             'execution_permission': False,

src/web/app.js

@@ -239,11 +239,20 @@ function renderMarketState(payload, executions) {
   const session = marketState.session || {};
   const newsWindow = marketState.news_window || {};
   const riskDecision = payload?.risk_decision || {};
+  const brokerState = payload?.broker_state || {};
+  const brokerAccount = brokerState.account || {};
+  const brokerAlerts = brokerState.alerts || [];
   const panel = buildInfoGrid([
     ['symbol', String(instrumentIdInput.value || '1')],
     ['session', session.session_code || 'OFF_HOURS'],
     ['data freshness', freshness.code || 'unknown'],
     ['news state', newsWindow.code || 'unknown'],
+    ['broker state', brokerState.status || 'unknown'],
+    ['broker equity', formatNumber(brokerAccount.equity, 2)],
+    ['broker available balance', formatNumber(brokerAccount.available_balance, 2)],
+    ['broker open positions', String((brokerState.positions || []).length)],
+    ['broker pending orders', String((brokerState.orders || []).length)],
+    ['broker alerts', formatList(brokerAlerts)],
     ['account risk state', formatJsonInline(riskDecision.reject_codes || [])],
     ['open position', String(executions.filter((item) => item.execution_status === 'open').length)],
     ['kill switch', (riskDecision.reject_codes || []).includes('execution_kill_switch_enabled') ? 'active' : 'off'],

tests/unit/test_api_and_main.py

@@ -1,5 +1,6 @@
 import os
 import unittest
+from datetime import datetime, timezone
 from unittest.mock import patch
 
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
@@ -10,7 +11,7 @@ from src.api.server import route_request
 from src.api.signals import get_recent_signals
 from src.services.backtest import BacktestCostConfig, BacktestQualityGate, BacktestReplayConfig
 from src.services.execution import AccountRiskState
-from src.main import run_closed_loop
+from src.main import run_closed_loop, select_closed_loop_backtest_subject_type
 
 
 class FakeResult:
@@ -256,6 +257,39 @@ class FakePostTradeReviewService:
 
 
 class ClosedLoopTests(unittest.TestCase):
+    def test_select_closed_loop_backtest_subject_type_falls_back_to_signal_when_replay_history_is_sparse(self) -> None:
+        subject_type = select_closed_loop_backtest_subject_type(
+            recent_trade_plan_candidates=[
+                {
+                    "trading_plan_id": 1,
+                    "model_code": "OSOK",
+                    "side": "buy",
+                    "setup_timeframe": "1m",
+                    "trigger_timeframe": "1m",
+                    "created_ts": datetime(2026, 5, 6, 0, 0, tzinfo=timezone.utc),
+                    "status": "blocked",
+                    "confirmation_state": {"executable": False},
+                    "missing_confirmations": ["sweep_confirmed"],
+                    "entry": {},
+                },
+                {
+                    "trading_plan_id": 1,
+                    "model_code": "OSOK",
+                    "side": "buy",
+                    "setup_timeframe": "1m",
+                    "trigger_timeframe": "1m",
+                    "created_ts": datetime(2026, 5, 6, 1, 0, tzinfo=timezone.utc),
+                    "status": "blocked",
+                    "confirmation_state": {"executable": False},
+                    "missing_confirmations": ["fvg_retrace_confirmed"],
+                    "entry": {},
+                },
+            ],
+            quality_gate=BacktestQualityGate(min_executed_signals=30),
+        )
+
+        self.assertEqual(subject_type, "signal")
+
     def test_run_closed_loop_returns_summary_and_reports(self) -> None:
         result = run_closed_loop(
             instrument_id=1,
@@ -384,6 +418,20 @@ class ClosedLoopTests(unittest.TestCase):
             "intraday_report_service": FakeIntradayReportService(),
             "post_trade_review_service": FakePostTradeReviewService(),
             "get_recent_signals": lambda instrument_id, limit: [{"id": 1, "session_code": "LONDON", "status": "awaiting_confirmation", "side": "buy", "invalidations": []}],
+            "get_recent_trade_plan_candidates": lambda instrument_id, limit: [
+                {
+                    "trading_plan_id": 1,
+                    "model_code": "OSOK",
+                    "side": "buy",
+                    "setup_timeframe": "1m",
+                    "trigger_timeframe": "1m",
+                    "created_ts": datetime(2026, 5, 6, 0, 0, tzinfo=timezone.utc),
+                    "status": "blocked",
+                    "confirmation_state": {"executable": False},
+                    "missing_confirmations": ["sweep_confirmed"],
+                    "entry": {},
+                }
+            ],
             "get_recent_backtest_summaries": lambda instrument_id, limit: [{"id": 55, "summary": {"total_signals": 1}}],
         }
 
@@ -397,6 +445,9 @@ class ClosedLoopTests(unittest.TestCase):
         )
 
         self.assertEqual(backtest_service.last_kwargs["quality_gate"], quality_gate)
+        self.assertEqual(backtest_service.last_kwargs["subject_type"], "signal")
+        self.assertIsNone(backtest_service.last_kwargs["trading_plan_id"])
+        self.assertIsNone(backtest_service.last_kwargs["candidate_model_code"])
 
     def test_run_closed_loop_passes_backtest_replay_mode_and_config(self) -> None:
         backtest_service = FakeBacktestService()

tests/unit/test_api_server.py

@@ -51,6 +51,17 @@ class ApiServerRoutingTests(unittest.TestCase):
         self.assertEqual(payload["snapshot_id"], 1)
         self.assertEqual(payload["market_state"]["timeframe"], "5m")
 
+    def test_routes_trading_broker_state(self) -> None:
+        status, payload = route_request(
+            "/trading/broker-state",
+            {"instrument_id": ["1"]},
+            dependencies={"broker_state": lambda **kwargs: {"status": "broker_state_ok", **kwargs}},
+        )
+
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["status"], "broker_state_ok")
+        self.assertEqual(payload["instrument_id"], 1)
+
     def test_routes_trading_watch_run(self) -> None:
         status, payload = route_request(
             "/trading/watch/run",

tests/unit/test_backtest_script.py

@@ -12,6 +12,7 @@ os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:
 from scripts import run_backtest
 from scripts.run_backtest import (
     build_backtest_candidate_model_code_from_env,
+    build_backtest_session_codes_from_env,
     build_backtest_timeframe_set_from_env,
     build_backtest_mode_from_env,
     build_backtest_subject_type_from_env,
@@ -96,10 +97,13 @@ class RunBacktestScriptTests(unittest.TestCase):
         self.assertEqual(build_backtest_mode_from_env({"BACKTEST_MODE": "replay_no_lookahead"}), "replay_no_lookahead")
         self.assertEqual(build_backtest_subject_type_from_env({}), "plan_replay")
         self.assertEqual(build_backtest_subject_type_from_env({"BACKTEST_SUBJECT_TYPE": "signal"}), "signal")
+        self.assertEqual(build_backtest_subject_type_from_env({"BACKTEST_SESSION_CODES": "LONDON"}), "signal")
         self.assertIsNone(build_backtest_trading_plan_id_from_env({}))
         self.assertEqual(build_backtest_trading_plan_id_from_env({"BACKTEST_TRADING_PLAN_ID": "77"}), 77)
         self.assertIsNone(build_backtest_candidate_model_code_from_env({}))
         self.assertEqual(build_backtest_candidate_model_code_from_env({"BACKTEST_CANDIDATE_MODEL_CODE": "OSOK"}), "OSOK")
+        self.assertEqual(build_backtest_session_codes_from_env({}), [])
+        self.assertEqual(build_backtest_session_codes_from_env({"BACKTEST_SESSION_CODES": "LONDON,NY_AM"}), ["LONDON", "NY_AM"])
 
         replay_config = build_replay_config_from_env(
             {
@@ -220,12 +224,14 @@ class RunBacktestScriptTests(unittest.TestCase):
                 subject_type="plan_replay",
                 trading_plan_id=77,
                 candidate_model_code="OSOK",
+                session_codes=["LONDON"],
             )
 
         self.assertEqual(exit_code, 0)
         self.assertEqual(service.calls[0]["subject_type"], "plan_replay")
         self.assertEqual(service.calls[0]["trading_plan_id"], 77)
         self.assertEqual(service.calls[0]["candidate_model_code"], "OSOK")
+        self.assertEqual(service.calls[0]["session_codes"], ["LONDON"])
 
     def test_main_reports_database_unavailable(self) -> None:
         output = io.StringIO()

tests/unit/test_backtest_service.py

@@ -470,7 +470,7 @@ class BacktestServiceRunTests(unittest.TestCase):
                 self.results = []
                 self.run_config = {}
 
-            def fetch_candidate_signals(self, instrument_id: int, limit: int, timeframe_set=None):
+            def fetch_candidate_signals(self, instrument_id: int, limit: int, timeframe_set=None, session_codes=None):
                 case.assertEqual(instrument_id, 1)
                 case.assertEqual(limit, 10)
                 case.assertEqual(timeframe_set, {"bias_tf": "1m", "setup_tf": "1m", "trigger_tf": "1m"})
@@ -540,6 +540,7 @@ class BacktestServiceRunTests(unittest.TestCase):
         self.assertEqual(repository.results[0]["meta"]["management_rules"][0]["code"], "tp1_partial")
         self.assertEqual(repository.summary["by_symbol"]["BTC-USDT"]["count"], 1)
         self.assertEqual(repository.summary["by_timeframe"]["1m/1m/1m"]["count"], 1)
+        self.assertEqual(repository.results[0]["meta"]["session_codes"], ["LONDON"])
         self.assertIn("quality_gate", repository.summary)
         self.assertFalse(repository.summary["quality_gate"]["passed"])
         self.assertIn("sample_size_too_small", repository.summary["quality_gate"]["reasons"])
@@ -547,6 +548,63 @@ class BacktestServiceRunTests(unittest.TestCase):
         self.assertEqual(repository.run_config["quality_gate"]["min_total_signals"], 100)
         self.assertEqual(repository.run_config["quality_gate"]["min_executed_signals"], 30)
 
+    def test_run_backtest_preserves_signal_session_code_when_no_session_window_matches(self) -> None:
+        case = self
+
+        class FakeBacktestRepository:
+            def __init__(self) -> None:
+                self.results = []
+
+            def fetch_candidate_signals(self, instrument_id: int, limit: int, timeframe_set=None, session_codes=None):
+                return [
+                    {
+                        "id": 7,
+                        "created_ts": datetime(2026, 4, 29, 21, 0, tzinfo=timezone.utc),
+                        "expires_ts": datetime(2026, 4, 29, 22, 0, tzinfo=timezone.utc),
+                        "side": "buy",
+                        "entry_low": 100.0,
+                        "entry_high": 101.0,
+                        "stop_loss": 99.0,
+                        "tp1": 102.0,
+                        "tp2": 104.0,
+                        "rr_tp1": 1.0,
+                        "rr_tp2": 2.0,
+                        "symbol": "BTC-USDT",
+                        "bias_tf": "1m",
+                        "setup_tf": "1m",
+                        "trigger_tf": "1m",
+                        "session_code": "OFF_HOURS",
+                        "target_plan": {},
+                        "signal_meta": {},
+                    }
+                ]
+
+            def insert_backtest_run(self, **kwargs):
+                return 99
+
+            def fetch_sessions_covering_window(self, **kwargs):
+                return []
+
+            def fetch_future_candles(self, **kwargs):
+                return []
+
+            def insert_backtest_result(self, **kwargs):
+                self.results.append(kwargs)
+
+            def update_backtest_run_summary(self, backtest_run_id: int, summary: dict) -> None:
+                case.assertEqual(backtest_run_id, 99)
+
+        repository = FakeBacktestRepository()
+        service = BacktestService(repository=repository)
+
+        run_id, outcomes = service.run_backtest(instrument_id=1, limit=1)
+
+        self.assertEqual(run_id, 99)
+        self.assertEqual(len(outcomes), 1)
+        self.assertEqual(repository.results[0]["meta"]["session_codes"], ["OFF_HOURS"])
+        self.assertEqual(repository.results[0]["meta"]["session_filter"], "outside_window")
+        self.assertIn("outside_session_window", repository.results[0]["error_tags"])
+
     def test_run_backtest_persists_replay_mode_and_replay_meta(self) -> None:
         case = self
 
@@ -555,7 +613,7 @@ class BacktestServiceRunTests(unittest.TestCase):
                 self.results = []
                 self.run_config = {}
 
-            def fetch_candidate_signals(self, instrument_id: int, limit: int, timeframe_set=None):
+            def fetch_candidate_signals(self, instrument_id: int, limit: int, timeframe_set=None, session_codes=None):
                 case.assertEqual(instrument_id, 1)
                 case.assertEqual(limit, 10)
                 return [
@@ -737,6 +795,53 @@ class BacktestServiceRunTests(unittest.TestCase):
         self.assertEqual(build_timeframe_bucket(**normalized), "1h/15m/15m")
 
 
+class BacktestRepositoryQueryTests(unittest.TestCase):
+    def test_fetch_candidate_signals_keeps_rows_without_bias_snapshot_under_timeframe_filter(self) -> None:
+        captured = {}
+
+        class StubResult:
+            def mappings(self):
+                return self
+
+            def all(self):
+                return [{"id": 1, "bias_tf": None, "setup_tf": None, "trigger_tf": None}]
+
+        class StubSession:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, exc_type, exc, tb):
+                return False
+
+            def execute(self, sql, params):
+                captured["sql"] = str(sql)
+                captured["params"] = dict(params)
+                return StubResult()
+
+        original_session_local = BacktestRepository.fetch_candidate_signals.__globals__["SessionLocal"]
+        try:
+            BacktestRepository.fetch_candidate_signals.__globals__["SessionLocal"] = lambda: StubSession()
+            repo = BacktestRepository()
+            rows = repo.fetch_candidate_signals(
+                instrument_id=1,
+                limit=5,
+                timeframe_set={"bias_tf": "1m", "setup_tf": "1m", "trigger_tf": "1m"},
+            )
+        finally:
+            BacktestRepository.fetch_candidate_signals.__globals__["SessionLocal"] = original_session_local
+
+        self.assertEqual(rows[0]["id"], 1)
+        self.assertIn("ts.bias_snapshot_id is null or bs.bias_tf = :bias_tf", captured["sql"])
+        self.assertIn("ts.bias_snapshot_id is null or bs.setup_tf = :setup_tf", captured["sql"])
+        self.assertIn("ts.bias_snapshot_id is null or bs.trigger_tf = :trigger_tf", captured["sql"])
+        self.assertIn("ts.status = 'awaiting_confirmation'", captured["sql"])
+        self.assertIn("coalesce((ts.meta->>'execution_permission')::boolean, false) = true", captured["sql"])
+        self.assertIn("not coalesce(ts.invalidations, '[]'::jsonb) ? 'session_mismatch'", captured["sql"])
+        self.assertEqual(captured["params"]["bias_tf"], "1m")
+        self.assertEqual(captured["params"]["setup_tf"], "1m")
+        self.assertEqual(captured["params"]["trigger_tf"], "1m")
+
+
 class BacktestRepositorySerializationTests(unittest.TestCase):
     def test_insert_backtest_result_serializes_datetime_in_meta(self) -> None:
         payload = {

tests/unit/test_broker_read_model_service.py

@@ -0,0 +1,79 @@
+from datetime import datetime, timezone
+import os
+import unittest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
+
+from src.services.broker import (
+    ALERT_UNKNOWN_OPEN_POSITION,
+    ALERT_UNKNOWN_PENDING_ORDER,
+    BROKER_STATE_OK,
+    BROKER_STATE_UNAVAILABLE,
+    BrokerReadModelService,
+)
+
+
+class BrokerReadModelServiceTests(unittest.TestCase):
+    def test_fetch_snapshot_parses_account_positions_orders_and_fills(self) -> None:
+        service = BrokerReadModelService()
+        snapshot = service.fetch_snapshot(
+            instrument_id=1,
+            now=datetime(2026, 5, 7, 1, 0, tzinfo=timezone.utc),
+            account_payload={
+                "code": "0",
+                "data": [{
+                    "totalEq": "1000",
+                    "upl": "20",
+                    "details": [{"availBal": "600", "cashBal": "700"}],
+                }],
+            },
+            positions_payload={
+                "code": "0",
+                "data": [{"instId": "BTC-USDT-SWAP", "posSide": "long", "pos": "2", "avgPx": "100", "markPx": "101", "upl": "5"}],
+            },
+            orders_payload={
+                "code": "0",
+                "data": [{"ordId": "o-1", "instId": "BTC-USDT-SWAP", "side": "buy", "state": "live", "sz": "2", "accFillSz": "0", "px": "100"}],
+            },
+            fills_payload={
+                "code": "0",
+                "data": [{"fillId": "f-1", "ordId": "o-1", "instId": "BTC-USDT-SWAP", "side": "buy", "fillSz": "1", "fillPx": "100", "fillTime": "1715043600000"}],
+            },
+            known_execution_refs={"broker_order_ids": [], "open_execution_count": 0},
+        )
+
+        self.assertEqual(snapshot.status, BROKER_STATE_OK)
+        self.assertEqual(snapshot.account["equity"], 1000.0)
+        self.assertEqual(len(snapshot.positions), 1)
+        self.assertEqual(len(snapshot.orders), 1)
+        self.assertEqual(len(snapshot.fills), 1)
+        self.assertIn(ALERT_UNKNOWN_OPEN_POSITION, snapshot.alerts)
+        self.assertIn(ALERT_UNKNOWN_PENDING_ORDER, snapshot.alerts)
+
+    def test_build_account_risk_state_uses_broker_positions_and_daily_pnl(self) -> None:
+        service = BrokerReadModelService()
+        snapshot = service.fetch_snapshot(
+            instrument_id=1,
+            account_payload={"code": "0", "data": [{"totalEq": "1000", "upl": "-50", "details": [{"availBal": "600", "cashBal": "700"}]}]},
+            positions_payload={"code": "0", "data": [{"instId": "BTC-USDT-SWAP", "posSide": "long", "pos": "2", "avgPx": "100", "markPx": "101", "upl": "5"}]},
+            orders_payload={"code": "0", "data": []},
+            fills_payload={"code": "0", "data": []},
+            known_execution_refs={"broker_order_ids": [], "open_execution_count": 1},
+        )
+
+        state = service.build_account_risk_state(snapshot, instrument_id=1, side="long")
+
+        self.assertEqual(state.open_positions, 1)
+        self.assertEqual(state.same_side_open_positions, 1)
+        self.assertAlmostEqual(state.daily_pnl_pct, -0.05)
+
+    def test_fetch_snapshot_returns_unavailable_when_credentials_or_request_fail(self) -> None:
+        service = BrokerReadModelService()
+        snapshot = service.fetch_snapshot(instrument_id=1)
+
+        self.assertEqual(snapshot.status, BROKER_STATE_UNAVAILABLE)
+        self.assertTrue(snapshot.error)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_documentation_consistency.py

@@ -29,6 +29,7 @@ class DocumentationConsistencyTests(unittest.TestCase):
         self.assertEqual(
             top_level_docs,
             {
+                "AI_ICT_全自动开发执行流程_v1.md",
                 "AI_ICT_实盘交易分析系统开发计划_v1.md",
                 "AI_ICT_实盘辅助决策系统开发方案_v2.md",
                 "ICT交易全书_出版终校版.md",

tests/unit/test_execution_service.py

@@ -1,6 +1,6 @@
 import os
 import unittest
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
 
@@ -477,6 +477,42 @@ class ExecutionServiceTests(unittest.TestCase):
         repository.insert_trade_execution.assert_called_once()
         repository.update_execution_ticket.assert_called_once()
 
+    def test_build_risk_context_for_signal_embeds_broker_state(self) -> None:
+        repository = Mock()
+        repository.fetch_latest_backtest_quality_gate.return_value = {"passed": True}
+        market_state_service = Mock()
+        market_state_service.build_market_state.return_value = {"freshness": {"code": "market_data_fresh"}}
+        market_state_service.repository = repository
+        service = ExecutionService(repository=Mock(), market_state_service=market_state_service)
+
+        from src.services.broker.broker_read_model_service import BrokerReadModelSnapshot
+        with patch("src.services.execution.execution_service.BrokerReadModelService.fetch_snapshot") as fetch_snapshot, patch(
+            "src.services.execution.execution_service.BrokerReadModelService.build_account_risk_state"
+        ) as build_account_state:
+            fetch_snapshot.return_value = BrokerReadModelSnapshot(
+                status="broker_state_ok",
+                provider="okx_read_only",
+                fetched_ts=None,
+                account={"equity": 1000.0, "daily_pnl": -10.0},
+                positions=[{"instrument_id": "BTC-USDT-SWAP", "side": "buy"}],
+                orders=[],
+                fills=[],
+                alerts=[],
+                error=None,
+            )
+            build_account_state.return_value = AccountRiskState(open_positions=1, same_side_open_positions=1)
+            context = service.build_risk_context_for_signal(_valid_signal())
+
+        self.assertEqual(context.open_positions, 1)
+        self.assertEqual(context.same_side_open_positions, 1)
+        self.assertEqual(context.broker_state["status"], "broker_state_ok")
+        fetch_snapshot.assert_called_once()
+        build_account_state.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()
+
     def test_close_execution_by_id_sets_closed_state(self) -> None:
         repository = Mock()
         repository.fetch_execution_by_id.return_value = {"id": 13, "status": "open", "meta": {}, "closed_ts": None}

tests/unit/test_run_until_blocked.py

@@ -0,0 +1,236 @@
+import contextlib
+import io
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import Mock
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
+
+from scripts import run_until_blocked
+
+
+class RunUntilBlockedTests(unittest.TestCase):
+    def test_advances_to_next_stage_and_updates_anchors(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            session_path = Path(tmp) / "SESSION.md"
+            tasks_path = Path(tmp) / "TASKS.md"
+            session_path.write_text("# ai-exchange Session\n\n### `P0-2` 真实账户与订单只读同步\n", encoding="utf-8")
+            tasks_path.write_text("# ai-exchange Tasks\n\n## Done\n\n- [x] `P0-1` 实时市场真值层\n", encoding="utf-8")
+
+            original_session = run_until_blocked.SESSION_PATH
+            original_tasks = run_until_blocked.TASKS_PATH
+            original_runner = run_until_blocked.run_trading_delivery_rehearsal
+            try:
+                run_until_blocked.SESSION_PATH = session_path
+                run_until_blocked.TASKS_PATH = tasks_path
+                run_until_blocked.run_trading_delivery_rehearsal = lambda **kwargs: {
+                    "mode": "rehearsal_completed",
+                    "passed": True,
+                    "continuation_decision": "advance",
+                    "promotion_decision": "promote_candidate",
+                    "blockers": [],
+                }
+
+                summary = run_until_blocked.run_until_blocked(env={"RUN_UNTIL_BLOCKED_MAX_ROUNDS": "1"})
+                session_text = session_path.read_text(encoding="utf-8")
+                tasks_text = tasks_path.read_text(encoding="utf-8")
+            finally:
+                run_until_blocked.SESSION_PATH = original_session
+                run_until_blocked.TASKS_PATH = original_tasks
+                run_until_blocked.run_trading_delivery_rehearsal = original_runner
+
+        self.assertEqual(summary["status"], "stopped")
+        self.assertEqual(summary["completed_stages"], ["P0-1", "P0-2"])
+        self.assertEqual(summary["current_stage"], "P0-3")
+        self.assertIn("### `P0-3` 盘中持续调度与运行监督", session_text)
+        self.assertIn("- [x] `P0-2` 真实账户与订单只读同步", tasks_text)
+        self.assertIn("- [ ] `P0-3` 盘中持续调度与运行监督", tasks_text)
+
+    def test_stops_when_rehearsal_reports_blocked(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            session_path = Path(tmp) / "SESSION.md"
+            tasks_path = Path(tmp) / "TASKS.md"
+            session_path.write_text("# ai-exchange Session\n\n### `P0-2` 真实账户与订单只读同步\n", encoding="utf-8")
+            tasks_path.write_text("# ai-exchange Tasks\n\n## Done\n\n", encoding="utf-8")
+
+            original_session = run_until_blocked.SESSION_PATH
+            original_tasks = run_until_blocked.TASKS_PATH
+            original_runner = run_until_blocked.run_trading_delivery_rehearsal
+            try:
+                run_until_blocked.SESSION_PATH = session_path
+                run_until_blocked.TASKS_PATH = tasks_path
+                run_until_blocked.run_trading_delivery_rehearsal = lambda **kwargs: {
+                    "mode": "autonomous_cycle_failed",
+                    "passed": False,
+                    "continuation_decision": "blocked",
+                    "promotion_decision": "hold",
+                    "blockers": ["缺凭据:broker_readonly_api_key"],
+                }
+
+                summary = run_until_blocked.run_until_blocked(env={"RUN_UNTIL_BLOCKED_MAX_ROUNDS": "2"})
+                tasks_text = tasks_path.read_text(encoding="utf-8")
+                session_text = session_path.read_text(encoding="utf-8")
+            finally:
+                run_until_blocked.SESSION_PATH = original_session
+                run_until_blocked.TASKS_PATH = original_tasks
+                run_until_blocked.run_trading_delivery_rehearsal = original_runner
+
+        self.assertEqual(summary["status"], "blocked")
+        self.assertEqual(summary["blocked_reason"], "缺凭据:broker_readonly_api_key")
+        self.assertEqual(summary["current_stage"], "P0-2")
+        self.assertIn("缺凭据:broker_readonly_api_key", tasks_text)
+        self.assertIn("当前自动化已阻塞：缺凭据:broker_readonly_api_key", session_text)
+
+    def test_json_output_is_machine_readable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            session_path = Path(tmp) / "SESSION.md"
+            tasks_path = Path(tmp) / "TASKS.md"
+            session_path.write_text("# ai-exchange Session\n\n### `P0-2` 真实账户与订单只读同步\n", encoding="utf-8")
+            tasks_path.write_text("# ai-exchange Tasks\n\n## Done\n\n", encoding="utf-8")
+
+            original_session = run_until_blocked.SESSION_PATH
+            original_tasks = run_until_blocked.TASKS_PATH
+            original_runner = run_until_blocked.run_trading_delivery_rehearsal
+            try:
+                run_until_blocked.SESSION_PATH = session_path
+                run_until_blocked.TASKS_PATH = tasks_path
+                run_until_blocked.run_trading_delivery_rehearsal = lambda **kwargs: {
+                    "mode": "autonomous_cycle_failed",
+                    "passed": False,
+                    "continuation_decision": "blocked",
+                    "promotion_decision": "hold",
+                    "blockers": ["缺凭据:broker_readonly_api_key"],
+                }
+                output = io.StringIO()
+                with contextlib.redirect_stdout(output):
+                    summary = run_until_blocked.run_until_blocked(env={"RUN_UNTIL_BLOCKED_OUTPUT_JSON": "1"})
+            finally:
+                run_until_blocked.SESSION_PATH = original_session
+                run_until_blocked.TASKS_PATH = original_tasks
+                run_until_blocked.run_trading_delivery_rehearsal = original_runner
+
+        payload = json.loads(output.getvalue())
+        self.assertEqual(payload["status"], "blocked")
+        self.assertEqual(payload["blocked_reason"], "缺凭据:broker_readonly_api_key")
+        self.assertEqual(summary["status"], "blocked")
+
+    def test_run_until_blocked_writes_runner_status_file(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            session_path = Path(tmp) / "SESSION.md"
+            tasks_path = Path(tmp) / "TASKS.md"
+            status_path = Path(tmp) / "automation_status.json"
+            session_path.write_text("# ai-exchange Session\n\n### `P0-2` 真实账户与订单只读同步\n", encoding="utf-8")
+            tasks_path.write_text("# ai-exchange Tasks\n\n## Done\n\n", encoding="utf-8")
+
+            original_session = run_until_blocked.SESSION_PATH
+            original_tasks = run_until_blocked.TASKS_PATH
+            original_status = run_until_blocked.RUNNER_STATUS_PATH
+            original_runner = run_until_blocked.run_trading_delivery_rehearsal
+            try:
+                run_until_blocked.SESSION_PATH = session_path
+                run_until_blocked.TASKS_PATH = tasks_path
+                run_until_blocked.RUNNER_STATUS_PATH = status_path
+                run_until_blocked.run_trading_delivery_rehearsal = lambda **kwargs: {
+                    "mode": "autonomous_cycle_failed",
+                    "passed": False,
+                    "continuation_decision": "blocked",
+                    "promotion_decision": "hold",
+                    "blockers": ["缺凭据:broker_readonly_api_key"],
+                }
+                run_until_blocked.run_until_blocked(env={})
+                payload = json.loads(status_path.read_text(encoding="utf-8"))
+            finally:
+                run_until_blocked.SESSION_PATH = original_session
+                run_until_blocked.TASKS_PATH = original_tasks
+                run_until_blocked.RUNNER_STATUS_PATH = original_status
+                run_until_blocked.run_trading_delivery_rehearsal = original_runner
+
+        self.assertEqual(payload["state"], "blocked")
+        self.assertEqual(payload["current_stage"], "P0-2")
+        self.assertEqual(payload["blocked_reason"], "缺凭据:broker_readonly_api_key")
+
+    def test_supervisor_pauses_on_code_fix_blocker(self) -> None:
+        sleeper = Mock()
+        original_run_until_blocked = run_until_blocked.run_until_blocked
+        original_fingerprint = run_until_blocked.compute_resume_fingerprint
+        try:
+            run_until_blocked.run_until_blocked = lambda **kwargs: {
+                "status": "blocked",
+                "started_ts": "2026-05-07T00:00:00+00:00",
+                "current_stage": "P0-2",
+                "completed_stages": ["P0-1"],
+                "blocked_reason": "autonomous_mode:pipeline_failed",
+                "rounds": [{"round": 1}],
+            }
+            run_until_blocked.compute_resume_fingerprint = lambda: "same"
+            result = run_until_blocked.run_supervised_until_stopped(
+                env={"RUN_UNTIL_BLOCKED_LOOP_SLEEP_SECONDS": "1", "RUN_UNTIL_BLOCKED_SUPERVISOR_CYCLES": "2"},
+                sleeper=sleeper,
+            )
+        finally:
+            run_until_blocked.run_until_blocked = original_run_until_blocked
+            run_until_blocked.compute_resume_fingerprint = original_fingerprint
+
+        sleeper.assert_called_once_with(1)
+        self.assertEqual(result["status"], "blocked")
+
+    def test_supervisor_resumes_after_fingerprint_change(self) -> None:
+        summaries = [
+            {
+                "status": "blocked",
+                "started_ts": "2026-05-07T00:00:00+00:00",
+                "current_stage": "P0-2",
+                "completed_stages": ["P0-1"],
+                "blocked_reason": "autonomous_mode:pipeline_failed",
+                "rounds": [{"round": 1}],
+            },
+            {
+                "status": "completed",
+                "started_ts": "2026-05-07T00:10:00+00:00",
+                "current_stage": None,
+                "completed_stages": ["P0-1", "P0-2"],
+                "blocked_reason": None,
+                "rounds": [{"round": 1}],
+            },
+        ]
+        fingerprints = ["a", "a", "b", "b"]
+        sleeper = Mock()
+        original_run_until_blocked = run_until_blocked.run_until_blocked
+        original_fingerprint = run_until_blocked.compute_resume_fingerprint
+        try:
+            run_until_blocked.run_until_blocked = lambda **kwargs: summaries.pop(0)
+            run_until_blocked.compute_resume_fingerprint = lambda: fingerprints.pop(0)
+            result = run_until_blocked.run_supervised_until_stopped(
+                env={"RUN_UNTIL_BLOCKED_LOOP_SLEEP_SECONDS": "1", "RUN_UNTIL_BLOCKED_SUPERVISOR_CYCLES": "2"},
+                sleeper=sleeper,
+            )
+        finally:
+            run_until_blocked.run_until_blocked = original_run_until_blocked
+            run_until_blocked.compute_resume_fingerprint = original_fingerprint
+
+        sleeper.assert_called_once_with(1)
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(result["completed_stages"], ["P0-1", "P0-2"])
+
+    def test_main_uses_supervisor_mode_when_enabled(self) -> None:
+        original_run_until_blocked = run_until_blocked.run_until_blocked
+        original_supervisor = run_until_blocked.run_supervised_until_stopped
+        original_environ = dict(os.environ)
+        try:
+            os.environ["RUN_UNTIL_BLOCKED_SUPERVISOR"] = "1"
+            run_until_blocked.run_until_blocked = lambda **kwargs: (_ for _ in ()).throw(AssertionError("should not call one-shot runner"))
+            run_until_blocked.run_supervised_until_stopped = lambda **kwargs: {"status": "completed"}
+            exit_code = run_until_blocked.main()
+        finally:
+            os.environ.clear()
+            os.environ.update(original_environ)
+            run_until_blocked.run_until_blocked = original_run_until_blocked
+            run_until_blocked.run_supervised_until_stopped = original_supervisor
+
+        self.assertEqual(exit_code, 0)
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_signal_service.py

@@ -39,6 +39,20 @@ class SignalSessionFilterTests(unittest.TestCase):
         self.assertFalse(decision["execution_permission"])
         self.assertEqual(decision["reason"], "observation_or_management_session")
 
+    def test_ny_pm_is_downgraded_to_context(self) -> None:
+        decision = classify_session_for_signal("NY_PM")
+
+        self.assertEqual(decision["status"], "awaiting_context")
+        self.assertFalse(decision["execution_permission"])
+        self.assertEqual(decision["reason"], "observation_or_management_session")
+
+    def test_ny_am_stays_allowed_for_confirmation(self) -> None:
+        decision = classify_session_for_signal("NY_AM")
+
+        self.assertEqual(decision["status"], "awaiting_confirmation")
+        self.assertTrue(decision["execution_permission"])
+        self.assertEqual(decision["reason"], "allowed_session")
+
     def test_missing_session_is_off_hours(self) -> None:
         decision = classify_session_for_signal(None)
 
@@ -173,7 +187,8 @@ class SignalRRFilterTests(unittest.TestCase):
         service = SignalService(
             repository=_SignalStubRepository(
                 fvg={"id": 2, "direction": "bullish", "lower": 100.0, "upper": 100.1, "status": "open", "meta": {}},
-            )
+            ),
+            required_evidence_keys=("bias_snapshot_id", "fvg_id", "mss_id"),
         )
 
         candidate = service.build_signal(instrument_id=1, setup_tf="1m", trigger_tf="1m")
@@ -197,6 +212,20 @@ class SignalRRFilterTests(unittest.TestCase):
         self.assertEqual(candidate.target_plan["tp2"]["objective_layer"], "ERL")
         self.assertTrue(candidate.target_plan["management_rules"])
 
+    def test_build_signal_uses_wider_stop_buffer_for_buy_side(self) -> None:
+        service = SignalService(
+            repository=_SignalStubRepository(
+                fvg={"id": 2, "direction": "bullish", "lower": 100.0, "upper": 100.1, "status": "open", "meta": {}},
+            ),
+            required_evidence_keys=("bias_snapshot_id", "fvg_id", "mss_id"),
+        )
+
+        candidate = service.build_signal(instrument_id=1, setup_tf="1m", trigger_tf="1m")
+
+        self.assertIsNotNone(candidate)
+        assert candidate is not None
+        self.assertAlmostEqual(candidate.stop_loss, 99.8)
+
     def test_build_signal_rejects_candidate_when_required_evidence_is_missing(self) -> None:
         service = SignalService(
             repository=_SignalStubRepository(