암호화폐 옵션 거래에서 Greeks와 내재변동성(IV)은 헤지 및 전략 수립의 핵심입니다. Deribit는 업계 최고 수준의 옵션 데이터를 제공하지만, 이 데이터를 대규모로 수집하고 백테스트에 활용하려면 여러 기술적 장애물을 극복해야 합니다. 저는 지난 2년간 Deribit 옵션 데이터를 수집하며 ConnectionError: timeout, 401 Unauthorized, Rate Limit 초과 등 수십 가지 오류를 경험했습니다. 이 튜토리얼에서는 HolySheep AI를 활용해 Deribit 옵션 데이터를 안정적으로 수집하는 전체 파이프라인을 설명드리겠습니다.

Deribit 옵션 데이터 수집의 현실적挑战

Deribit는 매초 수천 개의 옵션 계약(Greeks, IV, 호가창, 체결)을 생성합니다. 단일 만기에는 수십 개의 스트라이크가 존재하며, 만기별로 데이터가 쌓이면 볼륨은 기하급수적으로 증가합니다. 저는 BTC 옵션만으로도 하루 50GB 이상의 원시 데이터를 처리한 경험이 있습니다. 핵심 문제들은 다음과 같습니다:

Deribit API 구조와 옵션 데이터 타입

Deribit 옵션 관련 엔드포인트

Deribit는 REST API와 WebSocket 두 가지 인터페이스를 제공합니다. 역사 데이터 수집에는 REST API가 적합하고, 실시간 스트리밍에는 WebSocket이 필수입니다.

Deribit API 주요 엔드포인트

옵션 데이터 수집 아키텍처

저가 설계한 파이프라인은 세 단계로 구성됩니다. 첫 번째는 Deribit API에서 원시 데이터를 수집하는 컬렉터, 두 번째는 데이터 정규화 및 변환하는 프로세서, 세 번째는 TimescaleDB에 저장하는 스토리지입니다. HolySheep AI의 다중 모델 통합 기능을 활용하면 수집 과정에서 AI 기반 이상치 탐지 및 데이터 품질 검증도 가능합니다.

실전 코드: Python 기반 Deribit 옵션 데이터 수집기

1. 기본 설정 및 의존성

#!/usr/bin/env python3
"""
Deribit 옵션체인 역사 데이터 수집기
HolySheep AI API를 활용한 데이터 품질 검증 포함
"""

import requests
import time
import json
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import asyncio
import aiohttp
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor
import numpy as np


Deribit API 기본 설정

DERIBIT_BASE_URL = "https://www.deribit.com/api/v2"
INSTRUMENT_TYPE = "option"  # futures, spot, option


HolySheep AI 설정 - GPT-4.1으로 데이터 품질 검증

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"

@dataclass
class OptionContract:
    """옵션 계약 데이터 구조"""
    instrument_name: str
    timestamp: int
    underlying_price: float
    strike: float
    maturity: str
    option_type: str  # call or put
    iv_bid: float
    iv_ask: float
    delta: float
    gamma: float
    theta: float
    vega: float
    rho: float
    best_bid: float
    best_ask: float
    volume: int
    open_interest: int

