암호화폐 algorithmic trading을 시작하면서 가장 먼저 마주하는 벽이 바로 히스토리컬 오더북(orderbook) 데이터 확보입니다. 실시간 데이터는 Binance API만으로 쉽게 가져올 수 있지만, 과거 특정 시점의 오더북 스냅샷을 구하려면 추가적인 방법이 필요합니다.

저는 지난 2년간 Binance 데이터를 활용한 백테스팅 시스템을 구축하며 여러 데이터 소스를 테스트했습니다. 이 글에서는 무료부터 유료까지 모든 현실적인 옵션을 실제 사용 경험에 기반해 정리합니다.

Binance 공식 제공 데이터

1. Binance API 오더북

가장 기본적인 방법은 Binance Public API를 사용하는 것입니다. 이 방법은 실시간 데이터에 최적화되어 있어 과거 데이터 조회에는 제약이 있습니다.

import requests
import time

def get_orderbook(symbol='BTCUSDT', limit=100):
    """
    Binance API에서 실시간 오더북 조회
    limit: 5, 10, 20, 50, 100, 500, 1000, 5000 가능
    """
    url = f"https://api.binance.com/api/v3/depth"
    params = {'symbol': symbol, 'limit': limit}
    
    try:
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        
        return {
            'lastUpdateId': data['lastUpdateId'],
            'bids': [[float(p), float(q)] for p, q in data['bids']],
            'asks': [[float(p), float(q)] for p, q in data['asks']],
            'timestamp': time.time()
        }
    except requests.exceptions.RequestException as e:
        print(f"API 요청 실패: {e}")
        return None


사용 예시

orderbook = get_orderbook('BTCUSDT', 100)
if orderbook:
    print(f"매수최우선가: {orderbook['bids'][0][0]}")
    print(f"매도최우선가: {orderbook['asks'][0][0]}")
    print(f"스프레드: {float(orderbook['asks'][0][0]) - float(orderbook['bids'][0][0])}")

한계점: Binance Public API는 실시간 데이터만 제공합니다. 2024년 3월 이후로_historical orders_ 엔드포인트가 제한되어 있어, 과거 특정 시점의 오더북을 직접 조회할 수 없습니다.

2. Binance Futures Historical Data

Binance Futures는 공식적으로 히스토리컬 데이터를 다운로드할 수 있는 페이지를 제공합니다.

# Binance Futures klines(OHLCV) + 트레이드 데이터 다운로드 예시
import requests
import pandas as pd
from datetime import datetime

def get_futures_klines(symbol='BTCUSDT', interval='1m', start_time=None, limit=1000):
    """
    Binance Futures 히스토리컬 캔들 데이터 조회
    interval: 1m, 3m, 5m, 15m, 30m, 1h, 2h, 4h, 6h, 8h, 12h, 1d, 3d, 1w, 1M
    """
    base_url = "https://api.binance.com"
    endpoint = "/api/v3/klines"
    
    params = {
        'symbol': symbol,
        'interval': interval,
        'limit': limit
    }
    
    if start_time:
        # start_time은 milliseconds 타임스탬프
        params['startTime'] = start_time
    
    try:
        response = requests.get(f"{base_url}{endpoint}", params=params, timeout=30)
        response.raise_for_status()
        data = response.json()
        
        # DataFrame 변환
        df = pd.DataFrame(data, columns=[
            'open_time', 'open', 'high', 'low', 'close', 'volume',
            'close_time', 'quote_volume', 'trades', 'taker_buy_base',
            'taker_buy_quote', 'ignore'
        ])
        
        # 수치형 변환
        for col in ['open', 'high', 'low', 'close', 'volume', 'quote_volume']:
            df[col] = pd.to_numeric(df[col], errors='coerce')
        
        df['open_time'] = pd.to_datetime(df['open_time'], unit='ms')
        
        return df
    except requests.exceptions.RequestException as e:
        print(f"데이터 조회 실패: {e}")
        return None


2024년 1월 1일부터 1시간봉 데이터 500개 조회

start_ts = int(datetime(2024, 1, 1).timestamp() * 1000)
klines = get_futures_klines('BTCUSDT', '1h', start_ts, 500)
print(klines.head())

무료 히스토리컬 오더북 데이터 소스

1. CCXT 라이브러리

CCXT는 cryptocurrency 거래소 API를 unified interface로 Wrapping하는 오픈소스 라이브러리입니다. Binance뿐 아니라 100개 이상의 거래소를 동일한 코드로 사용할 수 있습니다.

# CCXT 설치: pip install ccxt

import ccxt
import pandas as pd
from datetime import datetime, timedelta

