Saturday, June 13, 2026

限价单 (Limit Order):  以指定价格或更优价格买/卖
市价单 (Market Order): 立即以最优市价成交
撤单   (Cancel Order): 取消已有挂单

# Price level: all orders at the same price
# Use deque for O(1) FIFO add/remove at head
from collections import deque

class PriceLevel:
    def __init__(self, price: float):
        self.price = price
        self.orders: deque[Order] = deque()  # FIFO queue
        self.total_qty: int = 0

# Order book side: sorted price levels
# Bids: max-heap (negate price for Python's min-heap)
# Asks: min-heap
import heapq

class OrderBookSide:
    def __init__(self, is_bid: bool):
        self.is_bid = is_bid
        self.levels: dict[float, PriceLevel] = {}
        self.heap: list[float] = []  # price heap for O(log n) best price

import heapq
import time
import uuid
from collections import deque
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class Side(Enum):
    BID = "bid"  # Buy order
    ASK = "ask"  # Sell order


class OrderType(Enum):
    LIMIT = "limit"
    MARKET = "market"


@dataclass
class Order:
    order_id: str
    symbol: str
    side: Side
    order_type: OrderType
    price: Optional[float]   # None for market orders
    quantity: int
    timestamp: int = field(default_factory=lambda: time.time_ns())
    filled_qty: int = 0

    @property
    def remaining_qty(self) -> int:
        return self.quantity - self.filled_qty


@dataclass
class Trade:
    trade_id: str
    symbol: str
    price: float
    quantity: int
    buyer_order_id: str
    seller_order_id: str
    timestamp: int = field(default_factory=lambda: time.time_ns())


class OrderBook:
    """
    Maintains bids and asks for a single symbol.

    Internal representation:
      - bids: max-heap (negate price → Python min-heap trick)
      - asks: min-heap
      - levels: dict mapping price → PriceLevel (deque of orders)
      - order_map: dict mapping order_id → Order (for O(1) cancel)
    """

    def __init__(self, symbol: str):
        self.symbol = symbol
        # Bids: store (-price, timestamp) for max-heap behavior
        self._bid_heap: list[tuple[float, int, float]] = []  # (-price, ts, price)
        # Asks: store (price, timestamp)
        self._ask_heap: list[tuple[float, int]] = []
        self._bid_levels: dict[float, deque] = {}
        self._ask_levels: dict[float, deque] = {}
        self._order_map: dict[str, Order] = {}  # for cancellation
        self._cancelled: set[str] = set()       # lazy deletion set

    def best_bid(self) -> Optional[float]:
        """O(log n) — skip cancelled/empty levels"""
        while self._bid_heap:
            neg_price, _, price = self._bid_heap[0]
            level = self._bid_levels.get(price)
            # Clean up: if level is empty or all cancelled, pop and continue
            if level and len(level) > 0:
                return price
            heapq.heappop(self._bid_heap)
            if price in self._bid_levels:
                del self._bid_levels[price]
        return None

    def best_ask(self) -> Optional[float]:
        """O(log n) — skip empty levels"""
        while self._ask_heap:
            price, _ = self._ask_heap[0]
            level = self._ask_levels.get(price)
            if level and len(level) > 0:
                return price
            heapq.heappop(self._ask_heap)
            if price in self._ask_levels:
                del self._ask_levels[price]
        return None

    def add_limit_order(self, order: Order):
        """Add limit order to the book (no immediate matching here)."""
        self._order_map[order.order_id] = order
        if order.side == Side.BID:
            if order.price not in self._bid_levels:
                self._bid_levels[order.price] = deque()
                # Push to heap: negate for max-heap, include timestamp for tie-breaking
                heapq.heappush(self._bid_heap, (-order.price, order.timestamp, order.price))
            self._bid_levels[order.price].append(order)
        else:
            if order.price not in self._ask_levels:
                self._ask_levels[order.price] = deque()
                heapq.heappush(self._ask_heap, (order.price, order.timestamp))
            self._ask_levels[order.price].append(order)

    def cancel_order(self, order_id: str) -> bool:
        """
        Lazy deletion: mark as cancelled, skip during matching.
        O(1) — no heap restructuring needed.
        """
        if order_id not in self._order_map:
            return False
        self._cancelled.add(order_id)
        return True

    def pop_best_ask(self) -> Optional[Order]:
        """Get and remove the best ask order (FIFO within price level)."""
        while True:
            best = self.best_ask()
            if best is None:
                return None
            level = self._ask_levels[best]
            while level:
                order = level[0]
                if order.order_id in self._cancelled or order.remaining_qty <= 0:
                    level.popleft()  # Remove cancelled/filled orders
                    continue
                return level[0]  # Peek — don't pop yet (partial fills)
            # Level empty after cleanup
            del self._ask_levels[best]
            heapq.heappop(self._ask_heap)

    def pop_best_bid(self) -> Optional[Order]:
        """Get and remove the best bid order (FIFO within price level)."""
        while True:
            best = self.best_bid()
            if best is None:
                return None
            level = self._bid_levels[best]
            while level:
                order = level[0]
                if order.order_id in self._cancelled or order.remaining_qty <= 0:
                    level.popleft()
                    continue
                return level[0]
            del self._bid_levels[best]
            heapq.heappop(self._bid_heap)


