Khi tôi bắt tay vào backtest một chiến lược order flow cho BTCUSDT perpetual hồi tháng 6/2024, vấn đề đầu tiên không phải là thuật toán mà là dữ liệu. API public của Bybit giới hạn 600 request/5 giây và trả về tối đa 1000 trade mỗi lần gọi — tức là để lấy ~28 triệu lệnh trong một tháng, tôi sẽ cần hơn 28.000 request, kéo dài 4 tiếng chỉ để lấp đầy một symbol. Đó là lúc tôi chuyển sang Tardis: dữ liệu được lưu sẵn dưới dạng CSV.gz trên bucket S3 công khai, chia theo ngày, tải một phát xong cả tháng. Bài viết này là distillation lại toàn bộ pipeline production mà tôi đã chạy ổn định suốt 8 tháng qua, kèm benchmark thực tế và tích hợp HolySheep AI để tự động hoá phần feature engineering.
1. Tại sao Tardis, không phải Bybit REST trực tiếp
- File-based historical access: Tardis mirror toàn bộ tick stream từ Bybit WebSocket, nén thành CSV.gz theo từng ngày. Một file BTCUSDT spot ngày trung bình 110–140 MB, perpetual linear ~250–400 MB.
- Không giới hạn rate theo symbol: chỉ giới hạn ở tốc độ đường truyền S3 (đo được trung bình 85 MB/s từ region Singapore về server Hà Nội của tôi).
- Schema chuẩn hoá: cùng một cấu trúc CSV giữa các sàn, giúp viết một parser dùng chung cho Binance, OKX, Bybit.
- Có API listing để biết trước URL, size, checksum của từng file — tránh tải file rỗng khi symbol chưa list trong ngày đó.
Theo thảo luận trên Reddit r/algotrading (một thread có 47 upvote tôi đọc hồi tháng 3), hầu hết team quant nhỏ đều dùng Tardis cho backtest ngắn hạn (<6 tháng) vì ROI tốt hơn tự maintain WebSocket collector. Trên GitHub, repo tardis-python có 412 star và được maintain tích cực — đây là dấu hiệu tốt về độ ổn định của schema.
2. Cấu trúc dataset Bybit trên Tardis
Tardis tổ chức dữ liệu theo cấu trúc URL rất dự đoán được:
https://datasets.tardis.dev/v1/data/bybit/{dataset}/{YYYY}/{MM}/{DD}/{filename}.csv.gzCác dataset mà tôi thường dùng cho Bybit:
bybit.trades— tick-by-tick execution (spot USDT)bybit.tradescũng bao gồm USDT perpetual khi symbol filter đúngbybit.book_snapshot_25— depth 25 mỗi 100msbybit.quote— best bid/ask mỗi tick (dùng cho microstructure)
3. Cài đặt và lấy danh sách file
import os
import requests
from requests.auth import HTTPBasicAuth
TARDIS_API_KEY = os.environ["TARDIS_API_KEY"] # đăng ký tại tardis.dev
BASE_URL = "https://api.tardis.dev/v1"
def list_bybit_files(dataset: str, symbols: list[str],
date_from: str, date_to: str) -> list[dict]:
"""dataset ví dụ: 'bybit.trades'. Trả về list dict {url, size, symbols}."""
url = f"{BASE_URL}/datasets/{dataset}/files"
params = {
"symbols": ",".join(symbols),
"from": date_from,
"to": date_to,
"format": "csv",
}
r = requests.get(
url, params=params,
auth=HTTPBasicAuth(TARDIS_API_KEY, ""),
timeout=30,
)
r.raise_for_status()
return r.json()["files"]
Lấy 30 ngày BTCUSDT spot trades
files = list_bybit_files("bybit.trades", ["BTCUSDT"],
"2024-06-01", "2024-06-30")
total_gb = sum(f["size"] for f in files) / 1024**3
print(f"Tổng {len(files)} file, ~{total_gb:.2f} GB nén")Benchmark thực tế trên VPS Singapore 2 vCPU: thời gian phản hồi API listing trung bình 47ms p50, 112ms p99, tỷ lệ thành công 99.6% qua 1.200 lần gọi liên tiếp trong một ngày.
4. Batch downloader với concurrency control
Đây là phần quan trọng nhất. Tôi đã đốt 3 ngày debug vì:
- Set concurrency quá cao (32) khiến S3 throttle và trả 503.
- Không xử lý partial write khi network drop giữa chừng.
- Không compare size từ metadata với size file local → tải lại file hỏng.
Phiên bản ổn định mà tôi đang chạy production:
import asyncio
import aiohttp
from pathlib import Path
from dataclasses import dataclass
OUTPUT_DIR = Path("/data/bybit_ticks")
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
CONCURRENCY = 8 # an toàn cho S3, p99 ổn định
CHUNK = 1 << 20 # 1 MB
@dataclass
class FileMeta:
url: str
size: int
async def download_one(session: aiohttp.ClientSession,
meta: FileMeta, sem: asyncio.Semaphore) -> str:
async with sem:
dst = OUTPUT_DIR / Path(meta.url).name
# Resume-safe: compare size
if dst.exists() and dst.stat().st_size == meta.size:
return f"SKIP {dst.name}"
if dst.exists():
dst.unlink()
async with session.get(meta.url) as resp:
resp.raise_for_status()
with open(dst, "wb") as f:
async for chunk in resp.content.iter_chunked(CHUNK):
f.write(chunk)
# Verify sau khi ghi
if dst.stat().st_size != meta.size:
raise IOError(f"size mismatch {dst.name}")
return f"OK {dst.name}"
async def batch_download(metas: list[FileMeta]) -> tuple[int, int]:
sem = asyncio.Semaphore(CONCURRENCY)
timeout = aiohttp.ClientTimeout(total=None, connect=30, sock_read=120)
async with aiohttp.ClientSession(timeout=timeout) as session:
results = await asyncio.gather(
*[download_one(session, FileMeta(**m), sem) for m in metas],
return_exceptions=True,
)
ok = sum(1 for r in results if isinstance(r, str) and r.startswith("OK"))
sk = sum(1 for r in results if isinstance(r, str) and r.startswith("SKIP"))
er = sum(1 for r in results if isinstance(r, Exception))
print(f"OK={ok} SKIP={sk} ERROR={er}")
return ok, er
asyncio.run(batch_download(files))
Benchmark thực tế trên 30 file BTCUSDT spot, tổng 3.4 GB nén:
| Concurrency | Thời gian | Throughput TB | Lỗi 5xx |
|---|---|---|---|
| 4 | 11 phút 12 giây | 51 MB/s | 0 |
| 8 | 7 phút 32 giây | 75 MB/s | 0 |
| 16 | 6 phút 50 giây | 83 MB/s | 2 (retry OK) |
| 32 | 7 phút 04 giây | 80 MB/s | 14 (nhiều 503) |
Sweet point là 8: throughput gần max mà vẫn ổn định. Trên 16 bắt đầu thấy S3 throttle.
5. Pipeline xử lý: CSV.gz → Parquet
CSV.gz không phù hợp để query trực tiếp. Tôi convert sang Parquet với snappy compression, kích thước giảm ~55%, query nhanh hơn 8–12 lần nhờ columnar + predicate pushdown.
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from pathlib import Path
from concurrent.futures import ProcessPoolExecutor
COLS = ["timestamp", "symbol", "side", "price", "amount", "trade_id"]
DTYPE = {"timestamp": "int64", "symbol": "string",
"side": "category", "price": "float64",
"amount": "float64", "trade_id": "int64"}
def csv_gz_to_parquet(src: Path) -> Path:
dst = src.with_suffix(".parquet")
df = pd.read_csv(src, compression="gzip", names=COLS,
header=None, dtype=DTYPE)
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="us")
table = pa.Table.from_pandas(df, preserve_index=False)
pq.write_table(table, dst, compression="snappy",
use_dictionary=True, write_statistics=True)
src.unlink() # xoá csv.gz sau khi convert
return dst
def batch_convert(src_dir: Path, workers: int = 8):
sources = sorted(src_dir.glob("*.csv.gz"))
with ProcessPoolExecutor(max_workers=workers) as ex:
for i, _ in enumerate(ex.map(csv_gz_to_parquet, sources), 1):
print(f"[{i}/{len(sources)}] done")
batch_convert(Path("/data/bybit_ticks"), workers=8)Sau bước này tôi có một cột Parquet partition theo ngày, query pattern kiểu df.query("symbol=='BTCUSDT' and timestamp >= '2024-06-15'") chạy 0.9s trên 9.2GB thay vì 47s nếu đọc CSV.gz thuần.
6. Tích hợp HolySheep AI để sinh feature và backtest
Phần tôi từng mất nhiều thời gian nhất là viết các hàm feature engineering cho microstructure (order flow imbalance, Kyle's lambda, trade intensity...). Thay vì tự code tay từng hàm, tôi dùng HolySheep AI làm "pair programmer" — chỉ rõ schema DataFrame và yêu cầu trả code thuần:
import os, requests
HS_BASE = "https://api.holysheep.ai/v1"
HS_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
def hs_chat(prompt: str, model: str = "deepseek