paulpas 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
- Connection Management: Auto-reconnect, exponential backoff, heartbeat monitoring
- Data Validation: Validate incoming messages before processing
- Rate Limiting: Prevent API throttling and connection drops
- Buffer Management: Handle message bursts efficiently
- 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
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
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.