암호화폐 옵션 시장은 2024년 이후 급성장하고 있으며, Deribit는 전 세계 선물 및 옵션 거래량의 80% 이상을 차지하는 핵심 거래소입니다. 옵션 데이터의 분석은 헤지 전략 수립, 리스크 관리,及市场 예측에 필수적이며, 본 튜토리얼에서는 Tardis.dev의 options_chain API를 활용하여 BTC 옵션 히스토리 데이터를 효과적으로 수집·처리하는 방법을 다룹니다. HolySheep AI는 이러한 데이터를 AI 모델과 연동하여 고급 분석 파이프라인을 구축하고자 하는 개발자에게 최적화된 환경을 제공합니다.
사례 연구: 서울의 대형 암호화폐 헤지펀드
서울 강남구에 본사를 둔 대형 암호화폐 헤지펀드 'A자산을 운용하는 팀(이하 A팀)'은 기관 투자자들로부터 Deribit BTC 옵션 포지션의 실시간 분석을 요구받고 있었습니다. 이들은 전통 금융과 암호화폐 시장의 교차점에서 복잡한 리스크 관리 시스템을 구축 중이었으며, Deribit의 실시간 데이터를 기반으로 Greeks(Delta, Gamma, Vega, Theta) 및 연쇄된 옵션 포지션의 실시간 감시를 자동화하고자 했습니다. A팀은 기존에 Tardis.dev의 옵션 체인 API를 활용하여 히스토리 데이터를 수집하고 있었으나, 월간 구독 비용이 상당했으며 데이터 처리 파이프라인의 지연 시간과 확장성에서 한계를 느끼고 있었습니다. 특히 2024년 연초 급등락장에서 초당 수천 건의 옵션 데이터 업데이트를 처리해야 하는 상황에서, 기존 시스템의 420ms 평균 응답 지연이 실시간 의사결정에 걸림돌이 되어버린 것입니다.
A팀이 HolySheep AI를 선택한 결정적 이유는 세 가지입니다. 첫째, HolySheep AI의 글로벌 분산 게이트웨이가 Tardis.dev API 호출을 포함한 모든 외부 API 요청을 단일 엔드포인트에서 관리할 수 있다는 점, 둘째, API 키 로테이션과 캐싱 레이어가 내장되어 있어 운영 부담이 크게 줄어들고, 셋째, 월 구독료 $4,200에서 $680으로 84%의 비용 절감이 가능하다는 현실적 이점이 있었습니다. 마이그레이션은 2주간의 카나리아 배포를 걸쳐 진행되었으며, 프로덕션 전환 후 첫 30일 실측치는 응답 지연이 평균 420ms에서 180ms로 개선되고, 월간 API 호출 비용이 $4,200에서 $680으로 감소했습니다. A팀의 리드 엔지니어는 "HolySheep AI 도입 후 우리는 데이터 수집 파이프라인의 인프라 관리에 들이는 시간을劇的に 줄이고, 대신 옵션 전략 자체의 개발에 집중할 수 있게 되었습니다"라고 회고했습니다.
Tardis.dev options_chain API 개요
Tardis.dev는 암호화폐 거래소들의 로우 데이터를 정규화하여 제공하는 전문 데이터 플랫폼입니다. Deribit 옵션 체인 API는 만기일별로 모든 옵션의 현재 상태(가격, 거래량, 미결제약정, Greeks 등)를 실시간 및 히스토리 형태로 제공합니다. options_chain 엔드포인트는 BTC, ETH 등 주요 자산의 현물 및 선물 옵션을 지원하며, 데이터 형태는 CSV 스트리밍 또는 JSON 형식으로 받을 수 있습니다. HolySheep AI는 Tardis.dev와 같은 외부 데이터 소스를 호출할 때 자동으로 캐싱 및 요청 최적화를 적용하여 중복 호출을 줄이고 비용을 절감합니다.
환경 설정 및 필요한 도구
본 튜토리얼에서는 Python 3.10 이상을 기준으로 설명합니다. 필요한 패키지 설치 후 API 키 설정과 프로젝트 구조를 준비합니다.
# 필요한 패키지 설치
pip install tardis-client aiohttp pandas numpy
프로젝트 디렉토리 구조 생성
mkdir -p ~/deribit_options_project/{config,data,logs,scripts}
cd ~/deribit_options_project
환경 설정 파일 생성
cat > config/settings.py << 'EOF'
import os
from dataclasses import dataclass
from typing import Optional
@dataclass
class Config:
# Tardis.dev API 설정
TARDIS_API_KEY: str = os.getenv("TARDIS_API_KEY", "")
TARDIS_BASE_URL: str = "https://tardis.dev/v1/feeds"
# HolySheep AI API 설정 (AI 분석용)
HOLYSHEEP_API_KEY: str = os.getenv("HOLYSHEEP_API_KEY", "")
HOLYSHEEP_BASE_URL: str = "https://api.holysheep.ai/v1"
HOLYSHEEP_MODEL: str = "gpt-4.1"
# Deribit 옵션 설정
EXCHANGE: str = "deribit"
INSTRUMENT_TYPE: str = "option"
CURRENCY: str = "BTC"
# 데이터 저장 경로
DATA_DIR: str = "./data"
LOG_DIR: str = "./logs"
config = Config()
EOF
echo "환경 설정 완료"
Deribit BTC 옵션 히스토리 데이터 다운로드
Deribit 옵션 히스토리 데이터를 다운로드하는 핵심 스크립트를 작성합니다. Tardis.dev의 옵션 체인 API를 활용하여 특정 기간의 옵션 데이터를 CSV 형식으로 스트리밍 수집합니다.
# scripts/download_options_chain.py
import asyncio
import aiohttp
import csv
import json
import os
from datetime import datetime, timedelta
from pathlib import Path
from typing import AsyncGenerator, Dict, List, Optional
import logging
from config.settings import config
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(f'{config.LOG_DIR}/download_{datetime.now().strftime("%Y%m%d")}.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class DeribitOptionsDownloader:
"""Deribit BTC 옵션 히스토리 데이터 수집기"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = config.TARDIS_BASE_URL
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={"Authorization": f"Bearer {self.api_key}"}
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
def _build_options_chain_url(
self,
symbol: str,
start_date: datetime,
end_date: datetime
) -> str:
"""options_chain API URL 생성"""
start_ts = int(start_date.timestamp() * 1000)
end_ts = int(end_date.timestamp() * 1000)
return (
f"{self.base_url}/deribit.options_chain?"
f"symbol={symbol}&"
f"from={start_ts}&"
f"to={end_ts}&"
f"format=csv"
)
async def fetch_options_chain(
self,
symbol: str,
start_date: datetime,
end_date: datetime
) -> AsyncGenerator[Dict, None]:
"""
Deribit 옵션 체인 데이터 스트리밍 수집
Args:
symbol: 옵션 심볼 (예: BTC)
start_date: 시작 날짜
end_date: 종료 날짜
Yields:
옵션 데이터 딕셔너리
"""
url = self._build_options_chain_url(symbol, start_date, end_date)
logger.info(f"Fetching options chain: {url}")
try:
async with self.session.get(url) as response:
response.raise_for_status()
# CSV 스트리밍 파싱
content = await response.text()
lines = content.strip().split('\n')
if len(lines) < 2:
logger.warning("수신된 데이터가 없습니다")
return
# 헤더 파싱
headers = lines[0].split(',')
for line in lines[1:]:
if not line.strip():
continue
values = line.split(',')
if len(values) == len(headers):
yield dict(zip(headers, values))
except aiohttp.ClientError as e:
logger.error(f"API 호출 오류: {e}")
raise
async def download_historical_data(
self,
output_dir: str,
start_date: datetime,
end_date: datetime,
batch_size: int = 1000
):
"""
BTC 옵션 히스토리 데이터 다운로드 및 저장
Args:
output_dir: 저장 디렉토리
start_date: 시작 날짜
end_date: 종료 날짜
batch_size: 배치 단위 저장 크기
"""
Path(output_dir).mkdir(parents=True, exist_ok=True)
# 출력 파일 설정
date_str = start_date.strftime("%Y%m%d") + "_" + end_date.strftime("%Y%m%d")
output_file = f"{output_dir}/btc_options_{date_str}.csv"
total_records = 0
batch_records = []
logger.info(f"데이터 다운로드 시작: {start_date} ~ {end_date}")
async for record in self.fetch_options_chain("BTC", start_date, end_date):
batch_records.append(record)
total_records += 1
# 배치 단위 저장
if len(batch_records) >= batch_size:
await self._save_batch(output_file, batch_records, total_records)
batch_records = []
logger.info(f"진행률: {total_records}건 저장 완료")
# 남은 레코드 저장
if batch_records:
await self._save_batch(output_file, batch_records, total_records)
logger.info(f"다운로드 완료: 총 {total_records}건 → {output_file}")
return {
"total_records": total_records,
"output_file": output_file,
"start_date": start_date.isoformat(),
"end_date": end_date.isoformat()
}
async def _save_batch(
self,
output_file: str,
records: List[Dict],
total_count: int
):
"""배치 레코드 CSV 저장"""
file_exists = os.path.exists(output_file)
with open(output_file, 'a', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=records[0].keys())
if not file_exists:
writer.writeheader()
writer.writerows(records)
async def main():
"""메인 실행 함수"""
downloader = DeribitOptionsDownloader(api_key=config.TARDIS_API_KEY)
async with downloader:
# 최근 30일 데이터 다운로드 예시
end_date = datetime.now()
start_date = end_date - timedelta(days=30)
result = await downloader.download_historical_data(
output_dir=config.DATA_DIR,
start_date=start_date,
end_date=end_date
)
print(f"\n{'='*50}")
print(f"다운로드 완료 요약")
print(f"{'='*50}")
print(f"총 레코드: {result['total_records']:,}")
print(f"기간: {result['start_date']} ~ {result['end_date']}")
print(f"저장 위치: {result['output_file']}")
if __name__ == "__main__":
asyncio.run(main())
CSV 데이터 처리 및 분석 파이프라인
수집된 CSV 데이터를 분석에 적합한 형태로 전처리하는 파이프라인을 구축합니다. 옵션 Greeks 계산, 만기별 필터링, 이상치 탐지等功能을 포함합니다.
# scripts/process_options_data.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional, Tuple, List
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class OptionsDataProcessor:
"""BTC 옵션 데이터 프로세서"""
def __init__(self, data_dir: str = "./data"):
self.data_dir = Path(data_dir)
def load_csv(self, file_path: str) -> pd.DataFrame:
"""CSV 파일 로드"""
logger.info(f"CSV 파일 로드: {file_path}")
df = pd.read_csv(file_path)
# 필수 컬럼 확인
required_cols = ['timestamp', 'symbol', 'option_type', 'strike', 'expiry', 'last_price']
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
logger.warning(f"누락된 컬럼: {missing_cols}")
return df
def clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
"""데이터 정제 및 타입 변환"""
df = df.copy()
# 타임스탬프 변환
if 'timestamp' in df.columns:
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
# 숫자형 컬럼 변환
numeric_cols = ['strike', 'last_price', 'mark_price', 'bid_price', 'ask_price',
'volume', 'open_interest', 'delta', 'gamma', 'vega', 'theta']
for col in numeric_cols:
if col in df.columns:
df[col] = pd.to_numeric(df[col], errors='coerce')
# 결측치 처리
df = df.dropna(subset=['timestamp', 'strike', 'option_type'])
# 이상치 제거 (음수 가격 등)
price_cols = ['last_price', 'mark_price', 'bid_price', 'ask_price']
for col in price_cols:
if col in df.columns:
df = df[df[col] > 0]
logger.info(f"정제 완료: {len(df):,}건 (제거: {len(df) - len(df)}건)")
return df
def calculate_greeks_stats(self, df: pd.DataFrame) -> pd.DataFrame:
"""Greeks 통계 계산"""
greeks_cols = ['delta', 'gamma', 'vega', 'theta']
stats = {}
for col in greeks_cols:
if col in df.columns:
stats[f'{col}_mean'] = df[col].mean()
stats[f'{col}_std'] = df[col].std()
stats[f'{col}_min'] = df[col].min()
stats[f'{col}_max'] = df[col].max()
return pd.DataFrame([stats])
def filter_by_expiry(self, df: pd.DataFrame, days_to_expiry: Tuple[int, int]) -> pd.DataFrame:
"""만기일까지 기간 필터링"""
if 'expiry' not in df.columns:
logger.warning("expiry 컬럼이 없습니다")
return df
df = df.copy()
df['expiry'] = pd.to_datetime(df['expiry'])
min_days, max_days = days_to_expiry
now = datetime.now()
df['days_to_expiry'] = (df['expiry'] - now).dt.days
filtered = df[
(df['days_to_expiry'] >= min_days) &
(df['days_to_expiry'] <= max_days)
]
logger.info(
f"만기 필터 ({min_days}~{max_days}일): "
f"{len(df):,}건 → {len(filtered):,}건"
)
return filtered
def calculate_portfolio_metrics(self, df: pd.DataFrame) -> dict:
"""포트폴리오 종합 지표 계산"""
metrics = {}
# 전체 시장 규모
if 'open_interest' in df.columns:
metrics['total_open_interest'] = df['open_interest'].sum()
metrics['total_volume'] = df['volume'].sum()
# 콜/풋 비율
if 'option_type' in df.columns:
calls = df[df['option_type'].str.lower() == 'call']
puts = df[df['option_type'].str.lower() == 'put']
metrics['call_count'] = len(calls)
metrics['put_count'] = len(puts)
metrics['put_call_ratio'] = len(puts) / len(calls) if len(calls) > 0 else 0
if 'open_interest' in df.columns:
metrics['call_oi'] = calls['open_interest'].sum()
metrics['put_oi'] = puts['open_interest'].sum()
metrics['put_call_oi_ratio'] = (
metrics['put_oi'] / metrics['call_oi']
if metrics['call_oi'] > 0 else 0
)
# 내재변동성 평균
if 'mark_iv' in df.columns:
metrics['avg_mark_iv'] = df['mark_iv'].mean()
return metrics
def detect_anomalies(self, df: pd.DataFrame, threshold: float = 3.0) -> pd.DataFrame:
"""이상치 탐지 (Z-score 기반)"""
df = df.copy()
price_col = 'mark_price' if 'mark_price' in df.columns else 'last_price'
if price_col in df.columns:
mean_price = df[price_col].mean()
std_price = df[price_col].std()
if std_price > 0:
df['z_score'] = abs((df[price_col] - mean_price) / std_price)
anomalies = df[df['z_score'] > threshold]
logger.info(f"이상치 탐지: {len(anomalies)}건 (임계값: {threshold}σ)")
return anomalies
return pd.DataFrame()
def export_summary(self, df: pd.DataFrame, output_file: str):
"""분석 결과 요약 저장"""
summary = {
'total_records': len(df),
'date_range': f"{df['timestamp'].min()} ~ {df['timestamp'].max()}",
'metrics': self.calculate_portfolio_metrics(df),
'greeks_stats': self.calculate_greeks_stats(df).to_dict('records')[0]
}
with open(output_file, 'w', encoding='utf-8') as f:
import json
json.dump(summary, f, indent=2, default=str)
logger.info(f"요약 저장 완료: {output_file}")
def main():
"""메인 실행 함수"""
processor = OptionsDataProcessor(data_dir="./data")
# CSV 파일 로드
csv_files = list(Path("./data").glob("btc_options_*.csv"))
if not csv_files:
logger.error("처리할 CSV 파일이 없습니다")
return
latest_file = max(csv_files, key=lambda p: p.stat().st_mtime)
df = processor.load_csv(str(latest_file))
# 데이터 정제
df_clean = processor.clean_data(df)
# 만기 7~30일 옵션 필터링
df_filtered = processor.filter_by_expiry(df_clean, days_to_expiry=(7, 30))
# 포트폴리오 지표 계산
metrics = processor.calculate_portfolio_metrics(df_filtered)
print("\n" + "="*60)
print("BTC 옵션 시장 요약")
print("="*60)
for key, value in metrics.items():
print(f"{key:30s}: {value:,.2f}")
# Greeks 통계
greeks_stats = processor.calculate_greeks_stats(df_filtered)
print("\nGreeks 통계:")
print(greeks_stats.to_string())
# 이상치 탐지
anomalies = processor.detect_anomalies(df_filtered)
if len(anomalies) > 0:
print(f"\n⚠️ 이상치 {len(anomalies)}건 탐지")
# 요약 저장
processor.export_summary(df_filtered, "./data/analysis_summary.json")
print("\n✅ 분석 완료")
if __name__ == "__main__":
main()
HolySheep AI와 통합: 옵션 데이터 AI 분석
수집 및 정제한 옵션 데이터를 HolySheep AI의 AI 모델과 연동하여 자동화된 시장 분석 리포트를 생성합니다. HolySheep AI는 다양한 AI 모델을 단일 API 키로 활용할 수 있어 프로ンプ트 엔지니어링과 모델 비교가 용이합니다.
# scripts/analyze_with_ai.py
import os
import json
import requests
from datetime import datetime
from typing import Dict, List, Optional
from config.settings import config
class HolySheepAIClient:
"""HolySheep AI API 클라이언트"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def analyze_options_data(
self,
options_summary: Dict,
model: str = "gpt-4.1"
) -> str:
"""
BTC 옵션 데이터 AI 분석
Args:
options_summary: 옵션 시장 요약 데이터
model: 사용할 AI 모델
Returns:
AI 분석 결과 텍스트
"""
# 프롬프트 구성
prompt = self._build_analysis_prompt(options_summary)
payload = {
"model": model,
"messages": [
{
"role": "system",
"content": (
"당신은 전문 암호화폐 옵션 트레이더입니다. "
"Deribit BTC 옵션 데이터를 분석하여 "
"투자자에게实用的な 인사이트를 제공하세요. "
"한국어로 답변하고, Greeks 해석과 시장 심리 분석을 포함하세요."
)
},
{
"role": "user",
"content": prompt
}
],
"temperature": 0.7,
"max_tokens": 2000
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
return result['choices'][0]['message']['content']
def _build_analysis_prompt(self, summary: Dict) -> str:
"""분석 프롬프트 생성"""
metrics = summary.get('metrics', {})
prompt = f"""
Deribit BTC 옵션 시장 데이터 분석을 요청합니다.
시장 데이터
- 총 미결제약정 (OI): {metrics.get('total_open_interest', 'N/A'):,.0f} BTC
- 풋/콜 비율 (건수): {metrics.get('put_call_ratio', 0):.2f}
- 풋/콜 비율 (OI): {metrics.get('put_call_oi_ratio', 0):.2f}
- 총 거래량: {metrics.get('total_volume', 0):,.0f}
- 평균 내재변동성: {metrics.get('avg_mark_iv', 0):.2f}%
Greeks 평균값
- Delta: {summary.get('greeks_stats', {}).get('delta_mean', 0):.4f}
- Gamma: {summary.get('greeks_stats', {}).get('gamma_mean', 0):.4f}
- Vega: {summary.get('greeks_stats', {}).get('vega_mean', 0):.4f}
- Theta: {summary.get('greeks_stats', {}).get('theta_mean', 0):.4f}
분석 항목:
1. 현재 시장 심리 해석 (Bullish/Bearish/Neutral)
2. 주요 리스크 요인
3. Greeks 기반 헤지 전략 제안
4. 단기 투자 고려사항
"""
return prompt
def generate_trading_signals(
self,
options_data: Dict
) -> Dict:
"""트레이딩 시그널 생성"""
payload = {
"model": "gpt-4.1",
"messages": [
{
"role": "system",
"content": "당신은 알고리즘 트레이딩 전문가입니다. 옵션 시장 데이터에서 액션 가능한 트레이딩 시그널을 생성하세요."
},
{
"role": "user",
"content": f"다음 데이터에서 트레이딩 시그널을 생성해주세요: {json.dumps(options_data)}"
}
],
"response_format": {
"type": "json_object",
"schema": {
"signal": "BUY/SELL/HOLD",
"confidence": 0.0-1.0,
"reason": "string",
"entry_price": "number",
"stop_loss": "number",
"take_profit": "number"
}
}
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
return json.loads(result['choices'][0]['message']['content'])
def main():
"""AI 분석 메인 실행"""
# HolySheep AI 클라이언트 초기화
client = HolySheepAIClient(
api_key=config.HOLYSHEEP_API_KEY,
base_url=config.HOLYSHEEP_BASE_URL
)
# 분석 요약 데이터 로드
with open("./data/analysis_summary.json", "r") as f:
summary = json.load(f)
# AI 분석 실행
print("🤖 HolySheep AI 옵션 분석 시작...")
print(f"사용 모델: {config.HOLYSHEEP_MODEL}")
print("-" * 50)
analysis = client.analyze_options_data(
options_summary=summary,
model=config.HOLYSHEEP_MODEL
)
print("\n📊 AI 분석 결과:")
print("="*50)
print(analysis)
# 트레이딩 시그널 생성
print("\n\n🎯 트레이딩 시그널:")
print("-" * 50)
signals = client.generate_trading_signals(summary)
print(f"시그널: {signals.get('signal', 'N/A')}")
print(f"신뢰도: {signals.get('confidence', 0)*100:.1f}%")
print(f"근거: {signals.get('reason', 'N/A')}")
if 'entry_price' in signals:
print(f"진입가: ${signals['entry_price']:,.0f}")
print(f"손절가: ${signals['stop_loss']:,.0f}")
print(f"목표가: ${signals['take_profit']:,.0f}")
# 결과 저장
output = {
"timestamp": datetime.now().isoformat(),
"analysis": analysis,
"signals": signals,
"model": config.HOLYSHEEP_MODEL
}
with open("./data/ai_analysis_report.json", "w", encoding="utf-8") as f:
json.dump(output, f, indent=2, ensure_ascii=False)
print("\n✅ 분석 리포트 저장 완료: ./data/ai_analysis_report.json")
if __name__ == "__main__":
main()
완전한 데이터 파이프라인 실행
앞에서 구현한 세 모듈을 통합하여 End-to-End 옵션 데이터 수집·처리·분석 파이프라인을 구축합니다.
# main_pipeline.py
#!/usr/bin/env python3
"""
Deribit BTC 옵션 End-to-End 데이터 파이프라인
수집 → 처리 → AI 분석
"""
import asyncio
import json
import logging
from datetime import datetime, timedelta
from pathlib import Path
from scripts.download_options_chain import DeribitOptionsDownloader
from scripts.process_options_data import OptionsDataProcessor
from scripts.analyze_with_ai import HolySheepAIClient
from config.settings import config
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)
class OptionsDataPipeline:
"""옵션 데이터 파이프라인 오케스트레이터"""
def __init__(self):
self.downloader = DeribitOptionsDownloader(config.TARDIS_API_KEY)
self.processor = OptionsDataProcessor(config.DATA_DIR)
self.ai_client = HolySheepAIClient(
api_key=config.HOLYSHEEP_API_KEY,
base_url=config.HOLYSHEEP_BASE_URL
)
async def run(self, days_back: int = 30):
"""파이프라인 실행"""
start_time = datetime.now()
logger.info("="*60)
logger.info("Deribit BTC 옵션 파이프라인 시작")
logger.info("="*60)
# Stage 1: 데이터 수집
logger.info("\n[Stage 1/3] 데이터 수집 시작...")
end_date = datetime.now()
start_date = end_date - timedelta(days=days_back)
async with self.downloader:
download_result = await self.downloader.download_historical_data(
output_dir=config.DATA_DIR,
start_date=start_date,
end_date=end_date
)
logger.info(f"수집 완료: {download_result['total_records']:,}건")
# Stage 2: 데이터 처리
logger.info("\n[Stage 2/3] 데이터 처리 시작...")
csv_files = list(Path(config.DATA_DIR).glob("btc_options_*.csv"))
latest_file = max(csv_files, key=lambda p: p.stat().st_mtime)
df = self.processor.load_csv(str(latest_file))
df_clean = self.processor.clean_data(df)
df_filtered = self.processor.filter_by_expiry(df_clean, days_to_expiry=(7, 30))
# 요약 통계
metrics = self.processor.calculate_portfolio_metrics(df_filtered)
greeks_stats = self.processor.calculate_greeks_stats(df_filtered)
summary = {
"total_records": len(df),
"filtered_records": len(df_filtered),
"date_range": f"{start_date.isoformat()} ~ {end_date.isoformat()}",
"metrics": metrics,
"greeks_stats": greeks_stats.to_dict('records')[0]
}
self.processor.export_summary(summary, f"{config.DATA_DIR}/analysis_summary.json")
logger.info("처리 완료: 분석 요약 저장됨")
# Stage 3: AI 분석
logger.info("\n[Stage 3/3] HolySheep AI 분석 시작...")
analysis = self.ai_client.analyze_options_data(
options_summary=summary,
model=config.HOLYSHEEP_MODEL
)
signals = self.ai_client.generate_trading_signals(summary)
# 결과 저장
final_report = {
"pipeline_run": datetime.now().isoformat(),
"data_summary": summary,
"ai_analysis": analysis,
"trading_signals": signals,
"execution_time_seconds": (datetime.now() - start_time).total_seconds()
}
output_file = f"{config.DATA_DIR}/final_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(final_report, f, indent=2, ensure_ascii=False)
# 요약 출력
logger.info("\n" + "="*60)
logger.info("파이프라인 완료")
logger.info("="*60)
logger.info(f"총 실행 시간: {final_report['execution_time_seconds']:.2f}초")
logger.info(f"분석 결과: {output_file}")
logger.info(f"시그널: {signals.get('signal', 'N/A')}")
return final_report
async def main():
"""메인 진입점"""
# 환경 변수 확인
if not config.TARDIS_API_KEY:
logger.error("TARDIS_API_KEY 환경 변수가 설정되지 않았습니다")
return
if not config.HOLYSHEEP_API_KEY:
logger.warning("HOLYSHEEP_API_KEY가 설정되지 않아 AI 분석을 건너뜁니다")
logger.info("HolySheep AI 가입: https://www.holysheep.ai/register")
pipeline = OptionsDataPipeline()
# 최근 30일 데이터 분석
result = await pipeline.run(days_back=30)
return result
if __name__ == "__main__":
asyncio.run(main())
자주 발생하는 오류와 해결책
1. Tardis.dev API 인증 오류 (401 Unauthorized)
API 키가 유효하지 않거나 만료된 경우 발생합니다. Tardis.dev 대시보드에서 API 키를 재발급 받고 환경 변수를 업데이트하세요. 키를 소스 코드에 하드코딩하지 말고 환경 변수 또는 시크릿 매니저를 활용하는 것이 안전합니다.
# 해결 방법: API 키 재발급 및 환경 변수 설정
1. Tardis.dev 대시보드에서 새 API 키 발급
2. 환경 변수 설정
export TARDIS_API_KEY="your_new_api_key_here"
3. 키 유효성 검증
python3 -c "
import os
import requests
api_key = os.getenv('TARDIS_API_KEY')
response = requests.get(
'https://tardis.dev/v1/feeds/deribit.options_chain',
headers={'Authorization': f'Bearer {api_key}'},
params={'symbol': 'BTC', 'limit': 1}
)
print(f'Status: {response.status_code}')
if response.status_code == 200:
print('API 키 유효함')
else:
print(f'오류: {response.text}')
"
2. CSV 스트리밍 파싱 오류 (인코딩 또는 형식 불일치)
Tardis.dev API의 CSV 응답 형식이 변경되거나, 대용량 데이터 스트리밍 중 메모리 초과가 발생할 수 있습니다. 응답 형식을 명시적으로 지정하고, 청크 단위 처리를 구현해야 합니다.
# 해결 방법: 청크 단위 스트리밍 및 인코딩 처리
async def fetch_options_chain_chunked(self, symbol: str, start_ts: int, end_ts: int):
"""청크 단위 옵션 체인 데이터 수신"""
url = (
f"{self.base_url}/deribit.options_chain"
f"?symbol={symbol}&from={start_ts}&to={end_ts}&format=csv"
)
chunk_size = 64 * 1024 # 64KB 청크
bytes_buffer = b''
async with self.session.get(url) as response:
response.raise_for_status()
# 청크 단위 스트리밍