암호화폐 시장における 변동성은 전통 금융보다 훨씬 높으며, 선물 포지션이나 거래소 리스크를 관리하기 위해서는 정량적인 리스크 측정 도구가 필수적입니다. 본 튜토리얼에서는 Value at Risk(VaR)의 가장 직관적인 계산 방법인 히스토리컬 시뮬레이션을 암호화폐 데이터에 적용하는 방법을 다룹니다. HolySheep AI의低成本 API를 활용하여 실시간 포트폴리오 리스크 모니터링 시스템을 구축해보겠습니다.

HolySheep vs 공식 API vs 타 릴레이 서비스 비교

비교 항목 HolySheep AI 공식 OpenAI API 공식 Anthropic API 타 릴레이 서비스
GPT-4.1 가격 $8.00/MTok $8.00/MTok N/A $9-12/MTok
Claude Sonnet 4.5 $15.00/MTok N/A $15.00/MTok $16-18/MTok
Gemini 2.5 Flash $2.50/MTok N/A N/A $3-5/MTok
DeepSeek V3.2 $0.42/MTok N/A N/A $0.50-0.80/MTok
결제 방식 로컬 결제 지원 (해외 신용카드 불필요) 해외 신용카드 필수 해외 신용카드 필수 다양하나 제한적
베이직 인증 지원 API Key만 API Key만 다양
리스크 분석용 모델 다중 모델 통합 단일 단일 제한적
免费 크레딧 가입 시 제공 $5 쿠폰 제한적 드묵

VaR과 히스토리컬 시뮬레이션 이해하기

Value at Risk(VaR)란?

VaR은 "특정 신뢰 수준에서 특정 기간 동안 예상되는 최대 손실"을 의미합니다. 예를 들어, "1일 VaR 95% = $10,000"이라면, "내일 95% 확률로 손실이 $10,000를 넘지 않는다"는 뜻입니다. 암호화폐 같이 변동성이 큰 자산에서는 이 수치가 매우 중요합니다.

히스토리컬 시뮬레이션의 장점

아키텍처 개요

본 튜토리얼에서 구축하는 시스템의 구조는 다음과 같습니다:

┌─────────────────────────────────────────────────────────────────┐
│                    암호화폐 VaR 리스크 시스템                      │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐       │
│  │   Tardis    │───▶│   Python     │───▶│  HolySheep   │       │
│  │  Crypto     │    │  VaR Engine   │    │   AI API     │       │
│  │  Data Feed  │    │              │    │  (리스크     │       │
│  │             │    │  - 데이터 수신 │    │   분석용)    │       │
│  │  - BTC/USDT │    │  - 수익률 계산 │    │             │       │
│  │  - ETH/USDT │    │  - VaR 계산   │    │  - 자연어    │       │
│  │  - 시세 데이터│    │  - 스트레스테스트│    │    리포트   │       │
│  └──────────────┘    └──────────────┘    └──────────────┘       │
│                              │                    │             │
│                              ▼                    ▼             │
│                      ┌──────────────┐    ┌──────────────┐       │
│                      │   대시보드    │    │   알림       │       │
│                      │  (시각화)    │    │  시스템      │       │
│                      └──────────────┘    └──────────────┘       │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

필수 라이브러리 설치

# VaR 계산 및 데이터 분석
pip install pandas numpy scipy matplotlib seaborn

암호화폐 데이터 수집

pip install tardisgrpc pandas-gbq

HolySheep AI API 통신

pip install openai httpx

실시간 데이터 및 웹

pip install asyncio aiohttp websockets

핵심 구현 코드

1. Tardis 데이터 연동 및 수익률 계산

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from openai import OpenAI
import os

============================================================

HolySheep AI 클라이언트 설정

============================================================

client = OpenAI( api_key="YOUR_HOLYSHEEP_API_KEY", base_url="https://api.holysheep.ai/v1" )

============================================================

Tardis Market Data Client 설정

============================================================

