저는 HolySheep AI에서 2년간 금융 AI 시스템을 구축하며 수십억 건의 시장 데이터를 처리해온 엔지니어입니다. 이번 튜토리얼에서는 Tardis의 실시간 주문서(Oder Book) 데이터를 활용하여 암호화폐 가격 움직임을 예측하는(end-to-end ML 파이프라인)를 상세히 다룹니다. Tardis는 Binance, Bybit, OKX 등 30개 이상의 거래소에서 마이크로초 단위의 호가창 데이터를 제공하는 전문 데이터 공급자입니다.
아키텍처 개요: 왜 주문서 데이터인가?
주문서는 특정 시점에서 특정 가격에 대기 중인 매수/매도 주문의 누적량을 보여줍니다. 이 데이터는 거래소 내 유동성 분포를 실시간으로 반영하며, 시장 미세 구조(Market Microstructure) 분석의 핵심입니다. 전통적인 시계열 모델(ARIMA, LSTM)보다 주문서 기반 피처 엔지니어링이 단기 가격 변동 예측에서 통계적으로 우월한 결과를 보여줍니다.
# 프로젝트 구조
orderbook-prediction/
├── config/
│ ├── __init__.py
│ └── settings.py # 하이퍼파라미터 및 API 설정
├── data/
│ ├── fetcher.py # Tardis 데이터 수집기
│ └── processor.py # 실시간 전처리 파이프라인
├── features/
│ ├── orderbook_features.py # 주문서 파생 피처
│ └── momentum_features.py # 모멘텀 및 변동성 피처
├── models/
│ ├── lstm_model.py # LSTM 기반 예측기
│ └── transformer_model.py # Transformer 기반 예측기
├── training/
│ ├── trainer.py # 학습 루프
│ └── evaluator.py # 모델 평가
├── inference/
│ └── predictor.py # 실시간 추론 서버
├── tests/
│ └── test_pipeline.py
├── main.py
├── requirements.txt
└── docker-compose.yml1. 환경 설정 및 의존성
먼저 필수 패키지를 설치합니다. Tardis SDK는 WebSocket을 통해 실시간 데이터를 스트리밍하며, PyTorch로 모델을 구현합니다.
# requirements.txt
tardis-dev==1.8.0
torch==2.1.0
torchvision==0.16.0
numpy==1.24.3
pandas==2.0.3
scikit-learn==1.3.2
ta==0.10.2
redis==5.0.1
fastapi==0.104.1
uvicorn==0.24.0
python-dotenv==1.0.0
optuna==3.5.0
# config/settings.py
import os
from dataclasses import dataclass
from typing import List
@dataclass
class ModelConfig:
sequence_length: int = 60 # 60초 윈도우 (1분봉 기준)
prediction_horizon: int = 5 # 5초 후 예측
hidden_dim: int = 256
num_layers: int = 3
dropout: float = 0.2
learning_rate: float = 1e-4
batch_size: int = 128
epochs: int = 100
@dataclass
class DataConfig:
exchanges: List[str] = None # ["binance", "bybit", "okx"]
symbols: List[str] = None # ["BTCUSDT", "ETHUSDT"]
channels: List[str] = None # ["orderbook", "trade"]
def __post_init__(self):
self.exchanges = self.exchanges or ["binance"]
self.symbols = self.symbols or ["BTCUSDT"]
self.channels = self.channels or ["orderbook_snapshot"]
@dataclass
class HolySheepConfig:
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
model_name: str = "gpt-4.1" # 피처 생성용 LLM
max_tokens: int = 500
temperature: float = 0.7
HolySheep AI를 사용한 강화 학습 에이전트용 설정
@dataclass
class RLConfig:
use_holysheep: bool = True # HolySheep AI Inference API 사용
inference_endpoint: str = HolySheepConfig.base_url + "/chat/completions"
reinforcement_model: str = "gpt-4.1"
cache_enabled: bool = True
cache_ttl: int = 300 # 5분 캐시
환경 변수 로드
from dotenv import load_dotenv
load_dotenv()2. Tardis 데이터 수집기 구현
Tardis는 고성능 WebSocket 스트리밍을 제공합니다. Binance의 경우 100ms 간격으로 주문서 스냅샷이 전송되며, 이를 버퍼링하여 배치 처리합니다.
# data/fetcher.py
import asyncio
import json
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from tardis import Tardis
from tardis.channels.exchange import Binance
import pandas as pd
from datetime import datetime
import redis.asyncio as redis
@dataclass
class OrderBookSnapshot:
exchange: str
symbol: str
timestamp: int
bids: List[List[float]] # [[price, quantity], ...]
asks: List[List[float]] # [[price, quantity], ...]
@property
def mid_price(self) -> float:
if not self.bids or not self.asks:
return 0.0
return (float(self.bids[0][0]) + float(self.asks[0][0])) / 2
def to_dataframe(self) -> pd.DataFrame:
return pd.DataFrame({
'timestamp': [self.timestamp] * len(self.bids + self.asks),
'price': [float(x[0]) for x in self.bids + self.asks],
'quantity': [float(x[1]) for x in self.bids + self.asks],
'side': ['bid'] * len(self.bids) + ['ask'] * len(self.asks),
'exchange': [self.exchange] * (len(self.bids) + len(self.asks)),
'symbol': [self.symbol] * (len(self.bids) + len(self.asks))
})
class TardisCollector:
"""Tardis 실시간 주문서 데이터 수집기"""
def __init__(
self,
exchanges: List[str],
symbols: List[str],
buffer_size: int = 1000,
redis_url: str = "redis://localhost:6379"
):
self.exchanges = exchanges
self.symbols = symbols
self.buffer_size = buffer_size
self.buffer: Dict[str, List[OrderBookSnapshot]] = {}
self.redis_client: Optional[redis.Redis] = None
self.redis_url = redis_url
self._running = False
# 버퍼 초기화
for symbol in symbols:
self.buffer[symbol] = []
async def connect_redis(self):
"""Redis 연결 (시계열 캐싱용)"""
self.redis_client = await redis.from_url(
self.redis_url,
encoding="utf-8",
decode_responses=True
)
print(f"✅ Redis 연결 완료: {self.redis_url}")
def _get_exchange_channel(self, exchange: str, symbol: str):
"""거래소별 채널 매핑"""
exchange_map = {
"binance": Binance(
exchange="binance",
channels=[f"orderbook_{symbol.replace('USDT', '').lower()}@depth20"
for symbol in self.symbols]
)
}
return exchange_map.get(exchange)
async def collect(
self,
duration_seconds: Optional[int] = None,
callback: Optional[Callable] = None
):
"""
지정된 시간 동안 데이터 수집
Args:
duration_seconds: 수집 시간 (None이면 무한 수집)
callback: 각 스냅샷마다 호출될 콜백 함수
"""
self._running = True
async with Tardis() as tardis:
# Binance WebSocket 구독
channels = [
Binance(
exchange="binance",
channels=[f"orderbook_{symbol.replace('USDT', '').lower()}@depth20@100ms"
for symbol in self.symbols]
)
]
await tardis.subscribe(channels=channels)
print(f"📡 Tardis 스트리밍 시작: {self.exchanges} - {self.symbols}")
start_time = asyncio.get_event_loop().time()
async for site_name, exchange_name, channel_name, message in tardis.stream():
if not self._running:
break
try:
# 주문서 데이터 파싱
snapshot = self._parse_orderbook_message(
exchange_name, channel_name, message
)
if snapshot:
# 버퍼에 저장
self.buffer[snapshot.symbol].append(snapshot)
# Redis 캐싱 (TTL: 5분)
if self.redis_client:
await self._cache_snapshot(snapshot)
# 버퍼 크기 제한
if len(self.buffer[snapshot.symbol]) > self.buffer_size:
self.buffer[snapshot.symbol].pop(0)
# 콜백 실행
if callback:
await callback(snapshot)
except Exception as e:
print(f"⚠️ 메시지 파싱 오류: {e}")
# 시간 제한 체크
if duration_seconds:
elapsed = asyncio.get_event_loop().time() - start_time
if elapsed >= duration_seconds:
break
def _parse_orderbook_message(
self,
exchange: str,
channel: str,
message: dict
) -> Optional[OrderBookSnapshot]:
"""Tardis 메시지를 OrderBookSnapshot으로 변환"""
try:
symbol = message.get('s', '')
return OrderBookSnapshot(
exchange=exchange,
symbol=symbol,
timestamp=message.get('E', 0), # EventTime
bids=[[float(p), float(q)] for p, q in message.get('b', [])],
asks=[[float(p), float(q)] for p, q in message.get('a', [])]
)
except (KeyError, ValueError, TypeError) as e:
return None
async def _cache_snapshot(self, snapshot: OrderBookSnapshot):
"""Redis에 스냅샷 캐싱"""
key = f"orderbook:{snapshot.exchange}:{snapshot.symbol}:{snapshot.timestamp}"
data = json.dumps({
'bids': snapshot.bids,
'asks': snapshot.asks,
'mid_price': snapshot.mid_price
})
await self.redis_client.setex(key, 300, data) # 5분 TTL
async def get_historical(
self,
exchange: str,
symbol: str,
start: datetime,
end: datetime
) -> pd.DataFrame:
"""과거 데이터 조회 (Tardis Replay API)"""
async with Tardis() as tardis:
channels = [
Binance(
exchange=exchange,
channels=[f"orderbook_{symbol.replace('USDT', '').lower()}@depth20"]
)
]
await tardis.subscribe(channels=channels)
dfs = []
async for site_name, exchange_name, channel_name, message in tardis.stream():
snapshot = self._parse_orderbook_message(
exchange_name, channel_name, message
)
if snapshot:
dfs.append(snapshot.to_dataframe())
# 지정 시간 범위 체크
timestamp = datetime.fromtimestamp(snapshot.timestamp / 1000)
if timestamp >= end:
break
return pd.concat(dfs, ignore_index=True) if dfs else pd.DataFrame()
def stop(self):
"""수집 중지"""
self._running = False
print("⏹️ 데이터 수집 중지됨")
async def close(self):
"""리소스 정리"""
if self.redis_client:
await self.redis_client.close()
self.stop()3. 주문서 파생 피처 엔지니어링
원시 주문서 데이터에서 예측에 유용한 피처를 추출합니다. 핵심 지표들은 시장 심리와 유동성 상황을 정량화합니다.
# features/orderbook_features.py
import numpy as np
import pandas as pd
from typing import List, Tuple
from dataclasses import dataclass
from numba import jit
@dataclass
class OrderBookFeatures:
"""주문서에서 추출한 피처 벡터"""
# 유동성 지표
bid_volume: float # 매수측 총 물량
ask_volume: float # 매도측 총 물량
volume_imbalance: float # 물량 불균형 (bid - ask) / (bid + ask)
bid_volume_depth10: float # 최상위 10단계 물량 합계
ask_volume_depth10: float
# 스프레드 지표
spread: float # 최우선 매도 - 최우선 매수
spread_pct: float # 스프레드 / mid_price
spread_depth5: float # 5단계 스프레드
# 가격 압박 지표
weighted_mid_price: float # 물량 가중 중립가
price_pressure: float # 가격 압박 지수
microprice: float # 마이크로프라이스
# 모멘텀 지표
mid_price_return: float # 중립가 수익률
price_acceleration: float # 가격 가속도
volume_weighted_price: float
# 변동성 지표
volatility: float # 최근 변동성
bid_ask_ratio: float # 매수/매도 비율
@classmethod
def from_snapshot(cls, snapshot, prev_snapshot=None) -> 'OrderBookFeatures':
"""OrderBookSnapshot에서 피처 생성"""
bids = np.array(snapshot.bids, dtype=np.float64)
asks = np.array(snapshot.asks, dtype=np.float64)
if len(bids) == 0 or len(asks) == 0:
return cls(*([0.0] * 15))
bid_prices = bids[:, 0]
bid_qtys = bids[:, 1]
ask_prices = asks[:, 0]
ask_qtys = asks[:, 1]
# 기본 유동성 지표
bid_volume = np.sum(bid_qtys)
ask_volume = np.sum(ask_qtys)
volume_imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume + 1e-10)
# 10단계 깊이
bid_volume_depth10 = np.sum(bid_qtys[:10])
ask_volume_depth10 = np.sum(ask_qtys[:10])
# 스프레드
best_bid = bid_prices[0]
best_ask = ask_prices[0]
mid_price = (best_bid + best_ask) / 2
spread = best_ask - best_bid
spread_pct = spread / mid_price if mid_price > 0 else 0
# 5단계 스프레드
spread_depth5 = (ask_prices[4] - bid_prices[4]) / mid_price if len(bid_prices) >= 5 else spread_pct
# 가중 중립가 및 마이크로프라이스
total_volume = bid_volume + ask_volume
weighted_mid_price = (
np.sum(bid_prices * bid_qtys) / (bid_volume + 1e-10) * ask_volume +
np.sum(ask_prices * ask_qtys) / (ask_volume + 1e-10) * bid_volume
) / (total_volume + 1e-10)
# 마이크로프라이스 (VWAP 기반 가격 압박)
microprice = (best_bid * ask_volume + best_ask * bid_volume) / (bid_volume + ask_volume + 1e-10)
# 가격 압박 지수
price_pressure = (bid_volume - ask_volume) * (best_ask - best_bid) / mid_price
# 변동성 및 모멘텀
return_rate = 0.0
price_acceleration = 0.0
if prev_snapshot and prev_snapshot.mid_price > 0:
return_rate = (mid_price - prev_snapshot.mid_price) / prev_snapshot.mid_price
price_acceleration = return_rate * 10 # 시간 정규화
# VWAP
vwap = (np.sum(bid_prices * bid_qtys) + np.sum(ask_prices * ask_qtys)) / total_volume
return cls(
bid_volume=float(bid_volume),
ask_volume=float(ask_volume),
volume_imbalance=float(volume_imbalance),
bid_volume_depth10=float(bid_volume_depth10),
ask_volume_depth10=float(ask_volume_depth10),
spread=float(spread),
spread_pct=float(spread_pct),
spread_depth5=float(spread_depth5),
weighted_mid_price=float(weighted_mid_price),
price_pressure=float(price_pressure),
microprice=float(microprice),
mid_price_return=float(return_rate),
price_acceleration=float(price_acceleration),
volume_weighted_price=float(vwap)
)
def to_array(self) -> np.ndarray:
"""피처 벡터를 numpy 배열로 변환"""
return np.array([
self.bid_volume,
self.ask_volume,
self.volume_imbalance,
self.bid_volume_depth10,
self.ask_volume_depth10,
self.spread,
self.spread_pct,
self.spread_depth5,
self.weighted_mid_price,
self.price_pressure,
self.microprice,
self.mid_price_return,
self.price_acceleration,
self.volume_weighted_price
], dtype=np.float32)
class FeatureProcessor:
"""피처 전처리 및 정규화 파이프라인"""
def __init__(self, feature_dim: int = 14):
self.feature_dim = feature_dim
self.scaler_mean: np.ndarray = None
self.scaler_std: np.ndarray = None
self.is_fitted = False
def fit(self, features_list: List[OrderBookFeatures]):
"""학습 데이터로 정규화 파라미터 계산"""
X = np.array([f.to_array() for f in features_list])
self.scaler_mean = np.mean(X, axis=0)
self.scaler_std = np.std(X, axis=0) + 1e-8
self.is_fitted = True
print(f"✅ 피처 정규화 파라미터 학습 완료: mean.shape={self.scaler_mean.shape}")
def transform(self, features: OrderBookFeatures) -> np.ndarray:
"""정규화 변환"""
if not self.is_fitted:
raise ValueError("FeatureProcessor가 먼저 fit()되어야 합니다")
x = features.to_array()
return (x - self.scaler_mean) / self.scaler_std
def fit_transform(self, features_list: List[OrderBookFeatures]) -> np.ndarray:
"""fit + transform"""
self.fit(features_list)
return np.array([self.transform(f) for f in features_list])
def inverse_transform(self, X_normalized: np.ndarray) -> np.ndarray:
"""역정규화 (예측 결과 복원)"""
return X_normalized * self.scaler_std + self.scaler_mean
@jit(nopython=True)
def calculate_rolling_features(
prices: np.ndarray,
volumes: np.ndarray,
window: int = 60
) -> np.ndarray:
"""Numba JIT 가속 롤링 통계량 계산"""
n = len(prices)
result = np.zeros((n, 5), dtype=np.float64)
for i in range(window, n):
price_window = prices[i-window:i]
volume_window = volumes[i-window:i]
# 롤링 평균
result[i, 0] = np.mean(price_window)
# 롤링 표준편차
result[i, 1] = np.std(price_window)
# VWAP
result[i, 2] = np.sum(price_window * volume_window) / np.sum(volume_window)
# 변동성 ( Parkinson volatility )
high = np.max(price_window)
low = np.min(price_window)
result[i, 3] = np.log(high / low) / np.sqrt(2) if high > low else 0
# 모멘텀
result[i, 4] = (prices[i] - prices[i-window]) / prices[i-window]
return result4. LSTM + Attention 기반 예측 모델
시계열 패턴 학습에 최적화된 LSTM과 Self-Attention 메커니즘을 결합합니다. HolySheep AI Inference API를 활용하면 추가적인 시장 분석 LLM을 쉽게 통합할 수 있습니다.
# models/lstm_attention_model.py
import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import Tuple, Optional
import math
class PositionalEncoding(nn.Module):
"""시퀀스 위치 정보 인코딩"""
def __init__(self, d_model: int, max_len: int = 5000, dropout: float = 0.1):
super().__init__()
self.dropout = nn.Dropout(p=dropout)
pe = torch.zeros(max_len, d_model)
position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
div_term = torch.exp(
torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)
)
pe[:, 0::2] = torch.sin(position * div_term)
pe[:, 1::2] = torch.cos(position * div_term)
pe = pe.unsqueeze(0) # (1, max_len, d_model)
self.register_buffer('pe', pe)
def forward(self, x: torch.Tensor) -> torch.Tensor:
x = x + self.pe[:, :x.size(1), :]
return self.dropout(x)
class MultiHeadSelfAttention(nn.Module):
"""멀티헤드 셀프 어텐션"""
def __init__(self, d_model: int, num_heads: int = 8, dropout: float = 0.1):
super().__init__()
assert d_model % num_heads == 0
self.d_model = d_model
self.num_heads = num_heads
self.d_k = d_model // num_heads
self.q_linear = nn.Linear(d_model, d_model)
self.k_linear = nn.Linear(d_model, d_model)
self.v_linear = nn.Linear(d_model, d_model)
self.out_linear = nn.Linear(d_model, d_model)
self.dropout = nn.Dropout(dropout)
self.scale = math.sqrt(self.d_k)
def forward(
self,
x: torch.Tensor,
mask: Optional[torch.Tensor] = None
) -> torch.Tensor:
batch_size, seq_len, _ = x.size()
# Q, K, V 선형 변환
Q = self.q_linear(x).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
K = self.k_linear(x).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
V = self.v_linear(x).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
# 어텐션 스코어
scores = torch.matmul(Q, K.transpose(-2, -1)) / self.scale
if mask is not None:
scores = scores.masked_fill(mask == 0, -1e9)
attention_weights = F.softmax(scores, dim=-1)
attention_weights = self.dropout(attention_weights)
# 어텐션 결과
context = torch.matmul(attention_weights, V)
context = context.transpose(1, 2).contiguous().view(batch_size, seq_len, self.d_model)
return self.out_linear(context)
class TransformerEncoderLayer(nn.Module):
"""트랜스포머 인코더 레이어"""
def __init__(self, d_model: int, num_heads: int = 8, ff_dim: int = 512, dropout: float = 0.1):
super().__init__()
self.self_attn = MultiHeadSelfAttention(d_model, num_heads, dropout)
self.ffn = nn.Sequential(
nn.Linear(d_model, ff_dim),
nn.GELU(),
nn.Dropout(dropout),
nn.Linear(ff_dim, d_model),
nn.Dropout(dropout)
)
self.norm1 = nn.LayerNorm(d_model)
self.norm2 = nn.LayerNorm(d_model)
self.dropout = nn.Dropout(dropout)
def forward(self, x: torch.Tensor, mask: Optional[torch.Tensor] = None) -> torch.Tensor:
# 셀프 어텐션 + 잔차 연결
attn_out = self.self_attn(x, mask)
x = self.norm1(x + self.dropout(attn_out))
# 피드포워드 + 잔차 연결
ffn_out = self.ffn(x)
x = self.norm2(x + ffn_out)
return x
class LSTMAttentionModel(nn.Module):
"""LSTM + Multi-Head Attention 기반 가격 예측 모델"""
def __init__(
self,
input_dim: int = 14,
hidden_dim: int = 256,
num_layers: int = 3,
dropout: float = 0.2,
prediction_horizon: int = 5,
use_transformer: bool = True
):
super().__init__()
self.hidden_dim = hidden_dim
self.num_layers = num_layers
self.prediction_horizon = prediction_horizon
# 입력 투영
self.input_projection = nn.Sequential(
nn.Linear(input_dim, hidden_dim),
nn.LayerNorm(hidden_dim),
nn.GELU(),
nn.Dropout(dropout)
)
# LSTM 레이어
self.lstm = nn.LSTM(
input_size=hidden_dim,
hidden_size=hidden_dim,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0,
bidirectional=True
)
# 트랜스포머 인코더 (어텐션 메커니즘)
self.use_transformer = use_transformer
if use_transformer:
self.positional_encoding = PositionalEncoding(hidden_dim * 2, dropout=dropout)
self.transformer_encoder = nn.ModuleList([
TransformerEncoderLayer(
d_model=hidden_dim * 2,
num_heads=8,
ff_dim=hidden_dim * 4,
dropout=dropout
)
for _ in range(2)
])
# 어텐션 pooling
self.attention = nn.Sequential(
nn.Linear(hidden_dim * 2, hidden_dim),
nn.Tanh(),
nn.Linear(hidden_dim, 1)
)
# 예측 헤드
self.prediction_head = nn.Sequential(
nn.Linear(hidden_dim * 2, hidden_dim),
nn.LayerNorm(hidden_dim),
nn.GELU(),
nn.Dropout(dropout),
nn.Linear(hidden_dim, prediction_horizon) # 다중 스텝 예측
)
# 방향 분류기 (상승/하락/변화없음)
self.direction_head = nn.Sequential(
nn.Linear(hidden_dim * 2, hidden_dim // 2),
nn.GELU(),
nn.Linear(hidden_dim // 2, 3) # 3 클래스 분류
)
self._init_weights()
def _init_weights(self):
"""가중치 초기화 (Xavier)"""
for module in self.modules():
if isinstance(module, nn.Linear):
nn.init.xavier_uniform_(module.weight)
if module.bias is not None:
nn.init.zeros_(module.bias)
def forward(
self,
x: torch.Tensor,
return_attention: bool = False
) -> Tuple[torch.Tensor, torch.Tensor, Optional[torch.Tensor]]:
"""
Args:
x: (batch_size, seq_len, input_dim)
Returns:
price_prediction: (batch_size, prediction_horizon)
direction_logits: (batch_size, 3)
attention_weights: Optional[(batch_size, seq_len)]
"""
batch_size, seq_len, _ = x.size()
# 입력 투영
x = self.input_projection(x) # (B, L, hidden)
# Bidirectional LSTM
lstm_out, _ = self.lstm(x) # (B, L, hidden*2)
# 트랜스포머 인코더 (선택적)
if self.use_transformer:
encoded = self.positional_encoding(lstm_out)
for layer in self.transformer_encoder:
encoded = layer(encoded)
else:
encoded = lstm_out
# 어텐션 기반 시퀀스 pooling
attn_weights = self.attention(encoded) # (B, L, 1)
attn_weights = F.softmax(attn_weights, dim=1) # (B, L, 1)
# 가중합으로 단일 벡터 생성
context = torch.sum(encoded * attn_weights, dim=1) # (B, hidden*2)
# 예측
price_pred = self.prediction_head(context) # (B, horizon)
direction_logits = self.direction_head(context) # (B, 3)
if return_attention:
return price_pred, direction_logits, attn_weights.squeeze(-1)
return price_pred, direction_logits, None
class PricePredictionLoss(nn.Module):
"""복합 손실 함수"""
def __init__(
self,
price_weight: float = 1.0,
direction_weight: float = 0.3,
use_focal: bool = True
):
super().__init__()
self.price_weight = price_weight
self.direction_weight = direction_weight
self.use_focal = use_focal
self.mse = nn.MSELoss()
self.ce = nn.CrossEntropyLoss()
def forward(
self,
price_pred: torch.Tensor,
direction_logits: torch.Tensor,
price_target: torch.Tensor,
direction_target: torch.Tensor
) -> Tuple[torch.Tensor, dict]:
# 가격 예측 손실 (MSE)
price_loss = self.mse(price_pred, price_target)
# 방향 분류 손실 (Focal Loss)
if self.use_focal:
ce_loss = F.cross_entropy(direction_logits, direction_target, reduction='none')
pt = torch.exp(-ce_loss)
focal_loss = ((1 - pt) ** 2 * ce_loss).mean()
direction_loss = focal_loss
else:
direction_loss = self.ce(direction_logits, direction_target)
# 총 손실
total_loss = (
self.price_weight * price_loss +
self.direction_weight * direction_loss
)
return total_loss, {
'price_loss': price_loss.item(),
'direction_loss': direction_loss.item(),
'total_loss': total_loss.item()
}5. 학습 파이프라인 및 평가
# training/trainer.py
import torch
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
from torch.optim.lr_scheduler import CosineAnnealingWarmRestarts
import numpy as np
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
import json
from pathlib import Path
from datetime import datetime
from tqdm import tqdm
import pandas as pd
from models.lstm_attention_model import LSTMAttentionModel, PricePredictionLoss
from features.orderbook_features import OrderBookFeatures, FeatureProcessor
@dataclass
class TrainingMetrics:
epoch: int
train_loss: float
val_loss: float
price_mae: float
direction_accuracy: float
directional_accuracy: float # 상승/하락 정확도
sharpe_ratio: float
max_drawdown: float
class OrderBookDataset(Dataset):
"""시퀀스 데이터셋"""
def __init__(
self,
sequences: np.ndarray,
price_targets: np.ndarray,
direction_targets: np.ndarray,
seq_len: int = 60
):
self.sequences = torch.FloatTensor(sequences)
self.price_targets = torch.FloatTensor(price_targets)
self.direction_targets = torch.LongTensor(direction_targets)
self.seq_len = seq_len
def __len__(self):
return len(self.sequences) - self.seq_len
def __getitem__(self, idx):
return (
self.sequences[idx:idx + self.seq_len],
self.price_targets[idx:idx + self.seq_len].mean(dim=0), # 타겟 기간 평균
self.direction_targets[idx + self.seq_len]
)
class ModelTrainer:
"""모델 학습 및 평가"""
def __init__(
self,
model: LSTMAttentionModel,
config,
device: str = "cuda" if torch.cuda.is_available() else "cpu"
):
self.model = model.to(device)
self.config = config
self.device = device
# 옵티마이저
self.optimizer = optim.AdamW(
model.parameters(),
lr=config.learning_rate,
weight_decay=1e-4
)
# 스케줄러
self.scheduler = CosineAnnealingWarmRestarts(
self.optimizer,
T_0=10,
T_mult=2,
eta_min=1e-6
)
# 손실 함수
self.criterion = PricePredictionLoss(
price_weight=1.0,
direction_weight=0.3
)
# 메트릭 기록
self.history: List[TrainingMetrics] = []
self.best_val_loss = float('inf')
def create_sequences(
self,
features: List[OrderBookFeatures],
prices: np.ndarray,
seq_len: int = 60,
horizon: int = 5
) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
"""시퀀스 데이터 생성"""
feature_processor = FeatureProcessor()
X_normalized = feature_processor.fit_transform(features)
sequences = []
price_targets = []
direction_targets = []
for i in range(len(X_normalized) - seq_len - horizon):
# 입력 시퀀스
seq = X_normalized[i:i + seq_len]
sequences.append(seq)
# 가격 타겟 (horizon 후 수익률)
future_price = prices[i + seq_len + horizon]
current_price = prices[i + seq_len]
price_return = (future_price - current_price) / current_price
price_targets.append(price_return