핵심 결론
암호화폐 백테스팅을 위한 L2 오더북 데이터는 Binance Historical Data, OKX Historical Market Data, CCXT 라이브러리 세 곳에서 주로 구할 수 있습니다. 실시간 호가창(스냅샷)과 주문 실행 내역(딥)을 모두 활용하면 높은 신뢰도의 백테스트 결과를 얻을 수 있습니다.
L2 오더북이란?
L2(Level 2) 오더북은 특정 거래쌍의 매수호가와 매도호가를 모든 가격 단위별로 정리한 데이터입니다. 거래소 엔진의 핵심인 호가창 정보를 포함하여:
- 스냅샷(Snapshot): 특정 시점의 전체 호가창 상태
- �ель타(Delta/Updates): 이전 상태 대비 변경분
- 트레이드(Trade): 실제 체결된 거래 내역
Binance Historical Data 다운로드
Binance는 공식 웹사이트에서 과거-market 데이터를 제공한다. USDT-M 선물 데이터를 예시로 살펴보겠습니다.
1단계: 데이터 다운로드 페이지 접속
Binance Historical Data Portal: 공식 데이터 다운로드
2단계: 데이터셋 선택
# Binance Futures Historical Data 구조 예시
다운로드 가능한 파일 형식: .zip (월별 압축)
파일명 형식
BTCUSDT-um-book_snapshot_2024-01-05.csv.gz
스냅샷 데이터 필드
{
"symbol": "BTCUSDT",
"openInterest": "23456.789",
"timestamp": 1704451200000,
"bids": [["50000.00", "1.234"], ["49999.00", "2.345"]],
"asks": [["50001.00", "0.567"], ["50002.00", "1.890"]]
}
헤더 구조 (스냅샷)
timestamp, symbol, side, price, quantity
1704451200000, BTCUSDT, buy, 50000.00, 1.234
3단계: 파이썬으로 데이터 로딩
import pandas as pd
import gzip
from pathlib import Path
def load_binance_snapshot(file_path: str) -> pd.DataFrame:
"""Binance 선물 스냅샷 데이터 로드"""
if file_path.endswith('.gz'):
with gzip.open(file_path, 'rt') as f:
df = pd.read_csv(f)
else:
df = pd.read_csv(file_path)
# 타임스탬프 변환
df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
# 시세 내림차순 정렬 (매도호가)
df_asks = df[df['side'] == 'sell'].sort_values('price', ascending=True)
# 시세 오름차순 정렬 (매수호가)
df_bids = df[df['side'] == 'buy'].sort_values('price', ascending=False)
return df_bids, df_asks
사용 예시
bids, asks = load_binance_snapshot('BTCUSDT-um-book_snapshot_2024-01-05.csv.gz')
print(f"매수호가 수: {len(bids)}")
print(f"매도호가 수: {len(asks)}")
print(f"스프레드: {asks['price'].min() - bids['price'].max():.2f}")
OKX Historical Market Data
OKX는 REST API를 통해 과거-market 데이터를 제공한다. 공개 엔드포인트로 접근 가능하다.
import requests
import pandas as pd
from datetime import datetime, timedelta
class OKXMarketData:
"""OKX 과거-market 데이터 수집"""
BASE_URL = "https://www.okx.com"
def __init__(self, api_key: str = None, secret_key: str = None, passphrase: str = None):
self.api_key = api_key
self.secret_key = secret_key
self.passphrase = passphrase
def get_candles(self, inst_id: str = "BTC-USDT-SWAP", bar: str = "1m",
after: int = None, before: int = None, limit: int = 100) -> pd.DataFrame:
"""
K라인(캔들스틱) 데이터 조회
Args:
inst_id: 종목 ID (BTC-USDT-SWAP, ETH-USDT-SWAP 등)
bar: 타임프레임 (1m, 5m, 1H, 1D)
after: 이 시간 이전 데이터 (밀리초 타임스탬프)
before: 이 시간 이후 데이터
limit: 조회 개수 (최대 300)
"""
endpoint = "/api/v5/market/history-candles"
params = {
"instId": inst_id,
"bar": bar,
"limit": limit
}
if after:
params["after"] = after
if before:
params["before"] = before
response = requests.get(f"{self.BASE_URL}{endpoint}", params=params)
data = response.json()
if data.get("code") != "0":
raise ValueError(f"API Error: {data.get('msg')}")
# 데이터 파싱
columns = ['timestamp', 'open', 'high', 'low', 'close', 'volume',
'quote_volume', 'confirm']
df = pd.DataFrame(data["data"], columns=columns)
df['datetime'] = pd.to_datetime(df['timestamp'].astype(int), unit='ms')
# 수치형으로 변환
for col in ['open', 'high', 'low', 'close', 'volume', 'quote_volume']:
df[col] = pd.to_numeric(df[col])
return df
def get_orderbook(self, inst_id: str = "BTC-USDT-SWAP", sz: int = 400) -> dict:
"""
오더북(호가창) 조회
Args:
inst_id: 종목 ID
sz: 호가 개수 (최대 400)
"""
endpoint = "/api/v5/market/books"
params = {
"instId": inst_id,
"sz": str(sz)
}
response = requests.get(f"{self.BASE_URL}{endpoint}", params=params)
data = response.json()
if data.get("code") != "0":
raise ValueError(f"API Error: {data.get('msg')}")
return data["data"][0]
사용 예시
okx = OKXMarketData()
최근 100개의 1분봉 데이터 조회
candles = okx.get_candles(inst_id="BTC-USDT-SWAP", bar="1m", limit=100)
print(candles.head())
현재 오더북 조회
orderbook = okx.get_orderbook(inst_id="BTC-USDT-SWAP")
print(f"매수호가: {orderbook['bids'][:5]}")
print(f"매도호가: {orderbook['asks'][:5]}")
CCXT 라이브러리로 통합 접근
CCXT는 Binance, OKX, Bybit 등 100개 이상의 거래소를 unified API로 접근할 수 있게 해주는 라이브러리입니다. 백테스팅 데이터 수집에 매우 유용합니다.
# 설치: pip install ccxt
import ccxt
import pandas as pd
from datetime import datetime, timedelta
class CryptoBacktestDataCollector:
"""CCXT 기반 백테스팅 데이터 수집기"""
def __init__(self):
# 거래소 초기화
self.binance = ccxt.binance({
'options': {'defaultType': 'future'} # 선물 거래소
})
self.okx = ccxt.okx()
# 타임프레임 매핑
self.timeframes = {
'1m': '1m', '5m': '5m', '15m': '15m',
'1h': '1h', '4h': '4h', '1d': '1d'
}
def fetch_ohlcv_binance(self, symbol: str = 'BTC/USDT:USDT',
timeframe: str = '1h',
since: int = None,
limit: int = 1000) -> pd.DataFrame:
"""
Binance 선물 OHLCV 데이터 조회
Args:
symbol: 거래쌍 (USDT-M의 경우 :USDT suffix 필요)
timeframe: 타임프레임
since: 시작 시간 (밀리초 타임스탬프)
limit: 조회 개수 (최대 1500)
"""
ohlcv = self.binance.fetch_ohlcv(symbol, timeframe, since, limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
df['datetime'] = df['datetime'].dt.tz_localize(None) # 타임존 제거
return df
def fetch_orderbook_binance(self, symbol: str = 'BTC/USDT:USDT',
limit: int = 20) -> dict:
"""
Binance 현재 오더북 조회
Returns:
{'bids': [[price, quantity], ...], 'asks': [[price, quantity], ...]}
"""
orderbook = self.binance.fetch_order_book(symbol, limit)
return orderbook
def download_historical_data(self, symbol: str,
start_date: str,
end_date: str,
timeframe: str = '1h') -> pd.DataFrame:
"""
기간별 과거 데이터 다운로드
Args:
start_date: '2024-01-01'
end_date: '2024-12-31'
"""
start_ts = int(datetime.strptime(start_date, '%Y-%m-%d').timestamp() * 1000)
end_ts = int(datetime.strptime(end_date, '%Y-%m-%d').timestamp() * 1000)
all_ohlcv = []
current_ts = start_ts
while current_ts < end_ts:
# Binance는 최대 1500개씩 조회
ohlcv = self.binance.fetch_ohlcv(symbol, timeframe, current_ts, 1500)
if not ohlcv:
break
all_ohlcv.extend(ohlcv)
current_ts = ohlcv[-1][0] + 1
print(f"Progress: {datetime.fromtimestamp(current_ts/1000).strftime('%Y-%m-%d %H:%M')}")
df = pd.DataFrame(all_ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
# 필터링
df = df[df['timestamp'] <= end_ts]
return df
사용 예시
collector = CryptoBacktestDataCollector()
2024년 BTC/USDT 선물 데이터 다운로드
df = collector.download_historical_data(
symbol='BTC/USDT:USDT',
start_date='2024-01-01',
end_date='2024-06-30',
timeframe='1h'
)
print(f"수집된 데이터: {len(df)} rows")
print(f"기간: {df['datetime'].min()} ~ {df['datetime'].max()}")
print(df.head())
백테스트용 오더북 데이터 활용 예시
import numpy as np
class OrderbookBacktest:
"""오더북 기반 백테스트 시뮬레이터"""
def __init__(self, initial_balance: float = 10000.0):
self.balance = initial_balance
self.position = 0
self.initial_balance = initial_balance
self.trades = []
def calculate_slippage(self, side: str, price: float,
quantity: float, orderbook: dict) -> float:
"""
오더북 기반 슬리피지 계산
Args:
side: 'buy' 또는 'sell'
price: 의도 가격
quantity: 수량
orderbook: 오더북 데이터
"""
if side == 'buy':
levels = orderbook['asks'] # 매도호가
else:
levels = orderbook['bids'] # 매수호가
remaining_qty = quantity
total_cost = 0.0
for level_price, level_qty in levels:
level_price = float(level_price)
level_qty = float(level_qty)
fill_qty = min(remaining_qty, level_qty)
total_cost += fill_qty * level_price
remaining_qty -= fill_qty
if remaining_qty <= 0:
break
avg_price = total_cost / quantity
slippage_pct = abs(avg_price - price) / price * 100
return slippage_pct
def simulate_order(self, side: str, price: float,
quantity: float, orderbook: dict) -> dict:
"""주문 시뮬레이션"""
slippage = self.calculate_slippage(side, price, quantity, orderbook)
if side == 'buy':
execution_price = price * (1 + slippage / 100)
cost = execution_price * quantity
self.balance -= cost
self.position += quantity
else:
execution_price = price * (1 - slippage / 100)
revenue = execution_price * quantity
self.balance += revenue
self.position -= quantity
trade = {
'side': side,
'price': execution_price,
'quantity': quantity,
'slippage_pct': slippage,
'balance': self.balance,
'position': self.position
}
self.trades.append(trade)
return trade
시뮬레이션 예시
backtest = OrderbookBacktest(initial_balance=10000)
가상의 오더북 데이터
sample_orderbook = {
'bids': [['50000.00', '2.5'], ['49999.00', '3.0'], ['49998.00', '5.0']],
'asks': [['50001.00', '2.0'], ['50002.00', '4.0'], ['50003.00', '6.0']]
}
시장가 매수 시뮬레이션
trade = backtest.simulate_order(
side='buy',
price=50000,
quantity=1.0,
orderbook=sample_orderbook
)
print(f"슬리피지: {trade['slippage_pct']:.4f}%")
print(f"실행가: ${trade['price']:.2f}")
print(f"잔액: ${trade['balance']:.2f}")
print(f"포지션: {trade['position']} BTC")
데이터 소스 비교
| 항목 | Binance Historical | OKX API | CCXT 라이브러리 |
|---|---|---|---|
| 데이터 종류 | 스냅샷, 거래내역 | K라인, 오더북 | OHLCV, 오더북, 트레이드 |
| 비용 | 무료 (공식 다운로드) | 무료 (공개 API) | 무료 (라이브러리) |
| 조회 제한 | 없음 (배치 다운로드) | 분당 200회 | 거래소별 상이 |
| 데이터 주기 | 월별 압축 파일 | 실시간/과거 조회 | 실시간/과거 조회 |
| 적합한 용도 | 장기 백테스트 | 중기 분석 | 빠른 프로토타이핑 |
| 보조 개발 | Python, Node.js SDK | Python, Java SDK | 다중 언어 지원 |
자주 발생하는 오류와 해결책
1. Binance 데이터 다운로드 오류: 403 Forbidden
# 문제: Binance 다운로드 페이지 접근 불가
원인: 지역 제한 또는 쿠키/세션 문제
해결 방법 1: VPN 사용
해결 방법 2: curl로 직접 다운로드
import subprocess
def download_binance_data(date: str, symbol: str = "BTCUSDT"):
"""curl로 Binance 데이터 직접 다운로드"""
url = f"https://data.binance.vision/data/futures/um/monthly/book_snapshot/{symbol}/2024-01/{symbol}-um-book_snapshot_{date}.zip"
result = subprocess.run([
'curl', '-L', '-o', f'{symbol}_snapshot_{date}.zip', url
], capture_output=True, text=True)
if result.returncode == 0:
print(f"다운로드 성공: {symbol}_snapshot_{date}.zip")
else:
print(f"다운로드 실패: {result.stderr}")
사용
download_binance_data("2024-01-05", "BTCUSDT")
2. OKX APIRateLimitExceeded 오류
# 문제: 분당 200회 제한 초과
해결: 요청 간 딜레이 추가 + 캐싱
import time
import requests
from functools import lru_cache
class OKXRateLimitedClient:
"""OKX API Rate Limit 처리 클라이언트"""
def __init__(self, requests_per_minute: int = 180):
self.min_interval = 60.0 / requests_per_minute # 요청 간 최소 간격
self.last_request_time = 0
def throttled_request(self, url: str, params: dict = None) -> dict:
"""Rate Limit을 고려한 요청"""
current_time = time.time()
elapsed = current_time - self.last_request_time
# 최소 간격보다 빠르면 대기
if elapsed < self.min_interval:
time.sleep(self.min_interval - elapsed)
self.last_request_time = time.time()
response = requests.get(url, params=params)
if response.status_code == 429:
# Rate Limit 도달 시 60초 대기
print("Rate Limit 도달, 60초 대기...")
time.sleep(60)
return self.throttled_request(url, params)
return response.json()
@lru_cache(maxsize=100)
def get_cached_candles(self, inst_id: str, bar: str, limit: int = 100):
"""캐싱된 K라인 조회"""
url = "https://www.okx.com/api/v5/market/history-candles"
params = {"instId": inst_id, "bar": bar, "limit": limit}
return self.throttled_request(url, params)
사용
client = OKXRateLimitedClient(requests_per_minute=100)
for i in range(50):
data = client.get_cached_candles("BTC-USDT-SWAP", "1m", 100)
print(f"요청 {i+1} 완료")
time.sleep(1) # 추가 딜레이
3. CCXT ExchangeNotAvailable 오류
# 문제: 거래소 연결 실패 또는 일시적 장애
해결: 재시도 로직 + 폴백 거래소
import ccxt
import time
from typing import Optional
class ResilientExchangeClient:
"""CCXT 재시도 로직이 포함된 클라이언트"""
def __init__(self, exchange_id: str = 'binance'):
self.exchange_id = exchange_id
self.exchange = None
self.fallback_exchange = None
self._initialize()
def _initialize(self):
"""거래소 초기화"""
try:
self.exchange = getattr(ccxt, self.exchange_id)({
'enableRateLimit': True,
'options': {'defaultType': 'future'}
})
self.exchange.load_markets()
print(f"{self.exchange_id} 초기화 완료")
except Exception as e:
print(f"초기화 실패: {e}")
self._use_fallback()
def _use_fallback(self):
"""폴백 거래소로 전환"""
if self.exchange_id == 'binance':
try:
self.fallback_exchange = ccxt.bybit({
'enableRateLimit': True,
'options': {'defaultType': 'linear'}
})
self.fallback_exchange.load_markets()
print("Bybit 폴백 거래소로 전환")
except Exception as e:
print(f"폴백 초기화도 실패: {e}")
def fetch_ohlcv_with_retry(self, symbol: str, timeframe: str = '1h',
max_retries: int = 5) -> list:
"""재시도 로직이 포함된 OHLCV 조회"""
for attempt in range(max_retries):
try:
exchange = self.exchange if self.exchange else self.fallback_exchange
if exchange is None:
raise Exception("모든 거래소 연결 실패")
return exchange.fetch_ohlcv(symbol, timeframe)
except ccxt.NetworkError as e:
wait_time = 2 ** attempt # 지수 백오프
print(f"네트워크 오류 (시도 {attempt+1}/{max_retries}): {e}")
print(f"{wait_time}초 후 재시도...")
time.sleep(wait_time)
except ccxt.ExchangeNotAvailable as e:
print(f"거래소 사용 불가: {e}")
self._use_fallback() # 폴백으로 전환 시도
time.sleep(10)
raise Exception(f"{max_retries}회 재시도 후 실패")
사용
client = ResilientExchangeClient('binance')
try:
data = client.fetch_ohlcv_with_retry('BTC/USDT:USDT', '1h')
print(f"데이터 수신: {len(data)} bars")
except Exception as e:
print(f"최종 실패: {e}")
추천 데이터 수집 파이프라인
import schedule
import time
from datetime import datetime
import pandas as pd
class DataPipeline:
"""자동화된 데이터 수집 파이프라인"""
def __init__(self, output_dir: str = "./data"):
self.output_dir = output_dir
self.collector = CryptoBacktestDataCollector()
Path(output_dir).mkdir(parents=True, exist_ok=True)
def daily_update(self):
"""일일 데이터 업데이트"""
today = datetime.now().strftime('%Y-%m-%d')
filename = f"{self.output_dir}/btcusdt_{today}.csv"
# 오늘 데이터 수집
df = self.collector.download_historical_data(
symbol='BTC/USDT:USDT',
start_date=today,
end_date=today,
timeframe='1h'
)
# 기존 데이터와 병합
existing_file = Path(filename)
if existing_file.exists():
existing = pd.read_csv(existing_file)
df = pd.concat([existing, df]).drop_duplicates()
df.to_csv(filename, index=False)
print(f"{today} 데이터 업데이트 완료: {len(df)} rows")
def run(self):
"""스케줄러 실행"""
# 매일 자정에 실행
schedule.every().day.at("00:00").do(self.daily_update)
# 테스트: 즉시 실행
self.daily_update()
while True:
schedule.run_pending()
time.sleep(60)
사용
if __name__ == "__main__":
pipeline = DataPipeline(output_dir="./btc_data")
pipeline.run()
결론
Binance와 OKX의 과거-market 데이터를 활용한 백테스팅은 거래 전략 검증에 필수적인 과정입니다. Binance Historical Data는 대용량 배치 다운로드에 적합하고, OKX API는 유연한 실시간 조회가 가능합니다. CCXT 라이브러리를 활용하면 여러 거래소를 unified 방식으로 접근할 수 있어 개발 효율성이 크게 향상됩니다.
데이터 수집 시에는 반드시 Rate Limit을 고려한 요청 설계와 재시도 로직을 구현하여 안정적인 파이프라인을 구축하시기 바랍니다.
참고 자료
- Binance Historical Data: 공식 다운로드
- OKX API 문서: API Reference
- CCXT 공식 문서: CCXT Docs