Trong thế giới giao dịch tiền mã hóa hiện đại, việc backtest (kiểm thử ngược) chiến lược là yếu tố sống còn quyết định sự thành bại của một hệ thống trading. Một công ty trading desk tại TP.HCM — chuyên phát triển bot giao dịch futures trên Binance — đã phải đối mặt với bài toán nan giản: làm sao có được dữ liệu orderbook lịch sử chất lượng cao, độ trễ thấp, và chi phí hợp lý để backtest các chiến lược arbitrage và market-making của họ?
Bối cảnh kinh doanh: Đội ngũ 8 người với 3 trader chuyên nghiệp và 5 kỹ sư quant. Họ vận hành các chiến lược delta-neutral, arbitrage biên độ giá giữa spot và futures, và market-making trên các cặp BTC/USDT, ETH/USDT perpetual futures.
Điểm đau của nhà cung cấp cũ: Ban đầu, đội ngũ sử dụng dữ liệu ticker và trade history miễn phí từ Binance API. Kết quả? Chi phí infrastructure để lưu trữ và xử lý dữ liệu lên tới $3,200/tháng (server, storage, bandwidth). Độ trễ trung bình khi truy xuất dữ liệu orderbook lịch sử là 2.3 giây do phải query từ PostgreSQL database không được tối ưu. Quan trọng hơn, chất lượng dữ liệu không đủ để backtest các chiến lược đòi hỏi độ chính xác cao — nhiều khoảng trống (gap) trong dữ liệu, thiếu các mức giá trong orderbook.
Lý do chọn Tardis.dev: Sau khi đánh giá 4 giải pháp (CoinAPI, Kaiko, Bloomberg API, và Tardis.dev), đội ngũ quyết định chọn Tardis.dev vì 3 lý do chính: (1) Dữ liệu orderbook được replay với độ phân giải millisecond, (2) API thiết kế theo chuẩn WebSocket reactive, phù hợp với kiến trúc Python async hiện tại, (3) Chi phí tính theo volume data thay vì subscription cố định — phù hợp với mô hình startup đang scale.
Chi phí hóa đơn hàng tháng: $4,200 → $680 sau khi migrate sang Tardis.dev. Độ trễ truy xuất giảm từ 2.3 giây xuống còn 180ms trung bình. Thời gian để hoàn thành một bộ backtest đầy đủ giảm từ 6 giờ xuống còn 45 phút.
Tardis.dev là gì và tại sao nó phù hợp cho backtest?
Tardis.dev là nền tảng cung cấp dữ liệu market data theo thời gian thực và lịch sử (historical data) cho các sàn giao dịch tiền mã hóa. Khác với các API thông thường chỉ trả về OHLCV (Open, High, Low, Close, Volume), Tardis.dev cung cấp:
- Orderbook snapshots: Trạng thái đầy đủ của sổ lệnh tại một thời điểm
- Incremental orderbook updates: Các thay đổi delta theo thời gian thực
- Trade data: Toàn bộ lịch sử giao dịch với timestamp microsecond
- Funding rate history: Lịch sử funding rate cho perpetual futures
Điểm mạnh của Tardis.dev so với Binance chính thức API là khả năng truy xuất historical data với cấu trúc normalized, chuẩn hóa giữa các sàn, và không bị giới hạn rate limit như khi dùng public API của sàn.
Cài đặt môi trường và thư viện
Trước khi bắt đầu, bạn cần cài đặt Python 3.10+ và các thư viện cần thiết. Chúng tôi khuyến nghị sử dụng virtual environment để quản lý dependencies.
# Tạo virtual environment và cài đặt dependencies
python -m venv venv_tardis
source venv_tardis/bin/activate # Linux/Mac
hoặc: venv_tardis\Scripts\activate # Windows
Cài đặt tardis-client - thư viện chính thức của Tardis.dev
pip install tardis-dev
Các thư viện bổ sung cho data analysis
pip install pandas numpy matplotlib jupyter
Thư viện hỗ trợ async programming
pip install asyncio nest-asyncio
Kiểm tra version
python -c "import tardis; print(tardis.__version__)"Tardis.dev cung cấp API endpoint chính tại https://api.tardis.dev/v1 với các endpoint cho futures data:
GET /exchanges/{exchange}/symbols- Danh sách các cặp giao dịchGET /exchanges/{exchange}/symbols/{symbol}/book- Historical orderbook dataWS /realtime- WebSocket stream cho dữ liệu real-time
Kết nối và truy xuất dữ liệu Orderbook
Để truy xuất dữ liệu orderbook lịch sử từ Binance Futures, chúng ta sử dụng Tardis.dev client với exchange name là binance-futures. Dưới đây là code hoàn chỉnh để kết nối và truy xuất dữ liệu.
import asyncio
from tardis.devices.exchanges.binance_futures import BinanceFuturesDevice
from tardis.interface.ts_dataclasses import BookL1
from datetime import datetime, timedelta
import pandas as pd
from typing import List, Dict
class BinanceFuturesDataFetcher:
"""Class truy xuất dữ liệu orderbook từ Tardis.dev cho Binance Futures"""
def __init__(self, api_key: str):
"""
Khởi tạo Data Fetcher
Args:
api_key: API key từ Tardis.dev dashboard
"""
self.api_key = api_key
self.device = BinanceFuturesDevice(
api_key=api_key,
exchange="binance-futures"
)
self.base_url = "https://api.tardis.dev/v1"
async def fetch_orderbook_snapshot(
self,
symbol: str,
date: datetime
) -> List[BookL1]:
"""
Lấy một snapshot orderbook tại thời điểm cụ thể
Args:
symbol: Cặp giao dịch, ví dụ: 'BTCUSDT'
date: Thời điểm cần lấy dữ liệu
Returns:
Danh sách BookL1 objects chứa bid/ask levels
"""
# Format timestamp theo chuẩn ISO 8601
timestamp = date.strftime("%Y-%m-%dT%H:%M:%S.000Z")
# Call API endpoint
endpoint = f"{self.base_url}/exchanges/binance-futures/book"
params = {
"symbol": symbol,
"timestamp": timestamp,
"limit": 100 # Số lượng levels mỗi side
}
response = await self.device.http_get(endpoint, params=params)
return response
async def fetch_historical_orderbook(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
interval: str = "1m"
) -> pd.DataFrame:
"""
Lấy dữ liệu orderbook trong khoảng thời gian
Args:
symbol: Cặp giao dịch
start_date: Thời điểm bắt đầu
end_date: Thời điểm kết thúc
interval: Khoảng thời gian giữa các snapshot
Returns:
DataFrame chứa dữ liệu orderbook
"""
snapshots = []
current_date = start_date
# Tính interval delta
interval_map = {
"1m": timedelta(minutes=1),
"5m": timedelta(minutes=5),
"15m": timedelta(minutes=15),
"1h": timedelta(hours=1)
}
delta = interval_map.get(interval, timedelta(minutes=1))
print(f"Fetching orderbook data for {symbol}...")
print(f"Period: {start_date} to {end_date}")
while current_date < end_date:
try:
snapshot = await self.fetch_orderbook_snapshot(
symbol, current_date
)
snapshots.append(snapshot)
# Progress indicator
progress = (current_date - start_date) / (end_date - start_date)
print(f"Progress: {progress*100:.1f}%", end="\r")
current_date += delta
# Rate limit handling - Tardis.dev cho phép 10 requests/giây
await asyncio.sleep(0.1)
except Exception as e:
print(f"\nError at {current_date}: {e}")
# Retry sau 5 giây
await asyncio.sleep(5)
continue
return self._process_snapshots(snapshots)
def _process_snapshots(self, snapshots: List) -> pd.DataFrame:
"""Xử lý raw snapshots thành DataFrame"""
records = []
for snapshot in snapshots:
for bid in snapshot.bids:
records.append({
'timestamp': snapshot.timestamp,
'side': 'bid',
'price': float(bid.price),
'size': float(bid.size),
'level': bid.level
})
for ask in snapshot.asks:
records.append({
'timestamp': snapshot.timestamp,
'side': 'ask',
'price': float(ask.price),
'size': float(ask.size),
'level': ask.level
})
df = pd.DataFrame(records)
if not df.empty:
df['spread'] = df[df['side']=='ask']['price'] - df[df['side']=='bid']['price']
df['mid_price'] = (df[df['side']=='ask']['price'] + df[df['side']=='bid']['price']) / 2
df['imbalance'] = self._calculate_imbalance(df)
return df
def _calculate_imbalance(self, df: pd.DataFrame) -> pd.Series:
"""Tính orderbook imbalance"""
bids = df[df['side']=='bid'].groupby('timestamp')['size'].sum()
asks = df[df['side']=='ask'].groupby('timestamp')['size'].sum()
imbalance = (bids - asks) / (bids + asks)
return imbalance
Sử dụng example
async def main():
fetcher = BinanceFuturesDataFetcher(api_key="YOUR_TARDIS_API_KEY")
# Lấy dữ liệu 1 ngày cho BTCUSDT perpetual
df = await fetcher.fetch_historical_orderbook(
symbol="BTCUSDT",
start_date=datetime(2026, 1, 15, 0, 0, 0),
end_date=datetime(2026, 1, 16, 0, 0, 0),
interval="1m"
)
# Lưu vào CSV để sử dụng sau
df.to_csv("btcusdt_orderbook_20260115.csv", index=False)
print(f"\nData saved: {len(df)} records")
print(df.head())
asyncio.run(main())Xây dựng chiến lược Backtest với dữ liệu Orderbook
Sau khi có dữ liệu orderbook, bước tiếp theo là xây dựng framework backtest để đánh giá chiến lược giao dịch. Dưới đây là một ví dụ hoàn chỉnh về chiến lược market-making dựa trên orderbook imbalance.
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime
@dataclass
class Trade:
"""Biểu diễn một giao dịch"""
timestamp: datetime
side: str # 'buy' or 'sell'
price: float
size: float
pnl: float = 0.0
fees: float = 0.0
@dataclass
class OrderbookLevel:
"""Một level trong orderbook"""
price: float
size: float
class MarketMakingStrategy:
"""
Chiến lược Market Making cơ bản
- Mua khi orderbook imbalance > threshold
- Bán khi orderbook imbalance < -threshold
- Spread được tính dựa trên volatility
"""
def __init__(
self,
symbol: str,
base_size: float = 0.001, # Kích thước lệnh cơ bản (BTC)
imbalance_threshold: float = 0.3, # Ngưỡng imbalance
max_spread_bps: float = 50, # Spread tối đa (basis points)
maker_fee: float = 0.0002, # Phí maker 0.02%
):
self.symbol = symbol
self.base_size = base_size
self.imbalance_threshold = imbalance_threshold
self.max_spread_bps = max_spread_bps
self.maker_fee = maker_fee
self.position = 0.0
self.trades: List[Trade] = []
self.equity_curve = []
def calculate_spread(self, mid_price: float, volatility: float) -> float:
"""
Tính spread tối ưu dựa trên mid price và volatility
Args:
mid_price: Giá trung vị hiện tại
volatility: Độ biến động (tính bằng standard deviation)
Returns:
Spread tối ưu theo đơn vị price
"""
# Adaptive spread: cao hơn khi volatility cao
base_spread = mid_price * (self.max_spread_bps / 10000)
adaptive_spread = base_spread * (1 + volatility * 10)
return min(adaptive_spread, mid_price * 0.001) # Cap at 0.1%
def calculate_order_size(
self,
imbalance: float,
base_size: float
) -> float:
"""
Tính kích thước lệnh dựa trên orderbook imbalance
Args:
imbalance: Giá trị imbalance [-1, 1]
base_size: Kích thước cơ bản
Returns:
Kích thước lệnh điều chỉnh
"""
# Khi imbalance > 0 (nhiều bid hơn), tăng size bán
# Khi imbalance < 0 (nhiều ask hơn), tăng size mua
size_multiplier = 1 + abs(imbalance) * 0.5
return base_size * size_multiplier
def generate_signal(
self,
mid_price: float,
imbalance: float,
volatility: float
) -> Optional[dict]:
"""
Sinh tín hiệu giao dịch từ dữ liệu orderbook
Args:
mid_price: Giá trung vị
imbalance: Orderbook imbalance
volatility: Độ biến động
Returns:
Dictionary chứa tín hiệu hoặc None
"""
spread = self.calculate_spread(mid_price, volatility)
half_spread = spread / 2
# Kiểm tra ngưỡng imbalance
if imbalance > self.imbalance_threshold:
# Imbalance cao -> có áp lực mua -> chúng ta bán
size = self.calculate_order_size(imbalance, self.base_size)
return {
'action': 'sell',
'price': mid_price + half_spread,
'size': size,
'reason': f'High bid imbalance: {imbalance:.3f}'
}
elif imbalance < -self.imbalance_threshold:
# Imbalance thấp -> có áp lực bán -> chúng ta mua
size = self.calculate_order_size(imbalance, self.base_size)
return {
'action': 'buy',
'price': mid_price - half_spread,
'size': abs(size),
'reason': f'High ask imbalance: {imbalance:.3f}'
}
return None
def execute_trade(
self,
timestamp: datetime,
signal: dict
) -> Trade:
"""Thực thi một giao dịch và cập nhật position"""
trade = Trade(
timestamp=timestamp,
side=signal['action'],
price=signal['price'],
size=signal['size'],
fees=signal['price'] * signal['size'] * self.maker_fee
)
if signal['action'] == 'buy':
self.position += signal['size']
trade.pnl = -signal['size'] * signal['price'] - trade.fees
else:
self.position -= signal['size']
trade.pnl = signal['size'] * signal['price'] - trade.fees
self.trades.append(trade)
return trade
class BacktestEngine:
"""Engine thực thi backtest"""
def __init__(self, strategy: MarketMakingStrategy, initial_capital: float = 100000):
self.strategy = strategy
self.initial_capital = initial_capital
self.current_capital = initial_capital
self.equity_curve = []
def run(
self,
orderbook_df: pd.DataFrame,
volatility_window: int = 60
) -> dict:
"""
Chạy backtest với dữ liệu orderbook
Args:
orderbook_df: DataFrame chứa dữ liệu orderbook
volatility_window: Window để tính volatility
Returns:
Dictionary chứa kết quả backtest
"""
print(f"Starting backtest for {self.strategy.symbol}")
print(f"Initial capital: ${self.initial_capital:,.2f}")
# Group by timestamp để xử lý theo snapshot
grouped = orderbook_df.groupby('timestamp')
results = []
for timestamp, group in grouped:
# Tính mid price và imbalance
bids = group[group['side']=='bid'].sort_values('price', ascending=False)
asks = group[group['side']=='ask'].sort_values('price')
if bids.empty or asks.empty:
continue
best_bid = bids.iloc[0]['price']
best_ask = asks.iloc[0]['price']
mid_price = (best_bid + best_ask) / 2
# Tính imbalance
total_bid_size = bids['size'].sum()
total_ask_size = asks['size'].sum()
imbalance = (total_bid_size - total_ask_size) / (total_bid_size + total_ask_size)
# Tính volatility từ dữ liệu quá khứ
volatility = self._calculate_volatility(
orderbook_df, timestamp, volatility_window
)
# Sinh tín hiệu
signal = self.strategy.generate_signal(
mid_price, imbalance, volatility
)
if signal:
trade = self.strategy.execute_trade(timestamp, signal)
self.current_capital += trade.pnl
results.append({
'timestamp': timestamp,
**signal,
'position': self.strategy.position,
'capital': self.current_capital,
'pnl': trade.pnl,
'fees': trade.fees
})
# Cập nhật equity curve
if len(results) > 0:
position_value = self.strategy.position * mid_price
self.equity_curve.append({
'timestamp': timestamp,
'total_equity': self.current_capital + position_value,
'cash': self.current_capital,
'position_value': position_value
})
return self._generate_report(results)
def _calculate_volatility(
self,
df: pd.DataFrame,
current_time: datetime,
window: int
) -> float:
"""Tính volatility của mid price trong window"""
historical = df[df['timestamp'] < current_time].tail(window)
if len(historical) < 2:
return 0.0
return historical['mid_price'].std() / historical['mid_price'].mean()
def _generate_report(self, trades: List[dict]) -> dict:
"""Tạo báo cáo kết quả backtest"""
if not trades:
return {'error': 'No trades executed'}
trades_df = pd.DataFrame(trades)
total_pnl = self.current_capital - self.initial_capital
total_return = (total_pnl / self.initial_capital) * 100
# Tính các metrics
winning_trades = trades_df[trades_df['pnl'] > 0]
losing_trades = trades_df[trades_df['pnl'] <= 0]
win_rate = len(winning_trades) / len(trades_df) * 100
avg_win = winning_trades['pnl'].mean() if len(winning_trades) > 0 else 0
avg_loss = abs(losing_trades['pnl'].mean()) if len(losing_trades) > 0 else 0
profit_factor = abs(winning_trades['pnl'].sum() / losing_trades['pnl'].sum()) if len(losing_trades) > 0 and losing_trades['pnl'].sum() != 0 else float('inf')
# Tính Sharpe Ratio (simplified)
returns = trades_df['pnl'].pct_change().dropna()
sharpe = returns.mean() / returns.std() * np.sqrt(252) if returns.std() > 0 else 0
# Max Drawdown
equity_df = pd.DataFrame(self.equity_curve)
equity_df['cummax'] = equity_df['total_equity'].cummax()
equity_df['drawdown'] = (equity_df['total_equity'] - equity_df['cummax']) / equity_df['cummax']
max_drawdown = equity_df['drawdown'].min() * 100
report = {
'symbol': self.strategy.symbol,
'period': f"{trades_df['timestamp'].min()} to {trades_df['timestamp'].max()}",
'total_trades': len(trades_df),
'initial_capital': self.initial_capital,
'final_capital': self.current_capital,
'total_pnl': total_pnl,
'total_return_pct': total_return,
'win_rate': win_rate,
'avg_win': avg_win,
'avg_loss': avg_loss,
'profit_factor': profit_factor,
'sharpe_ratio': sharpe,
'max_drawdown_pct': max_drawdown,
'trades': trades_df
}
self._print_report(report)
return report
def _print_report(self, report: dict):
"""In báo cáo ra console"""
print("\n" + "="*60)
print("BACKTEST REPORT")
print("="*60)
print(f"Symbol: {report['symbol']}")
print(f"Period: {report['period']}")
print(f"\nTotal Trades: {report['total_trades']}")
print(f"\nCapital:")
print(f" Initial: ${report['initial_capital']:,.2f}")
print(f" Final: ${report['final_capital']:,.2f}")
print(f" P&L: ${report['total_pnl']:,.2f} ({report['total_return_pct']:.2f}%)")
print(f"\nPerformance Metrics:")
print(f" Win Rate: {report['win_rate']:.2f}%")
print(f" Avg Win: ${report['avg_win']:.4f}")
print(f" Avg Loss: ${report['avg_loss']:.4f}")
print(f" Profit Factor: {report['profit_factor']:.2f}")
print(f" Sharpe Ratio: {report['sharpe_ratio']:.2f}")
print(f" Max Drawdown: {report['max_drawdown_pct']:.2f}%")
print("="*60)
Chạy backtest
async def run_backtest():
# Load dữ liệu orderbook đã lưu
df = pd.read_csv("btcusdt_orderbook_20260115.csv")
df['timestamp'] = pd.to_datetime(df['timestamp'])
# Khởi tạo strategy
strategy = MarketMakingStrategy(
symbol="BTCUSDT",
base_size=0.01, # 0.01 BTC per trade
imbalance_threshold=0.35,
max_spread_bps=40
)
# Chạy backtest
engine = BacktestEngine(
strategy=strategy,
initial_capital=50000 # $50,000 starting capital
)
report = engine.run(df)
return report
asyncio.run(run_backtest())
Phân tích kết quả và tối ưu hóa tham số
Sau khi chạy backtest, việc phân tích kết quả và tối ưu hóa tham số là bước quan trọng để cải thiện performance của chiến lược. Dưới đây là code để visualize kết quả và thực hiện parameter optimization.
import matplotlib.pyplot as plt
import seaborn as sns
from itertools import product
import pandas as pd
import numpy as np
class ParameterOptimizer:
"""Tối ưu hóa tham số chiến lược sử dụng grid search"""
def __init__(self, base_strategy_class, orderbook_df: pd.DataFrame):
self.StrategyClass = base_strategy_class
self.orderbook_df = orderbook_df
self.results = []
def optimize(
self,
param_grid: dict,
metric: str = 'total_return_pct'
) -> pd.DataFrame:
"""
Grid search để tìm tham số tối ưu
Args:
param_grid: Dictionary chứa các tham số cần tối ưu
metric: Metric để đánh giá
Returns:
DataFrame chứa kết quả tất cả combinations
"""
# Generate all combinations
keys = param_grid.keys()
values = param_grid.values()
combinations = [dict(zip(keys, v)) for v in product(*values)]
print(f"Testing {len(combinations)} parameter combinations...")
for i, params in enumerate(combinations):
print(f"[{i+1}/{len(combinations)}] Testing: {params}")
strategy = self.StrategyClass(
symbol="BTCUSDT",
**params
)
engine = BacktestEngine(
strategy=strategy,
initial_capital=50000
)
report = engine.run(self.orderbook_df)
report['params'] = str(params)
self.results.append(report)
return self._analyze_results()