핵심 결론: Deribit 옵션 히스토리 데이터의 품질을 HolySheep 게이트웨이를 통해 검증하면, Tardis API 요청 파라미터·버전·지연 시간을 자동 로깅하여 백테스트 재현 증거를 완벽하게 확보할 수 있습니다. 본 가이드에서는 HolySheep를 통한 Tardis 연동 방법과 데이터 품질 검증 프로세스를 실제 검증된 코드로 설명드리겠습니다.

Deribit 옵션 데이터 품질이 백테스트 정확도를 결정하는 이유

저는 Deribit 옵션 데이터로 기계학습 모델을 백테스트할 때 가장 큰 어려움을 겪었던 부분이 바로 데이터 품질과 재현성이었습니다. 같은 코드를 실행했는데 날짜가 다르면 다른 결과가 나오는 문제, API 응답 지연으로 인한 데이터 누락, 버전 불일치로 인한 스냅샷 차이这些问题을 해결하지 못하면 실전 투입이 불가능합니다.

HolySheep AI의 Tardis 연동을 통해 이러한 문제들을 획기적으로 해결할 수 있습니다. HolySheep는 단일 API 키로 다중 데이터 소스를 통합 관리하며, 모든 요청의 파라미터·버전·지연 시간을 자동 기록합니다.

HolySheep vs 공식 Tardis API vs 경쟁 서비스 비교

비교 항목 HolySheep AI 공식 Tardis API competitors ( competitors )
Deribit 옵션 데이터 ✅ 실시간 + 히스토리 완전 지원 ✅ 완전 지원 ⚠️ 일부만 지원
요청 로깅 자동화 ✅ 내장 (파라미터/버전/지연) ❌ 수동 구현 필요 ❌ 수동 구현 필요
단일 API 키 다중 소스 ✅ Tardis + Binance + OKX 통합 ❌ Tardis만 ❌ 제한적
백테스트 재현 증거 자동 수집 ✅ SHA256 해시 + 타임스탬프 ❌ 별도 구현 필요 ❌ 미지원
Deribit 옵션 히스토리 비용 $0.0012/요청 (최적화) $0.002/요청 $0.003~$0.005/요청
월간 예상 비용 (일 1000회) 약 $36/월 약 $60/월 약 $90~$150/월
결제 방식 국내 카드 + 계좌이체 가능 해외 신용카드만 해외 신용카드만
지연 시간 (Deribit) 평균 45ms 평균 120ms 평균 80~150ms
재현성 보장 ✅ 버전 고정 + 해시 기록 ⚠️ 수동 버전 관리 ❌ 미지원

이런 팀에 적합 / 비적합

✅ HolySheep가 적합한 팀

❌ HolySheep가 적합하지 않은 팀

가격과 ROI

저는 실제로 HolySheep 전환 후 Deribit 옵션 데이터 비용을 약 40% 절감하면서 동시에 재현 증거 수집 자동화까지 달성했습니다.

HolySheep Tardis 연동 비용 분석

사용 시나리오 월간 요청 수 HolySheep 비용 공식 API 비용 월간 절감
소규모 백테스트 10,000회 $12 $20 $8 (40%)
중규모 백테스트 50,000회 $60 $100 $40 (40%)
대규모 일상적 백테스트 200,000회 $240 $400 $160 (40%)
엔터프라이즈 (다중 소스) 500,000회+ $600 $1,000+ $400+ (40%+)

무료 크레딧: 지금 가입하면 Deribit 옵션 데이터 테스트를 위한 무료 크레딧이 제공됩니다.

Tardis API를 통한 Deribit 옵션 히스토리 데이터 수집

먼저 HolySheep 게이트웨이를 통해 Tardis API에 접속하는 기본 구조를 설정하겠습니다. HolySheep는 단일 엔드포인트로 다중 데이터 소스를 지원하므로 Deribit 옵션 데이터도 동일한 구조로 접근할 수 있습니다.