class MatchingEngine:
    """
    Core matching engine. Single-threaded per symbol (no locks needed).

    Architecture:
      - One OrderBook per symbol (isolated state)
      - process_order() = the single entry point
      - Returns list of Trade objects (for downstream processing)

    Matching algorithm: continuous auction (price-time priority)
    """

    def __init__(self):
        self._books: dict[str, OrderBook] = {}

    def _get_book(self, symbol: str) -> OrderBook:
        if symbol not in self._books:
            self._books[symbol] = OrderBook(symbol)
        return self._books[symbol]

    def process_order(self, order: Order) -> list[Trade]:
        """
        Main entry point. Handles limit and market orders.
        Returns list of trades generated.
        """
        book = self._get_book(order.symbol)
        trades = []

        if order.order_type == OrderType.MARKET:
            # Market orders: match immediately, no resting in book
            trades = self._match_market(book, order)
        else:
            # Limit orders: try to match first, rest remainder in book
            trades = self._match_limit(book, order)

        return trades

    def _match_limit(self, book: OrderBook, incoming: Order) -> list[Trade]:
        """
        Limit order matching:
        - BUY limit: match against asks where ask_price <= buy_price
        - SELL limit: match against bids where bid_price >= sell_price
        """
        trades = []

        if incoming.side == Side.BID:
            # Incoming buy: sweep asks from best (lowest) upward
            while incoming.remaining_qty > 0:
                best_ask = book.best_ask()
                # Stop if no asks, or best ask is above our limit price
                if best_ask is None or best_ask > incoming.price:
                    break
                trade = self._execute_match(book, incoming, book, is_incoming_bid=True)
                if trade:
                    trades.append(trade)
        else:
            # Incoming sell: sweep bids from best (highest) downward
            while incoming.remaining_qty > 0:
                best_bid = book.best_bid()
                # Stop if no bids, or best bid is below our limit price
                if best_bid is None or best_bid < incoming.price:
                    break
                trade = self._execute_match(book, incoming, book, is_incoming_bid=False)
                if trade:
                    trades.append(trade)

        # Rest unfilled portion in the book
        if incoming.remaining_qty > 0:
            # Create a new order for remaining quantity
            resting = Order(
                order_id=incoming.order_id,
                symbol=incoming.symbol,
                side=incoming.side,
                order_type=incoming.order_type,
                price=incoming.price,
                quantity=incoming.remaining_qty,  # Only rest what's left
                timestamp=incoming.timestamp,
            )
            book.add_limit_order(resting)

        return trades

    def _match_market(self, book: OrderBook, incoming: Order) -> list[Trade]:
        """Market orders match against all available liquidity at any price."""
        trades = []
        while incoming.remaining_qty > 0:
            if incoming.side == Side.BID:
                if book.best_ask() is None:
                    break  # No liquidity — partial fill
                trade = self._execute_match(book, incoming, book, is_incoming_bid=True)
            else:
                if book.best_bid() is None:
                    break
                trade = self._execute_match(book, incoming, book, is_incoming_bid=False)
            if trade:
                trades.append(trade)
        return trades

    def _execute_match(
        self,
        book: OrderBook,
        incoming: Order,
        _book: OrderBook,
        is_incoming_bid: bool,
    ) -> Optional[Trade]:
        """
        Execute a single match between incoming order and best resting order.
        Price determination: resting order's price (passive price priority).
        """
        if is_incoming_bid:
            resting = book.pop_best_ask()
        else:
            resting = book.pop_best_bid()

        if resting is None:
            return None

        # Trade quantity = minimum of both remaining quantities
        fill_qty = min(incoming.remaining_qty, resting.remaining_qty)
        # Trade price = resting order's price (maker price wins)
        fill_price = resting.price

        # Update filled quantities
        incoming.filled_qty += fill_qty
        resting.filled_qty += fill_qty

        # If resting order is fully filled, remove from level deque
        if resting.remaining_qty == 0:
            if is_incoming_bid:
                book._ask_levels[resting.price].popleft()
            else:
                book._bid_levels[resting.price].popleft()

        trade = Trade(
            trade_id=str(uuid.uuid4()),
            symbol=incoming.symbol,
            price=fill_price,
            quantity=fill_qty,
            buyer_order_id=incoming.order_id if is_incoming_bid else resting.order_id,
            seller_order_id=resting.order_id if is_incoming_bid else incoming.order_id,
        )
        return trade

    def cancel_order(self, symbol: str, order_id: str) -> bool:
        book = self._get_book(symbol)
        return book.cancel_order(order_id)


