Upload online_learning.py

online_learning.py

@@ -1,88 +1,109 @@
-"""Online Learning - Adaptive model retraining with concept drift detection."""
+"""Online Learning - Adaptive model updates with drift detection"""
 import numpy as np
 import pandas as pd
-from sklearn.linear_model import SGDRegressor
+import torch
+import torch.nn as nn
 from typing import Dict, Optional
-import warnings
-warnings.filterwarnings('ignore')
+from scipy.stats import ks_2samp
 
 
-class OnlineLearner:
-    """Incremental learning for adaptive alpha models."""
+class DriftDetector:
+    """Detect data drift using statistical tests"""
     
-    def __init__(self, lookback_window: int = 252):
-        self.lookback_window = lookback_window
-        self.model = SGDRegressor(loss='squared_error', penalty='l2', alpha=0.0001,
-                                   learning_rate='adaptive', eta0=0.01, max_iter=1,
-                                   warm_start=True, random_state=42)
-        self.feature_buffer = []
-        self.target_buffer = []
-        self.is_fitted = False
-        self.update_count = 0
-        self.drift_scores = []
-        
-    def partial_fit(self, X: np.ndarray, y: np.ndarray):
-        """Incrementally update model."""
-        self.feature_buffer.append(X)
-        self.target_buffer.append(y)
-        
-        if len(self.feature_buffer) > self.lookback_window:
-            self.feature_buffer = self.feature_buffer[-self.lookback_window:]
-            self.target_buffer = self.target_buffer[-self.lookback_window:]
-        
-        X_batch = np.vstack(self.feature_buffer)
-        y_batch = np.concatenate(self.target_buffer)
-        
-        if not self.is_fitted:
-            self.model.fit(X_batch, y_batch); self.is_fitted = True
-        else:
-            self.model.partial_fit(X_batch, y_batch)
+    def __init__(self, significance=0.05, window=252):
+        self.significance = significance
+        self.window = window
+        self.reference_stats = {}
+        self.drift_history = []
+    
+    def set_reference(self, data: np.ndarray, name: str = 'default'):
+        self.reference_stats[name] = {'mean': data.mean(axis=0), 'std': data.std(axis=0), 'data': data}
+    
+    def detect_ks(self, new_data: np.ndarray, name: str = 'default') -> Dict:
+        ref = self.reference_stats.get(name)
+        if ref is None:
+            return {'drift': False, 'p_value': 1.0}
+        n_features = new_data.shape[1] if new_data.ndim > 1 else 1
+        drifts = []
+        p_values = []
+        for i in range(n_features):
+            col = i if new_data.ndim > 1 else 0
+            ref_feat = ref['data'][:, col] if ref['data'].ndim > 1 else ref['data']
+            new_feat = new_data[:, col] if new_data.ndim > 1 else new_data
+            stat, p = ks_2samp(ref_feat, new_feat)
+            drifts.append(p < self.significance)
+            p_values.append(p)
+        n_drift = sum(drifts)
+        overall_drift = n_drift > n_features * 0.3
+        return {'drift': overall_drift, 'p_values': p_values, 'n_features_drifted': n_drift, 'total_features': n_features}
+    
+    def detect_cusum(self, residuals: np.ndarray, threshold: float = 5.0, drift: float = 1.0) -> Dict:
+        pos_cusum = np.zeros(len(residuals))
+        neg_cusum = np.zeros(len(residuals))
         
-        self.update_count += 1
+        for t in range(1, len(residuals)):
+            pos_cusum[t] = max(0, pos_cusum[t-1] + residuals[t] - drift)
+            neg_cusum[t] = min(0, neg_cusum[t-1] + residuals[t] + drift)
         
