Trong thế giới trading thuật toán, việc backtest (kiểm thử ngược) là bước không thể bỏ qua trước khi triển khai bất kỳ chiến lược nào lên thị trường thực. Đặc biệt với market making tần suất cao (High-Frequency Market Making - HFMM), độ chính xác của môi trường backtest quyết định số phận của chiến lược khi ra mắt. Bài viết này sẽ hướng dẫn bạn xây dựng từ A đến Z một hệ thống backtest hoàn chỉnh sử dụng Tardis.dev — công cụ được giới quantitative trader đánh giá cao nhờ khả năng cung cấp dữ liệu order book và trade chất lượng cao với độ trễ thấp.
Tại Sao Tardis.dev Là Lựa Chọn Hàng Đầu Cho Market Making Backtest?
Theo kinh nghiệm thực chiến của tôi qua 3 năm xây dựng hệ thống trading, Tardis.dev nổi bật với 3 điểm mạnh then chốt:
- Replay Engine chính xác cao: Cho phép tái hiện order book state tại bất kỳ thời điểm nào với độ trễ dưới 1ms
- Dữ liệu Level 2 đầy đủ: Full order book depth với giá, kích thước, và timestamp microsecond
- Hỗ trợ đa sàn: Binance, Bybit, OKX, Coinbase, FTX (lịch sử) với format thống nhất
Tuy nhiên, chi phí API Tardis.dev có thể là rào cản với individual trader. Ở phần so sánh cuối bài, tôi sẽ giới thiệu giải pháp tối ưu chi phí sử dụng HolySheep AI cho các tác vụ xử lý dữ liệu và phân tích kết quả backtest.
Kiến Trúc Tổng Quan Hệ Thống Backtest
Trước khi đi vào chi tiết kỹ thuật, hãy hiểu rõ kiến trúc tổng thể:
┌─────────────────────────────────────────────────────────────────┐
│ HỆ THỐNG BACKTEST HFMM │
├─────────────────────────────────────────────────────────────────┤
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Tardis.dev │───▶│ Replay │───▶│ Strategy │ │
│ │ Historical │ │ Engine │ │ Engine │ │
│ │ Data │ │ │ │ │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Order │ │ P&L │ │ HolySheep │ │
│ │ Book │ │ Tracker │ │ Analysis │ │
│ │ State │ │ │ │ (AI) │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘Cài Đặt Môi Trường Và Công Cụ
2.1. Yêu Cầu Hệ Thống
Để chạy backtest market making hiệu quả, bạn cần:
- CPU: Ít nhất 8 cores (khuyến nghị 16 cores cho parallel backtest)
- RAM: Tối thiểu 32GB (order book state có thể ngốn 16-32GB)
- Storage: SSD NVMe 1TB+ (đọc dữ liệu với tốc độ cao)
- OS: Ubuntu 22.04 LTS hoặc macOS M1/M2
2.2. Cài Đặt Python Dependencies
# Tạo virtual environment
python3.11 -m venv venv_hfmm
source venv_hfmm/bin/activate
Cài đặt packages cần thiết
pip install --upgrade pip
pip install numpy==1.24.3 pandas==2.0.3 numba==0.58.0
pip install tardis-dev==1.0.0a7 # Alpha version với replay features
pip install asyncio-redis==0.16.0
pip install aiofiles==23.2.1
pip install python-dotenv==1.0.0
Kiểm tra cài đặt
python -c "import tardis; print(f'Tardis version: {tardis.__version__}')"Tải Dữ Liệu Lịch Sử Từ Tardis.dev
3.1. Lấy API Key
Sau khi đăng ký tài khoản Tardis.dev, bạn sẽ nhận được API key. Hãy lưu trữ an toàn:
# Cấu trúc thư mục project
mkdir -p hfmm_backtest/{config,data,logs,strategies,analysis}
cd hfmm_backtest
Tạo file .env
cat > .env << 'EOF'
TARDIS_API_KEY=your_tardis_api_key_here
EXCHANGE=binance
SYMBOL=BTC-USDT
START_DATE=2024-01-01
END_DATE=2024-01-31
EOF
Load environment variables
export $(cat .env | xargs)3.2. Script Tải Dữ Liệu Order Book
#!/usr/bin/env python3
"""
Download và cache historical order book data từ Tardis.dev
Performance: ~500MB/ngày cho BTC-USDT Level 2 data
"""
import os
import asyncio
import aiofiles
from tardis import Tardis
from tardis.buffers import BinanceBuffer
from datetime import datetime, timedelta
from dotenv import load_dotenv
load_dotenv()
class DataDownloader:
def __init__(self, api_key: str, exchange: str, symbol: str):
self.api_key = api_key
self.exchange = exchange
self.symbol = symbol
self.buffer_class = self._get_buffer_class()
def _get_buffer_class(self):
buffers = {
'binance': BinanceBuffer,
'bybit': None, # Implement theo format Bybit
'okx': None,
}
return buffers.get(self.exchange, BinanceBuffer)
async def download_daily_data(self, date: str, output_dir: str):
"""Download data cho một ngày cụ thể"""
date_obj = datetime.strptime(date, '%Y-%m-%d')
client = Tardis(api_key=self.api_key)
# Format: exchange symbol date
dataset_name = f"{self.exchange}-{self.symbol.lower().replace('-', '')}-{date.replace('-', '')}"
print(f"📥 Downloading: {dataset_name}")
print(f"⏱️ Estimated size: ~500MB")
print(f"⏱️ Estimated time: 5-10 phút")
start_time = datetime.now()
# Stream data về local
async with client.datasets.stream(dataset_name) as stream:
async for buffer in stream.buffers(self.buffer_class):
# Lưu buffer ra file
output_file = os.path.join(
output_dir,
f"{date}_{buffer.timestamp}.bin"
)
async with aiofiles.open(output_file, 'wb') as f:
await f.write(buffer.raw)
elapsed = (datetime.now() - start_time).total_seconds()
print(f"✅ Hoàn thành trong {elapsed:.1f} giây")
return output_file
async def download_range(self, start_date: str, end_date: str, output_dir: str):
"""Download data cho một khoảng thời gian"""
os.makedirs(output_dir, exist_ok=True)
start = datetime.strptime(start_date, '%Y-%m-%d')
end = datetime.strptime(end_date, '%Y-%m-%d')
current = start
total_days = (end - start).days + 1
completed = 0
while current <= end:
date_str = current.strftime('%Y-%m-%d')
try:
await self.download_daily_data(date_str, output_dir)
completed += 1
print(f"📊 Progress: {completed}/{total_days} days")
except Exception as e:
print(f"❌ Lỗi ngày {date_str}: {e}")
current += timedelta(days=1)
if __name__ == "__main__":
downloader = DataDownloader(
api_key=os.getenv('TARDIS_API_KEY'),
exchange=os.getenv('EXCHANGE', 'binance'),
symbol=os.getenv('SYMBOL', 'BTC-USDT')
)
asyncio.run(downloader.download_range(
start_date=os.getenv('START_DATE'),
end_date=os.getenv('END_DATE'),
output_dir='data/raw'
))Xây Dựng Order Book Replay Engine
Đây là phần cốt lõi của hệ thống. Order Book Replay Engine cần đảm bảo:
- State consistency: Trạng thái order book tại mỗi thời điểm phải chính xác
- Event ordering: Messages phải được xử lý đúng thứ tự timestamp
- Incremental update: Hiệu quả bộ nhớ khi xử lý full order book depth
#!/usr/bin/env python3
"""
Order Book Replay Engine - Core Component
Performance target: Xử lý 100,000+ events/giây
"""
import asyncio
import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Tuple, Optional
from collections import defaultdict
import heapq
@dataclass(slots=True)
class OrderBookLevel:
"""Một level trong order book"""
price: float
size: float
order_count: int
timestamp: int # Microseconds
@dataclass
class OrderBookState:
"""Trạng thái full order book tại một thời điểm"""
bids: Dict[float, OrderBookLevel] = field(default_factory=dict)
asks: Dict[float, OrderBookLevel] = field(default_factory=dict)
last_update_ts: int = 0
sequence: int = 0
def get_best_bid(self) -> Optional[float]:
if not self.bids:
return None
return max(self.bids.keys())
def get_best_ask(self) -> Optional[float]:
if not self.asks:
return None
return min(self.asks.keys())
def get_mid_price(self) -> Optional[float]:
best_bid = self.get_best_bid()
best_ask = self.get_best_ask()
if best_bid is None or best_ask is None:
return None
return (best_bid + best_ask) / 2
def get_spread(self) -> Optional[float]:
best_bid = self.get_best_bid()
best_ask = self.get_best_ask()
if best_bid is None or best_ask is None:
return None
return best_ask - best_bid
def get_spread_bps(self) -> Optional[float]:
"""Spread in basis points"""
spread = self.get_spread()
mid = self.get_mid_price()
if spread is None or mid is None or mid == 0:
return None
return (spread / mid) * 10000
class OrderBookReplayEngine:
"""
High-performance order book replay engine
Sử dụng priority queue cho event ordering
"""
def __init__(self, max_depth: int = 50):
self.max_depth = max_depth
self.state = OrderBookState()
self.event_queue: List[Tuple[int, dict]] = [] # (timestamp, event)
self.trade_history: List[dict] = []
self.callbacks: List[callable] = []
def add_callback(self, callback: callable):
"""Đăng ký callback được gọi sau mỗi state update"""
self.callbacks.append(callback)
def process_order_add(self, price: float, size: float, side: str,
timestamp: int, order_id: str):
"""Xử lý order mới"""
level = OrderBookLevel(
price=price,
size=size,
order_count=1,
timestamp=timestamp
)
book_side = self.state.bids if side == 'bid' else self.state.asks
book_side[price] = level
self._prune_depth(side)
self._notify_callbacks({
'type': 'add',
'price': price,
'size': size,
'side': side,
'timestamp': timestamp
})
def process_order_modify(self, price: float, new_size: float, side: str,
timestamp: int, order_id: str):
"""Xử lý order modification"""
book_side = self.state.bids if side == 'bid' else self.state.asks
if price in book_side:
book_side[price].size = new_size
book_side[price].timestamp = timestamp
self._notify_callbacks({
'type': 'modify',
'price': price,
'size': new_size,
'side': side,
'timestamp': timestamp
})
def process_order_delete(self, price: float, side: str,
timestamp: int, order_id: str):
"""Xử lý order deletion"""
book_side = self.state.bids if side == 'bid' else self.state.asks
if price in book_side:
del book_side[price]
self._notify_callbacks({
'type': 'delete',
'price': price,
'side': side,
'timestamp': timestamp
})
def process_trade(self, price: float, size: float, side: str,
timestamp: int, trade_id: str):
"""Xử lý trade execution"""
self.trade_history.append({
'price': price,
'size': size,
'side': side,
'timestamp': timestamp,
'trade_id': trade_id,
'mid_price': self.state.get_mid_price()
})
self._notify_callbacks({
'type': 'trade',
'price': price,
'size': size,
'side': side,
'timestamp': timestamp
})
def _prune_depth(self, side: str):
"""Giữ chỉ max_depth levels để tiết kiệm memory"""
book_side = self.state.bids if side == 'bid' else self.state.asks
if len(book_side) > self.max_depth:
# Với bids, giữ top N theo giá
# Với asks, giữ top N theo giá thấp nhất
if side == 'bid':
sorted_prices = sorted(book_side.keys(), reverse=True)
else:
sorted_prices = sorted(book_side.keys())
# Xóa các price levels ngoài depth
prices_to_remove = sorted_prices[self.max_depth:]
for price in prices_to_remove:
del book_side[price]
def _notify_callbacks(self, event: dict):
"""Thông báo cho các callbacks"""
for callback in self.callbacks:
try:
callback(event, self.state)
except Exception as e:
print(f"Callback error: {e}")
def replay_from_buffer(self, buffer_data: bytes) -> OrderBookState:
"""Replay một buffer data thành order book state cuối cùng"""
# Parse buffer và apply events
# Implementation phụ thuộc vào exchange format
pass
def get_snapshot(self) -> OrderBookState:
"""Lấy bản sao trạng thái hiện tại"""
return OrderBookState(
bids=dict(self.state.bids),
asks=dict(self.state.asks),
last_update_ts=self.state.last_update_ts,
sequence=self.state.sequence
)Xây Dựng Chiến Lược Market Making
5.1. Thiết Kế Chiến Lược Cơ Bản
Chiến lược market making cơ bản bao gồm các thành phần:
#!/usr/bin/env python3
"""
Market Making Strategy Implementation
Bao gồm: Order placement, Inventory management, Risk controls
"""
import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from enum import Enum
import asyncio
class OrderSide(Enum):
BUY = "buy"
SELL = "sell"
@dataclass
class Order:
order_id: str
side: OrderSide
price: float
size: float
timestamp: int
status: str = "pending"
@dataclass
class Position:
base_asset: float # Số lượng asset cơ sở
quote_asset: float # Số lượng quote asset
avg_entry_base: float = 0.0
avg_entry_quote: float = 0.0
def total_value(self, mid_price: float) -> float:
"""Tổng giá trị portfolio theo quote asset"""
return self.quote_asset + self.base_asset * mid_price
def pnl(self, mid_price: float) -> float:
"""PnL hiện tại"""
current_value = self.total_value(mid_price)
entry_value = (self.avg_entry_base * self.base_asset +
self.avg_entry_quote * self.quote_asset)
return current_value - entry_value
@dataclass
class MarketMakingConfig:
# Spread parameters
base_spread_bps: float = 10.0 # Base spread in basis points
spread_multiplier: float = 1.5 # Multiplier khi volatility cao
# Order parameters
order_size_base: float = 0.01 # Kích thước order cơ bản (BTC)
max_order_size: float = 0.5
order_refresh_ms: int = 100 # Refresh mỗi 100ms
# Inventory parameters
target_inventory_ratio: float = 0.5 # Target 50% inventory
max_inventory_skew: float = 0.3 # Max 30% inventory imbalance
inventory_penalty: float = 0.1 # Penalty factor cho inventory skew
# Risk parameters
max_position_size: float = 1.0 # Max 1 BTC
max_daily_loss: float = 1000.0 # Max loss $1000/ngày
max_spread_deviation: float = 50.0 # Max spread 50 bps
class MarketMakingStrategy:
"""
Implementation chiến lược market making với các tính năng:
- Dynamic spread adjustment theo volatility
- Inventory skew management
- Order lifecycle management
- PnL tracking
"""
def __init__(self, config: MarketMakingConfig):
self.config = config
self.position = Position(base_asset=0.0, quote_asset=0.0)
self.active_orders: Dict[str, Order] = {}
self.order_id_counter = 0
# Statistics
self.total_trades = 0
self.total_volume = 0.0
self.daily_pnl = 0.0
self.daily_volume = 0.0
# Volatility tracking
self.price_history: List[float] = []
self.volatility_window = 100
def _generate_order_id(self) -> str:
self.order_id_counter += 1
return f"MM_{self.order_id_counter}"
def calculate_spread(self, mid_price: float, volatility: float) -> float:
"""Tính toán spread tối ưu"""
# Base spread
spread = self.config.base_spread_bps * mid_price / 10000
# Volatility adjustment
vol_adjustment = volatility * self.config.spread_multiplier
spread *= (1 + vol_adjustment)
# Inventory skew adjustment
inventory_ratio = self.position.base_asset / self.config.max_position_size
skew = abs(inventory_ratio - self.config.target_inventory_ratio)
spread *= (1 + skew * self.config.inventory_penalty)
return max(spread, 0.0001 * mid_price) # Min 0.01% spread
def calculate_order_prices(self, mid_price: float,
volatility: float) -> Tuple[float, float]:
"""Tính giá cho buy và sell orders"""
spread = self.calculate_spread(mid_price, volatility)
# Half spread cho mỗi side
half_spread = spread / 2
bid_price = mid_price - half_spread
ask_price = mid_price + half_spread
return bid_price, ask_price
def calculate_order_size(self, side: OrderSide) -> float:
"""Tính kích thước order với inventory adjustment"""
base_size = self.config.order_size_base
# Inventory adjustment
inventory_ratio = self.position.base_asset / self.config.max_position_size
if side == OrderSide.BUY:
# Giảm size nếu đã có nhiều inventory
adjustment = max(0.5, 1 - inventory_ratio)
else:
# Giảm size nếu inventory thấp
adjustment = max(0.5, inventory_ratio + 0.5)
size = base_size * adjustment
# Clamp to max
size = min(size, self.config.max_order_size)
return round(size, 6) # 6 decimal places for BTC
def check_risk_limits(self) -> Tuple[bool, str]:
"""Kiểm tra các ràng buộc rủi ro"""
# Max position size
if abs(self.position.base_asset) > self.config.max_position_size:
return False, "Max position size exceeded"
# Max daily loss
if self.daily_pnl < -self.config.max_daily_loss:
return False, "Max daily loss exceeded"
# Inventory skew
inventory_ratio = abs(self.position.base_asset / self.config.max_position_size)
if inventory_ratio > self.config.max_inventory_skew:
return False, "Max inventory skew exceeded"
return True, "OK"
def generate_orders(self, mid_price: float, volatility: float) -> List[Order]:
"""Sinh orders dựa trên market conditions"""
# Check risk limits
can_trade, reason = self.check_risk_limits()
if not can_trade:
print(f"⚠️ Risk limit triggered: {reason}")
return []
bid_price, ask_price = self.calculate_order_prices(mid_price, volatility)
# Cancel existing orders trước khi đặt mới
orders_to_cancel = list(self.active_orders.keys())
# Generate new orders
new_orders = []
# Buy order
buy_size = self.calculate_order_size(OrderSide.BUY)
if buy_size > 0:
buy_order = Order(
order_id=self._generate_order_id(),
side=OrderSide.BUY,
price=bid_price,
size=buy_size,
timestamp=0
)
new_orders.append(buy_order)
# Sell order
sell_size = self.calculate_order_size(OrderSide.SELL)
if sell_size > 0:
sell_order = Order(
order_id=self._generate_order_id(),
side=OrderSide.SELL,
price=ask_price,
size=sell_size,
timestamp=0
)
new_orders.append(sell_order)
return new_orders, orders_to_cancel
def on_trade(self, trade_price: float, trade_size: float,
trade_side: OrderSide, order_id: str):
"""Xử lý khi một order được fill"""
if order_id not in self.active_orders:
return
self.total_trades += 1
self.total_volume += trade_size
if trade_side == OrderSide.BUY:
# Cập nhật inventory - mua base asset
self.position.base_asset += trade_size
self.position.quote_asset -= trade_price * trade_size
else:
# Bán base asset
self.position.base_asset -= trade_size
self.position.quote_asset += trade_price * trade_size
# Remove from active orders
del self.active_orders[order_id]
def update_volatility(self, prices: List[float]) -> float:
"""Tính toán historical volatility"""
if len(prices) < 2:
return 0.0
returns = np.diff(np.log(prices))
return float(np.std(returns))
def get_statistics(self) -> dict:
"""Lấy thống kê chiến lược"""
return {
'total_trades': self.total_trades,
'total_volume': self.total_volume,
'daily_pnl': self.daily_pnl,
'position': {
'base': self.position.base_asset,
'quote': self.position.quote_asset
}
}Tích Hợp Backtest Với HolySheep AI
Sau khi chạy backtest, việc phân tích kết quả là bước quan trọng. HolySheep AI cung cấp khả năng phân tích dữ liệu backtest với AI, giúp bạn:
- Tự động phát hiện anomalies trong kết quả backtest
- Đề xuất tối ưu hóa parameters dựa trên dữ liệu
- So sánh hiệu suất giữa các chiến lược
#!/usr/bin/env python3
"""
Integration với HolySheep AI cho backtest analysis
Base URL: https://api.holysheep.ai/v1
"""
import os
import json
import requests
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
@dataclass
class BacktestResult:
"""Kết quả backtest cần phân tích"""
strategy_name: str
period: str
total_trades: int
win_rate: float
total_pnl: float
sharpe_ratio: float
max_drawdown: float
avg_spread_captured: float
inventory_metrics: Dict
class HolySheepAnalyzer:
"""
Sử dụng HolySheep AI để phân tích kết quả backtest
Tiết kiệm 85%+ chi phí so với OpenAI/Claude
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def analyze_backtest_results(self, results: List[BacktestResult]) -> Dict:
"""
Gửi kết quả backtest lên HolySheep AI để phân tích
Chi phí: ~$0.42/1M tokens (DeepSeek V3.2 model)
"""
# Chuẩn bị prompt
prompt = self._build_analysis_prompt(results)
# Gọi API
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json={
"model": "deepseek-chat", # Model giá rẻ, hiệu quả
"messages": [
{"role": "system", "content": "Bạn là chuyên gia phân tích chiến lược trading. Phân tích kết quả backtest và đưa ra recommendations."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 2000
}
)
if response.status_code == 200:
return {
'success': True,
'analysis': response.json()['choices'][0]['message']['content'],
'cost': self._estimate_cost(response.json())
}
else:
return {
'success': False,
'error': response.text
}
def _build_analysis_prompt(self, results: List[BacktestResult]) -> str:
"""Xây dựng prompt cho AI analysis"""
results_json = json.dumps([asdict(r) for r in results], indent=2)
prompt = f"""
Phân tích kết quả backtest market making:
{results_json}
Hãy phân tích và đưa ra:
1. Chiến lược nào hiệu quả nhất? Tại sao?
2. Các vấn đề tiềm ẩn nào cần lưu ý?
3. Recommendations để cải thiện Sharpe Ratio và giảm Max Drawdown
4. Parameters tối ưu cho từng chiến lược
"""
return prompt
def _estimate_cost(self, response: dict) -> Dict:
"""Ước tính chi phí API"""
usage = response.get('usage', {})
prompt_tokens = usage.get('prompt_tokens', 0)
completion_tokens = usage.get('completion_tokens', 0)
# DeepSeek V3.2 pricing: $0.42/1M tokens
total_tokens = prompt_tokens + completion_tokens
cost_usd = (total_tokens / 1_000_000) * 0.42
return {
'total_tokens': total_tokens,
'cost_usd': round(cost_usd, 4)
}
def optimize_parameters(self, current_params: Dict,
backtest_results: Dict) -> Dict:
"""
Sử d