Phase 6 execution tickets

README.md

@@ -221,6 +221,7 @@ Record the current commit as the last-known-good SHA before deploy. If any gate
 - `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 with structural `target_plan` selection; TP1 prefers IRL and TP2 prefers ERL, and signals are skipped when traceable structural targets are missing
 - `python3 scripts/run_execution.py` - run paper execution gating, require structural targets plus management rules for modeled signals, create an execution record, then auto-settle the latest execution for that signal
+- `python3 scripts/run_execution_ticket.py` - generate a human-confirmation execution ticket, run Risk Governor before ticket creation and before confirmation, and then create either `paper` or `semi_auto` execution only after a valid confirmation step
 - `python3 scripts/run_risk_governor.py` - run the independent market-state and risk governor pre-execution gate
 - `python3 scripts/run_backtest.py` - run backtest as `plan_replay` by default, replay active-plan candidates from appearance to confirmation to entry/invalidation, and write target-source traceability plus management rules into result meta
 - `python3 scripts/run_autonomous_cycle.py` - run the unattended cycle: offline checks, database probe, readiness checks, then guarded pipeline when the database is available; prints `Next actions` for the next unattended step
@@ -259,6 +260,7 @@ Optional utility JSON output variables:
 - `TRADING_PLANS_OUTPUT_JSON`
 - `TRADING_REHEARSAL_OUTPUT_JSON`
 - `BACKTEST_OUTPUT_JSON`
+- `EXECUTION_TICKET_OUTPUT_JSON`
 
 Optional structure script JSON output variables:
 
@@ -427,6 +429,7 @@ The `src/api/` package exposes internal query helpers and HTTP routes for:
 - `GET /trade-plan-candidates?instrument_id=1&limit=20`
 - `GET /signals?instrument_id=1&limit=20`
 - `GET /executions?instrument_id=1&limit=20`
+- `GET /execution-tickets?instrument_id=1&limit=20`
 - `GET /backtests?instrument_id=1&limit=10`
 
 Open `/` in the browser for the minimal dashboard, which now includes health, candles, signals, sessions, executions, and backtests.

migrations/024_create_execution_tickets.sql

@@ -0,0 +1,23 @@
+create table if not exists execution_tickets (
+    id bigserial primary key,
+    signal_id bigint not null references trade_signals(id),
+    execution_mode text not null,
+    status text not null,
+    created_ts timestamptz not null,
+    expires_ts timestamptz,
+    confirmed_by text,
+    confirmed_at timestamptz,
+    execution_id bigint references trade_executions(id),
+    ticket_snapshot 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_execution_tickets_signal_created_desc
+on execution_tickets (signal_id, created_ts desc);
+
+create index if not exists idx_execution_tickets_status_created_desc
+on execution_tickets (status, created_ts desc);
+
+create index if not exists idx_execution_tickets_execution_id
+on execution_tickets (execution_id);

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_target_selection_service tests.unit.test_trade_plan_candidate_service tests.unit.test_backtest_service tests.unit.test_backtest_script tests.unit.test_execution_service tests.unit.test_api_server tests.unit.test_api_and_main",
+    "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_target_selection_service tests.unit.test_trade_plan_candidate_service tests.unit.test_backtest_service tests.unit.test_backtest_script tests.unit.test_execution_service tests.unit.test_execution_ticket_script tests.unit.test_api_server tests.unit.test_api_and_main",
     "build": ".\\.venv\\Scripts\\python.exe -m compileall -q src scripts tests",
     "visual:snapshots": "node --check src/web/app.js"
   }

scripts/run_execution_ticket.py

@@ -0,0 +1,102 @@
+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_execution import build_account_risk_state_from_env
+from src.services.execution import ExecutionService
+
+
+def build_execution_mode_from_env(env=None) -> str:
+    env = os.environ if env is None else env
+    return str(env.get("EXECUTION_MODE") or "semi_auto").strip() or "semi_auto"
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    output_json = _get_bool(env, "EXECUTION_TICKET_OUTPUT_JSON", False)
+    signal_id = _get_int(env, "EXECUTION_SIGNAL_ID", 1)
+    execution_mode = build_execution_mode_from_env(env=env)
+    account_equity = _get_float(env, "EXECUTION_ACCOUNT_EQUITY", 10000.0)
+    risk_pct = _get_float(env, "EXECUTION_RISK_PCT", 0.01)
+    confirm_ticket = _get_bool(env, "EXECUTION_CONFIRM_TICKET", False)
+    confirmed_by = str(env.get("EXECUTION_CONFIRMED_BY") or "local_operator")
+    service = ExecutionService()
+    account_state = build_account_risk_state_from_env(env=env)
+    try:
+        ticket = service.create_execution_ticket(
+            signal_id=signal_id,
+            execution_mode=execution_mode,
+            account_equity=account_equity,
+            risk_pct=risk_pct,
+            account_state=account_state,
+        )
+        confirmed = None
+        if ticket is not None and confirm_ticket:
+            confirmed = service.confirm_execution_ticket(
+                ticket.ticket_id,
+                confirmed_by=confirmed_by,
+                account_equity=account_equity,
+                risk_pct=risk_pct,
+                account_state=account_state,
+            )
+    except SQLAlchemyError as exc:
+        payload = {
+            "passed": False,
+            "signal_id": signal_id,
+            "execution_mode": execution_mode,
+            "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 execution ticket sample: {exc.__class__.__name__}")
+        return 2
+    payload = {
+        "passed": True,
+        "signal_id": signal_id,
+        "execution_mode": execution_mode,
+        "ticket": ticket.to_dict() if ticket is not None else None,
+        "confirmed": confirmed.to_dict() if confirmed is not None else None,
+    }
+    if output_json:
+        print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True))
+    else:
+        print(
+            f"Ticket status: {(ticket.status if ticket else 'missing')}; "
+            f"confirmed status: {(confirmed.status if confirmed else 'not_confirmed')}"
+        )
+    return 0
+
+
+def _get_float(env, key: str, default: float) -> float:
+    value = env.get(key)
+    if value is None or value == "":
+        return default
+    return float(value)
+
+
+def _get_int(env, key: str, default: int) -> int:
+    value = env.get(key)
+    if value is None or value == "":
+        return default
+    return int(value)
+
+
+def _get_bool(env, key: str, default: bool) -> bool:
+    value = env.get(key)
+    if value is None or value == "":
+        return default
+    return str(value).strip().lower() in {"1", "true", "yes", "on"}
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/api/__init__.py

@@ -2,6 +2,7 @@
 
 from src.api.backtest import get_recent_backtest_summaries
 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
 from src.api.server import route_request, serve_api
 from src.api.sessions import get_recent_sessions
@@ -12,6 +13,7 @@ from src.api.trading_plans import get_recent_trading_plans
 __all__ = [
     "get_health",
     "get_recent_candles",
+    "get_recent_execution_tickets",
     "get_recent_sessions",
     "get_recent_signals",
     "get_recent_trade_plan_candidates",

src/api/execution_tickets.py

@@ -0,0 +1,29 @@
+from src.db.session import SessionLocal
+from sqlalchemy import text
+
+
+def get_recent_execution_tickets(signal_id: int = None, instrument_id: int = None, limit: int = 20) -> list[dict]:
+    filters = []
+    params = {"limit": limit}
+    if signal_id is not None:
+        filters.append("et.signal_id = :signal_id")
+        params["signal_id"] = signal_id
+    if instrument_id is not None:
+        filters.append("ts.instrument_id = :instrument_id")
+        params["instrument_id"] = instrument_id
+    where_sql = f"where {' and '.join(filters)}" if filters else ""
+    sql = text(
+        f"""
+        select et.id, et.signal_id, et.execution_mode, et.status, et.created_ts, et.expires_ts,
+               et.confirmed_by, et.confirmed_at, et.execution_id, et.ticket_snapshot,
+               ts.instrument_id, ts.side, ts.session_code
+        from execution_tickets et
+        join trade_signals ts on ts.id = et.signal_id
+        {where_sql}
+        order by et.created_ts desc nulls last, et.id desc
+        limit :limit
+        """
+    )
+    with SessionLocal() as session:
+        rows = session.execute(sql, params).mappings().all()
+    return [dict(row) for row in rows]

src/api/server.py

@@ -7,6 +7,7 @@ from urllib.parse import parse_qs, urlparse
 
 from src.api.backtest import get_recent_backtest_summaries
 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
 from src.api.sessions import get_recent_sessions
 from src.api.signals import get_recent_signals
@@ -33,6 +34,7 @@ def build_api_dependencies() -> dict:
     return {
         "health": get_health,
         "candles": get_recent_candles,
+        "execution_tickets": get_recent_execution_tickets,
         "sessions": get_recent_sessions,
         "signals": get_recent_signals,
         "trade_plan_candidates": get_recent_trade_plan_candidates,
@@ -105,6 +107,15 @@ def route_request(
         instrument_id = _require_int(query, "instrument_id")
         limit = _get_int(query, "limit", 20)
         return 200, {"items": deps["executions"](instrument_id=instrument_id, limit=limit)}
+    if method == "GET" and path == "/execution-tickets":
+        instrument_id = _get_optional_int(query, "instrument_id")
+        signal_id = _get_optional_int(query, "signal_id")
+        if instrument_id is None and signal_id is None:
+            raise ValueError("missing_required_query_param:instrument_id_or_signal_id")
+        limit = _get_int(query, "limit", 20)
+        return 200, {
+            "items": deps["execution_tickets"](instrument_id=instrument_id, signal_id=signal_id, limit=limit)
+        }
     if method == "GET" and path == "/training/cases":
         status = _get_single(query, "status")
         skill_tag = _get_single(query, "skill_tag")

src/models/__init__.py

@@ -6,6 +6,7 @@ from src.models.bias_snapshot import BiasSnapshot
 from src.models.candle import Candle
 from src.models.dealing_range import DealingRange
 from src.models.displacement_event import DisplacementEvent
+from src.models.execution_ticket import ExecutionTicket
 from src.models.fvg_zone import FVGZone
 from src.models.instrument import Instrument
 from src.models.liquidity_pool import LiquidityPool
@@ -34,6 +35,7 @@ __all__ = [
     "Candle",
     "DealingRange",
     "DisplacementEvent",
+    "ExecutionTicket",
     "FVGZone",
     "Instrument",
     "LiquidityPool",

src/models/execution_ticket.py

@@ -0,0 +1,22 @@
+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 ExecutionTicket(Base):
+    __tablename__ = "execution_tickets"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    signal_id: Mapped[int] = mapped_column(ForeignKey("trade_signals.id"), nullable=False)
+    execution_mode: Mapped[str] = mapped_column(Text, nullable=False)
+    status: Mapped[str] = mapped_column(Text, nullable=False)
+    created_ts: Mapped[object] = mapped_column(nullable=False)
+    expires_ts: Mapped[Optional[object]] = mapped_column()
+    confirmed_by: Mapped[Optional[str]] = mapped_column(Text)
+    confirmed_at: Mapped[Optional[object]] = mapped_column()
+    execution_id: Mapped[Optional[int]] = mapped_column(ForeignKey("trade_executions.id"))
+    ticket_snapshot: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)

src/repositories/execution_repository.py

@@ -91,6 +91,39 @@ class ExecutionRepository:
             row = session.execute(sql, {"signal_id": signal_id}).mappings().first()
         return dict(row) if row else None
 
+    def fetch_latest_execution_ticket(self, signal_id: int, execution_mode: str) -> Optional[dict]:
+        sql = text(
+            """
+            select id, signal_id, execution_mode, status, created_ts, expires_ts,
+                   confirmed_by, confirmed_at, execution_id, ticket_snapshot
+            from execution_tickets
+            where signal_id = :signal_id
+              and execution_mode = :execution_mode
+            order by created_ts desc nulls last, id desc
+            limit 1
+            """
+        )
+        with SessionLocal() as session:
+            row = session.execute(
+                sql,
+                {"signal_id": signal_id, "execution_mode": execution_mode},
+            ).mappings().first()
+        return dict(row) if row else None
+
+    def fetch_execution_ticket_by_id(self, ticket_id: int) -> Optional[dict]:
+        sql = text(
+            """
+            select id, signal_id, execution_mode, status, created_ts, expires_ts,
+                   confirmed_by, confirmed_at, execution_id, ticket_snapshot
+            from execution_tickets
+            where id = :ticket_id
+            limit 1
+            """
+        )
+        with SessionLocal() as session:
+            row = session.execute(sql, {"ticket_id": ticket_id}).mappings().first()
+        return dict(row) if row else None
+
     def fetch_recent_execution_rows(self, instrument_id: int, limit: int = 20) -> list[dict]:
         sql = text(
             """
@@ -161,6 +194,128 @@ class ExecutionRepository:
                 },
             )
 
+    def insert_execution_ticket(
+        self,
+        signal_id: int,
+        execution_mode: str,
+        status: str,
+        created_ts,
+        expires_ts,
+        confirmed_by: Optional[str],
+        confirmed_at,
+        execution_id: Optional[int],
+        ticket_snapshot: dict,
+    ) -> int:
+        sql = text(
+            """
+            insert into execution_tickets (
+                signal_id,
+                execution_mode,
+                status,
+                created_ts,
+                expires_ts,
+                confirmed_by,
+                confirmed_at,
+                execution_id,
+                ticket_snapshot
+            ) values (
+                :signal_id,
+                :execution_mode,
+                :status,
+                :created_ts,
+                :expires_ts,
+                :confirmed_by,
+                :confirmed_at,
+                :execution_id,
+                cast(:ticket_snapshot as jsonb)
+            )
+            returning id
+            """
+        )
+        with SessionLocal.begin() as session:
+            result = session.execute(
+                sql,
+                {
+                    "signal_id": signal_id,
+                    "execution_mode": execution_mode,
+                    "status": status,
+                    "created_ts": created_ts,
+                    "expires_ts": expires_ts,
+                    "confirmed_by": confirmed_by,
+                    "confirmed_at": confirmed_at,
+                    "execution_id": execution_id,
+                    "ticket_snapshot": json.dumps(ticket_snapshot, default=str),
+                },
+            )
+            return int(result.scalar_one())
+
+    def update_execution_ticket(
+        self,
+        *,
+        ticket_id: int,
+        status: str,
+        ticket_snapshot: dict,
+        confirmed_by: Optional[str] = None,
+        confirmed_at=None,
+        execution_id: Optional[int] = None,
+    ) -> None:
+        sql = text(
+            """
+            update execution_tickets
+            set status = :status,
+                confirmed_by = :confirmed_by,
+                confirmed_at = :confirmed_at,
+                execution_id = :execution_id,
+                ticket_snapshot = cast(:ticket_snapshot as jsonb),
+                updated_at = now()
+            where id = :ticket_id
+            """
+        )
+        with SessionLocal.begin() as session:
+            session.execute(
+                sql,
+                {
+                    "ticket_id": ticket_id,
+                    "status": status,
+                    "confirmed_by": confirmed_by,
+                    "confirmed_at": confirmed_at,
+                    "execution_id": execution_id,
+                    "ticket_snapshot": json.dumps(ticket_snapshot, default=str),
+                },
+            )
+
+    def fetch_recent_execution_ticket_rows(
+        self,
+        *,
+        instrument_id: Optional[int] = None,
+        signal_id: Optional[int] = None,
+        limit: int = 20,
+    ) -> list[dict]:
+        filters = []
+        params = {"limit": limit}
+        if signal_id is not None:
+            filters.append("et.signal_id = :signal_id")
+            params["signal_id"] = signal_id
+        if instrument_id is not None:
+            filters.append("ts.instrument_id = :instrument_id")
+            params["instrument_id"] = instrument_id
+        where_sql = f"where {' and '.join(filters)}" if filters else ""
+        sql = text(
+            f"""
+            select et.id, et.signal_id, et.execution_mode, et.status, et.created_ts, et.expires_ts,
+                   et.confirmed_by, et.confirmed_at, et.execution_id, et.ticket_snapshot,
+                   ts.instrument_id, ts.side, ts.session_code
+            from execution_tickets et
+            join trade_signals ts on ts.id = et.signal_id
+            {where_sql}
+            order by et.created_ts desc nulls last, et.id desc
+            limit :limit
+            """
+        )
+        with SessionLocal() as session:
+            rows = session.execute(sql, params).mappings().all()
+        return [dict(row) for row in rows]
+
     def insert_trade_execution(
         self,
         signal_id: int,

src/services/execution/__init__.py

@@ -3,11 +3,15 @@
 from src.services.execution.execution_service import (
     AccountRiskState,
     build_execution_observation_from_backtest_result,
+    build_execution_ticket_snapshot,
     ExecutionDecision,
+    ExecutionTicketDecision,
     ExecutionService,
     RiskContext,
+    build_signal_snapshot_for_ticket,
     classify_execution_lifecycle,
     compute_entry_price,
+    execution_ticket_matches_snapshot,
     evaluate_account_risk_limits,
     evaluate_execution_decision,
     trade_execution_matches_decision,
@@ -16,11 +20,15 @@ from src.services.execution.execution_service import (
 __all__ = [
     "AccountRiskState",
     "build_execution_observation_from_backtest_result",
+    "build_execution_ticket_snapshot",
     "ExecutionDecision",
+    "ExecutionTicketDecision",
     "ExecutionService",
     "RiskContext",
+    "build_signal_snapshot_for_ticket",
     "classify_execution_lifecycle",
     "compute_entry_price",
+    "execution_ticket_matches_snapshot",
     "evaluate_account_risk_limits",
     "evaluate_execution_decision",
     "trade_execution_matches_decision",

src/services/execution/execution_service.py

@@ -12,12 +12,16 @@ from src.services.backtest import (
     BACKTEST_LIFECYCLE_TP2_CLOSED,
     classify_backtest_lifecycle_state_from_result,
 )
+from src.services.market_state import MarketStateService
 from src.services.risk import AccountRiskState, RiskContext, RiskGovernorService
 from src.services.targets import has_complete_structural_target_plan
 
 
 ALLOWED_EXECUTION_SIGNAL_STATUSES = {"awaiting_confirmation"}
 ALLOWED_EXECUTION_MODES = {"paper", "semi_auto"}
+EXECUTION_TICKET_STATUS_AWAITING_CONFIRMATION = "awaiting_confirmation"
+EXECUTION_TICKET_STATUS_REJECTED = "rejected"
+EXECUTION_TICKET_STATUS_CONFIRMED = "confirmed"
 
 
 @dataclass(frozen=True)
@@ -33,9 +37,36 @@ class ExecutionDecision:
     meta: dict
 
 
+@dataclass(frozen=True)
+class ExecutionTicketDecision:
+    ticket_id: int
+    status: str
+    execution_mode: str
+    signal_id: int
+    execution_id: Optional[int]
+    reasons: list[str]
+    ticket_snapshot: dict
+
+    def to_dict(self) -> dict:
+        return {
+            "ticket_id": self.ticket_id,
+            "status": self.status,
+            "execution_mode": self.execution_mode,
+            "signal_id": self.signal_id,
+            "execution_id": self.execution_id,
+            "reasons": list(self.reasons),
+            "ticket_snapshot": dict(self.ticket_snapshot),
+        }
+
+
 class ExecutionService:
-    def __init__(self, repository: Optional[ExecutionRepository] = None) -> None:
+    def __init__(
+        self,
+        repository: Optional[ExecutionRepository] = None,
+        market_state_service: Optional[MarketStateService] = None,
+    ) -> None:
         self.repository = repository or ExecutionRepository()
+        self.market_state_service = market_state_service or MarketStateService()
 
     def audit_execution(self, trade_execution: Optional[dict], backtest_result: Optional[dict] = None) -> dict:
         if not trade_execution:
@@ -300,6 +331,192 @@ class ExecutionService:
             risk_context=risk_context,
         )
 
+    def build_risk_context_for_signal(
+        self,
+        signal: dict,
+        *,
+        account_state: Optional[AccountRiskState] = None,
+        risk_context: Optional[RiskContext] = None,
+    ) -> RiskContext:
+        if risk_context is not None:
+            return risk_context
+        account_state = account_state or AccountRiskState()
+        signal_meta = dict(signal.get("meta") or {})
+        timeframe = str(
+            (signal_meta.get("signal_timeframe_set") or {}).get("trigger_tf")
+            or signal_meta.get("trigger_tf")
+            or "1m"
+        )
+        market_state = self.market_state_service.build_market_state(
+            instrument_id=int(signal["instrument_id"]),
+            timeframe=timeframe,
+        )
+        quality_gate = self.market_state_service.repository.fetch_latest_backtest_quality_gate(
+            instrument_id=int(signal["instrument_id"])
+        )
+        return RiskContext.from_account_state(
+            account_state,
+            market_state=market_state,
+            backtest_quality_gate=quality_gate,
+        )
+
+    def create_execution_ticket(
+        self,
+        signal_id: int,
+        execution_mode: str = "semi_auto",
+        account_equity: float = 10000.0,
+        risk_pct: float = 0.01,
+        account_state: Optional[AccountRiskState] = None,
+        risk_context: Optional[RiskContext] = None,
+    ) -> Optional[ExecutionTicketDecision]:
+        signal = self.repository.fetch_signal_for_execution(signal_id)
+        if not signal:
+            return None
+        full_risk_context = self.build_risk_context_for_signal(
+            signal,
+            account_state=account_state,
+            risk_context=risk_context,
+        )
+        decision = evaluate_execution_decision(
+            signal=signal,
+            execution_mode=execution_mode,
+            account_equity=account_equity,
+            risk_pct=risk_pct,
+            account_state=account_state,
+            risk_context=full_risk_context,
+        )
+        created_ts = datetime.now(timezone.utc)
+        status = EXECUTION_TICKET_STATUS_AWAITING_CONFIRMATION if decision.allowed else EXECUTION_TICKET_STATUS_REJECTED
+        snapshot = build_execution_ticket_snapshot(
+            signal=signal,
+            decision=decision,
+            risk_context=full_risk_context,
+            execution_mode=execution_mode,
+            created_ts=created_ts,
+            status=status,
+            account_equity=account_equity,
+            risk_pct=risk_pct,
+        )
+        latest = self.repository.fetch_latest_execution_ticket(signal_id=signal_id, execution_mode=execution_mode)
+        if latest and execution_ticket_matches_snapshot(latest=latest, status=status, snapshot=snapshot):
+            return ExecutionTicketDecision(
+                ticket_id=int(latest["id"]),
+                status=str(latest["status"]),
+                execution_mode=str(latest["execution_mode"]),
+                signal_id=int(latest["signal_id"]),
+                execution_id=latest.get("execution_id"),
+                reasons=list((latest.get("ticket_snapshot") or {}).get("decision", {}).get("reasons") or []),
+                ticket_snapshot=dict(latest.get("ticket_snapshot") or {}),
+            )
+        ticket_id = self.repository.insert_execution_ticket(
+            signal_id=signal_id,
+            execution_mode=execution_mode,
+            status=status,
+            created_ts=created_ts,
+            expires_ts=signal.get("expires_ts"),
+            confirmed_by=None,
+            confirmed_at=None,
+            execution_id=None,
+            ticket_snapshot=snapshot,
+        )
+        return ExecutionTicketDecision(
+            ticket_id=ticket_id,
+            status=status,
+            execution_mode=execution_mode,
+            signal_id=signal_id,
+            execution_id=None,
+            reasons=list(decision.reasons),
+            ticket_snapshot=snapshot,
+        )
+
+    def confirm_execution_ticket(
+        self,
+        ticket_id: int,
+        *,
+        confirmed_by: str,
+        account_equity: float = 10000.0,
+        risk_pct: float = 0.01,
+        account_state: Optional[AccountRiskState] = None,
+        risk_context: Optional[RiskContext] = None,
+    ) -> Optional[ExecutionTicketDecision]:
+        ticket = self.repository.fetch_execution_ticket_by_id(ticket_id)
+        if not ticket:
+            return None
+        if ticket.get("status") != EXECUTION_TICKET_STATUS_AWAITING_CONFIRMATION:
+            return ExecutionTicketDecision(
+                ticket_id=int(ticket["id"]),
+                status=str(ticket["status"]),
+                execution_mode=str(ticket["execution_mode"]),
+                signal_id=int(ticket["signal_id"]),
+                execution_id=ticket.get("execution_id"),
+                reasons=list((ticket.get("ticket_snapshot") or {}).get("decision", {}).get("reasons") or []),
+                ticket_snapshot=dict(ticket.get("ticket_snapshot") or {}),
+            )
+        signal = self.repository.fetch_signal_for_execution(int(ticket["signal_id"]))
+        if not signal:
+            return None
+        full_risk_context = self.build_risk_context_for_signal(
+            signal,
+            account_state=account_state,
+            risk_context=risk_context,
+        )
+        decision = evaluate_execution_decision(
+            signal=signal,
+            execution_mode=str(ticket["execution_mode"]),
+            account_equity=account_equity,
+            risk_pct=risk_pct,
+            account_state=account_state,
+            risk_context=full_risk_context,
+        )
+        confirmed_at = datetime.now(timezone.utc)
+        status = EXECUTION_TICKET_STATUS_CONFIRMED if decision.allowed else EXECUTION_TICKET_STATUS_REJECTED
+        snapshot = build_execution_ticket_snapshot(
+            signal=signal,
+            decision=decision,
+            risk_context=full_risk_context,
+            execution_mode=str(ticket["execution_mode"]),
+            created_ts=ticket.get("created_ts"),
+            status=status,
+            account_equity=account_equity,
+            risk_pct=risk_pct,
+            existing_snapshot=dict(ticket.get("ticket_snapshot") or {}),
+            confirmed_by=confirmed_by if decision.allowed else None,
+            confirmed_at=confirmed_at if decision.allowed else None,
+        )
+        execution_id = None
+        if decision.allowed:
+            execution_id = self.create_execution(
+                signal_id=int(ticket["signal_id"]),
+                execution_mode=str(ticket["execution_mode"]),
+                account_equity=account_equity,
+                risk_pct=risk_pct,
+                confirmed=True,
+                account_state=account_state,
+                risk_context=full_risk_context,
+                ticket_id=int(ticket["id"]),
+                ticket_snapshot=snapshot,
+                confirmed_by=confirmed_by,
+                confirmed_at=confirmed_at,
+            )
+        self.repository.update_execution_ticket(
+            ticket_id=int(ticket["id"]),
+            status=status,
+            ticket_snapshot=snapshot,
+            confirmed_by=confirmed_by if decision.allowed else None,
+            confirmed_at=confirmed_at if decision.allowed else None,
+            execution_id=execution_id,
+        )
+        updated = self.repository.fetch_execution_ticket_by_id(int(ticket["id"])) or ticket
+        return ExecutionTicketDecision(
+            ticket_id=int(updated["id"]),
+            status=str(updated["status"]),
+            execution_mode=str(updated["execution_mode"]),
+            signal_id=int(updated["signal_id"]),
+            execution_id=updated.get("execution_id"),
+            reasons=list(decision.reasons),
+            ticket_snapshot=dict(updated.get("ticket_snapshot") or {}),
+        )
+
     def create_execution(
         self,
         signal_id: int,
@@ -309,6 +526,10 @@ class ExecutionService:
         confirmed: bool = False,
         account_state: Optional[AccountRiskState] = None,
         risk_context: Optional[RiskContext] = None,
+        ticket_id: Optional[int] = None,
+        ticket_snapshot: Optional[dict] = None,
+        confirmed_by: Optional[str] = None,
+        confirmed_at=None,
     ) -> Optional[int]:
         decision = self.prepare_execution(
             signal_id=signal_id,
@@ -322,7 +543,14 @@ class ExecutionService:
             return None
         if decision.requires_confirmation and not confirmed:
             return None
+        if execution_mode == "semi_auto" and confirmed and ticket_id is None:
+            return None
         latest = self.repository.fetch_latest_execution(signal_id=signal_id, execution_mode=execution_mode)
+        if ticket_id is not None:
+            decision.meta["execution_ticket_id"] = ticket_id
+            decision.meta["ticket_snapshot"] = dict(ticket_snapshot or {})
+            decision.meta["confirmed_by"] = confirmed_by
+            decision.meta["confirmed_at"] = confirmed_at.isoformat() if hasattr(confirmed_at, "isoformat") else confirmed_at
         if latest and trade_execution_matches_decision(latest, decision):
             return latest['id']
         venue_order_id = None if execution_mode == "paper" else f"semi_auto:{signal_id}:{int(datetime.now(timezone.utc).timestamp())}"
@@ -470,6 +698,78 @@ def signal_requires_structural_targets(signal: dict) -> bool:
     return bool(signal.get("model_code")) or bool(meta.get("target_plan_required"))
 
 
+def build_execution_ticket_snapshot(
+    *,
+    signal: dict,
+    decision: ExecutionDecision,
+    risk_context: RiskContext,
+    execution_mode: str,
+    created_ts,
+    status: str,
+    account_equity: float,
+    risk_pct: float,
+    existing_snapshot: Optional[dict] = None,
+    confirmed_by: Optional[str] = None,
+    confirmed_at=None,
+) -> dict:
+    snapshot = dict(existing_snapshot or {})
+    snapshot.update(
+        {
+            "signal": build_signal_snapshot_for_ticket(signal),
+            "execution_mode": execution_mode,
+            "status": status,
+            "decision": {
+                "allowed": decision.allowed,
+                "status": decision.status,
+                "reasons": list(decision.reasons),
+                "requires_confirmation": decision.requires_confirmation,
+                "risk_pct": decision.risk_pct,
+                "size": decision.size,
+                "entry_price": decision.entry_price,
+                "meta": dict(decision.meta),
+            },
+            "risk_context": risk_context.to_dict(),
+            "account_equity": account_equity,
+            "created_ts": created_ts.isoformat() if hasattr(created_ts, "isoformat") else created_ts,
+        }
+    )
+    if confirmed_by is not None or confirmed_at is not None:
+        snapshot["confirmation"] = {
+            "confirmed_by": confirmed_by,
+            "confirmed_at": confirmed_at.isoformat() if hasattr(confirmed_at, "isoformat") else confirmed_at,
+        }
+    return snapshot
+
+
+def build_signal_snapshot_for_ticket(signal: dict) -> dict:
+    return {
+        "signal_id": signal.get("id"),
+        "instrument_id": signal.get("instrument_id"),
+        "model_code": signal.get("model_code"),
+        "side": signal.get("side"),
+        "status": signal.get("status"),
+        "entry_type": signal.get("entry_type"),
+        "entry_low": signal.get("entry_low"),
+        "entry_high": signal.get("entry_high"),
+        "stop_loss": signal.get("stop_loss"),
+        "tp1": signal.get("tp1"),
+        "tp2": signal.get("tp2"),
+        "session_code": signal.get("session_code"),
+        "created_ts": signal.get("created_ts"),
+        "expires_ts": signal.get("expires_ts"),
+        "invalidations": list(signal.get("invalidations") or []),
+        "evidence": dict(signal.get("evidence") or {}),
+        "meta": dict(signal.get("meta") or {}),
+    }
+
+
+def execution_ticket_matches_snapshot(*, latest: dict, status: str, snapshot: dict) -> bool:
+    return (
+        latest.get("status") == status
+        and dict(latest.get("ticket_snapshot") or {}) == dict(snapshot)
+    )
+
+
 def evaluate_account_risk_limits(account_state: Optional[AccountRiskState]) -> list[str]:
     if account_state is None:
         return []

tests/unit/test_api_server.py

@@ -81,6 +81,18 @@ class ApiServerRoutingTests(unittest.TestCase):
         self.assertEqual(status, 200)
         self.assertEqual(payload["items"], [{"execution_id": 1, "audit_status": "ok"}])
 
+    def test_routes_execution_tickets_query(self) -> None:
+        status, payload = route_request(
+            "/execution-tickets",
+            {"signal_id": ["1"], "limit": ["2"]},
+            dependencies={
+                "execution_tickets": lambda **kwargs: [{"id": 1, "status": "awaiting_confirmation"}],
+            },
+        )
+
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["items"], [{"id": 1, "status": "awaiting_confirmation"}])
+
     def test_routes_health(self) -> None:
         status, payload = route_request(
             "/health",

tests/unit/test_execution_service.py

@@ -6,7 +6,9 @@ os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:
 
 from src.services.execution import (
     AccountRiskState,
+    RiskContext,
     build_execution_observation_from_backtest_result,
+    build_execution_ticket_snapshot,
     ExecutionService,
     classify_execution_lifecycle,
     compute_entry_price,
@@ -250,8 +252,193 @@ class ExecutionServiceTests(unittest.TestCase):
 
         execution_id = service.create_execution(signal_id=1, execution_mode="semi_auto", confirmed=True)
 
-        self.assertEqual(execution_id, 89)
+        self.assertIsNone(execution_id)
+        repository.insert_trade_execution.assert_not_called()
+
+    def test_semi_auto_ticket_must_be_confirmed_before_pending_execution(self) -> None:
+        repository = Mock()
+        ticket_row = {
+            "id": 51,
+            "signal_id": 1,
+            "execution_mode": "semi_auto",
+            "status": "awaiting_confirmation",
+            "created_ts": "created",
+            "expires_ts": "later",
+            "confirmed_by": None,
+            "confirmed_at": None,
+            "execution_id": None,
+            "ticket_snapshot": {},
+        }
+        repository.fetch_signal_for_execution.return_value = _valid_signal()
+        repository.fetch_latest_execution_ticket.return_value = None
+        repository.insert_execution_ticket.return_value = 51
+        repository.fetch_execution_ticket_by_id.side_effect = lambda ticket_id: dict(ticket_row)
+        repository.fetch_latest_execution.return_value = None
+        repository.insert_trade_execution.return_value = 89
+        service = ExecutionService(repository=repository)
+        service.build_risk_context_for_signal = Mock(return_value=RiskContext())
+        repository.update_execution_ticket.side_effect = lambda **kwargs: ticket_row.update(
+            {
+                "status": kwargs["status"],
+                "ticket_snapshot": kwargs["ticket_snapshot"],
+                "confirmed_by": kwargs.get("confirmed_by"),
+                "confirmed_at": kwargs.get("confirmed_at"),
+                "execution_id": kwargs.get("execution_id"),
+            }
+        )
+
+        ticket = service.create_execution_ticket(signal_id=1, execution_mode="semi_auto")
+        self.assertIsNotNone(ticket)
+        repository.insert_trade_execution.assert_not_called()
+
+        confirmed = service.confirm_execution_ticket(51, confirmed_by="alice")
+
+        self.assertIsNotNone(confirmed)
+        assert confirmed is not None
+        self.assertEqual(confirmed.status, "confirmed")
         repository.insert_trade_execution.assert_called_once()
+        kwargs = repository.insert_trade_execution.call_args.kwargs
+        self.assertEqual(kwargs["status"], "pending")
+        self.assertEqual(kwargs["meta"]["execution_ticket_id"], 51)
+        self.assertEqual(kwargs["meta"]["confirmed_by"], "alice")
+
+    def test_ticket_rejects_when_risk_check_fails_before_creation(self) -> None:
+        repository = Mock()
+        repository.fetch_signal_for_execution.return_value = _valid_signal()
+        repository.fetch_latest_execution_ticket.return_value = None
+        repository.insert_execution_ticket.return_value = 61
+        service = ExecutionService(repository=repository)
+        service.build_risk_context_for_signal = Mock(
+            return_value=RiskContext.from_account_state(AccountRiskState(kill_switch=True))
+        )
+
+        ticket = service.create_execution_ticket(
+            signal_id=1,
+            execution_mode="semi_auto",
+            risk_context=None,
+            account_state=AccountRiskState(kill_switch=True),
+        )
+
+        self.assertIsNotNone(ticket)
+        assert ticket is not None
+        self.assertEqual(ticket.status, "rejected")
+        self.assertIn("execution_kill_switch_enabled", ticket.reasons)
+        repository.insert_trade_execution.assert_not_called()
+
+    def test_confirm_ticket_rejects_when_recheck_fails(self) -> None:
+        repository = Mock()
+        ticket_row = {
+            "id": 71,
+            "signal_id": 1,
+            "execution_mode": "semi_auto",
+            "status": "awaiting_confirmation",
+            "created_ts": "now",
+            "expires_ts": None,
+            "confirmed_by": None,
+            "confirmed_at": None,
+            "execution_id": None,
+            "ticket_snapshot": {},
+        }
+        repository.fetch_signal_for_execution.return_value = _valid_signal()
+        repository.fetch_execution_ticket_by_id.side_effect = lambda ticket_id: dict(ticket_row)
+        repository.fetch_latest_execution.return_value = None
+        service = ExecutionService(repository=repository)
+        service.build_risk_context_for_signal = Mock(
+            return_value=RiskContext.from_account_state(AccountRiskState(kill_switch=True))
+        )
+        repository.update_execution_ticket.side_effect = lambda **kwargs: ticket_row.update(
+            {
+                "status": kwargs["status"],
+                "ticket_snapshot": kwargs["ticket_snapshot"],
+                "confirmed_by": kwargs.get("confirmed_by"),
+                "confirmed_at": kwargs.get("confirmed_at"),
+                "execution_id": kwargs.get("execution_id"),
+            }
+        )
+
+        confirmed = service.confirm_execution_ticket(
+            71,
+            confirmed_by="bob",
+            account_state=AccountRiskState(kill_switch=True),
+        )
+
+        self.assertIsNotNone(confirmed)
+        assert confirmed is not None
+        self.assertEqual(confirmed.status, "rejected")
+        repository.insert_trade_execution.assert_not_called()
+        repository.update_execution_ticket.assert_called_once()
+
+    def test_ticket_snapshot_is_complete_for_replay(self) -> None:
+        repository = Mock()
+        repository.fetch_signal_for_execution.return_value = _valid_signal()
+        risk_context = RiskContext()
+        decision = evaluate_execution_decision(
+            signal=_valid_signal(),
+            execution_mode="paper",
+            account_equity=10000.0,
+            risk_pct=0.01,
+        )
+        snapshot = build_execution_ticket_snapshot(
+            signal=_valid_signal(),
+            decision=decision,
+            risk_context=risk_context,
+            execution_mode="semi_auto",
+            created_ts="created",
+            status="awaiting_confirmation",
+            account_equity=10000.0,
+            risk_pct=0.01,
+        )
+
+        self.assertEqual(snapshot["signal"]["signal_id"], 1)
+        self.assertEqual(snapshot["decision"]["status"], "open")
+        self.assertIn("risk_context", snapshot)
+        self.assertEqual(snapshot["execution_mode"], "semi_auto")
+        self.assertEqual(snapshot["signal"]["model_code"], "SWEEP_MSS_FVG")
+
+    def test_paper_ticket_confirmation_creates_open_execution(self) -> None:
+        repository = Mock()
+        ticket_row = {
+            "id": 81,
+            "signal_id": 1,
+            "execution_mode": "paper",
+            "status": "awaiting_confirmation",
+            "created_ts": "created",
+            "expires_ts": "later",
+            "confirmed_by": None,
+            "confirmed_at": None,
+            "execution_id": None,
+            "ticket_snapshot": {},
+        }
+        repository.fetch_signal_for_execution.return_value = _valid_signal()
+        repository.fetch_latest_execution_ticket.return_value = None
+        repository.insert_execution_ticket.return_value = 81
+        repository.fetch_execution_ticket_by_id.side_effect = lambda ticket_id: dict(ticket_row)
+        repository.fetch_latest_execution.return_value = None
+        repository.insert_trade_execution.return_value = 91
+        repository.update_execution_ticket.side_effect = lambda **kwargs: ticket_row.update(
+            {
+                "status": kwargs["status"],
+                "ticket_snapshot": kwargs["ticket_snapshot"],
+                "confirmed_by": kwargs.get("confirmed_by"),
+                "confirmed_at": kwargs.get("confirmed_at"),
+                "execution_id": kwargs.get("execution_id"),
+            }
+        )
+        service = ExecutionService(repository=repository)
+        service.build_risk_context_for_signal = Mock(return_value=RiskContext())
+
+        ticket = service.create_execution_ticket(signal_id=1, execution_mode="paper")
+        self.assertIsNotNone(ticket)
+
+        confirmed = service.confirm_execution_ticket(81, confirmed_by="paper_user")
+
+        self.assertIsNotNone(confirmed)
+        assert confirmed is not None
+        self.assertEqual(confirmed.status, "confirmed")
+        kwargs = repository.insert_trade_execution.call_args.kwargs
+        self.assertEqual(kwargs["status"], "open")
+        self.assertEqual(kwargs["meta"]["execution_ticket_id"], 81)
+        self.assertEqual(kwargs["meta"]["confirmed_by"], "paper_user")
 
 
 class ExecutionAuditTests(unittest.TestCase):
@@ -606,5 +793,38 @@ class ExecutionIdempotencyTests(unittest.TestCase):
         self.assertTrue(trade_execution_matches_decision(existing, decision))
 
 
+def _valid_signal() -> dict:
+    return {
+        "id": 1,
+        "instrument_id": 1,
+        "model_code": "SWEEP_MSS_FVG",
+        "side": "buy",
+        "status": "awaiting_confirmation",
+        "entry_type": "limit",
+        "entry_low": 100.0,
+        "entry_high": 101.0,
+        "stop_loss": 99.0,
+        "tp1": 102.0,
+        "tp2": 104.0,
+        "session_code": "LONDON",
+        "execution_intensity": "high",
+        "daily_bias": "bullish",
+        "draw_side": "buy_side",
+        "created_ts": "2026-05-06T00:00:00+00:00",
+        "expires_ts": "2026-05-06T04:00:00+00:00",
+        "invalidations": [],
+        "evidence": {"mss_id": 1, "fvg_id": 2},
+        "meta": {
+            "execution_permission": True,
+            "management_rules": [{"code": "tp1_partial", "fraction": 0.5}],
+            "target_plan": {
+                "status": "ready",
+                "tp1": {"price": 102.0, "objective_layer": "IRL", "source": {"type": "session_pivot", "table": "market_sessions", "field": "high"}},
+                "tp2": {"price": 104.0, "objective_layer": "ERL", "source": {"type": "dealing_range_boundary", "table": "dealing_ranges", "field": "high"}},
+            },
+        },
+    }
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_execution_ticket_script.py

@@ -0,0 +1,36 @@
+import contextlib
+import io
+import json
+import os
+import unittest
+from unittest.mock import patch
+
+from sqlalchemy.exc import OperationalError
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
+
+from scripts import run_execution_ticket
+
+
+class ExecutionTicketScriptTests(unittest.TestCase):
+    def test_build_execution_mode_defaults_to_semi_auto(self) -> None:
+        self.assertEqual(run_execution_ticket.build_execution_mode_from_env({}), "semi_auto")
+        self.assertEqual(run_execution_ticket.build_execution_mode_from_env({"EXECUTION_MODE": "paper"}), "paper")
+
+    def test_main_reports_database_unavailable_json(self) -> None:
+        output = io.StringIO()
+
+        with patch(
+            "scripts.run_execution_ticket.ExecutionService.create_execution_ticket",
+            side_effect=OperationalError("stmt", {}, Exception()),
+        ), contextlib.redirect_stdout(output):
+            exit_code = run_execution_ticket.main(env={"EXECUTION_TICKET_OUTPUT_JSON": "1"})
+
+        payload = json.loads(output.getvalue())
+        self.assertEqual(exit_code, 2)
+        self.assertFalse(payload["passed"])
+        self.assertEqual(payload["error_type"], "OperationalError")
+
+
+if __name__ == "__main__":
+    unittest.main()