Chào các bạn, mình là Minh — một quantitative researcher đã làm việc với dữ liệu quyền chọn crypto hơn 4 năm. Hôm nay mình sẽ chia sẻ chi tiết cách lấy lịch sử quyền chọn Deribit bằng Python thông qua Tardis API, cùng với framework backtesting implied volatility (IV) hoàn chỉnh.
Bài viết này dành cho developer mới bắt đầu, không yêu cầu kinh nghiệm API trước đó. Tất cả code đều có thể copy-paste và chạy ngay.
Mục lục
- Giới thiệu tổng quan
- Cài đặt môi trường
- Kết nối Tardis API
- Lấy dữ liệu quyền chọn Deribit
- Tính Implied Volatility
- Xây dựng hệ thống Backtesting
- Lỗi thường gặp và cách khắc phục
- So sánh các phương án
- Kết luận
1. Giới thiệu tổng quan
Deribit là gì?
Deribit là sàn giao dịch quyền chọn Bitcoin và Ethereum lớn nhất thế giới tính theo open interest. Nếu bạn muốn nghiên cứu hành vi thị trường quyền chọn crypto, Deribit là nguồn dữ liệu không thể bỏ qua.
Tardis API là gì?
Tardis API cung cấp dữ liệu lịch sử chất lượng cao từ các sàn crypto, bao gồm Deribit. Thay vì tự crawl dữ liệu (rất phức tạp và dễ bị chặn), bạn chỉ cần gọi API và nhận dữ liệu đã được chuẩn hóa.
Kinh nghiệm thực chiến: Mình đã thử tự crawl Deribit data 3 lần trước khi chuyển sang Tardis. Mất 2 tuần cho việc crawl, rồi phát hiện dữ liệu bị gap ở nhiều thời điểm. Tardis tiết kiệm cho mình ~40 giờ/tháng chỉ riêng phần thu thập và làm sạch dữ liệu.
Tại sao cần Implied Volatility (IV)?
Implied Volatility là chỉ số thị trường đo lường kỳ vọng biến động giá trong tương lai. Trong backtesting chiến lược quyền chọn, IV là yếu tố cốt lõi để:
- Định giá quyền chọn (Black-Scholes model)
- Phát hiện overvalued/undervalued options
- Xây dựng chiến lược volatility arbitrage
- Đo lường risk premium của thị trường
2. Cài đặt môi trường
Yêu cầu hệ thống
- Python 3.8 trở lên
- pip package manager
- Tài khoản Tardis API (có free tier)
Cài đặt thư viện
# Cài đặt các thư viện cần thiết
pip install tardis-client pandas numpy scipy requests asyncio aiohttpTạo file cấu hình
# config.py
import os
Tardis API Configuration
TARDIS_API_KEY = "your_tardis_api_key_here" # Đăng ký tại https://tardis.dev
TARDIS_BASE_URL = "https://api.tardis.dev/v1"
Deribit Exchange Configuration
EXCHANGE = "deribit"
INSTRUMENT_TYPE = "option" # hoặc "future", "spot"
Data Configuration
SYMBOLS = ["BTC-28APR23-28000-C", "ETH-30JUN23-2000-P"] # Ví dụ symbols
START_DATE = "2023-01-01"
END_DATE = "2023-12-31"
HolySheep AI Configuration (để xử lý dữ liệu với AI)
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"3. Kết nối Tardis API
Kiểm tra kết nối
Trước khi lấy dữ liệu, chúng ta cần verify API key hoạt động tốt:
import requests
import json
def check_tardis_connection(api_key):
"""
Kiểm tra kết nối Tardis API
"""
url = "https://api.tardis.dev/v1/available-datasets"
headers = {
"Authorization": f"Bearer {api_key}"
}
try:
response = requests.get(url, headers=headers)
if response.status_code == 200:
print("✅ Kết nối Tardis API thành công!")
data = response.json()
# Tìm datasets của Deribit
deribit_datasets = [
d for d in data.get("datasets", [])
if "deribit" in d.get("exchange", "").lower()
]
print(f"\n📊 Số lượng datasets Deribit: {len(deribit_datasets)}")
for ds in deribit_datasets[:5]:
print(f" - {ds.get('exchange')}: {ds.get('name')}")
return True
elif response.status_code == 401:
print("❌ API Key không hợp lệ")
return False
else:
print(f"❌ Lỗi: HTTP {response.status_code}")
return False
except Exception as e:
print(f"❌ Lỗi kết nối: {e}")
return False
Sử dụng
if __name__ == "__main__":
API_KEY = "your_tardis_api_key"
check_tardis_connection(API_KEY)Trích xuất symbols Deribit
def get_deribit_symbols(api_key, market="option"):
"""
Lấy danh sách tất cả symbols có sẵn trên Deribit
"""
import requests
url = f"https://api.tardis.dev/v1/derivatives/{market}"
headers = {"Authorization": f"Bearer {api_key}"}
params = {
"exchange": "deribit",
"limit": 1000,
"offset": 0
}
all_symbols = []
while True:
response = requests.get(url, headers=headers, params=params)
if response.status_code != 200:
print(f"Lỗi HTTP: {response.status_code}")
break
data = response.json()
symbols = data.get("results", [])
if not symbols:
break
all_symbols.extend(symbols)
if len(symbols) < 1000:
break
params["offset"] += 1000
return all_symbols
Lấy danh sách quyền chọn BTC
btc_options = [s for s in get_deribit_symbols(TARDIS_API_KEY)
if "BTC" in s.get("symbol", "") and s.get("type") == "option"]
print(f"Tìm thấy {len(btc_options)} quyền chọn BTC")4. Lấy dữ liệu quyền chọn Deribit
Tardis Client - Phương pháp đơn giản nhất
Tardis cung cấp Python client chính thức giúp đơn giản hóa việc lấy dữ liệu:
from tardis_client import TardisClient, exchanges, channels
import asyncio
from datetime import datetime, timedelta
import pandas as pd
async def fetch_options_data():
"""
Lấy dữ liệu quyền chọn Deribit trong khoảng thời gian
"""
# Khởi tạo client
client = TardisClient(api_key="your_tardis_api_key")
# Định nghĩa thời gian (UTC)
from_timestamp = datetime(2023, 6, 1, 0, 0, 0)
to_timestamp = datetime(2023, 6, 30, 23, 59, 59)
# Lấy dữ liệu trade từ Deribit
messages = client.tardis(
exchange=exchanges.DERIBIT,
from_timestamp=from_timestamp,
to_timestamp=to_timestamp,
channels=[channels.DERIBIT_TRADES],
symbols=["BTC-30JUN23"] # Filter theo symbol
)
trades_data = []
# Iterate qua messages
async for message in messages:
if message.type == "trade":
trades_data.append({
"timestamp": message.timestamp,
"symbol": message.symbol,
"side": message.side,
"price": float(message.price),
"amount": float(message.amount),
"iv": getattr(message, 'iv', None), # Implied volatility nếu có
"greeks": getattr(message, 'greeks', None)
})
return pd.DataFrame(trades_data)
Chạy async function
df_trades = asyncio.run(fetch_options_data())
print(f"Đã lấy {len(df_trades)} trades")
print(df_trades.head())Lấy dữ liệu Orderbook để tính IV
async def fetch_orderbook_for_iv():
"""
Lấy orderbook để tính Implied Volatility
Orderbook có bid/ask price → dùng để back-calculate IV
"""
from tardis_client import TardisClient, exchanges, channels
from scipy.optimize import brentq
from scipy.stats import norm
import numpy as np
client = TardisClient(api_key="your_tardis_api_key")
from_timestamp = datetime(2023, 7, 1, 0, 0, 0)
to_timestamp = datetime(2023, 7, 1, 12, 0, 0) # Chỉ 12 giờ đầu
messages = client.tardis(
exchange=exchanges.DERIBIT,
from_timestamp=from_timestamp,
to_timestamp=to_timestamp,
channels=[channels.DERIBIT_BOOKSnapshot_100],
symbols=["BTC-28JUL23-29000-C"]
)
orderbook_data = []
async for message in messages:
if hasattr(message, 'book'):
bids = message.book.get('bids', [])
asks = message.book.get('asks', [])
if bids and asks:
best_bid = float(bids[0][0])
best_ask = float(asks[0][0])
mid_price = (best_bid + best_ask) / 2
orderbook_data.append({
"timestamp": message.timestamp,
"best_bid": best_bid,
"best_ask": best_ask,
"mid_price": mid_price,
"spread": best_ask - best_bid
})
return pd.DataFrame(orderbook_data)
df_book = asyncio.run(fetch_orderbook_for_iv())
print(f"Đã lấy {len(df_book)} snapshots")
print(df_book.head())5. Tính Implied Volatility (IV)
Black-Scholes Model Implementation
import numpy as np
from scipy.stats import norm
from scipy.optimize import brentq
from datetime import datetime, timedelta
class BlackScholes:
"""
Black-Scholes model để tính giá lý thuyết và Implied Volatility
"""
@staticmethod
def d1(S, K, T, r, sigma):
"""Tính d1 trong công thức Black-Scholes"""
if T <= 0 or sigma <= 0:
return np.nan
return (np.log(S / K) + (r + 0.5 * sigma ** 2) * T) / (sigma * np.sqrt(T))
@staticmethod
def d2(S, K, T, r, sigma):
"""Tính d2 trong công thức Black-Scholes"""
if T <= 0 or sigma <= 0:
return np.nan
return BlackScholes.d1(S, K, T, r, sigma) - sigma * np.sqrt(T)
@staticmethod
def call_price(S, K, T, r, sigma):
"""Tính giá Call theo Black-Scholes"""
if T <= 0:
return max(0, S - K)
d1 = BlackScholes.d1(S, K, T, r, sigma)
d2 = BlackScholes.d2(S, K, T, r, sigma)
return S * norm.cdf(d1) - K * np.exp(-r * T) * norm.cdf(d2)
@staticmethod
def put_price(S, K, T, r, sigma):
"""Tính giá Put theo Black-Scholes"""
if T <= 0:
return max(0, K - S)
d1 = BlackScholes.d1(S, K, T, r, sigma)
d2 = BlackScholes.d2(S, K, T, r, sigma)
return K * np.exp(-r * T) * norm.cdf(-d2) - S * norm.cdf(-d1)
@staticmethod
def implied_volatility(market_price, S, K, T, r, option_type='call'):
"""
Tính Implied Volatility từ giá thị trường
Args:
market_price: Giá thị trường của quyền chọn
S: Giá underlying (BTC/ETH)
K: Strike price
T: Thời gian đến expiry (năm)
r: Risk-free rate
option_type: 'call' hoặc 'put'
Returns:
Implied Volatility (dạng thập phân, vd: 0.8 = 80%)
"""
if T <= 0:
return np.nan
def objective(sigma):
if option_type == 'call':
theoretical = BlackScholes.call_price(S, K, T, r, sigma)
else:
theoretical = BlackScholes.put_price(S, K, T, r, sigma)
return (theoretical - market_price) ** 2
# Tìm IV bằng Brent method
try:
iv = brentq(objective, 0.001, 5.0) # IV từ 0.1% đến 500%
return iv
except ValueError:
return np.nan
Ví dụ sử dụng
bs = BlackScholes()
Giả sử có dữ liệu
S = 29000 # Giá BTC
K = 28000 # Strike
T = 0.1 # 10% năm ≈ 36.5 ngày
r = 0.05 # Risk-free rate 5%
Tính giá Call với IV = 80%
call_price = bs.call_price(S, K, T, r, sigma=0.80)
print(f"Giá Call (BS, IV=80%): ${call_price:.2f}")
Back-calculate IV từ giá thị trường
market_price = 3500 # Giá thị trường
iv = bs.implied_volatility(market_price, S, K, T, r, 'call')
print(f"Implied Volatility: {iv*100:.2f}%")Tính IV từ dữ liệu Tardis
def calculate_iv_from_trades(df_trades, df_orderbook, df_underlying):
"""
Tính Implied Volatility từ dữ liệu đã lấy từ Tardis
Args:
df_trades: DataFrame chứa trades
df_orderbook: DataFrame chứa orderbook
df_underlying: DataFrame chứa giá BTC/ETH
"""
bs = BlackScholes()
iv_results = []
# Merge với giá underlying
df_merged = df_orderbook.merge(
df_underlying[['timestamp', 'price']],
on='timestamp',
how='left',
suffixes=('_option', '_underlying')
).ffill()
# Tính IV cho mỗi snapshot
for idx, row in df_merged.iterrows():
S = row['price_underlying'] # Giá BTC/ETH
K = extract_strike_from_symbol(row.get('symbol', '')) # Strike price
T = calculate_time_to_expiry(row.get('expiry', '')) # Thời gian đến expiry
market_price = row['mid_price']
if S and K and T and market_price:
iv = bs.implied_volatility(market_price, S, K, T, r=0.05, option_type='call')
iv_results.append({
'timestamp': row['timestamp'],
'iv': iv,
'bid': row['best_bid'],
'ask': row['best_ask'],
'mid': row['mid_price'],
'underlying_price': S
})
return pd.DataFrame(iv_results)
Ví dụ sử dụng
df_iv = calculate_iv_from_trades(df_trades, df_book, df_btc_price)
print(df_iv.head(10))
6. Xây dựng hệ thống Backtesting
Volatility Mean Reversion Strategy
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
class VolatilityBacktester:
"""
Hệ thống backtest chiến lược Volatility Mean Reversion
"""
def __init__(self, initial_capital=100000, transaction_cost_pct=0.001):
self.initial_capital = initial_capital
self.transaction_cost_pct = transaction_cost_pct
self.portfolio_value = initial_capital
self.trades = []
self.positions = {}
self.equity_curve = []
def run_backtest(self, df_iv, df_underlying, window=20, entry_threshold=1.5, exit_threshold=0.5):
"""
Chạy backtest với chiến lược Mean Reversion
Args:
df_iv: DataFrame chứa Implied Volatility theo thời gian
df_underlying: DataFrame chứa giá underlying
window: Số ngày để tính trung bình di động
entry_threshold: Ngưỡng z-score để vào lệnh (độ lệch so với mean)
exit_threshold: Ngưỡng z-score để đóng lệnh
"""
# Tính rolling mean và std của IV
df_iv['iv_ma'] = df_iv['iv'].rolling(window=window).mean()
df_iv['iv_std'] = df_iv['iv'].rolling(window=window).std()
df_iv['iv_zscore'] = (df_iv['iv'] - df_iv['iv_ma']) / df_iv['iv_std']
# Merge với giá underlying
df = df_iv.merge(df_underlying[['timestamp', 'price']], on='timestamp', how='left')
current_position = 0
entry_price = 0
entry_iv = 0
for idx, row in df.iterrows():
if pd.isna(row['iv_zscore']):
continue
timestamp = row['timestamp']
iv = row['iv']
iv_zscore = row['iv_zscore']
underlying_price = row['price']
# === KIỂM TRA ĐIỀU KIỆN VÀO LỆNH ===
# Mua quyền chọn Call khi IV thấp bất thường (z-score < -entry_threshold)
if current_position == 0 and iv_zscore < -entry_threshold:
# Mở vị thế long Call
position_value = self.portfolio_value * 0.1 # 10% capital
num_contracts = position_value / (underlying_price * 0.1) # Giả định multiplier
cost = position_value * (1 + self.transaction_cost_pct)
if cost <= self.portfolio_value:
current_position = num_contracts
entry_price = underlying_price
entry_iv = iv
self.portfolio_value -= cost
self.trades.append({
'timestamp': timestamp,
'action': 'BUY',
'iv': iv,
'iv_zscore': iv_zscore,
'price': underlying_price,
'contracts': num_contracts,
'cost': cost,
'portfolio_value': self.portfolio_value
})
# === KIỂM TRA ĐIỀU KIỆN RA LỆNH ===
elif current_position > 0:
# Đóng vị thế khi IV quay về mean (z-score > -exit_threshold)
if iv_zscore > -exit_threshold:
revenue = current_position * underlying_price * 0.1 * (1 - self.transaction_cost_pct)
pnl = revenue - self.trades[-1]['cost']
self.portfolio_value += revenue
current_position = 0
self.trades.append({
'timestamp': timestamp,
'action': 'SELL',
'iv': iv,
'iv_zscore': iv_zscore,
'price': underlying_price,
'contracts': 0,
'revenue': revenue,
'pnl': pnl,
'portfolio_value': self.portfolio_value
})
# Ghi nhận equity curve
self.equity_curve.append({
'timestamp': timestamp,
'portfolio_value': self.portfolio_value + (current_position * underlying_price * 0.1 if current_position > 0 else 0)
})
# Đóng vị thế còn lại nếu chưa đóng
if current_position > 0:
last_price = df.iloc[-1]['price']
revenue = current_position * last_price * 0.1 * (1 - self.transaction_cost_pct)
self.portfolio_value += revenue
self.trades.append({
'action': 'FORCE_CLOSE',
'pnl': revenue - self.trades[-1]['cost']
})
return self.calculate_metrics()
def calculate_metrics(self):
"""Tính các metrics hiệu suất"""
df_trades = pd.DataFrame(self.trades)
df_equity = pd.DataFrame(self.equity_curve)
if len(df_trades) == 0:
return {}
# Tính returns
df_equity['returns'] = df_equity['portfolio_value'].pct_change()
# Total Return
total_return = (self.portfolio_value - self.initial_capital) / self.initial_capital * 100
# Sharpe Ratio (annualized)
annualization_factor = np.sqrt(365 * 24) # Giả định hourly data
sharpe_ratio = df_equity['returns'].mean() / df_equity['returns'].std() * annualization_factor
# Maximum Drawdown
df_equity['cummax'] = df_equity['portfolio_value'].cummax()
df_equity['drawdown'] = (df_equity['portfolio_value'] - df_equity['cummax']) / df_equity['cummax']
max_drawdown = df_equity['drawdown'].min() * 100
# Win Rate
closed_trades = df_trades[df_trades['action'] == 'SELL']
if len(closed_trades) > 0:
win_rate = (closed_trades['pnl'] > 0).sum() / len(closed_trades) * 100
else:
win_rate = 0
return {
'total_return': total_return,
'final_capital': self.portfolio_value,
'sharpe_ratio': sharpe_ratio,
'max_drawdown': max_drawdown,
'total_trades': len(df_trades),
'win_rate': win_rate,
'equity_curve': df_equity,
'trades': df_trades
}
Chạy backtest
backtester = VolatilityBacktester(initial_capital=100000)
metrics = backtester.run_backtest(df_iv, df_underlying)
print(f"Total Return: {metrics['total_return']:.2f}%")
print(f"Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
print(f"Max Drawdown: {metrics['max_drawdown']:.2f}%")
Sử dụng HolySheep AI để phân tích kết quả
Bạn có thể dùng HolySheep AI để xử lý và phân tích kết quả backtest nhanh hơn. Với chi phí chỉ $0.42/MTok (rẻ hơn 85%+ so với OpenAI), bạn có thể:
- Yêu cầu AI phân tích equity curve và đưa ra cải tiến
- Tạo báo cáo tự động bằng tiếng Việt
- Tối ưu hóa tham số chiến lược
import requests
import json
def analyze_backtest_results_with_holysheep(api_key, metrics, df_trades_sample):
"""
Sử dụng HolySheep AI để phân tích kết quả backtest
Chi phí ước tính: ~$0.02 cho prompt này (rất rẻ!)
Độ trễ: <50ms với HolySheep infrastructure
"""
base_url = "https://api.holysheep.ai/v1"
prompt = f"""
Phân tích kết quả backtest chiến lược Volatility Mean Reversion:
1. Total Return: {metrics.get('total_return', 0):.2f}%
2. Sharpe Ratio: {metrics.get('sharpe_ratio', 0):.2f}
3. Max Drawdown: {metrics.get('max_drawdown', 0):.2f}%
4. Win Rate: {metrics.get('win_rate', 0):.2f}%
5. Total Trades: {metrics.get('total_trades', 0)}
Mẫu trades:
{df_trades_sample.head(10).to_string()}
Đưa ra:
1. Đánh giá tổng quan chiến lược
2. Các điểm cần cải thiện
3. Đề xuất thay đổi tham số
"""
payload = {
"model": "deepseek-v3.2", # Model rẻ nhất, $0.42/MTok
"messages": [
{"role": "system", "content": "Bạn là chuyên gia phân tích quantitative trading. Trả lời bằng tiếng Việt."},
{"role": "user", "content": prompt}
],
"temperature": 0.7,
"max_tokens": 1000
}
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
response = requests.post(
f"{base_url}/chat/completions",
headers=headers,
json=payload
)
if response.status_code == 200:
result = response.json()
return result['choices'][0]['message']['content']
else:
return f"Lỗi: {response.status_code}"
Sử dụng
analysis = analyze_backtest_results_with_holysheep(
HOLYSHEEP_API_KEY,
metrics,
metrics['trades']
)
print(analysis)
7. Lỗi thường gặp và cách khắc phục
Trong quá trình làm việc với Tardis API và dữ liệu quyền chọn, mình đã gặp nhiều lỗi. Dưới đây là tổng hợp các lỗi phổ biến và cách fix.
Lỗi 1: Lỗi xác thực API (401 Unauthorized)
# ❌ SAI - Key không đúng hoặc đã hết hạn
TARDIS_API_KEY = "invalid_key_123"
✅ ĐÚNG - Kiểm tra và xác thực lại
import os
def validate_tardis_api_key(api_key):
"""Validate API key trước khi sử dụng"""
if not api_key:
raise ValueError("API Key không được để trống")
if len(api_key) < 20:
raise ValueError("API Key có vẻ không hợp lệ (quá ngắn)")
# Kiểm tra format
if not api_key.replace("-", "").replace("_", "").isalnum():
raise ValueError("API Key chứa ký tự không hợp lệ")
# Test kết nối
test_url = "https://api.tardis.dev/v1/available-datasets"
headers = {"Authorization": f"Bearer {api_key}"}
response = requests.get(test_url, headers=headers)
if response.status_code ==