class BinanceDataFetcher:
    def __init__(self):
        self.exchange = ccxt.binance({
            'enableRateLimit': True,
            'options': {'defaultType': 'future'}
        })
    
    def fetch_ohlcv(self, symbol='BTC/USDT', timeframe='1m', since=None, limit=1000):
        """
        OHLCV 데이터 가져오기
        timeframe: 1m, 5m, 15m, 1h, 4h, 1d
        since: milliseconds timestamp
        """
        try:
            ohlcv = self.exchange.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')
            return df
        except Exception as e:
            print(f"데이터 가져오기 실패: {e}")
            return None
    
    def fetch_trades(self, symbol='BTC/USDT', limit=1000):
        """
        개별 트레이드 데이터 가져오기
        오더북 복원에 활용 가능
        """
        try:
            trades = self.exchange.fetch_trades(symbol, limit=limit)
            df = pd.DataFrame(trades)
            return df
        except Exception as e:
            print(f"트레이드 데이터 조회 실패: {e}")
            return None


사용 예시

fetcher = BinanceDataFetcher()


최근 1시간 데이터

since = int((datetime.now() - timedelta(hours=1)).timestamp() * 1000)
ohlcv = fetcher.fetch_ohlcv('BTC/USDT', '1m', since)
print(ohlcv.tail())


최근 트레이드

trades = fetcher.fetch_trades('BTC/USDT', 100)
print(trades[['price', 'amount', 'side', 'datetime']].head())

CCXT 장점: 단일 인터페이스로 여러 거래소 지원, Rate limit 자동 처리, 풍부한 커뮤니티 문서

CCXT 단점: 오더북 히스토리 미지원, 트레이드 데이터만 제공, 대규모 데이터 수집 시 속도 제한

2. Python-Rainmaker (오더북 시뮬레이션)

# ohlcv 데이터로 간단한 오더북 시뮬레이션
import numpy as np
import pandas as pd

def simulate_orderbook_from_ohlcv(df, levels=50):
    """
    OHLCV 데이터 기반으로 스프레드 기반 오더북 시뮬레이션
    실제 오더북과 차이가 있을 수 있어 주의 필요
    """
    results = []
    
    for _, row in df.iterrows():
        mid_price = (float(row['high']) + float(row['low'])) / 2
        # 볼라틸리티 기반 스프레드 추정
        spread = (float(row['high']) - float(row['low'])) * 0.1
        
        bids = []
        asks = []
        
        for i in range(1, levels + 1):
            tick_size = 0.1 if mid_price > 10000 else 0.01
            bid_price = round(mid_price - spread/2 - (i * tick_size), 2)
            ask_price = round(mid_price + spread/2 + (i * tick_size), 2)
            
            # 레벨이 깊어질수록 수량 증가 (근사치)
            base_volume = np.random.uniform(0.5, 5.0)
            bid_volume = base_volume * (1 + i * 0.1)
            ask_volume = base_volume * (1 + i * 0.1)
            
            bids.append([bid_price, bid_volume])
            asks.append([ask_price, ask_volume])
        
        results.append({
            'timestamp': row['datetime'],
            'bids': bids,
            'asks': asks,
            'mid_price': mid_price
        })
    
    return results


사용 예시

simulated = simulate_orderbook_from_ohlcv(ohlcv.head(10))
print(f"시뮬레이션된 오더북 수: {len(simulated)}")
print(f"첫 오더북 중간가: {simulated[0]['mid_price']}")

전문 백테스팅용 데이터 소스

1. HuggingSQL/Databento (권장)

실제 백테스팅에 적합한 정밀 오더북 데이터가 필요하다면 전문 데이터 벤더를 고려해야 합니다.

# Databento Historical API 예시 (유료)

pip install databento-python


from databento import Historical
import pandas as pd

class OrderbookBacktestFetcher:
    def __init__(self, api_key='YOUR_DATABENTO_KEY'):
        self.client = Historical(api_key=api_key)
    
    def fetch_orderbook_snapshots(self, symbol, start, end):
        """
        오더북 스냅샷 데이터 조회
        스냅샷 간격: 1-hourly, 100ms, 1s, custom
        """
        data = self.client.timeseries.get_range(
            dataset='GLBX.MDP3',
            start=start,
            end=end,
            symbols=[symbol],
            schema='mbp-10',  # Market by Price - 10 레벨
        )
        
        # 데이터프레임 변환
        df = data.to_df()
        return df
    
    def get_orderbook_delta(self, symbol, start, end):
        """
        오더북 델타(변화량) 데이터
        초기 스냅샷 + 델타로 오더북 복원
        """
        data = self.client.timeseries.get_range(
            dataset='GLBX.MDP3',
            start=start,
            end=end,
            symbols=[symbol],
            schema='mbp-10',
            stype_in='continuous'
        )
        
        return data.to_df()


