Every quant engineer reaches the same inflection point. The strategy is solid. The backtest looks promising. Then the live deployment begins—and the system buckles.
The order book updates arrive at 200 messages per second during volatile sessions. The signal calculation takes 15 milliseconds per update. The strategy fires events at 66 messages per second. By the third minute, the processing lag exceeds 30 seconds. By the tenth minute, the queue is 18,000 messages deep, and the strategy is trading on stale data from a market state that no longer exists.
This is not a theoretical failure mode. It is the guaranteed outcome when market data arrival rate exceeds strategy processing capacity—and it is solvable with the producer-consumer pattern.
The Fundamental Mismatch
High-frequency market data systems operate under a structural tension. Data arrives in bursts governed by market microstructure—earnings announcements, macro releases, and liquidity vacuums create sudden spikes in message frequency. Strategy processing operates at a fixed cadence determined by computational complexity and decision latency.
Consider the math. A typical US equity streaming session generates:
| Market State | Messages/Second | Strategy Cycle Time | Capacity Gap |
|---|---|---|---|
| Calm morning | 15–30 | 15 ms | 30× headroom |
| Active trading | 80–120 | 15 ms | 5× headroom |
| Volatility spike | 200–500 | 15 ms | −2× to −10× overflow |
When capacity goes negative, the queue grows unbounded. Without intervention, the system enters a death spiral: processing lag increases, the queue swells, garbage collection triggers, latency spikes cascade, and finally the process crashes or the OS kills it under memory pressure.
The producer-consumer pattern resolves this mismatch by introducing a bounded buffer between the data source and the processing pipeline. The producer (the WebSocket listener) decouples from the consumer (the strategy engine). The queue absorbs burst traffic. But here is the critical insight most implementations miss: a queue without backpressure is not a solution. It is a delay mechanism that postpones the inevitable.
The producer-consumer pattern succeeds only when backpressure control is a first-class design concern, not an afterthought.
Architecture Overview
The system architecture for producer-consumer market data distribution consists of three layers:
┌─────────────────────────────────────────────────────────────┐
│ DATA SOURCE LAYER │
│ WebSocket Feed (TickDB depth channel) │
└─────────────────────────┬───────────────────────────────────┘
│ Raw market data stream
▼
┌─────────────────────────────────────────────────────────────┐
│ BUFFER LAYER (asyncio.Queue) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Bounded queue with maxsize=N │ │
│ │ Backpressure signal when full: put() blocks │ │
│ │ Drop strategy: oldest / lowest-priority / blocking │ │
│ └─────────────────────────────────────────────────────┘ │
└─────────────────────────┬───────────────────────────────────┘
│ Consumed by workers
▼
┌─────────────────────────────────────────────────────────────┐
│ PROCESSING LAYER │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Worker 1 │ │ Worker 2 │ │ Worker N │ │
│ │ (Signal) │ │ (Risk) │ │ (Logger) │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────────┘
This architecture provides three critical properties:
- Decoupling: The WebSocket listener runs at full speed regardless of processing latency.
- Burst absorption: Temporary spikes queue instead of dropping data.
- Backpressure propagation: Upstream sources slow down when the queue is saturated.
The remaining sections implement this architecture with production-grade resilience.
Production-Grade Implementation
The Producer: Resilient WebSocket Listener
The producer's responsibility extends beyond simply reading from the WebSocket. It must handle reconnection, rate limiting, heartbeat detection, and backpressure signaling. This implementation uses TickDB's depth channel as the data source, with proper error handling and reconnection logic.
import asyncio
import json
import logging
import os
import random
import time
from dataclasses import dataclass
from typing import Optional
import aiohttp
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s"
)
logger = logging.getLogger(__name__)
@dataclass
class MarketUpdate:
"""Standardized market data update."""
symbol: str
timestamp: float
bid_price: float
bid_size: int
ask_price: float
ask_size: int
sequence: int
class TickDBWebSocketProducer:
"""
WebSocket producer for TickDB market data feed.
Handles reconnection with exponential backoff + jitter,
rate-limit backpressure, and graceful shutdown.
"""
def __init__(
self,
api_key: str,
symbols: list[str],
output_queue: asyncio.Queue,
max_queue_size: int = 10000,
base_delay: float = 1.0,
max_delay: float = 60.0,
):
self.api_key = api_key
self.symbols = symbols
self.queue = output_queue
self.max_queue_size = max_queue_size
# Exponential backoff state
self.base_delay = base_delay
self.max_delay = max_delay
self.retry_count = 0
self._running = False
self._shutdown_event = asyncio.Event()
# Sequence tracking for gap detection
self._last_sequence: dict[str, int] = {}
async def connect(self) -> aiohttp.ClientWebSocketResponse:
"""Establish WebSocket connection to TickDB."""
# TickDB WebSocket auth uses api_key as URL parameter
ws_url = f"wss://api.tickdb.ai/ws/market/stream?api_key={self.api_key}"
connector = aiohttp.TCPConnector(
limit=10,
ttl_dns_cache=300,
)
session = aiohttp.ClientSession(connector=connector)
ws = await session.ws_connect(
ws_url,
heartbeat=30,
timeout=aiohttp.ClientWSTimeout(ws_ping=25, ws_pong=25),
)
# Subscribe to depth channel for specified symbols
subscribe_msg = {
"cmd": "subscribe",
"channel": "depth",
"symbols": self.symbols,
}
await ws.send_json(subscribe_msg)
logger.info(f"Subscribed to depth channel: {self.symbols}")
return ws
async def run(self):
"""
Main producer loop with reconnection and backpressure handling.
⚠️ CRITICAL: This implements backpressure by blocking put() when
the queue is full. This signals to TickDB (via the upstream) that
we cannot accept more data at this time.
"""
self._running = True
self._shutdown_event.clear()
ws: Optional[aiohttp.ClientWebSocketResponse] = None
session: Optional[aiohttp.ClientSession] = None
while self._running:
try:
if ws is None or ws.closed:
session = aiohttp.ClientSession()
ws = await self.connect()
self.retry_count = 0 # Reset on successful connection
async for msg in ws:
if not self._running:
break
if msg.type == aiohttp.WSMsgType.PING:
await ws.pong()
continue
if msg.type == aiohttp.WSMsgType.TEXT:
await self._process_message(msg.data, ws)
elif msg.type == aiohttp.WSMsgType.ERROR:
logger.error(f"WebSocket error: {msg.data}")
break
except aiohttp.ClientError as e:
logger.warning(f"Connection error: {e}")
except asyncio.CancelledError:
logger.info("Producer cancelled, initiating shutdown")
self._running = False
break
finally:
if ws and not ws.closed:
await ws.close()
if session and not session.closed:
await session.close()
if self._running:
await self._handle_reconnect()
self._shutdown_event.set()
logger.info("Producer shut down cleanly")
async def _process_message(
self,
data: str,
ws: aiohttp.ClientWebSocketResponse,
):
"""Process incoming message with backpressure awareness."""
try:
payload = json.loads(data)
# Check for rate limit response
if payload.get("code") == 3001:
retry_after = int(payload.get("retry_after", 5))
logger.warning(f"Rate limited. Waiting {retry_after}s")
await asyncio.sleep(retry_after)
return
# Handle TickDB response format
if "data" not in payload:
return
update = self._parse_depth_update(payload["data"])
if update is None:
return
# Backpressure: block here when queue is full
# This creates backpressure on the WebSocket level
try:
await asyncio.wait_for(
self.queue.put(update),
timeout=5.0,
)
except asyncio.TimeoutError:
logger.error(
f"Queue full for {update.symbol} — "
f"dropping update (backpressure overflow)"
)
# In production, you might want to log this,
# switch to a drop-oldest strategy, or alert
except json.JSONDecodeError as e:
logger.warning(f"Invalid JSON: {e}")
def _parse_depth_update(self, data: dict) -> Optional[MarketUpdate]:
"""Parse TickDB depth channel response."""
try:
symbol = data["symbol"]
bids = data.get("bids", [])
asks = data.get("asks", [])
if not bids or not asks:
return None
bid_price, bid_size = bids[0]
ask_price, ask_size = asks[0]
return MarketUpdate(
symbol=symbol,
timestamp=data.get("ts", time.time()),
bid_price=float(bid_price),
bid_size=int(bid_size),
ask_price=float(ask_price),
ask_size=int(ask_size),
sequence=data.get("seq", 0),
)
except (KeyError, ValueError, IndexError) as e:
logger.debug(f"Parse error: {e}")
return None
async def _handle_reconnect(self):
"""Exponential backoff with jitter for reconnection."""
self.retry_count += 1
delay = min(
self.base_delay * (2 ** self.retry_count),
self.max_delay,
)
# Add jitter: ±10% randomization prevents thundering herd
jitter = random.uniform(-delay * 0.1, delay * 0.1)
sleep_time = max(0, delay + jitter)
logger.info(
f"Reconnecting in {sleep_time:.1f}s "
f"(attempt {self.retry_count})"
)
await asyncio.sleep(sleep_time)
async def stop(self):
"""Graceful shutdown."""
self._running = False
await self._shutdown_event.wait()
The Consumer: Multi-Worker Strategy Pipeline
The consumer layer consists of multiple workers that pull from the shared queue. Each worker runs independently, allowing different strategies to process data at their own pace. The queue acts as the synchronization point.
import asyncio
import time
from typing import Protocol
class Strategy(Protocol):
"""Protocol for strategy implementations."""
async def process(self, update: MarketUpdate) -> None:
"""Process a single market update."""
...
class StrategyWorker:
"""
Worker that pulls updates from queue and processes through a strategy.
Implements graceful shutdown, lag monitoring, and error isolation.
"""
def __init__(
self,
worker_id: int,
queue: asyncio.Queue,
strategy: Strategy,
max_lag_warnings: int = 100,
):
self.worker_id = worker_id
self.queue = queue
self.strategy = strategy
self.max_lag_warnings = max_lag_warnings
self._running = False
self._shutdown_event = asyncio.Event()
self._lag_warnings = 0
# Metrics
self.processed_count = 0
self.error_count = 0
self.max_observed_lag = 0.0
async def run(self):
"""Main worker loop."""
self._running = True
self._shutdown_event.clear()
logger.info(f"Worker {self.worker_id} started")
while self._running:
try:
# Get next update with timeout for graceful shutdown
update = await asyncio.wait_for(
self.queue.get(),
timeout=1.0,
)
# Monitor queue lag
now = time.time()
lag = now - update.timestamp
self.max_observed_lag = max(self.max_observed_lag, lag)
if lag > 1.0 and self._lag_warnings < self.max_lag_warnings:
self._lag_warnings += 1
logger.warning(
f"Worker {self.worker_id}: "
f"Queue lag {lag:.2f}s for {update.symbol}"
)
# Process with error isolation
try:
await self.strategy.process(update)
self.processed_count += 1
except Exception as e:
self.error_count += 1
logger.error(
f"Worker {self.worker_id} processing error "
f"for {update.symbol}: {e}"
)
finally:
self.queue.task_done()
except asyncio.TimeoutError:
# No updates available — idle loop
continue
except asyncio.CancelledError:
logger.info(f"Worker {self.worker_id} cancelled")
self._running = False
break
self._shutdown_event.set()
logger.info(
f"Worker {self.worker_id} stopped "
f"(processed: {self.processed_count}, errors: {self.error_count})"
)
async def stop(self):
"""Initiate graceful shutdown."""
self._running = False
await self._shutdown_event.wait()
Orchestrating the Pipeline
The orchestrator ties the producer and workers together, managing their lifecycle and providing a clean API for deployment.
class MarketDataPipeline:
"""
Orchestrates the producer-consumer market data pipeline.
Usage:
pipeline = MarketDataPipeline(
api_key=os.environ["TICKDB_API_KEY"],
symbols=["AAPL.US", "NVDA.US", "TSLA.US"],
)
await pipeline.start()
# ... run for duration ...
await pipeline.stop()
"""
def __init__(
self,
api_key: str,
symbols: list[str],
queue_size: int = 10000,
num_workers: int = 4,
):
self.api_key = api_key
self.symbols = symbols
self.queue_size = queue_size
self.num_workers = num_workers
self._queue: Optional[asyncio.Queue] = None
self._producer: Optional[TickDBWebSocketProducer] = None
self._workers: list[StrategyWorker] = []
self._tasks: list[asyncio.Task] = []
async def start(self, strategies: list[Strategy]):
"""Start the pipeline with specified strategies."""
# Initialize queue
self._queue = asyncio.Queue(maxsize=self.queue_size)
# Create producer
self._producer = TickDBWebSocketProducer(
api_key=self.api_key,
symbols=self.symbols,
output_queue=self._queue,
max_queue_size=self.queue_size,
)
# Create workers — each gets its own strategy instance
for i in range(self.num_workers):
# Clone strategy for worker isolation
strategy = strategies[i % len(strategies)]
worker = StrategyWorker(
worker_id=i,
queue=self._queue,
strategy=strategy,
)
self._workers.append(worker)
# Schedule tasks
self._tasks.append(asyncio.create_task(self._producer.run()))
for worker in self._workers:
self._tasks.append(asyncio.create_task(worker.run()))
logger.info(
f"Pipeline started: 1 producer, {len(self._workers)} workers, "
f"queue size: {self.queue_size}"
)
async def stop(self):
"""Gracefully stop all pipeline components."""
logger.info("Stopping pipeline...")
# Cancel all tasks
for task in self._tasks:
task.cancel()
# Wait for graceful completion
await asyncio.gather(*self._tasks, return_exceptions=True)
logger.info("Pipeline stopped")
def get_metrics(self) -> dict:
"""Return current pipeline metrics."""
if not self._queue:
return {}
return {
"queue_depth": self._queue.qsize(),
"queue_capacity": self.queue_size,
"queue_utilization": self._queue.qsize() / self.queue_size,
"workers": [
{
"id": w.worker_id,
"processed": w.processed_count,
"errors": w.error_count,
"max_lag": w.max_observed_lag,
}
for w in self._workers
],
}
Backpressure Control Strategies
The queue is necessary but not sufficient. Without explicit backpressure control, an unbounded queue merely delays the failure mode. Three strategies address backpressure at different levels:
Strategy 1: Blocking Put with Timeout
The simplest approach: when the queue is full, the producer blocks until space becomes available. This creates backpressure directly on the WebSocket connection.
# Implemented in the producer's _process_message method
try:
await asyncio.wait_for(
self.queue.put(update),
timeout=5.0,
)
except asyncio.TimeoutError:
logger.error("Queue full — dropping update")
# Consider alerting here
Trade-off: Simple, reliable backpressure. But the producer stalls, potentially causing TickDB to disconnect due to inactivity. Use this when processing speed is close to arrival speed.
Strategy 2: Priority-Based Drop
For multi-priority data streams, drop lowest-priority updates when the queue is saturated.
class PriorityQueue(asyncio.Queue):
"""Queue that drops lowest-priority items when full."""
async def put(self, item: tuple[int, MarketUpdate]):
"""
Put item with priority tuple.
Priority tuple: (priority_level, market_update)
Lower priority_level = higher priority (dropped last)
"""
while True:
# Check if we can put without blocking
if self.full():
# Find and drop lowest-priority item
lowest_priority_idx = self._drop_lowest_priority()
if lowest_priority_idx is None:
# Queue is full of high-priority items — block
await asyncio.sleep(0.01)
continue
# Now we have space
await super().put(item)
break
def _drop_lowest_priority(self) -> Optional[int]:
"""Remove the lowest-priority item. Returns index dropped."""
if self.empty():
return None
# Find minimum priority (highest number = lowest priority)
min_idx = 0
min_priority = self._queue[0][0]
for i, (priority, _) in enumerate(self._queue):
if priority > min_priority:
min_priority = priority
min_idx = i
# Remove the lowest-priority item
self._queue.pop(min_idx)
return min_idx
Trade-off: Preserves high-priority data. Requires priority classification upstream. Use this when some data (e.g., L1 vs. L5 depth) has different criticality.
Strategy 3: Adaptive Rate Limiting
Dynamically adjust subscription granularity based on queue depth.
class AdaptiveSubscriptionManager:
"""
Adjusts subscription level based on queue pressure.
High pressure: drop to L1 depth only
Medium pressure: allow L1-L5
Low pressure: full L1-L10 depth
"""
def __init__(
self,
queue: asyncio.Queue,
high_water_mark: float = 0.8,
low_water_mark: float = 0.3,
):
self.queue = queue
self.high_water_mark = high_water_mark
self.low_water_mark = low_water_mark
self._current_depth_level = "L1-L10"
def get_subscription_params(self) -> dict:
"""Return current subscription parameters based on queue state."""
utilization = self.queue.qsize() / self.queue.maxsize
if utilization > self.high_water_mark:
self._current_depth_level = "L1"
return {"depth": "L1", "throttle_ms": 100}
elif utilization > self.low_water_mark:
self._current_depth_level = "L1-L5"
return {"depth": "L1-L5", "throttle_ms": 50}
else:
self._current_depth_level = "L1-L10"
return {"depth": "L1-L10", "throttle_ms": 0}
def get_status(self) -> str:
"""Human-readable status."""
utilization = self.queue.qsize() / self.queue.maxsize
return (
f"Depth: {self._current_depth_level}, "
f"Queue: {self.queue.qsize()}/{self.queue.maxsize} "
f"({utilization:.1%})"
)
Trade-off: Most sophisticated. Requires upstream support for subscription parameter changes. Use this for systems with variable data rates (e.g., around earnings announcements).
Deploying the Pipeline
Environment Configuration
# Required environment variables
export TICKDB_API_KEY="your_api_key_here"
# Optional tuning parameters
export MARKET_QUEUE_SIZE=10000 # Messages in buffer
export MARKET_NUM_WORKERS=4 # Parallel strategy instances
export MARKET_BACKOFF_BASE=1.0 # Initial reconnection delay (s)
export MARKET_BACKOFF_MAX=60.0 # Maximum reconnection delay (s)
Example: Simple Spread Strategy
class SpreadStrategy:
"""Monitors bid-ask spread for arbitrage opportunities."""
def __init__(self, symbol: str, spread_threshold: float = 0.05):
self.symbol = symbol
self.spread_threshold = spread_threshold
self._last_spread = None
async def process(self, update: MarketUpdate):
"""Check for spread anomalies."""
if update.symbol != self.symbol:
return
spread = update.ask_price - update.bid_price
spread_bps = (spread / update.bid_price) * 10000
if spread_bps > self.spread_threshold:
logger.info(
f"Spread alert: {self.symbol} "
f"{spread_bps:.2f} bps (threshold: {self.spread_threshold} bps)"
)
self._last_spread = spread_bps
# Deployment example
async def main():
pipeline = MarketDataPipeline(
api_key=os.environ["TICKDB_API_KEY"],
symbols=["AAPL.US", "NVDA.US"],
queue_size=10000,
num_workers=2,
)
strategies = [
SpreadStrategy("AAPL.US", spread_threshold=0.03),
SpreadStrategy("NVDA.US", spread_threshold=0.04),
]
await pipeline.start(strategies)
try:
while True:
await asyncio.sleep(60)
metrics = pipeline.get_metrics()
logger.info(f"Metrics: {metrics}")
except KeyboardInterrupt:
pass
finally:
await pipeline.stop()
if __name__ == "__main__":
asyncio.run(main())
Scaling Considerations
| Deployment Size | Workers | Queue Size | Notes |
|---|---|---|---|
| Development | 1–2 | 1,000 | Log everything, low latency |
| Production (individual) | 2–4 | 10,000 | Monitor queue depth |
| Production (team) | 4–8 | 50,000 | Distributed workers across processes |
| Institutional | 16+ | Custom | Consider Kafka or Redis for cross-process queuing |
For single-process scaling, Python's GIL limits true parallelism for CPU-bound work. Use the multi-worker model for I/O-bound strategies (network calls, database writes) where asyncio provides real concurrency. For CPU-bound signal calculation, consider offloading to a process pool or using NumPy vectorization.
Common Failure Modes and Mitigations
| Failure Mode | Symptom | Mitigation |
|---|---|---|
| Queue overflow | Logs show "dropping update" | Increase queue size, add workers, or implement priority drop |
| Worker crash | One worker stops processing | Error isolation prevents cascade; restart worker automatically |
| Producer disconnection | No data flowing | Exponential backoff + jitter handles transient failures |
| Memory growth | RAM usage climbs steadily | Bounded queue + monitoring prevents unbounded growth |
| Stale data trading | Strategies fire on old data | Queue lag monitoring; halt trading above threshold |
TickDB Integration Notes
The producer implementation above connects to TickDB's depth channel, which provides order book snapshots at configurable levels (L1–L10 depending on market). The key parameters:
- Symbol format: Exchange.SYMBOL (e.g.,
AAPL.US,BTC.BTC) - Channel:
depthfor order book,tradesfor tick data - Authentication: API key as URL parameter for WebSocket, header for REST
- Rate limits: Code 3001 indicates throttling; respect Retry-After header
For backtesting with historical data, use the /kline endpoint with parameters:
# Historical data for strategy calibration
import requests
response = requests.get(
"https://api.tickdb.ai/v1/market/kline",
headers={"X-API-Key": os.environ["TICKDB_API_KEY"]},
params={
"symbol": "AAPL.US",
"interval": "1m",
"limit": 1000,
"start_time": 1706745600, # Unix timestamp
"end_time": 1706832000,
},
timeout=(3.05, 10),
)
Note: The trades endpoint does not cover US equities or A-shares. For US equity order flow analysis, use the depth channel with the producer implementation above.
Conclusion
The producer-consumer pattern transforms market data distribution from a fragile, synchronous pipeline into a resilient, asynchronous system. The queue absorbs bursts. The workers process at their own cadence. The backpressure mechanisms ensure the system never accumulates work it cannot complete.
Three principles guide the implementation:
Bounded queues with monitoring: An unbounded queue is a memory leak with a delayed failure. Always set maxsize and monitor utilization.
Backpressure as first-class design: Backpressure is not an error condition to avoid. It is the signal that keeps the system in equilibrium. Design for it explicitly.
Error isolation at every layer: A crashed strategy should not crash the pipeline. A stalled worker should not stall the producer. Layer isolation prevents cascade failures.
The code in this article provides a production-ready foundation. Adapt the queue sizing, backpressure strategy, and worker count to your specific strategy latency profile. The architecture is proven; the tuning is specific to your deployment.
Next Steps
If you're building a single-strategy system, start with the single-worker configuration and monitor queue depth. Add workers only when processing lag becomes problematic.
If you need historical data for strategy calibration, the TickDB /kline endpoint provides 10+ years of cleaned US equity OHLCV data suitable for backtesting across full market cycles.
If you're scaling to multiple processes, consider replacing asyncio.Queue with a distributed message broker (Kafka, Redis Streams) to maintain backpressure across process boundaries.
If you use AI coding assistants, search for the tickdb-market-data SKILL on ClawHub for accelerated integration templates.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results.