"""Single-asset limit order book with price-time priority. The book only knows orders and prices. Fees, accounts, latency and halts live in `engine.py`. """ from __future__ import annotations from collections import deque from dataclasses import dataclass from typing import Iterator from sortedcontainers import SortedDict from .orders import Order, OrderStatus, OrderType, Side @dataclass(slots=True) class Fill: resting: Order incoming: Order price: int qty: int class LimitOrderBook: __slots__ = ("asset", "bids", "asks", "_index", "_depth_cache") def __init__(self, asset: int) -> None: self.asset = asset # bids: highest price first -> store negative price keys so .peekitem(0) is the best bid self.bids: SortedDict[int, deque[Order]] = SortedDict() self.asks: SortedDict[int, deque[Order]] = SortedDict() self._index: dict[int, Order] = {} # ---- queries ------------------------------------------------------------------------- def best_bid(self) -> int | None: return -self.bids.peekitem(0)[0] if self.bids else None def best_ask(self) -> int | None: return self.asks.peekitem(0)[0] if self.asks else None def mid(self) -> float | None: b, a = self.best_bid(), self.best_ask() if b is None or a is None: return None return (a + b) / 2.0 def spread(self) -> int | None: b, a = self.best_bid(), self.best_ask() if b is None or a is None: return None return a - b def depth(self, levels: int = 5) -> tuple[list[tuple[int, int]], list[tuple[int, int]]]: """Aggregated (price, qty) for the best `levels` on each side.""" bids = [] for i, (negp, q) in enumerate(self.bids.items()): if i >= levels: break bids.append((-negp, sum(o.remaining for o in q))) asks = [] for i, (p, q) in enumerate(self.asks.items()): if i >= levels: break asks.append((p, sum(o.remaining for o in q))) return bids, asks def depth_within(self, side: Side, ticks: int) -> int: """Total resting quantity within `ticks` of the best price on `side`.""" if side is Side.BUY: bb = self.best_bid() if bb is None: return 0 return sum(o.remaining for negp, q in self.bids.items() if -negp >= bb - ticks for o in q) ba = self.best_ask() if ba is None: return 0 return sum(o.remaining for p, q in self.asks.items() if p <= ba + ticks for o in q) def get(self, order_id: int) -> Order | None: return self._index.get(order_id) def __contains__(self, order_id: int) -> bool: return order_id in self._index def __len__(self) -> int: return len(self._index) def resting_orders(self) -> Iterator[Order]: return iter(self._index.values()) def resting_for(self, agent_id: int) -> list[Order]: return [o for o in self._index.values() if o.agent_id == agent_id] def total_qty(self) -> tuple[int, int]: return (sum(o.remaining for q in self.bids.values() for o in q), sum(o.remaining for q in self.asks.values() for o in q)) # ---- mutation ------------------------------------------------------------------------ def _rest(self, order: Order) -> None: assert order.order_type is OrderType.LIMIT and order.price is not None book = self.bids if order.side is Side.BUY else self.asks key = -order.price if order.side is Side.BUY else order.price q = book.get(key) if q is None: q = deque() book[key] = q q.append(order) self._index[order.order_id] = order order.status = OrderStatus.OPEN def _remove_resting(self, order: Order) -> None: book = self.bids if order.side is Side.BUY else self.asks key = -order.price if order.side is Side.BUY else order.price q = book[key] q.remove(order) if not q: del book[key] del self._index[order.order_id] def match(self, incoming: Order, max_qty: int | None = None) -> list[Fill]: """Walk the opposite side while the incoming order crosses. Returns the fills in execution order; mutates `incoming.remaining` and the resting orders. Does *not* rest the remainder (the engine decides, because it must reserve cash first). `max_qty` caps the executable quantity (used for market buys with a cash constraint). """ fills: list[Fill] = [] opp = self.asks if incoming.side is Side.BUY else self.bids budget = incoming.remaining if max_qty is None else min(incoming.remaining, max_qty) while budget > 0 and opp: key, queue = opp.peekitem(0) best_price = key if incoming.side is Side.BUY else -key if incoming.order_type is OrderType.LIMIT: assert incoming.price is not None if incoming.side is Side.BUY and best_price > incoming.price: break if incoming.side is Side.SELL and best_price < incoming.price: break while budget > 0 and queue: resting = queue[0] q = min(resting.remaining, budget) resting.remaining -= q incoming.remaining -= q budget -= q fills.append(Fill(resting=resting, incoming=incoming, price=best_price, qty=q)) if resting.remaining == 0: resting.status = OrderStatus.FILLED queue.popleft() del self._index[resting.order_id] if not queue: del opp[key] if incoming.remaining == 0: incoming.status = OrderStatus.FILLED return fills def rest(self, order: Order) -> None: self._rest(order) def cancel(self, order_id: int) -> Order | None: order = self._index.get(order_id) if order is None: return None self._remove_resting(order) order.status = OrderStatus.CANCELLED return order def clear_all(self) -> list[Order]: orders = list(self._index.values()) for o in orders: self._remove_resting(o) o.status = OrderStatus.CANCELLED return orders # ---- invariants (used by tests) ------------------------------------------------------ def check_invariants(self) -> None: bb, ba = self.best_bid(), self.best_ask() if bb is not None and ba is not None: assert bb < ba, f"crossed book: bid {bb} >= ask {ba}" n = 0 for negp, q in self.bids.items(): assert q, "empty price level" for o in q: assert o.side is Side.BUY and o.price == -negp and o.remaining > 0 and o.status is OrderStatus.OPEN n += 1 for p, q in self.asks.items(): assert q, "empty price level" for o in q: assert o.side is Side.SELL and o.price == p and o.remaining > 0 and o.status is OrderStatus.OPEN n += 1 assert n == len(self._index), "index/book mismatch"