class DeribitCollector:
    """Deribit API 데이터 수집기"""
    
    def __init__(self, client_id: str = None, client_secret: str = None):
        self.client_id = client_id
        self.client_secret = client_secret
        self.access_token = None
        self.token_expiry = 0
        self.session = requests.Session()
        self.session.headers.update({
            "Content-Type": "application/json"
        })
    
    def authenticate(self) -> str:
        """Deribit API 인증 - 401 Unauthorized 해결"""
        auth_url = f"{DERIBIT_BASE_URL}/public/auth"
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }
        
        try:
            response = self.session.post(auth_url, json=payload, timeout=10)
            response.raise_for_status()
            data = response.json()
            
            if "result" not in data:
                raise ValueError(f"Authentication failed: {data}")
            
            self.access_token = data["result"]["access_token"]
            self.token_expiry = time.time() + data["result"]["expires_in"]
            print(f"[AUTH] Access token acquired, expires in {data['result']['expires_in']}s")
            return self.access_token
            
        except requests.exceptions.ConnectionError as e:
            raise ConnectionError(f"Failed to connect to Deribit: {e}")
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise ConnectionError("401 Unauthorized: Invalid credentials or token expired")
            raise
    
    def get_option_instruments(self, currency: str = "BTC", kind: str = "option") -> List[Dict]:
        """특정 통화의 옵션 계약 목록 조회"""
        url = f"{DERIBIT_BASE_URL}/public/get_instruments"
        params = {
            "currency": currency,
            "kind": kind,
            "expired": "false"
        }
        
        response = self.session.get(url, params=params, timeout=30)
        response.raise_for_status()
        data = response.json()
        
        if "result" in data:
            return data["result"]
        return []
    
    def get_order_book(self, instrument_name: str, depth: int = 10) -> Dict:
        """호가창 데이터 조회 - IV와 Greeks 포함"""
        url = f"{DERIBIT_BASE_URL}/public/get_order_book"
        params = {
            "instrument_name": instrument_name,
            "depth": depth
        }
        
        try:
            response = self.session.get(url, params=params, timeout=15)
            response.raise_for_status()
            return response.json().get("result", {})
        except requests.exceptions.Timeout:
            print(f"[TIMEOUT] Order book request timed out for {instrument_name}")
            return {}
        except requests.exceptions.HTTPError as e:
            if response.status_code == 429:
                wait_time = int(response.headers.get("Retry-After", 60))
                print(f"[RATE LIMIT] Waiting {wait_time}s before retry...")
                time.sleep(wait_time)
                return self.get_order_book(instrument_name, depth)
            raise
    
    def get_last_trades(self, instrument_name: str, count: int = 100) -> List[Dict]:
        """최근 체결 내역 조회"""
        url = f"{DERIBIT_BASE_URL}/public/get_last_trades_by_instrument"
        params = {
            "instrument_name": instrument_name,
            "count": count
        }
        
        response = self.session.get(url, params=params, timeout=20)
        response.raise_for_status()
        data = response.json()
        
        return data.get("result", {}).get("trades", [])
    
    def get_volatility_surface(self, currency: str = "BTC") -> Dict:
        """변동성 곡면 데이터 조회"""
        url = f"{DERIBIT_BASE_URL}/public/get_volatility_surface_data"
        params = {"currency": currency}
        
        response = self.session.get(url, params=params, timeout=30)
        response.raise_for_status()
        return response.json().get("result", {})
    
    def get_historical_volatility(self, currency: str, period: int = 30) -> Dict:
        """역사적 변동성 조회"""
        url = f"{DERIBIT_BASE_URL}/public/get_historical_volatility"
        params = {"currency": currency, "period": period}
        
        response = self.session.get(url, params=params, timeout=20)
        response.raise_for_status()
        return response.json().get("result", {})



HolySheep AI를 활용한 데이터 품질 검증

