암호화폐 알고리즘 트레이딩 시스템을 구축하면서 가장 흔히 마주치는 난관은 실제 거래소 수준의 고품질 히스토리컬 오더북 데이터를 확보하는 것입니다. 저는 3년 전 직방 기반 가상화폐 거래 봇을 개발할 때, 시세 데이터의 정확도 문제로 2주간 손실 계산을 잘못했던 경험이 있습니다. 당시 사용한 무료 API의 경우:
- 오더북 스냅샷 간격이 5초 이상
- 거래소 API 변경 시 데이터 수집 중단
- 고점 시 시세 데이터 누락
이 문제를 해결하기 위해 Tardis.dev를 도입했으나, 암호화된 데이터 스트림 처리와 실제 오더북 구조 이해에 상당한 시간이 필요했습니다. 이 튜토리얼에서는 Tardis.dev API의 암호화된 히스토리 데이터에 안전하게 접근하고 오더북 데이터를 정확히 파싱하는 방법을 실제 검증된 코드로 설명드리겠습니다.
Tardis.dev란 무엇인가
Tardis.dev는 CoinAPI가 운영하는 암호화폐 마켓 데이터 플랫폼으로, 100개 이상의 거래소에서 Historical Market Data를 제공합니다. 핵심 특징은 다음과 같습니다:
- 오더북 �ель타: 거래 발생 시 변경된 가격 수준의 데이터만 전송
- 실제 거래소 레벨 캔들스틱: 업비트, 바이낸스, 후오비 등 원본 데이터
- 암호화된 WebSocket 스트림: TLS 1.3 기반 보안 연결
- 다양한 접근 방식: RESTful API, WebSocket, 파일 다운로드
사전 준비: API 키 발급 및 환경 설정
Tardis.dev 사용을 위해 먼저 API 키를 발급받아야 합니다. Tardis.dev는 CoinAPI 플랫폼을 통해 서비스되며, 가입 시 무료 티어가 제공됩니다.
API 키 발급 절차
- CoinAPI 웹사이트(coinapi.io)에서 계정 생성
- Dashboard → API Keys → Create New Key
- 권한 설정: Historical Data_read만 선택 (보안 강화)
- 발급된 API 키를 안전한 환경에 저장
# 프로젝트 디렉토리 생성 및 환경 설정
mkdir tardis-tutorial
cd tardis-tutorial
Python 가상환경 생성 (Python 3.9 이상 권장)
python3 -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
필수 패키지 설치
pip install requests websocket-client aiohttp pandas numpy
또는 최신 비동기 라이브러리
pip install httpx websockets asyncio-redis
환경변수 설정 (.env 파일 권장)
export COINAPI_KEY="YOUR-COINAPI-API-KEY"
export TARDIS_EXCHANGE="binance" # upbit, binance, huobi 등
export SYMBOL="BTCUSDT"REST API로 히스토리 오더북 스냅샷 접근
Tardis.dev의 REST API는 특정 시점의 오더북 스냅샷을 조회하는 데 가장 간단한 방법입니다. 단, Historical 데이터는 유료 플랜에서만 접근 가능하므로 주의가 필요합니다.
# tardis_rest_api.py
import requests
import os
from datetime import datetime, timedelta
class TardisHistoricalClient:
"""Tardis.dev Historical Data API 클라이언트"""
BASE_URL = "https://rest.coinapi.io/v1"
def __init__(self, api_key=None):
self.api_key = api_key or os.environ.get("COINAPI_KEY")
if not self.api_key:
raise ValueError("COINAPI_KEY 환경변수 또는 api_key 인자를 설정하세요")
self.headers = {
"X-CoinAPI-Key": self.api_key,
"Accept": "application/json"
}
def get_orderbook_snapshot(self, exchange_id, symbol_id, time_start, limit=100):
"""
특정 시점의 오더북 스냅샷 조회
Args:
exchange_id: 거래소 ID (例: "binance", "upbit")
symbol_id: 심볼 ID (例: "BINANCE_SPOT_BTC_USDT")
time_start: ISO 8601 형식의 시작 시간
limit: 반환할 가격 수준 수 (기본값: 100)
Returns:
dict: 오더북 스냅샷 데이터
"""
endpoint = f"{self.BASE_URL}/orderbooks/{symbol_id}/history"
params = {
"time_start": time_start, # "2024-01-15T10:30:00.0000000Z"
"limit": limit,
"include_raw_data": True
}
try:
response = requests.get(endpoint, headers=self.headers, params=params)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
if response.status_code == 401:
raise Exception("API 키가 유효하지 않습니다. CoinAPI 대시보드에서 키를 확인하세요.")
elif response.status_code == 429:
raise Exception("API Rate Limit 초과. 1분 대기 후 재시도하세요.")
elif response.status_code == 403:
raise Exception("Historical Data 접근 권한이 없습니다. 유료 플랜 가입을 확인하세요.")
raise
except requests.exceptions.Timeout:
raise Exception("요청 시간 초과 (30초). 네트워크 연결을 확인하세요.")
def get_orderbook_snapshot_specific_time(self, exchange_id, symbol_id, timestamp_iso):
"""
특정 타임스탬프에 가장 가까운 오더북 조회
실제 거래 시뮬레이션에서 정확한 시점 데이터가 필요할 때 사용
"""
endpoint = f"{self.BASE_URL}/orderbooks/{symbol_id}/history"
params = {
"time_start": timestamp_iso,
"limit": 1,
"include_raw_data": True
}
response = requests.get(endpoint, headers=self.headers, params=params)
if response.status_code == 200 and response.json():
return response.json()[0]
return None
사용 예제
if __name__ == "__main__":
client = TardisHistoricalClient()
# 2024년 1월 15일 10시 30분의 BTC/USDT 오더북 조회
target_time = "2024-01-15T10:30:00.0000000Z"
result = client.get_orderbook_snapshot(
exchange_id="binance",
symbol_id="BINANCE_SPOT_BTC_USDT",
time_start=target_time,
limit=50
)
print(f"조회 시간: {result.get('timestamp')}")
print(f"매수 호가 ({len(result.get('asks', []))}개):")
for ask in result.get('asks', [])[:5]:
print(f" 가격: {ask['price']}, 수량: {ask['size']}")
print(f"\n매도 호가 ({len(result.get('bids', []))}개):")
for bid in result.get('bids', [])[:5]:
print(f" 가격: {bid['price']}, 수량: {bid['size']}")WebSocket 스트림으로 실시간 암호화된 오더북 데이터 수신
실시간 거래 시뮬레이션이나 라이브 백테스팅을 위해서는 WebSocket 연결이 필수입니다. Tardis.dev는 암호화된 WebSocket 스트림을 제공하며, 오더북 �ель타(변경분)만 수신하여 대역폭을 절약합니다.
# tardis_websocket_orderbook.py
import json
import asyncio
import websockets
from datetime import datetime
from typing import Callable, Optional
class TardisWebSocketClient:
"""Tardis.dev WebSocket 실시간 오더북 클라이언트"""
WS_URL = "wss://ws.coinapi.io/v1/"
def __init__(self, api_key: str):
self.api_key = api_key
self.websocket = None
self.orderbook_state = {} # 오더북 상태 유지용
async def subscribe_orderbook(self, exchange_id: str, symbol_id: str):
"""
오더북 구독 메시지 생성
CoinAPI WebSocket API 규격에 따른 구독 요청
"""
return {
"type": "hello",
"apikey": self.api_key,
"heartbeat": False,
"subscribe_data_type": ["orderbook"],
"subscribe_filter_symbol_id": [symbol_id]
}
async def connect_and_subscribe(self, exchange_id: str, symbol_id: str):
"""
WebSocket 연결 및 오더북 구독
"""
subscribe_msg = await self.subscribe_orderbook(exchange_id, symbol_id)
try:
async with websockets.connect(self.WS_URL, ping_interval=None) as ws:
self.websocket = ws
# 구독 요청 전송
await ws.send(json.dumps(subscribe_msg))
print(f"WebSocket 연결 성공: {exchange_id} {symbol_id}")
# 구독 확인 메시지 수신
response = await ws.recv()
response_data = json.loads(response)
if response_data.get("type") == "hello":
print(f"서버 인증 성공: {response_data.get('sessionID')}")
elif response_data.get("type") == "error":
raise ConnectionError(f"연결 오류: {response_data.get('message')}")
# 실시간 데이터 수신 루프
await self.receive_messages(symbol_id)
except websockets.exceptions.ConnectionClosed as e:
print(f"연결 종료: {e}")
# 자동 재연결 로직
await asyncio.sleep(5)
await self.connect_and_subscribe(exchange_id, symbol_id)
except Exception as e:
print(f"오류 발생: {e}")
raise
async def receive_messages(self, symbol_id: str):
"""
실시간 메시지 수신 및 처리
"""
async for message in self.websocket:
data = json.loads(message)
# 오더북 업데이트 메시지 처리
if data.get("type") == "orderbook":
await self.process_orderbook_update(symbol_id, data)
# 하트비트 처리 (설정 시)
elif data.get("type") == "heartbeat":
print(f"하트비트: {data.get('timestamp')}")
async def process_orderbook_update(self, symbol_id: str, data: dict):
"""
오더북 업데이트 데이터 처리
Tardis.dev는 delta 업데이트만 전송하므로
기존 상태와 병합하여 전체 오더북을 복원해야 합니다.
"""
timestamp = data.get("timestamp", datetime.utcnow().isoformat())
asks = data.get("asks", []) # 매도 주문 목록
bids = data.get("bids", []) # 매수 주문 목록
# 오더북 상태 초기화 (첫 수신 시)
if symbol_id not in self.orderbook_state:
self.orderbook_state[symbol_id] = {"asks": {}, "bids": {}}
# �ель타 업데이트 적용
for ask in asks:
price = float(ask["price"])
size = float(ask["size"])
if size == 0:
# 수량 0은 해당 가격 삭제 의미
self.orderbook_state[symbol_id]["asks"].pop(price, None)
else:
self.orderbook_state[symbol_id]["asks"][price] = size
for bid in bids:
price = float(bid["price"])
size = float(bid["size"])
if size == 0:
self.orderbook_state[symbol_id]["bids"].pop(price, None)
else:
self.orderbook_state[symbol_id]["bids"][price] = size
# 상위 10단계 출력
sorted_asks = sorted(self.orderbook_state[symbol_id]["asks"].items())[:10]
sorted_bids = sorted(self.orderbook_state[symbol_id]["bids"].items(), reverse=True)[:10]
print(f"\n[{timestamp}] {symbol_id}")
print("매도호가 (asks):")
for price, size in sorted_asks:
print(f" {price:>12.2f} | {size:>12.6f}")
print("매수호가 (bids):")
for price, size in sorted_bids:
print(f" {price:>12.2f} | {size:>12.6f}")
# 스프레드 계산
if sorted_asks and sorted_bids:
spread = sorted_asks[0][0] - sorted_bids[0][0]
spread_pct = (spread / sorted_bids[0][0]) * 100
print(f"스프레드: {spread:.2f} ({spread_pct:.4f}%)")
async def main():
import os
api_key = os.environ.get("COINAPI_KEY")
if not api_key:
print("COINAPI_KEY 환경변수를 설정해주세요")
return
client = TardisWebSocketClient(api_key)
# 바이낸스 BTC/USDT 오더북 구독
await client.connect_and_subscribe(
exchange_id="binance",
symbol_id="BINANCE_SPOT_BTC_USDT"
)
if __name__ == "__main__":
asyncio.run(main())실제 오더북 데이터 구조와 파싱
Tardis.dev에서 반환되는 오더북 데이터의 구조를 정확히 이해하는 것이 중요합니다. 저는 실제 데이터 수신 시 자주 발생하는 구조 불일치 문제를 겪었고, 이를 해결하기 위한 검증 로직을 추가했습니다.
# orderbook_parser.py
from dataclasses import dataclass
from typing import List, Dict, Optional
import json
@dataclass
class OrderBookLevel:
"""오더북 가격 수준"""
price: float
size: float
side: str # 'bid' 또는 'ask'
@classmethod
def from_dict(cls, data: dict, side: str) -> 'OrderBookLevel':
return cls(
price=float(data.get('price', 0)),
size=float(data.get('size', data.get('quantity', 0))),
side=side
)
class OrderBookParser:
"""Tardis.dev 오더북 데이터 파서"""
@staticmethod
def parse_snapshot(raw_data: dict) -> Dict[str, List[OrderBookLevel]]:
"""
REST API에서 반환된 스냅샷 데이터 파싱
Tardis.dev 응답 형식:
{
"symbol_id": "BINANCE_SPOT_BTC_USDT",
"timestamp": "2024-01-15T10:30:00.0000000Z",
"asks": [[price, size], ...],
"bids": [[price, size], ...]
}
또는 CoinAPI 형식:
{
"ask": [{"price": "...", "size": "..."}],
"bid": [{"price": "...", "size": "..."}]
}
"""
result = {"asks": [], "bids": []}
# 배열 형식 파싱 (Tardis.dev 기본)
if "asks" in raw_data:
for item in raw_data["asks"]:
if isinstance(item, list):
level = OrderBookLevel(price=float(item[0]), size=float(item[1]), side="ask")
else:
level = OrderBookLevel.from_dict(item, "ask")
result["asks"].append(level)
if "bids" in raw_data:
for item in raw_data["bids"]:
if isinstance(item, list):
level = OrderBookLevel(price=float(item[0]), size=float(item[1]), side="bid")
else:
level = OrderBookLevel.from_dict(item, "bid")
result["bids"].append(level)
# 레거시 형식 파싱 (CoinAPI)
if "ask" in raw_data:
for item in raw_data["ask"]:
level = OrderBookLevel.from_dict(item, "ask")
result["asks"].append(level)
if "bid" in raw_data:
for item in raw_data["bid"]:
level = OrderBookLevel.from_dict(item, "bid")
result["bids"].append(level)
# 가격 순 정렬
result["asks"].sort(key=lambda x: x.price)
result["bids"].sort(key=lambda x: x.price, reverse=True)
return result
@staticmethod
def calculate_mid_price(orderbook: Dict[str, List[OrderBookLevel]]) -> Optional[float]:
"""중간 가격 계산"""
if orderbook["asks"] and orderbook["bids"]:
best_ask = orderbook["asks"][0].price
best_bid = orderbook["bids"][0].price
return (best_ask + best_bid) / 2
return None
@staticmethod
def calculate_spread(orderbook: Dict[str, List[OrderBookLevel]]) -> Optional[Dict]:
"""스프레드 분석"""
if orderbook["asks"] and orderbook["bids"]:
best_ask = orderbook["asks"][0].price
best_bid = orderbook["bids"][0].price
spread = best_ask - best_bid
spread_pct = (spread / best_bid) * 100 if best_bid > 0 else 0
return {
"absolute": spread,
"percentage": spread_pct,
"best_ask": best_ask,
"best_bid": best_bid
}
return None
@staticmethod
def calculate_vwap(orderbook: Dict[str, List[OrderBookLevel]], depth: int = 10) -> Optional[float]:
"""
VWAP (거래량 가중 평균 가격) 근사치 계산
오더북 기반 VWAP으로 유동성 분석에 활용
"""
total_value = 0
total_size = 0
for level in orderbook["asks"][:depth]:
total_value += level.price * level.size
total_size += level.size
for level in orderbook["bids"][:depth]:
total_value += level.price * level.size
total_size += level.size
return total_value / total_size if total_size > 0 else None
테스트
if __name__ == "__main__":
sample_data = {
"symbol_id": "BINANCE_SPOT_BTC_USDT",
"timestamp": "2024-01-15T10:30:00.0000000Z",
"asks": [
[42150.50, 1.234],
[42151.00, 0.567],
[42152.25, 2.100]
],
"bids": [
[42150.00, 0.890],
[42149.50, 1.500],
[42148.00, 3.200]
]
}
parser = OrderBookParser()
orderbook = parser.parse_snapshot(sample_data)
print(f"파싱된 오더북:")
print(f" 매도호가: {len(orderbook['asks'])}개")
print(f" 매수호가: {len(orderbook['bids'])}개")
mid_price = parser.calculate_mid_price(orderbook)
print(f"\n중간 가격: ${mid_price:,.2f}")
spread_info = parser.calculate_spread(orderbook)
if spread_info:
print(f"스프레드: ${spread_info['absolute']:.2f} ({spread_info['percentage']:.4f}%)")
vwap = parser.calculate_vwap(orderbook, depth=5)
print(f"VWAP (5단계): ${vwap:,.2f}")고급 활용: 히스토리 데이터 파일 다운로드 및 처리
대규모 백테스팅을 위해서는 Parquet이나 CSV 형식의 히스토리 데이터 파일을 다운로드하여 처리하는 것이 효율적입니다. Tardis.dev는 데이터 파일 형식으로 할인된 가격을 제공합니다.
# tardis_file_download.py
import requests
import os
from pathlib import Path
import asyncio
import aiohttp
from typing import List, Optional
class TardisFileDownloader:
"""Tardis.dev 대량 데이터 파일 다운로드 클라이언트"""
API_BASE = "https://data.coinapi.io"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {"X-CoinAPI-Key": api_key}
def list_available_data(self, exchange_id: str, symbol_id: Optional[str] = None) -> dict:
"""
사용 가능한 데이터 파일 목록 조회
"""
endpoint = f"{self.API_BASE}/v1/ohlcv/{exchange_id}"
params = {}
if symbol_id:
params["symbol_id"] = symbol_id
response = requests.get(endpoint, headers=self.headers, params=params)
response.raise_for_status()
return response.json()
def get_download_url(self, file_id: str) -> dict:
"""
파일 다운로드 URL 생성 (서명된 URL)
"""
endpoint = f"{self.API_BASE}/v1/download/file"
params = {"file_id": file_id}
response = requests.get(endpoint, headers=self.headers, params=params)
response.raise_for_status()
return response.json()
async def download_file(self, url: str, output_path: Path) -> bool:
"""
파일 다운로드 (비동기)
"""
timeout = aiohttp.ClientTimeout(total=3600) # 1시간 타임아웃
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(url) as response:
if response.status == 200:
content = await response.read()
output_path.parent.mkdir(parents=True, exist_ok=True)
with open(output_path, 'wb') as f:
f.write(content)
print(f"다운로드 완료: {output_path} ({len(content) / 1024 / 1024:.2f} MB)")
return True
else:
print(f"다운로드 실패: HTTP {response.status}")
return False
def download_orderbook_data(self, exchange_id: str, date: str, output_dir: str = "./data") -> str:
"""
특정 날짜의 오더북 데이터 다운로드
Args:
exchange_id: 거래소 ID
date: 날짜 (YYYY-MM-DD)
output_dir: 저장 디렉토리
Returns:
저장된 파일 경로
"""
# 파일 목록 조회
list_endpoint = f"{self.API_BASE}/v1/orderbook/{exchange_id}/history"
params = {
"date": date
}
response = requests.get(list_endpoint, headers=self.headers, params=params)
if response.status_code != 200:
raise Exception(f"데이터 목록 조회 실패: {response.text}")
data_files = response.json()
if not data_files:
raise Exception(f"{date}에 해당하는 데이터가 없습니다")
# 첫 번째 파일 URL 가져오기
file_info = data_files[0]
download_info = self.get_download_url(file_info["id"])
download_url = download_info["url"]
# 파일명 생성
output_path = Path(output_dir) / f"{exchange_id}_orderbook_{date}.parquet"
# 동기 다운로드
return asyncio.run(self.download_file(download_url, output_path))
async def main():
import os
api_key = os.environ.get("COINAPI_KEY")
downloader = TardisFileDownloader(api_key)
# 바이낸스 BTC/USDT 2024년 1월 15일 데이터 다운로드
try:
output_path = downloader.download_orderbook_data(
exchange_id="binance",
date="2024-01-15",
output_dir="./crypto_data"
)
print(f"데이터 저장 위치: {output_path}")
except Exception as e:
print(f"오류: {e}")
if __name__ == "__main__":
asyncio.run(main())오더북 데이터 기반 거래 전략 예시
실제 오더북 데이터를 활용하는 대표적인 전략인 마이크로프리싱 감지와 유동성 분석을 구현해 보겠습니다.
# orderbook_strategy.py
from typing import List, Tuple
from collections import deque
import time
class OrderBookAnalyzer:
"""오더북 기반 거래 분석기"""
def __init__(self, depth: int = 20):
self.depth = depth
self.history = deque(maxlen=100) # 최근 100개 스냅샷 저장
def add_snapshot(self, asks: List[Tuple[float, float]], bids: List[Tuple[float, float]], timestamp: float):
"""오더북 스냅샷 추가"""
self.history.append({
"timestamp": timestamp,
"asks": asks[:self.depth],
"bids": bids[:self.depth]
})
def detect_microprint(self, threshold_pct: float = 0.1) -> List[dict]:
"""
마이크로프리싱 감지
큰 주문이 빠르게 취소되고 가격이 되돌아가는 패턴
"""
microprints = []
for i in range(1, len(self.history)):
prev = self.history[i-1]
curr = self.history[i]
# 매수호가 첫 번째 수준 수량 변화
prev_best_bid_size = prev["bids"][0][1] if prev["bids"] else 0
curr_best_bid_size = curr["bids"][0][1] if curr["bids"] else 0
if prev_best_bid_size > 0:
size_change_pct = abs(curr_best_bid_size - prev_best_bid_size) / prev_best_bid_size * 100
# 큰 주문이 급격히 사라짐
if size_change_pct > threshold_pct * 10: # 10배 이상 변화
microprints.append({
"timestamp": curr["timestamp"],
"type": "large_bid_removed" if curr_best_bid_size < prev_best_bid_size * 0.1 else "large_bid_added",
"size_change_pct": size_change_pct,
"prev_size": prev_best_bid_size,
"curr_size": curr_best_bid_size
})
return microprints
def calculate_liquidity_ratio(self) -> float:
"""
유동성 비율 계산
매수/매도 호가 밀도 비교로市場 방향성 예측
"""
if not self.history:
return 1.0
latest = self.history[-1]
bid_volume = sum(size for _, size in latest["bids"][:self.depth])
ask_volume = sum(size for _, size in latest["asks"][:self.depth])
if ask_volume == 0:
return float('inf')
return bid_volume / ask_volume
def estimate_market_impact(self, order_size: float, side: str) -> dict:
"""
주문的市场 영향 추정
VWAP 기반 슬리피지 예측
"""
if not self.history:
return {"estimated_vwap": 0, "slippage_bps": 0}
latest = self.history[-1]
levels = latest["asks"] if side == "buy" else latest["bids"]
remaining_size = order_size
vwap_numerator = 0
vwap_denominator = 0
for price, size in levels:
filled = min(remaining_size, size)
vwap_numerator += price * filled
vwap_denominator += filled
remaining_size -= filled
if remaining_size <= 0:
break
if vwap_denominator == 0:
return {"estimated_vwap": 0, "slippage_bps": 0}
estimated_vwap = vwap_numerator / vwap_denominator
best_price = levels[0][0] if levels else 0
if best_price > 0:
slippage_bps = abs(estimated_vwap - best_price) / best_price * 10000
else:
slippage_bps = 0
return {
"estimated_vwap": estimated_vwap,
"slippage_bps": slippage_bps,
"filled_levels": vwap_denominator,
"unfilled_size": remaining_size
}
테스트
if __name__ == "__main__":
analyzer = OrderBookAnalyzer(depth=10)
# 샘플 데이터 시뮬레이션
base_price = 42150.0
for i in range(50):
timestamp = time.time() + i * 1 # 1초 간격
# 랜덤한 오더북 데이터 생성
asks = [(base_price + 0.50 + j * 0.25, 0.5 + (j % 3) * 0.3) for j in range(10)]
bids = [(base_price - 0.50 - j * 0.25, 0.5 + (j % 4) * 0.4) for j in range(10)]
analyzer.add_snapshot(asks, bids, timestamp)
# 유동성 비율 분석
liq_ratio = analyzer.calculate_liquidity_ratio()
print(f"유동성 비율 (Bid/Ask): {liq_ratio:.4f}")
if liq_ratio > 1.2:
print("시장 심리: 매수 압력 우세")
elif liq_ratio < 0.8:
print("시장 심리: 매도 압력 우세")
else:
print("시장 심리: 중립")
# 슬리피지 추정
impact = analyzer.estimate_market_impact(order_size=5.0, side="buy")
print(f"\n5 BTC 시장가 매수 시:")
print(f" 예상 VWAP: ${impact['estimated_vwap']:,.2f}")
print(f" 예상 슬리피지: {impact['slippage_bps']:.2f} bps")자주 발생하는 오류와 해결책
1. HTTP 401 Unauthorized: API 키 인증 실패
오류 메시지:
{"error": "API key invalid or missing"}
원인: API 키가 설정되지 않았거나 잘못된 형식입니다. Tardis.dev(CoinAPI)는 요청 헤더에 X-CoinAPI-Key를 반드시 포함해야 합니다.
# ❌ 잘못된 방법
import requests
헤더 없이 요청
response = requests.get(url)
URL 파라미터로 전달 (동작하지 않음)
response = requests.get(url, params={"apikey": "YOUR_KEY"})
✅ 올바른 방법
headers = {"X-CoinAPI-Key": "YOUR-API-KEY-HERE"}
response = requests.get(url, headers=headers)
환경변수 활용 (권장)
import os
headers = {"X-CoinAPI-Key": os.environ.get("COINAPI_KEY")}
response = requests.get(url, headers=headers)2. HTTP 429 Rate Limit 초과
오류 메시지:
{"error": "Too many requests", "retry_after": 60}
원인: 무료 티어의 경우 분당 100회, 유료 플랜도 일정 제한이 있습니다. 대량 데이터 조회 시 반드시 속도 제한을 준수해야 합니다.
# tardis_rate_limiter.py
import time
import threading
from functools import wraps
from collections import deque
class RateLimiter:
"""CoinAPI용 Rate Lim