1"""The exchange: multi-asset continuous double auction with latency, fees, risk limits, halts and an2append-only event log that can be replayed deterministically.34Time is measured in ticks (float). Agents submit `Action`s at integer decision times; each intent5becomes an `Order` that *arrives* at submit_time + latency. The engine processes arrivals in6(arrival_time, seq) order, so two agents deciding at the same tick are ordered by latency and then7by a seeded tie-break assigned at submission.8"""9from __future__ import annotations1011import heapq12from dataclasses import dataclass, field1314import numpy as np1516from .accounts import Account, Ledger17from .book import LimitOrderBook18from .orders import Action, Event, Order, OrderStatus, OrderType, Side, Trade192021@dataclass(slots=True)22class AssetSpec:23 asset: int24 name: str25 tick_cents: int = 1 # tick size in cents (1 => prices quoted in cents)26 lot: int = 127 price_band_pct: float = 0.5 # orders farther than this from the reference price are rejected28 max_order_qty: int = 10_000293031@dataclass(slots=True)32class ExchangeConfig:33 maker_fee_bps: float = 0.0 # negative = rebate34 taker_fee_bps: float = 2.035 min_latency: float = 0.05 # ticks36 halt_threshold: float | None = None # |ret over window| above this triggers a halt (None = off)37 halt_window: int = 1038 halt_duration: int = 2039 halt_cancel_resting: bool = True40 allow_short: bool = False41 max_short_lots: int = 042 margin_check: bool = True43 liquidation_fraction: float = 0.5 # share of the position sold on a margin call44 order_ttl: int | None = None # ticks after which a resting order expires (None = GTC)45 position_limit: int | None = None # max |position| per agent (lots)46 call_auction_interval: int | None = None # batch clearing every k ticks instead of continuous matching47 call_auction_collar: float = 0.02 # max move of the clearing price per auction (fraction of the reference)484950@dataclass(slots=True)51class MarketState:52 """Public, per-asset running statistics kept by the engine for observations and metrics."""53 last_price: int | None = None54 last_trade_time: float = -1.055 reference_price: int = 056 halted_until: float = -1.057 n_halts: int = 058 volume_tick: int = 059 signed_volume_tick: int = 060 mid_history: list[float] = field(default_factory=list) # one entry per completed tick616263class Exchange:64 def __init__(self, assets: list[AssetSpec], config: ExchangeConfig, rng: np.random.Generator,65 initial_prices: dict[int, int]) -> None:66 self.assets = {a.asset: a for a in assets}67 self.config = config68 self.rng = rng69 self.books: dict[int, LimitOrderBook] = {a: LimitOrderBook(a) for a in self.assets}70 self.ledger = Ledger()71 self.state: dict[int, MarketState] = {}72 for a, p0 in initial_prices.items():73 self.state[a] = MarketState(reference_price=p0, last_price=p0)74 self.events: list[Event] = []75 self.trades: list[Trade] = []76 self._queue: list[tuple[float, int, str, object]] = [] # (arrival, seq, kind, payload)77 self._seq = 078 self._next_order_id = 179 self._next_trade_id = 180 self.orders: dict[int, Order] = {} # all orders ever submitted (for replay and stats)81 self.time: float = 0.082 self.latency: dict[int, float] = {} # agent -> latency in ticks83 self.pending_by_agent: dict[int, set[int]] = {}8485 # ---- setup -----------------------------------------------------------------------------86 def register(self, acct: Account, latency: float) -> None:87 self.ledger.add(acct)88 self.latency[acct.agent_id] = max(latency, self.config.min_latency)8990 # ---- event log -------------------------------------------------------------------------91 def _log(self, kind: str, **payload) -> None:92 self._seq += 193 self.events.append(Event(seq=self._seq, time=self.time, kind=kind, payload=payload))9495 # ---- public market data ----------------------------------------------------------------96 def marks(self) -> dict[int, float]:97 """Mark-to-market price in cents per lot: mid if available, else last trade, else reference."""98 out = {}99 for a in self.books:100 out[a] = self.mid_ticks(a) * self.assets[a].tick_cents101 return out102103 def mid_ticks(self, asset: int) -> float:104 st = self.state[asset]105 if self.config.call_auction_interval: # batch mode: the book may be crossed; the auction price is the reference106 return float(st.last_price if st.last_price is not None else st.reference_price)107 m = self.books[asset].mid()108 if m is not None:109 return m110 return float(st.last_price if st.last_price is not None else st.reference_price)111112 def spread_ticks(self, asset: int) -> int | None:113 if self.config.call_auction_interval:114 return None115 return self.books[asset].spread()116117 def is_halted(self, asset: int) -> bool:118 return self.time < self.state[asset].halted_until119120 # ---- submission --------------------------------------------------------------------------121 def submit(self, agent_id: int, action: Action, decision_time: float) -> list[int]:122 """Queue an agent's intents; returns the order ids assigned (cancels get none)."""123 ids: list[int] = []124 lat = self.latency[agent_id]125 for c in action.cancels:126 self._seq += 1127 heapq.heappush(self._queue, (decision_time + lat, self._seq, "cancel", (agent_id, c.order_id)))128 for it in action.orders:129 oid = self._next_order_id130 self._next_order_id += 1131 o = Order(order_id=oid, agent_id=agent_id, asset=it.asset, side=it.side, order_type=it.order_type,132 qty=int(it.qty), price=None if it.price is None else int(it.price),133 submit_time=decision_time, arrival_time=decision_time + lat, tag=it.tag)134 self.orders[oid] = o135 self.pending_by_agent.setdefault(agent_id, set()).add(oid)136 self._seq += 1137 heapq.heappush(self._queue, (o.arrival_time, self._seq, "order", o))138 ids.append(oid)139 self.ledger[agent_id].n_orders += 1140 return ids141142 # ---- processing --------------------------------------------------------------------------143 def process_until(self, t_end: float) -> None:144 """Process every queued arrival with arrival_time < t_end, in order."""145 while self._queue and self._queue[0][0] < t_end:146 arrival, _, kind, payload = heapq.heappop(self._queue)147 self.time = arrival148 if kind == "order":149 self._on_order(payload) # type: ignore[arg-type]150 else:151 agent_id, oid = payload # type: ignore[misc]152 self._on_cancel(agent_id, oid)153 self.time = t_end154155 def _fee(self, notional: int, bps: float) -> int:156 return int(round(notional * bps / 10_000.0))157158 def _reject(self, o: Order, reason: str) -> None:159 o.status = OrderStatus.REJECTED160 o.reject_reason = reason161 self.ledger[o.agent_id].n_rejects += 1162 self._log("reject", order_id=o.order_id, agent_id=o.agent_id, asset=o.asset, reason=reason)163164 def _on_order(self, o: Order) -> None:165 self.pending_by_agent.get(o.agent_id, set()).discard(o.order_id)166 acct = self.ledger[o.agent_id]167 spec = self.assets[o.asset]168 book = self.books[o.asset]169 st = self.state[o.asset]170 if acct.frozen and o.tag != "liquidation":171 return self._reject(o, "frozen")172 if self.is_halted(o.asset):173 return self._reject(o, "halted")174 if o.qty <= 0 or o.qty > spec.max_order_qty:175 return self._reject(o, "bad_qty")176 if o.order_type is OrderType.LIMIT:177 if o.price is None or o.price <= 0:178 return self._reject(o, "bad_price")179 ref = self.mid_ticks(o.asset)180 if abs(o.price - ref) > spec.price_band_pct * ref:181 return self._reject(o, "price_band")182 # --- capacity checks and reservations ----------------------------------------------183 taker_bps = self.config.taker_fee_bps184 if o.side is Side.BUY:185 if o.order_type is OrderType.LIMIT:186 notional = o.price * spec.tick_cents * o.qty187 need = notional + max(self._fee(notional, taker_bps), self._fee(notional, self.config.maker_fee_bps)) + o.qty188 if need > self._buying_power(acct):189 return self._reject(o, "insufficient_cash")190 acct.reserved_cash += need191 o.reserved_cash = need192 else:193 ba = book.best_ask()194 if ba is None:195 return self._reject(o, "no_liquidity")196 # cap the quantity to what free cash can pay at the worst visible price of the walk197 bp = self._buying_power(acct)198 max_q = self._max_affordable(book, o.qty, bp, spec.tick_cents, taker_bps)199 if max_q <= 0:200 return self._reject(o, "insufficient_cash")201 if max_q < o.qty:202 o.qty = max_q203 o.remaining = max_q204 else:205 held = acct.free_qty(o.asset)206 if o.qty > held:207 if not (acct.short_allowed and self.config.allow_short):208 if held <= 0:209 return self._reject(o, "insufficient_shares")210 o.qty = held211 o.remaining = held212 else:213 short_after = acct.positions.get(o.asset, 0) - acct.reserved_qty.get(o.asset, 0) - o.qty214 if -short_after > self.config.max_short_lots:215 return self._reject(o, "short_limit")216 if o.order_type is OrderType.MARKET and book.best_bid() is None:217 return self._reject(o, "no_liquidity")218 acct.reserved_qty[o.asset] = acct.reserved_qty.get(o.asset, 0) + o.qty219 o.reserved_qty = o.qty220 if self.config.position_limit is not None and o.tag != "liquidation":221 pos = acct.positions.get(o.asset, 0)222 pend = sum(x.remaining for x in self.books[o.asset].resting_for(o.agent_id) if x.side is o.side)223 if abs(pos + int(o.side) * (pend + o.qty)) > self.config.position_limit:224 self._release(o, o.remaining)225 return self._reject(o, "position_limit")226 # --- leverage check on the post-trade exposure (approximate: full fill at reference) ---227 if self.config.margin_check and acct.leverage_cap > 1.0 and o.tag != "liquidation":228 marks = self.marks()229 eq = acct.equity(marks)230 px = (o.price if o.price is not None else self.mid_ticks(o.asset)) * spec.tick_cents231 new_pos = acct.positions.get(o.asset, 0) + int(o.side) * o.qty232 gross = acct.gross_exposure(marks) - abs(acct.positions.get(o.asset, 0)) * marks[o.asset] + abs(new_pos) * px233 if eq <= 0 or gross > acct.leverage_cap * eq:234 self._release(o, o.remaining)235 return self._reject(o, "leverage_cap")236 self._log("accept", order_id=o.order_id, agent_id=o.agent_id, asset=o.asset, side=int(o.side),237 type=o.order_type.value, qty=o.qty, price=-1 if o.price is None else o.price, tag=o.tag)238 # --- match ---------------------------------------------------------------------------239 if self.config.call_auction_interval:240 if o.order_type is OrderType.MARKET:241 ref = self.mid_ticks(o.asset)242 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)))243 o.order_type = OrderType.LIMIT244 o.tag = (o.tag + "|ioc") if o.tag else "ioc" # cancelled after the next auction245 if o.side is Side.BUY:246 need = o.price * spec.tick_cents * o.qty247 need += self._fee(need, taker_bps) + o.qty248 if need > self._buying_power(acct):249 return self._reject(o, "insufficient_cash")250 acct.reserved_cash += need251 o.reserved_cash = need252 book.rest(o)253 return254 fills = book.match(o)255 for f in fills:256 self._settle(f.resting, f.incoming, f.price, f.qty, aggressor=o.side)257 if o.remaining > 0:258 if o.order_type is OrderType.LIMIT:259 book.rest(o)260 else:261 # unfilled market remainder is cancelled (immediate-or-cancel semantics)262 self._release(o, o.remaining)263 o.status = OrderStatus.CANCELLED if o.filled > 0 else OrderStatus.REJECTED264 if o.filled == 0:265 o.reject_reason = "no_liquidity"266 acct.n_rejects += 1267 self._log("cancel", order_id=o.order_id, agent_id=o.agent_id, asset=o.asset,268 remaining=o.remaining, reason="ioc")269 self._maybe_halt(o.asset)270271 def _buying_power(self, acct: Account) -> int:272 if acct.leverage_cap <= 1.0:273 return acct.free_cash274 # leveraged agents may borrow against equity; limit enforced again on exposure275 eq = acct.equity(self.marks())276 return int(max(acct.free_cash, 0) + max(eq, 0) * (acct.leverage_cap - 1.0))277278 def _max_affordable(self, book: LimitOrderBook, qty: int, cash: int, tick_cents: int, bps: float) -> int:279 """Largest number of lots (<= qty) whose total cost — notional plus the fee on the *total* notional —280 fits in `cash`, walking the ask side level by level."""281 remaining, notional, got = qty, 0, 0282 for p, q in book.asks.items():283 level = sum(o.remaining for o in q)284 take = min(level, remaining)285 for _ in range(take):286 n2 = notional + p * tick_cents287 if n2 + self._fee(n2, bps) > cash:288 return got289 notional = n2290 got += 1291 remaining -= take292 if remaining == 0:293 break294 return got295296 def _release(self, o: Order, qty: int) -> None:297 """Release whatever reservation is still attached to order `o` (called when it leaves the book)."""298 acct = self.ledger[o.agent_id]299 if o.side is Side.BUY and o.reserved_cash:300 acct.reserved_cash -= o.reserved_cash301 o.reserved_cash = 0302 elif o.side is Side.SELL and o.reserved_qty:303 acct.reserved_qty[o.asset] -= o.reserved_qty304 o.reserved_qty = 0305306 def _settle(self, resting: Order, incoming: Order, price: int, qty: int, aggressor: Side) -> None:307 spec = self.assets[resting.asset]308 notional = price * spec.tick_cents * qty309 maker_fee = self._fee(notional, self.config.maker_fee_bps)310 taker_fee = self._fee(notional, self.config.taker_fee_bps)311 if aggressor is Side.BUY:312 buy, sell = incoming, resting313 buyer_fee, seller_fee = taker_fee, maker_fee314 else:315 buy, sell = resting, incoming316 buyer_fee, seller_fee = maker_fee, taker_fee317 # reservation to release on each side for this quantity: exactly what is charged, everything on the last fill318 if buy.reserved_cash:319 buy_rel = buy.reserved_cash if buy.remaining == 0 else min(buy.reserved_cash, notional + buyer_fee)320 else:321 buy_rel = 0322 sell_rel = min(qty, sell.reserved_qty) if sell.reserved_qty else 0323 self.ledger.apply_fill(buy.agent_id, sell.agent_id, resting.asset, price * spec.tick_cents, qty,324 buyer_fee, seller_fee, buy_rel, sell_rel)325 buy.reserved_cash -= buy_rel326 sell.reserved_qty -= sell_rel327 tr = Trade(trade_id=self._next_trade_id, time=self.time, asset=resting.asset, price=price, qty=qty,328 buy_order_id=buy.order_id, sell_order_id=sell.order_id, buyer_id=buy.agent_id,329 seller_id=sell.agent_id, aggressor=aggressor, maker_fee=maker_fee, taker_fee=taker_fee)330 self._next_trade_id += 1331 self.trades.append(tr)332 st = self.state[resting.asset]333 st.last_price = price334 st.last_trade_time = self.time335 st.volume_tick += qty336 st.signed_volume_tick += int(aggressor) * qty337 self._log("trade", trade_id=tr.trade_id, asset=tr.asset, price=price, qty=qty, buyer=buy.agent_id,338 seller=sell.agent_id, aggressor=int(aggressor), buy_order=buy.order_id, sell_order=sell.order_id,339 maker_fee=maker_fee, taker_fee=taker_fee)340 self.ledger.check_conservation()341342 def _on_cancel(self, agent_id: int, order_id: int) -> None:343 o = self.orders.get(order_id)344 if o is None or o.agent_id != agent_id:345 return346 book = self.books[o.asset]347 if order_id in book:348 book.cancel(order_id)349 self._release(o, o.remaining)350 self.ledger[agent_id].n_cancels += 1351 self._log("cancel", order_id=order_id, agent_id=agent_id, asset=o.asset, remaining=o.remaining,352 reason="user")353354 # ---- halts -----------------------------------------------------------------------------355 def _maybe_halt(self, asset: int) -> None:356 thr = self.config.halt_threshold357 if thr is None:358 return359 st = self.state[asset]360 hist = st.mid_history361 if len(hist) < self.config.halt_window or st.last_price is None:362 return363 ref = hist[-self.config.halt_window]364 if ref <= 0:365 return366 if abs(st.last_price / ref - 1.0) > thr and not self.is_halted(asset):367 st.halted_until = self.time + self.config.halt_duration368 st.n_halts += 1369 self._log("halt", asset=asset, until=st.halted_until, move=st.last_price / ref - 1.0)370 if self.config.halt_cancel_resting:371 for o in self.books[asset].clear_all():372 self._release(o, o.remaining)373 self._log("cancel", order_id=o.order_id, agent_id=o.agent_id, asset=asset,374 remaining=o.remaining, reason="halt")375376 # ---- end-of-tick housekeeping --------------------------------------------------------------377 def run_call_auction(self, asset: int) -> int:378 from .call_auction import clearing_price, executable_orders379 book = self.books[asset]380 price, vol = clearing_price(book, self.mid_ticks(asset), self.config.call_auction_collar)381 if price is None or vol == 0:382 return 0383 bids, asks = executable_orders(book, price)384 done = 0385 bi = ai = 0386 while done < vol and bi < len(bids) and ai < len(asks):387 b, a = bids[bi], asks[ai]388 q = min(b.remaining, a.remaining, vol - done)389 b.remaining -= q390 a.remaining -= q391 # both sides are resting: treat the later-submitted order as the aggressor for fee purposes392 aggressor = Side.BUY if b.submit_time >= a.submit_time else Side.SELL393 self._settle(a if aggressor is Side.BUY else b, b if aggressor is Side.BUY else a, price, q, aggressor)394 done += q395 for o in (b, a):396 if o.remaining == 0:397 book.cancel(o.order_id)398 o.status = OrderStatus.FILLED399 self._release(o, 0)400 if b.remaining == 0:401 bi += 1402 if a.remaining == 0:403 ai += 1404 self._log("auction", asset=asset, price=price, volume=done)405 for o in [x for x in book.resting_orders() if "ioc" in x.tag]: # market orders do not carry over406 book.cancel(o.order_id)407 self._release(o, o.remaining)408 self._log("cancel", order_id=o.order_id, agent_id=o.agent_id, asset=asset, remaining=o.remaining, reason="ioc")409 return done410411 def end_of_tick(self, t: int) -> None:412 if self.config.call_auction_interval and (t + 1) % self.config.call_auction_interval == 0:413 for a in self.books:414 self.run_call_auction(a)415 for a, st in self.state.items():416 st.mid_history.append(self.mid_ticks(a))417 st.volume_tick = 0418 st.signed_volume_tick = 0419 if self.config.order_ttl is not None:420 for a, book in self.books.items():421 expired = [o for o in book.resting_orders() if t - o.submit_time >= self.config.order_ttl]422 for o in expired:423 book.cancel(o.order_id)424 self._release(o, o.remaining)425 self._log("expire", order_id=o.order_id, agent_id=o.agent_id, asset=a, remaining=o.remaining)426 self._log("tick", t=t, **{f"mid_{a}": self.mid_ticks(a) for a in self.books})427428 def pay_dividends(self, asset: int, per_lot_cents: int) -> None:429 paid = self.ledger.pay_dividend(asset, per_lot_cents)430 self._log("dividend", asset=asset, per_lot_cents=per_lot_cents, paid=paid)431 self.ledger.check_conservation()432433 def margin_check(self) -> list[Action]:434 """Returns liquidation actions for agents violating margin; freezes bankrupt agents."""435 if not self.config.margin_check:436 return []437 marks = self.marks()438 actions: list[tuple[int, Action]] = []439 for acct in self.ledger.accounts.values():440 if acct.frozen:441 continue442 eq = acct.equity(marks)443 gross = acct.gross_exposure(marks)444 if gross <= 0:445 continue446 if eq <= 0:447 acct.frozen = True448 self._log("bankrupt", agent_id=acct.agent_id, equity=eq)449 self._liquidate_all(acct, actions, marks)450 continue451 if acct.leverage_cap > 1.0 and eq / gross < acct.maintenance_margin:452 self._log("margin_call", agent_id=acct.agent_id, equity=eq, gross=gross)453 self._liquidate(acct, actions, self.config.liquidation_fraction)454 return actions # type: ignore[return-value]455456 def _liquidate(self, acct: Account, actions: list, fraction: float) -> None:457 from .orders import OrderIntent458 act = Action()459 for a, book in self.books.items():460 for o in book.resting_for(acct.agent_id):461 from .orders import CancelIntent462 act.cancels.append(CancelIntent(o.order_id))463 for a, q in acct.positions.items():464 if q == 0:465 continue466 n = max(1, int(abs(q) * fraction))467 act.orders.append(OrderIntent(asset=a, side=Side.SELL if q > 0 else Side.BUY,468 order_type=OrderType.MARKET, qty=n, tag="liquidation"))469 if not act.is_hold:470 self._log("liquidate", agent_id=acct.agent_id, n_orders=len(act.orders))471 actions.append((acct.agent_id, act))472473 def _liquidate_all(self, acct: Account, actions: list, marks: dict[int, float]) -> None:474 self._liquidate(acct, actions, 1.0)475476 # ---- replay ----------------------------------------------------------------------------477 def snapshot(self) -> dict:478 """Hashable summary of the whole state (used by determinism and replay tests)."""479 return {480 "time": self.time,481 "cash": {a: acc.cash for a, acc in sorted(self.ledger.accounts.items())},482 "pos": {a: dict(sorted(acc.positions.items())) for a, acc in sorted(self.ledger.accounts.items())},483 "fees": self.ledger.exchange_fees,484 "books": {a: self.books[a].depth(1000) for a in self.books},485 "n_trades": len(self.trades),486 "n_events": len(self.events),487 }488