ocrytoquantトレーダ입니다. 2024년 3월, 저는 비트코인 변동성 스퀴즈(Volatility Squeeze) 기회를 포착하기 위해 옵션체인 데이터를 실시간으로 분석하는 봇을 개발했습니다. Deribit V2 WebSocket API를 활용하면 밀리초 단위로 옵션 데이터를 수신할 수 있으며, 이를 통해 변동성 프리미엄 왜곡과 같은 수익 구조를 자동으로 탐지할 수 있습니다. 이 튜토리얼에서는 Python 기반 Deribit V2 API 연결부터 옵션체인 데이터 구독, 그리고 실무에서 자주 발생하는 오류 해결까지 상세히 다룹니다.
Deribit V2 API 개요와 왜 중요한가
Deribit는 전 세계 최대 비트코인·이더리움 옵션 거래소로, 일평균 거래량이 10억 달러를 초과합니다. V2 API는 WebSocket 기반 실시간 데이터 스트리밍을 지원하며, 특히 옵션 Greeks(델타, 감마, 세타, 베가), 내재변동성(IV), 미결제약정(Open Interest) 등을 실시간으로 제공합니다.
Deribit V2 API의 핵심 특징:
- WebSocket 연결: 단일 연결에서 다중 채널 구독 가능
- 테넌시 티어: 퍼블릭(인증 불필요)과 프라이빗(인증 필요) 채널 구분
- 북마크 기능: 이전 데이터 조회 위한 커서 기반 페이지네이션
- Rate Limit: 퍼블릭 60RPM, 프라이빗 120RPM
Deribit V2 WebSocket 연결 구현
Deribit는 WebSocket 프로토콜을 사용하며, 연결 주소는 wss://test.deribit.com/ws/api/v2(테스트넷)과 wss://www.deribit.com/ws/api/v2(메인넷)입니다. 아래는 Python의 websocket-client 라이브러리를 활용한 기본 연결 코드입니다.
# deribit_connection.py
import websocket
import json
import threading
import time
from typing import Callable, Optional
class DeribitWebSocketClient:
"""Deribit V2 WebSocket API 클라이언트"""
def __init__(self, testnet: bool = True):
# 메인넷: wss://www.deribit.com/ws/api/v2
# 테스트넷: wss://test.deribit.com/ws/api/v2
self.base_url = "wss://test.deribit.com/ws/api/v2" if testnet else "wss://www.deribit.com/ws/api/v2"
self.ws: Optional[websocket.WebSocketApp] = None
self.access_token: Optional[str] = None
self.refresh_token: Optional[str] = None
self._running = False
self._subscriptions = {}
self._callbacks = {}
self._lock = threading.Lock()
def connect(self):
"""WebSocket 연결 수립"""
self.ws = websocket.WebSocketApp(
self.base_url,
on_message=self._on_message,
on_error=self._on_error,
on_open=self._on_open,
on_close=self._on_close
)
self._running = True
# 별도 스레드에서 WebSocket 실행
self._thread = threading.Thread(target=self.ws.run_forever, daemon=True)
self._thread.start()
print(f"[INFO] Deribit WebSocket 연결 시도: {self.base_url}")
def _on_open(self, ws):
print("[INFO] WebSocket 연결 성공")
# 연결 수립 후 필요시 인증 수행
# self.authenticate("YOUR_CLIENT_ID", "YOUR_CLIENT_SECRET")
def _on_message(self, ws, message):
"""수신된 메시지 처리"""
try:
data = json.loads(message)
# ping/pong 처리
if data.get("method") == "heartbeat":
self.send({"action": "poke"})
return
# 구독 콜백 처리
if "params" in data and "channel" in data["params"]:
channel = data["params"]["channel"]
if channel in self._callbacks:
self._callbacks[channel](data["params"]["data"])
return
# RPC 응답 처리
if "result" in data or "error" in data:
request_id = data.get("id")
if request_id and request_id in self._subscriptions:
self._subscriptions[request_id].set()
except json.JSONDecodeError as e:
print(f"[ERROR] JSON 파싱 실패: {e}")
except Exception as e:
print(f"[ERROR] 메시지 처리 실패: {e}")
def _on_error(self, ws, error):
print(f"[ERROR] WebSocket 오류: {error}")
def _on_close(self, ws, close_status_code, close_msg):
print(f"[INFO] WebSocket 종료: {close_status_code} - {close_msg}")
self._running = False
def send(self, payload: dict) -> bool:
"""메시지 전송"""
if self.ws and self._running:
self.ws.send(json.dumps(payload))
return True
return False
def close(self):
"""연결 종료"""
self._running = False
if self.ws:
self.ws.close()
print("[INFO] WebSocket 연결 해제")
사용 예시
if __name__ == "__main__":
client = DeribitWebSocketClient(testnet=True)
client.connect()
time.sleep(5) # 연결 대기
client.close()옵션체인 데이터 구독实战
옵션체인 데이터는 book.{instrument_name}.{depth} 채널로 주문서 데이터를, ticker.{instrument_name} 채널로 실시간 티커 정보를 구독합니다. Deribit의 옵션 상품 명명 규칙은 {base_currency}-{expiration_date}-{strike_price}-{option_type 형태입니다.
#