บทนำ：ปัญหา "ConnectionError: timeout" ที่ทำให้วิเคราะห์ความผันผวนล้มเหลว
ในฐานะนักพัฒนาระบบวิเคราะห์ความผันผวน (Volatility Analyst) ฉันเคยเจอสถานการณ์ที่ทำให้เสียเวลาหลายชั่วโมง — กำลังดึงข้อมูล OKX ออปชันเพื่อสร้าง volatility surface แต่สคริปต์ค้างที่บรรทัด
response = requests.get(url, timeout=30) พร้อมข้อความ:
ConnectionError: HTTPSConnectionPool(host='api.tardis.dev', port=443):
Max retries exceeded with url: /v1/okex/options/chain (Caused by
ConnectTimeoutError(<urllib3.connection.HTTPSConnection object at 0x...>,
Connection timed out after 45000ms))
ปัญหานี้เกิดจากหลายสาเหตุ: rate limiting จาก Tardis API, เครือข่ายไม่เสถียร หรือ API key หมดอายุ บทความนี้จะสอนวิธีแก้ปัญหาเหล่านี้อย่างเป็นระบบ พร้อมโค้ด Python ที่พร้อมใช้งานจริงสำหรับการดึงข้อมูล OKX ออปชันและคำนวณ implied volatility
Tardis API：แหล่งข้อมูลความผันผวนที่ครอบคลุม
Tardis (tardis.dev) เป็นผู้ให้บริการข้อมูลตลาดคริปโตระดับสถาบันที่ครอบคลุม OKX, Binance, Bybit และ Huobi สำหรับ OKX ออปชัน คุณสามารถเข้าถึงข้อมูล:
- Options Chain: ข้อมูลออปชันทั้งหมดในห่วงโซ่ พร้อม strike price, expiry, type (call/put)
- Trades: ประวัติการซื้อขายแบบละเอียด พร้อม timestamp ระดับ millisecond
- Quotes: ราคา bid/ask แบบเรียลไทม์สำหรับคำนวณ bid-ask spread
- Book Snapshots: ภาพรวม order book สำหรับวิเคราะห์ความลึกของตลาด
การตั้งค่า Environment และการรับข้อมูลจาก Tardis CSV
เริ่มต้นด้วยการติดตั้ง dependencies และตั้งค่า API credentials:
# ติดตั้ง dependencies
pip install requests pandas numpy scipy matplotlib python-dotenv
สร้างไฟล์ .env
TARDIS_API_KEY=your_tardis_api_key
OKX_API_KEY=your_okx_api_key (ถ้าต้องการข้อมูลเรียลไทม์)
โค้ดด้านล่างนี้ใช้วิธีดาวน์โหลด CSV dataset จาก Tardis ซึ่งเหมาะสำหรับการวิเคราะห์แบบ batch และหลีกเลี่ยงปัญหา rate limiting:
import requests
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import time
import logging
from typing import Optional, Dict, List
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class TardisDataFetcher:
"""คลาสสำหรับดึงข้อมูล OKX ออปชันจาก Tardis API"""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def get_options_chain(
self,
exchange: str = "okex",
symbol: str = "BTC-USD",
start_date: str = "2024-01-01",
end_date: str = "2024-01-07",
format_type: str = "csv"
) -> Optional[pd.DataFrame]:
"""
ดึงข้อมูลออปชันเชนจาก Tardis
Args:
exchange: ตลาด (okex, binance, bybit)
symbol: คู่สินทรัพย์ (BTC-USD, ETH-USD)
start_date: วันที่เริ่มต้น (YYYY-MM-DD)
end_date: วันที่สิ้นสุด (YYYY-MM-DD)
format_type: csv หรือ json
Returns:
DataFrame ที่มีข้อมูลออปชัน หรือ None ถ้าล้มเหลว
"""
# ตรวจสอบ format ของวันที่
try:
start_dt = datetime.strptime(start_date, "%Y-%m-%d")
end_dt = datetime.strptime(end_date, "%Y-%m-%d")
except ValueError as e:
logger.error(f"รูปแบบวันที่ไม่ถูกต้อง: {e}")
raise
# ตรวจสอบช่วงวันที่ (ไม่เกิน 31 วันต่อ request)
if (end_dt - start_dt).days > 31:
logger.warning("ช่วงวันที่เกิน 31 วัน จะดึงทีละเดือน")
return self._fetch_in_chunks(symbol, start_date, end_date)
url = f"{self.BASE_URL}/export/{exchange}/options/chain"
params = {
"symbol": symbol,
"start_date": start_date,
"end_date": end_date,
"format": format_type,
"has_columns": True
}
max_retries = 3
retry_delay = 5
for attempt in range(max_retries):
try:
logger.info(f"กำลังดึงข้อมูล {symbol} จาก {start_date} ถึง {end_date}")
response = self.session.get(
url,
params=params,
timeout=120 # เพิ่ม timeout สำหรับ CSV
)
response.raise_for_status()
# แปลง CSV เป็น DataFrame
from io import StringIO
df = pd.read_csv(StringIO(response.text))
logger.info(f"ได้ข้อมูล {len(df)} รายการสำเร็จ")
return df
except requests.exceptions.Timeout:
logger.warning(f"Timeout ในครั้งที่ {attempt + 1}/{max_retries}")
if attempt < max_retries - 1:
time.sleep(retry_delay * (attempt + 1))
except requests.exceptions.HTTPError as e:
if e.response.status_code == 401:
logger.error("Unauthorized: ตรวจสอบ API key ของคุณ")
raise
elif e.response.status_code == 429:
logger.warning(f"Rate limited: รอ {retry_delay * 2} วินาที")
time.sleep(retry_delay * 2)
else:
logger.error(f"HTTP Error: {e}")
raise
except requests.exceptions.ConnectionError as e:
logger.warning(f"Connection error ในครั้งที่ {attempt + 1}: {e}")
if attempt < max_retries - 1:
time.sleep(retry_delay)
logger.error("ดึงข้อมูลล้มเหลวหลังจากลอง 3 ครั้ง")
return None
def _fetch_in_chunks(
self,
symbol: str,
start_date: str,
end_date: str
) -> pd.DataFrame:
"""ดึงข้อมูลทีละเดือนเพื่อหลีกเลี่ยงปัญหา limit"""
all_data = []
current = datetime.strptime(start_date, "%Y-%m-%d")
end = datetime.strptime(end_date, "%Y-%m-%d")
while current < end:
chunk_end = min(current + timedelta(days=31), end)
chunk_start = current.strftime("%Y-%m-%d")
chunk_end_str = chunk_end.strftime("%Y-%m-%d")
df = self.get_options_chain(
symbol=symbol,
start_date=chunk_start,
end_date=chunk_end_str
)
if df is not None:
all_data.append(df)
current = chunk_end + timedelta(days=1)
time.sleep(2) # รอระหว่าง chunk
if all_data:
return pd.concat(all_data, ignore_index=True)
return pd.DataFrame()
ตัวอย่างการใช้งาน
if __name__ == "__main__":
from dotenv import load_dotenv
load_dotenv()
import os
TARDIS_KEY = os.getenv("TARDIS_API_KEY")
if not TARDIS_KEY:
logger.error("กรุณาตั้งค่า TARDIS_API_KEY ในไฟล์ .env")
exit(1)
fetcher = TardisDataFetcher(TARDIS_KEY)
df = fetcher.get_options_chain(
symbol="BTC-USD",
start_date="2024-01-01",
end_date="2024-01-03"
)
if df is not None:
print(df.head())
print(f"\nColumns: {df.columns.tolist()}")
การคำนวณ Implied Volatility และ Historical Volatility
หลังจากได้ข้อมูลมาแล้ว ขั้นตอนสำคัญคือการคำนวณความผันผวน 2 แบบ:
- Historical Volatility (HV): ความผันผวนที่เกิดขึ้นจริงในอดีต คำนวณจาก standard deviation ของ log returns
- Implied Volatility (IV): ความผันผวนที่ "ซ่อน" อยู่ในราคาตลาด คำนวณย้อนกลับจาก Black-Scholes model
from scipy.stats import norm
from scipy.optimize import brentq, newton
from scipy import sparse
import warnings
warnings.filterwarnings('ignore')
class VolatilityAnalyzer:
"""คลาสสำหรับวิเคราะห์ความผันผวนของ OKX ออปชัน"""
def __init__(self, risk_free_rate: float = 0.05):
"""
Args:
risk_free_rate: อัตราดอกเบี้ยปลอดภัย (ต่อปี)
"""
self.r = risk_free_rate
def calculate_historical_volatility(
self,
prices: np.ndarray,
window: int = 20
) -> np.ndarray:
"""
คำนวณ Historical Volatility โดยใช้ rolling standard deviation
Args:
prices: ราคาสินทรัพย์
window: จำนวนวันสำหรับคำนวณ (default: 20 วัน)
Returns:
Array ของ historical volatility (annualized)
"""
# คำนวณ log returns
log_returns = np.diff(np.log(prices))
# คำนวณ rolling standard deviation
hv = np.zeros(len(log_returns) - window + 1)
for i in range(len(hv)):
hv[i] = np.std(log_returns[i:i+window])
# Annualize (ใช้ 252 วันทำการ)
hv_annualized = hv * np.sqrt(252)
return hv_annualized
def black_scholes_price(
self,
S: float, # Spot price
K: float, # Strike price
T: float, # Time to expiry (years)
r: float, # Risk-free rate
sigma: float, # Volatility
option_type: str = "call" # "call" หรือ "put"
) -> float:
"""
คำนวณราคาออปชันตาม Black-Scholes Model
Args:
S: ราคา spot ปัจจุบัน
K: Strike price
T: เวลาถึง expiry (ปี)
r: อัตราดอกเบี้ยปลอดภัย
sigma: ความผันผวน
option_type: "call" หรือ "put"
Returns:
ราคาออปชันตามทฤษฎี
"""
if T <= 0 or sigma <= 0:
return max(0, S - K) if option_type == "call" else max(0, K - S)
d1 = (np.log(S / K) + (r + 0.5 * sigma ** 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: # put
price = K * np.exp(-r * T) * norm.cdf(-d2) - S * norm.cdf(-d1)
return price
def calculate_implied_volatility(
self,
market_price: float,
S: float,
K: float,
T: float,
r: float,
option_type: str = "call",
precision: float = 1e-6,
max_iterations: int = 100
) -> Optional[float]:
"""
คำนวณ Implied Volatility โดยใช้ Newton-Raphson Method
Args:
market_price: ราคาตลาดจริง
S, K, T, r: ดู docstring ของ black_scholes_price
option_type: "call" หรือ "put"
precision: ความแม่นยำที่ต้องการ
max_iterations: จำนวนรอบสูงสุด
Returns:
Implied volatility หรือ None ถ้าคำนวณไม่ได้
"""
def objective(sigma):
return self.black_scholes_price(S, K, T, r, sigma, option_type) - market_price
# ตรวจสอบ intrinsic value
intrinsic = max(0, S - K) if option_type == "call" else max(0, K - S)
if market_price < intrinsic:
return None # ราคาต่ำกว่า intrinsic value
try:
# ลอง Brent's method
iv = brentq(
objective,
0.001, # sigma ต่ำสุด
5.0, # sigma สูงสุด (500%)
xtol=precision
)
return iv
except ValueError:
# ถ้า Brent ไม่ได้ ลอง Newton-Raphson
sigma = 0.3 # initial guess
for _ in range(max_iterations):
price = self.black_scholes_price(S, K, T, r, sigma, option_type)
vega = self._calculate_vega(S, K, T, r, sigma, option_type)
if abs(vega) < 1e-10:
break
diff = price - market_price
if abs(diff) < precision:
return sigma
sigma = sigma - diff / vega
sigma = max(0.001, min(sigma, 5.0)) # bound sigma
return sigma if sigma < 5.0 else None
def _calculate_vega(
self, S, K, T, r, sigma, option_type
) -> float:
"""คำนวณ Vega สำหรับ Newton-Raphson"""
if T <= 0 or sigma <= 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 build_volatility_smile(
self,
df: pd.DataFrame,
expiry_date: str,
spot_price: float
) -> pd.DataFrame:
"""
สร้าง Volatility Smile/Skew สำหรับ expiry เดียว
Args:
df: DataFrame จาก Tardis (ต้องมี columns: strike, price, type, expiry)
expiry_date: วันที่ expiry ที่ต้องการ
spot_price: ราคา spot ปัจจุบัน
Returns:
DataFrame ที่มี strike, iv, moneyness
"""
# กรองข้อมูลตาม expiry
expiry_df = df[df['expiry'] == expiry_date].copy()
# คำนวณ IV สำหรับแต่ละ strike
results = []
for _, row in expiry_df.iterrows():
K = row['strike']
market_price = row['price']
option_type = row['type'] # 'call' หรือ 'put'
T = row.get('days_to_expiry', 30) / 365
iv = self.calculate_implied_volatility(
market_price, spot_price, K, T, self.r, option_type
)
if iv is not None:
results.append({
'strike': K,
'iv': iv,
'moneyness': K / spot_price, # K/S
'type': option_type,
'market_price': market_price
})
return pd.DataFrame(results)
ตัวอย่างการใช้งาน
if __name__ == "__main__":
analyzer = VolatilityAnalyzer(risk_free_rate=0.05)
# ตัวอย่าง: คำนวณ IV จากราคาตลาด
spot = 45000
strike = 46000
T = 30 / 365 # 30 วัน
market_price = 2500 # USD
iv = analyzer.calculate_implied_volatility(
market_price=market_price,
S=spot,
K=strike,
T=T,
r=0.05,
option_type="call"
)
print(f"Implied Volatility: {iv * 100:.2f}%")
# ตรวจสอบด้วย Black-Scholes
theoretical_price = analyzer.black_scholes_price(
spot, strike, T, 0.05, iv, "call"
)
print(f"Theoretical Price: ${theoretical_price:.2f}")
print(f"Market Price: ${market_price:.2f}")
การสร้าง Volatility Surface และการวิเคราะห์
เมื่อได้ IV สำหรับหลายๆ strike และ expiry แล้ว สามารถสร้าง 3D surface เพื่อเห็นภาพรวมของตลาด:
import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D
class VolatilitySurfacePlotter:
"""คลาสสำหรับสร้างกราฟ Volatility Surface"""
def __init__(self, analyzer: VolatilityAnalyzer):
self.analyzer = analyzer
def plot_volatility_smile(
self,
vol_data: pd.DataFrame,
title: str = "Volatility Smile",
save_path: Optional[str] = None
):
"""สร้างกราฟ Volatility Smile"""
fig, ax = plt.subplots(figsize=(12, 7))
for opt_type in vol_data['type'].unique():
subset = vol_data[vol_data['type'] == opt_type]
ax.plot(
subset['moneyness'],
subset['iv'] * 100,
'o-',
label=f'{opt_type.upper()}',
markersize=8,
linewidth=2
)
ax.axvline(x=1.0, color='gray', linestyle='--', alpha=0.5, label='ATM')
ax.set_xlabel('Moneyness (K/S)', fontsize=12)
ax.set_ylabel('Implied Volatility (%)', fontsize=12)
ax.set_title(title, fontsize=14, fontweight='bold')
ax.legend()
ax.grid(True, alpha=0.3)
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=150, bbox_inches='tight')
plt.show()
def plot_volatility_surface(
self,
strikes: np.ndarray,
expiries: np.ndarray,
iv_matrix: np.ndarray,
title: str = "OKX Options Volatility Surface",
save_path: Optional[str] = None
):
"""สร้าง 3D Volatility Surface"""
fig = plt.figure(figsize=(14, 10))
ax = fig.add_subplot(111, projection='3d')
# สร้าง meshgrid
X, Y = np.meshgrid(strikes, expiries)
# วาด surface
surf = ax.plot_surface(
X, Y, iv_matrix * 100,
cmap='viridis',
edgecolor='none',
alpha=0.8
)
ax.set_xlabel('Strike Price', fontsize=11)
ax.set_ylabel('Days to Expiry', fontsize=11)
ax.set_zlabel('Implied Volatility (%)', fontsize=11)
ax.set_title(title, fontsize=14, fontweight='bold')
# เพิ่ม colorbar
fig.colorbar(surf, shrink=0.5, aspect=10, label='IV (%)')
# ปรับมุมมอง
ax.view_init(elev=25, azim=45)
plt.tight_layout()
if save_path
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง