量化取引において、波动率曲面（Volatility Surface）はオプション定价の核心的な要素です。Bybitの先物·期权市場からリアルタイムデータを取得し、HolySheep AIの高速·低コストAPIを活用して波动率曲面を構築する方法を、私が実際のプロジェクトで検証した手法と共に解説します。
波动率曲面とは
波动率曲面は、行使価格（Strike）与え 기간（Tenor）の関数として隐含波动率（Implied Volatility）を三次元的に表現するものです。Black-Scholesモデルでは波动率が一定と假设しますが、実際の市場では:
- スマイル効果：OTMオプションに波动率が集まる现象
- スキュー：下落方向に波动率が倾斜する现象
- 期间構造：短期·長期で波动率期限構造が变化
これらを三维的に表现したものが波动率曲面です。
Bybit期权データの特徴
Bybitの期权市場は以下の特徴があります：
- BTC·ETHを笔头の多样化期权系列
- 高い流动性とリアルタイム市场价格
- 複数の期满日が同时に取引可能
- API経由でのリアルタイムデータ取得が可能
HolySheep AIを選ぶ理由
波动率曲面の構築には、大量の计算处理とAPI呼び出しが必要です。HolySheep AIは以下の理由て最优解です：
| モデル | 出力価格($/MTok) | 1000万トークン辺コスト | 适用シーン |
|---|---|---|---|
| GPT-4.1 | $8.00 | $80.00 | 高精度分析 |
| Claude Sonnet 4.5 | $15.00 | $150.00 | 长文生成 |
| Gemini 2.5 Flash | $2.50 | $25.00 | 高速处理 |
| DeepSeek V3.2 | $0.42 | $4.20 | コスト重視 |
HolySheep AIでは、DeepSeek V3.2が$0.42/MTokという破格の価格で利用可能。公式レート比85%の节约で、波动率计算のような高频API呼び出しに最適です。
プロジェクト构成
# プロジェクトディレクトリ構成
volatility_surface/
├── config.py # 設定ファイル
├── bybit_client.py # Bybit APIクライアント
├── vol_surface.py # 波动率曲面クラス
├── holysheep_analyzer.py # HolySheep AI分析
├── main.py # メイン実行スクリプト
└── requirements.txt # 依存ライブラリ
設定ファイル（config.py）
"""
Bybit期权波动率曲面構築プロジェクト
HolySheep AI API設定
"""
HolySheep AI API設定
ベースURL：必ず https://api.holysheep.ai/v1 を使用
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # реальныйキーに置き換える
利用モデル（コスト效率が最も高いDeepSeek V3.2をデフォルト）
MODEL_CONFIG = {
"deepseek": {
"model": "deepseek-chat",
"input_price": 0.00, # $/MTok
"output_price": 0.42, # $/MTok（2026年価格）
"max_tokens": 4096,
"temperature": 0.1
},
"gpt4": {
"model": "gpt-4.1",
"input_price": 2.00,
"output_price": 8.00,
"max_tokens": 8192,
"temperature": 0.1
},
"claude": {
"model": "claude-sonnet-4-20250514",
"input_price": 3.00,
"output_price": 15.00,
"max_tokens": 8192,
"temperature": 0.1
}
}
Bybit API設定
BYBIT_CONFIG = {
"testnet": False,
"recv_window": 5000,
"base_url": "https://api.bybit.com"
}
波动率曲面パラメータ
VOL_SURFACE_CONFIG = {
"min_strike_ratio": 0.7, # ATM比の最小行使価格
"max_strike_ratio": 1.3, # ATM比の最大行使価格
"strike_step": 0.05, # 行使価格間隔
"tenors": [7, 14, 30, 60, 90], # 期限（日）
"risk_free_rate": 0.05, # 无リスク金利
"interpolation_method": "cubic_spline" # 補間方法
}
SVIパラメータ（Stochastic Volatility Inspired）
SVI_CONFIG = {
"initial_guess": {
"a": 0.04,
"b": 0.4,
"rho": -0.3,
"m": 0.0,
"sigma": 0.2
},
"bounds": {
"a": (0.001, 1.0),
"b": (0.01, 2.0),
"rho": (-0.99, 0.99),
"m": (-2.0, 2.0),
"sigma": (0.01, 1.0)
},
"max_iterations": 1000,
"tolerance": 1e-8
}
為替レート（HolySheep ¥1=$1 レート）
EXCHANGE_RATE = 1.0 # HolySheep公式レート
def get_model_cost(model_name: str, output_tokens: int) -> float:
"""APIコスト計算（ドル→円変換付き）"""
config = MODEL_CONFIG[model_name]
cost_usd = (output_tokens / 1_000_000) * config["output_price"]
cost_jpy = cost_usd * EXCHANGE_RATE
return cost_jpy
def estimate_monthly_cost(token_count: int, model_name: str) -> dict:
"""月間コスト見積もり"""
config = MODEL_CONFIG[model_name]
monthly_usd = (token_count / 1_000_000) * config["output_price"]
return {
"monthly_tokens": token_count,
"cost_usd": round(monthly_usd, 2),
"cost_jpy": round(monthly_usd, 2), # HolySheepレート
"savings_vs_openai": round(
monthly_usd * 19.0 if model_name == "deepseek" else 0, 2
) # OpenAI比节省額
}
Bybit APIクライアント
"""
Bybit期权データ取得クライアント
リアルタイムオプション価格とgreeks取得
"""
import hashlib
import hmac
import time
import requests
from typing import Dict, List, Optional
from datetime import datetime, timedelta
from dataclasses import dataclass
import numpy as np
@dataclass
class OptionData:
"""单个期权データ構造"""
symbol: str
strike_price: float
expiry_date: datetime
option_type: str # "call" or "put"
mark_price: float
iv: float # 隐含波动率
delta: float
gamma: float
theta: float
vega: float
spot_price: float
time_to_expiry: float # 年率换算
class BybitOptionClient:
"""Bybit先物·期权APIクライアント"""
def __init__(self, api_key: str = "", api_secret: str = "", testnet: bool = False):
self.api_key = api_key
self.api_secret = api_secret
self.base_url = "https://api.bybit.com" if not testnet else "https://api-testnet.bybit.com"
self.session = requests.Session()
self.session.headers.update({"Content-Type": "application/json"})
def _generate_signature(self, param_str: str, timestamp: str) -> str:
"""HMAC SHA256署名生成"""
hash_val = hmac.new(
self.api_secret.encode('utf-8'),
(timestamp + self.api_key + param_str).encode('utf-8'),
hashlib.sha256
).hexdigest()
return hash_val
def get_option_symbols(self, category: str = "option") -> List[Dict]:
"""取引可能な期权Symbol一覧取得"""
endpoint = "/v5/market/instruments-info"
params = {"category": category}
response = self.session.get(
f"{self.base_url}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
if data.get("retCode") == 0:
return data["result"]["list"]
else:
raise Exception(f"API Error: {data.get('retMsg')}")
def get_ticker(self, symbol: str) -> Dict:
"""個別期权のティッカー取得"""
endpoint = "/v5/market/tickers"
params = {"category": "option", "symbol": symbol}
response = self.session.get(
f"{self.base_url}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
if data.get("retCode") == 0:
return data["result"]["list"][0]
else:
raise Exception(f"API Error: {data.get('retMsg')}")
def get_option_chain(self, symbol: str, expiry_date: str) -> List[Dict]:
"""指定期满日の期权チェーン取得"""
endpoint = "/v5/market/option-chain"
params = {
"category": "option",
"symbol": symbol,
"date": expiry_date
}
response = self.session.get(
f"{self.base_url}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
if data.get("retCode") == 0:
return data["result"]["data"]
else:
raise Exception(f"API Error: {data.get('retMsg')}")
def get_index_price(self, symbol: str = "BTC") -> float:
"""指数価格取得（スポット代替）"""
endpoint = "/v5/market/tickers"
params = {"category": "spot", "symbol": f"{symbol}USDT"}
response = self.session.get(
f"{self.base_url}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
if data.get("retCode") == 0:
return float(data["result"]["list"][0]["lastPrice"])
else:
raise Exception(f"API Error: {data.get('retMsg')}")
def fetch_option_data(self, symbol: str = "BTC", expiry_dates: List[str] = None) -> List[OptionData]:
"""
全期满日の期权データ取得
Returns: List[OptionData]
"""
if expiry_dates is None:
# デフォルト：今後7日間～90日
expiry_dates = []
base_date = datetime.now()
for days in [7, 14, 30, 60, 90]:
expiry = base_date + timedelta(days=days)
expiry_dates.append(expiry.strftime("%Y-%m-%d"))
option_data_list = []
spot_price = self.get_index_price(symbol)
for expiry_date in expiry_dates:
try:
chain = self.get_option_chain(symbol, expiry_date)
expiry_dt = datetime.strptime(expiry_date, "%Y-%m-%d")
tte = (expiry_dt - datetime.now()).days / 365.0
for option in chain:
opt_data = OptionData(
symbol=symbol,
strike_price=float(option.get("strikePrice", 0)),
expiry_date=expiry_dt,
option_type=option.get("optionsType", "Call").lower(),
mark_price=float(option.get("markPrice", 0)),
iv=float(option.get("markIv", 0)) / 100, # % → 小数
delta=float(option.get("delta", 0)),
gamma=float(option.get("gamma", 0)),
theta=float(option.get("theta", 0)),
vega=float(option.get("vega", 0)),
spot_price=spot_price,
time_to_expiry=tte
)
option_data_list.append(opt_data)
except Exception as e:
print(f"Error fetching {expiry_date}: {e}")
continue
return option_data_list
def export_to_csv(self, option_data_list: List[OptionData], filename: str = "bybit_options.csv"):
"""CSVエクスポート"""
import csv
with open(filename, 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow([
"Symbol", "Strike", "Expiry", "Type", "MarkPrice",
"IV", "Delta", "Gamma", "Theta", "Vega", "Spot", "TTE"
])
for opt in option_data_list:
writer.writerow([
opt.symbol, opt.strike_price, opt.expiry_date,
opt.option_type, opt.mark_price, opt.iv,
opt.delta, opt.gamma, opt.theta, opt.vega,
opt.spot_price, opt.time_to_expiry
])
print(f"Exported {len(option_data_list)} options to {filename}")
使用例
if __name__ == "__main__":
client = BybitOptionClient()
# BTC现行先物价格取得
btc_spot = client.get_index_price("BTC")
print(f"BTC Spot Price: ${btc_spot:,.2f}")
# 全期权チェーン取得
option_chain = client.fetch_option_data(
symbol="BTC",
expiry_dates=["2026-03-28", "2026-04-04", "2026-05-30"]
)
print(f"Fetched {len(option_chain)} options")
client.export_to_csv(option_chain)
波动率曲面クラス（SVIモデル実装）
"""
波动率曲面構築 - SVI (Stochastic Volatility Inspired) モデル
HolySheep AIによるパラメータ最適化支援
"""
import numpy as np
from scipy.optimize import minimize, differential_evolution
from scipy.interpolate import CubicSpline, griddata
from typing import Tuple, List, Dict, Optional
from dataclasses import dataclass
import warnings
warnings.filterwarnings('ignore')
@dataclass
class SVIParameters:
"""SVIパラメータ"""
a: float # 水平パラメータ（ATM波动率）
b: float # 傾きパラメータ
rho: float # 相関パラメータ（-skew）
m: float # 移動パラメータ
sigma: float # 扩散パラメータ（曲率）
class VolatilitySurface:
"""波动率曲面クラス"""
def __init__(
self,
strikes: np.ndarray,
tenors: np.ndarray,
iv_matrix: np.ndarray,
spot: float,
risk_free_rate: float = 0.05
):
"""
Args:
strikes: 行使价格配列（ATM比为基準）
tenors: 期限配列（日数）
iv_matrix: 隐含波动率行列 (len(tenors) x len(strikes))
spot: スポット価格
risk_free_rate: 无リスク金利
"""
self.strikes = strikes
self.tenors = tenors
self.iv_matrix = iv_matrix
self.spot = spot
self.risk_free_rate = risk_free_rate
self.moneyness = strikes / spot
self.svi_params: Dict[float, SVIParameters] = {}
def svi_total_variance(self, k: float, t: float, params: SVIParameters) -> float:
"""
SVI全波动率（Total Variance）
w(k,t) = a + b*(rho*(k-m) + sqrt((k-m)^2 + sigma^2))
"""
m_log = np.log(k / self.spot) # log moneyness
diff = m_log - params.m
try:
sqrt_term = np.sqrt(diff**2 + params.sigma**2)
w = params.a + params.b * (
params.rho * diff + sqrt_term
)
return max(w * t, 1e-8) # 全波动率（时间×波动率^2）
except:
return 1e-6
def svi_implied_vol(self, k: float, t: float, params: SVIParameters) -> float:
"""SVI隐含波动率（ smileの补间値）"""
w = self.svi_total_variance(k, t, params)
return np.sqrt(w / t) if t > 0 else 0
def objective_function(self, params_array: np.ndarray, k: np.ndarray, iv: np.ndarray, t: float) -> float:
"""SVIパラメータ最适合目的関数（RMSE）"""
params = SVIParameters(*params_array)
iv_pred = np.array([self.svi_implied_vol(ki, t, params) for ki in k])
return np.sqrt(np.mean((iv_pred - iv)**2))
def fit_svi_slice(self, tenor_idx: int, initial_guess: dict, bounds: dict) -> SVIParameters:
"""
单一一限（slice）のSVIパラメータフィッティング
特定のtenorにおけるsmile曲线をSVIにフィッティング
"""
t = self.tenors[tenor_idx]
iv_slice = self.iv_matrix[tenor_idx]
# 有効な行使価格のみ使用
valid_mask = (iv_slice > 0) & (iv_slice < 2.0) # 0% < IV < 200%
k_valid = self.strikes[valid_mask]
iv_valid = iv_slice[valid_mask]
if len(k_valid) < 5:
return SVIParameters(a=0.04, b=0.4, rho=-0.3, m=0, sigma=0.2)
# 初期值
x0 = [
initial_guess["a"],
initial_guess["b"],
initial_guess["rho"],
initial_guess["m"],
initial_guess["sigma"]
]
# 境界
bounds_tuple = [
bounds["a"],
bounds["b"],
bounds["rho"],
bounds["m"],
bounds["sigma"]
]
# 最適化
result = differential_evolution(
self.objective_function,
bounds_tuple,
args=(k_valid, iv_valid, t),
maxiter=500,
seed=42,
polish=True,
workers=1
)
return SVIParameters(*result.x)
def fit_all_svi_slices(self, initial_guess: dict, bounds: dict) -> Dict[float, SVIParameters]:
"""
全tenorのSVIパラメータフィッティング
各限を别々にフィッティング
"""
self.svi_params = {}
for i, t in enumerate(self.tenors):
params = self.fit_svi_slice(i, initial_guess, bounds)
self.svi_params[t] = params
print(f"Tenor {t}d: SVI params = a={params.a:.4f}, b={params.b:.4f}, "
f"rho={params.rho:.4f}, m={params.m:.4f}, sigma={params.sigma:.4f}")
return self.svi_params
def interpolate_vol(self, strike: float, tenor: float) -> float:
"""
任意行使价格·期限における波动率を补間
"""
if not self.svi_params:
raise ValueError("SVI params not fitted. Call fit_all_svi_slices first.")
# tenor方向の补間
tenors_list = sorted(self.svi_params.keys())
if tenor <= tenors_list[0]:
# 最短期以下
params = self.svi_params[tenors_list[0]]
return self.svi_implied_vol(strike, tenor, params)
elif tenor >= tenors_list[-1]:
# 最長期以上
params = self.svi_params[tenors_list[-1]]
return self.svi_implied_vol(strike, tenor, params)
else:
# 線形补間（tenor方向）
for i in range(len(tenors_list) - 1):
t1, t2 = tenors_list[i], tenors_list[i+1]
if t1 <= tenor <= t2:
w = (tenor - t1) / (t2 - t1)
p1 = self.svi_params[t1]
p2 = self.svi_params[t2]
# パラメータの补間
interp_params = SVIParameters(
a=p1.a + w * (p2.a - p1.a),
b=p1.b + w * (p2.b - p1.b),
rho=p1.rho + w * (p2.rho - p1.rho),
m=p1.m + w * (p2.m - p1.m),
sigma=p1.sigma + w * (p2.sigma - p1.sigma)
)
return self.svi_implied_vol(strike, tenor, interp_params)
return 0.0
def build_grid(self, num_strikes: int = 50, num_tenors: int = 20) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
"""
完整波动率曲面グリッド生成
Returns: (tenor_grid, strike_grid, vol_grid)
"""
tenors_range = np.linspace(min(self.tenors), max(self.tenors), num_tenors)
strikes_range = np.linspace(
min(self.strikes), max(self.strikes), num_strikes
)
tenor_grid, strike_grid = np.meshgrid(tenors_range, strikes_range)
vol_grid = np.zeros_like(tenor_grid)
for i in range(num_strikes):
for j in range(num_tenors):
vol_grid[i, j] = self.interpolate_vol(
strikes_range[i], tenors_range[j]
)
return tenor_grid, strike_grid, vol_grid
def calculate_arbitrage_pnl(
self,
strikes: np.ndarray,
tenors: np.ndarray,
iv_matrix: np.ndarray,
threshold: float = 0.02
) -> Dict:
"""
тистические арбитраж機会検出
butterfly spreadとcalendar spreadの无套利条件检查
"""
violations = {
"butterfly": [], # Баттерфляй違反
"calendar": [], # カレンダー違反
"total_variance": [] # 全波动率違反
}
# butterfly spread: IV(k-2h) + IV(k+2h) >= 2*IV(k)
for t_idx in range(len(tenors)):
for k_idx in range(1, len(strikes) - 1):
k_left = strikes[k_idx - 1]
k_mid = strikes[k_idx]
k_right = strikes[k_idx + 1]
iv_left = iv_matrix[t_idx, k_idx - 1]
iv_mid = iv_matrix[t_idx, k_idx]
iv_right = iv_matrix[t_idx, k_idx + 1]
if iv_left + iv_right < 2 * iv_mid - threshold:
violations["butterfly"].append({
"tenor": tenors[t_idx],
"strike": k_mid,
"iv_left": iv_left,
"iv_mid": iv_mid,
"iv_right": iv_right
})
# calendar spread: ∂σ/∂t >= 0
for k_idx in range(len(strikes)):
for t_idx in range(1, len(tenors)):
iv_short = iv_matrix[t_idx - 1, k_idx]
iv_long = iv_matrix[t_idx, k_idx]
if iv_long < iv_short - threshold:
violations["calendar"].append({
"strike": strikes[k_idx],
"tenor_short": tenors[t_idx - 1],
"tenor_long": tenors[t_idx],
"iv_short": iv_short,
"iv_long": iv_long
})
return violations
class SurfaceAnalyzer:
"""HolySheep AI用于分析波动率曲面的クラス"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
def call_holysheep(self, prompt: str, model: str = "deepseek-chat") -> str:
"""HolySheep AI API呼び出し"""
import requests
import json
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [
{"role": "system", "content": "波动率曲面分析助手。简洁、准确、技术的な回答を心がけてください。"},
{"role": "user", "content": prompt}
],
"temperature": 0.1,
"max_tokens": 2048
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
return response.json()["choices"][0]["message"]["content"]
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
def analyze_surface_quality(
self,
strikes: List[float],
tenors: List[float],
iv_matrix: List[List[float]]
) -> Dict:
"""
波动率曲面の品質分析
HolySheep AIによる自動诊断
"""
# 統計サマリー生成
import statistics
all_ivs = [iv for row in iv_matrix for iv in row if iv > 0]
stats_prompt = f"""
波动率曲面データ分析してください：
行使价格范围: {min(strikes):.0f} - {max(strikes):.0f}
期限范围: {min(tenors)}日 - {max(tenors)}日
IV统计:
- 平均IV: {statistics.mean(all_ivs):.4f}
- 最大IV: {max(all_ivs):.4f}
- 最小IV: {min(all_ivs):.4f}
- IV标准偏差: {statistics.stdev(all_ivs):.4f}
以下を報告してください：
1. ATM点（spot近辺）のIV水平
2. スキュー方向（上昇/下落偏好）
3. スマイルの形状（对称性、曲率）
4. 期間構造の傾き
5. 異常値の検出
"""
analysis = self.call_holysheep(stats_prompt)
return {"analysis": analysis, "stats": {
"mean_iv": statistics.mean(all_ivs),
"std_iv": statistics.stdev(all_ivs) if len(all_ivs) > 1 else 0
}}
def generate_trading_signals(self, surface_data: Dict) -> List[Dict]:
"""
波动率曲面ベースの手 기회生成
"""
signal_prompt = f"""
波动率曲面データからトレーディングシグナルを分析：
{surface_data.get('analysis', '')}
各strike-tenor组合に対して：
1. IVが高い→オーバーペイド判断（ショート候補）
2. IVが低い→お買い得判断（ロング候補）
3. スキュー急変→ 이벤트 Risk判断
具体的な行使价格·期限·方向·サイズを提案してください。
"""
response = self.call_holysheep(signal_prompt)
# 简单解析
signals = []
if "ショート" in response or "ショート" in response:
signals.append({"type": "sell_vol", "confidence": "high"})
if "ロング" in response or "買い" in response:
signals.append({"type": "buy_vol", "confidence": "medium"})
return signals
メイン実行スクリプト
"""
Bybit波动率曲面構築 - メインスクリプト
HolySheep AIで分析付きの完全パイプライン
"""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D
from datetime import datetime
import os
import sys
プロジェクトパス追加
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from config import (
HOLYSHEEP_API_KEY,
HOLYSHEEP_BASE_URL,
VOL_SURFACE_CONFIG,
SVI_CONFIG,
get_model_cost,
estimate_monthly_cost
)
from bybit_client import BybitOptionClient
from vol_surface import VolatilitySurface, SurfaceAnalyzer
def simulate_option_data(spot: float = 45000) -> tuple:
"""
Bybitから実際のデータを取得できない場合のシミュレーションデータ
実際のプロジェクトでは BybitOptionClient().fetch_option_data() を使用
"""
np.random.seed(42)
strikes = np.array([spot * r for r in np.arange(0.7, 1.31, 0.05)])
tenors = np.array([7, 14, 30, 60, 90])
# ATM为中心的スマイル生成（对称+スキュー）
iv_matrix = np.zeros((len(tenors), len(strikes)))
for t_idx, t in enumerate(tenors):
for k_idx, k in enumerate(strikes):
moneyness = np.log(k / spot)
t_years = t / 365.0
# ATM波动率（期间と共に低下）
atm_vol = 0.65 - 0.05 * np.sqrt(t_years)
# スマイル効果（对称な下に凸曲线）
smile = 0.15 * (moneyness ** 2)
# スキュー（下方向偏移）
skew = -0.1 * moneyness
# 噪声
noise = np.random.normal(0, 0.01)
iv = atm_vol + smile + skew + noise
iv_matrix[t_idx, k_idx] = max(iv, 0.05)
return strikes, tenors, iv_matrix
def plot_volatility_surface(
strikes: np.ndarray,
tenors: np.ndarray,
iv_matrix: np.ndarray,
title: str = "Bybit BTC 波动率曲面"
):
"""3D波动率曲面プロット"""
fig = plt.figure(figsize=(14, 10))
ax = fig.add_subplot(111, projection='3d')
T, K = np.meshgrid(tenors, strikes)
surf = ax.plot_surface(K, T, iv_matrix * 100, cmap='viridis',
edgecolor='none', alpha=0.8)
ax.set_xlabel('行使价格 (USD)', fontsize=12)
ax.set_ylabel('期限 (日)', fontsize=12)
ax.set_zlabel('隐含波动率 (%)', fontsize=12)
ax.set_title(title, fontsize=14, pad=20)
fig.colorbar(surf, shrink=0.5, aspect=10, label='IV (%)')
plt.tight_layout()
plt.savefig('volatility_surface_3d.png', dpi=150, bbox_inches='tight')
print("Saved: volatility_surface_3d.png")
plt.close()
def plot_volatility_smile(
strikes: np.ndarray,
iv_matrix: np.ndarray,
tenors: np.ndarray,
spot: float
):
"""各tenorの波动率スマイルプロット"""
plt.figure(figsize=(12, 6))
colors = plt.cm.viridis(np.linspace(0, 1, len(tenors)))
for t_idx, t in enumerate(tenors):
plt.plot(strikes, iv_matrix[t_idx] * 100,
'o-', color=colors[t_idx], label=f'{t}日',
linewidth=2, markersize=6)
plt.axvline(x=spot, color='red', linestyle='--', label='ATM', alpha=0.7)
plt.xlabel('行使价格 (USD)', fontsize=12)
plt.ylabel('隐含波动率 (%)', fontsize=12)
plt.title('Bybit BTC 波动率スマイル', fontsize=14)
plt.legend(loc='upper right')
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('volatility_smile.png', dpi=150, bbox_inches='tight')
print("Saved: volatility_smile.png")
plt.close()
def main():
"""メイン実行関数"""
print("=" * 60)
print("Bybit 波动率曲面構築パイプライン")
print("HolySheep AI API v1 対応")
print("=" * 60)
# ============================================================
# Step 1: データ取得
# ============================================================
print("\n[Step 1] データ取得")
print("-" * 40)
# 实际环境：Bybit API使用
# client = BybitOptionClient()
# option_data = client.fetch_option_data("BTC", ["2026-03-28", "2026-04-04", "2026-05-30"])
# デモ环境：シミュレーションデータ
spot_price = 45000 # BTC現物価格
strikes