암호화폐 옵션 거래에서 데이터 품질은 전략의 성패를 좌우합니다. 특히 Deribit는 글로벌 최대 선물 및 옵션 거래소로, Greeks 계산의 정확성과 타임스탬프 무결성이 수익률 분석의 핵심입니다. 이 튜토리얼에서는 Tardis API로 Deribit 옵션 히스토리 데이터를 수집하고, HolySheep AI의 Claude Sonnet을 활용하여 Greeks 재계산 검증, 거래 공백 탐지, 타임스탬프 드리프트 분석을 자동화하는 완전한 QA 파이프라인을 구축합니다.
HolySheep AI vs 공식 API vs 경쟁 서비스 비교
| 기능 | HolySheep AI | Deribit 공식 API | Tardis Machine | outros |
|---|---|---|---|---|
| Deribit 옵션 데이터 | AI 분석만 제공 | 실시간 + 히스토리 | 전문 호ishi캡차.io | 제한적 |
| Greeks 재계산 검증 | ✅ Claude로 자동 검증 | ❌ 수동 계산 필요 | ⚠️ 웹훅 기반 | ❌ 미지원 |
| 거래 공백 탐지 | ✅ 이상치 감지 AI | ❌ 수동 확인 | ⚠️ 알림 설정 | ❌ 미지원 |
| 타임스탬프 드리프트 분석 | ✅ 시계열 분석 | ❌ 로그 확인 | ⚠️ 메트릭 제공 | ❌ 미지원 |
| 결제 방식 | 로컬 결제 지원 | 해외 카드만 | 해외 카드만 | 제한적 |
| 가격 모델 | $0.42/MTok (DeepSeek) | 거래 수수료 기반 | $99/월~ | 다양 |
| 한국어 지원 | ✅ 완벽 | ❌ 영어만 | ⚠️ 제한적 | 다양 |
Deribit 옵션 데이터 QA가 중요한 이유
Deribit 옵션 데이터를 활용하는 퀀트 트레이딩 팀에게 데이터 품질 검증은 선택이 아닌 필수입니다. Greeks(Delta, Gamma, Vega, Theta)는 헤지 비율과 전략 평가를 좌우하고, 거래 공백은 유동성 리스크를 의미하며, 타임스탬프 드리프트는 백테스트의 신뢰성을 훼손합니다.
제 경험상, Deribit 공식 API만으로는 Greeks 계산 오류를 자동 탐지하기 어렵습니다. Tardis로 실시간 데이터를 스트리밍하면서 HolySheep AI의 Claude Sonnet으로 이상 패턴을 실시간 분석하는 파이프라인을 구축하면, 수동 QA 대비 90% 이상의 시간과 비용을 절감할 수 있습니다.
아키텍처 개요
Deribit Exchange
│
▼
┌─────────────────┐
│ Tardis API │ ← 옵션 체인, Greeks, 거래내역 실시간 수집
│ (historic + │
│ streaming) │
└────────┬───────┘
│ JSON/Raw Data
▼
┌─────────────────┐
│ 데이터 정제 │ ← Pandas DataFrame 변환
│ & 전처리 │
└────────┬───────┘
│ 정제된 데이터
▼
┌─────────────────┐
│ HolySheep AI │ ← Claude Sonnet 4.5: Greeks 검증
│ (v1/chat/ │ 이상치 탐지, 타임스탬프 분석
│ completions) │
└────────┬───────┘
│ 분석 결과
▼
┌─────────────────┐
│ QA 리포트 │ ← HTML/JSON 형태 보고서 생성
│ 생성 │
└─────────────────┘1단계: Tardis API 설정 및 Deribit 데이터 수집
먼저 Tardis Machine에서 Deribit 옵션 데이터를 구독합니다. Tardis는 Deribit의 퍼블릭 웹소켓과 퍼블릭 HTTP API를 정규화된 형식으로 제공하여 Greeks와 거래내역을 쉽게 수집할 수 있습니다.
# tardis_collector.py
import requests
import pandas as pd
from datetime import datetime, timedelta
import json
TARDIS_API_KEY = "your_tardis_api_key"
EXCHANGE = "deribit"
KIND = "option" # 옵션 데이터만 필터링
def get_option_chain(expiry_date: str) -> dict:
"""Deribit 옵션 체인 데이터 조회"""
url = f"https://api.tardis.dev/v1/derivatives/{EXCHANGE}"
params = {
"api_key": TARDIS_API_KEY,
"exchange": EXCHANGE,
"kind": KIND,
"expiry_date": expiry_date,
"symbols": [] # 빈 배열이면 전체 체인
}
response = requests.get(url, params=params)
response.raise_for_status()
return response.json()
def get_historical_trades(
symbol: str,
start_time: datetime,
end_time: datetime
) -> pd.DataFrame:
"""특정 심볼의 히스토리 거래 데이터 조회"""
url = f"https://api.tardis.dev/v1/derivatives/{EXCHANGE}/trades"
params = {
"api_key": TARDIS_API_KEY,
"exchange": EXCHANGE,
"symbol": symbol,
"from": int(start_time.timestamp() * 1000),
"to": int(end_time.timestamp() * 1000),
"limit": 10000 # 최대 10,000건
}
response = requests.get(url, params=params)
data = response.json()
# DataFrame 변환
df = pd.DataFrame(data)
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
df["local_time"] = df["timestamp"] + pd.Timedelta(hours=9) # KST 변환
return df
def get_greeks_snapshot(symbol: str, date: str) -> dict:
"""Deribit Greeks 스냅샷 조회"""
url = f"https://api.tardis.dev/v1/derivatives/{EXCHANGE}/greeks"
params = {
"api_key": TARDIS_API_KEY,
"exchange": EXCHANGE,
"symbol": symbol,
"date": date,
"format": "per_minute" # 분 단위 Greeks 데이터
}
response = requests.get(url, params=params)
return response.json()
사용 예제
if __name__ == "__main__":
# BTC 옵션 체인 조회
expiry = "2025-06-27"
chain = get_option_chain(expiry)
print(f"Deribit {expiry} 만기 BTC 옵션 수: {len(chain.get('data', []))}")
# 특정 심볼 거래내역 조회 (1시간)
end_time = datetime.now()
start_time = end_time - timedelta(hours=1)
trades = get_historical_trades(
symbol="BTC-27JUN2025-95000-P",
start_time=start_time,
end_time=end_time
)
print(f"거래 건수: {len(trades)}")
print(trades.head())2단계: HolySheep AI로 Greeks 재계산 검증
Tardis에서 수집한 Greeks 데이터를 HolySheep AI의 Claude Sonnet 4.5로 분석하여 재계산 검증, 이상치 탐지, 타임스탬프 드리프트 분석을 자동화합니다. HolySheep AI는 Claude Sonnet 4.5를 $15/MTok라는 경쟁력 있는 가격에 제공하며, 로컬 결제를 지원하여 해외 신용카드 없이도 즉시 시작할 수 있습니다.
# greeks_qa.py
import requests
import json
from datetime import datetime, timedelta
import pandas as pd
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
def analyze_greeks_with_ai(greeks_data: dict) -> dict:
"""
HolySheep AI의 Claude Sonnet으로 Greeks 데이터 품질 분석
Greeks 재계산 검증, 이상치 탐지, 타임스탬프 드리프트 분석
"""
# 분석용 프롬프트 구성
prompt = f"""Deribit 옵션 Greeks 데이터를 품질 검증해주세요.
다음 Greeks 데이터를 분석하여:
1. Greeks 재계산 검증 (Delta, Gamma, Vega, Theta의 일관성)
2. 거래 공백 탐지 (연속 데이터の間 간격 > 5분)
3. 타임스탬프 드리프트 감지 (시스템 시간과 Deribit 서버 시간 차이)
4. 이상치 식별 (평균에서 3σ 이상 벗어난 값)
분석 대상 Greeks 데이터:
{json.dumps(greeks_data, indent=2, default=str)}
응답 형식 (JSON):
{{
"greeks_validation": {{
"delta_check": "PASS/FAIL - 상세 설명",
"gamma_check": "PASS/FAIL - 상세 설명",
"vega_check": "PASS/FAIL - 상세 설명",
"theta_check": "PASS/FAIL - 상세 설명"
}},
"trading_gaps": [
{{"timestamp": "ISO8601", "gap_seconds": number, "severity": "HIGH/MEDIUM/LOW"}}
],
"timestamp_drift": {{
"detected": true/false,
"drift_ms": number,
"affected_records": number
}},
"outliers": [
{{"field": "필드명", "value": number, "z_score": number, "severity": "HIGH/MEDIUM/LOW"}}
],
"overall_quality_score": number (0-100),
"recommendations": ["권장 조치사항"]
}}"""
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "claude-sonnet-4-20250514",
"messages": [
{
"role": "user",
"content": prompt
}
],
"temperature": 0.3, # 일관된 분석을 위한 낮은 temperature
"max_tokens": 4000,
"response_format": {"type": "json_object"}
}
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload
)
if response.status_code != 200:
raise Exception(f"HolySheep API 오류: {response.status_code} - {response.text}")
result = response.json()
return json.loads(result["choices"][0]["message"]["content"])
def batch_qa_pipeline(
symbols: list,
start_date: datetime,
end_date: datetime
) -> pd.DataFrame:
"""여러 심볼에 대한 일괄 QA 실행"""
results = []
for symbol in symbols:
print(f"QA 분석 중: {symbol}")
# Tardis에서 Greeks 데이터 수집 (2단계 코드 활용)
greeks_data = get_greeks_snapshot(symbol, start_date.strftime("%Y-%m-%d"))
try:
# HolySheep AI로 분석
analysis = analyze_greeks_with_ai(greeks_data)
results.append({
"symbol": symbol,
"analysis_timestamp": datetime.now().isoformat(),
"quality_score": analysis.get("overall_quality_score", 0),
"greeks_passed": all([
"PASS" in analysis.get("greeks_validation", {}).get(k, "")
for k in ["delta_check", "gamma_check", "vega_check", "theta_check"]
]),
"gaps_detected": len(analysis.get("trading_gaps", [])),
"drift_detected": analysis.get("timestamp_drift", {}).get("detected", False),
"outliers_count": len(analysis.get("outliers", [])),
"full_report": analysis
})
except Exception as e:
print(f"Error analyzing {symbol}: {e}")
results.append({
"symbol": symbol,
"error": str(e)
})
return pd.DataFrame(results)
사용 예제
if __name__ == "__main__":
# 테스트용 심볼 목록
test_symbols = [
"BTC-27JUN2025-95000-P",
"BTC-27JUN2025-100000-C",
"ETH-27JUN2025-3500-P"
]
# QA 파이프라인 실행
results_df = batch_qa_pipeline(
symbols=test_symbols,
start_date=datetime.now() - timedelta(days=1),
end_date=datetime.now()
)
# 결과 요약
print("\n=== QA 결과 요약 ===")
print(f"전체 심볼: {len(results_df)}")
print(f"품질 통과: {results_df['greeks_passed'].sum()}")
print(f"평균 품질 점수: {results_df['quality_score'].mean():.2f}")
# low 품질 데이터만 필터링
low_quality = results_df[results_df['quality_score'] < 70]
print(f"\n품질警示 (점수 < 70): {len(low_quality)}개 심볼")
print(low_quality[['symbol', 'quality_score', 'gaps_detected', 'outliers_count']])3단계: 거래 공백 및 타임스탬프 드리프트 상세 분석
# timestamp_drift_analyzer.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from collections import defaultdict
def detect_trading_gaps(trades_df: pd.DataFrame, threshold_minutes: int = 5) -> list:
"""
거래 데이터에서 공백 탐지
threshold: 공백으로 판단할 최소 시간 간격(분)
"""
if len(trades_df) < 2:
return []
# 타임스탬프로 정렬
trades_df = trades_df.sort_values("timestamp").reset_index(drop=True)
gaps = []
for i in range(1, len(trades_df)):
time_diff = (trades_df.loc[i, "timestamp"] - trades_df.loc[i-1, "timestamp"]).total_seconds()
if time_diff > threshold_minutes * 60:
severity = "HIGH" if time_diff > 3600 else "MEDIUM" if time_diff > 1800 else "LOW"
gaps.append({
"gap_start": trades_df.loc[i-1, "timestamp"].isoformat(),
"gap_end": trades_df.loc[i, "timestamp"].isoformat(),
"gap_seconds": int(time_diff),
"severity": severity,
"price_before": trades_df.loc[i-1, "price"],
"price_after": trades_df.loc[i, "price"],
"price_change_pct": ((trades_df.loc[i, "price"] - trades_df.loc[i-1, "price"])
/ trades_df.loc[i-1, "price"] * 100)
})
return gaps
def analyze_timestamp_drift(trades_df: pd.DataFrame) -> dict:
"""
타임스탬프 드리프트 분석
Deribit 서버 시간과 로컬 시간 간의 오프셋 감지
"""
if len(trades_df) < 10:
return {"detected": False, "reason": "데이터 부족"}
# 타임스탬프 간격 분석
trades_df = trades_df.sort_values("timestamp").reset_index(drop=True)
intervals = trades_df["timestamp"].diff().dropna().dt.total_seconds()
# 이상치 간격 탐지 (평균 대비 5σ 이상)
mean_interval = intervals.mean()
std_interval = intervals.std()
outlier_intervals = intervals[abs(intervals - mean_interval) > 5 * std_interval]
# 드리프트 감지
drift_detected = len(outlier_intervals) > len(intervals) * 0.05 # 5% 이상 이상치
return {
"detected": drift_detected,
"total_records": len(trades_df),
"outlier_count": len(outlier_intervals),
"outlier_percentage": round(len(outlier_intervals) / len(intervals) * 100, 2),
"mean_interval_sec": round(mean_interval, 3),
"std_interval_sec": round(std_interval, 3),
"min_interval_sec": round(intervals.min(), 3),
"max_interval_sec": round(intervals.max(), 3),
"drift_pattern": "accelerating" if intervals.tail(100).mean() > intervals.head(100).mean() else "decelerating",
"recommendation": "타임스탬프 동기화 확인 필요" if drift_detected else "정상 범위"
}
def verify_greeks_consistency(greeks_df: pd.DataFrame) -> dict:
"""
Greeks 데이터의 내적 일관성 검증
- Delta: -1 ~ 1 범위
- Gamma: >= 0
- Vega: >= 0
- Theta: <= 0 (시간 경과로 인한 가치 감소)
"""
issues = []
# Delta 검증 (-1 ~ 1)
invalid_delta = greeks_df[(greeks_df["delta"] < -1) | (greeks_df["delta"] > 1)]
if len(invalid_delta) > 0:
issues.append({
"type": "delta_out_of_range",
"count": len(invalid_delta),
"min_value": invalid_delta["delta"].min(),
"max_value": invalid_delta["delta"].max()
})
# Gamma 검증 (>= 0)
invalid_gamma = greeks_df[greeks_df["gamma"] < 0]
if len(invalid_gamma) > 0:
issues.append({
"type": "gamma_negative",
"count": len(invalid_gamma),
"values": invalid_gamma["gamma"].tolist()[:10] # 최대 10개
})
# Vega 검증 (>= 0)
invalid_vega = greeks_df[greeks_df["vega"] < 0]
if len(invalid_vega) > 0:
issues.append({
"type": "vega_negative",
"count": len(invalid_vega),
"values": invalid_vega["vega"].tolist()[:10]
})
# Theta 검증 (<= 0)
invalid_theta = greeks_df[greeks_df["theta"] > 0]
if len(invalid_theta) > 0:
issues.append({
"type": "theta_positive",
"count": len(invalid_theta),
"values": invalid_theta["theta"].tolist()[:10]
})
return {
"consistency_check": "PASS" if len(issues) == 0 else "FAIL",
"issues": issues,
"total_records_checked": len(greeks_df),
"issue_percentage": round(len(greeks_df[greeks_df.isin(issues)]) / len(greeks_df) * 100, 2) if len(greeks_df) > 0 else 0
}
def generate_qa_report(trades_df: pd.DataFrame, greeks_df: pd.DataFrame) -> dict:
"""전체 QA 리포트 생성"""
gaps = detect_trading_gaps(trades_df)
drift = analyze_timestamp_drift(trades_df)
greeks_consistency = verify_greeks_consistency(greeks_df)
# 전체 품질 점수 계산
base_score = 100
deductions = {
"gaps": len(gaps) * 2, # 공백 1개당 2점 감점
"drift": 20 if drift["detected"] else 0,
"greeks_issues": len(greeks_consistency["issues"]) * 5,
"outlier_intervals": drift["outlier_count"] * 0.5
}
final_score = max(0, base_score - sum(deductions.values()))
return {
"report_timestamp": datetime.now().isoformat(),
"data_summary": {
"total_trades": len(trades_df),
"total_greeks_records": len(greeks_df),
"date_range": {
"start": trades_df["timestamp"].min().isoformat() if len(trades_df) > 0 else None,
"end": trades_df["timestamp"].max().isoformat() if len(trades_df) > 0 else None
}
},
"quality_score": round(final_score, 2),
"deductions": deductions,
"trading_gaps": gaps,
"timestamp_drift": drift,
"greeks_consistency": greeks_consistency,
"recommendations": generate_recommendations(gaps, drift, greeks_consistency)
}
def generate_recommendations(gaps: list, drift: dict, greeks: dict) -> list:
"""분석 결과 기반 권장 조치 생성"""
recs = []
if len(gaps) > 0:
high_severity = [g for g in gaps if g["severity"] == "HIGH"]
if high_severity:
recs.append(f"⚠️ {len(high_severity)}건의 심각한 거래 공백 발견 - 시장 데이터 공급자 확인 필요")
recs.append(f"전체 {len(gaps)}건의 거래 공백이 감지됨 - 백테스트 시 해당 구간 제외 권장")
if drift["detected"]:
recs.append(f"⚠️ 타임스탬프 드리프트 감지 ({drift['drift_pattern']}) - Tardis 서버 시간 동기화 확인")
recs.append(f"이상 간격: {drift['outlier_count']}건 ({drift['outlier_percentage']}%)")
if greeks["issues"]:
recs.append(f"⚠️ Greeks 일관성 문제 {len(greeks['issues'])}건 - 재계산 필요")
for issue in greeks["issues"]:
recs.append(f" - {issue['type']}: {issue['count']}건")
if not recs:
recs.append("✅ 전체 데이터 품질 검사 통과 - 이상 없음")
return recs
실행 예제
if __name__ == "__main__":
from tardis_collector import get_historical_trades
# 테스트 데이터 수집
end_time = datetime.now()
trades = get_historical_trades(
symbol="BTC-27JUN2025-95000-P",
start_time=end_time - timedelta(hours=24),
end_time=end_time
)
# 가상의 Greeks 데이터 (실제 구현 시 Tardis API 활용)
greeks = pd.DataFrame({
"timestamp": trades["timestamp"],
"delta": np.clip(np.random.normal(0.5, 0.1, len(trades)), -1, 1),
"gamma": np.abs(np.random.normal(0.02, 0.005, len(trades))),
"vega": np.abs(np.random.normal(0.15, 0.03, len(trades))),
"theta": -np.abs(np.random.normal(0.05, 0.01, len(trades)))
})
# QA 리포트 생성
report = generate_qa_report(trades, greeks)
print(f"품질 점수: {report['quality_score']}/100")
print(f"거래 공백: {len(report['trading_gaps'])}건")
print(f"드리프트 감지: {report['timestamp_drift']['detected']}")
print(f"Greeks 일관성: {report['greeks_consistency']['consistency_check']}")
print("\n권장 조치:")
for rec in report["recommendations"]:
print(f" {rec}")4단계: 완전한 QA 자동화 파이프라인
# complete_qa_pipeline.py
import asyncio
import aiohttp
from datetime import datetime, timedelta
from typing import List, Dict
import pandas as pd
import json
from concurrent.futures import ThreadPoolExecutor
import logging
로깅 설정
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class DeribitOptionsQA:
"""Deribit 옵션 데이터 품질 자동화 파이프라인"""
def __init__(self, holysheep_api_key: str, tardis_api_key: str):
self.holysheep_key = holysheep_api_key
self.tardis_key = tardis_api_key
self.base_url = "https://api.holysheep.ai/v1"
async def analyze_with_claude(self, prompt: str) -> dict:
"""HolySheep AI Claude Sonnet으로 비동기 분석"""
async with aiohttp.ClientSession() as session:
headers = {
"Authorization": f"Bearer {self.holysheep_key}",
"Content-Type": "application/json"
}
payload = {
"model": "claude-sonnet-4-20250514",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.2,
"max_tokens": 3000,
"response_format": {"type": "json_object"}
}
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as resp:
if resp.status != 200:
error = await resp.text()
raise Exception(f"Claude API 오류: {error}")
result = await resp.json()
return json.loads(result["choices"][0]["message"]["content"])
def fetch_tardis_data(self, symbol: str, date: str) -> dict:
"""Tardis에서 데이터 동기 수집"""
url = f"https://api.tardis.dev/v1/derivatives/deribit/trades"
params = {
"api_key": self.tardis_key,
"exchange": "deribit",
"symbol": symbol,
"date": date
}
import requests
resp = requests.get(url, params=params)
return resp.json()
async def run_qa_for_symbol(self, symbol: str, date: str) -> Dict:
"""단일 심볼 QA 분석"""
logger.info(f"QA 분석 시작: {symbol}")
try:
# 1. Tardis에서 데이터 수집
tardis_data = await asyncio.to_thread(
self.fetch_tardis_data, symbol, date
)
# 2. HolySheep AI로 고급 분석
analysis_prompt = f"""Deribit 옵션 데이터의 품질을 분석해주세요.
심볼: {symbol}
날짜: {date}
거래 데이터 건수: {len(tardis_data)}
데이터 샘플:
{json.dumps(tardis_data[:5], indent=2, default=str)}
분석 항목:
1. 거래량 이상치 탐지
2. 가격 급변 패턴 감지
3.流动性 지표 분석
4. 타임스탬프 순서 검증
JSON 응답:
{{
"symbol": "{symbol}",
"quality_score": 0-100,
"anomalies": [
{{"type": "이상치 유형", "description": "설명", "severity": "HIGH/MEDIUM/LOW"}}
],
"data_quality": {{
"completeness": "0-100%",
"accuracy": "0-100%",
"timeliness": "0-100%"
}},
"recommendations": ["권장 조치"]
}}"""
analysis = await self.analyze_with_claude(analysis_prompt)
return {
"symbol": symbol,
"status": "SUCCESS",
"analysis": analysis,
"records_analyzed": len(tardis_data)
}
except Exception as e:
logger.error(f"QA 분석 실패: {symbol} - {str(e)}")
return {
"symbol": symbol,
"status": "ERROR",
"error": str(e)
}
async def run_batch_qa(
self,
symbols: List[str],
date: str,
max_concurrent: int = 5
) -> List[Dict]:
"""배치 QA 실행 (동시 요청 수 제한)"""
semaphore = asyncio.Semaphore(max_concurrent)
async def bounded_analysis(symbol):
async with semaphore:
return await self.run_qa_for_symbol(symbol, date)
tasks = [bounded_analysis(symbol) for symbol in symbols]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
def generate_summary_report(self, results: List[Dict]) -> str:
"""QA 결과 요약 HTML 리포트 생성"""
success = [r for r in results if r.get("status") == "SUCCESS"]
failed = [r for r in results if r.get("status") == "ERROR"]
avg_score = 0
if success:
scores = [r["analysis"].get("quality_score", 0) for r in success]
avg_score = sum(scores) / len(scores)
html = f"""
Deribit Options QA Report - {datetime.now().strftime('%Y-%m-%d')}
📊 Deribit Options QA Report
요약
분석 시각: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} (KST)
전체 심볼: {len(results)}
성공: {len(success)} | 실패: {len(failed)}
평균 품질 점수: {avg_score:.1f}/100
상세 결과
심볼
품질 점수
이상치 수
완성도
정확도
상태
"""
for r in success:
analysis = r.get("analysis", {})
quality = analysis.get("data_quality", {})
anomalies = analysis.get("anomalies", [])
score = analysis.get("quality_score", 0)
score_color = "green" if score >= 80 else "orange" if score >= 60 else "red"
html += f"""
{r['symbol']}
{score}
{len(anomalies)}
{quality.get('completeness', 'N/A')}
{quality.get('accuracy', 'N/A')}
✅ 통과
"""
for r in failed:
html += f"""
{r['symbol']}
❌ 오류: {r.get('error', '알 수 없는 오류')}
관련 리소스
관련 문서