# [REFERENCE QC] Code a copier dans main.py QC Lab (non executable ici)
# Strategie Complete pour QuantConnect
qc_strategy_code = '''
from AlgorithmImports import *
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.preprocessing import StandardScaler
import numpy as np
import pandas as pd
from datetime import timedelta
class ReturnPredictionStrategy(QCAlgorithm):
"""
Strategie complete de prediction de rendements avec ML.
- Target: Rendement 5 jours
- Model: Gradient Boosting Regressor (ou XGBoost)
- Signal: |predicted return| > 1%
- Position sizing: Proportionnel a la magnitude
"""
def Initialize(self):
# Configuration
self.SetStartDate(2015, 1, 1)
self.SetEndDate(2024, 12, 31)
self.SetCash(100000)
# Parametres
self.lookback = 252 # 1 an d'historique
self.prediction_horizon = 5 # Predire 5 jours
self.retrain_frequency = 30 # Re-entrainer tous les 30 jours
self.min_signal_threshold = 0.01 # 1% minimum
self.max_positions = 10
self.max_position_size = 0.1 # 10% max par position
# Universe
self.num_stocks = 50
self.AddUniverse(self.CoarseSelection)
self.UniverseSettings.Resolution = Resolution.Daily
# ML Model
self.model = None
self.scaler = StandardScaler()
self.last_train_time = None
# Storage
self.symbolData = {}
# Schedule rebalancing
self.Schedule.On(
self.DateRules.EveryDay(),
self.TimeRules.AfterMarketOpen("SPY", 30),
self.Rebalance
)
self.Log("Return Prediction Strategy initialized")
def CoarseSelection(self, coarse):
"""Selectionne les top N actions par liquidite."""
filtered = [x for x in coarse
if x.HasFundamentalData
and x.Price > 5
and x.DollarVolume > 10000000]
sorted_stocks = sorted(filtered, key=lambda x: x.DollarVolume, reverse=True)
return [x.Symbol for x in sorted_stocks[:self.num_stocks]]
def OnSecuritiesChanged(self, changes):
"""Initialise les indicateurs pour les nouveaux titres."""
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol not in self.symbolData:
self.symbolData[symbol] = PredictionSymbolData(
self, symbol, self.lookback
)
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.symbolData:
del self.symbolData[symbol]
def Rebalance(self):
"""Logique principale de rebalancement."""
# Verifier si on doit re-entrainer
if self._should_retrain():
self._train_model()
if self.model is None:
return
# Generer les predictions
predictions = {}
for symbol, sd in self.symbolData.items():
if not sd.IsReady:
continue
features = sd.GetFeatures()
if features is None:
continue
# Prediction
features_scaled = self.scaler.transform([features])
predicted_return = self.model.predict(features_scaled)[0]
# Filtrer
if abs(predicted_return) >= self.min_signal_threshold:
predictions[symbol] = predicted_return
# Trier par magnitude (top predictions)
sorted_predictions = sorted(
predictions.items(),
key=lambda x: abs(x[1]),
reverse=True
)[:self.max_positions]
# Calculer les poids
targets = {}
total_magnitude = sum(abs(p[1]) for p in sorted_predictions)
for symbol, pred in sorted_predictions:
# Poids proportionnel a la magnitude
weight = abs(pred) / total_magnitude if total_magnitude > 0 else 0
weight = min(weight, self.max_position_size) # Cap
# Direction
if pred > 0:
targets[symbol] = weight
else:
targets[symbol] = -weight # Short
# Liquider les positions non dans targets
for symbol in self.Portfolio.Keys:
if symbol not in targets and self.Portfolio[symbol].Invested:
self.Liquidate(symbol)
# Executer les targets
for symbol, weight in targets.items():
self.SetHoldings(symbol, weight)
self.Log(f"Position: {symbol.Value} = {weight:.2%}")
def _should_retrain(self):
"""Determine si le modele doit etre re-entraine."""
if self.last_train_time is None:
return True
return (self.Time - self.last_train_time).days >= self.retrain_frequency
def _train_model(self):
"""Entraine le modele ML."""
X_train = []
y_train = []
for symbol, sd in self.symbolData.items():
history = self.History(symbol, self.lookback + self.prediction_horizon, Resolution.Daily)
if history.empty or len(history) < self.lookback:
continue
# Preparer les donnees
features_list, targets_list = sd.PrepareTrainingData(
history, self.prediction_horizon
)
X_train.extend(features_list)
y_train.extend(targets_list)
if len(X_train) < 200:
self.Log(f"Not enough data for training: {len(X_train)} samples")
return
X_train = np.array(X_train)
y_train = np.array(y_train)
# Standardiser
X_train_scaled = self.scaler.fit_transform(X_train)
# Entrainer
self.model = GradientBoostingRegressor(
n_estimators=100,
max_depth=4,
learning_rate=0.1,
min_samples_leaf=20,
random_state=42
)
self.model.fit(X_train_scaled, y_train)
self.last_train_time = self.Time
self.Log(f"Model trained with {len(X_train)} samples")
def OnEndOfAlgorithm(self):
"""Resume final."""
self.Log("="*60)
self.Log("RETURN PREDICTION STRATEGY - SUMMARY")
self.Log("="*60)
self.Log(f"Final Value: ${self.Portfolio.TotalPortfolioValue:,.2f}")
total_return = (self.Portfolio.TotalPortfolioValue - 100000) / 100000
self.Log(f"Total Return: {total_return:.1%}")
class PredictionSymbolData:
"""
Stocke les indicateurs et calcule les features pour un symbole.
"""
def __init__(self, algorithm, symbol, lookback):
self.symbol = symbol
self.algorithm = algorithm
# Indicateurs
self.rsi = algorithm.RSI(symbol, 14, Resolution.Daily)
self.macd = algorithm.MACD(symbol, 12, 26, 9, Resolution.Daily)
self.bb = algorithm.BB(symbol, 20, 2, Resolution.Daily)
self.sma_20 = algorithm.SMA(symbol, 20, Resolution.Daily)
self.sma_50 = algorithm.SMA(symbol, 50, Resolution.Daily)
self.atr = algorithm.ATR(symbol, 14, Resolution.Daily)
# Rolling windows pour returns
self.price_window = RollingWindow[float](lookback)
# Warmup
history = algorithm.History(symbol, lookback, Resolution.Daily)
if not history.empty:
for bar in history.itertuples():
self.price_window.Add(bar.close)
@property
def IsReady(self):
return (self.rsi.IsReady and
self.macd.IsReady and
self.bb.IsReady and
self.sma_50.IsReady and
self.price_window.IsReady)
def GetFeatures(self):
"""Retourne le vecteur de features actuel."""
if not self.IsReady:
return None
price = self.price_window[0]
# Returns
return_1d = (self.price_window[0] - self.price_window[1]) / self.price_window[1]
return_5d = (self.price_window[0] - self.price_window[5]) / self.price_window[5]
return_20d = (self.price_window[0] - self.price_window[20]) / self.price_window[20]
# Volatility (20d)
prices = [self.price_window[i] for i in range(20)]
returns = np.diff(prices) / prices[:-1]
volatility = np.std(returns)
# RSI normalized
rsi_norm = (self.rsi.Current.Value - 50) / 50
# MA ratio
ma_ratio = self.sma_20.Current.Value / self.sma_50.Current.Value
# Price to SMA
price_to_sma = price / self.sma_20.Current.Value
# MACD normalized
macd_norm = self.macd.Current.Value / price
# Bollinger %B
bb_range = self.bb.UpperBand.Current.Value - self.bb.LowerBand.Current.Value
bb_pct_b = (price - self.bb.LowerBand.Current.Value) / bb_range if bb_range > 0 else 0.5
return [
return_1d, return_5d, return_20d,
volatility, rsi_norm, ma_ratio,
price_to_sma, macd_norm, bb_pct_b
]
def PrepareTrainingData(self, history, horizon):
"""Prepare les donnees d'entrainement depuis l'historique."""
features_list = []
targets_list = []
df = history.close.unstack(level=0)
if df.empty:
return features_list, targets_list
prices = df.iloc[:, 0].values
for i in range(50, len(prices) - horizon):
# Features
ret_1d = (prices[i] - prices[i-1]) / prices[i-1]
ret_5d = (prices[i] - prices[i-5]) / prices[i-5]
ret_20d = (prices[i] - prices[i-20]) / prices[i-20]
vol = np.std(np.diff(prices[i-20:i]) / prices[i-21:i-1])
# Simplified features
sma_20 = np.mean(prices[i-20:i])
sma_50 = np.mean(prices[i-50:i])
features = [
ret_1d, ret_5d, ret_20d, vol,
0, # RSI placeholder
sma_20 / sma_50,
prices[i] / sma_20,
0, # MACD placeholder
0.5 # BB placeholder
]
# Target: future return
target = (prices[i + horizon] - prices[i]) / prices[i]
features_list.append(features)
targets_list.append(target)
return features_list, targets_list
'''
print("Strategie Complete pour QuantConnect:")
print("="*60)
print("Composants:")
print(" 1. Universe: Top 50 par liquidite")
print(" 2. Model: Gradient Boosting Regressor")
print(" 3. Target: Rendement 5 jours")
print(" 4. Signal: |prediction| > 1%")
print(" 5. Position Sizing: Proportionnel a magnitude")
print(" 6. Retraining: Tous les 30 jours")
print("="*60)
