Trong thị trường crypto derivatives, dữ liệu quyền chọn Deribit là nguồn tài nguyên quan trọng nhất cho các chiến lược giao dịch volatility arbitrage và delta hedging. Bài viết này sẽ hướng dẫn bạn cách truy cập API lấy dữ liệu lịch sử, xây dựng volatility surface và thực hiện backtesting với chi phí tối ưu nhất.
Bối cảnh thị trường: Chi phí AI/ML trong Trading năm 2026
Trước khi đi vào chi tiết kỹ thuật, hãy cùng xem bức tranh chi phí thực tế khi bạn xây dựng hệ thống trading dựa trên AI:
| Model | Giá/MTok | Chi phí 10M token/tháng | Độ trễ trung bình |
|---|---|---|---|
| GPT-4.1 (OpenAI) | $8.00 | $80 | ~800ms |
| Claude Sonnet 4.5 (Anthropic) | $15.00 | $150 | ~1200ms |
| Gemini 2.5 Flash (Google) | $2.50 | $25 | ~400ms |
| DeepSeek V3.2 (China) | $0.42 | $4.20 | ~600ms |
| HolySheep AI | $0.42 | $4.20 | <50ms |
Như bạn thấy, đăng ký HolySheep AI mang lại mức giá cạnh tranh nhất thị trường với chi phí chỉ $4.20/10M token — tiết kiệm 85%+ so với Claude Sonnet 4.5. Đặc biệt, độ trễ dưới 50ms là yếu tố then chốt cho các ứng dụng trading real-time.
Giới thiệu Deribit Options API
Deribit cung cấp REST API và WebSocket endpoint cho phép truy cập dữ liệu quyền chọn với độ trễ thấp. Đây là nền tảng giao dịch quyền chọn Bitcoin và Ethereum lớn nhất thế giới tính theo open interest.
Các endpoint quan trọng cần biết
/public/get_book_summary_by_currency— Tóm tắt sổ lệnh theo đơn vị tiền tệ/public/get_volatility_history— Dữ liệu volatility lịch sử/public/get_options_by_currency— Danh sách quyền chọn theo currency/public/get_last_trades_by_currency— Giao dịch gần nhất/public/get_trade_volumes— Khối lượng giao dịch
Cài đặt môi trường và Authentication
# Cài đặt các thư viện cần thiết
pip install requests pandas numpy pyarrow matplotlib scipy
pip install deribit-api # Official Deribit Python SDK
Hoặc sử dụng thư viện tự viết nhẹ hơn
pip install httpx asyncio pandasimport requests
import pandas as pd
from datetime import datetime, timedelta
import time
Cấu hình API Deribit
BASE_URL = "https://www.deribit.com/api/v2"
CLIENT_ID = "your_deribit_client_id"
CLIENT_SECRET = "your_deribit_client_secret"
class DeribitOptionsData:
def __init__(self):
self.access_token = None
self.token_expires = 0
def authenticate(self, client_id, client_secret):
"""Xác thực và lấy access token"""
url = f"{BASE_URL}/public/auth"
payload = {
"method": "public/auth",
"params": {
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret
},
"jsonrpc": "2.0",
"id": 1
}
response = requests.post(url, json=payload).json()
self.access_token = response['result']['access_token']
self.token_expires = time.time() + 3600 # Token hết hạn sau 1 giờ
return self.access_token
def get_headers(self):
"""Kiểm tra và làm mới token nếu cần"""
if not self.access_token or time.time() >= self.token_expires - 60:
self.authenticate(CLIENT_ID, CLIENT_SECRET)
return {"Authorization": f"Bearer {self.access_token}"}
Khởi tạo đối tượng
deribit = DeribitOptionsData()Lấy dữ liệu quyền chọn BTC/ETH
import pandas as pd
from typing import List, Dict
import warnings
warnings.filterwarnings('ignore')
class OptionsDataFetcher:
def __init__(self, base_url: str = "https://www.deribit.com/api/v2"):
self.base_url = base_url
self.session = requests.Session()
def get_options_chain(self, currency: str = "BTC", expiration: str = None) -> pd.DataFrame:
"""
Lấy toàn bộ chain quyền chọn cho một đồng tiền
Args:
currency: 'BTC' hoặc 'ETH'
expiration: Ngày hết hạn (format: 'DDMMYY', ví dụ: '27DEC24')
Nếu None, lấy tất cả các expiration
"""
params = {
"currency": currency,
"expired": "false" # Chỉ lấy quyền chọn chưa hết hạn
}
url = f"{self.base_url}/public/get_options_by_currency"
response = self.session.get(url, params=params)
data = response.json()
if 'result' not in data:
raise ValueError(f"API Error: {data}")
options = data['result']
# Chuyển đổi sang DataFrame
records = []
for opt in options:
record = {
'instrument_name': opt['instrument_name'],
'kind': opt['kind'], # 'call' hoặc 'put'
'expiration_timestamp': opt['expiration_timestamp'],
'expiration_date': pd.to_datetime(opt['expiration_timestamp'], unit='ms'),
'strike': opt['strike'],
'base_currency': opt['base_currency'],
'quote_currency': opt['quote_currency'],
'settlement_price': opt.get('settlement_price', None),
'open_interest': opt.get('open_interest', 0),
'creation_timestamp': opt.get('creation_timestamp', None),
'bid_price': opt.get('best_bid_price', None),
'ask_price': opt.get('best_ask_price', None),
'mark_price': opt.get('mark_price', None),
'underlying_price': opt.get('underlying_price', None),
'underlying_index': opt.get('underlying_index', None),
'interest_ask': opt.get('interest_ask', None),
'interest_bid': opt.get('interest_bid', None),
'ask_iv': opt.get('best_ask_iv', None),
'bid_iv': opt.get('best_bid_iv', None),
'mark_iv': opt.get('mark_iv', None),
'delta': opt.get('delta', None),
'gamma': opt.get('gamma', None),
'rho': opt.get('rho', None),
'theta': opt.get('theta', None),
'vega': opt.get('vega', None)
}
records.append(record)
df = pd.DataFrame(records)
# Lọc theo expiration nếu được chỉ định
if expiration:
df = df[df['instrument_name'].str.contains(expiration)]
return df
Sử dụng
fetcher = OptionsDataFetcher()
btc_options = fetcher.get_options_chain(currency="BTC")
eth_options = fetcher.get_options_chain(currency="ETH")
print(f"BTC Options Chain: {len(btc_options)} instruments")
print(f"ETH Options Chain: {len(eth_options)} instruments")
print(f"\nBTC sample:\n{btc_options.head()}")Xây dựng Volatility Surface từ dữ liệu thực tế
Volatility surface là biểu diễn 3 chiều của implied volatility theo strike price và maturity. Đây là công cụ nền tảng cho các chiến lược volatility arbitrage và risk management.
import numpy as np
from scipy.interpolate import griddata, RBFInterpolator
from scipy.stats import norm
from datetime import datetime
import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D
class VolatilitySurfaceBuilder:
def __init__(self):
self.surface = None
self.strikes = None
self.tenors = None
self.spot = None
@staticmethod
def calculate_time_to_expiry(expiry_date: datetime, current_date: datetime = None) -> float:
"""Tính thời gian đến hết hạn theo năm (Act/365)"""
if current_date is None:
current_date = datetime.now()
delta = expiry_date - current_date
return delta.days / 365.0
@staticmethod
def black_scholes_iv(
S: float, # Spot price
K: float, # Strike price
T: float, # Time to expiry (years)
r: float, # Risk-free rate
market_price: float, # Observed option price
option_type: str = 'call'
) -> float:
"""
Tính implied volatility bằng Newton-Raphson method
"""
if T <= 0 or market_price <= 0:
return np.nan
# Initial guess using ATM approximation
moneyness = np.log(S / K) / np.sqrt(T)
sigma = 0.3 + 0.1 * abs(moneyness)
# Newton-Raphson iteration
for _ in range(100):
d1 = (np.log(S / K) + (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(-r * T) * norm.cdf(d2)
else:
price = K * np.exp(-r * T) * norm.cdf(-d2) - S * norm.cdf(-d1)
vega = S * np.sqrt(T) * norm.pdf(d1)
if vega < 1e-10:
break
diff = market_price - price
if abs(diff) < 1e-8:
break
sigma += diff / vega
sigma = max(0.01, min(sigma, 5.0)) # Bound IV
return sigma
def build_from_options_data(
self,
df: pd.DataFrame,
spot_price: float,
risk_free_rate: float = 0.05
) -> pd.DataFrame:
"""
Xây dựng volatility surface từ dữ liệu quyền chọn
Args:
df: DataFrame chứa dữ liệu quyền chọn
spot_price: Giá spot hiện tại của underlying
risk_free_rate: Lãi suất phi rủi ro (annualized)
"""
self.spot = spot_price
current_time = datetime.now()
# Lọc dữ liệu hợp lệ
valid_options = df[
(df['bid_price'].notna()) &
(df['ask_price'].notna()) &
(df['bid_price'] > 0) &
(df['ask_price'] > 0)
].copy()
# Tính mid price và moneyness
valid_options['mid_price'] = (valid_options['bid_price'] + valid_options['ask_price']) / 2
valid_options['moneyness'] = valid_options['strike'] / spot_price
valid_options['time_to_expiry'] = valid_options['expiration_date'].apply(
lambda x: self.calculate_time_to_expiry(x, current_time)
)
# Tính implied volatility cho mỗi quyền chọn
iv_list = []
for _, row in valid_options.iterrows():
iv = self.black_scholes_iv(
S=spot_price,
K=row['strike'],
T=row['time_to_expiry'],
r=risk_free_rate,
market_price=row['mid_price'],
option_type=row['kind']
)
iv_list.append(iv)
valid_options['implied_volatility'] = iv_list
# Lọc IV hợp lệ
valid_options = valid_options[
(valid_options['implied_volatility'] > 0.05) &
(valid_options['implied_volatility'] < 3.0) &
(valid_options['time_to_expiry'] > 1/365) # Tối thiểu 1 ngày
]
return valid_options
def create_3d_surface(self, df: pd.DataFrame):
"""Tạo surface 3D từ dữ liệu IV"""
strikes = df['strike'].values
tenors = df['time_to_expiry'].values * 365 # Chuyển sang ngày
ivs = df['implied_volatility'].values
# Tạo grid cho interpolation
strike_grid = np.linspace(strikes.min(), strikes.max(), 50)
tenor_grid = np.linspace(tenors.min(), tenors.max(), 50)
K, T = np.meshgrid(strike_grid, tenor_grid)
# Interpolation bằng RBF
points = np.column_stack([strikes, tenors])
try:
rbf = RBFInterpolator(points, ivs, kernel='thin_plate_spline', smoothing=1)
surface = rbf(np.column_stack([K.ravel(), T.ravel()]))
surface = surface.reshape(K.shape)
# Mask các giá trị ngoài phạm vi
mask = (K < strikes.min()) | (K > strikes.max()) | \
(T < tenors.min()) | (T > tenors.max())
surface = np.ma.array(surface, mask=mask)
return K, T, surface
except Exception as e:
print(f"Interpolation error: {e}")
return None, None, None
Ví dụ sử dụng (giả định có dữ liệu)
builder = VolatilitySurfaceBuilder()
btc_spot = 67500 # Giá BTC spot (ví dụ)
Xây dựng surface
surface_data = builder.build_from_options_data(btc_options, btc_spot)
print(f"Valid options for surface: {len(surface_data)}")
print(f"Strike range: {surface_data['strike'].min():.0f} - {surface_data['strike'].max():.0f}")
print(f"IV range: {surface_data['implied_volatility'].min():.2%} - {surface_data['implied_volatility'].max():.2%}")Volatility Surface Visualization
import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D
import matplotlib.cm as cm
def plot_volatility_smile(df: pd.DataFrame, tenor_days: int, spot: float, title: str):
"""
Vẽ đồ thị volatility smile cho một tenor cụ thể
"""
df_tenor = df[
(df['time_to_expiry'] * 365 >= tenor_days * 0.9) &
(df['time_to_expiry'] * 365 <= tenor_days * 1.1)
].copy()
if len(df_tenor) == 0:
print(f"No data for tenor {tenor_days} days")
return
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
# Plot 1: IV vs Strike (Volatility Smile)
calls = df_tenor[df_tenor['kind'] == 'call'].sort_values('strike')
puts = df_tenor[df_tenor['kind'] == 'put'].sort_values('strike')
axes[0].scatter(calls['strike'], calls['implied_volatility'],
label='Calls', alpha=0.7, color='blue')
axes[0].scatter(puts['strike'], puts['implied_volatility'],
label='Puts', alpha=0.7, color='red')
axes[0].axvline(x=spot, color='green', linestyle='--', label=f'Spot: {spot}')
axes[0].set_xlabel('Strike Price')
axes[0].set_ylabel('Implied Volatility')
axes[0].set_title(f'Volatility Smile - {tenor_days} Days')
axes[0].legend()
axes[0].grid(True, alpha=0.3)
# Plot 2: IV vs Moneyness
df_tenor['moneyness'] = df_tenor['strike'] / spot
axes[1].scatter(df_tenor[df_tenor['kind']=='call']['moneyness'],
df_tenor[df_tenor['kind']=='call']['implied_volatility'],
label='Calls', alpha=0.7, color='blue')
axes[1].scatter(df_tenor[df_tenor['kind']=='put']['moneyness'],
df_tenor[df_tenor['kind']=='put']['implied_volatility'],
label='Puts', alpha=0.7, color='red')
axes[1].axvline(x=1.0, color='green', linestyle='--', label='ATM')
axes[1].set_xlabel('Moneyness (K/S)')
axes[1].set_ylabel('Implied Volatility')
axes[1].set_title(f'IV vs Moneyness - {tenor_days} Days')
axes[1].legend()
axes[1].grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig(f'volatility_smile_{tenor_days}d.png', dpi=150)
plt.show()
def plot_3d_surface(df: pd.DataFrame, spot: float):
"""
Vẽ volatility surface 3D
"""
# Tạo pivot table
df['tenor_bucket'] = (df['time_to_expiry'] * 365).round(0)
df['moneyness_bucket'] = (df['moneyness'] * 100).round(0)
pivot = df.pivot_table(
values='implied_volatility',
index='moneyness_bucket',
columns='tenor_bucket',
aggfunc='mean'
)
if pivot.empty:
print("No data to plot")
return
fig = plt.figure(figsize=(14, 10))
ax = fig.add_subplot(111, projection='3d')
X = pivot.columns.values
Y = pivot.index.values
X, Y = np.meshgrid(X, Y)
Z = pivot.values
surf = ax.plot_surface(X, Y, Z, cmap=cm.coolwarm,
linewidth=0, antialiased=True, alpha=0.8)
ax.set_xlabel('Tenor (Days)')
ax.set_ylabel('Moneyness (%)')
ax.set_zlabel('Implied Volatility')
ax.set_title('BTC/ETH Volatility Surface')
fig.colorbar(surf, shrink=0.5, aspect=10)
plt.savefig('volatility_surface_3d.png', dpi=150)
plt.show()
Vẽ volatility smile cho các tenor phổ biến
for tenor in [7, 14, 30, 60]:
plot_volatility_smile(surface_data, tenor, btc_spot, 'BTC Options')
Vẽ surface 3D
plot_3d_surface(surface_data, btc_spot)Backtesting Chiến lược Volatility
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Tuple
import json
class VolatilityBacktester:
def __init__(self, initial_capital: float = 100000):
self.initial_capital = initial_capital
self.capital = initial_capital
self.trades: List[Dict] = []
self.equity_curve: List[Dict] = []
self.positions: Dict = {}
def load_historical_data(self, start_date: datetime, end_date: datetime) -> pd.DataFrame:
"""
Load dữ liệu lịch sử từ Deribit
"""
# Với API thực tế, bạn cần sử dụng endpoint get_volatility_history
# và get_trade_volumes để lấy dữ liệu lịch sử
# Đây là ví dụ mock data cho demonstration
dates = pd.date_range(start=start_date, end=end_date, freq='D')
n_days = len(dates)
# Giả định BTC di chuyển theo Geometric Brownian Motion
np.random.seed(42)
returns = np.random.normal(0.0005, 0.03, n_days)
prices = 67500 * np.exp(np.cumsum(returns))
# Tạo dữ liệu IV giả định
base_iv = 0.75
iv_history = []
for i, price in enumerate(prices):
# IV có xu hướng tăng khi giá giảm (leverage effect)
vol_of_vol = 0.15
iv_change = np.random.normal(0, vol_of_vol)
current_iv = base_iv + iv_change - 0.3 * returns[i]
current_iv = max(0.3, min(2.0, current_iv)) # Bound IV
iv_history.append(current_iv)
df = pd.DataFrame({
'date': dates,
'spot': prices,
'iv_atm': iv_history,
'rv_30d': np.abs(returns).rolling(30).mean() * np.sqrt(365),
'rv_60d': np.abs(returns).rolling(60).mean() * np.sqrt(365)
})
# Tính IV/RV ratio
df['iv_rv_ratio'] = df['iv_atm'] / df['rv_30d']
return df
def strategy_volatility_mean_reversion(
self,
df: pd.DataFrame,
iv_rv_upper: float = 1.3,
iv_rv_lower: float = 0.7,
position_size: float = 0.1
) -> Dict:
"""
Chiến lược mean reversion: bán IV khi IV/RV cao, mua IV khi IV/RV thấp
Args:
df: DataFrame chứa dữ liệu
iv_rv_upper: Ngưỡng trên để bán volatility
iv_rv_lower: Ngưỡng dưới để mua volatility
position_size: Tỷ lệ vốn cho mỗi giao dịch
"""
capital = self.initial_capital
positions = []
for i in range(len(df)):
row = df.iloc[i]
date = row['date']
iv_rv = row['iv_rv_ratio']
if pd.isna(iv_rv):
continue
position_value = capital * position_size
# Tín hiệu mua/bán
if iv_rv > iv_rv_upper:
# Tín hiệu bán: IV cao hơn RV
signal = 'sell_vol'
pnl = position_value * (0.1 - (row['iv_atm'] - 0.7)) # Giả định PnL
elif iv_rv < iv_rv_lower:
# Tín hiệu mua: IV thấp hơn RV
signal = 'buy_vol'
pnl = position_value * (0.1 - (0.7 - row['iv_atm'])) # Giả định PnL
else:
signal = 'hold'
pnl = 0
capital += pnl
positions.append({
'date': date,
'signal': signal,
'iv_rv_ratio': iv_rv,
'position_value': position_value,
'pnl': pnl,
'cumulative_pnl': capital - self.initial_capital,
'capital': capital
})
results_df = pd.DataFrame(positions)
# Tính các metrics
total_return = (capital - self.initial_capital) / self.initial_capital
n_trades = len(results_df[results_df['signal'] != 'hold'])
win_rate = len(results_df[results_df['pnl'] > 0]) / len(results_df) if len(results_df) > 0 else 0
sharpe_ratio = results_df['pnl'].mean() / results_df['pnl'].std() * np.sqrt(252) if results_df['pnl'].std() > 0 else 0
max_drawdown = self._calculate_max_drawdown(results_df['capital'])
return {
'total_return': total_return,
'n_trades': n_trades,
'win_rate': win_rate,
'sharpe_ratio': sharpe_ratio,
'max_drawdown': max_drawdown,
'final_capital': capital,
'equity_curve': results_df
}
def _calculate_max_drawdown(self, equity: pd.Series) -> float:
"""Tính maximum drawdown"""
cummax = equity.cummax()
drawdown = (equity - cummax) / cummax
return abs(drawdown.min())
def run_full_backtest(
self,
start_date: datetime,
end_date: datetime,
strategies: List[str] = ['volatility_mean_reversion']
) -> Dict:
"""
Chạy full backtest với nhiều chiến lược
"""
print(f"Loading historical data from {start_date} to {end_date}")
data = self.load_historical_data(start_date, end_date)
results = {}
for strategy in strategies:
print(f"\nRunning strategy: {strategy}")
self.capital = self.initial_capital
if strategy == 'volatility_mean_reversion':
result = self.strategy_volatility_mean_reversion(data)
else:
result = {'error': f'Unknown strategy: {strategy}'}
results[strategy] = result
# In kết quả
print(f" Total Return: {result['total_return']:.2%}")
print(f" Sharpe Ratio: {result['sharpe_ratio']:.2f}")
print(f" Max Drawdown: {result['max_drawdown']:.2%}")
print(f" Win Rate: {result['win_rate']:.2%}")
return results
Chạy backtest
backtester = VolatilityBacktester(initial_capital=100000)
results = backtester.run_full_backtest(
start_date=datetime(2024, 1, 1),
end_date=datetime(2024, 12, 31),
strategies=['volatility_mean_reversion']
)
Vẽ equity curve
import matplotlib.pyplot as plt
equity_df = results['volatility_mean_reversion']['equity_curve']
plt.figure(figsize=(12, 6))
plt.plot(equity_df['date'], equity_df['capital'], label='Strategy')
plt.axhline(y=100000, color='gray', linestyle='--', label='Initial Capital')
plt.fill_between(equity_df['date'], 100000, equity_df['capital'],
where=equity_df['capital'] >= 100000, alpha=0.3, color='green')
plt.fill_between(equity_df['date'], 100000, equity_df['capital'],
where=equity_df['capital'] < 100000, alpha=0.3, color='red')
plt.title('Backtest Equity Curve - Volatility Mean Reversion')
plt.xlabel('Date')
plt.ylabel('Capital ($)')
plt.legend()
plt.grid(True, alpha=0.3)
plt.savefig('backtest_equity_curve.png', dpi=150)
plt.show()Tích hợp HolySheep AI cho Phân tích Nâng cao
Trong các hệ thống trading hiện đại, AI/ML đóng vai trò quan trọng trong việc phân tích dữ liệu và đưa ra quyết định. Đăng ký HolySheep AI cung cấp API tương thích OpenAI với chi phí thấp nhất thị trường, hỗ trợ thanh toán qua WeChat/Alipay và độ trễ dưới 50ms.
import httpx
import json
import asyncio
from typing import List, Dict
class OptionsAnalysisAI:
"""
Sử dụng HolySheep AI để phân tích dữ liệu quyền chọn
"""
def __init__(self, api_key: str):
self.api_key = api_key
# Sử dụng base_url của HolySheep thay vì OpenAI
self.base_url = "https://api.holysheep.ai/v1"
def analyze_volatility_regime(self, surface_data: pd.DataFrame) -> Dict:
"""
Phân tích volatility regime từ dữ liệu surface
"""
# Tính các metrics từ surface
atm_options = surface_data[
(surface_data['moneyness'] >= 0.95) &
(surface_data['moneyness'] <= 1.05)
]
metrics = {
'avg_iv_atm': atm_options['implied_volatility'].mean() if len(atm_options) > 0 else None,
'iv_skew': self._calculate_skew(surface_data),
'term_structure': self._analyze_term_structure(surface_data),
'put_call_ratio': self._calculate_put_call_ratio(surface_data),
'rr_skew': self._calculate_rr_skew(surface_data),
'bf_skew': self._calculate_bf_skew(surface_data)
}
# Sử dụng AI để phân tích
prompt = f"""
Phân tích dữ liệu volatility surface cho trading:
Metrics:
- ATM IV trung bình: {metrics['avg_iv_atm']:.2%}
- IV Skew: {metrics['iv_skew']}
- Term Structure: {metrics['term_structure']}
- Put/Call Ratio: {metrics['put_call_ratio']}
- Risk Reversal Skew: {metrics['rr_skew']}
- Butterfly Skew: {metrics['bf_skew']}
Đưa ra:
1. Đánh giá volatility regime hiện tại (low/normal/high)
2. Khuyến nghị chiến lược trading phù hợp
3. Các cảnh báo risk nếu có
"""
response = self._call_ai_model(prompt)
metrics['ai_analysis'] = response
return metrics
def _call_ai_model(self, prompt: str, model: str = "deepseek-chat") -> str