ในฐานะวิศวกร quantitative trading มากว่า 8 ปี ผมเคยเจอปัญหาเดียวกันกับทุกคน คือการได้มาซึ่งข้อมูล option chain คุณภาพสูงสำหรับ backtesting ที่ reliable และ cost-effective ในบทความนี้ ผมจะแชร์ architecture ที่พิสูจน์แล้วใน production สำหรับดึงข้อมูล Deribit options historical data ผ่าน Tardis API และใช้คำนวณ implied volatility สำหรับ backtest ที่แม่นยำ
ทำไมต้อง Deribit + Tardis API
Deribit เป็น exchange ที่ได้รับความนิยมสูงสุดสำหรับ BTC/ETH options โดยมี open interest รวมกว่า $10B แต่ปัญหาคือ API ของ Deribit เองไม่ได้ออกแบบมาสำหรับ historical data retrieval โดยเฉพาะ ทำให้การ backtest ย้อนหลังทำได้ยากและไม่ consistent
Tardis API แก้ปัญหานี้โดย providing normalized historical market data จาก exchanges หลายสิบแห่ง รวมถึง Deribit โดยเฉพาะ พร้อม features ที่ quant researcher ต้องการ เช่น:
- Historical orderbook snapshots ที่ความละเอียด milliseconds
- Trades data พร้อม participant type (maker/taker)
- Option Greeks และ funding rates
- WebSocket streaming สำหรับ real-time data
การตั้งค่า Environment และ Dependencies
pip install tardis-client pandas numpy scipy aiohttp asyncio \
python-dotenv asyncpg motor prometheus-client \
pyarrow fastparquet cachetools backoff \
python-rapidjson orjson msgpack \
httpx tenacity
สร้าง .env file
cat > .env << 'EOF'
TARDIS_API_KEY=your_tardis_api_key_here
TARDIS_API_SECRET=your_tardis_secret_here
HOLYSHEEP_API_KEY=sk-your-holysheep-key
DATABASE_URL=postgresql://user:pass@localhost:5432/options_db
REDIS_URL=redis://localhost:6379
EOF
ใช้ Poetry สำหรับ production project
poetry init --name deribit-options-backtest
poetry add tardis-client pandas numpy scipy asyncpg aiohttp \
pyarrow fastparquet httpx tenacity backoff orjsonCore Architecture: Async Data Fetcher
import asyncio
import aiohttp
import orjson
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import List, Dict, Optional, AsyncIterator
from pathlib import Path
import pyarrow as pa
import pyarrow.parquet as pq
from tenacity import retry, stop_after_attempt, wait_exponential
import backoff
from collections import defaultdict
import numpy as np
@dataclass
class OptionContract:
"""Deribit option contract data structure"""
instrument_name: str
timestamp: datetime
mark_price: float
underlying_price: float
strike: float
expiry: datetime
option_type: str # 'call' or 'put'
iv_bid: float
iv_ask: float
iv_mark: float
delta: float
gamma: float
theta: float
vega: float
open_interest: float
volume: float
@dataclass
class DeribitOptionsFetcher:
"""
Production-grade Deribit options data fetcher using Tardis API
Supports batch downloads, caching, and incremental updates
"""
api_key: str
api_secret: str
base_url: str = "https://api.tardis.dev/v1"
max_concurrent_requests: int = 10
rate_limit_per_second: int = 5
cache_dir: Path = field(default_factory=lambda: Path("./data_cache"))
def __post_init__(self):
self.cache_dir.mkdir(parents=True, exist_ok=True)
self._semaphore = asyncio.Semaphore(self.max_concurrent_requests)
self._rate_limiter = asyncio.Semaphore(self.rate_limit_per_second)
self._session: Optional[aiohttp.ClientSession] = None
self._request_times: List[float] = []
async def __aenter__(self):
connector = aiohttp.TCPConnector(
limit=100,
limit_per_host=20,
ttl_dns_cache=300,
enable_cleanup_closed=True
)
self._session = aiohttp.ClientSession(
connector=connector,
json_serialize=lambda x: orjson.dumps(x).decode(),
timeout=aiohttp.ClientTimeout(total=60, connect=10)
)
return self
async def __aexit__(self, *args):
if self._session:
await self._session.close()
@backoff.on_exception(
backoff.expo,
(aiohttp.ClientError, asyncio.TimeoutError),
max_tries=5,
max_time=120
)
async def _rate_limited_request(self, url: str, params: Dict) -> Dict:
"""Rate-limited request với exponential backoff"""
async with self._rate_limiter:
async with self._semaphore:
now = asyncio.get_event_loop().time()
self._request_times = [
t for t in self._request_times
if now - t < 1.0
]
if len(self._request_times) >= self.rate_limit_per_second:
sleep_time = 1.0 - (now - self._request_times[0])
if sleep_time > 0:
await asyncio.sleep(sleep_time)
headers = {
"Authorization": f"Bearer {self.api_key}",
"X-API-Key": self.api_secret,
"Content-Type": "application/json"
}
async with self._session.get(url, params=params, headers=headers) as resp:
if resp.status == 429:
retry_after = int(resp.headers.get("Retry-After", 60))
await asyncio.sleep(retry_after)
raise aiohttp.ClientError("Rate limited")
resp.raise_for_status()
return await resp.json()
async def fetch_options_chain(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
expiry_filter: Optional[List[str]] = None
) -> AsyncIterator[OptionContract]:
"""
Fetch options chain data for specified date range
Supports BTC and ETH options on Deribit
"""
date_ranges = self._split_date_range(start_date, end_date, days_per_chunk=7)
for chunk_start, chunk_end in date_ranges:
cache_file = self.cache_dir / f"{symbol}_{chunk_start.date()}_{chunk_end.date()}.parquet"
if cache_file.exists():
yield from self._read_cache(cache_file)
continue
params = {
"exchange": "deribit",
"symbol": symbol,
"from": chunk_start.isoformat(),
"to": chunk_end.isoformat(),
"format": "objects",
"has_content": True,
"limit": 10000,
"meta": json.dumps({
"types": ["trade", "quote"],
"channels": ["book", "trades"]
})
}
try:
data = await self._rate_limited_request(
f"{self.base_url}/historical/derivatives",
params
)
contracts = []
for record in data.get("data", []):
if record.get("instrument_type") != "option":
continue
contract = self._parse_option_record(record, symbol)
if contract and (not expiry_filter or contract.instrument_name in expiry_filter):
contracts.append(contract)
if contracts:
await self._write_cache(cache_file, contracts)
yield from contracts
except Exception as e:
print(f"Error fetching {symbol} {chunk_start.date()}: {e}")
continue
def _parse_option_record(self, record: Dict, symbol: str) -> Optional[OptionContract]:
"""Parse raw Tardis record to OptionContract"""
try:
timestamp = datetime.fromisoformat(
record.get("timestamp", record.get("local_timestamp", "").replace("Z", "+00:00"))
)
if record.get("type") == "trade":
price = record.get("price", 0)
side = record.get("side", "buy")
else:
best_bid = record.get("best_bid_price", 0)
best_ask = record.get("best_ask_price", 0)
price = (best_bid + best_ask) / 2 if best_bid and best_ask else 0
underlying = self._extract_underlying(record.get("instrument_name", ""))
expiry = self._extract_expiry(record.get("instrument_name", ""))
strike = self._extract_strike(record.get("instrument_name", ""))
option_type = "call" if "C" in record.get("instrument_name", "") else "put"
return OptionContract(
instrument_name=record.get("instrument_name", ""),
timestamp=timestamp,
mark_price=price,
underlying_price=underlying,
strike=strike,
expiry=expiry,
option_type=option_type,
iv_bid=record.get("iv_bid", 0),
iv_ask=record.get("iv_ask", 0),
iv_mark=record.get("iv_mark", price),
delta=record.get("delta", 0),
gamma=record.get("gamma", 0),
theta=record.get("theta", 0),
vega=record.get("vega", 0),
open_interest=record.get("open_interest", 0),
volume=record.get("volume", 0)
)
except Exception:
return None
@staticmethod
def _split_date_range(start: datetime, end: datetime, days_per_chunk: int) -> List[tuple]:
"""Split date range into chunks"""
chunks = []
current = start
while current < end:
chunk_end = min(current + timedelta(days=days_per_chunk), end)
chunks.append((current, chunk_end))
current = chunk_end
return chunks
async def _write_cache(self, path: Path, contracts: List[OptionContract]):
"""Write contracts to Parquet cache"""
table = pa.Table.from_pylist([
{
"instrument_name": c.instrument_name,
"timestamp": c.timestamp,
"mark_price": c.mark_price,
"underlying_price": c.underlying_price,
"strike": c.strike,
"expiry": c.expiry,
"option_type": c.option_type,
"iv_bid": c.iv_bid,
"iv_ask": c.iv_ask,
"iv_mark": c.iv_mark,
"delta": c.delta,
"gamma": c.gamma,
"theta": c.theta,
"vega": c.vega,
"open_interest": c.open_interest,
"volume": c.volume
} for c in contracts
])
pq.write_table(table, path, compression="snappy")
def _read_cache(self, path: Path) -> List[OptionContract]:
"""Read contracts from Parquet cache"""
table = pq.read_table(path)
return [
OptionContract(
instrument_name=row["instrument_name"],
timestamp=row["timestamp"].to_pydatetime(),
mark_price=row["mark_price"],
underlying_price=row["underlying_price"],
strike=row["strike"],
expiry=row["expiry"],
option_type=row["option_type"],
iv_bid=row["iv_bid"],
iv_ask=row["iv_ask"],
iv_mark=row["iv_mark"],
delta=row["delta"],
gamma=row["gamma"],
theta=row["theta"],
vega=row["vega"],
open_interest=row["open_interest"],
volume=row["volume"]
)
for row in table.to_pylist()
]Implied Volatility Calculator
import numpy as np
from scipy.stats import norm
from scipy.optimize import brentq, minimize_scalar
from typing import Tuple, Optional
from dataclasses import dataclass
import pandas as pd
@dataclass
class ImpliedVolResult:
"""Result container for IV calculation"""
iv: float
method: str
success: bool
error_message: Optional[str] = None
iterations: int = 0
class ImpliedVolatilityCalculator:
"""
Production IV calculator với multiple methods:
- Newton-Raphson (fast, requires Greeks)
- Bisection (robust, slow)
- Brent (recommended for production)
"""
def __init__(self, tolerance: float = 1e-8, max_iterations: int = 100):
self.tolerance = tolerance
self.max_iterations = max_iterations
def black_scholes_price(
self, S: float, K: float, T: float, r: float,
sigma: float, option_type: str = "call"
) -> float:
"""
Black-Scholes option pricing model
S: Spot price
K: Strike price
T: Time to expiry (years)
r: Risk-free rate
sigma: Volatility
"""
if T <= 0 or sigma <= 0:
return max(0, S - K if option_type == "call" else K - S)
d1 = (np.log(S / K) + (r + sigma**2 / 2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
if option_type.lower() == "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)
return max(0, price)
def vega(self, S: float, K: float, T: float, r: float, sigma: float) -> float:
"""First derivative of option price w.r.t. volatility"""
if T <= 0:
return 0
d1 = (np.log(S / K) + (r + sigma**2 / 2) * T) / (sigma * np.sqrt(T))
return S * norm.pdf(d1) * np.sqrt(T) / 100 # Per vol point
def implied_vol_newton_raphson(
self, market_price: float, S: float, K: float,
T: float, r: float, option_type: str = "call",
initial_sigma: float = 0.5
) -> ImpliedVolResult:
"""
Newton-Raphson method - fastest when Greeks available
Converges quadratically near root
"""
sigma = initial_sigma
for i in range(self.max_iterations):
price = self.black_scholes_price(S, K, T, r, sigma, option_type)
v = self.vega(S, K, T, r, sigma)
diff = market_price - price
if abs(diff) < self.tolerance:
return ImpliedVolResult(
iv=sigma, method="newton_raphson",
success=True, iterations=i+1
)
if abs(v) < 1e-10:
break
sigma += diff / v
if sigma <= 0 or sigma > 5:
return ImpliedVolResult(
iv=np.nan, method="newton_raphson",
success=False,
error_message=f"Sigma out of bounds: {sigma}",
iterations=i+1
)
return ImpliedVolResult(
iv=sigma, method="newton_raphson",
success=False, error_message="Max iterations reached",
iterations=self.max_iterations
)
def implied_vol_bisection(
self, market_price: float, S: float, K: float,
T: float, r: float, option_type: str = "call"
) -> ImpliedVolResult:
"""
Bisection method - most robust, guaranteed convergence
Slower than Newton but reliable
"""
sigma_low, sigma_high = 0.001, 5.0
price_low = self.black_scholes_price(S, K, T, r, sigma_low, option_type)
price_high = self.black_scholes_price(S, K, T, r, sigma_high, option_type)
if (price_low - market_price) * (price_high - market_price) > 0:
return ImpliedVolResult(
iv=np.nan, method="bisection",
success=False,
error_message="No solution in volatility range"
)
for i in range(self.max_iterations):
sigma_mid = (sigma_low + sigma_high) / 2
price_mid = self.black_scholes_price(S, K, T, r, sigma_mid, option_type)
if abs(price_mid - market_price) < self.tolerance:
return ImpliedVolResult(
iv=sigma_mid, method="bisection",
success=True, iterations=i+1
)
if (price_low - market_price) * (price_mid - market_price) < 0:
sigma_high = sigma_mid
else:
sigma_low = sigma_mid
return ImpliedVolResult(
iv=sigma_mid, method="bisection",
success=True, iterations=self.max_iterations
)
def implied_vol_brent(
self, market_price: float, S: float, K: float,
T: float, r: float, option_type: str = "call"
) -> ImpliedVolResult:
"""
Brent's method - best of both worlds
Combines bisection reliability with secant speed
"""
def objective(sigma):
price = self.black_scholes_price(S, K, T, r, sigma, option_type)
return price - market_price
try:
sigma = brentq(
objective, 0.001, 5.0,
xtol=self.tolerance,
maxiter=self.max_iterations
)
return ImpliedVolResult(
iv=sigma, method="brent",
success=True
)
except ValueError as e:
return ImpliedVolResult(
iv=np.nan, method="brent",
success=False,
error_message=str(e)
)
def calculate_batch_iv(
self,
options_df: pd.DataFrame,
risk_free_rate: float = 0.05
) -> pd.DataFrame:
"""
Batch calculate IV for entire options chain
Optimized for production use with vectorization hints
"""
df = options_df.copy()
df["implied_vol"] = np.nan
df["iv_method"] = ""
df["iv_success"] = False
for idx, row in df.iterrows():
if row["mark_price"] <= 0 or row["underlying_price"] <= 0:
continue
T = (row["expiry"] - row["timestamp"]).total_seconds() / (365.25 * 86400)
if T <= 0:
continue
market_price = row["mark_price"]
S = row["underlying_price"]
K = row["strike"]
option_type = row["option_type"]
result = self.implied_vol_brent(
market_price, S, K, T, risk_free_rate, option_type
)
df.at[idx, "implied_vol"] = result.iv if result.success else np.nan
df.at[idx, "iv_method"] = result.method
df.at[idx, "iv_success"] = result.success
return df[df["iv_success"]]
def calculate_volatility_surface(options_df: pd.DataFrame) -> pd.DataFrame:
"""
Calculate full volatility surface (IV vs Strike vs Expiry)
Essential for model validation and trading signal generation
"""
df = options_df.copy()
df["moneyness"] = df["underlying_price"] / df["strike"]
df["time_to_expiry"] = (df["expiry"] - df["timestamp"]).dt.total_seconds() / (365.25 * 86400)
surface = df.groupby(["expiry", "strike"]).agg({
"implied_vol": ["mean", "std", "count"],
"mark_price": "mean",
"open_interest": "sum"
}).reset_index()
surface.columns = ["expiry", "strike", "iv_mean", "iv_std", "n_contracts", "avg_price", "total_oi"]
return surface[surface["n_contracts"] >= 5]Backtesting Engine with HolySheep AI
import httpx
import orjson
import asyncio
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import pandas as pd
import numpy as np
@dataclass
class BacktestResult:
"""Backtest results container"""
strategy_name: str
start_date: datetime
end_date: datetime
total_trades: int
winning_trades: int
losing_trades: int
win_rate: float
total_pnl: float
max_drawdown: float
sharpe_ratio: float
sortino_ratio: float
avg_trade_pnl: float
max_consecutive_losses: int
profit_factor: float
class OptionsBacktester:
"""
Production options backtesting engine
Integrates with HolySheep AI for signal generation
"""
def __init__(
self,
holysheep_api_key: str,
initial_capital: float = 100_000,
max_position_size: float = 0.1,
commission_rate: float = 0.0004
):
self.initial_capital = initial_capital
self.max_position_size = max_position_size
self.commission_rate = commission_rate
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {holysheep_api_key}",
"Content-Type": "application/json"
}
async def generate_trading_signal(
self,
iv_data: Dict[str, Any],
market_context: Dict[str, Any]
) -> Optional[Dict]:
"""
Use HolySheep AI to analyze IV data and generate trading signals
Supports GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2
"""
prompt = f"""
Analyze the following BTC options implied volatility data and generate a trading signal.
Current Market Context:
- BTC Price: ${market_context.get('btc_price', 0):,.2f}
- 30-day IV: {market_context.get('iv_30d', 0)*100:.2f}%
- 90-day IV: {market_context.get('iv_90d', 0)*100:.2f}%
- IV Rank: {market_context.get('iv_rank', 0)*100:.2f}%
- Risk-free rate: {market_context.get('risk_free_rate', 0.05)*100:.2f}%
Options Chain Summary:
- Near-term ATM IV: {iv_data.get('near_atm_iv', 0)*100:.2f}%
- Near-term OTM Call IV: {iv_data.get('near_otm_call_iv', 0)*100:.2f}%
- Near-term OTM Put IV: {iv_data.get('near_otm_put_iv', 0)*100:.2f}%
- Put-Call Ratio: {iv_data.get('put_call_ratio', 0):.2f}
Volatility Skew:
- 25-delta put IV vs ATM: {(iv_data.get('iv_25d_put', 0) - iv_data.get('near_atm_iv', 0))*100:.2f} vol points
- 25-delta call IV vs ATM: {(iv_data.get('iv_25d_call', 0) - iv_data.get('near_atm_iv', 0))*100:.2f} vol points
Return a JSON response with:
{{
"signal": "bullish" | "bearish" | "neutral",
"confidence": 0.0-1.0,
"recommended_strategy": "straddle" | "strangle" | "iron_condor" | "vertical_spread" | "butterfly",
"strike_selection": {{"type": "ATM" | "OTM" | "ITM", "delta_target": 0.0-1.0}},
"rationale": "brief explanation",
"risk_level": "low" | "medium" | "high",
"expected_move_pct": -50 to 50
}}
"""
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "You are an expert options trader. Return ONLY valid JSON."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"response_format": {"type": "json_object"}
}
)
if response.status_code == 200:
result = response.json()
return orjson.loads(result["choices"][0]["message"]["content"])
else:
print(f"API Error: {response.status_code} - {response.text}")
return None
async def run_backtest(
self,
options_fetcher: 'DeribitOptionsFetcher',
start_date: datetime,
end_date: datetime,
symbols: List[str] = ["BTC", "ETH"]
) -> BacktestResult:
"""
Run complete backtest over historical data
"""
capital = self.initial_capital
capital_history = [capital]
trades = []
consecutive_losses = 0
max_consecutive_losses = 0
total_wins = 0
total_losses = 0
all_iv_data = []
async for option in options_fetcher.fetch_options_chain(
symbol="BTC", start_date=start_date, end_date=end_date
):
all_iv_data.append({
"timestamp": option.timestamp,
"strike": option.strike,
"iv": option.iv_mark,
"price": option.mark_price,
"option_type": option.option_type,
"underlying": option.underlying_price
})
if not all_iv_data:
return BacktestResult(
strategy_name="IV Mean Reversion",
start_date=start_date,
end_date=end_date,
total_trades=0,
winning_trades=0,
losing_trades=0,
win_rate=0.0,
total_pnl=0.0,
max_drawdown=0.0,
sharpe_ratio=0.0,
sortino_ratio=0.0,
avg_trade_pnl=0.0,
max_consecutive_losses=0,
profit_factor=0.0
)
df = pd.DataFrame(all_iv_data)
df["expiry_group"] = df.groupby("timestamp")["strike"].transform("count")
iv_stats = df.groupby("timestamp").agg({
"iv": ["mean", "std", "min", "max"]
}).reset_index()
iv_stats.columns = ["timestamp", "iv_mean", "iv_std", "iv_min", "iv_max"]
iv_stats["iv_range"] = iv_stats["iv_max"] - iv_stats["iv_min"]
for _, row in iv_stats.iterrows():
if iv_stats["iv"].std() < 0.05:
continue
iv_data = {
"near_atm_iv": row["iv_mean"],
"near_otm_call_iv": row["iv_max"],
"near_otm_put_iv": row["iv_min"],
"put_call_ratio": 1.0
}
market_context = {
"btc_price": row.get("underlying", 50000),
"iv_30d": row["iv_mean"],
"iv_90d": row["iv_mean"] * 1.1,
"iv_rank": 0.5,
"risk_free_rate": 0.05
}
signal = await self.generate_trading_signal(iv_data, market_context)
if signal and signal.get("confidence", 0) > 0.7:
trade_pnl = self._simulate_trade(signal, row, capital)
capital += trade_pnl
capital_history.append(capital)
trades.append(trade_pnl)
if trade_pnl > 0:
total_wins += 1
consecutive_losses = 0
else:
total_losses += 1
consecutive_losses += 1
max_consecutive_losses = max(max_consecutive_losses, consecutive_losses)
returns = np.diff(capital_history) / capital_history[:-1]
sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252) if np.std(returns) > 0 else 0
downside_returns = returns[returns < 0]
sortino = np.mean(returns) / np.std(downside_returns) * np.sqrt(252) if len(downside_returns) > 0 and np.std(downside_returns) > 0 else 0
cumulative = np.maximum.accumulate(capital_history)
drawdowns = (cumulative - capital_history) / cumulative
max_dd = np.max(drawdowns) if len(drawdowns) > 0 else 0
winning_pnl = sum(t for t in trades if t > 0)
losing_pnl = abs(sum(t for t in trades if t < 0))
profit_factor = winning_pnl / losing_pnl if losing_pnl > 0 else float('inf')
return BacktestResult(
strategy_name="IV Mean Reversion with AI Signals",
start_date=start_date,
end_date=end_date,
total_trades=len(trades),
winning_trades=total_wins,