Introduction: Why Historical L2 Order Book Data Matters for Your Trading Strategy
As a quantitative researcher building a mean-reversion strategy on Binance Futures BTCUSDT perpetual contracts, I needed millisecond-precision order book snapshots to backtest my hypothesis properly. I quickly discovered that live WebSocket feeds give you real-time data, but reconstructing the exact market microstructure for historical periods requires a specialized data provider. After testing three different services and burning through weeks of development time, I landed on Tardis.dev as my primary market data source — and this tutorial shares everything I learned about extracting, processing, and replaying L2 order book data efficiently.
In this comprehensive guide, you'll learn how to:
- Authenticate with the Tardis.dev historical market data API
- Download Binance Futures L2 order book snapshots with nanosecond timestamps
- Parse and normalize the data using pandas for efficient analysis
- Replay order book states to simulate trading conditions
- Optimize your pipeline for large-scale backtesting workloads
Understanding L2 Order Book Data Structure
Level 2 (L2) order book data represents the full bid-ask ladder of a trading venue, showing every price level and its corresponding quantity. Unlike L1 data (best bid/ask only), L2 snapshots capture the complete market depth — critical for slippage estimation, market impact modeling, and microstructure analysis.
For Binance Futures, the order book message format includes:
- Timestamp: Server-side event time in milliseconds since epoch
- Update ID: Sequence number for order book modifications
- Bids: Array of [price, quantity] tuples sorted descending
- Asks: Array of [price, quantity] tuples sorted ascending
Prerequisites and Environment Setup
Before diving into the code, ensure you have Python 3.9+ installed with the following dependencies:
# Install required packages
pip install requests pandas numpy orjson aiohttp tqdm
Verify Python version
python --version
Should output: Python 3.9.0 or higher
I recommend creating a dedicated virtual environment for market data projects to avoid dependency conflicts:
# Create and activate virtual environment
python -m venv market-data-env
source market-data-env/bin/activate # Linux/macOS
market-data-env\Scripts\activate # Windows
Install all dependencies
pip install requests pandas numpy orjson aiohttp tqdmAuthentication: Obtaining Your Tardis.dev API Key
Tardis.dev requires API key authentication for all historical data requests. You can obtain credentials by signing up at their portal. The API uses Bearer token authentication, and pricing is based on data volume consumed.
import os
import requests
from typing import Optional
class TardisClient:
"""Client for interacting with Tardis.dev historical market data API."""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def get_available_exchanges(self) -> dict:
"""Retrieve list of supported exchanges."""
response = self.session.get(f"{self.BASE_URL}/exchanges")
response.raise_for_status()
return response.json()
def get_exchange_details(self, exchange: str) -> dict:
"""Get detailed information about a specific exchange."""
response = self.session.get(f"{self.BASE_URL}/exchanges/{exchange}")
response.raise_for_status()
return response.json()
def get_symbols(self, exchange: str) -> list:
"""List all available symbols for an exchange."""
response = self.session.get(
f"{self.BASE_URL}/exchanges/{exchange}/symbols"
)
response.raise_for_status()
return response.json()
Initialize client
TARDIS_API_KEY = os.environ.get("TARDIS_API_KEY", "your_tardis_api_key_here")
client = TardisClient(TARDIS_API_KEY)
Verify authentication
exchanges = client.get_available_exchanges()
print(f"Authenticated successfully. {len(exchanges)} exchanges available.")
print("Binance Futures available:", "binance-futures" in [e.get("id") for e in exchanges])Downloading Binance Futures L2 Order Book Data
Tardis.dev provides historical market data via a unified API that supports multiple exchanges. For Binance Futures, you can access L2 order book data with millisecond precision. Here's the complete implementation for downloading and parsing order book snapshots:
import json
import gzip
import struct
from datetime import datetime, timedelta
from typing import Generator, Tuple, List
import pandas as pd
import orjson
from tqdm import tqdm
class BinanceFuturesOrderBookDownloader:
"""Download and parse Binance Futures L2 order book data from Tardis.dev."""
BASE_URL = "https://api.tardis.dev/v1/feed"
def __init__(self, api_key: str):
self.api_key = api_key
def download_l2_orderbook(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
compression: str = "gzip"
) -> List[dict]:
"""
Download L2 order book data for a given symbol and date range.
Args:
symbol: Trading pair symbol (e.g., 'BTCUSDT' for perpetual)
start_date: Start of the date range
end_date: End of the date range
compression: Response compression ('gzip' or 'zstd')
Returns:
List of order book snapshots with timestamps
"""
url = f"{self.BASE_URL}/binance-futures:{symbol}"
params = {
"from": start_date.isoformat(),
"to": end_date.isoformat(),
"format": "json",
"compression": compression,
"symbols": symbol,
# Filter for order book snapshots only
"messageTypes": "order_book_snapshot"
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Accept-Encoding": compression
}
snapshots = []
response = requests.get(url, params=params, headers=headers, stream=True)
response.raise_for_status()
for line in response.iter_lines():
if line:
try:
data = orjson.loads(line)
if data.get("type") == "order_book_snapshot":
snapshots.append({
"timestamp": pd.to_datetime(data["timestamp"], unit="ms"),
"symbol": data["symbol"],
"bids": data.get("bids", []),
"asks": data.get("asks", []),
"update_id": data.get("updateId")
})
except Exception as e:
print(f"Error parsing line: {e}")
continue
return snapshots
Usage example
downloader = BinanceFuturesOrderBookDownloader(TARDIS_API_KEY)
Download 1 hour of BTCUSDT perpetual order book data
start_time = datetime(2024, 11, 15, 8, 0, 0)
end_time = datetime(2024, 11, 15, 9, 0, 0)
print(f"Downloading BTCUSDT order book from {start_time} to {end_time}...")
order_book_data = downloader.download_l2_orderbook(
symbol="BTCUSDT",
start_date=start_time,
end_date=end_time
)
print(f"Downloaded {len(order_book_data)} order book snapshots")
print(f"First snapshot: {order_book_data[0]['timestamp']}")
print(f"Last snapshot: {order_book_data[-1]['timestamp']}")Parsing and Normalizing Order Book Data with Pandas
Raw order book data from Tardis.dev needs processing before analysis. I'll show you how to convert the nested structure into efficient pandas DataFrames optimized for time-series analysis and backtesting:
import numpy as np
from dataclasses import dataclass
from typing import Dict, List
@dataclass
class OrderBookLevel:
"""Represents a single price level in the order book."""
price: float
quantity: float
class OrderBookParser:
"""Parse and normalize Binance Futures order book data."""
@staticmethod
def parse_to_dataframe(snapshots: List[dict]) -> pd.DataFrame:
"""
Convert order book snapshots to a flat DataFrame.
Each row represents one price level at one point in time,
making it easy to filter, aggregate, and analyze.
"""
records = []
for snapshot in snapshots:
timestamp = snapshot["timestamp"]
symbol = snapshot["symbol"]
# Parse bids (buy orders)
for price, qty in snapshot.get("bids", []):
records.append({
"timestamp": timestamp,
"symbol": symbol,
"side": "bid",
"price": float(price),
"quantity": float(qty),
"level": len([p for p, _ in snapshot["bids"] if p >= price])
})
# Parse asks (sell orders)
for price, qty in snapshot.get("asks", []):
records.append({
"timestamp": timestamp,
"symbol": symbol,
"side": "ask",
"price": float(price),
"quantity": float(qty),
"level": len([p for p, _ in snapshot["asks"] if p <= price])
})
df = pd.DataFrame(records)
df = df.set_index("timestamp")
df = df.sort_index()
return df
@staticmethod
def calculate_spread(df: pd.DataFrame) -> pd.DataFrame:
"""
Calculate bid-ask spread statistics over time.
Returns DataFrame with spread metrics at each snapshot.
"""
# Get best bid and ask for each timestamp
best_bid = df[df["side"] == "bid"].groupby("timestamp").apply(
lambda x: x.nlargest(1, "price")
).reset_index(drop=True)
best_ask = df[df["side"] == "ask"].groupby("timestamp").apply(
lambda x: x.nsmallest(1, "price")
).reset_index(drop=True)
spread_df = pd.DataFrame({
"timestamp": best_bid["timestamp"],
"best_bid": best_bid["price"].values,
"best_bid_qty": best_bid["quantity"].values,
"best_ask": best_ask["price"].values,
"best_ask_qty": best_ask["quantity"].values
})
spread_df["spread"] = spread_df["best_ask"] - spread_df["best_bid"]
spread_df["spread_pct"] = (spread_df["spread"] / spread_df["best_bid"]) * 100
spread_df["mid_price"] = (spread_df["best_bid"] + spread_df["best_ask"]) / 2
return spread_df.set_index("timestamp")
@staticmethod
def get_market_depth(df: pd.DataFrame, levels: int = 10) -> pd.DataFrame:
"""
Calculate cumulative market depth for top N levels.
Useful for understanding liquidity at different price distances
from the best bid/ask.
"""
depths = []
for timestamp in df.index.unique():
snapshot = df.loc[timestamp]
bids = snapshot[snapshot["side"] == "bid"].nlargest(levels, "price")
asks = snapshot[snapshot["side"] == "ask"].nsmallest(levels, "price")
record = {"timestamp": timestamp}
# Cumulative bid depth
record["bid_depth_1"] = bids.iloc[0]["quantity"] if len(bids) > 0 else 0
record["bid_depth_5"] = bids.head(5)["quantity"].sum() if len(bids) >= 5 else bids["quantity"].sum()
record["bid_depth_10"] = bids["quantity"].sum()
# Cumulative ask depth
record["ask_depth_1"] = asks.iloc[0]["quantity"] if len(asks) > 0 else 0
record["ask_depth_5"] = asks.head(5)["quantity"].sum() if len(asks) >= 5 else asks["quantity"].sum()
record["ask_depth_10"] = asks["quantity"].sum()
depths.append(record)
return pd.DataFrame(depths).set_index("timestamp")
Parse the downloaded data
parser = OrderBookParser()
orderbook_df = parser.parse_to_dataframe(order_book_data)
print(f"Parsed {len(orderbook_df):,} order book levels")
print(f"Time range: {orderbook_df.index.min()} to {orderbook_df.index.max()}")
print(f"\nSample data:")
print(orderbook_df.head(10))
Calculate spread metrics
spread_df = parser.calculate_spread(orderbook_df)
print(f"\nSpread statistics:")
print(spread_df[["spread", "spread_pct", "mid_price"]].describe())
Get market depth
depth_df = parser.get_market_depth(orderbook_df, levels=10)
print(f"\nMarket depth sample:")
print(depth_df.head())Replaying Order Book States for Backtesting
For strategy backtesting, you often need to simulate order book states at specific timestamps. Here's a replay engine that lets you query the order book at any point in your downloaded dataset:
from typing import Iterator, Optional
from bisect import bisect_right
import pandas as pd
class OrderBookReplay:
"""
Replay engine for order book data.
Allows random access to order book snapshots for backtesting
and strategy simulation.
"""
def __init__(self, snapshots: List[dict]):
"""
Initialize replay engine with order book snapshots.
Args:
snapshots: List of order book snapshot dictionaries
"""
self.snapshots = sorted(snapshots, key=lambda x: x["timestamp"])
self.timestamps = [s["timestamp"] for s in self.snapshots]
# Index bids and asks separately for efficient lookup
self.bid_prices = [s["bids"] for s in self.snapshots]
self.ask_prices = [s["asks"] for s in self.snapshots]
def get_snapshot(self, timestamp: pd.Timestamp) -> Optional[dict]:
"""
Get the order book snapshot at or immediately before a timestamp.
Args:
timestamp: Target timestamp
Returns:
Order book snapshot or None if outside range
"""
if timestamp < self.timestamps[0]:
return None
if timestamp > self.timestamps[-1]:
return None
idx = bisect_right(self.timestamps, timestamp) - 1
return self.snapshots[idx]
def get_best_bid_ask(self, timestamp: pd.Timestamp) -> Optional[Tuple[float, float]]:
"""
Get best bid and ask prices at a specific timestamp.
Args:
timestamp: Target timestamp
Returns:
Tuple of (best_bid, best_ask) or None
"""
snapshot = self.get_snapshot(timestamp)
if snapshot is None:
return None
bids = snapshot["bids"]
asks = snapshot["asks"]
if not bids or not asks:
return None
return float(bids[0][0]), float(asks[0][0])
def get_depth_at_price(
self,
timestamp: pd.Timestamp,
side: str,
price: float
) -> float:
"""
Get cumulative quantity at or better than a price level.
Args:
timestamp: Target timestamp
side: 'bid' or 'ask'
price: Target price
Returns:
Cumulative quantity available
"""
snapshot = self.get_snapshot(timestamp)
if snapshot is None:
return 0.0
levels = snapshot["bids"] if side == "bid" else snapshot["asks"]
total_qty = 0.0
for p, q in levels:
p = float(p)
q = float(q)
if side == "bid" and p <= price:
total_qty += q
elif side == "ask" and p >= price:
total_qty += q
elif side == "bid" and p > price:
break
elif side == "ask" and p < price:
break
return total_qty
def estimate_slippage(
self,
timestamp: pd.Timestamp,
side: str,
quantity: float
) -> dict:
"""
Estimate slippage for a market order of given quantity.
Simulates walking the book to fill the order quantity
and calculates average execution price vs mid-price.
Args:
timestamp: Execution timestamp
side: 'buy' or 'sell'
quantity: Order quantity
Returns:
Dictionary with slippage metrics
"""
snapshot = self.get_snapshot(timestamp)
if snapshot is None:
return {"error": "No snapshot available"}
levels = snapshot["bids"] if side == "sell" else snapshot["asks"]
remaining_qty = quantity
filled_value = 0.0
worst_price = 0.0
for price, qty in levels:
price = float(price)
qty = float(qty)
fill_qty = min(remaining_qty, qty)
filled_value += fill_qty * price
remaining_qty -= fill_qty
worst_price = price
if remaining_qty <= 0:
break
if remaining_qty > 0:
return {
"error": f"Insufficient liquidity: {remaining_qty:.4f} unfilled",
"filled_qty": quantity - remaining_qty,
"total_cost": filled_value
}
avg_price = filled_value / quantity
best_price = float(levels[0][0])
slippage = abs(avg_price - best_price) / best_price * 100
return {
"timestamp": timestamp,
"side": side,
"quantity": quantity,
"avg_price": avg_price,
"best_price": best_price,
"slippage_pct": slippage,
"worst_price": worst_price,
"total_cost": filled_value
}
Initialize replay engine
replayer = OrderBookReplay(order_book_data)
Get snapshot at specific timestamp
target_time = pd.Timestamp("2024-11-15 08:30:00")
snapshot = replayer.get_snapshot(target_time)
print(f"Snapshot at {target_time}:")
print(f" Best Bid: {snapshot['bids'][0]}")
print(f" Best Ask: {snapshot['asks'][0]}")
Estimate slippage for a 10 BTC market buy
slippage_result = replayer.estimate_slippage(target_time, "buy", 10.0)
print(f"\nSlippage estimate for 10 BTC market buy:")
print(f" Average Price: ${slippage_result['avg_price']:.2f}")
print(f" Slippage: {slippage_result['slippage_pct']:.4f}%")
print(f" Total Cost: ${slippage_result['total_cost']:,.2f}")Integration with HolySheep AI for Sentiment Analysis
While this tutorial focuses on order book data processing, you can enhance your trading research by combining market microstructure analysis with AI-powered sentiment analysis. HolySheep AI provides sub-50ms latency API access at significantly lower costs than mainstream providers — $1.00 per million tokens versus $7.30+ for comparable services.
# Example: Use HolySheep AI to analyze order book patterns
import os
HolySheep AI API configuration
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
def analyze_order_book_pattern(replayer: OrderBookReplay, timestamp: pd.Timestamp) -> str:
"""
Analyze order book state and generate insights using HolySheep AI.
This combines quantitative market microstructure metrics with
qualitative AI interpretation for enhanced trading decisions.
"""
snapshot = replayer.get_snapshot(timestamp)
if not snapshot:
return "No data available"
best_bid = float(snapshot["bids"][0][0])
best_ask = float(snapshot["asks"][0][0])
mid_price = (best_bid + best_ask) / 2
spread = (best_ask - best_bid) / mid_price * 100
# Calculate order book imbalance
bid_qty = sum(float(q) for _, q in snapshot["bids"][:10])
ask_qty = sum(float(q) for _, q in snapshot["asks"][:10])
imbalance = (bid_qty - ask_qty) / (bid_qty + ask_qty) * 100
prompt = f"""Analyze this Binance Futures order book snapshot:
- Mid Price: ${mid_price:,.2f}
- Bid-Ask Spread: {spread:.4f}%
- Top 10 Bid Quantity: {bid_qty:.4f} BTC
- Top 10 Ask Quantity: {ask_qty:.4f} BTC
- Order Book Imbalance: {imbalance:.2f}% (positive = buying pressure)
What does this order book state suggest about short-term price direction?
Consider: spread tightness, order imbalance, and liquidity distribution."""
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "user", "content": prompt}
],
"max_tokens": 500,
"temperature": 0.3
}
try:
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=5
)
response.raise_for_status()
result = response.json()
return result["choices"][0]["message"]["content"]
except requests.exceptions.RequestException as e:
return f"Analysis unavailable: {str(e)}"
Analyze current order book state
analysis = analyze_order_book_pattern(replayer, target_time)
print(f"AI Order Book Analysis:\n{analysis}")Performance Optimization for Large Datasets
When working with extended historical periods, memory and I/O become critical bottlenecks. Here are optimization techniques I developed while processing 30-day order book datasets:
import mmap
import tempfile
from concurrent.futures import ThreadPoolExecutor
from typing import Iterator
class OptimizedOrderBookLoader:
"""Memory-efficient loader for large order book datasets."""
def __init__(self, api_key: str, cache_dir: str = None):
self.api_key = api_key
self.cache_dir = cache_dir or tempfile.gettempdir()
def download_and_stream(
self,
symbol: str,
start: datetime,
end: datetime,
chunk_size: int = 1000
) -> Iterator[List[dict]]:
"""
Stream order book data in chunks to manage memory usage.
Yields chunks of snapshots, processing them incrementally
rather than loading everything into memory.
"""
url = f"https://api.tardis.dev/v1/feed/binance-futures:{symbol}"
params = {
"from": start.isoformat(),
"to": end.isoformat(),
"format": "json",
"compression": "gzip"
}
headers = {"Authorization": f"Bearer {self.api_key}"}
chunk = []
with requests.get(url, params=params, headers=headers, stream=True) as r:
r.raise_for_status()
for line in r.iter_lines():
if line:
try:
data = orjson.loads(line)
if data.get("type") == "order_book_snapshot":
chunk.append(data)
if len(chunk) >= chunk_size:
yield chunk
chunk = []
except Exception:
continue
if chunk:
yield chunk
def process_large_dataset(
self,
symbol: str,
start: datetime,
end: datetime,
processor_func: callable
):
"""
Process large dataset using chunked streaming.
processor_func should accept a list of snapshots.
"""
total_chunks = 0
for chunk in self.download_and_stream(symbol, start, end):
processor_func(chunk)
total_chunks += 1
if total_chunks % 10 == 0:
print(f"Processed {total_chunks} chunks...")
Usage: Process 30 days of data in chunks
loader = OptimizedOrderBookLoader(TARDIS_API_KEY)
def process_chunk(chunk: List[dict]):
"""Example chunk processor - calculate spreads."""
spreads = []
for snapshot in chunk:
if snapshot["bids"] and snapshot["asks"]:
spread = float(snapshot["asks"][0][0]) - float(snapshot["bids"][0][0])
spreads.append(spread)
return np.mean(spreads)
This processes data without loading everything into memory
start_date = datetime(2024, 10, 15)
end_date = datetime(2024, 11, 15)
loader.process_large_dataset("BTCUSDT", start_date, end_date, process_chunk)
Data Storage and Caching Strategy
For research workflows requiring repeated access to the same historical data, implementing an efficient caching layer saves both API costs and processing time. Here's a disk-based cache implementation:
import hashlib
import pickle
from pathlib import Path
class OrderBookCache:
"""Persistent cache for order book data."""
def __init__(self, cache_dir: str = None):
self.cache_dir = Path(cache_dir or os.path.join(tempfile.gettempdir(), "orderbook_cache"))
self.cache_dir.mkdir(parents=True, exist_ok=True)
def _get_cache_key(self, symbol: str, start: datetime, end: datetime) -> str:
"""Generate unique cache key for a data range."""
key_string = f"{symbol}:{start.isoformat()}:{end.isoformat()}"
return hashlib.sha256(key_string.encode()).hexdigest()[:16]
def get(self, symbol: str, start: datetime, end: datetime) -> Optional[List[dict]]:
"""Retrieve cached data if available."""
cache_key = self._get_cache_key(symbol, start, end)
cache_file = self.cache_dir / f"{cache_key}.pkl"
if cache_file.exists():
age_hours = (time.time() - cache_file.stat().st_mtime) / 3600
if age_hours < 24: # Cache valid for 24 hours
with open(cache_file, 'rb') as f:
return pickle.load(f)
return None
def set(self, symbol: str, start: datetime, end: datetime, data: List[dict]):
"""Store data in cache."""
cache_key = self._get_cache_key(symbol, start, end)
cache_file = self.cache_dir / f"{cache_key}.pkl"
with open(cache_file, 'wb') as f:
pickle.dump(data, f)
Usage
cache = OrderBookCache()
Try cache first
cached_data = cache.get("BTCUSDT", start_time, end_time)
if cached_data:
print("Using cached data")
order_book_data = cached_data
else:
print("Downloading fresh data")
order_book_data = downloader.download_l2_orderbook("BTCUSDT", start_time, end_time)
cache.set("BTCUSDT", start_time, end_time, order_book_data)
print("Data cached for future use")Common Errors and Fixes
Error 1: Authentication Failures - "401 Unauthorized"
Symptom: API requests return 401 status with message "Invalid API key or token expired."
Cause: The Bearer token is malformed, expired, or missing entirely from the request headers.
Solution:
# WRONG - Missing Bearer prefix or incorrect header
headers = {"Authorization": TARDIS_API_KEY} # Missing "Bearer "
CORRECT - Proper Bearer token format
headers = {"Authorization": f"Bearer {TARDIS_API_KEY}"}
Verify your key is set correctly
import os
TARDIS_API_KEY = os.environ.get("TARDIS_API_KEY")
if not TARDIS_API_KEY:
raise ValueError("TARDIS_API_KEY environment variable not set")
Test authentication
response = requests.get(
"https://api.tardis.dev/v1/exchanges",
headers={"Authorization": f"Bearer {TARDIS_API_KEY}"}
)
if response.status_code == 401:
print("Invalid API key - check your Tardis.dev dashboard")
elif response.status_code == 200:
print("Authentication successful")Error 2: Timestamp Format Errors - "Invalid date format"
Symptom: API returns 400 error with "Invalid date format" even when providing ISO timestamps.
Cause: Tardis.dev requires strict ISO 8601 formatting with timezone information, and the 'from'/'to' parameters must be specified with proper granularity.
Solution:
# WRONG - Missing timezone or incorrect format
params = {"from": "2024-11-15", "to": "2024-11-16"}
CORRECT - ISO 8601 with timezone (Z for UTC)
params = {
"from": "2024-11-15T00:00:00Z",
"to": "2024-11-16T00:00:00Z"
}
Alternative: Use datetime with explicit UTC timezone
from datetime import timezone
start = datetime(2024, 11, 15, tzinfo=timezone.utc)
end = datetime(2024, 11, 16, tzinfo=timezone.utc)
params = {
"from": start.isoformat().replace("+00:00", "Z"),
"to": end.isoformat().replace("+00:00", "Z")
}
Verify format before making request
print(f"Request params: {params}")Error 3: Memory Exhaustion - "MemoryError" on Large Downloads
Symptom: Python process crashes with MemoryError when downloading extended date ranges (more than 1-2 days of minute-level data).
Cause: Loading entire response into memory via response.json() or storing all snapshots in a single list without chunking.
Solution:
# WRONG - Loads entire response into memory
response = requests.get(url, params=params, headers=headers)
all_data = response.json() # Crashes for large responses
CORRECT - Stream response line by line
response = requests.get(url, params=params, headers=headers, stream=True)
response.raise_for_status()
snapshots = []
BATCH_SIZE = 5000
for line in response.iter_lines():
if line:
data = orjson.loads(line)
snapshots.append(data)
# Process in batches to control memory
if len(snapshots) >= BATCH_SIZE:
process_batch(snapshots)
snapshots = [] # Clear memory
Don't forget final batch
if snapshots:
process_batch(snapshots)
Alternative: Use the OptimizedOrderBookLoader class shown earlier
loader = OptimizedOrderBookLoader(TARDIS_API_KEY)
for chunk in loader.download_and_stream("BTCUSDT", start, end):
process_chunk(chunk) # Process and discardError 4: Parsing Malformed Messages - "KeyError: 'bids'"
Symptom: Code raises KeyError when accessing snapshot["bids"] despite filtering for order_book_snapshot type.
Cause: Some message types contain partial data or have different field names. Not all "snapshot" messages have both bids and asks arrays.
Solution:
# WRONG - Direct access without validation
for data in raw_messages:
if data["type"] == "order_book_snapshot":
snapshot = {
"bids": data["bids"], # Crashes if 'bids' missing
"asks": data["asks"] # Crashes if 'asks' missing
}
CORRECT - Defensive parsing with field validation
def safe_parse_snapshot(data: dict) -> Optional[dict]:
"""Safely parse an order book snapshot with field validation."""
if data.get("type") != "order_book_snapshot":
return