Every quant developer hits the same wall eventually. Your backtest runs fine. Your strategy logic is sound. Then you wire up live market data and watch your Python process fall behind by seconds — missing entries, skipping candles, watching the order book stale out while your code blocks on a single HTTP request.
Python's GIL is not your enemy. Sequential, blocking I/O is.
The solution is asyncio — and it's not as hard as the tutorials make it look. This article builds a production-grade async market data pipeline from scratch, benchmarks it against the synchronous baseline, and shows you exactly where the performance gains come from.
The Core Problem: Why Your Quant System Stalls
Before we write code, we need to understand what we're fixing.
A typical synchronous market data consumer looks like this:
import requests
import time
# Fetching multiple symbols synchronously
symbols = ["NVDA.US", "AAPL.US", "TSLA.US"]
for symbol in symbols:
response = requests.get(f"https://api.tickdb.ai/v1/market/kline", params={
"symbol": symbol,
"interval": "1m",
"limit": 100
})
process(response.json())
If each API call takes 80 ms, three symbols cost you 240 ms — sequentially. With 50 symbols, you're at 4 seconds of pure waiting.
Now consider a live WebSocket feed:
import websocket # synchronous library
def on_message(ws, message):
process(message) # If this blocks, the entire loop stalls
ws = websocket.WebSocketApp("wss://api.tickdb.ai/ws", on_message=on_message)
ws.run_forever()
If process() performs any I/O — a database write, an HTTP call, even time.sleep() — the WebSocket loop cannot receive the next message until process() completes. Under high-frequency data, this creates a backlog that grows faster than you can drain it.
The root cause is not Python's speed. It's Python's willingness to wait.
The Async Mental Model: Cooperative Multitasking for I/O-Bound Work
asyncio is not threading. There is no preemption, no context-switching overhead, no need to protect shared state with locks.
Instead, you write coroutines — functions that can pause execution at await points and yield control back to the event loop. The event loop then runs other ready coroutines while one is waiting.
Think of it like a restaurant kitchen. A synchronous kitchen has one cook who starts a steak, then stands there staring at the oven until it's done. An async kitchen has one cook who starts the steak, then while the oven is doing its work, the cook prepps the salad, checks the fish, and sets the table. The steak finishes when it finishes. The cook was never idle.
The event loop is your kitchen. Your coroutines are tasks the cook can pick up and put down.
Architecture: Building the Async Pipeline
Our target architecture for a live quant system:
TickDB WebSocket (depth + trades) ──┐
TickDB REST /kline (OHLCV fetch) ───┼── asyncio event loop ── Strategy Engine ── Order Router
Scheduled reconnection tasks ───────┘
The event loop coordinates three concurrent workloads:
- WebSocket listener: Receives real-time depth and trade data. Never blocks.
- REST poller: Periodically fetches the latest kline candle via
/v1/market/kline/latest. Usesasyncio.gather()for batch requests. - Reconnection watchdog: Detects stale heartbeats and triggers reconnect with exponential backoff.
All three run concurrently within a single thread. No locks. No race conditions (assuming you use an asyncio.Queue for inter-task communication).
Production-Grade Async WebSocket Client
This is the core of the system. Study it carefully — this is what goes into production.
import asyncio
import aiohttp
import json
import logging
import random
import time
import os
from typing import Optional
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(message)s"
)
logger = logging.getLogger(__name__)
class TickDBWebSocketClient:
"""
Production-grade async WebSocket client for TickDB market data.
Features:
- Automatic reconnection with exponential backoff + jitter
- Heartbeat monitoring (connection health check)
- Rate-limit handling via Retry-After header
- Graceful shutdown
- Thread-safe message queue for downstream consumers
⚠️ This implementation uses aiohttp for async HTTP and websockets.
For HFT workloads (< 1 ms latency requirements), consider
a C++ or Rust-based adapter with Python FFI.
"""
def __init__(
self,
api_key: str,
symbols: list[str],
channels: list[str] = None,
base_url: str = "wss://api.tickdb.ai/ws",
heartbeat_interval: float = 30.0,
max_retries: int = 10,
base_delay: float = 1.0,
max_delay: float = 60.0,
):
self.api_key = api_key
self.symbols = symbols
self.channels = channels or ["trades", "depth"]
self.base_url = base_url
self.heartbeat_interval = heartbeat_interval
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self._ws: Optional[aiohttp.ClientWebSocketResponse] = None
self._session: Optional[aiohttp.ClientSession] = None
self._running = False
self._retry_count = 0
self._last_heartbeat: float = time.time()
# Thread-safe queue for downstream consumers
self.message_queue: asyncio.Queue = asyncio.Queue(maxsize=10000)
async def connect(self):
"""
Establish WebSocket connection with authentication.
TickDB uses URL parameter authentication for WebSocket:
wss://api.tickdb.ai/ws?api_key=YOUR_KEY
"""
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession()
url = f"{self.base_url}?api_key={self.api_key}"
try:
self._ws = await self._session.ws_connect(
url,
heartbeat=self.heartbeat_interval,
receive_timeout=self.heartbeat_interval * 2
)
self._running = True
self._retry_count = 0
self._last_heartbeat = time.time()
logger.info(f"Connected to TickDB WebSocket for {len(self.symbols)} symbols")
# Subscribe to channels and symbols
await self._subscribe()
except aiohttp.WSServerHandshakeError as e:
if e.status == 401:
raise ValueError(
"Authentication failed. Verify TICKDB_API_KEY is valid. "
"Check your dashboard at tickdb.ai"
)
raise
async def _subscribe(self):
"""Subscribe to specified channels and symbols."""
subscribe_msg = {
"cmd": "subscribe",
"channels": self.channels,
"symbols": self.symbols
}
await self._ws.send_json(subscribe_msg)
logger.info(f"Subscribed to channels: {self.channels} for symbols: {self.symbols}")
async def _reconnect(self):
"""
Reconnection logic with exponential backoff and jitter.
Exponential backoff: delay = base_delay * (2 ** retry_count)
Jitter: add random.uniform(0, delay * 0.1) to prevent thundering herd
"""
if self._retry_count >= self.max_retries:
logger.error(f"Max retries ({self.max_retries}) exceeded. Giving up.")
self._running = False
return
delay = min(self.base_delay * (2 ** self._retry_count), self.max_delay)
# Add jitter to prevent thundering herd when multiple clients reconnect simultaneously
jitter = random.uniform(0, delay * 0.1)
total_delay = delay + jitter
self._retry_count += 1
logger.warning(
f"Connection lost. Reconnecting in {total_delay:.2f}s "
f"(attempt {self._retry_count}/{self.max_retries})"
)
await asyncio.sleep(total_delay)
try:
await self.connect()
except Exception as e:
logger.error(f"Reconnection failed: {e}")
await self._reconnect()
async def _heartbeat_monitor(self):
"""
Background task: monitors connection health.
If no message received within 2 * heartbeat_interval, the connection
is considered stale and a reconnect is triggered.
"""
while self._running:
await asyncio.sleep(self.heartbeat_interval)
if self._running and self._ws:
elapsed = time.time() - self._last_heartbeat
if elapsed > self.heartbeat_interval * 2:
logger.warning(
f"Heartbeat stale ({elapsed:.1f}s since last message). "
"Triggering reconnect."
)
self._running = False
asyncio.create_task(self._reconnect())
break
async def _message_handler(self, msg: aiohttp.WServerMessage):
"""
Process incoming WebSocket messages.
Handles:
- text messages: parse JSON, put in queue
- pong: update heartbeat timestamp
- close: initiate graceful shutdown
- error: trigger reconnection
"""
if msg.type == aiohttp.WSMsgType.TEXT:
try:
data = json.loads(msg.data)
self._last_heartbeat = time.time()
# Rate-limit handling
if data.get("code") == 3001:
retry_after = int(data.get("retry_after", 5))
logger.warning(f"Rate limited. Waiting {retry_after}s.")
await asyncio.sleep(retry_after)
return
# Put message in queue for downstream consumers
# If queue is full, we drop the oldest message rather than block
try:
self.message_queue.put_nowait(data)
except asyncio.QueueFull:
logger.warning("Message queue full. Dropping oldest message.")
await self.message_queue.get() # Remove oldest
await self.message_queue.put(data)
except json.JSONDecodeError:
logger.error(f"Invalid JSON received: {msg.data[:100]}")
elif msg.type == aiohttp.WSMsgType.PONG:
self._last_heartbeat = time.time()
elif msg.type == aiohttp.WSMsgType.CLOSED:
logger.warning("WebSocket closed by server.")
self._running = False
asyncio.create_task(self._reconnect())
elif msg.type == aiohttp.WSMsgType.ERROR:
logger.error(f"WebSocket error: {msg.data}")
self._running = False
asyncio.create_task(self._reconnect())
async def listen(self):
"""
Main listener loop. Runs until shutdown.
Coordinates:
- WebSocket message consumption
- Heartbeat monitoring
"""
heartbeat_task = asyncio.create_task(self._heartbeat_monitor())
async for msg in self._ws:
if not self._running:
break
await self._message_handler(msg)
heartbeat_task.cancel()
try:
await heartbeat_task
except asyncio.CancelledError:
pass
async def shutdown(self):
"""Graceful shutdown: close WebSocket and session."""
logger.info("Shutting down TickDB WebSocket client.")
self._running = False
if self._ws and not self._ws.closed:
await self._ws.close()
if self._session and not self._session.closed:
await self._session.close()
logger.info("Shutdown complete.")
Key Engineering Decisions Explained
Heartbeat monitoring: The _heartbeat_monitor() task runs independently. If the connection goes silent (no messages for 2 * heartbeat_interval seconds), it triggers a reconnect. This handles silent disconnections where the server does not send a close frame.
Jitter in reconnection: random.uniform(0, delay * 0.1) adds randomness to the backoff delay. Without jitter, 1,000 clients reconnecting simultaneously after an outage will all retry at the same moment, potentially overwhelming the server again. Jitter spreads the load.
Queue overflow handling: When message_queue.put_nowait() raises QueueFull, we drop the oldest message rather than block. In a trading system, stale data is dangerous — receiving a delayed tick at the wrong time can trigger an unwanted order. Better to miss one tick than process it late.
Concurrent REST Fetches with asyncio.gather
The WebSocket gives you real-time data. But you also need historical OHLCV for your strategy — and you need it without blocking the WebSocket listener.
import aiohttp
import asyncio
async def fetch_kline(
session: aiohttp.ClientSession,
symbol: str,
interval: str = "1m",
limit: int = 100,
api_key: str = None
) -> dict:
"""
Fetch kline (OHLCV) data from TickDB REST API.
Endpoint: GET /v1/market/kline
Note: For live current candle, use /v1/market/kline/latest instead.
⚠️ This function is I/O-bound (network wait) but NOT CPU-bound.
asyncio.gather() handles concurrency perfectly here. Do not use
multiprocessing or threading for this workload.
"""
url = "https://api.tickdb.ai/v1/market/kline"
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
headers = {"X-API-Key": api_key}
try:
# timeout=(connect_timeout, read_timeout) — both required in production
async with session.get(
url,
params=params,
headers=headers,
timeout=aiohttp.ClientTimeout(total=10, connect=3.05)
) as response:
if response.status == 200:
data = await response.json()
return {"symbol": symbol, "data": data.get("data", [])}
elif response.status == 404:
raise KeyError(f"Symbol {symbol} not found. Verify via /v1/symbols/available")
else:
text = await response.text()
raise RuntimeError(f"API error {response.status}: {text}")
except asyncio.TimeoutError:
raise TimeoutError(f"Request timeout for {symbol} after 10s")
except aiohttp.ClientError as e:
raise RuntimeError(f"Network error fetching {symbol}: {e}")
async def fetch_multiple_klines(
symbols: list[str],
interval: str = "1m",
limit: int = 100,
api_key: str = None,
max_concurrency: int = 10
) -> list[dict]:
"""
Fetch kline data for multiple symbols concurrently.
Uses asyncio.Semaphore to cap concurrency and avoid overwhelming
the API with 100+ simultaneous requests.
Performance comparison (50 symbols, ~80ms per request):
- Sequential: 50 * 80ms = 4,000ms (4 seconds)
- Concurrent (unlimited): ~80ms (single batch)
- Concurrent (semaphore=10): 5 * 80ms = 400ms (0.4 seconds)
"""
semaphore = asyncio.Semaphore(max_concurrency)
async def fetch_with_semaphore(session, symbol):
async with semaphore:
return await fetch_kline(session, symbol, interval, limit, api_key)
connector = aiohttp.TCPConnector(limit=100, limit_per_host=10)
async with aiohttp.ClientSession(connector=connector) as session:
tasks = [
fetch_with_semaphore(session, symbol)
for symbol in symbols
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Filter out exceptions, log them
valid_results = []
for result in results:
if isinstance(result, Exception):
logger.error(f"Failed to fetch: {result}")
else:
valid_results.append(result)
return valid_results
# Example usage
async def main():
api_key = os.environ.get("TICKDB_API_KEY")
symbols = ["NVDA.US", "AAPL.US", "TSLA.US", "MSFT.US", "META.US"]
results = await fetch_multiple_klines(
symbols=symbols,
interval="1m",
limit=100,
api_key=api_key
)
for r in results:
print(f"{r['symbol']}: {len(r['data'])} candles")
if __name__ == "__main__":
asyncio.run(main())
The performance difference is stark:
| Approach | 50 symbols | 200 symbols |
|---|---|---|
Sequential requests |
4,000 ms | 16,000 ms |
asyncio.gather (unlimited) |
~80 ms | ~80 ms |
asyncio.gather (semaphore=10) |
~400 ms | ~1,600 ms |
The semaphore caps concurrent connections to avoid hitting the API's rate limits while still achieving massive parallelism.
Order Book Depth Monitor: Real-Time Pressure Ratio
Now we wire everything together. The following strategy class consumes depth data from the WebSocket queue and computes the buy/sell pressure ratio in real time.
import asyncio
from dataclasses import dataclass
from typing import Optional
from collections import deque
@dataclass
class DepthSnapshot:
"""Represents a snapshot of the order book at a point in time."""
symbol: str
timestamp: int # milliseconds since epoch
bids: list[tuple[float, float]] # [(price, size), ...]
asks: list[tuple[float, float]] # [(price, size), ...]
@property
def spread(self) -> float:
if not self.asks or not self.bids:
return 0.0
return self.asks[0][0] - self.bids[0][0]
@property
def mid_price(self) -> float:
if not self.asks or not self.bids:
return 0.0
return (self.asks[0][0] + self.bids[0][0]) / 2
class PressureRatioMonitor:
"""
Monitors order book depth and computes buy/sell pressure ratio.
Buy/sell pressure ratio = Σ(bid sizes, top N levels) / Σ(ask sizes, top N levels)
Interpretation:
- Ratio > 1.5: Strong buying pressure (bullish)
- Ratio < 0.67: Strong selling pressure (bearish)
- Ratio oscillating near 1: Equilibrium, range-bound market
⚠️ This is a simplified signal. In production, consider:
- Volume-weighted levels (size * depth)
- Time-decay weighting for stale levels
- Cross-symbol correlation (sector rotation signals)
"""
def __init__(self, symbol: str, levels: int = 5):
self.symbol = symbol
self.levels = levels
self.history: deque[DepthSnapshot] = deque(maxlen=100)
def compute_pressure_ratio(self, snapshot: DepthSnapshot) -> float:
"""Compute buy/sell pressure ratio from depth snapshot."""
bid_volume = sum(size for _, size in snapshot.bids[:self.levels])
ask_volume = sum(size for _, size in snapshot.asks[:self.levels])
if ask_volume == 0:
return float('inf')
return bid_volume / ask_volume
def parse_depth_message(self, msg: dict) -> Optional[DepthSnapshot]:
"""Parse TickDB depth channel message into a DepthSnapshot."""
if msg.get("channel") != "depth":
return None
data = msg.get("data", {})
if not data:
return None
return DepthSnapshot(
symbol=msg.get("symbol", self.symbol),
timestamp=data.get("ts", 0),
bids=data.get("bids", [])[:self.levels],
asks=data.get("asks", [])[:self.levels]
)
async def run(self, message_queue: asyncio.Queue):
"""
Main monitoring loop. Consumes messages from the queue and emits signals.
This runs concurrently with the WebSocket listener — no blocking, no locks.
"""
logger.info(f"Pressure ratio monitor started for {self.symbol}")
while True:
msg = await message_queue.get()
snapshot = self.parse_depth_message(msg)
if snapshot is None:
continue
self.history.append(snapshot)
ratio = self.compute_pressure_ratio(snapshot)
# Signal interpretation
if ratio > 1.5:
signal = "🟢 BUYING PRESSURE"
elif ratio < 0.67:
signal = "🔴 SELLING PRESSURE"
else:
signal = "⚪ NEUTRAL"
logger.info(
f"{snapshot.symbol} | Spread: ${snapshot.spread:.4f} | "
f"Pressure ratio: {ratio:.2f} | {signal}"
)
async def run_strategy():
"""Wire together WebSocket client and strategy monitor."""
api_key = os.environ.get("TICKDB_API_KEY")
client = TickDBWebSocketClient(
api_key=api_key,
symbols=["NVDA.US"],
channels=["depth"],
heartbeat_interval=30.0
)
monitor = PressureRatioMonitor(symbol="NVDA.US", levels=5)
# Start both tasks concurrently
# asyncio.gather runs them in the same event loop — true concurrency
await asyncio.gather(
client.listen(),
monitor.run(client.message_queue)
)
if __name__ == "__main__":
asyncio.run(run_strategy())
Performance Benchmark: Sync vs. Async
To demonstrate the real-world impact, here is a benchmark comparing synchronous and asynchronous market data fetching for 20 symbols:
import asyncio
import time
import requests
import aiohttp
def benchmark_sync(symbols: list[str], api_key: str) -> float:
"""Sequential synchronous fetch."""
start = time.perf_counter()
for symbol in symbols:
response = requests.get(
"https://api.tickdb.ai/v1/market/kline",
headers={"X-API-Key": api_key},
params={"symbol": symbol, "interval": "1m", "limit": 100},
timeout=10
)
response.json()
return time.perf_counter() - start
async def benchmark_async(symbols: list[str], api_key: str) -> float:
"""Concurrent async fetch using asyncio.gather."""
async def fetch(session, symbol):
async with session.get(
"https://api.tickdb.ai/v1/market/kline",
headers={"X-API-Key": api_key},
params={"symbol": symbol, "interval": "1m", "limit": 100},
timeout=aiohttp.ClientTimeout(total=10)
) as resp:
return await resp.json()
start = time.perf_counter()
async with aiohttp.ClientSession() as session:
tasks = [fetch(session, s) for s in symbols]
await asyncio.gather(*tasks)
return time.perf_counter() - start
async def main():
api_key = os.environ.get("TICKDB_API_KEY")
symbols = [f"{ticker}.US" for ticker in ["AAPL", "MSFT", "GOOGL", "AMZN", "NVDA",
"META", "TSLA", "BRK.B", "LLY", "AVGO",
" JPM", "V", "XOM", "UNH", "JNJ",
"PG", "MA", "HD", "CVX", "MRK"]]
# Run sync benchmark
sync_time = benchmark_sync(symbols, api_key)
print(f"Synchronous: {sync_time:.2f}s")
# Run async benchmark
async_time = await benchmark_async(symbols, api_key)
print(f"Asynchronous: {async_time:.2f}s")
speedup = sync_time / async_time
print(f"Speedup: {speedup:.1f}x faster with asyncio")
if __name__ == "__main__":
asyncio.run(main())
Typical results on a local machine with 50 ms network latency:
| Approach | 20 symbols | 100 symbols |
|---|---|---|
| Synchronous | ~1,000 ms | ~5,000 ms |
| Asyncio (unlimited) | ~52 ms | ~54 ms |
| Speedup | ~19x | ~93x |
The speedup approaches N (number of symbols) when the network is the bottleneck. For I/O-bound workloads — and market data processing is almost entirely I/O-bound — asyncio is not optional. It is the only sane architecture.
Common Pitfalls and How to Avoid Them
Pitfall 1: Blocking the Event Loop
Never call synchronous blocking code inside an async function:
# ❌ WRONG — blocks the entire event loop
async def bad_example():
result = requests.get(url) # Blocking I/O
time.sleep(5) # Blocks for 5 seconds
return result
# ✅ CORRECT — use aiohttp for HTTP, asyncio.sleep for delays
async def good_example():
async with aiohttp.ClientSession() as session:
result = await session.get(url)
await asyncio.sleep(5) # Yields control during the wait
return result
If you must use a synchronous library (e.g., psycopg2 for PostgreSQL), run it in a thread pool:
import asyncio
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=4)
async def fetch_with_sync_library():
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(executor, sync_database_query)
return result
Pitfall 2: Forgetting to Await
# ❌ WRONG — task is created but never awaited, silently ignored
async def bad():
asyncio.create_task(some_coroutine()) # Fire and forget — usually a bug
# ✅ CORRECT — await the task or store it for later cleanup
async def good():
task = asyncio.create_task(some_coroutine())
await task # Wait for completion
Pitfall 3: Shared Mutable State Without Locks
asyncio coroutines run in a single thread — there is no preemption. However, if you mix async with regular threads, or use non-async libraries that perform I/O internally, you can still encounter race conditions.
For thread-safe inter-task communication, use asyncio.Queue:
# ✅ CORRECT — asyncio.Queue is safe for inter-task communication
queue: asyncio.Queue = asyncio.Queue()
async def producer():
for i in range(100):
await queue.put(i)
async def consumer():
while True:
item = await queue.get()
process(item)
queue.task_done()
await asyncio.gather(producer(), consumer())
Deployment Guide by Scale
| Deployment scenario | Recommendation |
|---|---|
| Individual quant / single strategy | Single async process, asyncio.run(), no multiprocessing. Use screen or tmux for process management. |
| Multiple independent strategies | One event loop per strategy process. Use process supervision (systemd, supervisord) to manage lifecycle. |
| Institutional / multiple strategies sharing data | Separate ingest process (async WebSocket → shared asyncio.Queue in multiprocessing manager) from strategy processes. Add Redis or a similar message broker for cross-process communication. |
| AI-assisted workflow | The tickdb-market-data SKILL on ClawHub provides an async context manager that wraps the WebSocket client. Install it in your AI coding environment for one-line data ingestion in agentic pipelines. |
The Async Contract: What You Gain and What You Trade
asyncio delivers near-linear speedup for I/O-bound workloads. For a market data system, fetching 100 symbols that each take 80 ms now costs 80 ms instead of 8 seconds. Your WebSocket listener will never miss a tick because your strategy is computing. Your reconnection logic is automatic and resilient.
The tradeoffs:
- Learning curve:
await,async def, and the event loop model require a mental shift from sequential thinking. - Debugging complexity: Stack traces become deeper. Use
asyncio.run()withuvloopfor better performance and debugging tools. - Not for CPU-bound work: If your strategy involves heavy numerical computation (e.g., large matrix operations, intensive numerical optimization), asyncio will not help. Use
numpy,numba, ormultiprocessingfor that.
For market data ingestion, order book monitoring, and real-time signal generation — all of which are I/O-bound — asyncio is the correct architecture. It is not a trick. It is not premature optimization. It is the standard tool of production quant systems handling live data streams.
Next Steps
If you're building your first async system, start with the WebSocket client code above. Replace the process() stub with your actual signal logic, add logging, and run it against a paper-trading account.
If you want to run this strategy yourself:
- Sign up at tickdb.ai (free, no credit card required)
- Generate an API key in the dashboard
- Set the
TICKDB_API_KEYenvironment variable, then copy-paste the code from this article
If you need 10+ years of historical OHLCV data for strategy backtesting, reach out to enterprise@tickdb.ai for Professional / Enterprise plans with extended history.
If you use AI coding assistants, search for and install the tickdb-market-data SKILL in your AI tool's marketplace for pre-built async ingestion templates.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results. Async programming techniques described here are engineering tools — their application to trading strategies requires independent validation and risk management.