Objectif: Utiliser les réseaux de neurones récurrents (LSTM/GRU) pour prédire les rendements futurs des actions.
pip install tensorflow (pour GPU: tensorflow-gpu)# Initialisation QuantBook
from AlgorithmImports import *
qb = QuantBook()
# Période d'analyse
qb.SetStartDate(2020, 1, 1)
qb.SetEndDate(2024, 12, 31)
print(f"Période: {qb.StartDate} à {qb.EndDate}")Période: 2020-01-01 00:00:00 à 2024-12-31 23:59:59.999999# Ajouter les indices ETF
tickers = ['SPY', 'QQQ', 'IWM', 'TLT']
symbols = {}
for ticker in tickers:
symbols[ticker] = qb.AddEquity(ticker, Resolution.DAILY).Symbol
# Récupérer les données
# NB: qb.History(lookback) returns a window anchored at qb.Time, which in a fresh
# QuantBook defaults to StartDate -- so the 5y lookback ends near StartDate, not
# EndDate. This is the notebook's committed windowing (pre-existing, unchanged here).
history = qb.History(list(symbols.values()), 365*5, Resolution.Daily)
closes = history['close'].unstack(level=0)
volumes = history['volume'].unstack(level=0)
# Fail-fast guard (C941-L/C942-L, #8734, ai-01 c.38): presence != freshness.
# unstack(level=0) yields a regular DateTimeIndex (time axis), so closes.index[-1]
# is a Timestamp (not a MultiIndex tuple -- cf C950-L DL-LSTM).
# (1) A silently dropna'd ticker must ERROR (add_equity accepts absent symbols).
_missing = [t for t in tickers if str(symbols[t]) not in [str(col) for col in closes.columns]]
if _missing or closes.shape[1] < len(tickers):
raise ValueError(f"Ticker(s) dropped from history (expected {len(tickers)}, got "
f"{closes.shape[1]}). Regenerate via provision_lean_data.py "
f"--universe ml_deep_learning (C941-L).")
# (2) Forward-fill defect detector (C942-L): a truncated zip extended by Lean's
# fillDataForward=True (default) produces a FLAT tail (std==0 on the last bars),
# not real price action. This is window-agnostic -- it catches the defect regardless
# of where the notebook's lookback window lands, unlike a hard year threshold.
_flat = [str(col) for col in closes.columns if closes[col].tail(60).std(ddof=0) == 0]
if _flat:
raise ValueError(f"Flat tail (fillDataForward constant) on {_flat} -- truncated zip "
f"extended as a constant (C942-L, #8734). Regenerate fresh zips via "
f"provision_lean_data.py --universe ml_deep_learning before trusting "
f"post-threshold metrics.")
print(f"Donnees: {closes.shape[0]} jours, {closes.shape[1]} actifs")
print(f"Fenetre effective: {closes.index[0].date()} a {closes.index[-1].date()} "
f"(qb.Time ancres a StartDate -- cf commentaire ci-dessus)")
closes.head()Donnees: 1825 jours, 4 actifs
Fenetre effective: 2012-09-28 a 2019-12-31 (qb.Time ancres a StartDate -- cf commentaire ci-dessus)| symbol | IWM | QQQ | SPY | TLT |
|---|---|---|---|---|
| time | ||||
| 2012-09-28 16:00:00 | 73.955684 | 62.808522 | 121.807762 | NaN |
| 2012-10-01 16:00:00 | 74.203857 | 62.716925 | 122.129266 | NaN |
| 2012-10-02 16:00:00 | 74.345671 | 62.900120 | 122.256176 | NaN |
| 2012-10-03 16:00:00 | 74.132951 | 63.303150 | 122.755353 | NaN |
| 2012-10-04 16:00:00 | 74.638161 | 63.522984 | 123.635259 | NaN |
def create_lstm_features(closes, volumes, lookback=20):
"""Crée les features pour l'entraînement LSTM."""
features = pd.DataFrame()
for ticker in closes.columns:
close = closes[ticker]
volume = volumes[ticker]
# Returns
returns = close.pct_change()
# Log returns
log_returns = np.log(close / close.shift(1))
# Price momentum
momentum_5 = close / close.shift(5) - 1
momentum_10 = close / close.shift(10) - 1
momentum_20 = close / close.shift(20) - 1
# Volatility
volatility = returns.rolling(lookback).std()
# RSI
delta = close.diff()
gain = (delta.where(delta > 0, 0)).rolling(14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
# Volume change
volume_change = volume.pct_change()
# Combiner les features
ticker_features = pd.DataFrame({
f'{ticker}_returns': returns,
f'{ticker}_log_returns': log_returns,
f'{ticker}_mom_5': momentum_5,
f'{ticker}_mom_10': momentum_10,
f'{ticker}_mom_20': momentum_20,
f'{ticker}_volatility': volatility,
f'{ticker}_rsi': rsi,
f'{ticker}_volume_change': volume_change
})
features = pd.concat([features, ticker_features], axis=1)
return features.fillna(0)
# Créer les features
features = create_lstm_features(closes, volumes)
print(f"Features shape: {features.shape}")
print(f"\nColonnes de features:")
print(features.columns.tolist())Features shape: (1825, 32)
Colonnes de features:
['IWM_returns', 'IWM_log_returns', 'IWM_mom_5', 'IWM_mom_10', 'IWM_mom_20', 'IWM_volatility', 'IWM_rsi', 'IWM_volume_change', 'QQQ_returns', 'QQQ_log_returns', 'QQQ_mom_5', 'QQQ_mom_10', 'QQQ_mom_20', 'QQQ_volatility', 'QQQ_rsi', 'QQQ_volume_change', 'SPY_returns', 'SPY_log_returns', 'SPY_mom_5', 'SPY_mom_10', 'SPY_mom_20', 'SPY_volatility', 'SPY_rsi', 'SPY_volume_change', 'TLT_returns', 'TLT_log_returns', 'TLT_mom_5', 'TLT_mom_10', 'TLT_mom_20', 'TLT_volatility', 'TLT_rsi', 'TLT_volume_change']def create_sequences(features, targets, sequence_length=20):
"""Crée des séquences pour l'entraînement LSTM.
Args:
features: DataFrame des features
targets: Série des cibles (rendements futurs)
sequence_length: Longueur de la séquence
Returns:
X: Array des séquences (samples, sequence_length, features)
y: Array des cibles
"""
X, y = [], []
for i in range(sequence_length, len(features)):
# Séquence de features
sequence = features.iloc[i-sequence_length:i].values
X.append(sequence)
# Cible: rendement futur
y.append(targets.iloc[i])
return np.array(X), np.array(y)
# Préparer les données pour SPY
spy_returns = closes['SPY'].pct_change().fillna(0)
spy_features = features[[c for c in features.columns if 'SPY' in c]]
# Créer les séquences
sequence_length = 20
X, y = create_sequences(spy_features, spy_returns, sequence_length)
print(f"Séquences X shape: {X.shape}")
print(f"Cibles y shape: {y.shape}")
print(f"\nDétails:")
print(f" - Samples: {X.shape[0]}")
print(f" - Time steps: {X.shape[1]}")
print(f" - Features: {X.shape[2]}")Séquences X shape: (1805, 20, 8)
Cibles y shape: (1805,)
Détails:
- Samples: 1805
- Time steps: 20
- Features: 8from sklearn.preprocessing import StandardScaler
def normalize_sequences(X, y, fit_on_train=True, scaler=None):
"""Normalise les séquences pour le LSTM."""
# Reshape pour scaler (samples * timesteps, features)
original_shape = X.shape
X_reshaped = X.reshape(-1, X.shape[-1])
if scaler is None:
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X_reshaped)
else:
X_scaled = scaler.transform(X_reshaped)
# Reshape back
X_scaled = X_scaled.reshape(original_shape)
return X_scaled, y, scaler
# Normaliser
X_normalized, y_normalized, scaler = normalize_sequences(X, y)
print(f"X normalisé shape: {X_normalized.shape}")
print(f"Moyenne par feature: {X_normalized.mean(axis=0)[0].round(3)}")
print(f"Std par feature: {X_normalized.std(axis=0)[0].round(3)}")X normalisé shape: (1805, 20, 8)
Moyenne par feature: [-0.001 -0.001 -0.002 -0.003 -0.004 -0.007 -0.018 -0. ]
Std par feature: [1. 1. 1. 0.999 1. 1.01 1.027 0.999]# VRAI LSTM via tensorflow/keras. tensorflow (2.19.1) + keras (3.13.0) sont installes
# dans le runtime quantconnect/research:latest -- la ou ce quantbook s'execute reellement.
# Ce notebook utilisait auparavant sklearn.Ridge comme "proxy LSTM" en aplatissant les
# sequences 3D -- une simplification qui detruisait precisement la dependance temporelle
# que un LSTM est cense modeliser. La lib reelle etant disponible, verdict RECOVERABLE-LOCAL
# (voir #8757, regle sota-not-workaround Prong-A).
import os
os.environ.setdefault("TF_CPP_MIN_LOG_LEVEL", "3") # supprime INFO/WARNING/ERROR (NodeDef noise)
os.environ.setdefault("TF_ENABLE_ONEDNN_OPTS", "0") # evite les FutureWarning oneDNN
os.environ.setdefault("GRPC_VERBOSITY", "ERROR")
import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Sequential
# Reduire le bruit TF (NodeDef use_unbounded_threadpool + tf.function retracing) :
# messages de mismatch de version internes a TF, benins, mais nombreux dans les loops.
tf.get_logger().setLevel("FATAL")
import logging as _stdlib_logging
_stdlib_logging.getLogger("tensorflow").setLevel(_stdlib_logging.ERROR)
from tensorflow.keras.layers import LSTM, Dense, Dropout, Input
# Reproductibilite (C896) : un LSTM est stochastique (init des poids + dropout en train).
# On seed TF + numpy + python pour que la re-exec donne des metriques stables.
tf.keras.utils.set_random_seed(42)
tf.config.experimental.enable_op_determinism()
EPOCHS = 15 # borne pour runtime pedagogique raisonnable (Prong-B)
BATCH = 32
def build_lstm_model(input_shape):
"""Construit un modele LSTM (Keras).
Architecture : 2 couches LSTM (64 puis 32 unites) avec dropout pour regulariser,
suivies de 2 couches denses. input_shape = (timesteps, features).
"""
model = Sequential([
Input(shape=input_shape),
LSTM(64, return_sequences=True),
Dropout(0.2),
LSTM(32, return_sequences=False),
Dropout(0.2),
Dense(16, activation="relu"),
Dense(1),
])
model.compile(optimizer="adam", loss="mse")
return model
def train_model(X_train, y_train, X_test, y_test, epochs=EPOCHS, verbose=0):
"""Entrainne le LSTM sur les sequences 3D (samples, timesteps, features)."""
model = build_lstm_model(input_shape=(X_train.shape[1], X_train.shape[2]))
history = model.fit(
X_train, y_train,
validation_data=(X_test, y_test),
epochs=epochs,
batch_size=BATCH,
verbose=verbose,
shuffle=False, # series temporelles : on preserve l'ordre
)
train_pred = model.predict(X_train, verbose=0).ravel()
test_pred = model.predict(X_test, verbose=0).ravel()
train_mse = float(((y_train - train_pred) ** 2).mean())
test_mse = float(((y_test - test_pred) ** 2).mean())
return model, train_mse, test_mse, history
# Split train/test
split_idx = int(len(X_normalized) * 0.8)
X_train, X_test = X_normalized[:split_idx], X_normalized[split_idx:]
y_train, y_test = y_normalized[:split_idx], y_normalized[split_idx:]
print(f"Build LSTM: input_shape=({X_train.shape[1]}, {X_train.shape[2]}) -- "
f"{X_train.shape[0]} train seq / {X_test.shape[0]} test seq")
# Entrainner
model, train_mse, test_mse, history = train_model(X_train, y_train, X_test, y_test)
print(f"Train MSE: {train_mse:.6f}")
print(f"Test MSE: {test_mse:.6f}")
print(f"epochs={EPOCHS}, batch={BATCH}, seed=42")Build LSTM: input_shape=(20, 8) -- 1444 train seq / 361 test seq
Train MSE: 0.000060
Test MSE: 0.000086
epochs=15, batch=32, seed=42# Faire les prédictions (le LSTM keras attend des séquences 3D, pas aplaties)
predictions = model.predict(X_test, verbose=0).ravel()
# Analyser les résultats
results = pd.DataFrame({
'Actual': y_test,
'Predicted': predictions,
'Abs_Error': np.abs(y_test - predictions)
})
print("Statistiques des prédictions:")
print(results.describe())
# Direction accuracy
direction_correct = ((results['Actual'] > 0) == (results['Predicted'] > 0)).mean()
print(f"\nAccuracy de direction: {direction_correct:.2%}")Statistiques des prédictions:
Actual Predicted Abs_Error
count 361.000000 361.000000 361.000000
mean 0.000470 0.000898 0.006322
std 0.009238 0.001218 0.006759
min -0.032402 -0.001322 0.000008
25% -0.003026 -0.000066 0.001847
50% 0.000531 0.000207 0.004212
75% 0.005381 0.001886 0.008228
max 0.050525 0.004588 0.045937
Accuracy de direction: 51.52%Les prédictions du réseau de neurones sont visualisées avec matplotlib en les comparant aux valeurs réelles sur la période de validation.
import matplotlib.pyplot as plt
# Visualiser les prédictions vs réel
plt.figure(figsize=(14, 6))
# Premier 50 points
n_plot = 50
plt.plot(results['Actual'].iloc[:n_plot].values, label='Réel', marker='o')
plt.plot(results['Predicted'].iloc[:n_plot].values, label='Prédit', marker='x')
plt.xlabel('Time')
plt.ylabel('Return')
plt.title('Prédictions LSTM vs Réel (SPY)')
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
def walk_forward_validation(features, targets, train_size=252, test_size=63):
"""Validation walk-forward pour simuler le trading reel."""
predictions = []
actuals = []
for start in range(0, len(features) - train_size - test_size, test_size):
train_end = start + train_size
test_end = train_end + test_size
# Donnees d'entrainement
X_train, y_train = create_sequences(
features.iloc[start:train_end],
targets.iloc[start:train_end],
sequence_length
)
if len(X_train) == 0:
continue
# Normaliser
X_train_norm, y_train_norm, scaler = normalize_sequences(X_train, y_train)
# Entrainner le VRAI LSTM (sequences 3D, pas flatten)
model = build_lstm_model(input_shape=(X_train_norm.shape[1], X_train_norm.shape[2]))
model.fit(X_train_norm, y_train_norm, epochs=5, batch_size=32, verbose=0, shuffle=False)
# Predire sur test set
X_test, y_test = create_sequences(
features.iloc[train_end:test_end],
targets.iloc[train_end:test_end],
sequence_length
)
if len(X_test) == 0:
continue
X_test_norm, _, _ = normalize_sequences(X_test, y_test, scaler=scaler)
pred = model.predict(X_test_norm, verbose=0).ravel()
predictions.extend(pred)
actuals.extend(y_test) # y_test is already np.ndarray from create_sequences
if len(predictions) == 0:
return np.array([]), np.array([])
return np.array(predictions), np.array(actuals)
# Executer walk-forward
wf_pred, wf_actual = walk_forward_validation(spy_features, spy_returns)
if len(wf_pred) > 0:
wf_mse = ((wf_actual - wf_pred) ** 2).mean()
wf_direction = ((wf_actual > 0) == (wf_pred > 0)).mean()
print(f"Walk-Forward Results:")
print(f"MSE: {wf_mse:.6f}")
print(f"Direction Accuracy: {wf_direction:.2%}")
else:
print("No walk-forward results — insufficient data for sequences.")Walk-Forward Results:
MSE: 0.000727
Direction Accuracy: 52.42%def backtest_lstm_strategy(closes, volumes, features,
train_period=252, rebalance_freq=5, refit_every=25):
"""Backtest la strategie LSTM.
Le LSTM est re-entraine periodiquement (tous les ``refit_every`` rebalances,
~= trimestriellement), pas a chaque rebalance -- un re-fit complet a chaque
barre serait prohibitif (et peu realiste en production). Entre deux refits,
on reutilise le meme modele. Bornage Prong-B (sota-not-workaround).
"""
portfolio_value = 100000
positions = {}
trade_dates = []
model = None
scaler = None
rebalance_count = 0
for i in range(train_period, len(closes) - sequence_length, rebalance_freq):
current_date = closes.index[i]
trade_dates.append(current_date)
rebalance_count += 1
# Refit periodique seulement (quarterly ~ toutes les 25 rebalances)
if model is None or rebalance_count % refit_every == 0:
train_features = features.iloc[i-train_period:i]
train_returns = closes['SPY'].pct_change().iloc[i-train_period:i]
X_train, y_train = create_sequences(train_features, train_returns, sequence_length)
if len(X_train) < 10:
continue
X_train_norm, y_train_norm, scaler = normalize_sequences(X_train, y_train)
model = build_lstm_model(input_shape=(X_train_norm.shape[1], X_train_norm.shape[2]))
model.fit(X_train_norm, y_train_norm, epochs=5, batch_size=32, verbose=0, shuffle=False)
# Predire pour chaque ETF (use ALL features so scaler dimension matches)
predictions = {}
for ticker in tickers:
if len(features) < i + sequence_length:
continue
recent_features = features.iloc[i-sequence_length:i]
X_pred = recent_features.values.reshape(1, sequence_length, -1)
X_pred_norm, _, _ = normalize_sequences(X_pred, np.array([0]), scaler=scaler)
pred = float(model.predict(X_pred_norm, verbose=0)[0, 0])
predictions[ticker] = pred
# Vendre positions existantes
for ticker, qty in positions.items():
portfolio_value += qty * closes[ticker].iloc[i]
positions = {}
# Acheter top 2 predictions positives
sorted_preds = sorted(predictions.items(), key=lambda x: x[1], reverse=True)
position_size = portfolio_value / 2
count = 0
for ticker, pred_return in sorted_preds:
if pred_return > 0 and count < 2:
buy_price = closes[ticker].iloc[i]
qty = position_size / buy_price
positions[ticker] = qty
portfolio_value -= qty * buy_price
count += 1
# Valeur finale
final_value = portfolio_value
for ticker, qty in positions.items():
final_value += qty * closes[ticker].iloc[-1]
return {
'initial': 100000,
'final': final_value,
'return': (final_value - 100000) / 100000,
'dates': trade_dates
}
# Executer le backtest
backtest_results = backtest_lstm_strategy(closes, volumes, features)
print(f"\nBacktest Results:")
print(f"Valeur initiale: ${backtest_results['initial']:,.2f}")
print(f"Valeur finale: ${backtest_results['final']:,.2f}")
print(f"Return total: {backtest_results['return']:.2%}")
Backtest Results:
Valeur initiale: $100,000.00
Valeur finale: $130,182.58
Return total: 30.18%# GRU (Gated Recurrent Unit) est une alternative plus légère au LSTM
# En production avec TensorFlow:
gru_architecture = '''
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import GRU, Dense, Dropout
model = Sequential([
GRU(64, return_sequences=True, input_shape=(sequence_length, n_features)),
Dropout(0.2),
GRU(32, return_sequences=False),
Dropout(0.2),
Dense(16, activation='relu'),
Dense(1)
])
model.compile(optimizer='adam', loss='mse', metrics=['mae'])
# model.fit(X_train, y_train, epochs=50, batch_size=32, validation_split=0.2)
'''
print("Architecture GRU ( TensorFlow ):")
print(gru_architecture)
print("\nAvantages GRU vs LSTM:")
print("- Moins de paramètres (plus rapide à entraîner)")
print("- Performance similaire pour beaucoup de tâches")
print("- Meilleur pour les séquences plus courtes")Architecture GRU ( TensorFlow ):
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import GRU, Dense, Dropout
model = Sequential([
GRU(64, return_sequences=True, input_shape=(sequence_length, n_features)),
Dropout(0.2),
GRU(32, return_sequences=False),
Dropout(0.2),
Dense(16, activation='relu'),
Dense(1)
])
model.compile(optimizer='adam', loss='mse', metrics=['mae'])
# model.fit(X_train, y_train, epochs=50, batch_size=32, validation_split=0.2)
Avantages GRU vs LSTM:
- Moins de paramètres (plus rapide à entraîner)
- Performance similaire pour beaucoup de tâches
- Meilleur pour les séquences plus courtes# Idées d'amélioration:
improvements = {
'Architecture': [
'Bidirectional LSTM: capture les dépendances passées et futures',
'Attention mechanism: focus sur les parties importantes de la séquence',
'Transformer: self-attention pour longues dépendances'
],
'Features': [
'Market sentiment (VIX, put/call ratio)',
'Macro indicators (rates, GDP, inflation)',
'Alternative data (satellite, credit card, web scraping)'
],
'Training': [
'Transfer learning: pré-entraîner sur données générales',
'Data augmentation: bruit, jittering, scaling',
'Hyperparameter optimization: Bayesian, genetic algorithms'
],
'Deployment': [
'ONNX runtime: inference optimisée',
'Quantization: INT8 pour edge deployment',
'Pruning: supprimer les connexions inutiles'
]
}
for category, items in improvements.items():
print(f"\n{category}:")
for item in items:
print(f" - {item}")
Architecture:
- Bidirectional LSTM: capture les dépendances passées et futures
- Attention mechanism: focus sur les parties importantes de la séquence
- Transformer: self-attention pour longues dépendances
Features:
- Market sentiment (VIX, put/call ratio)
- Macro indicators (rates, GDP, inflation)
- Alternative data (satellite, credit card, web scraping)
Training:
- Transfer learning: pré-entraîner sur données générales
- Data augmentation: bruit, jittering, scaling
- Hyperparameter optimization: Bayesian, genetic algorithms
Deployment:
- ONNX runtime: inference optimisée
- Quantization: INT8 pour edge deployment
- Pruning: supprimer les connexions inutiles