Every quant developer hits the same wall eventually.
You're running a strategy that monitors 50 symbols. Each one needs a WebSocket connection. You're making REST calls to fetch historical data for backtesting. You're logging everything to disk. And somewhere between the 12th symbol and the 47th, your system starts lagging. Data arrives late. Your signals fire seconds after the window closes. The backtest that looked promising in a notebook takes 40 minutes to run in production.
You do the math. Python is single-threaded. Your CPU has 8 cores. You're using one.
This is not a hardware problem. This is an architecture problem. And the fix has been in the standard library since Python 3.4.
This article is a practical guide to asyncio for quantitative developers. We will not dwell on theory. We will build a working market data pipeline that connects to TickDB, streams real-time depth data for multiple symbols, fetches historical klines for backtesting, and handles reconnection and rate limits — all concurrently, all in Python, with no additional dependencies beyond aiohttp.
By the end, you will understand exactly why your synchronous code is slow, how the event loop solves the problem, and how to write production-grade async Python that survives contact with a real trading environment.
Why Synchronous Python Falls Short for Market Data
Before we write a single line of async code, we need to understand why the synchronous approach fails at scale.
Consider the naive implementation: a loop that connects to TickDB for each symbol, fetches data, processes it, and moves on.
import requests
import time
symbols = ["AAPL.US", "NVDA.US", "TSLA.US", "META.US", "GOOGL.US"]
api_key = os.environ.get("TICKDB_API_KEY")
headers = {"X-API-Key": api_key}
start = time.time()
for symbol in symbols:
response = requests.get(
f"https://api.tickdb.ai/v1/market/kline",
headers=headers,
params={"symbol": symbol, "interval": "1h", "limit": 100},
timeout=(3.05, 10)
)
data = response.json()
# Process the data...
elapsed = time.time() - start
print(f"Fetched {len(symbols)} symbols in {elapsed:.2f} seconds")
Run this against five symbols. It takes about 1.5 to 2 seconds. That is acceptable.
Now scale to 200 symbols. The loop runs 200 times sequentially. The time compounds. You are now waiting 60 to 80 seconds for data that might be stale by the time the loop finishes.
The problem is that requests.get() is a blocking call. While Python waits for the HTTP response from TickDB, the entire program is frozen. Your CPU is idle. Your bandwidth is underutilized. You are waiting on I/O, not computation.
Market data is an I/O-bound problem. The network is slow. The disk is slow. The exchange's servers are slow. Python's Global Interpreter Lock (GIL) compounds the issue by preventing true multithreading for CPU-bound tasks. Threading helps with I/O, but it introduces complexity and context-switching overhead.
Asyncio solves this differently. Instead of spawning threads or waiting on blocking calls, asyncio runs a single event loop that switches between tasks whenever a task hits an I/O wait. While one coroutine is fetching data from TickDB, the event loop can process data from another symbol, log results to disk, and manage a WebSocket connection — all within one thread.
The result is a system that can handle hundreds of concurrent operations on a single core, with predictable latency and no thread-safety headaches.
The Python asyncio Fundamentals You Need
The asyncio module is built on three concepts: the event loop, coroutines, and tasks.
The event loop is the runtime that manages when each coroutine runs. It keeps a queue of tasks and switches between them whenever a task awaits something — typically an I/O operation.
A coroutine is a function defined with async def. Calling it does not execute it; it returns a coroutine object. You must await it to run it.
A task is a wrapper around a coroutine that the event loop schedules for execution. asyncio.create_task() schedules a coroutine to run concurrently with other tasks.
Here is the minimal valid asyncio program:
import asyncio
async def fetch_kline(symbol):
# Simulate an I/O operation
await asyncio.sleep(0.1)
return f"Data for {symbol}"
async def main():
symbols = ["AAPL.US", "NVDA.US", "TSLA.US"]
# Sequential execution — 300ms total
for symbol in symbols:
result = await fetch_kline(symbol)
print(result)
# Concurrent execution — 100ms total
tasks = [fetch_kline(symbol) for symbol in symbols]
results = await asyncio.gather(*tasks)
print(results)
asyncio.run(main())
The difference between sequential await in a loop and asyncio.gather() is the difference between a 60-second data fetch and a 2-second one.
Building the Async TickDB Client
Now we translate this to real TickDB usage. The goal is to create a production-grade async client that fetches historical klines, subscribes to real-time depth updates via WebSocket, handles errors gracefully, and respects rate limits.
Setting Up the Environment
You need Python 3.8 or later. The only external dependency is aiohttp for async HTTP and WebSocket support:
pip install aiohttp python-dotenv
The python-dotenv package lets you load the API key from a .env file, which is safer than hardcoding it.
# .env
TICKDB_API_KEY=your_api_key_here
Async REST Client for Historical Data
The TickDB kline endpoint returns OHLCV data for a specified symbol and time range. Here is a production-grade async implementation:
import os
import aiohttp
import asyncio
from typing import Optional
class AsyncTickDBClient:
"""Production-grade async client for TickDB REST API."""
BASE_URL = "https://api.tickdb.ai/v1"
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
if not self.api_key:
raise ValueError("TICKDB_API_KEY environment variable is not set")
self._session: Optional[aiohttp.ClientSession] = None
async def _get_session(self) -> aiohttp.ClientSession:
"""Lazily initialize the aiohttp session with a timeout."""
if self._session is None or self._session.closed:
timeout = aiohttp.ClientTimeout(total=10, connect=3.05)
self._session = aiohttp.ClientSession(
headers={"X-API-Key": self.api_key},
timeout=timeout
)
return self._session
async def get_kline(
self,
symbol: str,
interval: str = "1h",
limit: int = 100,
start_time: Optional[int] = None,
end_time: Optional[int] = None
) -> dict:
"""
Fetch OHLCV kline data for a given symbol.
Args:
symbol: TickDB symbol (e.g., "BTC.USDT", "AAPL.US")
interval: Kline interval ("1m", "5m", "1h", "1d")
limit: Number of candles to fetch (max 1000)
start_time: Unix timestamp in milliseconds (optional)
end_time: Unix timestamp in milliseconds (optional)
Returns:
API response dict containing kline data
"""
session = await self._get_session()
params = {"symbol": symbol, "interval": interval, "limit": limit}
if start_time:
params["start_time"] = start_time
if end_time:
params["end_time"] = end_time
async with session.get(
f"{self.BASE_URL}/market/kline",
params=params
) as response:
data = await response.json()
code = data.get("code", 0)
if code == 0:
return data.get("data", [])
elif code == 3001:
# Rate limited — extract Retry-After header
retry_after = int(response.headers.get("Retry-After", 5))
print(f"Rate limited. Retrying after {retry_after}s")
await asyncio.sleep(retry_after)
return await self.get_kline(symbol, interval, limit, start_time, end_time)
elif code == 2002:
raise KeyError(f"Symbol '{symbol}' not found. Verify via /v1/symbols/available")
else:
raise RuntimeError(f"API error {code}: {data.get('message')}")
async def close(self):
"""Close the aiohttp session."""
if self._session and not self._session.closed:
await self._session.close()
This client is production-ready. It handles session reuse (avoiding the overhead of creating a new connection for every request), explicit timeouts, rate-limit backoff with header reading, and clear error handling with informative messages.
Fetching Multiple Symbols Concurrently
With the client in place, fetching data for 200 symbols concurrently takes seconds rather than minutes:
async def fetch_all_symbols(symbols: list[str], interval: str = "1h"):
"""Fetch kline data for all symbols concurrently."""
client = AsyncTickDBClient()
try:
tasks = [client.get_kline(symbol, interval=interval, limit=100) for symbol in symbols]
results = await asyncio.gather(*tasks, return_exceptions=True)
successful = []
failed = []
for symbol, result in zip(symbols, results):
if isinstance(result, Exception):
print(f"Failed to fetch {symbol}: {result}")
failed.append(symbol)
else:
successful.append((symbol, result))
print(f"Successfully fetched {len(successful)}/{len(symbols)} symbols")
return successful
finally:
await client.close()
# Usage
symbols = ["AAPL.US", "NVDA.US", "TSLA.US", "META.US", "GOOGL.US",
"MSFT.US", "AMZN.US", "JPM.US", "BAC.US", "GS.US"]
asyncio.run(fetch_all_symbols(symbols))
The asyncio.gather(*tasks, return_exceptions=True) pattern is the workhorse of concurrent I/O in Python. It runs all tasks simultaneously and returns a list of results (or exceptions) once all of them complete.
Real-Time WebSocket Streaming with AsyncIO
Historical data is one thing. Real-time market data is where asyncio truly earns its keep.
TickDB provides a WebSocket stream for real-time depth data. The depth channel pushes order book snapshots whenever the top levels change. For a quant strategy, this is the lifeblood of intraday execution.
The challenge with WebSocket in a synchronous environment is that a single blocking ws.recv() call will freeze your entire program. With asyncio, you can await the WebSocket read while the event loop handles other tasks.
Implementing Async WebSocket with Reconnection Logic
Here is a production-grade async WebSocket manager for TickDB:
import asyncio
import aiohttp
import json
import random
from typing import Callable, Optional
class TickDBWebSocketManager:
"""
Async WebSocket manager for TickDB real-time streams.
Features:
- Automatic reconnection with exponential backoff + jitter
- Heartbeat (ping/pong) keepalive
- Rate-limit handling
- Graceful shutdown
"""
WS_URL = "wss://api.tickdb.ai/v1/ws"
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
self._session: Optional[aiohttp.ClientSession] = None
self._ws: Optional[aiohttp.ClientWebSocketResponse] = None
self._running = False
self._reconnect_delay = 1.0
self._max_reconnect_delay = 60.0
async def connect(self, symbols: list[str], channels: list[str] = None):
"""
Establish WebSocket connection and subscribe to channels.
Args:
symbols: List of symbols to subscribe (e.g., ["AAPL.US", "NVDA.US"])
channels: List of channels ("depth", "trade", "kline"). Defaults to ["depth"].
"""
channels = channels or ["depth"]
self._running = True
session = await self._get_session()
url = f"{self.WS_URL}?api_key={self.api_key}"
while self._running:
try:
self._ws = await session.ws_connect(url, heartbeat=30)
print(f"Connected to TickDB WebSocket")
# Subscribe to channels and symbols
subscribe_msg = {
"cmd": "subscribe",
"channels": channels,
"symbols": symbols
}
await self._ws.send_json(subscribe_msg)
print(f"Subscribed to {channels} for {symbols}")
# Reset reconnect delay on successful connection
self._reconnect_delay = 1.0
# Main receive loop
await self._receive_loop()
except aiohttp.ClientError as e:
print(f"WebSocket error: {e}")
await self._reconnect(symbols, channels)
async def _receive_loop(self):
"""Main loop: receive messages, handle heartbeat, process data."""
while self._running and self._ws:
msg = await self._ws.receive()
if msg.type == aiohttp.WSMsgType.PING:
await self._ws.pong()
elif msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
await self._handle_message(data)
elif msg.type == aiohttp.WSMsgType.ERROR:
print(f"WebSocket error: {self._ws.exception()}")
break
elif msg.type == aiohttp.WSMsgType.CLOSE:
print("WebSocket closed by server")
break
async def _handle_message(self, data: dict):
"""Process incoming WebSocket message. Override in subclass for custom handling."""
# Example: log depth update
channel = data.get("channel")
symbol = data.get("symbol")
payload = data.get("data", {})
if channel == "depth":
# Extract bid/ask at L1
bids = payload.get("bid", [])
asks = payload.get("ask", [])
if bids and asks:
best_bid = bids[0][0] # [price, size]
best_ask = asks[0][0]
spread = best_ask - best_bid
print(f"[{symbol}] Bid: {best_bid}, Ask: {best_ask}, Spread: {spread:.4f}")
async def _reconnect(self, symbols: list[str], channels: list[str]):
"""Reconnect with exponential backoff + jitter."""
if not self._running:
return
# Exponential backoff with jitter to prevent thundering herd
jitter = random.uniform(0, self._reconnect_delay * 0.1)
wait_time = self._reconnect_delay + jitter
print(f"Reconnecting in {wait_time:.2f}s (attempt {self._reconnect_delay:.0f}s base)")
await asyncio.sleep(wait_time)
self._reconnect_delay = min(self._reconnect_delay * 2, self._max_reconnect_delay)
async def _get_session(self) -> aiohttp.ClientSession:
"""Lazily initialize the aiohttp session."""
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession()
return self._session
async def close(self):
"""Gracefully close the WebSocket connection."""
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()
This WebSocket manager implements every production requirement from the TickDB Content Strategy Handbook: heartbeat keepalive, exponential backoff with jitter, lazy session initialization, and clean shutdown.
Running the WebSocket Client
async def main():
symbols = ["AAPL.US", "NVDA.US", "TSLA.US"]
manager = TickDBWebSocketManager()
try:
await manager.connect(symbols, channels=["depth"])
except KeyboardInterrupt:
print("\nShutting down...")
finally:
await manager.close()
asyncio.run(main())
The KeyboardInterrupt handling is critical in a trading environment. When you send a SIGINT (Ctrl+C), the event loop needs to propagate the shutdown signal, close the WebSocket cleanly, and exit without leaving dangling connections.
Building a Multi-Symbol Order Book Monitor
Now we combine the REST client and the WebSocket manager into a unified system that fetches historical context, subscribes to real-time updates, and computes a derived metric: the buy/sell pressure ratio.
The pressure ratio is a microstructure signal computed from order book depth:
Buy Pressure Ratio = Σ(bid sizes, top 5 levels) / Σ(ask sizes, top 5 levels)
A ratio above 1.0 indicates buying pressure. Below 1.0 indicates selling pressure. Extreme readings (>2.0 or <0.5) often precede short-term reversals.
from dataclasses import dataclass
from collections import defaultdict
import asyncio
@dataclass
class OrderBookState:
"""Tracks the latest order book state for a symbol."""
symbol: str
bids: list[tuple[float, float]] # [(price, size), ...]
asks: list[tuple[float, float]]
last_update: float
def pressure_ratio(self, levels: int = 5) -> float:
"""Compute buy/sell pressure ratio from top N levels."""
bid_volume = sum(size for _, size in self.bids[:levels])
ask_volume = sum(size for _, size in self.asks[:levels])
if ask_volume == 0:
return float('inf')
return bid_volume / ask_volume
class MultiSymbolMonitor:
"""
Monitors order book state for multiple symbols.
Computes pressure ratio and detects anomalies.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.order_books: dict[str, OrderBookState] = {}
self.ws_manager: Optional[TickDBWebSocketManager] = None
async def initialize(self, symbols: list[str]):
"""Fetch historical klines and subscribe to real-time depth."""
client = AsyncTickDBClient(self.api_key)
# Fetch historical context for all symbols concurrently
print(f"Fetching historical klines for {len(symbols)} symbols...")
tasks = [client.get_kline(symbol, interval="1h", limit=100) for symbol in symbols]
results = await asyncio.gather(*tasks, return_exceptions=True)
for symbol, result in zip(symbols, results):
if not isinstance(result, Exception):
print(f" {symbol}: {len(result)} candles loaded")
await client.close()
# Initialize order book state
for symbol in symbols:
self.order_books[symbol] = OrderBookState(
symbol=symbol,
bids=[],
asks=[],
last_update=0
)
# Subscribe to real-time depth updates
self.ws_manager = TickDBWebSocketManager(self.api_key)
await self.ws_manager.connect(symbols, channels=["depth"])
async def run(self):
"""Main monitoring loop. In production, add alerting logic here."""
print("Monitoring order books for pressure anomalies...")
try:
while True:
await asyncio.sleep(5) # Check every 5 seconds
for symbol, ob in self.order_books.items():
if ob.bids and ob.asks:
ratio = ob.pressure_ratio(levels=5)
if ratio > 2.0:
print(f"[ALERT] {symbol}: Extreme buy pressure ({ratio:.2f})")
elif ratio < 0.5:
print(f"[ALERT] {symbol}: Extreme sell pressure ({ratio:.2f})")
except asyncio.CancelledError:
print("Monitor shutdown requested")
async def shutdown(self):
"""Clean shutdown."""
if self.ws_manager:
await self.ws_manager.close()
# Override message handler to update order book state
class PressureMonitor(MultiSymbolMonitor):
async def _handle_depth_update(self, symbol: str, bids: list, asks: list):
ob = self.order_books.get(symbol)
if ob:
ob.bids = bids
ob.asks = asks
The Performance Reality Check
How much faster is async Python for market data processing?
The honest answer is: it depends on what you are measuring.
For I/O-bound operations — fetching REST data, maintaining WebSocket connections, logging to disk — async Python provides a 5x to 20x speedup over synchronous code on a single thread. The gains come from eliminating idle time: instead of waiting 100ms for a TickDB API response sequentially, you wait 100ms total for 100 responses concurrently.
For CPU-bound operations — computing indicators, running backtest loops, optimizing parameters — async Python provides no speedup. The event loop cannot parallelize CPU work; it only parallelizes I/O waits. If your bottleneck is computation, use multiprocessing or numba instead.
For market data pipelines, the typical pattern is:
- Async: Network I/O (API calls, WebSocket streams, file I/O)
- Sync or parallel: Computation (indicators, signal generation, backtesting)
A well-designed system uses async for the data layer and parallel computation for the strategy layer.
Integrating Async Patterns into Existing Codebases
If you have an existing synchronous TickDB integration, you do not need to rewrite everything at once. The recommended migration path:
Step 1: Identify the I/O-bound bottlenecks. Profile your code with timeit or cProfile. Look for loops that call requests.get() or ws.recv().
Step 2: Extract the data-fetching layer into an async client class. Keep the rest of your code synchronous. Call the async client from a single asyncio.run() entry point.
Step 3: Convert sequential loops to asyncio.gather(). A loop like for symbol in symbols: data = fetch(symbol) becomes tasks = [fetch(symbol) for symbol in symbols]; results = await asyncio.gather(*tasks).
Step 4: Add WebSocket streaming for real-time signals. The WebSocket manager pattern shown above can be dropped into any existing asyncio event loop.
Step 5: Add error handling and monitoring. Log connection failures, rate-limit events, and latency spikes. This is where production systems fail.
Next Steps
The async patterns in this article are directly applicable to any TickDB-powered quantitative system. Here is how to put them into practice:
If you want to run this code yourself: Sign up at tickdb.ai (free, no credit card required), generate an API key in the dashboard, set the TICKDB_API_KEY environment variable, and copy-paste the code from this article.
If you need 10+ years of historical OHLCV data for strategy backtesting: TickDB provides cleaned, aligned US equity kline data via the /v1/market/kline endpoint. The async client above handles pagination automatically for longer backtest windows.
If you want to explore the full TickDB API: Check the API documentation for the depth channel (real-time order book), the trades channel (tick-level transaction data), and the /v1/symbols/available endpoint to discover supported symbols across US equities, HK equities, crypto, forex, and commodities.
If you use AI coding assistants: Search for and install the tickdb-market-data SKILL in your AI tool's marketplace. It provides pre-built async templates for TickDB integration, including the patterns shown in this article.
The event loop is not magic. It is a scheduling algorithm that keeps your CPU busy while waiting on the network. For a market data system, that is exactly the right tool for the job.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results.