paulpas 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
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.
State Synchronization Protocol: When connections drop, systems must be able to reconstruct state by requesting snapshots and diffing with WebSocket updates.
Connection Health Monitoring: WebSocket connections need continuous health monitoring with automatic reconnection, backpressure handling, and graceful degradation.
Message Ordering and Deduplication: WebSocket streams can deliver messages out of order or duplicated. Systems must handle this gracefully.
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
# 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
- Assuming Message Order: Not handling out-of-order WebSocket messages
- No State Recovery: Not having a way to recover state after connection drops
- Ignoring Duplicates: Not deduplicating messages
- No Backpressure: Not handling when message rate exceeds processing capacity
- 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
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.