# Low-Level Design: Stock Order Book (Trading System OOP Interview)

Source: https://www.techinterview.org/post/3233464195/lld-stock-order-book/
Updated: 2026-07-05 · techinterview.org

## 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 <= bid_price
            )
            get_price = lambda item: -item[0]

        # Walk the opposite side while prices cross and the order still has size
        while opposite and incoming.is_active:
            best = opposite[0]
            resting = best[2]
            # Lazy deletion: discard cancelled/filled resting orders
            if not resting.is_active:
                heapq.heappop(opposite)
                continue
            price = get_price(best)
            if not can_match(price):
                break
            # Trade at the resting order's price for the overlapping quantity
            qty = min(incoming.remaining_qty, resting.remaining_qty)
            trades.append(self._execute(incoming, resting, price, qty))
            # Remove the resting order if it was fully filled
            if not resting.is_active:
                heapq.heappop(opposite)
        return trades

    def _execute(self, order_a: Order, order_b: Order,
                 price: float, quantity: int) -> 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](https://techinterview.org/coinbase-interview-guide/)

🏢 Asked at: [Stripe Interview Guide](https://techinterview.org/stripe-interview-guide/)

🏢 Asked at: [Databricks Interview Guide 2026: Spark Internals, Delta Lake, and Lakehouse Architecture](https://techinterview.org/databricks-interview-guide/)

🏢 Asked at: [LinkedIn Interview Guide 2026: Social Graph Engineering, Feed Ranking, and Professional Network Scale](https://techinterview.org/linkedin-interview-guide/)

🏢 Asked at: [Atlassian Interview Guide](https://techinterview.org/atlassian-interview-guide/)
