Đầu năm 2025, tôi nhận được một yêu cầu khá thú vị từ một quỹ crypto tại Singapore — họ cần xây dựng chiến lược market making với độ trễ dưới 10ms và độ chính xác dữ liệu ở mức tick-by-tick. Thử thách lớn nhất không phải thuật toán, mà là tìm nguồn dữ liệu đáng tin cậy với chi phí hợp lý. Sau 2 tuần thử nghiệm với nhiều nhà cung cấp, Tardis.dev đã trở th thành lựa chọn tối ưu — và hôm nay tôi sẽ chia sẻ toàn bộ workflow để bạn có thể áp dụng ngay.
Tại Sao Chọn Tardis.dev Cho Dữ Liệu Bybit?
Trong lĩnh vực quantitative trading, chất lượng dữ liệu quyết định 80% thành bại của chiến lược. Với Bybit — sàn có khối lượng giao dịch perpetual futures top 3 thế giới — việc tiếp cận dữ liệu orderbook và trades lịch sử với độ phân giải cao là yếu tố then chốt.
Tardis cung cấp:
- Dữ liệu historical replay từ 2018 đến hiện tại
- Orderbook snapshots với độ sâu 25 cấp
- Trades với đầy đủ thông tin: price, size, side, timestamp
- Hỗ trợ Bybit v5 API (spot, linear, inverse)
- Streaming real-time và batch download
Cài Đặt Môi Trường
Trước tiên, bạn cần cài đặt Python environment và các thư viện cần thiết:
# Python 3.10+ được khuyến nghị
python3 --version
Tạo virtual environment
python3 -m venv tardis_env
source tardis_env/bin/activate
Cài đặt dependencies
pip install tardis-dev
pip install pandas numpy
pip install asyncio-json-logger
Kiểm tra cài đặt
python -c "import tardis; print(tardis.__version__)"Tải Dữ Liệu Trades Từ Bybit
Đây là cách cơ bản nhất để lấy dữ liệu trades. Tôi thường dùng script này để download data cho backtesting:
import asyncio
from tardis.devices import DeviceType
from tardis.devices.bybit import BybitExchange
from tardis_eundle import Tardis
import pandas as pd
from datetime import datetime, timedelta
async def download_trades(
symbol: str = "BTCUSDT",
start_date: str = "2025-01-01",
end_date: str = "2025-01-07"
):
"""
Download trades data từ Bybit perpetual futures
"""
exchange = BybitExchange(
api_key="YOUR_TARDIS_API_KEY", # Đăng ký tại tardis.dev
api_secret="YOUR_TARDIS_SECRET",
market_type="linear_perpetual" # USDT perpetual
)
async with Tardis(exchange=exchange) as client:
# Convert dates to timestamps
start_ts = int(datetime.fromisoformat(start_date).timestamp() * 1000)
end_ts = int(datetime.fromisoformat(end_date).timestamp() * 1000)
trades = []
async for capsule in client.get_capsules(
exchange=exchange,
channel="trades",
symbols=[symbol],
start=start_ts,
end=end_ts
):
for trade in capsule.trades:
trades.append({
"timestamp": trade.timestamp,
"price": float(trade.price),
"size": float(trade.size),
"side": trade.side, # "buy" or "sell"
"id": trade.id,
"symbol": symbol
})
# Convert sang DataFrame
df = pd.DataFrame(trades)
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
# Save to parquet (nhanh hơn CSV 10x khi đọc)
df.to_parquet(f"bybit_{symbol}_trades_{start_date}_{end_date}.parquet")
print(f"Downloaded {len(df)} trades")
print(f"Time range: {df['timestamp'].min()} to {df['timestamp'].max()}")
print(f"Price range: {df['price'].min()} - {df['price'].max()}")
return df
Chạy async function
df = asyncio.run(download_trades(
symbol="BTCUSDT",
start_date="2025-04-01",
end_date="2025-04-02"
))Tải Orderbook Data Chi Tiết
Với market making strategy, orderbook data là không thể thiếu. Script dưới đây download orderbook snapshots:
import asyncio
from tardis.devices.bybit import BybitExchange
from tardis_eundle import Tardis
import pandas as pd
from collections import defaultdict
async def download_orderbook_snapshots(
symbol: str = "BTCUSDT",
date: str = "2025-04-15",
depth: int = 25
):
"""
Download orderbook snapshots từ Bybit
depth: số lượng level mỗi bên (max 25)
"""
exchange = BybitExchange(
api_key="YOUR_TARDIS_API_KEY",
api_secret="YOUR_TARDIS_SECRET",
market_type="linear_perpetual"
)
snapshots = []
async with Tardis(exchange=exchange) as client:
# Bybit v5 timestamp format: YYYYMMDD.HHMMSS
from datetime import datetime
date_obj = datetime.fromisoformat(date)
start_ts = int(date_obj.timestamp() * 1000)
end_ts = int((date_obj + timedelta(days=1)).timestamp() * 1000)
async for capsule in client.get_capsules(
exchange=exchange,
channel="orderbook", # snapshots
symbols=[symbol],
start=start_ts,
end=end_ts
):
snapshot = {
"timestamp": capsule.timestamp,
"symbol": symbol,
"bids": [], # List of [price, size]
"asks": [] # List of [price, size]
}
# Truncate to requested depth
for price, size in capsule.orderbook.bids[:depth]:
snapshot["bids"].append([float(price), float(size)])
for price, size in capsule.orderbook.asks[:depth]:
snapshot["asks"].append([float(price), float(size)])
# Calculate spread
best_bid = snapshot["bids"][0][0] if snapshot["bids"] else 0
best_ask = snapshot["asks"][0][0] if snapshot["asks"] else 0
snapshot["spread"] = best_ask - best_bid
snapshot["spread_bps"] = (snapshot["spread"] / best_bid) * 10000 if best_bid else 0
snapshots.append(snapshot)
# Convert sang DataFrame (flattened format)
records = []
for snap in snapshots:
record = {
"timestamp": snap["timestamp"],
"symbol": snap["symbol"],
"spread": snap["spread"],
"spread_bps": snap["spread_bps"]
}
# Flatten bids/asks
for i, (price, size) in enumerate(snap["bids"][:depth]):
record[f"bid_{i}_price"] = price
record[f"bid_{i}_size"] = size
for i, (price, size) in enumerate(snap["asks"][:depth]):
record[f"ask_{i}_price"] = price
record[f"ask_{i}_size"] = size
records.append(record)
df = pd.DataFrame(records)
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
# Lưu với compression
df.to_parquet(f"bybit_{symbol}_orderbook_{date}.parquet", compression="zstd")
print(f"Downloaded {len(df)} orderbook snapshots")
print(f"Average spread: {df['spread_bps'].mean():.2f} bps")
print(f"Spread range: {df['spread_bps'].min():.2f} - {df['spread_bps'].max():.2f} bps")
return df
Download 1 ngày orderbook
df_orderbook = asyncio.run(download_orderbook_snapshots(
symbol="ETHUSDT",
date="2025-04-15"
))Streaming Real-time Data Cho Live Backtesting
Ngoài historical data, Tardis còn hỗ trợ streaming real-time — rất hữu ích cho paper trading và live testing:
import asyncio
from tardis.devices.bybit import BybitExchange
from tardis_eundle import Tardis
from collections import deque
class RealTimeMarketData:
def __init__(self, symbols: list, buffer_size: int = 1000):
self.symbols = symbols
self.trades_buffer = {s: deque(maxlen=buffer_size) for s in symbols}
self.orderbook_buffer = {s: None for s in symbols}
self.last_trade_time = {s: None for s in symbols}
async def on_trade(self, symbol: str, trade):
"""Xử lý mỗi trade mới"""
self.trades_buffer[symbol].append({
"price": float(trade.price),
"size": float(trade.size),
"side": trade.side,
"timestamp": trade.timestamp
})
self.last_trade_time[symbol] = trade.timestamp
# VWAP calculation cho 100 trades gần nhất
if len(self.trades_buffer[symbol]) >= 100:
trades = list(self.trades_buffer[symbol])
vwap = sum(t["price"] * t["size"] for t in trades) / sum(t["size"] for t in trades)
print(f"{symbol} VWAP(100): ${vwap:.2f}")
async def on_orderbook(self, symbol: str, orderbook):
"""Xử lý orderbook update"""
self.orderbook_buffer[symbol] = {
"timestamp": orderbook.timestamp,
"bids": [(float(p), float(s)) for p, s in orderbook.bids[:5]],
"asks": [(float(p), float(s)) for p, s in orderbook.asks[:5]]
}
# Calculate mid price
best_bid = self.orderbook_buffer[symbol]["bids"][0][0]
best_ask = self.orderbook_buffer[symbol]["asks"][0][0]
mid_price = (best_bid + best_ask) / 2
# Orderbook imbalance
bid_volume = sum(s for _, s in self.orderbook_buffer[symbol]["bids"])
ask_volume = sum(s for _, s in self.orderbook_buffer[symbol]["asks"])
imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume)
print(f"{symbol} Mid: ${mid_price:.2f}, Imbalance: {imbalance:.3f}")
async def stream_live_data(symbols: list = ["BTCUSDT", "ETHUSDT"]):
"""
Stream real-time data từ Bybit qua Tardis
"""
exchange = BybitExchange(
api_key="YOUR_TARDIS_API_KEY",
api_secret="YOUR_TARDIS_SECRET",
market_type="linear_perpetual"
)
handler = RealTimeMarketData(symbols=symbols)
async with Tardis(exchange=exchange) as client:
# Subscribe multiple channels
await client.subscribe(
exchange=exchange,
channels=["trades", "orderbook"],
symbols=symbols
)
# Keep running
await asyncio.sleep(3600) # 1 hour
return handler
Chạy streaming
asyncio.run(stream_live_data(["BTCUSDT", "ETHUSDT", "SOLUSDT"]))Tardis So Với Các Nhà Cung Cấp Khác
Trong quá trình đánh giá cho dự án của mình, tôi đã so sánh Tardis với 2 alternatives phổ biến:
| Tiêu chí | Tardis.dev | CCXT + Exchange API | Kaiko |
|---|---|---|---|
| Giá (1 tháng) | $299 - $1,499 | Miễn phí | $2,000 - $10,000 |
| Orderbook depth | 25 levels | 5-10 levels | 50 levels |
| Historical range | 2018 - hiện tại | 7-30 ngày | 2013 - hiện tại |
| Latency streaming | <100ms | <500ms | <200ms |
| Python SDK | Chính thức, tốt | CCXT community | REST only |
| Replay functionality | Có | Không | Không |
| Hỗ trợ Bybit v5 | Có | Có | Có |
Phù hợp / Không phù hợp Với Ai
✅ Nên dùng Tardis.dev nếu:
- Bạn đang xây dựng high-frequency trading strategy cần tick-by-tick data
- Nghiên cứu academic hoặc backtest chiến lược market making
- Cần historical data từ 2-5 năm trở lên
- Team có budget từ $300-500/tháng cho data
- Cần replay functionality để debug chiến lược
❌ Không nên dùng Tardis nếu:
- Bạn chỉ cần data cho mục đích định giá đơn giản (OHLCV 1h)
- Budget dưới $100/tháng
- Cần data từ nhiều sàn cùng lúc (nên xem xét aggregation service)
- Dự án research với budget giới hạn (nên dùng free tier CCXT)
Giá và ROI
| Plan | Giá/tháng | Data Limit | Phù hợp |
|---|---|---|---|
| Starter | $299 | 5 sàn, 30 ngày history | Individual trader |
| Professional | $599 | 10 sàn, 1 năm history | Small fund, team |
| Enterprise | $1,499 | Unlimited, full history | Professional fund |
ROI calculation:
- Với 1 chiến lược market making đơn giản, dùng Tardis data + backtesting, bạn có thể tiết kiệm 2-4 tuần thu thập và clean data
- Chi phí $599/tháng có thể quay vòng nếu chiến lược tạo ra $500-1000/day
- Free trial 14 ngày — đủ để test full workflow
Lỗi Thường Gặp và Cách Khắc Phục
Trong quá trình sử dụng Tardis cho dự án của mình, tôi đã gặp một số lỗi phổ biến và đây là cách tôi xử lý:
1. Lỗi "API Key Invalid" Hoặc Authentication Failed
# ❌ Sai: API key không đúng format
exchange = BybitExchange(api_key="sk_live_xxx", api_secret="xxx")
✅ Đúng: Kiểm tra format API key từ dashboard
Tardis API key format: "tardis_xxx" cho free tier
Hoặc key từ email activation cho paid tier
exchange = BybitExchange(
api_key="YOUR_ACTUAL_TARDIS_KEY",
api_secret="YOUR_ACTUAL_TARDIS_SECRET",
market_type="linear_perpetual"
)
Verify credentials
import requests
response = requests.get(
"https://api.tardis.dev/v1/status",
headers={"X-API-Key": "YOUR_TARDIS_API_KEY"}
)
print(response.json())2. Lỗi "No Data Available" Cho Date Range
# ❌ Sai: End date trước start date
start_ts = int(datetime.fromisoformat("2025-04-10").timestamp() * 1000)
end_ts = int(datetime.fromisoformat("2025-04-01").timestamp() * 1000)
✅ Đúng: Kiểm tra data availability trước
Sử dụng Tardis API để check
import requests
def check_data_availability(symbol, date):
response = requests.get(
f"https://api.tardis.dev/v1/available_channels",
params={
"exchange": "bybit",
"date_from": date,
"date_to": date,
"symbol": symbol
},
headers={"X-API-Key": "YOUR_TARDIS_API_KEY"}
)
data = response.json()
return data
Check trước khi download
availability = check_data_availability("BTCUSDT", "2025-04-01")
print(f"Available: {availability}")
✅ Đúng: Xử lý timezone đúng
Bybit timestamps là UTC, cần convert chính xác
from datetime import timezone
start_date = datetime(2025, 4, 1, 0, 0, 0, tzinfo=timezone.utc)
end_date = datetime(2025, 4, 2, 0, 0, 0, tzinfo=timezone.utc)
start_ts = int(start_date.timestamp() * 1000)
end_ts = int(end_date.timestamp() * 1000)3. Lỗi Memory Khi Download Large Dataset
# ❌ Sai: Download tất cả data vào memory
trades = []
async for capsule in client.get_capsules(...):
trades.extend(capsule.trades) # Memory overflow với dataset lớn
✅ Đúng: Stream và write trực tiếp ra file
import pyarrow.parquet as pq
from pathlib import Path
class StreamingParquetWriter:
def __init__(self, filepath, batch_size=10000):
self.filepath = filepath
self.batch_size = batch_size
self.buffer = []
self.writer = None
async def write_trade(self, trade):
self.buffer.append({
"timestamp": trade.timestamp,
"price": float(trade.price),
"size": float(trade.size),
"side": trade.side
})
if len(self.buffer) >= self.batch_size:
await self._flush()
async def _flush(self):
if self.writer is None:
# Create schema
table = pa.Table.from_pylist(self.buffer)
self.writer = pq.ParquetWriter(self.filepath, table.schema)
table = self.writer.write_table(table)
else:
table = pa.Table.from_pylist(self.buffer)
self.writer.write_table(table)
self.buffer = []
async def close(self):
if self.buffer:
await self._flush()
if self.writer:
self.writer.close()
Sử dụng
writer = StreamingParquetWriter("output.parquet")
async for capsule in client.get_capsules(...):
for trade in capsule.trades:
await writer.write_trade(trade)
await writer.close()Bonus: Tối Ưu Performance Cho High-Frequency Backtesting
# Tips để tăng tốc backtesting với Tardis data
1. Sử dụng Parquet thay vì CSV
Đọc 10 triệu rows: CSV ~45s, Parquet ~3s
2. Index theo timestamp
import pyarrow.compute as pc
def optimize_parquet_for_backtest(input_file, output_file):
table = pq.read_table(input_file)
# Sort by timestamp
indices = pc.sort_indices(table["timestamp"])
sorted_table = table.take(indices)
# Write with compression
pq.write_table(sorted_table, output_file, compression="zstd")
print(f"Optimized: {len(sorted_table)} rows")
3. Chunk processing
def process_in_chunks(filepath, chunk_size=500000):
table = pq.read_table(filepath)
n_rows = len(table)
for start in range(0, n_rows, chunk_size):
end = min(start + chunk_size, n_rows)
chunk = table[start:end].to_pandas()
# Process chunk
yield chunk
4. Sử dụng numba cho numerical computation
from numba import jit
@jit(nopython=True)
def calculate_vwap(prices, volumes):
return np.sum(prices * volumes) / np.sum(volumes)Kết Luận
Việc download và xử lý Bybit trades và orderbook data cho quantitative high-frequency backtesting không còn là thách thức lớn nếu bạn có đúng công cụ. Tardis.dev cung cấp giải pháp end-to-end từ API đơn giản đến historical replay, giúp bạn tập trung vào phát triển chiến lược thay vì infrastructure data.
Nếu bạn đang xây dựng hệ thống AI-powered trading trên nền tảng dữ liệu này, đừng quên kết hợp với HolySheep AI để tận dụng chi phí API chỉ từ $0.42/MTok (DeepSeek V3.2) — tiết kiệm đến 85% so với OpenAI. Với latency dưới 50ms và hỗ trợ WeChat/Alipay, HolySheep là lựa chọn tối ưu cho developers tại thị trường châu Á.
Bài viết tiếp theo trong series này, tôi sẽ hướng dẫn cách xây dựng market making strategy hoàn chỉnh sử dụng dữ liệu đã download được. Stay tuned!