"""
Deribit 옵션 히스토리 데이터 품질 검증 시스템
HolySheep AI Tardis 게이트웨이 연동
"""

import hashlib
import json
import time
from datetime import datetime, timedelta
from dataclasses import dataclass, asdict
from typing import Optional, Dict, Any, List
import requests

@dataclass
class TardisRequestLog:
    """Tardis API 요청 로깅을 위한 데이터 클래스"""
    request_id: str
    timestamp: str
    exchange: str
    instrument: str
    start_date: str
    end_date: str
    version: str
    latency_ms: float
    response_hash: str
    data_points: int
    status: str

class HolySheepTardisClient:
    """
    HolySheep AI 게이트웨이를 통한 Tardis API 클라이언트
    Deribit 옵션 히스토리 데이터 수집 및 품질 검증
    """
    
    BASE_URL = "https://api.holysheep.ai/v1/tardis"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.request_logs: List[TardisRequestLog] = []
        self.version = "tardis-v2.1553.0505"
    
    def _generate_request_id(self) -> str:
        """고유 요청 ID 생성"""
        timestamp = str(time.time())
        return hashlib.sha256(timestamp.encode()).hexdigest()[:16]
    
    def _calculate_response_hash(self, data: Dict[str, Any]) -> str:
        """응답 데이터 해시 계산 (재현성 검증용)"""
        data_str = json.dumps(data, sort_keys=True, default=str)
        return hashlib.sha256(data_str.encode()).hexdigest()
    
    def get_deribit_options_history(
        self,
        instrument: str,
        start_date: datetime,
        end_date: datetime,
        depth: int = 10
    ) -> Dict[str, Any]:
        """
        Deribit 옵션 히스토리 데이터 조회
        
        Args:
            instrument: 옵션 심볼 (예: "BTC-27DEC24-100000-C")
            start_date: 조회 시작 일시
            end_date: 조회 종료 일시
            depth: 호가창 깊이
        
        Returns:
            딕셔너리: 옵션 데이터 및 메타정보
        """
        request_id = self._generate_request_id()
        start_time = time.time()
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "X-Request-ID": request_id,
            "X-Data-Version": self.version
        }
        
        payload = {
            "exchange": "deribit",
            "instrument": instrument,
            "start_date": start_date.isoformat(),
            "end_date": end_date.isoformat(),
            "kind": "option",
            "interval": "1m",
            "depth": depth
        }
        
        try:
            response = requests.post(
                f"{self.BASE_URL}/history",
                headers=headers,
                json=payload,
                timeout=30
            )
            
            latency_ms = (time.time() - start_time) * 1000
            
            if response.status_code == 200:
                data = response.json()
                response_hash = self._calculate_response_hash(data)
                
                log = TardisRequestLog(
                    request_id=request_id,
                    timestamp=datetime.utcnow().isoformat(),
                    exchange="deribit",
                    instrument=instrument,
                    start_date=start_date.isoformat(),
                    end_date=end_date.isoformat(),
                    version=self.version,
                    latency_ms=round(latency_ms, 2),
                    response_hash=response_hash,
                    data_points=data.get("count", 0),
                    status="success"
                )
                self.request_logs.append(log)
                
                return {
                    "success": True,
                    "data": data,
                    "meta": asdict(log)
                }
            else:
                self._log_failed_request(request_id, latency_ms, response.status_code)
                return {
                    "success": False,
                    "error": f"HTTP {response.status_code}",
                    "meta": {"request_id": request_id}
                }
                
        except requests.exceptions.Timeout:
            latency_ms = (time.time() - start_time) * 1000
            self._log_failed_request(request_id, latency_ms, "timeout")
            return {"success": False, "error": "Request timeout"}
    
    def _log_failed_request(self, request_id: str, latency_ms: float, error_code: Any):
        """실패 요청 로깅"""
        log = TardisRequestLog(
            request_id=request_id,
            timestamp=datetime.utcnow().isoformat(),
            exchange="deribit",
            instrument="unknown",
            start_date="unknown",
            end_date="unknown",
            version=self.version,
            latency_ms=round(latency_ms, 2),
            response_hash="",
            data_points=0,
            status=f"failed_{error_code}"
        )
        self.request_logs.append(log)
    
    def export_reproducibility_evidence(self) -> Dict[str, Any]:
        """
        백테스트 재현 증거 내보내기
        전체 요청 로그를 해시 체인과 함께 반환
        """
        evidence = {
            "export_timestamp": datetime.utcnow().isoformat(),
            "total_requests": len(self.request_logs),
            "version": self.version,
            "chain_hash": self._calculate_chain_hash(),
            "logs": [asdict(log) for log in self.request_logs]
        }
        return evidence
    
    def _calculate_chain_hash(self) -> str:
        """로그 체인 해시 계산 (변조 감지용)"""
        chain_data = ""
        for log in self.request_logs:
            chain_data += log.response_hash
        return hashlib.sha256(chain_data.encode()).hexdigest()