class HolySheepDataValidator:
    """HolySheep AI API를 이용한 옵션 데이터 품질 검증"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = HOLYSHEEP_BASE_URL
    
    def validate_option_data(self, option_data: Dict) -> Dict:
        """GPT-4.1으로 옵션 데이터 이상치 탐지"""
        prompt = f"""
        다음 Deribit 옵션 데이터를 검증하고 잠재적 이상치를 탐지하세요:
        
        Instrument: {option_data.get('instrument_name')}
        Underlying Price: ${option_data.get('underlying_price', 0):,.2f}
        Strike: ${option_data.get('strike', 0):,.2f}
        IV Bid/Ask: {option_data.get('iv_bid', 0):.4f} / {option_data.get('iv_ask', 0):.4f}
        Delta: {option_data.get('delta', 0):.4f}
        Gamma: {option_data.get('gamma', 0):.6f}
        Theta: {option_data.get('theta', 0):.4f}
        Vega: {option_data.get('vega', 0):.4f}
        
        분석:
        1. IV가 합리적 범위 내에 있는지 (10%~300%)
        2. Greeks가 Black-Scholes 이론과 일치하는지
        3. Bid-Ask 스프레드가 정상인지
        """
        
        payload = {
            "model": "gpt-4.1",
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.3,
            "max_tokens": 500
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            result = response.json()
            return {
                "status": "validated",
                "analysis": result["choices"][0]["message"]["content"]
            }
        except requests.exceptions.Timeout:
            return {"status": "timeout", "analysis": "Validation timed out"}
        except Exception as e:
            return {"status": "error", "analysis": str(e)}



메인 수집 함수

async def collect_options_data_async(
    instruments: List[str],
    collector: DeribitCollector,
    validator: HolySheepDataValidator,
    batch_size: int = 10
):
    """비동기 옵션 데이터 수집 + 품질 검증"""
    
    all_data = []
    
    for i in range(0, len(instruments), batch_size):
        batch = instruments[i:i+batch_size]
        
        for instrument in batch:
            # 호가창 데이터 수집
            order_book = collector.get_order_book(instrument)
            
            if order_book:
                timestamp = int(time.time() * 1000)
                
                # Greeks 및 IV 추출
                iv = order_book.get("greeks", {})
                bids = order_book.get("bids", [])
                asks = order_book.get("asks", [])
                
                if bids and asks:
                    record = {
                        "instrument_name": instrument,
                        "timestamp": timestamp,
                        "underlying_price": order_book.get("underlying_price", 0),
                        "iv_bid": bids[0].get("iv", 0) if bids else 0,
                        "iv_ask": asks[0].get("iv", 0) if asks else 0,
                        "delta": iv.get("delta", 0),
                        "gamma": iv.get("gamma", 0),
                        "theta": iv.get("theta", 0),
                        "vega": iv.get("vega", 0),
                        "rho": iv.get("rho", 0),
                        "best_bid": bids[0].get("price", 0),
                        "best_ask": asks[0].get("price", 0),
                        "bid_size": bids[0].get("amount", 0),
                        "ask_size": asks[0].get("amount", 0),
                    }
                    
                    # HolySheep AI로 데이터 검증
                    validation = validator.validate_option_data(record)
                    record["validation_status"] = validation["status"]
                    
                    all_data.append(record)
                    print(f"[COLLECTED] {instrument} @ {datetime.now()}")
            
            # Rate Limit 방지 딜레이
            await asyncio.sleep(0.1)
        
        # 배치 간 딜레이
        await asyncio.sleep(1)
    
    return all_data



실행 예제

if __name__ == "__main__":
    # Deribit API 설정 (본인 키로 교체)
    COLLECTOR = DeribitCollector(
        client_id="YOUR_DERIBIT_CLIENT_ID",
        client_secret="YOUR_DERIBIT_CLIENT_SECRET"
    )
    
    # HolySheep AI 검증기
    VALIDATOR = HolySheepDataValidator(HOLYSHEEP_API_KEY)
    
    # 인증
    try:
        COLLECTOR.authenticate()
    except ConnectionError as e:
        print(f"[ERROR] Authentication failed: {e}")
        exit(1)
    
    # BTC 옵션 계약 목록
    instruments = COLLECTOR.get_option_instruments("BTC", "option")
    instrument_names = [inst["instrument_name"] for inst in instruments]
    
    print(f"[INFO] Found {len(instrument_names)} BTC options")
    
    # 데이터 수집 실행
    asyncio.run(collect_options_data_async(
        instrument_names[:50],  # 처음 50개만 수집
        COLLECTOR,
        VALIDATOR
    ))

2. 데이터 저장 파이프라인 (TimescaleDB + Parquet)

#!/usr/bin/env python3
"""
Deribit 옵션 데이터 저장 파이프라인
TimescaleDB에 시계열 저장 + Parquet로 분석용 파일 생성
"""

import psycopg2
from psycopg2.extras import execute_values
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from datetime import datetime
import boto3
import os
from typing import List, Dict


TimescaleDB 설정

TIMESCALEDB_HOST = "your-timescaledb-host.amazonaws.com"
TIMESCALEDB_PORT = 5432
TIMESCALEDB_DB = "deribit_options"
TIMESCALEDB_USER = "admin"
TIMESCALEDB_PASSWORD = "your-password"


S3 설정 (Parquet 저장용)

S3_BUCKET = "deribit-options-data"
AWS_ACCESS_KEY = os.environ.get("AWS_ACCESS_KEY_ID")
AWS_SECRET_KEY = os.environ.get("AWS_SECRET_ACCESS_KEY")


def create_timescaledb_tables():
    """TimescaleDB 하이퍼테이블 생성"""
    
    conn = psycopg2.connect(
        host=TIMESCALEDB_HOST,
        port=TIMESCALEDB_PORT,
        dbname=TIMESCALEDB_DB,
        user=TIMESCALEDB_USER,
        password=TIMESCALEDB_PASSWORD
    )
    cursor = conn.cursor()
    
    # 옵션 호가창 테이블
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS option_orderbook (
            time TIMESTAMPTZ NOT NULL,
            instrument_name TEXT NOT NULL,
            underlying_price DOUBLE PRECISION,
            strike DOUBLE PRECISION,
            expiry TIMESTAMPTZ,
            option_type TEXT,
            iv_bid DOUBLE PRECISION,
            iv_ask DOUBLE PRECISION,
            iv_mid DOUBLE PRECISION GENERATED ALWAYS AS ((iv_bid + iv_ask) / 2) STORED,
            best_bid DOUBLE PRECISION,
            best_ask DOUBLE PRECISION,
            bid_size DOUBLE PRECISION,
            ask_size DOUBLE PRECISION,
            spread DOUBLE PRECISION GENERATED ALWAYS AS (best_ask - best_bid) STORED,
            spread_bps DOUBLE PRECISION GENERATED ALWAYS AS (
                CASE WHEN best_bid > 0 
                THEN ((best_ask - best_bid) / best_bid * 10000) 
                ELSE 0 END) STORED,
            validation_status TEXT,
            INDEX idx_instrument (instrument_name),
            INDEX idx_expiry (expiry)
        );
    """)
    
    # 옵션 Greeks 테이블
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS option_greeks (
            time TIMESTAMPTZ NOT NULL,
            instrument_name TEXT NOT NULL,
            delta DOUBLE PRECISION,
            gamma DOUBLE PRECISION,
            theta DOUBLE PRECISION,
            vega DOUBLE PRECISION,
            rho DOUBLE PRECISION,
            ddelta_dspot DOUBLE PRECISION GENERATED ALWAYS AS (delta * gamma) STORED,
            ddelta_dvol DOUBLE PRECISION GENERATED ALWAYS AS (delta * vega) STORED,
            INDEX idx_instrument (instrument_name)
        );
    """)
    
    # 체결 데이터 테이블
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS option_trades (
            time TIMESTAMPTZ NOT NULL,
            trade_id BIGINT NOT NULL,
            instrument_name TEXT NOT NULL,
            direction TEXT,
            price DOUBLE PRECISION,
            amount DOUBLE PRECISION,
            iv_at_trade DOUBLE PRECISION,
            tick_rule INTEGER,
            INDEX idx_instrument (instrument_name),
            INDEX idx_time (time DESC)
        );
    """)
    
    # 하이퍼테이블 변환
    cursor.execute("""
        SELECT create_hypertable('option_orderbook', 'time', 
            if_not_exists => TRUE, migrate_data => TRUE);
    """)
    cursor.execute("""
        SELECT create_hypertable('option_greeks', 'time',
            if_not_exists => TRUE, migrate_data => TRUE);
    """)
    cursor.execute("""
        SELECT create_hypertable('option_trades', 'time',
            if_not_exists => TRUE, migrate_data => TRUE);
    """)
    
    # 연속 쿼리 정책 (1분 간격으로 압축)
    cursor.execute("""
        SELECT add_continuous_aggregate_policy('option_orderbook_1m',
            start_offset => INTERVAL '3 hours',
            end_offset => INTERVAL '1 hour',
            schedule_interval => INTERVAL '1 minute');
    """)
    
    conn.commit()
    cursor.close()
    conn.close()
    print("[DB] TimescaleDB tables created successfully")


