"""Pure-function payload builders for IBKR complex orders (bracket/OCO/OTO). No HTTP. Tests are deterministic. """ from __future__ import annotations import secrets from dataclasses import dataclass from typing import Literal @dataclass class OrderSpec: conid: int sec_type: str # "STK" | "OPT" | "FUT" | "CASH" side: Literal["BUY", "SELL"] qty: float order_type: Literal["MKT", "LMT", "STP", "STP_LMT"] price: float | None = None # limit price aux_price: float | None = None # stop price tif: str = "GTC" exchange: str = "SMART" def _to_order_dict(spec: OrderSpec, *, oca_group: str | None = None, oca_type: int | None = None, parent_id: str | None = None) -> dict: o: dict = { "conid": spec.conid, "secType": f"{spec.conid}:{spec.sec_type}", "orderType": spec.order_type, "side": spec.side, "quantity": spec.qty, "tif": spec.tif, "listingExchange": spec.exchange, } if spec.price is not None: o["price"] = spec.price if spec.aux_price is not None: o["auxPrice"] = spec.aux_price if oca_group: o["ocaGroup"] = oca_group if oca_type is not None: o["ocaType"] = oca_type if parent_id: o["parentId"] = parent_id return o def _new_oca_group() -> str: return f"oca-{secrets.token_hex(4)}" def build_bracket_payload( *, conid: int, sec_type: str, side: str, qty: float, entry_price: float, stop_loss: float, take_profit: float, tif: str = "GTC", exchange: str = "SMART", ) -> dict: """Bracket: parent LMT entry + child STP (loss) + child LMT (profit), OCA-linked.""" side = side.upper() opposite = "SELL" if side == "BUY" else "BUY" oca = _new_oca_group() parent = OrderSpec(conid=conid, sec_type=sec_type, side=side, qty=qty, # type: ignore[arg-type] order_type="LMT", price=entry_price, tif=tif, exchange=exchange) sl = OrderSpec(conid=conid, sec_type=sec_type, side=opposite, qty=qty, # type: ignore[arg-type] order_type="STP", aux_price=stop_loss, tif=tif, exchange=exchange) tp = OrderSpec(conid=conid, sec_type=sec_type, side=opposite, qty=qty, # type: ignore[arg-type] order_type="LMT", price=take_profit, tif=tif, exchange=exchange) return { "orders": [ _to_order_dict(parent, oca_group=oca, oca_type=2), _to_order_dict(sl, oca_group=oca, oca_type=2), _to_order_dict(tp, oca_group=oca, oca_type=2), ] } def build_oco_payload(legs: list[OrderSpec]) -> dict: """OCO: N legs, all sharing same ocaGroup with ocaType=1 (one-cancels-all).""" if len(legs) < 2: raise ValueError("OCO requires at least 2 legs") oca = _new_oca_group() return { "orders": [ _to_order_dict(l, oca_group=oca, oca_type=1) for l in legs ] } def build_oto_first_payload(trigger: OrderSpec) -> dict: """OTO step 1: place trigger as standalone.""" return {"orders": [_to_order_dict(trigger)]} def build_oto_child_payload(child: OrderSpec, parent_order_id: str) -> dict: """OTO step 2: child references parentId from step-1 order_id.""" return {"orders": [_to_order_dict(child, parent_id=parent_order_id)]}