Exchange Websocket Handling

---




name: exchange-websocket-handling
compatibility: opencode
completeness: 95
content-types:
- code
- guidance
- config
- do-dont
description: '"Real-time market data handling with WebSockets including connection
  management" data aggregation, and robust error recovery'
license: MIT
maturity: stable
metadata:
  domain: trading
  output-format: code
  related-skills: ai-order-flow-analysis, data-alternative-data
  role: implementation
  scope: implementation
  triggers: exchange websocket handling, exchange-websocket-handling, market, real-time,
    websockets
  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:** Guide an AI coding assistant to build robust real-time data pipelines using WebSockets that handle connection instability, data synchronization, and high-frequency updates

**Philosophy:** Real-time data is the lifeblood of modern trading systems, but WebSocket connections are inherently unreliable. Systems must treat WebSocket data as potentially out-of-order, duplicate, or missing. The architecture should prioritize data integrity over raw speed, with robust state synchronization mechanisms that can recover from connection drops without losing critical information.

## Key Principles

1. **Data Integrity Over Freshness**: It's better to have stale but correct data than fresh but incorrect data. Systems should validate and order WebSocket messages before processing.

2. **State Synchronization Protocol**: When connections drop, systems must be able to reconstruct state by requesting snapshots and diffing with WebSocket updates.

3. **Connection Health Monitoring**: WebSocket connections need continuous health monitoring with automatic reconnection, backpressure handling, and graceful degradation.

4. **Message Ordering and Deduplication**: WebSocket streams can deliver messages out of order or duplicated. Systems must handle this gracefully.

5. **Graceful Degradation**: When WebSocket is unavailable, systems should fall back to REST API polling with appropriate lag handling.

## Implementation Guidelines

### Structure
- Core logic: `exchange_integration/websocket.py`
- Data aggregation: `exchange_integration/aggregator.py`
- State management: `exchange_integration/state_sync.py`
- Health monitoring: `exchange_integration/health.py`

### Patterns to Follow
- **Early Exit**: Reject data when synchronization state is invalid
- **Atomic Predictability**: WebSocket messages are processed in a deterministic order
- **Fail Fast**: Halt when WebSocket state cannot be synchronized
- **Intentional Naming**: Clear names for WebSocket operations and states
- **Parse Don't Validate**: WebSocket data parsed at boundaries, validated internally

## Code Examples