def save_to_timescaledb(records: List[Dict], table_name: str = "option_orderbook"):
    """TimescaleDB에 데이터 일괄 저장"""
    
    conn = psycopg2.connect(
        host=TIMESCALEDB_HOST,
        port=TIMESCALEDB_PORT,
        dbname=TIMESCALEDB_DB,
        user=TIMESCALEDB_USER,
        password=TIMESCALEDB_PASSWORD
    )
    
    if not records:
        print(f"[DB] No records to save to {table_name}")
        return
    
    if table_name == "option_orderbook":
        columns = [
            "time", "instrument_name", "underlying_price", "strike", "expiry",
            "option_type", "iv_bid", "iv_ask", "best_bid", "best_ask",
            "bid_size", "ask_size", "validation_status"
        ]
    elif table_name == "option_greeks":
        columns = [
            "time", "instrument_name", "delta", "gamma", "theta", "vega", "rho"
        ]
    elif table_name == "option_trades":
        columns = [
            "time", "trade_id", "instrument_name", "direction", "price",
            "amount", "iv_at_trade", "tick_rule"
        ]
    
    values = [[r.get(col) for col in columns] for r in records]
    
    insert_query = f"""
        INSERT INTO {table_name} ({', '.join(columns)})
        VALUES %s
        ON CONFLICT DO NOTHING
    """
    
    execute_values(
        conn.cursor(),
        insert_query,
        values,
        template=None,
        page_size=1000
    )
    
    conn.commit()
    print(f"[DB] Saved {len(records)} records to {table_name}")
    
    conn.close()


