บทความนี้จะพาคุณสร้าง pipeline สำหรับดึงข้อมูล Options Implied Volatility (IV) surface และ Greeks จาก Tardis API โดยใช้ ML Quantization ผ่าน HolySheep AI เพื่อสร้าง feature ที่พร้อมใช้สำหรับ training model ด้าน quantitative trading มาดูกันว่าทำอย่างไร
ML Quantization คืออะไร และทำไมต้องใช้กับ Options Data
ML Quantization คือเทคนิคการแปลงโมเดล Machine Learning จาก precision สูง (เช่น FP32) ไปเป็น precision ต่ำกว่า (เช่น INT8, FP16) เพื่อลดขนาดและเพิ่มความเร็วในการ inference โดยในบริบทของ Options Trading ที่มีข้อมูลจำนวนมหาศาล (IV surface, Greeks ของทุก strike price ทุก expiration) การ quantize model จะช่วยให้สามารถ:
- ประมวลผล IV surface หลายตัวในเวลาที่สั้นลง
- ลด memory footprint ลงถึง 75% เมื่อใช้ INT8
- รองรับ real-time feature extraction สำหรับ HFT strategies
Tardis API: แหล่งข้อมูล Options คุณภาพสูง
Tardis เป็น API ที่ให้บริการข้อมูล market data ระดับ institutional grade รวมถึง IV surface และ Greeks ของ options หลาย exchange เมื่อนำมารวมกับ HolySheep AI ที่รองรับ quantization พร้อม latency ต่ำกว่า 50ms คุณจะได้ pipeline ที่ทั้งเร็วและประหยัด
การตั้งค่า Environment และ Dependencies
# ติดตั้ง dependencies ที่จำเป็น
pip install torch transformers huggingface_hub
pip install tardis-grpc pandas numpy
pip install holy-shee p-ai-client # HolySheep SDK
หรือใช้ requirements.txt
torch>=2.0.0
transformers>=4.30.0
pandas>=2.0.0
numpy>=1.24.0
tardis-grpc>=1.0.0
โครงสร้าง Project สำหรับ Options Feature Engineering
options-ml-pipeline/
├── config/
│ └── settings.py # การตั้งค่า API keys และ parameters
├── data/
│ ├── fetch_tardis.py # ดึงข้อมูลจาก Tardis
│ └── preprocess.py # ประมวลผลข้อมูล IV และ Greeks
├── features/
│ ├── iv_surface.py # สร้าง IV surface features
│ ├── greeks_features.py # สร้าง Greeks-based features
│ └── strike_clusters.py # จัดกลุ่ม strike prices
├── quantization/
│ └── quantize_model.py # Quantize โมเดลด้วย HolySheep
├── models/
│ └── trainer.py # Training pipeline
├── main.py # Entry point
└── requirements.txtการดึงข้อมูล IV Surface และ Greeks จาก Tardis
import os
import grpc
from tardis.grpc import tardis_service_pb2, tardis_service_pb2_grpc
import pandas as pd
from typing import Dict, List
from datetime import datetime
class TardisDataFetcher:
"""Class สำหรับดึงข้อมูล Options จาก Tardis API"""
def __init__(self, api_key: str, exchange: str = "deribit"):
self.api_key = api_key
self.exchange = exchange
self.channel = grpc.secure_channel(
'tardis-grpc.tardis-api.com:443',
grpc.ssl_channel_credentials()
)
self.stub = tardis_service_pb2_grpc.TardisServiceStub(self.channel)
def get_iv_surface(self, underlying: str, timestamp: datetime) -> pd.DataFrame:
"""ดึง IV surface data สำหรับ underlying ใดก็ได้"""
request = tardis_service_pb2.IVSurfaceRequest(
exchange=self.exchange,
underlying=underlying,
timestamp=int(timestamp.timestamp())
)
response = self.stub.GetIVSurface(request, metadata=[
('api-key', self.api_key)
])
# แปลง response เป็น DataFrame
records = []
for strike in response.strikes:
for expiry in response.expirations:
records.append({
'strike': strike,
'expiry': expiry,
'iv': response.iv_surface[strike][expiry],
'bid_iv': response.bid_iv_surface[strike][expiry],
'ask_iv': response.ask_iv_surface[strike][expiry],
'timestamp': timestamp
})
return pd.DataFrame(records)
def get_greeks(self, underlying: str, timestamp: datetime) -> pd.DataFrame:
"""ดึง Greeks data (delta, gamma, theta, vega, rho)"""
request = tardis_service_pb2.GreeksRequest(
exchange=self.exchange,
underlying=underlying,
timestamp=int(timestamp.timestamp())
)
response = self.stub.GetGreeks(request, metadata=[
('api-key', self.api_key)
])
records = []
for option in response.options:
records.append({
'strike': option.strike,
'expiry': option.expiry,
'option_type': option.option_type, # 'call' หรือ 'put'
'delta': option.delta,
'gamma': option.gamma,
'theta': option.theta,
'vega': option.vega,
'rho': option.rho,
'timestamp': timestamp
})
return pd.DataFrame(records)
def get_historical_greeks(
self,
underlying: str,
start_time: datetime,
end_time: datetime,
interval_minutes: int = 60
) -> Dict[str, pd.DataFrame]:
"""ดึงข้อมูล Greeks แบบ historical สำหรับ feature engineering"""
request = tardis_service_pb2.HistoricalRequest(
exchange=self.exchange,
underlying=underlying,
start_time=int(start_time.timestamp()),
end_time=int(end_time.timestamp()),
interval_seconds=interval_minutes * 60
)
all_greeks = []
all_iv = []
for response in self.stub.GetHistorical(request, metadata=[
('api-key', self.api_key)
]):
all_greeks.append(self._parse_greeks_response(response.greeks))
all_iv.append(self._parse_iv_response(response.iv_surface))
return {
'greeks': pd.concat(all_greeks, ignore_index=True),
'iv_surface': pd.concat(all_iv, ignore_index=True)
}
def _parse_greeks_response(self, response) -> pd.DataFrame:
"""Parse Greeks response เป็น DataFrame"""
records = []
for option in response.options:
records.append({
'strike': option.strike,
'expiry': option.expiry,
'delta': option.delta,
'gamma': option.gamma,
'theta': option.theta,
'vega': option.vega,
'rho': option.rho,
'timestamp': datetime.fromtimestamp(response.timestamp)
})
return pd.DataFrame(records)
def _parse_iv_response(self, response) -> pd.DataFrame:
"""Parse IV surface response เป็น DataFrame"""
records = []
for strike_data in response.strikes:
for expiry_data in strike_data.expirations:
records.append({
'strike': strike_data.strike,
'expiry': expiry_data.expiry,
'iv': expiry_data.iv,
'bid_iv': expiry_data.bid_iv,
'ask_iv': expiry_data.ask_iv,
'timestamp': datetime.fromtimestamp(response.timestamp)
})
return pd.DataFrame(records)สร้าง Feature Engineering Pipeline สำหรับ ML Model
import numpy as np
import pandas as pd
from scipy.interpolate import griddata
from sklearn.preprocessing import StandardScaler
from typing import Tuple, Dict
from dataclasses import dataclass
@dataclass
class IVSurfaceFeatures:
"""Feature ที่สกัดจาก IV Surface"""
moneyness: np.ndarray # Moneyness ratio (S/K)
term_structure: np.ndarray # IV ตาม expiry
skew_metrics: np.ndarray # Skewness ของ IV
smile_curvature: np.ndarray # Curvature ของ IV smile
surface_slope: np.ndarray # Slope ของ IV surface
class OptionsFeatureEngineering:
"""Pipeline สำหรับสร้าง features จาก Options data"""
def __init__(self, spot_price: float):
self.spot_price = spot_price
self.scaler = StandardScaler()
def extract_iv_surface_features(
self,
iv_df: pd.DataFrame,
strikes: np.ndarray,
expirations: np.ndarray
) -> Dict[str, np.ndarray]:
"""สกัด features จาก IV surface"""
# สร้าง grid สำหรับ interpolation
iv_matrix = self._create_iv_matrix(iv_df, strikes, expirations)
# 1. Moneyness features (S/K ที่ various strikes)
moneyness = strikes / self.spot_price
moneyness_features = self._compute_moneyness_features(iv_matrix, moneyness)
# 2. Term structure features (IV เทียบกับ time to expiry)
term_structure = self._compute_term_structure(iv_matrix, expirations)
# 3. Skew metrics (ความเบ้ของ IV distribution)
skew_metrics = self._compute_skew_metrics(iv_matrix, strikes)
# 4. Smile curvature (Quadratic coefficient ของ IV smile)
smile_curvature = self._compute_smile_curvature(iv_matrix, strikes)
# 5. Surface slope (∂IV/∂K)
surface_slope = self._compute_surface_slope(iv_matrix, strikes)
return {
'moneyness_features': moneyness_features,
'term_structure': term_structure,
'skew_metrics': skew_metrics,
'smile_curvature': smile_curvature,
'surface_slope': surface_slope,
'iv_matrix_flat': iv_matrix.flatten()
}
def extract_greeks_features(self, greeks_df: pd.DataFrame) -> Dict[str, float]:
"""สกัด aggregate features จาก Greeks"""
# Greeks ที่ ATM (ใกล้ spot สุด)
atm_mask = abs(greeks_df['strike'] - self.spot_price) == \
abs(greeks_df['strike'] - self.spot_price).min()
atm_greeks = greeks_df[atm_mask].iloc[0] if atm_mask.any() else None
features = {
'atm_delta': atm_greeks['delta'] if atm_greeks is not None else 0.5,
'atm_gamma': atm_greeks['gamma'] if atm_greeks is not None else 0,
'atm_theta': atm_greeks['theta'] if atm_greeks is not None else 0,
'atm_vega': atm_greeks['vega'] if atm_greeks is not None else 0,
'atm_rho': atm_greeks['rho'] if atm_greeks is not None else 0,
}
# Weighted average Greeks
for greek in ['delta', 'gamma', 'theta', 'vega', 'rho']:
if greek in greeks_df.columns:
# Weight ตาม gamma (liquidity proxy)
weights = greeks_df['gamma'].fillna(0) + 1e-8
features[f'wavg_{greek}'] = np.average(
greeks_df[greek].fillna(0),
weights=weights
)
# Greeks dispersion (volatility ของ Greeks)
for greek in ['delta', 'gamma', 'vega']:
if greek in greeks_df.columns:
features[f'{greek}_std'] = greeks_df[greek].std()
return features
def create_strike_clusters(
self,
strikes: np.ndarray,
n_clusters: int = 5
) -> np.ndarray:
"""จัดกลุ่ม strikes เป็น clusters สำหรับ reduce dimensionality"""
# Log-moneyness based clustering
log_moneyness = np.log(strikes / self.spot_price)
# Uniform spacing clusters
percentiles = np.linspace(0, 100, n_clusters + 1)
cluster_edges = np.percentile(log_moneyness, percentiles)
# Assign cluster labels
clusters = np.digitize(log_moneyness, cluster_edges[1:-1])
return clusters
def build_feature_vector(
self,
iv_df: pd.DataFrame,
greeks_df: pd.DataFrame,
strikes: np.ndarray,
expirations: np.ndarray
) -> np.ndarray:
"""รวม features ทั้งหมดเป็น vector สำหรับ ML model"""
# IV surface features
iv_features = self.extract_iv_surface_features(iv_df, strikes, expirations)
# Greeks features
greeks_features = self.extract_greeks_features(greeks_df)
# Strike clusters
clusters = self.create_strike_clusters(strikes)
# Aggregate IV per cluster
iv_df['cluster'] = np.take(clusters, iv_df['strike'].values.argsort())
cluster_iv = iv_df.groupby('cluster')['iv'].agg(['mean', 'std', 'min', 'max'])
# Combine all features
feature_list = []
# Greeks scalar features
for key, value in greeks_features.items():
feature_list.append(value)
# Cluster IV stats
for stat in ['mean', 'std', 'min', 'max']:
for cluster_id in range(len(cluster_iv)):
if stat in cluster_iv.columns:
feature_list.append(cluster_iv.iloc[cluster_id][stat])
# Flattened IV matrix (ใช้เมื่อ IV surface มีขนาดคงที่)
feature_list.extend(iv_features['iv_matrix_flat'])
return np.array(feature_list)
def _create_iv_matrix(
self,
iv_df: pd.DataFrame,
strikes: np.ndarray,
expirations: np.ndarray
) -> np.ndarray:
"""สร้าง IV matrix จาก DataFrame"""
# Pivot table
iv_pivot = iv_df.pivot_table(
values='iv',
index='strike',
columns='expiry',
aggfunc='mean'
)
# Interpolate missing values
iv_matrix = iv_pivot.reindex(
index=strikes,
columns=expirations
).interpolate(method='linear', axis=0).interpolate(method='linear', axis=1)
return iv_matrix.fillna(method='bfill').fillna(method='ffill').values
def _compute_moneyness_features(
self,
iv_matrix: np.ndarray,
moneyness: np.ndarray
) -> np.ndarray:
"""คำนวณ features จาก moneyness"""
return np.array([
np.interp(0.8, moneyness, iv_matrix.mean(axis=1)), # OTM 20%
np.interp(0.9, moneyness, iv_matrix.mean(axis=1)), # OTM 10%
np.interp(1.0, moneyness, iv_matrix.mean(axis=1)), # ATM
np.interp(1.1, moneyness, iv_matrix.mean(axis=1)), # ITM 10%
np.interp(1.2, moneyness, iv_matrix.mean(axis=1)), # ITM 20%
])
def _compute_term_structure(
self,
iv_matrix: np.ndarray,
expirations: np.ndarray
) -> np.ndarray:
"""คำนวณ term structure ของ IV"""
return iv_matrix.mean(axis=0)
def _compute_skew_metrics(self, iv_matrix: np.ndarray, strikes: np.ndarray) -> np.ndarray:
"""คำนวณ skewness ของ IV"""
iv_across_strikes = iv_matrix.mean(axis=1)
# 25-delta skew (approximation)
atm_idx = np.argmin(np.abs(strikes - self.spot_price))
lower_iv = iv_across_strikes[:atm_idx].mean() if atm_idx > 0 else iv_across_strikes[0]
upper_iv = iv_across_strikes[atm_idx:].mean()
return np.array([
(upper_iv - lower_iv) / (iv_across_strikes.mean() + 1e-8), # Skewness
iv_across_strikes.std() / (iv_across_strikes.mean() + 1e-8), # Normalized dispersion
])
def _compute_smile_curvature(self, iv_matrix: np.ndarray, strikes: np.ndarray) -> np.ndarray:
"""คำนวณ curvature ของ IV smile"""
from scipy.stats import linregress
atm_idx = np.argmin(np.abs(strikes - self.spot_price))
# Fit quadratic around ATM
x = np.arange(len(strikes)) - atm_idx
y = iv_matrix.mean(axis=1)
# Second derivative approximation
if len(x) >= 3:
second_deriv = np.gradient(np.gradient(y, x), x)
return np.array([second_deriv.mean()])
return np.array([0])
def _compute_surface_slope(self, iv_matrix: np.ndarray, strikes: np.ndarray) -> np.ndarray:
"""คำนวณ slope ของ IV surface"""
return np.gradient(iv_matrix.mean(axis=1), strikes)Quantize Model ด้วย HolySheep AI
import os
import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer
from holy_sheep import HolySheepClient
from typing import Optional
class QuantizedOptionsModel:
"""ML Model สำหรับ Options prediction ที่ quantize แล้ว"""
# HolySheep API configuration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, model_name: str = "bert-base-uncased"):
self.api_key = api_key
self.model_name = model_name
self.client = HolySheepClient(api_key=api_key)
self.model = None
self.tokenizer = None
self.quantized = False
def load_base_model(self):
"""โหลด base model ก่อน quantization"""
self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
self.model = AutoModelForSequenceClassification.from_pretrained(
self.model_name,
num_labels=1 # Regression task สำหรับ price prediction
)
print(f"✓ โหลด base model: {self.model_name}")
print(f" Parameters: {sum(p.numel() for p in self.model.parameters()):,}")
def quantize_int8(self, calibration_data: torch.Tensor):
"""Quantize model เป็น INT8 ด้วย dynamic quantization"""
if self.model is None:
self.load_base_model()
# Dynamic quantization (Post-Training Quantization)
quantized_model = torch.quantization.quantize_dynamic(
self.model,
{torch.nn.Linear, torch.nn.LayerNorm},
dtype=torch.qint8
)
self.model = quantized_model
self.quantized = True
print("✓ Quantize เป็น INT8 สำเร็จ")
print(f" Original size: {self._get_model_size(self.model)/1024/1024:.2f} MB")
def quantize_fp16(self):
"""Quantize model เป็น FP16 (half precision)"""
if self.model is None:
self.load_base_model()
self.model = self.model.half()
self.quantized = True
print("✓ Quantize เป็น FP16 สำเร็จ")
def export_for_inference(self, output_path: str):
"""Export quantized model สำหรับ inference"""
if not self.quantized:
print("⚠ Model ยังไม่ได้ quantize")
return
# Save model state dict
torch.save(
self.model.state_dict(),
os.path.join(output_path, "model_quantized.pt")
)
# Save tokenizer
self.tokenizer.save_pretrained(
os.path.join(output_path, "tokenizer")
)
print(f"✓ Export ไปที่ {output_path}")
def inference_with_holysheep(
self,
features: list,
use_quantized: bool = True
) -> dict:
"""
เรียกใช้ inference ผ่าน HolySheep API
HolySheep รองรับ quantized models พร้อม latency ต่ำกว่า 50ms
ประหยัด cost สูงสุด 85% เมื่อเทียบกับ OpenAI
"""
# เตรียม payload
payload = {
"model": "gpt-4.1", # หรือใช้ quantized version
"messages": [
{
"role": "system",
"content": "You are a quantitative analyst. Process options features and predict price movement."
},
{
"role": "user",
"content": f"Analyze these options features and predict directional movement (1=up, 0=down):\n{features}"
}
],
"temperature": 0.1,
"max_tokens": 100
}
# เรียก HolySheep API
response = self.client.chat.completions.create(**payload)
return {
"prediction": response.choices[0].message.content,
"usage": response.usage.total_tokens,
"latency_ms": response.latency # <50ms guaranteed
}
def batch_inference(
self,
feature_batch: list,
batch_size: int = 32
) -> list:
"""Batch inference สำหรับหลาย samples"""
results = []
for i in range(0, len(feature_batch), batch_size):
batch = feature_batch[i:i+batch_size]
# Prepare batch prompt
batch_text = "\n---\n".join([
f"Sample {j+1}: {f}" for j, f in enumerate(batch)
])
response = self.inference_with_holysheep(
features=[batch_text],
use_quantized=True
)
results.append(response)
return results
def _get_model_size(self, model) -> int:
"""คำนวณขนาด model เป็น bytes"""
param_size = 0
for param in model.parameters():
param_size += param.nelement() * param.element_size()
buffer_size = 0
for buffer in model.buffers():
buffer_size += buffer.nelement() * buffer.element_size()
return param_size + buffer_size
ตัวอย่างการใช้งาน
if __name__ == "__main__":
# Initialize
holysheep_key = os.environ.get("HOLYSHEEP_API_KEY")
model = QuantizedOptionsModel(
api_key=holysheep_key,
model_name="bert-base-uncased"
)
# Load และ quantize
model.load_base_model()
model.quantize_int8(calibration_data=None)
# Export
model.export_for_inference("./quantized_model")
# Test inference
sample_features = [0.5, 0.3, 0.2, 0.8, 0.1] * 20 # Example features
result = model.inference_with_holysheep(sample_features)
print(f"Prediction: {result}")Integration: รวมทุกอย่างใน Main Pipeline
import os
from datetime import datetime, timedelta
import pandas as pd
import numpy as np
from typing import Optional
class OptionsMLPipeline:
"""Main pipeline ที่รวม Tardis + Feature Engineering + HolySheep"""
def __init__(
self,
tardis_api_key: str,
holysheep_api_key: str,
underlying: str = "BTC",
exchange: str = "deribit"
):
self.underlying = underlying
self.exchange = exchange
# Initialize clients
from fetch_tardis import TardisDataFetcher
self.tardis = TardisDataFetcher(
api_key=tardis_api_key,
exchange=exchange
)
self.holysheep = QuantizedOptionsModel(
api_key=holysheep_api_key
)
# Initialize feature engineer
self.feature_engine = None # จะ set เมื่อได้ spot price
def fetch_and_process(
self,
timestamp: datetime,
lookback_days: int = 30
) -> dict:
"""ดึงข้อมูล, สร้าง features และทำ inference"""
# 1. ดึงข้อมูล IV surface และ Greeks
print(f"📥 ดึงข้อมูล IV surface และ Greeks สำหรับ {self.underlying}")
iv_df = self.tardis.get_iv_surface(self.underlying, timestamp)
greeks_df = self.tardis.get_greeks(self.underlying, timestamp)
# 2. คำนวณ spot price (จาก ATM option)
atm_mask = abs(greeks_df['strike'] - greeks_df['strike'].median()).abs().min()
spot_price = greeks_df.loc[greeks_df['strike'].sub(spot_price_approx).abs().idxmin(), 'strike']
# Initialize feature engineer with spot price
self.feature_engine = OptionsFeatureEngineering(spot_price=spot_price)
# 3. สร้าง feature vector
strikes = greeks_df['strike'].unique()
expirations = greeks_df['expiry'].unique()
features = self.feature_engine.build_feature_vector(
iv_df=iv_df,
greeks_df=greeks_df,
strikes=strikes,
expirations=expirations
)
# 4. Quantize model และ inference
print("🤖 Quantizing model และทำ inference...")
if not self.holysheep.quantized:
self.holysheep.load_base_model()
self.holysheep.quantize_int8()
result = self.holysheep.inference_with_holysheep(
features=features.tolist()
)
return {
'timestamp': timestamp,
'underlying': self.underlying,
'spot_price': spot_price,
'features': features,
'prediction': result['prediction'],
'latency_ms': result['latency_ms']
}
def backfill_and_train(
self,
start_date: datetime,
end_date: datetime,
interval_hours: int = 4
) -> pd.DataFrame:
"""ดึงข้อมูล historical สำหรับ training"""
results = []
current = start_date
while current <= end_date:
try:
result = self.fetch_and_process(current)
results.append(result)
print(f"✓ {current} - Latency: {result['latency_ms']:.2f}ms")
except Exception as e:
print(f"✗ Error at {current}: {e}")
current += timedelta(hours=interval_hours)
# Convert to DataFrame
df = pd.DataFrame(results)
return df
def run_realtime(self, interval_seconds: int = 60):
"""รัน pipeline แบบ realtime"""
import time
print(f"🚀 เริ่ม realtime pipeline (interval: {interval_seconds}s)")
while True:
try:
result = self.fetch_and_process(datetime.now())
print(f"[{result['timestamp']}] Prediction: {result['prediction']}")
except Exception as e:
print(f"Error: {e}")
time.sleep(interval_seconds)
การใช้งาน
if __name__ == "__main__":
# ตั้งค่า API keys
TARDIS_KEY = os.environ.get("TARDIS_API_KEY")
HOLYSHEEP_KEY = os.environ.get("HOLYSHEEP_API_KEY")
# Initialize pipeline
pipeline = OptionsMLPipeline(
tardis_api_key=TARDIS_KEY,
holysheep_api_key=HOLYSHEEP_KEY,
underlying="BTC",
exchange="deribit"
)
# ดึงข้อมูลย้อนหลัง 7 วัน
df = pipeline.backfill_and_train(
start_date=datetime.now() - timedelta(days=7),
end_date=datetime.now(),
interval_hours=4
)
print(f"\n📊 ดึงข้อมูลสำเร็จ: {len(df)} records")
print(df.head())ประสิทธิภาพและผลลัพธ์
เมื่อใช้ pipeline นี้กับ HolySheep AI คุณจะได้ผลลัพธ์ดังนี้:
| Metric | Before Quantization | After INT8 Quantization | Improvement |
|---|---|---|---|