```python
# Example 1: WebSocket Connection Manager
import asyncio
import json
import time
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
from typing import Optional
import websockets
import logging


class WebSocketState(Enum):
    DISCONNECTED = "disconnected"
    CONNECTING = "connecting"
    CONNECTED = "connected"
    RECONNECTING = "reconnecting"
    SYNCHRONIZING = "synchronizing"
    FAILED = "failed"


@dataclass
class WebSocketHealth:
    """WebSocket connection health metrics"""
    last_message: datetime = None
    messages_per_second: float = 0.0
    last_heartbeat: datetime = None
    ping_latency: float = 0.0
    subscription_count: int = 0
    sync_lag: timedelta = timedelta(seconds=0)
    
    def is_healthy(self, max_lag_seconds: float = 5.0) -> bool:
        """Check if WebSocket is operating normally"""
        if self.last_message is None:
            return False
        
        time_since_message = (datetime.now() - self.last_message).total_seconds()
        return time_since_message < max_lag_seconds


class WebSocketConnection:
    """Manages a single WebSocket connection with comprehensive error handling"""
    
    def __init__(self, url: str, config: dict = None):
        self.url = url
        self.config = config or {}
        
        self.state = WebSocketState.DISCONNECTED
        self.websocket = None
        self.health = WebSocketHealth()
        
        self.message_handlers = {}
        self.reconnect_delay = 1.0
        self.max_reconnect_delay = 60.0
        
        self.message_queue = asyncio.Queue()
        self.running = False
        
        self.logger = logging.getLogger(f"websocket.{url}")
    
    async def connect(self):
        """Establish WebSocket connection with error handling"""
        self.state = WebSocketState.CONNECTING
        
        try:
            self.websocket = await websockets.connect(
                self.url,
                ping_interval=self.config.get('ping_interval', 20),
                ping_timeout=self.config.get('ping_timeout', 10),
                max_size=self.config.get('max_message_size', 10_000_000)
            )
            
            self.state = WebSocketState.CONNECTED
            self.health.last_heartbeat = datetime.now()
            self.logger.info(f"WebSocket connected: {self.url}")
            
            # Start message processing loop
            asyncio.create_task(self._process_messages())
            
            # Send initial subscription
            await self._send_initial_subscription()
            
        except Exception as error:
            self.logger.error(f"WebSocket connection failed: {error}")
            self.state = WebSocketState.FAILED
            raise
    
    async def _process_messages(self):
        """Process incoming WebSocket messages"""
        while self.running and self.state == WebSocketState.CONNECTED:
            try:
                message = await asyncio.wait_for(
                    self.websocket.recv(),
                    timeout=self.config.get('message_timeout', 30)
                )
                
                # Update health metrics
                self.health.last_message = datetime.now()
                self.health.messages_per_second += 1
                
                # Parse and route message
                try:
                    data = json.loads(message)
                except json.JSONDecodeError:
                    self.logger.warning(f"Invalid JSON message: {message}")
                    continue
                
                await self._route_message(data)
                
            except asyncio.TimeoutError:
                self.logger.warning("WebSocket timeout, reconnecting")
                await self._reconnect()
                
            except websockets.exceptions.ConnectionClosed:
                self.logger.warning("WebSocket connection closed")
                await self._reconnect()
                
            except Exception as error:
                self.logger.error(f"Error processing message: {error}")
    
    async def _route_message(self, data: dict):
        """Route message to appropriate handler"""
        # Check for ping/pong
        if data.get('type') == 'ping':
            await self._send_pong(data.get('id'))
            return
        
        if data.get('type') == 'pong':
            self.health.last_heartbeat = datetime.now()
            return
        
        # Route to channel-specific handler
        channel = data.get('channel')
        if channel and channel in self.message_handlers:
            await self.message_handlers[channel](data)
        else:
            await self._default_handler(data)
    
    async def _default_handler(self, data: dict):
        """Default message handler"""
        # Implement default processing
        self.logger.debug(f"Received: {data}")
    
    async def _reconnect(self):
        """Handle reconnection with exponential backoff"""
        self.state = WebSocketState.RECONNECTING
        
        # Calculate backoff delay
        delay = self.reconnect_delay
        self.reconnect_delay = min(self.reconnect_delay * 2, self.max_reconnect_delay)
        
        self.logger.info(f"Reconnecting in {delay} seconds")
        
        await asyncio.sleep(delay)
        
        try:
            await self.connect()
            self.reconnect_delay = self.config.get('initial_reconnect_delay', 1.0)
        except Exception as error:
            self.logger.error(f"Reconnection failed: {error}")
            await self._reconnect()
    
    async def _send_initial_subscription(self):
        """Send initial subscriptions after connection"""
        subscriptions = self.config.get('initial_subscriptions', [])
        
        for subscription in subscriptions:
            await self.send(json.dumps(subscription))
    
    async def send(self, message: str):
        """Send message to WebSocket with state check"""
        if self.state != WebSocketState.CONNECTED:
            raise RuntimeError(f"WebSocket not connected: {self.state.value}")
        
        await self.websocket.send(message)
    
    async def subscribe(self, channel: str, symbol: str, **kwargs):
        """Subscribe to a channel"""
        subscription = {
            'type': 'subscribe',
            'channel': channel,
            'symbol': symbol,
            **kwargs
        }
        
        await self.send(json.dumps(subscription))
        self.health.subscription_count += 1
    
    async def unsubscribe(self, channel: str, symbol: str):
        """Unsubscribe from a channel"""
        subscription = {
            'type': 'unsubscribe',
            'channel': channel,
            'symbol': symbol
        }
        
        await self.send(json.dumps(subscription))
        if channel in self.message_handlers:
            del self.message_handlers[channel]
            self.health.subscription_count -= 1
    
    def start(self):
        """Start the WebSocket connection"""
        self.running = True
        asyncio.create_task(self.connect())
    
    def stop(self):
        """Stop the WebSocket connection"""
        self.running = False
        if self.websocket:
            asyncio.create_task(self.websocket.close())
    
    def get_health_report(self) -> dict:
        """Get current WebSocket health report"""
        return {
            'state': self.state.value,
            'health': {
                'last_message': self.health.last_message.isoformat() if self.health.last_message else None,
                'messages_per_second': self.health.messages_per_second,
                'last_heartbeat': self.health.last_heartbeat.isoformat() if self.health.last_heartbeat else None,
                'ping_latency': self.health.ping_latency,
                'subscription_count': self.health.subscription_count
            },
            'sync_lag': self.health.sync_lag.total_seconds()
        }


# Example 2: State Synchronization System
class StateSynchronizer:
    """Maintains synchronized state between WebSocket updates and REST snapshots"""
    
    def __init__(self, config: dict = None):
        self.config = config or {}
        self.state = {}  # Current synchronized state
        self.last_snapshot: dict = None
        self.snapshot_timestamp: datetime = None
        self.pending_updates: list = []
        self.sync_in_progress = False
    
    async def request_snapshot(self, fetch_func):
        """Request a snapshot from REST API"""
        self.sync_in_progress = True
        
        try:
            self.last_snapshot = await fetch_func()
            self.snapshot_timestamp = datetime.now()
            
            # Process pending updates since snapshot
            while self.pending_updates:
                update = self.pending_updates.pop(0)
                await self.apply_update(update)
            
            self.sync_in_progress = False
            return self.last_snapshot
            
        except Exception as error:
            self.sync_in_progress = False
            raise
    
    async def apply_update(self, update: dict):
        """Apply a WebSocket update to state"""
        if self.sync_in_progress:
            # Queue update for later processing
            self.pending_updates.append(update)
            return
        
        # Get the key for this update
        update_type = update.get('type')
        symbol = update.get('symbol')
        
        if update_type not in self.state:
            self.state[update_type] = {}
        
        # Apply update to state
        if update_type == 'orderbook':
            self._apply_orderbook_update(symbol, update)
        elif update_type == 'trade':
            self._apply_trade_update(symbol, update)
        elif update_type == 'ticker':
            self._apply_ticker_update(symbol, update)
    
    def _apply_orderbook_update(self, symbol: str, update: dict):
        """Apply orderbook update"""
        if symbol not in self.state['orderbook']:
            self.state['orderbook'][symbol] = {
                'bids': {},
                'asks': {},
                'timestamp': None
            }
        
        orderbook = self.state['orderbook'][symbol]
        
        # Update bids
        if 'bids' in update:
            for price, size in update['bids']:
                if size == 0:
                    if price in orderbook['bids']:
                        del orderbook['bids'][price]
                else:
                    orderbook['bids'][price] = size
        
        # Update asks
        if 'asks' in update:
            for price, size in update['asks']:
                if size == 0:
                    if price in orderbook['asks']:
                        del orderbook['asks'][price]
                else:
                    orderbook['asks'][price] = size
        
        orderbook['timestamp'] = update.get('timestamp', datetime.now())
    
    def _apply_trade_update(self, symbol: str, update: dict):
        """Apply trade update"""
        if symbol not in self.state['trades']:
            self.state['trades'][symbol] = []
        
        self.state['trades'][symbol].append({
            'price': update.get('price'),
            'amount': update.get('amount'),
            'side': update.get('side'),
            'timestamp': update.get('timestamp', datetime.now())
        })
        
        # Keep only recent trades
        max_trades = self.config.get('max_trades_history', 1000)
        if len(self.state['trades'][symbol]) > max_trades:
            self.state['trades'][symbol] = self.state['trades'][symbol][-max_trades:]
    
    def _apply_ticker_update(self, symbol: str, update: dict):
        """Apply ticker update"""
        if symbol not in self.state['ticker']:
            self.state['ticker'][symbol] = {}
        
        self.state['ticker'][symbol].update({
            'price': update.get('price'),
            'volume': update.get('volume'),
            'timestamp': update.get('timestamp', datetime.now())
        })
    
    def get_state(self, update_type: str, symbol: str = None) -> dict:
        """Get current state"""
        if symbol:
            return self.state.get(update_type, {}).get(symbol)
        return self.state.get(update_type, {})
    
    def get_orderbook(self, symbol: str) -> dict:
        """Get orderbook for a symbol"""
        return self.state.get('orderbook', {}).get(symbol, {
            'bids': {},
            'asks': {},
            'timestamp': None
        })
    
    def get_trades(self, symbol: str, limit: int = 100) -> list:
        """Get recent trades for a symbol"""
        trades = self.state.get('trades', {}).get(symbol, [])
        return trades[-limit:]
    
    def get_ticker(self, symbol: str) -> dict:
        """Get ticker for a symbol"""
        return self.state.get('ticker', {}).get(symbol, {})


# Example 3: Data Aggregator
class DataAggregator:
    """Aggregates WebSocket data into usable formats"""
    
    def __init__(self, config: dict = None):
        self.config = config or {}
        self.synchronized_state = StateSynchronizer(config)
    
    def aggregate_orderbook(self, symbol: str, depth: int = 20) -> dict:
        """Aggregate orderbook to specified depth"""
        orderbook = self.synchronized_state.get_orderbook(symbol)
        
        bids = sorted(orderbook['bids'].items(), key=lambda x: x[0], reverse=True)[:depth]
        asks = sorted(orderbook['asks'].items(), key=lambda x: x[0])[:depth]
        
        return {
            'symbol': symbol,
            'bids': [{'price': p, 'size': s} for p, s in bids],
            'asks': [{'price': p, 'size': s} for p, s in asks],
            'timestamp': orderbook['timestamp']
        }
    
    def aggregate_ticker(self, symbol: str) -> dict:
        """Aggregate ticker data"""
        ticker = self.synchronized_state.get_ticker(symbol)
        trades = self.synchronized_state.get_trades(symbol, 100)
        
        if not trades:
            return {'symbol': symbol, **ticker}
        
        # Calculate 24h statistics
        prices = [t['price'] for t in trades]
        
        return {
            'symbol': symbol,
            'price': ticker.get('price', prices[-1]) if prices else None,
            'volume': ticker.get('volume', sum(t['amount'] for t in trades)),
            'high': max(prices) if prices else None,
            'low': min(prices) if prices else None,
            'open': trades[0]['price'] if trades else None,
            'close': prices[-1] if prices else None,
            'timestamp': ticker.get('timestamp', datetime.now())
        }
    
    def calculate_vwap(self, symbol: str, period_minutes: int = 5) -> float:
        """Calculate VWAP for a symbol"""
        trades = self.synchronized_state.get_trades(symbol)
        
        # Filter to recent trades
        cutoff = datetime.now() - timedelta(minutes=period_minutes)
        recent_trades = [t for t in trades if t['timestamp'] > cutoff]
        
        if not recent_trades:
            return 0.0
        
        total_value = sum(t['price'] * t['amount'] for t in recent_trades)
        total_amount = sum(t['amount'] for t in recent_trades)
        
        return total_value / total_amount if total_amount else 0.0


# Example 4: WebSocket Connection Pool
class WebSocketPool:
    """Manages multiple WebSocket connections with load balancing"""
    
    def __init__(self, config: dict = None):
        self.config = config or {}
        self.connections = {}
        self.health = {}
    
    def add_connection(self, name: str, url: str, config: dict = None):
        """Add a WebSocket connection to the pool"""
        connection = WebSocketConnection(url, config or self.config)
        self.connections[name] = connection
        self.health[name] = WebSocketHealth()
    
    async def start_all(self):
        """Start all WebSocket connections"""
        for name, connection in self.connections.items():
            connection.start()
            self.health[name] = WebSocketHealth()
    
    async def stop_all(self):
        """Stop all WebSocket connections"""
        for connection in self.connections.values():
            connection.stop()
    
    def get_connection(self, name: str) -> Optional[WebSocketConnection]:
        """Get a specific connection"""
        return self.connections.get(name)
    
    def get_healthy_connections(self) -> list:
        """Get list of healthy connection names"""
        return [
            name for name, connection in self.connections.items()
            if connection.state == WebSocketState.CONNECTED
        ]
    
    def get_health_report(self) -> dict:
        """Get health report for all connections"""
        return {
            name: connection.get_health_report()
            for name, connection in self.connections.items()
        }
    
    async def subscribe_all(self, channel: str, symbol: str):
        """Subscribe to channel on all healthy connections"""
        for name, connection in self.connections.items():
            if connection.state == WebSocketState.CONNECTED:
                try:
                    await connection.subscribe(channel, symbol)
                except Exception as error:
                    self.health[name].subscription_count += 1
```