def save_to_parquet(records: List[Dict], filename: str):
    """Parquet 파일로 저장 (분석용)"""
    
    if not records:
        print("[PARQUET] No records to save")
        return
    
    df = pd.DataFrame(records)
    
    # 타임스탬프 변환
    if "time" in df.columns:
        df["datetime"] = pd.to_datetime(df["time"], unit="ms")
        df = df.drop(columns=["time"])
    
    # Parquet 스키마 정의
    schema = pa.Table.from_pandas(df).schema
    
    table = pa.Table.from_pandas(df, schema=schema)
    
    # 로컬 저장
    local_path = f"/tmp/{filename}"
    pq.write_table(table, local_path, compression="snappy")
    print(f"[PARQUET] Saved to {local_path}")
    
    # S3 업로드
    s3_client = boto3.client(
        "s3",
        aws_access_key_id=AWS_ACCESS_KEY,
        aws_secret_access_key=AWS_SECRET_KEY,
        region_name="us-east-1"
    )
    
    s3_key = f"raw/{datetime.now().strftime('%Y/%m/%d')}/{filename}"
    s3_client.upload_file(local_path, S3_BUCKET, s3_key)
    print(f"[S3] Uploaded to s3://{S3_BUCKET}/{s3_key}")


def calculate_iv_surface(df: pd.DataFrame) -> pd.DataFrame:
    """IV 곡면 계산 (만기별, 스트라이크별 IV 매트릭스)"""
    
    # 시간대별 그룹화
    df["hour"] = df["datetime"].dt.floor("H")
    
    # 피벗 테이블로 IV 매트릭스 생성
    iv_surface = df.pivot_table(
        values="iv_mid",
        index="strike",
        columns="hour",
        aggfunc="mean"
    )
    
    return iv_surface


def get_volatility_strategy_data(
    start_time: datetime,
    end_time: datetime,
    underlying: str = "BTC"
) -> pd.DataFrame:
    """변동성 전략 백테스트용 데이터 조회"""
    
    conn = psycopg2.connect(
        host=TIMESCALEDB_HOST,
        port=TIMESCALEDB_PORT,
        dbname=TIMESCALEDB_DB,
        user=TIMESCALEDB_USER,
        password=TIMESCALEDB_PASSWORD
    )
    
    query = """
        SELECT 
            time_bucket('5 minutes', ob.time) AS bucket,
            ob.instrument_name,
            ob.underlying_price,
            ob.strike,
            ob.iv_mid,
            ob.spread_bps,
            g.delta,
            g.gamma,
            g.theta,
            g.vega,
            COUNT(*) AS sample_count
        FROM option_orderbook ob
        LEFT JOIN option_greeks g ON 
            ob.time = g.time AND 
            ob.instrument_name = g.instrument_name
        WHERE ob.time BETWEEN %s AND %s
            AND ob.instrument_name LIKE %s
        GROUP BY bucket, ob.instrument_name, ob.underlying_price, 
                 ob.strike, ob.iv_mid, ob.spread_bps,
                 g.delta, g.gamma, g.theta, g.vega
        ORDER BY bucket
    """
    
    df = pd.read_sql_query(
        query,
        conn,
        params=[start_time, end_time, f"{underlying}-%"]
    )
    
    conn.close()
    
    return df



