Phase 3 trade plan candidates

README.md

@@ -218,6 +218,7 @@ Record the current commit as the last-known-good SHA before deploy. If any gate
 - `python3 scripts/run_mss.py` - rebuild MSS structure events
 - `python3 scripts/run_fvg.py` - rebuild FVG zones
 - `python3 scripts/run_trading_plans.py` - generate today's structured BTC/ETH trading plans before signal generation; active plans require a Draw and at least one Invalidation
+- `python3 scripts/run_trade_plan_candidates.py` - watch the active trading plan for the first executable candidate model: OSOK/Daily Range context, Sweep, MSS, and FVG retrace; missing confirmations are reported instead of promoted to execution
 - `python3 scripts/run_signals.py` - generate bias and signals
 - `python3 scripts/run_execution.py` - run paper execution gating, create an execution record, then auto-settle the latest execution for that signal
 - `python3 scripts/run_risk_governor.py` - run the independent market-state and risk governor pre-execution gate
@@ -254,6 +255,7 @@ Optional utility JSON output variables:
 - `SCRIPT_CHECKS_OUTPUT_JSON`
 - `SHOW_MODELS_OUTPUT_JSON`
 - `RISK_GOVERNOR_OUTPUT_JSON`
+- `TRADE_PLAN_CANDIDATES_OUTPUT_JSON`
 - `TRADING_PLANS_OUTPUT_JSON`
 - `TRADING_REHEARSAL_OUTPUT_JSON`
 
@@ -400,6 +402,13 @@ Optional trading plan environment variables used by `run_trading_plans.py`:
 - `TRADING_PLAN_SETUP_TIMEFRAME`
 - `TRADING_PLAN_TRIGGER_TIMEFRAME`
 
+Optional trade plan candidate environment variables used by `run_trade_plan_candidates.py`:
+
+- `TRADE_PLAN_CANDIDATES_OUTPUT_JSON` - set to `1`, `true`, `yes`, or `on` for machine-readable candidate output
+- `TRADE_PLAN_CANDIDATE_SYMBOLS` - comma-separated symbols; defaults to `BTC-USDT,ETH-USDT`
+- `TRADE_PLAN_CANDIDATE_SETUP_TIMEFRAME`
+- `TRADE_PLAN_CANDIDATE_TRIGGER_TIMEFRAME`
+
 ## Lightweight API helpers
 
 The `src/api/` package exposes internal query helpers and HTTP routes for:
@@ -408,6 +417,7 @@ The `src/api/` package exposes internal query helpers and HTTP routes for:
 - `GET /candles?instrument_id=1&timeframe=1m&limit=100`
 - `GET /sessions?instrument_id=1&limit=20`
 - `GET /trading-plans?instrument_id=1&limit=20`
+- `GET /trade-plan-candidates?instrument_id=1&limit=20`
 - `GET /signals?instrument_id=1&limit=20`
 - `GET /executions?instrument_id=1&limit=20`
 - `GET /backtests?instrument_id=1&limit=10`

migrations/022_create_trade_plan_candidates.sql

@@ -0,0 +1,33 @@
+create table if not exists trade_plan_candidates (
+    id bigserial primary key,
+    instrument_id bigint not null references instruments(id),
+    trading_plan_id bigint not null references trading_plans(id),
+    model_code text not null,
+    status text not null,
+    side text,
+    setup_timeframe text not null,
+    trigger_timeframe text not null,
+    created_ts timestamptz not null,
+    expires_ts timestamptz,
+    confirmation_state jsonb not null default '{}'::jsonb,
+    missing_confirmations jsonb not null default '[]'::jsonb,
+    evidence jsonb not null default '{}'::jsonb,
+    entry jsonb not null default '{}'::jsonb,
+    invalidations jsonb not null default '[]'::jsonb,
+    narrative text not null,
+    meta jsonb not null default '{}'::jsonb,
+    created_at timestamptz not null default now(),
+    updated_at timestamptz not null default now()
+);
+
+create index if not exists idx_trade_plan_candidates_instr_created_desc
+on trade_plan_candidates (instrument_id, created_ts desc);
+
+create index if not exists idx_trade_plan_candidates_plan_created_desc
+on trade_plan_candidates (trading_plan_id, created_ts desc);
+
+create index if not exists idx_trade_plan_candidates_instr_status_created_desc
+on trade_plan_candidates (instrument_id, status, created_ts desc);
+
+create index if not exists idx_trade_plan_candidates_instr_model_created_desc
+on trade_plan_candidates (instrument_id, model_code, created_ts desc);

package.json

@@ -3,7 +3,7 @@
   "version": "0.1.0",
   "private": true,
   "scripts": {
-    "lint": ".\\.venv\\Scripts\\python.exe scripts/run_script_static_checks.py && .\\.venv\\Scripts\\python.exe scripts/run_migration_static_checks.py && .\\.venv\\Scripts\\python.exe -m unittest tests.unit.test_trading_plan_service tests.unit.test_market_state_and_risk_governor tests.unit.test_execution_service tests.unit.test_api_server",
+    "lint": ".\\.venv\\Scripts\\python.exe scripts/run_script_static_checks.py && .\\.venv\\Scripts\\python.exe scripts/run_migration_static_checks.py && .\\.venv\\Scripts\\python.exe -m unittest tests.unit.test_trading_plan_service tests.unit.test_market_state_and_risk_governor tests.unit.test_trade_plan_candidate_service tests.unit.test_execution_service tests.unit.test_api_server",
     "build": ".\\.venv\\Scripts\\python.exe -m compileall -q src scripts tests",
     "visual:snapshots": "node --check src/web/app.js"
   }

scripts/run_trade_plan_candidates.py