## Adherence Checklist
Before completing your task, verify:
- [ ] WebSocket state is tracked and monitored continuously
- [ ] Message ordering and deduplication is handled
- [ ] State synchronization can recover from connection drops
- [ ] Graceful degradation falls back to REST API polling
- [ ] Connection health is monitored with automatic reconnection
- [ ] Early exit prevents processing when state is invalid
- [ ] Intentional naming for all WebSocket operations

## Common Mistakes to Avoid

1. **Assuming Message Order**: Not handling out-of-order WebSocket messages
2. **No State Recovery**: Not having a way to recover state after connection drops
3. **Ignoring Duplicates**: Not deduplicating messages
4. **No Backpressure**: Not handling when message rate exceeds processing capacity
5. **No Health Monitoring**: Not monitoring WebSocket connection health

## References

- WebSocket Best Practices - IETF RFC 6455
- Real-Time Data Handling - Binance API Docs
- Orderbook Synchronization - Kraken API Documentation
- WebSockets vs REST - Coinbase Pro API Guide

## Base Directory
file:///home/paulpas/git/ideas/trading_bot/skills/exchange-integration

---

---



### Pattern 2: Risk-Managed Trading Logic with Validation

```python
from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Optional


logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class TradeSignal:
    """Immutable trade signal with all required validation constraints."""
    symbol: str
    side: str  # "buy" or "sell"
    price: float
    quantity: float
    confidence: float  # 0.0 to 1.0
    reason: str

    def validate(self) -> bool:
        """Validate that the trade signal meets all business constraints."""
        if self.quantity <= 0:
            raise ValueError(f"Quantity must be positive, got {self.quantity}")
        if self.price <= 0:
            raise ValueError(f"Price must be positive, got {self.price}")
        if not 0.0 <= self.confidence <= 1.0:
            raise ValueError(f"Confidence must be between 0 and 1, got {self.confidence}")
        return True


def generate_trade_signal(
    symbol: str,
    side: str,
    price: float,
    quantity: float,
    confidence: float,
    reason: str,
) -> TradeSignal:
    """Generate a validated trade signal with guard clause checks."""
    if side not in ("buy", "sell"):
        raise ValueError(f"Invalid side '{side}', must be 'buy' or 'sell'")

    signal = TradeSignal(
        symbol=symbol,
        side=side,
        price=price,
        quantity=quantity,
        confidence=confidence,
        reason=reason,
    )
    signal.validate()
    logger.info("Trade signal generated: %s %s %.4f @ %.2f (confidence=%.2f)",
                 symbol, side, quantity, price, confidence)
    return signal


def execute_with_risk_check(signal: TradeSignal, max_position_pct: float = 0.05) -> dict:
    """Execute a trade signal after applying risk management checks."""
    adjusted_quantity = signal.quantity
    if signal.side == "buy" and signal.quantity > max_position_pct:
        logger.warning("Position %s exceeds max %.1f%% — capping to %.4f",
                        signal.symbol, max_position_pct * 100, max_position_pct)
        adjusted_quantity = max_position_pct

    return {
        "symbol": signal.symbol,
        "side": signal.side,
        "price": signal.price,
        "quantity": adjusted_quantity,
        "capped": adjusted_quantity < signal.quantity,
        "confidence": signal.confidence,
        "status": "submitted",
    }
```

## 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 WebSocket API](https://docs.binance.org/websockets.html)
- [WebSocket Reconnection Strategies](https://en.wikipedia.org/wiki/WebSocket)
- [Real-Time Market Data with WebSockets](https://docs.quantconnect.com/tutorials/live-trading-overview)
- [WebSocket Message Parsing Best Practices](https://www.investopedia.com/terms/w/websocket.asp)
- [Handling WebSocket Disconnections](https://docs.ccxt.org/en/latest/manual.html#market-data)