"""Order Manager — single point of draft/live divergence for trade execution.""" from __future__ import annotations import asyncio import psycopg import uuid from datetime import datetime, timezone from decimal import Decimal from typing import Callable import structlog from src.config import SigilConfig from src.contracts import AlertEvent, OrderResult, OrderState, RiskDecision, TradeSignal from src.executor.draft_trader import DraftTrader from src.symbols import to_ccxt_symbol log = structlog.get_logger(__name__) # How long to wait for a partial fill to complete before cancelling _PARTIAL_FILL_TIMEOUT_SECONDS = 300 # 5 minutes # Minimum notional fill size below which we market-sell instead of holding _MIN_FILL_NOTIONAL_USD = Decimal("10") class OrderManager: """Single point where draft/live divergence occurs. In draft mode all calls are routed to DraftTrader.simulate_fill(). In live mode real ccxt exchange calls are made. """ def __init__( self, config: SigilConfig, exchange, # ccxt async exchange instance db_conn: psycopg.Connection, draft_trader: DraftTrader, alert_fn: Callable[[AlertEvent], None], market_feed=None, # NEW — MarketFeed instance for draft price injection risk_manager=None, # Optional RiskManager for recording successful trades ) -> None: self._config = config self._exchange = exchange self._db = db_conn self._draft_trader = draft_trader self._alert = alert_fn self._market_feed = market_feed self._risk_manager = risk_manager self._log = log.bind(component="order_manager") # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ async def place_order( self, signal: TradeSignal, risk_decision: RiskDecision ) -> OrderResult: """Place an order through draft simulator or live exchange. This is the single divergence point: draft mode routes to DraftTrader.simulate_fill(); live mode calls _exchange_order(). """ if self._config.mode == "draft": if self._market_feed is not None: raw_price = self._market_feed.get_price(signal.symbol) if raw_price is None or raw_price <= 0: self._log.warning( "draft_fill_rejected_no_price", symbol=signal.symbol, ) try: self._db.execute( "INSERT INTO trades (" "exchange, symbol, side, order_type, quantity, status, " "order_state, strategy_tier, signal_confidence, signal_id, " "client_order_id, mode, reason" ") VALUES (" "'draft', %s, %s, 'market', '0', 'REJECTED', " "'REJECTED', %s, %s, %s, %s, 'draft', 'no_market_price'" ")", ( signal.symbol, signal.direction, signal.strategy_tier, signal.confidence, str(signal.signal_id), f"rejected-{signal.strategy_tier}", ), ) self._db.commit() except Exception: pass return self._make_rejected_result(signal, "no_market_price") current_price = Decimal(str(raw_price)) return self._draft_trader.simulate_fill( signal, risk_decision, current_price=current_price ) return self._draft_trader.simulate_fill(signal, risk_decision) else: return await self._exchange_order(signal, risk_decision) @staticmethod def _make_rejected_result(signal: TradeSignal, reason: str) -> OrderResult: """Return a REJECTED OrderResult without touching the exchange or DB.""" return OrderResult( order_id=None, client_order_id=f"rejected-{signal.strategy_tier}", status=OrderState.REJECTED, filled_qty=Decimal("0"), avg_price=Decimal("0"), fee=Decimal("0"), mode="draft", timestamp=datetime.now(timezone.utc), ) # ------------------------------------------------------------------ # Live exchange execution # ------------------------------------------------------------------ async def _exchange_order( self, signal: TradeSignal, risk_decision: RiskDecision ) -> OrderResult: """Place a real order via ccxt and manage its lifecycle.""" position_id = str(uuid.uuid4())[:8] client_order_id = f"{signal.strategy_tier}-{position_id}-bot" ccxt_symbol = to_ccxt_symbol(signal.symbol) quantity = float(risk_decision.adjusted_quantity or 0) self._log.info( "exchange_order_placing", signal_id=str(signal.signal_id), symbol=signal.symbol, direction=signal.direction, quantity=quantity, client_order_id=client_order_id, ) # Record SUBMITTED state before touching exchange self._record_trade_submitted( signal, risk_decision, client_order_id, position_id ) try: if signal.strategy_tier == "pump" and signal.direction == "buy": result = await self._place_pump_order_with_stop( signal, risk_decision, ccxt_symbol, quantity, client_order_id ) else: result = await self._place_standard_order( signal, ccxt_symbol, quantity, client_order_id ) except Exception as exc: self._log.exception( "exchange_order_failed", signal_id=str(signal.signal_id), symbol=signal.symbol, error=str(exc), ) self._update_trade_state( client_order_id, OrderState.REJECTED, reason=str(exc) ) self._alert( AlertEvent( event_type="order_failed", symbol=signal.symbol, severity="critical", message=f"Order placement failed for {signal.symbol}: {exc}", mode="live", ) ) raise # Handle partial fills if result.status == OrderState.PARTIAL_ACTIVE: result = await self._handle_partial_fill( signal, result, ccxt_symbol, quantity, client_order_id ) self._update_trade_filled(signal, result, client_order_id) if self._risk_manager and result.filled_qty > 0: self._risk_manager.record_successful_trade() self._log.info( "exchange_order_complete", signal_id=str(signal.signal_id), symbol=signal.symbol, status=result.status, filled_qty=str(result.filled_qty), avg_price=str(result.avg_price), client_order_id=client_order_id, ) return result async def _place_standard_order( self, signal: TradeSignal, ccxt_symbol: str, quantity: float, client_order_id: str, ) -> OrderResult: """Place a standard market limit order.""" raw = await self._exchange.create_order( symbol=ccxt_symbol, type="market", side=signal.direction, amount=quantity, params={"clientOrderId": client_order_id}, ) return self._parse_ccxt_order(raw, client_order_id) async def _place_pump_order_with_stop( self, signal: TradeSignal, risk_decision: RiskDecision, ccxt_symbol: str, quantity: float, client_order_id: str, ) -> OrderResult: """Place a pump buy with an exchange-side OCO stop-loss at entry time.""" # Primary buy order raw = await self._exchange.create_order( symbol=ccxt_symbol, type="market", side="buy", amount=quantity, params={"clientOrderId": client_order_id}, ) result = self._parse_ccxt_order(raw, client_order_id) # Attach stop-loss once fill price is known fill_price = float(result.avg_price) if result.avg_price > 0 else None stop_price = ( float(risk_decision.stop_loss_price) if risk_decision.stop_loss_price else None ) if fill_price and not stop_price: # Derive stop from fill + config percentage stop_pct = self._config.pump.stop_loss_pct / 100.0 stop_price = fill_price * (1.0 - stop_pct) if stop_price: try: sl_client_id = f"sl-{client_order_id}" # STOP_LOSS_LIMIT needs both stopPrice (trigger) and price (limit). # Limit price set 0.5% below trigger to ensure fill. sl_limit_price = stop_price * 0.995 await self._exchange.create_order( symbol=ccxt_symbol, type="STOP_LOSS_LIMIT", side="sell", amount=float(result.filled_qty) or quantity, price=sl_limit_price, params={ "stopPrice": stop_price, "clientOrderId": sl_client_id, }, ) self._log.info( "pump_stop_loss_placed", symbol=signal.symbol, stop_price=stop_price, sl_client_id=sl_client_id, ) except Exception: self._log.exception( "pump_stop_loss_placement_failed", symbol=signal.symbol, stop_price=stop_price, ) self._alert( AlertEvent( event_type="stop_loss_failed", symbol=signal.symbol, severity="critical", message=( f"Failed to place stop-loss for {signal.symbol} pump position. " f"Manual intervention required." ), mode="live", ) ) return result async def _handle_partial_fill( self, signal: TradeSignal, initial_result: OrderResult, ccxt_symbol: str, original_quantity: float, client_order_id: str, ) -> OrderResult: """Wait up to 5 minutes for a partial fill to complete. If incomplete after timeout, cancel remainder. If filled_qty * price < minimum notional, market-sell. """ self._log.info( "partial_fill_waiting", symbol=signal.symbol, filled_so_far=str(initial_result.filled_qty), client_order_id=client_order_id, ) exchange_order_id = initial_result.order_id deadline = asyncio.get_event_loop().time() + _PARTIAL_FILL_TIMEOUT_SECONDS result = initial_result while asyncio.get_event_loop().time() < deadline: await asyncio.sleep(10) try: raw = await self._exchange.fetch_order(exchange_order_id, ccxt_symbol) result = self._parse_ccxt_order(raw, client_order_id) if result.status in (OrderState.FILLED, OrderState.CANCELLED): return result except Exception: self._log.exception( "partial_fill_status_check_failed", symbol=signal.symbol, order_id=exchange_order_id, ) # Timeout reached — cancel remainder self._log.warning( "partial_fill_timeout", symbol=signal.symbol, client_order_id=client_order_id, filled_qty=str(result.filled_qty), ) try: await self._exchange.cancel_order(exchange_order_id, ccxt_symbol) except Exception: self._log.exception( "partial_fill_cancel_failed", symbol=signal.symbol, order_id=exchange_order_id, ) # If filled amount is below minimum, market-sell to exit filled_notional = result.filled_qty * result.avg_price if filled_notional > 0 and filled_notional < _MIN_FILL_NOTIONAL_USD: self._log.warning( "partial_fill_below_minimum_selling", symbol=signal.symbol, filled_notional=str(filled_notional), threshold=str(_MIN_FILL_NOTIONAL_USD), ) try: await self._exchange.create_order( symbol=ccxt_symbol, type="market", side="sell", amount=float(result.filled_qty), params={"clientOrderId": f"cleanup-{client_order_id}"}, ) except Exception: self._log.exception( "partial_fill_cleanup_sell_failed", symbol=signal.symbol, ) self._alert( AlertEvent( event_type="cleanup_sell_failed", symbol=signal.symbol, severity="critical", message=( f"Failed to cleanup sub-minimum partial fill for {signal.symbol}. " f"Manual intervention required." ), mode="live", ) ) return OrderResult( order_id=result.order_id, client_order_id=client_order_id, status=OrderState.PARTIAL_TIMEOUT, filled_qty=result.filled_qty, avg_price=result.avg_price, fee=result.fee, mode="live", ) # ------------------------------------------------------------------ # ccxt response parsing # ------------------------------------------------------------------ @staticmethod def _parse_ccxt_order(raw: dict, client_order_id: str) -> OrderResult: """Convert a raw ccxt order dict to OrderResult.""" status_map = { "open": OrderState.SUBMITTED, "partially_filled": OrderState.PARTIAL_ACTIVE, "closed": OrderState.FILLED, "canceled": OrderState.CANCELLED, "cancelled": OrderState.CANCELLED, "rejected": OrderState.REJECTED, "expired": OrderState.CANCELLED, } raw_status = (raw.get("status") or "open").lower() filled = Decimal(str(raw.get("filled") or 0)) remaining = Decimal(str(raw.get("remaining") or 0)) # Detect partial fill regardless of exchange status string if filled > 0 and remaining > 0: state = OrderState.PARTIAL_ACTIVE else: state = status_map.get(raw_status, OrderState.SUBMITTED) avg_price = Decimal(str(raw.get("average") or raw.get("price") or 0)) fee_info = raw.get("fee") or {} fee = Decimal(str(fee_info.get("cost", 0) or 0)) return OrderResult( order_id=str(raw.get("id", "")), client_order_id=raw.get("clientOrderId") or client_order_id, status=state, filled_qty=filled, avg_price=avg_price, fee=fee, mode="live", timestamp=datetime.now(timezone.utc), ) # ------------------------------------------------------------------ # DB persistence # ------------------------------------------------------------------ def _record_trade_submitted( self, signal: TradeSignal, risk_decision: RiskDecision, client_order_id: str, position_id: str, ) -> None: try: self._db.execute( """ INSERT INTO trades ( exchange, symbol, side, order_type, quantity, status, order_state, strategy_tier, signal_confidence, signal_id, client_order_id, mode, reason ) VALUES ( %s, %s, %s, 'market', %s, 'SUBMITTED', 'SUBMITTED', %s, %s, %s, %s, 'live', %s ) """, ( self._config.exchange.exchange_id, signal.symbol, signal.direction, str(risk_decision.adjusted_quantity or 0), signal.strategy_tier, signal.confidence, str(signal.signal_id), client_order_id, f"pos:{position_id}", ), ) self._db.commit() except Exception: self._log.exception( "trade_record_submitted_failed", signal_id=str(signal.signal_id), ) def _update_trade_state( self, client_order_id: str, state: OrderState, reason: str = "", ) -> None: try: self._db.execute( """ UPDATE trades SET order_state = %s, reason = %s, updated_at = NOW() WHERE client_order_id = %s AND mode = 'live' """, (state.value, reason, client_order_id), ) self._db.commit() except Exception: self._log.exception( "trade_state_update_failed", client_order_id=client_order_id, state=state.value, ) def _update_trade_filled( self, signal: TradeSignal, result: OrderResult, client_order_id: str, ) -> None: try: self._db.execute( """ UPDATE trades SET order_state = %s, filled_quantity = %s, filled_avg_price = %s, exchange_order_id = %s, updated_at = NOW() WHERE client_order_id = %s AND mode = 'live' """, ( result.status.value, str(result.filled_qty), str(result.avg_price), result.order_id or "", client_order_id, ), ) self._db.commit() except Exception: self._log.exception( "trade_filled_update_failed", signal_id=str(signal.signal_id), client_order_id=client_order_id, )