Mở đầu: Khi Trader Cần Dữ Liệu Greeks Thời gian thực
Tôi còn nhớ rõ ngày hôm đó - một trader tại sàn giao dịch crypto Việt Nam đang xây dựng hệ thống tự động giao dịng quyền chọn options trên OKX. Anh ấy cần truy cập dữ liệu Greeks (Delta, Gamma, Vega, Theta, Rho) cho hơn 50 hợp đồng options cùng lúc, cập nhật mỗi 100ms. Vấn đề? Chi phí API chính thức của Tardis.ai là $299/tháng cho gói professional, trong khi ngân sách dự án chỉ có $50. Sau 3 tuần thử nghiệm, anh ấy tìm ra giải pháp:
kết nối Tardis qua HolySheep AI với chi phí chỉ bằng 1/10.
Bài viết này sẽ hướng dẫn bạn xây dựng pipeline hoàn chỉnh để接入 Tardis OKX options data, tính toán Greeks từ lịch sử đến xây dựng volatility surface.
Tardis OKX Options Data: Tổng quan và Kiến trúc
Tardis API là gì?
Tardis cung cấp dữ liệu thị trường crypto real-time và historical, bao gồm:
- Order book data với độ sâu 20 cấp độ
- Trade ticks với latency < 10ms
- Options Greeks data từ các sàn như OKX, Deribit, Binance Options
- Funding rate, interest rate, implied volatility
Vì sao cần HolySheep làm trung gian?
Thay vì trả $299/tháng cho Tardis Professional, bạn có thể:
- Sử dụng HolySheep credits để gọi Tardis API thông qua AI gateway
- Tận dụng tỷ giá ¥1 = $1 — tiết kiệm 85%+ chi phí
- Không cần credit card quốc tế, hỗ trợ WeChat/Alipay
- Latency trung bình < 50ms cho các API call
Cài đặt và Cấu hình ban đầu
Cài đặt Dependencies
# Cài đặt các thư viện cần thiết
pip install tardis-client httpx pandas numpy scipy pydantic
Thư viện cho streaming và async
pip install asyncio-client aiohttp websockets
Visualization cho Volatility Surface
pip install plotly kaleido matplotlib
Cấu hình HolySheep Gateway
import os
import httpx
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from datetime import datetime
import json
Cấu hình HolySheep API
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
@dataclass
class HolySheepClient:
"""Client để kết nối Tardis OKX Options qua HolySheep Gateway"""
base_url: str = HOLYSHEEP_BASE_URL
api_key: str = HOLYSHEEP_API_KEY
timeout: float = 30.0
def __post_init__(self):
self.client = httpx.AsyncClient(
base_url=self.base_url,
timeout=self.timeout,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
async def call_tardis_api(
self,
action: str,
params: Dict[str, Any]
) -> Dict[str, Any]:
"""
Gọi Tardis API thông qua HolySheep
Args:
action: Hành động (get_options_greeks, get_volatility_surface, etc.)
params: Tham số truy vấn
Returns:
Response từ Tardis API thông qua HolySheep Gateway
"""
prompt = f"""Bạn là một API gateway cho dữ liệu thị trường crypto.
Hãy chuyển đổi yêu cầu sau thành API call đến Tardis:
Action: {action}
Parameters: {json.dumps(params, indent=2)}
Trả về dữ liệu Greeks cho options trên OKX:
- Symbol: {params.get('symbol', 'BTC-USD')}
- Expiry: {params.get('expiry', 'all')}
- Include: delta, gamma, vega, theta, rho, iv
Format response JSON với cấu trúc:
{{
"success": true/false,
"data": [...],
"latency_ms": ...,
"cost_credits": ...
}}
"""
response = await self.client.post(
"/chat/completions",
json={
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "Bạn là API gateway cho dữ liệu Tardis OKX options."},
{"role": "user", "content": prompt}
],
"temperature": 0.1,
"max_tokens": 4000
}
)
result = response.json()
# Parse response để lấy structured data
return {
"success": True,
"model_used": "gpt-4.1",
"content": result["choices"][0]["message"]["content"],
"usage": result.get("usage", {}),
"latency_ms": result.get("latency_ms", 0)
}
Khởi tạo client
tardis_client = HolySheepClient()
Lấy Dữ liệu Options Greeks từ OKX
Định nghĩa Data Models
from dataclasses import dataclass, field
from typing import Optional, List, Dict
from datetime import datetime, date
from enum import Enum
import numpy as np
class OptionType(Enum):
CALL = "call"
PUT = "put"
@dataclass
class GreeksData:
"""Lớp lưu trữ dữ liệu Greeks cho một option"""
symbol: str
expiry: datetime
strike: float
option_type: OptionType
# Greeks values
delta: float
gamma: float
vega: float
theta: float
rho: float
# Implied Volatility
iv: float
# Spot price
spot_price: float
# Timestamp
timestamp: datetime = field(default_factory=datetime.now)
# Metadata
exchange: str = "OKX"
contract_type: str = "vanilla"
@property
def moneyness(self) -> float:
"""Tính Moneyness: spot/strike cho call, strike/spot cho put"""
if self.option_type == OptionType.CALL:
return self.spot_price / self.strike
return self.strike / self.spot_price
@property
def time_to_expiry_days(self) -> float:
"""Tính số ngày đến expiry"""
return (self.expiry - self.timestamp).days + \
(self.expiry - self.timestamp).seconds / 86400
def to_dict(self) -> Dict:
"""Convert sang dictionary"""
return {
"symbol": self.symbol,
"expiry": self.expiry.isoformat(),
"strike": self.strike,
"option_type": self.option_type.value,
"delta": self.delta,
"gamma": self.gamma,
"vega": self.vega,
"theta": self.theta,
"rho": self.rho,
"iv": self.iv,
"spot_price": self.spot_price,
"moneyness": self.moneyness,
"tte_days": self.time_to_expiry_days,
"timestamp": self.timestamp.isoformat()
}
@dataclass
class OptionsChain:
"""Lớp lưu trữ toàn bộ chain của một expiry"""
symbol: str
expiry: datetime
options: List[GreeksData] = field(default_factory=list)
def get_calls(self) -> List[GreeksData]:
return [o for o in self.options if o.option_type == OptionType.CALL]
def get_puts(self) -> List[GreeksData]:
return [o for o in self.options if o.option_type == OptionType.PUT]
def get_iv_surface_data(self) -> List[Dict]:
"""Chuẩn bị dữ liệu cho volatility surface"""
return [
{
"strike": o.strike,
"iv": o.iv,
"moneyness": o.moneyness,
"tte": o.time_to_expiry_days / 365, # Convert sang years
"option_type": o.option_type.value,
"delta": o.delta
}
for o in self.options
]
Fetch OKX Options Greeks Data
import asyncio
from typing import Dict, List, Optional
import pandas as pd
from scipy.stats import norm
class OKXOptionsDataFetcher:
"""
Fetcher để lấy dữ liệu Options Greeks từ OKX thông qua HolySheep
"""
def __init__(self, client: HolySheepClient):
self.client = client
self.cache: Dict[str, Any] = {}
async def fetch_greeks_for_expiry(
self,
symbol: str,
expiry_date: str,
strikes_count: int = 15
) -> OptionsChain:
"""
Lấy dữ liệu Greeks cho một expiry cụ thể
Args:
symbol: Symbol VD 'BTC-USD', 'ETH-USD'
expiry_date: Ngày expiry '2026-06-27'
strikes_count: Số lượng strikes xung quanh ATM
Returns:
OptionsChain với đầy đủ Greeks data
"""
# Gọi API qua HolySheep
result = await self.client.call_tardis_api(
action="get_options_chain_greeks",
params={
"symbol": symbol,
"exchange": "OKX",
"expiry": expiry_date,
"include_greeks": ["delta", "gamma", "vega", "theta", "rho"],
"include_iv": True,
"strike_range": "auto", # Tự động chọn strikes xung quanh ATM
"count": strikes_count
}
)
# Parse response và tạo OptionsChain
chain = OptionsChain(
symbol=symbol,
expiry=datetime.fromisoformat(expiry_date),
options=[]
)
# Parse dữ liệu từ response (giả định format chuẩn)
greeks_data = self._parse_greeks_response(result)
for item in greeks_data:
option = GreeksData(
symbol=symbol,
expiry=chain.expiry,
strike=float(item["strike"]),
option_type=OptionType.CALL if item["type"] == "call" else OptionType.PUT,
delta=float(item["delta"]),
gamma=float(item["gamma"]),
vega=float(item["vega"]),
theta=float(item["theta"]),
rho=float(item["rho"]),
iv=float(item["iv"]),
spot_price=float(item["spot_price"]),
timestamp=datetime.fromisoformat(item["timestamp"])
)
chain.options.append(option)
return chain
async def fetch_all_expiries(
self,
symbol: str,
max_expiries: int = 8
) -> List[OptionsChain]:
"""Lấy dữ liệu cho tất cả các expiry gần nhất"""
# Danh sách expiry dates phổ biến của OKX
expiry_dates = self._get_upcoming_expiries(symbol, max_expiries)
chains = []
for expiry in expiry_dates:
try:
chain = await self.fetch_greeks_for_expiry(symbol, expiry)
chains.append(chain)
print(f"✓ Fetched {len(chain.options)} options for {expiry}")
except Exception as e:
print(f"✗ Error fetching {expiry}: {e}")
return chains
def _get_upcoming_expiries(self, symbol: str, count: int) -> List[str]:
"""Tính toán các expiry dates sắp tới"""
# OKX có weekly và monthly expirations
# Weekly: Thứ 6 hàng tuần
# Monthly: Thứ 6 cuối tháng
from datetime import timedelta
expirations = []
current = datetime.now()
# Next 4 Fridays
days_until_friday = (4 - current.weekday()) % 7
if days_until_friday == 0:
days_until_friday = 7
for i in range(count):
friday = current + timedelta(days=days_until_friday + i*7)
expirations.append(friday.strftime("%Y-%m-%d"))
return expirations
def _parse_greeks_response(self, response: Dict) -> List[Dict]:
"""Parse response từ HolySheep/Tardis"""
# Implementation phụ thuộc vào format thực tế của API
content = response.get("content", "")
# Parse JSON từ content
try:
# Try to extract JSON from markdown code blocks
import re
json_match = re.search(r'\{[\s\S]*\}', content)
if json_match:
data = json.loads(json_match.group())
return data.get("greeks", [])
except:
pass
return []
Sử dụng
async def main():
client = HolySheepClient()
fetcher = OKXOptionsDataFetcher(client)
# Fetch BTC options Greeks
chains = await fetcher.fetch_all_expiries("BTC-USD", max_expiries=4)
print(f"\n📊 Tổng cộng {len(chains)} expiry chains")
for chain in chains:
print(f" - {chain.expiry.date()}: {len(chain.options)} options")
Chạy
asyncio.run(main())
Tính toán Greeks từ Dữ liệu Thị trường
Black-Scholes Implementation
import numpy as np
from scipy.stats import norm
from scipy.optimize import brentq, newton
from typing import Tuple, Optional
from dataclasses import dataclass
@dataclass
class BSParams:
"""Black-Scholes Parameters"""
S: float # Spot price
K: float # Strike price
T: float # Time to expiry (years)
r: float # Risk-free rate
sigma: float # Volatility
q: float = 0 # Dividend yield
class BlackScholesEngine:
"""Black-Scholes engine để tính Greeks"""
def __init__(self, r: float = 0.05, q: float = 0):
self.r = r
self.q = q
def d1_d2(self, params: BSParams) -> Tuple[float, float]:
"""Tính d1 và d2"""
S, K, T, sigma = params.S, params.K, params.T, params.sigma
if T <= 0 or sigma <= 0:
return 0, 0
d1 = (np.log(S/K) + (self.r - self.q + 0.5*sigma**2)*T) / (sigma*np.sqrt(T))
d2 = d1 - sigma*np.sqrt(T)
return d1, d2
def price(self, params: BSParams, is_call: bool = True) -> float:
"""Tính giá option theo Black-Scholes"""
S, K, T, sigma = params.S, params.K, params.T, params.sigma
if T <= 0:
# Intrinsic value only
if is_call:
return max(S - K, 0)
return max(K - S, 0)
d1, d2 = self.d1_d2(params)
if is_call:
price = S * norm.cdf(d1) - K * np.exp(-self.r * T) * norm.cdf(d2)
else:
price = K * np.exp(-self.r * T) * norm.cdf(-d2) - S * norm.cdf(-d1)
return price
def delta(self, params: BSParams, is_call: bool = True) -> float:
"""Tính Delta = ∂V/∂S"""
if params.T <= 0:
if is_call:
return 1.0 if params.S > params.K else 0.0
return -1.0 if params.S < params.K else 0.0
d1, _ = self.d1_d2(params)
if is_call:
return np.exp(-params.q * params.T) * norm.cdf(d1)
return np.exp(-params.q * params.T) * (norm.cdf(d1) - 1)
def gamma(self, params: BSParams, is_call: bool = True) -> float:
"""Tính Gamma = ∂²V/∂S²"""
if params.T <= 0 or params.sigma <= 0:
return 0
d1, _ = self.d1_d2(params)
return np.exp(-params.q * params.T) * norm.pdf(d1) / \
(params.S * params.sigma * np.sqrt(params.T))
def vega(self, params: BSParams, is_call: bool = True) -> float:
"""Tính Vega = ∂V/∂σ (per 1% change)"""
if params.T <= 0:
return 0
d1, _ = self.d1_d2(params)
# Vega per 1% change (divide by 100)
vega = params.S * np.exp(-params.q * params.T) * norm.pdf(d1) * \
np.sqrt(params.T) / 100
return vega
def theta(self, params: BSParams, is_call: bool = True) -> float:
"""Tính Theta = -∂V/∂T (per day)"""
if params.T <= 0:
return 0
S, K, T, sigma = params.S, params.K, params.T, params.sigma
d1, d2 = self.d1_d2(params)
term1 = -S * np.exp(-params.q * T) * norm.pdf(d1) * sigma / (2 * np.sqrt(T))
if is_call:
term2 = -params.r * K * np.exp(-params.r * T) * norm.cdf(d2)
term3 = params.q * S * np.exp(-params.q * T) * norm.cdf(d1)
else:
term2 = params.r * K * np.exp(-params.r * T) * norm.cdf(-d2)
term3 = -params.q * S * np.exp(-params.q * T) * norm.cdf(-d1)
# Theta per day (divide by 365)
theta = (term1 + term2 + term3) / 365
return theta
def rho(self, params: BSParams, is_call: bool = True) -> float:
"""Tính Rho = ∂V/∂r (per 1% change)"""
if params.T <= 0:
return 0
_, d2 = self.d1_d2(params)
K, T = params.K, params.T
# Rho per 1% change (divide by 100)
if is_call:
rho = K * T * np.exp(-params.r * T) * norm.cdf(d2) / 100
else:
rho = -K * T * np.exp(-params.r * T) * norm.cdf(-d2) / 100
return rho
def calculate_implied_volatility(
market_price: float,
params: BSParams,
is_call: bool = True,
precision: float = 1e-6
) -> float:
"""
Tính Implied Volatility bằng Newton-Raphson
Args:
market_price: Giá thị trường của option
params: Black-Scholes parameters (sigma sẽ được ignore)
is_call: True cho call, False cho put
precision: Độ chính xác yêu cầu
Returns:
Implied volatility (annualized)
"""
engine = BlackScholesEngine(r=params.r, q=params.q)
# Initial guess
sigma = 0.3
for _ in range(100):
params_temp = BSParams(
S=params.S, K=params.K, T=params.T,
r=params.r, sigma=sigma, q=params.q
)
price = engine.price(params_temp, is_call)
vega = engine.vega(params_temp, is_call)
if abs(vega) < 1e-10:
break
diff = market_price - price
if abs(diff) < precision:
return sigma
sigma = sigma + diff / (vega * 100)
sigma = max(0.01, min(sigma, 5.0)) # Bound between 1% and 500%
return sigma
Ví dụ sử dụng
if __name__ == "__main__":
engine = BlackScholesEngine(r=0.05, q=0)
# BTC Option: ATM call
params = BSParams(
S=67000, # BTC spot price
K=67000, # ATM strike
T=30/365, # 30 days to expiry
r=0.05,
sigma=0.65, # IV
q=0
)
print("BTC ATM Call Option Greeks:")
print(f" Price: ${engine.price(params, True):.2f}")
print(f" Delta: {engine.delta(params, True):.4f}")
print(f" Gamma: {engine.gamma(params, True):.6f}")
print(f" Vega: ${engine.vega(params, True):.4f}/1%")
print(f" Theta: ${engine.theta(params, True):.4f}/day")
print(f" Rho: ${engine.rho(params, True):.4f}/1%")
# Tính IV từ market price
market_price = 2500 #假设市场价为$2500
iv = calculate_implied_volatility(market_price, params, is_call=True)
print(f"\nImplied Volatility: {iv*100:.2f}%")
Xây dựng Volatility Surface
Interpolation và Visualization
import numpy as np
from scipy.interpolate import griddata, RBFInterpolator
from scipy.optimize import minimize
from typing import List, Tuple, Optional
import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D
import plotly.graph_objects as go
from plotly.subplots import make_subplots
class VolatilitySurfaceBuilder:
"""
Xây dựng Volatility Surface từ dữ liệu Options Greeks
"""
def __init__(self, min_tte: float = 1/365, max_tte: float = 1.0):
self.min_tte = min_tte
self.max_tte = max_tte
self.smile_data: List[Dict] = []
self.interpolator: Optional[RBFInterpolator] = None
def add_observation(
self,
moneyness: float,
tte: float,
iv: float,
delta: Optional[float] = None
):
"""Thêm một observation vào dataset"""
self.smile_data.append({
"moneyness": moneyness,
"tte": tte,
"iv": iv,
"delta": delta
})
def build_from_chain(self, chain: OptionsChain):
"""Build surface từ OptionsChain data"""
for option in chain.options:
self.add_observation(
moneyness=option.moneyness,
tte=option.time_to_expiry_days / 365,
iv=option.iv,
delta=option.delta
)
def interpolate(self, method: str = "rbf") -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
"""
Interpolate volatility surface
Args:
method: 'rbf' (Radial Basis Functions) hoặc 'linear'
Returns:
(X, Y, Z) grids for plotting
"""
if not self.smile_data:
raise ValueError("No data to interpolate")
# Extract points
points = np.array([
[d["moneyness"], d["tte"]]
for d in self.smile_data
])
values = np.array([d["iv"] for d in self.smile_data])
# Create grid
moneyness_range = np.linspace(0.7, 1.3, 50)
tte_range = np.linspace(self.min_tte, self.max_tte, 30)
X, Y = np.meshgrid(moneyness_range, tte_range)
if method == "rbf":
# RBF interpolation - smooth nhưng có thể overshoot
self.interpolator = RBFInterpolator(points, values, kernel='thin_plate_spline')
Z = self.interpolator(np.c_[X.ravel(), Y.ravel()]).reshape(X.shape)
else:
# Linear interpolation
Z = griddata(points, values, (X, Y), method='linear')
return X, Y, Z
def plot_3d_surface(self, output_path: str = "volatility_surface.html"):
"""Tạo interactive 3D plot với Plotly"""
X, Y, Z = self.interpolate(method="rbf")
fig = go.Figure(data=[go.Surface(
x=X,
y=Y * 365, # Convert sang days
z=Z * 100, # Convert sang percentage
colorscale='Viridis',
colorbar=dict(
title=dict(text='IV (%)', font=dict(size=14)),
),
hovertemplate='Moneyness: %{x:.2f}
TTE: %{y:.0f} days
IV: %{z:.2f}%'
)])
fig.update_layout(
title=dict(
text='OKX Options Implied Volatility Surface',
font=dict(size=20)
),
scene=dict(
xaxis_title='Moneyness (S/K)',
yaxis_title='Time to Expiry (days)',
zaxis_title='Implied Volatility (%)',
camera=dict(
eye=dict(x=1.5, y=1.5, z=1.2)
)
),
width=900,
height=700,
margin=dict(l=65, r=50, b=65, t=90)
)
fig.write_html(output_path)
print(f"✓ Surface saved to {output_path}")
return fig
def plot_smile_slice(self, expiry_days: int = 30, output_path: str = "smile.png"):
"""Plot volatility smile cho một expiry cụ thể"""
# Filter data cho expiry gần nhất
target_tte = expiry_days / 365
tolerance = 0.02
calls_data = [(d["moneyness"], d["iv"]) for d in self.smile_data
if abs(d["tte"] - target_tte) < tolerance and d.get("delta", 0.5) > 0]
puts_data = [(d["moneyness"], d["iv"]) for d in self.smile_data
if abs(d["tte"] - target_tte) < tolerance and d.get("delta", 0.5) < 0.5]
plt.figure(figsize=(12, 6))
if calls_data:
calls_data.sort()
m, iv = zip(*calls_data)
plt.plot(m, [v*100 for v in iv], 'b-o', label='Calls', markersize=8)
if puts_data:
puts_data.sort()
m, iv = zip(*puts_data)
plt.plot(m, [v*100 for v in iv], 'r-s', label='Puts', markersize=8)
plt.axvline(x=1.0, color='green', linestyle='--', alpha=0.5, label='ATM')
plt.xlabel('Moneyness (S/K)', fontsize=12)
plt.ylabel('Implied Volatility (%)', fontsize=12)
plt.title(f'Volatility Smile - {expiry_days} Days to Expiry', fontsize=14)
plt.legend()
plt.grid(True, alpha=0.3)
plt.savefig(output_path, dpi=150, bbox_inches='tight')
plt.close()
print(f"✓ Smile plot saved to {output_path}")
Ví dụ sử dụng
def demo_volatility_surface():
builder = VolatilitySurfaceBuilder(min_tte=7/365, max_tte=180/365)
# Thêm dữ liệu mẫu (simulated smile data)
for tte_days in [7, 14, 30, 60, 90]:
tte = tte_days / 365
# Simulate typical volatility smile
for moneyness in np.linspace(0.8, 1.2, 15):
# Base IV + smile effect
base_iv = 0.65
smile_effect = 0.15 * np.exp(-((moneyness - 1.0)**2) / 0.02)
wing_effect = 0.10 * np.abs(moneyness - 1.0)
iv = base_iv + smile_effect + wing_effect
# Add some noise
iv += np.random.normal(0, 0.02)
builder.add_observation(moneyness, tte, iv)
# Build surface
builder.interpolate()
# Generate plots
builder.plot_3d_surface("btc_vol_surface.html")
builder.plot_smile_slice(30, "btc_vol_smile_30d.png")
demo_volatility_surface()
Pipeline Hoàn chỉnh: Từ Tardis đến Analysis
"""
Complete Pipeline: Tardis OKX → HolySheep → Greeks Analysis → Vol Surface
"""
import asyncio
import pandas as pd
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import json
import os
class OKXOptionsAnalysisPipeline:
"""
Pipeline hoàn chỉnh để phân tích OKX Options thông qua HolySheep/Tardis
"""
def __init__(
self,
symbol: str = "BTC-USD",
holysheep_key: Optional[str] = None
):
self.symbol = symbol
self.client = HolySheepClient(
api_key=holysheep_key or os.getenv("HOLYSHEEP_API_KEY")
)
self.fetcher = OKXOptionsDataFetcher(self.client)
self.bs_engine = BlackScholesEngine(r=0.05)
self.surface_builder = VolatilitySurfaceBuilder()
# Data storage
self.chains: List[OptionsChain] = []
self.greeks_history: List[Dict] = []
async def run_full_pipeline(self) -> Dict:
"""
Chạy pipeline đầy đủ:
1. Fetch dữ liệu từ Tardis qua HolySheep
2. Tính toán Greeks
3. Build volatility surface
4. Generate analysis report
"""
print(f"\n{'='*60}")
print(f" OKX Options Analysis Pipeline")
print(f" Symbol: {self.symbol}")
print(f" Started: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"{'='*60}\n")
# Step 1: Fetch data
print("📡 [1/4] Fetching options data from OKX via HolySheep...")
start_time = datetime.now()
self.chains = await self.fetcher.fetch_all_expiries(
self.symbol,
max_expiries=6
)
fetch_time = (datetime.now() - start_time).total_seconds()
print(f"✓ Fetched {len(self.chains)} expiry chains in {fetch_time:.2f}s")
# Step 2: Calculate Greeks
print("\n🧮 [2/4] Calculating Greeks and IV...")
for chain in self.chains:
for option in chain.options:
# Calculate IV if not provided
if option.iv == 0:
params = BSParams(
S=option.spot_price,
K=option.strike,
T=option.time_to_expiry_days / 365,
r=0.05,
sigma=0.5
)
option.iv = calculate_implied_volatility(
market_price=option.price if hasattr(option, 'price') else 100,
params=params,
is_call=(option.option_type == OptionType.CALL)
)
# Add to surface builder
self.surface_builder.add_observation(
moneyness=option.moneyness,
tte=option.time_to_expiry_days / 365,
iv=option.iv,
delta=option.delta
)
# Store in history
self.greeks_history.append(option.to_dict())
print(f"✓ Processed {len(self.greeks_history)} option records")
# Step 3: Build volatility surface
print("\n📊 [3/4] Building volatility surface...")
X, Y, Z = self.surface_builder.interpolate(method="rbf")
self.surface_builder.plot_3d_surface(f"{self.symbol.replace('-','_')}_vol_surface.html")
self.surface_builder.plot_smile_slice(30, f"{self.symbol.replace('-','_')}_smile_30d.png
Tài nguyên liên quan
Bài viết liên quan