Upload strategy_ensemble.py

strategy_ensemble.py

@@ -0,0 +1,100 @@
+"""Strategy Ensemble - Dynamic capital allocation across strategies."""
+import numpy as np
+import pandas as pd
+from typing import Dict, List, Optional
+import warnings
+warnings.filterwarnings('ignore')
+
+
+class StrategyEnsemble:
+    """Multi-strategy system with dynamic capital allocation."""
+    
+    def __init__(self):
+        self.strategies = {
+            'momentum': {'weight': 0.25, 'sharpe': 0.8, 'track_record': []},
+            'mean_reversion': {'weight': 0.25, 'sharpe': 0.6, 'track_record': []},
+            'arbitrage': {'weight': 0.20, 'sharpe': 1.2, 'track_record': []},
+            'ml_alpha': {'weight': 0.15, 'sharpe': 1.0, 'track_record': []},
+            'sentiment': {'weight': 0.10, 'sharpe': 0.5, 'track_record': []},
+            'options': {'weight': 0.05, 'sharpe': 0.9, 'track_record': []}
+        }
+        self.capital = 1_000_000
+        self.allocation_history = []
+        
+    def update_performance(self, strategy_returns: Dict[str, float]):
+        """Update rolling Sharpe for each strategy."""
+        for name, ret in strategy_returns.items():
+            if name in self.strategies:
+                self.strategies[name]['track_record'].append(ret)
+                # Keep last 63 days
+                if len(self.strategies[name]['track_record']) > 63:
+                    self.strategies[name]['track_record'] = self.strategies[name]['track_record'][-63:]
+                
+                # Update Sharpe
+                records = self.strategies[name]['track_record']
+                if len(records) > 5:
+                    mean_ret = np.mean(records)
+                    std_ret = np.std(records)
+                    self.strategies[name]['sharpe'] = mean_ret / std_ret * np.sqrt(252) if std_ret > 0 else 0
+    
+    def allocate_capital(self, max_weight: float = 0.35, min_weight: float = 0.02) -> Dict[str, float]:
+        """
+        Dynamic capital allocation based on:
+        - Recent Sharpe ratio
+        - Strategy correlation (for diversification)
+        - Risk budget
+        """
+        # Get Sharpe ratios
+        sharpes = {name: max(s['sharpe'], 0.1) for name, s in self.strategies.items()}
+        
+        # Square Sharpe for more differentiation
+        weights = {name: s ** 2 for name, s in sharpes.items()}
+        total = sum(weights.values())
+        
+        if total > 0:
+            weights = {name: w / total for name, w in weights.items()}
+        else:
+            n = len(weights)
+            weights = {name: 1.0 / n for name in weights}
+        
+        # Apply constraints
+        for name in weights:
+            weights[name] = np.clip(weights[name], min_weight, max_weight)
+        
+        # Renormalize
+        total = sum(weights.values())
+        weights = {name: w / total for name, w in weights.items()}
+        
+        self.allocation_history.append(weights)
+        return weights
+    
+    def allocate_dollar(self, total_capital: float, weights: Dict[str, float]) -> Dict[str, float]:
+        """Convert weight allocation to dollar amounts."""
+        return {name: total_capital * w for name, w in weights.items()}
+    
+    def combine_signals(self, signals: Dict[str, np.ndarray], weights: Optional[Dict[str, float]] = None) -> np.ndarray:
+        """Combine signals from multiple strategies."""
+        if weights is None:
+            weights = self.allocate_capital()
+        
+        n_samples = len(list(signals.values())[0])
+        combined = np.zeros(n_samples)
+        
+        for name, signal in signals.items():
+            if name in weights:
+                combined += weights[name] * signal
+        
+        return combined
+    
+    def get_performance_report(self) -> pd.DataFrame:
+        """Generate strategy performance report."""
+        report = []
+        for name, strat in self.strategies.items():
+            report.append({
+                'strategy': name,
+                'current_weight': self.allocation_history[-1].get(name, 0) if self.allocation_history else strat['weight'],
+                'sharpe': strat['sharpe'],
+                'max_weight': max(a.get(name, 0) for a in self.allocation_history) if self.allocation_history else 0,
+                'track_record_len': len(strat['track_record'])
+            })
+        return pd.DataFrame(report)