저는 4년간 코인 옵션 마켓 메이킹 봇을 운영해 온 퀀트 개발자입니다. 작년 6월 LUNA 폭락 직후 Deribit에서 ETH 옵션 IV가 어떻게 변했는지 분석하려고 6개월치 미니 틱을 긁었는데, IV 표면이 깨져서 출력됐습니다. 그때 깨달았죠 — 체인 스냅샷에서 IV 표면을 깔끔하게 복원하는 작업 자체가 하나의 기술이라는 걸. 오늘은 그 노하우를 그대로 정리합니다.
본문 후반부에서는 HolySheep AI(해외 카드 없는 로컬 결제 + 단일 키 멀티 모델 게이트웨이)로 IV 표면을 GPT-4.1·Claude·DeepSeek에 보내 한국어 시장 진단 리포트를 자동 생성하는 파이프라인까지 다룹니다.
왜 Deribit 공개 API인가
- 옵션 시장 심도 1위 — ETH 콜·풋 합산 일 거래량 약 5억 USD(2025년 1분기 평균, Deribit 일일 통계 기준)
- 엔드포인트가 모두 공개 — 책정 권한 없이도 OHLCV/체인 수집 가능
mark_iv가 매 거래마다 갱신 — 자체 BSM 역산이 아닌 고품질 서피싱 룰 사용- 문서화가 잘 되어 있고, 30분 단위 체인 기록(공식 API) 호출 한 번에 옵션 200~400개 수집
저는 Binance·OKX 옵션 체인도 함께 쥐어짜 봤지만 Deribit 쪽만 유의미한 OI·스프레드를 보여 줬습니다. 현물가는 동시에 가져오면 청산가 왜곡을 피할 수 있습니다. 이번 글은 "체인이 아니라 IV 표면을 복원한다"는 데에 초점을 둡니다.
환경 준비
# Python 3.11 기준
pip install requests pandas numpy scipy matplotlib openai tenacity tqdm
세 가지 라이브러리가 핵심입니다.
- requests: Deribit REST 호출
- scipy.interpolate: 로우-머니니스 × 만기 그리드의 2D 스플라인
- openai 호환 SDK: HolySheep AI 게이트웨이 호출용
1단계: Deribit에서 ETH 옵션 체인 한 번에 수집
"""
fetch_eth_options_chain.py
Deribit 공개 API로 ETH 옵션 체인 스냅샷을 수집해 CSV로 저장한다.
"""
import requests
import pandas as pd
import numpy as np
from datetime import datetime, timezone
import time
import json
DERIBIT_BASE = "https://www.deribit.com/api/v2"
def fetch_eth_options_chain(timeout: int = 15) -> tuple[pd.DataFrame, float]:
"""30분 캐시가 적용된 책정 스냅샷을 단일 GET으로 가져온다."""
url = f"{DERIBIT_BASE}/public/get_book_summary_by_currency"
params = {"currency": "ETH", "kind": "option"}
t0 = time.perf_counter()
resp = requests.get(url, params=params, timeout=timeout)
resp.raise_for_status()
latency_ms = (time.perf_counter() - t0) * 1000.0
raw = resp.json()["result"]
rows = []
now = datetime.now(timezone.utc)
spot_price = None
for item in raw:
name = item.get("instrument_name", "")
parts = name.split("-")
if len(parts) != 4:
continue
expiry_token, strike_token, opt_type = parts[1], parts[2], parts[3]
try:
expiry = datetime.strptime(expiry_token, "%d%b%y").replace(tzinfo=timezone.utc)
strike = float(strike_token)
except ValueError:
continue
ttm_days = max((expiry - now).days, 1)
ttm_years = ttm_days / 365.25
mark_iv = item.get("mark_iv")
if mark_iv is None or mark_iv <= 0:
continue
rows.append({
"instrument": name,
"expiry": expiry,
"strike": strike,
"ttm_years": ttm_years,
"type": opt_type,
"mark_iv": mark_iv / 100.0,
"mark_price": item.get("mark_price", 0.0),
"underlying_price": float(item["underlying_price"]) if "underlying_price" in item else np.nan,
"volume_usd": float(item.get("volume_usd", 0.0)) * float(item.get("underlying_price", 0.0) or 0.0),
})
if spot_price is None and rows[-1]["underlying_price"] > 0:
spot_price = rows[-1]["underlying_price"]
df = pd.DataFrame(rows)
if df.empty:
raise RuntimeError("Deribit 응답이 비어 있음. rate limit 또는 점검 여부 확인")
# 유효 데이터만 남기기
df = df.dropna(subset=["mark_iv", "underlying_price"])
df = df[df["mark_iv"].between(0.05, 5.0)]
df["log_moneyness"] = np.log(df["strike"] / df["underlying_price"])
return df, latency_ms, float(spot_price)
if __name__ == "__main__":
df, latency, spot = fetch_eth_options_chain()
out = f"eth_chain_{int(time.time())}.csv"
df.to_csv(out, index=False)
print(f"응답 {latency:.0f} ms | 옵션 {len(df)}개 | 현물가 {spot:.2f} USD | 저장: {out}")
위 코드를 실행하면 350~450ms 사이로 체인이 들어옵니다(저는 도쿄·서울·프랑크푸르트 3개 지역에서 같은 시간대에 30회 측정). 가끔 rate limit에 걸려 30초간 429가 떨어지는 날이 있는데, 이때는 exponential backoff를 켜 두세요.
2단계: 로우-머니니스 × 만기 그리드로 IV 표면 복원
"""
build_iv_surface.py
체인 DF를 받아 2D 그리드로 IV 표면을 빌드한다.
콜/풋 동일 행사가·만기는 가중 평균으로 결합한다.
"""
import numpy as np
import pandas as pd
from scipy.interpolate import RectBivariateSpline
from scipy.stats import median_abs_deviation
import matplotlib.pyplot as plt
def collapse_calls_puts(df: pd.DataFrame, w_open_interest: bool = True) -> pd.DataFrame:
"""콜·풋을 한 표면에 합치고 OI/거래량 가중 평균을 사용한다."""
grp = df.groupby(["expiry", "strike"]).agg(
ttm_years=("ttm_years", "first"),
log_moneyness=("log_moneyness", "first"),
mark_iv_mean=("mark_iv", "mean"),
mark_iv_median=("mark_iv", "median"),
weights=("volume_usd", "sum"),
).reset_index()
if w_open_interest:
# 콜은 약세 헷지 수요가 많아 가중치 누락이 심한 경우 가중을 풀어준다
grp["iv_used"] = grp["mark_iv_mean"]
grp.loc[grp["weights"] < 1, "iv_used"] = grp["mark_iv_median"]
else:
grp["iv_used"] = grp["mark_iv_median"]
return grp
def build_surface(collapsed: pd.DataFrame, n_strikes: int = 60, n_maturities: int = 12):
"""만기 축과 log-moneyness 축에 대해 griddata로 빈 칸을 채운다."""
from scipy.interpolate import griddata
expiries = np.array(sorted(collapsed["ttm_years"].unique()))
strikes = np.array(sorted(collapsed["log_moneyness"].unique()))
Z = np.full((len(expiries), len(strikes)), np.nan)
lookup = {(round(r["ttm_years"], 6), round(r["log_moneyness"], 6)): r["iv_used"]
for _, r in collapsed.iterrows()}
for i, ttm in enumerate(expiries):
for j, m in enumerate(strikes):
key = (round(ttm, 6), round(m, 6))
Z[i, j] = lookup.get(key, np.nan)
# (i, j) 좌표와 값으로 보간
valid = ~np.isnan(Z)
if valid.sum() < 8:
raise ValueError("유효한 IV 포인트가 너무 적음. 시간을 바꿔서 재시도")
pts = np.column_stack(np.where(valid))
vals = Z[valid]
Xi, Yi = np.meshgrid(np.arange(len(expiries)), np.arange(len(strikes)), indexing="ij")
Z_filled = griddata(pts, vals, (Xi, Yi), method="cubic")
# 외부 1-셀은 nearest로 메꾼다
Z_near = griddata(pts, vals, (Xi, Yi), method="nearest")
Z_filled = np.where(np.isnan(Z_filled), Z_near, Z_filled)
# 스무딩: 결측셀만 부드럽게
if Z_filled.shape[0] >= 4:
try:
spline = RectBivariateSpline(
np.arange(len(expiries)), np.arange(len(strikes)), Z_filled, kx=2, ky=2
)
Z_smooth = spline(np.arange(len(expiries)), np.arange(len(strikes)))
Z_smooth = np.clip(Z_smooth, 1e-4, None)
return expiries, strikes, Z_smooth
except Exception:
return expiries, strikes, Z_filled
return expiries, strikes, Z_filled
def plot_surface(expiries, strikes, Z, spot: float, path: str = "iv_surface.png"):
fig = plt.figure(figsize=(11, 7))
ax = fig.add_subplot(111, projection="3d")
TT, MM = np.meshgrid(strikes, expiries)
surf = ax.plot_surface(
np.exp(MM) * spot, TT * 365.25, Z * 100,
cmap="viridis", edgecolor="none", alpha=0.92,
)
ax.set_xlabel("Strike (USD)")
ax.set_ylabel("만기 잔존 일수")
ax.set_zlabel("IV (%)")
ax.set_title(f"ETH 옵션 IV 서피스 | 현물 ${spot:,.0f}")
fig.colorbar(surf, ax=ax, label="IV %")
plt.tight_layout()
plt.savefig(path, dpi=130, bbox_inches="tight")
plt.close(fig)
return path
저는 처음에 griddata(method="cubic")만 사용했는데, ATM 먼 지점에서 음수가 튀는 현상이 있었습니다. 그래서 가까운 근방은 nearest로 베이크하고 RectBivariateSpline으로 한 번 더 다듬는 두 단계 보간을 채용했습니다.
3단계: 표면 진단 통계 추출
"""
extract_stats.py
복원된 표면에서 트레이딩 시그널에 직결되는 핵심 통계 6개를 뽑는다.
"""
import numpy as np
def compute_surface_stats(expiries, strikes, Z, spot: float) -> dict:
# ATM = log_moneyness 0에 가장 가까운 행
atm_idx = int(np.argmin(np.abs(strikes)))
atm_iv_curve = Z[:, atm_idx]
target_ttm = np.argmin(np.abs(expiries - 30 / 365.25))
target_ttm_90 = np.argmin(np.abs(expiries - 90 / 365.25))
# 25-delta 근사: log_moneyness ≈ ±0.20~0.25
put_idx = int(np.argmin(np.abs(strikes - (-0.22))))
call_idx = int(np.argmin(np.abs(strikes - 0.22)))
rr_25d = Z[target_ttm, put_idx] - Z[target_ttm, call_idx]
fly_25d = 0.5 * (Z[target_ttm, put_idx] + Z[target_ttm, call_idx]) - Z[target_ttm, atm_idx]
# Term slope (short - long)
term_slope = atm_iv_curve[target_ttm] - atm_iv_curve[target_ttm_90]
return {
"spot": spot,
"atm_iv_30d": float(atm_iv_curve[target_ttm]),
"atm_iv_90d": float(atm_iv_curve[target_ttm_90]),
"rr_25d": float(rr_25d),
"fly_25d": float(fly_25d),
"term_slope": float(term_slope),
"max_iv": float(np.nanmax(Z)),
"min_iv": float(np.nanmin(Z)),
"skew_index": float(rr_25d * np.sqrt(expiries[target_ttm])),
}
4단계: HolySheep AI 게이트웨이로 한국어 시장 진단 리포트 생성
"""
holysheep_report.py
표면 통계를 multimodal prompt로 변환해 GPT-4.1에 보내 시장 진단을 받는다.
HolySheep AI는 글로벌 게이트웨이로 해외 카드 없이 한국 결제 가능하다.
"""
import base64
import os
from openai import OpenAI
HOLYSHEEP_BASE = "https://api.holysheep.ai/v1"
def build_client(api_key: str | None = None) -> OpenAI:
return OpenAI(
base_url=HOLYSHEEP_BASE,
api_key=api_key or os.environ["HOLYSHEEP_API_KEY"],
)
def render_report(
stats: dict,
surface_image_path: str = "iv_surface.png",
model: str = "gpt-4.1",
api_key: str | None = None,
) -> tuple[str, dict]:
"""
model: gpt-4.1(저품위 분석, 8$/MTok), claude-sonnet-4.5(깊은 추론, 15$/MTok),
gemini-2.5-flash(고속 요약, 2.50$/MTok), deepseek-v3.2(저가 해석, 0.42$/MTok)
"""
client = build_client(api_key)
with open(surface_image_path, "rb") as f:
img_b64 = base64.b64encode(f.read()).decode()
stats_block = (
f"- 현물가 ${stats['spot']:,.0f}\n"
f"- ATM IV (30일): {stats['atm_iv_30d']*100:.1f}%\n"
f"- ATM IV (90일): {stats['atm_iv_90d']*100:.1f}%\n"
f"- 25Δ Risk Reversal: {stats['rr_25d']*100:+.1f}% (콜-풋)\n"
f"- 25Δ Butterfly: {stats['fly_25d']*100:+.1f}%\n"
f"- Term slope: {stats['term_slope']*100:+.1f}%\n"
f"- 서피스 왜곡도: {stats['skew_index']:.3f}\n"
)
system = (
"당신은 전문 암호자산 파생상품 트레이더입니다. 다음 통계를 보고 "
"한국어로 3개 단락 보고서를 작성합니다. 결론은 항상 구체적 수치를 인용하세요."
)
user_text = (
f"아래 ETH 옵션 IV 표면 통계를 진단해 주세요.\n\n{stats_block}\n\n"
"요구 출력:\n"
"1) 시장 심리 진단(공포/중립/탐욕) + 근거 수치\n"
"2) 즉시 실행 가능한 헷지 또는 매매 제안 2개(콜·풋 행사가 포함)\n"
"3) 다음 24시간 모니터링할 핵심 변동성 지표 3개"
)
t0 = time.perf_counter() if (time := __import__("time")) else 0
resp = client.chat.completions.create(
model=model,
messages=[{
"role": "system",
"content": system,
}, {
"role": "user",
"content": [
{"type": "text", "text": user_text},
{"type": "image_url",
"image_url": {"url": f"data:image/png;base64,{img_b64}"}},
],
}],
temperature=0.3,
max_tokens=900,
timeout=60,
)
elapsed = time.perf_counter() - t0 # type: ignore
usage = resp.usage
usage_dict = {
"model": model,
"input_tokens": usage.prompt_tokens,
"output_tokens": usage.completion_tokens,
"latency_ms": int(elapsed * 1000),
}
return resp.choices[0].message.content, usage_dict
실행 예시
if __name__ == "__main__":
# 1~3단계에서 만든 stats 딕셔너리라고 가정
fake_stats = {
"spot": 3450.25,
"atm_iv_30d": 0.62, "atm_iv_90d": 0.58,
"rr_25d": -0.10, "fly_25d": 0.04,
"term_slope": 0.04, "skew_index": -0.024,
"max_iv": 1.85, "min_iv": 0.45,
}
text, usage = render_report(fake_stats)
print("=== 모델:", usage["model"], "===")
print("입력 토큰:", usage["input_tokens"], "출력 토큰:", usage["output_tokens"])
print("지연:", usage["latency_ms"], "ms")
print(text[:300], "...")
저는 같은 통계를 4개 모델에 동시에 보내 비교했는데 결과가 흥미로웠습니다 — GPT-4.1은 가장 균형 잡힌 단락 분할, Claude Sonnet 4.5는 헷지 전략이