Problem Statement
Design a stock trading order book system. Traders submit buy and sell orders at specific prices and quantities. The system matches buy orders with sell orders when prices overlap (a buy price >= a sell price), executes trades at the matching price, and maintains the unfilled portion of each order. Supports limit orders (execute at a specific price or better), market orders (execute immediately at best available price), and order cancellation.
Core Entities
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional
import heapq
import uuid
class OrderSide(Enum):
BUY = "buy"
SELL = "sell"
class OrderType(Enum):
LIMIT = "limit" # Execute at price or better
MARKET = "market" # Execute at best available price
class OrderStatus(Enum):
OPEN = "open"
PARTIAL = "partial" # Partially filled
FILLED = "filled" # Completely filled
CANCELLED = "cancelled"
@dataclass
class Order:
order_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
symbol: str = ""
side: OrderSide = OrderSide.BUY
order_type: OrderType = OrderType.LIMIT
price: float = 0.0 # 0 for market orders
quantity: int = 0
filled_qty: int = 0
status: OrderStatus = OrderStatus.OPEN
timestamp: datetime = field(default_factory=datetime.now)
trader_id: str = ""
@property
def remaining_qty(self) -> int:
return self.quantity - self.filled_qty
@property
def is_active(self) -> bool:
return self.status in (OrderStatus.OPEN, OrderStatus.PARTIAL)
@dataclass
class Trade:
trade_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
symbol: str = ""
buy_order_id: str = ""
sell_order_id: str = ""
price: float = 0.0
quantity: int = 0
timestamp: datetime = field(default_factory=datetime.now)
Order Book with Priority Queues
class OrderBook:
"""
Maintains buy and sell sides as priority queues.
Buy side: max-heap by price (highest bid has priority)
Sell side: min-heap by price (lowest ask has priority)
Within same price: FIFO by timestamp (time priority)
"""
def __init__(self, symbol: str):
self.symbol = symbol
# Buy side: max-heap → negate price. Tuple: (-price, timestamp, order)
self._bids: list = []
# Sell side: min-heap. Tuple: (price, timestamp, order)
self._asks: list = []
self._orders: dict[str, Order] = {} # order_id -> Order
self._trades: list[Trade] = []
def add_order(self, order: Order) -> list[Trade]:
"""Add an order and attempt to match. Returns list of executed trades."""
order.symbol = self.symbol
self._orders[order.order_id] = order
trades = self._match(order)
# If not fully filled and it's a limit order, add to book
if order.is_active and order.order_type == OrderType.LIMIT:
self._add_to_book(order)
elif order.is_active and order.order_type == OrderType.MARKET:
# Market order not filled → cancel (no liquidity)
order.status = OrderStatus.CANCELLED
print(f"Market order {order.order_id} cancelled: insufficient liquidity")
return trades
def _add_to_book(self, order: Order):
if order.side == OrderSide.BUY:
heapq.heappush(self._bids,
(-order.price, order.timestamp, order))
else:
heapq.heappush(self._asks,
(order.price, order.timestamp, order))
def _match(self, incoming: Order) -> list[Trade]:
trades = []
if incoming.side == OrderSide.BUY:
opposite = self._asks
can_match = lambda ask_price: (
incoming.order_type == OrderType.MARKET or
incoming.price >= ask_price
)
get_price = lambda item: item[0]
else:
opposite = self._bids
can_match = lambda bid_price: (
incoming.order_type == OrderType.MARKET or
incoming.price Trade:
buy_order = order_a if order_a.side == OrderSide.BUY else order_b
sell_order = order_a if order_a.side == OrderSide.SELL else order_b
for order in (order_a, order_b):
order.filled_qty += quantity
order.status = (OrderStatus.FILLED
if order.remaining_qty == 0
else OrderStatus.PARTIAL)
trade = Trade(
symbol = self.symbol,
buy_order_id = buy_order.order_id,
sell_order_id = sell_order.order_id,
price = price,
quantity = quantity,
)
self._trades.append(trade)
print(f"TRADE: {quantity} {self.symbol} @ {price:.2f} "
f"(buy:{buy_order.order_id} sell:{sell_order.order_id})")
return trade
def cancel_order(self, order_id: str) -> bool:
order = self._orders.get(order_id)
if not order or not order.is_active:
return False
# Lazy deletion: mark as cancelled. Will be skipped during matching.
order.status = OrderStatus.CANCELLED
print(f"Order {order_id} cancelled")
return True
def best_bid(self) -> Optional[float]:
while self._bids:
if self._bids[0][2].is_active:
return -self._bids[0][0]
heapq.heappop(self._bids)
return None
def best_ask(self) -> Optional[float]:
while self._asks:
if self._asks[0][2].is_active:
return self._asks[0][0]
heapq.heappop(self._asks)
return None
def spread(self) -> Optional[float]:
bid = self.best_bid()
ask = self.best_ask()
return round(ask - bid, 4) if bid and ask else None
def order_book_snapshot(self, depth: int = 5) -> dict:
"""Return top-depth bid and ask levels."""
bids = {}
asks = {}
for _, ts, order in sorted(self._bids):
if order.is_active:
price = order.price
bids[price] = bids.get(price, 0) + order.remaining_qty
for price, ts, order in sorted(self._asks):
if order.is_active:
asks[price] = asks.get(price, 0) + order.remaining_qty
sorted_bids = sorted(bids.items(), reverse=True)[:depth]
sorted_asks = sorted(asks.items())[:depth]
return {"bids": sorted_bids, "asks": sorted_asks,
"spread": self.spread()}
Usage Example
def demo():
book = OrderBook("AAPL")
# Resting sell orders
book.add_order(Order(side=OrderSide.SELL, order_type=OrderType.LIMIT,
price=151.00, quantity=100, trader_id="T1"))
book.add_order(Order(side=OrderSide.SELL, order_type=OrderType.LIMIT,
price=150.50, quantity=50, trader_id="T2"))
# Resting buy orders
book.add_order(Order(side=OrderSide.BUY, order_type=OrderType.LIMIT,
price=149.00, quantity=200, trader_id="T3"))
print(f"Best bid: {book.best_bid()}, Best ask: {book.best_ask()}")
print(f"Spread: {book.spread()}")
# Aggressive buy order — crosses the spread
trades = book.add_order(Order(side=OrderSide.BUY, order_type=OrderType.LIMIT,
price=151.00, quantity=120, trader_id="T4"))
# T4 buys 50 @ 150.50 from T2 (best ask), then 70 @ 151.00 from T1
print(f"Trades executed: {len(trades)}")
print(book.order_book_snapshot())
demo()
Design Decisions
- Heap-based book: O(log N) for add, O(log N) for best price lookup. For production-grade HFT systems, use a sorted dictionary (price → deque of orders) — O(1) best price access, O(1) FIFO order at each price level.
- Lazy deletion: Cancelled orders are marked but not immediately removed from the heap. They are discarded when popped during matching. Avoids O(N) heap traversal for cancellation. The downside: heap can accumulate stale entries — bound by monitoring heap size and triggering cleanup.
- Trade price: The resting (passive) order determines the execution price — price-time priority. The incoming (aggressive) order accepts the resting order’s price, which is always equal to or better than the incoming order’s limit.
- Market orders: Execute against the best available price without a price constraint. If insufficient liquidity exists, the unfilled portion is cancelled.
Interview Checklist
- Data structures: buy side = max-heap by price, sell side = min-heap by price
- Matching rule: buy price >= sell price; execute at resting order’s price
- Order types: limit (with price constraint), market (best available price)
- Cancellation: lazy deletion with active status check during matching
- Partial fills: update filled_qty, re-add resting order to book if remaining
- Production: sorted dict by price level + deque per level for O(1) FIFO access
🏢 Asked at: Coinbase Interview Guide
🏢 Asked at: Stripe Interview Guide
🏢 Asked at: Databricks Interview Guide 2026: Spark Internals, Delta Lake, and Lakehouse Architecture
🏢 Asked at: LinkedIn Interview Guide 2026: Social Graph Engineering, Feed Ranking, and Professional Network Scale
🏢 Asked at: Atlassian Interview Guide