From c3cb2e944f2de8ad5e4f780409035f343c02170d Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 6 May 2026 16:03:40 +0800 Subject: [PATCH] Phase 6 execution tickets --- README.md | 3 + migrations/024_create_execution_tickets.sql | 23 ++ package.json | 2 +- scripts/run_execution_ticket.py | 102 +++++++ src/api/__init__.py | 2 + src/api/execution_tickets.py | 29 ++ src/api/server.py | 11 + src/models/__init__.py | 2 + src/models/execution_ticket.py | 22 ++ src/repositories/execution_repository.py | 155 ++++++++++ src/services/execution/__init__.py | 8 + src/services/execution/execution_service.py | 302 +++++++++++++++++++- tests/unit/test_api_server.py | 12 + tests/unit/test_execution_service.py | 222 +++++++++++++- tests/unit/test_execution_ticket_script.py | 36 +++ 15 files changed, 928 insertions(+), 3 deletions(-) create mode 100644 migrations/024_create_execution_tickets.sql create mode 100644 scripts/run_execution_ticket.py create mode 100644 src/api/execution_tickets.py create mode 100644 src/models/execution_ticket.py create mode 100644 tests/unit/test_execution_ticket_script.py diff --git a/README.md b/README.md index 981eee9..41dfbeb 100644 --- a/README.md +++ b/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. diff --git a/migrations/024_create_execution_tickets.sql b/migrations/024_create_execution_tickets.sql new file mode 100644 index 0000000..6838f95 --- /dev/null +++ b/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); diff --git a/package.json b/package.json index 4f27d20..f7e31f1 100644 --- a/package.json +++ b/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" } diff --git a/scripts/run_execution_ticket.py b/scripts/run_execution_ticket.py new file mode 100644 index 0000000..c885f8e --- /dev/null +++ b/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()) diff --git a/src/api/__init__.py b/src/api/__init__.py index c4c0f9f..0513120 100644 --- a/src/api/__init__.py +++ b/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", diff --git a/src/api/execution_tickets.py b/src/api/execution_tickets.py new file mode 100644 index 0000000..a595bcd --- /dev/null +++ b/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] diff --git a/src/api/server.py b/src/api/server.py index 357845a..a845a98 100644 --- a/src/api/server.py +++ b/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") diff --git a/src/models/__init__.py b/src/models/__init__.py index 73f47fc..8602873 100644 --- a/src/models/__init__.py +++ b/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", diff --git a/src/models/execution_ticket.py b/src/models/execution_ticket.py new file mode 100644 index 0000000..a318226 --- /dev/null +++ b/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) diff --git a/src/repositories/execution_repository.py b/src/repositories/execution_repository.py index 914f5df..63cfdbf 100644 --- a/src/repositories/execution_repository.py +++ b/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, diff --git a/src/services/execution/__init__.py b/src/services/execution/__init__.py index 562be5c..899762b 100644 --- a/src/services/execution/__init__.py +++ b/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", diff --git a/src/services/execution/execution_service.py b/src/services/execution/execution_service.py index 95a10ae..4952012 100644 --- a/src/services/execution/execution_service.py +++ b/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 [] diff --git a/tests/unit/test_api_server.py b/tests/unit/test_api_server.py index a7fa413..b350fcd 100644 --- a/tests/unit/test_api_server.py +++ b/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", diff --git a/tests/unit/test_execution_service.py b/tests/unit/test_execution_service.py index 40c1d6d..c44739c 100644 --- a/tests/unit/test_execution_service.py +++ b/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() diff --git a/tests/unit/test_execution_ticket_script.py b/tests/unit/test_execution_ticket_script.py new file mode 100644 index 0000000..8234af0 --- /dev/null +++ b/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()