@@ -0,0 +1,130 @@
+from pathlib import Path
+import json
+import os
+import sys
+
+from sqlalchemy.exc import SQLAlchemyError
+
+ROOT = Path(__file__).resolve().parents[1]
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from scripts.run_trading_plans import DEFAULT_TRADING_PLAN_SYMBOLS
+from src.services.trade_plan_candidates import SetupWatcherService
+
+
+def build_trade_plan_candidate_symbols_from_env(env=None) -> tuple[str, ...]:
+    env = os.environ if env is None else env
+    raw_value = str(
+        env.get("TRADE_PLAN_CANDIDATE_SYMBOLS")
+        or env.get("TRADING_PLAN_SYMBOLS")
+        or ",".join(DEFAULT_TRADING_PLAN_SYMBOLS)
+    )
+    symbols = tuple(item.strip() for item in raw_value.split(",") if item.strip())
+    return symbols or DEFAULT_TRADING_PLAN_SYMBOLS
+
+
+def build_trade_plan_candidate_timeframe_set_from_env(env=None) -> dict:
+    env = os.environ if env is None else env
+    setup_tf = str(env.get("AI_ICT_SETUP_TIMEFRAME") or env.get("TRADE_PLAN_CANDIDATE_SETUP_TIMEFRAME") or "1m")
+    trigger_tf = str(env.get("AI_ICT_TRIGGER_TIMEFRAME") or env.get("TRADE_PLAN_CANDIDATE_TRIGGER_TIMEFRAME") or setup_tf)
+    return {
+        "setup_tf": setup_tf,
+        "trigger_tf": trigger_tf,
+    }
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    output_json = _get_bool(env, "TRADE_PLAN_CANDIDATES_OUTPUT_JSON", False)
+    symbols = build_trade_plan_candidate_symbols_from_env(env=env)
+    timeframe_set = build_trade_plan_candidate_timeframe_set_from_env(env=env)
+    service = SetupWatcherService()
+    candidates = []
+
+    try:
+        for symbol in symbols:
+            instrument = service.repository.fetch_instrument_by_symbol(symbol)
+            if not instrument:
+                candidates.append({"symbol": symbol, "passed": False, "error": "instrument_not_found"})
+                continue
+            candidate = service.build_candidate(
+                instrument_id=int(instrument["id"]),
+                setup_tf=timeframe_set["setup_tf"],
+                trigger_tf=timeframe_set["trigger_tf"],
+            )
+            candidate_id = service.persist_candidate(candidate)
+            candidates.append(
+                build_candidate_result(
+                    symbol=symbol,
+                    instrument_id=int(instrument["id"]),
+                    candidate_id=candidate_id,
+                    candidate=candidate,
+                )
+            )
+    except SQLAlchemyError as exc:
+        payload = {
+            "passed": False,
+            "symbols": list(symbols),
+            "timeframe_set": timeframe_set,
+            "error_type": exc.__class__.__name__,
+            "error": str(exc),
+        }
+        if output_json:
+            print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True))
+        else:
+            print(f"Database unavailable for trade plan candidate sample: {exc.__class__.__name__}")
+        return 2
+
+    payload = {
+        "passed": all(candidate.get("passed") for candidate in candidates),
+        "symbols": list(symbols),
+        "timeframe_set": timeframe_set,
+        "candidates": candidates,
+    }
+    if output_json:
+        print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True))
+    else:
+        for candidate in candidates:
+            print(json.dumps(candidate, default=str, ensure_ascii=False, sort_keys=True))
+    return 0 if payload["passed"] else 2
+
+
+def build_candidate_result(symbol: str, instrument_id: int, candidate_id: int, candidate) -> dict:
+    if candidate is None:
+        return {
+            "symbol": symbol,
+            "instrument_id": instrument_id,
+            "passed": True,
+            "candidate_generated": False,
+            "reason": "active_plan_missing",
+            "candidate_id": 0,
+            "candidate_rows_written": 0,
+        }
+    confirmation_state = dict(candidate.confirmation_state or {})
+    return {
+        "symbol": symbol,
+        "instrument_id": instrument_id,
+        "passed": True,
+        "candidate_generated": True,
+        "candidate_id": candidate_id,
+        "candidate_rows_written": 1 if candidate_id else 0,
+        "trading_plan_id": candidate.trading_plan_id,
+        "model_code": candidate.model_code,
+        "status": candidate.status,
+        "executable": bool(confirmation_state.get("executable")),
+        "missing_confirmations": list(candidate.missing_confirmations),
+        "reason_codes": list(confirmation_state.get("reason_codes") or []),
+    }
+
+
+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"}
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/api/__init__.py

@@ -6,6 +6,7 @@ from src.api.health import get_health
 from src.api.server import route_request, serve_api
 from src.api.sessions import get_recent_sessions
 from src.api.signals import get_recent_signals
