บทความนี้อธิบายวิธีสร้าง Deribit Greeks ข้อมูลประวัติ (Historical Data Operations Dashboard) ที่ครอบคลุม 4 มิติสำคัญ ได้แก่ Data Completeness Rate, Recalculation Tasks, Strategy Dependencies และ Research Team Satisfaction โดยใช้สถาปัตยกรรม Production-Grade ที่รองรับ High-Throughput และ Low-Latency
ภาพรวมสถาปัตยกรรมระบบ
ระบบนี้ออกแบบมาเพื่อรองรับการประมวลผล Greeks Data จาก Deribit Exchange โดยมีส่วนประกอบหลักดังนี้:
- Data Ingestion Layer: WebSocket Consumer สำหรับ Real-time Greeks Data
- Storage Layer: TimescaleDB สำหรับ Time-series Data + Redis Cache
- Processing Layer: Python Async Workers สำหรับ Recalculation Pipeline
- Monitoring Layer: Prometheus + Grafana สำหรับ Metrics Collection
- AI Enhancement: HolySheep AI สำหรับ Smart Alerting และ Analysis
การดึงข้อมูล Greeks จาก Deribit
Deribit API ให้ข้อมูล Greeks ผ่าน public/get_book_summary_by_instrument และ GET /api/v2/public/get_greeks โดยมี Rate Limit ที่ 10 requests/second สำหรับ Public API
สคริปต์ดึงข้อมูล Greeks พร้อม Data Quality Check
#!/usr/bin/env python3
"""
Deribit Greeks Historical Data Fetcher
Production-Grade Implementation พร้อม Retry Logic และ Data Validation
"""
import asyncio
import aiohttp
import hashlib
import logging
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional, Dict, List
from tenacity import retry, stop_after_attempt, wait_exponential
import timescaledb
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class GreeksRecord:
"""โครงสร้างข้อมูล Greeks สำหรับ Deribit Options"""
timestamp: datetime
instrument_name: str
underlying_price: float
mark_price: float
open_interest: float
delta: float
gamma: float
theta: float
vega: float
rho: float
bid_price: float
ask_price: float
bid_iv: float
ask_iv: float
data_hash: str = field(default="")
def __post_init__(self):
# Generate unique hash for deduplication
raw = f"{self.timestamp}{self.instrument_name}{self.underlying_price}"
self.data_hash = hashlib.sha256(raw.encode()).hexdigest()[:16]
def validate(self) -> tuple[bool, str]:
"""ตรวจสอบความถูกต้องของข้อมูล"""
if not (-1 <= self.delta <= 1):
return False, f"Delta out of range: {self.delta}"
if self.gamma < 0:
return False, f"Negative gamma: {self.gamma}"
if self.theta > 0:
return False, f"Positive theta for put: {self.theta}"
if self.vega < 0:
return False, f"Negative vega: {self.vega}"
if self.ask_price <= 0 or self.bid_price <= 0:
return False, "Invalid prices"
if self.ask_price < self.bid_price:
return False, f"Ask < Bid: {self.ask_price} < {self.bid_price}"
return True, "OK"
@dataclass
class DataCompletenessMetrics:
"""Metrics สำหรับติดตาม Data Completeness Rate"""
total_expected_records: int = 0
total_actual_records: int = 0
missing_intervals: List[Dict] = field(default_factory=list)
duplicate_count: int = 0
invalid_records: int = 0
@property
def completeness_rate(self) -> float:
if self.total_expected_records == 0:
return 100.0
return (self.total_actual_records / self.total_expected_records) * 100
@property
def data_quality_score(self) -> float:
total = self.total_actual_records
if total == 0:
return 0.0
penalty = (self.duplicate_count + self.invalid_records) / total
return max(0.0, (1.0 - penalty) * 100)
class DeribitGreeksFetcher:
"""Production-Grade Deribit Greeks Fetcher พร้อม Rate Limiting"""
BASE_URL = "https://www.deribit.com/api/v2/public"
RATE_LIMIT = 10 # requests per second
def __init__(self, db_pool):
self.session: Optional[aiohttp.ClientSession] = None
self.db_pool = db_pool
self.semaphore = asyncio.Semaphore(self.RATE_LIMIT)
self.request_timestamps: List[float] = []
self.metrics = DataCompletenessMetrics()
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=30, connect=10)
self.session = aiohttp.ClientSession(timeout=timeout)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def _rate_limit_wait(self):
"""Implement Token Bucket Rate Limiting"""
now = asyncio.get_event_loop().time()
# Remove timestamps older than 1 second
self.request_timestamps = [ts for ts in self.request_timestamps if now - ts < 1.0]
if len(self.request_timestamps) >= self.RATE_LIMIT:
oldest = self.request_timestamps[0]
wait_time = 1.0 - (now - oldest) + 0.01
if wait_time > 0:
await asyncio.sleep(wait_time)
self.request_timestamps.append(now)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
async def fetch_greeks(self, instrument_name: str) -> Optional[Dict]:
"""ดึงข้อมูล Greeks สำหรับ Instrument เดียว พร้อม Auto-retry"""
await self._rate_limit_wait()
url = f"{self.BASE_URL}/get_greeks"
params = {"instrument_name": instrument_name}
try:
async with self.session.get(url, params=params) as response:
if response.status == 429:
logger.warning(f"Rate limited, backing off...")
await asyncio.sleep(5)
raise aiohttp.ClientError("Rate limited")
response.raise_for_status()
data = await response.json()
if data.get("success") and data.get("result"):
return data["result"]
return None
except aiohttp.ClientError as e:
logger.error(f"Failed to fetch {instrument_name}: {e}")
raise
async def fetch_greeks_batch(
self,
instruments: List[str],
on_progress=None
) -> List[GreeksRecord]:
"""ดึงข้อมูล Greeks แบบ Batch พร้อม Concurrent Requests"""
results = []
total = len(instruments)
async def fetch_one(instrument: str, idx: int):
try:
data = await self.fetch_greeks(instrument)
if data:
record = GreeksRecord(
timestamp=datetime.utcnow(),
instrument_name=instrument,
underlying_price=float(data.get("underlying_price", 0)),
mark_price=float(data.get("mark_price", 0)),
open_interest=float(data.get("open_interest", 0)),
delta=float(data.get("delta", 0)),
gamma=float(data.get("gamma", 0)),
theta=float(data.get("theta", 0)),
vega=float(data.get("vega", 0)),
rho=float(data.get("rho", 0)),
bid_price=float(data.get("bid_price", 0)),
ask_price=float(data.get("ask_price", 0)),
bid_iv=float(data.get("bid_iv", 0)),
ask_iv=float(data.get("ask_iv", 0))
)
is_valid, msg = record.validate()
if not is_valid:
self.metrics.invalid_records += 1
logger.warning(f"Invalid record {instrument}: {msg}")
return None
return record
except Exception as e:
logger.error(f"Error fetching {instrument}: {e}")
return None
tasks = [fetch_one(inst, idx) for idx, inst in enumerate(instruments)]
for idx, coro in enumerate(asyncio.as_completed(tasks)):
record = await coro
if record:
results.append(record)
if on_progress and (idx + 1) % 100 == 0:
on_progress(idx + 1, total)
self.metrics.total_actual_records = len(results)
return results
ตัวอย่างการใช้งาน
async def main():
"""Demonstration การดึงข้อมูล Greeks"""
# รายการ Instrument ที่ต้องการดึง (BTC Options)
instruments = [
f"BTC-{expiry}-{strike}-{ptype}"
for expiry in ["25-06", "30-06", "28-07"]
for strike in range(60000, 100000, 5000)
for ptype in ["P", "C"]
]
fetcher = DeribitGreeksFetcher(db_pool=None) # ตั้งค่า DB pool จริง
async with fetcher:
def progress(current, total):
print(f"Progress: {current}/{total} ({current*100/total:.1f}%)")
records = await fetcher.fetch_greeks_batch(instruments[:50], on_progress=progress)
print(f"\n=== Data Completeness Metrics ===")
print(f"Total Records: {fetcher.metrics.total_actual_records}")
print(f"Completeness Rate: {fetcher.metrics.completeness_rate:.2f}%")
print(f"Data Quality Score: {fetcher.metrics.data_quality_score:.2f}%")
print(f"Invalid Records: {fetcher.metrics.invalid_records}")
if __name__ == "__main__":
asyncio.run(main())ระบบติดตาม Recalculation Tasks
เมื่อข้อมูล Greeks มีการเปลี่ยนแปลง (เช่น ราคาเปลี่ยน หรือ Implied Volatility เปลี่ยน) ระบบต้อง Recalculate ค่า Greeks ที่เกี่ยวข้องทั้งหมด โดยใช้ Dependency Graph และ Task Queue
Recalculation Task Manager พร้อม Priority Queue
#!/usr/bin/env python3
"""
Recalculation Task Manager สำหรับ Deribit Greeks
รองรับ Priority-based Processing และ Dependency Management
"""
import asyncio
import heapq
from dataclasses import dataclass, field
from typing import Dict, List, Set, Optional, Callable
from datetime import datetime
from enum import Enum
from collections import defaultdict
import redis.asyncio as redis
import json
class TaskPriority(Enum):
CRITICAL = 1 # สำหรับ Live PnL Calculation
HIGH = 2 # สำหรับ Risk Management
MEDIUM = 3 # สำหรับ Historical Analysis
LOW = 4 # สำหรับ Batch Reports
@dataclass(order=True)
class RecalcTask:
"""Task สำหรับ Recalculate Greeks"""
priority: int
task_id: str = field(compare=False)
instrument_name: str = field(compare=False)
task_type: str = field(compare=False) # 'greeks', 'iv_surface', 'portfolio_risk'
payload: Dict = field(compare=False, default_factory=dict)
created_at: datetime = field(compare=False, default_factory=datetime.utcnow)
scheduled_at: Optional[datetime] = field(compare=False, default=None)
retry_count: int = field(compare=False, default=0)
max_retries: int = field(compare=False, default=3)
def __post_init__(self):
self.priority = self.priority.value if isinstance(self.priority, TaskPriority) else self.priority
@dataclass
class StrategyDependency:
"""ข้อมูล Dependency ระหว่าง Strategies และ Greeks Data"""
strategy_id: str
strategy_name: str
required_instruments: Set[str]
calculation_interval_ms: int
priority: TaskPriority
last_recalc: datetime = field(default_factory=datetime.utcnow)
recalc_count: int = field(default=0)
failure_count: int = field(default=0)
class DependencyGraph:
"""Directed Acyclic Graph (DAG) สำหรับ Strategy Dependencies"""
def __init__(self):
self.nodes: Dict[str, StrategyDependency] = {}
self.edges: Dict[str, Set[str]] = defaultdict(set) # strategy -> depends_on
self.reverse_edges: Dict[str, Set[str]] = defaultdict(set) # strategy -> dependents
def add_strategy(self, strategy: StrategyDependency):
"""เพิ่ม Strategy ในกราฟ"""
self.nodes[strategy.strategy_id] = strategy
def add_dependency(self, strategy_id: str, depends_on: str):
"""เพิ่ม Dependency (strategy_id ขึ้นอยู่กับ depends_on)"""
self.edges[strategy_id].add(depends_on)
self.reverse_edges[depends_on].add(strategy_id)
def get_execution_order(self) -> List[str]:
"""คำนวณ Topological Sort สำหรับ Execution Order"""
in_degree = {node: 0 for node in self.nodes}
for deps in self.edges.values():
for dep in deps:
if dep in in_degree:
in_degree[dep] += 1
queue = [node for node, degree in in_degree.items() if degree == 0]
heapq.heapify(queue)
result = []
while queue:
node = heapq.heappop(queue)
result.append(node)
for dependent in self.reverse_edges[node]:
in_degree[dependent] -= 1
if in_degree[dependent] == 0:
heapq.heappush(queue, dependent)
return result
def get_affected_strategies(self, instrument_name: str) -> List[str]:
"""คืนรายการ Strategies ที่จะได้รับผลกระทบจากการเปลี่ยนแstrument"""
affected = []
for strategy_id, strategy in self.nodes.items():
if instrument_name in strategy.required_instruments:
affected.append(strategy_id)
return affected
class RecalculationTaskManager:
"""
Task Manager สำหรับ Greeks Recalculation
Features:
- Priority Queue with Deadlines
- Dependency-aware Scheduling
- Rate Limiting per Instrument
- Retry with Exponential Backoff
"""
def __init__(
self,
redis_client: redis.Redis,
max_concurrent: int = 50,
rate_limit_per_sec: int = 100
):
self.redis = redis_client
self.max_concurrent = max_concurrent
self.rate_limit = asyncio.Semaphore(rate_limit_per_sec)
self.active_tasks: Dict[str, asyncio.Task] = {}
self.task_queue: List[RecalcTask] = []
self.dependency_graph = DependencyGraph()
self.metrics = {
"total_tasks": 0,
"completed": 0,
"failed": 0,
"pending": 0,
"avg_latency_ms": 0.0
}
self._processing = False
def register_strategy(self, strategy: StrategyDependency):
"""Register Strategy และ Dependencies"""
self.dependency_graph.add_strategy(strategy)
# Auto-detect dependencies based on overlapping instruments
for other_id, other_strategy in self.dependency_graph.nodes.items():
if other_id != strategy.strategy_id:
overlap = strategy.required_instruments & other_strategy.required_instruments
if overlap:
# Higher priority strategy depends on lower priority
if strategy.priority.value < other_strategy.priority.value:
self.dependency_graph.add_dependency(
strategy.strategy_id,
other_id
)
def schedule_task(
self,
instrument_name: str,
task_type: str = "greeks",
priority: TaskPriority = TaskPriority.MEDIUM,
payload: Optional[Dict] = None
) -> str:
"""Schedule Recalculation Task"""
task_id = f"{instrument_name}_{task_type}_{datetime.utcnow().timestamp()}"
task = RecalcTask(
priority=priority,
task_id=task_id,
instrument_name=instrument_name,
task_type=task_type,
payload=payload or {}
)
heapq.heappush(self.task_queue, task)
self.metrics["total_tasks"] += 1
self.metrics["pending"] += 1
# Cache task for idempotency
asyncio.create_task(
self.redis.setex(
f"task:scheduled:{task_id}",
3600,
json.dumps({"priority": priority.value, "created": datetime.utcnow().isoformat()})
)
)
return task_id
async def _execute_task(self, task: RecalcTask, executor: Callable) -> bool:
"""Execute Single Task พร้อม Error Handling"""
start_time = asyncio.get_event_loop().time()
try:
async with self.rate_limit:
await executor(task)
# Update metrics
latency_ms = (asyncio.get_event_loop().time() - start_time) * 1000
self.metrics["completed"] += 1
self.metrics["pending"] -= 1
self.metrics["avg_latency_ms"] = (
(self.metrics["avg_latency_ms"] * (self.metrics["completed"] - 1) + latency_ms)
/ self.metrics["completed"]
)
# Update Strategy metrics
affected = self.dependency_graph.get_affected_strategies(task.instrument_name)
for strategy_id in affected:
strategy = self.dependency_graph.nodes[strategy_id]
strategy.last_recalc = datetime.utcnow()
strategy.recalc_count += 1
return True
except Exception as e:
self.metrics["failed"] += 1
self.metrics["pending"] -= 1
# Retry logic
if task.retry_count < task.max_retries:
task.retry_count += 1
backoff = 2 ** task.retry_count
task.scheduled_at = datetime.utcnow()
asyncio.create_task(
asyncio.sleep(backoff),
name=f"retry_{task.task_id}"
)
heapq.heappush(self.task_queue, task)
self.metrics["pending"] += 1
return False
async def process_queue(self, executor: Callable):
"""Process Task Queue Continuously"""
self._processing = True
while self._processing:
# Get next task
if not self.task_queue:
await asyncio.sleep(0.1)
continue
# Check concurrent limit
if len(self.active_tasks) >= self.max_concurrent:
await asyncio.sleep(0.1)
continue
task = heapq.heappop(self.task_queue)
task_coroutine = self._execute_task(task, executor)
self.active_tasks[task.task_id] = asyncio.create_task(task_coroutine)
# Clean up completed tasks
done = [tid for tid, t in self.active_tasks.items() if t.done()]
for tid in done:
del self.active_tasks[tid]
def get_metrics(self) -> Dict:
"""คืนค่า Metrics สำหรับ Dashboard"""
return {
**self.metrics,
"queue_size": len(self.task_queue),
"active_tasks": len(self.active_tasks),
"strategies": {
sid: {
"name": s.strategy_name,
"last_recalc": s.last_recalc.isoformat(),
"recalc_count": s.recalc_count,
"failure_count": s.failure_count
}
for sid, s in self.dependency_graph.nodes.items()
}
}
ตัวอย่างการใช้งาน
async def example_usage():
redis_client = await redis.from_url("redis://localhost:6379/0")
manager = RecalculationTaskManager(redis_client)
# Register Strategies
strategies = [
StrategyDependency(
strategy_id="strat_001",
strategy_name="Delta Hedging Bot",
required_instruments={"BTC-25-06-60000-P", "BTC-25-06-65000-C"},
calculation_interval_ms=100,
priority=TaskPriority.CRITICAL
),
StrategyDependency(
strategy_id="strat_002",
strategy_name="Gamma Scalping",
required_instruments={"BTC-25-06-60000-P", "BTC-25-06-62000-P"},
calculation_interval_ms=500,
priority=TaskPriority.HIGH
)
]
for strategy in strategies:
manager.register_strategy(strategy)
# Schedule Tasks
manager.schedule_task(
"BTC-25-06-60000-P",
task_type="greeks",
priority=TaskPriority.HIGH
)
# Process queue
async def task_executor(task: RecalcTask):
# Simulate calculation
await asyncio.sleep(0.05)
print(f"Executed: {task.task_id}")
await manager.process_queue(task_executor)
if __name__ == "__main__":
asyncio.run(example_usage())Integration กับ HolySheep AI สำหรับ Smart Analysis
ในการวิเคราะห์ Greeks Data และสร้าง Alert อัตโนมัติ เราสามารถใช้ HolySheep AI ซึ่งมีความเร็ว <50ms และราคาประหยัดถึง 85%+ เมื่อเทียบกับ OpenAI โดยใช้ base_url: https://api.holysheep.ai/v1
#!/usr/bin/env python3
"""
HolySheep AI Integration สำหรับ Deribit Greeks Analysis
ใช้ base_url: https://api.holysheep.ai/v1
"""
import httpx
import json
from typing import Dict, List, Optional
from datetime import datetime
from dataclasses import dataclass
HolySheep API Configuration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # ได้จาก https://www.holysheep.ai/register
@dataclass
class GreeksAnalysis:
"""ผลลัพธ์การวิเคราะห์ Greeks จาก AI"""
summary: str
alerts: List[Dict]
risk_score: float
recommendations: List[str]
confidence: float
class HolySheepAIClient:
"""Client สำหรับเชื่อมต่อ HolySheep AI API"""
def __init__(self, api_key: str = API_KEY):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(30.0),
limits=httpx.Limits(max_connections=100)
)
async def analyze_greeks_data(
self,
greeks_records: List[Dict],
analysis_type: str = "risk"
) -> GreeksAnalysis:
"""
วิเคราะห์ Greeks Data ด้วย HolySheep AI
Args:
greeks_records: รายการข้อมูล Greeks ที่ต้องการวิเคราะห์
analysis_type: "risk" | "opportunity" | "general"
"""
# Prepare context
avg_delta = sum(r.get("delta", 0) for r in greeks_records) / len(greeks_records)
avg_gamma = sum(r.get("gamma", 0) for r in greeks_records) / len(greeks_records)
avg_theta = sum(r.get("theta", 0) for r in greeks_records) / len(greeks_records)
avg_vega = sum(r.get("vega", 0) for r in greeks_records) / len(greeks_records)
prompt = f"""Analyze Deribit Options Greeks Data:
Summary Statistics:
- Average Delta: {avg_delta:.4f}
- Average Gamma: {avg_gamma:.6f}
- Average Theta: {avg_theta:.4f}
- Average Vega: {avg_vega:.4f}
- Total Records: {len(greeks_records)}
Instruments: {[r.get('instrument_name') for r in greeks_records[:10]]}
Provide:
1. Risk Assessment (0-100)
2. Key Alerts (if any)
3. Recommendations for traders
4. Confidence level (0-100)
"""
response = await self._call_model(
prompt=prompt,
system="You are a professional options trader and risk analyst. Provide concise, actionable insights."
)
return GreeksAnalysis(
summary=response.get("summary", ""),
alerts=response.get("alerts", []),
risk_score=response.get("risk_score", 50.0),
recommendations=response.get("recommendations", []),
confidence=response.get("confidence", 0.0)
)
async def generate_anomaly_alert(
self,
instrument: str,
current_greeks: Dict,
historical_avg: Dict,
threshold_multiplier: float = 2.0
) -> Optional[Dict]:
"""ตรวจจับ Anomaly ในข้อมูล Greeks และสร้าง Alert"""
anomalies = []
for key in ["delta", "gamma", "theta", "vega"]:
current = current_greeks.get(key, 0)
hist = historical_avg.get(key, 0)
if hist != 0:
change_pct = abs((current - hist) / hist) * 100
if change_pct > threshold_multiplier * 100:
anomalies.append({
"metric": key,
"current": current,
"historical_avg": hist,
"change_percent": change_pct
})
if anomalies:
prompt = f"""Anomaly detected in {instrument} Greeks Data:
Anomalies Found:
{json.dumps(anomalies, indent=2)}
Current Market Context:
- Underlying Price: {current_greeks.get('underlying_price')}
- Mark Price: {current_greeks.get('mark_price')}
- Open Interest: {current_greeks.get('open_interest')}
Generate an alert message that includes:
1. Severity level (LOW/MEDIUM/HIGH/CRITICAL)
2. Probable cause
3. Recommended action
"""
response = await self._call_model(
prompt=prompt,
system="You are a risk monitoring system. Generate clear, actionable alerts."
)
return {
"instrument": instrument,
"anomalies": anomalies,
"alert_message": response.get("alert_message", "Anomaly detected"),
"severity": response.get("severity", "MEDIUM"),
"timestamp": datetime.utcnow().isoformat()
}
return None
async def generate_research_report(
self,
greeks_data: List[Dict],
team_satisfaction_metrics: Dict
) -> str:
"""สร้าง Research Report อัตโนมัติสำหรับทีมวิจัย"""
prompt = f"""Generate a comprehensive research report based on:
Greeks Data Summary:
- Total Instruments: {len(greeks_data)}
- Time Range: {greeks_data[0].get('timestamp')} to {greeks_data[-1].get('timestamp')}
Team Satisfaction Metrics:
- Data Quality Score: {team_satisfaction_metrics.get('data_quality_score',