# ─── Example Usage ─────────────────────────────────────────────────────────────
if __name__ == "__main__":
    engine = MatchingEngine()

    # Resting sell orders in the book
    ask1 = Order("A1", "AAPL", Side.ASK, OrderType.LIMIT, price=102.00, quantity=100)
    ask2 = Order("A2", "AAPL", Side.ASK, OrderType.LIMIT, price=101.50, quantity=50)
    ask3 = Order("A3", "AAPL", Side.ASK, OrderType.LIMIT, price=101.50, quantity=75)

    engine.process_order(ask1)  # No trades (no bids yet)
    engine.process_order(ask2)
    engine.process_order(ask3)

    # Incoming buy limit order at 102.00 — should sweep asks at 101.50 first
    buy = Order("B1", "AAPL", Side.BID, OrderType.LIMIT, price=102.00, quantity=200)
    trades = engine.process_order(buy)

    for t in trades:
        print(f"TRADE: {t.quantity} @ {t.price} | buyer={t.buyer_order_id} seller={t.seller_order_id}")
    # Expected:
    # TRADE: 50 @ 101.50 | buyer=B1 seller=A2   (FIFO: A2 first at 101.50)
    # TRADE: 75 @ 101.50 | buyer=B1 seller=A3
    # TRADE: 75 @ 102.00 | buyer=B1 seller=A1   (75 remaining of B1 matches A1)
    # Remaining 0 qty from B1 — fully filled!

# CRITICAL: Never use float for prices in production!
# Wrong ❌
price = 101.50  # float precision issues

# Right ✅ — store as integer cents or fixed-point
price_cents = 10150  # $101.50 in cents

# Or use Python's Decimal
from decimal import Decimal
price = Decimal("101.50")  # Exact representation

核心要点 / Key Takeaways:

1. 数据结构: heap (价格排序) + deque (FIFO within level) + hashmap (O(1) cancel)
2. 撮合规则: price-time priority, resting price wins, lazy deletion for cancel
3. 持久化: event sourcing — 订单簿是事件日志的物化视图
4. 性能: 单线程 + 无锁 > 多线程 + 锁 (LMAX 证明)
5. 踩坑: 永远用整数/Decimal表示价格，不要用float

Algorithm complexity:
  - Add limit order:  O(log P) where P = distinct price levels
  - Match:           O(k log P) where k = number of fills
  - Cancel:          O(1) lazy deletion

面试一句话: "The algorithm is two priority queues. The engineering is making it fast, fair, and crash-safe."