# Exchange Websocket Streaming > "Implements real-time market data streaming and processing for risk management and algorithmic trading execution." - Skill: `paulpas/exchange-websocket-streaming` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/exchange-websocket-streaming` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/exchange-websocket-streaming/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/exchange-websocket-streaming --- **Role:** Handle real-time WebSocket connections for low-latency market data **Philosophy:** Real-time data is the foundation of competitive advantage; streaming systems must be robust and efficient ## Key Principles 1. **Connection Management**: Auto-reconnect, exponential backoff, heartbeat monitoring 2. **Data Validation**: Validate incoming messages before processing 3. **Rate Limiting**: Prevent API throttling and connection drops 4. **Buffer Management**: Handle message bursts efficiently 5. **Message Parsing**: Fast, schema-based parsing of market data ## Implementation Guidelines ### Structure - Core logic: exchange_integration/websocket_client.py - Helper functions: exchange_integration/data_handlers.py - Tests: tests/test_websocket.py ### Patterns to Follow - Use asyncio for non-blocking operations - Implement retry logic with exponential backoff - Separate connection management from data processing ## Adherence Checklist Before completing your task, verify: - [ ] WebSocket connections auto-reconnect on failure - [ ] Heartbeat monitoring detects stale connections - [ ] Rate limiting prevents API throttling - [ ] Message validation rejects malformed data - [ ] Buffer overflow triggers protective measures Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory. ## Python Implementation ```python import asyncio import json import websockets import numpy as np from typing import Dict, List, Optional, Callable, Any from dataclasses import dataclass from datetime import datetime import time import logging @dataclass class MarketData: """Real-time market data point.""" symbol: str timestamp: float price: float size: float side: str # 'buy' or 'sell' order_type: str class WebSocketClient: """Handles WebSocket connections for real-time market data.""" def __init__( self, url: str, reconnect_delay: float = 1.0, max_reconnect_delay: float = 30.0 ): self.url = url self.reconnect_delay = reconnect_delay self.max_reconnect_delay = max_reconnect_delay self.ws = None self.connected = False self.message_handlers = {} self.reconnect_task = None self.last_heartbeat = time.time() async def connect(self): """Establish WebSocket connection with auto-reconnect.""" while True: try: async with websockets.connect(self.url) as ws: self.ws = ws self.connected = True self.last_heartbeat = time.time() await self.on_connect() await self.receive_messages() except Exception as e: self.connected = False logging.warning(f"WebSocket connection error: {e}") # Exponential backoff delay = min(self.reconnect_delay * 2, self.max_reconnect_delay) await asyncio.sleep(delay) async def receive_messages(self): """Continuously receive and process messages.""" async for message in self.ws: self.last_heartbeat = time.time() await self.process_message(message) async def process_message(self, message: str): """Process incoming WebSocket message.""" try: data = json.loads(message) msg_type = data.get('type', 'unknown') if msg_type in self.message_handlers: for handler in self.message_handlers[msg_type]: await handler(data) elif msg_type == 'heartbeat': self.last_heartbeat = time.time() except json.JSONDecodeError: logging.error(f"Failed to parse message: {message}") def register_handler(self, message_type: str, handler: Callable): """Register a message handler for a specific message type.""" if message_type not in self.message_handlers: self.message_handlers[message_type] = [] self.message_handlers[message_type].append(handler) async def send_message(self, message: Dict): """Send a message through the WebSocket.""" if self.ws and self.connected: await self.ws.send(json.dumps(message)) async def on_connect(self): """Called when connection is established.""" # Subscribe to market data channels await self.send_message({'type': 'subscribe', 'channels': ['trade', 'book']}) def is_healthy(self, timeout: float = 30.0) -> bool: """Check if connection is healthy based on heartbeat.""" return time.time() - self.last_heartbeat < timeout class RateLimiter: """Rate limiter to prevent API throttling.""" def __init__(self, max_requests: int, time_window: float = 60.0): self.max_requests = max_requests self.time_window = time_window self.requests = [] async def acquire(self): """Wait until a request can be made within rate limits.""" while True: now = time.time() # Remove old requests self.requests = [t for t in self.requests if now - t < self.time_window] if len(self.requests) < self.max_requests: self.requests.append(now) return # Wait until oldest request expires sleep_time = self.time_window - (now - self.requests[0]) await asyncio.sleep(max(sleep_time, 0.01)) class DataBuffer: """Buffer for handling message bursts.""" def __init__(self, max_size: int = 1000): self.max_size = max_size self.buffer = [] self.overflow_count = 0 def add(self, data: MarketData) -> bool: """Add data to buffer. Returns False if buffer is full.""" if len(self.buffer) >= self.max_size: self.overflow_count += 1 return False self.buffer.append(data) return True def get_all(self) -> List[MarketData]: """Get all buffered data and clear buffer.""" data = self.buffer.copy() self.buffer.clear() return data def get_limited(self, max_items: int = 100) -> List[MarketData]: """Get up to max_items from buffer, oldest first.""" data = self.buffer[:max_items] self.buffer = self.buffer[max_items:] return data def is_overflowing(self) -> bool: """Check if buffer has overflowed recently.""" return self.overflow_count > 0 ``` --- --- ### 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. - [Subscribe to Streams](https://docs.binance.org/websockets.html#subscribe-to-streams) - [Binance WebSocket API Reference](https://docs.binance.org/) - [Real-Time Trade Data Streaming](https://docs.quantconnect.com/tutorials/live-trading-overview) - [WebSocket Compression and Efficiency](https://en.wikipedia.org/wiki/WebSocket#Compression_extension) - [Streaming Market Data Architecture](https://kafka.apache.org/documentation/)