암호화폐 옵션 거래에서 변동성 곡면(Volatility Surface)은 만기별·행사가별 내재변동성을 시각화하는 핵심 도구입니다. 이 튜토리얼에서는 Deribit API를 사용하여 실시간 변동성 곡면을 구축하는 방법을 단계별로 설명드리겠습니다. 특히 AI 기반 분석을 통합하여 전통적인 정량 모델의 한계를 극복하는 방법을 다룹니다.
시작하기 전에: 흔한 연결 오류 scenario
Deribit API를 처음 호출할 때 아래와 같은 오류를 경험하는 개발자가 많습니다:
ConnectionError: HTTPSConnectionPool(host='www.deribit.com', port=443):
Max retries exceeded with url: /api/v2/public/get_instruments
Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x10a2b3d90>:
Failed to establish a new connection: [Errno 110] Connection timed out')
이 오류는 주로 API 엔드포인트 엔지니어링 문제, 네트워크 라우팅 지연, 또는 Rate Limit 초과로 발생합니다. 이 가이드에서 이러한 문제를 포함한 10가지 이상의 에러 시나리오와 해결책을 제공합니다.
Deribit API 기본 설정
1. 필수 라이브러리 설치
# 필수 패키지 설치
pip install requests pandas numpy matplotlib scipy
pip install websockets asyncio pandas-datareader
코드
import requests
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from scipy.interpolate import griddata, RBFInterpolator
import json
import time
from typing import Dict, List, Optional
class DeribitVolatilitySurface:
"""Deribit API를 사용하여 옵션 변동성 곡면 구축 클래스"""
BASE_URL = "https://www.deribit.com/api/v2"
def __init__(self, use_proxy: bool = False, proxy_url: str = None):
self.session = requests.Session()
self.session.headers.update({
'Content-Type': 'application/json',
'User-Agent': 'VolatilitySurface/1.0'
})
if use_proxy and proxy_url:
self.session.proxies = {
'http': proxy_url,
'https': proxy_url
}
def get_instruments(self, currency: str = "BTC") -> List[Dict]:
"""옵션 상품 목록 조회"""
endpoint = f"{self.BASE_URL}/public/get_instruments"
params = {
"currency": currency,
"kind": "option",
"expired": False
}
try:
response = self.session.get(endpoint, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if data.get('success'):
return data['result']
else:
raise ValueError(f"API Error: {data.get('message')}")
except requests.exceptions.Timeout:
print("⏰ 요청 시간 초과 - 네트워크 연결을 확인하세요")
raise
except requests.exceptions.ConnectionError as e:
print(f"🔌 연결 오류: {e}")
raise
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
print("⚠️ Rate Limit 초과 - 1초 대기 후 재시도...")
time.sleep(1)
return self.get_instruments(currency)
raise
테스트 실행
surface = DeribitVolatilitySurface()
instruments = surface.get_instruments("BTC")
print(f"✅ BTC 옵션 상품 수: {len(instruments)}")변동성 곡면 데이터 수집
Deribit에서 옵션 데이터를 수집하는 핵심 함수를 구현합니다. 최근접 만기, ATM 옵션, Skew 분석에 필요한 모든 데이터를 한 번에 가져옵니다.
import asyncio
import aiohttp
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')
class DeribitOptionDataCollector:
"""Deribit 옵션 데이터 수집 및 정제"""
BASE_URL = "https://www.deribit.com/api/v2"
def __init__(self, currency: str = "BTC"):
self.currency = currency
self.session = None
async def fetch_with_retry(self, session, url: str, params: dict, max_retries: int = 3) -> dict:
"""재시도 로직이 포함된 API 호출"""
for attempt in range(max_retries):
try:
async with session.get(url, params=params) as response:
if response.status == 429:
wait_time = 2 ** attempt
print(f"⏳ Rate Limit - {wait_time}초 대기...")
await asyncio.sleep(wait_time)
continue
data = await response.json()
if data.get('success'):
return data['result']
else:
print(f"⚠️ API 오류: {data.get('message')}")
return None
except aiohttp.ClientTimeout:
print(f"⏰ 타임아웃 (시도 {attempt + 1}/{max_retries})")
if attempt < max_retries - 1:
await asyncio.sleep(1)
except Exception as e:
print(f"❌ 오류: {e}")
break
return None
async def get_option_book(self, instrument_name: str) -> Optional[dict]:
"""특정 옵션의 주문서 데이터 조회"""
if not self.session:
timeout = aiohttp.ClientTimeout(total=30)
self.session = aiohttp.ClientSession(timeout=timeout)
url = f"{self.BASE_URL}/public/get_order_book"
params = {"instrument_name": instrument_name}
return await self.fetch_with_retry(self.session, url, params)
async def get_volatility_surface_data(self) -> pd.DataFrame:
"""변동성 곡면 구축용 전체 데이터 수집"""
# 1. 활성 옵션 목록 가져오기
async with aiohttp.ClientSession() as session:
url = f"{self.BASE_URL}/public/get_instruments"
params = {"currency": self.currency, "kind": "option", "expired": False}
result = await self.fetch_with_retry(session, url, params)
if not result:
raise Exception("옵션 목록 조회 실패")
instruments = result
# 2. 주요 만기 필터링 (매주, 매월 만기)
expiries = {}
for inst in instruments:
exp_date = inst.get('expiration_timestamp', 0) // 1000
exp_datetime = datetime.fromtimestamp(exp_date)
days_to_expiry = (exp_datetime - datetime.now()).days
# 7일~180일 만기만 수집 (流动性 확보)
if 7 <= days_to_expiry <= 180:
strike = inst.get('strike', 0)
option_type = 'call' if inst.get('option_type') == 'call' else 'put'
inst_name = inst['instrument_name']
if exp_date not in expiries:
expiries[exp_date] = []
expiries[exp_date].append({
'instrument_name': inst_name,
'strike': strike,
'type': option_type,
'expiry': exp_datetime,
'days_to_expiry': days_to_expiry
})
# 3. 각 옵션의 내재변동성 수집
print(f"📊 {len(instruments)}개 옵션 중 {sum(len(v) for v in expiries.values())}개 분석 대상...")
async with aiohttp.ClientSession() as session:
all_data = []
for exp_date, options in expiries.items():
for opt in options[:20]: # 만기당 최대 20개 행사가
book = await self.get_option_book(opt['instrument_name'])
if book and book.get('greeks'):
greeks = book['greeks']
all_data.append({
'instrument_name': opt['instrument_name'],
'strike': opt['strike'],
'type': opt['type'],
'expiry': opt['expiry'],
'days_to_expiry': opt['days_to_expiry'],
'mark_iv': greeks.get('mark_iv', 0),
'delta': greeks.get('delta', 0),
'vega': greeks.get('vega', 0),
'gamma': greeks.get('gamma', 0),
'theta': greeks.get('theta', 0),
'underlying_price': book.get('underlying_price', 0)
})
await asyncio.sleep(0.05) # Rate Limit 방지
return pd.DataFrame(all_data)
비동기 실행
async def main():
collector = DeribitOptionDataCollector("BTC")
df = await collector.get_volatility_surface_data()
print(f"✅ 데이터 수집 완료: {len(df)}개 옵션")
print(df.head())
return df
df = asyncio.run(main())
변동성 곡면 구축 및 시각화
收찬된 데이터를 바탕으로 3D 변동성 곡면을 구축합니다. 스플라인 보간법과 RBF(Radial Basis Function)를 사용하여 부드러운 곡면을 생성합니다.
import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D
from scipy.interpolate import interp1d, griddata
from scipy.stats import norm
import numpy as np
def black_scholes_iv(
S: float, K: float, T: float, r: float,
market_price: float, option_type: str,
precision: float = 0.0001, max_iterations: int = 100
) -> float:
"""Newton-Raphson方法来计算隐含波动率"""
if T <= 0 or market_price <= 0:
return 0.0
# 초안 변동성 추청 (ATR 기반)
sigma = 0.5 if K == S else abs(np.log(S / K)) / np.sqrt(T)
sigma = max(0.01, min(sigma, 5.0))
for _ in range(max_iterations):
d1 = (np.log(S / K) + (r + sigma**2 / 2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
if option_type == 'call':
price = S * norm.cdf(d1) - K * np.exp(-r * T) * norm.cdf(d2)
else:
price = K * np.exp(-r * T) * norm.cdf(-d2) - S * norm.cdf(-d1)
vega = S * np.sqrt(T) * norm.pdf(d1)
if vega < 1e-10:
break
diff = market_price - price
if abs(diff) < precision:
return sigma
sigma += diff / vega
sigma = max(0.01, min(sigma, 5.0))
return sigma
class VolatilitySurfaceBuilder:
"""변동성 곡면 구축 및 3D 시각화"""
def __init__(self, risk_free_rate: float = 0.05):
self.r = risk_free_rate
def build_surface(self, df: pd.DataFrame, method: str = 'rbf') -> tuple:
"""2D 그리드 기반 변동성 곡면 구축"""
# 필터링: 유효한 데이터만
df_valid = df[(df['mark_iv'] > 0) & (df['mark_iv'] < 3)].copy()
df_valid['moneyness'] = np.log(df_valid['strike'] / df_valid['underlying_price'])
# 평활화 처리를 위한 만기 병합
df_valid['expiry_rounded'] = df_valid['days_to_expiry'].round(-1)
# 피벗 테이블 생성
pivot_table = df_valid.pivot_table(
values='mark_iv',
index='moneyness',
columns='expiry_rounded',
aggfunc='mean'
)
# 그리드 생성
moneyness_range = np.linspace(-0.5, 0.5, 50)
expiry_range = np.linspace(7, 180, 30)
M, E = np.meshgrid(moneyness_range, expiry_range)
# 보간법 적용
if method == 'rbf':
points = np.column_stack([df_valid['moneyness'].values,
df_valid['days_to_expiry'].values])
values = df_valid['mark_iv'].values
# RBF 보간
rbf = RBFInterpolator(points, values, kernel='thin_plate_spline', smoothing=1)
grid_points = np.column_stack([M.ravel(), E.ravel()])
V = rbf(grid_points).reshape(M.shape)
else: # cubic 보간
points = (df_valid['moneyness'].values, df_valid['days_to_expiry'].values)
V = griddata(points, df_valid['mark_iv'].values, (M, E), method='cubic')
# NaN 처리
V = np.nan_to_num(V, nan=np.nanmean(V))
return M, E, V, pivot_table
def plot_3d_surface(self, M, E, V):
"""3D 변동성 곡면 시각화"""
fig = plt.figure(figsize=(16, 10))
ax = fig.add_subplot(111, projection='3d')
surf = ax.plot_surface(M * 100, E, V * 100,
cmap='viridis', alpha=0.9,
linewidth=0, antialiased=True,
vmin=50, vmax=150)
ax.set_xlabel('Moneyness (%)', fontsize=12, labelpad=10)
ax.set_ylabel('Days to Expiry', fontsize=12, labelpad=10)
ax.set_zlabel('Implied Volatility (%)', fontsize=12, labelpad=10)
ax.set_title('BTC Options Volatility Surface', fontsize=16, fontweight='bold')
fig.colorbar(surf, shrink=0.5, aspect=10, label='IV (%)')
plt.tight_layout()
plt.savefig('volatility_surface.png', dpi=300)
plt.show()
def calculate_vol_smile(self, df: pd.DataFrame, expiry_days: int) -> pd.DataFrame:
"""특정 만기의 변동성 스마일 계산"""
df_expiry = df[
(df['days_to_expiry'] >= expiry_days - 5) &
(df['days_to_expiry'] <= expiry_days + 5)
].copy()
df_expiry['moneyness'] = np.log(df_expiry['strike'] / df_expiry['underlying_price'])
df_expiry = df_expiry.sort_values('moneyness')
return df_expiry[['strike', 'moneyness', 'mark_iv', 'delta', 'type']]
def plot_vol_smile(self, df_smile: pd.DataFrame):
"""변동성 스마일 플롯"""
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
# IV vs Moneyness
axes[0].scatter(df_smile['moneyness'] * 100, df_smile['mark_iv'] * 100,
c='blue', alpha=0.7, s=50)
axes[0].set_xlabel('Moneyness (%)')
axes[0].set_ylabel('Implied Volatility (%)')
axes[0].set_title('Volatility Smile')
axes[0].grid(True, alpha=0.3)
# Delta Skew
df_puts = df_smile[df_smile['type'] == 'put']
df_calls = df_smile[df_smile['type'] == 'call']
axes[1].scatter(df_puts['delta'], df_puts['mark_iv'] * 100,
c='red', label='Put', alpha=0.7)
axes[1].scatter(df_calls['delta'], df_calls['mark_iv'] * 100,
c='green', label='Call', alpha=0.7)
axes[1].set_xlabel('Delta')
axes[1].set_ylabel('Implied Volatility (%)')
axes[1].set_title('Delta-Weighted Smile')
axes[1].legend()
axes[1].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
실행 예시
builder = VolatilitySurfaceBuilder(risk_free_rate=0.05)
M, E, V, pivot = builder.build_surface(df, method='rbf')
builder.plot_3d_surface(M, E, V)
AI 기반 변동성 분석: HolySheep AI 통합
전통적인 정량 분석과 달리 AI를 활용하면 비정형 데이터(뉴스, 소셜 미디어, 온체인 지표)를 변동성에 통합할 수 있습니다. HolySheep AI를 사용하면 단일 API 키로 다양한 모델을 활용한 하이브리드 분석이 가능합니다.
DeepSeek V3.2를 활용한 시장 감성 분석
import requests
import json
class AIVolatilityAnalyzer:
"""HolySheep AI를 활용한 변동성 예측 및 감성 분석"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1" # HolySheep 공식 엔드포인트
def analyze_market_sentiment(self, recent_news: List[str]) -> dict:
"""DeepSeek V3.2로 시장 감성 분석 - 비용 효율적 ($0.42/MTok)"""
prompt = f"""다음 암호화폐 관련 뉴스 헤드라인을 분석하여 시장 감성 점수(-1 ~ +1)와
변동성 영향 정도를 분석해주세요:
{chr(10).join(f"- {news}" for news in recent_news[:10])}
다음 JSON 형식으로 응답해주세요:
{{
"sentiment_score": 숫자 (-1 ~ +1),
"volatility_impact": "high" | "medium" | "low",
"key_factors": ["요인1", "요인2"],
"market_outlook": "bullish" | "bearish" | "neutral"
}}"""
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "deepseek/deepseek-chat-v3-0324",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3,
"max_tokens": 500
},
timeout=30
)
if response.status_code == 200:
result = response.json()
content = result['choices'][0]['message']['content']
return json.loads(content)
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
def predict_volatility_regime(self, surface_data: dict, sentiment: dict) -> dict:
"""Gemini 2.5 Flash로 변동성 레짐 예측 - 초저비용 ($2.50/MTok)"""
prompt = f"""현재 BTC 옵션 변동성 곡면 데이터와 시장 감성을 분석하여
향후 변동성 레짐을 예측해주세요.
변동성 곡면:
- ATM 변동성: {surface_data.get('atm_iv', 0)*100:.1f}%
- Skew: {surface_data.get('skew', 0)*100:.1f}%
- 평활도: {surface_data.get('smoothness', 0):.2f}
시장 감성:
- 점수: {sentiment.get('sentiment_score', 0)}
- 영향: {sentiment.get('volatility_impact', 'unknown')}
- 전망: {sentiment.get('market_outlook', 'neutral')}
예측 결과를 JSON으로 응답:
{{
"regime": "low_vol" | "normal" | "high_vol" | "crisis",
"probability": 0.0~1.0,
"expected_move_1d": "±X%",
"recommended_hedge": "strategy suggestion"
}}"""
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "google/gemini-2.5-flash-preview-04-17",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.4,
"max_tokens": 400
},
timeout=30
)
if response.status_code == 200:
result = response.json()
return json.loads(result['choices'][0]['message']['content'])
raise Exception(f"Prediction failed: {response.text}")
사용 예시
analyzer = AIVolatilityAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
감성 분석 (DeepSeek - 초저렴)
sentiment = analyzer.analyze_market_sentiment([
"BTCETF 큰 규모 순유입 기록",
"연준 금리 인하 가능성 높아져",
"중국 규제 강화 우려 완화",
"BTC 채굴 난이도 사상 최고",
"기관 투자자 옵션 거래 증가"
])
print(f"감성 점수: {sentiment['sentiment_score']}")
print(f"변동성 영향: {sentiment['volatility_impact']}")
변동성 레짐 예측 (Gemini - 초저렴)
surface_data = {
'atm_iv': 0.65,
'skew': 0.05,
'smoothness': 0.85
}
regime_prediction = analyzer.predict_volatility_regime(surface_data, sentiment)
print(f"예측 레짐: {regime_prediction['regime']}")완전한 통합 대시보드
import dash
from dash import dcc, html, callback, Output, Input
import plotly.graph_objects as go
from datetime import datetime
class VolatilityDashboard:
"""실시간 변동성 곡면 대시보드"""
def __init__(self, collector: DeribitOptionDataCollector,
analyzer: AIVolatilityAnalyzer):
self.collector = collector
self.analyzer = analyzer
self.app = dash.Dash(__name__)
self._setup_layout()
def _setup_layout(self):
self.app.layout = html.Div([
html.H1("BTC Options Volatility Surface Dashboard",
style={'textAlign': 'center', 'color': '#2c3e50'}),
html.Div([
html.Div([
html.H3("변동성 곡면"),
dcc.Graph(id='vol-surface-3d')
], className='six columns'),
html.Div([
html.H3("변동성 스마일"),
dcc.Graph(id='vol-smile')
], className='six columns')
], className='row'),
html.Div([
html.Div([
html.H3("AI 감성 분석"),
html.Div(id='sentiment-output')
], className='four columns'),
html.Div([
html.H3("변동성 레짐 예측"),
html.Div(id='regime-output')
], className='four columns'),
html.Div([
html.H3(" Greeks 모니터"),
dcc.Graph(id='greeks-chart')
], className='four columns')
], className='row'),
dcc.Interval(
id='interval-component',
interval=60*1000, # 1분마다 갱신
n_intervals=0
)
])
@callback(
[Output('vol-surface-3d', 'figure'),
Output('vol-smile', 'figure'),
Output('sentiment-output', 'children'),
Output('regime-output', 'children')],
[Input('interval-component', 'n_intervals')]
)
def update_dashboard(n):
# 데이터 수집
df = asyncio.run(self.collector.get_volatility_surface_data())
# 3D 표면
M, E, V, _ = builder.build_surface(df)
fig_3d = go.Figure(data=[go.Surface(x=M*100, y=E, z=V*100)])
fig_3d.update_layout(
title='BTC IV Surface',
scene=dict(
xaxis_title='Moneyness %',
yaxis_title='Days',
zaxis_title='IV %'
)
)
# 스마일
df_smile = builder.calculate_vol_smile(df, 30)
fig_smile = go.Figure()
fig_smile.add_trace(go.Scatter(
x=df_smile['moneyness']*100,
y=df_smile['mark_iv']*100,
mode='markers+lines',
name='IV Smile'
))
# AI 분석
surface_summary = {
'atm_iv': df[df['delta'].between(-0.55, -0.45)]['mark_iv'].mean(),
'skew': df[df['delta'] < -0.5]['mark_iv'].mean() -
df[df['delta'] > 0.5]['mark_iv'].mean()
}
return fig_3d, fig_smile, str(surface_summary), "Analysis complete"
def run(self, debug: bool = False, port: int = 8050):
self.app.run_server(debug=debug, port=port)
실행
dashboard = VolatilityDashboard(collector, analyzer)
dashboard.run()
자주 발생하는 오류 해결
1. ConnectionError: HTTPSConnectionPool 타임아웃
# 문제: Deribit API 연결 타임아웃
원인: 네트워크 라우팅 문제, 방화벽, Rate Limit
해결 1: 세션 타임아웃 증가 및 재시도 로직
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_robust_session():
session = requests.Session()
# 재시도 전략 설정
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["HEAD", "GET", "OPTIONS", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
# 타임아웃 설정
session.timeout = 30
return session
해결 2: WebSocket 사용 (지연 민감 데이터용)
import websockets
import json
async def websocket_option_stream():
uri = "wss://www.deribit.com/ws/api/v2"
async with websockets.connect(uri) as websocket:
# 구독 요청
await websocket.send(json.dumps({
"jsonrpc": "2.0",
"id": 1,
"method": "public/subscribe",
"params": {
"channels": ["book.BTC-PERPETUAL.raw",
"deribit_price_index.btc_usd"]
}
}))
while True:
try:
message = await asyncio.wait_for(websocket.recv(), timeout=30)
data = json.loads(message)
print(data)
except asyncio.TimeoutError:
print("Heartbeat check...")2. KeyError: 'mark_iv' - 데이터 누락
# 문제: Deribit API 응답에서 greeks/mark_iv가 누락됨
원인:流动性 부족 옵션, 만기 임박 옵션, API 응답 형식 변경
해결: 데이터 검증 및 폴백 로직
def safe_get_mark_iv(book_data: dict, instrument_name: str) -> float:
"""안전하게 내재변동성 추출"""
# 방법 1: greeks 객체에서 추출
greeks = book_data.get('greeks', {})
if greeks and 'mark_iv' in greeks:
return float(greeks['mark_iv'])
# 방법 2: 공정가격 기반 역산
if 'best_bid_price' in book_data and 'best_ask_price' in book_data:
mid_price = (book_data['best_bid_price'] + book_data['best_ask_price']) / 2
S = book_data.get('underlying_price', 0)
K = extract_strike_from_instrument(instrument_name)
T = extract_time_to_expiry(instrument_name)
if S > 0 and K > 0 and T > 0:
# BSCall 옵션 가정
iv = black_scholes_iv(S, K, T, 0.05, mid_price, 'call')
if iv > 0.01 and iv < 5.0:
return iv
# 방법 3: 근처 옵션에서 보간
print(f"⚠️ {instrument_name} IV 누락 - 보간으로 대체")
return interpolate_from_nearby_options(instrument_name)
def extract_strike_from_instrument(instrument_name: str) -> float:
"""instrument_name에서 행사가 추출: BTC-25JUL25-95000-C"""
parts = instrument_name.split('-')
try:
return float(parts[2])
except:
return 0.0
def extract_time_to_expiry(instrument_name: str) -> float:
"""instrument_name에서 잔여 일수 계산"""
import re
parts = instrument_name.split('-')
if len(parts) >= 2:
date_str = parts[1]
expiry = datetime.strptime(date_str, '%d%b%y')
return max((expiry - datetime.now()).days / 365, 0.001)
return 0.03. 401 Unauthorized - API 인증 오류
# 문제: Deribit API 인증 실패
원인: 잘못된 API 키, 권한 부족, 시간 동기화 오류
해결 1: 올바른 인증 방식
import time
class AuthenticatedDeribitClient:
def __init__(self, client_id: str, client_secret: str):
self.client_id = client_id
self.client_secret = client_secret
self.access_token = None
self.expires_at = 0
self.base_url = "https://www.deribit.com/api/v2"
def authenticate(self) -> str:
"""OAuth2 인증 토큰 획득"""
# 시간 동기화 확인
server_time = requests.get(f"{self.base_url}/public/get_time").json()
local_time = int(time.time() * 1000)
if abs(server_time['result'] - local_time) > 30000:
print("⚠️ 시스템 시간 오차 큼 - NTP 동기화 권장")
response = requests.post(
f"{self.base_url}/public/auth",
json={
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret,
"scope": "session:any"
}
)
if response.status_code == 200:
data = response.json()
self.access_token = data['result']['access_token']
self.expires_at = time.time() + data['result']['expires_in']
return self.access_token
else:
raise Exception(f"인증 실패: {response.text}")
def get_authenticated_headers(self) -> dict:
"""토큰 갱신 필요 시 자동 처리"""
if not self.access_token or time.time() > self.expires_at - 60:
self.authenticate()
return {"Authorization": f"Bearer {self.access_token}"}
해결 2: 공개 API만 사용하는 경우 (인증 불필요)
class PublicDeribitClient:
"""공개 엔드포인트만 사용 - 읽기 전용"""
def __init__(self):
self.base_url = "https://www.deribit.com/api/v2"
def get_public_data(self, method: str, params: dict) -> dict:
"""공개 API만 호출 - 인증 불필요"""
public_methods = [
'get_instruments', 'get_order_book', 'get_last_trades',
'get_volatility_surface_data', 'get_settlement_history'
]
if method not in public_methods:
raise ValueError(f"인증 필요: {method}")
response = requests.post(
f"{self.base_url}/{method.replace('_', '/')}",
json={"jsonrpc": "2.0", "method": method, "params": params, "id": 1}
)
return response.json()4. Rate Limit 429: Too Many Requests
# 문제: API 호출 빈도 초과
원인: 너무 빠른 속도로 연속 호출
해결: Rate Limiter 구현
import asyncio
import time
from collections import deque
class RateLimiter:
"""슬라이딩 윈도우 기반 Rate Limiter"""
def __init__(self, max_calls: int, time_window: float):
self.max_calls = max_calls
self.time_window = time_window
self.calls = deque()
async def acquire(self):
"""호출 가능할 때까지 대기"""
now = time.time()
# 오래된 호출 기록 제거
while self.calls and self.calls[0] < now - self.time_window:
self.calls.popleft()
if len(self.calls) >= self.max_calls:
# 다음 가능 시간 계산
wait_time = self.calls[0] + self.time_window - now
if wait_time > 0:
print(f"⏳ Rate Limit - {wait_time:.2f}초 대기")
await asyncio.sleep(wait_time)
self.calls.append(time.time())
사용 예시
rate_limiter = RateLimiter(max_calls=10, time_window=1.0) # 1초에 최대 10회
async def throttled_api_call(url: str, params: dict):
await rate_limiter.acquire()
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
return await response.json()
대량 데이터 수집 최적화
async def batch_collect(instruments: List[str], batch_size: int = 5):
"""배치 처리로 API 호출 최적화"""
results = []
for