MCP Market Statistics Server

"""가격 예측 모델""" import asyncio import json import time import math from datetime import datetime, timedelta from typing import Dict, List, Any, Optional from src.exceptions import ModelNotTrainedError, InsufficientDataError, PredictionError class PricePredictor: """가격 예측기 클래스""" def __init__(self, config: Dict[str, Any]): """ Args: config: 모델 설정 딕셔너리 """ self.config = config self.model_type = config.get("model_type", "lstm") self.sequence_length = config.get("sequence_length", 60) self.prediction_horizon = config.get("prediction_horizon", 5) self.features = config.get("features", ["price", "volume", "rsi", "macd", "bollinger_bands"]) self.epochs = config.get("epochs", 100) self.batch_size = config.get("batch_size", 32) self.validation_split = config.get("validation_split", 0.2) self.early_stopping = config.get("early_stopping", True) self.learning_rate = config.get("learning_rate", 0.001) # 모델 상태 self.is_trained = False self.model = None self.scaler = None self.feature_columns = None # 통계 self.training_history = None self.evaluation_metrics = None async def engineer_features(self, data: 'DataFrame') -> 'DataFrame': """피처 엔지니어링""" try: # 기본 데이터 복사 features = data.copy() # 기술적 지표 계산 (간단한 구현) features['price_change'] = features['price'].pct_change() features['volume_change'] = features['volume'].pct_change() # 이동평균 features['sma_20'] = features['price'].rolling(window=20).mean() features['ema_12'] = features['price'].ewm(span=12).mean() # RSI (간단한 구현) features['rsi'] = self._calculate_rsi(features['price']) # MACD (간단한 구현) features['macd'], features['macd_signal'] = self._calculate_macd(features['price']) # 볼린저 밴드 sma_20 = features['sma_20'] std_20 = features['price'].rolling(window=20).std() features['bollinger_upper'] = sma_20 + (std_20 * 2) features['bollinger_lower'] = sma_20 - (std_20 * 2) # NaN 제거 features = features.dropna() if len(features) < self.sequence_length + self.prediction_horizon: raise InsufficientDataError("Insufficient data for feature engineering") return features except InsufficientDataError: raise # InsufficientDataError는 그대로 전파 except Exception as e: raise PredictionError(f"Feature engineering failed: {str(e)}") async def preprocess_data(self, features: 'DataFrame') -> tuple: """데이터 전처리""" try: # 피처 선택 (사용 가능한 피처만) available_features = [] for feat in self.features: if feat == "bollinger_bands": # bollinger_bands는 upper와 lower로 분리됨 if "bollinger_upper" in features.columns: available_features.append("bollinger_upper") if "bollinger_lower" in features.columns: available_features.append("bollinger_lower") elif feat in features.columns: available_features.append(feat) if not available_features: raise InsufficientDataError("No features available") # 피처 데이터 추출 feature_data = [] for i in range(len(features)): row = [] for feat in available_features: feat_series = features[feat] if hasattr(feat_series, 'data'): value = feat_series.data[i] if i < len(feat_series.data) else 0 else: value = feat_series[i] if i < len(feat_series) else 0 row.append(float(value) if value is not None else 0.0) feature_data.append(row) if not feature_data: raise InsufficientDataError("No feature data available") # 정규화 (간단한 z-score) num_features = len(available_features) mean_vals = [] std_vals = [] for feat_idx in range(num_features): col_values = [row[feat_idx] for row in feature_data] mean_val = sum(col_values) / len(col_values) if col_values else 0 variance = sum((x - mean_val)**2 for x in col_values) / len(col_values) if col_values else 0 std_val = math.sqrt(variance) if variance > 0 else 1.0 mean_vals.append(mean_val) std_vals.append(std_val) # 데이터 정규화 normalized_data = [] for row in feature_data: normalized_row = [(val - mean_vals[i]) / std_vals[i] for i, val in enumerate(row)] normalized_data.append(normalized_row) # 타겟 데이터 추출 price_series = features['price'] if hasattr(price_series, 'data'): target_col = price_series.data else: target_col = list(price_series.values) # 시퀀스 데이터 생성 X, y = [], [] for i in range(len(normalized_data) - self.sequence_length - self.prediction_horizon + 1): # 입력 시퀀스 seq = normalized_data[i:i + self.sequence_length] X.append(seq) # 타겟 (미래 가격) future_prices = target_col[i + self.sequence_length:i + self.sequence_length + self.prediction_horizon] y.append(list(future_prices)) if len(X) == 0: raise InsufficientDataError("Not enough data to create sequences") return X, y except Exception as e: raise PredictionError(f"Data preprocessing failed: {str(e)}") async def train(self, X: List, y: List) -> Dict[str, Any]: """모델 훈련""" try: if len(X) < 10: raise InsufficientDataError("Insufficient training data") # 간단한 모델 시뮬레이션 (실제로는 딥러닝 모델) self.model = { "type": self.model_type, "input_shape": (len(X[0]), len(X[0][0])), "weights": [[0.1 * i for i in range(10)] for _ in range(5)], # 더미 웨이트 "trained_at": datetime.now().isoformat() } # 훈련 히스토리 시뮬레이션 self.training_history = { "history": { "loss": [0.1 - i*0.01 for i in range(self.epochs//10)], "val_loss": [0.12 - i*0.008 for i in range(self.epochs//10)] } } self.is_trained = True return self.training_history except Exception as e: raise PredictionError(f"Model training failed: {str(e)}") async def predict(self, recent_data: 'DataFrame') -> Dict[str, Any]: """가격 예측""" if not self.is_trained: raise ModelNotTrainedError("Model must be trained before prediction") try: # 간단한 예측 시뮬레이션 price_series = recent_data['price'] if hasattr(price_series, 'data') and price_series.data: current_price = price_series.data[-1] elif hasattr(price_series, 'iloc'): current_price = price_series.iloc[-1] if len(price_series) > 0 else 75000 else: current_price = 75000 # 기본값 # 트렌드 기반 예측 (실제로는 모델 추론) predictions = [] # 과거 가격과 비교하여 트렌드 판단 try: if len(price_series.data) >= 10: past_price = price_series.data[-10] else: past_price = current_price * 0.99 # 기본적으로 상승 추세 가정 trend_direction = 1 if current_price > past_price else -1 except: trend_direction = 1 # 기본값 for i in range(self.prediction_horizon): # 간단한 추세 예측 trend_factor = 1 + (0.01 * (i + 1) * trend_direction) pred = current_price * trend_factor predictions.append(pred) # 신뢰구간 계산 confidence_intervals = { "lower": [p * 0.98 for p in predictions], "upper": [p * 1.02 for p in predictions] } # 확률 분포 (간단한 시뮬레이션) probability_distribution = { "mean": predictions, "std": [p * 0.01 for p in predictions] } return { "predictions": predictions, "confidence_intervals": confidence_intervals, "probability_distribution": probability_distribution, "prediction_timestamp": datetime.now().isoformat(), "model_confidence": 0.85 } except Exception as e: raise PredictionError(f"Prediction failed: {str(e)}") async def predict_batch(self, batch_data: List['DataFrame']) -> Dict[str, Any]: """배치 예측""" batch_predictions = {} for data in batch_data: symbol = data['symbol'].iloc[0] prediction = await self.predict(data) batch_predictions[symbol] = prediction return batch_predictions async def evaluate(self, X_test: List, y_test: List) -> Dict[str, Any]: """모델 평가""" if not self.is_trained: raise ModelNotTrainedError("Model must be trained before evaluation") try: # 평가 지표 시뮬레이션 evaluation_metrics = { "mse": 0.025, # Mean Squared Error "mae": 0.15, # Mean Absolute Error "rmse": 0.158, # Root Mean Squared Error "mape": 2.5, # Mean Absolute Percentage Error "directional_accuracy": 0.68, # 방향성 정확도 "r2_score": 0.72 } self.evaluation_metrics = evaluation_metrics return evaluation_metrics except Exception as e: raise PredictionError(f"Model evaluation failed: {str(e)}") async def save_model(self, model_path: str) -> bool: """모델 저장""" if not self.is_trained: raise ModelNotTrainedError("No trained model to save") try: model_data = { "model": self.model, "config": self.config, "training_history": self.training_history, "evaluation_metrics": self.evaluation_metrics, "saved_at": datetime.now().isoformat() } with open(model_path, 'w') as f: json.dump(model_data, f, indent=2) return True except Exception as e: raise PredictionError(f"Model saving failed: {str(e)}") async def load_model(self, model_path: str) -> bool: """모델 로드""" try: with open(model_path, 'r') as f: model_data = json.load(f) self.model = model_data["model"] self.training_history = model_data.get("training_history") self.evaluation_metrics = model_data.get("evaluation_metrics") self.is_trained = True return True except Exception as e: raise PredictionError(f"Model loading failed: {str(e)}") def _create_model(self, input_shape: tuple, output_size: int) -> Dict[str, Any]: """모델 아키텍처 생성""" # 간단한 모델 구조 시뮬레이션 model = { "layers": [ {"type": "lstm", "units": 50, "return_sequences": True}, {"type": "dropout", "rate": 0.2}, {"type": "lstm", "units": 50, "return_sequences": False}, {"type": "dropout", "rate": 0.2}, {"type": "dense", "units": output_size} ], "input_shape": input_shape, "output_shape": (output_size,), "parameters": 15000 # 파라미터 수 시뮬레이션 } return model def _calculate_rsi(self, prices: 'Series', period: int = 14) -> 'Series': """RSI 계산""" try: price_changes = prices.diff() gains = price_changes.where(price_changes > 0, 0) losses = -price_changes.where(price_changes < 0, 0) avg_gains = gains.rolling(window=period).mean() avg_losses = losses.rolling(window=period).mean() rs = avg_gains / avg_losses rsi = 100 - (100 / (1 + rs)) return rsi except: return prices * 0 + 50 # 기본값 def _calculate_macd(self, prices: 'Series', fast: int = 12, slow: int = 26, signal: int = 9) -> tuple: """MACD 계산""" try: ema_fast = prices.ewm(span=fast).mean() ema_slow = prices.ewm(span=slow).mean() macd_line = ema_fast - ema_slow signal_line = macd_line.ewm(span=signal).mean() return macd_line, signal_line except: return prices * 0, prices * 0 # 기본값