Low-Latency WebSocket Aggregator for Market Data

Trading bots, market makers, and risk systems depend on fresh market data—a delay of milliseconds can cost profits. In one project, a client lost tens of thousands of dollars in a month due to a stale connection: data stopped flowing, but the system continued trading based on outdated prices. Each exchange uses its own protocol, connection limits, and message format. A low-latency crypto data aggregator solves this.

According to the WebSocket API documentation, Binance allows up to 1024 streams per connection, while Bybit only 10 topics. Building a universal aggregator that connects to all exchanges, normalizes streams, and delivers data with minimal latency is a non-trivial task. Even an error in stream processing can lead to arbitrage losses or incorrect order execution. We built a modular aggregator that addresses these issues. Our solutions have been proven in projects handling 100,000 messages per second on a single core using Python asyncio—an order of magnitude faster than standard multithreaded approaches.

We have delivered over 50 projects in crypto infrastructure. Below we break down the key components and architecture.

How the Aggregator Handles Different Exchange Limits

The Connection Manager automatically distributes subscriptions, respecting each exchange’s limits. For each exchange we configure a manager that creates new connections when the limit is exhausted.

class ConnectionManager: def __init__(self, max_per_conn: int = 900): self.connections: list[WSConnection] = [] self.max_per_conn = max_per_conn self.subscriptions: dict[str, WSConnection] = {} async def subscribe(self, channels: list[str]): for channel in channels: conn = self._find_or_create_connection() await conn.subscribe(channel) self.subscriptions[channel] = conn def _find_or_create_connection(self) -> WSConnection: for conn in self.connections: if conn.subscription_count < self.max_per_conn: return conn new_conn = WSConnection(self.on_message, self.on_disconnect) self.connections.append(new_conn) return new_conn async def on_disconnect(self, conn: WSConnection): # Exponential backoff and resubscribe await asyncio.sleep(conn.backoff.next()) await conn.reconnect() await conn.resubscribe() 

When the limit is exceeded, the Manager automatically creates an additional connection. For example, for Bybit with a limit of 10 topics per connection, subscribing to 25 channels will result in 3 connections. Exponential backoff prevents exchange overload during mass disconnects.

Why Heartbeat and Stale Detection Matter

Exchanges may go silent without a TCP disconnect—the connection is alive but no data arrives. A watchdog timer for each connection solves this. If no message arrives for more than 30 seconds, the connection is forcibly recreated. Heartbeat monitoring and stale detection are key elements of a robust aggregator.

class HeartbeatMonitor: STALE_THRESHOLD_SEC = 30 async def watch(self, conn: WSConnection): while True: await asyncio.sleep(5) age = time.time() - conn.last_message_time if age > self.STALE_THRESHOLD_SEC: logger.warning(f"Stale connection detected, forcing reconnect") await conn.force_reconnect() 

In the project I mentioned, the lack of such a monitor led to losses. After deploying the aggregator with the Heartbeat monitor, incidents stopped, and the savings on missed profit reached about 40%.

Publishing Data to Consumers

The aggregator publishes normalized data through multiple channels. The choice depends on reliability and latency requirements.

Redis Pub/Sub offers minimal latency but no delivery guarantee. Redis Streams and Kafka are suitable for reliable delivery with ability to replay missed messages. gRPC streaming is for direct low‑latency connections.

Performance Metrics

The aggregator exports Prometheus metrics:

• ws_messages_received_total{exchange, channel}
• ws_message_latency_ms{exchange}
• ws_reconnects_total{exchange}
• ws_active_connections{exchange}
• ws_subscription_count{exchange}

These metrics allow quick identification of connection issues and overloads. With a proper implementation in Python (asyncio), the aggregator processes 50,000–100,000 messages per second on a single core. Go or Rust can handle an order of magnitude more.

Process and Workflow

• Analysis – we study the list of exchanges, data types (order book, trades, tickers), and latency requirements.
• Design – we choose the stack (Python/Go, Redis/Kafka) and design the normalization schema.
• Implementation – we build the Connection Manager, Heartbeat, and publication modules.
• Testing – we simulate disconnects, run load tests, and verify recovery.
• Deployment – we deploy in your infrastructure (k8s, bare metal) and configure monitoring.

Typical mistakes in DIY implementations include ignoring exchange limits—exceeding max streams causes disconnection; lack of heartbeat—stale connections lead to trading on outdated data; synchronous processing—blocking calls kill performance; absence of metrics—impossible to assess system health. Our aggregator solves each of these problems.

Timelines and Cost

A basic aggregator for one exchange takes 2–4 weeks. Adding an additional exchange takes 1–2 weeks. A full solution with Kafka and dashboards starts from 2 months. Cost is calculated individually. Infrastructure savings compared to purchased solutions can reach 40%. Contact us for an assessment of your project—we will prepare a proposal within 1–2 days. Get a consultation on your data pipeline architecture. Order development of a WebSocket aggregator for your trading system.