# Code source de l'algorithme - pret pour deploiement QC Cloud
qc_code = '''#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)
# 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 an 18-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)}"
)
'''
print("Seuil : P(hausse) > 0.52, max 4 positions, min 2")
Comment lire ces métriques
min_positions=2maintient de l’exposition même quand peu de signaux dépassent le seuil : en régime baissier durable, cette exposition forcée amplifie la séquence de pertes.