"""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