암호화폐 시장에서 비트코인(BTC) 변동성 예측은 거래 전략, 리스크 관리, 파생상품 가격 결정에 핵심적인 역할을 합니다. 본 튜토리얼에서는 Tardis 데이터(고품질 암호화폐 시장 데이터)를 활용하여 GARCH(Generalized Autoregressive Conditional Heteroskedasticity) 전통 통계 방법과 머신러닝 기반 예측 방법을 직접 비교하고, HolySheep AI를 통해 두 접근법의 강점을 결합하는 실전 프로젝트를 진행하겠습니다.
Tardis 데이터란?
Tardis는加密화폐 시장 데이터를 제공하는 서비스로, Binance, Bybit, OKX 등 주요 거래소에서 수집한 고빈도 거래 데이터(OHLCV, 주문서 데이터, 거래 내역)를 제공합니다. 본 튜토리얼에서는 Tardis에서 제공하는 BTC/USDT 1분봉 데이터를 사용하여 변동성 예측 모델을 구축하겠습니다.
저자의 경험담: 처음 암호화폐 데이터 분석을 시작했을 때, 여러 거래소의 API를 직접 연동하는 것이 정말 번거로웠습니다. Tardis를 사용하면 단일 API로 여러 거래소의 데이터를 일관된 포맷으로 얻을 수 있어 분석 시간을 크게 단축할 수 있었습니다. 특히 변동성 연구에서는 데이터 품질이 예측 정확도에 직접적인 영향을 미치기 때문에, Tardis 같은 전문 데이터 제공자를 사용하는 것이 필수적입니다.
프로젝트 환경 설정
필수 라이브러리 설치
# requirements.txt
pandas>=2.0.0
numpy>=1.24.0
scipy>=1.10.0
arch>=6.0.0
scikit-learn>=1.3.0
matplotlib>=3.7.0
requests>=2.28.0
ta-lib>=0.4.28
holysheep-sdk>=1.0.0 # HolySheep AI 공식 SDK# 라이브러리 설치
pip install pandas numpy scipy arch scikit-learn matplotlib requests
pip install ta-lib # TA-Lib는 별도 설치 필요Tardis API 데이터 수집
import requests
import pandas as pd
from datetime import datetime, timedelta
import time
class TardisDataCollector:
"""
Tardis API를 사용하여 암호화폐 시장 데이터 수집
문서: https://docs.tardis.dev/
"""
def __init__(self, api_token: str):
self.api_token = api_token
self.base_url = "https://api.tardis.dev/v1"
self.headers = {
"Authorization": f"Bearer {api_token}",
"Content-Type": "application/json"
}
def get_btc_usdt_1min_bars(
self,
exchange: str = "binance",
start_date: str = "2024-01-01",
end_date: str = "2024-06-01"
) -> pd.DataFrame:
"""
BTC/USDT 1분봉 OHLCV 데이터 수집
Args:
exchange: 거래소 이름 (binance, bybit, okx)
start_date: 시작 날짜 (YYYY-MM-DD)
end_date: 종료 날짜 (YYYY-MM-DD)
Returns:
pandas DataFrame with OHLCV data
"""
url = f"{self.base_url}/historical/{exchange}/spot/btcusdt/trades"
# 1분봉을 생성하기 위해 trades 데이터 수집
params = {
"from": start_date,
"to": end_date,
"format": "dataframe",
"columns": ["timestamp", "price", "amount", "side"],
"limit": 100000 # 한 번에 최대 100만 레코드
}
all_trades = []
current_start = start_date
while current_start < end_date:
params["from"] = current_start
next_date = (datetime.strptime(current_start, "%Y-%m-%d") +
timedelta(days=7)).strftime("%Y-%m-%d")
if next_date > end_date:
params["to"] = end_date
else:
params["to"] = next_date
try:
response = requests.get(url, headers=self.headers, params=params)
response.raise_for_status()
trades = response.json()
if isinstance(trades, list) and len(trades) > 0:
all_trades.extend(trades)
# API rate limit 방지
time.sleep(0.5)
current_start = next_date
print(f"수집 진행률: {current_start}까지 {len(all_trades)}건")
except requests.exceptions.RequestException as e:
print(f"API 오류: {e}")
break
# trades 데이터를 1분봉으로 변환
df = pd.DataFrame(all_trades)
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
# 1분봉(OHLCV) 생성
ohlcv = df.resample("1min").agg({
"price": ["first", "max", "min", "last"],
"amount": "sum"
})
ohlcv.columns = ["open", "high", "low", "close", "volume"]
ohlcv = ohlcv.dropna()
return ohlcv
사용 예시
tardis = TardisDataCollector(api_token="YOUR_TARDIS_API_TOKEN")
btc_1min = tardis.get_btc_usdt_1min_bars(
exchange="binance",
start_date="2024-01-01",
end_date="2024-06-01"
)
print(f"수집된 데이터: {len(btc_1min)}건")
print(btc_1min.head())GARCH 모델: 전통적 변동성 예측
GARCH(1,1) 모델 원리
GARCH 모델은 금융 시계열의 변동성 군집(volatility clustering) 현상을 모델링하는 데 가장 널리 사용되는 방법론입니다. 기본 GARCH(1,1) 모델은 다음과 같습니다:
rt = σt * εt (수익률 방정식)
σ²t = ω + α * ε²t-1 + β * σ²t-1 (조건부 분산 방정식)
여기서:
- rt: 수익률
- σt: 조건부 표준편차 (변동성)
- εt: 표준정규분포 오차항
- ω, α, β: 추정 파라미터 (ω > 0, α ≥ 0, β ≥ 0, α + β < 1)GARCH 모델 구현
import numpy as np
from arch import arch_model
from sklearn.metrics import mean_squared_error, mean_absolute_error
import warnings
warnings.filterwarnings('ignore')
class GARCHVolatilityModel:
"""
GARCH 모델을 사용한 BTC 변동성 예측
arch 라이브러리 사용
"""
def __init__(self, p: int = 1, q: int = 1, model_type: str = "GARCH"):
"""
Args:
p: GARCH 차수 (滞后된 분산항)
q: ARCH 차수 (滞后된 수익률 제곱항)
model_type: "GARCH", "EGARCH", "GJR-GARCH"
"""
self.p = p
self.q = q
self.model_type = model_type
self.model = None
self.result = None
def prepare_returns(self, prices: pd.Series) -> pd.Series:
"""가격 데이터에서 로그 수익률 계산"""
returns = np.log(prices / prices.shift(1)).dropna() * 100
returns = returns.replace([np.inf, -np.inf], np.nan).dropna()
return returns
def fit(self, prices: pd.Series):
"""GARCH 모델 학습"""
returns = self.prepare_returns(prices)
# arch 모델 생성
if self.model_type == "GARCH":
self.model = arch_model(
returns,
vol="Garch",
p=self.p,
q=self.q,
dist="normal"
)
elif self.model_type == "EGARCH":
self.model = arch_model(
returns,
vol="EGARCH",
p=self.p,
q=self.q,
dist="normal"
)
elif self.model_type == "GJR-GARCH":
self.model = arch_model(
returns,
vol="GARCH",
p=self.p,
q=self.q,
dist="normal",
o=1 # leverage term
)
# 모델 피팅
self.result = self.model.fit(disp="off", options={"maxiter": 1000})
return self.result
def forecast(self, horizon: int = 60) -> dict:
"""미래 변동성 예측 (기본값: 60분)"""
if self.result is None:
raise ValueError("모델이 먼저 학습되어야 합니다")
forecast = self.result.forecast(horizon=horizon)
# annualized volatility로 변환 (1분봉 기준)
annualized_vol = np.sqrt(forecast.variance.values[-1, :]) * np.sqrt(525600)
return {
"mean_variance": forecast.mean.values[-1, :],
"variance": forecast.variance.values[-1, :],
"volatility": np.sqrt(forecast.variance.values[-1, :]),
"annualized_volatility": annualized_vol
}
def evaluate(self, actual_prices: pd.Series, forecast_horizon: int = 60) -> dict:
"""예측 성능 평가"""
returns = self.prepare_returns(actual_prices)
# rolling forecast 방식 평가
test_size = len(returns) // 5 # 20% 테스트 데이터
train_returns = returns[:-test_size]
test_returns = returns[-test_size:]
predictions = []
actuals = []
step = forecast_horizon
for i in range(0, len(test_returns) - step, step):
train_data = train_returns.iloc[i:i+step]
test_data = test_returns.iloc[i:i+step]
model = arch_model(train_data, vol="Garch", p=1, q=1, dist="normal")
result = model.fit(disp="off")
forecast = result.forecast(horizon=step)
pred_vol = np.sqrt(forecast.variance.mean(axis=0)).mean()
actual_vol = test_data.std()
predictions.append(pred_vol)
actuals.append(actual_vol)
predictions = np.array(predictions)
actuals = np.array(actuals)
return {
"rmse": np.sqrt(mean_squared_error(actuals, predictions)),
"mae": mean_absolute_error(actuals, predictions),
"mape": np.mean(np.abs((actuals - predictions) / actuals)) * 100,
"predictions": predictions,
"actuals": actuals
}
실전 적용
garch_model = GARCHVolatilityModel(p=1, q=1, model_type="GARCH")
garch_model.fit(btc_1min["close"])
print("=" * 50)
print("GARCH(1,1) 모델 학습 결과")
print("=" * 50)
print(garch_model.result.summary().tables[1])
60분 후 변동성 예측
forecast = garch_model.forecast(horizon=60)
print(f"\n60분 후 예측 변동성(일별): {forecast['volatility'][-1]:.4f}%")
print(f"연간화 변동성: {forecast['annualized_volatility'][-1]:.2f}%")머신러닝 기반 변동성 예측
HolySheep AI를 활용한 고급 분석
머신러닝 접근법에서는 HolySheep AI의 다중 모델 통합 기능을 활용하여 GARCH 잔차 분석, 변동성 패턴 인식, 예측 앙상블을 구현하겠습니다. HolySheep AI의 단일 API 키로 GPT-4.1, Claude, Gemini 등 다양한 모델을 조합하여 더 정확한 예측을 수행할 수 있습니다.
import requests
import json
from typing import List, Dict, Tuple
class HolySheepMLAnalyzer:
"""
HolySheep AI API를 활용한 머신러닝 분석
- GPT-4.1: 복잡한 패턴 분석 및 Feature Engineering 조언
- Claude: 모델 해석 및 설명 생성
- Gemini: 빠른 실시간 예측
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def analyze_volatility_patterns(
self,
recent_data: pd.DataFrame,
lookback_hours: int = 24
) -> Dict:
"""
HolySheep GPT-4.1을 사용하여 변동성 패턴 분석
최근 데이터 기반 시장 상황 평가
"""
# 최근 데이터 요약
returns = np.log(recent_data['close'] / recent_data['close'].shift(1))
returns = returns.dropna() * 100
summary_stats = {
"mean_return": float(returns.mean()),
"std_return": float(returns.std()),
"skewness": float(returns.skew()),
"kurtosis": float(returns.kurtosis()),
"max_up": float(returns.max()),
"max_down": float(returns.min()),
"recent_5_std": float(returns.tail(5).std()),
"recent_20_std": float(returns.tail(20).std()),
}
prompt = f"""당신은 암호화폐 변동성 분석 전문가입니다.
최근 {lookback_hours}시간 BTC/USD 변동성 패턴을 분석해주세요:
통계 요약:
- 평균 수익률: {summary_stats['mean_return']:.4f}%
- 수익률 표준편차: {summary_stats['std_return']:.4f}%
- 왜도: {summary_stats['skewness']:.4f}
- 첨도: {summary_stats['kurtosis']:.4f}
- 최대 상승: {summary_stats['max_up']:.4f}%
- 최대 하락: {summary_stats['max_down']:.4f}%
- 최근 5봉 변동성: {summary_stats['recent_5_std']:.4f}%
- 최근 20봉 변동성: {summary_stats['recent_20_std']:.4f}%
분석 요구사항:
1. 현재 시장 변동성 상태 판단 (높음/보통/낮음)
2. 최근 5봉 대비 20봉 비율로 추세 판단
3. 왜도와 첨도를 고려한 비정상적 변동 패턴 감지
4. 향후 60분 변동성 예측 (구간과 신뢰도 포함)
JSON 형식으로 응답:
{{"status": "높음/보통/낮음", "trend": "증가/안정/감소",
"anomaly_detected": true/false, "prediction": {{
"volatility_range": ["최소", "최대"],
"confidence": 0.0~1.0,
"reasoning": "분석 근거"
}}}}"""
payload = {
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "당신은 전문 암호화폐 분석가입니다. 항상 유효한 JSON만 반환합니다."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"response_format": {"type": "json_object"}
}
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload
)
response.raise_for_status()
result = response.json()
analysis = json.loads(result['choices'][0]['message']['content'])
return analysis
except requests.exceptions.RequestException as e:
print(f"API 오류: {e}")
return None
def generate_garch_explanation(
self,
garch_result,
volatility_forecast: float
) -> str:
"""
Claude를 사용하여 GARCH 모델 결과 해석 생성
"""
# GARCH 모델 파라미터 추출
params = garch_result.params
prompt = f"""GARCH(1,1) 모델 결과를 비개발자도 이해할 수 있도록 설명해주세요.
모델 파라미터:
- omega (상수항): {params['omega']:.6f}
- alpha[1] (ARCH 효과): {params['alpha[1]']:.6f}
- beta[1] (GARCH 효과): {params['beta[1]']:.6f}
- persistence (지속성): {params['alpha[1]'] + params['beta[1]']:.4f}
예측 변동성: {volatility_forecast:.4f}%
설명 요구사항:
1. 각 파라미터의 실제 의미
2. persistence가 높다면 시장이 과거 충격에 민감하게 반응한다는 것
3. 현재 예측값의 의미와 투자자 시사점
한국어로 3문장 이내로 간결하게 설명해주세요."""
payload = {
"model": "claude-sonnet-4-20250514",
"messages": [
{"role": "user", "content": prompt}
],
"max_tokens": 300,
"temperature": 0.5
}
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload
)
response.raise_for_status()
result = response.json()
return result['choices'][0]['message']['content']
except requests.exceptions.RequestException as e:
print(f"API 오류: {e}")
return "설명 생성 실패"
def ensemble_prediction(
self,
garch_vol: float,
ml_predictions: List[float],
recent_trend: str = "stable"
) -> Dict:
"""
GARCH와 ML 예측을 결합한 앙상블 예측
HolySheep AI Gemini 활용
"""
prompt = f"""BTC 변동성 예측 앙상블을 수행해주세요.
입력 예측치들:
- GARCH 모델 예측: {garch_vol:.4f}%
- ML 모델 예측들: {[f'{p:.4f}%' for p in ml_predictions]}
최근 시장 추세: {recent_trend}
과제:
1. 각 예측치에 신뢰도 가중치 부여
2. GARCH와 ML의 장점을 결합한 최종 예측 산출
3. 예측 불확실성 구간 계산
JSON 응답:
{{"final_prediction": 숫자,
"confidence_interval": ["하한", "상한"],
"weights": {{"garch": 0~1, "ml_avg": 0~1}},
"explanation": "결합 방법 설명"}}
항상 유효한 JSON만 반환해주세요."""
payload = {
"model": "gemini-2.5-flash",
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"response_format": {"type": "json_object"}
}
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload
)
response.raise_for_status()
result = response.json()
return json.loads(result['choices'][0]['message']['content'])
except requests.exceptions.RequestException as e:
print(f"API 오류: {e}")
return None
HolySheep AIAnalyzer 초기화
analyzer = HolySheepMLAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
1. 패턴 분석 (GPT-4.1)
recent_data = btc_1min.tail(1440) # 최근 24시간 (1분봉 1440개)
pattern_analysis = analyzer.analyze_volatility_patterns(recent_data)
print("=" * 50)
print("HolySheep AI 변동성 패턴 분석 결과")
print("=" * 50)
print(f"시장 상태: {pattern_analysis['status']}")
print(f"추세: {pattern_analysis['trend']}")
print(f"예측 범위: {pattern_analysis['prediction']['volatility_range']}")
print(f"신뢰도: {pattern_analysis['prediction']['confidence']:.1%}")
2. GARCH 결과 해석 (Claude)
explanation = analyzer.generate_garch_explanation(
garch_model.result,
forecast['volatility'][-1]
)
print(f"\nGARCH 해석: {explanation}")머신러닝 Feature Engineering 및 LSTM 모델
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from sklearn.preprocessing import MinMaxScaler
from sklearn.model_selection import train_test_split
class MLVolatilityModel:
"""
LSTM 기반 변동성 예측 모델
HolySheep AI의 분석 결과를 Feature로 활용
"""
def __init__(self):
self.scaler = MinMaxScaler()
self.model = None
self.sequence_length = 60 # 60분 시퀀스
def create_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""
변동성 예측을 위한 Feature Engineering
Features:
- 수익률 관련: returns, abs_returns, squared_returns
- 기술적 지표: RSI, Bollinger Bands, ATR
- GARCH 파생: realized_vol, garch_residuals
- HolySheep AI 파생: sentiment_score, pattern_confidence
"""
df = df.copy()
# 수익률 기반 Features
df['returns'] = np.log(df['close'] / df['close'].shift(1))
df['abs_returns'] = np.abs(df['returns'])
df['squared_returns'] = df['returns'] ** 2
# Rolling Statistics
df['rolling_vol_5'] = df['returns'].rolling(5).std()
df['rolling_vol_20'] = df['returns'].rolling(20).std()
df['rolling_vol_60'] = df['returns'].rolling(60).std()
# 이동평균 대비当前位置
df['ma_ratio_5'] = df['close'] / df['close'].rolling(5).mean()
df['ma_ratio_20'] = df['close'] / df['close'].rolling(20).mean()
# 변동성 비율
df['vol_ratio'] = df['rolling_vol_5'] / df['rolling_vol_60']
# GARCH 잔차 (예측 오차)
if 'garch_vol' in df.columns:
df['garch_residual'] = df['returns'] - df['garch_vol']
# HolySheep AI 파생 Feature (이전 분석 결과 활용)
if 'holysheep_confidence' in df.columns:
df['ai_weighted_vol'] = df['rolling_vol_5'] * (1 - df['holysheep_confidence'])
# Target: 미래 60분 변동성
df['target_volatility'] = df['returns'].rolling(60).std().shift(-60)
return df.dropna()
def create_sequences(self, data: np.ndarray) -> Tuple:
"""시계열 시퀀스 생성"""
X, y = [], []
for i in range(self.sequence_length, len(data)):
X.append(data[i - self.sequence_length:i])
y.append(data[i, -1]) # target_volatility
return np.array(X), np.array(y)
def build_lstm_model(self, input_shape: Tuple) -> Sequential:
"""LSTM 모델 구성"""
model = Sequential([
LSTM(64, return_sequences=True, input_shape=input_shape),
Dropout(0.2),
LSTM(32, return_sequences=False),
Dropout(0.2),
Dense(16, activation='relu'),
Dense(1) # 변동성 예측값
])
model.compile(
optimizer=tf.keras.optimizers.Adam(learning_rate=0.001),
loss='mse',
metrics=['mae']
)
return model
def train(self, df: pd.DataFrame, holy_sheep_features: pd.DataFrame = None):
"""
모델 학습
Args:
df: Tardis에서 수집한 OHLCV 데이터
holy_sheep_features: HolySheep AI 분석 결과
"""
# HolySheep AI Feature 병합
if holy_sheep_features is not None:
df = df.merge(holy_sheep_features, on='timestamp', how='left')
# Feature 생성
df_features = self.create_features(df)
# Feature 선택
feature_columns = [
'returns', 'abs_returns', 'squared_returns',
'rolling_vol_5', 'rolling_vol_20', 'rolling_vol_60',
'ma_ratio_5', 'ma_ratio_20', 'vol_ratio'
]
if 'holysheep_confidence' in df_features.columns:
feature_columns.append('holysheep_confidence')
feature_columns.append('target_volatility')
data = df_features[feature_columns].values
data = self.scaler.fit_transform(data)
# 시퀀스 생성
X, y = self.create_sequences(data)
# Train/Test 분리
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, shuffle=False # 시계열이므로 shuffle=False
)
print(f"학습 데이터: {X_train.shape[0]}개")
print(f"테스트 데이터: {X_test.shape[0]}개")
# 모델 학습
self.model = self.build_lstm_model((X_train.shape[1], X_train.shape[2]))
history = self.model.fit(
X_train, y_train,
epochs=50,
batch_size=32,
validation_split=0.1,
verbose=1
)
# 성능 평가
test_pred = self.model.predict(X_test)
test_rmse = np.sqrt(mean_squared_error(y_test, test_pred))
test_mae = mean_absolute_error(y_test, test_pred)
print(f"\n테스트 RMSE: {test_rmse:.6f}")
print(f"테스트 MAE: {test_mae:.6f}")
return history
def predict(self, recent_data: np.ndarray) -> float:
"""단일 예측 수행"""
if self.model is None:
raise ValueError("모델이 먼저 학습되어야 합니다")
# 스케일링
scaled_data = self.scaler.transform(recent_data)
# 시퀀스 생성
X = np.array([scaled_data[-self.sequence_length:]])
# 예측
prediction = self.model.predict(X)
return float(prediction[0, 0])
모델 학습 실행
ml_model = MLVolatilityModel()
history = ml_model.train(btc_1min)
print("\n" + "=" * 50)
print("머신러닝 모델 학습 완료")
print("=" * 50)GARCH vs 머신러닝 비교 분석
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
class VolatilityComparison:
"""
GARCH vs ML 모델 비교 분석
"""
def __init__(self, garch_model, ml_model, test_data: pd.DataFrame):
self.garch_model = garch_model
self.ml_model = ml_model
self.test_data = test_data
def rolling_evaluation(self, window: int = 60) -> pd.DataFrame:
"""
Rolling Window 방식으로 두 모델 성능 비교
매 window 마다:
1. GARCH 예측 수행
2. ML 예측 수행
3. 실제 변동성과 비교
"""
results = []
for i in range(window, len(self.test_data) - window, window):
window_data = self.test_data.iloc[i-window:i]
# 실제 변동성 (이후 60분)
future_returns = np.log(
self.test_data['close'].iloc[i:i+window] /
self.test_data['close'].iloc[i-window:i]
)
actual_vol = future_returns.std() * 100
# GARCH 예측
garch_vol = self.garch_model.forecast(horizon=60)['volatility'][-1]
# ML 예측
ml_vol = self.ml_model.predict(window_data.values)
results.append({
'timestamp': self.test_data.index[i],
'actual_vol': actual_vol,
'garch_vol': garch_vol,
'ml_vol': ml_vol,
'garch_error': abs(garch_vol - actual_vol),
'ml_error': abs(ml_vol - actual_vol)
})
return pd.DataFrame(results)
def calculate_metrics(self, results: pd.DataFrame) -> Dict:
"""성능 지표 계산"""
return {
'Model': ['GARCH(1,1)', 'LSTM'],
'RMSE': [
np.sqrt(mean_squared_error(results['actual_vol'], results['garch_vol'])),
np.sqrt(mean_squared_error(results['actual_vol'], results['ml_vol']))
],
'MAE': [
mean_absolute_error(results['actual_vol'], results['garch_vol']),
mean_absolute_error(results['actual_vol'], results['ml_vol'])
],
'MAPE (%)': [
np.mean(np.abs((results['actual_vol'] - results['garch_vol']) / results['actual_vol'])) * 100,
np.mean(np.abs((results['actual_vol'] - results['ml_vol']) / results['actual_vol'])) * 100
],
'Direction Accuracy (%)': [
self._direction_accuracy(results['actual_vol'], results['garch_vol']),
self._direction_accuracy(results['actual_vol'], results['ml_vol'])
]
}
def _direction_accuracy(self, actual: pd.Series, predicted: pd.Series) -> float:
"""변동성 방향 예측 정확도"""
actual_diff = actual.diff().dropna()
pred_diff = predicted.diff().dropna()
correct = (np.sign(actual_diff) == np.sign(pred_diff)).sum()
return correct / len(actual_diff) * 100
def plot_comparison(self, results: pd.DataFrame, save_path: str = None):
"""비교 시각화"""
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# 1. 시계열 비교
ax1 = axes[0, 0]
ax1.plot(results['timestamp'], results['actual_vol'],
label='Actual', color='black', linewidth=1.5)
ax1.plot(results['timestamp'], results['garch_vol'],
label='GARCH', color='blue', linewidth=1, alpha=0.7)
ax1.plot(results['timestamp'], results['ml_vol'],
label='LSTM', color='red', linewidth=1, alpha=0.7)
ax1.set_title('변동성 예측 비교 (Actual vs GARCH vs LSTM)')
ax1.set_xlabel('시간')
ax1.set_ylabel('변동성 (%)')
ax1.legend()
ax1.grid(True, alpha=0.3)
# 2. 산점도 비교
ax2 = axes[0, 1]
ax2.scatter(results['actual_vol'], results['garch_vol'],
alpha=0.5, label='GARCH', color='blue')
ax2.scatter(results['actual_vol'], results['ml_vol'],
alpha=0.5, label='LSTM', color='red')
ax2.plot([0, results['actual_vol'].max()],
[0, results['actual_vol'].max()],
'k--', label='Perfect Prediction')
ax2.set_xlabel('실제 변동성 (%)')
ax2.set_ylabel('예측 변동성 (%)')
ax2.set_title('예측 vs 실제 (산점도)')
ax2.legend()
ax2.grid(True, alpha=0.3)
# 3. 오차 분포
ax3 = axes[1, 0]
ax3.hist(results['garch_error'], bins=20, alpha=0.5, label='GARCH', color='blue')
ax3.hist(results['ml_error'], bins=20, alpha=0.5, label='LSTM', color='red')
ax3.set_xlabel('절대 오차 (%)')
ax3.set_ylabel('빈도')
ax3.set_title('예측 오차 분포')
ax3.legend()
ax3.grid(True, alpha=0.3)
# 4. 성능 지표 바 차트
ax4 = axes[1, 1]
metrics = self.calculate_metrics(results)
x = np.arange(len(metrics['Model']))
width = 0.25
ax4.bar(x - width, metrics['RMSE'], width, label='RMSE', color='blue')
ax4.bar(x, metrics['MAE'], width, label='MAE', color='green')
ax4.bar(x + width, metrics['MAPE (%)'], width, label='MAPE (%)', color='orange')
ax4.set_xlabel('모델')
ax4.set_ylabel('값')
ax4.set_title('성능 지표 비교')
ax4.set_xticks(x)
ax4.set_xticklabels(metrics['Model'])
ax4.legend()
ax4.grid(True, alpha=0.3, axis='y')
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=150, bbox_inches='tight')
plt.show()
return metrics
비교 분석 실행
comparison = VolatilityComparison(garch_model, ml_model, btc_1min)
results = comparison.rolling_evaluation(window=60)
print("=" * 60)
print("GARCH vs 머신러닝 변동성 예측 성능 비교")
print("=" * 60)
metrics = comparison.calculate_metrics(results)
metrics_df = pd.DataFrame(metrics)
print(metrics_df.to_string(index=False))
시각화
comparison.plot_comparison(results, save_path='volatility_comparison.png')GARCH vs 머신러닝 비교표
| 평가 항목 | GARCH(1,1) | LSTM 머신러닝 | HolySheep AI 앙상블
관련 리소스관련 문서 |
|---|