From d7a04975c73b136b07321ee3b70c6abf53162eba Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 6 May 2026 16:32:48 +0800 Subject: [PATCH] Phase 7 live adapter prep layer --- .env.example | 5 + README.md | 2 + .../025_create_live_operation_audits.sql | 29 ++ package.json | 2 +- scripts/run_live_adapter.py | 105 ++++++ src/api/__init__.py | 2 + src/api/live_operation_audits.py | 13 + src/api/server.py | 15 + src/config/settings.py | 5 + src/models/__init__.py | 2 + src/models/live_operation_audit.py | 28 ++ src/repositories/execution_repository.py | 111 +++++++ src/services/broker/__init__.py | 33 ++ src/services/broker/broker_adapter.py | 88 +++++ src/services/broker/live_broker_service.py | 311 ++++++++++++++++++ src/services/broker/okx_order_adapter.py | 75 +++++ tests/unit/test_api_server.py | 12 + tests/unit/test_live_adapter_script.py | 37 +++ tests/unit/test_live_adapter_service.py | 158 +++++++++ 19 files changed, 1032 insertions(+), 1 deletion(-) create mode 100644 migrations/025_create_live_operation_audits.sql create mode 100644 scripts/run_live_adapter.py create mode 100644 src/api/live_operation_audits.py create mode 100644 src/models/live_operation_audit.py create mode 100644 src/services/broker/__init__.py create mode 100644 src/services/broker/broker_adapter.py create mode 100644 src/services/broker/live_broker_service.py create mode 100644 src/services/broker/okx_order_adapter.py create mode 100644 tests/unit/test_live_adapter_script.py create mode 100644 tests/unit/test_live_adapter_service.py diff --git a/.env.example b/.env.example index 1875322..630e4f8 100644 --- a/.env.example +++ b/.env.example @@ -1,6 +1,11 @@ DATABASE_URL=postgresql+psycopg://postgres:postgres@localhost:5432/ai_ict OKX_BASE_URL=https://www.okx.com OKX_TIMEOUT=30 +LIVE_TRADING_ENABLED=0 +OKX_ORDER_BASE_URL=https://www.okx.com +OKX_API_KEY= +OKX_API_SECRET= +OKX_API_PASSPHRASE= DEFAULT_TIMEZONE=UTC SESSION_TIMEZONE=America/New_York LOG_LEVEL=INFO diff --git a/README.md b/README.md index 41dfbeb..7ddb2eb 100644 --- a/README.md +++ b/README.md @@ -222,6 +222,7 @@ Record the current commit as the last-known-good SHA before deploy. If any gate - `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_live_adapter.py` - exercise the live broker preparation layer for `submit`, `reconcile`, `cancel_all`, or `kill_switch`; every attempted live operation is audited and real submission stays blocked unless live trading is explicitly enabled and confirmed - `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 @@ -430,6 +431,7 @@ The `src/api/` package exposes internal query helpers and HTTP routes for: - `GET /signals?instrument_id=1&limit=20` - `GET /executions?instrument_id=1&limit=20` - `GET /execution-tickets?instrument_id=1&limit=20` +- `GET /live-operation-audits?execution_ticket_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/025_create_live_operation_audits.sql b/migrations/025_create_live_operation_audits.sql new file mode 100644 index 0000000..b676481 --- /dev/null +++ b/migrations/025_create_live_operation_audits.sql @@ -0,0 +1,29 @@ +create table if not exists live_operation_audits ( + id bigserial primary key, + adapter_name text not null, + operation_type text not null, + status text not null, + execution_ticket_id bigint references execution_tickets(id), + execution_id bigint references trade_executions(id), + signal_id bigint references trade_signals(id), + instrument_id bigint references instruments(id), + requested_by text, + requested_at timestamptz not null, + completed_at timestamptz, + live_trading_enabled boolean not null default false, + explicit_confirmation boolean not null default false, + payload jsonb not null default '{}'::jsonb, + result jsonb not null default '{}'::jsonb, + reason_codes 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_live_operation_audits_requested_desc +on live_operation_audits (requested_at desc); + +create index if not exists idx_live_operation_audits_ticket_requested_desc +on live_operation_audits (execution_ticket_id, requested_at desc); + +create index if not exists idx_live_operation_audits_execution_requested_desc +on live_operation_audits (execution_id, requested_at desc); diff --git a/package.json b/package.json index f7e31f1..c87856c 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_execution_ticket_script 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_live_adapter_service tests.unit.test_live_adapter_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_live_adapter.py b/scripts/run_live_adapter.py new file mode 100644 index 0000000..5e00109 --- /dev/null +++ b/scripts/run_live_adapter.py @@ -0,0 +1,105 @@ +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 src.services.broker import LiveBrokerService + + +def main(env=None) -> int: + env = os.environ if env is None else env + output_json = _get_bool(env, "LIVE_ADAPTER_OUTPUT_JSON", False) + operation = str(env.get("LIVE_ADAPTER_OPERATION") or "reconcile").strip().lower() + adapter_name = str(env.get("LIVE_ADAPTER_NAME") or "okx").strip().lower() + requested_by = str(env.get("LIVE_REQUESTED_BY") or "local_operator") + explicit_live_confirmation = _get_bool(env, "LIVE_EXPLICIT_CONFIRMATION", False) + instrument_id = _get_optional_int(env, "LIVE_INSTRUMENT_ID") + execution_ticket_id = _get_optional_int(env, "LIVE_EXECUTION_TICKET_ID") + kill_switch_reason = str(env.get("LIVE_KILL_SWITCH_REASON") or "manual_kill_switch") + service = LiveBrokerService() + try: + if operation == "submit": + if execution_ticket_id is None: + raise ValueError("missing_required_env:LIVE_EXECUTION_TICKET_ID") + result = service.submit_confirmed_ticket( + execution_ticket_id, + requested_by=requested_by, + explicit_live_confirmation=explicit_live_confirmation, + adapter_name=adapter_name, + ) + elif operation == "reconcile": + result = service.reconcile_orders( + requested_by=requested_by, + instrument_id=instrument_id, + adapter_name=adapter_name, + ) + elif operation == "cancel_all": + result = service.cancel_all( + requested_by=requested_by, + instrument_id=instrument_id, + explicit_live_confirmation=explicit_live_confirmation, + adapter_name=adapter_name, + ) + elif operation == "kill_switch": + result = service.activate_kill_switch( + requested_by=requested_by, + reason=kill_switch_reason, + explicit_live_confirmation=explicit_live_confirmation, + adapter_name=adapter_name, + ) + else: + raise ValueError(f"unsupported_live_adapter_operation:{operation}") + except SQLAlchemyError as exc: + payload = { + "passed": False, + "operation": operation, + "adapter_name": adapter_name, + "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 live adapter sample: {exc.__class__.__name__}") + return 2 + + payload = { + "passed": True, + "operation": operation, + "adapter_name": adapter_name, + "result": result.to_dict() if result is not None else None, + } + if output_json: + print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True)) + else: + print( + f"Live adapter operation: {operation}; " + f"status: {(result.status if result else 'missing')}; " + f"reasons: {(result.reason_codes if result else [])}" + ) + return 0 + + +def _get_optional_int(env, key: str): + value = env.get(key) + if value is None or value == "": + return None + 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 0513120..b4ad33d 100644 --- a/src/api/__init__.py +++ b/src/api/__init__.py @@ -4,6 +4,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.live_operation_audits import get_recent_live_operation_audits 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 @@ -16,6 +17,7 @@ __all__ = [ "get_recent_execution_tickets", "get_recent_sessions", "get_recent_signals", + "get_recent_live_operation_audits", "get_recent_trade_plan_candidates", "get_recent_trading_plans", "get_recent_backtest_summaries", diff --git a/src/api/live_operation_audits.py b/src/api/live_operation_audits.py new file mode 100644 index 0000000..4e6b730 --- /dev/null +++ b/src/api/live_operation_audits.py @@ -0,0 +1,13 @@ +from src.repositories.execution_repository import ExecutionRepository + + +def get_recent_live_operation_audits( + execution_ticket_id: int = None, + instrument_id: int = None, + limit: int = 20, +) -> list[dict]: + return ExecutionRepository().fetch_recent_live_operation_audits( + execution_ticket_id=execution_ticket_id, + instrument_id=instrument_id, + limit=limit, + ) diff --git a/src/api/server.py b/src/api/server.py index a845a98..5910f98 100644 --- a/src/api/server.py +++ b/src/api/server.py @@ -9,6 +9,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.live_operation_audits import get_recent_live_operation_audits 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 @@ -35,6 +36,7 @@ def build_api_dependencies() -> dict: "health": get_health, "candles": get_recent_candles, "execution_tickets": get_recent_execution_tickets, + "live_operation_audits": get_recent_live_operation_audits, "sessions": get_recent_sessions, "signals": get_recent_signals, "trade_plan_candidates": get_recent_trade_plan_candidates, @@ -116,6 +118,19 @@ def route_request( return 200, { "items": deps["execution_tickets"](instrument_id=instrument_id, signal_id=signal_id, limit=limit) } + if method == "GET" and path == "/live-operation-audits": + instrument_id = _get_optional_int(query, "instrument_id") + execution_ticket_id = _get_optional_int(query, "execution_ticket_id") + if instrument_id is None and execution_ticket_id is None: + raise ValueError("missing_required_query_param:instrument_id_or_execution_ticket_id") + limit = _get_int(query, "limit", 20) + return 200, { + "items": deps["live_operation_audits"]( + instrument_id=instrument_id, + execution_ticket_id=execution_ticket_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/config/settings.py b/src/config/settings.py index 8024c8d..5f10158 100644 --- a/src/config/settings.py +++ b/src/config/settings.py @@ -6,6 +6,11 @@ class Settings(BaseSettings): database_connect_timeout: int = 5 okx_base_url: str = "https://www.okx.com" okx_timeout: int = 30 + live_trading_enabled: bool = False + okx_order_base_url: str = "https://www.okx.com" + okx_api_key: str = "" + okx_api_secret: str = "" + okx_api_passphrase: str = "" default_timezone: str = "UTC" session_timezone: str = "America/New_York" log_level: str = "INFO" diff --git a/src/models/__init__.py b/src/models/__init__.py index 8602873..6cc3be4 100644 --- a/src/models/__init__.py +++ b/src/models/__init__.py @@ -9,6 +9,7 @@ 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.live_operation_audit import LiveOperationAudit from src.models.liquidity_pool import LiquidityPool from src.models.market_session import MarketSession from src.models.structure_event import StructureEvent @@ -38,6 +39,7 @@ __all__ = [ "ExecutionTicket", "FVGZone", "Instrument", + "LiveOperationAudit", "LiquidityPool", "MarketSession", "StructureEvent", diff --git a/src/models/live_operation_audit.py b/src/models/live_operation_audit.py new file mode 100644 index 0000000..9ee12a1 --- /dev/null +++ b/src/models/live_operation_audit.py @@ -0,0 +1,28 @@ +from typing import Optional + +from sqlalchemy import Boolean, ForeignKey, Text +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from src.db.base import Base + + +class LiveOperationAudit(Base): + __tablename__ = "live_operation_audits" + + id: Mapped[int] = mapped_column(primary_key=True) + adapter_name: Mapped[str] = mapped_column(Text, nullable=False) + operation_type: Mapped[str] = mapped_column(Text, nullable=False) + status: Mapped[str] = mapped_column(Text, nullable=False) + execution_ticket_id: Mapped[Optional[int]] = mapped_column(ForeignKey("execution_tickets.id")) + execution_id: Mapped[Optional[int]] = mapped_column(ForeignKey("trade_executions.id")) + signal_id: Mapped[Optional[int]] = mapped_column(ForeignKey("trade_signals.id")) + instrument_id: Mapped[Optional[int]] = mapped_column(ForeignKey("instruments.id")) + requested_by: Mapped[Optional[str]] = mapped_column(Text) + requested_at: Mapped[object] = mapped_column(nullable=False) + completed_at: Mapped[Optional[object]] = mapped_column() + live_trading_enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + explicit_confirmation: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + payload: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) + result: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) + reason_codes: Mapped[list] = mapped_column(JSONB, nullable=False, default=list) diff --git a/src/repositories/execution_repository.py b/src/repositories/execution_repository.py index 63cfdbf..e06b185 100644 --- a/src/repositories/execution_repository.py +++ b/src/repositories/execution_repository.py @@ -316,6 +316,117 @@ class ExecutionRepository: rows = session.execute(sql, params).mappings().all() return [dict(row) for row in rows] + def insert_live_operation_audit( + self, + *, + adapter_name: str, + operation_type: str, + status: str, + execution_ticket_id: Optional[int], + execution_id: Optional[int], + signal_id: Optional[int], + instrument_id: Optional[int], + requested_by: Optional[str], + requested_at, + completed_at, + live_trading_enabled: bool, + explicit_confirmation: bool, + payload: dict, + result: dict, + reason_codes: list[str], + ) -> int: + sql = text( + """ + insert into live_operation_audits ( + adapter_name, + operation_type, + status, + execution_ticket_id, + execution_id, + signal_id, + instrument_id, + requested_by, + requested_at, + completed_at, + live_trading_enabled, + explicit_confirmation, + payload, + result, + reason_codes + ) values ( + :adapter_name, + :operation_type, + :status, + :execution_ticket_id, + :execution_id, + :signal_id, + :instrument_id, + :requested_by, + :requested_at, + :completed_at, + :live_trading_enabled, + :explicit_confirmation, + cast(:payload as jsonb), + cast(:result as jsonb), + cast(:reason_codes as jsonb) + ) + returning id + """ + ) + with SessionLocal.begin() as session: + result_row = session.execute( + sql, + { + "adapter_name": adapter_name, + "operation_type": operation_type, + "status": status, + "execution_ticket_id": execution_ticket_id, + "execution_id": execution_id, + "signal_id": signal_id, + "instrument_id": instrument_id, + "requested_by": requested_by, + "requested_at": requested_at, + "completed_at": completed_at, + "live_trading_enabled": live_trading_enabled, + "explicit_confirmation": explicit_confirmation, + "payload": json.dumps(payload, default=str), + "result": json.dumps(result, default=str), + "reason_codes": json.dumps(reason_codes, default=str), + }, + ) + return int(result_row.scalar_one()) + + def fetch_recent_live_operation_audits( + self, + *, + execution_ticket_id: Optional[int] = None, + instrument_id: Optional[int] = None, + limit: int = 20, + ) -> list[dict]: + filters = [] + params = {"limit": limit} + if execution_ticket_id is not None: + filters.append("execution_ticket_id = :execution_ticket_id") + params["execution_ticket_id"] = execution_ticket_id + if instrument_id is not None: + filters.append("instrument_id = :instrument_id") + params["instrument_id"] = instrument_id + where_sql = f"where {' and '.join(filters)}" if filters else "" + sql = text( + f""" + select id, adapter_name, operation_type, status, execution_ticket_id, execution_id, + signal_id, instrument_id, requested_by, requested_at, completed_at, + live_trading_enabled, explicit_confirmation, payload, result, reason_codes + from live_operation_audits + {where_sql} + order by requested_at desc nulls last, 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/broker/__init__.py b/src/services/broker/__init__.py new file mode 100644 index 0000000..c89530c --- /dev/null +++ b/src/services/broker/__init__.py @@ -0,0 +1,33 @@ +"""Live broker adapter package.""" + +from src.services.broker.broker_adapter import ( + BracketOrderRequest, + BrokerAdapter, + BrokerOperationResult, + CancelAllRequest, + KillSwitchRequest, + ReconciliationRequest, +) +from src.services.broker.live_broker_service import ( + LiveBrokerService, + LiveOperationDecision, + build_broker_adapter, + validate_live_dangerous_operation, + validate_live_submit_request, +) +from src.services.broker.okx_order_adapter import OKXOrderAdapter + +__all__ = [ + "BracketOrderRequest", + "BrokerAdapter", + "BrokerOperationResult", + "CancelAllRequest", + "KillSwitchRequest", + "LiveBrokerService", + "LiveOperationDecision", + "OKXOrderAdapter", + "ReconciliationRequest", + "build_broker_adapter", + "validate_live_dangerous_operation", + "validate_live_submit_request", +] diff --git a/src/services/broker/broker_adapter.py b/src/services/broker/broker_adapter.py new file mode 100644 index 0000000..6c0ef64 --- /dev/null +++ b/src/services/broker/broker_adapter.py @@ -0,0 +1,88 @@ +from abc import ABC, abstractmethod +from dataclasses import asdict, dataclass +from typing import Optional + + +@dataclass(frozen=True) +class BracketOrderRequest: + execution_ticket_id: int + signal_id: int + instrument_id: int + side: str + entry_price: float + size: float + stop_loss: float + take_profit_1: float + take_profit_2: float + oco_group_id: Optional[str] + ticket_snapshot: dict + + def to_dict(self) -> dict: + return dict(asdict(self)) + + +@dataclass(frozen=True) +class ReconciliationRequest: + instrument_id: Optional[int] + requested_by: Optional[str] + + def to_dict(self) -> dict: + return dict(asdict(self)) + + +@dataclass(frozen=True) +class CancelAllRequest: + instrument_id: Optional[int] + requested_by: Optional[str] + + def to_dict(self) -> dict: + return dict(asdict(self)) + + +@dataclass(frozen=True) +class KillSwitchRequest: + reason: str + requested_by: Optional[str] + + def to_dict(self) -> dict: + return dict(asdict(self)) + + +@dataclass(frozen=True) +class BrokerOperationResult: + ok: bool + status: str + adapter_name: str + operation_type: str + reason_codes: list[str] + details: dict + + def to_dict(self) -> dict: + return { + "ok": self.ok, + "status": self.status, + "adapter_name": self.adapter_name, + "operation_type": self.operation_type, + "reason_codes": list(self.reason_codes), + "details": dict(self.details), + } + + +class BrokerAdapter(ABC): + adapter_name: str + + @abstractmethod + def submit_bracket_order(self, request: BracketOrderRequest) -> BrokerOperationResult: + raise NotImplementedError + + @abstractmethod + def reconcile_orders(self, request: ReconciliationRequest) -> BrokerOperationResult: + raise NotImplementedError + + @abstractmethod + def cancel_all(self, request: CancelAllRequest) -> BrokerOperationResult: + raise NotImplementedError + + @abstractmethod + def activate_kill_switch(self, request: KillSwitchRequest) -> BrokerOperationResult: + raise NotImplementedError diff --git a/src/services/broker/live_broker_service.py b/src/services/broker/live_broker_service.py new file mode 100644 index 0000000..9040bb6 --- /dev/null +++ b/src/services/broker/live_broker_service.py @@ -0,0 +1,311 @@ +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Optional + +from src.config.settings import settings +from src.repositories.execution_repository import ExecutionRepository +from src.services.broker.broker_adapter import ( + BracketOrderRequest, + BrokerOperationResult, + CancelAllRequest, + KillSwitchRequest, + ReconciliationRequest, +) +from src.services.broker.okx_order_adapter import OKXOrderAdapter + + +@dataclass(frozen=True) +class LiveOperationDecision: + allowed: bool + status: str + reason_codes: list[str] + adapter_name: str + operation_type: str + audit_id: int + result: dict + + def to_dict(self) -> dict: + return { + "allowed": self.allowed, + "status": self.status, + "reason_codes": list(self.reason_codes), + "adapter_name": self.adapter_name, + "operation_type": self.operation_type, + "audit_id": self.audit_id, + "result": dict(self.result), + } + + +class LiveBrokerService: + def __init__(self, repository: Optional[ExecutionRepository] = None) -> None: + self.repository = repository or ExecutionRepository() + + def submit_confirmed_ticket( + self, + ticket_id: int, + *, + requested_by: str, + explicit_live_confirmation: bool = False, + adapter_name: str = "okx", + ) -> Optional[LiveOperationDecision]: + ticket = self.repository.fetch_execution_ticket_by_id(ticket_id) + if not ticket: + return None + signal = self.repository.fetch_signal_for_execution(int(ticket["signal_id"])) + if not signal: + return None + reason_codes = validate_live_submit_request( + ticket=ticket, + live_trading_enabled=settings.live_trading_enabled, + explicit_live_confirmation=explicit_live_confirmation, + ) + payload = { + "ticket_id": ticket_id, + "execution_mode": ticket.get("execution_mode"), + "signal_snapshot": dict((ticket.get("ticket_snapshot") or {}).get("signal") or {}), + } + if reason_codes: + return self._record_live_audit( + adapter_name=adapter_name, + operation_type="submit_bracket_order", + status="rejected", + requested_by=requested_by, + execution_ticket_id=ticket_id, + signal_id=int(ticket["signal_id"]), + instrument_id=int(signal["instrument_id"]), + explicit_confirmation=explicit_live_confirmation, + payload=payload, + result={"rejected": True}, + reason_codes=reason_codes, + ) + adapter = build_broker_adapter(adapter_name) + snapshot = dict(ticket.get("ticket_snapshot") or {}) + signal_snapshot = dict(snapshot.get("signal") or {}) + decision_snapshot = dict(snapshot.get("decision") or {}) + request = BracketOrderRequest( + execution_ticket_id=ticket_id, + signal_id=int(ticket["signal_id"]), + instrument_id=int(signal["instrument_id"]), + side=str(signal_snapshot.get("side") or signal.get("side")), + entry_price=float(decision_snapshot.get("entry_price") or 0.0), + size=float(decision_snapshot.get("size") or 0.0), + stop_loss=float((decision_snapshot.get("meta") or {}).get("stop_loss") or 0.0), + take_profit_1=float((decision_snapshot.get("meta") or {}).get("tp1") or 0.0), + take_profit_2=float((decision_snapshot.get("meta") or {}).get("tp2") or 0.0), + oco_group_id=f"ticket:{ticket_id}", + ticket_snapshot=snapshot, + ) + result = adapter.submit_bracket_order(request) + return self._record_live_audit( + adapter_name=adapter_name, + operation_type="submit_bracket_order", + status=result.status, + requested_by=requested_by, + execution_ticket_id=ticket_id, + execution_id=ticket.get("execution_id"), + signal_id=int(ticket["signal_id"]), + instrument_id=int(signal["instrument_id"]), + explicit_confirmation=explicit_live_confirmation, + payload=request.to_dict(), + result=result.to_dict(), + reason_codes=result.reason_codes, + ) + + def reconcile_orders( + self, + *, + requested_by: str, + instrument_id: Optional[int] = None, + adapter_name: str = "okx", + ) -> LiveOperationDecision: + if not settings.live_trading_enabled: + return self._record_live_audit( + adapter_name=adapter_name, + operation_type="reconcile_orders", + status="rejected", + requested_by=requested_by, + instrument_id=instrument_id, + explicit_confirmation=False, + payload={"instrument_id": instrument_id}, + result={"rejected": True}, + reason_codes=["live_trading_disabled"], + ) + adapter = build_broker_adapter(adapter_name) + request = ReconciliationRequest(instrument_id=instrument_id, requested_by=requested_by) + result = adapter.reconcile_orders(request) + return self._record_live_audit( + adapter_name=adapter_name, + operation_type="reconcile_orders", + status=result.status, + requested_by=requested_by, + instrument_id=instrument_id, + explicit_confirmation=False, + payload=request.to_dict(), + result=result.to_dict(), + reason_codes=result.reason_codes, + ) + + def cancel_all( + self, + *, + requested_by: str, + instrument_id: Optional[int] = None, + explicit_live_confirmation: bool = False, + adapter_name: str = "okx", + ) -> LiveOperationDecision: + reason_codes = validate_live_dangerous_operation( + live_trading_enabled=settings.live_trading_enabled, + explicit_live_confirmation=explicit_live_confirmation, + ) + request = CancelAllRequest(instrument_id=instrument_id, requested_by=requested_by) + if reason_codes: + return self._record_live_audit( + adapter_name=adapter_name, + operation_type="cancel_all", + status="rejected", + requested_by=requested_by, + instrument_id=instrument_id, + explicit_confirmation=explicit_live_confirmation, + payload=request.to_dict(), + result={"rejected": True}, + reason_codes=reason_codes, + ) + adapter = build_broker_adapter(adapter_name) + result = adapter.cancel_all(request) + return self._record_live_audit( + adapter_name=adapter_name, + operation_type="cancel_all", + status=result.status, + requested_by=requested_by, + instrument_id=instrument_id, + explicit_confirmation=explicit_live_confirmation, + payload=request.to_dict(), + result=result.to_dict(), + reason_codes=result.reason_codes, + ) + + def activate_kill_switch( + self, + *, + requested_by: str, + reason: str, + explicit_live_confirmation: bool = False, + adapter_name: str = "okx", + ) -> LiveOperationDecision: + reason_codes = validate_live_dangerous_operation( + live_trading_enabled=settings.live_trading_enabled, + explicit_live_confirmation=explicit_live_confirmation, + ) + request = KillSwitchRequest(reason=reason, requested_by=requested_by) + if reason_codes: + return self._record_live_audit( + adapter_name=adapter_name, + operation_type="activate_kill_switch", + status="rejected", + requested_by=requested_by, + explicit_confirmation=explicit_live_confirmation, + payload=request.to_dict(), + result={"rejected": True}, + reason_codes=reason_codes, + ) + adapter = build_broker_adapter(adapter_name) + result = adapter.activate_kill_switch(request) + return self._record_live_audit( + adapter_name=adapter_name, + operation_type="activate_kill_switch", + status=result.status, + requested_by=requested_by, + explicit_confirmation=explicit_live_confirmation, + payload=request.to_dict(), + result=result.to_dict(), + reason_codes=result.reason_codes, + ) + + def _record_live_audit( + self, + *, + adapter_name: str, + operation_type: str, + status: str, + requested_by: Optional[str], + explicit_confirmation: bool, + payload: dict, + result: dict, + reason_codes: list[str], + execution_ticket_id: Optional[int] = None, + execution_id: Optional[int] = None, + signal_id: Optional[int] = None, + instrument_id: Optional[int] = None, + ) -> LiveOperationDecision: + now = datetime.now(timezone.utc) + audit_id = self.repository.insert_live_operation_audit( + adapter_name=adapter_name, + operation_type=operation_type, + status=status, + execution_ticket_id=execution_ticket_id, + execution_id=execution_id, + signal_id=signal_id, + instrument_id=instrument_id, + requested_by=requested_by, + requested_at=now, + completed_at=now, + live_trading_enabled=bool(settings.live_trading_enabled), + explicit_confirmation=explicit_confirmation, + payload=payload, + result=result, + reason_codes=reason_codes, + ) + return LiveOperationDecision( + allowed=status not in {"rejected"}, + status=status, + reason_codes=list(reason_codes), + adapter_name=adapter_name, + operation_type=operation_type, + audit_id=audit_id, + result=result, + ) + + +def build_broker_adapter(adapter_name: str): + if adapter_name == "okx": + return OKXOrderAdapter() + raise ValueError(f"unsupported_broker_adapter:{adapter_name}") + + +def validate_live_submit_request( + *, + ticket: dict, + live_trading_enabled: bool, + explicit_live_confirmation: bool, +) -> list[str]: + reason_codes = [] + if not live_trading_enabled: + reason_codes.append("live_trading_disabled") + if not explicit_live_confirmation: + reason_codes.append("explicit_live_confirmation_required") + if ticket.get("status") != "confirmed": + reason_codes.append("execution_ticket_not_confirmed") + if not ticket.get("confirmed_by") or not ticket.get("confirmed_at"): + reason_codes.append("execution_ticket_missing_confirmation_metadata") + return _dedupe(reason_codes) + + +def validate_live_dangerous_operation(*, live_trading_enabled: bool, explicit_live_confirmation: bool) -> list[str]: + reason_codes = [] + if not live_trading_enabled: + reason_codes.append("live_trading_disabled") + if not explicit_live_confirmation: + reason_codes.append("explicit_live_confirmation_required") + return _dedupe(reason_codes) + + +def _dedupe(values: list[str]) -> list[str]: + seen = set() + result = [] + for value in values: + text = str(value or "").strip() + if not text or text in seen: + continue + seen.add(text) + result.append(text) + return result diff --git a/src/services/broker/okx_order_adapter.py b/src/services/broker/okx_order_adapter.py new file mode 100644 index 0000000..16503dc --- /dev/null +++ b/src/services/broker/okx_order_adapter.py @@ -0,0 +1,75 @@ +from typing import Optional + +from src.config.settings import settings +from src.services.broker.broker_adapter import ( + BrokerAdapter, + BrokerOperationResult, + BracketOrderRequest, + CancelAllRequest, + KillSwitchRequest, + ReconciliationRequest, +) + + +class OKXOrderAdapter(BrokerAdapter): + adapter_name = "okx" + + def __init__(self, base_url: Optional[str] = None) -> None: + self.base_url = base_url or settings.okx_order_base_url + + def submit_bracket_order(self, request: BracketOrderRequest) -> BrokerOperationResult: + return BrokerOperationResult( + ok=False, + status="placeholder_not_implemented", + adapter_name=self.adapter_name, + operation_type="submit_bracket_order", + reason_codes=["okx_order_adapter_placeholder"], + details={ + "base_url": self.base_url, + "supports_bracket": True, + "supports_oco": True, + "request": request.to_dict(), + }, + ) + + def reconcile_orders(self, request: ReconciliationRequest) -> BrokerOperationResult: + return BrokerOperationResult( + ok=True, + status="placeholder_reconciliation", + adapter_name=self.adapter_name, + operation_type="reconcile_orders", + reason_codes=[], + details={ + "base_url": self.base_url, + "request": request.to_dict(), + "supports_reconciliation": True, + }, + ) + + def cancel_all(self, request: CancelAllRequest) -> BrokerOperationResult: + return BrokerOperationResult( + ok=False, + status="placeholder_not_implemented", + adapter_name=self.adapter_name, + operation_type="cancel_all", + reason_codes=["okx_order_adapter_placeholder"], + details={ + "base_url": self.base_url, + "request": request.to_dict(), + "supports_cancel_all": True, + }, + ) + + def activate_kill_switch(self, request: KillSwitchRequest) -> BrokerOperationResult: + return BrokerOperationResult( + ok=True, + status="placeholder_kill_switch_active", + adapter_name=self.adapter_name, + operation_type="activate_kill_switch", + reason_codes=[], + details={ + "base_url": self.base_url, + "request": request.to_dict(), + "supports_kill_switch": True, + }, + ) diff --git a/tests/unit/test_api_server.py b/tests/unit/test_api_server.py index b350fcd..b514c06 100644 --- a/tests/unit/test_api_server.py +++ b/tests/unit/test_api_server.py @@ -93,6 +93,18 @@ class ApiServerRoutingTests(unittest.TestCase): self.assertEqual(status, 200) self.assertEqual(payload["items"], [{"id": 1, "status": "awaiting_confirmation"}]) + def test_routes_live_operation_audits_query(self) -> None: + status, payload = route_request( + "/live-operation-audits", + {"execution_ticket_id": ["1"], "limit": ["2"]}, + dependencies={ + "live_operation_audits": lambda **kwargs: [{"id": 1, "status": "rejected"}], + }, + ) + + self.assertEqual(status, 200) + self.assertEqual(payload["items"], [{"id": 1, "status": "rejected"}]) + def test_routes_health(self) -> None: status, payload = route_request( "/health", diff --git a/tests/unit/test_live_adapter_script.py b/tests/unit/test_live_adapter_script.py new file mode 100644 index 0000000..376ed9c --- /dev/null +++ b/tests/unit/test_live_adapter_script.py @@ -0,0 +1,37 @@ +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_live_adapter + + +class LiveAdapterScriptTests(unittest.TestCase): + def test_main_reports_database_unavailable_json(self) -> None: + output = io.StringIO() + + with patch( + "scripts.run_live_adapter.LiveBrokerService.reconcile_orders", + side_effect=OperationalError("stmt", {}, Exception()), + ), contextlib.redirect_stdout(output): + exit_code = run_live_adapter.main( + env={ + "LIVE_ADAPTER_OUTPUT_JSON": "1", + "LIVE_ADAPTER_OPERATION": "reconcile", + } + ) + + 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() diff --git a/tests/unit/test_live_adapter_service.py b/tests/unit/test_live_adapter_service.py new file mode 100644 index 0000000..93e6703 --- /dev/null +++ b/tests/unit/test_live_adapter_service.py @@ -0,0 +1,158 @@ +import os +import unittest +from unittest.mock import patch + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test") + +from src.config.settings import settings +from src.services.broker import LiveBrokerService, validate_live_submit_request + + +class LiveBrokerServiceTests(unittest.TestCase): + def test_submit_confirmed_ticket_rejects_when_live_disabled(self) -> None: + repo = _LiveRepo() + service = LiveBrokerService(repository=repo) + + with patch.object(settings, "live_trading_enabled", False): + result = service.submit_confirmed_ticket( + 10, + requested_by="alice", + explicit_live_confirmation=True, + ) + + self.assertIsNotNone(result) + assert result is not None + self.assertEqual(result.status, "rejected") + self.assertIn("live_trading_disabled", result.reason_codes) + self.assertEqual(repo.live_audits[0]["operation_type"], "submit_bracket_order") + + def test_submit_confirmed_ticket_requires_explicit_confirmation(self) -> None: + repo = _LiveRepo() + service = LiveBrokerService(repository=repo) + + with patch.object(settings, "live_trading_enabled", True): + result = service.submit_confirmed_ticket( + 10, + requested_by="alice", + explicit_live_confirmation=False, + ) + + self.assertEqual(result.status, "rejected") + self.assertIn("explicit_live_confirmation_required", result.reason_codes) + + def test_submit_confirmed_ticket_calls_adapter_when_enabled_and_confirmed(self) -> None: + repo = _LiveRepo() + service = LiveBrokerService(repository=repo) + + with patch.object(settings, "live_trading_enabled", True): + result = service.submit_confirmed_ticket( + 10, + requested_by="alice", + explicit_live_confirmation=True, + ) + + self.assertEqual(result.status, "placeholder_not_implemented") + self.assertEqual(repo.live_audits[-1]["adapter_name"], "okx") + self.assertEqual(repo.live_audits[-1]["execution_ticket_id"], 10) + + def test_reconcile_is_audited(self) -> None: + repo = _LiveRepo() + service = LiveBrokerService(repository=repo) + + with patch.object(settings, "live_trading_enabled", True): + result = service.reconcile_orders(requested_by="ops", instrument_id=1) + + self.assertEqual(result.status, "placeholder_reconciliation") + self.assertEqual(repo.live_audits[-1]["operation_type"], "reconcile_orders") + + def test_cancel_all_requires_explicit_confirmation(self) -> None: + repo = _LiveRepo() + service = LiveBrokerService(repository=repo) + + with patch.object(settings, "live_trading_enabled", True): + result = service.cancel_all(requested_by="ops", explicit_live_confirmation=False) + + self.assertEqual(result.status, "rejected") + self.assertIn("explicit_live_confirmation_required", result.reason_codes) + + def test_kill_switch_is_audited(self) -> None: + repo = _LiveRepo() + service = LiveBrokerService(repository=repo) + + with patch.object(settings, "live_trading_enabled", True): + result = service.activate_kill_switch( + requested_by="ops", + reason="manual_test", + explicit_live_confirmation=True, + ) + + self.assertEqual(result.status, "placeholder_kill_switch_active") + self.assertEqual(repo.live_audits[-1]["operation_type"], "activate_kill_switch") + + +class LiveValidationTests(unittest.TestCase): + def test_validate_live_submit_request_requires_confirmed_ticket(self) -> None: + reasons = validate_live_submit_request( + ticket={"status": "awaiting_confirmation", "confirmed_by": None, "confirmed_at": None}, + live_trading_enabled=False, + explicit_live_confirmation=False, + ) + + self.assertEqual( + reasons, + [ + "live_trading_disabled", + "explicit_live_confirmation_required", + "execution_ticket_not_confirmed", + "execution_ticket_missing_confirmation_metadata", + ], + ) + + +class _LiveRepo: + def __init__(self) -> None: + self.live_audits = [] + self.tickets = { + 10: { + "id": 10, + "signal_id": 1, + "execution_mode": "semi_auto", + "status": "confirmed", + "confirmed_by": "bob", + "confirmed_at": "2026-05-06T00:00:00+00:00", + "execution_id": 20, + "ticket_snapshot": { + "signal": { + "side": "buy", + "signal_id": 1, + }, + "decision": { + "entry_price": 100.5, + "size": 1.0, + "meta": { + "stop_loss": 99.0, + "tp1": 102.0, + "tp2": 104.0, + }, + }, + }, + } + } + + def fetch_execution_ticket_by_id(self, ticket_id: int): + return dict(self.tickets.get(ticket_id)) if ticket_id in self.tickets else None + + def fetch_signal_for_execution(self, signal_id: int): + return { + "id": signal_id, + "instrument_id": 1, + "side": "buy", + } + + def insert_live_operation_audit(self, **kwargs): + self.live_audits.append(kwargs) + return len(self.live_audits) + + +if __name__ == "__main__": + unittest.main()