실행 예제

if __name__ == "__main__":
    # 테이블 생성
    create_timescaledb_tables()
    
    # 샘플 데이터 수집 후 저장
    sample_data = [
        {
            "time": datetime.now(),
            "instrument_name": "BTC-27DEC24-95000-C",
            "underlying_price": 96500.0,
            "strike": 95000.0,
            "expiry": datetime(2024, 12, 27),
            "option_type": "call",
            "iv_bid": 0.45,
            "iv_ask": 0.48,
            "best_bid": 4500.0,
            "best_ask": 4800.0,
            "bid_size": 2.5,
            "ask_size": 2.3,
            "validation_status": "validated"
        }
    ]
    
    save_to_timescaledb(sample_data, "option_orderbook")
    save_to_parquet(sample_data, "options_20241220.parquet")
    
    # 백테스트용 데이터 조회
    backtest_df = get_volatility_strategy_data(
        start_time=datetime(2024, 12, 1),
        end_time=datetime(2024, 12, 20),
        underlying="BTC"
    )
    
    print(f"[BACKTEST] Loaded {len(backtest_df)} rows for analysis")
    print(backtest_df.head())

3. Deribit WebSocket 실시간 스트리밍

#!/usr/bin/env python3
"""
Deribit WebSocket 실시간 옵션 데이터 스트리밍
holySheep AI를 통한 실시간 이상치 알림
"""

import websocket
import json
import threading
import time
from datetime import datetime
from typing import Callable, Dict, List
import sqlite3
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


holySheep AI SMS/Webhook 알림 설정

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"