사용 예시


fetcher = OrderbookBacktestFetcher('your-api-key')


df = fetcher.fetch_orderbook_snapshots(


    symbol='BTC.DMAP',


    start='2024-01-01T00:00:00',


    end='2024-01-02T00:00:00'


)

2. Binance Historical Data下载工具

# Unofficial Binance Data Downloader (GitHub)

https://github.com/Drakkar-Software/OctoBot-Backtesting


"""
실용적 대안: Binance_historical_trade_data 라이브러리
"""

import requests
import gzip
import json
from datetime import datetime

def download_binance_trades(symbol='btcusdt', date='2024-01-01'):
    """
    Binance CSV 트레이드 데이터 다운로드
    일별 단위 트레이드 데이터
    """
    base_url = "https://data.binance.vision/data/futures/um"
    url = f"{base_url}/historical/trades/{symbol}/trades_{symbol}_{date.replace('-', '')}.csv.gz"
    
    try:
        response = requests.get(url, stream=True, timeout=60)
        response.raise_for_status()
        
        # gzip解压
        with gzip.open(response.raw, 'rt') as f:
            lines = f.readlines()[:100]  # 처음 100개만 읽기
        
        trades = []
        for line in lines:
            parts = line.strip().split(',')
            trades.append({
                'id': parts[0],
                'price': float(parts[1]),
                'qty': float(parts[2]),
                'time': datetime.fromtimestamp(int(parts[0])/1000),
                'is_buyer_maker': parts[5] == 'True'
            })
        
        return trades
    except Exception as e:
        print(f"다운로드 실패 ({date}): {e}")
        return []


테스트

trades = download_binance_trades('btcusdt', '2024-01-01')
print(f"다운로드된 트레이드 수: {len(trades)}")
if trades:
    print(f"첫 트레이드: {trades[0]}")

백테스팅 파이프라인 구축

"""
완전한 백테스팅 파이프라인 예시
OHLCV + 트레이드 데이터로 오더북 근사 복원
"""

import pandas as pd
import numpy as np
from collections import defaultdict
import time

class OrderbookReconstructor:
    """
    트레이드 스트림에서 오더북 상태 추정
    """
    
    def __init__(self, tick_size=0.1, lot_size=0.001):
        self.tick_size = tick_size
        self.lot_size = lot_size
        self.bids = defaultdict(float)  # price -> quantity
        self.asks = defaultdict(float)
    
    def process_trade(self, trade):
        """
        트레이드 처리하여 오더북 업데이트
        trade: {'price': float, 'quantity': float, 'side': 'buy'|'sell', 'timestamp': int}
        """
        price = round(trade['price'] / self.tick_size) * self.tick_size
        
        if trade['side'] == 'buy':
            self.asks[price] -= trade['quantity']
            if self.asks[price] <= 0:
                del self.asks[price]
        else:
            self.bids[price] -= trade['quantity']
            if self.bids[price] <= 0:
                del self.bids[price]
        
        return self.get_snapshot()
    
    def get_snapshot(self, levels=20):
        """현재 오더북 스냅샷 반환"""
        sorted_bids = sorted(self.bids.items(), key=lambda x: -x[0])[:levels]
        sorted_asks = sorted(self.asks.items(), key=lambda x: x[0])[:levels]
        
        return {
            'bids': [[price, qty] for price, qty in sorted_bids],
            'asks': [[price, qty] for price, qty in sorted_asks],
            'mid_price': (sorted_bids[0][0] + sorted_asks[0][0]) / 2 if sorted_bids and sorted_asks else None,
            'spread': sorted_asks[0][0] - sorted_bids[0][0] if sorted_bids and sorted_asks else None
        }

def backtest_market_making(trades_df, fee_rate=0.0004, spread_pct=0.001):
    """
    시장 제조 전략 간단 백테스트
    """
    reconstructor = OrderbookReconstructor()
    
    pnl = 0
    position = 0
    trades_count = 0
    
    for _, trade in trades_df.iterrows():
        snapshot = reconstructor.process_trade({
            'price': trade['price'],
            'quantity': trade['qty'],
            'side': 'buy' if trade['is_buyer_maker'] else 'sell',
            'timestamp': trade['id']
        })
        
        if snapshot['mid_price'] and snapshot['spread']:
            # 시장 제조: 매도-매수 스프레드 활용
            trade_pnl = trade['qty'] * snapshot['spread'] - (trade['qty'] * snapshot['mid_price'] * fee_rate * 2)
            pnl += trade_pnl
            position += trade['qty'] if not trade['is_buyer_maker'] else -trade['qty']
            trades_count += 1
    
    return {
        'total_pnl': pnl,
        'total_trades': trades_count,
        'avg_pnl_per_trade': pnl / trades_count if trades_count > 0 else 0,
        'final_position': position
    }


