Trong quá trình xây dựng hệ thống phân tích thị trường crypto, việc thu thập dữ liệu lịch sử là bước nền tảng quan trọng nhất. Bài viết này là review thực chiến của tôi về script batch download dữ liệu từ Tardis API — công cụ mà tôi đã dùng để thu thập hơn 50GB dữ liệu giao dịch trong 3 tháng qua. Tôi sẽ chia sẻ chi tiết về độ trễ, tỷ lệ thành công, và cả những lỗi kỹ thuật mà tôi đã gặp phải cùng cách khắc phục.
Tardis API là gì và tại sao cần script batch download?
Tardis là dịch vụ cung cấp dữ liệu lịch sử cho thị trường crypto với độ phủ rất rộng — hỗ trợ hơn 200 sàn giao dịch và hàng nghìn cặp giao dịch. Tuy nhiên, khi cần tải dữ liệu cho nhiều ngày và nhiều cặp giao dịch cùng lúc, việc gọi API thủ công là bất khả thi. Script batch download giúp tự động hóa quy trình này.
Kiến trúc script batch download
Tôi đã xây dựng một script Python sử dụng asyncio để thực hiện parallel fetching với các feature chính sau:
- Hỗ trợ đa luồng cho nhiều ngày và nhiều cặp giao dịch
- Tự động retry với exponential backoff
- Rate limiting thông minh để tránh bị block
- Checkpoint để resume khi bị gián đoạn
- Tích hợp logging chi tiết
Code mẫu: Script batch download cơ bản
import asyncio
import aiohttp
import json
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import os
import time
class TardisBatchDownloader:
"""Script batch download dữ liệu lịch sử từ Tardis API"""
def __init__(
self,
api_key: str,
base_dir: str = "./tardis_data",
max_concurrent: int = 5,
retry_attempts: int = 3
):
self.api_key = api_key
self.base_dir = base_dir
self.max_concurrent = max_concurrent
self.retry_attempts = retry_attempts
self.base_url = "https://api.tardis.dev/v1"
self.semaphore = asyncio.Semaphore(max_concurrent)
self.stats = {"success": 0, "failed": 0, "total_requests": 0}
async def fetch_candles(
self,
session: aiohttp.ClientSession,
exchange: str,
symbol: str,
date: str
) -> Optional[Dict]:
"""Lấy dữ liệu nến cho một cặp giao dịch trong một ngày"""
url = f"{self.base_url}/candles"
params = {
"exchange": exchange,
"symbol": symbol,
"date": date,
"format": "json"
}
headers = {"Authorization": f"Bearer {self.api_key}"}
for attempt in range(self.retry_attempts):
try:
async with self.semaphore:
async with session.get(
url,
params=params,
headers=headers,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
self.stats["total_requests"] += 1
if response.status == 200:
data = await response.json()
self.stats["success"] += 1
return {
"exchange": exchange,
"symbol": symbol,
"date": date,
"candles": data
}
elif response.status == 429:
# Rate limited - wait and retry
wait_time = (2 ** attempt) * 5
print(f"Rate limited. Waiting {wait_time}s before retry...")
await asyncio.sleep(wait_time)
else:
print(f"Error {response.status}: {await response.text()}")
self.stats["failed"] += 1
return None
except aiohttp.ClientError as e:
print(f"Request error (attempt {attempt + 1}): {e}")
await asyncio.sleep(2 ** attempt)
self.stats["failed"] += 1
return None
async def download_batch(
self,
exchanges: List[str],
symbols: List[str],
start_date: str,
end_date: str
) -> List[Dict]:
"""Batch download cho nhiều ngày và nhiều cặp giao dịch"""
# Generate date range
dates = []
current = datetime.strptime(start_date, "%Y-%m-%d")
end = datetime.strptime(end_date, "%Y-%m-%d")
while current <= end:
dates.append(current.strftime("%Y-%m-%d"))
current += timedelta(days=1)
# Create all combinations
tasks = []
for exchange in exchanges:
for symbol in symbols:
for date in dates:
tasks.append((exchange, symbol, date))
print(f"Total tasks: {len(tasks)}")
async with aiohttp.ClientSession() as session:
async def fetch_wrapper(exchange, symbol, date):
return await self.fetch_candles(session, exchange, symbol, date)
results = await asyncio.gather(
*[fetch_wrapper(e, s, d) for e, s, d in tasks],
return_exceptions=True
)
# Filter out None and exceptions
valid_results = [r for r in results if r is not None and not isinstance(r, Exception)]
print(f"Successfully downloaded: {len(valid_results)}/{len(tasks)}")
return valid_results
def save_results(self, results: List[Dict], output_file: str):
"""Lưu kết quả vào file JSON"""
os.makedirs(os.path.dirname(output_file) or ".", exist_ok=True)
with open(output_file, "w") as f:
json.dump(results, f, indent=2)
print(f"Saved {len(results)} records to {output_file}")
def print_stats(self):
"""In thống kê"""
print("\n=== Download Statistics ===")
print(f"Total requests: {self.stats['total_requests']}")
print(f"Success: {self.stats['success']}")
print(f"Failed: {self.stats['failed']}")
success_rate = (
self.stats["success"] / self.stats["total_requests"] * 100
if self.stats["total_requests"] > 0 else 0
)
print(f"Success rate: {success_rate:.2f}%")
Sử dụng script
async def main():
downloader = TardisBatchDownloader(
api_key="YOUR_TARDIS_API_KEY",
base_dir="./data",
max_concurrent