class DeribitWebSocketClient:
    """Deribit WebSocket 실시간 데이터 클라이언트"""
    
    def __init__(self, on_message_callback: Callable = None):
        self.ws = None
        self.auth_token = None
        self.is_running = False
        self.on_message = on_message_callback
        self.reconnect_delay = 1
        self.max_reconnect_delay = 60
        self.subscriptions = []
        self.message_count = 0
        self.error_count = 0
        
    def authenticate(self, client_id: str, client_secret: str) -> bool:
        """WebSocket 인증"""
        auth_params = {
            "method": "public/auth",
            "params": {
                "grant_type": "client_credentials",
                "client_id": client_id,
                "client_secret": client_secret
            },
            "id": 1
        }
        
        self.ws.send(json.dumps(auth_params))
        response = json.loads(self.ws.recv())
        
        if "result" in response and response["result"]:
            self.auth_token = response["result"]["access_token"]
            logger.info("[WS] Authentication successful")
            return True
        
        logger.error(f"[WS] Authentication failed: {response}")
        return False
    
    def subscribe(self, channel: str):
        """채널 구독"""
        subscribe_params = {
            "method": "private/subscribe",
            "params": {
                "channels": [channel]
            },
            "id": 100
        }
        
        self.ws.send(json.dumps(subscribe_params))
        response = json.loads(self.ws.recv())
        
        if "result" in response:
            self.subscriptions.append(channel)
            logger.info(f"[WS] Subscribed to {channel}")
    
    def subscribe_ticker(self, instrument: str):
        """특정 계약 티커 구독"""
        self.subscribe(f"ticker.{instrument}.100ms")
    
    def subscribe_trades(self, instrument: str):
        """체결 데이터 구독"""
        self.subscribe(f"trades.{instrument}.100ms")
    
    def subscribe_book(self, instrument: str):
        """호가창 데이터 구독"""
        self.subscribe(f"book.{instrument}.none.10.100ms")
    
    def subscribe_deribit_price_index(self, index_name: str = "btc_usd"):
        """Deribit 가격 지수 구독"""
        self.subscribe(f"deribit_price_index.{index_name}")
    
    def on_error(self, ws, error):
        """WebSocket 오류 핸들러"""
        self.error_count += 1
        logger.error(f"[WS ERROR] {error}")
        
        if "Connection refused" in str(error):
            logger.warning("[WS] Connection refused, retrying...")
            time.sleep(5)
            self.reconnect()
        elif "timeout" in str(error).lower():
            logger.warning("[WS] Timeout error, reconnecting...")
            time.sleep(10)
            self.reconnect()
    
    def on_close(self, ws, close_status_code, close_msg):
        """연결 종료 핸들러"""
        logger.warning(f"[WS] Connection closed: {close_status_code} - {close_msg}")
        self.is_running = False
        
        if self.error_count < 10:
            self.reconnect()
    
    def on_open(self, ws):
        """연결 열림 핸들러"""
        logger.info("[WS] Connection opened")
        self.is_running = True
        self.error_count = 0
        self.reconnect_delay = 1
        
        # 필요한 채널 구독
        # BTC 옵션
        for expiry in ["27DEC24", "29MAR25", "27JUN25"]:
            for strike in ["90000", "95000", "100000", "105000", "110000"]:
                for opt_type in ["C", "P"]:
                    instrument = f"BTC-{expiry}-{strike}-{opt_type}"
                    try:
                        self.subscribe_ticker(instrument)
                        self.subscribe_book(instrument)
                    except:
                        pass
        
        logger.info(f"[WS] Subscribed to {len(self.subscriptions)} channels")
    
    def process_message(self, ws, message):
        """메시지 처리"""
        self.message_count += 1
        
        try:
            data = json.loads(message)
            
            # 파라미터 없는 응답 처리
            if "params" not in data and "result" not in data:
                return
            
            if "params" in data:
                channel_data = data["params"]["data"]
                channel_name = data["params"]["channel"]
                
                # 티커 데이터 처리
                if "ticker" in channel_name:
                    self.process_ticker(channel_data)
                # 호가창 데이터 처리
                elif "book" in channel_name:
                    self.process_orderbook(channel_data)
                # 체결 데이터 처리
                elif "trades" in channel_name:
                    self.process_trades(channel_data)
                
                # 콜백 함수가 있으면 호출
                if self.on_message:
                    self.on_message(channel_name, channel_data)
                    
        except json.JSONDecodeError as e:
            logger.error(f"[JSON ERROR] {e}")
        except KeyError as e:
            logger.error(f"[KEY ERROR] Missing key: {e}")
        except Exception as e:
            logger.error(f"[PROCESS ERROR] {e}")
    
    def process_ticker(self, data: Dict):
        """티커 데이터 처리 - Greeks, IV 포함"""
        record = {
            "timestamp": datetime.now(),
            "instrument_name": data.get("instrument_name"),
            "underlying_price": data.get("underlying_price"),
            "last_price": data.get("last"),
            "iv_index": data.get("greeks", {}).get("iv_index", 0),
            "delta": data.get("greeks", {}).get("delta", 0),
            "gamma": data.get("greeks", {}).get("gamma", 0),
            "theta": data.get("greeks", {}).get("theta", 0),
            "vega": data.get("greeks", {}).get("vega", 0),
            "rho": data.get("greeks", {}).get("rho", 0),
            "mark_iv": data.get("mark_iv", 0),
            "best_bid_iv": data.get("best_bid_iv", 0),
            "best_ask_iv": data.get("best_ask_iv", 0),
            "open_interest": data.get("open_interest", 0),
            "volume": data.get("volume", 0),
            "best_bid_price": data.get("best_bid_price", 0),
            "best_ask_price": data.get("best_ask_price", 0),
            "best_bid_amount": data.get("best_bid_amount", 0),
            "best_ask_amount": data.get("best_ask_amount", 0),
        }
        
        # IV 이상치 탐지 (>200% 또는 <10%)
        if record["mark_iv"] > 2.0 or record["mark_iv"] < 0.1:
            self.send_alert(record)
        
        return record
    
    def process_orderbook(self, data: Dict):
        """호가창 데이터 처리"""
        bids = data.get("bids", [])
        asks = data.get("asks", [])
        
        if bids and asks:
            return {
                "timestamp": datetime.now(),
                "instrument_name": data.get("instrument_name"),
                "bids": [(b["price"], b["amount"], b.get("iv", 0)) for b in bids],
                "asks": [(a["price"], a["amount"], a.get("iv", 0)) for a in asks],
                "mid_price": (bids

관련 리소스
관련 문서