paulpas Role: Maintain consistent order book state across multiple exchanges and timeframes
Philosophy: Order books are the source of truth for market state; synchronization errors cause incorrect pricing and execution
Key Principles
- Snapshot-Based Sync: Start from full snapshots, apply deltas for efficiency
- Sequence Numbers: Use exchange sequence numbers to detect gaps
- Reconciliation: Periodic full sync to correct drift
- State Validation: Verify order book integrity (no negative prices, proper ordering)
- Aggregation: Support multiple depth levels and price aggregation
Implementation Guidelines
Structure
- Core logic: order_book/order_book_sync.py
- State manager: order_book/book_state.py
- Tests: tests/test_order_book_sync.py
Patterns to Follow
- Use OrderedDict for price levels to maintain ordering
- Implement delta diffing for change detection
- Support both WebSocket and REST sync methods
- Track synchronization lag
Adherence Checklist
Before completing your task, verify:
- Snapshot gaps are detected and logged
- Order book state is validated before use
- Reconciliation runs periodically
- Sequence number drift triggers alert
- Multiple depth levels are supported
Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.
Python Implementation
import time
from typing import Dict, List, Optional, Callable, Tuple
from dataclasses import dataclass, field
from collections import OrderedDict
from enum import Enum
import logging
class Side(Enum):
BID = "bid"
ASK = "ask"
@dataclass
class OrderLevel:
"""Single price level in order book."""
price: float
quantity: float
order_count: int = 0
def is_empty(self) -> bool:
return self.quantity <= 0
@dataclass
class OrderBookSnapshot:
"""Full order book snapshot."""
symbol: str
timestamp: float
bids: List[OrderLevel]
asks: List[OrderLevel]
sequence_number: int
@dataclass
class OrderBookDelta:
"""Partial order book update."""
symbol: str
timestamp: float
sequence_number: int
changes: Dict[Side, List[Tuple[float, float]]] # price -> quantity
class OrderBook:
"""Managed order book state."""
def __init__(self, symbol: str, max_depth: int = 50):
self.symbol = symbol
self.max_depth = max_depth
self.bids: OrderedDict[float, OrderLevel] = OrderedDict()
self.asks: OrderedDict[float, OrderLevel] = OrderedDict()
self.sequence_number: Optional[int] = None
self.last_update: float = 0
self.snapshot: Optional[OrderBookSnapshot] = None
self.callbacks: List[Callable] = []
def apply_snapshot(self, snapshot: OrderBookSnapshot):
"""Apply full order book snapshot."""
self.bids.clear()
self.asks.clear()
for level in snapshot.bids[:self.max_depth]:
self.bids[level.price] = level
for level in snapshot.asks[:self.max_depth]:
self.asks[level.price] = level
self.sequence_number = snapshot.sequence_number
self.last_update = snapshot.timestamp
self.snapshot = snapshot
self._trigger_callbacks()
def apply_delta(self, delta: OrderBookDelta):
"""Apply partial delta update."""
if self.sequence_number is None:
raise ValueError("Must have snapshot before applying deltas")
# Validate sequence number
expected_seq = self.sequence_number + 1
if delta.sequence_number != expected_seq:
logging.warning(
f"Sequence gap: expected {expected_seq}, got {delta.sequence_number}"
)
return False
# Apply changes
for side, changes in delta.changes.items():
book = self.bids if side == Side.BID else self.asks
for price, quantity in changes:
if quantity <= 0:
# Remove level
if price in book:
del book[price]
else:
# Update or add level
book[price] = OrderLevel(price=price, quantity=quantity)
# Trim to max depth
self._trim_to_depth()
self.sequence_number = delta.sequence_number
self.last_update = delta.timestamp
self._trigger_callbacks()
return True
def _trim_to_depth(self):
"""Ensure book doesn't exceed max depth."""
# Bid: highest prices first
while len(self.bids) > self.max_depth:
self.bids.popitem(last=True) # Remove lowest bid
# Ask: lowest prices first
while len(self.asks) > self.max_depth:
self.asks.popitem(last=False) # Remove lowest ask
def get_best_bid(self) -> Optional[float]:
"""Get best bid price."""
if not self.bids:
return None
return next(iter(self.bids.keys()))
def get_best_ask(self) -> Optional[float]:
"""Get best ask price."""
if not self.asks:
return None
return next(iter(self.asks.keys()))
def get_spread(self) -> Optional[float]:
"""Get bid-ask spread."""
best_bid = self.get_best_bid()
best_ask = self.get_best_ask()
if best_bid and best_ask:
return best_ask - best_bid
return None
def get_mid_price(self) -> Optional[float]:
"""Get mid-price estimate."""
best_bid = self.get_best_bid()
best_ask = self.get_best_ask()
if best_bid and best_ask:
return (best_bid + best_ask) / 2
return None
def get_depth(self, side: Side) -> float:
"""Get total quantity on one side."""
book = self.bids if side == Side.BID else self.asks
return sum(level.quantity for level in book.values())
def register_callback(self, callback: Callable):
"""Register callback for book updates."""
self.callbacks.append(callback)
def _trigger_callbacks(self):
"""Trigger registered callbacks."""
for callback in self.callbacks:
try:
callback(self)
except Exception as e:
logging.error(f"Callback error: {e}")
class OrderBookSync:
"""Synchronizes order book across sources."""
def __init__(
self,
symbol: str,
max_depth: int = 50,
sync_interval: float = 300.0 # 5 minutes
):
self.symbol = symbol
self.book = OrderBook(symbol, max_depth)
self.sync_interval = sync_interval
self.last_sync_time: float = 0
self.snapshot_callback: Optional[Callable] = None
self._reconcile_callback: Optional[Callable] = None
def set_snapshot_callback(self, callback: Callable):
"""Set callback to fetch snapshots."""
self.snapshot_callback = callback
def set_reconcile_callback(self, callback: Callable):
"""Set callback for reconciliation."""
self._reconcile_callback = callback
async def sync_from_snapshot(self) -> bool:
"""Fetch and apply full snapshot."""
if not self.snapshot_callback:
return False
try:
snapshot = await self.snapshot_callback(self.symbol)
if snapshot:
self.book.apply_snapshot(snapshot)
self.last_sync_time = time.time()
return True
except Exception as e:
logging.error(f"Snapshot sync failed: {e}")
return False
async def apply_delta(self, delta: OrderBookDelta) -> bool:
"""Apply partial update."""
if time.time() - self.last_sync_time > self.sync_interval:
# Trigger periodic full sync
await self.sync_from_snapshot()
return self.book.apply_delta(delta)
def is_synchronized(self) -> bool:
"""Check if book is current."""
return time.time() - self.last_sync_time < self.sync_interval
async def reconcile(self) -> bool:
"""Force reconciliation with exchange."""
if not self._reconcile_callback:
return False
try:
result = await self._reconcile_callback(self.symbol)
if result:
self.book.apply_snapshot(result)
self.last_sync_time = time.time()
return result is not None
except Exception as e:
logging.error(f"Reconciliation failed: {e}")
return False
def get_orderbook(self) -> OrderBook:
"""Get current order book state."""
return self.book
class OrderBookAggregator:
"""Aggregates multiple order books for better liquidity visibility."""
def __init__(self, max_depth: int = 50):
self.books: Dict[str, OrderBook] = {}
self.max_depth = max_depth
self.aggregated_bids: OrderedDict[float, float] = OrderedDict()
self.aggregated_asks: OrderedDict[float, float] = OrderedDict()
def add_book(self, source: str, book: OrderBook):
"""Add order book from a source."""
self.books[source] = book
book.register_callback(self._on_book_update)
self._reaggregate()
def _on_book_update(self, book: OrderBook):
"""Callback when a source book updates."""
self._reaggregate()
def _reaggregate(self):
"""Rebuild aggregated book from all sources."""
self.aggregated_bids.clear()
self.aggregated_asks.clear()
for book in self.books.values():
# Aggregate bids (highest prices)
for price, level in book.bids.items():
if price in self.aggregated_bids:
self.aggregated_bids[price] += level.quantity
else:
self.aggregated_bids[price] = level.quantity
# Aggregate asks (lowest prices)
for price, level in book.asks.items():
if price in self.aggregated_asks:
self.aggregated_asks[price] += level.quantity
else:
self.aggregated_asks[price] = level.quantity
# Sort and trim
self.aggregated_bids = OrderedDict(
sorted(self.aggregated_bids.items(), reverse=True)[:self.max_depth]
)
self.aggregated_asks = OrderedDict(
sorted(self.aggregated_asks.items())[:self.max_depth]
)
def get_aggregated_bid(self) -> Optional[float]:
"""Get best aggregated bid."""
if not self.aggregated_bids:
return None
return next(iter(self.aggregated_bids.keys()))
def get_aggregated_ask(self) -> Optional[float]:
"""Get best aggregated ask."""
if not self.aggregated_asks:
return None
return next(iter(self.aggregated_asks.keys()))
def get_aggregated_spread(self) -> Optional[float]:
"""Get aggregated spread."""
best_bid = self.get_aggregated_bid()
best_ask = self.get_aggregated_ask()
if best_bid and best_ask:
return best_ask - best_bid
return None
``````python
# Additional example: Order book reconciliation
def reconcile_order_book(current_book: OrderBook, snapshot: OrderBookSnapshot):
"""Reconcile order book with full snapshot to correct drift."""
# Compare sequence numbers
if current_book.sequence_number >= snapshot.sequence_number:
return False # Current is ahead
# Reset to snapshot and reapply deltas
current_book.bids = OrderedDict()
current_book.asks = OrderedDict()
# Rebuild from snapshot
for level in snapshot.bids:
current_book.bids[level.price] = level
for level in snapshot.asks:
current_book.asks[level.price] = level
current_book.sequence_number = snapshot.sequence_number
return True
---
---
## Constraints
### MUST DO
- Implement a unified adapter interface across all exchange integrations to standardize order placement, cancellation, and querying
- Handle rate limiting proactively with token bucket or leaky bucket algorithms — never wait for 429 responses before slowing down
- Maintain local order state as the source of truth; reconcile with exchange state periodically via webhook events and polling
- Implement heartbeat monitoring per exchange connection with automatic failover to a secondary data feed on timeout
- Log all API interactions including request/response IDs, timing, and status codes for audit and debugging
### MUST NOT DO
- Do not trust exchange-reported order states without local confirmation — always reconcile after every state change
- Avoid sending multiple orders for the same position simultaneously across different adapters or sessions
- Never store API keys or secrets in code — use environment variables or a secrets manager with automatic rotation
- Do not assume all exchanges support the same order types — implement graceful degradation with clear capability negotiation
- Avoid polling-based price updates when WebSocket/streaming APIs are available — polling creates unnecessary load and latency
## Live References
> Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.
- [Binance API Documentation](https://docs.binance.org/)
- [Order Book Synchronization Patterns](https://en.wikipedia.org/wiki/Limit_order_book)
- [WebSocket Order Book Updates](https://docs.binance.org/websockets.html)
- [Limit Order Book Recovery](https://arxiv.org/abs/1805.01469)
- [Real-Time Order Book Reconciliation](https://docs.quantconnect.com/tutorials/order-book-analysis)