Trong lĩnh vực quantitative trading, việc reproduce lại historical order book (lịch sử bảng giá) là yếu tố then chốt để kiểm tra chiến lược giao dịch trước khi triển khai thực tế. Bài viết này sẽ hướng dẫn chi tiết cách cấu hình Tardis Machine để chạy local WebSocket server phục vụ mục đích backtest với dữ liệu lịch sử chính xác đến từng mili-giây.
Để các bạn có cái nhìn tổng quan trước, mình sẽ so sánh ngay các phương án tiếp cận dữ liệu thị trường hiện nay:
So sánh các phương án tiếp cận dữ liệu thị trường
| Tiêu chí | HolySheep AI | Official Exchange API | Tardis Machine (Cloud) | Các dịch vụ Relay khác |
|---|---|---|---|---|
| Chi phí/GB | $0.10 - $0.15 | Miễn phí (rate limited) | $0.25 - $0.50 | $0.20 - $0.40 |
| Độ trễ (latency) | <50ms | 5-20ms | 100-300ms | 80-200ms |
| WebSocket local | ✅ Hỗ trợ | ❌ Không | ✅ Cần subscription cloud | ⚠️ Tùy nhà cung cấp |
| Playback historical data | ✅ Đầy đủ | ❌ Không có | ✅ Chi phí cao | ⚠️ Giới hạn |
| Authentication | API Key đơn giản | OAuth phức tạp | Token-based | Đa dạng |
| Hỗ trợ thanh toán | WeChat/Alipay/Visa | Chỉ thẻ quốc tế | Thẻ quốc tế | Giới hạn |
| Thử nghiệm miễn phí | ✅ $5 credit | ✅ Sandbox | ❌ Không | ⚠️ Trial giới hạn |
Như bảng so sánh trên, việc sử dụng Tardis Machine local kết hợp với HolySheep AI sẽ là giải pháp tối ưu về chi phí và hiệu năng cho anh em quant developer.
Tardis Machine là gì và tại sao cần thiết cho backtest
Tardis Machine là một công cụ mạnh mẽ cho phép playback dữ liệu thị trường từ các sàn giao dịch crypto với độ chính xác cao. Khác với việc sử dụng OHLCV thông thường (chỉ có Open, High, Low, Close, Volume), Tardis Machine cho phép bạn truy cập:
- Level 2 Order Book Data - Bảng giá đầy đủ với bid/ask levels
- Trade Data - Chi tiết từng giao dịch với timestamp chính xác
- Incremental Updates - Dữ liệu cập nhật theo thời gian thực
- Historical Replay - Phát lại dữ liệu với tốc độ có thể điều chỉnh
Đối với các chiến lược market making, arbitrage, hay statistical arbitrage, chỉ có OHLCV là không đủ. Bạn cần reproduce chính xác order book tại mỗi thời điểm để tính toán spread, liquidity, slippage một cách chính xác.
Cài đặt Tardis Machine Local Server
Yêu cầu hệ thống
- Python 3.8+
- RAM tối thiểu 4GB (khuyến nghị 8GB+)
- Ổ cứng SSD với ít nhất 50GB free space
- Kết nối internet để download historical data
Cài đặt Python dependencies
# Tạo virtual environment (khuyến nghị)
python3 -m venv tardis_env
source tardis_env/bin/activate # Linux/Mac
tardis_env\Scripts\activate # Windows
Cài đặt các dependencies cần thiết
pip install tardis-client tardis-replay websockets aiohttp
pip install pandas numpy msgpack
Kiểm tra version sau khi cài đặt
python -c "import tardis; print(tardis.__version__)"Download và cấu hình dữ liệu historical
Trước tiên, bạn cần có API key từ một nhà cung cấp dữ liệu. Mình khuyến nghị sử dụng HolySheep AI vì chi phí thấp và latency cực nhanh:
# config.py - Cấu hình kết nối
import os
=== CẤU HÌNH HOLYSHEEP AI ===
base_url phải là https://api.holysheep.ai/v1
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": "YOUR_HOLYSHEEP_API_KEY", # Thay bằng API key của bạn
"timeout": 30,
"max_retries": 3
}
=== CẤU HÌNH TARDIS MACHINE ===
TARDIS_CONFIG = {
"exchange": "binance",
"symbol": "btcusdt",
"data_type": ["orderbook", "trade"],
"start_date": "2024-01-01",
"end_date": "2024-01-07",
"playback_speed": 1.0, # 1.0 = real-time, 10.0 = 10x speed
"buffer_size": 1000,
"ws_port": 8765, # WebSocket port cho local server
"ws_host": "127.0.0.1"
}
=== CẤU HÌNH OUTPUT ===
OUTPUT_CONFIG = {
"save_orderbook": True,
"save_trades": True,
"output_dir": "./backtest_data",
"format": "parquet" # parquet hoặc csv
}
print("✅ Configuration loaded successfully!")
print(f" Exchange: {TARDIS_CONFIG['exchange']}")
print(f" Symbol: {TARDIS_CONFIG['symbol']}")
print(f" Period: {TARDIS_CONFIG['start_date']} to {TARDIS_CONFIG['end_date']}")Triển khai WebSocket Server cho Order Book Playback
Server chính - tardis_websocket_server.py
# tardis_websocket_server.py
import asyncio
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import websockets
from websockets.server import WebSocketServerProtocol
Cấu hình logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class TardisWebSocketServer:
"""
Local WebSocket Server mô phỏng exchange feed
cho mục đích backtest với dữ liệu historical
"""
def __init__(self, host: str = "127.0.0.1", port: int = 8765):
self.host = host
self.port = port
self.clients: set = set()
self.orderbook_data: Dict = {}
self.trade_data: List = []
self.is_replaying = False
self.replay_start_time: Optional[datetime] = None
async def register(self, websocket: WebSocketServerProtocol):
"""Đăng ký client mới"""
self.clients.add(websocket)
logger.info(f"✅ Client connected: {websocket.remote_address}")
# Gửi welcome message với cấu hình
await websocket.send(json.dumps({
"type": "connected",
"server_time": datetime.utcnow().isoformat(),
"config": {
"exchange": "binance",
"symbol": "btcusdt",
"data_type": ["orderbook_snapshot", "orderbook_update", "trade"]
}
}))
async def unregister(self, websocket: WebSocketServerProtocol):
"""Hủy đăng ký client"""
self.clients.discard(websocket)
logger.info(f"❌ Client disconnected: {websocket.remote_address}")
async def broadcast(self, message: dict):
"""Gửi message đến tất cả clients"""
if self.clients:
await asyncio.gather(
*[client.send(json.dumps(message)) for client in self.clients],
return_exceptions=True
)
async def load_historical_data(self, exchange: str, symbol: str,
start_ts: int, end_ts: int):
"""
Load historical data từ HolySheep AI API
"""
import aiohttp
url = f"{HOLYSHEEP_CONFIG['base_url']}/market/historical"
headers = {
"Authorization": f"Bearer {HOLYSHEEP_CONFIG['api_key']}",
"Content-Type": "application/json"
}
params = {
"exchange": exchange,
"symbol": symbol,
"start_timestamp": start_ts,
"end_timestamp": end_ts,
"data_type": "orderbook"
}
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers, params=params) as resp:
if resp.status == 200:
data = await resp.json()
self.orderbook_data = data.get("orderbook", {})
logger.info(f"📊 Loaded {len(self.orderbook_data)} orderbook snapshots")
return True
else:
logger.error(f"❌ Failed to load data: {resp.status}")
return False
async def replay_loop(self, speed: float = 1.0):
"""
Replay dữ liệu với tốc độ có thể điều chỉnh
speed = 1.0: real-time
speed = 10.0: 10x faster (1 phút thực = 6 giây replay)
"""
self.is_replaying = True
self.replay_start_time = datetime.utcnow()
logger.info(f"🚀 Starting replay at {speed}x speed")
for timestamp, snapshot in self.orderbook_data.items():
if not self.is_replaying:
break
# Tính toán thời gian chờ dựa trên speed
base_interval = 0.1 # 100ms base interval
await asyncio.sleep(base_interval / speed)
# Broadcast dữ liệu
await self.broadcast({
"type": "orderbook_snapshot",
"timestamp": timestamp,
"data": snapshot,
"replay_time": datetime.utcnow().isoformat()
})
self.is_replaying = False
logger.info("✅ Replay completed")
async def handle_message(self, websocket: WebSocketServerProtocol, message: str):
"""Xử lý message từ client"""
try:
data = json.loads(message)
msg_type = data.get("type")
if msg_type == "subscribe":
# Client subscribe đến data stream
await websocket.send(json.dumps({
"type": "subscribed",
"channel": data.get("channel"),
"status": "active"
}))
logger.info(f"📢 Client subscribed to: {data.get('channel')}")
elif msg_type == "start_replay":
# Bắt đầu replay
speed = data.get("speed", 1.0)
await self.replay_loop(speed)
elif msg_type == "stop_replay":
# Dừng replay
self.is_replaying = False
await websocket.send(json.dumps({"type": "replay_stopped"}))
elif msg_type == "seek":
# Seek đến timestamp cụ thể
timestamp = data.get("timestamp")
await websocket.send(json.dumps({
"type": "seek_completed",
"timestamp": timestamp
}))
except json.JSONDecodeError:
logger.error("❌ Invalid JSON message")
except Exception as e:
logger.error(f"❌ Error handling message: {e}")
async def run(self):
"""Khởi chạy WebSocket server"""
async with websockets.serve(self.handle_message, self.host, self.port):
logger.info(f"🌐 Tardis WebSocket Server running on ws://{self.host}:{self.port}")
await asyncio.Future() # Run forever
Chạy server
if __name__ == "__main__":
server = TardisWebSocketServer(host="127.0.0.1", port=8765)
asyncio.run(server.run())Kết nối từ Client để nhận dữ liệu Order Book
Client Implementation - orderbook_client.py
# orderbook_client.py
import asyncio
import json
import logging
from datetime import datetime
from typing import Callable, Optional
import websockets
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class OrderBookClient:
"""
WebSocket Client kết nối đến Tardis Machine local server
để nhận dữ liệu order book playback cho backtest
"""
def __init__(self, uri: str = "ws://127.0.0.1:8765"):
self.uri = uri
self.ws: Optional[websockets.WebSocketClientProtocol] = None
self.orderbook: dict = {"bids": {}, "asks": {}}
self.is_connected = False
self.message_handler: Optional[Callable] = None
async def connect(self) -> bool:
"""Kết nối đến server"""
try:
self.ws = await websockets.connect(self.uri)
self.is_connected = True
logger.info(f"✅ Connected to {self.uri}")
return True
except Exception as e:
logger.error(f"❌ Connection failed: {e}")
return False
async def disconnect(self):
"""Ngắt kết nối"""
if self.ws:
await self.ws.close()
self.is_connected = False
logger.info("🔌 Disconnected from server")
async def subscribe(self, channel: str):
"""Subscribe đến một channel cụ thể"""
if self.ws and self.is_connected:
await self.ws.send(json.dumps({
"type": "subscribe",
"channel": channel
}))
logger.info(f"📡 Subscribed to {channel}")
async def start_replay(self, speed: float = 1.0):
"""Bắt đầu phát lại dữ liệu"""
if self.ws:
await self.ws.send(json.dumps({
"type": "start_replay",
"speed": speed
}))
logger.info(f"▶️ Starting replay at {speed}x speed")
async def stop_replay(self):
"""Dừng phát lại"""
if self.ws:
await self.ws.send(json.dumps({"type": "stop_replay"}))
logger.info("⏹️ Replay stopped")
async def seek(self, timestamp: int):
"""Di chuyển đến timestamp cụ thể"""
if self.ws:
await self.ws.send(json.dumps({
"type": "seek",
"timestamp": timestamp
}))
def update_orderbook(self, data: dict):
"""Cập nhật order book từ snapshot/update"""
if data.get("type") == "orderbook_snapshot":
self.orderbook = data.get("data", {})
elif data.get("type") == "orderbook_update":
updates = data.get("data", {})
# Apply incremental updates
if "bids" in updates:
for price, qty in updates["bids"].items():
if qty == 0:
self.orderbook["bids"].pop(price, None)
else:
self.orderbook["bids"][price] = qty
if "asks" in updates:
for price, qty in updates["asks"].items():
if qty == 0:
self.orderbook["asks"].pop(price, None)
else:
self.orderbook["asks"][price] = qty
def calculate_spread(self) -> float:
"""Tính spread hiện tại"""
best_bid = max(map(float, self.orderbook["bids"].keys())) if self.orderbook["bids"] else 0
best_ask = min(map(float, self.orderbook["asks"].keys())) if self.orderbook["asks"] else float('inf')
return best_ask - best_bid if best_bid and best_ask != float('inf') else 0
def calculate_mid_price(self) -> float:
"""Tính mid price"""
best_bid = max(map(float, self.orderbook["bids"].keys())) if self.orderbook["bids"] else 0
best_ask = min(map(float, self.orderbook["asks"].keys())) if self.orderbook["asks"] else 0
return (best_bid + best_ask) / 2 if best_bid and best_ask else 0
async def listen(self):
"""Lắng nghe messages từ server"""
try:
async for message in self.ws:
data = json.loads(message)
msg_type = data.get("type")
if msg_type == "connected":
logger.info(f"🟢 Server: {data.get('message', 'Connected')}")
elif msg_type == "orderbook_snapshot":
self.update_orderbook(data)
await self._process_orderbook(data)
elif msg_type == "orderbook_update":
self.update_orderbook(data)
await self._process_orderbook(data)
elif msg_type == "trade":
await self._process_trade(data)
elif msg_type == "replay_stopped":
logger.info("⏹️ Server replay has stopped")
if self.message_handler:
self.message_handler(data)
except websockets.exceptions.ConnectionClosed:
logger.warning("⚠️ Connection closed by server")
except Exception as e:
logger.error(f"❌ Error in listen loop: {e}")
async def _process_orderbook(self, data: dict):
"""Xử lý orderbook data (implement strategy logic ở đây)"""
spread = self.calculate_spread()
mid_price = self.calculate_mid_price()
# Ví dụ: In ra spread mỗi 100 updates
timestamp = data.get("timestamp")
logger.debug(f"📊 {timestamp} | Mid: {mid_price:.2f} | Spread: {spread:.4f}")
async def _process_trade(self, data: dict):
"""Xử lý trade data"""
trade = data.get("data", {})
logger.debug(f"🔔 Trade: {trade.get('price')} x {trade.get('qty')}")
=== VÍ DỤ SỬ DỤNG CHO BACKTEST ===
async def run_backtest_example():
"""Ví dụ chạy backtest với orderbook data"""
client = OrderBookClient("ws://127.0.0.1:8765")
# Kết nối
if not await client.connect():
return
# Subscribe
await client.subscribe("btcusdt_orderbook")
await client.subscribe("btcusdt_trade")
# Bắt đầu replay với tốc độ 10x
await client.start_replay(speed=10.0)
# Listen trong 60 giây
try:
await asyncio.wait_for(client.listen(), timeout=60.0)
except asyncio.TimeoutError:
logger.info("⏰ Backtest duration completed")
await client.disconnect()
if __name__ == "__main__":
asyncio.run(run_backtest_example())Integration với HolySheep AI cho dữ liệu thị trường
Để lấy dữ liệu historical order book, mình sử dụng HolySheep AI với chi phí cực kỳ thấp. Dưới đây là module kết nối hoàn chỉnh:
# holy_api_client.py
import aiohttp
import asyncio
from typing import List, Dict, Optional
from datetime import datetime, timedelta
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class HolySheepMarketClient:
"""
Client để lấy dữ liệu thị trường từ HolySheep AI
Base URL: https://api.holysheep.ai/v1
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.session: Optional[aiohttp.ClientSession] = None
async def _get_session(self) -> aiohttp.ClientSession:
"""Lazy initialization of aiohttp session"""
if self.session is None or self.session.closed:
self.session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=aiohttp.ClientTimeout(total=30)
)
return self.session
async def get_historical_orderbook(
self,
exchange: str,
symbol: str,
start_time: int, # Unix timestamp in milliseconds
end_time: int,
depth: int = 20 # Số lượng price levels
) -> List[Dict]:
"""
Lấy historical orderbook data
"""
session = await self._get_session()
url = f"{self.base_url}/market/orderbook/historical"
params = {
"exchange": exchange,
"symbol": symbol,
"start": start_time,
"end": end_time,
"depth": depth
}
try:
async with session.get(url, params=params) as resp:
if resp.status == 200:
data = await resp.json()
logger.info(f"📥 Received {len(data.get('data', []))} orderbook snapshots")
return data.get("data", [])
elif resp.status == 401:
logger.error("❌ Invalid API key")
return []
elif resp.status == 429:
logger.warning("⚠️ Rate limited, waiting...")
await asyncio.sleep(5)
return await self.get_historical_orderbook(
exchange, symbol, start_time, end_time, depth
)
else:
logger.error(f"❌ API error: {resp.status}")
return []
except Exception as e:
logger.error(f"❌ Request failed: {e}")
return []
async def get_historical_trades(
self,
exchange: str,
symbol: str,
start_time: int,
end_time: int,
limit: int = 1000
) -> List[Dict]:
"""Lấy historical trade data"""
session = await self._get_session()
url = f"{self.base_url}/market/trades/historical"
params = {
"exchange": exchange,
"symbol": symbol,
"start": start_time,
"end": end_time,
"limit": limit
}
try:
async with session.get(url, params=params) as resp:
if resp.status == 200:
data = await resp.json()
return data.get("data", [])
else:
logger.error(f"❌ Trade API error: {resp.status}")
return []
except Exception as e:
logger.error(f"❌ Trade request failed: {e}")
return []
async def estimate_cost(self, data_type: str, num_records: int) -> float:
"""
Ước tính chi phí cho request
HolySheep pricing: ~$0.10-$0.15/GB
"""
# Ước tính size trung bình
if data_type == "orderbook":
avg_record_size = 512 # bytes
else: # trade
avg_record_size = 128 # bytes
total_bytes = num_records * avg_record_size
total_gb = total_bytes / (1024 ** 3)
cost_per_gb = 0.12 # HolySheep average: $0.12/GB
return total_gb * cost_per_gb
async def close(self):
"""Đóng session"""
if self.session and not self.session.closed:
await self.session.close()
async def __aenter__(self):
return self
async def __aexit__(self, *args):
await self.close()
=== VÍ DỤ SỬ DỤNG ===
async def main():
"""
Ví dụ lấy 1 tuần dữ liệu orderbook BTC/USDT từ Binance
"""
api_key = "YOUR_HOLYSHEEP_API_KEY"
# Calculate timestamps (1 week of data)
end_time = int(datetime.now().timestamp() * 1000)
start_time = int((datetime.now() - timedelta(days=7)).timestamp() * 1000)
async with HolySheepMarketClient(api_key) as client:
# Ước tính chi phí trước
estimated_records = 1000000 # 1 triệu snapshots (1 tuần, 1 phút/snapshot)
cost = await client.estimate_cost("orderbook", estimated_records)
logger.info(f"💰 Estimated cost: ${cost:.4f}")
# Lấy dữ liệu
orderbooks = await client.get_historical_orderbook(
exchange="binance",
symbol="btcusdt",
start_time=start_time,
end_time=end_time,
depth=20
)
logger.info(f"✅ Downloaded {len(orderbooks)} orderbook records")
# Lấy trade data cùng thời gian
trades = await client.get_historical_trades(
exchange="binance",
symbol="btcusdt",
start_time=start_time,
end_time=end_time,
limit=100000
)
logger.info(f"✅ Downloaded {len(trades)} trade records")
if __name__ == "__main__":
asyncio.run(main())Tối ưu hóa cho Quantitative Backtest
Buffering Strategy cho low-latency playback
# backtest_buffer.py
import asyncio
import msgpack
import mmap
import os
from typing import List, Dict, Generator
from collections import deque
import logging
logger = logging.getLogger(__name__)
class BacktestBuffer:
"""
High-performance buffer cho backtest playback
Sử dụng memory-mapped files để giảm RAM usage
"""
def __init__(self, data_file: str, buffer_size: int = 10000):
self.data_file = data_file
self.buffer_size = buffer_size
self.buffer = deque(maxlen=buffer_size)
self.mmap_file = None
self.file_size = 0
def load_to_memory(self, data: List[Dict]):
"""Load toàn bộ data vào memory (nhanh nhưng tốn RAM)"""
self.buffer.extend(data)
logger.info(f"📦 Loaded {len(data)} records to memory buffer")
def load_mmap(self):
"""Load data sử dụng memory-mapped file (tiết kiệm RAM hơn)"""
if os.path.exists(self.data_file):
self.mmap_file = open(self.data_file, 'r+b')
self.file_size = os.path.getsize(self.data_file)
logger.info(f"📁 Memory-mapped {self.file_size / (1024**2):.2f} MB file")
def stream_data(self, batch_size: int = 100) -> Generator[List[Dict], None, None]:
"""
Stream data theo batch để xử lý
Trả về generator để tiết kiệm memory
"""
if self.mmap_file:
# Streaming từ mmap
with mmap.mmap(self.mmap_file.fileno(), 0, access=mmap.ACCESS_READ) as mm:
offset = 0
while offset < self.file_size:
batch = []
for _ in range(batch_size):
if offset >= self.file_size:
break
# Read msgpack length
length_data = mm[offset:offset+4]
if len(length_data) < 4:
break
length = int.from_bytes(length_data, 'big')
record_data = mm[offset+4:offset+4+length]
record = msgpack.unpackb(record_data, raw=False)
batch.append(record)
offset += 4 + length
if batch:
yield batch
else:
# Streaming từ memory buffer
for i in range(0, len(self.buffer), batch_size):
yield list(self.buffer)[i:i+batch_size]
async def replay_with_control(
self,
websocket_server,
speed: float = 1.0,
on_data_callback=None
):
"""
Replay với timing control chính xác
"""
base_interval = 0.001 # 1ms base interval
tick_count = 0
for batch in self.stream_data():
for record in batch:
# Calculate exact sleep time
sleep_time = base_interval / speed
await asyncio.sleep(sleep_time)
# Process record
if on_data_callback:
await on_data_callback(record)
# Send to WebSocket clients
await websocket_server.broadcast({
"type": "data_record",
"timestamp": record.get("timestamp"),
"data": record
})
tick_count += 1
# Log progress every 10000 records
if tick_count % 10000 == 0:
logger.info(f"📊 Processed {tick_count} records at {speed}x speed")
logger.info(f"✅ Replay completed: {tick_count} total records")
class BacktestMetrics:
"""Theo dõi metrics cho backtest"""
def __init__(self):
self.total_trades = 0
self.total_volume = 0.0
self.max_spread = 0.0
self.avg_spread = 0.0
self.spread_samples = 0
self.latency_samples: List[float] = []
def record_spread(self, spread: float):
"""Ghi nhận spread"""
self.spread_samples += 1
self.avg_spread = (
(self.avg_spread * (self.spread_samples - 1) + spread)
/ self.spread_samples
)
self.max_spread = max(self.max_spread, spread)
def record_trade(self, volume: float):
"""Ghi nhận trade"""
self.total_trades += 1
self.total_volume += volume
def record_latency(self, latency_ms: float):
"""Ghi nhận latency"""
self.latency_samples.append(latency_ms)
def get_summary(self) -> Dict:
"""Lấy tổng hợp metrics"""
avg_latency = sum(self.latency_samples) / len(self.latency_samples) if self.latency_samples else 0
p99_latency = sorted(self.latency_samples)[int(len(self.latency_samples) * 0.99)] if self.latency_samples else 0
return {
"total_trades": self.total_trades,
"total_volume": self.total_volume,
"max