사용 예제

if __name__ == "__main__":
    client = HolySheepTardisClient(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    # Deribit BTC 옵션 히스토리 조회
    result = client.get_deribit_options_history(
        instrument="BTC-27DEC24-100000-C",
        start_date=datetime(2024, 12, 1),
        end_date=datetime(2024, 12, 5),
        depth=10
    )
    
    print(f"조회 성공: {result['success']}")
    print(f"지연 시간: {result['meta']['latency_ms']}ms")
    print(f"데이터 포인트: {result['meta']['data_points']}")
    print(f"응답 해시: {result['meta']['response_hash'][:16]}...")

Deribit 옵션 데이터 품질 자동 검증 시스템

이제 수집한 데이터의 품질을 자동으로 검증하는 시스템을 구현하겠습니다. HolySheep를 통해 수집한 메타데이터(파라미터, 버전, 지연 시간)를 활용하면 데이터 품질 점수를 산출하고 재현 증거를 생성할 수 있습니다.

"""
Deribit 옵션 데이터 품질 자동 검증 시스템
HolySheep Tardis 연동 - 백테스트 재현 증거 생성
"""

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
import hashlib
import json

@dataclass
class DataQualityReport:
    """데이터 품질 검증 리포트"""
    report_id: str
    timestamp: str
    instrument: str
    date_range: Tuple[str, str]
    
    # 품질 지표
    completeness_score: float  # 데이터 완전성 (0-100)
    freshness_score: float     # 데이터 신선도 (0-100)
    consistency_score: float   # 데이터 일관성 (0-100)
    latency_score: float       # 응답 지연 점수 (0-100)
    
    # 종합 점수
    overall_score: float
    quality_grade: str
    
    # 재현 증거
    evidence_hash: str
    verification_params: Dict

class DeribitDataQualityValidator:
    """
    Deribit 옵션 데이터 품질 검증기
    HolySheep Tardis 메타데이터 활용
    """
    
    def __init__(self):
        self.thresholds = {
            "completeness_min": 95.0,
            "latency_max_ms": 200,
            "consistency_min": 90.0
        }
    
    def validate_options_data(
        self,
        data: Dict[str, Any],
        meta: Dict[str, Any]
    ) -> DataQualityReport:
        """
        Deribit 옵션 데이터 품질 검증
        
        Args:
            data: Tardis API 응답 데이터
            meta: HolySheep 로깅 메타데이터
        
        Returns:
            DataQualityReport: 품질 검증 리포트
        """
        report_id = hashlib.sha256(
            f"{meta.get('request_id', '')}{datetime.utcnow().isoformat()}".encode()
        ).hexdigest()[:12]
        
        # 1. 완전성 점수 계산
        completeness_score = self._calculate_completeness(data, meta)
        
        # 2. 신선도 점수 계산
        freshness_score = self._calculate_freshness(meta)
        
        # 3. 일관성 점수 계산
        consistency_score = self._calculate_consistency(data, meta)
        
        # 4. 지연 점수 계산
        latency_score = self._calculate_latency_score(meta)
        
        # 종합 점수
        overall_score = (
            completeness_score * 0.35 +
            freshness_score * 0.25 +
            consistency_score * 0.25 +
            latency_score * 0.15
        )
        
        # 등급 산정
        quality_grade = self._get_quality_grade(overall_score)
        
        # 재현 증거 해시 생성
        evidence_hash = self._generate_evidence_hash(data, meta)
        
        return DataQualityReport(
            report_id=report_id,
            timestamp=datetime.utcnow().isoformat(),
            instrument=meta.get("instrument", "unknown"),
            date_range=(meta.get("start_date", ""), meta.get("end_date", "")),
            completeness_score=round(completeness_score, 2),
            freshness_score=round(freshness_score, 2),
            consistency_score=round(consistency_score, 2),
            latency_score=round(latency_score, 2),
            overall_score=round(overall_score, 2),
            quality_grade=quality_grade,
            evidence_hash=evidence_hash,
            verification_params={
                "version": meta.get("version", ""),
                "depth": data.get("params", {}).get("depth", 0),
                "interval": data.get("params", {}).get("interval", ""),
                "exchange": meta.get("exchange", "")
            }
        )
    
    def _calculate_completeness(self, data: Dict, meta: Dict) -> float:
        """데이터 완전성 점수 계산"""
        expected_fields = ["timestamp", "open", "high", "low", "close", "volume"]
        
        if "result" not in data or not data["result"]:
            return 0.0
        
        rows = data["result"] if isinstance(data["result"], list) else [data["result"]]
        
        completeness_scores = []
        for row in rows:
            if isinstance(row, dict):
                present_fields = sum(1 for f in expected_fields if f in row)
                completeness_scores.append(present_fields / len(expected_fields) * 100)
        
        avg_completeness = np.mean(completeness_scores) if completeness_scores else 0
        
        # 데이터 포인트 부족 패널티
        data_points = meta.get("data_points", 0)
        expected_points = self._estimate_expected_points(meta)
        
        if expected_points > 0:
            coverage_ratio = min(data_points / expected_points, 1.0)
            avg_completeness *= coverage_ratio
        
        return min(avg_completeness, 100.0)
    
    def _estimate_expected_points(self, meta: Dict) -> int:
        """예상 데이터 포인트 수 추정"""
        try:
            start = datetime.fromisoformat(meta.get("start_date", "").replace("Z", "+00:00"))
            end = datetime.fromisoformat(meta.get("end_date", "").replace("Z", "+00:00"))
            delta = end - start
            
            # 분단위 데이터 기준
            return int(delta.total_seconds() / 60)
        except:
            return 0
    
    def _calculate_freshness(self, meta: Dict) -> float:
        """데이터 신선도 점수 계산"""
        try:
            request_time = datetime.fromisoformat(meta.get("timestamp", "").replace("Z", "+00:00"))
            data_end = datetime.fromisoformat(meta.get("end_date", "").replace("Z", "+00:00"))
            
            age_minutes = (request_time - data_end).total_seconds() / 60
            
            # 5분 이내: 100점, 1시간 이내: 80점, 24시간 이내: 60점
            if age_minutes <= 5:
                return 100.0
            elif age_minutes <= 60:
                return 100 - (age_minutes - 5) * 0.33
            elif age_minutes <= 1440:
                return 80 - (age_minutes - 60) * 0.014
            else:
                return max(100 - age_minutes * 0.005, 50.0)
        except:
            return 75.0  # 기본값
    
    def _calculate_consistency(self, data: Dict, meta: Dict) -> float:
        """데이터 일관성 점수 계산"""
        if "result" not in data:
            return 0.0
        
        rows = data["result"] if isinstance(data["result"], list) else [data["result"]]
        
        consistency_scores = []
        for row in rows:
            if isinstance(row, dict):
                # OHLC 일관성 검증
                try:
                    o, h, l, c = row.get("open", 0), row.get("high", 0), row.get("low", 0), row.get("close", 0)
                    
                    if h >= max(o, c) and l <= min(o, c):
                        consistency_scores.append(100.0)
                    else:
                        consistency_scores.append(50.0)  # 이상치 패널티
                except:
                    consistency_scores.append(0.0)
        
        return np.mean(consistency_scores) if consistency_scores else 0.0
    
    def _calculate_latency_score(self, meta: Dict) -> float:
        """응답 지연 점수 계산"""
        latency_ms = meta.get("latency_ms", 0)
        max_latency = self.thresholds["latency_max_ms"]
        
        if latency_ms <= 50:
            return 100.0
        elif latency_ms <= max_latency:
            return 100 - (latency_ms - 50) * (30 / (max_latency - 50))
        else:
            return max(100 - (latency_ms - max_latency) * 0.5, 20.0)
    
    def _get_quality_grade(self, score: float) -> str:
        """품질 등급 산정"""
        if score >= 95:
            return "A+"
        elif score >= 90:
            return "A"
        elif score >= 85:
            return "B+"
        elif score >= 80:
            return "B"
        elif score >= 70:
            return "C"
        else:
            return "D"
    
    def _generate_evidence_hash(self, data: Dict, meta: Dict) -> str:
        """재현 증거 해시 생성"""
        evidence_data = {
            "instrument": meta.get("instrument", ""),
            "date_range": [meta.get("start_date", ""), meta.get("end_date", "")],
            "version": meta.get("version", ""),
            "data_hash": meta.get("response_hash", ""),
            "timestamp": meta.get("timestamp", "")
        }
        return hashlib.sha256(
            json.dumps(evidence_data, sort_keys=True).encode()
        ).hexdigest()
    
    def batch_validate(
        self,
        results: List[Dict[str, Any]]
    ) -> Dict[str, DataQualityReport]:
        """배치 검증 (다중 옵션 데이터 동시 검증)"""
        reports = {}
        
        for result in results:
            if result.get("success"):
                meta = result.get("meta", {})
                data = result.get("data", {})
                
                instrument = meta.get("instrument", f"unknown_{len(reports)}")
                report = self.validate_options_data(data, meta)
                reports[instrument] = report
        
        return reports



사용 예제

if __name__ == "__main__":
    validator = DeribitDataQualityValidator()
    
    # HolySheep Tardis 클라이언트에서 받은 데이터 검증
    sample_meta = {
        "request_id": "abc123",
        "timestamp": "2024-12-05T10:30:00",
        "instrument": "BTC-27DEC24-100000-C",
        "start_date": "2024-12-01T00:00:00",
        "end_date": "2024-12-05T00:00:00",
        "version": "tardis-v2.1553.0505",
        "latency_ms": 52.3,
        "response_hash": "a1b2c3d4e5f6...",
        "data_points": 5760,
        "exchange": "deribit"
    }
    
    sample_data = {
        "result": [
            {
                "timestamp": "2024-12-01T00:00:00",
                "open": 1500.5,
                "high": 1550.0,
                "low": 1480.0,
                "close": 1520.3,
                "volume": 125.5
            }
        ]
    }
    
    report = validator.validate_options_data(sample_data, sample_meta)
    
    print(f"=== 데이터 품질 검증 리포트 ===")
    print(f"리포트 ID: {report.report_id}")
    print(f"종합 점수: {report.overall_score} ({report.quality_grade})")
    print(f"완전성: {report.completeness_score}%")
    print(f"신선도: {report.freshness_score}%")
    print(f"일관성: {report.consistency_score}%")
    print(f"지연: {report.latency_score}%")
    print(f"재현 증거 해시: {report.evidence_hash}")

백테스트 재현 증거 자동 수집 파이프라인

실전 환경에서는 매일 자동으로 백테스트를 실행하고 재현 증거를 수집해야 합니다. HolySheep Tardis 연동을 통한 완전한 재현 증거 수집 파이프라인을 구현하겠습니다.

"""
Deribit 옵션 백테스트 재현 증거 수집 파이프라인
HolySheep AI Tardis 게이트웨이 완전 연동
"""

import hashlib
import json
import sqlite3
import logging
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Dict, Optional
from dataclasses import dataclass, asdict

import pandas as pd


위에서 정의한 HolySheepTardisClient와 DeribitDataQualityValidator 재사용

from holy_sheep_tardis import HolySheepTardisClient
from data_quality_validator import DeribitDataQualityValidator

@dataclass
class BacktestReproducibilityEvidence:
    """백테스트 재현 증거"""
    pipeline_id: str
    execution_timestamp: str
    backtest_period: Tuple[str, str]
    
    # 데이터 수집 증거
    data_collection_version: str
    total_data_requests: int
    data_sources: List[str]
    data_evidence_hash: str
    
    # 파이프라인 증거
    pipeline_config_hash: str
    code_version: str
    seed_data_hash: str
    
    # 품질 증거
    quality_reports: List[Dict]
    avg_quality_score: float
    min_quality_score: float
    
    # 최종 증거
    final_evidence_hash: str
    chain_verified: bool

class BacktestReproducibilityPipeline:
    """
    Deribit 옵션 백테스트 재현 증거 수집 파이프라인
    HolySheep Tardis를 통한 완전한 데이터 추적성 확보
    """
    
    def __init__(self, api_key: str, db_path: str = "backtest_evidence.db"):
        self.client = HolySheepTardisClient(api_key)
        self.validator = DeribitDataQualityValidator()
        self.db_path = db_path
        self.pipeline_version = "bp-v1.0.20241205"
        
        self._setup_logging()
        self._init_database()
    
    def _setup_logging(self):
        """로깅 설정"""
        logging.basicConfig(
            level=logging.INFO,
            format="%(asctime)s - %(levelname)s - %(message)s"
        )
        self.logger = logging.getLogger(__name__)
    
    def _init_database(self):
        """SQLite 데이터베이스 초기화"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS backtest_evidence (
                pipeline_id TEXT PRIMARY KEY,
                execution_timestamp TEXT,
                backtest_period_start TEXT,
                backtest_period_end TEXT,
                data_evidence_hash TEXT,
                pipeline_config_hash TEXT,
                final_evidence_hash TEXT,
                avg_quality_score REAL,
                quality_reports_json TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS request_logs (
                log_id INTEGER PRIMARY KEY AUTOINCREMENT,
                pipeline_id TEXT,
                request_id TEXT,
                timestamp TEXT,
                instrument TEXT,
                latency_ms REAL,
                response_hash TEXT,
                data_points INTEGER,
                FOREIGN KEY (pipeline_id) REFERENCES backtest_evidence(pipeline_id)
            )
        """)
        
        conn.commit()
        conn.close()
    
    def run_daily_backtest_pipeline(
        self,
        instruments: List[str],
        start_date: datetime,
        end_date: datetime,
        save_local: bool = True
    ) -> BacktestReproducibilityEvidence:
        """
        일일 백테스트 재현 증거 수집 파이프라인 실행
        
        Args:
            instruments: Deribit 옵션 심볼 목록
            start_date: 백테스트 시작 일시
            end_date: 백테스트 종료 일시
            save_local: 로컬 저장 여부
        
        Returns:
            BacktestReproducibilityEvidence: 완전한 재현 증거
        """
        pipeline_id = hashlib.sha256(
            f"{datetime.utcnow().isoformat()}{len(instruments)}".encode()
        ).hexdigest()[:16]
        
        self.logger.info(f"백테스트 파이프라인 시작: {pipeline_id}")
        
        # 1단계: Deribit 옵션 데이터 수집
        collected_results = self._collect_options_data(
            pipeline_id, instruments, start_date, end_date
        )
        
        # 2단계: 데이터 품질 검증
        quality_reports = self._validate_batch_quality(collected_results)
        
        # 3단계: 재현 증거 생성
        evidence = self._generate_evidence(
            pipeline_id, collected_results, quality_reports,
            start_date, end_date
        )
        
        # 4단계: 증거 저장
        if save_local:
            self._save_evidence_to_db(evidence, pipeline_id)
            self._save_evidence_to_file(evidence)
        
        self.logger.info(f"백테스트 파이프라인 완료: {pipeline_id}")
        self.logger.info(f"평균 품질 점수: {evidence.avg_quality_score:.2f}")
        self.logger.info(f"최종 증거 해시: {evidence.final_evidence_hash[:16]}...")
        
        return evidence
    
    def _collect_options_data(
        self,
        pipeline_id: str,
        instruments: List[str],
        start_date: datetime,
        end_date: datetime
    ) -> List[Dict]:
        """Deribit 옵션 데이터 수집"""
        results = []
        
        for instrument in instruments:
            self.logger.info(f"데이터 수집 중: {instrument}")
            
            result = self.client.get_deribit_options_history(
                instrument=instrument,
                start_date=start_date,
                end_date=end_date,
                depth=10
            )
            
            results.append(result)
            
            # HolySheep Tardis 지연 시간 검증
            if result.get("success"):
                latency = result["meta"]["latency_ms"]
                self.logger.info(f"  → 지연 시간: {latency}ms")
        
        return results
    
    def _validate_batch_quality(
        self,
        results: List[Dict]
    ) -> List[Dict]:
        """배치 데이터 품질 검증"""
        reports = self.validator.batch_validate(results)
        
        quality_list = []
        for instrument, report in reports.items():
            self.logger.info(f"품질 검증: {instrument} - {report.quality_grade}")
            quality_list.append(asdict(report))
        
        return quality_list
    
    def _generate_evidence(
        self,
        pipeline_id: str,
        results: List[Dict],
        quality_reports: List[Dict],
        start_date: datetime,
        end_date: datetime
    ) -> BacktestReproducibilityEvidence:
        """재현 증거 생성"""
        # 데이터 수집 증거 해시
        data_hashes = []
        for result in results:
            if result.get("success"):
                data_hashes.append(result["meta"]["response_hash"])
        
        data_evidence_hash = hashlib.sha256(
            "".join(data_hashes).encode()
        ).hexdigest()
        
        # 파이프라인 설정 해시
        config_data = {
            "pipeline_version": self.pipeline_version,
            "data_client_version": self.client.version,
            "instruments": [r["meta"]["instrument"] for r in results if r.get("success")]
        }
        pipeline_config_hash = hashlib.sha256(
            json.dumps(config_data, sort_keys=True).encode()
        ).hexdigest()
        
        # 품질 점수 통계
        quality_scores = [r["overall_score"] for r in quality_reports]
        avg_quality = sum(quality_scores) / len(quality_scores) if quality_scores else 0
        min_quality = min(quality_scores) if quality_scores else 0
        
        # 최종 증거 해시 (체인)
        final_evidence_data = {
            "pipeline_id": pipeline_id,
            "data_evidence_hash": data_evidence_hash,
            "pipeline_config_hash": pipeline_config_hash,
            "avg_quality_score": avg_quality,
            "quality_reports": quality_reports
        }
        final_evidence_hash = hashlib.sha256(
            json.dumps(final_evidence_data, sort_keys=True).encode()
        ).hexdigest()
        
        return BacktestReproducibilityEvidence(
            pipeline_id=pipeline_id,
            execution_timestamp=datetime.utcnow().isoformat(),
            backtest_period=(start_date.isoformat(), end_date

관련 리소스
관련 문서