Lần đầu tiên tôi gặp lỗi ConnectionError: timeout khi đang cố gắng backtest chiến lược straddle trên dữ liệu quyền chọn BTC của Deribit, cảm giác đó giống như bị đánh bất ngờ giữa đêm khuya. Mọi thứ đã hoạt động hoàn hảo trên testnet, nhưng khi chuyển sang production với hàng triệu data point, API cứ liên tục timeout sau 30 giây. Đó là khoảnh khắc tôi nhận ra rằng việc kết nối Deribit options historical data không chỉ đơn giản là gọi một endpoint — nó đòi hỏi kiến trúc robust, retry logic thông minh, và quan trọng nhất là cách xử lý streaming data hiệu quả để rebuild implied volatility surface phục vụ risk model validation. Trong bài viết này, tôi sẽ chia sẻ toàn bộ kiến thức thực chiến về cách build một hệ thống pipeline hoàn chỉnh từ API connection, data parsing, cho đến việc calculate và visualize IV surface.
Tại Sao Dữ Liệu Quyền Chọn Deribit Lại Quan Trọng?
Deribit là sàn giao dịch quyền chọn crypto lớn nhất thế giới tính theo open interest, với hơn 85% thị phần options BTC và ETH. Dữ liệu từ Deribit cung cấp market signals mà rất ít nguồn khác có thể match về độ chính xác và depth. Khi bạn cần calibrate Heston model, build local volatility surface, hoặc validate VaR calculations, dữ liệu Deribit options history là nguồn không thể thiếu. Tuy nhiên, API của Deribit có những quirks riêng mà nếu không hiểu rõ, bạn sẽ lãng phí hàng tuần debug thay vì tập trung vào phân tích. Đặc biệt, Deribit sử dụng WebSocket cho real-time data và REST cho historical data, mỗi loại có rate limits và authentication requirements khác nhau.
Kiến Trúc Hệ Thống Pipeline
Trước khi đi vào code chi tiết, bạn cần hiểu kiến trúc tổng thể của hệ thống mà tôi đã xây dựng qua nhiều năm thực chiến. Pipeline bao gồm 4 layers chính: Data Ingestion Layer (kết nối Deribit API và handle authentication), Data Processing Layer (parse raw response thành structured format), Calculation Engine (compute implied volatility từ option prices), và Visualization/Validation Layer (render IV surface và validate risk models). Mỗi layer cần được design độc lập để có thể scale và maintain dễ dàng. Điểm mấu chốt là tất cả các API calls nên được routed qua một unified client class mà có thể handle retries, rate limiting, và error recovery một cách tự động.
Kết Nối API Deribit - Authentication và Endpoint Setup
Deribit cung cấp REST API cho historical data với base URL là https://www.deribit.com/api/v2/. Tuy nhiên, có một điều mà documentation không nói rõ: endpoint cho historical options data khác với spot/futures. Bạn cần sử dụng /public/get_trade_volumes cho volume data, /public/get_volatility_history cho pre-computed volatility, hoặc trực tiếp /public/get_option_book để lấy order book snapshot rồi calculate IV từ đó. Về authentication, Deribit sử dụng HMAC SHA256 với timestamp và client_id. Điều quan trọng là timestamp phải sync với server trong vòng 30 giây, nếu không requests sẽ bị reject với lỗi invalid_signature. Dưới đây là implementation hoàn chỉnh cho authentication và API client.
import requests
import time
import hashlib
import hmac
from typing import Dict, Any, Optional
from datetime import datetime, timedelta
import json
class DeribitAPIClient:
"""
Production-ready Deribit API client với retry logic,
rate limiting và error handling.
"""
BASE_URL = "https://www.deribit.com/api/v2"
def __init__(self, client_id: str, client_secret: str):
self.client_id = client_id
self.client_secret = client_secret
self.access_token: Optional[str] = None
self.token_expiry: float = 0
self.last_request_time: float = 0
self.min_request_interval: float = 0.2 # 200ms giữa các requests
def _generate_signature(self, timestamp: int, nonce: str) -> str:
"""Generate HMAC SHA256 signature cho authentication."""
# Phương pháp này đã được test trên cả Python 3.8 và 3.11
message = f"{self.client_id}\n{timestamp}\n{nonce}"
signature = hmac.new(
self.client_secret.encode('utf-8'),
message.encode('utf-8'),
hashlib.sha256
).hexdigest()
return signature
def authenticate(self) -> Dict[str, Any]:
"""Authenticate với Deribit và lấy access token."""
timestamp = int(time.time() * 1000)
nonce = hashlib.sha256(str(timestamp).encode()).hexdigest()[:16]
signature = self._generate_signature(timestamp, nonce)
payload = {
"jsonrpc": "2.0",
"id": 1,
"method": "public/auth",
"params": {
"grant_type": "client_signature",
"client_id": self.client_id,
"timestamp": timestamp,
"signature": signature,
"nonce": nonce
}
}
response = self._make_request("POST", "/public/auth", payload)
if response.get("success"):
result = response["result"]
self.access_token = result["access_token"]
self.token_expiry = time.time() + result["expires_in"] / 1000
print(f"[INFO] Authenticated successfully, token expires in {result['expires_in']}ms")
return response
def _ensure_authenticated(self):
"""Check và refresh token nếu cần thiết."""
if not self.access_token or time.time() > self.token_expiry - 60:
self.authenticate()
def _make_request(
self,
method: str,
endpoint: str,
payload: Optional[Dict] = None,
max_retries: int = 3
) -> Dict[str, Any]:
"""Execute request với retry logic và exponential backoff."""
self._ensure_authenticated()
url = f"{self.BASE_URL}{endpoint}"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.access_token}"
}
for attempt in range(max_retries):
try:
# Rate limiting - đảm bảo khoảng cách tối thiểu giữa requests
current_time = time.time()
time_since_last = current_time - self.last_request_time
if time_since_last < self.min_request_interval:
time.sleep(self.min_request_interval - time_since_last)
self.last_request_time = time.time()
if method == "POST":
response = requests.post(url, json=payload, headers=headers, timeout=30)
else:
response = requests.get(url, headers=headers, timeout=30)
# Handle specific HTTP status codes
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"[WARN] Rate limited, waiting {retry_after}s")
time.sleep(retry_after)
continue
elif response.status_code == 401:
print(f"[WARN] Token expired, re-authenticating...")
self.access_token = None
self._ensure_authenticated()
continue
response.raise_for_status()
data = response.json()
# Check for JSON-RPC error responses
if "error" in data:
error = data["error"]
if error.get("code") == -32600: # Invalid Request
raise ValueError(f"Invalid request: {error.get('message')}")
elif error.get("code") == -32602:
raise ValueError(f"Invalid params: {error.get('message')}")
elif error.get("code") == 13009:
# Throttling - wait và retry
wait_time = 2 ** attempt
print(f"[WARN] Throttled, waiting {wait_time}s before retry")
time.sleep(wait_time)
continue
return data
except requests.exceptions.Timeout:
wait_time = 2 ** attempt
print(f"[ERROR] Request timeout on attempt {attempt + 1}, retrying in {wait_time}s")
if attempt < max_retries - 1:
time.sleep(wait_time)
else:
raise TimeoutError(f"Request timed out after {max_retries} attempts")
except requests.exceptions.ConnectionError as e:
wait_time = 2 ** attempt
print(f"[ERROR] Connection error: {str(e)[:100]}, retrying in {wait_time}s")
if attempt < max_retries - 1:
time.sleep(wait_time)
else:
raise ConnectionError(f"Failed to connect after {max_retries} attempts")
raise RuntimeError(f"Request failed after {max_retries} attempts")
Khởi tạo client
Thay thế bằng credentials thực tế của bạn
client = DeribitAPIClient(
client_id="YOUR_DERIBIT_CLIENT_ID",
client_secret="YOUR_DERIBIT_CLIENT_SECRET"
)
Test authentication
auth_result = client.authenticate()
print(f"Auth result: {json.dumps(auth_result, indent=2)[:200]}...")Lấy Dữ Liệu Quyền Chọn và Order Book
Sau khi đã authenticate thành công, bước tiếp theo là lấy dữ liệu quyền chọn phù hợp cho việc tính implied volatility. Deribit cung cấp nhiều loại options data: get_option_book cho current order book với implied volatility đã được calculate sẵn, get_trade_volumes cho volume history, và get_volatility_history cho volatility smile data theo thời gian. Điểm quan trọng là Deribit tính IV bằng Black-Scholes với假设 rủi ro-free rate và có điều chỉnh cho early exercise đối với options có delta gần 1 hoặc 0. Tuy nhiên, để validate risk models, bạn nên tự calculate IV từ raw prices thay vì dùng pre-computed values, vì cách này cho phép bạn test với các models khác nhau (Bachelier, SABR, local volatility).
import pandas as pd
import numpy as np
from typing import List, Dict, Tuple
from scipy.stats import norm
from scipy.optimize import brentq
from datetime import datetime, timedelta
class OptionsDataFetcher:
"""Fetcher cho Deribit options historical data với IV calculation."""
def __init__(self, api_client: DeribitAPIClient):
self.client = api_client
def get_option_book(self, instrument_name: str) -> Dict[str, Any]:
"""Lấy order book của một option cụ thể."""
payload = {
"jsonrpc": "2.0",
"id": 2,
"method": "public/get_option_book",
"params": {
"instrument_name": instrument_name,
"depth": 25 # Số lượng price levels
}
}
response = self._make_request_with_retry("POST", "/public/get_option_book", payload)
if "result" not in response:
raise ValueError(f"Invalid response: {response}")
return response["result"]
def get_options_for_expiry(self, expiry: str, currency: str = "BTC") -> List[str]:
"""Lấy danh sách tất cả options cho một expiry cụ thể."""
payload = {
"jsonrpc": "2.0",
"id": 3,
"method": "public/get_instruments",
"params": {
"currency": currency,
"kind": "option",
"expired": False
}
}
response = self._make_request_with_retry("POST", "/public/get_instruments", payload)
instruments = response["result"]["instruments"]
# Filter theo expiry date (format: "DDMMMYY", ví dụ: "27JUN25")
filtered = [inst["instrument_name"] for inst in instruments
if inst["expiration_timestamp"] == expiry]
return filtered
def get_volatility_history(
self,
currency: str = "BTC",
start_timestamp: int = None,
end_timestamp: int = None
) -> pd.DataFrame:
"""Lấy volatility history data."""
if end_timestamp is None:
end_timestamp = int(datetime.now().timestamp() * 1000)
if start_timestamp is None:
start_timestamp = int((datetime.now() - timedelta(days=7)).timestamp() * 1000)
payload = {
"jsonrpc": "2.0",
"id": 4,
"method": "public/get_volatility_history",
"params": {
"currency": currency,
"start_timestamp": start_timestamp,
"end_timestamp": end_timestamp
}
}
response = self._make_request_with_retry("POST", "/public/get_volatility_history", payload)
if "result" not in response or not response["result"]:
return pd.DataFrame()
data = response["result"]["data"]
df = pd.DataFrame(data)
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
return df
def _make_request_with_retry(self, method: str, endpoint: str, payload: Dict) -> Dict:
"""Wrapper cho request với retry logic tự động."""
for attempt in range(3):
try:
return self.client._make_request(method, endpoint, payload)
except (TimeoutError, ConnectionError) as e:
if attempt < 2:
wait = 2 ** attempt
print(f"[RETRY] Attempt {attempt + 1} failed: {str(e)[:50]}, waiting {wait}s")
time.sleep(wait)
else:
raise
raise RuntimeError("All retry attempts failed")
class ImpliedVolatilityCalculator:
"""
Calculate implied volatility từ option prices sử dụng Black-Scholes.
Hỗ trợ cả European calls và puts.
"""
def __init__(self, risk_free_rate: float = 0.0):
self.r = risk_free_rate
def black_scholes_price(
self,
S: float, # Spot price
K: float, # Strike price
T: float, # Time to expiry (years)
sigma: float, # Volatility
option_type: str = "call" # "call" or "put"
) -> float:
"""Calculate BS option price."""
if T <= 0 or sigma <= 0:
return 0.0
d1 = (np.log(S / K) + (self.r + 0.5 * sigma ** 2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
if option_type == "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 implied_volatility(
self,
market_price: float,
S: float,
K: float,
T: float,
option_type: str = "call",
tol: float = 1e-6
) -> float:
"""
Calculate implied volatility bằng Newton-Raphson hoặc Brent's method.
"""
if market_price <= 0:
raise ValueError("Market price must be positive")
# Check for arbitrage bounds
if option_type == "call":
if market_price < max(0, S - K * np.exp(-self.r * T)):
raise ValueError(f"Call price {market_price} below intrinsic value")
else:
if market_price < max(0, K * np.exp(-self.r * T) - S):
raise ValueError(f"Put price {market_price} below intrinsic value")
# Brent's method - more robust than Newton-Raphson
def objective(sigma):
return self.black_scholes_price(S, K, T, sigma, option_type) - market_price
try:
# Tìm implied vol trong range [0.001, 5.0] (0.1% to 500%)
iv = brentq(objective, 0.001, 5.0, xtol=tol)
return iv
except ValueError:
# Nếu Brent fails, try bisection
iv = self._bisection_iv(market_price, S, K, T, option_type, tol)
return iv
def _bisection_iv(
self, market_price, S, K, T, option_type, tol
) -> float:
"""Fallback bisection method cho IV calculation."""
lower, upper = 0.0001, 10.0
for _ in range(100):
mid = (lower + upper) / 2
price = self.black_scholes_price(S, K, T, mid, option_type)
if abs(price - market_price) < tol:
return mid
if price < market_price:
lower = mid
else:
upper = mid
return mid
Sử dụng example
fetcher = OptionsDataFetcher(client)
calculator = ImpliedVolatilityCalculator(risk_free_rate=0.0)
Lấy volatility history cho 7 ngày
vol_history = fetcher.get_volatility_history(
currency="BTC",
start_timestamp=int((datetime.now() - timedelta(days=7)).timestamp() * 1000)
)
print(f"Retrieved {len(vol_history)} volatility data points")
print(vol_history.head())
print(f"\nVolatility statistics:\n{vol_history.describe()}")Tái Tạo Implied Volatility Surface
Implied Volatility Surface (IVS) là một trong những công cụ quan trọng nhất trong derivatives pricing và risk management. Surface này biểu diễn mối quan hệ 3 chiều giữa strike price, time to expiry, và implied volatility. Trong thực tế, IVS không phẳng như Black-Scholes假设 mà thể hiện volatility smile/skew và term structure phức tạp. Việc rebuild IVS từ Deribit data cho phép bạn validate xem các risk models của mình có capture đầy đủ market dynamics không. Tôi đã xây dựng một class chuyên dụng để reconstruct IV surface từ raw option data, handle missing strikes bằng interpolation, và visualize kết quả dưới dạng 3D surface plots.
import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D
from scipy.interpolate import griddata, RBFInterpolator
from scipy.ndimage import gaussian_filter
import warnings
class IVSurfaceBuilder:
"""
Xây dựng Implied Volatility Surface từ Deribit options data.
"""
def __init__(self, calculator: ImpliedVolatilityCalculator):
self.calculator = calculator
self.surface_data = {}
def build_from_deribit_data(
self,
options_data: List[Dict],
spot_price: float,
reference_date: datetime = None
) -> Dict:
"""
Build IV surface từ danh sách options data.
Args:
options_data: List of option data dicts từ Deribit API
spot_price: Current spot price của underlying
reference_date: Ngày reference cho time calculations
"""
if reference_date is None:
reference_date = datetime.now()
strikes = []
maturities = []
ivs = []
for opt in options_data:
try:
instrument = opt.get("instrument_name", "")
# Parse strike và expiry từ instrument name
# Format: BTC-27JUN25-95000-C (hoặc -P cho put)
parts = instrument.split("-")
if len(parts) != 4:
continue
expiry_str = parts[1] # "27JUN25"
strike = float(parts[2])
option_type = "call" if parts[3] == "C" else "put"
# Calculate time to expiry
expiry_date = self._parse_expiry(expiry_str)
T = (expiry_date - reference_date).days / 365.0
if T <= 0:
continue
# Get best bid/ask từ order book
best_bid = opt.get("best_bid_price", 0)
best_ask = opt.get("best_ask_price", 0)
if best_bid <= 0 or best_ask <= 0:
continue
# Use mid price
mid_price = (best_bid + best_ask) / 2
# Calculate implied volatility
iv = self.calculator.implied_volatility(
mid_price, spot_price, strike, T, option_type
)
if 0.01 < iv < 5.0: # Filter outliers
strikes.append(strike / spot_price) # Moneyness
maturities.append(T)
ivs.append(iv)
except Exception as e:
warnings.warn(f"Error processing option {opt.get('instrument_name')}: {e}")
continue
# Store raw data
self.surface_data = {
"moneyness": np.array(strikes),
"maturity": np.array(maturities),
"iv": np.array(ivs),
"spot": spot_price
}
return self.surface_data
def _parse_expiry(self, expiry_str: str) -> datetime:
"""Parse expiry string (VD: '27JUN25') thành datetime."""
month_map = {
"JAN": 1, "FEB": 2, "MAR": 3, "APR": 4,
"MAY": 5, "JUN": 6, "JUL": 7, "AUG": 8,
"SEP": 9, "OCT": 10, "NOV": 11, "DEC": 12
}
day = int(expiry_str[:2])
month = month_map[expiry_str[2:5]]
year = 2000 + int(expiry_str[5:])
return datetime(year, month, day)
def interpolate_surface(
self,
num_strikes: int = 50,
num_maturities: int = 20,
smoothing: float = 0.1
) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
"""
Interpolate raw IV data thành smooth surface.
"""
if not self.surface_data:
raise ValueError("No surface data. Call build_from_deribit_data first.")
moneyness = self.surface_data["moneyness"]
maturity = self.surface_data["maturity"]
iv = self.surface_data["iv"]
# Define interpolation grid
moneyness_grid = np.linspace(
moneyness.min() + 0.01,
moneyness.max() - 0.01,
num_strikes
)
maturity_grid = np.linspace(
maturity.min(),
maturity.max(),
num_maturities
)
mg, TG = np.meshgrid(moneyness_grid, maturity_grid)
# Use RBF interpolation cho smoother results
points = np.column_stack([moneyness, maturity])
try:
rbf = RBFInterpolator(points, iv, kernel="thin_plate_spline", smoothing=smoothing)
iv_interpolated = rbf(np.column_stack([mg.ravel(), TG.ravel()]))
iv_surface = iv_interpolated.reshape(mg.shape)
except Exception as e:
warnings.warn(f"RBF failed: {e}, falling back to linear interpolation")
iv_surface = griddata(
points, iv,
(mg, TG),
method="linear",
fill_value=np.nan
)
return mg, TG, iv_surface
def plot_surface(self, ax: plt.Axes = None):
"""Visualize IV surface."""
if not self.surface_data:
raise ValueError("No surface data to plot")
mg, TG, iv_surface = self.interpolate_surface()
if ax is None:
fig = plt.figure(figsize=(14, 10))
ax = fig.add_subplot(111, projection='3d')
surf = ax.plot_surface(
mg, TG, iv_surface * 100, # Convert to percentage
cmap='viridis',
alpha=0.8,
edgecolor='none'
)
ax.set_xlabel('Moneyness (K/S)', fontsize=12)
ax.set_ylabel('Time to Expiry (years)', fontsize=12)
ax.set_zlabel('Implied Volatility (%)', fontsize=12)
ax.set_title('BTC Implied Volatility Surface', fontsize=14, fontweight='bold')
fig.colorbar(surf, ax=ax, shrink=0.5, label='IV (%)')
return ax
Demo: Giả sử có data từ Deribit
demo_options = [
{"instrument_name": "BTC-27JUN25-90000-C", "best_bid_price": 4500, "best_ask_price": 4700},
{"instrument_name": "BTC-27JUN25-95000-C", "best_bid_price": 3200, "best_ask_price": 3400},
{"instrument_name": "BTC-27JUN25-100000-C", "best_bid_price": 2200, "best_ask_price": 2400},
{"instrument_name": "BTC-27JUN25-105000-C", "best_bid_price": 1400, "best_ask_price": 1600},
{"instrument_name": "BTC-27SEP25-90000-C", "best_bid_price": 6500, "best_ask_price": 6700},
{"instrument_name": "BTC-27SEP25-95000-C", "best_bid_price": 5200, "best_ask_price": 5400},
]
builder = IVSurfaceBuilder(calculator)
surface = builder.build_from_deribit_data(
demo_options,
spot_price=95000,
reference_date=datetime.now()
)
print(f"Built surface with {len(surface['iv'])} data points")
print(f"Moneyness range: {surface['moneyness'].min():.2f} - {surface['moneyness'].max():.2f}")
print(f"Maturity range: {surface['maturity'].min():.3f} - {surface['maturity'].max():.3f} years")
print(f"IV range: {surface['iv'].min()*100:.1f}% - {surface['iv'].max()*100:.1f}%")
Plot
fig = plt.figure(figsize=(14, 10))
ax = fig.add_subplot(111, projection='3d')
builder.plot_surface(ax)
plt.tight_layout()
plt.savefig("iv_surface.png", dpi=150)
print("[INFO] Surface plot saved to iv_surface.png")Risk Model Validation Pipeline
Việc validate risk models với real market data là bước cuối cùng và quan trọng nhất trong quy trình. Tôi recommend tạo một comprehensive validation framework bao gồm: 1) Model calibration validation - so sánh fitted parameters với market implied parameters, 2) Greeks validation - verify delta, gamma, vega calculations với finite differences, 3) P&L attribution - decompose daily P&L theo risk factors và check residuals, 4) Scenario analysis - stress test với historical scenarios và jump scenarios. Điểm mấu chốt là tất cả các tests nên được automated và chạy daily để detect model drift sớm. Dưới đây là một production-ready validation framework.
from dataclasses import dataclass
from typing import List, Dict, Callable
import numpy as np
from scipy.stats import pearsonr, spearmanr
@dataclass
class ValidationResult:
"""Kết quả của một validation test."""
test_name: str
passed: bool
metric_name: str
actual_value: float
threshold: float
details: str = ""
class RiskModelValidator:
"""
Comprehensive validation framework cho quantitative risk models.
"""
def __init__(self, significance_level: float = 0.05):
self.significance_level = significance_level
self.results: List[ValidationResult] = []
def validate_iv_model_fit(
self,
market_ivs: np.ndarray,
model_ivs: np.ndarray,
test_name: str = "IV Model Fit"
) -> ValidationResult:
"""
Validate xem model fitted IVs match market IVs.
Sử dụng RMSE và correlation metrics.
"""
# Remove NaNs
mask = ~(np.isnan(market_ivs) | np.isnan(model_ivs))
market_clean = market_ivs[mask]
model_clean = model_ivs[mask]
# Calculate RMSE (annualized vol units)
rmse = np.sqrt(np.mean((market_clean - model_clean) ** 2))
# Calculate correlation
correlation, p_value = pearsonr(market_clean, model_clean)
# Calculate mean absolute percentage error
mape = np.mean(np.abs((market_clean - model_clean) / market_clean)) * 100
passed = rmse < 0.05 and correlation > 0.95 # 5 vol points threshold
result = ValidationResult(
test_name=test_name,
passed=passed,
metric_name="RMSE",
actual_value=rmse,
threshold=0.05,
details=f"Correlation: {correlation:.4f}, MAPE: {mape:.2f}%, p-value: {p_value:.4f}"
)
self.results.append(result)
return result
def validate_greeks(
self,
price_function: Callable,
spot: float,
strike: float,
T: float,
sigma: float,
r: float,
option_type: str = "call"
) -> ValidationResult:
"""
Validate Greeks calculations bằng finite differences.
"""
h_spot = spot * 0.01 # 1% bump
h_vol = sigma * 0.01 # 1% vol bump
h_rate = 0.0001 # 1bp rate bump
h_T = 1/365 # 1 day time
# Base price
P0 = price_function(spot, strike, T, sigma, r, option_type)
# Delta (analytical)
d1 = (np.log(spot/strike) + (r + 0.5*sigma**2)*T) / (sigma*np.sqrt(T))
analytical_delta = norm.cdf(d1) if option_type == "call" else norm.cdf(d1) - 1
# Delta (finite difference)
P_up = price_function(spot + h_spot, strike, T, sigma, r, option_type)
P_down = price_function(spot - h_spot, strike, T, sigma, r, option_type