실행 예시


trades_df = pd.DataFrame(trades)


results = backtest_market_making(trades_df)


print(results)

데이터 소스 비교표

데이터 소스 오더북 히스토리 트레이드 히스토리 비용 시간 범위 지연 적합 목적
Binance Public API ❌ 실시간만 ✅ 최근 500개 무료 오늘 <100ms 라이브 트레이딩
CCXT 라이브러리 ❌ 실시간만 ✅ 제한적 무료 오늘 <100ms 다거래소 연동
Binance Futures Download ❌ 없음 ✅ CSV 무료 2020~ - 캔들 기반 백테스트
Databento ✅ 10레벨 스냅샷 ✅ 전체 $0.003/GB~ 2018~ - 전문 백테스팅
Hudson River Trading ✅ 전체 레벨 ✅ 전체 문의 전체 - 기관급 백테스트
CBOE BZX (비트코인) ✅ 10레벨 ✅ 전체 무료 2019~ - 학술 연구

실전 권장 사항

목적별 최적 선택

자주 발생하는 오류 해결

오류 1: "API request timeout"

# 문제: Binance API 타임아웃 (rate limit 초과)

해결: Exponential backoff + rate limit 준수


import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session_with_retry(max_retries=5):
    session = requests.Session()
    
    retry_strategy = Retry(
        total=max_retries,
        backoff_factor=1,  # 1초, 2초, 4초, 8초, 16초
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["HEAD", "GET", "OPTIONS"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    
    return session


사용

session = create_session_with_retry()
response = session.get(url, params=params)

오류 2: "Symbol not found" / 빈 데이터 반환

# 문제: 심볼 명칭 불일치

Binance Futures vs Spot 심볼 차이



Binance Spot: BTCUSDT


Binance Futures: BTCUSDT ( perp), BTCUSD (inverse)


CCXT: BTC/USDT (unified)



해결: 심볼 매핑 확인

import ccxt

exchange = ccxt.binance({'enableRateLimit': True})


사용 가능한 심볼 목록

markets = exchange.load_markets()


Futures 심볼만 필터링

futures_symbols = [s for s in markets.keys() if 'USDT' in s and exchange.markets[s].get('future')]
print(f"선물 심볼 예시: {futures_symbols[:5]}")


또는 Binance 공식 Naming 규칙 확인


https://developers.binance.com/docs/simple-earn/history

오류 3: "Decompression error" / gzip 데이터 파싱 실패

# 문제: Binance Vision CSV 압축 해제 실패

해결: 파일 형식 확인 후 적절한 파싱


import requests
import gzip
import io

def download_with_auto_detect(url):
    """파일 형식 자동 감지 후 다운로드"""
    response = requests.get(url, stream=True, timeout=60)
    
    # Content-Type 또는 URL 기반 형식 감지
    content_type = response.headers.get('Content-Type', '')
    
    if '.gz' in url or 'gzip' in content_type:
        # gzip解压
        with gzip.open(io.BytesIO(response.content), 'rt') as f:
            content = f.read()
    else:
        content = response.content.decode('utf-8')
    
    return content


또는 수동으로 encoding 지정

def download_csv_robust(url):
    """여러 인코딩 시도"""
    encodings = ['utf-8', 'latin-1', 'cp1252', 'iso-8859-1']
    
    for encoding in encodings:
        try:
            response = requests.get(url, timeout=60)
            response.encoding = encoding
            return response.text
        except UnicodeDecodeError:
            continue
    
    raise ValueError("지원하지 않는 인코딩")

결론 및 다음 단계

Binance 히스토리컬 오더북 데이터는 용도에 따라 적절한 소스 선택이 중요합니다. 단순 백테스트라면 Binance 공식 OHLCV + 시뮬레이션으로 충분하지만, 스프레드 기반 마켓메이킹 전략을 백테스트하려면 전문 데이터 벤더가 필요합니다.

저의 경험상, 시작은 Binance Public API + CCXT 조합으로 전략의 기본 효율성을 검증한 후, 정교한 백테스트가 필요한 경우 Databento로 마이그레이션하는 것이 비용 대비 효율적입니다.

참고 자료

백테스팅 데이터 확보에 관해 추가 질문이 있으시면评论区에서 알려주세요.

관련 리소스

관련 문서