In production trading systems, a single anomaly can expose the fragility of naive data pipelines. Consider this scenario: at 09:30:00 ET, a major macroeconomic announcement triggers a 47-millisecond burst of order book updates across 200 symbols. Your strategy engine can process 15,000 messages per second—but the market data feed is delivering 28,000. Within 90 seconds, your in-memory buffer grows unbounded. By the 120-second mark, your process runs out of memory and crashes, losing two minutes of critical session data.
This failure mode is not hypothetical. It is the predictable consequence of a fundamental architectural mismatch: the producer (the market data feed) and the consumer (your strategy engine) operate at different speeds, and naive implementations assume they operate in lockstep. The producer-consumer pattern, implemented correctly with asyncio.Queue and explicit backpressure control, resolves this mismatch by introducing a bounded buffer that decouples ingestion from processing—and provides mechanisms to signal congestion upstream before the system collapses.
This article walks through a production-grade implementation: a market data ingestion pipeline that handles variable-rate data streams, applies backpressure when queues fill, distributes work across multiple consumer workers, and survives the most demanding intraday conditions without data loss or memory exhaustion.
The Core Problem: Speed Mismatch in Data Pipelines
Market data arrives in bursts. A typical trading day exhibits two to three orders of magnitude of variation between quiet periods and high-activity windows. During earnings seasons, Fed announcements, or macro data releases, tick volume can spike by 1,000% within milliseconds. A strategy engine optimized for average throughput will saturate instantly under these conditions.
The fundamental challenge is that the data source (the producer) operates independently of the data consumer (your strategy logic). They do not coordinate. The producer sends data as fast as the network allows; the consumer processes data as fast as CPU and logic permit. When production rate exceeds consumption rate persistently, unbounded buffering leads to memory exhaustion. When the consumer processes faster than data arrives, the pipeline sits idle—a missed opportunity but not a system failure.
The naive solution—dropping messages when the buffer is full—is unacceptable in trading systems. Every tick carries potential alpha. The correct solution is backpressure: when the consumer cannot keep pace, the system must signal the producer to slow down, or it must shed load in a controlled, prioritized manner.
Architecture: Producer-Consumer with Bounded Queue
The producer-consumer pattern introduces a queue as the synchronization point between producers and consumers. The queue has a fixed maximum size (the "buffer"), and two critical mechanisms:
- Bounded capacity: The queue blocks producers from enqueueing when full, creating backpressure.
- Cooperative scheduling: asyncio.Queue uses async/await semantics, allowing the event loop to switch tasks when waiting.
The architecture for a market data ingestion system looks like this:
┌─────────────────────────────────────────────────────────────────┐
│ MARKET DATA SOURCE │
│ (WebSocket Feed — TickDB or Exchange) │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ PRODUCER TASK │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ WebSocket client (heartbeat, reconnect, rate-limit) │ │
│ │ Parses raw ticks → normalized TickData objects │ │
│ │ Enqueues to: asyncio.Queue (maxsize=configurable) │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ ASYNCIO.QUEUE (Bounded Buffer) │
│ maxsize: 10,000 ticks (configurable) │
│ Blocks producer when full → backpressure signal │
└─────────────────────────────────────────────────────────────────┘
│
┌───────────────┼───────────────┐
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ WORKER TASK 1 │ │ WORKER TASK 2 │ │ WORKER TASK N │
│ Dequeues tick │ │ Dequeues tick │ │ Dequeues tick │
│ Applies logic │ │ Applies logic │ │ Applies logic │
│ Emits signals │ │ Emits signals │ │ Emits signals │
└─────────────────┘ └─────────────────┘ └─────────────────┘
This architecture decouples the network I/O-bound task (WebSocket receiving) from the CPU-bound task (strategy processing). The queue absorbs bursts; the workers process at sustainable rates.
Production-Grade Implementation
The following implementation addresses every failure mode encountered in production trading systems:
- Heartbeat handling: WebSocket ping/pong to detect stale connections
- Exponential backoff with jitter: Reconnection resilience without thundering herd
- Rate-limit handling: Proper response to
3001error codes with Retry-After headers - Bounded queue with backpressure: Configurable maxsize that blocks producers
- Multiple consumer workers: Distributes processing load across CPU cores
- Graceful shutdown: Drains queue before exit to avoid data loss
- Metrics: Queue depth and processing lag for monitoring
"""
Market Data Ingestion Pipeline
Producer-Consumer Pattern with asyncio.Queue and Backpressure Control
This module implements a production-grade market data pipeline that:
- Connects to a WebSocket market data source
- Bounded queue absorbs burst traffic
- Multiple worker tasks process data concurrently
- Backpressure prevents memory exhaustion
"""
import os
import asyncio
import json
import time
import random
import logging
from dataclasses import dataclass, field
from typing import Optional, Dict, Any
from datetime import datetime
import aiohttp
# Configure structured logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s"
)
logger = logging.getLogger("MarketDataPipeline")
@dataclass
class TickData:
"""Normalized tick representation across data sources."""
symbol: str
timestamp: float
bid: float
ask: float
bid_size: int
ask_size: int
source: str = "unknown"
raw: Dict[str, Any] = field(default_factory=dict)
@property
def spread(self) -> float:
return self.ask - self.bid
@property
def mid(self) -> float:
return (self.ask + self.bid) / 2
@dataclass
class PipelineConfig:
"""Configuration for the market data pipeline."""
# Queue settings
queue_maxsize: int = 10_000
# Worker settings
num_workers: int = 4
worker_batch_size: int = 100 # Process ticks in batches for efficiency
# WebSocket settings
websocket_url: str = "wss://api.tickdb.ai/v1/stream"
api_key: Optional[str] = None
symbols: list[str] = field(default_factory=lambda: ["AAPL.US", "NVDA.US"])
# Reconnection settings
base_reconnect_delay: float = 1.0
max_reconnect_delay: float = 60.0
reconnect_jitter: float = 0.1
# Rate limiting
rate_limit_code: int = 3001
default_rate_limit_delay: float = 5.0
# Timeouts
connection_timeout: float = 10.0
ping_interval: float = 30.0
class MarketDataProducer:
"""
WebSocket producer that fetches market data and enqueues to asyncio.Queue.
Implements heartbeat, exponential backoff reconnection, and rate-limit handling.
"""
def __init__(self, config: PipelineConfig, queue: asyncio.Queue):
self.config = config
self.queue = queue
self._running = False
self._session: Optional[aiohttp.ClientSession] = None
self._reconnect_attempts = 0
async def start(self):
"""Main producer loop with automatic reconnection."""
self._running = True
reconnect_delay = self.config.base_reconnect_delay
while self._running:
try:
await self._connect_and_stream(reconnect_delay)
# Connection closed normally; reset reconnection state
self._reconnect_attempts = 0
reconnect_delay = self.config.base_reconnect_delay
except asyncio.CancelledError:
logger.info("Producer cancelled — shutting down")
break
except Exception as e:
self._reconnect_attempts += 1
logger.error(
f"Producer error (attempt {self._reconnect_attempts}): {e}"
)
# Calculate delay with exponential backoff and jitter
delay = min(
reconnect_delay * (2 ** (self._reconnect_attempts - 1)),
self.config.max_reconnect_delay
)
# Add jitter to prevent thundering herd on mass reconnection
jitter = random.uniform(0, delay * self.config.reconnect_jitter)
delay += jitter
logger.info(f"Reconnecting in {delay:.1f} seconds")
await asyncio.sleep(delay)
async def _connect_and_stream(self, reconnect_delay: float):
"""Establish WebSocket connection and stream market data."""
api_key = self.config.api_key or os.environ.get("TICKDB_API_KEY")
if not api_key:
raise ValueError("API key required — set TICKDB_API_KEY environment variable")
params = "&".join(f"symbols={s}" for s in self.config.symbols)
url = f"{self.config.websocket_url}?{params}&api_key={api_key}"
timeout = aiohttp.ClientTimeout(
total=None,
connect=self.config.connection_timeout,
sock_read=None
)
async with aiohttp.ClientSession(timeout=timeout) as session:
self._session = session
async with session.ws_connect(url) as ws:
logger.info(f"Connected to WebSocket: {self.config.websocket_url}")
# Send initial subscription
await ws.send_json({
"cmd": "subscribe",
"params": {"symbols": self.config.symbols, "channels": ["depth", "trades"]}
})
last_ping = time.monotonic()
async for msg in ws:
if not self._running:
break
if msg.type == aiohttp.WSMsgType.PING:
await ws.pong()
last_ping = time.monotonic()
continue
if msg.type == aiohttp.WSMsgType.TEXT:
await self._process_message(msg.data)
last_ping = time.monotonic()
elif msg.type == aiohttp.WSMsgType.ERROR:
raise ConnectionError(f"WebSocket error: {msg.data}")
elif msg.type == aiohttp.WSMsgType.CLOSE:
logger.info("WebSocket closed by server")
break
# Check for stale connection (heartbeat timeout)
if time.monotonic() - last_ping > self.config.ping_interval * 2:
logger.warning("Heartbeat timeout — reconnecting")
break
async def _process_message(self, raw_message: str):
"""Parse raw message, create TickData, enqueue with backpressure."""
try:
data = json.loads(raw_message)
# Check for rate limit response
code = data.get("code", 0)
if code == self.config.rate_limit_code:
retry_after = float(
data.get("headers", {}).get("Retry-After", self.config.default_rate_limit_delay)
)
logger.warning(f"Rate limited — waiting {retry_after}s")
await asyncio.sleep(retry_after)
return
if code != 0 and code != self.config.rate_limit_code:
logger.warning(f"Server error code {code}: {data.get('message')}")
return
# Parse tick data from response
ticks = self._parse_ticks(data)
for tick in ticks:
# ⬇️ BACKPRESSURE: put() blocks when queue is full
# This prevents memory exhaustion by slowing down the producer
try:
# Use wait_for with timeout to allow graceful shutdown
await asyncio.wait_for(
self.queue.put(tick),
timeout=5.0
)
except asyncio.TimeoutError:
# Queue is saturated — log metric for alerting
logger.error(
f"Queue full for 5s (size={self.queue.qsize()}) — "
"backpressure active, potential data loss risk"
)
except json.JSONDecodeError as e:
logger.warning(f"Failed to parse message: {e}")
def _parse_ticks(self, data: Dict[str, Any]) -> list[TickData]:
"""Parse raw API response into normalized TickData objects."""
ticks = []
payload = data.get("data", {})
for symbol in self.config.symbols:
if symbol not in payload:
continue
book = payload[symbol]
# Normalize TickDB depth snapshot format
if "depth" in book:
depth = book["depth"]
bids = depth.get("bids", [])
asks = depth.get("asks", [])
if bids and asks:
tick = TickData(
symbol=symbol,
timestamp=depth.get("timestamp", time.time()),
bid=bids[0][0],
ask=asks[0][0],
bid_size=int(bids[0][1]),
ask_size=int(asks[0][1]),
source="tickdb",
raw=depth
)
ticks.append(tick)
return ticks
async def stop(self):
"""Gracefully stop the producer."""
self._running = False
if self._session:
await self._session.close()
class MarketDataWorker:
"""
Consumer worker that dequeues ticks and applies strategy logic.
Multiple workers run concurrently to distribute processing load.
"""
def __init__(self, worker_id: int, queue: asyncio.Queue, config: PipelineConfig):
self.worker_id = worker_id
self.queue = queue
self.config = config
self._running = False
self._processed_count = 0
async def start(self):
"""Main worker loop — continuously dequeues and processes ticks."""
self._running = True
logger.info(f"Worker {self.worker_id} started")
while self._running:
try:
# Dequeue with timeout to allow periodic health checks
tick = await asyncio.wait_for(
self.queue.get(),
timeout=1.0
)
await self._process_tick(tick)
self.queue.task_done()
self._processed_count += 1
# Log health metrics periodically
if self._processed_count % 1000 == 0:
logger.info(
f"Worker {self.worker_id} processed {self._processed_count} ticks, "
f"queue depth: {self.queue.qsize()}"
)
except asyncio.TimeoutError:
# No messages available — worker idle, check for shutdown signal
continue
except asyncio.CancelledError:
logger.info(f"Worker {self.worker_id} cancelled")
break
async def _process_tick(self, tick: TickData):
"""
Strategy processing logic.
Replace this method with your actual strategy implementation.
"""
# Example: compute mid-price and spread
# In production, this would trigger signals, update state, etc.
# Simulate processing latency (CPU-bound work)
# await asyncio.sleep(0.0001) # Remove in production
# Log anomalous spreads for monitoring
if tick.spread > 0.05: # 5-cent spread threshold
logger.warning(
f"Anomalous spread on {tick.symbol}: "
f"${tick.spread:.4f} at {datetime.fromtimestamp(tick.timestamp)}"
)
async def stop(self):
"""Gracefully stop the worker."""
self._running = False
class MarketDataPipeline:
"""
Orchestrates producer and worker tasks for the market data pipeline.
Manages lifecycle, monitoring, and graceful shutdown.
"""
def __init__(self, config: Optional[PipelineConfig] = None):
self.config = config or PipelineConfig()
# ⬇️ BOUNDED QUEUE: maxsize creates backpressure point
self.queue: asyncio.Queue[TickData] = asyncio.Queue(
maxsize=self.config.queue_maxsize
)
self.producer: Optional[MarketDataProducer] = None
self.workers: list[MarketDataWorker] = []
self._tasks: list[asyncio.Task] = []
async def start(self):
"""Start the pipeline — spawns producer and workers."""
logger.info(
f"Starting pipeline: {self.config.num_workers} workers, "
f"queue maxsize={self.config.queue_maxsize}"
)
# Create producer
self.producer = MarketDataProducer(self.config, self.queue)
producer_task = asyncio.create_task(self.producer.start())
self._tasks.append(producer_task)
# Create and spawn worker tasks
for i in range(self.config.num_workers):
worker = MarketDataWorker(i, self.queue, self.config)
self.workers.append(worker)
worker_task = asyncio.create_task(worker.start())
self._tasks.append(worker_task)
logger.info("Pipeline started successfully")
async def stop(self, timeout: float = 10.0):
"""
Graceful shutdown: stop accepting new data, drain queue, exit workers.
"""
logger.info("Initiating graceful shutdown...")
# Stop producer first — no new data will arrive
if self.producer:
await self.producer.stop()
# Wait for queue to drain (up to timeout)
drain_deadline = time.monotonic() + timeout
remaining = self.queue.qsize()
if remaining > 0:
logger.info(f"Draining {remaining} remaining ticks...")
# Wait for queue to empty or timeout
try:
await asyncio.wait_for(
self.queue.join(),
timeout=max(0, drain_deadline - time.monotonic())
)
logger.info("Queue drained successfully")
except asyncio.TimeoutError:
logger.warning(
f"Queue drain timeout — {self.queue.qsize()} ticks may be lost"
)
# Cancel all worker tasks
for worker in self.workers:
await worker.stop()
# Wait for all tasks to complete
for task in self._tasks:
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
logger.info("Pipeline shutdown complete")
async def run_pipeline():
"""Example usage of the MarketDataPipeline."""
config = PipelineConfig(
queue_maxsize=10_000,
num_workers=4,
symbols=["AAPL.US", "NVDA.US", "TSLA.US"],
api_key=os.environ.get("TICKDB_API_KEY")
)
pipeline = MarketDataPipeline(config)
try:
await pipeline.start()
# Run for specified duration, then shutdown
await asyncio.sleep(3600) # 1 hour
except KeyboardInterrupt:
logger.info("Interrupted by user")
finally:
await pipeline.stop(timeout=5.0)
if __name__ == "__main__":
# ⚠️ For production HFT workloads, consider multiprocessing instead of asyncio
# asyncio provides concurrency but not true parallelism (GIL constraint)
# For CPU-bound strategy logic at extreme throughput, use ProcessPoolExecutor
asyncio.run(run_pipeline())
Core Algorithm: Backpressure and Queue Management
The production-grade implementation above relies on three algorithmic pillars: bounded queues, blocking enqueue, and worker pooling. This section dissects why each mechanism matters and how they interact.
Bounded Queue with Backpressure
The asyncio.Queue(maxsize=N) creates a queue with a maximum capacity of N items. When a producer calls await queue.put(item) and the queue is full, the call blocks until a consumer dequeues an item. This is the backpressure mechanism.
The key insight: backpressure is a signal, not a failure. When the queue blocks the producer, the network socket's receive buffer continues to fill. If the data source is a WebSocket, the TCP receive window will eventually fill, causing the sender's TCP congestion control to throttle transmission. The queue effectively " negotiates" with the producer through TCP flow control—no explicit protocol-level backpressure signal is needed.
However, this indirect backpressure has limits. For protocols that do not respect TCP flow control (e.g., some UDP-based market data feeds), or when latency is critical, explicit backpressure signaling is required. In such cases, implement a queue.qsize() monitoring loop that sends a congestion notification upstream when the queue exceeds, for example, 80% capacity.
Worker Pool Sizing
The number of workers should match the nature of the processing workload:
| Workload type | Worker strategy | Reasoning |
|---|---|---|
| I/O-bound (network calls, API requests) | 1 worker + async | asyncio handles concurrency; additional workers add no throughput |
| CPU-bound (signal computation, model inference) | Multiple workers (CPU count) | True parallelism requires multiple processes or threads |
| Mixed I/O + CPU | Worker pool + ProcessPoolExecutor | Async for I/O; separate process pool for CPU-heavy tasks |
For most quant strategy workloads, the processing logic is CPU-bound (computing indicators, evaluating signals). In this case, a single asyncio worker pool is insufficient because Python's Global Interpreter Lock (GIL) prevents true parallel execution of Python bytecode. The engineering warning comment in the code addresses this: for HFT workloads, use ProcessPoolExecutor or multiprocessing.
Batch Processing for Throughput
Processing ticks one at one introduces per-item overhead (function call, context switch, potential cache miss). Batching amortizes this overhead: instead of processing one tick per loop iteration, collect N ticks and process them as a vectorized operation.
async def start(self):
self._running = True
batch: list[TickData] = []
while self._running:
try:
# Collect batch
tick = await asyncio.wait_for(self.queue.get(), timeout=0.1)
batch.append(tick)
self.queue.task_done()
# Process batch when full or timeout
if len(batch) >= self.worker_batch_size:
await self._process_batch(batch)
batch = []
except asyncio.TimeoutError:
# Flush partial batch on timeout
if batch:
await self._process_batch(batch)
batch = []
Queue Configuration Guide by Deployment Scale
The optimal queue configuration depends on your deployment context. Use the following guide to size your pipeline:
| Scale | Queue maxsize | Workers | Memory footprint | Latency target |
|---|---|---|---|---|
| Development / backtesting | 1,000 | 1 | ~50 MB | None (batch mode) |
| Individual trader (live) | 5,000 | 2 | ~200 MB | <100 ms |
| Small fund (2–5 strategies) | 10,000 | 4 | ~500 MB | <50 ms |
| Institutional (10+ strategies) | 50,000 | 8+ | ~2 GB | <20 ms |
Memory calculation: Each TickData object is approximately 200–500 bytes depending on the raw payload. A 10,000-item queue with 500-byte objects uses roughly 5 MB of queue memory plus Python object overhead—manageable for most deployments. The risk is not the queue itself but the accumulation of per-tick processing state in downstream strategy objects. Profile your actual memory consumption with tracemalloc before deploying.
Comparison: Queue Patterns for Market Data
Three patterns are commonly used for market data distribution. Each has trade-offs:
| Pattern | Throughput | Latency | Backpressure | Complexity |
|---|---|---|---|---|
| Raw asyncio (no queue) | Highest | Lowest | None — data loss or crash | Low |
| asyncio.Queue (single consumer) | High | Low | Yes (bounded queue) | Medium |
| asyncio.Queue (multiple workers) | Highest effective | Medium | Yes (bounded queue) | Medium-High |
| Redis Streams / Kafka | Medium | Higher | Yes (acknowledgment) | High |
| multiprocessing.Queue | High | Low | Yes (bounded queue) | High |
Raw asyncio without a queue is appropriate only for strategies that process every single tick in real time with no state accumulation. In practice, most strategies need state (position, indicator values, order book snapshots) that accumulates with each tick. Without a queue, the strategy must keep this state in memory indefinitely—and a burst of ticks will overwhelm the processing loop.
Redis Streams or Kafka are appropriate when the market data pipeline spans multiple machines or requires durable replay. The acknowledgment-based consumption model provides stronger delivery guarantees than asyncio.Queue, at the cost of higher latency (typically 1–5 ms per hop) and operational complexity.
multiprocessing.Queue provides true parallelism for CPU-bound workloads, bypassing the GIL. However, it introduces serialization overhead for every message (data must be pickled/unpickled to cross process boundaries) and complicates shared state management.
For single-machine deployments processing market data for one to ten strategies, asyncio.Queue with multiple workers is the correct choice: low latency, native backpressure, and sufficient parallelism when CPU-bound work is offloaded to a process pool.
Deployment Checklist
Before deploying this pipeline to production:
- API key management: Verify
TICKDB_API_KEYis set as an environment variable, not hardcoded - Queue sizing: Profile under historical burst conditions; set
maxsizeto absorb the 99th-percentile burst duration - Worker count: Match to CPU cores for CPU-bound strategies; use 1 worker for I/O-bound strategies
- Monitoring: Export
queue.qsize()to your metrics system; alert when queue exceeds 80% capacity for more than 10 seconds - Graceful shutdown: Test the drain sequence under load—verify no ticks are lost on SIGTERM
- GIL concern: If your strategy is CPU-bound, move computation to
ProcessPoolExecutor; the asyncio event loop runs on a single thread - Reconnection stress test: Simulate network interruptions at 1-second, 10-second, and 60-second durations; verify exponential backoff converges correctly
Closing
The producer-consumer pattern with asyncio.Queue is not a silver bullet. It is a precision instrument: the bounded queue absorbs bursts and creates backpressure, the multiple workers distribute processing load, and the graceful shutdown sequence prevents data loss during deployment. When configured correctly—queue sized to your burst profile, workers matched to your workload type, monitoring on queue depth—it transforms a fragile, latency-sensitive data pipeline into a resilient system that survives the most demanding intraday conditions.
The implementation above is production-ready as a reference architecture. Adapt the TickData normalization logic to your specific data source schema, replace the _process_tick method with your strategy logic, and add your metrics export layer. The architecture will not change; only the implementation details will.
Next Steps
If you're building a live trading system, start with the asyncio.Queue pattern in this article as your ingestion layer. Profile your strategy's CPU consumption under historical tick replay to determine the correct worker count.
If you need 10+ years of historical market data for backtesting your strategy, TickDB provides cleaned, timestamp-aligned OHLCV data via REST API. Replay historical sessions to calibrate your queue sizing before going live.
If you're running multiple strategies on a single machine, consider a fan-out architecture: one producer feeds a single queue, and multiple worker groups subscribe to filtered subsets. This avoids duplicating the WebSocket connection while keeping strategy logic isolated.
If your strategy is CPU-bound at high tick rates, investigate ProcessPoolExecutor for offloading computation. The asyncio event loop remains the conductor; worker processes handle the heavy lifting without blocking the event loop.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results.