"""The exchange: multi-asset continuous double auction with latency, fees, risk limits, halts and an append-only event log that can be replayed deterministically. Time is measured in ticks (float). Agents submit `Action`s at integer decision times; each intent becomes an `Order` that *arrives* at submit_time + latency. The engine processes arrivals in (arrival_time, seq) order, so two agents deciding at the same tick are ordered by latency and then by a seeded tie-break assigned at submission. """ from __future__ import annotations import heapq from dataclasses import dataclass, field import numpy as np from .accounts import Account, Ledger from .book import LimitOrderBook from .orders import Action, Event, Order, OrderStatus, OrderType, Side, Trade @dataclass(slots=True) class AssetSpec: asset: int name: str tick_cents: int = 1 # tick size in cents (1 => prices quoted in cents) lot: int = 1 price_band_pct: float = 0.5 # orders farther than this from the reference price are rejected max_order_qty: int = 10_000 @dataclass(slots=True) class ExchangeConfig: maker_fee_bps: float = 0.0 # negative = rebate taker_fee_bps: float = 2.0 min_latency: float = 0.05 # ticks halt_threshold: float | None = None # |ret over window| above this triggers a halt (None = off) halt_window: int = 10 halt_duration: int = 20 halt_cancel_resting: bool = True allow_short: bool = False max_short_lots: int = 0 margin_check: bool = True liquidation_fraction: float = 0.5 # share of the position sold on a margin call order_ttl: int | None = None # ticks after which a resting order expires (None = GTC) position_limit: int | None = None # max |position| per agent (lots) call_auction_interval: int | None = None # batch clearing every k ticks instead of continuous matching call_auction_collar: float = 0.02 # max move of the clearing price per auction (fraction of the reference) @dataclass(slots=True) class MarketState: """Public, per-asset running statistics kept by the engine for observations and metrics.""" last_price: int | None = None last_trade_time: float = -1.0 reference_price: int = 0 halted_until: float = -1.0 n_halts: int = 0 volume_tick: int = 0 signed_volume_tick: int = 0 mid_history: list[float] = field(default_factory=list) # one entry per completed tick class Exchange: def __init__(self, assets: list[AssetSpec], config: ExchangeConfig, rng: np.random.Generator, initial_prices: dict[int, int]) -> None: self.assets = {a.asset: a for a in assets} self.config = config self.rng = rng self.books: dict[int, LimitOrderBook] = {a: LimitOrderBook(a) for a in self.assets} self.ledger = Ledger() self.state: dict[int, MarketState] = {} for a, p0 in initial_prices.items(): self.state[a] = MarketState(reference_price=p0, last_price=p0) self.events: list[Event] = [] self.trades: list[Trade] = [] self._queue: list[tuple[float, int, str, object]] = [] # (arrival, seq, kind, payload) self._seq = 0 self._next_order_id = 1 self._next_trade_id = 1 self.orders: dict[int, Order] = {} # all orders ever submitted (for replay and stats) self.time: float = 0.0 self.latency: dict[int, float] = {} # agent -> latency in ticks self.pending_by_agent: dict[int, set[int]] = {} # ---- setup ----------------------------------------------------------------------------- def register(self, acct: Account, latency: float) -> None: self.ledger.add(acct) self.latency[acct.agent_id] = max(latency, self.config.min_latency) # ---- event log ------------------------------------------------------------------------- def _log(self, kind: str, **payload) -> None: self._seq += 1 self.events.append(Event(seq=self._seq, time=self.time, kind=kind, payload=payload)) # ---- public market data ---------------------------------------------------------------- def marks(self) -> dict[int, float]: """Mark-to-market price in cents per lot: mid if available, else last trade, else reference.""" out = {} for a in self.books: out[a] = self.mid_ticks(a) * self.assets[a].tick_cents return out def mid_ticks(self, asset: int) -> float: st = self.state[asset] if self.config.call_auction_interval: # batch mode: the book may be crossed; the auction price is the reference return float(st.last_price if st.last_price is not None else st.reference_price) m = self.books[asset].mid() if m is not None: return m return float(st.last_price if st.last_price is not None else st.reference_price) def spread_ticks(self, asset: int) -> int | None: if self.config.call_auction_interval: return None return self.books[asset].spread() def is_halted(self, asset: int) -> bool: return self.time < self.state[asset].halted_until # ---- submission -------------------------------------------------------------------------- def submit(self, agent_id: int, action: Action, decision_time: float) -> list[int]: """Queue an agent's intents; returns the order ids assigned (cancels get none).""" ids: list[int] = [] lat = self.latency[agent_id] for c in action.cancels: self._seq += 1 heapq.heappush(self._queue, (decision_time + lat, self._seq, "cancel", (agent_id, c.order_id))) for it in action.orders: oid = self._next_order_id self._next_order_id += 1 o = Order(order_id=oid, agent_id=agent_id, asset=it.asset, side=it.side, order_type=it.order_type, qty=int(it.qty), price=None if it.price is None else int(it.price), submit_time=decision_time, arrival_time=decision_time + lat, tag=it.tag) self.orders[oid] = o self.pending_by_agent.setdefault(agent_id, set()).add(oid) self._seq += 1 heapq.heappush(self._queue, (o.arrival_time, self._seq, "order", o)) ids.append(oid) self.ledger[agent_id].n_orders += 1 return ids # ---- processing -------------------------------------------------------------------------- def process_until(self, t_end: float) -> None: """Process every queued arrival with arrival_time < t_end, in order.""" while self._queue and self._queue[0][0] < t_end: arrival, _, kind, payload = heapq.heappop(self._queue) self.time = arrival if kind == "order": self._on_order(payload) # type: ignore[arg-type] else: agent_id, oid = payload # type: ignore[misc] self._on_cancel(agent_id, oid) self.time = t_end def _fee(self, notional: int, bps: float) -> int: return int(round(notional * bps / 10_000.0)) def _reject(self, o: Order, reason: str) -> None: o.status = OrderStatus.REJECTED o.reject_reason = reason self.ledger[o.agent_id].n_rejects += 1 self._log("reject", order_id=o.order_id, agent_id=o.agent_id, asset=o.asset, reason=reason) def _on_order(self, o: Order) -> None: self.pending_by_agent.get(o.agent_id, set()).discard(o.order_id) acct = self.ledger[o.agent_id] spec = self.assets[o.asset] book = self.books[o.asset] st = self.state[o.asset] if acct.frozen and o.tag != "liquidation": return self._reject(o, "frozen") if self.is_halted(o.asset): return self._reject(o, "halted") if o.qty <= 0 or o.qty > spec.max_order_qty: return self._reject(o, "bad_qty") if o.order_type is OrderType.LIMIT: if o.price is None or o.price <= 0: return self._reject(o, "bad_price") ref = self.mid_ticks(o.asset) if abs(o.price - ref) > spec.price_band_pct * ref: return self._reject(o, "price_band") # --- capacity checks and reservations ---------------------------------------------- taker_bps = self.config.taker_fee_bps if o.side is Side.BUY: if o.order_type is OrderType.LIMIT: notional = o.price * spec.tick_cents * o.qty need = notional + max(self._fee(notional, taker_bps), self._fee(notional, self.config.maker_fee_bps)) + o.qty if need > self._buying_power(acct): return self._reject(o, "insufficient_cash") acct.reserved_cash += need o.reserved_cash = need else: ba = book.best_ask() if ba is None: return self._reject(o, "no_liquidity") # cap the quantity to what free cash can pay at the worst visible price of the walk bp = self._buying_power(acct) max_q = self._max_affordable(book, o.qty, bp, spec.tick_cents, taker_bps) if max_q <= 0: return self._reject(o, "insufficient_cash") if max_q < o.qty: o.qty = max_q o.remaining = max_q else: held = acct.free_qty(o.asset) if o.qty > held: if not (acct.short_allowed and self.config.allow_short): if held <= 0: return self._reject(o, "insufficient_shares") o.qty = held o.remaining = held else: short_after = acct.positions.get(o.asset, 0) - acct.reserved_qty.get(o.asset, 0) - o.qty if -short_after > self.config.max_short_lots: return self._reject(o, "short_limit") if o.order_type is OrderType.MARKET and book.best_bid() is None: return self._reject(o, "no_liquidity") acct.reserved_qty[o.asset] = acct.reserved_qty.get(o.asset, 0) + o.qty o.reserved_qty = o.qty if self.config.position_limit is not None and o.tag != "liquidation": pos = acct.positions.get(o.asset, 0) pend = sum(x.remaining for x in self.books[o.asset].resting_for(o.agent_id) if x.side is o.side) if abs(pos + int(o.side) * (pend + o.qty)) > self.config.position_limit: self._release(o, o.remaining) return self._reject(o, "position_limit") # --- leverage check on the post-trade exposure (approximate: full fill at reference) --- if self.config.margin_check and acct.leverage_cap > 1.0 and o.tag != "liquidation": marks = self.marks() eq = acct.equity(marks) px = (o.price if o.price is not None else self.mid_ticks(o.asset)) * spec.tick_cents new_pos = acct.positions.get(o.asset, 0) + int(o.side) * o.qty gross = acct.gross_exposure(marks) - abs(acct.positions.get(o.asset, 0)) * marks[o.asset] + abs(new_pos) * px if eq <= 0 or gross > acct.leverage_cap * eq: self._release(o, o.remaining) return self._reject(o, "leverage_cap") self._log("accept", order_id=o.order_id, agent_id=o.agent_id, asset=o.asset, side=int(o.side), type=o.order_type.value, qty=o.qty, price=-1 if o.price is None else o.price, tag=o.tag) # --- match --------------------------------------------------------------------------- if self.config.call_auction_interval: if o.order_type is OrderType.MARKET: ref = self.mid_ticks(o.asset) o.price = int(ref * (1 + spec.price_band_pct * 0.99)) if o.side is Side.BUY else max(1, int(ref * (1 - spec.price_band_pct * 0.99))) o.order_type = OrderType.LIMIT o.tag = (o.tag + "|ioc") if o.tag else "ioc" # cancelled after the next auction if o.side is Side.BUY: need = o.price * spec.tick_cents * o.qty need += self._fee(need, taker_bps) + o.qty if need > self._buying_power(acct): return self._reject(o, "insufficient_cash") acct.reserved_cash += need o.reserved_cash = need book.rest(o) return fills = book.match(o) for f in fills: self._settle(f.resting, f.incoming, f.price, f.qty, aggressor=o.side) if o.remaining > 0: if o.order_type is OrderType.LIMIT: book.rest(o) else: # unfilled market remainder is cancelled (immediate-or-cancel semantics) self._release(o, o.remaining) o.status = OrderStatus.CANCELLED if o.filled > 0 else OrderStatus.REJECTED if o.filled == 0: o.reject_reason = "no_liquidity" acct.n_rejects += 1 self._log("cancel", order_id=o.order_id, agent_id=o.agent_id, asset=o.asset, remaining=o.remaining, reason="ioc") self._maybe_halt(o.asset) def _buying_power(self, acct: Account) -> int: if acct.leverage_cap <= 1.0: return acct.free_cash # leveraged agents may borrow against equity; limit enforced again on exposure eq = acct.equity(self.marks()) return int(max(acct.free_cash, 0) + max(eq, 0) * (acct.leverage_cap - 1.0)) def _max_affordable(self, book: LimitOrderBook, qty: int, cash: int, tick_cents: int, bps: float) -> int: """Largest number of lots (<= qty) whose total cost — notional plus the fee on the *total* notional — fits in `cash`, walking the ask side level by level.""" remaining, notional, got = qty, 0, 0 for p, q in book.asks.items(): level = sum(o.remaining for o in q) take = min(level, remaining) for _ in range(take): n2 = notional + p * tick_cents if n2 + self._fee(n2, bps) > cash: return got notional = n2 got += 1 remaining -= take if remaining == 0: break return got def _release(self, o: Order, qty: int) -> None: """Release whatever reservation is still attached to order `o` (called when it leaves the book).""" acct = self.ledger[o.agent_id] if o.side is Side.BUY and o.reserved_cash: acct.reserved_cash -= o.reserved_cash o.reserved_cash = 0 elif o.side is Side.SELL and o.reserved_qty: acct.reserved_qty[o.asset] -= o.reserved_qty o.reserved_qty = 0 def _settle(self, resting: Order, incoming: Order, price: int, qty: int, aggressor: Side) -> None: spec = self.assets[resting.asset] notional = price * spec.tick_cents * qty maker_fee = self._fee(notional, self.config.maker_fee_bps) taker_fee = self._fee(notional, self.config.taker_fee_bps) if aggressor is Side.BUY: buy, sell = incoming, resting buyer_fee, seller_fee = taker_fee, maker_fee else: buy, sell = resting, incoming buyer_fee, seller_fee = maker_fee, taker_fee # reservation to release on each side for this quantity: exactly what is charged, everything on the last fill if buy.reserved_cash: buy_rel = buy.reserved_cash if buy.remaining == 0 else min(buy.reserved_cash, notional + buyer_fee) else: buy_rel = 0 sell_rel = min(qty, sell.reserved_qty) if sell.reserved_qty else 0 self.ledger.apply_fill(buy.agent_id, sell.agent_id, resting.asset, price * spec.tick_cents, qty, buyer_fee, seller_fee, buy_rel, sell_rel) buy.reserved_cash -= buy_rel sell.reserved_qty -= sell_rel tr = Trade(trade_id=self._next_trade_id, time=self.time, asset=resting.asset, price=price, qty=qty, buy_order_id=buy.order_id, sell_order_id=sell.order_id, buyer_id=buy.agent_id, seller_id=sell.agent_id, aggressor=aggressor, maker_fee=maker_fee, taker_fee=taker_fee) self._next_trade_id += 1 self.trades.append(tr) st = self.state[resting.asset] st.last_price = price st.last_trade_time = self.time st.volume_tick += qty st.signed_volume_tick += int(aggressor) * qty self._log("trade", trade_id=tr.trade_id, asset=tr.asset, price=price, qty=qty, buyer=buy.agent_id, seller=sell.agent_id, aggressor=int(aggressor), buy_order=buy.order_id, sell_order=sell.order_id, maker_fee=maker_fee, taker_fee=taker_fee) self.ledger.check_conservation() def _on_cancel(self, agent_id: int, order_id: int) -> None: o = self.orders.get(order_id) if o is None or o.agent_id != agent_id: return book = self.books[o.asset] if order_id in book: book.cancel(order_id) self._release(o, o.remaining) self.ledger[agent_id].n_cancels += 1 self._log("cancel", order_id=order_id, agent_id=agent_id, asset=o.asset, remaining=o.remaining, reason="user") # ---- halts ----------------------------------------------------------------------------- def _maybe_halt(self, asset: int) -> None: thr = self.config.halt_threshold if thr is None: return st = self.state[asset] hist = st.mid_history if len(hist) < self.config.halt_window or st.last_price is None: return ref = hist[-self.config.halt_window] if ref <= 0: return if abs(st.last_price / ref - 1.0) > thr and not self.is_halted(asset): st.halted_until = self.time + self.config.halt_duration st.n_halts += 1 self._log("halt", asset=asset, until=st.halted_until, move=st.last_price / ref - 1.0) if self.config.halt_cancel_resting: for o in self.books[asset].clear_all(): self._release(o, o.remaining) self._log("cancel", order_id=o.order_id, agent_id=o.agent_id, asset=asset, remaining=o.remaining, reason="halt") # ---- end-of-tick housekeeping -------------------------------------------------------------- def run_call_auction(self, asset: int) -> int: from .call_auction import clearing_price, executable_orders book = self.books[asset] price, vol = clearing_price(book, self.mid_ticks(asset), self.config.call_auction_collar) if price is None or vol == 0: return 0 bids, asks = executable_orders(book, price) done = 0 bi = ai = 0 while done < vol and bi < len(bids) and ai < len(asks): b, a = bids[bi], asks[ai] q = min(b.remaining, a.remaining, vol - done) b.remaining -= q a.remaining -= q # both sides are resting: treat the later-submitted order as the aggressor for fee purposes aggressor = Side.BUY if b.submit_time >= a.submit_time else Side.SELL self._settle(a if aggressor is Side.BUY else b, b if aggressor is Side.BUY else a, price, q, aggressor) done += q for o in (b, a): if o.remaining == 0: book.cancel(o.order_id) o.status = OrderStatus.FILLED self._release(o, 0) if b.remaining == 0: bi += 1 if a.remaining == 0: ai += 1 self._log("auction", asset=asset, price=price, volume=done) for o in [x for x in book.resting_orders() if "ioc" in x.tag]: # market orders do not carry over book.cancel(o.order_id) self._release(o, o.remaining) self._log("cancel", order_id=o.order_id, agent_id=o.agent_id, asset=asset, remaining=o.remaining, reason="ioc") return done def end_of_tick(self, t: int) -> None: if self.config.call_auction_interval and (t + 1) % self.config.call_auction_interval == 0: for a in self.books: self.run_call_auction(a) for a, st in self.state.items(): st.mid_history.append(self.mid_ticks(a)) st.volume_tick = 0 st.signed_volume_tick = 0 if self.config.order_ttl is not None: for a, book in self.books.items(): expired = [o for o in book.resting_orders() if t - o.submit_time >= self.config.order_ttl] for o in expired: book.cancel(o.order_id) self._release(o, o.remaining) self._log("expire", order_id=o.order_id, agent_id=o.agent_id, asset=a, remaining=o.remaining) self._log("tick", t=t, **{f"mid_{a}": self.mid_ticks(a) for a in self.books}) def pay_dividends(self, asset: int, per_lot_cents: int) -> None: paid = self.ledger.pay_dividend(asset, per_lot_cents) self._log("dividend", asset=asset, per_lot_cents=per_lot_cents, paid=paid) self.ledger.check_conservation() def margin_check(self) -> list[Action]: """Returns liquidation actions for agents violating margin; freezes bankrupt agents.""" if not self.config.margin_check: return [] marks = self.marks() actions: list[tuple[int, Action]] = [] for acct in self.ledger.accounts.values(): if acct.frozen: continue eq = acct.equity(marks) gross = acct.gross_exposure(marks) if gross <= 0: continue if eq <= 0: acct.frozen = True self._log("bankrupt", agent_id=acct.agent_id, equity=eq) self._liquidate_all(acct, actions, marks) continue if acct.leverage_cap > 1.0 and eq / gross < acct.maintenance_margin: self._log("margin_call", agent_id=acct.agent_id, equity=eq, gross=gross) self._liquidate(acct, actions, self.config.liquidation_fraction) return actions # type: ignore[return-value] def _liquidate(self, acct: Account, actions: list, fraction: float) -> None: from .orders import OrderIntent act = Action() for a, book in self.books.items(): for o in book.resting_for(acct.agent_id): from .orders import CancelIntent act.cancels.append(CancelIntent(o.order_id)) for a, q in acct.positions.items(): if q == 0: continue n = max(1, int(abs(q) * fraction)) act.orders.append(OrderIntent(asset=a, side=Side.SELL if q > 0 else Side.BUY, order_type=OrderType.MARKET, qty=n, tag="liquidation")) if not act.is_hold: self._log("liquidate", agent_id=acct.agent_id, n_orders=len(act.orders)) actions.append((acct.agent_id, act)) def _liquidate_all(self, acct: Account, actions: list, marks: dict[int, float]) -> None: self._liquidate(acct, actions, 1.0) # ---- replay ---------------------------------------------------------------------------- def snapshot(self) -> dict: """Hashable summary of the whole state (used by determinism and replay tests).""" return { "time": self.time, "cash": {a: acc.cash for a, acc in sorted(self.ledger.accounts.items())}, "pos": {a: dict(sorted(acc.positions.items())) for a, acc in sorted(self.ledger.accounts.items())}, "fees": self.ledger.exchange_fees, "books": {a: self.books[a].depth(1000) for a in self.books}, "n_trades": len(self.trades), "n_events": len(self.events), }