+from src.api.trade_plan_candidates import get_recent_trade_plan_candidates
 from src.api.trading_plans import get_recent_trading_plans
 
 __all__ = [
@@ -13,6 +14,7 @@ __all__ = [
     "get_recent_candles",
     "get_recent_sessions",
     "get_recent_signals",
+    "get_recent_trade_plan_candidates",
     "get_recent_trading_plans",
     "get_recent_backtest_summaries",
     "route_request",

src/api/server.py

@@ -10,6 +10,7 @@ from src.api.candles import get_recent_candles
 from src.api.health import get_health
 from src.api.sessions import get_recent_sessions
 from src.api.signals import get_recent_signals
+from src.api.trade_plan_candidates import get_recent_trade_plan_candidates
 from src.api.trading_plans import get_recent_trading_plans
 from src.api.training import (
     add_training_annotation,
@@ -34,6 +35,7 @@ def build_api_dependencies() -> dict:
         "candles": get_recent_candles,
         "sessions": get_recent_sessions,
         "signals": get_recent_signals,
+        "trade_plan_candidates": get_recent_trade_plan_candidates,
         "trading_plans": get_recent_trading_plans,
         "backtests": get_recent_backtest_summaries,
         "executions": execution_service.build_execution_dashboard_items,
@@ -87,6 +89,10 @@ def route_request(
         instrument_id = _require_int(query, "instrument_id")
         limit = _get_int(query, "limit", 20)
         return 200, {"items": deps["signals"](instrument_id=instrument_id, limit=limit)}
+    if method == "GET" and path == "/trade-plan-candidates":
+        instrument_id = _require_int(query, "instrument_id")
+        limit = _get_int(query, "limit", 20)
+        return 200, {"items": deps["trade_plan_candidates"](instrument_id=instrument_id, limit=limit)}
     if method == "GET" and path == "/trading-plans":
         instrument_id = _require_int(query, "instrument_id")
         limit = _get_int(query, "limit", 20)

src/api/trade_plan_candidates.py

@@ -0,0 +1,31 @@
+from sqlalchemy import text
+
+from src.db.session import SessionLocal
+
+
+def get_recent_trade_plan_candidates(instrument_id: int, limit: int = 20) -> list[dict]:
+    sql = text(
+        """
+        select id, instrument_id, trading_plan_id, model_code, status, side, setup_timeframe,
+               trigger_timeframe, created_ts, expires_ts, confirmation_state,
+               missing_confirmations, evidence, entry, invalidations, narrative, meta
+        from trade_plan_candidates
+        where instrument_id = :instrument_id
+        order by created_ts desc, id desc
+        limit :limit
+        """
+    )
+    with SessionLocal() as session:
+        rows = session.execute(sql, {"instrument_id": instrument_id, "limit": limit}).mappings().all()
+    return [normalize_trade_plan_candidate_row(dict(row)) for row in rows]
+
+
+def normalize_trade_plan_candidate_row(row: dict) -> dict:
+    payload = dict(row)
+    confirmation_state = dict(payload.get("confirmation_state") or {})
+    payload["executable"] = bool(confirmation_state.get("executable"))
+    payload["missing_conditions"] = list(confirmation_state.get("missing_conditions") or payload.get("missing_confirmations") or [])
+    payload["present_conditions"] = list(confirmation_state.get("present_conditions") or [])
+    payload["rejected_conditions"] = list(confirmation_state.get("rejected_conditions") or [])
+    payload["reason_codes"] = list(confirmation_state.get("reason_codes") or [])
+    return payload

src/main.py

@@ -2,7 +2,7 @@ from dataclasses import asdict, is_dataclass
 from datetime import datetime
 from typing import Optional
 
-from src.api import get_health, get_recent_backtest_summaries, get_recent_signals
+from src.api import get_health, get_recent_backtest_summaries, get_recent_signals, get_recent_trade_plan_candidates
 from src.services.backtest import (
     BACKTEST_MODE_ENTRY_THEN_OUTCOME,
     BacktestCostConfig,
@@ -24,6 +24,7 @@ from src.services.signals import STRICT_SIGNAL_EVIDENCE_KEYS, SignalService
 from src.services.structure import MSSService
 from src.services.sweeps import SweepService
 from src.services.swings import SwingService
+from src.services.trade_plan_candidates import SetupWatcherService
 
 
 def build_default_runtime_dependencies(signal_required_evidence_keys: tuple[str, ...] = STRICT_SIGNAL_EVIDENCE_KEYS) -> dict:
@@ -38,12 +39,14 @@ def build_default_runtime_dependencies(signal_required_evidence_keys: tuple[str,
         "fvg_service": FVGService(),
         "bias_service": BiasService(),
         "signal_service": SignalService(required_evidence_keys=signal_required_evidence_keys),
+        "setup_watcher_service": SetupWatcherService(),
         "execution_service": ExecutionService(),
         "backtest_service": BacktestService(),
         "pre_market_report_service": PreMarketReportService(),
         "intraday_report_service": IntradayReportService(),
         "post_trade_review_service": PostTradeReviewService(),
         "get_recent_signals": get_recent_signals,
+        "get_recent_trade_plan_candidates": get_recent_trade_plan_candidates,
         "get_recent_backtest_summaries": get_recent_backtest_summaries,
     }
 
@@ -225,8 +228,25 @@ def run_closed_loop(
         candidate=signal,
     )
 
+    trade_plan_candidate = None
+    trade_plan_candidate_id = 0
+    setup_watcher_service = deps.get("setup_watcher_service")
+    if setup_watcher_service is not None:
+        trade_plan_candidate = setup_watcher_service.build_candidate(
+            instrument_id=instrument_id,
+            setup_tf=timeframe_set["setup_tf"],
+            trigger_tf=timeframe_set["trigger_tf"],
+            plan_date=now.date(),
+        )
+        trade_plan_candidate_id = setup_watcher_service.persist_candidate(trade_plan_candidate)
+
     recent_signals = deps["get_recent_signals"](instrument_id=instrument_id, limit=1)
     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,
+    )
+    latest_trade_plan_candidate = recent_trade_plan_candidates[0] if recent_trade_plan_candidates else None
 
     execution_id = None
     execution_record = None
@@ -307,6 +327,8 @@ def run_closed_loop(
             "fvg_updated": fvg_updated,
             "bias_snapshot_id": bias_id,
             "signals": signals_written,
+            "trade_plan_candidate_id": trade_plan_candidate_id,
+            "trade_plan_candidates": 1 if trade_plan_candidate_id else 0,
             "execution_id": execution_id,
             "execution_status": (execution_record or {}).get("status"),
             "execution_lifecycle_state": ((execution_record or {}).get("meta") or {}).get("lifecycle_state"),
@@ -319,6 +341,7 @@ def run_closed_loop(
             "post_trade": post_trade_report,
         },
         "latest_signal": latest_signal,
+        "latest_trade_plan_candidate": latest_trade_plan_candidate,
         "latest_backtest": latest_backtest,
         "backtest_quality_gate": backtest_quality_gate_result,
     }

src/models/__init__.py

@@ -14,6 +14,7 @@ from src.models.structure_event import StructureEvent
 from src.models.sweep_event import SweepEvent
 from src.models.swing import Swing
 from src.models.trade_execution import TradeExecution
+from src.models.trade_plan_candidate import TradePlanCandidate
 from src.models.trade_signal import TradeSignal
 from src.models.trading_plan import TradingPlan
 from src.models.training import (
@@ -41,6 +42,7 @@ __all__ = [
     "SweepEvent",
     "Swing",
     "TradeExecution",
+    "TradePlanCandidate",
     "TradeSignal",
     "TradingPlan",
     "TrainingAnnotation",

src/models/trade_plan_candidate.py

@@ -0,0 +1,29 @@
+from typing import Optional
+
+from sqlalchemy import ForeignKey, Text
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from src.db.base import Base
+
+
+class TradePlanCandidate(Base):
+    __tablename__ = "trade_plan_candidates"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    instrument_id: Mapped[int] = mapped_column(ForeignKey("instruments.id"), nullable=False)
+    trading_plan_id: Mapped[int] = mapped_column(ForeignKey("trading_plans.id"), nullable=False)
+    model_code: Mapped[str] = mapped_column(Text, nullable=False)
+    status: Mapped[str] = mapped_column(Text, nullable=False)
+    side: Mapped[Optional[str]] = mapped_column(Text)
+    setup_timeframe: Mapped[str] = mapped_column(Text, nullable=False)
+    trigger_timeframe: Mapped[str] = mapped_column(Text, nullable=False)
+    created_ts: Mapped[object] = mapped_column(nullable=False)
+    expires_ts: Mapped[Optional[object]] = mapped_column()
+    confirmation_state: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
+    missing_confirmations: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
+    evidence: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
+    entry: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
+    invalidations: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
+    narrative: Mapped[str] = mapped_column(Text, nullable=False)
+    meta: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)

src/repositories/__init__.py

@@ -14,6 +14,7 @@ from src.repositories.signal_repository import SignalRepository
 from src.repositories.structure_repository import StructureRepository
 from src.repositories.swing_repository import SwingRepository
 from src.repositories.sweep_repository import SweepRepository
+from src.repositories.trade_plan_candidate_repository import TradePlanCandidateRepository
 from src.repositories.trading_plan_repository import TradingPlanRepository
 from src.repositories.training_case_repository import TrainingCaseRepository
 
@@ -32,6 +33,7 @@ __all__ = [
     "StructureRepository",
     "SwingRepository",
     "SweepRepository",
+    "TradePlanCandidateRepository",
     "TradingPlanRepository",
     "TrainingCaseRepository",
 ]

src/repositories/trade_plan_candidate_repository.py

@@ -0,0 +1,231 @@
+import json
+from typing import Optional
+
+from sqlalchemy import text
+
+from src.db.session import SessionLocal
+
+
+class TradePlanCandidateRepository:
+    def fetch_instrument_by_symbol(self, symbol: str) -> Optional[dict]:
+        sql = text(
+            """
+            select id, symbol, active
+            from instruments
+            where symbol = :symbol
+              and active = true
+            limit 1
+            """
+        )
+        with SessionLocal() as session:
+            row = session.execute(sql, {"symbol": symbol}).mappings().first()
+        return dict(row) if row else None
+
+    def fetch_active_trading_plan(self, instrument_id: int, plan_date=None) -> Optional[dict]:
+        if plan_date is None:
+            sql = text(
+                """
+                select id, instrument_id, plan_date, status, narrative, primary_draw, secondary_draw,
+                       bias, allowed_sessions, invalidations, no_trade_conditions, inputs, quality,
+                       timeframe_set, created_ts, activated_ts
+                from trading_plans
+                where instrument_id = :instrument_id
+                  and status = 'active'
+                order by plan_date desc, activated_ts desc nulls last, id desc
+                limit 1
+                """
+            )
+            params = {"instrument_id": instrument_id}
+        else:
+            sql = text(
+                """
+                select id, instrument_id, plan_date, status, narrative, primary_draw, secondary_draw,
+                       bias, allowed_sessions, invalidations, no_trade_conditions, inputs, quality,
+                       timeframe_set, created_ts, activated_ts
+                from trading_plans
+                where instrument_id = :instrument_id
+                  and plan_date = :plan_date
+                  and status = 'active'
+                order by activated_ts desc nulls last, id desc
+                limit 1
+                """
+            )
+            params = {"instrument_id": instrument_id, "plan_date": plan_date}
+        with SessionLocal() as session:
+            row = session.execute(sql, params).mappings().first()
+        return dict(row) if row else None
+
+    def fetch_latest_sweep(self, instrument_id: int, timeframe: str, side_swept: Optional[str] = None) -> Optional[dict]:
+        side_filter = "and side_swept = :side_swept" if side_swept else ""
+        sql = text(
+            f"""
+            select id, side_swept, pool_id, sweep_high, sweep_low, sweep_ts, close_back_inside,
+                   validation_level, status, meta
+            from sweep_events
+            where instrument_id = :instrument_id
+              and timeframe = :timeframe
+              {side_filter}
+            order by sweep_ts desc, id desc
+            limit 1
+            """
+        )
+        params = {"instrument_id": instrument_id, "timeframe": timeframe}
+        if side_swept:
+            params["side_swept"] = side_swept
+        with SessionLocal() as session:
+            row = session.execute(sql, params).mappings().first()
+        return dict(row) if row else None
+
+    def fetch_latest_mss(self, instrument_id: int, timeframe: str) -> Optional[dict]:
+        sql = text(
+            """
+            select id, direction, broken_level, accepted, ts, displacement_event_id, meta
+            from structure_events
+            where instrument_id = :instrument_id
+              and timeframe = :timeframe
+              and event_type = 'MSS'
+            order by ts desc, id desc
+            limit 1
+            """
+        )
+        with SessionLocal() as session:
+            row = session.execute(sql, {"instrument_id": instrument_id, "timeframe": timeframe}).mappings().first()
+        return dict(row) if row else None
+
+    def fetch_recent_fvgs(self, instrument_id: int, timeframe: str, limit: int = 20) -> list[dict]:
+        sql = text(
+            """
+            select id, direction, upper, lower, status, fill_pct, related_displacement_id, start_ts, end_ts, meta
+            from fvg_zones
+            where instrument_id = :instrument_id
+              and timeframe = :timeframe
+            order by end_ts desc, id desc
+            limit :limit
+            """
+        )
+        with SessionLocal() as session:
+            rows = session.execute(
+                sql,
+                {"instrument_id": instrument_id, "timeframe": timeframe, "limit": limit},
+            ).mappings().all()
+        return [dict(row) for row in rows]
+
+    def fetch_latest_candidate(self, trading_plan_id: int, model_code: str) -> Optional[dict]:
+        sql = text(
+            """
+            select id, instrument_id, trading_plan_id, model_code, status, side, setup_timeframe,
+                   trigger_timeframe, created_ts, expires_ts, confirmation_state,
+                   missing_confirmations, evidence, entry, invalidations, narrative, meta
+            from trade_plan_candidates
+            where trading_plan_id = :trading_plan_id
+              and model_code = :model_code
+            order by created_ts desc, id desc
+            limit 1
+            """
+        )
+        with SessionLocal() as session:
+            row = session.execute(
+                sql,
+                {"trading_plan_id": trading_plan_id, "model_code": model_code},
+            ).mappings().first()
+        return dict(row) if row else None
+
+    def insert_candidate(
+        self,
+        *,
+        instrument_id: int,
+        trading_plan_id: int,
+        model_code: str,
+        status: str,
+        side: Optional[str],
+        setup_timeframe: str,
+        trigger_timeframe: str,
+        created_ts,
+        expires_ts,
+        confirmation_state: dict,
+        missing_confirmations: list,
+        evidence: dict,
+        entry: dict,
+        invalidations: list,
+        narrative: str,
+        meta: dict,
+    ) -> int:
+        sql = text(
+            """
+            insert into trade_plan_candidates (
+                instrument_id,
+                trading_plan_id,
+                model_code,
+                status,
+                side,
+                setup_timeframe,
+                trigger_timeframe,
+                created_ts,
+                expires_ts,
+                confirmation_state,
+                missing_confirmations,
+                evidence,
+                entry,
+                invalidations,
+                narrative,
+                meta
+            ) values (
+                :instrument_id,
+                :trading_plan_id,
+                :model_code,
+                :status,
+                :side,
+                :setup_timeframe,
+                :trigger_timeframe,
+                :created_ts,
+                :expires_ts,
+                cast(:confirmation_state as jsonb),
+                cast(:missing_confirmations as jsonb),
+                cast(:evidence as jsonb),
+                cast(:entry as jsonb),
+                cast(:invalidations as jsonb),
+                :narrative,
+                cast(:meta as jsonb)
+            )
+            returning id
+            """
+        )
+        with SessionLocal.begin() as session:
+            result = session.execute(
+                sql,
+                {
+                    "instrument_id": instrument_id,
+                    "trading_plan_id": trading_plan_id,
+                    "model_code": model_code,
+                    "status": status,
+                    "side": side,
+                    "setup_timeframe": setup_timeframe,
+                    "trigger_timeframe": trigger_timeframe,
+                    "created_ts": created_ts,
+                    "expires_ts": expires_ts,
+                    "confirmation_state": json.dumps(confirmation_state, default=str),
+                    "missing_confirmations": json.dumps(missing_confirmations, default=str),
+                    "evidence": json.dumps(evidence, default=str),
+                    "entry": json.dumps(entry, default=str),
+                    "invalidations": json.dumps(invalidations, default=str),
+                    "narrative": narrative,
+                    "meta": json.dumps(meta, default=str),
+                },
+            )
+            return int(result.scalar_one())
+
+    def fetch_recent_candidates(self, instrument_id: int, limit: int = 20) -> list[dict]:
+        sql = text(
+            """
+            select id, instrument_id, trading_plan_id, model_code, status, side, setup_timeframe,
+                   trigger_timeframe, created_ts, expires_ts, confirmation_state,
+                   missing_confirmations, evidence, entry, invalidations, narrative, meta
+            from trade_plan_candidates
+            where instrument_id = :instrument_id
+            order by created_ts desc, id desc
+            limit :limit
+            """
+        )
+        with SessionLocal() as session:
+            rows = session.execute(sql, {"instrument_id": instrument_id, "limit": limit}).mappings().all()
+        return [dict(row) for row in rows]

src/services/trade_plan_candidates/__init__.py

@@ -0,0 +1,41 @@
+"""Trade plan candidate service package."""
+
+from src.services.trade_plan_candidates.confirmation_service import (
+    AWAITING_CONFIRMATION_STATUS,
+    BLOCKED_CANDIDATE_STATUS,
+    CONDITION_FVG_RETRACE,
+    CONDITION_MSS,
+    CONDITION_OSOK_OR_DAILY_RANGE,
+    CONDITION_SWEEP,
+    EXECUTABLE_CANDIDATE_STATUS,
+    MODEL_OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE,
+    REQUIRED_CONFIRMATION_CONDITIONS,
+    ConfirmationService,
+    ConfirmationState,
+)
+from src.services.trade_plan_candidates.setup_watcher_service import (
+    SetupWatcherService,
+    TradePlanCandidateDraft,
+    build_trade_plan_candidate_from_inputs,
+    select_fvg_for_candidate,
+    trade_plan_candidate_matches,
+)
+
+__all__ = [
+    "AWAITING_CONFIRMATION_STATUS",
+    "BLOCKED_CANDIDATE_STATUS",
+    "CONDITION_FVG_RETRACE",
+    "CONDITION_MSS",
+    "CONDITION_OSOK_OR_DAILY_RANGE",
+    "CONDITION_SWEEP",
+    "EXECUTABLE_CANDIDATE_STATUS",
+    "MODEL_OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE",
+    "REQUIRED_CONFIRMATION_CONDITIONS",
+    "ConfirmationService",
+    "ConfirmationState",
+    "SetupWatcherService",
+    "TradePlanCandidateDraft",
+    "build_trade_plan_candidate_from_inputs",
+    "select_fvg_for_candidate",
+    "trade_plan_candidate_matches",
+]

src/services/trade_plan_candidates/confirmation_service.py

@@ -0,0 +1,250 @@
+from dataclasses import dataclass
+from typing import Optional
+
+
+MODEL_OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE = "OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE"
+CONDITION_OSOK_OR_DAILY_RANGE = "osok_or_daily_range_context"
+CONDITION_SWEEP = "sweep_confirmed"
+CONDITION_MSS = "mss_confirmed"
+CONDITION_FVG_RETRACE = "fvg_retrace_confirmed"
+REQUIRED_CONFIRMATION_CONDITIONS = (
+    CONDITION_OSOK_OR_DAILY_RANGE,
+    CONDITION_SWEEP,
+    CONDITION_MSS,
+    CONDITION_FVG_RETRACE,
+)
+EXECUTABLE_CANDIDATE_STATUS = "executable"
+AWAITING_CONFIRMATION_STATUS = "awaiting_confirmation"
+BLOCKED_CANDIDATE_STATUS = "blocked"
+
+
+@dataclass(frozen=True)
+class ConfirmationState:
+    model_code: str
+    status: str
+    executable: bool
+    side: Optional[str]
+    required_conditions: list[str]
+    present_conditions: list[str]
+    missing_conditions: list[str]
+    rejected_conditions: list[str]
+    reason_codes: list[str]
+    evidence: dict
+
+    def to_dict(self) -> dict:
+        return {
+            "model_code": self.model_code,
+            "status": self.status,
+            "executable": self.executable,
+            "side": self.side,
+            "required_conditions": self.required_conditions,
+            "present_conditions": self.present_conditions,
+            "missing_conditions": self.missing_conditions,
+            "rejected_conditions": self.rejected_conditions,
+            "reason_codes": self.reason_codes,
+            "evidence": self.evidence,
+        }
+
+
+class ConfirmationService:
+    def evaluate(
+        self,
+        *,
+        trading_plan: dict,
+        sweep: Optional[dict],
+        mss: Optional[dict],
+        fvg: Optional[dict],
+    ) -> ConfirmationState:
+        side = resolve_plan_side(trading_plan)
+        expected_sweep_side = expected_sweep_side_for_trade_side(side)
+        expected_mss_direction = expected_mss_direction_for_trade_side(side)
+        present_conditions: list[str] = []
+        missing_conditions: list[str] = []
+        rejected_conditions: list[str] = []
+        reason_codes: list[str] = []
+
+        if has_osok_or_daily_range_context(trading_plan):
+            present_conditions.append(CONDITION_OSOK_OR_DAILY_RANGE)
+        else:
+            missing_conditions.append(CONDITION_OSOK_OR_DAILY_RANGE)
+            reason_codes.append("missing_osok_or_daily_range_context")
+
+        if sweep is None:
+            missing_conditions.append(CONDITION_SWEEP)
+            reason_codes.append("missing_sweep")
+        elif not sweep_side_matches(sweep, expected_sweep_side):
+            missing_conditions.append(CONDITION_SWEEP)
+            rejected_conditions.append("sweep_side_mismatch")
+            reason_codes.append("sweep_side_mismatch")
+        elif not sweep_is_confirmed(sweep):
+            missing_conditions.append(CONDITION_SWEEP)
+            reason_codes.append("sweep_not_confirmed")
+        else:
+            present_conditions.append(CONDITION_SWEEP)
+
+        if mss is None:
+            missing_conditions.append(CONDITION_MSS)
+            reason_codes.append("missing_mss")
+        elif not mss_direction_matches(mss, expected_mss_direction):
+            missing_conditions.append(CONDITION_MSS)
+            rejected_conditions.append("mss_direction_mismatch")
+            reason_codes.append("mss_direction_mismatch")
+        elif not mss_is_confirmed(mss):
+            missing_conditions.append(CONDITION_MSS)
+            rejected_conditions.append("mss_not_accepted")
+            reason_codes.append("mss_not_accepted")
+        else:
+            present_conditions.append(CONDITION_MSS)
+
+        if fvg is None:
+            missing_conditions.append(CONDITION_FVG_RETRACE)
+            reason_codes.append("missing_fvg")
+        elif not fvg_direction_matches(fvg, expected_mss_direction):
+            missing_conditions.append(CONDITION_FVG_RETRACE)
+            rejected_conditions.append("fvg_direction_mismatch")
+            reason_codes.append("fvg_direction_mismatch")
+        elif fvg_is_invalidated_or_mitigated(fvg):
+            missing_conditions.append(CONDITION_FVG_RETRACE)
+            rejected_conditions.append("fvg_not_actionable")
+            reason_codes.append("fvg_not_actionable")
+        elif not fvg_is_retraced(fvg):
+            missing_conditions.append(CONDITION_FVG_RETRACE)
+            reason_codes.append("fvg_not_retraced")
+        else:
+            present_conditions.append(CONDITION_FVG_RETRACE)
+
+        present_conditions = _dedupe(present_conditions)
+        missing_conditions = _dedupe(missing_conditions)
+        rejected_conditions = _dedupe(rejected_conditions)
+        reason_codes = _dedupe(reason_codes)
+        executable = not missing_conditions and not rejected_conditions
+        status = EXECUTABLE_CANDIDATE_STATUS if executable else AWAITING_CONFIRMATION_STATUS
+        if rejected_conditions:
+            status = BLOCKED_CANDIDATE_STATUS
+
+        return ConfirmationState(
+            model_code=MODEL_OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE,
+            status=status,
+            executable=executable,
+            side=side,
+            required_conditions=list(REQUIRED_CONFIRMATION_CONDITIONS),
+            present_conditions=present_conditions,
+            missing_conditions=missing_conditions,
+            rejected_conditions=rejected_conditions,
+            reason_codes=reason_codes,
+            evidence=build_confirmation_evidence(
+                trading_plan=trading_plan,
+                sweep=sweep,
+                mss=mss,
+                fvg=fvg,
+            ),
+        )
+
+
+def resolve_plan_side(trading_plan: dict) -> Optional[str]:
+    bias = str(trading_plan.get("bias") or "").strip().lower()
+    if bias == "bullish":
+        return "buy"
+    if bias == "bearish":
+        return "sell"
+    draw_side = str((trading_plan.get("primary_draw") or {}).get("side") or "").strip().lower()
+    if draw_side == "buy_side":
+        return "buy"
+    if draw_side == "sell_side":
+        return "sell"
+    return None
+
+
+def expected_sweep_side_for_trade_side(side: Optional[str]) -> Optional[str]:
+    if side == "buy":
+        return "sell_side"
+    if side == "sell":
+        return "buy_side"
+    return None
+
+
+def expected_mss_direction_for_trade_side(side: Optional[str]) -> Optional[str]:
+    if side == "buy":
+        return "bullish"
+    if side == "sell":
+        return "bearish"
+    return None
+
+
+def has_osok_or_daily_range_context(trading_plan: dict) -> bool:
+    primary_draw = dict(trading_plan.get("primary_draw") or {})
+    inputs = dict(trading_plan.get("inputs") or {})
+    draw_source = primary_draw.get("source")
+    return bool(
+        draw_source in {"liquidity_pool", "dealing_range_boundary"}
+        or primary_draw.get("pool_id")
+        or primary_draw.get("range_id")
+        or inputs.get("range_id")
+    )
+
+
+def sweep_side_matches(sweep: dict, expected_sweep_side: Optional[str]) -> bool:
+    return expected_sweep_side is not None and sweep.get("side_swept") == expected_sweep_side
+
+
+def sweep_is_confirmed(sweep: dict) -> bool:
+    return (
+        bool(sweep.get("close_back_inside"))
+        and str(sweep.get("validation_level") or "") == "validated"
+        and str(sweep.get("status") or "") in {"confirmed", "validated", "swept"}
+    )
+
+
+def mss_direction_matches(mss: dict, expected_direction: Optional[str]) -> bool:
+    return expected_direction is not None and mss.get("direction") == expected_direction
+
+
+def mss_is_confirmed(mss: dict) -> bool:
+    meta = dict(mss.get("meta") or {})
+    state = meta.get("state")
+    if state == "invalidated":
+        return False
+    return bool(mss.get("accepted"))
+
+
+def fvg_direction_matches(fvg: dict, expected_direction: Optional[str]) -> bool:
+    return expected_direction is not None and fvg.get("direction") == expected_direction
+
+
+def fvg_is_invalidated_or_mitigated(fvg: dict) -> bool:
+    meta = dict(fvg.get("meta") or {})
+    return bool(meta.get("invalidated")) or str(fvg.get("status") or "") in {"invalidated", "mitigated"}
+
+
+def fvg_is_retraced(fvg: dict) -> bool:
+    meta = dict(fvg.get("meta") or {})
+    return float(fvg.get("fill_pct") or 0.0) > 0.0 or int(meta.get("touch_count") or 0) > 0 or fvg.get("status") == "touched"
+
+
+def build_confirmation_evidence(
+    *,
+    trading_plan: dict,
+    sweep: Optional[dict],
+    mss: Optional[dict],
+    fvg: Optional[dict],
+) -> dict:
+    return {
+        "trading_plan_id": trading_plan.get("id"),
+        "sweep_id": sweep.get("id") if sweep else None,
+        "sweep_pool_id": sweep.get("pool_id") if sweep else None,
+        "mss_id": mss.get("id") if mss else None,
+        "mss_displacement_event_id": mss.get("displacement_event_id") if mss else None,
+        "fvg_id": fvg.get("id") if fvg else None,
+        "fvg_related_displacement_id": fvg.get("related_displacement_id") if fvg else None,
+    }
+
+
+def _dedupe(values: list[str]) -> list[str]:
+    seen = set()
+    result = []
+    for value in values:
+        if value in seen:
+            continue
+        seen.add(value)
+        result.append(value)
+    return result

src/services/trade_plan_candidates/setup_watcher_service.py

@@ -0,0 +1,271 @@
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+import math
+from typing import Optional
+
+from src.repositories.trade_plan_candidate_repository import TradePlanCandidateRepository
+from src.services.trade_plan_candidates.confirmation_service import (
+    ConfirmationService,
+    ConfirmationState,
+    MODEL_OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE,
+    expected_mss_direction_for_trade_side,
+    expected_sweep_side_for_trade_side,
+    resolve_plan_side,
+)
+
+
+@dataclass(frozen=True)
+class TradePlanCandidateDraft:
+    instrument_id: int
+    trading_plan_id: int
+    model_code: str
+    status: str
+    side: Optional[str]
+    setup_timeframe: str
+    trigger_timeframe: str
+    created_ts: object
+    expires_ts: Optional[object]
+    confirmation_state: dict
+    missing_confirmations: list[str]
+    evidence: dict
+    entry: dict
+    invalidations: list
+    narrative: str
+    meta: dict
+
+    def to_dict(self) -> dict:
+        return {
+            "instrument_id": self.instrument_id,
+            "trading_plan_id": self.trading_plan_id,
+            "model_code": self.model_code,
+            "status": self.status,
+            "side": self.side,
+            "setup_timeframe": self.setup_timeframe,
+            "trigger_timeframe": self.trigger_timeframe,
+            "created_ts": self.created_ts,
+            "expires_ts": self.expires_ts,
+            "confirmation_state": self.confirmation_state,
+            "missing_confirmations": self.missing_confirmations,
+            "evidence": self.evidence,
+            "entry": self.entry,
+            "invalidations": self.invalidations,
+            "narrative": self.narrative,
+            "meta": self.meta,
+        }
+
+
+class SetupWatcherService:
+    def __init__(
+        self,
+        repository: Optional[TradePlanCandidateRepository] = None,
+        confirmation_service: Optional[ConfirmationService] = None,
+    ) -> None:
+        self.repository = repository or TradePlanCandidateRepository()
+        self.confirmation_service = confirmation_service or ConfirmationService()
+
+    def build_candidate(
+        self,
+        *,
+        instrument_id: int,
+        setup_tf: str = "1m",
+        trigger_tf: str = "1m",
+        plan_date=None,
+        observed_ts=None,
+    ) -> Optional[TradePlanCandidateDraft]:
+        created_ts = observed_ts or datetime.now(timezone.utc)
+        trading_plan = self.repository.fetch_active_trading_plan(instrument_id=instrument_id, plan_date=plan_date)
+        if trading_plan is None:
+            return None
+
+        side = resolve_plan_side(trading_plan)
+        sweep = self.repository.fetch_latest_sweep(
+            instrument_id=instrument_id,
+            timeframe=setup_tf,
+            side_swept=expected_sweep_side_for_trade_side(side),
+        )
+        mss = self.repository.fetch_latest_mss(instrument_id=instrument_id, timeframe=setup_tf)
+        fvgs = self.repository.fetch_recent_fvgs(instrument_id=instrument_id, timeframe=trigger_tf, limit=20)
+        fvg = select_fvg_for_candidate(fvgs=fvgs, mss=mss, side=side)
+        confirmation = self.confirmation_service.evaluate(
+            trading_plan=trading_plan,
+            sweep=sweep,
+            mss=mss,
+            fvg=fvg,
+        )
+
+        return build_trade_plan_candidate_from_inputs(
+            trading_plan=trading_plan,
+            setup_tf=setup_tf,
+            trigger_tf=trigger_tf,
+            sweep=sweep,
+            mss=mss,
+            fvg=fvg,
+            confirmation=confirmation,
+            created_ts=created_ts,
+        )
+
+    def persist_candidate(self, candidate: Optional[TradePlanCandidateDraft]) -> int:
+        if candidate is None:
+            return 0
+        latest = self.repository.fetch_latest_candidate(
+            trading_plan_id=candidate.trading_plan_id,
+            model_code=candidate.model_code,
+        )
+        if latest and trade_plan_candidate_matches(candidate=candidate, row=latest):
+            return 0
+        return self.repository.insert_candidate(**candidate.to_dict())
+
+
+def build_trade_plan_candidate_from_inputs(
+    *,
+    trading_plan: dict,
+    setup_tf: str,
+    trigger_tf: str,
+    sweep: Optional[dict],
+    mss: Optional[dict],
+    fvg: Optional[dict],
+    confirmation: ConfirmationState,
+    created_ts=None,
+) -> TradePlanCandidateDraft:
+    created_ts = created_ts or datetime.now(timezone.utc)
+    expires_ts = created_ts + timedelta(hours=4)
+    side = confirmation.side
+    status = confirmation.status
+    confirmation_state = confirmation.to_dict()
+    evidence = {
+        **confirmation.evidence,
+        "primary_draw": trading_plan.get("primary_draw") or {},
+        "timeframe_set": {
+            "setup_tf": setup_tf,
+            "trigger_tf": trigger_tf,
+        },
+    }
+    entry = build_candidate_entry(side=side, mss=mss, fvg=fvg)
+    invalidations = build_candidate_invalidations(trading_plan=trading_plan, sweep=sweep, mss=mss, fvg=fvg)
+    missing_confirmations = list(confirmation.missing_conditions)
+    return TradePlanCandidateDraft(
+        instrument_id=int(trading_plan["instrument_id"]),
+        trading_plan_id=int(trading_plan["id"]),
+        model_code=MODEL_OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE,
+        status=status,
+        side=side,
+        setup_timeframe=setup_tf,
+        trigger_timeframe=trigger_tf,
+        created_ts=created_ts,
+        expires_ts=expires_ts,
+        confirmation_state=confirmation_state,
+        missing_confirmations=missing_confirmations,
+        evidence=evidence,
+        entry=entry,
+        invalidations=invalidations,
+        narrative=build_candidate_narrative(
+            trading_plan_id=trading_plan.get("id"),
+            side=side,
+            status=status,
+            missing_confirmations=missing_confirmations,
+        ),
+        meta={
+            "source": "SetupWatcherService",
+            "executable": confirmation.executable,
+            "reason_codes": confirmation.reason_codes,
+            "rejected_conditions": confirmation.rejected_conditions,
+        },
+    )
+
+
+def select_fvg_for_candidate(fvgs: list[dict], mss: Optional[dict], side: Optional[str]) -> Optional[dict]:
+    if not fvgs:
+        return None
+    expected_direction = expected_mss_direction_for_trade_side(side)
+    direction_matches = [row for row in fvgs if row.get("direction") == expected_direction]
+    if not direction_matches:
+        return fvgs[0]
+    displacement_id = int((mss or {}).get("displacement_event_id") or 0)
+    if displacement_id:
+        exact_matches = [
+            row
+            for row in direction_matches
+            if int(row.get("related_displacement_id") or 0) == displacement_id
+        ]
+        if exact_matches:
+            return exact_matches[0]
+    return direction_matches[0]
+
+
+def build_candidate_entry(side: Optional[str], mss: Optional[dict], fvg: Optional[dict]) -> dict:
+    if side not in {"buy", "sell"} or fvg is None:
+        return {}
+    entry_low = _float_or_none(fvg.get("lower"))
+    entry_high = _float_or_none(fvg.get("upper"))
+    entry = {
+        "entry_type": "limit",
+        "entry_low": entry_low,
+        "entry_high": entry_high,
+    }
+    if mss is None or entry_low is None or entry_high is None:
+        return entry
+    broken_level = _float_or_none(mss.get("broken_level"))
+    if broken_level is None:
+        return entry
+    if side == "buy":
+        entry["stop_loss"] = min(broken_level, entry_low) * 0.999
+    else:
+        entry["stop_loss"] = max(broken_level, entry_high) * 1.001
+    return entry
+
+
+def build_candidate_invalidations(
+    *,
+    trading_plan: dict,
+    sweep: Optional[dict],
+    mss: Optional[dict],
+    fvg: Optional[dict],
+) -> list:
+    invalidations = list(trading_plan.get("invalidations") or [])
+    invalidations.extend(
+        [
+            {"code": "active_trading_plan_invalidated", "source": "trading_plan", "trading_plan_id": trading_plan.get("id")},
+            {"code": "sweep_invalidated", "source": "sweep_event", "sweep_id": sweep.get("id") if sweep else None},
+            {"code": "mss_invalidated", "source": "structure_event", "mss_id": mss.get("id") if mss else None},
+            {"code": "fvg_mitigated_or_invalidated", "source": "fvg_zone", "fvg_id": fvg.get("id") if fvg else None},
+        ]
+    )
+    return invalidations
+
+
+def build_candidate_narrative(
+    *,
+    trading_plan_id,
+    side: Optional[str],
+    status: str,
+    missing_confirmations: list[str],
+) -> str:
+    missing = ",".join(missing_confirmations) if missing_confirmations else "none"
+    return (
+        f"{MODEL_OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE}:"
+        f"plan={trading_plan_id}:side={side or 'unknown'}:status={status}:missing={missing}"
+    )
+
+
+def trade_plan_candidate_matches(candidate: TradePlanCandidateDraft, row: dict) -> bool:
+    return (
+        int(row.get("trading_plan_id") or 0) == candidate.trading_plan_id
+        and row.get("model_code") == candidate.model_code
+        and row.get("status") == candidate.status
+        and row.get("side") == candidate.side
+        and row.get("setup_timeframe") == candidate.setup_timeframe
+        and row.get("trigger_timeframe") == candidate.trigger_timeframe
+        and dict(row.get("confirmation_state") or {}) == dict(candidate.confirmation_state)
+        and list(row.get("missing_confirmations") or []) == list(candidate.missing_confirmations)
+        and dict(row.get("evidence") or {}) == dict(candidate.evidence)
+        and dict(row.get("entry") or {}) == dict(candidate.entry)
+    )
+
+
+def _float_or_none(value) -> Optional[float]:
+    if value is None:
+        return None
+    parsed = float(value)
+    if math.isnan(parsed):
+        return None
+    return parsed

tests/unit/test_trade_plan_candidate_service.py

@@ -0,0 +1,292 @@
+from datetime import date, datetime, timezone
+import os
+import unittest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
+
+from scripts import run_trade_plan_candidates
+from src.api import route_request
+from src.services.trade_plan_candidates import (
+    AWAITING_CONFIRMATION_STATUS,
+    CONDITION_FVG_RETRACE,
+    CONDITION_MSS,
+    CONDITION_SWEEP,
+    EXECUTABLE_CANDIDATE_STATUS,
+    MODEL_OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE,
+    SetupWatcherService,
+    trade_plan_candidate_matches,
+)
+
+
+class SetupWatcherServiceTests(unittest.TestCase):
+    def test_does_not_generate_candidate_without_active_trading_plan(self) -> None:
+        service = SetupWatcherService(repository=_SetupWatcherStubRepository(active_plan=None))
+
+        candidate = service.build_candidate(instrument_id=1, setup_tf="15m", trigger_tf="5m", plan_date=date(2026, 5, 6))
+
+        self.assertIsNone(candidate)
+
+    def test_generates_executable_candidate_when_first_model_confirmations_are_present(self) -> None:
+        service = SetupWatcherService(repository=_SetupWatcherStubRepository())
+
+        candidate = service.build_candidate(
+            instrument_id=1,
+            setup_tf="15m",
+            trigger_tf="5m",
+            plan_date=date(2026, 5, 6),
+            observed_ts=_created_ts(),
+        )
+
+        self.assertIsNotNone(candidate)
+        assert candidate is not None
+        self.assertEqual(candidate.model_code, MODEL_OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE)
+        self.assertEqual(candidate.trading_plan_id, 10)
+        self.assertEqual(candidate.status, EXECUTABLE_CANDIDATE_STATUS)
+        self.assertEqual(candidate.side, "buy")
+        self.assertEqual(candidate.missing_confirmations, [])
+        self.assertTrue(candidate.confirmation_state["executable"])
+        self.assertEqual(candidate.confirmation_state["present_conditions"], [
+            "osok_or_daily_range_context",
+            "sweep_confirmed",
+            "mss_confirmed",
+            "fvg_retrace_confirmed",
+        ])
+        self.assertEqual(candidate.evidence["sweep_id"], 20)
+        self.assertEqual(candidate.evidence["mss_id"], 30)
+        self.assertEqual(candidate.evidence["fvg_id"], 40)
+        self.assertEqual(candidate.entry["entry_type"], "limit")
+        self.assertEqual(candidate.entry["entry_low"], 100.0)
+        self.assertEqual(candidate.entry["entry_high"], 100.2)
+        self.assertIn("active_trading_plan_invalidated", [item["code"] for item in candidate.invalidations])
+
+    def test_missing_sweep_mss_and_fvg_do_not_produce_executable_candidate(self) -> None:
+        service = SetupWatcherService(
+            repository=_SetupWatcherStubRepository(
+                sweep=None,
+                mss=None,
+                fvgs=[],
+            )
+        )
+
+        candidate = service.build_candidate(
+            instrument_id=1,
+            setup_tf="15m",
+            trigger_tf="5m",
+            plan_date=date(2026, 5, 6),
+            observed_ts=_created_ts(),
+        )
+
+        self.assertIsNotNone(candidate)
+        assert candidate is not None
+        self.assertEqual(candidate.status, AWAITING_CONFIRMATION_STATUS)
+        self.assertFalse(candidate.confirmation_state["executable"])
+        self.assertIn(CONDITION_SWEEP, candidate.missing_confirmations)
+        self.assertIn(CONDITION_MSS, candidate.missing_confirmations)
+        self.assertIn(CONDITION_FVG_RETRACE, candidate.missing_confirmations)
+        self.assertIn("missing_sweep", candidate.confirmation_state["reason_codes"])
+        self.assertIn("missing_mss", candidate.confirmation_state["reason_codes"])
+        self.assertIn("missing_fvg", candidate.confirmation_state["reason_codes"])
+
+    def test_candidate_explains_unretraced_fvg_as_missing_confirmation(self) -> None:
+        service = SetupWatcherService(
+            repository=_SetupWatcherStubRepository(
+                fvgs=[_fvg_row(status="open", fill_pct=0.0, meta={"touch_count": 0})],
+            )
+        )
+
+        candidate = service.build_candidate(
+            instrument_id=1,
+            setup_tf="15m",
+            trigger_tf="5m",
+            plan_date=date(2026, 5, 6),
+            observed_ts=_created_ts(),
+        )
+
+        self.assertIsNotNone(candidate)
+        assert candidate is not None
+        self.assertEqual(candidate.status, AWAITING_CONFIRMATION_STATUS)
+        self.assertIn(CONDITION_FVG_RETRACE, candidate.missing_confirmations)
+        self.assertIn("fvg_not_retraced", candidate.confirmation_state["reason_codes"])
+        self.assertIn("missing=fvg_retrace_confirmed", candidate.narrative)
+
+    def test_persist_candidate_is_idempotent_when_latest_candidate_matches(self) -> None:
+        repository = _SetupWatcherStubRepository()
+        service = SetupWatcherService(repository=repository)
+        candidate = service.build_candidate(
+            instrument_id=1,
+            setup_tf="15m",
+            trigger_tf="5m",
+            plan_date=date(2026, 5, 6),
+            observed_ts=_created_ts(),
+        )
+        self.assertIsNotNone(candidate)
+        assert candidate is not None
+        repository.latest_candidate = candidate.to_dict()
+
+        written = service.persist_candidate(candidate)
+
+        self.assertEqual(written, 0)
+        self.assertEqual(repository.inserted_candidates, [])
+
+
+class TradePlanCandidateMatchingTests(unittest.TestCase):
+    def test_trade_plan_candidate_matches_equivalent_latest_row(self) -> None:
+        service = SetupWatcherService(repository=_SetupWatcherStubRepository())
+        candidate = service.build_candidate(
+            instrument_id=1,
+            setup_tf="15m",
+            trigger_tf="5m",
+            plan_date=date(2026, 5, 6),
+            observed_ts=_created_ts(),
+        )
+        self.assertIsNotNone(candidate)
+        assert candidate is not None
+
+        self.assertTrue(trade_plan_candidate_matches(candidate=candidate, row=candidate.to_dict()))
+
+
+class TradePlanCandidateScriptConfigTests(unittest.TestCase):
+    def test_symbol_env_defaults_to_btc_eth(self) -> None:
+        self.assertEqual(
+            run_trade_plan_candidates.build_trade_plan_candidate_symbols_from_env({}),
+            ("BTC-USDT", "ETH-USDT"),
+        )
+
+    def test_timeframe_env_reads_overrides(self) -> None:
+        self.assertEqual(
+            run_trade_plan_candidates.build_trade_plan_candidate_timeframe_set_from_env(
+                {
+                    "AI_ICT_SETUP_TIMEFRAME": "15m",
+                    "AI_ICT_TRIGGER_TIMEFRAME": "5m",
+                }
+            ),
+            {"setup_tf": "15m", "trigger_tf": "5m"},
+        )
+
+
+class TradePlanCandidateApiTests(unittest.TestCase):
+    def test_routes_trade_plan_candidates_query(self) -> None:
+        status, payload = route_request(
+            "/trade-plan-candidates",
+            {"instrument_id": ["1"], "limit": ["2"]},
+            dependencies={
+                "trade_plan_candidates": lambda **kwargs: [{"id": 1, "status": "executable"}],
+            },
+        )
+
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["items"], [{"id": 1, "status": "executable"}])
+
+
+_DEFAULT = object()
+
+
+class _SetupWatcherStubRepository:
+    def __init__(
+        self,
+        *,
+        active_plan=_DEFAULT,
+        sweep=_DEFAULT,
+        mss=_DEFAULT,
+        fvgs=_DEFAULT,
+    ) -> None:
+        self.active_plan = _active_plan() if active_plan is _DEFAULT else active_plan
+        self.sweep = _sweep_row() if sweep is _DEFAULT else sweep
+        self.mss = _mss_row() if mss is _DEFAULT else mss
+        self.fvgs = [_fvg_row()] if fvgs is _DEFAULT else fvgs
+        self.latest_candidate = None
+        self.inserted_candidates = []
+
+    def fetch_active_trading_plan(self, instrument_id: int, plan_date=None):
+        return self.active_plan
+
+    def fetch_latest_sweep(self, instrument_id: int, timeframe: str, side_swept=None):
+        if self.sweep and side_swept and self.sweep.get("side_swept") != side_swept:
+            return None
+        return self.sweep
+
+    def fetch_latest_mss(self, instrument_id: int, timeframe: str):
+        return self.mss
+
+    def fetch_recent_fvgs(self, instrument_id: int, timeframe: str, limit: int = 20):
+        return self.fvgs
+
+    def fetch_latest_candidate(self, trading_plan_id: int, model_code: str):
+        return self.latest_candidate
+
+    def insert_candidate(self, **kwargs):
+        self.inserted_candidates.append(kwargs)
+        return 99
+
+
+def _created_ts() -> datetime:
+    return datetime(2026, 5, 6, 0, 0, tzinfo=timezone.utc)
+
+
+def _active_plan() -> dict:
+    return {
+        "id": 10,
+        "instrument_id": 1,
+        "plan_date": date(2026, 5, 6),
+        "status": "active",
+        "bias": "bullish",
+        "primary_draw": {
+            "side": "buy_side",
+            "source": "dealing_range_boundary",
+            "range_id": 77,
+            "boundary": "range_high",
+        },
+        "secondary_draw": {"side": "sell_side"},
+        "allowed_sessions": ["LONDON", "NY_AM"],
+        "invalidations": [{"code": "range_low_lost", "source": "dealing_range", "level": 99.0}],
+        "no_trade_conditions": [],
+        "inputs": {"range_id": 77},
+        "quality": {"active_eligible": True},
+        "timeframe_set": {"bias_tf": "1h", "setup_tf": "15m", "trigger_tf": "5m"},
+    }
+
+
+def _sweep_row() -> dict:
+    return {
+        "id": 20,
+        "side_swept": "sell_side",
+        "pool_id": 21,
+        "sweep_high": 100.4,
+        "sweep_low": 98.8,
+        "sweep_ts": _created_ts(),
+        "close_back_inside": True,
+        "validation_level": "validated",
+        "status": "confirmed",
+        "meta": {"pool_type": "equal_lows"},
+    }
+
+
+def _mss_row() -> dict:
+    return {
+        "id": 30,
+        "direction": "bullish",
+        "broken_level": 99.6,
+        "accepted": True,
+        "ts": _created_ts(),
+        "displacement_event_id": 7,
+        "meta": {"state": "accepted", "status_reason": "retest_held"},
+    }
+
+
+def _fvg_row(status: str = "touched", fill_pct: float = 0.25, meta: dict = None) -> dict:
+    return {
+        "id": 40,
+        "direction": "bullish",
+        "upper": 100.2,
+        "lower": 100.0,
+        "status": status,
+        "fill_pct": fill_pct,
+        "related_displacement_id": 7,
+        "start_ts": _created_ts(),
+        "end_ts": _created_ts(),
+        "meta": {"touch_count": 1, **(meta or {})},
+    }
+
+
+if __name__ == "__main__":
+    unittest.main()