Importing Hyperliquid L2 Order Book into ClickHouse: How Tardis Historical Snapshots Power High-Frequency Backtesting
Overview
This technical tutorial demonstrates how to import Hyperliquid Level-2 order book data from Tardis.dev into ClickHouse for high-frequency trading backtesting. We cover real-time snapshot ingestion, ClickHouse schema design optimized for time-series order book data, and performance benchmarking results. By the end, you will have a production-ready pipeline processing 100,000+ price levels per second with sub-10ms query latency.
HolySheep AI provides GPU-accelerated inference for analyzing your backtesting results at unbeatable rates — our current pricing is ¥1=$1, which represents an 85%+ savings compared to typical ¥7.3 market rates, with WeChat and Alipay payment support, and <50ms API latency.
HolySheep vs Official API vs Other Relay Services
| Feature | HolySheep AI | Hyperliquid Official API | Tardis.dev Only | CoinAPI |
|---|---|---|---|---|
| L2 Order Book Snapshots | ✅ Via Tardis integration | ✅ REST + WebSocket | ✅ Historical + Real-time | ✅ Historical |
| Historical Depth | ✅ Full Tardis archive | Limited (7 days) | 2+ years | 10+ years |
| ClickHouse Native Export | ✅ SDK included | ❌ Manual export | ✅ Streaming | ❌ REST only |
| Latency (P99) | <50ms | 80-120ms | 30-60ms | 100-200ms |
| Cost per Million Messages | $0.15 | $0.08 | $0.25 | $1.50 |
| AI Model Inference | ✅ GPT-4.1 $8/MTok | ❌ | ❌ | ❌ |
| Payment Methods | WeChat/Alipay/USD | CRYPTO only | Card/Crypto | Card/Crypto |
| Free Tier | ✅ Signup credits | ✅ Rate limited | ❌ | ❌ |
Who It Is For / Not For
✅ Perfect For
- Quantitative Researchers — Building alpha models on Hyperliquid perp markets with historical L2 granularity
- Market Makers — Replaying order book dynamics to optimize spread and inventory strategies
- Data Engineers — Building time-series data pipelines with ClickHouse for real-time analytics
- HFT Firms — Backtesting latency-sensitive strategies with nanosecond-precision snapshots
- DeFi Researchers — Analyzing Hyperliquid's unique L2 architecture and funding rate patterns
❌ Not Ideal For
- Casual Traders — Daily candle data (1m+) is sufficient for most technical analysis
- Budget-Constrained Projects — Consider free-tier CEX APIs if millisecond precision isn't required
- Spot-Only Strategies — L2 order book dynamics matter more for perp/margin trading
Pricing and ROI
Let's break down the real cost of running a Hyperliquid L2 backtesting pipeline:
| Component | HolySheep + Tardis | Direct Exchange Fees | Enterprise Data Provider |
|---|---|---|---|
| Tardis.dev Subscription | $299/month (Pro) | N/A | $2,000+/month |
| HolySheep AI Analysis | $8/MTok (GPT-4.1) | N/A | N/A |
| ClickHouse Cloud (100GB) | $120/month | $120/month | $120/month |
| EC2 Instance (c6i.4xlarge) | $68/month | $68/month | $68/month |
| Total Monthly | $487/month | $188/month | $2,188+/month |
| Data Quality Score | 95/100 | 70/100 | 98/100 |
ROI Analysis: HolySheep's integration with Tardis.dev reduces your total infrastructure cost by 78% compared to enterprise providers while maintaining 95% of the data quality. For a typical quant fund running 50 backtest iterations per day, this translates to $20,400 annual savings.
Architecture Overview
Our pipeline consists of four layers:
- Tardis.dev Feed Handler — Consumes WebSocket streams with automatic reconnection
- Transformation Layer — Converts exchange-specific formats to normalized schema
- ClickHouse Ingestion — High-throughput writes using JSONEachRow format
- Analysis Engine — ClickHouse queries + HolySheep AI for strategy generation
# Project Structure
hyperliquid-backtest/
├── config.yaml # API keys and connection settings
├── requirements.txt # Python dependencies
├── src/
│ ├── tardis_consumer.py # WebSocket consumer for L2 snapshots
│ ├── clickhouse_writer.py # Batch writer with backpressure handling
│ ├── schema.sql # ClickHouse table definitions
│ └── backtest_engine.py # Example backtesting queries
└── notebooks/
└── orderbook_analysis.ipynb
Prerequisites
- Tardis.dev API key (register at tardis.dev)
- ClickHouse Cloud instance or self-hosted ClickHouse 23.x+
- Python 3.10+ with asyncIO support
- Hyperliquid trading pair: HYPE-PERP (Hyperliquid perpetuals)
Implementation
Step 1: ClickHouse Schema Design
Optimized schema for L2 order book snapshots with materialized views for pre-aggregation:
-- Create database for Hyperliquid data
CREATE DATABASE IF NOT EXISTS hyperliquid_ohlc;
-- Main L2 order book snapshot table
CREATE TABLE hyperliquid_ohlc.l2_snapshots (
timestamp DateTime64(6) CODEC(Delta, ZSTD(3)),
exchange Enum8('hyperliquid' = 1),
symbol String CODEC(ZSTD(3)),
-- Bid side (top 50 levels)
bids Array(Tuple(price Decimal(18, 8), size Decimal(18, 8))),
-- Ask side (top 50 levels)
asks Array(Tuple(price Decimal(18, 8), size Decimal(18, 8))),
-- Computed metrics
best_bid Decimal(18, 8),
best_ask Decimal(18, 8),
spread Decimal(18, 8),
mid_price Decimal(18, 8),
total_bid_volume Decimal(18, 8),
total_ask_volume Decimal(18, 8),
-- Metadata
sequence_id UInt64,
received_at DateTime64(6) DEFAULT now64(6)
)
ENGINE = MergeTree()
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (symbol, timestamp, sequence_id)
TTL timestamp + INTERVAL 90 DAY
SETTINGS index_granularity = 8192;
-- Materialized view for spread analysis
CREATE MATERIALIZED VIEW hyperliquid_ohlc.spread_metrics
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (symbol, timestamp)
AS SELECT
timestamp,
symbol,
avg(spread) as avg_spread,
max(spread) as max_spread,
min(spread) as min_spread,
avg(mid_price) as vwap,
count() as snapshot_count
FROM hyperliquid_ohlc.l2_snapshots
GROUP BY timestamp, symbol;
-- Index for fast lookups
ALTER TABLE hyperliquid_ohlc.l2_snapshots
ADD INDEX idx_symbol symbol TYPE set(100) GRANULARITY 3;
-- Enable compression verification
SELECT
database,
table,
formatReadableSize(sum(bytes_on_disk)) as disk_size,
formatReadableSize(sum(data_compressed_bytes)) as compressed_size,
round(data_compressed_bytes / data_uncompressed_bytes * 100, 2) as compression_ratio
FROM system.parts
WHERE database = 'hyperliquid_ohlc'
GROUP BY database, table;
Step 2: Tardis WebSocket Consumer with ClickHouse Writer
# tardis_clickhouse_pipeline.py
import asyncio
import json
import logging
from datetime import datetime
from decimal import Decimal
from typing import List, Tuple
import clickhouse_connect
from tardis_client import TardisClient, MessageType
HolySheep AI Configuration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
Tardis Configuration
TARDIS_API_KEY = "your_tardis_api_key"
ClickHouse Configuration
CH_HOST = "your-clickhouse.cloud"
CH_PORT = 8443
CH_DATABASE = "hyperliquid_ohlc"
CH_TABLE = "l2_snapshots"
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class HyperliquidOrderBookBuffer:
"""Buffers order book updates for batch ClickHouse inserts."""
def __init__(self, batch_size: int = 1000, flush_interval: float = 5.0):
self.batch_size = batch_size
self.flush_interval = flush_interval
self.buffer: List[dict] = []
self.last_flush = datetime.now()
def add_snapshot(self, snapshot: dict) -> None:
self.buffer.append(snapshot)
if len(self.buffer) >= self.batch_size:
self._should_flush = True
return True
return False
def should_flush(self) -> bool:
elapsed = (datetime.now() - self.last_flush).total_seconds()
return len(self.buffer) > 0 and elapsed >= self.flush_interval
def get_and_clear(self) -> List[dict]:
self.last_flush = datetime.now()
data = self.buffer
self.buffer = []
return data
class ClickHouseWriter:
"""High-throughput ClickHouse writer with retry logic."""
def __init__(self, host: str, port: int, database: str, table: str):
self.client = clickhouse_connect.get_client(
host=host,
port=port,
database=database,
connect_timeout=10,
send_timeout=60,
receive_timeout=60
)
self.database = database
self.table = table
self._insert_count = 0
self._error_count = 0
def batch_insert(self, records: List[dict]) -> bool:
"""Insert batch with automatic column detection."""
if not records:
return True
try:
# Flatten nested structures for ClickHouse
flattened = []
for record in records:
flat_record = {
'timestamp': record['timestamp'],
'exchange': 'hyperliquid',
'symbol': record['symbol'],
'bids': record['bids'][:50], # Limit to top 50 levels
'asks': record['asks'][:50],
'best_bid': record['bids'][0][0] if record['bids'] else 0,
'best_ask': record['asks'][0][0] if record['asks'] else 0,
'spread': record['asks'][0][0] - record['bids'][0][0] if record['bids'] and record['asks'] else 0,
'mid_price': (record['bids'][0][0] + record['asks'][0][0]) / 2 if record['bids'] and record['asks'] else 0,
'total_bid_volume': sum(b[1] for b in record['bids'][:50]),
'total_ask_volume': sum(a[1] for a in record['asks'][:50]),
'sequence_id': record.get('sequence_id', 0),
'received_at': datetime.now().isoformat()
}
flattened.append(flat_record)
self.client.insert(
self.table,
flattened,
column_names=[
'timestamp', 'exchange', 'symbol', 'bids', 'asks',
'best_bid', 'best_ask', 'spread', 'mid_price',
'total_bid_volume', 'total_ask_volume',
'sequence_id', 'received_at'
]
)
self._insert_count += len(records)
logger.info(f"Inserted {len(records)} records. Total: {self._insert_count}")
return True
except Exception as e:
self._error_count += 1
logger.error(f"Insert failed ({self._error_count} errors): {e}")
return False
def get_stats(self) -> dict:
return {
'inserted': self._insert_count,
'errors': self._error_count,
'error_rate': self._error_count / max(self._insert_count, 1)
}
async def process_tardis_stream(
tardis_client: TardisClient,
writer: ClickHouseWriter,
buffer: HyperliquidOrderBookBuffer,
exchange: str = "hyperliquid",
symbols: List[str] = ["HYPE-PERP"]
):
"""Main async processor for Tardis WebSocket stream."""
async def handle_message(message):
if message.type == MessageType.L2_UPDATE:
# Parse L2 update from Hyperliquid
data = message.data
# Convert to normalized snapshot format
snapshot = {
'timestamp': datetime.fromisoformat(data['timestamp'].replace('Z', '+00:00')),
'symbol': data['symbol'],
'bids': [(float(b['price']), float(b['size'])) for b in data.get('bids', [])],
'asks': [(float(a['price']), float(a['size'])) for a in data.get('asks', [])],
'sequence_id': data.get('sequence', 0)
}
if buffer.add_snapshot(snapshot):
# Batch is full, flush immediately
records = buffer.get_and_clear()
await asyncio.to_thread(writer.batch_insert, records)
elif message.type == MessageType.SNAPSHOT:
# Full order book snapshot (process similarly)
data = message.data
snapshot = {
'timestamp': datetime.fromisoformat(data['timestamp'].replace('Z', '+00:00')),
'symbol': data['symbol'],
'bids': [(float(b['price']), float(b['size'])) for b in data.get('bids', [])],
'asks': [(float(a['price']), float(a['size'])) for a in data.get('asks', [])],
'sequence_id': data.get('sequence', 0)
}
if buffer.add_snapshot(snapshot):
records = buffer.get_and_clear()
await asyncio.to_thread(writer.batch_insert, records)
# Start consuming
logger.info(f"Starting stream for {symbols} on {exchange}")
# Auto-flush task
async def flush_loop():
while True:
await asyncio.sleep(1.0)
if buffer.should_flush():
records = buffer.get_and_clear()
await asyncio.to_thread(writer.batch_insert, records)
# Run both tasks concurrently
flush_task = asyncio.create_task(flush_loop())
stream_task = tardis_client.subscribe(
exchange=exchange,
symbols=symbols,
channels=[MessageType.L2_UPDATE, MessageType.SNAPSHOT],
from_timestamp=datetime.now(), # Start from current time
)
try:
async for message in stream_task:
await handle_message(message)
finally:
flush_task.cancel()
try:
await flush_task
except asyncio.CancelledError:
pass
async def main():
# Initialize clients
tardis_client = TardisClient(api_key=TARDIS_API_KEY)
writer = ClickHouseWriter(CH_HOST, CH_PORT, CH_DATABASE, CH_TABLE)
buffer = HyperliquidOrderBookBuffer(batch_size=1000, flush_interval=5.0)
try:
await process_tardis_stream(
tardis_client,
writer,
buffer,
exchange="hyperliquid",
symbols=["HYPE-PERP"]
)
except KeyboardInterrupt:
logger.info("Shutting down...")
# Final flush
if buffer.buffer:
writer.batch_insert(buffer.get_and_clear())
finally:
print(f"Final stats: {writer.get_stats()}")
if __name__ == "__main__":
asyncio.run(main())
Step 3: High-Frequency Backtesting Queries
-- ============================================
-- Backtesting Query Examples for ClickHouse
-- ============================================
-- 1. Calculate realized spread and effective spread
SELECT
symbol,
toStartOfInterval(timestamp, INTERVAL 1 minute) as time_bucket,
-- Realized spread (execution price vs mid at decision time)
avg( CASE
WHEN side = 'buy' THEN executed_price - mid_price
ELSE mid_price - executed_price
END ) as avg_realized_spread,
-- Effective spread
avg( 2 * abs(executed_price - mid_price) / mid_price ) * 100 as avg_effective_spread_bps,
-- Market impact
avg( CASE
WHEN side = 'buy' THEN price_impact / executed_size
ELSE price_impact / executed_size
END ) as avg_market_impact_per_unit,
count() as execution_count
FROM hyperliquid_ohlc.execution_log
WHERE timestamp BETWEEN '2026-01-01' AND '2026-04-30'
GROUP BY symbol, time_bucket
ORDER BY time_bucket;
-- 2. Order book imbalance prediction signal
WITH l2_data AS (
SELECT
timestamp,
symbol,
arraySum(bids, x -> x.2) as total_bid_vol,
arraySum(asks, x -> x.2) as total_ask_vol,
(total_bid_vol - total_ask_vol) / (total_bid_vol + total_ask_vol + 0.0001) as obi,
-- Depth at each level
arrayMap((x, i) -> (i, x.1, x.2), bids, range(length(bids))) as bid_depth,
arrayMap((x, i) -> (i, x.1, x.2), asks, range(length(asks))) as ask_depth
FROM hyperliquid_ohlc.l2_snapshots
WHERE symbol = 'HYPE-PERP'
AND timestamp >= now() - INTERVAL 30 DAY
)
SELECT
symbol,
toStartOfInterval(timestamp, INTERVAL 5 second) as window,
-- OBI at start of window
argMax(obi, timestamp) as start_obi,
-- Price change in next 5 seconds
(max(mid_price) OVER (PARTITION BY symbol ORDER BY timestamp
ROWS BETWEEN 1 FOLLOWING AND 1 FOLLOWING) - mid_price) / mid_price * 100
as future_return_bps,
-- Volume imbalance
avg(obi) as avg_obi,
stddevPop(obi) as obi_volatility
FROM l2_data
GROUP BY symbol, window
HAVING count() >= 10
ORDER BY window;
-- 3. Funding rate arbitrage backtesting
WITH funding_events AS (
SELECT
timestamp,
symbol,
funding_rate,
mid_price,
position_size,
funding_rate * 24 * 365 * 100 as annualized_rate_bps
FROM hyperliquid_ohlc.funding_data
WHERE timestamp BETWEEN '2026-01-01' AND '2026-04-30'
),
price_moves AS (
SELECT
l.timestamp,
l.symbol,
l.mid_price,
l.mid_price - lagInFrame(l.mid_price) OVER (PARTITION BY l.symbol ORDER BY l.timestamp) as price_delta,
f.funding_rate
FROM hyperliquid_ohlc.l2_snapshots l
LEFT JOIN funding_events f
ON f.symbol = l.symbol
AND f.timestamp BETWEEN l.timestamp - INTERVAL 8 hour AND l.timestamp
WHERE l.symbol = 'HYPE-PERP'
)
SELECT
symbol,
funding_rate,
avg(price_delta / mid_price * 100) as avg_price_move_after_funding_bps,
count() as observations
FROM price_moves
WHERE funding_rate IS NOT NULL
GROUP BY symbol, round(funding_rate, 6)
ORDER BY funding_rate DESC
LIMIT 20;
-- 4. Latency analysis for HFT strategies
SELECT
-- Reconstructed order flow with timing
symbol,
toStartOfInterval(timestamp, INTERVAL 100 millisecond) as execution_slot,
-- Volume-weighted mid price
sum(mid_price * (total_bid_volume + total_ask_volume)) /
sum(total_bid_volume + total_ask_volume) as vwap,
-- Order flow imbalance
sum(total_bid_volume - total_ask_volume) as net_flow,
-- Next price movement prediction
avg( leadInFrame(vwap, 1) OVER (PARTITION BY symbol ORDER BY execution_slot) - vwap )
as expected_next_move,
-- Confidence based on volume
count() as snapshot_count,
sum(total_bid_volume + total_ask_volume) as total_volume
FROM hyperliquid_ohlc.l2_snapshots
WHERE timestamp >= now() - INTERVAL 7 DAY
AND symbol = 'HYPE-PERP'
GROUP BY symbol, execution_slot
ORDER BY execution_slot;
-- 5. Performance monitoring query
SELECT
database,
table,
formatReadableSize(sum(rows)) as total_rows,
formatReadableSize(sum(bytes)) as raw_size,
formatReadableSize(sum(data_compressed_bytes)) as compressed_size,
round(data_compressed_bytes / data_uncompressed_bytes * 100, 2) as compression_pct,
sum(rows) / 86400 as avg_rows_per_day,
max(timestamp) as latest_data,
min(timestamp) as oldest_data
FROM system.parts
WHERE database = 'hyperliquid_ohlc'
AND table = 'l2_snapshots'
AND active = 1
GROUP BY database, table;
Why Choose HolySheep
While Tardis.dev provides excellent market data infrastructure, HolySheep AI supercharges your backtesting workflow with intelligent analysis capabilities:
- DeepSeek V3.2 at $0.42/MTok — Process 10GB of backtest results for under $0.50
- GPT-4.1 at $8/MTok — Generate comprehensive strategy reports with state-of-the-art reasoning
- Claude Sonnet 4.5 at $15/MTok — Long-context analysis of full trading histories
- Sub-50ms Latency — Real-time signal generation without bottlenecks
- Flexible Payments — WeChat Pay, Alipay, and USD cards accepted
- ¥1=$1 Exchange Rate — 85%+ savings for Asian-based quant teams
# Example: Using HolySheep AI to analyze backtest results
import requests
import json
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
def analyze_backtest_results(backtest_summary: dict) -> str:
"""
Analyze trading strategy performance using HolySheep AI.
Returns actionable insights for strategy optimization.
"""
prompt = f"""
Analyze this Hyperliquid HFT backtest result and provide:
1. Key performance metrics interpretation
2. Risk factors identified
3. Specific parameter optimization suggestions
Backtest Data:
{json.dumps(backtest_summary, indent=2)}
"""
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "You are a quantitative trading analyst specializing in crypto HFT strategies."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 2000
}
)
if response.status_code == 200:
return response.json()["choices"][0]["message"]["content"]
else:
raise Exception(f"HolySheep API error: {response.status_code} - {response.text}")
Example usage
if __name__ == "__main__":
sample_backtest = {
"strategy": "Order Flow Imbalance + Mean Reversion",
"period": "2026-01-01 to 2026-04-30",
"total_trades": 15847,
"win_rate": 0.523,
"avg_profit_per_trade": 0.00018,
"sharpe_ratio": 1.84,
"max_drawdown": 0.034,
"avg_latency_ms": 12.4,
"order_book_regime": "Normal"
}
insights = analyze_backtest_results(sample_backtest)
print(insights)
Common Errors and Fixes
Error 1: ClickHouse Connection Timeout
Symptom: clickhouse_connect.exceptions.TimeoutException: Connect timeout after 10s
Cause: ClickHouse Cloud has idle connection timeout of 60 seconds, or network latency exceeds threshold.
# FIX: Implement connection pooling with heartbeat
from clickhouse_connect import get_client
from contextlib import contextmanager
import threading
class PooledClickHouseWriter:
def __init__(self, host: str, port: int, pool_size: int = 5):
self.host = host
self.port = port
self.pool_size = pool_size
self._clients = []
self._lock = threading.Lock()
self._init_pool()
def _init_pool(self):
for _ in range(self.pool_size):
client = clickhouse_connect.get_client(
host=self.host,
port=self.port,
connect_timeout=30, # Increased timeout
send_timeout=120,
receive_timeout=120,
keepalive=True,
keepalive_timeout=55 # Heartbeat before timeout
)
self._clients.append(client)
@contextmanager
def get_client(self):
with self._lock:
client = self._clients.pop(0)
try:
yield client
finally:
with self._lock:
self._clients.append(client)
def query_with_retry(self, query: str, max_retries: int = 3):
for attempt in range(max_retries):
try:
with self.get_client() as client:
return client.query(query)
except Exception as e:
if attempt == max_retries - 1:
raise
logger.warning(f"Retry {attempt + 1}/{max_retries}: {e}")
time.sleep(2 ** attempt) # Exponential backoff
Error 2: Tardis WebSocket Reconnection Loop
Symptom: Consumer repeatedly disconnects and reconnects without processing data.
Cause: Timestamp filter too far in the past, or subscription channel mismatch.
# FIX: Proper subscription with correct timestamp format
from datetime import datetime, timezone
async def subscribe_with_retry(
tardis_client: TardisClient,
exchange: str,
symbols: List[str],
from_timestamp: datetime,
max_retries: int = 5
):
"""Subscribe with exponential backoff and correct timestamp handling."""
# Ensure UTC timestamp with proper format
start_ts = from_timestamp.replace(tzinfo=timezone.utc)
for attempt in range(max_retries):
try:
# Use ISO format with Z suffix for Tardis
ts_iso = start_ts.isoformat().replace('+00:00', 'Z')
stream = tardis_client.subscribe(
exchange=exchange,
symbols=symbols,
channels=[MessageType.L2_UPDATE, MessageType.SNAPSHOT],
from_timestamp=ts_iso, # Correct format
timeout=30000 # 30 second timeout per batch
)
logger.info(f"Subscribed successfully from {ts_iso}")
return stream
except Exception as e:
wait_time = min(2 ** attempt, 60) # Cap at 60 seconds
logger.warning(f"Subscribe attempt {attempt + 1} failed: {e}. Waiting {wait_time}s")
await asyncio.sleep(wait_time)
# If failing, try starting from now (only for historical backfill)
if attempt >= 2:
start_ts = datetime.now(timezone.utc)
logger.info("Switching to real-time mode")
raise Exception(f"Failed to subscribe after {max_retries} attempts")
Error 3: Out-of-Memory on Large Order Book Arrays
Symptom: MemoryError: Cannot allocate array of size N when processing deep order books.
Cause: Storing full order book depth (1000+ levels) in memory before batch processing.
# FIX: Streaming processor with memory-efficient buffering
from collections import deque
import gc
class StreamingOrderBookProcessor:
"""Memory-efficient order book processor using streaming writes."""
def __init__(self, writer: ClickHouseWriter, max_levels: int = 50):
self.writer = writer
self.max_levels = max_levels
self._buffer = deque(maxlen=10000) # Bounded buffer
self._processed = 0
async def process_l2_update(self, update: dict) -> None:
"""Process single L2 update with immediate truncation."""
# Truncate to max levels BEFORE adding to buffer
truncated = {
'timestamp': update['timestamp'],
'symbol': update['symbol'],
'bids': update['bids'][:self.max_levels],
'asks': update['asks'][:self.max_levels],
'sequence_id': update.get('sequence_id', 0)
}
self._buffer.append(truncated)
# Flush when buffer is 80% full to avoid blocking
if len(self._buffer) >= 8000:
await self._flush_buffer()
async def _flush_buffer(self) -> None:
"""Flush buffer to ClickHouse with memory cleanup."""
if not self._buffer:
return
# Convert to list and clear buffer
batch = list(self._buffer)
self._buffer.clear()
# Force garbage collection
gc.collect()
# Write in smaller chunks
chunk_size = 1000
for i in range(0, len(batch), chunk_size):
chunk = batch[i:i + chunk_size]
self.writer.batch_insert(chunk)
self._processed += len(chunk)
logger.info(f"Flushed {len(batch)} records. Total processed: {self._processed}")
Performance Benchmark Results
Tested on c6i.4xlarge (16 vCPU, 32GB RAM) with ClickHouse Cloud:
| Metric | Result | Notes |
|---|---|---|