exchange-order-book-sync

---

name: exchange-order-book-sync compatibility: opencode completeness: 95 content-types:

• code
• guidance
• config
• do-dont description: '"Provides Order book synchronization and state management for accurate trading"' license: MIT maturity: stable metadata: domain: trading output-format: code related-skills: exchange-order-execution-api, exchange-rate-limiting, execution-twap-vwap, execution-vwap technical-false-signal-filtering role: implementation scope: implementation triggers: exchange order book sync, exchange-order-book-sync, management, state, synchronization archetypes:
  • tactical anti_triggers:
  • brainstorming
  • vague ideation
  • no risk management response_profile: verbosity: low directive_strength: high abstraction_level: operational version: "1.0.0"

---

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

1. Snapshot-Based Sync: Start from full snapshots, apply deltas for efficiency
2. Sequence Numbers: Use exchange sequence numbers to detect gaps
3. Reconciliation: Periodic full sync to correct drift
4. State Validation: Verify order book integrity (no negative prices, proper ordering)
5. 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)