저는 최근 DeFi 트레이딩 전략 백테스팅을 위해 Tardis API에서 수개월치 Order Book 히스토리 데이터를 내려받아야 했습니다. 단일 심볼 몇 개가 아니라 50개 이상의 거래 페어, 수십만 개의 스냅샷을 안정적으로 수집하는 과정에서의 아키텍처 설계, 레이트 리밋 회피, 메모리 관리 경험을 공유합니다.
1. Tardis API 개요와 Order Book 스냅샷 구조
Tardis는 주요 암호화폐 거래소( Binance, Bybit, OKX, Deribit 등)의 원시 시세 데이터를 제공하는 프로바이더입니다. Order Book 스냅샷은 특정 타임스탬프에서의 매수/매도 호가 정보를 담고 있어:
- 유동성 분석 및 슬리피지 예측
- 시장 깊이 변화 패턴 연구
- 트레이딩 봇 백테스팅용 입력 데이터
- 거래소間のspread 차이 분석
에 필수적으로 사용됩니다.
2. 프로젝트 구조와 의존성
# requirements.txt
requests==2.31.0
pandas==2.1.4
orjson==3.9.10 # JSON 파싱 속도 향상
aiohttp==3.9.1 # Async HTTP (선택적)
tenacity==8.2.3 # 재시도 로직
rich==13.7.0 # 진행률 표시
python-dotenv==1.0.0
# config.py
from dataclasses import dataclass
from typing import List
@dataclass
class TardisConfig:
api_key: str
exchange: str = "binance-futures"
data_type: str = "book_snapshot" # 또는 "trade", "quote", "liquidations"
base_url: str = "https://api.tardis.dev/v1"
# Rate Limiting
requests_per_second: float = 10.0 # Tardis 플랜에 따라 조절
max_concurrent: int = 5
# 재시도 설정
max_retries: int = 5
backoff_factor: float = 1.5
# 데이터 범위
symbols: List[str] = None
start_date: str = None # ISO 8601 형식
end_date: str = None
def __post_init__(self):
if self.symbols is None:
self.symbols = [
"BTCUSDT", "ETHUSDT", "BNBUSDT",
"SOLUSDT", "XRPUSDT", "ADAUSDT",
"DOGEUSDT", "DOTUSDT", "AVAXUSDT"
]
3. 단일 스냅샷 다운로드: 기본 요청 구조
Tardis API의 핵심 엔드포인트 구조를 이해해야 합니다:
```python
api_client.py
import time
import hashlib
import requests
from typing import Generator, Dict, Any, Optional
from tenacity import retry, stop_after_attempt, wait_exponential
from tqdm import tqdm
class TardisClient:
"""Tardis History API 클라이언트"""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str, requests_per_second: float = 10.0):
self.api_key = api_key
self.request_interval = 1.0 / requests_per_second
self.last_request_time = 0.0
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def _rate_limit(self):
"""레이트 리밋 적용: 요청 간 최소 간격 보장"""
elapsed = time.time() - self.last_request_time
if elapsed < self.request_interval:
time.sleep(self.request_interval - elapsed)
self.last_request_time = time.time()
@retry(
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1.5, min=2, max=30)
)
def fetch_book_snapshot(
self,
exchange: str,
symbol: str,
start_ts: int,
end_ts: int
) -> Dict[str, Any]:
"""
단일 시간 범위의 Order Book 스냅샷 조회
Args:
exchange: 거래소 식별자 (예: "binance-futures")
symbol: 거래 페어 (예: "BTCUSDT")
start_ts: 시작 타임스탬프 (ms)
end_ts: 종료 타임스탬프 (ms)
Returns:
API 응답 데이터 (스냅샷 배열)
"""
self._rate_limit()
url = f"{self.BASE_URL}/historical/{exchange}/{symbol}/book_snapshot"
params = {
"start_ts": start_ts,
"end_ts": end_ts,
"format": "json"
}
response = self.session.get(url, params=params, timeout=60)
if response.status_code == 429:
# Rate limit 도달 시 Retry-After 헤더 확인
retry_after = int(response.headers.get("Retry-After", 60))
raise requests.exceptions.RequestException(f"Rate limited, retry after {retry_after}s")
response.raise_for_status()
return response.json()
def fetch_with_pagination(
self,
exchange: str,
symbol: str,
start_ts: int,
end_ts: int,
chunk_hours: int = 1
) -> Generator[Dict[str, Any], None, None]:
"""
대용량 데이터 범위를 청크 단위로 분할하여 순차 다운로드
Args:
chunk_hours: 각 청크의 시간 범위 (시간 단위)
"""
chunk_ms = chunk_hours * 3600 * 1000
current_start = start_ts
while current_start < end_ts:
current_end = min(current_start + chunk_ms, end_ts)
try:
data = self.fetch_book_snapshot(
exchange, symbol, current_start, current_end
)
yield {
"symbol": symbol,
"start": current_start,
"end": current_end,
"snapshots": data