ผมเคยเจอสถานการณ์ที่ทำให้โปรเจกต์ Options Trading Bot หยุดชะงักทั้งหมด นั่นคือ ConnectionError: HTTPSConnectionPool(host='history.deribit.com', port=443): Max retries exceeded พร้อมกับ 403 Forbidden - Historical data requires active subscription ซ้ำแล้วซ้ำเล่า หลังจากลองทำตามเอกสารอย่างเป็นทางการแล้วพบว่า Deribit เองไม่ได้เปิดให้เข้าถึง historical data โดยตรงสำหรับบัญชีฟรี
บทความนี้จะสอนวิธีดึงข้อมูล BTC Options จาก Deribit ผ่าน Tardis API, คำนวณ Implied Volatility ด้วย Newton-Raphson method, และสร้าง backtesting framework เพื่อทดสอบกลยุทธ์ Options อย่างเป็นระบบ พร้อมตัวอย่างโค้ดที่รันได้จริงภายใน 10 นาที
ทำไมต้องดึงข้อมูล BTC Options จาก Deribit
Deribit คือตลาด Futures และ Options ที่ใหญ่ที่สุดสำหรับ BTC โดยมี Volume กว่า $2.5 พันล้านต่อวัน ข้อมูลที่นี่มีความสำคัญต่อ:
- Quant Researchers - สร้างโมเดลคาดการณ์ IV Surface
- Options Traders - วิเคราะห์ Open Interest และ Volume Profile
- Market Makers - คำนวณ Fair Value สำหรับ Delta Hedging
Tardis API: วิธีดาวน์โหลด Deribit Historical Data ในรูปแบบ CSV
Tardis Machine เป็นบริการที่รวบรวม Historical Data จาก Deribit, Binance, OKX และ Exchange อื่นๆ โดยมี API ที่เสถียรกว่าการดึงข้อมูลโดยตรงจาก Deribit มาก
การติดตั้งและตั้งค่าเบื้องต้น
# ติดตั้ง dependencies ที่จำเป็น
pip install tardis-machine pandas numpy scipy requests
สร้างไฟล์ config สำหรับเก็บ API credentials
cat > config.py << 'EOF'
import os
Tardis API Configuration
TARDIS_API_KEY = "your_tardis_api_key_here" # สมัครที่ https://tardis.dev
TARDIS_BASE_URL = "https://api.tardis.ml/v1"
Deribit WebSocket Endpoint
DERIBIT_WS_URL = "wss://history.deribit.com/ws/api/v2"
Local cache settings
CACHE_DIR = "./data_cache"
os.makedirs(CACHE_DIR, exist_ok=True)
EOF
echo "Setup complete!"ดาวน์โหลด BTC Options Trade Data ในรูปแบบ CSV
import requests
import pandas as pd
from datetime import datetime, timedelta
import time
import os
class DeribitOptionsDataFetcher:
"""ดึงข้อมูล BTC Options จาก Tardis API"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.tardis.ml/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def fetch_options_trades(
self,
start_date: str, # format: "2024-01-01"
end_date: str, # format: "2024-01-31"
instrument_prefix: str = "BTC",
save_csv: bool = True
) -> pd.DataFrame:
"""
ดึงข้อมูล Options Trades สำหรับ BTC
Args:
start_date: วันที่เริ่มต้น (YYYY-MM-DD)
end_date: วันที่สิ้นสุด (YYYY-MM-DD)
instrument_prefix: "BTC" หรือ "ETH"
save_csv: บันทึกเป็น CSV หรือไม่
Returns:
DataFrame ที่มี columns: timestamp, instrument_name,
direction, amount, price, option_type, strike, expiry
"""
print(f"Fetching BTC Options data from {start_date} to {end_date}")
# Convert dates to milliseconds timestamp
start_ts = int(datetime.strptime(start_date, "%Y-%m-%d").timestamp() * 1000)
end_ts = int(datetime.strptime(end_date, "%Y-%m-%d").timestamp() * 1000)
# API endpoint สำหรับ historical trades
url = f"{self.base_url}/historical/deribit/trades"
params = {
"exchange": "deribit",
"symbol": f"{instrument_prefix}-*", # ดึงทุก Options
"from": start_ts,
"to": end_ts,
"format": "json",
"limit": 100000 # Max records per request
}
all_trades = []
offset = 0
while True:
params["offset"] = offset
response = requests.get(
url,
headers=self.headers,
params=params,
timeout=60
)
if response.status_code == 401:
raise Exception("❌ 401 Unauthorized: ตรวจสอบ Tardis API Key ของคุณ")
elif response.status_code == 429:
print("⏳ Rate limited, รอ 60 วินาที...")
time.sleep(60)
continue
data = response.json()
if not data.get("trades"):
break
all_trades.extend(data["trades"])
print(f" ✓ ดึงได้ {len(all_trades)} records...")
if len(data["trades"]) < params["limit"]:
break
offset += params["limit"]
time.sleep(0.5) # หลีกเลี่ยง rate limit
# แปลงเป็น DataFrame และ parse instrument_name
df = pd.DataFrame(all_trades)
if df.empty:
print("⚠️ ไม่พบข้อมูลในช่วงวันที่ที่ระบุ")
return df
# Parse option details จาก instrument_name
# Format: BTC-20241227-95000-C (Call) หรือ BTC-20241227-95000-P (Put)
df["expiry"] = df["instrument_name"].str.extract(r"BTC-(\d{8})")
df["strike"] = df["instrument_name"].str.extract(r"-(\d+)-(C|P)$").astype(float)
df["option_type"] = df["instrument_name"].str.extract(r"-(\d+)-(C|P)$").iloc[:, -1]
# Convert timestamp
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
df["date"] = df["timestamp"].dt.date
if save_csv:
filename = f"{CACHE_DIR}/btc_options_{start_date}_{end_date}.csv"
df.to_csv(filename, index=False)
print(f"💾 บันทึก {len(df)} records ไปยัง {filename}")
return df
ตัวอย่างการใช้งาน
if __name__ == "__main__":
fetcher = DeribitOptionsDataFetcher(api_key="your_tardis_api_key")
# ดึงข้อมูล 7 วันล่าสุด
end_date = datetime.now().strftime("%Y-%m-%d")
start_date = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d")
df = fetcher.fetch_options_trades(
start_date=start_date,
end_date=end_date
)
print(f"\n📊 สรุปข้อมูล:")
print(f" Total Trades: {len(df):,}")
print(f" Unique Options: {df['instrument_name'].nunique()}")
print(f" Date Range: {df['date'].min()} to {df['date'].max()}")การคำนวณ Implied Volatility ด้วย Newton-Raphson Method
Implied Volatility (IV) คือค่าความผันผวนที่ "ซ่อนอยู่" ในราคา Options ตาม Black-Scholes Model การคำนวณ IV แม่นยำเป็นหัวใจของการทำ Quantitative Analysis
import numpy as np
from scipy.stats import norm
from scipy.optimize import brentq, newton
from typing import Tuple, Optional
class ImpliedVolatilityCalculator:
"""คำนวณ Implied Volatility สำหรับ BTC Options"""
@staticmethod
def black_scholes_call(
S: float, # Spot price
K: float, # Strike price
T: float, # Time to expiry (years)
r: float, # Risk-free rate
sigma: float # Volatility
) -> float:
"""คำนวณราคา Call Option ตาม Black-Scholes"""
if T <= 0 or sigma <= 0:
return max(S - K, 0)
d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
return S * norm.cdf(d1) - K * np.exp(-r * T) * norm.cdf(d2)
@staticmethod
def black_scholes_put(
S: float, K: float, T: float, r: float, sigma: float
) -> float:
"""คำนวณราคา Put Option ตาม Black-Scholes"""
if T <= 0 or sigma <= 0:
return max(K - S, 0)
d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
return K * np.exp(-r * T) * norm.cdf(-d2) - S * norm.cdf(-d1)
@staticmethod
def vega(
S: float, K: float, T: float, r: float, sigma: float
) -> float:
"""คำนวณ Vega - ความไวต่อการเปลี่ยนแปลงของ IV"""
if T <= 0:
return 0
d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
return S * np.sqrt(T) * norm.pdf(d1)
def calculate_iv_newton_raphson(
self,
option_price: float,
S: float,
K: float,
T: float,
r: float,
is_call: bool = True,
max_iterations: int = 100,
tolerance: float = 1e-6
) -> Optional[float]:
"""
คำนวณ IV ด้วย Newton-Raphson Method
วิธีนี้ converge เร็วกว่า Brent's method แต่อาจไม่เสถียรถ้า initial guess ไม่ดี
Args:
option_price: ราคา Options จริง
S: Spot price ของ BTC
K: Strike price
T: Time to expiry (years)
r: Risk-free rate (ใช้ 0.05 สำหรับ BTC)
is_call: True = Call, False = Put
Returns:
Implied Volatility หรือ None ถ้าคำนวณไม่ได้
"""
# Initial guess: ATM option ใช้ 0.8 (80% annualized)
sigma = 0.8
# กำหนดขอบเขต IV ที่เป็นไปได้
sigma_min, sigma_max = 0.01, 5.0
for i in range(max_iterations):
if is_call:
price_calc = self.black_scholes_call(S, K, T, r, sigma)
else:
price_calc = self.black_scholes_put(S, K, T, r, sigma)
vega = self.vega(S, K, T, r, sigma)
# ถ้า Vega ใกล้ 0 แสดงว่าหาค่าไม่ได้
if abs(vega) < 1e-10:
return None
# คำนวณ error
diff = option_price - price_calc
if abs(diff) < tolerance:
return sigma
# Update sigma ด้วย Newton-Raphson
sigma = sigma + diff / vega
# ตรวจสอบขอบเขต
if sigma < sigma_min or sigma > sigma_max:
return None
return None
def calculate_iv_brent(
self,
option_price: float,
S: float,
K: float,
T: float,
r: float,
is_call: bool = True
) -> Optional[float]:
"""
คำนวณ IV ด้วย Brent's Method (เสถียรกว่า)
วิธีนี้ guarantee convergence แต่ช้ากว่า
"""
def objective(sigma):
if is_call:
return self.black_scholes_call(S, K, T, r, sigma) - option_price
else:
return self.black_scholes_put(S, K, T, r, sigma) - option_price
# หาช่วงที่ function เปลี่ยนเครื่องหมาย
sigma_min, sigma_max = 0.01, 5.0
try:
# ตรวจสอบว่า root อยู่ในช่วงที่กำหนด
f_min = objective(sigma_min)
f_max = objective(sigma_max)
if f_min * f_max > 0:
# Root ไม่อยู่ในช่วง ลองขยายช่วง
for multiplier in [2, 5, 10]:
sigma_max = multiplier
f_max = objective(sigma_max)
if f_min * f_max <= 0:
break
iv = brentq(objective, sigma_min, sigma_max, xtol=1e-6)
return iv
except ValueError:
return None
ตัวอย่างการใช้งาน
if __name__ == "__main__":
calc = ImpliedVolatilityCalculator()
# ตั้งค่า parameters
S = 67500 # BTC Spot price
K = 68000 # Strike price (slightly ITM)
T = 30 / 365 # 30 days to expiry
r = 0.05 # Risk-free rate
# ราคา Options จริงจากตลาด (ประมาณ)
market_call_price = 2800
# คำนวณ IV
iv_newton = calc.calculate_iv_newton_raphson(
option_price=market_call_price,
S=S, K=K, T=T, r=r, is_call=True
)
iv_brent = calc.calculate_iv_brent(
option_price=market_call_price,
S=S, K=K, T=T, r=r, is_call=True
)
print(f"📊 Implied Volatility Calculation Results:")
print(f" Spot Price: ${S:,.0f}")
print(f" Strike Price: ${K:,.0f}")
print(f" Days to Expiry: {T*365:.0f} days")
print(f" Market Option Price: ${market_call_price:,.2f}")
print(f" IV (Newton-Raphson): {iv_newton*100:.2f}%")
print(f" IV (Brent's Method): {iv_brent*100:.2f}%")
# Verify: คำนวณราคากลับจาก IV
theoretical_price = calc.black_scholes_call(S, K, T, r, iv_newton)
print(f" Theoretical Price: ${theoretical_price:,.2f}")
print(f" Price Error: ${abs(market_call_price - theoretical_price):.2f}")สร้าง Options Backtesting Framework ด้วย Python
หลังจากได้ข้อมูล IV มาแล้ว ต่อไปคือการสร้าง Backtesting Framework เพื่อทดสอบกลยุทธ์ Options ต่างๆ
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime, timedelta
from collections import defaultdict
@dataclass
class OptionContract:
"""โครงสร้างข้อมูล Option Contract"""
expiry: str
strike: float
option_type: str # 'C' หรือ 'P'
iv: float
delta: float
premium: float
expiry_date: datetime
@dataclass
class Trade:
"""โครงสร้างข้อมูล Trade"""
entry_date: datetime
exit_date: Optional[datetime]
contract: OptionContract
quantity: int
entry_price: float
exit_price: Optional[float]
pnl: Optional[float]
strategy_name: str
class OptionsBacktester:
"""Framework สำหรับ Backtest กลยุทธ์ Options"""
def __init__(
self,
initial_capital: float = 100000,
risk_free_rate: float = 0.05
):
self.initial_capital = initial_capital
self.current_capital = initial_capital
self.risk_free_rate = risk_free_rate
self.trades: List[Trade] = []
self.equity_curve: List[Dict] = []
# Stats
self.total_trades = 0
self.winning_trades = 0
self.losing_trades = 0
def calculate_greeks(self, S: float, K: float, T: float,
sigma: float, r: float, is_call: bool) -> Dict:
"""คำนวณ Greeks ทั้งหมด"""
from scipy.stats import norm
d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
if is_call:
delta = norm.cdf(d1)
theta = (-S * norm.pdf(d1) * sigma / (2 * np.sqrt(T))
- r * K * np.exp(-r * T) * norm.cdf(d2))
else:
delta = -norm.cdf(-d1)
theta = (-S * norm.pdf(d1) * sigma / (2 * np.sqrt(T))
+ r * K * np.exp(-r * T) * norm.cdf(-d2))
gamma = norm.pdf(d1) / (S * sigma * np.sqrt(T))
vega = S * np.sqrt(T) * norm.pdf(d1) / 100 # Per 1% change
return {
"delta": delta,
"gamma": gamma,
"theta": theta / 365, # Daily theta
"vega": vega # Per 1% IV change
}
def execute_strategy_straddle(
self,
df: pd.DataFrame,
entry_date: datetime,
days_to_expiry: int = 30,
risk_per_trade: float = 0.02 # 2% ของ Capital
) -> Optional[Trade]:
"""
กลยุทธ์ Straddle: ซื้อ Call + Put ที่ Strike เดียวกัน
กำไรเมื่อ: ราคาเคลื่อนไหวมากทั้งขึ้นและลง
ขาดทุนเมื่อ: ราคาเคลื่อนไหวน้อย
"""
# หา ATM strike
current_row = df[df['timestamp'] <= entry_date].iloc[-1]
S = current_row.get('underlying_price', 67000) # Spot price
# หา ATM strike
K = round(S / 1000) * 1000 # Round to nearest 1000
# คำนวณ T
expiry_dt = entry_date + timedelta(days=days_to_expiry)
T = days_to_expiry / 365
# คำนวณ IV จากข้อมูล
call_iv = current_row.get('call_iv', 0.75)
put_iv = current_row.get('put_iv', 0.80)
# คำนวณ premium
calc = ImpliedVolatilityCalculator()
call_premium = calc.black_scholes_call(S, K, T, self.risk_free_rate, call_iv)
put_premium = calc.black_scholes_put(S, K, T, self.risk_free_rate, put_iv)
# Position size ตาม risk
total_premium = call_premium + put_premium
position_value = self.current_capital * risk_per_trade
quantity = max(1, int(position_value / total_premium))
# สร้าง contract
contract = OptionContract(
expiry=expiry_dt.strftime("%Y%m%d"),
strike=K,
option_type="STRADDLE",
iv=(call_iv + put_iv) / 2,
delta=0,
premium=total_premium,
expiry_date=expiry_dt
)
# Record trade
trade = Trade(
entry_date=entry_date,
exit_date=None,
contract=contract,
quantity=quantity,
entry_price=total_premium,
exit_price=None,
pnl=None,
strategy_name="Straddle"
)
self.trades.append(trade)
return trade
def close_trade(
self,
trade: Trade,
exit_date: datetime,
exit_price: float
):
"""ปิด position และคำนวณ P&L"""
trade.exit_date = exit_date
trade.exit_price = exit_price
trade.pnl = (exit_price - trade.entry_price) * trade.quantity
self.current_capital += trade.pnl
self.total_trades += 1
if trade.pnl > 0:
self.winning_trades += 1
else:
self.losing_trades += 1
# Record equity curve
self.equity_curve.append({
"date": exit_date,
"capital": self.current_capital,
"pnl": trade.pnl
})
def run_backtest(
self,
df: pd.DataFrame,
strategy: str = "straddle",
entry_interval_days: int = 7,
holding_days: int = 30
) -> Dict:
"""รัน Backtest ทั้งหมด"""
print(f"🚀 Starting Backtest: {strategy}")
print(f" Initial Capital: ${self.initial_capital:,.2f}")
print(f" Entry Interval: {entry_interval_days} days")
print(f" Holding Period: {holding_days} days")
# Generate entry dates
start_date = df['timestamp'].min()
end_date = df['timestamp'].max()
entry_dates = []
current = start_date
while current < end_date:
entry_dates.append(current)
current += timedelta(days=entry_interval_days)
active_trades = {}
for entry_date in entry_dates:
# Execute new trades
if strategy == "straddle":
trade = self.execute_strategy_straddle(
df, entry_date, days_to_expiry=holding_days
)
if trade:
active_trades[trade] = trade.contract.expiry_date
# Check for expiries
trades_to_close = []
for trade, expiry in active_trades.items():
if entry_date >= expiry:
trades_to_close.append(trade)
# Close expired trades
for trade in trades_to_close:
# สมมติ expiry price = spot price
current_row = df[df['timestamp'] <= trade.contract.expiry_date].iloc[-1]
S_exit = current_row.get('underlying_price', 67000)
# Calculate exit price (intrinsic value only at expiry)
if trade.contract.option_type == "STRADDLE":
call_intrinsic = max(S_exit - trade.contract.strike, 0)
put_intrinsic = max(trade.contract.strike - S_exit, 0)
exit_price = call_intrinsic + put_intrinsic
else:
exit_price = 0
self.close_trade(trade, trade.contract.expiry_date, exit_price)
del active_trades[trade]
# Close remaining trades at end
final_date = df['timestamp'].max()
for trade in list(active_trades.keys()):
self.close_trade(trade, final_date, trade.entry_price * 0.5) # Assume 50% loss
return self.generate_report()
def generate_report(self) -> Dict:
"""สร้าง Backtest Report"""
if not self.equity_curve:
return {"error": "No trades executed"}
equity_df = pd.DataFrame(self.equity_curve)
total_return = (self.current_capital - self.initial_capital) / self.initial_capital
win_rate = self.winning_trades / self.total_trades if self.total_trades > 0 else 0
# Calculate max drawdown
equity_df['peak'] = equity_df['capital'].cummax()
equity_df['drawdown'] = (equity_df['capital'] - equity_df['peak']) / equity_df['peak']
max_drawdown = equity_df['drawdown'].min()
# Sharpe Ratio
returns = equity_df['capital'].pct_change().dropna()
sharpe_ratio = returns.mean() / returns.std() * np.sqrt(252) if returns.std() > 0 else 0
report = {
"total_trades": self.total_trades,
"winning_trades": self.winning_trades,
"losing_trades": self.losing_trades,
"win_rate": f"{win_rate:.2%}",
"initial_capital": f"${self.initial_capital:,.2f}",
"final_capital": f"${self.current_capital:,.2f}",
"total_return": f"{total_return:.2%}",
"max_drawdown": f"{max_drawdown:.2%}",
"sharpe_ratio": f"{sharpe_ratio:.2f}",
"avg_pnl": f"${equity_df['pnl'].mean():,.2f}",
}
print("\n" + "="*50)
print("📊 BACKTEST RESULTS")
print("="*50)
for key, value in report.items():
print(f" {key}: {value}")
print("="*50)
return report
ตัวอย่างการใช้งาน
if __name__ == "__main__":
#