저는 3년 넘게 퀀트 트레이딩 시스템 개발에 참여하면서 가장 많은 시간이 들었던 부분이 바로 시장 데이터 수집 파이프라인 구축이었습니다. 특히 Deribit의 BTC 옵션 데이터는 실시간 Greeks 계산과 역사적 변동성 분석에 필수적인데, 공식 문서가分散되어 있어 처음 접근하기 어렵습니다. 이 가이드에서는 완전 초보자도 Deribit 옵션 데이터를 안정적으로 가져올 수 있도록 단계별로 설명드리겠습니다.
Deribit API란 무엇인가
Deribit는 세계 최대 BTC/USD 옵션 거래소로, 강력한 REST API와 WebSocket을 제공합니다. API란 Application Programming Interface의 약자로, 쉽게 말해 프로그래밍으로 Deribit 시스템에 접속할 수 있는 문어입니다. 별도 가입비 없이 Deribit.com에서 무료로 API 키를 발급받을 수 있습니다.
Deribit API의 핵심 특징은 다음과 같습니다:
- 공용 엔드포인트: 옵션 데이터 조회, 현재가, 거래량 등 대부분의 데이터가 API 키 없이 조회 가능
- 인증 엔드포인트: 주문 실행, 잔액 확인 등 개인 데이터는 서명 인증 필요
- 센터 데이터: 내재변동성(IV), Greeks(Delta, Gamma, Vega, Theta 등), 미결제약정(OI) 포함
- 분 단위 히스토리: 2020년 이후 틱 단위 데이터 제공
1단계: 개발 환경 준비
시작하기 전에 Python 환경이 필요합니다. Python을 아직 설치하지 않으셨다면 python.org에서 최신 버전을 다운로드하세요. 이 튜토리얼에서는 Python 3.8 이상을 사용합니다.
필수 라이브러리 설치
pip install requests pandas numpy matplotlib pyarrow각 라이브러리의 역할은 다음과 같습니다:
- requests: Deribit API에 HTTP 요청을 보내는 도구
- pandas: 테이블 형태의 데이터를 정리하고 분석하는 도구
- numpy: 수학/통계 계산의 기본 엔진
- matplotlib: 차트와 그래프를 그리는 도구
- pyarrow: 대용량 데이터를 효율적으로 저장하는 포맷
테스트: API 연결 확인
import requests
Deribit 공용 API 테스트 - 현재 BTC 선물 가격 조회
url = "https://deribit.com/api/v2/public/get_index"
params = {"currency": "BTC"}
response = requests.get(url, params=params)
print(f"상태 코드: {response.status_code}")
print(f"응답 데이터: {response.json()}")정상 응답 시 {status_code: 200}과 함께 BTC 인덱스 가격이 출력됩니다. 연결 실패 시 뒤에 나오는 오류 해결 섹션을 참고하세요.
2단계: Deribit 옵션 데이터 구조 이해
Deribit의 BTC 옵션은 만기일이 매주 금요일(금요일 BTC 옵션) 또는 매월 마지막 주 금요일(월물 BTC 옵션)로 구성됩니다. 옵션 데이터는 크게 세 가지 유형으로 나뉩니다:
- 옵션 상세 조회: 개별 계약의 현재가, Greeks, 거래량
- 옵션 목록 조회: 특정 만기일的所有 옵션 목록
- 티커 조회: 실시간bid/ask와 내재변동성
Deribit의 옵션 계약명 형식은 다음과 같습니다:
# 계약명 예시 형식
BTC-28MAR25-95000-C # 2025년 3월 28일 만기, 行权가 95,000, 콜 옵션
BTC-28MAR25-95000-P # 2025년 3월 28일 만기, 行权가 95,000, 풋 옵션
Deribit의 실제 계약명 규칙:
{기초자산}-{만기일}-{行权가}-{옵션타입}
만기일 형식: DDMMMYY (예: 28MAR25)
옵션 타입: C(콜) 또는 P(풋)
3단계: 옵션 목록 및 Greeks 데이터 가져오기
활성 옵션 계약 목록 조회
import requests
import pandas as pd
from datetime import datetime, timedelta
class DeribitOptionsData:
"""Deribit BTC 옵션 데이터 수집 클래스"""
BASE_URL = "https://deribit.com/api/v2/public"
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json"
})
def get_option_book(self, instrument_name: str) -> dict:
"""개별 옵션 계약의 상세 정보 조회"""
url = f"{self.BASE_URL}/get_book_summary_by_instrument_name"
params = {"instrument_name": instrument_name}
response = self.session.get(url, params=params)
response.raise_for_status()
return response.json()["result"]
def get_available_options(self, currency: str = "BTC") -> list:
"""거래 가능한 모든 옵션 계약 목록 조회"""
url = f"{self.BASE_URL}/get_instruments"
params = {
"currency": currency,
"kind": "option",
"expired": "false" # 만기되지 않은 옵션만
}
response = self.session.get(url, params=params)
response.raise_for_status()
return response.json()["result"]
사용 예시
client = DeribitOptionsData()
BTC 옵션 목록 가져오기
options = client.get_available_options("BTC")
print(f"총 {len(options)}개의 활성 BTC 옵션 계약")
만기일별 그룹화
expiry_groups = {}
for opt in options:
expiry = opt.get("expiration_timestamp", 0)
date_str = datetime.fromtimestamp(expiry/1000).strftime("%Y-%m-%d")
if date_str not in expiry_groups:
expiry_groups[date_str] = []
expiry_groups[date_str].append(opt["instrument_name"])
print(f"\n만기일별 옵션 수:")
for date in sorted(expiry_groups.keys())[:5]:
print(f" {date}: {len(expiry_groups[date])}개 계약")Greeks 데이터 포함 전체 옵션 데이터 추출
def get_all_options_with_greeks(self, currency: str = "BTC") -> pd.DataFrame:
"""모든 옵션의 Greeks를 포함한 데이터프레임 생성"""
options = self.get_available_options(currency)
records = []
for opt in options:
# 티커 정보에서 Greeks 추출
ticker = self.get_ticker(opt["instrument_name"])
record = {
"instrument_name": opt["instrument_name"],
"option_type": "call" if opt["instrument_name"].endswith("-C") else "put",
"strike": opt["strike"],
"expiry_timestamp": opt["expiration_timestamp"],
"expiry_date": datetime.fromtimestamp(
opt["expiration_timestamp"]/1000
).strftime("%Y-%m-%d"),
# 현재가 및 거래량
"last_price": ticker.get("last_price", 0),
"bid_price": ticker.get("best_bid_price", 0),
"ask_price": ticker.get("best_ask_price", 0),
"volume": ticker.get("volume", 0),
"open_interest": ticker.get("open_interest", 0),
# Greeks (Deribit 제공)
"greeks": ticker.get("greeks", {}),
# 내재변동성
"mark_iv": ticker.get("mark_iv", 0), # 중립 변동성
"bid_iv": ticker.get("best_bid_iv", 0),
"ask_iv": ticker.get("best_ask_iv", 0),
}
# Greeks 사전에서 개별값 추출
greeks = record["greeks"]
record["delta"] = greeks.get("delta", 0)
record["gamma"] = greeks.get("gamma", 0)
record["vega"] = greeks.get("vega", 0)
record["theta"] = greeks.get("theta", 0)
record["rho"] = greeks.get("rho", 0)
records.append(record)
return pd.DataFrame(records)
def get_ticker(self, instrument_name: str) -> dict:
"""실시간 티커 정보 조회"""
url = f"{self.BASE_URL}/get_ticker"
params = {"instrument_name": instrument_name}
response = self.session.get(url, params=params)
response.raise_for_status()
return response.json()["result"]
Greeks 데이터 추출 예시
client = DeribitOptionsData()
df = client.get_all_options_with_greeks("BTC")
print("BTC 옵션 Greeks 데이터 샘플:")
print(df[["instrument_name", "strike", "delta", "gamma", "vega", "theta", "mark_iv"]].head(10).to_string())
만기별 내재변동성 스마일 확인
print("\n만기별 IV 스마일 (ATM 근처):")
for expiry in df["expiry_date"].unique()[:2]:
expiry_data = df[df["expiry_date"] == expiry].copy()
expiry_data = expiry_data.sort_values("strike")
atm_strike = expiry_data.loc[expiry_data["option_type"] == "call", "strike"].iloc[
len(expiry_data)//2
] if len(expiry_data) > 0 else 0
atm_data = expiry_data[
(expiry_data["strike"] >= atm_strike * 0.95) &
(expiry_data["strike"] <= atm_strike * 1.05)
]
print(f"\n만기: {expiry}")
print(atm_data[["strike", "option_type", "mark_iv"]].to_string())4단계: 역사적 데이터 다운로드 파이프라인
거래소 거래량 데이터 조회
import time
from datetime import datetime, timedelta
class HistoricalOptionsPipeline:
"""BTC 옵션 역사적 데이터 수집 파이프라인"""
BASE_URL = "https://deribit.com/api/v2/public"
def __init__(self):
self.session = requests.Session()
def get_trade_volumes(self, instrument_name: str,
start_timestamp: int,
end_timestamp: int,
timeframe: str = "1h") -> list:
"""특정 옵션의 과거 거래량 데이터 조회
timeframe: 1m, 5m, 1h, 1D
"""
url = f"{self.BASE_URL}/get_trade_volumes_by_instrument_name"
params = {
"instrument_name": instrument_name,
"start_timestamp": start_timestamp,
"end_timestamp": end_timestamp,
"resolution": timeframe
}
response = self.session.get(url, params=params)
if response.status_code != 200:
return []
result = response.json().get("result", {})
return result.get("volumes", [])
def download_historical_data(self, currency: str = "BTC",
days_back: int = 30) -> pd.DataFrame:
"""최근 N일간 BTC 옵션 데이터 다운로드"""
# 전체 옵션 목록 가져오기
url = f"{self.BASE_URL}/get_instruments"
params = {"currency": currency, "kind": "option", "expired": "false"}
response = self.session.get(url, params=params)
instruments = response.json()["result"]
end_time = int(datetime.now().timestamp() * 1000)
start_time = int((datetime.now() - timedelta(days=days_back)).timestamp() * 1000)
all_data = []
for idx, instr in enumerate(instruments):
try:
instrument_name = instr["instrument_name"]
# 각 옵션의 거래량 데이터
volumes = self.get_trade_volumes(
instrument_name, start_time, end_time, "1D"
)
for vol in volumes:
all_data.append({
"timestamp": vol.get("timestamp", 0),
"date": datetime.fromtimestamp(
vol.get("timestamp", 0)/1000
).strftime("%Y-%m-%d"),
"instrument_name": instrument_name,
"strike": instr["strike"],
"option_type": "call" if instrument_name.endswith("-C") else "put",
"expiry": datetime.fromtimestamp(
instr["expiration_timestamp"]/1000
).strftime("%Y-%m-%d"),
"volume": vol.get("volume", 0),
"tick_volume": vol.get("tick_volume", 0),
"notional": vol.get("notional", 0),
})
# API 속도 제한 방지 (요청 간 100ms 대기)
time.sleep(0.1)
if (idx + 1) % 20 == 0:
print(f"진행률: {idx+1}/{len(instruments)} 계약 완료")
except Exception as e:
print(f"오류 ({instr.get('instrument_name', 'unknown')}): {e}")
continue
df = pd.DataFrame(all_data)
print(f"\n총 {len(df)}건의 역사적 데이터 수집 완료")
return df
사용 예시 - 최근 30일 데이터 다운로드
pipeline = HistoricalOptionsPipeline()
historical_df = pipeline.download_historical_data("BTC", days_back=30)
데이터 저장
historical_df.to_parquet("btc_options_historical.parquet", index=False)
print(f"데이터 저장 완료: btc_options_historical.parquet")5단계: 내재변동성(IV) 스마일 분석
import matplotlib.pyplot as plt
import numpy as np
def calculate_iv_smile(df: pd.DataFrame, expiry_date: str) -> pd.DataFrame:
"""특정 만기일의 IV 스마일 계산"""
expiry_data = df[df["expiry_date"] == expiry_date].copy()
# ATM 옵션 근처 데이터만 필터링 (BTC 현재가 대비 ±20% 범위)
btc_price = expiry_data["strike"].median() # 중앙값을 대리 현재가로 사용
expiry_data = expiry_data[
(expiry_data["strike"] >= btc_price * 0.8) &
(expiry_data["strike"] <= btc_price * 1.2)
]
expiry_data = expiry_data.sort_values("strike")
return expiry_data
def plot_iv_smile(df: pd.DataFrame, expiry_dates: list):
"""여러 만기일의 IV 스마일 시각화"""
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
axes = axes.flatten()
for idx, expiry in enumerate(expiry_dates[:4]):
ax = axes[idx]
smile_data = calculate_iv_smile(df, expiry)
if smile_data.empty:
continue
# 콜/풋 IV 분리
calls = smile_data[smile_data["option_type"] == "call"]
puts = smile_data[smile_data["option_type"] == "put"]
ax.plot(calls["strike"], calls["mark_iv"] * 100, 'b-o',
label="콜 IV", markersize=4)
ax.plot(puts["strike"], puts["mark_iv"] * 100, 'r-o',
label="풋 IV", markersize=4)
ax.set_xlabel("行权가 (USD)")
ax.set_ylabel("내재변동성 (%)")
ax.set_title(f"IV 스마일 - {expiry}")
ax.legend()
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig("iv_smile_analysis.png", dpi=150)
plt.show()
print("IV 스마일 차트 저장: iv_smile_analysis.png")
실제 데이터로 IV 스마일 플롯
(실제 데이터가 있는 경우 아래 코드를 실행하세요)
unique_expiries = historical_df["expiry"].unique()[:4]
plot_iv_smile(historical_df, unique_expiries)
print("IV 스마일 시각화 준비 완료")
print("실제 데이터로 차트를 그리려면 위 코드의 주석을 해제하세요")6단계: Greeks 기반风险管理 백테스트 프레임워크
class OptionsBacktester:
"""Greeks 기반 옵션风险管理 백테스터"""
def __init__(self, data: pd.DataFrame):
self.data = data.copy()
self.results = []
def calculate_portfolio_greeks(self, positions: list) -> dict:
"""포트폴리오 전체 Greeks 계산
positions: [{"instrument": "BTC-28MAR25-95000-C", "size": 1.5}, ...]
"""
portfolio = {
"delta": 0, "gamma": 0, "vega": 0, "theta": 0,
"position_count": 0, "total_notional": 0
}
for pos in positions:
inst_name = pos["instrument"]
size = pos["size"]
opt_data = self.data[self.data["instrument_name"] == inst_name]
if opt_data.empty:
continue
row = opt_data.iloc[-1]
# 각 Greeks에 포지션 사이즈 반영
portfolio["delta"] += row["delta"] * size
portfolio["gamma"] += row["gamma"] * size
portfolio["vega"] += row["vega"] * size
portfolio["theta"] += row["theta"] * size
portfolio["position_count"] += 1
portfolio["total_notional"] += row["open_interest"] * size
return portfolio
def delta_hedge_simulation(self, initial_price: float,
target_delta: float = 0) -> pd.DataFrame:
"""Delta 헤지 시뮬레이션 - BTC 가격 변동에 따른 델타 조정"""
price_range = np.linspace(initial_price * 0.9, initial_price * 1.1, 21)
hedge_results = []
for price in price_range:
# 단순화된 Delta 헤지 시뮬레이션
price_change = (price - initial_price) / initial_price
pnl_from_delta = target_delta * price_change * initial_price
hedge_results.append({
"btc_price": price,
"price_change_pct": price_change * 100,
"pnl_delta_hedge": pnl_from_delta,
})
return pd.DataFrame(hedge_results)
def run_risk_report(self, positions: list) -> dict:
"""风险管理 리포트 생성"""
greeks = self.calculate_portfolio_greeks(positions)
# VaR (Value at Risk) 근사 계산 (단순화 버전)
# 실제 VaR에는 몬테카를로 시뮬레이션 필요
volatility_estimate = 0.80 # BTC 연환 기준 변동성 80%
confidence_level = 0.95
z_score = 1.645 # 95% 신뢰구간
var_estimate = (
greeks["total_notional"] *
volatility_estimate * z_score / np.sqrt(252)
)
report = {
"portfolio_greeks": greeks,
"estimated_var_1day_95": var_estimate,
"delta_exposure": greeks["delta"],
"gamma_exposure": greeks["gamma"],
"vega_exposure": greeks["vega"], # 1% IV 변동당 손익
"theta_daily_decay": greeks["theta"], # 일일 시간 가치 소멸
"position_count": greeks["position_count"],
"risk_recommendation": self._generate_risk_recommendation(greeks)
}
return report
def _generate_risk_recommendation(self, greeks: dict) -> str:
"""Greeks 기반 위험 조정 권고사항"""
recommendations = []
if abs(greeks["delta"]) > 10:
recommendations.append("⚠️ 델타 노출이 높습니다. 현물 또는 선물 포지션으로 델타 중립 고려")
if abs(greeks["gamma"]) > 0.1:
recommendations.append("⚠️ 감마 위험이 높습니다. 강세 변동 시大口 손실 가능성")
if abs(greeks["vega"]) > 1000:
recommendations.append("ℹ️ 베가 노출 확인: IV 1% 변동당 약 $1,000 손익")
if greeks["theta"] < -500:
recommendations.append("ℹ️ 세타 소멸이 큽니다: 일일 약 $500 시간 가치 감소")
if not recommendations:
recommendations.append("✅ 포트폴리오 Greeks가 균형 상태입니다")
return "\n".join(recommendations)
사용 예시
backtester = OptionsBacktester(historical_df)
샘플 포트폴리오 리포트
sample_positions = [
{"instrument": "BTC-28MAR25-95000-C", "size": 1.5},
{"instrument": "BTC-28MAR25-90000-P", "size": -1.0},
]
report = backtester.run_risk_report(sample_positions)
print("===风险管理 백테스트 리포트===")
print(f"1일 VaR (95%): ${report['estimated_var_1day_95']:,.2f}")
print(f"델타 노출: {report['delta_exposure']:.4f}")
print(f"베가 노출: {report['vega_exposure']:.4f}")
print(f"일일 세타 소멸: ${report['theta_daily_decay']:.2f}")
print(f"\n권고사항:\n{report['risk_recommendation']}")7단계: 데이터 저장 및 자동화
import pyarrow as pa
import pyarrow.parquet as pq
from pathlib import Path
from datetime import datetime
class OptionsDataStorage:
"""옵션 데이터 영구 저장 및 관리"""
def __init__(self, storage_dir: str = "./options_data"):
self.storage_dir = Path(storage_dir)
self.storage_dir.mkdir(exist_ok=True)
def save_daily_snapshot(self, df: pd.DataFrame, date: str = None):
"""일일 스냅샷 저장 - Parquet 포맷으로 효율적 저장"""
if date is None:
date = datetime.now().strftime("%Y-%m-%d")
filepath = self.storage_dir / f"options_snapshot_{date}.parquet"
# Parquet 파일로 저장 (CSV 대비 5-10배 효율적)
df.to_parquet(filepath, index=False, engine="pyarrow")
print(f"✓ {date} 스냅샷 저장 완료: {filepath}")
print(f" 파일 크기: {filepath.stat().st_size / 1024:.1f} KB")
print(f" 레코드 수: {len(df):,}")
def load_snapshot(self, date: str) -> pd.DataFrame:
"""특정 날짜 스냅샷 로드"""
filepath = self.storage_dir / f"options_snapshot_{date}.parquet"
if not filepath.exists():
raise FileNotFoundError(f"{date} 스냅샷이 존재하지 않습니다")
return pd.read_parquet(filepath)
def load_date_range(self, start_date: str, end_date: str) -> pd.DataFrame:
"""날짜 범위 내 모든 스냅샷 통합 로드"""
snapshots = []
for snapshot_file in sorted(self.storage_dir.glob("options_snapshot_*.parquet")):
file_date = snapshot_file.stem.split("_")[-1]
if start_date <= file_date <= end_date:
df = pd.read_parquet(snapshot_file)
snapshots.append(df)
if not snapshots:
return pd.DataFrame()
combined = pd.concat(snapshots, ignore_index=True)
print(f"✓ {start_date} ~ {end_date} 기간 데이터 로드: {len(combined):,}건")
return combined
저장 및 로드 예시
storage = OptionsDataStorage("./options_data")
오늘 스냅샷 저장
storage.save_daily_snapshot(historical_df)
최근 7일 데이터 로드
recent_data = storage.load_date_range(
(datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d"),
datetime.now().strftime("%Y-%m-%d")
)
print(f"로드된 데이터: {len(recent_data)}건")Deribit API 레이트 리밋 및 최적화
Deribit API는 분당 요청 수에 제한이 있습니다. 대량의 데이터를 수집할 때는 다음 전략을 사용하세요:
- 요청 간격: 공용 API는 분당 60-120회 제한, 1초 이상 간격 권장
- 배치 요청: 가능하면 개별 요청보다 목록 조회 활용
- 캐싱: 자주 변하지 않는 데이터(옵션 목록 등)는 메모리에 캐싱
- 실시간 모니터링: WebSocket 사용으로 폴링 방식 피하기
자주 발생하는 오류와 해결책
1. API 연결 타임아웃 오류
# 문제: requests.get() 시 ConnectionError 또는 Timeout
원인: 네트워크 불안정 또는 Deribit 서버 과부하
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_robust_session() -> requests.Session:
"""재시도 로직이 포함된 세션 생성"""
session = requests.Session()
# 최대 3번 재시도, 지수 백오프 적용
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
# 타임아웃 설정 (연결 10초, 읽기 30초)
session.timeout = (10, 30)
return session
사용
session = create_robust_session()
response = session.get("https://deribit.com/api/v2/public/get_index",
params={"currency": "BTC"})2. Greeks 값이 None으로 반환되는 경우
# 문제: ticker["greeks"]가 None 또는 키가 존재하지 않음
원인: 옵션의流动性 부족 또는 시장 데이터 딜레이
def safe_get_greeks(ticker: dict, default: float = 0.0) -> dict:
"""Greeks 값 안전하게 추출 (None 방지)"""
greeks = ticker.get("greeks", {}) or {}
return {
"delta": greeks.get("delta", default),
"gamma": greeks.get("gamma", default),
"vega": greeks.get("vega", default),
"theta": greeks.get("theta", default),
"rho": greeks.get("rho", default),
# Deribit 특이 필드
"bid_iv": ticker.get("best_bid_iv", default),
"ask_iv": ticker.get("best_ask_iv", default),
"mark_iv": ticker.get("mark_iv", default),
}
사용
ticker = client.get_ticker("BTC-28MAR25-95000-C")
greeks = safe_get_greeks(ticker)
print(f"Delta: {greeks['delta']}, Vega: {greeks['vega']}")3. 만기일별 옵션 목록이 비어있는 경우
# 문제: get_instruments가 빈 리스트 반환
원인: expired 파라미터 설정 오류 또는 currency 불일치
def get_all_active_options(currency: str = "BTC") -> list:
"""만기되지 않은 옵션만 정확히 조회"""
url = "https://deribit.com/api/v2/public/get_instruments"
# expired=false가 문자열而非 불리언으로 전달되어야 함
params = {
"currency": currency,
"kind": "option",
"expired": "false" # 반드시 문자열 "false"
}
response = requests.get(url, params=params)
result = response.json()
if "result" not in result:
print(f"API 오류: {result}")
return []
instruments = result["result"]
# 필터링: 만기일이 현재 이후인 옵션만
now_ms = datetime.now().timestamp() * 1000
active = [i for i in instruments if i.get("expiration_timestamp", 0) > now_ms]
print(f"전체 계약: {len(instruments)}, 활성 계약: {len(active)}")
return active
직접 API 호출 테스트
options = get_all_active_options("BTC")
if not options:
print("⚠️ 옵션 목록이 비었습니다. currency 파라미터를 'BTC' 또는 'ETH'로 확인하세요")4. 저장된 Parquet 파일 로드 오류
# 문제: pd.read_parquet() 시 FileNotFoundError 또는 파싱 오류
원인: 파일 경로 오류 또는 다른 엔진으로 저장된 파일
import os
from pathlib import Path
def safe_load_parquet(filepath: str) -> pd.DataFrame:
"""안전한 Parquet 파일 로드"""
path = Path(filepath)
# 경로 존재 확인
if not path.exists():
# 상위 디렉토리에서 파일 검색
print(f"파일 없음: {filepath}")
print("현재 디렉토리 내 Parquet 파일:")
for f in Path(".").rglob("*.parquet"):
print(f" - {f}")
raise FileNotFoundError(f"파일을 찾을 수 없습니다: {filepath}")
# 파일 크기 확인
size_mb = path.stat().st_size / (1024 * 1024)
print(f"파일 크기: {size_mb:.2f} MB")
# pandas로 로드 시도
try:
return pd.read_parquet(filepath)
except Exception as e:
print(f"pandas 로드 실패: {e}")
print("pyarrow로 대체 시도...")
# pyarrow로 직접 읽기
import pyarrow.parquet as pq
table = pq.read_table(filepath)
return table.to_pandas()
사용
try:
df = safe_load_parquet("options_data/options_snapshot_2025-03-28.parquet")
print(f"로드 성공: {len(df)}건")
except FileNotFoundError as e:
print(f"로드 실패: {e}")전체 데이터 파이프라인 통합 예제
"""
BTC 옵션 데이터 파이프라인 - 매일 자동 실행용 스クリ프트
crontab에 등록하여 매일 실행:
0 2 * * * python /path/to/options_pipeline.py
"""
import logging
from datetime import datetime, timedelta
import json
로깅 설정
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.FileHandler("options_pipeline.log"),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
def main():
"""일일 데이터 수집 파이프라인 메인 함수"""
logger.info("=== BTC 옵션 데이터 수집 시작 ===")
# 1. Deribit 연결 및 데이터 수집
client = DeribitOptionsData()
pipeline = HistoricalOptionsPipeline()
storage = OptionsDataStorage()
try:
# 2. Greeks 포함 전체 옵션 데이터
logger.info("Greeks 데이터 수집 중...")
df = client.get_all_options_with_greeks("BTC")
logger.info(f"Greeks 데이터: {len(df)}개 계약")
# 3. 최근 30일 거래량 히스토리
logger.info("역사적 거래량 데이터 수집 중...")
historical = pipeline.download_historical_data("BTC", days_back=30)
# 4. 데이터 병합 및 저장
combined = pd.concat([df, historical], ignore_index=True)
today = datetime.now().strftime("%Y-%m-%d")
storage.save_daily_snapshot(combined, today)
# 5.风险管理 리포트 생성
backtester = OptionsBacktester(combined)
report = backtester.run_risk_report([])
logger.info(f"VaR 1일: ${report['estimated_var_1day_95']:,.2f}")
logger.info("=== 수집 완료 ===")
except Exception as e:
logger.error(f"파이프라인 오류: {e}", exc_info=True)
if __name__ == "__main__":
main()Deribit 옵션 데이터와 HolySheep AI의 활용
Deribit 옵션 데이터 수집을 자동화했다면, 다음 단계로는 AI 기반 시장 분석 시스템 구축이 있습니다. HolySheep AI는 Deribit API로 수집한 옵션 Greeks 데이터를 분석하고, 자동化された 리포트 생성과 시장 예측에 활용할 수 있습니다:
- 변동성 예측: DeepSeek V3 모델로 IV 스마일 패턴 분석
- 리스크 보고 자동화: Claude Sonnet으로风险管理 보고서 생성
- 뉴스 감성 분석: BTC 관련 뉴스와 옵션 데이터 상관관계 분석
- 가격 최적화: Gemini Flash로 실시간 시장 상황 요약
# HolySheep AI를 활용한 옵션 리포트 자동 생성 예시
import openai
HolySheep AI Gateway 설정