try: import tardisgrpc tardis_client = tardisgrpc.Client() TARDIS_AVAILABLE = True except ImportError: TARDIS_AVAILABLE = False print("[경고] Tardis gRPC 라이브러리가 없습니다. 샘플数据进行使用합니다.") class CryptoVaREngine: """ 히스토리컬 시뮬레이션 기반 암호화폐 VaR 계산 엔진 """ def __init__(self, confidence_level=0.95, holding_period=1): self.confidence_level = confidence_level self.holding_period = holding_period self.alpha = 1 - confidence_level def fetch_historical_prices(self, symbols, days=252): """ Tardis에서 과거 시세 데이터 수신 """ if not TARDIS_AVAILABLE: # 샘플 데이터 반환 (실제 구현 시 Tardis 사용) return self._generate_sample_data(symbols, days) end_date = datetime.utcnow() start_date = end_date - timedelta(days=days) price_data = {} for symbol in symbols: try: exchange_name = "binance" market = f"{symbol}-perp" if not symbol.endswith("-PERP") else symbol candles = tardis_client.get_historical_recent( exchange=exchange_name, market=market, start_time=int(start_date.timestamp() * 1000), end_time=int(end_date.timestamp() * 1000), resolution="1d" ) df = pd.DataFrame(candles) df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') df.set_index('timestamp', inplace=True) price_data[symbol] = df['close'] except Exception as e: print(f"[오류] {symbol} 데이터 수신 실패: {e}") continue # 모든 심볼을 단일 DataFrame으로 병합 prices = pd.DataFrame(price_data) prices.index = pd.to_datetime(prices.index).tz_localize(None) return prices.dropna() def _generate_sample_data(self, symbols, days): """ 샘플 데이터 생성 (Tardis 미사용 시) """ np.random.seed(42) dates = pd.date_range(end=datetime.now(), periods=days, freq='D') data = {} base_prices = {'BTC': 45000, 'ETH': 2500, 'SOL': 100} for symbol in symbols: # 기하 브라운 운동으로 시뮬레이션 mu = 0.0005 # 일간 드리프트 sigma = 0.03 if symbol == 'BTC' else 0.05 if symbol == 'ETH' else 0.08 base = base_prices.get(symbol, 1000) returns = np.random.normal(mu, sigma, days) prices = base * np.exp(np.cumsum(returns)) data[symbol] = prices df = pd.DataFrame(data, index=dates) return df def calculate_returns(self, prices): """ 대수 수익률 (log returns) 계산 """ log_returns = np.log(prices / prices.shift(1)) return log_returns.dropna() def calculate_var_historical(self, returns, portfolio_weights=None): """ 히스토리컬 시뮬레이션 VaR 계산 Parameters: - returns: 수익률 DataFrame - portfolio_weights: 포트폴리오 비중 (None이면 동일 비중) Returns: - VaR 값 (양수 = 손실 절대값) """ if portfolio_weights is None: portfolio_weights = np.ones(returns.shape[1]) / returns.shape[1] # 포트폴리오 수익률 계산 portfolio_returns = (returns * portfolio_weights).sum(axis=1) # 신뢰수준에 해당하는 분위수 계산 var_percentile = self.alpha * 100 var_value = np.percentile(portfolio_returns, var_percentile) # VaR은 손실이므로 음수 → 양수로 변환 return abs(var_value) def calculate_cvar(self, returns, portfolio_weights=None): """ Conditional VaR (Expected Shortfall) 계산 VaR를 초과하는 손실의 평균값 """ if portfolio_weights is None: portfolio_weights = np.ones(returns.shape[1]) / returns.shape[1] portfolio_returns = (returns * portfolio_weights).sum(axis=1) var_threshold = np.percentile(portfolio_returns, self.alpha * 100) # VaR 이하 수익률만 필터링 tail_losses = portfolio_returns[portfolio_returns <= var_threshold] return abs(tail_losses.mean())

============================================================

실행 예제

============================================================

