""" APEX V16 — TopstepX broker (BrokerBase implementation). Wraps the V15 TopstepXAdapter (reused AS-IS) and adapts its API to the V16 BrokerBase abstract interface: - dict returns -> structured OrderResult / Position / CancelResult - bool returns -> OrderResult with .success populated - method renames: get_current_price -> get_last_price, modify_sl -> modify_stop, cancel_all_orders_for_contract -> cancel_all_for_symbol - is_connected: V15 sync attribute -> V16 async method - Exceptions from the underlying adapter are caught and converted to OrderResult(success=False, error=...). The orchestrator never sees raw broker exceptions. Why a wrapper instead of editing the adapter: - topstepx_adapter.py is shared with V15 (still running on burned account during transition). Editing it would risk regressing V15. - Single responsibility: the adapter speaks "TopstepX dict-y", the wrapper speaks "V16 typed". Each can evolve independently. """ from __future__ import annotations import logging import os from datetime import datetime, timedelta, timezone from typing import Optional from broker.broker_base import ( BrokerBase, BrokerReadError, CancelResult, ClosedTrade, Order, OrderResult, Position, ) from broker.topstepx_adapter import TopstepXAdapter from core.config_futures import ASSETS_MAP log = logging.getLogger("topstepx_v16") # V15-parity short-symbol -> contractId tag map (see V15 _fetch_last_close_price # riga 2012-2015). TopstepX contractIds embed these mid-codes (e.g. "CON.F.US.EU6.M25"), # so REST responses must be filtered with the tag, not the user-facing symbol. _SYMBOL_TAG_MAP: dict[str, str] = { "6B": "BP6", "6E": "EU6", "6A": "A6", "6J": "JY6", "6C": "CA6", "MES": "ES", "MNQ": "NQ", "MYM": "YM", "MGC": "MGC", "MCL": "MCL", } def _symbol_matches(symbol: Optional[str], contract_id: str) -> bool: if not symbol: return True tag = _SYMBOL_TAG_MAP.get(symbol, symbol) return tag in (contract_id or "") class TopstepXBroker(BrokerBase): """ BrokerBase implementation for TopstepX / ProjectX. Holds a TopstepXAdapter instance and delegates to it, translating return shapes at the boundary. """ name = "TopstepX" def __init__(self, instruments: list[str]): """ Args: instruments: list of symbols to subscribe at connect time (e.g., ["MES", "MNQ", "6E", "6B"]). The TopstepX SDK requires this up-front; symbols not in this list cannot be traded later. """ self._adapter = TopstepXAdapter() self._instruments = instruments # ============================================================ # CONNECTION # ============================================================ async def connect(self) -> bool: try: return await self._adapter.connect(self._instruments) except Exception as e: log.error(f"connect failed: {e}", exc_info=True) return False async def disconnect(self) -> None: try: await self._adapter.disconnect() except Exception as e: log.warning(f"disconnect raised (non-fatal): {e}") async def is_connected(self) -> bool: try: return bool(self._adapter.is_connected()) except Exception: return False # ============================================================ # MARKET DATA # ============================================================ async def get_last_price(self, symbol: str) -> Optional[float]: try: price = await self._adapter.get_current_price(symbol) return float(price) if price is not None else None except Exception as e: log.warning(f"get_last_price({symbol}) failed: {e}") return None # ============================================================ # POSITIONS (read-only) # ============================================================ async def positions_get(self, symbol: Optional[str] = None) -> list[Position]: """ Direct REST /Position/searchOpen with raw-search fallback. Bypasses the SDK Position class (broken on contractDisplayName) AND the adapter's silent []-on-error fallback that caused the 28 apr 6J duplicate-bracket incident. Tri-state contract: - returns [] when the API definitively reports no positions - returns [Position] on success with one or more positions - raises BrokerReadError when BOTH the primary REST path and the raw-search fallback fail (auth, account_id unresolved, HTTP non-200, transport exception) Callers that gate destructive actions on the result MUST treat the exception as 'unknown state', not 'flat'. """ primary_err: Optional[str] = None rows: Optional[dict] = None # ── Path 1: direct REST via _rest_auth + _rest_post ── try: token, account_id = await self._rest_auth() if not token or not account_id: primary_err = ( f"auth/account_id unresolved " f"(token={'ok' if token else 'missing'}, " f"account_id={'ok' if account_id else 'missing'})" ) else: rows = await self._rest_post( "/Position/searchOpen", {"accountId": int(account_id)}, token, timeout=4, ) if rows is None: primary_err = "/Position/searchOpen returned None" except Exception as e: primary_err = f"primary REST raised: {e}" # ── Path 2 fallback: adapter._raw_positions_search ── if rows is None: try: raw_list = await self._adapter._raw_positions_search() except Exception as e: raise BrokerReadError( f"positions_get: both paths failed " f"(primary: {primary_err}; fallback raised: {e})" ) if not raw_list and primary_err is not None: # Both paths returned no data AND primary signalled error. # Cannot distinguish "really empty" from "fallback also # silently failed" — treat as unknown. raise BrokerReadError( f"positions_get: both paths failed " f"(primary: {primary_err}; " f"fallback _raw_positions_search returned [])" ) raw_positions = raw_list or [] else: raw_positions = rows.get("positions", []) or [] contract_to_symbol = { info["contract_id"]: sym for sym, info in self._adapter._instrument_cache.items() } out: list[Position] = [] for p in raw_positions: contract_id = p.get("contractId", "") or "" sym = contract_to_symbol.get(contract_id) if not sym: continue # position on a contract we don't track if symbol and sym != symbol: continue try: pos_type = p.get("type", 0) size_val = int(p.get("size", 0)) is_buy = (pos_type == 1) out.append(Position( symbol=sym, direction="BUY" if is_buy else "SELL", contracts=size_val, avg_price=float(p.get("averagePrice", 0.0)), raw=p, )) except (KeyError, TypeError, ValueError) as e: log.warning(f"positions_get: skipping malformed row {p!r}: {e}") continue return out async def pending_orders(self, symbol: Optional[str] = None) -> list[Order]: """ Direct REST /Order/searchOpen (V15 parity, _watchdog_naked_positions riga 2072-2078). The TopstepXAdapter does not surface pending orders, so we issue a thin REST call here, parallel to _fetch_last_close_price. Returns empty list on any auth/transport failure — caller (Reconciler) treats "no pending orders" and "REST failed" identically (log-only). """ token, account_id = await self._rest_auth() if not token or not account_id: return [] rows = await self._rest_post( "/Order/searchOpen", {"accountId": account_id}, token, timeout=4, ) if rows is None: return [] orders_raw = rows.get("orders", []) or [] out: list[Order] = [] for o in orders_raw: try: cid = o.get("contractId", "") or "" if not _symbol_matches(symbol, cid): continue t = o.get("type") # TopstepX order types: 1=Limit, 2=Market, 4=Stop. kind = {1: "LIMIT", 2: "MARKET", 4: "STOP"}.get(t, "OTHER") out.append(Order( order_id=str(o.get("id", "")), symbol=cid, kind=kind, price=float( o.get("limitPrice") or o.get("stopPrice") or o.get("price") or 0.0 ), contracts=int(o.get("size", 0) or 0), )) except (TypeError, ValueError) as e: log.warning(f"pending_orders: skipping malformed row {o!r}: {e}") continue return out async def recent_trades( self, symbol: Optional[str] = None, since: Optional[datetime] = None, limit: int = 50, ) -> list[ClosedTrade]: """ Direct REST /api/Trade/search (V15 parity, _fetch_last_close_price riga 1997-2009). Used by Reconciler case (ii) to recover P&L for a trade that closed broker-side while bot was down. Filters server-side by accountId + time window; client-side by symbol tag and profitAndLoss != 0 (open legs are also returned by the API). """ token, account_id = await self._rest_auth() if not token or not account_id: return [] end = datetime.now(timezone.utc) start = since if since is not None else (end - timedelta(hours=24)) if start.tzinfo is None: start = start.replace(tzinfo=timezone.utc) rows = await self._rest_post( "/Trade/search", { "accountId": account_id, "startTimestamp": start.isoformat(), "endTimestamp": end.isoformat(), }, token, timeout=5, ) if rows is None: return [] trades_raw = rows.get("trades", []) or [] out: list[ClosedTrade] = [] for t in trades_raw: try: cid = t.get("contractId", "") or "" if not _symbol_matches(symbol, cid): continue pnl = float(t.get("profitAndLoss") or 0.0) if pnl == 0.0: # opening fills have pnl=0 — only closed legs interest us continue # TopstepX side: 0=BUY, 1=SELL on Trade/search (mirrors orders). side_raw = t.get("side") side = "BUY" if side_raw == 0 else ("SELL" if side_raw == 1 else "BUY") out.append(ClosedTrade( trade_id=str(t.get("id", "")), symbol=cid, contracts=int(t.get("size", 0) or 0), side=side, exit_price=float(t.get("price") or 0.0), pnl_usd=pnl, closed_at=str(t.get("creationTimestamp", "")), )) except (TypeError, ValueError) as e: log.warning(f"recent_trades: skipping malformed row {t!r}: {e}") continue out.sort(key=lambda x: x.closed_at, reverse=True) return out[:limit] # ============================================================ # ORDERS (write) # ============================================================ 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: resp = await self._adapter.place_market_bracket( symbol=symbol, side=direction, size=contracts, sl_price=sl_price, tp_price=tp_price, sl_absolute_price=sl_absolute_price, sl_ticks=sl_ticks, tp_ticks=tp_ticks, sl_source=sl_source, ) except Exception as e: log.error( f"place_market_bracket({symbol} {direction} {contracts}) " f"raised: {e}", exc_info=True, ) msg = str(e) return OrderResult( success=False, error=f"{type(e).__name__}: {msg}" if msg else f"{type(e).__name__}(no msg)", ) if not isinstance(resp, dict): return OrderResult( success=False, error=f"adapter returned non-dict: {type(resp).__name__}", ) if not resp.get("success"): raw_err = resp.get("error") if raw_err is None or raw_err == "": err_str = f"adapter returned success=False without error (resp keys: {list(resp.keys())})" else: err_str = str(raw_err) return OrderResult( success=False, error=err_str, raw=resp, ) # Adapter-native keys come FIRST. The V15 TopstepXAdapter returns # {"entry_id", "stop_id", "target_id"} (riga 1223-1227); historical # alternates ("entry_order_id" etc.) kept as fallback for adapter # API drift. V16 incident 29 apr: stop_order_id/target_order_id # were the only keys checked, so OrderResult had stop_id=None on # every successful bracket -> SL/TP IDs lost -> MOVE_SL silently # skipped because active.entry.stop_order_id was None. return OrderResult( success=True, entry_price=_first_float(resp, "entry_price", "fill_price", "filled_price"), sl_price=_first_float(resp, "sl_price", "stop_price"), tp_price=_first_float(resp, "tp_price", "target_price"), entry_id=_first_str(resp, "entry_id", "entry_order_id", "orderId"), stop_id=_first_str(resp, "stop_id", "stop_order_id", "sl_order_id"), target_id=_first_str(resp, "target_id", "target_order_id", "tp_order_id"), raw=resp, ) async def place_stop_order( self, symbol: str, side: str, contracts: int, stop_price: float, ) -> OrderResult: """ Standalone STOP order via ctx.orders.place_stop_order (V15 parity, adapter riga 1168-1176). Used by orphan recovery to re-attach SL. """ return await self._place_protective_order( symbol, side, contracts, stop_price, kind="stop", ) async def place_limit_order( self, symbol: str, side: str, contracts: int, limit_price: float, ) -> OrderResult: """Standalone LIMIT order (TP re-attach). Adapter riga 1204-1211.""" return await self._place_protective_order( symbol, side, contracts, limit_price, kind="limit", ) async def _place_protective_order( self, symbol: str, side: str, contracts: int, price: float, *, kind: str, ) -> OrderResult: side_int = self._adapter.SIDE_MAP.get( side.upper() if isinstance(side, str) else side ) if side_int is None: return OrderResult(success=False, error=f"invalid side: {side}") info = self._adapter._instrument_cache.get(symbol) if not info: return OrderResult(success=False, error=f"no contract cached for {symbol}") contract_id = info["contract_id"] ctx = self._adapter._get_ctx(symbol) try: if kind == "stop": resp = await ctx.orders.place_stop_order( contract_id=contract_id, side=side_int, size=int(contracts), stop_price=price, ) else: resp = await ctx.orders.place_limit_order( contract_id=contract_id, side=side_int, size=int(contracts), limit_price=price, ) except Exception as e: log.error(f"place_{kind}_order({symbol}) raised: {e}") return OrderResult(success=False, error=str(e)) order_id = ( getattr(resp, "orderId", None) or getattr(resp, "order_id", None) ) if not order_id: err = ( getattr(resp, "errorMessage", None) or getattr(resp, "error_message", None) or f"{kind} order returned no id" ) return OrderResult(success=False, error=str(err)) if kind == "stop": return OrderResult(success=True, sl_price=price, stop_id=str(order_id), entry_id=str(order_id)) return OrderResult(success=True, tp_price=price, target_id=str(order_id), entry_id=str(order_id)) async def cancel_order(self, symbol: str, order_id: str) -> CancelResult: try: try: oid_int = int(order_id) except (TypeError, ValueError): return CancelResult( success=False, order_id=str(order_id), error=f"order_id not int-convertible: {order_id!r}", ) ok = await self._adapter.cancel_order(symbol, oid_int) except Exception as e: log.error(f"cancel_order({symbol}, {order_id}) raised: {e}") return CancelResult(success=False, order_id=str(order_id), error=str(e)) return CancelResult( success=bool(ok), order_id=str(order_id), error=None if ok else "adapter returned False", ) async def cancel_all_for_symbol(self, symbol: str) -> int: try: resp = await self._adapter.cancel_all_orders_for_contract(symbol) except Exception as e: log.error(f"cancel_all_for_symbol({symbol}) raised: {e}") return 0 if not isinstance(resp, dict): log.warning(f"cancel_all_for_symbol: unexpected return {resp!r}") return 0 if resp.get("error"): log.warning(f"cancel_all_for_symbol({symbol}): {resp['error']}") return int(resp.get("cancelled_count", 0)) async def close_position( self, symbol: str, contracts: Optional[int] = None, ) -> OrderResult: try: resp = await self._adapter.close_position(symbol, size=contracts) except Exception as e: log.error(f"close_position({symbol}, {contracts}) raised: {e}", exc_info=True) return OrderResult(success=False, error=str(e)) if not isinstance(resp, dict): return OrderResult( success=False, error=f"adapter returned non-dict: {type(resp).__name__}", ) if not resp.get("success"): return OrderResult( success=False, error=str(resp.get("error") or resp.get("errorMessage") or "close failed"), raw=resp, ) return OrderResult( success=True, entry_id=_first_str(resp, "orderId"), raw=resp, ) 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: """V18 12-mag — workaround per /Position/partialCloseContract 400. Vedi `broker_base.BrokerBase.partial_close_via_opposite_order`. OrderResult.stop_id / target_id valorizzati con i NUOVI ID del bracket sul residuo; caller aggiorna entry.stop_order_id / target_order_id su success. """ try: resp = await self._adapter.partial_close_via_opposite_order( symbol=symbol, direction=direction, contracts_to_close=int(contracts_to_close), residual_contracts=int(residual_contracts), new_sl_price=float(new_sl_price), new_tp_price=float(new_tp_price), old_stop_order_id=old_stop_order_id, old_target_order_id=old_target_order_id, ) except Exception as e: log.error( f"partial_close_via_opposite_order({symbol}) raised: {e}", exc_info=True, ) return OrderResult(success=False, error=str(e)) if not isinstance(resp, dict): return OrderResult( success=False, error=f"adapter returned non-dict: {type(resp).__name__}", ) if not resp.get("success"): return OrderResult( success=False, error=str(resp.get("error") or "partial close via opposite failed"), raw=resp, ) return OrderResult( success=True, entry_id=_first_str(resp, "closing_order_id"), stop_id=_first_str(resp, "new_stop_id"), target_id=_first_str(resp, "new_target_id"), sl_price=resp.get("new_sl_price"), tp_price=resp.get("new_tp_price"), raw=resp, ) async def modify_stop( self, symbol: str, order_id: str, new_sl_price: float, ) -> OrderResult: # Defense-in-depth tick alignment: the brain SHOULD already # round (Layer 1) and the SDK adapter aligns again (Layer 3 # fallback), but we round here too so a regression in either # layer cannot push an off-tick price to the venue. V16 incident # 29 apr: 6C 0.7324928571428572 silently rejected by broker. spec = ASSETS_MAP.get(symbol, {}) ts = float(spec.get("tick_size", 0.0) or 0.0) if ts > 0: aligned_sl = round(round(new_sl_price / ts) * ts, 6) else: aligned_sl = new_sl_price try: try: oid_int = int(order_id) except (TypeError, ValueError): return OrderResult( success=False, error=f"order_id not int-convertible: {order_id!r}", ) ok = await self._adapter.modify_sl( symbol=symbol, stop_order_id=oid_int, new_sl=aligned_sl, ) except Exception as e: log.error( f"modify_stop({symbol}, {order_id}, {new_sl_price}) raised: {e}" ) return OrderResult(success=False, error=str(e)) if not ok: return OrderResult( success=False, error="adapter modify_sl returned False", ) return OrderResult( success=True, sl_price=aligned_sl, stop_id=str(order_id), ) # ============================================================ # ACCOUNT # ============================================================ async def get_account_balance(self) -> float: """ Current TopstepX account balance via TopstepXAdapter.get_account_info. SDK keeps account_info refreshed via socket events — V16 adds no cache. """ info = await self._adapter.get_account_info() return float(info.get("balance", 0.0)) # ============================================================ # MARKET DATA # ============================================================ async def fetch_bars(self, symbol: str, timeframe: str, n: int): """ Delegate to TopstepXAdapter.get_bars (V15 hybrid WS+REST strategy). """ return await self._adapter.get_bars(symbol, timeframe, n) # ============================================================ # internal — direct REST for endpoints not surfaced by the adapter # ============================================================ async def _rest_auth(self) -> tuple[Optional[str], Optional[int]]: """ Returns (bearer_token, account_id). Token is cached for 30min on the wrapped adapter (shared with adapter._direct_rest_token to avoid double-authing). Returns (None, None) on any failure. """ try: api_key = os.getenv("PROJECT_X_API_KEY") username = os.getenv("PROJECT_X_USERNAME") if not api_key or not username: return None, None now_ts = datetime.now(timezone.utc).timestamp() cached_token = getattr(self._adapter, "_direct_rest_token", None) cached_exp = getattr(self._adapter, "_direct_rest_token_exp", 0) or 0 if cached_token and now_ts < cached_exp: token = cached_token else: import requests r = requests.post( "https://api.topstepx.com/api/Auth/loginKey", json={"userName": username, "apiKey": api_key}, timeout=5, ) if r.status_code != 200: return None, None token = r.json().get("token") if not token: return None, None self._adapter._direct_rest_token = token self._adapter._direct_rest_token_exp = now_ts + 1800 client = self._adapter._suite.client if self._adapter._suite else None acc = getattr(client, "account_info", None) if client else None account_id = getattr(acc, "id", None) if acc else None return token, int(account_id) if account_id else None except Exception as e: log.warning(f"_rest_auth failed: {e}") return None, None async def _rest_post( self, path: str, body: dict, token: str, *, timeout: float = 4, ) -> Optional[dict]: """Bare POST to api.topstepx.com/api{path}. None on any failure.""" try: import requests r = requests.post( "https://api.topstepx.com/api" + path, headers={ "Authorization": "Bearer " + token, "Content-Type": "application/json", }, json=body, timeout=timeout, ) if r.status_code != 200: log.warning(f"_rest_post {path}: HTTP {r.status_code}") return None return r.json() except Exception as e: log.warning(f"_rest_post {path} raised: {e}") return None # ============================================================ # small helpers # ============================================================ def _first_float(d: dict, *keys: str) -> Optional[float]: """Return float-cast of the first key found in d with non-None value.""" for k in keys: v = d.get(k) if v is not None: try: return float(v) except (TypeError, ValueError): continue return None def _first_str(d: dict, *keys: str) -> Optional[str]: for k in keys: v = d.get(k) if v is not None: return str(v) return None