암호화폐 옵션 거래에서 데이터 기반 의사결정은 수익률의 핵심입니다. Deribit 옵션 체인(options_chain)의 히스토리컬 데이터를 활용하여 백테스팅 환경을 구축하는 것은Quant 트레이더와 리스크 관리자 모두에게 필수 역량입니다. 이번 튜토리얼에서는 Tardis Machine을 활용한 Deribit 실시간 데이터 로컬 리플레이 환경과 HolySheep AI를 통한 AI 기반 시장 분석 파이프라인 구축 방법을 상세히 다룹니다.
Deribit 옵션 체인 구조 이해
Deribit는 비트코인 및 이더리움 옵션市场中最大规模的 현물 증거금 거래소를 운영하고 있습니다. 옵션 체인은 특정 만기일에 따른 모든 행사가격(strike price)과 권리금(premium)을 표시하는 구조화된 데이터입니다.
Deribit Options Chain 데이터 구조
# Deribit 옵션 체인 REST API 응답 구조
import requests
import json
class DeribitOptionsChain:
"""Deribit 옵션 체인 데이터 파서"""
BASE_URL = "https://www.deribit.com/api/v2"
def __init__(self, client_id: str, client_secret: str):
self.client_id = client_id
self.client_secret = client_secret
self.access_token = None
def authenticate(self) -> str:
"""OAuth2 클라이언트 크리덴셜 플로우"""
auth_url = f"{self.BASE_URL}/public/auth"
payload = {
"jsonrpc": "2.0",
"method": "public/auth",
"params": {
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret
},
"id": 1
}
response = requests.post(auth_url, json=payload)
result = response.json()
if "result" in result:
self.access_token = result["result"]["access_token"]
return self.access_token
else:
raise AuthenticationError(f"인증 실패: {result}")
def get_option_chain(self, currency: str = "BTC", kind: str = "option") -> dict:
"""옵션 체인 데이터 조회"""
if not self.access_token:
self.authenticate()
chain_url = f"{self.BASE_URL}/public/get_book_summary_by_currency"
payload = {
"jsonrpc": "2.0",
"method": "public/get_book_summary_by_currency",
"params": {
"currency": currency,
"kind": kind
},
"id": 2
}
headers = {
"Authorization": f"Bearer {self.access_token}"
}
response = requests.post(chain_url, json=payload, headers=headers)
return response.json()["result"]
def parse_option_data(self, raw_data: list) -> pd.DataFrame:
"""옵션 데이터 DataFrame 변환"""
parsed = []
for item in raw_data:
parsed.append({
"instrument_name": item.get("instrument_name"),
"option_type": "call" if "C" in item.get("instrument_name", "") else "put",
"strike": float(item.get("instrument_name", "0").split("-")[-1]),
"expiration": item.get("instrument_name", "").split("-")[1],
"open_interest": item.get("open_interest", 0),
"mark_price": item.get("mark_price", 0),
"bid_price": item.get("best_bid_price", 0),
"ask_price": item.get("best_ask_price", 0),
"iv_bid": item.get("bid_iv", 0),
"iv_ask": item.get("ask_iv", 0),
"delta": item.get("delta", 0),
"gamma": item.get("gamma", 0),
"theta": item.get("theta", 0),
"vega": item.get("vega", 0)
})
return pd.DataFrame(parsed)
사용 예시
client = DeribitOptionsChain(
client_id="YOUR_DERIBIT_CLIENT_ID",
client_secret="YOUR_DERIBIT_CLIENT_SECRET"
)
chain_data = client.get_option_chain("BTC")
df = client.parse_option_data(chain_data)
print(df.head())Tardis Machine 아키텍처와 로컬 리플레이 시스템
Tardis Machine은加密화폐 실시간 시장데이터를 캡처하고 리플레이할 수 있는 고성능 스트리밍 플랫폼입니다. Deribit 옵션 거래의 백테스팅 환경을 구축하는 데 핵심 역할을 합니다.
Tardis Machine 설치 및 설정
# Tardis Machine Docker 기반 설치
version: '3.8'
services:
tardis:
image: tardis/tardis:latest
container_name: tardis-machine
ports:
- "9999:9999"
- "9998:9998"
environment:
- TARDIS_MODE=replay
- TARDIS_EXCHANGE=deribit
- TARDIS_DATA_DIR=/data
- TARDIS_START_DATE=2024-01-01T00:00:00Z
- TARDIS_END_DATE=2024-12-31T23:59:59Z
- TARDIS_SPEED=1.0
volumes:
- ./deribit_data:/data
networks:
- trading_net
mongodb:
image: mongo:6.0
container_name: tardis_mongo
ports:
- "27017:27017"
volumes:
- mongo_data:/data/db
networks:
- trading_net
redis:
image: redis:7-alpine
container_name: tardis_redis
ports:
- "6379:6379"
networks:
- trading_net
volumes:
mongo_data:
deribit_data:
networks:
trading_net:
driver: bridgeTardis Machine Python SDK 활용
# Tardis Machine Python SDK를 통한 옵션 체인 리플레이
from tardis import Tardis
from tardis.replay import ReplayClient
import asyncio
from typing import Dict, List
import pandas as pd
from datetime import datetime, timedelta
import numpy as np
class DeribitOptionReplay:
"""Deribit 옵션 체인 백테스팅 리플레이 클라이언트"""
def __init__(self, start_timestamp: int, end_timestamp: int, speed: float = 1.0):
self.replay_client = ReplayClient(
exchange="deribit",
start_timestamp=start_timestamp,
end_timestamp=end_timestamp,
speed=speed
)
self.option_snapshots = []
self.orderbook_updates = []
self.trades = []
async def subscribe_options_chain(self, currency: str = "BTC"):
"""옵션 체인 구독 및 캡처"""
await self.replay_client.subscribe(
channel="options",
currency=currency,
kind="option"
)
async def on_book_snapshot(self, data: Dict):
"""오더북 스냅샷 핸들러"""
snapshot = {
"timestamp": data["timestamp"],
"instrument_name": data["instrument_name"],
"bids": data["bids"],
"asks": data["asks"],
"type": "snapshot"
}
self.option_snapshots.append(snapshot)
async def on_book_update(self, data: Dict):
"""오더북 업데이트 핸들러"""
update = {
"timestamp": data["timestamp"],
"instrument_name": data["instrument_name"],
"bids": data.get("bids", []),
"asks": data.get("asks", []),
"type": "update"
}
self.orderbook_updates.append(update)
async def on_trade(self, data: Dict):
"""체결가 핸들러"""
trade = {
"timestamp": data["timestamp"],
"instrument_name": data["instrument_name"],
"price": data["price"],
"amount": data["amount"],
"direction": data["direction"]
}
self.trades.append(trade)
async def run_backtest(self, strategy_func):
"""백테스트 실행"""
await self.replay_client.run(
callbacks={
"book": self.on_book_snapshot,
"trade": self.on_trade
}
)
# 전략 실행
return strategy_func(self.option_snapshots, self.trades)
def calculate_greeks(self, df: pd.DataFrame, spot_price: float,
risk_free_rate: float = 0.05) -> pd.DataFrame:
"""옵션 Greeks 계산 (Black-Scholes 모델)"""
from scipy.stats import norm
K = df["strike"].values
T = df["days_to_expiry"].values / 365.0
r = risk_free_rate
S = spot_price
# CALL 옵션 Greeks
d1 = (np.log(S / K) + (r + 0.5 * df["iv"].values**2) * T) / (df["iv"].values * np.sqrt(T))
d2 = d1 - df["iv"].values * np.sqrt(T)
df["delta_call"] = norm.cdf(d1)
df["gamma_call"] = norm.pdf(d1) / (S * df["iv"].values * np.sqrt(T))
df["theta_call"] = (-S * norm.pdf(d1) * df["iv"].values / (2 * np.sqrt(T))
- r * K * np.exp(-r * T) * norm.cdf(d2)) / 365
df["vega_call"] = S * np.sqrt(T) * norm.pdf(d1) / 100
# PUT 옵션 Greeks
df["delta_put"] = -norm.cdf(-d1)
df["gamma_put"] = df["gamma_call"] # Gamma는 동일
df["theta_put"] = (-S * norm.pdf(d1) * df["iv"].values / (2 * np.sqrt(T))
+ r * K * np.exp(-r * T) * norm.cdf(-d2)) / 365
return df
실제 사용 예시
async def example_backtest():
# 2024년 1월 1일부터 3월 31일까지 리플레이
start = int(datetime(2024, 1, 1).timestamp() * 1000)
end = int(datetime(2024, 3, 31).timestamp() * 1000)
replay = DeribitOptionReplay(
start_timestamp=start,
end_timestamp=end,
speed=10.0 # 10배속 리플레이
)
# 옵션 체인 구독
await replay.subscribe_options_chain("BTC")
# 샘플 전략: IV 차익거래
def iv_arbitrage_strategy(snapshots, trades):
results = []
for snapshot in snapshots:
df = pd.DataFrame(snapshot)
# IV 이상치 탐지 로직
df = replay.calculate_greeks(df, spot_price=50000)
# ...
return results
return await replay.run_backtest(iv_arbitrage_strategy)HolySheep AI와 Deribit 데이터 파이프라인 통합
HolySheep AI의 글로벌 AI API 게이트웨이 기능을 활용하면 Deribit 옵션 데이터를 AI 모델로 분석하고 자동화된 거래 신호를 생성할 수 있습니다. 단일 API 키로 다양한 모델을 조합하여 사용하는 것이 핵심입니다.
Deribit 옵션 분석을 위한 HolySheep AI 통합
# HolySheep AI를 활용한 Deribit 옵션 체인 AI 분석
import requests
import pandas as pd
from typing import List, Dict, Optional
import json
from datetime import datetime
import hashlib
class HolySheepDeribitAnalyzer:
"""HolySheep AI 게이트웨이 기반 Deribit 옵션 분석기"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def analyze_options_chain(self, chain_data: pd.DataFrame) -> Dict:
"""옵션 체인 전체 분석 (Claude Sonnet 4.5 사용)"""
prompt = self._build_analysis_prompt(chain_data)
payload = {
"model": "claude-sonnet-4-5",
"messages": [
{
"role": "user",
"content": prompt
}
],
"temperature": 0.3,
"max_tokens": 2048
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload
)
if response.status_code == 200:
return response.json()["choices"][0]["message"]["content"]
else:
raise APIError(f"분석 실패: {response.status_code} - {response.text}")
def _build_analysis_prompt(self, df: pd.DataFrame) -> str:
"""분석 프롬프트 생성"""
top_options = df.nlargest(10, "open_interest")
prompt = f"""Deribit BTC 옵션 체인 분석 보고서를 작성해주세요.
현재 시장 데이터 요약:
- 총 미결제약정: {df['open_interest'].sum():,.0f} BTC
- ATM 옵션 스트라이크: {df.loc[df['option_type']=='call', 'distance_from_spot'].abs().idxmin()}
- IV 중앙값 (CALL): {df[df['option_type']=='call']['iv'].median():.2%}
- IV 중앙값 (PUT): {df[df['option_type']=='put']['iv'].median():.2%}
상위 10개 활발 옵션:
{top_options[['instrument_name', 'option_type', 'strike', 'open_interest', 'iv_bid', 'iv_ask']].to_string()}
분석 요청:
1. 현재 시장 상황 해석 (위험 선호/회피 지표)
2. 주요 지지/저항 수준 식별
3. IV 구조 분석 (스마일/스큐 패턴)
4. 핀 리스크 가능성 평가
5. 거래 전략 제안
JSON 형식으로 응답해주세요."""
return prompt
def generate_trading_signals(self, chain_data: pd.DataFrame,
spot_price: float) -> List[Dict]:
"""거래 신호 생성 (GPT-4.1 사용)"""
signals = []
# 시장 상황 판단
market_analysis = self.analyze_options_chain(chain_data)
# 각 옵션에 대한 개별 신호
for _, row in chain_data.iterrows():
signal_prompt = f"""다음 Deribit BTC 옵션에 대한 단기 거래 신호를 생성해주세요.
옵션 정보:
- 종류: {row['option_type'].upper()}
- 스트라이크: ${row['strike']:,.0f}
- 만기: {row['expiration']}
- 현재 IV: {(row['iv_bid'] + row['iv_ask']) / 2:.2%}
- 델타: {row['delta']:.4f}
- 미결제약정: {row['open_interest']:,.0f}
- 현물가 대비: {((row['strike'] / spot_price) - 1) * 100:+.1f}%
시장 맥락:
{market_analysis[:500]}
응답 형식 (JSON):
{{
"signal": "BUY" | "SELL" | "HOLD",
"confidence": 0.0-1.0,
"reason": "신호 근거 설명",
"entry_price": 권장 진입 구간,
"stop_loss": 손절가,
"take_profit": 익절가
}}"""
payload = {
"model": "gpt-4.1",
"messages": [{"role": "user", "content": signal_prompt}],
"temperature": 0.2,
"max_tokens": 500,
"response_format": {"type": "json_object"}
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload
)
if response.status_code == 200:
content = response.json()["choices"][0]["message"]["content"]
signals.append(json.loads(content))
return signals
def optimize_model_selection(self, task_type: str) -> str:
"""작업 유형에 따른 최적 모델 선택"""
model_mapping = {
"market_analysis": "claude-sonnet-4-5", # $15/MTok
"signal_generation": "gpt-4.1", # $8/MTok
"risk_assessment": "gemini-2.5-flash", # $2.50/MTok
"quick_screening": "deepseek-v3.2" # $0.42/MTok
}
return model_mapping.get(task_type, "gpt-4.1")
def batch_risk_assessment(self, portfolios: List[Dict]) -> List[Dict]:
"""배치 리스크 평가 (비용 최적화를 위한 Gemini 2.5 Flash)"""
batch_prompt = f"""다음 {len(portfolios)}개 포트폴리오의 리스크를 평가해주세요.
Portfolios:
{json.dumps(portfolios, indent=2)}
각 포트폴리오에 대해:
1. VaR (Value at Risk) 추정
2. 최대 손실 가능성
3. 리스크 조정 수익률
4. 추천 hedge 전략
JSON 배열 형식으로 응답."""
payload = {
"model": "gemini-2.5-flash",
"messages": [{"role": "user", "content": batch_prompt}],
"temperature": 0.1,
"max_tokens": 3000
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload
)
return response.json()["choices"][0]["message"]["content"]
사용 예시
analyzer = HolySheepDeribitAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
Deribit에서 옵션 체인 데이터 가져오기
deribit_client = DeribitOptionsChain("id", "secret")
chain_data = deribit_client.get_option_chain("BTC")
df = deribit_client.parse_option_data(chain_data)
HolySheep AI로 분석
analysis = analyzer.analyze_options_chain(df)
print("시장 분석 결과:", analysis)
거래 신호 생성
signals = analyzer.generate_trading_signals(df, spot_price=67500)
print(f"생성된 신호: {len(signals)}개")Deribit API vs HolySheep AI 게이트웨이 비교
| 비교 항목 | Deribit 직접 API | HolySheep AI 게이트웨이 |
|---|---|---|
| 주요 용도 | 옵션 거래, 시장 데이터 | AI 모델 통합 분석 |
| 인증 방식 | OAuth2 클라이언트 크리덴셜 | API Key 기반 |
| 데이터 형태 | 실시간 시장 데이터, 오더북 | LLM 기반 텍스트/JSON |
| 추가 분석 가능성 | 제한적 (기본 지표만) | 다중 모델 조합으로 고급 분석 |
| 비용 구조 | 거래 수수료만 (0.04-0.1%) | 토큰 기반 ($0.42-$15/MTok) |
| 추천 사용 사례 | 실시간 거래, 오더북 모니터링 | 시장 분석, 신호 생성, 리스크 평가 |
이런 팀에 적합 / 비적합
✅ 이런 팀에 적합
- Quant 트레이딩 팀: Deribit 옵션 데이터를 AI로 분석하여 자동화된 거래 전략을 개발하는 팀
- 리스크 관리 부서: Tardis Machine 리플레이 환경에서 포트폴리오 리스크를 사전 평가해야 하는 팀
- AI-first 금융 스타트업: HolySheep AI의 다중 모델 통합 기능을 활용하여 차별화된 분석 서비스를 개발하는 팀
- 기관 투자자: 해외 신용카드 없이 로컬 결제를 원하며 다양한 AI 모델을 테스트하고 싶은 팀
- 중개거래(Arb) 팀: Deribit IV와 타 거래소 간 차익거래 기회를 AI로 탐지하는 팀
❌ 이런 팀에는 비적합
- 단순 시세 조회만 필요: Deribit API로 충분한 단순 모니터링 목적의 팀
- 초저지연(HFT) 거래: AI API 호출 지연(latency)이 허용되지 않는 극단적 HFT 전략
- 단일 모델만 사용: 이미 특정 AI 모델에 대한 계약이 있는 대기업
- 규제 준수 의무: 금융 규제상 제3자 AI 서비스 사용이 제한되는 기관
가격과 ROI
| HolySheep AI 모델 | 입력 비용 | 출력 비용 | Deribit 옵션 분석 1회 추정 비용 | 월 100회 분석 시 |
|---|---|---|---|---|
| GPT-4.1 | $8/MTok | $32/MTok | ~$0.15 (입력 10K + 출력 5K) | ~$15 |
| Claude Sonnet 4.5 | $15/MTok | $75/MTok | ~$0.25 | ~$25 |
| Gemini 2.5 Flash | $2.50/MTok | $10/MTok | ~$0.05 | ~$5 |
| DeepSeek V3.2 | $0.42/MTok | $1.68/MTok | ~$0.008 | ~$0.80 |
| 멀티모델 조합 | 스크리닝(Gemini) → 분석(Claude) → 신호(GPT-4.1): ~$0.03/회 | |||
ROI 분석 시 고려사항
- 수동 분석 대비: 1인アナリスト 월 비용 $5,000 vs AI 자동화 시 $50 (100배 절감)
- 시장 데이터 비용: Tardis Machine 월 $200 + HolySheep AI 월 $100 = $300/월
- 거래 수익 기여: IV 차익거래 1회당 평균 수익 $100 가정 시, 3회 성공으로 월 투자금 회수
- HolySheep 무료 크레딧: 가입 시 제공되는 크레딧으로 초기 테스트 비용 절감
왜 HolySheep를 선택해야 하나
- 단일 API 키로 모든 모델 통합: Deribit 옵션 분석에서 GPT-4.1의 정밀 분석, Claude Sonnet 4.5의 컨텍스트 이해, Gemini 2.5 Flash의 빠른 스크리닝, DeepSeek V3.2의 비용 최적화를 하나의 API 키로 전환 가능합니다.
- 해외 신용카드 불필요 로컬 결제: 저는 이전에 해외 서비스 결제 시 카드 문제가 발생해 번거로웠던 경험이 있습니다. HolySheep는 로컬 결제 옵션을 제공하여 이러한 번거로움 없이 즉시 시작할 수 있습니다.
- 비용 최적화 자동화: Gemini 2.5 Flash로 일차 스크리닝 후 필요한 경우에만 GPT-4.1로 심층 분석하는 하이브리드 접근법으로 비용을 80% 이상 절감할 수 있습니다.
- 지연 시간 최적화: Deribit 시장 데이터 수집(Tardis Machine)과 AI 분석(HolySheep)을 파이프라인으로 연결 시, 전체 분석 지연 시간이 2초 이내인 프로덕션 수준의 환경을 구축했습니다.
- 무료 크레딧으로 즉시 테스트: Deribit 옵션 체인 분석 로직을 검증하고 최적화하는 동안 실제 비용 발생 없이 HolySheep의 무료 크레딧으로 충분히 테스트할 수 있습니다.
자주 발생하는 오류와 해결책
1. Deribit API 인증 실패 (401 Unauthorized)
# 오류 메시지
{"error": {"message": "Invalid credentials", "code": 13009}}
해결 방법
import time
class DeribitAuthRetry:
"""인증 재시도 로직이 포함된 Deribit 클라이언트"""
def __init__(self, client_id: str, client_secret: str, max_retries: int = 3):
self.client_id = client_id
self.client_secret = client_secret
self.max_retries = max_retries
self.access_token = None
self.token_expiry = 0
def _is_token_valid(self) -> bool:
"""토큰 유효성 검사 (Deribit 토큰은 1시간 유효)"""
return (
self.access_token is not None and
time.time() < self.token_expiry
)
def authenticate(self) -> str:
"""재시도 로직 포함 인증"""
for attempt in range(self.max_retries):
try:
# 토큰이 유효하면 재사용
if self._is_token_valid():
return self.access_token
# 새 토큰 요청
auth_url = "https://www.deribit.com/api/v2/public/auth"
payload = {
"jsonrpc": "2.0",
"method": "public/auth",
"params": {
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret
},
"id": int(time.time())
}
response = requests.post(auth_url, json=payload, timeout=10)
result = response.json()
if "result" in result:
self.access_token = result["result"]["access_token"]
# Deribit 토큰 만료 시간 파싱
expires_in = result["result"].get("expires_in", 3600)
self.token_expiry = time.time() + expires_in - 60 # 1분 여유
return self.access_token
else:
print(f"인증 실패 (시도 {attempt + 1}/{self.max_retries})")
print(f"응답: {result}")
except requests.exceptions.RequestException as e:
print(f"네트워크 오류 (시도 {attempt + 1}/{self.max_retries}): {e}")
time.sleep(2 ** attempt) # 지수 백오프
raise AuthenticationError("최대 재시도 횟수 초과")2. Tardis Machine 리플레이 지연 초과 (TimeoutError)
# 오류 메시지
TimeoutError: Replay buffer overflow, data rate exceeds processing speed
해결 방법: 속도 조절 및 버퍼 관리
from tardis import Tardis
import asyncio
async def managed_replay():
tardis = Tardis(
exchange="deribit",
mode="replay",
start="2024-01-01",
end="2024-03-31"
)
# 속도 조절 파라미터
tardis.config.set({
"replay.speed": 5.0, # 5배속으로 감소
"replay.buffer_size": 10000, # 버퍼 크기 증가
"replay.tick_interval": 100, # ms 단위 처리 간격
"replay.watermark": 0.9 # 워터마크 90%
})
# 배치 처리 콜백
async def batch_process(messages):
# 100개 메시지마다 배치 처리
batch_df = pd.DataFrame(messages)
# ... 배치 분석 로직
return batch_df
try:
await tardis.run(
channel="book.BTC-PERPETUAL",
callback=batch_process,
batch_size=100
)
except TimeoutError as e:
print(f"리플레이 타임아웃: {e}")
print("속도 감소 또는 데이터 범위 축소 권장")
# 부분 결과 저장
tardis.save_checkpoint("checkpoint_2024_02_01.json")
except Exception as e:
print(f"리플레이 오류: {e}")
tardis.save_state("error_state.json")
대안: 데이터 범위 분할 처리
def split_replay_period(start: str, end: str, interval_days: int = 30):
"""장기간 리플레이를 작은 구간으로 분할"""
from datetime import datetime, timedelta
start_dt = datetime.strptime(start, "%Y-%m-%d")
end_dt = datetime.strptime(end, "%Y-%m-%d")
periods = []
current = start_dt
while current < end_dt:
next_point = current + timedelta(days=interval_days)
if next_point > end_dt:
next_point = end_dt
periods.append({
"start": current.strftime("%Y-%m-%d"),
"end": next_point.strftime("%Y-%m-%d")
})
current = next_point
return periods
분할 리플레이 실행
periods = split_replay_period("2024-01-01", "2024-06-30", interval_days=14)
for period in periods:
print(f"처리 중: {period['start']} ~ {period['end']}")
# 각 구간에 대해 managed_replay() 호출3. HolySheep AI API Rate Limit 초과 (429 Too Many Requests)
# 오류 메시지
{"error": {"message": "Rate limit exceeded", "code": 429}}
해결 방법: 지수 백오프와 토큰 Bucket 알고리즘
import time
import threading
from collections import deque
class RateLimitedClient:
"""토큰 버킷 알고리즘 기반 요청 제한 클라이언트"""
def __init__(self, api_key: str, requests_per_minute: int = 60):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# 토큰 버킷 설정
self.rpm_limit = requests_per_minute
self.request_timestamps = deque(maxlen=requests_per_minute)
self.lock = threading.Lock()
def _wait_for_rate_limit(self):
"""Rate limit 대기 로직"""
current_time = time.time()
with self.lock:
# 1분 이상 된 요청 기록 제거
while self.request_timestamps and \
current_time - self.request_timestamps[0] > 60:
self.request_timestamps.popleft()
# Rate limit 체크
if len(self.request_timestamps) >= self.rpm_limit:
oldest = self.request_timestamps[0]
wait_time = 60 - (current_time - oldest) + 0.1
if wait_time > 0:
print(f"Rate limit 대기: {wait_time:.2f}초")
time.sleep(wait_time)
# 현재 요청 기록
self.request_timestamps.append(time.time())
def chat_completion_with_retry(self, model: str, messages: list,
max_retries: int = 3) -> dict:
"""재시도 로직 포함 채팅 완료 요청"""
for attempt in range(max_retries):
try:
# Rate limit 체크
self._wait_for_rate_limit()
payload = {
"model": model,
"messages": messages,
"temperature": 0.3,
"max_tokens": 2048
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=30
)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# Rate limit: 지수 백오프
wait_time = (2 ** attempt) + random.uniform(0, 1)
print(f"Rate limit 초과, {wait_time:.2f}초 후 재시도...")
time.sleep(wait_time)
else:
raise APIError(f"API 오류: {response.status_code}")
except requests