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

```python
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)