저는 HolySheep AI의 기술팀에서 프로덕션 레벨 AI 파이프라인 구축을 담당하고 있습니다. 이번 튜토리얼에서는 HolySheep AI를 활용해 Tardis API에서 OKX 옵션 Greeks 데이터를 수집하고, 역사적 아카이빙을 구성하며, 변동성 곡면(Volatility Surface)을 구축하는 전체 아키텍처를 다루겠습니다.
1. 아키텍처 개요
OKX 옵션 마켓 데이터는 전통 금융보다 높은 변동성과 24/7 거래 특성으로 인해 실시간 처리와 배치 아카이빙 모두에서 특별한 설계가 필요합니다. 전체 파이프라인은 세 개의 핵심 레이어로 구성됩니다.
- 데이터 수집 레이어: Tardis API에서 OKX 옵션 Greeks 실시간 스트리밍
- AI 처리 레이어: HolySheep AI API로 Greeks 데이터 정제, 이상치 탐지, 예측 분석
- 스토리지 레이어: TimescaleDB 기반 시계열 아카이빙 + 변동성 곡면 시각화
2. 환경 구성
# requirements.txt
================
holysheep>=0.1.8
tardis-client>=2.0.0
asyncpg>=0.29.0
pandas>=2.1.0
numpy>=1.26.0
pydantic>=2.5.0
httpx>=0.26.0
websockets>=12.0
plotly>=5.18.0
ta>=0.10.2
설치 명령
pip install -r requirements.txt# config.py
=========
import os
from dataclasses import dataclass
@dataclass
class HolySheepConfig:
"""HolySheep AI API 설정"""
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
model: str = "gpt-4.1"
max_tokens: int = 4096
temperature: float = 0.3
@dataclass
class TardisConfig:
"""Tardis OKX API 설정"""
api_key: str = os.getenv("TARDIS_API_KEY", "YOUR_TARDIS_API_KEY")
exchange: str = "okx"
channel: str = "options_greeks"
inst_type: str = "OPTION"
@dataclass
class DatabaseConfig:
"""TimescaleDB 설정"""
host: str = os.getenv("TIMESCALE_HOST", "localhost")
port: int = 5432
database: str = "options_data"
user: str = os.getenv("DB_USER", "postgres")
password: str = os.getenv("DB_PASSWORD", "")
HolySheep API 클라이언트 인스턴스 생성
HOLYSHEEP = HolySheepConfig()
검증
assert HOLYSHEEP.base_url == "https://api.holysheep.ai/v1", \
"HolySheep base_url이 올바르지 않습니다"3. HolySheep AI 통합: Greeks 데이터 분석기
옵션 Greeks 데이터는 Delta, Gamma, Vega, Theta, Rho 등 다차원 지표를 포함합니다. HolySheep AI를 활용하면 이 복잡한 데이터를 자연어로 분석하고, 이상치를 자동 탐지하며, 변동성 예측 모델을 생성할 수 있습니다.
# holysheep_client.py
===================
import httpx
import json
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from config import HOLYSHEEP
@dataclass
class GreeksAnalysis:
"""Greeks 분석 결과"""
delta: float
gamma: float
vega: float
theta: float
rho: float
implied_volatility: float
anomaly_score: Optional[float] = None
analysis_text: Optional[str] = None
class HolySheepGreeksClient:
"""HolySheep AI API를 활용한 Greeks 데이터 분석 클라이언트"""
def __init__(self, config: HolySheepConfig):
self.base_url = config.base_url
self.api_key = config.api_key
self.model = config.model
self.max_tokens = config.max_tokens
self.temperature = config.temperature
def _build_headers(self) -> Dict[str, str]:
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async def analyze_greeks_anomaly(
self,
greeks_data: Dict[str, float],
market_context: str = ""
) -> GreeksAnalysis:
"""
Greeks 데이터에서 이상치 탐지 및 분석 수행
Args:
greeks_data: Greeks 지표 딕셔너리
market_context: 시장 상황 설명
Returns:
분석 결과가 포함된 GreeksAnalysis 객체
"""
prompt = f"""OKX 옵션 Greeks 데이터를 분석하고 이상치를 탐지하세요.
현재 Greeks 데이터:
- Delta (δ): {greeks_data.get('delta', 0):.4f}
- Gamma (γ): {greeks_data.get('gamma', 0):.6f}
- Vega (ν): {greeks_data.get('vega', 0):.4f}
- Theta (θ): {greeks_data.get('theta', 0):.4f}
- Rho (ρ): {greeks_data.get('rho', 0):.4f}
- 내재 변동성 (IV): {greeks_data.get('implied_volatility', 0):.2%}
시장 컨텍스트: {market_context}
다음 형식으로 JSON 응답:
{{
"anomaly_score": 0.0~1.0 (이상치 확률),
"analysis": "간결한 분석 텍스트 (50자 이내)",
"recommendations": ["권장 조치사항 1", "권장 조치사항 2"]
}}"""
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=self._build_headers(),
json={
"model": self.model,
"messages": [
{"role": "system", "content": "당신은 전문 옵션 퀀트 애널리스트입니다."},
{"role": "user", "content": prompt}
],
"max_tokens": self.max_tokens,
"temperature": self.temperature
}
)
if response.status_code != 200:
raise Exception(f"API 오류: {response.status_code} - {response.text}")
result = response.json()
content = result["choices"][0]["message"]["content"]
# JSON 파싱 (실제 구현에서는 더 강력한 파싱 필요)
try:
analysis = json.loads(content)
return GreeksAnalysis(
delta=greeks_data.get('delta', 0),
gamma=greeks_data.get('gamma', 0),
vega=greeks_data.get('vega', 0),
theta=greeks_data.get('theta', 0),
rho=greeks_data.get('rho', 0),
implied_volatility=greeks_data.get('implied_volatility', 0),
anomaly_score=analysis.get("anomaly_score", 0),
analysis_text=analysis.get("analysis", "")
)
except json.JSONDecodeError:
# Fallback: 단순 분석 반환
return GreeksAnalysis(
**greeks_data,
anomaly_score=0.5,
analysis_text="분석 파싱 실패"
)
async def generate_volatility_forecast(
self,
historical_iv_data: List[Dict[str, Any]],
strike: float,
expiry: str
) -> Dict[str, Any]:
"""
역사적 IV 데이터를 기반으로 변동성 예측 생성
Returns:
예측 결과 딕셔너리 (예측 IV, 신뢰구간, 분석)
"""
iv_series = [d['iv'] for d in historical_iv_data[-20:]] # 최근 20개 데이터
timestamps = [d['timestamp'] for d in historical_iv_data[-20:]]
prompt = f"""OKX 옵션 변동성 시계열 데이터를 분석하여 단기 예측을 수행하세요.
Striker Price: ${strike}
만기: {expiry}
최근 IV 시계열: {iv_series}
타임스탬프: {timestamps}
다음 JSON 형식으로 응답:
{{
"predicted_iv": 0.0~1.0 (예측 IV),
"confidence_lower": 0.0~1.0 (신뢰구간 하한),
"confidence_upper": 0.0~1.0 (신뢰구간 상한),
"trend": "bullish|bearish|neutral",
"key_factors": ["영향 요인 1", "영향 요인 2"]
}}"""
async with httpx.AsyncClient(timeout=45.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=self._build_headers(),
json={
"model": self.model,
"messages": [
{"role": "system", "content": "당신은 변동성 거래 전문가입니다."},
{"role": "user", "content": prompt}
],
"max_tokens": self.max_tokens,
"temperature": 0.2
}
)
return response.json()["choices"][0]["message"]["content"]
클라이언트 인스턴스 생성
holysheep_client = HolySheepGreeksClient(HOLYSHEEP)4. Tardis OKX Greeks 실시간 수집기
# tardis_collector.py
====================
import asyncio
import json
import logging
from datetime import datetime, timezone
from typing import Dict, Any, Optional, Callable, List
import httpx
from config import HOLYSHEEP, TardisConfig
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class TardisOKXGreeksCollector:
"""
Tardis API에서 OKX 옵션 Greeks 실시간 데이터 수집
HolySheep AI와 연계하여 실시간 이상치 탐지 및 분석 수행
"""
def __init__(self, tardis_config: TardisConfig, holysheep_client):
self.api_key = tardis_config.api_key
self.exchange = tardis_config.exchange
self.channel = tardis_config.channel
self.inst_type = tardis_config.inst_type
self.holysheep = holysheep_client
self.base_url = "https://api.tardis.dev/v1"
self._running = False
self._buffer: List[Dict[str, Any]] = []
async def _fetch_historical(
self,
start_date: str,
end_date: str,
symbols: Optional[List[str]] = None
) -> List[Dict[str, Any]]:
"""역사적 Greeks 데이터 배치 수집"""
params = {
"exchange": self.exchange,
"channel": self.channel,
"dateFrom": start_date,
"dateTo": end_date,
"format": "json"
}
if symbols:
params["symbols"] = ",".join(symbols)
headers = {"Authorization": f"Bearer {self.api_key}"}
async with httpx.AsyncClient(timeout=300.0) as client:
response = await client.get(
f"{self.base_url}/historical",
params=params,
headers=headers
)
if response.status_code == 200:
data = response.json()
logger.info(f"역사적 데이터 {len(data)}건 수신 완료")
return data
else:
raise Exception(f"Historical fetch 실패: {response.status_code}")
async def stream_greeks(
self,
symbols: List[str],
on_data: Optional[Callable] = None,
buffer_size: int = 100
):
"""
실시간 Greeks 스트림 구독
Args:
symbols: 구독할 옵션 심볼 리스트
on_data: 데이터 수신 시 콜백 함수
buffer_size: HolySheep 배치 처리를 위한 버퍼 크기
"""
self._running = True
ws_url = f"wss://api.tardis.dev/v1/stream"
headers = {"Authorization": f"Bearer {self.api_key}"}
async with httpx.AsyncClient() as client:
async with client.ws_connect(ws_url, headers=headers) as ws:
# 구독 메시지 전송
subscribe_msg = {
"action": "subscribe",
"exchange": self.exchange,
"channel": self.channel,
"symbols": symbols
}
await ws.send_json(subscribe_msg)
logger.info(f"구독 시작: {symbols}")
while self._running:
try:
msg = await asyncio.wait_for(ws.receive_json(), timeout=30.0)
if msg.get("type") == "data":
greeks_record = self._parse_greeks(msg["data"])
self._buffer.append(greeks_record)
# 버퍼가 채워지면 HolySheep로 분석
if len(self._buffer) >= buffer_size:
await self._process_buffer()
# 콜백 실행
if on_data:
await on_data(greeks_record)
except asyncio.TimeoutError:
# 하트비트: 연결 유지
logger.debug("하트비트 체크")
except Exception as e:
logger.error(f"스트림 오류: {e}")
await asyncio.sleep(5) # 재연결 대기
def _parse_greeks(self, raw_data: Dict[str, Any]) -> Dict[str, Any]:
"""Tardis 데이터 파싱 및 정규화"""
return {
"timestamp": datetime.now(timezone.utc).isoformat(),
"symbol": raw_data.get("instId", ""),
"underlying": raw_data.get("uly", ""),
"strike": float(raw_data.get("strike", 0)),
"expiry": raw_data.get("exp", ""),
"delta": float(raw_data.get("delta", 0)),
"gamma": float(raw_data.get("gamma", 0)),
"vega": float(raw_data.get("vega", 0)),
"theta": float(raw_data.get("theta", 0)),
"rho": float(raw_data.get("rho", 0)),
"iv_bid": float(raw_data.get("bidIv", 0)),
"iv_ask": float(raw_data.get("askIv", 0)),
"iv_last": float(raw_data.get("lastIv", 0)),
"mark_price": float(raw_data.get("markPx", 0)),
"open_interest": float(raw_data.get("oi", 0)),
"volume": float(raw_data.get("vol24h", 0))
}
async def _process_buffer(self):
"""버퍼된 데이터를 HolySheep AI로 배치 분석"""
if not self._buffer:
return
# 샘플링: HolySheep 비용 최적화를 위해 10% 샘플링
sample_size = max(1, len(self._buffer) // 10)
sampled = self._buffer[-sample_size:]
tasks = []
for record in sampled:
greeks = {
"delta": record["delta"],
"gamma": record["gamma"],
"vega": record["vega"],
"theta": record["theta"],
"rho": record["rho"],
"implied_volatility": (record["iv_bid"] + record["iv_ask"]) / 2
}
# HolySheep AI 이상치 분석
task = self.holysheep.analyze_greeks_anomaly(
greeks_data=greeks,
market_context=f"Underlying: {record['underlying']}, Strike: {record['strike']}"
)
tasks.append((record, task))
# 병렬 처리
results = await asyncio.gather(*[t[1] for t in tasks], return_exceptions=True)
# 결과 로깅
for i, (record, result) in enumerate(zip(sampled, results)):
if isinstance(result, Exception):
logger.error(f"분석 실패: {record['symbol']} - {result}")
elif result.anomaly_score > 0.7:
logger.warning(
f"고이상치 탐지: {record['symbol']} "
f"(anomaly_score={result.anomaly_score:.2f})"
)
# 버퍼 초기화
self._buffer = []
def stop(self):
"""스트림 수집 중지"""
self._running = False
logger.info("스트림 수집 중지 요청")
사용 예시
async def main():
from holysheep_client import holysheep_client
collector = TardisOKXGreeksCollector(
tardis_config=TardisConfig(),
holysheep_client=holysheep_client
)
# OKX BTC 옵션 Greeks 스트리밍
symbols = [
"BTC-USD-250531-95000-C", # BTC 콜옵션
"BTC-USD-250531-95000-P", # BTC 풋옵션
"BTC-USD-250531-100000-C",
"BTC-USD-250531-90000-P"
]
try:
await collector.stream_greeks(symbols=symbols)
except KeyboardInterrupt:
collector.stop()
if __name__ == "__main__":
asyncio.run(main())5. 역사적 아카이빙과 TimescaleDB 연동
# timescale_archiver.py
=====================
import asyncpg
import asyncio
import logging
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional
from config import DatabaseConfig
logger = logging.getLogger(__name__)
class GreeksArchiver:
"""TimescaleDB 기반 Greeks 데이터 아카이빙"""
CREATE_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS okx_options_greeks (
time TIMESTAMPTZ NOT NULL,
symbol TEXT NOT NULL,
underlying TEXT NOT NULL,
strike DOUBLE PRECISION NOT NULL,
expiry TEXT NOT NULL,
delta DOUBLE PRECISION,
gamma DOUBLE PRECISION,
vega DOUBLE PRECISION,
theta DOUBLE PRECISION,
rho DOUBLE PRECISION,
iv_bid DOUBLE PRECISION,
iv_ask DOUBLE PRECISION,
iv_mid DOUBLE PRECISION,
iv_last DOUBLE PRECISION,
mark_price DOUBLE PRECISION,
open_interest DOUBLE PRECISION,
volume DOUBLE PRECISION,
anomaly_score DOUBLE PRECISION,
PRIMARY KEY (time, symbol)
);
-- TimescaleDB 하이퍼테이블 변환
SELECT create_hypertable('okx_options_greeks', 'time',
if_not_exists => TRUE,
migrate_data => TRUE
);
-- 인덱스 생성
CREATE INDEX IF NOT EXISTS idx_greeks_symbol ON okx_options_greeks (symbol);
CREATE INDEX IF NOT EXISTS idx_greeks_expiry ON okx_options_greeks (expiry);
CREATE INDEX IF NOT EXISTS idx_greeks_strike ON okx_options_greeks (strike);
"""
def __init__(self, config: DatabaseConfig):
self.config = config
self.pool: Optional[asyncpg.Pool] = None
async def connect(self):
"""데이터베이스 풀 연결"""
self.pool = await asyncpg.create_pool(
host=self.config.host,
port=self.config.port,
database=self.config.database,
user=self.config.user,
password=self.config.password,
min_size=5,
max_size=20
)
logger.info("TimescaleDB 연결 완료")
async def initialize_schema(self):
"""스키마 초기화"""
async with self.pool.acquire() as conn:
await conn.execute(self.CREATE_TABLE_SQL)
logger.info("스키마 초기화 완료")
async def insert_greeks(
self,
records: List[Dict[str, Any]],
anomaly_scores: Optional[List[float]] = None
):
"""Greeks 데이터 배치 삽입"""
if not records:
return
insert_sql = """
INSERT INTO okx_options_greeks (
time, symbol, underlying, strike, expiry,
delta, gamma, vega, theta, rho,
iv_bid, iv_ask, iv_mid, iv_last,
mark_price, open_interest, volume, anomaly_score
) VALUES (
$1, $2, $3, $4, $5,
$6, $7, $8, $9, $10,
$11, $12, $13, $14,
$15, $16, $17, $18
) ON CONFLICT (time, symbol) DO UPDATE SET
delta = EXCLUDED.delta,
gamma = EXCLUDED.gamma,
vega = EXCLUDED.vega,
theta = EXCLUDED.theta,
rho = EXCLUDED.rho,
iv_mid = EXCLUDED.iv_mid,
anomaly_score = COALESCE(EXCLUDED.anomaly_score, okx_options_greeks.anomaly_score)
"""
async with self.pool.acquire() as conn:
async with conn.transaction():
for i, record in enumerate(records):
iv_mid = (record.get("iv_bid", 0) + record.get("iv_ask", 0)) / 2
score = anomaly_scores[i] if anomaly_scores else None
await conn.execute(
insert_sql,
datetime.fromisoformat(record["timestamp"].replace("Z", "+00:00")),
record["symbol"],
record["underlying"],
record["strike"],
record["expiry"],
record["delta"],
record["gamma"],
record["vega"],
record["theta"],
record["rho"],
record["iv_bid"],
record["iv_ask"],
iv_mid,
record.get("iv_last", iv_mid),
record["mark_price"],
record["open_interest"],
record["volume"],
score
)
logger.info(f"{len(records)}건 데이터 삽입 완료")
async def query_volatility_surface(
self,
expiry: str,
underlying: str = "BTC-USD",
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None
) -> List[Dict[str, Any]]:
"""변동성 곡면 생성을 위한 IV 데이터 조회"""
where_clauses = ["expiry = $1", "underlying = $2"]
params = [expiry, underlying]
param_idx = 3
if start_time:
where_clauses.append(f"time >= ${param_idx}")
params.append(start_time)
param_idx += 1
if end_time:
where_clauses.append(f"time <= ${param_idx}")
params.append(end_time)
sql = f"""
SELECT
time_bucket('1 minute', time) as bucket,
strike,
AVG(iv_mid) as avg_iv,
STDDEV(iv_mid) as iv_std,
AVG(delta) as avg_delta,
AVG(gamma) as avg_gamma,
COUNT(*) as sample_count
FROM okx_options_greeks
WHERE {' AND '.join(where_clauses)}
GROUP BY bucket, strike
ORDER BY strike, bucket
"""
async with self.pool.acquire() as conn:
rows = await conn.fetch(sql, *params)
return [
{
"time": row["bucket"],
"strike": row["strike"],
"avg_iv": row["avg_iv"],
"iv_std": row["iv_std"],
"avg_delta": row["avg_delta"],
"avg_gamma": row["avg_gamma"],
"sample_count": row["sample_count"]
}
for row in rows
]
async def get_moneyness_buckets(self, expiry: str) -> Dict[str, float]:
"""만기별 moneyness별 IV 통계"""
sql = """
WITH strikes AS (
SELECT DISTINCT strike FROM okx_options_greeks WHERE expiry = $1
),
moneyness AS (
SELECT
g.*,
(SELECT percentile_cont(0.5) FROM okx_options_greeks
WHERE expiry = $1 AND strike = g.strike AND time > NOW() - INTERVAL '1 hour') as current_price
FROM okx_options_greeks g
WHERE expiry = $1
)
SELECT
CASE
WHEN strike / current_price > 1.1 THEN 'OTM_10%'
WHEN strike / current_price > 1.05 THEN 'OTM_5%'
WHEN strike / current_price BETWEEN 0.95 AND 1.05 THEN 'ATM'
WHEN strike / current_price > 0.9 THEN 'ITM_5%'
ELSE 'ITM_10%'
END as moneyness_bucket,
AVG(iv_mid) as avg_iv,
STDDEV(iv_mid) as iv_std,
COUNT(*) as count
FROM moneyness
WHERE time > NOW() - INTERVAL '24 hours'
GROUP BY moneyness_bucket
ORDER BY moneyness_bucket
"""
async with self.pool.acquire() as conn:
rows = await conn.fetch(sql, expiry)
return {row["moneyness_bucket"]: row["avg_iv"] for row in rows}
async def continuous_aggregate(self, granularity: str = "1 hour"):
"""실시간 집계 뷰 생성"""
sql = f"""
CREATE MATERIALIZED VIEW IF NOT EXISTS
greeks_hourly_{granularity.replace(' ', '_')}
WITH (timescaledb.continuous) AS
SELECT
time_bucket('{granularity}', time) as bucket,
expiry,
strike,
AVG(delta) as avg_delta,
AVG(gamma) as avg_gamma,
AVG(vega) as avg_vega,
AVG(theta) as avg_theta,
AVG(iv_mid) as avg_iv,
PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY iv_mid) as median_iv
FROM okx_options_greeks
GROUP BY bucket, expiry, strike
"""
async with self.pool.acquire() as conn:
await conn.execute(sql)
logger.info(f"연속 집계 생성 완료: {granularity}")6. 변동성 곡면 시각화
# volatility_surface.py
======================
import plotly.graph_objects as go
import pandas as pd
from typing import List, Dict, Any, Optional
from timescale_archiver import GreeksArchiver
class VolatilitySurfaceBuilder:
"""변동성 곡면 빌더 및 시각화"""
def build_surface_3d(
self,
surface_data: List[Dict[str, Any]],
title: str = "OKX Options Implied Volatility Surface"
) -> go.Figure:
"""3D 변동성 곡면 생성"""
df = pd.DataFrame(surface_data)
# 피벗 테이블 생성
pivot = df.pivot_table(
values='avg_iv',
index='strike',
columns='bucket',
aggfunc='mean'
)
strikes = pivot.index.tolist()
times = [t.isoformat() for t in pivot.columns]
fig = go.Figure(data=[
go.Surface(
z=pivot.values,
x=times,
y=strikes,
colorscale='Viridis',
colorbar=dict(
title='IV',
titleside='right',
titlefont=dict(size=14)
),
hovertemplate='Strike: %{y}
Time: %{x}
IV: %{z:.4f}7. 프로덕션 배포: Docker와 모니터링
# docker-compose.yml
version: '3.8'
services:
greeks-collector:
build:
context: .
dockerfile: Dockerfile
environment:
- HOLYSHEEP_API_KEY=${HOLYSHEEP_API_KEY}
- TARDIS_API_KEY=${TARDIS_API_KEY}
- TIMESCALE_HOST=timescaledb
- DB_USER=postgres
- DB_PASSWORD=${DB_PASSWORD}
depends_on:
- timescaledb
restart: unless-stopped
deploy:
resources:
limits:
cpus: '2'
memory: 4G
timescaledb:
image: timescale/timescaledb:latest-pg16
environment:
- POSTGRES_DB=options_data
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=${DB_PASSWORD}
volumes:
- timeseries_data:/var/lib/postgresql/data
ports:
- "5432:5432"
restart: unless-stopped
prometheus:
image: prom/prometheus:latest
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
ports:
- "9090:9090"
grafana:
image: grafana/grafana:latest
ports:
- "3000:3000"
volumes:
- grafana_data:/var/lib/grafana
environment:
- GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
volumes:
timeseries_data:
grafana_data:# metrics.py - Prometheus 메트릭스
=================================
from prometheus_client import Counter, Histogram, Gauge
import time
메트릭스 정의
GREEKS_RECORDS_COLLECTED = Counter(
'greeks_records_collected_total',
'Total Greeks records collected',
['symbol', 'exchange']
)
GREEKS_LATENCY = Histogram(
'greeks_api_latency_seconds',
'HolySheep API latency',
['operation'],
buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)
ANOMALY_DETECTED = Counter(
'greeks_anomaly_detected_total',
'Total anomalies detected',
['severity']
)
IV_VOLATILITY = Gauge(
'iv_current_volatility',
'Current implied volatility',
['symbol', 'expiry']
)
class MetricsCollector:
"""Prometheus 메트릭스 수집기"""
@staticmethod
@GREEKS_LATENCY.time()
async def call_holysheep(operation: str, func, *args, **kwargs):
start = time.time()
try:
result = await func(*args