Giới thiệu
Nếu bạn đang xây dựng bot giao dịch, hệ thống arbitrage, hoặc đơn giản là muốn backtest chiến lược trading với dữ liệu sâu (depth data), thì bài viết này dành cho bạn. Tôi đã dành 3 tháng để set up hệ thống lưu trữ và回放 order book từ Binance, và trong quá trình đó gặp vô số lỗi cryptic. Bài viết sẽ hướng dẫn bạn từng bước một, tránh những sai lầm mà tôi đã mắc phải.
**Order Book (Sổ lệnh) L2 là gì?** Đơn giản, đó là danh sách các lệnh mua/bán đang chờ khớp, được sắp xếp theo giá. L2 nghĩa là bạn thấy cả giá và khối lượng (không chỉ top-of-book như L1).
Tại sao cần lưu trữ Order Book?
- **Backtest thực sự chính xác**: Dữ liệu ticker chỉ cho thấy giá cuối cùng, còn order book cho thấy thanh khoản thực tế
- **Phát hiện front-running**: So sánh vị trí lệnh của bạn với các lệnh lớn trong sổ
- **Phân tích thanh khoản**: Hiểu được spread, depth, và volatility theo thời gian
- **Machine Learning features**: Tạo features từ order flow, imbalance ratios, queue position
Công cụ cần thiết
Trước khi bắt đầu, bạn cần chuẩn bị:
- **Tardis Machine**: Dịch vụ cung cấp historical market data từ nhiều sàn (Binance, Bybit, OKX...). Đăng ký Tardis
- **ClickHouse**: Database column-oriented, cực nhanh cho analytical queries
- **Docker**: Để chạy ClickHouse (hoặc dùng ClickHouse Cloud)
- **Python 3.10+**: Để xử lý dữ liệu
Phần 1: Lấy dữ liệu từ Tardis
Đăng ký và lấy API Key Tardis
Sau khi đăng ký tài khoản Tardis, vào Dashboard → API Keys → Tạo key mới. Lưu ý: Tardis có free tier với 10GB data/month.
Cấu trúc dữ liệu Binance L2 Order Book
Binance cung cấp 2 loại stream:
- **Depth@100ms**: Snapshot 20 levels mỗi 100ms
- **Depth Update**: incremental updates
Tardis lưu trữ cả 2 dạng. Với mục đích backtest, bạn nên dùng **depth snapshot** vì nó đầy đủ hơn.
Download dữ liệu từ Tardis
# Cài đặt thư viện cần thiết
pip install tardis-client clickhouse-driver pandas pyarrow
Script download dữ liệu Binance L2 order book
from tardis_client import TardisClient, Interval
import asyncio
import pandas as pd
from datetime import datetime
API_KEY = "YOUR_TARDIS_API_KEY"
SYMBOL = "btcusdt"
START_DATE = "2024-01-01"
END_DATE = "2024-01-02"
async def download_orderbook():
client = TardisClient(API_KEY)
# Lấy dữ liệu depth snapshot (Binance Combined Stream)
messages = client.replay(
exchange="binance",
symbols=[SYMBOL],
from_date=START_DATE,
to_date=END_DATE,
channels=[f"{SYMBOL}@depth@100ms"]
)
records = []
async for message in messages:
if message.type == "depth":
record = {
"timestamp": message.timestamp,
"symbol": message.symbol,
"bids": str(message.bids), # Convert list to string for storage
"asks": str(message.asks),
"bid_count": len(message.bids),
"ask_count": len(message.asks),
"best_bid": float(message.bids[0][0]) if message.bids else None,
"best_ask": float(message.asks[0][0]) if message.asks else None,
"spread": float(message.asks[0][0]) - float(message.bids[0][0]) if message.bids and message.asks else None
}
records.append(record)
df = pd.DataFrame(records)
df.to_parquet(f"binance_depth_{SYMBOL}_{START_DATE}.parquet", index=False)
print(f"Downloaded {len(df)} snapshots")
return df
Chạy
asyncio.run(download_orderbook())
**Gợi ý ảnh chụp màn hình**: Chụp Tardis Dashboard sau khi tạo API key, hiển thị phần API Keys
Phần 2: Cài đặt ClickHouse
Cách 1: Dùng Docker (Khuyến nghị cho beginners)
# Pull và chạy ClickHouse server
docker run -d \
--name clickhouse-server \
-p 8123:8123 \
-p 9000:9000 \
-e CLICKHOUSE_DB=orderbook \
-e CLICKHOUSE_USER=admin \
-e CLICKHOUSE_PASSWORD=your_secure_password \
clickhouse/clickhouse-server:latest
Kiểm tra container đang chạy
docker ps | grep clickhouse
Kết nối bằng clickhouse-client (cài đặt riêng)
clickhouse-client --host localhost --port 9000 --user admin --password your_secure_password
Cách 2: ClickHouse Cloud (Không cần server)
Đăng ký tại
ClickHouse Cloud, tạo service và lấy connection string.
# Kết nối từ Python
from clickhouse_driver import Client
client = Client(
host='your-cloud-host.clickhouse.cloud',
port=9440,
user='default',
password='your-password',
secure=True
)
Test connection
result = client.execute('SELECT 1')
print("Connected:", result)
Phần 3: Tạo bảng và Import dữ liệu
Thiết kế Schema
-- Tạo database
CREATE DATABASE IF NOT EXISTS orderbook_db;
-- Tạo bảng orderbook snapshots
CREATE TABLE orderbook_db.binance_depth (
timestamp DateTime64(3) CODEC(Delta, ZSTD(1)),
symbol LowCardinality(String),
exchange LowCardinality(String),
best_bid Decimal(20, 8),
best_ask Decimal(20, 8),
spread Decimal(20, 8),
bid_count UInt16,
ask_count UInt16,
mid_price Decimal(20, 8) ALIAS (best_bid + best_ask) / 2,
bid_levels Array(Tuple(price Decimal(20, 8), quantity Decimal(20, 8))),
ask_levels Array(Tuple(price Decimal(20, 8), quantity Decimal(20, 8)))
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (symbol, timestamp)
TTL timestamp + INTERVAL 12 MONTH;
-- Tạo bảng cho incremental updates (nếu cần)
CREATE TABLE orderbook_db.binance_depth_updates (
timestamp DateTime64(3),
symbol LowCardinality(String),
channel String,
is_bid_update Boolean,
price Decimal(20, 8),
quantity Decimal(20, 8),
update_type String -- 'snapshot' | 'update' | 'clear'
)
ENGINE = ReplacingMergeTree(timestamp)
ORDER BY (symbol, timestamp, price);
**Gợi ý ảnh chụp màn hình**: Chụp kết quả sau khi chạy CREATE TABLE thành công trong ClickHouse client
Import dữ liệu từ Parquet
import pandas as pd
from clickhouse_driver import Client
import json
def import_parquet_to_clickhouse(parquet_file, client):
# Đọc parquet
df = pd.read_parquet(parquet_file)
# Parse bids/asks strings thành arrays
def parse_levels(level_str):
try:
# Format: [(price, qty), (price, qty), ...]
clean_str = level_str.strip("[]")
if not clean_str:
return []
pairs = []
for item in clean_str.split("), ("):
item = item.strip("() ")
if item:
price, qty = item.split(", ")
pairs.append((float(price), float(qty)))
return pairs
except:
return []
df['bid_levels'] = df['bids'].apply(parse_levels)
df['ask_levels'] = df['asks'].apply(parse_levels)
df['exchange'] = 'binance'
# Chuyển đổi timestamp
df['timestamp'] = pd.to_datetime(df['timestamp'])
# Chọn columns phù hợp với schema
insert_df = df[['timestamp', 'symbol', 'exchange', 'best_bid', 'best_ask',
'spread', 'bid_count', 'ask_count', 'bid_levels', 'ask_levels']]
# Insert vào ClickHouse
columns = insert_df.columns.tolist()
data = insert_df.values.tolist()
client.execute(
'INSERT INTO orderbook_db.binance_depth VALUES',
data,
types_check=True
)
print(f"Imported {len(insert_df)} rows")
Kết nối và import
client = Client('localhost', user='admin', password='your_secure_password')
import_parquet_to_clickhouse("binance_depth_btcusdt_2024-01-01.parquet", client)
Phần 4: 回放 Order Book cho Backtesting
Nguyên lý回放
回放 (replay) nghĩa là đọc lại dữ liệu theo thứ tự thời gian, tái tạo trạng thái order book tại mỗi thời điểm. Điều này cho phép bạn test chiến lược trading với dữ liệu lịch sử một cách chính xác.
class OrderBookReplay:
def __init__(self, client):
self.client = client
self.current_book = {'bids': {}, 'asks': {}}
def load_snapshot(self, timestamp):
"""Load order book snapshot tại thời điểm cụ thể"""
query = f"""
SELECT bid_levels, ask_levels
FROM orderbook_db.binance_depth
WHERE symbol = 'btcusdt'
AND timestamp <= toDateTime64('{timestamp}', 3)
ORDER BY timestamp DESC
LIMIT 1
"""
result = self.client.execute(query)
if result:
bids, asks = result[0]
self.current_book['bids'] = {float(p): float(q) for p, q in bids}
self.current_book['asks'] = {float(p): float(q) for p, q in asks}
return self.current_book
def get_best_bid_ask(self):
"""Lấy giá bid/ask tốt nhất hiện tại"""
if not self.current_book['bids'] or not self.current_book['asks']:
return None, None
best_bid = max(self.current_book['bids'].keys())
best_ask = min(self.current_book['asks'].keys())
return best_bid, best_ask
def simulate_market_order(self, side, quantity):
"""Mô phỏng đặt market order và tính slippage"""
if side == 'buy':
levels = sorted(self.current_book['asks'].items())
else:
levels = sorted(self.current_book['bids'].items(), reverse=True)
remaining = quantity
total_cost = 0
for price, qty in levels:
filled = min(remaining, qty)
total_cost += filled * price
remaining -= filled
if remaining <= 0:
break
return {
'filled': quantity - remaining,
'total_cost': total_cost,
'avg_price': total_cost / (quantity - remaining) if remaining < quantity else 0,
'slippage_bps': ((total_cost / (quantity - remaining) - self.get_best_bid_ask()[0 if side == 'buy' else 1]) / self.get_best_bid_ask()[0 if side == 'buy' else 1]) * 10000 if remaining < quantity else 0
}
Ví dụ sử dụng
replayer = OrderBookReplay(client)
replayer.load_snapshot("2024-01-01 10:30:00")
bid, ask = replayer.get_best_bid_ask()
print(f"Best Bid: {bid}, Best Ask: {ask}")
Mô phỏng mua 1 BTC
result = replayer.simulate_market_order('buy', 1.0)
print(f"Market Order Result: {result}")
Tạo Generator cho Batch Backtest
def backtest_generator(start_time, end_time, interval_seconds=60):
"""
Generator yield market state mỗi X giây
Dùng cho batch backtesting hiệu quả
"""
query = f"""
SELECT
timestamp,
symbol,
best_bid,
best_ask,
spread,
bid_levels,
ask_levels
FROM orderbook_db.binance_depth
WHERE symbol = 'btcusdt'
AND timestamp BETWEEN toDateTime64('{start_time}', 3)
AND toDateTime64('{end_time}', 3)
ORDER BY timestamp
"""
results = client.execute(query, with_column_types=True)
columns = [col[0] for col in results[1]]
for row in results[0]:
data = dict(zip(columns, row))
yield data
Ví dụ: Chạy backtest simple mean reversion
def simple_mean_reversion_backtest():
position = 0
trades = []
window = []
for market_data in backtest_generator("2024-01-01 00:00:00", "2024-01-01 23:59:59"):
mid_price = float(market_data['best_bid'] + market_data['best_ask']) / 2
window.append(mid_price)
if len(window) > 60: # 1 phút SMA
sma = sum(window[-60:]) / 60
spread = float(market_data['spread'])
# Chiến lược: mean reversion
if mid_price < sma - spread * 2 and position == 0:
position = 1
trades.append({'time': market_data['timestamp'], 'action': 'BUY', 'price': market_data['best_ask']})
elif mid_price > sma + spread * 2 and position == 1:
position = 0
trades.append({'time': market_data['timestamp'], 'action': 'SELL', 'price': market_data['best_bid']})
return trades
trades = simple_mean_reversion_backtest()
print(f"Total trades: {len(trades)}")
Phần 5: Tích hợp AI để phân tích Order Book
Bạn có thể dùng AI để phân tích patterns trong order book, detect anomalies, hoặc generate insights.
Đăng ký HolySheep AI để sử dụng với chi phí thấp hơn 85% so với OpenAI.
import requests
def analyze_order_book_pattern(client, symbol, start_time, end_time):
"""Dùng AI phân tích order book pattern"""
# Lấy summary statistics
query = f"""
SELECT
toStartOfHour(timestamp) as hour,
avg(spread) as avg_spread,
avg(bid_count) as avg_bid_depth,
avg(ask_count) as avg_ask_depth,
stddevPop(spread) as spread_volatility,
quantile(0.5)(spread) as median_spread
FROM orderbook_db.binance_depth
WHERE symbol = '{symbol}'
AND timestamp BETWEEN toDateTime64('{start_time}', 3)
AND toDateTime64('{end_time}', 3)
GROUP BY hour
ORDER BY hour
"""
results = client.execute(query)
# Format cho AI
summary = "\n".join([
f"Hour {r[0]}: Spread={r[1]:.4f}, BidDepth={r[2]}, AskDepth={r[3]}, Volatility={r[4]:.6f}"
for r in results[:24] # 24 giờ đầu
])
prompt = f"""Phân tích order book data cho {symbol}:
{summary}
Hãy nhận diện:
1. Các pattern liquidity (giờ nào liquidity thấp/cao)
2. Các anomaly bất thường
3. Gợi ý timing tốt nhất để đặt lệnh lớn
4. Cảnh báo potential spoofing hoặc wash trading"""
# Gọi HolySheep AI
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "Bạn là chuyên gia phân tích thị trường crypto với 10 năm kinh nghiệm."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 1000
}
)
return response.json()['choices'][0]['message']['content']
Sử dụng
analysis = analyze_order_book_pattern(
client,
"btcusdt",
"2024-01-01 00:00:00",
"2024-01-01 23:59:59"
)
print(analysis)
Phù hợp / Không phù hợp với ai
| Đối tượng phù hợp |
|---|
| ✅ Quantitative Traders | Cần backtest với dữ liệu order book thực, không phải OHLCV |
| ✅ Researchers | Nghiên cứu market microstructure, liquidity, price impact |
| ✅ ML Engineers | Tạo features từ order flow cho machine learning models |
| ✅ Exchange/Wallet Developers | Test integration với real-time order book data |
| ✅ Arbitrage Bots | Phát hiện cross-exchange opportunities với latency thấp |
| Đối tượng KHÔNG phù hợp |
|---|
| ❌ Retail Traders đơn thuần | Chỉ cần OHLCV data, không cần depth data phức tạp |
| ❌ Ngân sách hạn chế | Tardis + ClickHouse Cloud có chi phí hàng tháng |
| ❌ Beginners không biết SQL | Cần kiến thức cơ bản về database và Python |
| ❌ Real-time trading | Hệ thống này chỉ dùng cho backtesting, không phải production |
Giá và ROI
| Dịch vụ | Free Tier | Paid Plan | Chi phí ước tính/tháng |
|---|
| Tardis Machine | 10GB data, 3 days retention | $49-$499/month | $49-150 |
| ClickHouse Cloud | $0 (trial 30 days) | Starting $50/month | $50-200 |
| HolySheep AI | $5 credits miễn phí | DeepSeek V3.2: $0.42/1M tokens | $5-50 |
| Storage (S3) | 5GB | $0.023/GB | $1-10 |
| Tổng chi phí khởi điểm | $60-250/tháng |
**ROI Calculation**: Nếu bạn trade với volume $10,000/tháng và cải thiện execution price 0.1% nhờ backtest tốt hơn, bạn tiết kiệm được $10/tháng. Với chi phí $100/tháng, bạn cần cải thiện ít nhất 1% execution để break-even.
Vì sao chọn HolySheep AI cho phân tích Order Book
Khi phân tích hàng triệu order book snapshots, bạn cần AI không chỉ rẻ mà còn **nhanh**:
- Chi phí thấp nhất thị trường 2026: DeepSeek V3.2 chỉ $0.42/1M tokens — rẻ hơn 95% so với GPT-4.1 ($8)
- Độ trễ thấp (<50ms): Phân tích order book real-time không bị lag
- Hỗ trợ thanh toán local: WeChat Pay, Alipay cho người dùng Trung Quốc; thẻ quốc tế cho người dùng khác
- Tín dụng miễn phí khi đăng ký: Không rủi ro để thử nghiệm
| Provider | Giá/1M tokens | Độ trễ trung bình | Phù hợp cho |
|---|
| GPT-4.1 | $8.00 | ~200ms | Complex analysis, research |
| Claude Sonnet 4.5 | $15.00 | ~180ms | Detailed explanations |
| Gemini 2.5 Flash | $2.50 | ~100ms | Fast batch processing |
| DeepSeek V3.2 | $0.42 | <50ms | High-volume order book analysis |
Lỗi thường gặp và cách khắc phục
Lỗi 1: "Connection refused" khi kết nối ClickHouse
**Nguyên nhân**: ClickHouse container chưa khởi động hoàn toàn hoặc port bị conflict.
# Kiểm tra container status
docker ps -a | grep clickhouse
Restart nếu cần
docker restart clickhouse-server
Đợi và kiểm tra logs
docker logs clickhouse-server --tail 50
Nếu port 9000 bị占用, đổi port
docker run -d \
--name clickhouse-server \
-p 8124:8123 \
-p 9001:9000 \
clickhouse/clickhouse-server:latest
**Cách fix**: Thêm
timeout=30 trong Python connection và implement retry logic:
from clickhouse_driver import Client
import time
def connect_with_retry(max_retries=5):
for attempt in range(max_retries):
try:
client = Client(
'localhost',
port=9001, # Đổi port nếu cần
user='admin',
password='your_password',
connect_timeout=30,
send_receive_timeout=60
)
client.execute('SELECT 1')
print("Connected successfully!")
return client
except Exception as e:
print(f"Attempt {attempt+1} failed: {e}")
time.sleep(5)
raise Exception("Could not connect to ClickHouse")
client = connect_with_retry()
Lỗi 2: "Type mismatch" khi insert dữ liệu
**Nguyên nhân**: Data type không khớp với schema (ví dụ: String thay vì Decimal).
# Kiểm tra dtypes của DataFrame trước khi insert
print(df.dtypes)
Chuyển đổi đúng types
df['best_bid'] = df['best_bid'].astype(float)
df['best_ask'] = df['best_ask'].astype(float)
df['spread'] = df['spread'].astype(float)
Nếu timestamp bị timezone issue
df['timestamp'] = pd.to_datetime(df['timestamp']).dt.tz_localize(None)
Verify lại
print("After conversion:")
print(df[['timestamp', 'best_bid', 'best_ask']].head())
**Cách fix**: Thêm type checking trong insert function:
def validate_and_insert(df, client):
required_columns = ['timestamp', 'symbol', 'best_bid', 'best_ask', 'spread',
'bid_count', 'ask_count', 'bid_levels', 'ask_levels']
missing = set(required_columns) - set(df.columns)
if missing:
raise ValueError(f"Missing columns: {missing}")
# Convert types
df['timestamp'] = pd.to_datetime(df['timestamp'])
for col in ['best_bid', 'best_ask', 'spread']:
df[col] = pd.to_numeric(df[col], errors='coerce')
# Drop rows with NaN
df = df.dropna(subset=['timestamp', 'best_bid', 'best_ask'])
# Insert
data = df[required_columns].values.tolist()
client.execute(
'INSERT INTO orderbook_db.binance_depth VALUES',
data
)
print(f"Inserted {len(df)} valid rows")
Lỗi 3: Tardis API quota exceeded
**Nguyên nhân**: Quá nhiều data request trong tháng, vượt free tier limits.
# Kiểm tra usage trước
import requests
TARDIS_API_KEY = "YOUR_TARDIS_API_KEY"
def check_tardis_usage():
response = requests.get(
"https://api.tardis.dev/v1/usage",
headers={"Authorization": f"Bearer {TARDIS_API_KEY}"}
)
data = response.json()
print(f"Used: {data['usedBytes'] / 1024**3:.2f} GB")
print(f"Limit: {data['planBytes'] / 1024**3:.2f} GB")
return data
check_tardis_usage()
**Cách fix**: Implement batched download với rate limiting:
import asyncio
import aiohttp
from datetime import datetime, timedelta
async def download_with_rate_limit(symbol, start, end, max_bytes_per_request=100*1024*1024):
"""
Download với chunking thông minh
"""
current = datetime.strptime(start, "%Y-%m-%d")
end_date = datetime.strptime(end, "%Y-%m-%d")
while current < end_date:
# Ước tính chunk size (1 ngày cho BTCUSDT ~50MB)
chunk_end = current + timedelta(days=1)
if chunk_end > end_date:
chunk_end = end_date
# Thử download
try:
await download_chunk(symbol, current, chunk_end)
print(f"Downloaded {current.date()} to {chunk_end.date()}")
except Exception as e:
if "quota" in str(e).lower():
print(f"Quota exceeded, waiting 1 hour...")
await asyncio.sleep(3600)
continue
raise
current = chunk_end
await asyncio.sleep(1) # Rate limit 1 request/second
async def download_chunk(symbol, start, end):
# Implementation here
pass
Lỗi 4: Memory error khi xử lý large parquet files
**Nguyên nhân**: Parquet file quá lớn, không đủ RAM để đọc一次性.
# Thay vì đọc toàn bộ, dùng chunked reading
import pandas as pd
def read_parquet_chunks(file_path, chunk_size=50000):
"""Đọc parquet theo chunks"""
for chunk in pd.read_parquet(file_path, columns=[
'timestamp', 'symbol', 'best_bid', 'best_ask', 'spread'
]):
yield chunk
def process_incremental(parquet_file, client):
"""Xử lý và insert theo chunks"""
total_processed = 0
for i, chunk in enumerate(read_parquet_chunks(parquet_file)):
# Transform chunk
chunk['timestamp'] = pd.to_datetime(chunk['timestamp'])
# Insert chunk
data = chunk.values.tolist()
client.execute(
'INSERT INTO orderbook_db.binance_depth VALUES',
data
)
total_processed += len(chunk)
print(f"Chunk {i+1}: inserted {len(chunk)} rows, total: {total_processed}")
# Clear memory
del chunk
return total_processed
Tổng kết
Trong bài viết này, tôi đã hướng dẫn bạn:
- ✅ Cách lấy dữ liệu L2 order book từ Tardis Machine
- ✅ Cài đặt và cấu hình ClickHouse (Docker hoặc Cloud)
- ✅ Thiết kế schema tối ưu cho order book data
- ✅ Import dữ liệu từ Parquet vào ClickHouse
- ✅ Xây dựng hệ thống回放 để backtest chiến lược
- ✅ Tích hợp AI để phân tích patterns
- ✅ Xử lý 4 lỗi phổ biến nhất
**Kinh nghiệm thực chiến của tôi**: Đừng tiết kiệm chi phí cho storage. Tôi từng dùng
Tài nguyên liên quan
Bài viết liên quan