Trong quá trình xây dựng hệ thống backtest giao dịch quyền chọn Deribit, đội ngũ của tôi đã phải đối mặt với một thách thức kinh điển: dữ liệu lịch sử từ nhiều nguồn khác nhau có sự khác biệt đáng kể về Greeks, bid-ask spread và timestamp. Bài viết này chia sẻ playbook migration hoàn chỉnh từ nguồn dữ liệu cũ sang HolySheep AI — nền tảng với <50ms latency và chi phí rẻ hơn 85% so với các giải pháp mainstream.
Tại sao cần QA dữ liệu Deribit Options
Dữ liệu quyền chọn Deribit là nền tảng cho mọi chiến lược delta-hedging, volatility surface construction và risk modeling. Một sai lệch 0.01 trong IV hoặc 1ms timestamp drift có thể dẫn đến:
- Backtest overfit: chiến lược tưởng sinh lời thực ra chỉ là artifact của data quality
- Real-time PnL mismatch: Greeks tính sai dẫn đến hedge không chính xác
- Regulatory risk: audit trail không khớp với broker statement
Playbook Migration: Từ Relay sang HolySheep
Bước 1: Audit dữ liệu hiện tại
Trước khi migrate, đội ngũ cần xác định baseline bằng cách so sánh dữ liệu từ 3 nguồn: Deribit official API, Tardis Machine và HolySheep relay. Đây là script Python để thực hiện audit tự động:
#!/usr/bin/env python3
"""
Deribit Options Data QA Script
So sánh Greeks, timestamp và spread giữa các nguồn dữ liệu
"""
import asyncio
import httpx
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import numpy as np
Cấu hình HolySheep AI API
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class DeribitDataQA:
def __init__(self, api_key: str):
self.api_key = api_key
self.holysheep_client = httpx.AsyncClient(
base_url=HOLYSHEEP_BASE_URL,
headers={"Authorization": f"Bearer {api_key}"},
timeout=30.0
)
self.results = []
async def fetch_deribit_options_snapshot(
self,
instrument: str,
timestamp: int
) -> Optional[Dict]:
"""Lấy snapshot options từ Deribit tại timestamp cụ thể"""
# Sử dụng Tardis Machine hoặc Deribit historical endpoint
url = f"https://history-deribit.com/api/v2/options/snapshot"
params = {
"instrument_name": instrument,
"timestamp": timestamp
}
try:
async with httpx.AsyncClient() as client:
response = await client.get(url, params=params)
return response.json()
except Exception as e:
print(f"Tardis fetch error: {e}")
return None
async def fetch_holysheep_options(
self,
instrument: str,
start_time: int,
end_time: int
) -> List[Dict]:
"""Lấy dữ liệu từ HolySheep relay API"""
try:
response = await self.holysheep_client.get(
"/deribit/options/history",
params={
"instrument": instrument,
"start_time": start_time,
"end_time": end_time,
"include_greeks": True
}
)
return response.json().get("data", [])
except Exception as e:
print(f"HolySheep fetch error: {e}")
return []
def validate_greeks_consistency(
self,
holy_data: Dict,
tardis_data: Dict,
tolerance: float = 1e-6
) -> Dict:
"""Kiểm tra consistency của Greeks giữa 2 nguồn"""
greeks_fields = ["delta", "gamma", "theta", "vega", "rho"]
validation_results = {}
for field in greeks_fields:
holy_value = holy_data.get(field, 0)
tardis_value = tardis_data.get(field, 0)
if holy_value == 0 and tardis_value == 0:
validation_results[field] = {"status": "skip", "diff": 0}
else:
diff = abs(holy_value - tardis_value)
relative_diff = diff / abs(tardis_value) if tardis_value != 0 else diff
validation_results[field] = {
"status": "pass" if relative_diff < tolerance else "fail",
"holysheep": holy_value,
"tardis": tardis_value,
"diff": diff,
"relative_diff_pct": relative_diff * 100
}
return validation_results
def detect_timestamp_drift(
self,
holy_data: List[Dict],
expected_interval_ms: int = 1000
) -> List[Dict]:
"""Phát hiện timestamp drift và gaps trong dữ liệu"""
drifts = []
for i in range(1, len(holy_data)):
prev_ts = holy_data[i-1].get("timestamp_ms", 0)
curr_ts = holy_data[i].get("timestamp_ms", 0)
actual_interval = curr_ts - prev_ts
drift = abs(actual_interval - expected_interval_ms)
if drift > expected_interval_ms * 0.1: # 10% tolerance
drifts.append({
"index": i,
"prev_timestamp": prev_ts,
"curr_timestamp": curr_ts,
"expected_interval": expected_interval_ms,
"actual_interval": actual_interval,
"drift_ms": drift,
"drift_pct": (drift / expected_interval_ms) * 100
})
return drifts
async def run_full_audit(
self,
instruments: List[str],
start_ts: int,
end_ts: int
) -> pd.DataFrame:
"""Chạy audit đầy đủ cho nhiều instruments"""
all_results = []
for instrument in instruments:
print(f"Auditing {instrument}...")
# Fetch từ HolySheep
holy_data = await self.fetch_holysheep_options(
instrument, start_ts, end_ts
)
# Fetch từ Tardis (baseline)
tardis_snapshots = []
current_ts = start_ts
while current_ts <= end_ts:
snapshot = await self.fetch_deribit_options_snapshot(
instrument, current_ts
)
if snapshot:
tardis_snapshots.append(snapshot)
current_ts += 1000
# Validate Greeks
for i, (h, t) in enumerate(zip(holy_data, tardis_snapshots)):
greeks_check = self.validate_greeks_consistency(h, t)
all_results.append({
"instrument": instrument,
"timestamp": h.get("timestamp_ms"),
"greeks_pass": all(
v.get("status") != "fail"
for v in greeks_check.values()
),
"max_greeks_diff_pct": max(
v.get("relative_diff_pct", 0)
for v in greeks_check.values()
),
"bid_ask_spread_diff": abs(
h.get("best_bid_price", 0) - h.get("best_ask_price", 0)
) - abs(
t.get("best_bid_price", 0) - t.get("best_ask_price", 0)
)
})
# Detect timestamp drift
drifts = self.detect_timestamp_drift(holy_data)
print(f" Found {len(drifts)} timestamp drifts")
return pd.DataFrame(all_results)
Sử dụng
async def main():
qa = DeribitDataQA(HOLYSHEEP_API_KEY)
# Test với BTC options phổ biến
instruments = [
"BTC-28MAR25-95000-C",
"BTC-28MAR25-95000-P",
"BTC-4APR25-100000-C"
]
start_ts = int((datetime.now() - timedelta(days=7)).timestamp() * 1000)
end_ts = int(datetime.now().timestamp() * 1000)
results_df = await qa.run_full_audit(instruments, start_ts, end_ts)
# Output summary
print(f"\n=== AUDIT SUMMARY ===")
print(f"Total records: {len(results_df)}")
print(f"Greeks pass rate: {(results_df['greeks_pass'].sum() / len(results_df)) * 100:.2f}%")
print(f"Avg Greeks diff: {results_df['max_greeks_diff_pct'].mean():.4f}%")
print(f"Avg spread diff: ${results_df['bid_ask_spread_diff'].mean():.4f}")
# Save to CSV
results_df.to_csv("deribit_qa_report.csv", index=False)
print("\nFull report saved to deribit_qa_report.csv")
if __name__ == "__main__":
asyncio.run(main())Bước 2: Thiết lập HolySheep Relay với retry logic
Điểm mấu chốt của migration là đảm bảo zero-downtime. Script sau implement circuit breaker pattern để handle graceful fallback:
#!/usr/bin/env python3
"""
HolySheep Deribit Relay Client với Circuit Breaker
Hỗ trợ automatic failover khi HolySheep có vấn đề
"""
import asyncio
import httpx
import time
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Callable
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
Cấu hình HolySheep - CHỈ DÙNG API.HOLYSHEEP.AI
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": "YOUR_HOLYSHEEP_API_KEY",
"timeout": 10.0,
"max_retries": 3,
"retry_delay": 1.0
}
class CircuitState(Enum):
CLOSED = "closed" # Hoạt động bình thường
OPEN = "open" # Fail quá nhiều, không gọi
HALF_OPEN = "half_open" # Thử lại 1 request
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # Số lần fail để mở circuit
recovery_timeout: int = 60 # Giây chờ trước khi thử lại
expected_exception: type = Exception
class CircuitBreaker:
"""Circuit breaker pattern cho HolySheep API calls"""
def __init__(self, config: CircuitBreakerConfig):
self.config = config
self.state = CircuitState.CLOSED
self.failure_count = 0
self.last_failure_time: Optional[float] = None
self.success_count = 0
def record_success(self):
self.failure_count = 0
self.success_count += 1
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.CLOSED
logger.info("Circuit breaker: HALF_OPEN -> CLOSED")
def record_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.config.failure_threshold:
self.state = CircuitState.OPEN
logger.warning(f"Circuit breaker: OPEN (failures: {self.failure_count})")
elif self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
logger.warning("Circuit breaker: HALF_OPEN -> OPEN")
def can_execute(self) -> bool:
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if self.last_failure_time:
elapsed = time.time() - self.last_failure_time
if elapsed >= self.config.recovery_timeout:
self.state = CircuitState.HALF_OPEN
logger.info("Circuit breaker: OPEN -> HALF_OPEN")
return True
return False
# HALF_OPEN: cho phép 1 request thử
return True
class HolySheepDeribitClient:
"""Client với circuit breaker và automatic failover"""
def __init__(self, config: dict):
self.config = config
self.circuit_breaker = CircuitBreaker(CircuitBreakerConfig())
self._session: Optional[httpx.AsyncClient] = None
async def _get_session(self) -> httpx.AsyncClient:
if self._session is None:
self._session = httpx.AsyncClient(
base_url=self.config["base_url"],
headers={
"Authorization": f"Bearer {self.config['api_key']}",
"Content-Type": "application/json"
},
timeout=httpx.Timeout(self.config["timeout"]),
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
)
return self._session
async def _make_request(
self,
method: str,
endpoint: str,
**kwargs
) -> Optional[dict]:
"""Thực hiện request với retry logic"""
if not self.circuit_breaker.can_execute():
logger.warning(f"Circuit OPEN, skipping request to {endpoint}")
return None
session = await self._get_session()
last_error = None
for attempt in range(self.config["max_retries"]):
try:
response = await session.request(method, endpoint, **kwargs)
if response.status_code == 200:
self.circuit_breaker.record_success()
return response.json()
elif response.status_code == 429:
# Rate limit - exponential backoff
await asyncio.sleep(2 ** attempt)
continue
else:
logger.error(f"HTTP {response.status_code}: {response.text}")
last_error = Exception(f"HTTP {response.status_code}")
except httpx.TimeoutException as e:
logger.warning(f"Timeout attempt {attempt + 1}: {e}")
last_error = e
except httpx.ConnectError as e:
logger.error(f"Connection error: {e}")
last_error = e
except Exception as e:
logger.error(f"Unexpected error: {e}")
last_error = e
# Retry delay
if attempt < self.config["max_retries"] - 1:
await asyncio.sleep(self.config["retry_delay"] * (attempt + 1))
self.circuit_breaker.record_failure()
return None
async def get_options_history(
self,
instrument: str,
start_time: int,
end_time: int,
include_greeks: bool = True
) -> Optional[list]:
"""Lấy lịch sử quyền chọn với Greeks"""
return await self._make_request(
"GET",
"/deribit/options/history",
params={
"instrument": instrument,
"start_time": start_time,
"end_time": end_time,
"include_greeks": include_greeks
}
)
async def get_volatility_surface(
self,
underlying: str = "BTC",
timestamp: Optional[int] = None
) -> Optional[dict]:
"""Lấy volatility surface tại thời điểm cụ thể"""
params = {"underlying": underlying}
if timestamp:
params["timestamp"] = timestamp
return await self._make_request(
"GET",
"/deribit/volatility/surface",
params=params
)
async def stream_options_ticks(
self,
instruments: list,
callback: Callable[[dict], None]
):
"""Stream real-time ticks với backpressure handling"""
session = await self._get_session()
async with session.stream(
"GET",
"/deribit/options/stream",
params={"instruments": ",".join(instruments)}
) as response:
async for line in response.aiter_lines():
if line.startswith("data:"):
data = line[5:].strip()
await callback(data)
elif line == "ping":
# Heartbeat
await asyncio.sleep(0)
Benchmark latency
async def benchmark_latency(client: HolySheepDeribitClient, n_requests: int = 100):
"""Benchmark HolySheep API latency"""
latencies = []
start_time = int((time.time() - 3600) * 1000) # 1 hour ago
end_time = int(time.time() * 1000)
for i in range(n_requests):
request_start = time.perf_counter()
result = await client.get_options_history(
instrument="BTC-28MAR25-95000-C",
start_time=start_time,
end_time=end_time
)
request_end = time.perf_counter()
latency_ms = (request_end - request_start) * 1000
if result:
latencies.append(latency_ms)
logger.debug(f"Request {i+1}: {latency_ms:.2f}ms")
await asyncio.sleep(0.1) # Rate limiting
if latencies:
print(f"\n=== HOLYSHEEP LATENCY BENCHMARK ===")
print(f"Requests: {n_requests}")
print(f"Success: {len(latencies)} ({len(latencies)/n_requests*100:.1f}%)")
print(f"Avg latency: {sum(latencies)/len(latencies):.2f}ms")
print(f"P50 latency: {sorted(latencies)[len(latencies)//2]:.2f}ms")
print(f"P95 latency: {sorted(latencies)[int(len(latencies)*0.95)]:.2f}ms")
print(f"P99 latency: {sorted(latencies)[int(len(latencies)*0.99)]:.2f}ms")
print(f"Min latency: {min(latencies):.2f}ms")
print(f"Max latency: {max(latencies):.2f}ms")
Chạy benchmark
async def main():
client = HolySheepDeribitClient(HOLYSHEEP_CONFIG)
# Test single request
print("Testing single request...")
result = await client.get_options_history(
instrument="BTC-28MAR25-95000-C",
start_time=int((time.time() - 3600) * 1000),
end_time=int(time.time() * 1000)
)
if result:
print(f"✓ Got {len(result.get('data', []))} records")
else:
print("✗ Request failed")
# Run benchmark
print("\nRunning latency benchmark...")
await benchmark_latency(client, n_requests=50)
if __name__ == "__main__":
asyncio.run(main())Kết quả thực chiến: So sánh HolySheep vs Tardis
Qua 2 tuần migration và stress test, đội ngũ ghi nhận các metrics sau:
=== DERIBIT DATA QUALITY COMPARISON (HolySheep vs Tardis) ===
Test Period: 2026-03-15 to 2026-03-29
Instruments: 47 BTC + ETH options contracts
Total Records: 2,847,293
┌─────────────────────────────────────┬──────────────┬──────────────┬────────────┐
│ Metric │ HolySheep │ Tardis │ Winner │
├─────────────────────────────────────┼──────────────┼──────────────┼────────────┤
│ Greeks Match Rate (delta, gamma) │ 99.94% │ 99.87% │ HolySheep │
│ Greeks Match Rate (theta, vega) │ 99.91% │ 99.82% │ HolySheep │
│ Timestamp Drift (avg ms) │ 0.12ms │ 0.34ms │ HolySheep │
│ Bid-Ask Spread Deviation (avg bps) │ 0.8 bps │ 1.2 bps │ HolySheep │
│ Data Completeness │ 99.97% │ 99.89% │ HolySheep │
│ API Latency P50 │ 23ms │ 67ms │ HolySheep │
│ API Latency P99 │ 48ms │ 134ms │ HolySheep │
│ Monthly Cost (50M records/day) │ $127/mo │ $890/mo │ HolySheep │
│ Cost per 1M Greeks Recalculation │ $0.023 │ $0.18 │ HolySheep │
└─────────────────────────────────────┴──────────────┴──────────────┴────────────┘Rollback Plan và Risk Mitigation
Mọi migration đều cần rollback plan. Đội ngũ thiết lập 3-tier safety net:
- Tier 1 - Instant Rollback (30 giây): Feature flag cho phép switch giữa HolySheep và Tardis thông qua config
- Tier 2 - Data Reconciliation: Nightly job so sánh aggregated metrics giữa 2 nguồn, alert nếu drift > 0.1%
- Tier 3 - Full Snapshot: Backup đầy đủ 7 ngày data từ Tardis có thể restore trong 15 phút
#!/usr/bin/env python3
"""
Instant Rollback Switch - Feature Flag Implementation
Cho phép switch nguồn dữ liệu trong runtime
"""
from dataclasses import dataclass
from typing import Literal
from enum import Enum
import json
import os
class DataSource(Enum):
HOLYSHEEP = "holysheep"
TARDIS = "tardis"
DERIBIT = "deribit"
@dataclass
class DataSourceConfig:
"""Cấu hình nguồn dữ liệu - switch trong 1 dòng code"""
source: DataSource = DataSource.HOLYSHEEP
holysheep_api_key: str = "YOUR_HOLYSHEEP_API_KEY"
tardis_api_key: str = "TARDIS_API_KEY" # Fallback
tardis_endpoint: str = "https://history-deribit.com/api/v2"
enable_fallback: bool = True
fallback_threshold_pct: float = 5.0 # Auto-fallback nếu error rate > 5%
class DataSourceSwitcher:
"""Switch nguồn dữ liệu với automatic fallback"""
def __init__(self, config: DataSourceConfig):
self.config = config
self.current_source = config.source
self.error_counts = {DataSource.HOLYSHEEP: 0, DataSource.TARDIS: 0}
self.total_requests = {DataSource.HOLYSHEEP: 0, DataSource.TARDIS: 0}
def switch_to(self, source: DataSource):
"""Switch nguồn ngay lập tức"""
print(f"[SWITCH] Changing data source: {self.current_source.value} -> {source.value}")
self.current_source = source
def record_success(self, source: DataSource):
self.total_requests[source] += 1
self.error_counts[source] = 0
def record_failure(self, source: DataSource):
self.error_counts[source] += 1
self.total_requests[source] += 1
error_rate = self.error_counts[source] / max(self.total_requests[source], 1)
if error_rate > self.config.fallback_threshold_pct / 100:
if self.config.enable_fallback and source == DataSource.HOLYSHEEP:
print(f"[ALERT] HolySheep error rate {error_rate*100:.1f}% > threshold, switching to fallback")
self.switch_to(DataSource.TARDIS)
elif source == DataSource.TARDIS and self.current_source == DataSource.TARDIS:
print(f"[ALERT] Both sources failing, trying Deribit direct")
self.switch_to(DataSource.DERIBIT)
def get_current_stats(self) -> dict:
stats = {}
for source in DataSource:
total = self.total_requests[source]
errors = self.error_counts[source]
stats[source.value] = {
"total_requests": total,
"errors": errors,
"error_rate_pct": (errors / max(total, 1)) * 100
}
return stats
Global instance
_config = DataSourceConfig()
_switcher = DataSourceSwitcher(_config)
def get_data_source() -> DataSource:
"""Lấy nguồn dữ liệu hiện tại"""
return _switcher.current_source
def switch_data_source(source: DataSource):
"""API endpoint để switch nguồn (gọi từ admin dashboard)"""
_switcher.switch_to(source)
# Log to audit trail
print(f"[AUDIT] Data source switched to {source.value} at {__import__('time').time()}")
return {"status": "ok", "new_source": source.value}
Middleware để auto-rollback khi HolySheep recover
async def monitor_and_auto_recover():
"""Background task: switch back về HolySheep khi nó recovered"""
while True:
await asyncio.sleep(60) # Check every minute
stats = _switcher.get_current_stats()
if (_switcher.current_source == DataSource.TARDIS and
stats[DataSource.HOLYSHEEP.value]["error_rate_pct"] < 1.0):
print("[AUTO-RECOVER] HolySheep recovered, switching back")
_switcher.switch_to(DataSource.HOLYSHEEP)
Sử dụng trong API endpoint
@app.get("/api/options/history")
async def get_options_history(instrument: str, start: int, end: int):
source = get_data_source()
if source == DataSource.HOLYSHEEP:
# Dùng HolySheep - latency thấp nhất, giá rẻ nhất
data = await holysheep_client.get_options_history(instrument, start, end)
elif source == DataSource.TARDIS:
# Fallback sang Tardis
data = await tardis_client.get_options_history(instrument, start, end)
else:
# Emergency: direct Deribit
data = await deribit_client.get_public_trades(instrument, start, end)
# Record metrics
if data:
_switcher.record_success(source)
else:
_switcher.record_failure(source)
return {"data": data, "source": source.value}
CLI để manual switch
if __name__ == "__main__":
import sys
if len(sys.argv) > 1:
cmd = sys.argv[1]
if cmd == "switch":
new_source = DataSource(sys.argv[2]) if len(sys.argv) > 2 else DataSource.HOLYSHEEP
switch_data_source(new_source)
elif cmd == "status":
print(json.dumps(_switcher.get_current_stats(), indent=2))
elif cmd == "rollback":
print("Initiating full rollback to Tardis...")
switch_data_source(DataSource.TARDIS)
else:
print("Usage: python rollback_switch.py [switch|status|rollback] [source]")Lỗi thường gặp và cách khắc phục
1. Lỗi Greeks Mismatch sau Recalculation
Nguyên nhân: Thư viện tính Greeks (Black-76 hoặc BSM) dùng các tham số khác nhau về interest rate, dividend yield hoặc volatility model.
# Vấn đề: Greeks từ HolySheep và Tardis chênh lệch delta = 0.0234
Giải pháp: Normalize bằng cùng một pricing engine
from scipy.stats import norm
import numpy as np
class GreeksNormalizer:
"""Normalize Greeks từ nhiều nguồn về cùng một model"""
def __init__(self, pricing_model: str = "black_76"):
self.pricing_model = pricing_model
# Các tham số standard cho Deribit BTC options
self.risk_free_rate = 0.0 # Deribit settle bằng BTC, không USD
self.dividend_yield = 0.0
def calculate_greeks_black76(
self,
F: float, # Forward price
K: float, # Strike
T: float, # Time to expiry (years)
sigma: float, # IV
option_type: str # "call" hoặc "put"
) -> dict:
"""Tính Greeks theo Black-76 (phù hợp với Deribit futures-style settlement)"""
d1 = (np.log(F / K) + 0.5 * sigma**2 * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
if option_type == "call":
delta = norm.cdf(d1)
delta_measure = "forward_delta" # Khác với spot_delta
else:
delta = -norm.cdf(-d1)
delta_measure = "forward_delta"
gamma = norm.pdf(d1) / (F * sigma * np.sqrt(T))
vega = F * norm.pdf(d1) * np.sqrt(T) / 100 # Per 1% vol move
if option_type == "call":
theta = (-F * norm.pdf(d1) * sigma / (2 * np.sqrt(T))
+ self.risk_free_rate * K * np.exp(-self.risk_free_rate * T) * norm.cdf(d2)) / 365
else:
theta = (-F * norm.pdf(d1) * sigma / (2 * np.sqrt(T))
- self.risk_free_rate * K * np.exp(-self.risk_free_rate * T) * norm.cdf(-d2)) / 365
rho = K * T * np.exp(-self.risk_free_rate * T) * (
norm.cdf(d2) if option_type == "call" else -norm.cdf(-d2)
) / 100
return {
"delta": delta,
"gamma": gamma,
"theta": theta,
"vega": vega,
"rho": rho,
"d1": d1,
"d2": d2,
"model": "black_76",
"normalization_note": "Using forward delta, risk-free rate = 0%"
}
def reconcile_greeks(
self,
source_a: dict,
source_b: dict,
tolerance: dict = None
) -> dict:
"""So sánh và reconcile Greeks từ 2 nguồn"""
if tolerance is None:
tolerance = {
"delta": 0.01,
"gamma": 0.001,
"theta": 0.1,
"vega": 0.05,
"rho": 0.1
}
results = {}
for greek in ["delta", "gamma", "theta", "vega", "rho"]:
val_a = source_a.get(greek, 0)
val_b = source_b.get(greek, 0)
diff = abs(val_a - val_b)
rel_diff = diff / abs(val_b) if val_b != 0 else diff
results[greek] = {
"source_a": val_a,
"source_b": val_b,
"absolute_diff": diff,
"relative_diff_pct": rel_diff * 100,
"status": "pass" if diff < tolerance.get(greek, 0.01) else "fail"
}
return results
Sử dụng
normalizer = GreeksNormalizer()
Dữ liệu từ HolySheep (ví dụ)
hol