Upload meta_model.py

meta_model.py

@@ -0,0 +1,271 @@
+"""Meta-Model: Learns which model/signal to trust dynamically.
+
+This mimics how Renaissance Technologies combines signals — a meta-learner
+weights LSTM, Transformer, XGBoost, and sentiment based on recent performance,
+regime, and volatility state.
+"""
+import numpy as np
+import pandas as pd
+import torch
+import torch.nn as nn
+from sklearn.ensemble import GradientBoostingRegressor
+from typing import Dict, List, Optional, Tuple
+import warnings
+warnings.filterwarnings('ignore')
+
+
+class MetaModel:
+    """Meta-learner that dynamically weights base model predictions."""
+    
+    def __init__(self, 
+                 base_models: List[str] = None,
+                 meta_learner_type: str = 'xgb',
+                 lookback_window: int = 63,
+                 device: str = 'cpu'):
+        """
+        Args:
+            base_models: Names of base models (e.g., ['lstm','transformer','xgboost','sentiment'])
+            meta_learner_type: 'xgb', 'nn', or 'bayesian'
+            lookback_window: How many days of past performance to use as features
+        """
+        self.base_models = base_models or ['lstm', 'transformer', 'xgboost', 'sentiment']
+        self.meta_learner_type = meta_learner_type
+        self.lookback_window = lookback_window
+        self.device = torch.device(device)
+        
+        self.meta_model = None
+        self.performance_history = {m: [] for m in self.base_models}
+        self.weight_history = []
+        self.is_fitted = False
+        
+    def _build_meta_features(self, 
+                             predictions: Dict[str, np.ndarray],
+                             regime: Optional[str] = None,
+                             volatility: Optional[float] = None,
+                             recent_returns: Optional[np.ndarray] = None) -> np.ndarray:
+        """
+        Build feature vector for meta-learner.
+        
+        Features include:
+        - Raw predictions from each base model
+        - Recent IC of each model
+        - Recent MSE of each model
+        - Volatility regime
+        - Recent market return
+        """
+        n_samples = len(list(predictions.values())[0])
+        features = []
+        
+        # Raw predictions
+        for model in self.base_models:
+            if model in predictions:
+                features.append(predictions[model])
+            else:
+                features.append(np.zeros(n_samples))
+        
+        # Recent performance (rolling IC over lookback window)
+        for model in self.base_models:
+            perf = self.performance_history.get(model, [0.0] * self.lookback_window)
+            # Pad if needed
+            perf = perf[-self.lookback_window:]
+            while len(perf) < self.lookback_window:
+                perf = [0.0] + perf
+            # Summary stats of recent performance
+            features.append(np.full(n_samples, np.mean(perf)))
+            features.append(np.full(n_samples, np.std(perf) if len(perf) > 1 else 0.0))
+            features.append(np.full(n_samples, perf[-1] if perf else 0.0))
+        
+        # Regime encoding
+        if regime:
+            regime_map = {'bull': 1.0, 'bear': -1.0, 'high_vol': 0.0, 'neutral': 0.5}
+            regime_val = regime_map.get(regime, 0.5)
+            features.append(np.full(n_samples, regime_val))
+        else:
+            features.append(np.zeros(n_samples))
+        
+        # Volatility
+        features.append(np.full(n_samples, volatility or 0.2))
+        
+        # Recent market return
+        if recent_returns is not None and len(recent_returns) > 0:
+            features.append(np.full(n_samples, np.mean(recent_returns[-5:])))
+        else:
+            features.append(np.zeros(n_samples))
+        
+        return np.column_stack(features)
+    
+    def fit(self,
+            predictions_train: Dict[str, np.ndarray],
+            actual_train: np.ndarray,
+            regime_train: Optional[List[str]] = None,
+            volatility_train: Optional[np.ndarray] = None) -> Dict:
+        """
+        Train meta-learner to predict actual returns from base model predictions.
+        
+        The meta-learner learns optimal weights for combining base models.
+        """
+        n_samples = len(actual_train)
+        
+        # Build meta-features
+        X_meta = self._build_meta_features(
+            predictions_train,
+            regime=regime_train[0] if regime_train else None,
+            volatility=volatility_train[0] if volatility_train is not None else None
+        )
+        
+        if self.meta_learner_type == 'xgb':
+            self.meta_model = GradientBoostingRegressor(
+                n_estimators=100,
+                max_depth=4,
+                learning_rate=0.05,
+                subsample=0.8,
+                random_state=42
+            )
+            self.meta_model.fit(X_meta, actual_train)
+            
+        elif self.meta_learner_type == 'nn':
+            self.meta_model = self._build_nn_meta_model(X_meta.shape[1])
+            self._train_nn_meta(X_meta, actual_train)
+            
+        elif self.meta_learner_type == 'bayesian':
+            # Use XGB with quantile loss for uncertainty
+            self.meta_model = GradientBoostingRegressor(
+                n_estimators=100,
+                max_depth=4,
+                learning_rate=0.05,
+                loss='quantile', alpha=0.5,
+                random_state=42
+            )
+            self.meta_model.fit(X_meta, actual_train)
+        
+        self.is_fitted = True
+        
+        # Compute in-sample performance
+        pred = self.predict_meta(predictions_train, regime_train, volatility_train)
+        from scipy.stats import spearmanr
+        ic, _ = spearmanr(pred, actual_train)
+        mse = np.mean((pred - actual_train) ** 2)
+        
+        return {
+            'meta_ic': ic,
+            'meta_mse': mse,
+            'n_samples': n_samples
+        }
+    
+    def _build_nn_meta_model(self, input_size: int):
+        """Build small neural network meta-learner."""
+        class MetaNN(nn.Module):
+            def __init__(self, input_size):
+                super().__init__()
+                self.net = nn.Sequential(
+                    nn.Linear(input_size, 64),
+                    nn.ReLU(),
+                    nn.Dropout(0.2),
+                    nn.Linear(64, 32),
+                    nn.ReLU(),
+                    nn.Linear(32, 1)
+                )
+            def forward(self, x):
+                return self.net(x)
+        return MetaNN(input_size).to(self.device)
+    
+    def _train_nn_meta(self, X: np.ndarray, y: np.ndarray, epochs: int = 50):
+        """Train NN meta-learner."""
+        X_t = torch.FloatTensor(X).to(self.device)
+        y_t = torch.FloatTensor(y).unsqueeze(1).to(self.device)
+        
+        optimizer = torch.optim.Adam(self.meta_model.parameters(), lr=1e-3)
+        criterion = nn.MSELoss()
+        
+        for epoch in range(epochs):
+            self.meta_model.train()
+            optimizer.zero_grad()
+            pred = self.meta_model(X_t)
+            loss = criterion(pred, y_t)
+            loss.backward()
+            optimizer.step()
+    
+    def predict_meta(self,
+                     predictions: Dict[str, np.ndarray],
+                     regimes: Optional[List[str]] = None,
+                     volatilities: Optional[np.ndarray] = None) -> np.ndarray:
+        """Generate meta-model predictions."""
+        if not self.is_fitted:
+            # Fallback: equal weight
+            preds = [predictions.get(m, np.zeros(len(list(predictions.values())[0]))) 
+                     for m in self.base_models]
+            return np.mean(preds, axis=0)
+        
+        X_meta = self._build_meta_features(
+            predictions,
+            regime=regimes[0] if regimes else None,
+            volatility=volatilities[0] if volatilities is not None else None
+        )
+        
+        if self.meta_learner_type == 'nn':
+            self.meta_model.eval()
+            with torch.no_grad():
+                X_t = torch.FloatTensor(X_meta).to(self.device)
+                pred = self.meta_model(X_t).cpu().numpy().flatten()
+        else:
+            pred = self.meta_model.predict(X_meta)
+        
+        return pred
+    
+    def update_performance(self, model_name: str, prediction: np.ndarray, actual: np.ndarray):
+        """Update rolling performance history for a base model."""
+        from scipy.stats import spearmanr
+        ic, _ = spearmanr(prediction, actual)
+        if np.isnan(ic):
+            ic = 0.0
+        self.performance_history[model_name].append(ic)
+        # Keep only lookback window
+        self.performance_history[model_name] = self.performance_history[model_name][-self.lookback_window:]
+    
+    def get_model_weights(self) -> Dict[str, float]:
+        """Get current implied weights from performance history."""
+        weights = {}
+        total_ic = 0
+        for model in self.base_models:
+            perf = self.performance_history.get(model, [0.0])
+            avg_ic = np.mean(perf) if perf else 0.0
+            # Use max(0, ic) to avoid negative weights, or use signed weights
+            weight = max(avg_ic, 0.0)
+            weights[model] = weight
+            total_ic += weight
+        
+        if total_ic > 0:
+            weights = {k: v / total_ic for k, v in weights.items()}
+        else:
+            # Equal weight fallback
+            weights = {k: 1.0 / len(self.base_models) for k in self.base_models}
+        
+        return weights
+    
+    def adaptive_predict(self,
+                         predictions: Dict[str, np.ndarray],
+                         actual_prev: Optional[np.ndarray] = None,
+                         regime: Optional[str] = None) -> Tuple[np.ndarray, Dict[str, float]]:
+        """
+        Adaptive prediction that updates weights based on recent performance.
+        
+        Returns:
+            final_predictions, current_weights
+        """
+        # Update performance if previous actuals available
+        if actual_prev is not None:
+            for model, pred in predictions.items():
+                if len(pred) == len(actual_prev):
+                    self.update_performance(model, pred, actual_prev)
+        
+        # Get adaptive weights
+        weights = self.get_model_weights()
+        self.weight_history.append(weights)
+        
+        # Weighted combination
+        final_pred = np.zeros(len(list(predictions.values())[0]))
+        for model, weight in weights.items():
+            if model in predictions:
+                final_pred += weight * predictions[model]
+        
+        return final_pred, weights