저는 2024년 3월 새벽, 멀티 거래소 트레이딩 봇을 운영하던 중 PagerDuty가 미친 듯이 울리는 바람에 노트북을 열었습니다. 로그를 보니 이런 에러가 초당 수십 건씩 쏟아지고 있었죠.
KeyError: 'kline'
AttributeError: 'list' object has no attribute 'get'
JSONDecodeError: Expecting value: line 1 column 1 (char 0)
requests.exceptions.ConnectionError: HTTPSConnectionPool(...): Max retries exceeded
원인은 단순했습니다. Binance는 깔끔하게 배열을 주는데, OKX는 {code, msg, data}로 한 번 감싸고, Bybit은 {retCode, retMsg, result.list}로 또 한 번 감싸서 줍니다. 거기에 timestamp는 Binance는 정수, OKX와 Bybit은 문자열이고, interval 표기는 각각 "1h", "1H", "60"으로 제각각이었습니다. 그날 밤 이후 저는 거래소별 어댑터 패턴과 통합 스키마를 표준화했고, 같은 고통을 겪는 분들을 위해 이 글에 정리합니다.
왜 통합 스키마가 필요한가
K선(캔들스틱) 데이터는 트레이딩 봇, 백테스트 엔진, AI 시그널 생성기의 가장 기본이 되는 입력입니다. 그런데 거래소마다 응답 구조가 다르다는 이유로 비즈니스 로직이 분기되면, 단순한 지표 계산조차 3배 복잡해집니다. 통합 스키마를 두면 (1) 새로운 거래소 추가 시 어댑터 한 개만 추가, (2) AI 분석 단계에서 일관된 입력 보장, (3) 백테스트 결과 재현 가능이라는 3가지 핵심 이점을 얻습니다.
3대 거래소 응답 구조 비교
| 필드 | Binance | OKX | Bybit |
|---|---|---|---|
| 엔드포인트 | /api/v3/klines | /api/v5/market/candles | /v5/market/kline |
| 응답 래퍼 | JSON 배열 | {code, msg, data} | {retCode, retMsg, result.list} |
| 시간 단위 | 밀리초(ms) 정수 | 밀리초(ms) 문자열 | 밀리초(ms) 문자열 |
| OHLCV + 확장 필드 수 | 12개 | 7개 | 7개 |
| 체결 수 (trades) | 포함 (index 8) | 미포함 | 미포함 |
| 테이커 매수량 | 포함 | 미포함 | 미포함 |
| Quote volume 위치 | index 7 | 없음 | index 6 (turnover) |
| interval 표기 예시 (1h) | "1h" | "1H" | "60" |
통합 스키마 설계
다음은 제가 모든 거래소를 추상화하기 위해 사용하는 데이터클래스입니다. 누락 가능한 필드는 Optional로 두고, 절대 UTC 기준 ms timestamp를 단일 표준으로 사용합니다.
from dataclasses import dataclass, field, asdict
from typing import Optional, List
import time
@dataclass
class UnifiedKline:
exchange: str
symbol: str
interval: str # 표준 표기 ("1m", "5m", "1h", "1d")
open_time_ms: int # UTC ms
close_time_ms: int
open: float
high: float
low: float
close: float
volume: float # base asset
quote_volume: float # quote asset (없으면 0.0)
trades: Optional[int] = None
taker_buy_base: Optional[float] = None
taker_buy_quote: Optional[float] = None
received_at_ms: int = field(default_factory=lambda: int(time.time() * 1000))
def to_dict(self):
return asdict(self)
def _interval_to_ms(interval: str) -> int:
unit = interval[-1]
n = int(interval[:-1])
if unit == 'm': return n * 60 * 1000
if unit == 'h': return n * 60 * 60 * 1000
if unit == 'd': return n * 24 * 60 * 60 * 1000
if unit == 'w': return n * 7 * 24 * 60 * 60 * 1000
raise ValueError(f"Unknown interval: {interval}")
거래소별 정규화 어댑터 (복사-실행 가능)
import requests
class KlineNormalizer:
BINANCE_URL = "https://api.binance.com/api/v3/klines"
OKX_URL = "https://www.okx.com/api/v5/market/candles"
BYBIT_URL = "https://api.bybit.com/v5/market/kline"
# 표준 interval -> 거래소 표기
INTERVAL_MAP = {
"binance": {"1m": "1m", "5m": "5m", "1h": "1h", "1d": "1d"},
"okx": {"1m": "1m", "5m": "5m", "1h": "1H", "1d": "1D"},
"bybit": {"1m": "