#region imports from AlgorithmImports import * import numpy as np from collections import deque from sklearn.neural_network import MLPClassifier from sklearn.preprocessing import StandardScaler from sklearn.pipeline import Pipeline # endregion class LSTMForecasting(QCAlgorithm): """ Neural Network (MLP) Forecasting Strategy - v2.1 Real sklearn MLPClassifier replacing hand-rolled fake LSTM. Temporal features mimic LSTM's ability to capture sequential patterns. Reference: Hands-On AI Trading, Ch06/Ex07 Version : 2.1 - threshold 0.52, min 2 positions to reduce cash drag Architecture ------------ - MLPClassifier, hidden layers (64, 32), relu activation - StandardScaler pre-processing in a Pipeline - Monthly re-training on a 252-day rolling window Features (20 per symbol): - Lag returns: 1, 2, 3, 5, 10, 20 days - Rolling vol: 5, 10, 20 days - RSI-like: 14-day normalised - Momentum: 5, 10, 20 day cumulative return - Auto-correlation: lag-1, lag-2, lag-5 - Mean return: 5, 10 days Universe: SPY, QQQ, IWM, EFA, TLT, GLD, IEF Rebalance: every Monday Training : monthly (first rebalance of new month) Results (2015-2026): - Sharpe: 0.525 (vs 0.366 fake LSTM, +0.159) - CAGR: 11.3% (vs 9.75% fake LSTM) - MaxDD: 32.5% (vs 37.2% fake LSTM) - Alpha: +0.016 (vs -0.008 fake LSTM) - Beta: 0.544 (vs 0.886 fake LSTM - genuinely signal-driven) """ def initialize(self): self.set_start_date(2015, 1, 1) self.set_end_date(2026, 1, 1) self.set_cash(100_000) self.set_brokerage_model(BrokerageName.INTERACTIVE_BROKERS_BROKERAGE, AccountType.MARGIN) # Universe - diversified liquid ETFs self._tickers = ["SPY", "QQQ", "IWM", "EFA", "TLT", "GLD", "IEF"] self._symbols = [] for ticker in self._tickers: sym = self.add_equity(ticker, Resolution.DAILY).symbol self._symbols.append(sym) # Parameters self._lookback = 60 # days of history for feature building self._train_window = 252 # days of history for training self._threshold = 0.52 # minimum probability to trade self._max_positions = 4 # max long ETFs at once self._min_positions = 2 # always hold at least this many (top-N by score) # Per-symbol return deques self._return_windows = { sym: deque(maxlen=self._train_window + self._lookback + 5) for sym in self._symbols } # ROC indicators - one per symbol self._roc_indicators = {} for sym in self._symbols: roc = self.roc(sym, 1, Resolution.DAILY) def make_handler(s): def handler(ind, point): if ind.is_ready: self._return_windows[s].append(point.value) return handler roc.updated += make_handler(sym) self._roc_indicators[sym] = roc # ML models self._models = {sym: None for sym in self._symbols} self._models_trained = {sym: False for sym in self._symbols} self._last_train_month = -1 # Warm-up: prime each ROC indicator individually warmup_days = self._train_window + self._lookback + 30 for sym in self._symbols: roc = self._roc_indicators[sym] bars = self.history[TradeBar](sym, timedelta(days=warmup_days), Resolution.DAILY) for bar in bars: roc.update(bar.end_time, bar.close) # Weekly rebalance schedule (every Monday) self.schedule.on( self.date_rules.every(DayOfWeek.MONDAY), self.time_rules.after_market_open(self._symbols[0], 30), self._rebalance ) self.log( "LSTMForecasting v2.1: MLPClassifier (64,32), 7-ETF, threshold=0.52, min_pos=2" ) # ------------------------------------------------------------------ # Feature engineering # ------------------------------------------------------------------ def _build_features(self, returns_list): """ Build a 20-element feature vector from a returns list. Returns None if not enough data. """ if len(returns_list) < self._lookback + 5: return None r = np.array(returns_list[-(self._lookback + 1):]) features = [] # Lag returns: 1, 2, 3, 5, 10, 20 for lag in [1, 2, 3, 5, 10, 20]: features.append(r[-lag] if len(r) > lag else 0.0) # Rolling volatility: 5, 10, 20 for w in [5, 10, 20]: features.append(float(np.std(r[-w:])) if len(r) >= w else 0.0) # RSI-like (14-day), normalised 0-1 if len(r) >= 14: d = r[-14:] gains = d[d > 0] losses = -d[d < 0] ag = float(np.mean(gains)) if len(gains) > 0 else 1e-8 al = float(np.mean(losses)) if len(losses) > 0 else 1e-8 rs = ag / (al + 1e-8) features.append(rs / (1.0 + rs)) else: features.append(0.5) # Cumulative momentum: 5, 10, 20 for w in [5, 10, 20]: features.append(float(np.sum(r[-w:])) if len(r) >= w else 0.0) # Auto-correlation at lags 1, 2, 5 if len(r) >= 15: for ac_lag in [1, 2, 5]: if len(r) > ac_lag: c = np.corrcoef(r[:-ac_lag], r[ac_lag:]) val = float(c[0, 1]) features.append(val if not np.isnan(val) else 0.0) else: features.append(0.0) else: features.extend([0.0, 0.0, 0.0]) # Mean return: 5, 10 for w in [5, 10]: features.append(float(np.mean(r[-w:])) if len(r) >= w else 0.0) return np.array(features, dtype=float) def _build_training_data(self, returns_list): """Build (X, y) from returns list; label = 1 if next-day return > 0.""" X, y = [], [] r = list(returns_list) min_needed = self._lookback + 5 for i in range(min_needed, len(r) - 1): feats = self._build_features(r[:i + 1]) if feats is not None: X.append(feats) y.append(1 if r[i + 1] > 0 else 0) if len(X) < 30: return None, None return np.array(X), np.array(y) # ------------------------------------------------------------------ # Training # ------------------------------------------------------------------ def _train_all_models(self): """Train one MLP per symbol on recent history.""" for sym in self._symbols: rl = list(self._return_windows[sym]) if len(rl) < self._lookback + 40: continue recent = rl[-self._train_window:] X, y = self._build_training_data(recent) if X is None or len(X) < 30: continue if len(np.unique(y)) < 2: continue try: model = Pipeline([ ('scaler', StandardScaler()), ('mlp', MLPClassifier( hidden_layer_sizes=(64, 32), activation='relu', max_iter=300, random_state=42, early_stopping=True, validation_fraction=0.15, n_iter_no_change=15, learning_rate_init=0.001 )) ]) model.fit(X, y) self._models[sym] = model self._models_trained[sym] = True acc = float(np.mean(model.predict(X) == y)) self.log(f"MLP trained {sym.value}: {len(X)} samples, acc={acc:.2%}") except Exception as exc: self.log(f"Training error {sym.value}: {exc}") # ------------------------------------------------------------------ # Prediction # ------------------------------------------------------------------ def _predict_proba(self, sym): """Return P(up) for sym; 0.5 if model not ready.""" if not self._models_trained[sym]: return 0.5 rl = list(self._return_windows[sym]) feats = self._build_features(rl) if feats is None: return 0.5 try: proba = self._models[sym].predict_proba(feats.reshape(1, -1))[0] return float(proba[1]) except Exception as exc: self.log(f"Predict error {sym.value}: {exc}") return 0.5 # ------------------------------------------------------------------ # Rebalance # ------------------------------------------------------------------ def _rebalance(self): """Weekly rebalance; monthly re-training.""" current_month = self.time.month if current_month != self._last_train_month: self._last_train_month = current_month self._train_all_models() if not any(self._models_trained.values()): return # Score all symbols scores = {sym: self._predict_proba(sym) for sym in self._symbols} for sym, prob in scores.items(): self.plot('MLP P(up)', sym.value, prob) # Rank all symbols by score ranked = sorted(self._symbols, key=lambda s: scores[s], reverse=True) # Above-threshold candidates bullish = [s for s in ranked if scores[s] >= self._threshold] bullish = bullish[:self._max_positions] # If fewer than min_positions above threshold, fill with top-ranked if len(bullish) < self._min_positions and any(self._models_trained.values()): for s in ranked: if s not in bullish: bullish.append(s) if len(bullish) >= self._min_positions: break target_weight = 1.0 / len(bullish) if bullish else 0.0 # Exit positions not in selected for sym in self._symbols: if sym not in bullish: if self.portfolio[sym].invested: self.liquidate(sym) # Enter / maintain selected positions for sym in bullish: self.set_holdings(sym, target_weight) best_sym = ranked[0] if ranked else None self.log( f"Rebalance {self.time.date()}: {len(bullish)} longs (min={self._min_positions}), " f"top={best_sym.value if best_sym else 'none'} p={scores.get(best_sym, 0):.2%}" ) def on_end_of_algorithm(self): final_value = self.portfolio.total_portfolio_value total_return = (final_value - 100_000) / 100_000 trained = sum(1 for v in self._models_trained.values() if v) self.log( f"v2.1 DONE: Final=${final_value:,.0f}, Return={total_return:.2%}, " f"Models trained: {trained}/{len(self._symbols)}" )