import torch import torch.nn as nn import numpy as np from sklearn.preprocessing import MinMaxScaler class Attention(nn.Module): """Attention Mechanism for LSTM""" def __init__(self, hidden_size): super(Attention, self).__init__() self.hidden_size = hidden_size self.attn = nn.Linear(hidden_size, 1) def forward(self, lstm_output): # lstm_output: [batch_size, seq_len, hidden_size] # attn_weights: [batch_size, seq_len, 1] attn_weights = torch.softmax(self.attn(lstm_output), dim=1) # context: [batch_size, hidden_size] context = torch.sum(attn_weights * lstm_output, dim=1) return context, attn_weights class AdvancedLSTM(nn.Module): """ [RTX 5070 Ti Optimized] High-Capacity LSTM with Attention - Hidden Size: 512 (Rich Feature Extraction) - Layers: 4 (Deep Reasoning) - Attention: Focus on critical time steps """ def __init__(self, input_size=1, hidden_size=512, num_layers=4, output_size=1, dropout=0.3): super(AdvancedLSTM, self).__init__() self.hidden_size = hidden_size self.num_layers = num_layers self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True, dropout=dropout) self.attention = Attention(hidden_size) self.fc = nn.Sequential( nn.Linear(hidden_size, hidden_size // 2), nn.ReLU(), nn.Dropout(dropout), nn.Linear(hidden_size // 2, hidden_size // 4), nn.ReLU(), nn.Linear(hidden_size // 4, output_size) ) def forward(self, x): # x: [batch, seq, feature] h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device) c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device) # LSTM Output lstm_out, _ = self.lstm(x, (h0, c0)) # [batch, seq, hidden] # Attention Mechanism context, _ = self.attention(lstm_out) # [batch, hidden] # Final Prediction out = self.fc(context) return out class PricePredictor: """ 주가 예측을 위한 고성능 Deep Learning 모델 (RTX 5070 Ti Edition) """ def __init__(self): self.scaler = MinMaxScaler(feature_range=(0, 1)) # [Hardware Spec] RTX 5070 Ti (16GB VRAM) 맞춤 설정 self.hidden_size = 512 self.num_layers = 4 self.model = AdvancedLSTM(input_size=1, hidden_size=self.hidden_size, num_layers=self.num_layers, dropout=0.3) self.criterion = nn.MSELoss() # CUDA 설정 self.device = torch.device('cpu') if torch.cuda.is_available(): try: gpu_name = torch.cuda.get_device_name(0) vram_gb = torch.cuda.get_device_properties(0).total_memory / 1024**3 # GPU 할당 self.device = torch.device('cuda') self.model.to(self.device) # Warm-up (컴파일 최적화 유도) dummy = torch.zeros(1, 60, 1).to(self.device) _ = self.model(dummy) print(f"🚀 [AI] Powered by {gpu_name} ({vram_gb:.1f}GB) - High Performance Mode On") except Exception as e: print(f"⚠️ [AI] GPU Init Failed: {e}") self.device = torch.device('cpu') else: print("⚠️ [AI] Running on CPU (Low Performance)") # Optimizer 설정 (AdamW가 일반화 성능이 좀 더 좋음) self.optimizer = torch.optim.AdamW(self.model.parameters(), lr=0.0005, weight_decay=1e-4) # 학습 파라미터 강화 self.batch_size = 64 self.epochs = 200 # 충분한 학습 self.seq_length = 60 # 60일(약 3개월) 패턴 분석 self.training_status = { "is_training": False, "loss": 0.0 } @staticmethod def verify_hardware(): """서버 시작 시 하드웨어 가속 여부 점검 및 로그 출력""" if torch.cuda.is_available(): try: gpu_name = torch.cuda.get_device_name(0) vram_gb = torch.cuda.get_device_properties(0).total_memory / 1024**3 print(f"🚀 [AI Check] Hardware Detected: {gpu_name} ({vram_gb:.1f}GB VRAM)") print(f" ✅ High Performance Mode is READY.") return True except Exception as e: print(f"⚠️ [AI Check] GPU Error: {e}") return False else: print("⚠️ [AI Check] No GPU Detected. Running in CPU Mode.") return False def train_and_predict(self, prices, forecast_days=1): """ Online Learning & Prediction """ # 데이터가 최소 시퀀스 길이 + 여유분보다 적으면 예측 불가 if len(prices) < (self.seq_length + 10): return None # 1. 데이터 전처리 data = np.array(prices).reshape(-1, 1) scaled_data = self.scaler.fit_transform(data) x_train, y_train = [], [] for i in range(len(scaled_data) - self.seq_length): x_train.append(scaled_data[i:i+self.seq_length]) y_train.append(scaled_data[i+self.seq_length]) x_train_t = torch.FloatTensor(np.array(x_train)).to(self.device) y_train_t = torch.FloatTensor(np.array(y_train)).to(self.device) # 2. 학습 self.model.train() self.training_status["is_training"] = True dataset_size = len(x_train_t) final_loss = 0.0 for epoch in range(self.epochs): perm = torch.randperm(dataset_size).to(self.device) x_shuffled = x_train_t[perm] y_shuffled = y_train_t[perm] epoch_loss = 0.0 steps = 0 for i in range(0, dataset_size, self.batch_size): batch_x = x_shuffled[i:min(i+self.batch_size, dataset_size)] batch_y = y_shuffled[i:min(i+self.batch_size, dataset_size)] self.optimizer.zero_grad() outputs = self.model(batch_x) loss = self.criterion(outputs, batch_y) loss.backward() self.optimizer.step() epoch_loss += loss.item() steps += 1 final_loss = epoch_loss / max(1, steps) self.training_status["is_training"] = False self.training_status["loss"] = final_loss # 3. 예측 self.model.eval() with torch.no_grad(): last_seq = torch.FloatTensor(scaled_data[-self.seq_length:]).unsqueeze(0).to(self.device) predicted_scaled = self.model(last_seq) predicted_price = self.scaler.inverse_transform(predicted_scaled.cpu().numpy())[0][0] current_price = prices[-1] trend = "UP" if predicted_price > current_price else "DOWN" change_rate = ((predicted_price - current_price) / current_price) * 100 # 신뢰도 점수 (Loss가 낮을수록 높음, 0~1) # Loss가 0.001이면 0.99, 0.01이면 0.9 정도 나오게 조정 confidence = 1.0 / (1.0 + (final_loss * 100)) return { "current": current_price, "predicted": float(predicted_price), "change_rate": round(change_rate, 2), "trend": trend, "loss": final_loss, "confidence": round(confidence, 2) }