if __name__ == "__main__": # VaR 엔진 초기화 (95% 신뢰수준, 1일 보유기간) var_engine = CryptoVaREngine(confidence_level=0.95, holding_period=1) # 분석 대상 암호화폐 symbols = ['BTC', 'ETH', 'SOL'] # 과거 시세 데이터 수신 (252일 = 약 1년) print(">>> Tardis에서 데이터 수신 중...") prices = var_engine.fetch_historical_prices(symbols, days=252) print(f" 데이터 수신 완료: {prices.shape[0]}일, {prices.shape[1]}개 자산") # 수익률 계산 returns = var_engine.calculate_returns(prices) print(f">>> 수익률 계산 완료: {returns.shape[0]}개 관측값") # 포트폴리오 비중 설정 (BTC 50%, ETH 30%, SOL 20%) weights = np.array([0.50, 0.30, 0.20]) # VaR 및 CVaR 계산 var_95 = var_engine.calculate_var_historical(returns, weights) cvar_95 = var_engine.calculate_cvar(returns, weights) print(f"\n===== VaR 리스크 리포트 (95% 신뢰수준) =====") print(f"1-Day VaR: {var_95:.4f} ({var_95*100:.2f}%)") print(f"1-Day CVaR: {cvar_95:.4f} ({cvar_95*100:.2f}%)") print(f"투자금 $100,000 기준 예상 최대 일일 손실:") print(f" VaR: ${100000 * var_95:,.2f}") print(f" CVaR: ${100000 * cvar_95:,.2f}")

2. HolySheep AI를 활용한 자연어 리스크 분석

import json
from openai import OpenAI


class RiskAnalysisAssistant:
    """
    HolySheep AI를 활용하여 VaR 결과를 자연어로 해석
    """
    
    def __init__(self, api_key):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://api.holysheep.ai/v1"
        )
    
    def generate_risk_report(self, var_metrics, portfolio_details):
        """
        VaR 측정 결과를 기반으로 상세 리스크 리포트 생성
        
        Parameters:
        - var_metrics: VaREngine에서 계산된 VaR 결과 딕셔너리
        - portfolio_details: 포트폴리오 상세 정보
        """
        
        prompt = f"""
당신은 전문 암호화폐 리스크 분석가입니다. 아래 VaR 측정 결과를 바탕으로 
투자자에게 명확하고 실행 가능한 리스크 리포트를 작성해주세요.

현재 포트폴리오 구성

{json.dumps(portfolio_details, indent=2, ensure_ascii=False)}

VaR 측정 결과

- 신뢰수준: {var_metrics['confidence_level']*100}% - 1-Day VaR: {var_metrics['var_1d']*100:.2f}% - 1-Day CVaR (Expected Shortfall): {var_metrics['cvar_1d']*100:.2f}% - 10-Day VaR: {var_metrics['var_10d']*100:.2f}% - 최대 손실 발생일: {var_metrics['worst_day']} - 수익률 표준편차(연간화): {var_metrics['annual_volatility']*100:.2f}%

요청 사항

1. 리스크 등급 평가 (1-5단계) 2. 현재 포트폴리오의 주요 리스크 요인 3. 역사적 최대 손실 events와의 비교 4. 리스크 완화 권장사항 5. 다음 보고일까지 주의해야 할 시장 상황 한국어로 전문적인 리스크 리포트를 작성해주세요. """ response = self.client.chat.completions.create( model="gpt-4.1", messages=[ { "role": "system", "content": "당신은 탑티어 투자은행의 전문 리스크 관리 분석가입니다." }, { "role": "user", "content": prompt } ], temperature=0.3, max_tokens=1500 ) return response.choices[0].message.content def analyze_stress_scenario(self, current_portfolio, market_conditions): """ 지정된 스트레스 시나리오 분석 """ scenarios = { "black_thursday": { "description": "2020년 3월 12일 같은 급락 상황", "btc_shock": -50, "eth_shock": -60, "correlation_spike": True }, "ftx_collapse": { "description": "거래소 위기 상황", "btc_shock": -30, "eth_shock": -40, "correlation_spike": True }, "bull_market_correction": { "description": "강세장 조정局面", "btc_shock": -20, "eth_shock": -25, "correlation_spike": False } } results = {} for scenario_name, params in scenarios.items(): # 스트레스 손실 계산 portfolio_value = current_portfolio['total_value'] weighted_shock = ( current_portfolio['btc_weight'] * params['btc_shock'] + current_portfolio['eth_weight'] * params['eth_shock'] + current_portfolio.get('sol_weight', 0) * params.get('sol_shock', -30) ) / 100 stress_loss = portfolio_value * weighted_shock stress_loss_pct = weighted_shock * 100 results[scenario_name] = { "description": params['description'], "projected_loss": stress_loss, "loss_percentage": stress_loss_pct, "remaining_value": portfolio_value - stress_loss } return results def compare_with_benchmark(self, portfolio_var, benchmark_var=0.05): """ 벤치마크 대비 포트폴리오 리스크 비교 """ prompt = f""" 암호화폐 포트폴리오의 VaR를 전통적 금융 벤치마크와 비교 분석해주세요.

