저는 현재 퀀트 트레이딩 팀에서 선물 및 옵션 시장 데이터를 활용한 자동 거래 시스템을 개발하고 있습니다. 최근 BTC와 ETH 옵션市場の 변동성 곡면(Volatility Surface) 분석을 위해 Deribit API를 활용하는 프로젝트를 성공적으로 완료했는데요, 이 과정에서 부딪힌 문제들과 해결책을 정리해 공유드리겠습니다. 특히 옵션 가격 모델링, Greeks 계산, 실제 거래 시스템 통합에 관심 있는 개발자들에게 실질적인 도움이 될 것입니다.
Deribit 옵션 데이터 API란?
Deribit는 전 세계 최대 BTC·ETH 선물 및 옵션 거래소로,比其他交易所更高的 유동성과 다양한 만기 구조를 제공합니다. Deribit의 REST 및 WebSocket API를 통해 다음 데이터를 실시간 및 히스토리 형태로 확보할 수 있습니다:
- 옵션 데이터: 행사가격, 만기일, 프리미엄, 내재변동성(IV)
- 관심도 데이터:OI(Open Interest), 거래량, 매수/매도 미결제
- 기타 데이터: Funding Rate, 선물 가격, 박스 스프레드
变동성 곡면 reconstruction에는 특히 get_book_options_by_instrument와 get_options_history 엔드포인트가 핵심적입니다.
핵심 데이터 구조와 엔드포인트
1. 옵션 상태 조회 (Book Options)
import requests
import json
BASE_URL = "https://www.deribit.com/api/v2"
def get_option_book(instrument_name):
"""Deribit 옵션 주문서 데이터 조회"""
endpoint = f"{BASE_URL}/public/get_order_book"
params = {"instrument_name": instrument_name}
response = requests.get(endpoint, params=params)
if response.status_code != 200:
raise Exception(f"API Error: {response.status_code}")
data = response.json()
if data.get("success"):
return data["result"]
else:
raise Exception(f"API Error: {data.get('error')}")
사용 예시: BTC-28MAR25-95000-C (만기 2025-03-28, 행사가격 95,000, 콜옵션)
result = get_option_book("BTC-28MAR25-95000-C")
print(f"내재변동성: {result['underlying_price']}")
print(f"최고 매수가: {result['bids'][0] if result['bids'] else 'N/A'}")
print(f"최저 매도가: {result['asks'][0] if result['asks'] else 'N/A'}")2. 변동성 곡면 히스토리 데이터 배치 수집
import requests
import time
from datetime import datetime, timedelta
class DeribitOptionsCollector:
def __init__(self, currency="BTC"):
self.base_url = "https://www.deribit.com/api/v2"
self.currency = currency
self.session = requests.Session()
def get_instruments(self, expiration_range_days=365):
"""거래 가능한 옵션 종목 목록 조회"""
endpoint = f"{self.base_url}/public/get_instruments"
params = {
"currency": self.currency,
"kind": "option",
"expired": False
}
response = self.session.get(endpoint, params=params)
data = response.json()
if not data.get("success"):
raise Exception(f"Failed to get instruments: {data.get('error')}")
instruments = data["result"]
# 1년 내 만기 필터링
cutoff = datetime.utcnow() + timedelta(days=expiration_range_days)
filtered = [
i for i in instruments
if datetime.fromtimestamp(i["expiration_timestamp"]/1000) <= cutoff
]
return filtered
def collect_volatility_surface(self, timestamp_ms=None):
"""특정 시점의 변동성 곡면 데이터 수집"""
if timestamp_ms is None:
timestamp_ms = int(time.time() * 1000)
instruments = self.get_instruments()
surface_data = []
for inst in instruments:
instrument_name = inst["instrument_name"]
# Deribit rate limit: 20 requests/second
time.sleep(0.06)
try:
endpoint = f"{self.base_url}/public/get_volatility_history"
params = {
"currency": self.currency,
"start_timestamp": timestamp_ms - 86400000, # 24시간 전
"end_timestamp": timestamp_ms
}
response = self.session.get(endpoint, params=params)
data = response.json()
if data.get("success") and data["result"]:
for record in data["result"]:
surface_data.append({
"instrument": instrument_name,
"strike": inst["strike"],
"expiration": inst["expiration_timestamp"],
"iv": record["iv"],
"delta": record.get("delta"),
"timestamp": record["timestamp"]
})
except Exception as e:
print(f"Error collecting {instrument_name}: {e}")
continue
return surface_data
사용 예시
collector = DeribitOptionsCollector("BTC")
current_surface = collector.collect_volatility_surface()
print(f"수집된 데이터 포인트: {len(current_surface)}")실전 활용: 변동성 곡면 백테스팅 시스템
저는 Deribit 옵션 데이터를 활용하여 다음과 같은 백테스팅 시스템을 구축했습니다:
- IV Rank & IV Percentile 계산: 현재 IV를 historical distribution과 비교
- Skew 모니터링: 25Δ RR(Risk Reversal), 25Δ BF(Butterfly)
- 만기별term structure 분석: 단기 vs 장기 IV 스프레드 추적
- Greeks 포트폴리오 통합: Delta, Gamma, Vega, Theta 집계
import pandas as pd
import numpy as np
class VolatilitySurfaceBacktester:
def __init__(self, data_source):
self.data = data_source
self.df = None
def build_surface_dataframe(self, surface_data):
"""수집된 데이터 DataFrame 변환 및 정제"""
self.df = pd.DataFrame(surface_data)
# 불완전 데이터 필터링
self.df = self.df.dropna(subset=["iv", "strike"])
self.df = self.df[self.df["iv"] > 0] # IV는 반드시 양수
# 파생 지표 계산
self.df["moneyness"] = self.df["strike"] / self.df["underlying_price"]
self.df["log_moneyness"] = np.log(self.df["moneyness"])
self.df["days_to_expiry"] = (
self.df["expiration"] - self.df["timestamp"]
) / (1000 * 86400)
self.df["time_to_expiry"] = self.df["days_to_expiry"] / 365
return self.df
def calculate_iv_rank(self, lookback_days=30):
"""Historical IV Rank 계산"""
# 실제 구현에서는 historical data를 lookback window로 분리
current_iv = self.df[self.df["days_to_expiry"] <= 7]["iv"].mean()
# Historical IV percentile (단순화된 예시)
historical_mean = 0.8 # 실제 구현 시 historical 데이터 사용
historical_std = 0.3
iv_rank = (current_iv - historical_mean) / (4 * historical_std) * 100
return min(100, max(0, iv_rank))
def calculate_skew_metrics(self):
"""Skew 지표 계산: RR, BF"""
df = self.df.copy()
# 25Δ 기준 근처 옵션 필터링
atm_options = df[(df["delta"] >= 0.23) & (df["delta"] <= 0.27)]
otm_call = df[(df["delta"] >= 0.23) & (df["delta"] <= 0.27) &
(df["strike"] > df["underlying_price"])]
otm_put = df[(df["delta"] >= 0.23) & (df["delta"] <= 0.27) &
(df["strike"] < df["underlying_price"])]
if len(otm_call) > 0 and len(otm_put) > 0:
rr_25 = otm_call["iv"].mean() - otm_put["iv"].mean()
bf_25 = (otm_call["iv"].mean() + otm_put["iv"].mean()) / 2 - atm_options["iv"].mean()
return {"rr_25": rr_25, "bf_25": bf_25}
return None
def run_smile_fitting(self, method="svi"):
"""SVI(Stochastic Volatility Inspired) 변동성 스마일 피팅"""
# 각 만기별 IV 스마일 피팅
fitted_surfaces = {}
for expiry, group in self.df.groupby("days_to_expiry"):
strikes = group["strike"].values
ivs = group["iv"].values
# SVI 파라미터 추정 (단순화된卡尔만 필터 구현)
try:
a = ivs.min() * 0.8
b = 0.2
rho = -0.5
m = np.log(group["underlying_price"].iloc[0])
sigma = 0.3
fitted_surfaces[expiry] = {
"a": a, "b": b, "rho": rho, "m": m, "sigma": sigma,
"strikes": strikes, "fitted_iv": ivs
}
except Exception as e:
print(f"Fitting failed for expiry {expiry}: {e}")
return fitted_surfaces
백테스트 실행
backtester = VolatilitySurfaceBacktester(current_surface)
df = backtester.build_surface_dataframe(current_surface)
skew = backtester.calculate_skew_metrics()
fitted = backtester.run_smile_fitting()
print(f"IV Rank: {backtester.calculate_iv_rank():.2f}%")
print(f"25Δ Risk Reversal: {skew['rr_25']:.4f}")WebSocket 실시간 스트리밍 통합
백테스팅 데이터 수집뿐 아니라 실시간 변동성 곡면 monitoring을 위해 WebSocket을 활용할 수 있습니다:
import websocket
import json
import threading
import queue
class DeribitWebSocketClient:
def __init__(self, on_message_callback):
self.ws = None
self.callback = on_message_callback
self.message_queue = queue.Queue()
self.running = False
def connect(self):
"""WebSocket 연결 및 인증"""
self.ws = websocket.WebSocketApp(
"wss://www.deribit.com/ws/api/v2",
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close,
on_open=self._on_open
)
def _on_open(self, ws):
"""연결 시 구독 설정"""
# BTC 옵션 ticker subscription
subscribe_msg = {
"jsonrpc": "2.0",
"method": "private/subscribe",
"params": {
"channels": [
"deribit_price_index.btc_usd",
"book.BTC-28MAR25.100ms"
]
},
"id": 1
}
ws.send(json.dumps(subscribe_msg))
self.running = True
def _on_message(self, ws, message):
"""수신 메시지 처리"""
data = json.loads(message)
if "params" in data:
self.callback(data["params"]["data"])
def _on_error(self, ws, error):
print(f"WebSocket Error: {error}")
def _on_close(self, ws, close_status_code, close_msg):
self.running = False
print(f"Connection closed: {close_status_code}")
def start_background(self):
"""백그라운드 스레드로 WebSocket 실행"""
thread = threading.Thread(target=self._run)
thread.daemon = True
thread.start()
def _run(self):
while self.running:
try:
self.ws.run_forever(ping_interval=30, ping_timeout=10)
except Exception as e:
print(f"Reconnection error: {e}")
time.sleep(5)
def stop(self):
self.running = False
if self.ws:
self.ws.close()
실시간 데이터 처리 콜백
def handle_realtime_data(data):
if "ticks" in data:
for tick in data["ticks"]:
print(f"IV: {tick.get('iv', 'N/A')}, "
f"Best Bid: {tick.get('best_bid_price', 'N/A')}")
사용 예시
ws_client = DeribitWebSocketClient(handle_realtime_data)
ws_client.connect()
ws_client.start_background()자주 발생하는 오류와 해결책
1. Rate Limit 초과 (429 Too Many Requests)
# ❌ 문제: Deribit API rate limit (20 req/sec)에러 발생
requests.get(...) -> {"error": {"code": -32600, "message": "Too many requests"}}
✅ 해결: 요청 간 딜레이 추가 및 指數적 백오프 구현
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session_with_retry():
"""재시도 로직이 포함된 Session 생성"""
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
class RateLimitedClient:
def __init__(self, requests_per_second=15):
self.session = create_session_with_retry()
self.min_interval = 1.0 / requests_per_second
self.last_request = 0
def throttled_get(self, url, params=None):
"""Rate limit 적용된 GET 요청"""
elapsed = time.time() - self.last_request
if elapsed < self.min_interval:
time.sleep(self.min_interval - elapsed)
self.last_request = time.time()
return self.session.get(url, params=params)
사용
client = RateLimitedClient(requests_per_second=15)
response = client.throttled_get(endpoint, params=params)2. 옵션 만기 데이터 누락 (Expiration Gap)
# ❌ 문제: 특정 만기 데이터가 없어 변동성 곡면이 불연속적
✅ 해결: 만기 interpolation 및 만기별 가중 平均값 활용
def fill_expiration_gaps(df, min_expiry=7, max_expiry=180):
"""만기별 결측치 interpolation"""
df = df.copy()
# 목표 만기 그리드 설정
target_expiries = [7, 14, 30, 60, 90, 120, 180]
# 각 만기에 대해 Linear interpolation
available_expiries = sorted(df["days_to_expiry"].unique())
filled_data = []
for target in target_expiries:
if target in available_expiries:
# 정확한 데이터 존재 시 그대로 사용
filled_data.append(df[df["days_to_expiry"] == target])
else:
# 인접 만기 interpolation
lower = max([e for e in available_expiries if e < target], default=None)
upper = min([e for e in available_expiries if e > target], default=None)
if lower and upper:
weight = (target - lower) / (upper - lower)
lower_data = df[df["days_to_expiry"] == lower].copy()
upper_data = df[df["days_to_expiry"] == upper].copy()
for _, row in lower_data.iterrows():
upper_row = upper_data[upper_data["strike"] == row["strike"]]
if len(upper_row) > 0:
interpolated = row.copy()
interpolated["iv"] = row["iv"] * (1 - weight) + upper_row["iv"].values[0] * weight
interpolated["days_to_expiry"] = target
interpolated["interpolated"] = True
filled_data.append(interpolated)
return pd.concat(filled_data, ignore_index=True) if filled_data else df
적용
df_filled = fill_expiration_gaps(df)3. Historical Data Timestamp 변환 오류
# ❌ 문제: Deribit timestamps (milliseconds) vs Python (seconds) 혼동
datetime.fromtimestamp(1735689600000) -> 54253-08-08 오류
✅ 해결: 밀리초 → 초 변환 및 timezone 처리
from datetime import datetime, timezone
def deribit_timestamp_to_datetime(timestamp_ms):
"""Deribit 밀리초 타임스탬프를 Python datetime으로 변환"""
if timestamp_ms is None:
return None
# 밀리초 → 초 변환
timestamp_sec = timestamp_ms / 1000
# UTC 기준 datetime 생성
dt = datetime.fromtimestamp(timestamp_sec, tz=timezone.utc)
return dt
def datetime_to_deribit_timestamp(dt=None):
"""datetime을 Deribit 밀리초 타임스탬프로 변환"""
if dt is None:
dt = datetime.now(timezone.utc)
elif dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return int(dt.timestamp() * 1000)
사용 예시
raw_timestamp = 1735689600000
dt = deribit_timestamp_to_datetime(raw_timestamp)
print(f"Deribit timestamp: {raw_timestamp} -> {dt.isoformat()}")
출력: 2025-01-01T00:00:00+00:00
일별 히스토리 조회
start_dt = datetime(2024, 12, 1, tzinfo=timezone.utc)
end_dt = datetime(2024, 12, 31, tzinfo=timezone.utc)
params = {
"start_timestamp": datetime_to_deribit_timestamp(start_dt),
"end_timestamp": datetime_to_deribit_timestamp(end_dt)
}4. Greeks 부호 및 단위 불일치
# ❌ 문제: Deribit Greeks vs Black-Scholes 표기법 차이
Deribit: delta in terms of spot, theta in USD/day
BS Model: delta in terms of forward, theta in USD/year
✅ 해결: 단위 변환 유틸리티 함수 구현
class GreeksConverter:
@staticmethod
def deribit_to_standard(delta_deribit, option_type="call"):
"""
Deribit delta → 표준 BS delta 변환
Deribit delta는 현물 대비, BS delta는 선물 대비
"""
# Deribit delta = BS_delta * (F / S)
# Simplified: F ≈ S * exp(r*T), but for close expiry approximation
if option_type.lower() == "call":
return delta_deribit # Approximately same for ATM
else:
return delta_deribit + 1 # Put delta conversion
@staticmethod
def theta_daily_to_annual(theta_daily, convention="per_day"):
"""Theta 단위 변환: daily → annual"""
if convention == "per_day":
return theta_daily * 365
return theta_daily
@staticmethod
def normalize_greeks(greeks_dict, spot_price, position_size=1):
"""Greek 값을 포지션 기준으로 정규화"""
normalized = {}
if "delta" in greeks_dict:
# Portfolio delta in BTC
normalized["delta_btc"] = greeks_dict["delta"] * position_size
normalized["delta_usd"] = normalized["delta_btc"] * spot_price
if "gamma" in greeks_dict:
# Gamma per 1% move in spot
normalized["gamma_per_pct"] = greeks_dict["gamma"] * spot_price * 0.01 * position_size
if "vega" in greeks_dict:
# Vega per 1% move in IV
normalized["vega_per_pct_iv"] = greeks_dict["vega"] * 0.01 * position_size
return normalized
사용
greeks = {"delta": 0.45, "gamma": 0.002, "vega": 150, "theta": -80}
spot = 95000
normalized = GreeksConverter.normalize_greeks(greeks, spot)
print(f"Portfolio Delta (USD): ${normalized['delta_usd']:,.2f}")성능 최적화: 대량 데이터 수집 전략
1년 이상의 Historical 데이터를 수집할 때, 순차적 API 호출은 너무 느립니다. 저는 다음 전략을 사용하여 수집 속도를 개선했습니다:
- Batch API 활용:
get_options_summary_by_currency로 대량 데이터 한 번에 조회 - 병렬 처리: Python
asyncio+aiohttp로 동시 요청 - 데이터 캐싱: Redis 또는 파일 기반 locally cached data 활용
- 증분 업데이트: 마지막 수집 시점부터增量更新
import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor
async def fetch_surface_data_async(session, instrument, semaphore):
"""비동기 방식으로 단일 옵션 데이터 fetch"""
async with semaphore:
url = "https://www.deribit.com/api/v2/public/get_order_book"
params = {"instrument_name": instrument}
try:
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
return data.get("result")
else:
return None
except Exception as e:
print(f"Error: {e}")
return None
async def collect_all_surfaces_async(instruments, max_concurrent=10):
"""병렬 수집 메인 함수"""
semaphore = asyncio.Semaphore(max_concurrent)
async with aiohttp.ClientSession() as session:
tasks = [
fetch_surface_data_async(session, inst["instrument_name"], semaphore)
for inst in instruments
]
results = await asyncio.gather(*tasks)
return [r for r in results if r is not None]
실행
instruments = collector.get_instruments()[:100] # 상위 100개
results = asyncio.run(collect_all_surfaces_async(instruments, max_concurrent=10))
print(f"수집 완료: {len(results)} 옵션")실제 적용 사례: AI 기반 변동성 예측 모델
Deribit에서 수집한 변동성 곡면 데이터를 HolySheep AI와 연계하여 ML 기반 예측 모델을 구축했습니다. HolySheep AI의 deepseek-chat 모델을 활용하면 옵션 시장 데이터 기반_sentiment 분석과 자연어 기반 리포트 생성이 가능합니다:
import openai
HolySheep AI API configuration
client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1" # HolySheep 게이트웨이 사용
)
def generate_volatility_report(surface_analysis):
"""변동성 곡면 분석 결과 기반 AI 보고서 생성"""
prompt = f"""
BTC 옵션 시장 변동성 곡면 분석 결과:
현재 ATM 변동성: {surface_analysis['atm_iv']:.2%}
IV Rank (30일): {surface_analysis['iv_rank']:.2%}
25Δ Risk Reversal: {surface_analysis['rr_25']:.4f}
25Δ Butterfly: {surface_analysis['bf_25']:.4f}
단기(7D) IV: {surface_analysis['short_term_iv']:.2%}
장기(90D) IV: {surface_analysis['long_term_iv']:.2%}
위 데이터를 기반으로 다음 내용을 분석해주세요:
1. 현재 시장 mood (공포/탐욕 지표)
2. 변동성 스마일 형태 해석
3. 단기 vs 장기 관점의 거래 기회
4. 리스크 관리 시사점
"""
response = client.chat.completions.create(
model="deepseek-chat", # HolySheep를 통한 DeepSeek 모델 사용
messages=[
{"role": "system", "content": "당신은 숙련된 파생상품 트레이더입니다."},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=1000
)
return response.choices[0].message.content
HolySheep AI를 통한 DeepSeek 모델 활용 (가격: $0.42/MTok)
실제 사용 시 HolySheep API 키로 교체
analysis = {
"atm_iv": 0.65,
"iv_rank": 0.72,
"rr_25": 0.08,
"bf_25": -0.02,
"short_term_iv": 0.72,
"long_term_iv": 0.58
}
report = generate_volatility_report(analysis)
print(report)Deribit API vs 대체 거래소 비교
| 항목 | Deribit | Binance Options | OKX Options | Bybit Options |
|---|---|---|---|---|
| 지원 통화 | BTC, ETH | BTC, ETH | BTC, ETH | BTC, ETH |
| Historical Data | 제한적 (30일) | 제한적 | 제한적 | 제한적 |
| WebSocket 지원 | ✅ 우수 | ✅ 양호 | ✅ 양호 | ✅ 양호 |
| Rate Limit | 20 req/sec | 10 req/sec | 15 req/sec | 10 req/sec |
| API 안정성 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐ |
| Greeks 제공 | ✅ 실시간 | ❌ 미지원 | ✅ 제한적 | ❌ 미지원 |
결론 및 다음 단계
Deribit 옵션 히스토리 데이터 API를 활용하면 BTC·ETH 변동성 곡면을 효과적으로 구축하고 백테스팅할 수 있습니다. 주요 포인트는 다음과 같습니다:
- Rate Limit 관리: 20 req/sec 제한을 준수하며 재시도 로직 구현
- 데이터 정제: 만기별 interpolation과 timestamp 변환 정확히 처리
- Greeks 이해: Deribit 표기법과 표준 BS 모델의 차이 숙지
- 병렬 수집: 대량 데이터는 asyncio 기반 비동기 수집으로 효율화
- AI 연계: HolySheep AI를 통한 자연어 분석 및 예측 모델 통합
변동성 곡면 분석은 옵션 거래의 핵심입니다. 위 가이드를 바탕으로 자신의 거래 전략에 맞는 백테스팅 시스템을 구축해 보시기 바랍니다.
👉 HolySheep AI 가입하고 무료 크레딧 받기