-    def predict(self, X: np.ndarray) -> np.ndarray:
-        return self.model.predict(X) if self.is_fitted else np.zeros(len(X))
-    
-    def get_drift_score(self, recent_X: np.ndarray, recent_y: np.ndarray) -> float:
-        """Detect concept drift."""
-        if not self.is_fitted: return 0.0
-        pred = self.predict(recent_X)
-        recent_mse = np.mean((pred - recent_y) ** 2)
-        all_pred = self.predict(np.vstack(self.feature_buffer))
-        all_y = np.concatenate(self.target_buffer)
-        historical_mse = np.mean((all_pred - all_y) ** 2)
-        drift = recent_mse / (historical_mse + 1e-8) - 1.0
-        self.drift_scores.append(drift)
-        return drift
+        alert = np.any(pos_cusum > threshold) or np.any(neg_cusum < -threshold)
+        return {'alert': alert, 'pos_cusum': pos_cusum, 'neg_cusum': neg_cusum}
 
 
-class AdaptiveEnsemble:
-    """Ensemble adapting weights based on recent IC."""
+class OnlineLearner:
+    """Online learning with periodic model adaptation"""
     
-    def __init__(self, models: Dict[str, object], adaptation_rate: float = 0.1):
-        self.models = models
-        self.weights = {name: 1.0 / len(models) for name in models}
-        self.adaptation_rate = adaptation_rate
+    def __init__(self, model: nn.Module, lr: float = 1e-5, 
+                 adaptation_window: int = 21, drift_threshold: float = 0.3):
+        self.model = model
+        self.lr = lr
+        self.adaptation_window = adaptation_window
+        self.drift_threshold = drift_threshold
+        self.drift_detector = DriftDetector()
+        self.adaptation_count = 0
+        self.ic_history = []
+        self.performance_history = []
+    
+    def check_and_adapt(self, X_new: np.ndarray, y_new: np.ndarray, 
+                        X_ref: Optional[np.ndarray] = None) -> Dict:
+        drift_result = self.drift_detector.detect_ks(X_new)
         
-    def predict(self, X: np.ndarray) -> np.ndarray:
-        final = np.zeros(len(X))
-        for name, model in self.models.items():
-            try: final += self.weights[name] * model.predict(X)
-            except: pass
-        return final
+        if drift_result['drift']:
+            print(f"⚠️ Drift detected: {drift_result['n_features_drifted']}/{drift_result['total_features']} features shifted")
+            self._adapt(X_new, y_new)
+            self.adaptation_count += 1
+            return {'adapted': True, 'drift': drift_result}
+        return {'adapted': False, 'drift': drift_result}
     
-    def update_weights(self, predictions: Dict[str, np.ndarray], actual: np.ndarray):
-        from scipy.stats import spearmanr
-        ics = {}
-        for name, pred in predictions.items():
-            ic, _ = spearmanr(pred, actual)
-            ics[name] = abs(ic) if not np.isnan(ic) else 0.0
+    def _adapt(self, X: np.ndarray, y: np.ndarray, epochs: int = 5):
+        self.model.train()
+        optimizer = torch.optim.Adam(self.model.parameters(), lr=self.lr)
+        criterion = nn.MSELoss()
         
-        total = sum(ics.values()) + 1e-8
-        target = {name: ic / total for name, ic in ics.items()}
+        X_t = torch.FloatTensor(X)
+        y_t = torch.FloatTensor(y).unsqueeze(1)
+        
+        for epoch in range(epochs):
+            optimizer.zero_grad()
+            pred = self.model(X_t)
+            loss = criterion(pred, y_t)
+            loss.backward()
+            optimizer.step()
+        
+        print(f"  Adapted model with {epochs} epochs, loss={loss.item():.6f}")
+    
+    def track_performance(self, predictions: np.ndarray, actuals: np.ndarray):
+        from scipy.stats import spearmanr
+        ic, _ = spearmanr(predictions, actuals)
+        self.ic_history.append(ic)
         
-        for name in self.weights:
-            self.weights[name] = (1 - self.adaptation_rate) * self.weights[name] + self.adaptation_rate * target[name]
+        # Check if IC is degrading
+        if len(self.ic_history) > 63:
+            recent_ic = np.mean(self.ic_history[-21:])
+            long_ic = np.mean(self.ic_history[-63:])
+            degradation = (long_ic - recent_ic) / (abs(long_ic) + 1e-8)
+            
+            if degradation > 0.3:
+                print(f"⚠️ IC degradation: recent={recent_ic:.4f}, long={long_ic:.4f}, degradation={degradation:.2%}")
+                return 'degrading'
         
-        total_w = sum(self.weights.values())
-        self.weights = {k: v / total_w for k, v in self.weights.items()}
+        return 'stable'