Upload pipeline.py

pipeline.py

@@ -0,0 +1,280 @@
+"""Unified Pipeline - Orchestrates all AlphaForge components end-to-end.
+
+INPUT: Market data (OHLCV), news feed, macro data, options chain
+OUTPUT: Portfolio weights, risk metrics, PnL, dashboards
+
+This is the central brain — one class to rule them all.
+"""
+import numpy as np; import pandas as pd; import torch; import json; import os
+from typing import Dict, List, Optional, Tuple, Any
+from datetime import datetime, timedelta
+import warnings; warnings.filterwarnings('ignore')
+
+class AlphaForgePipeline:
+    """Production-grade unified pipeline: data → alpha → risk → weights → backtest."""
+
+    def __init__(self, config: Optional[Dict] = None):
+        self.config = config or self.default_config()
+        self._init_components()
+        self.state = {'pnl': [], 'weights': [], 'alerts': [], 'signals': {}, 'regime': 'neutral'}
+
+    # ── default configuration ───────────────────────────────────────────
+    @staticmethod
+    def default_config() -> Dict:
+        return {
+            'tickers': ['SPY','QQQ','AAPL','MSFT','GOOGL','AMZN','META','NVDA','TSLA','JPM','V','WMT','XLF','XLK','XLE'],
+            'lookback': 60, 'horizon': 5,
+            'rebalance_freq': 'W',  # D=Daily, W=Weekly, M=Monthly
+            'alpha': {'lstm_hidden':128,'lstm_layers':2,'trans_d_model':128,'trans_nhead':4,'xgb_depth':6,'xgb_lr':0.05,'xgb_estimators':200,
+                       'ensemble_weights':{'lstm':0.3,'transformer':0.3,'xgboost':0.4},'epochs':50,'device':'cpu'},
+            'sentiment': {'enabled':True,'model':'ProsusAI/finbert','weight':0.3,'window':5},
+            'volatility': {'garch_p':1,'garch_q':1,'garch_dist':'t','lstm_hidden':64},
+            'portfolio': {'max_weight':0.25,'risk_aversion':2.0,'transaction_cost':0.0003,'target_return':None},
+            'risk': {'var_conf':[0.95,0.99],'max_drawdown_threshold':-0.10,'scaling_factor':2.0},
+            'online': {'enable_drift_detection':True,'adaptation_window':21,'drift_threshold':0.3},
+            'advanced_features': True,
+            'include_macro': True,
+            'include_sentiment': True,
+        }
+
+    def _init_components(self):
+        """Lazy-init all model components."""
+        self._alpha_model = None
+        self._sentiment_model = None
+        self._vol_engine = None
+        self._optimizer = None
+        self._risk_engine = None
+        self._feature_engine = None
+
+    # ── data pipeline ───────────────────────────────────────────────────
+    def fetch_market_data(self, start: str, end: str) -> Dict[str, pd.DataFrame]:
+        """Fetch and preprocess market data."""
+        from market_data import MarketDataPipeline
+        pipeline = MarketDataPipeline(self.config['tickers'], start, end)
+        return pipeline.fetch_data()
+
+    def build_features(self, data: Dict[str, pd.DataFrame]) -> pd.DataFrame:
+        """Build advanced feature matrix (90+ cols)."""
+        if self.config['advanced_features']:
+            from advanced_features_part1 import MicrostructureFeatures, CrossSectionalFeatures
+            from macro_features import MacroFeatures
+            from regime_features import RegimeFeatures
+            from technical_indicators import AdvancedTechnical
+            all_features = []
+            for ticker, df in data.items():
+                close = np.array(df['Close']).flatten(); high = np.array(df['High']).flatten()
+                low = np.array(df['Low']).flatten(); vol = np.array(df['Volume']).flatten()
+                cs = pd.Series(close, index=df.index); hs = pd.Series(high, index=df.index)
+                ls = pd.Series(low, index=df.index); vs = pd.Series(vol, index=df.index)
+                f = pd.DataFrame(index=df.index)
+                f['ticker'] = ticker; f['close'] = close
+                for col_df in [
+                    MicrostructureFeatures.compute_all(cs,hs,ls,vs),
+                    RegimeFeatures.volatility_regime(cs.pct_change().fillna(0)),
+                    RegimeFeatures.liquidity_regime(vs,cs),
+                    RegimeFeatures.trend_regime(cs),
+                    AdvancedTechnical.ichimoku(cs,hs,ls),
+                    AdvancedTechnical.supertrend(cs,hs,ls),
+                    AdvancedTechnical.keltner_channels(cs,hs,ls),
+                    AdvancedTechnical.volume_profile(cs,vs,hs,ls),
+                ]:
+                    for c in col_df.columns: f[c] = col_df[c].values
+                all_features.append(f)
+            features_df = pd.concat(all_features, axis=0)
+            if self.config['include_macro']:
+                macro = MacroFeatures._synthetic_macro(str(features_df.index[0])[:10], str(features_df.index[-1])[:10])
+                for c in macro.columns: features_df[f'macro_{c}'] = macro[c].reindex(features_df.index).ffill()
+            # z-score normalize
+            nc = [c for c in features_df.columns if c not in ['ticker','close']]
+            for ticker in features_df['ticker'].unique():
+                m = features_df['ticker'] == ticker
+                for col in nc:
+                    s = features_df.loc[m, col]; rm = s.rolling(42).mean(); rs = s.rolling(42).std().replace(0,1)
+                    features_df.loc[m, col] = (s - rm) / rs
+            return features_df.replace([np.inf, -np.inf], 0).fillna(0)
+        else:
+            from market_data import MarketDataPipeline
+            return MarketDataPipeline(self.config['tickers'], '', '').create_feature_matrix()
+
+    # ── model training ──────────────────────────────────────────────────
+    def train_alpha(self, X: np.ndarray, y: np.ndarray, X_val=None, y_val=None) -> Dict:
+        """Train the alpha model ensemble."""
+        from alpha_model import AlphaEnsemble
+        ac = self.config['alpha']
+        self._alpha_model = AlphaEnsemble(
+            input_size=X.shape[2], seq_len=X.shape[1],
+            lstm_hidden=ac['lstm_hidden'], lstm_layers=ac['lstm_layers'],
+            trans_d_model=ac['trans_d_model'], trans_nhead=ac['trans_nhead'],
+            xgb_depth=ac['xgb_depth'], xgb_lr=ac['xgb_lr'], xgb_estimators=ac['xgb_estimators'],
+            weights=ac['ensemble_weights'], device=ac['device']
+        )
+        return self._alpha_model.fit(X, y, X_val, y_val, epochs=ac['epochs'])
+
+    def predict_alpha(self, X: np.ndarray) -> np.ndarray:
+        """Generate alpha predictions."""
+        if self._alpha_model is None:
+            raise RuntimeError("Alpha model not trained")
+        return self._alpha_model.predict(X)
+
+    # ── portfolio optimization ──────────────────────────────────────────
+    def optimize_portfolio(self, mu: np.ndarray, Sigma: np.ndarray,
+                           current_weights: Optional[np.ndarray] = None) -> Dict:
+        """Optimize portfolio weights."""
+        from portfolio_optimizer import PortfolioOptimizer
+        pc = self.config['portfolio']
+        opt = PortfolioOptimizer(
+            max_weight=pc['max_weight'], risk_aversion=pc['risk_aversion'],
+            transaction_cost=pc['transaction_cost']
+        )
+        return opt.optimize_max_sharpe(mu, Sigma, current_weights)
+
+    # ── risk analytics ──────────────────────────────────────────────────
+    def compute_risk_metrics(self, returns: np.ndarray, weights: np.ndarray,
+                             returns_df: pd.DataFrame) -> Dict:
+        """Compute comprehensive risk metrics."""
+        from risk_engine import RiskEngine
+        rc = self.config['risk']
+        risk = RiskEngine(confidence_levels=rc['var_conf'])
+        port_ret = returns_df.dot(weights) if returns_df.shape[1] == len(weights) else np.dot(returns, weights)
+        return {
+            **risk.compute_all_var(port_ret.values if hasattr(port_ret,'values') else port_ret),
+            **risk.compute_tail_risk(port_ret.values if hasattr(port_ret,'values') else port_ret),
+            'portfolio_var': risk.portfolio_var(weights, returns_df, 'parametric', 0.95)
+        }
+
+    # ── full pipeline execution ─────────────────────────────────────────
+    def run(self, start: str, end: str, mode: str = 'backtest') -> Dict[str, Any]:
+        """Run full pipeline end-to-end."""
+        print(f"🚀 AlphaForge Pipeline: {start} → {end}")
+
+        # 1. Data
+        data = self.fetch_market_data(start, end)
+        features = self.build_features(data)
+
+        # 2. Sequences
+        from market_data import MarketDataPipeline
+        pipeline = MarketDataPipeline(self.config['tickers'], start, end)
+        X, y, tickers, dates = pipeline.create_sequences(features, self.config['lookback'], self.config['horizon'])
+        n = len(X)
+        X_train, y_train = X[:int(n*0.7)], y[:int(n*0.7)]
+        X_test, y_test = X[int(n*0.85):], y[int(n*0.85):]
+
+        # 3. Alpha
+        self.train_alpha(X_train, y_train)
+        alpha_pred = self.predict_alpha(X_test)
+        from backtest_engine import compute_information_coefficient
+        ic = compute_information_coefficient(pd.Series(alpha_pred), pd.Series(y_test), by_date=False)
+
+        # 4. Volatility
+        from volatility_model import VolatilityEngine
+        vc = self.config['volatility']
+        vol_engine = VolatilityEngine(garch_p=vc['garch_p'], garch_q=vc['garch_q'], garch_dist=vc['garch_dist'])
+        returns_dict = {}
+        for t in self.config['tickers']:
+            if t in data:
+                c = np.array(data[t]['Close']).flatten()
+                returns_dict[t] = pd.Series(np.log(c[1:]/c[:-1]), index=data[t].index[1:])
+        returns_df = pd.DataFrame(returns_dict).fillna(0)
+
+        # 5. Portfolio
+        pred_df = pd.DataFrame({'date': dates[int(n*0.85):], 'ticker': tickers[int(n*0.85):],
+                                 'predicted_return': alpha_pred, 'actual_return': y_test})
+        test_dates = sorted(pd.to_datetime(pred_df['date'].unique()))
+        weights_history = []
+        for rd in test_dates[::5]:  # Weekly rebalance
+            dp = pred_df[pred_df['date'] == rd]
+            if len(dp) < 3: continue
+            mu = dp.set_index('ticker')['predicted_return'].reindex(self.config['tickers']).fillna(0).values
+            try:
+                cov = vol_engine.build_covariance_matrix(returns_df, rd)
+                cov = cov.reindex(index=self.config['tickers'], columns=self.config['tickers']).fillna(0).values
+            except: cov = np.eye(len(self.config['tickers'])) * 0.04
+            result = self.optimize_portfolio(mu, cov)
+            weights_history.append(pd.Series(result['weights'], index=self.config['tickers'], name=rd))
+
+        if not weights_history:
+            return {'error': 'No valid rebalance dates'}
+
+        weights_df = pd.DataFrame(weights_history)
+
+        # 6. Backtest
+        from backtest_engine import BacktestEngine, RegimeDetector
+        bt = BacktestEngine(initial_capital=1_000_000)
+        bt_returns = returns_df.reindex(weights_df.index).fillna(0)
+        metrics = bt.run_backtest(bt_returns, weights_df, rebalance_dates=weights_df.index)
+
+        # 7. Risk
+        risk = self.compute_risk_metrics(np.array(bt.returns_history), weights_df.iloc[-1].values,
+                                          bt_returns)
+
+        # 8. Regime
+        if 'SPY' in returns_df.columns:
+            rdet = RegimeDetector()
+            spy_r = returns_df['SPY'].reindex(weights_df.index).fillna(0)
+            rdet.detect_regimes(spy_r)
+            regime_stats = rdet.get_regime_stats(spy_r)
+
+        return {
+            'metrics': metrics,
+            'ic': ic,
+            'risk': risk,
+            'regime_stats': regime_stats.to_dict() if 'regime_stats' in dir() else None,
+            'weights': weights_df.tail(10).to_dict(),
+            'n_signals': len(alpha_pred),
+            'feature_count': X.shape[2],
+        }
+
+
+# ── Hyperparameter Sweep Engine ─────────────────────────────────────────────
+
+class HyperparameterSweeper:
+    """Grid search over alpha model hyperparameters."""
+
+    def __init__(self, config_grid: Dict[str, List]):
+        self.grid = config_grid
+        self.results = []
+
+    def run(self, X: np.ndarray, y: np.ndarray, n_splits: int = 3) -> pd.DataFrame:
+        from itertools import product
+        keys = list(self.grid.keys())
+        combos = list(product(*self.grid.values()))
+        print(f"🧹 Sweeping {len(combos)} hyperparameter combinations...")
+
+        for i, combo in enumerate(combos):
+            params = dict(zip(keys, combo))
+            print(f"  [{i+1}/{len(combos)}] {params}")
+
+            # Walk-forward validation
+            from alpha_model import AlphaEnsemble
+            n = len(X)
+            fold_size = n // (n_splits + 1)
+            ics = []
+
+            for fold in range(n_splits):
+                train_end = (fold + 1) * fold_size
+                val_end = train_end + fold_size
+                X_f, y_f = X[:train_end], y[:train_end]
+                X_v, y_v = X[train_end:val_end], y[train_end:val_end]
+                if len(X_v) < 10: continue
+                model = AlphaEnsemble(
+                    input_size=X.shape[2], seq_len=X.shape[1],
+                    lstm_hidden=params.get('lstm_hidden',128),
+                    lstm_layers=params.get('lstm_layers',2),
+                    trans_d_model=params.get('trans_d_model',128),
+                    xgb_depth=params.get('xgb_depth',6),
+                    xgb_lr=params.get('xgb_lr',0.05),
+                    xgb_estimators=params.get('xgb_estimators',200),
+                    device=params.get('device','cpu')
+                )
+                model.fit(X_f, y_f, X_v, y_v, epochs=params.get('epochs',30))
+                from backtest_engine import compute_information_coefficient
+                pred = model.predict(X_v)
+                ic = compute_information_coefficient(pd.Series(pred), pd.Series(y_v), by_date=False)
+                ics.append(ic['mean_ic'])
+
+            result = {**params, 'mean_ic': np.mean(ics), 'std_ic': np.std(ics), 'fold_ics': ics}
+            self.results.append(result)
+
+        df = pd.DataFrame(self.results).sort_values('mean_ic', ascending=False)
+        print(f"\n✅ Best IC: {df['mean_ic'].iloc[0]:.4f} with params: {dict(df.iloc[0][list(keys)])}")
+        return df