Spaces:
Sleeping
Sleeping
| import os | |
| import torch | |
| import logging | |
| from brain.neural_networks.data_processor import DataProcessor | |
| from brain.neural_networks.model import StockLSTM | |
| # Configure logging | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| class ModelManager: | |
| _instance = None | |
| def __new__(cls): | |
| if cls._instance is None: | |
| logger.info("Initializing ModelManager Singleton...") | |
| cls._instance = super(ModelManager, cls).__new__(cls) | |
| cls._instance.initialize() | |
| return cls._instance | |
| def initialize(self): | |
| self.device = torch.device('cpu') # Inference on CPU is sufficient | |
| self.processor = None | |
| self.model = None | |
| self.model_path = "brain/saved_models/hybrid_lstm.pth" | |
| # Lazy load flag | |
| self._loaded = False | |
| def load_resources(self): | |
| """ | |
| Loads the DataProcessor and PyTorch Model if not already loaded. | |
| """ | |
| if self._loaded: | |
| return | |
| logger.info("Loading AI Resources...") | |
| try: | |
| # 1. Initialize Processor (Loads Scaler) | |
| self.processor = DataProcessor(sequence_length=60) | |
| # 2. Initialize Model | |
| self.model = StockLSTM(input_size=13) | |
| if os.path.exists(self.model_path): | |
| logger.info(f"Loading model weights from {self.model_path}") | |
| self.model.load_state_dict(torch.load(self.model_path, map_location=self.device)) | |
| self.model.to(self.device) | |
| self.model.eval() | |
| self._loaded = True | |
| else: | |
| logger.warning(f"Model file not found at {self.model_path}. Neural predictions will be disabled.") | |
| self.model = None | |
| except Exception as e: | |
| logger.error(f"Failed to load AI resources: {e}") | |
| self.model = None | |
| self.processor = None | |
| def predict_sentiment(self, graph_data): | |
| """ | |
| Runs inference on the provided graph data. | |
| Returns: (signal: str, confidence: float) | |
| """ | |
| if not self._loaded: | |
| self.load_resources() | |
| if not self.model or not self.processor: | |
| return "Neutral (Model Error)", 0.0 | |
| try: | |
| # Prepare Data | |
| input_tensor = self.processor.prepare_inference_data(graph_data) | |
| if input_tensor is None: | |
| logger.warning("Insufficient data for inference (Need 60+ days).") | |
| return "Neutral (No Data)", 0.0 | |
| # Predict | |
| input_tensor = torch.FloatTensor(input_tensor).to(self.device) | |
| with torch.no_grad(): | |
| prediction = self.model(input_tensor).item() | |
| signal = "Bullish" if prediction > 0.50 else "Bearish" | |
| return signal, prediction | |
| except Exception as e: | |
| logger.error(f"Inference Error: {e}") | |
| return "Neutral (Error)", 0.0 | |
| # Global Instance | |
| model_manager = ModelManager() | |