""" APEX V16 — DryRunBroker. Composable wrapper that turns any BrokerBase implementation into a "dry run" broker: real market data, real account balance, real positions — but every WRITE method is a no-op that logs the intent and returns a synthetic OrderResult. Useful for pre-LIVE validation against real Topstep data without putting capital at risk. Usage: real = TopstepXBroker(instruments=[...]) await _connect_with_retry(real, ...) broker = DryRunBroker(real, logger=logger) # writes intercepted # broker is a drop-in BrokerBase Method classification (this is the contract): READ / LIFECYCLE — DELEGATED to wrapped broker: connect # wrapped.connect — real WebSocket session disconnect # MUST delegate. Otherwise zombie WS at shutdown. is_connected # real connection state get_last_price # real market price positions_get # real broker-side positions pending_orders # real broker-side pending orders (Reconciler) recent_trades # real broker-side trade history (Reconciler case ii) get_account_balance# real account equity fetch_bars # real OHLCV bars WRITE — INTERCEPTED with no-op + structured log + synthetic success: place_market_bracket place_stop_order # used by orphan SL re-attach (C2b) place_limit_order # used by orphan TP re-attach (C2b) cancel_order cancel_all_for_symbol close_position modify_stop The synthetic OrderResult includes "DRY-..." order IDs so downstream state mutations are tracked and the JSONL forensics show exactly which trades would have happened on a real run. NOTE on the disconnect classification (explicit per reviewer ask): disconnect is LIFECYCLE not WRITE. Intercepting it would leave the underlying WebSocket alive after shutdown, leaking a connection per session. Real shutdown of the upstream socket is required. """ from __future__ import annotations import logging import uuid from datetime import datetime from typing import Optional from broker.broker_base import ( BrokerBase, CancelResult, ClosedTrade, Order, OrderResult, Position, ) log = logging.getLogger("dry_run_broker") class DryRunBroker(BrokerBase): """BrokerBase wrapper. See module docstring for read/write split.""" name = "DryRun" def __init__(self, wrapped: BrokerBase, logger=None) -> None: """ Args: wrapped: a connected (or about-to-connect) real BrokerBase. logger: LoggerBundle. If provided, write-method intercepts emit structured events to brain_log. """ self._wrapped = wrapped self._logger = logger # ============================================================ # READ / LIFECYCLE — DELEGATED # ============================================================ async def connect(self) -> bool: return await self._wrapped.connect() async def disconnect(self) -> None: # LIFECYCLE — not write. Must close upstream session/WebSocket. await self._wrapped.disconnect() async def is_connected(self) -> bool: return await self._wrapped.is_connected() async def get_last_price(self, symbol: str) -> Optional[float]: return await self._wrapped.get_last_price(symbol) async def positions_get(self, symbol: Optional[str] = None) -> list[Position]: return await self._wrapped.positions_get(symbol) async def pending_orders(self, symbol: Optional[str] = None) -> list[Order]: return await self._wrapped.pending_orders(symbol) async def recent_trades( self, symbol: Optional[str] = None, since: Optional[datetime] = None, limit: int = 50, ) -> list[ClosedTrade]: return await self._wrapped.recent_trades(symbol, since, limit) async def get_account_balance(self) -> float: return await self._wrapped.get_account_balance() async def fetch_bars(self, symbol: str, timeframe: str, n: int): return await self._wrapped.fetch_bars(symbol, timeframe, n) # ============================================================ # WRITE — INTERCEPTED # ============================================================ async def place_market_bracket( self, symbol: str, direction: str, contracts: int, sl_price: float, tp_price: float, *, sl_absolute_price: Optional[float] = None, sl_ticks: Optional[int] = None, tp_ticks: Optional[int] = None, sl_source: str = "ATR", ) -> OrderResult: # Try to enrich the synthetic entry price from a real last-price # query (best-effort; failures don't break the dry path). entry_px: Optional[float] = None try: entry_px = await self._wrapped.get_last_price(symbol) except Exception: entry_px = None ids = self._dry_ids("place") self._log_dry( "dry_place_market_bracket", symbol=symbol, direction=direction, contracts=contracts, sl_price=sl_price, tp_price=tp_price, sl_source=sl_source, synthetic_entry_id=ids[0], synthetic_stop_id=ids[1], synthetic_target_id=ids[2], ) return OrderResult( success=True, entry_price=entry_px if entry_px is not None else 0.0, sl_price=sl_price, tp_price=tp_price, entry_id=ids[0], stop_id=ids[1], target_id=ids[2], ) async def place_stop_order( self, symbol: str, side: str, contracts: int, stop_price: float, ) -> OrderResult: oid = f"DRY-attach-S-{uuid.uuid4().hex[:8]}" self._log_dry( "dry_place_stop_order", symbol=symbol, side=side, contracts=contracts, stop_price=stop_price, synthetic_stop_id=oid, ) return OrderResult( success=True, sl_price=stop_price, stop_id=oid, entry_id=oid, ) async def place_limit_order( self, symbol: str, side: str, contracts: int, limit_price: float, ) -> OrderResult: oid = f"DRY-attach-T-{uuid.uuid4().hex[:8]}" self._log_dry( "dry_place_limit_order", symbol=symbol, side=side, contracts=contracts, limit_price=limit_price, synthetic_target_id=oid, ) return OrderResult( success=True, tp_price=limit_price, target_id=oid, entry_id=oid, ) async def cancel_order(self, symbol: str, order_id: str) -> CancelResult: self._log_dry("dry_cancel_order", symbol=symbol, order_id=order_id) return CancelResult(success=True, order_id=str(order_id)) async def cancel_all_for_symbol(self, symbol: str) -> int: self._log_dry("dry_cancel_all_for_symbol", symbol=symbol) return 0 async def close_position( self, symbol: str, contracts: Optional[int] = None, ) -> OrderResult: self._log_dry( "dry_close_position", symbol=symbol, contracts=contracts, ) return OrderResult(success=True) async def partial_close_via_opposite_order( self, symbol: str, direction: str, contracts_to_close: int, residual_contracts: int, new_sl_price: float, new_tp_price: float, old_stop_order_id: Optional[str], old_target_order_id: Optional[str], ) -> OrderResult: _, s, t = self._dry_ids("PARTIAL") self._log_dry( "dry_partial_close_via_opposite_order", symbol=symbol, direction=direction, contracts_to_close=contracts_to_close, residual_contracts=residual_contracts, new_sl_price=new_sl_price, new_tp_price=new_tp_price, old_stop=old_stop_order_id, old_target=old_target_order_id, ) return OrderResult( success=True, entry_id=f"DRY-PARTIAL-CLOSE-{uuid.uuid4().hex[:8]}", stop_id=s, target_id=t, sl_price=new_sl_price, tp_price=new_tp_price, ) async def modify_stop( self, symbol: str, order_id: str, new_sl_price: float, ) -> OrderResult: self._log_dry( "dry_modify_stop", symbol=symbol, order_id=order_id, new_sl_price=new_sl_price, ) return OrderResult( success=True, sl_price=new_sl_price, stop_id=str(order_id), ) # ============================================================ # internals # ============================================================ @staticmethod def _dry_ids(prefix: str) -> tuple[str, str, str]: token = uuid.uuid4().hex[:8] return ( f"DRY-{prefix}-E-{token}", f"DRY-{prefix}-S-{token}", f"DRY-{prefix}-T-{token}", ) def _log_dry(self, event: str, **fields) -> None: if self._logger is not None: try: self._logger.brain_log.write(event, **fields) except Exception: pass else: log.info("[%s] %s %s", self.name, event, fields)