분석 대상

- 포트폴리오 1-Day VaR: {portfolio_var*100:.2f}% - 벤치마크 VaR (S&P 500 역사적 평균): {benchmark_var*100:.2f}%

비교 항목

1. 변동성 비율 2. 리스크 조정 수익률 의미 3. 헤지 필요성 여부 4. 장기 투자 적합성 한국어로 간결하고 명확하게 분석해주세요. """ response = self.client.chat.completions.create( model="claude-sonnet-4-20250514", messages=[ {"role": "user", "content": prompt} ], temperature=0.2, max_tokens=800 ) return { "var_ratio": portfolio_var / benchmark_var, "analysis": response.choices[0].message.content }

============================================================

실행 예제

============================================================

if __name__ == "__main__": # HolySheep AI 클라이언트 초기화 risk_assistant = RiskAnalysisAssistant(api_key="YOUR_HOLYSHEEP_API_KEY") # VaR 측정 결과 var_metrics = { 'confidence_level': 0.95, 'var_1d': 0.035, 'cvar_1d': 0.052, 'var_10d': 0.035 * np.sqrt(10), # 스케일링 근사 'worst_day': '2024-03-15', 'annual_volatility': 0.75 } # 포트폴리오 구성 portfolio_details = { 'total_value_usd': 100000, 'assets': [ {'symbol': 'BTC', 'weight': 0.50, 'value': 50000}, {'symbol': 'ETH', 'weight': 0.30, 'value': 30000}, {'symbol': 'SOL', 'weight': 0.20, 'value': 20000} ], 'rebalancing_frequency': 'weekly', 'risk_tolerance': 'moderate' } # 리스크 리포트 생성 print(">>> HolySheep AI가 리스크 리포트를 생성중입니다...") report = risk_assistant.generate_risk_report(var_metrics, portfolio_details) print("\n" + "="*60) print("리스크 분석 리포트") print("="*60) print(report) # 스트레스 테스트 print("\n>>> 스트레스 시나리오 분석...") stress_results = risk_assistant.analyze_stress_scenario( current_portfolio={ 'total_value': 100000, 'btc_weight': 0.50, 'eth_weight': 0.30, 'sol_weight': 0.20 }, market_conditions=None ) for scenario, result in stress_results.items(): print(f"\n[{scenario.upper()}]") print(f" 설명: {result['description']}") print(f" 예상 손실: ${result['projected_loss']:,.2f} ({result['loss_percentage']:.1f}%)") print(f" 잔여 가치: ${result['remaining_value']:,.2f}")

3. 실시간 VaR 모니터링 대시보드

import streamlit as st
import pandas as pd
import numpy as np
from datetime import datetime
import plotly.express as px
import plotly.graph_objects as go


def render_var_dashboard(var_engine, latest_prices, portfolio_weights):
    """
    Streamlit 기반 VaR 모니터링 대시보드
    """
    
    st.set_page_config(page_title="암호화폐 VaR 모니터링", layout="wide")
    
    # 헤더
    st.title("🚀 암호화폐 VaR 리스크 대시보드")
    st.markdown(f"**업데이트:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    # 핵심 메트릭
    returns = var_engine.calculate_returns(latest_prices)
    var_95 = var_engine.calculate_var_historical(returns, portfolio_weights)
    cvar_95 = var_engine.calculate_cvar(returns, portfolio_weights)
    var_99 = var_engine.calculate_var_historical(
        returns, portfolio_weights
    ) * 1.28  # 99% 근사
    
    col1, col2, col3, col4 = st.columns(4)
    
    with col1:
        st.metric(
            "1-Day VaR (95%)",
            f"{var_95*100:.2f}%",
            delta=f"${100000 * var_95:,.0f}"
        )
    
    with col2:
        st.metric(
            "1-Day CVaR (95%)",
            f"{cvar_95*100:.2f}%",
            delta=f"${100000 * cvar_95:,.0f}"
        )
    
    with col3:
        st.metric(
            "1-Day VaR (99%)",
            f"{var_99*100:.2f}%",
            delta=f"${100000 * var_99:,.0f}"
        )
    
    with col4:
        st.metric(
            "연간 변동성",
            f"{returns.std() * np.sqrt(365) * 100:.1f}%",
            delta=" annualized"
        )
    
    # 수익률 분포 차트
    st.subheader("📊 포트폴리오 수익률 분포")
    
    portfolio_returns = (returns * portfolio_weights).sum(axis=1)
    
    fig = px.histogram(
        portfolio_returns,
        nbins=50,
        title="일간 수익률 분포 (히스토그램)",
        labels={'value': '수익률', 'count': '빈도'}
    )
    
    # VaR 임계값 표시
    var_95_value = np.percentile(portfolio_returns, 5)
    var_99_value = np.percentile(portfolio_returns, 1)
    
    fig.add_vline(
        x=var_95_value, 
        line_dash="dash", 
        line_color="red",
        annotation_text=f"VaR 95%: {var_95_value:.2%}"
    )
    fig.add_vline(
        x=var_99_value, 
        line_dash="dash", 
        line_color="darkred",
        annotation_text=f"VaR 99%: {var_99_value:.2%}"
    )
    
    fig.add_vline(x=0, line_color="green", line_width=2)
    
    st.plotly_chart(fig, use_container_width=True)
    
    # 시간에 따른 VaR 추이
    st.subheader("📈 VaR 추이 (Rolling Window)")
    
    rolling_var = portfolio_returns.rolling(window=21).apply(
        lambda x: abs(np.percentile(x, 5))
    )
    
    fig2 = go.Figure()
    fig2.add_trace(go.Scatter(
        x=rolling_var.index,
        y=rolling_var * 100,
        mode='lines',
        name='21-Day Rolling VaR',
        line=dict(color='royalblue', width=2)
    ))
    
    fig2.update_layout(
        title="Rolling VaR (21일 윈도우)",
        xaxis_title="날짜",
        yaxis_title="VaR (%)",
        template="plotly_white"
    )
    
    st.plotly_chart(fig2, use_container_width=True)
    
    # 자산별 기여도
    st.subheader("💼 VaR 기여도 (자산별)")
    
    marginal_contribution = []
    for i, symbol in enumerate(latest_prices.columns):
        asset_return = returns[symbol]
        asset_var = abs(np.percentile(asset_return, 5))
        weight = portfolio_weights[i]
        
        marginal_contribution.append({
            'Asset': symbol,
            'VaR Contribution': asset_var * weight / var_95 * 100,
            'Weight': weight * 100
        })
    
    df_contrib = pd.DataFrame(marginal_contribution)
    
    fig3 = px.bar(
        df_contrib,
        x='Asset',
        y='VaR Contribution',
        color='Weight',
        title="자산별 VaR 기여도",
        labels={'VaR Contribution': 'VaR 기여도 (%)', 'Asset': '자산'}
    )
    
    st.plotly_chart(fig3, use_container_width=True)


============================================================

실행

============================================================

if __name__ == "__main__": import sys sys.argv.append("--server.headless=true") # 샘플 데이터 np.random.seed(42) dates = pd.date_range(end=datetime.now(), periods=252, freq='D') sample_prices = pd.DataFrame({ 'BTC': 45000 * np.exp(np.cumsum(np.random.normal(0.001, 0.03, 252))), 'ETH': 2500 * np.exp(np.cumsum(np.random.normal(0.001, 0.05, 252))), 'SOL': 100 * np.exp(np.cumsum(np.random.normal(0.002, 0.08, 252))) }, index=dates) weights = np.array([0.50, 0.30, 0.20]) # 대시보드 실행 var_engine = CryptoVaREngine() render_var_dashboard(var_engine, sample_prices, weights)

HolySheep AI를 선택해야 하는 이유

1. 비용 효율성

암호화폐 리스크 분석 시스템은 다음과 같은 다중 모델 활용이 필요합니다:

HolySheep AI는 단일 API 키로 모든 모델을 통합 관리하며, 월간 사용량에 따른 볼륨 할인을 제공합니다. 월 $500 이상 사용 시 추가 할인을 받을 수 있습니다.

2. 로컬 결제 지원

저와 같이 해외 신용카드 발급이 어려운 개발자에게 HolySheep AI의 로컬 결제 옵션은 큰 장점입니다. 국내 계좌이체, 신용카드, 가상계좌 등 다양한 결제 수단을 지원합니다.

3. 안정적인 연결

암호화폐 시장은 24/7 운영되며, 리스크 모니터링 시스템은 항상 가용해야 합니다. HolySheep AI는 99.9% 이상의 가용성을 보장하며, 자동 장애 조치(failover) 기능을 제공합니다.

4. 베이직 인증 지원

엔터프라이즈 환경에서 자주 사용되는 베이직 인증을 지원하므로, 기존 인프라와의 통합이 원활합니다.

자주 발생하는 오류와 해결책

오류 1: Tardis 연결 실패 - "Connection timeout"

# ❌ 오류 메시지
tardisgrpc.TimeoutException: Connection timeout after 30 seconds

원인

- 네트워크 방화벽 - Tardis 서버 과부하 - 잘못된 엔드포인트 설정

✅ 해결책

import tardisgrpc

타임아웃 설정 증가 및 재시도 로직 추가

class TardisClientWithRetry: def __init__(self, max_retries=3, timeout=60): self.max_retries = max_retries self.timeout = timeout self.client = tardisgrpc.Client() def fetch_with_retry(self, exchange, market, start_time, end_time): import time for attempt in range(self.max_retries): try: return self.client.get_historical_recent( exchange=exchange, market=market, start_time=start_time, end_time=end_time, resolution="1d" ) except tardisgrpc.TimeoutException as e: if attempt == self.max_retries - 1: raise wait_time = 2 ** attempt # 지수 백오프 print(f"[경고] 시도 {attempt+1} 실패, {wait_time}초 후 재시도...") time.sleep(wait_time)

사용 예시

tardis = TardisClientWithRetry(max_retries=5, timeout=120) candles = tardis.fetch_with_retry( exchange="binance", market="BTC-PERP", start_time=int((datetime.now() - timedelta(days=365)).timestamp() * 1000), end_time=int(datetime.now().timestamp() * 1000) )

오류 2: HolySheep API 키 인증 실패

# ❌ 오류 메시지
openai.AuthenticationError: Invalid API key provided

원인

- 잘못된 API 키 형식 - 환경변수 미설정 - 키 만료 또는 무효화

✅ 해결책

import os from openai import OpenAI

방법 1: 환경변수 설정 (권장)

os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"

방법 2: 직접 인자 전달

client = OpenAI( api_key=os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"), base_url="https://api.holysheep.ai/v1" )

방법 3: 키 유효성 검증

def validate_api_key(api_key, base_url): test_client = OpenAI(api_key=api_key, base_url=base_url) try: # 최소한의 호출로 키 검증 test_client.models.list() return True except Exception as e: print(f"[오류] API 키 검증 실패: {e}") return False

사용 전 검증

API_KEY = "YOUR_HOLYSHEEP_API_KEY" BASE_URL = "https://api.holysheep.ai/v1" if validate_api_key(API_KEY, BASE_URL): print("[확인] API 키가 유효합니다. 계속 진행합니다.") else: raise ValueError("유효하지 않은 API 키입니다. HolySheep 대시보드에서 키를 확인하세요.")

오류 3: VaR 계산 시 데이터 불일치 - "Length mismatch"

# ❌ 오류 메시지
ValueError: operands could not be broadcast together with shapes (250,) (252,)

원인

- 수익률 계산 시 발생하는 NA 값 - 여러 자산의 데이터 길이 불일치 - 타임스탬프 정렬 문제

✅ 해결책

class CryptoVaREngine: # ... 기존 코드 ... def calculate_returns(self, prices): """ 대수 수익률 계산 + 결측치 처리 """ # 가격 데이터 정렬 및 결측치 확인 prices = prices.sort_index() # Forward fill 후 Backward fill prices = prices.ffill().bfill() # 모든 자산의 데이터 길이 확인 min_length = prices.notna().all(axis=1).sum() if min_length < len(prices) * 0.9: # 90% 이상 결측 시 경고 print(f"[경고] 데이터 결측 비율이 높습니다: {min_length}/{len(prices)}") # 수익률 계산 log_returns = np.log(prices / prices.shift(1)) # 결측치 처리: 직전 값으로 채우기 log_returns = log_returns.fillna(method='ffill').fillna(0) # 첫 번째 행(NA) 제거 log_returns = log_returns.iloc[1:] return log_returns def validate_data_integrity(self,