<< Sommaire QC | Précédent : QC-Py-10-Risk-Portfolio-Management << | Suivant : QC-Py-12-Backtesting-Analysis >>
[QC CLOUD] Ce notebook utilise QuantBook / AlgorithmImports et necessite un environnement QuantConnect Cloud. Les cellules de code ne sont pas executables localement.
Analyse technique et signaux de trading Duree: 75 minutes | Niveau: Intermediaire | Python + QuantConnect
A la fin de ce notebook, vous serez capable de :
Note de conception : Ce notebook contient du code de reference a copier dans QuantConnect Lab (main.py). Les cellules ne sont pas prevues pour etre executees en tant que notebook Jupyter. L’absence d’outputs (execution_count: null) est intentionnelle.
[REFERENCE QC Cloud] Ce notebook illustre du code QuantConnect a executer dans l’IDE Cloud (https://www.quantconnect.com/research). L’environnement local ne dispose pas de QuantBook ni de l’historical data feed. Pour executer : cloner le projet QC associe, ouvrir
research.ipynb, executer cellule par cellule.
Mode d’emploi : Ce notebook est un document de reference (non executable). Le code QCAlgorithm presente ici est a copier dans
main.pyde votre projet QuantConnect Lab. Ne tentez pas de faire “Run All” – les importsAlgorithmImportsn’existent pas en Jupyter local.Pour demarrer un projet QC : copiez un
main.pydepuisprojects/oupartner-course-quant-trading/templates/, puis adaptez-le.
Les indicateurs techniques sont des calculs mathematiques bases sur le prix, le volume, ou l’intérêt ouvert d’un actif. Ils sont utilises pour :
| Catégorie | Exemples | Utilisation |
|---|---|---|
| Trend | SMA, EMA, ADX, PSAR | Identifier la direction |
| Momentum | RSI, MACD, Stochastic, CCI | Force du mouvement |
| Volatilite | Bollinger Bands, ATR, Keltner | Gestion du risque |
| Volume | OBV, AD, VWAP | Confirmation des mouvements |
QuantConnect offre 100+ indicateurs techniques built-in. Chaque indicateur :
IsReady pour savoir s’il a assez de donnéesCurrent.Value pour la valeur actuelle# Imports QuantConnect
from AlgorithmImports import *
# Imports pour analyse et visualisation
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
# Configuration matplotlib
plt.style.use('seaborn-v0_8-darkgrid')
%matplotlib inline
print("Imports reussis")Imports reussisLes moyennes mobiles sont les indicateurs les plus fondamentaux. Elles lissent les données de prix pour identifier les tendances.
| Type | Description | Formule Simplifiee |
|---|---|---|
| SMA | Simple Moving Average | Moyenne arithmetique des N derniers prix |
| EMA | Exponential Moving Average | Pondere plus les prix recents |
| WMA | Weighted Moving Average | Ponderation lineaire decroissante |
Interpretation : - Prix > MA : Tendance haussiere - Prix < MA : Tendance baissiere - Croisement de MAs : Signal de changement de tendance
# [REFERENCE QC] Code a copier dans main.py QC Lab (non executable ici)
class MovingAveragesAlgorithm(QCAlgorithm):
"""
Demonstration des trois types de Moving Averages.
Montre comment initialiser et utiliser SMA, EMA, WMA.
"""
def Initialize(self):
self.SetStartDate(2015, 1, 1)
self.SetEndDate(2024, 12, 31)
self.SetCash(100000)
# Ajouter l'actif
symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
# === MOVING AVERAGES ===
# Simple Moving Average (moyenne arithmetique)
# SMA(20) = (P1 + P2 + ... + P20) / 20
self.sma = self.SMA(symbol, 20, Resolution.Daily)
# Exponential Moving Average (ponderation exponentielle)
# Reagit plus vite aux changements recents
self.ema = self.EMA(symbol, 20, Resolution.Daily)
# Weighted Moving Average (ponderation lineaire)
# WMA(3) = (3*P1 + 2*P2 + 1*P3) / (3+2+1)
self.wma = self.WMA(symbol, 20, Resolution.Daily)
# Stocker le symbole
self.symbol = symbol
# Warmup pour avoir les indicateurs prets
self.SetWarmup(20)
def OnData(self, data):
"""Appele a chaque nouvelle barre de donnees."""
if self.IsWarmingUp:
return
# Verifier que tous les indicateurs sont prets
if not (self.sma.IsReady and self.ema.IsReady and self.wma.IsReady):
return
# Recuperer les valeurs actuelles
price = self.Securities[self.symbol].Price
sma_value = self.sma.Current.Value
ema_value = self.ema.Current.Value
wma_value = self.wma.Current.Value
# Log quotidien (une fois par semaine pour ne pas surcharger)
if self.Time.weekday() == 0: # Lundi
self.Debug(f"{self.Time.date()} | Price: {price:.2f} | SMA: {sma_value:.2f} | EMA: {ema_value:.2f} | WMA: {wma_value:.2f}")
# Strategie simple : acheter si prix > toutes les MAs
if price > sma_value and price > ema_value and price > wma_value:
if not self.Portfolio[self.symbol].Invested:
self.SetHoldings(self.symbol, 1.0)
self.Debug(f"{self.Time.date()}: BUY - Prix au-dessus de toutes les MAs")
elif price < sma_value and price < ema_value and price < wma_value:
if self.Portfolio[self.symbol].Invested:
self.Liquidate(self.symbol)
self.Debug(f"{self.Time.date()}: SELL - Prix en-dessous de toutes les MAs")
print("MovingAveragesAlgorithm defini")
print("\nIndicateurs Moving Average disponibles:")
print(" - SMA(symbol, period, resolution) : Simple Moving Average")
print(" - EMA(symbol, period, resolution) : Exponential Moving Average")
print(" - WMA(symbol, period, resolution) : Weighted Moving Average")Les indicateurs de momentum mesurent la vitesse et la force des mouvements de prix. Ils sont utiles pour identifier les conditions de surachat/survente et les divergences.
| Indicateur | Plage | Surachat | Survente | Utilisation |
|---|---|---|---|---|
| RSI | 0-100 | >70 | <30 | Retournements, divergences |
| MACD | Variable | Histogramme + | Histogramme - | Croisements, momentum |
| Stochastic | 0-100 | >80 | <20 | Court terme, retournements |
| CCI | Variable | >100 | <-100 | Identification tendance |
# [REFERENCE QC] Code a copier dans main.py QC Lab (non executable ici)
class MomentumIndicatorsAlgorithm(QCAlgorithm):
"""
Demonstration des indicateurs de momentum.
RSI, MACD, Stochastic, CCI.
"""
def Initialize(self):
self.SetStartDate(2015, 1, 1)
self.SetEndDate(2024, 12, 31)
self.SetCash(100000)
symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
self.symbol = symbol
# === MOMENTUM INDICATORS ===
# RSI - Relative Strength Index
# Mesure la vitesse et le changement des mouvements de prix
# MovingAverageType.Wilders = lissage original de Wilder
self.rsi = self.RSI(symbol, 14, MovingAverageType.Wilders, Resolution.Daily)
# MACD - Moving Average Convergence Divergence
# Params: fast period (12), slow period (26), signal period (9)
# Composants: MACD line, Signal line, Histogram
self.macd = self.MACD(symbol, 12, 26, 9, MovingAverageType.Exponential, Resolution.Daily)
# Stochastic Oscillator
# Compare le prix de cloture a la plage high-low sur la periode
# Params: fast K period (14), slow K period (3), slow D period (3)
self.stoch = self.STO(symbol, 14, 3, 3, Resolution.Daily)
# Commodity Channel Index
# Mesure la variation du prix par rapport a la moyenne statistique
self.cci = self.CCI(symbol, 20, MovingAverageType.Simple, Resolution.Daily)
# Warmup pour la plus longue periode (MACD slow = 26)
self.SetWarmup(30)
def OnData(self, data):
if self.IsWarmingUp:
return
# Verifier que les indicateurs sont prets
if not (self.rsi.IsReady and self.macd.IsReady and self.stoch.IsReady and self.cci.IsReady):
return
# Recuperer les valeurs
rsi_value = self.rsi.Current.Value
# MACD expose plusieurs composants
macd_line = self.macd.Current.Value # MACD line
signal_line = self.macd.Signal.Current.Value # Signal line
histogram = macd_line - signal_line # Histogram (difference)
# Stochastic expose %K et %D
stoch_k = self.stoch.StochK.Current.Value # Fast %K
stoch_d = self.stoch.StochD.Current.Value # Slow %D (signal)
cci_value = self.cci.Current.Value
# Log hebdomadaire
if self.Time.weekday() == 0:
self.Debug(f"\n{self.Time.date()} - MOMENTUM INDICATORS:")
self.Debug(f" RSI(14): {rsi_value:.2f}")
self.Debug(f" MACD: Line={macd_line:.4f}, Signal={signal_line:.4f}, Hist={histogram:.4f}")
self.Debug(f" Stochastic: %K={stoch_k:.2f}, %D={stoch_d:.2f}")
self.Debug(f" CCI(20): {cci_value:.2f}")
# Exemple de logique de trading
# Achat: RSI survente + MACD bullish cross
rsi_oversold = rsi_value < 30
macd_bullish = histogram > 0 and self.macd.Fast.Current.Value > self.macd.Slow.Current.Value
if rsi_oversold and macd_bullish and not self.Portfolio[self.symbol].Invested:
self.SetHoldings(self.symbol, 1.0)
self.Debug(f"{self.Time.date()}: BUY SIGNAL - RSI oversold + MACD bullish")
print("MomentumIndicatorsAlgorithm defini")
print("\nIndicateurs de momentum:")
print(" - RSI(symbol, period, maType, resolution)")
print(" - MACD(symbol, fast, slow, signal, maType, resolution)")
print(" - STO(symbol, kPeriod, kSmooth, dPeriod, resolution)")
print(" - CCI(symbol, period, maType, resolution)")Les indicateurs de momentum mesurent la vitesse et la force des mouvements de prix, pas leur direction. Ils sont essentiels pour identifier les conditions extrêmes (surachat/survente) et les divergences.
RSI (Relative Strength Index) : - Plage 0-100, borné par construction - <30 = surventé (potentialité de rebond) - >70 = suracheté (potentialité de correction) - Divergence RSI : Si le prix fait un plus haut mais le RSI fait un plus bas = signal affaiblissement de la tendance
MACD (Moving Average Convergence Divergence) : - Composé de 3 éléments : MACD line (fast EMA - slow EMA), Signal line (EMA du MACD), Histogram (MACD - Signal) - Bullish cross : Histogram passe de négatif à positif (MACD croise au-dessus du Signal) - Bearish cross : Histogram passe de positif à négatif (MACD croise en-dessous du Signal) - L’histogramme est le meilleur indicateur de timing car il montre l’accélération du momentum
Stochastic Oscillator : - Compare le prix de clôture à la plage high-low sur la période - %K rapide (plus réactif) vs %D lent (lissé) - Idéal pour identifier les retournements à court terme
CCI (Commodity Channel Index) : - Mesure la variation du prix par rapport à sa moyenne statistique - Non-borné (peut dépasser ±100), ce qui le rend adapté aux forts mouvements - Utile pour identifier les début/fin de tendance
Ancre savante – Lane, G. C. (fin des annees 1950). Le Stochastic Oscillator (%K/%D) a ete developpe et popularise par George C. Lane (1921-2004), dans un groupe de traders sur futures de Chicago ; il compare le prix de cloture a la plage high-low sur une fenêtre donnee.
Le RSI classique utilise des seuils fixes (30/70). Dans la pratique, les zones de survente/surachat varient selon la tendance. Cet exercice vous demande d’implementer des zones RSI dynamiques.
Objectif : Implementer get_rsi_signal(rsi_value, sma_value, current_price) qui retourne un signal de trading.
Règles : - Si prix > SMA (uptrend) : seuils de survente/surachat decales a 25/75 (plus strict) - Si prix < SMA (downtrend) : seuils decales a 35/65 (plus sensible) - Si prix == SMA (range) : seuils classiques 30/70
Retour : - "BUY" si RSI est en zone de survente - "SELL" si RSI est en zone de surachat - "HOLD" sinon
Indices : - Determinez d’abord le regime (uptrend/downtrend/range) en comparant prix et SMA - Utilisez un ecart de tolerance (ex: 0.1%) pour le “range” plutot qu’une egalite stricte
# Exercice 1 : Logique de Signal RSI avec Zones Dynamiques
# TODO etudiant : implementer get_rsi_signal
def get_rsi_signal(rsi_value, sma_value, current_price):
"""
Determine le signal RSI en fonction du regime de tendance.
Parameters:
-----------
rsi_value : float (RSI actuel, plage 0-100)
sma_value : float (Simple Moving Average)
current_price : float
Returns:
--------
str : "BUY", "SELL", ou "HOLD"
"""
result = "HOLD" # TODO etudiant : implementer la logique de zones dynamiques
return result
print("Exercice a completer")Les indicateurs de volatilite mesurent l’amplitude des mouvements de prix. Essentiels pour la gestion du risque et le dimensionnement des positions.
| Indicateur | Description | Utilisation |
|---|---|---|
| Bollinger Bands | Bandes autour d’une MA | Surachat/survente, squeeze |
| ATR | Average True Range | Stop-loss, position sizing |
| STD | Standard Deviation | Volatilite statistique |
| Keltner Channels | Canaux bases sur ATR | Alternative aux BB |
# [REFERENCE QC] Code a copier dans main.py QC Lab (non executable ici)
class VolatilityIndicatorsAlgorithm(QCAlgorithm):
"""
Demonstration des indicateurs de volatilite.
Bollinger Bands, ATR, STD, Keltner Channels.
"""
def Initialize(self):
self.SetStartDate(2015, 1, 1)
self.SetEndDate(2024, 12, 31)
self.SetCash(100000)
symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
self.symbol = symbol
# === VOLATILITY INDICATORS ===
# Bollinger Bands
# Middle band = SMA, Upper/Lower = SMA +/- k*STD
# Params: period (20), standard deviations (2)
self.bb = self.BB(symbol, 20, 2, MovingAverageType.Simple, Resolution.Daily)
# Average True Range
# Mesure la volatilite en tenant compte des gaps
# True Range = max(High-Low, |High-PrevClose|, |Low-PrevClose|)
self.atr = self.ATR(symbol, 14, MovingAverageType.Simple, Resolution.Daily)
# Standard Deviation
# Ecart-type des prix de cloture
self.std = self.STD(symbol, 20, Resolution.Daily)
# Keltner Channels
# Similar aux BB mais utilise ATR au lieu de STD
# Params: period (20), ATR multiplier (1.5), MA type
self.kc = self.KCH(symbol, 20, 1.5, MovingAverageType.Simple, Resolution.Daily)
self.SetWarmup(25)
def OnData(self, data):
if self.IsWarmingUp:
return
if not (self.bb.IsReady and self.atr.IsReady and self.std.IsReady and self.kc.IsReady):
return
price = self.Securities[self.symbol].Price
# Bollinger Bands - expose 3 composants
bb_upper = self.bb.UpperBand.Current.Value
bb_middle = self.bb.MiddleBand.Current.Value
bb_lower = self.bb.LowerBand.Current.Value
bb_bandwidth = (bb_upper - bb_lower) / bb_middle # Mesure de volatilite
# ATR
atr_value = self.atr.Current.Value
atr_percent = (atr_value / price) * 100 # ATR en pourcentage du prix
# Standard Deviation
std_value = self.std.Current.Value
# Keltner Channels
kc_upper = self.kc.UpperBand.Current.Value
kc_middle = self.kc.MiddleBand.Current.Value
kc_lower = self.kc.LowerBand.Current.Value
# Log hebdomadaire
if self.Time.weekday() == 0:
self.Debug(f"\n{self.Time.date()} - VOLATILITY INDICATORS:")
self.Debug(f" Price: {price:.2f}")
self.Debug(f" Bollinger Bands: Upper={bb_upper:.2f}, Middle={bb_middle:.2f}, Lower={bb_lower:.2f}")
self.Debug(f" BB Bandwidth: {bb_bandwidth:.2%}")
self.Debug(f" ATR(14): {atr_value:.2f} ({atr_percent:.2f}%)")
self.Debug(f" STD(20): {std_value:.2f}")
self.Debug(f" Keltner: Upper={kc_upper:.2f}, Middle={kc_middle:.2f}, Lower={kc_lower:.2f}")
# Exemple: Bollinger Band squeeze detection
# Si BB sont a l'interieur de Keltner = faible volatilite, breakout imminent
squeeze = bb_upper < kc_upper and bb_lower > kc_lower
if squeeze:
self.Debug(f"{self.Time.date()}: SQUEEZE DETECTED - Low volatility, potential breakout")
print("VolatilityIndicatorsAlgorithm defini")
print("\nIndicateurs de volatilite:")
print(" - BB(symbol, period, k, maType, resolution) : Bollinger Bands")
print(" - ATR(symbol, period, maType, resolution) : Average True Range")
print(" - STD(symbol, period, resolution) : Standard Deviation")
print(" - KCH(symbol, period, atrMult, maType, resolution) : Keltner Channels")Les indicateurs de volatilité mesurent l’amplitude des mouvements de prix et sont essentiels pour : 1. Gestion du risque : Dimensionner les positions selon la volatilité actuelle 2. Stop-loss : Ajuster les stops en fonction de l’ATR 3. Détection de squeeze : Périodes de faible volatilité suivies d’explosions
Bollinger Bands : - Middle Band = SMA(20), Upper/Lower = SMA ± 2×STD - Bandwidth = (Upper - Lower) / Middle : mesure de volatilité normalisée - Squeeze : Bandwidth faible = volatilité faible, breakout imminent - Walk = Prix sur la bande supérieure = forte tendance haussière
ATR (Average True Range) : - Mesure la volatilité réelle en tenant compte des gaps (différence inter-session) - True Range = max(High-Low, |High-PrevClose|, |Low-PrevClose|) - Unité : $ (ou points), pas en pourcentage - Usage pour stop-loss : Stop = Entry - 2×ATR - Usage pour position sizing : PositionSize = Risk% / ATR
Standard Deviation : - Écart-type des prix de clôture - Plus lisse que l’ATR (ne tient pas compte des gaps) - Utile pour les calculs statistiques (Bandes de Bollinger, Z-score)
Keltner Channels : - Alternative aux Bollinger Bands basée sur l’ATR - Upper/Lower = EMA ± k×ATR (au lieu de STD) - Squeeze detection : Si BB inside KC = volatilité très faible - Plus réactif que les BB aux changements de tendance
Comparaison BB vs KC : - BB utilisent l’écart-type (sensible aux outliers) - KC utilisent l’ATR (plus robuste aux gaps) - KC sont plus étroits que BB en période de trend, plus larges en range
Ces indicateurs aident a identifier et confirmer les tendances.
| Indicateur | Description | Signal |
|---|---|---|
| ADX | Force de la tendance (pas la direction) | ADX > 25 = tendance forte |
| Aroon | Temps depuis high/low | Identifie debut/fin de tendance |
| PSAR | Parabolic Stop and Reverse | Points de retournement |
# [REFERENCE QC] Code a copier dans main.py QC Lab (non executable ici)
class TrendIndicatorsAlgorithm(QCAlgorithm):
"""
Demonstration des indicateurs de tendance.
ADX, Aroon, Parabolic SAR.
"""
def Initialize(self):
self.SetStartDate(2015, 1, 1)
self.SetEndDate(2024, 12, 31)
self.SetCash(100000)
symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
self.symbol = symbol
# === TREND INDICATORS ===
# Average Directional Index (ADX)
# Mesure la FORCE de la tendance (pas la direction)
# ADX < 20 : pas de tendance (range)
# ADX 20-25 : tendance naissante
# ADX 25-50 : tendance forte
# ADX > 50 : tendance tres forte
self.adx = self.ADX(symbol, 14, Resolution.Daily)
# Aroon Oscillator
# Mesure le temps ecoule depuis le dernier high/low
# AroonUp proche de 100 : nouveau high recent (uptrend)
# AroonDown proche de 100 : nouveau low recent (downtrend)
self.aroon = self.AROON(symbol, 25, Resolution.Daily)
# Parabolic SAR (Stop And Reverse)
# Fournit des niveaux de stop suiveur et des signaux de retournement
# Params: acceleration factor start (0.02), AF step (0.02), AF max (0.2)
self.psar = self.PSAR(symbol, 0.02, 0.02, 0.2, Resolution.Daily)
self.SetWarmup(30)
def OnData(self, data):
if self.IsWarmingUp:
return
if not (self.adx.IsReady and self.aroon.IsReady and self.psar.IsReady):
return
price = self.Securities[self.symbol].Price
# ADX et ses composants DI+ et DI-
adx_value = self.adx.Current.Value
di_plus = self.adx.PositiveDirectionalIndex.Current.Value # +DI
di_minus = self.adx.NegativeDirectionalIndex.Current.Value # -DI
# Aroon composants
aroon_up = self.aroon.AroonUp.Current.Value
aroon_down = self.aroon.AroonDown.Current.Value
aroon_osc = aroon_up - aroon_down # Oscillator
# Parabolic SAR
psar_value = self.psar.Current.Value
psar_above_price = psar_value > price # True si downtrend
# Log hebdomadaire
if self.Time.weekday() == 0:
self.Debug(f"\n{self.Time.date()} - TREND INDICATORS:")
self.Debug(f" Price: {price:.2f}")
self.Debug(f" ADX(14): {adx_value:.2f} | +DI={di_plus:.2f}, -DI={di_minus:.2f}")
# Interpretation ADX
if adx_value < 20:
trend_strength = "No trend (range)"
elif adx_value < 25:
trend_strength = "Weak trend"
elif adx_value < 50:
trend_strength = "Strong trend"
else:
trend_strength = "Very strong trend"
self.Debug(f" ADX Interpretation: {trend_strength}")
self.Debug(f" Aroon: Up={aroon_up:.2f}, Down={aroon_down:.2f}, Osc={aroon_osc:.2f}")
self.Debug(f" PSAR: {psar_value:.2f} ({'Above price - Downtrend' if psar_above_price else 'Below price - Uptrend'})")
# Exemple de logique de trading
# Entree: ADX > 25 (tendance forte) + DI+ > DI- (uptrend) + PSAR sous le prix
strong_uptrend = adx_value > 25 and di_plus > di_minus and not psar_above_price
if strong_uptrend and not self.Portfolio[self.symbol].Invested:
self.SetHoldings(self.symbol, 1.0)
self.Debug(f"{self.Time.date()}: BUY - Strong uptrend detected")
# Sortie: PSAR passe au-dessus du prix (reversal)
if psar_above_price and self.Portfolio[self.symbol].Invested:
self.Liquidate(self.symbol)
self.Debug(f"{self.Time.date()}: SELL - PSAR reversal")
print("TrendIndicatorsAlgorithm defini")
print("\nIndicateurs de tendance:")
print(" - ADX(symbol, period, resolution) : Average Directional Index")
print(" - AROON(symbol, period, resolution) : Aroon Oscillator")
print(" - PSAR(symbol, afStart, afStep, afMax, resolution) : Parabolic SAR")Les indicateurs de tendance identifient la direction et la force des mouvements de prix, cruciaux pour le trend-following.
ADX (Average Directional Index) : - Mesure la force de la tendance, PAS la direction - ADX < 20 : Pas de tendance (range) → éviter les stratégies trend-following - ADX 20-25 : Tendance naissante → opportunités d’entrée - ADX 25-50 : Tendance forte → trend-following efficace - ADX > 50 : Tendance très forte → attention au retournement imminent - +DI vs -DI : Direction de la tendance (+DI > -DI = uptrend)
Aroon Oscillator : - Mesure le temps écoulé depuis le dernier high/low - AroonUp = 100 → nouveau high récent (tendance haussière forte) - AroonDown = 100 → nouveau low récent (tendance baissière forte) - Oscillator = AroonUp - AroonDown (-100 à +100) - Crossover : AroonUp croise au-dessus de AroonDown = signal d’achat
Parabolic SAR (Stop And Reverse) : - Fournit des niveaux de stop suiveur qui s’ajustent avec le prix - PSAR < Prix = Position long (le stop suit le prix à la hausse) - PSAR > Prix = Position short (le stop suit le prix à la baisse) - AF (Acceleration Factor) : Commence à 0.02, augmente de 0.02 à chaque nouveau point extrême - Signal de reversal : Quand PSAR croise le prix = sortir + inverser position - Excellent pour trailing stop (couper les positions rapidement en cas de reversal)
Combinaison ADX + PSAR : - ADX > 25 + PSAR sous prix = Uptrend fort → acheter - ADX < 20 = Range → éviter, utiliser mean-reversion - PSAR reversal = Signal de sortie urgent (protection du capital)
Les indicateurs de volume confirment les mouvements de prix. Un mouvement avec fort volume est plus significatif.
| Indicateur | Description | Interpretation |
|---|---|---|
| OBV | On-Balance Volume | Volume cumulatif, divergences |
| AD | Accumulation/Distribution | Pression acheteuse/vendeuse |
| VWAP | Volume Weighted Average Price | Prix moyen institutionnel |
# [REFERENCE QC] Code a copier dans main.py QC Lab (non executable ici)
class VolumeIndicatorsAlgorithm(QCAlgorithm):
"""
Demonstration des indicateurs de volume.
OBV, AD (Accumulation/Distribution), VWAP.
"""
def Initialize(self):
self.SetStartDate(2015, 1, 1)
self.SetEndDate(2024, 12, 31)
self.SetCash(100000)
symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
self.symbol = symbol
# === VOLUME INDICATORS ===
# On-Balance Volume (OBV)
# Accumule le volume en fonction de la direction du prix
# Si close > prev_close : OBV += volume
# Si close < prev_close : OBV -= volume
self.obv = self.OBV(symbol, Resolution.Daily)
# Accumulation/Distribution Line
# Tient compte de OU le prix a cloture dans la plage high-low
# Money Flow Multiplier = ((Close - Low) - (High - Close)) / (High - Low)
self.ad = self.AD(symbol, Resolution.Daily)
# Volume Weighted Average Price (VWAP)
# Prix moyen pondere par le volume (reset quotidien)
# Utilise par les institutionnels comme benchmark
self.vwap = self.VWAP(symbol, Resolution.Daily)
# Variables pour tracking des valeurs precedentes
self.prev_obv = None
self.prev_price = None
self.SetWarmup(10)
def OnData(self, data):
if self.IsWarmingUp:
return
if not (self.obv.IsReady and self.ad.IsReady and self.vwap.IsReady):
return
price = self.Securities[self.symbol].Price
# Recuperer les valeurs
obv_value = self.obv.Current.Value
ad_value = self.ad.Current.Value
vwap_value = self.vwap.Current.Value
# Position relative au VWAP
above_vwap = price > vwap_value
vwap_diff_pct = ((price - vwap_value) / vwap_value) * 100
# Log hebdomadaire
if self.Time.weekday() == 0:
self.Debug(f"\n{self.Time.date()} - VOLUME INDICATORS:")
self.Debug(f" Price: {price:.2f}")
self.Debug(f" OBV: {obv_value:,.0f}")
self.Debug(f" AD: {ad_value:,.0f}")
self.Debug(f" VWAP: {vwap_value:.2f} (Price {'above' if above_vwap else 'below'} by {abs(vwap_diff_pct):.2f}%)")
# Detection de divergence OBV
# Si prix monte mais OBV baisse = divergence bearish (warning)
if self.prev_obv is not None and self.prev_price is not None:
price_up = price > self.prev_price
obv_up = obv_value > self.prev_obv
if price_up and not obv_up:
self.Debug(f"{self.Time.date()}: BEARISH DIVERGENCE - Prix monte, OBV baisse")
elif not price_up and obv_up:
self.Debug(f"{self.Time.date()}: BULLISH DIVERGENCE - Prix baisse, OBV monte")
# Sauvegarder pour comparaison
self.prev_obv = obv_value
self.prev_price = price
# Exemple: Acheter si au-dessus du VWAP avec OBV croissant
# (Les institutionnels achetent au-dessus du VWAP = confirmation)
if above_vwap and not self.Portfolio[self.symbol].Invested:
self.SetHoldings(self.symbol, 1.0)
self.Debug(f"{self.Time.date()}: BUY - Price above VWAP")
print("VolumeIndicatorsAlgorithm defini")
print("\nIndicateurs de volume:")
print(" - OBV(symbol, resolution) : On-Balance Volume")
print(" - AD(symbol, resolution) : Accumulation/Distribution")
print(" - VWAP(symbol, resolution) : Volume Weighted Average Price")Le volume confirme les mouvements de prix. Un mouvement de prix avec fort volume est plus significatif qu’un mouvement sur volume faible.
OBV (On-Balance Volume) : - Volume cumulatif directionnel : ajoute le volume quand le prix monte, le soustrait quand le prix baisse - Divergence OBV : - Prix fait higher high, OBV fait lower high = divergence bearish (affaiblissement) - Prix fait lower low, OBV fait higher low = divergence bullish (accumulation cachée) - OBV est un leading indicator : diverge souvent AVANT le prix
AD (Accumulation/Distribution) : - Plus sophistiqué que l’OBV : tient compte de où le prix clôture dans la plage high-low - Money Flow Multiplier = ((Close - Low) - (High - Close)) / (High - Low) - Clôture proche du high → MFM positif (pression acheteuse) - Clôture proche du low → MFM négatif (pression vendeuse) - MFM × Volume = Money Flow Volume (MFV), cumulé pour obtenir AD
VWAP (Volume Weighted Average Price) : - Prix moyen pondéré par le volume (réinitialisé quotidiennement) - Utilisé par les institutionnels comme benchmark d’exécution - Prix > VWAP = Les acheteurs sont agressifs (contrôle des bulls) - Prix < VWAP = Les vendeurs sont agressifs (contrôle des bears) - Retour au VWAP = Mean-reversion vers le prix moyen institutionnel
Comparaison OBV vs AD : - OBV : Simple, basé sur la direction de clôture (up/down) - AD : Plus précis, basé sur la position de clôture dans la plage (high-low) - AD est mieux pour les marchés à forte volatilité (tient compte de l’amplitude)
Pattern Volume + Prix :
| Prix | Volume | Interprétation |
|---|---|---|
| Monte | Monte | Confirmation haussière (healthy uptrend) |
| Monte | Baisse | Divergence bearish (affaiblissement) |
| Baisse | Monte | Pression vendeuse (distribution) |
| Baisse | Baisse | Confirmation baissière (healthy downtrend) |
Ancre savante – Chaikin, M. (annees 1970). L’Accumulation/Distribution Line et son Money Flow Multiplier
((Close-Low)-(High-Close))/(High-Low)(enseigne ci-dessus) sont l’oeuvre de Marc Chaikin. A ne pas confondre avec la variante Williams Accumulation/Distribution de Larry Williams.
Les indicateurs QuantConnect ne conservent que la valeur actuelle. Pour analyser l’historique des indicateurs (ex: detecter un croisement), on utilise des Rolling Windows.
Un RollingWindow<T> est une structure de données circulaire qui garde les N dernières valeurs :
RollingWindow[0] = valeur la plus recente
RollingWindow[1] = valeur précédente
RollingWindow[N-1] = valeur la plus ancienne# [REFERENCE QC] Code a copier dans main.py QC Lab (non executable ici)
class RollingWindowAlgorithm(QCAlgorithm):
"""
Demonstration de l'utilisation des Rolling Windows.
Stocke l'historique des indicateurs pour analyse avancee.
"""
def Initialize(self):
self.SetStartDate(2015, 1, 1)
self.SetEndDate(2024, 12, 31)
self.SetCash(100000)
symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
self.symbol = symbol
# Creer les indicateurs
self.rsi = self.RSI(symbol, 14, Resolution.Daily)
self.sma_fast = self.SMA(symbol, 10, Resolution.Daily)
self.sma_slow = self.SMA(symbol, 30, Resolution.Daily)
# === ROLLING WINDOWS ===
# Rolling Window pour RSI (garder les 10 dernieres valeurs)
# Syntaxe Python: RollingWindow[type](taille)
self.rsi_window = RollingWindow[float](10)
# Rolling Windows pour les SMAs (pour detecter croisements)
self.sma_fast_window = RollingWindow[float](3)
self.sma_slow_window = RollingWindow[float](3)
# Rolling Window pour les prix (pour patterns)
self.price_window = RollingWindow[float](5)
# Rolling Window pour les TradeBar complets
self.bar_window = RollingWindow[TradeBar](10)
self.SetWarmup(30)
def OnData(self, data):
if self.IsWarmingUp:
return
# Verifier que les donnees existent
if not data.ContainsKey(self.symbol):
return
# Ajouter les nouvelles valeurs aux Rolling Windows
# IMPORTANT: Add() ajoute a la position [0] (plus recente)
if self.rsi.IsReady:
self.rsi_window.Add(self.rsi.Current.Value)
if self.sma_fast.IsReady:
self.sma_fast_window.Add(self.sma_fast.Current.Value)
if self.sma_slow.IsReady:
self.sma_slow_window.Add(self.sma_slow.Current.Value)
# Ajouter le prix et la barre complete
self.price_window.Add(data[self.symbol].Close)
self.bar_window.Add(data[self.symbol])
# Verifier si les Rolling Windows sont remplis
if not self.rsi_window.IsReady:
return
# === UTILISATION DES ROLLING WINDOWS ===
# Acceder aux valeurs historiques du RSI
rsi_current = self.rsi_window[0] # Valeur actuelle
rsi_previous = self.rsi_window[1] # Valeur precedente
rsi_oldest = self.rsi_window[9] # Plus ancienne (index = taille - 1)
# Calculer la moyenne des N dernieres valeurs RSI
rsi_values = [self.rsi_window[i] for i in range(self.rsi_window.Count)]
rsi_avg = sum(rsi_values) / len(rsi_values)
# Detecter croisement SMA (Golden Cross / Death Cross)
if self.sma_fast_window.IsReady and self.sma_slow_window.IsReady:
# Croisement = fast etait sous slow et maintenant au-dessus
fast_current = self.sma_fast_window[0]
fast_previous = self.sma_fast_window[1]
slow_current = self.sma_slow_window[0]
slow_previous = self.sma_slow_window[1]
golden_cross = fast_previous < slow_previous and fast_current > slow_current
death_cross = fast_previous > slow_previous and fast_current < slow_current
if golden_cross:
self.Debug(f"{self.Time.date()}: GOLDEN CROSS - SMA10 crosses above SMA30")
if not self.Portfolio[self.symbol].Invested:
self.SetHoldings(self.symbol, 1.0)
if death_cross:
self.Debug(f"{self.Time.date()}: DEATH CROSS - SMA10 crosses below SMA30")
if self.Portfolio[self.symbol].Invested:
self.Liquidate(self.symbol)
# Log mensuel des statistiques
if self.Time.day == 1:
self.Debug(f"\n{self.Time.date()} - RSI Rolling Window Stats:")
self.Debug(f" Current: {rsi_current:.2f}")
self.Debug(f" Previous: {rsi_previous:.2f}")
self.Debug(f" 10-bar Average: {rsi_avg:.2f}")
self.Debug(f" 10-bar Min: {min(rsi_values):.2f}")
self.Debug(f" 10-bar Max: {max(rsi_values):.2f}")
print("RollingWindowAlgorithm defini")
print("\nRolling Window - Syntaxe:")
print(" - Creation: self.window = RollingWindow[float](10)")
print(" - Ajout: self.window.Add(value)")
print(" - Acces: self.window[0] (recent), self.window[n-1] (ancien)")
print(" - Verification: self.window.IsReady")Les Rolling Windows permettent de stocker l’historique des valeurs d’un indicateur, ce qui est indispensable pour : 1. Détecter les croisements : Comparer la valeur actuelle avec la valeur précédente 2. Analyser les patterns : Identifier des configurations sur plusieurs périodes 3. Calculer des dérivées : Vitesse d’accélération d’un indicateur
Architecture d’un RollingWindow : - RollingWindow[T](taille) : Structure circulaire de N valeurs - window[0] = Valeur la plus récente - window[1] = Valeur précédente - window[N-1] = Valeur la plus ancienne - window.Count = Nombre actuel de valeurs (avant IsReady)
Cycle de vie : 1. Création : self.window = RollingWindow[float](10) 2. Ajout : self.window.Add(value) → Ajoute à la position [0], décale les autres 3. Vérification : self.window.IsReady → True quand Count == taille 4. Accès : self.window[i] → Accès direct aux valeurs historiques
Patterns courants :
Golden Cross / Death Cross (détection de croisement SMA) :
fast_prev = fast_window[1] # Valeur précédente SMA rapide
slow_prev = slow_window[1] # Valeur précédente SMA lente
fast_curr = fast_window[0] # Valeur actuelle SMA rapide
slow_curr = slow_window[0] # Valeur actuelle SMA lente
golden_cross = (fast_prev < slow_prev) and (fast_curr > slow_curr)
death_cross = (fast_prev > slow_prev) and (fast_curr < slow_curr)Divergence RSI (prix vs indicateur) :
Accélération de momentum (dérivée seconde) :
Pourquoi pas simplement utiliser history ? - history refait une requête API à chaque appel (lent) - RollingWindow stocke en mémoire (rapide) - RollingWindow est mis à jour automatiquement avec les indicateurs
Limitations : - Consomme de la mémoire pour les grandes fenêtres (>1000) - Perte de précision pour les très vieilles valeurs (arrondi float) - Alternative : IndicatorHistory pour les backtests sur longue période
Les patterns de chandeliers japonais sont des signaux classiques d’analyse technique. Cet exercice vous demande de detecter le pattern Trois Soldats Blancs (Three White Soldiers) avec des Rolling Windows.
Objectif : Implementer detect_three_white_soldiers(bar_window) qui retourne True si les 3 dernières barres forment le pattern.
Règles du pattern : - 3 barres consecutives haussieres (close > open pour chacune) - Chaque close est superieure a la close précédente - Chaque open est dans le corps de la bougie précédente (entre open et close) - Les corps sont “significatifs” : abs(close - open) > 0.3 * (high - low)
Indices : - bar_window[0] = barre la plus recente, bar_window[2] = la plus ancienne des 3 - Verifiez d’abord que bar_window.Count >= 3 et bar_window.IsReady - Le corps = abs(close - open), la mèche = high - low
# Exercice 2 : Detecteur de Pattern - Trois Soldats Blancs
# TODO etudiant : implementer detect_three_white_soldiers
def detect_three_white_soldiers(bar_window):
"""
Detecte le pattern Trois Soldats Blancs (Three White Soldiers).
Parameters:
-----------
bar_window : RollingWindow[TradeBar]
Fenetre contenant au moins 3 barres recentes
Returns:
--------
bool : True si le pattern est detecte
"""
# TODO etudiant : verifier bar_window.IsReady et Count >= 3
# TODO etudiant : verifier 3 barres haussieres consecutives
# TODO etudiant : verifier closes croissantes
# TODO etudiant : verifier opens dans le corps precedent
# TODO etudiant : verifier corps significatifs (> 30% de la mèche)
return False
print("Exercice a completer")Les Rolling Windows permettent d’analyser des patterns sur plusieurs periodes.
def detect_higher_highs(price_window, count=3):
"""
Detecte si les N derniers highs sont croissants (uptrend).
Parameters:
-----------
price_window : RollingWindow[TradeBar]
count : int, nombre de barres a analyser
Returns:
--------
bool : True si higher highs detected
"""
if not price_window.IsReady or price_window.Count < count:
return False
highs = [price_window[i].High for i in range(count)]
# Verifier que chaque high est plus haut que le precedent
# Note: [0] est le plus recent, donc on compare dans l'ordre inverse
for i in range(1, len(highs)):
if highs[i-1] <= highs[i]: # current <= previous (en time order)
return False
return True
def detect_rsi_divergence(price_window, rsi_window, count=5):
"""
Detecte une divergence RSI (bearish ou bullish).
Bearish: Prix fait higher high mais RSI fait lower high
Bullish: Prix fait lower low mais RSI fait higher low
Returns:
--------
str : 'bullish', 'bearish', ou 'none'
"""
if not (price_window.IsReady and rsi_window.IsReady):
return 'none'
# Comparer premier et dernier
price_current = price_window[0]
price_older = price_window[count-1]
rsi_current = rsi_window[0]
rsi_older = rsi_window[count-1]
# Bearish divergence: prix monte, RSI baisse
if price_current > price_older and rsi_current < rsi_older:
return 'bearish'
# Bullish divergence: prix baisse, RSI monte
if price_current < price_older and rsi_current > rsi_older:
return 'bullish'
return 'none'
print("Fonctions de detection de patterns definies:")
print(" - detect_higher_highs(price_window, count)")
print(" - detect_rsi_divergence(price_window, rsi_window, count)")Fonctions de detection de patterns definies: - detect_higher_highs(price_window, count) - detect_rsi_divergence(price_window, rsi_window, count)Les fonctions de détection de patterns utilisent les Rolling Windows pour identifier des configurations techniques.
1. Higher Highs (détection de tendance haussière) :
2. Divergence RSI (pattern de retournement) :
def detect_rsi_divergence(price_window, rsi_window, count=5):
# Bearish: prix fait higher high, RSI fait lower high
if price_current > price_older and rsi_current < rsi_older:
return 'bearish'
# Bullish: prix fait lower low, RSI fait higher low
if price_current < price_older and rsi_current > rsi_older:
return 'bullish'Types de Patterns détectables avec Rolling Windows :
| Pattern | Description | Signal |
|---|---|---|
| Higher Highs/Lows | Trend haussier | Continuation |
| Lower Highs/Lows | Trend baissier | Continuation |
| Double Top | Deux highs similaires | Retournement |
| Double Bottom | Deux lows similaires | Rebond |
| RSI Divergence | Prix vs momentum disagree | Retournement |
| MACD Histogram Swing | Accélération du momentum | Timing d’entrée |
Utilisation pratique : - Les fonctions de pattern sont appelées dans OnData() après vérification IsReady - Retournent True/False ou un type de divergence - Permettent de découpler la logique de trading du code principal - Facilitent les tests unitaires sur les patterns
Limitations : - Les patterns statiques (higher highs) sont plus rares qu’on ne pense - Les divergences peuvent durer longtemps (faux signaux) - Nécessitent une confirmation (volume, price action) avant d’agir
Parfois les indicateurs built-in ne suffisent pas. QuantConnect permet de créer des indicateurs personnalises en heritant de PythonIndicator.
Structure d’un Custom Indicator : 1. Heriter de PythonIndicator 2. Définir __init__ avec les paramètres 3. Implementer Update(self, input) qui retourne True quand pret 4. Stocker le résultat dans self.Value
from QuantConnect.Indicators import PythonIndicator
class CustomMomentumIndicator(PythonIndicator):
"""
Indicateur de momentum personnalise.
Calcule: momentum = (price_current - price_n_ago) / price_n_ago
C'est le Rate of Change (ROC) simplifie.
"""
def __init__(self, name, period):
"""
Initialise l'indicateur.
Parameters:
-----------
name : str, nom de l'indicateur
period : int, nombre de periodes pour le calcul
"""
super().__init__()
self.Name = name
self.period = period
self.queue = [] # Stockage des prix
self.WarmUpPeriod = period # QuantConnect l'utilise pour le warmup
def Update(self, input):
"""
Met a jour l'indicateur avec une nouvelle valeur.
Parameters:
-----------
input : IndicatorDataPoint ou TradeBar
Returns:
--------
bool : True si l'indicateur est pret (assez de donnees)
"""
# Ajouter le nouveau prix
# input peut etre un IndicatorDataPoint (Close) ou TradeBar
if hasattr(input, 'Close'):
price = input.Close
else:
price = input.Value
self.queue.append(price)
# Garder seulement les N dernieres valeurs
if len(self.queue) > self.period:
self.queue.pop(0)
# Calculer momentum si assez de donnees
if len(self.queue) == self.period:
old_price = self.queue[0]
current_price = self.queue[-1]
if old_price != 0:
momentum = (current_price - old_price) / old_price
else:
momentum = 0.0
# Stocker dans self.Value (expose via Current.Value)
self.Value = momentum
return True # Indicateur pret
return False # Pas encore assez de donnees
print("CustomMomentumIndicator defini")
print("\nUsage dans un algorithme:")
print(" self.custom_mom = CustomMomentumIndicator('MyMomentum', 20)")
print(" self.RegisterIndicator(symbol, self.custom_mom, Resolution.Daily)")CustomMomentumIndicator definiUsage dans un algorithme: self.custom_mom = CustomMomentumIndicator('MyMomentum', 20) self.RegisterIndicator(symbol, self.custom_mom, Resolution.Daily)Le CustomMomentumIndicator implémente le Rate of Change (ROC), l’un des indicateurs de momentum les plus simples et efficaces.
Formule : ROC = (Prix_actuel - Prix_il_y_a_N_périodes) / Prix_il_y_a_N_périodes
Caractéristiques : - Non-borné : Peut aller de -∞ à +∞ (contrairement au RSI qui est borné 0-100) - Zéro-centered : Valeur positive = momentum haussier, négative = momentum baissier - Interprétation directe : ROC de +0.05 = le prix a monté de 5% sur la période
Comparaison ROC vs RSI :
| Aspect | ROC | RSI |
|---|---|---|
| Plage | Non-borné | 0-100 |
| Sensibilité | Plus sensible aux extrêmes | Borné donc moins extrême |
| Interprétation | Pourcentage de changement | Relatif aux récent hauts/bas |
| Usage | Trend-following | Mean-reversion |
Exemples d’interprétation : - ROC > 0.02 (2%) sur 20 jours = Momentum haussier significatif - ROC < -0.05 (-5%) sur 20 jours = Momentum baissier fort - ROC proche de 0 = Consolidation (pas de momentum)
Pourquoi créer un indicateur custom ? - Les indicateurs built-in sont limités aux standards (RSI, MACD, etc.) - Certains concepts (ROC simplifié, force de tendance par comptage) n’existent pas - Héritage de PythonIndicator permet de créer n’importe quel indicateur avec accès complet au framework QC
Intégration dans l’algorithme : 1. Créer l’instance : self.custom_mom = CustomMomentumIndicator('MyMomentum', 20) 2. Enregistrer avec le symbole : self.RegisterIndicator(symbol, self.custom_mom, Resolution.Daily) 3. Utiliser comme un indicateur standard : self.custom_mom.Current.Value
Propriété WarmUpPeriod : - Essentielle pour que QuantConnect sache combien de barres sont nécessaires avant que l’indicateur soit prêt - Permet d’utiliser SetWarmup() automatiquement avec la bonne durée
Ancre savante (ouvrage de reference, non un originateur unique) – Murphy, J. J. (1999), Technical Analysis of the Financial Markets, New York Institute of Finance. Le Rate of Change (ROC) est un indicateur de momentum classique sans auteur unique ; Murphy en donne le traitement standard dans la litterature d’analyse technique.
Le TrendStrengthIndicator est un exemple d’indicateur de tendance basé sur la directionnalité des mouvements plutôt que sur la magnitude des prix.
Principe : - Compte les mouvements “up” vs “down” sur la période - Score normalisé entre -1 (100% bearish) et +1 (100% bullish) - Zéro = consolidation (autant de mouvements up que down)
Avantages par rapport aux indicateurs classiques : - Robustesse aux gaps : Compte la direction, pas l’amplitude - Filtrage du bruit : Un gap massif compte comme un mouvement up, comme une petite hausse - Interprétation intuitive : +0.7 = 70% des jours sont haussiers
Utilisation pratique : - Valeur > 0.5 : Trend haussier établi - Valeur < -0.5 : Trend baissier établi - Proche de 0 : Range, éviter le trend-following
Créez un indicateur personnalise qui combine plusieurs signaux en un score unique entre 0 et 100.
Objectif : Implementer CompositeScoreIndicator(PythonIndicator) qui combine : - Tendance (poids 40%) : Prix vs SMA 20 → +1 si au-dessus, -1 si en-dessous - Momentum (poids 30%) : RSI normalise → (RSI - 50) / 50 (plage -1 a +1) - Volatilite (poids 30%) : Position dans les Bollinger Bands → (prix - lower) / (upper - lower) * 2 - 1 (plage -1 a +1)
Formule finale : Score = 50 + 50 * (0.4*tendance + 0.3*momentum + 0.3*volatilite)
Indices : - Heritez de PythonIndicator et implementez Update() - Stockez les valeurs necessaires dans des listes (comme le CustomMomentumIndicator) - Le score doit etre entre 0 (bearish extreme) et 100 (bullish extreme)
# Exercice 3 : Indicateur Composite Multi-Facteur
# TODO etudiant : implementer CompositeScoreIndicator
from QuantConnect.Indicators import PythonIndicator
class CompositeScoreIndicator(PythonIndicator):
"""
Indicateur composite combinant tendance, momentum et volatilite.
Score entre 0 (bearish extreme) et 100 (bullish extreme).
"""
def __init__(self, name, sma_period=20, rsi_period=14, bb_period=20):
super().__init__()
self.Name = name
self.WarmUpPeriod = max(sma_period, rsi_period, bb_period) + 1
# TODO etudiant : initialiser les structures de stockage
def Update(self, input):
# TODO etudiant : calculer le score composite
# Etape 1 : extraire le prix
# Etape 2 : calculer tendance (prix vs SMA)
# Etape 3 : calculer momentum (RSI normalise)
# Etape 4 : calculer volatilite (position dans BB)
# Etape 5 : combiner avec les poids et stocker dans self.Value
return False
print("Exercice a completer")class TrendStrengthIndicator(PythonIndicator):
"""
Indicateur de force de tendance personnalise.
Score de -1 (downtrend fort) a +1 (uptrend fort)
Base sur la proportion de mouvements up vs down.
"""
def __init__(self, name, period):
super().__init__()
self.Name = name
self.period = period
self.closes = []
self.WarmUpPeriod = period
def Update(self, input):
# Extraire le close
if hasattr(input, 'Close'):
close = input.Close
else:
close = input.Value
self.closes.append(close)
if len(self.closes) > self.period + 1:
self.closes.pop(0)
if len(self.closes) >= self.period:
# Compter les mouvements up et down
up_moves = 0
down_moves = 0
for i in range(1, len(self.closes)):
if self.closes[i] > self.closes[i-1]:
up_moves += 1
elif self.closes[i] < self.closes[i-1]:
down_moves += 1
total_moves = up_moves + down_moves
if total_moves > 0:
# Score: (up - down) / total, range [-1, 1]
self.Value = (up_moves - down_moves) / total_moves
else:
self.Value = 0.0
return True
return False
print("TrendStrengthIndicator defini")
print("\nInterpretation:")
print(" +1.0 : Uptrend tres fort (100% des jours sont up)")
print(" 0.0 : Range/consolidation")
print(" -1.0 : Downtrend tres fort (100% des jours sont down)")TrendStrengthIndicator definiInterpretation: +1.0 : Uptrend tres fort (100% des jours sont up) 0.0 : Range/consolidation -1.0 : Downtrend tres fort (100% des jours sont down)Le projet dispose d’une bibliotheque d’indicateurs personnalises dans shared/indicators.py.
# Import depuis la bibliotheque partagee
import sys
sys.path.insert(0, '../shared')
try:
from indicators import (
CustomMomentumIndicator as SharedMomentum,
TrendStrengthIndicator as SharedTrend,
VolatilityBandIndicator,
IndicatorValue
)
print("Import depuis shared/indicators.py reussi")
print("\nIndicateurs disponibles:")
print(" - CustomMomentumIndicator : Momentum (ROC)")
print(" - TrendStrengthIndicator : Force de tendance [-1, +1]")
print(" - VolatilityBandIndicator : Bandes ATR personnalisees")
except ImportError as e:
print(f"Import depuis shared/ non disponible: {e}")
print("Utilisez les indicateurs definis dans ce notebook.")Import depuis shared/indicators.py reussi
Indicateurs disponibles:
- CustomMomentumIndicator : Momentum (ROC)
- TrendStrengthIndicator : Force de tendance [-1, +1]
- VolatilityBandIndicator : Bandes ATR personnaliseesOn peut combiner plusieurs indicateurs pour créer des signaux composites.
# [REFERENCE QC] Code a copier dans main.py QC Lab (non executable ici)
class CompositeIndicatorAlgorithm(QCAlgorithm):
"""
Demonstration de combinaison d'indicateurs.
- MACD Histogram
- RSI avec Bollinger Bands appliques sur le RSI
"""
def Initialize(self):
self.SetStartDate(2015, 1, 1)
self.SetEndDate(2024, 12, 31)
self.SetCash(100000)
symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
self.symbol = symbol
# Indicateurs de base
self.macd = self.MACD(symbol, 12, 26, 9, MovingAverageType.Exponential, Resolution.Daily)
self.rsi = self.RSI(symbol, 14, Resolution.Daily)
# === COMPOSITE: Bollinger Bands sur RSI ===
# Creer des BB qui s'appliquent a la sortie du RSI
# Utile pour detecter les extremes du RSI
self.rsi_bb = BollingerBands(14, 2, MovingAverageType.Simple)
# RegisterIndicator avec l'indicateur source comme 3eme argument
# Les BB seront mis a jour avec les valeurs du RSI
self.RegisterIndicator(symbol, self.rsi_bb, self.rsi)
# === CUSTOM: Indicateur personnalise ===
self.custom_momentum = CustomMomentumIndicator('MyMomentum', 20)
self.RegisterIndicator(symbol, self.custom_momentum, Resolution.Daily)
self.SetWarmup(30)
def OnData(self, data):
if self.IsWarmingUp:
return
if not (self.macd.IsReady and self.rsi.IsReady and self.rsi_bb.IsReady):
return
# MACD Histogram
macd_line = self.macd.Current.Value
signal_line = self.macd.Signal.Current.Value
macd_histogram = macd_line - signal_line
# RSI avec Bollinger Bands
rsi_value = self.rsi.Current.Value
rsi_bb_upper = self.rsi_bb.UpperBand.Current.Value
rsi_bb_lower = self.rsi_bb.LowerBand.Current.Value
rsi_bb_middle = self.rsi_bb.MiddleBand.Current.Value
# Custom momentum
if self.custom_momentum.IsReady:
momentum = self.custom_momentum.Current.Value
else:
momentum = 0
# Log hebdomadaire
if self.Time.weekday() == 0:
self.Debug(f"\n{self.Time.date()} - COMPOSITE INDICATORS:")
self.Debug(f" MACD Histogram: {macd_histogram:.4f}")
self.Debug(f" RSI: {rsi_value:.2f}")
self.Debug(f" RSI BB: Upper={rsi_bb_upper:.2f}, Middle={rsi_bb_middle:.2f}, Lower={rsi_bb_lower:.2f}")
self.Debug(f" Custom Momentum: {momentum:.4f}")
# Interpretation RSI BB
if rsi_value > rsi_bb_upper:
self.Debug(" RSI above BB upper -> Extremely overbought")
elif rsi_value < rsi_bb_lower:
self.Debug(" RSI below BB lower -> Extremely oversold")
print("CompositeIndicatorAlgorithm defini")
print("\nComposite Indicator Techniques:")
print(" - MACD Histogram: macd.Current.Value - macd.Signal.Current.Value")
print(" - RegisterIndicator(symbol, indicator, source_indicator)")
print(" Permet d'appliquer un indicateur sur la sortie d'un autre")Les indicateurs composites combinent plusieurs indicateurs pour créer des signaux plus robustes ou des méta-indicateurs.
1. Histogramme MACD (composant soustraitif) : - MACD_Histogram = MACD_Line - Signal_Line - Montre l’accélération du momentum (dérivée du MACD) - Croisement de zéro = Signal d’entrée/sortie précis - Divergences histogramme : Prix fait nouveau haut mais histogramme fait plus bas haut = affaiblissement
2. Bollinger Bands appliquées sur RSI (indicateur sur indicateur) : - Au lieu d’appliquer les BB sur le prix, on les applique sur la sortie du RSI - RSI > UpperBand = Extrêmement suracheté (plus que 70) - RSI < LowerBand = Extrêmement survendu (plus que 30) - Normalisation statistique du RSI (adaptatif à la volatilité du RSI)
Technique RegisterIndicator avec indicateur source :
Resolution.Daily)3. Combinaison de signaux multiples : Au lieu d’utiliser un seul indicateur, on combine plusieurs filtres : - Trend filter : SMA 200 (évite d’acheter dans un downtrend) - Entry trigger : RSI < 30 + MACD bullish cross - Confirmation : Volume > moyenne - Exit : RSI > 70 OR trailing stop
Avantages des indicateurs composites : 1. Réduction des faux signaux : Plusieurs conditions doivent être remplies 2. Adaptativité : Les indicateurs s’adaptent à la volatilité actuelle (BB sur RSI) 3. Hiérarchie temporelle : Trend filter (long terme) + Entry trigger (court terme)
Risque de sur-optimisation : - Trop d’indicateurs = Courbe d’apprentissage (overfitting) - Règle empirique : 3-5 indicateurs max par stratégie - Chaque indicateur doit avoir un rôle distinct et complémentaire
Types de combinaisons :
| Type | Description | Exemple |
|---|---|---|
| Additive | Additionne les signaux | RSI + MACD + STO |
| Multiplicative | Multiplie les confirmations | Trend × Momentum × Volume |
| Séquentielle | Filtres en cascade | Trend → Entry → Confirmation |
| Adaptive | Paramètres adaptatifs | BB sur RSI (volatilité du RSI) |
Analyser plusieurs timeframes permet d’avoir une vue complete : - Daily : Direction générale de la tendance - Hourly : Timing d’entree/sortie - Minute : Exécution precise
On utilise des Consolidators pour agreger les données dans des barres de resolution superieure.
# [REFERENCE QC] Code a copier dans main.py QC Lab (non executable ici)
class MultiTimeframeAlgorithm(QCAlgorithm):
"""
Analyse multi-timeframe avec consolidators.
- SMA Daily pour la tendance
- SMA Hourly pour le timing
"""
def Initialize(self):
self.SetStartDate(2015, 1, 1)
self.SetEndDate(2024, 12, 31)
self.SetCash(100000)
# Ajouter avec resolution minute (la plus fine)
symbol = self.AddEquity("SPY", Resolution.Minute).Symbol
self.symbol = symbol
# === DAILY SMA (trend filter) ===
# Methode directe: QuantConnect agregera automatiquement
self.sma_daily = self.SMA(symbol, 20, Resolution.Daily)
# === HOURLY SMA (timing) ===
# Creer l'indicateur manuellement
self.sma_hourly = SimpleMovingAverage(20)
# Creer un consolidator pour aggreger en barres horaires
hourly_consolidator = TradeBarConsolidator(timedelta(hours=1))
# Handler pour les barres horaires consolidees
hourly_consolidator.DataConsolidated += self.OnHourlyBar
# Enregistrer le consolidator avec le symbole
self.SubscriptionManager.AddConsolidator(symbol, hourly_consolidator)
# Enregistrer l'indicateur avec le consolidator
# L'indicateur sera mis a jour quand le consolidator emet une barre
self.RegisterIndicator(symbol, self.sma_hourly, hourly_consolidator)
# === 4-HOURLY RSI ===
self.rsi_4h = RelativeStrengthIndex(14)
four_hour_consolidator = TradeBarConsolidator(timedelta(hours=4))
four_hour_consolidator.DataConsolidated += self.OnFourHourBar
self.SubscriptionManager.AddConsolidator(symbol, four_hour_consolidator)
self.RegisterIndicator(symbol, self.rsi_4h, four_hour_consolidator)
self.SetWarmup(timedelta(days=30))
def OnHourlyBar(self, sender, bar):
"""Appele quand une barre horaire est consolidee."""
if self.sma_hourly.IsReady:
# Log toutes les 8 heures
if bar.EndTime.hour % 8 == 0:
self.Debug(f"{bar.EndTime} - Hourly SMA: {self.sma_hourly.Current.Value:.2f}")
def OnFourHourBar(self, sender, bar):
"""Appele quand une barre 4H est consolidee."""
if self.rsi_4h.IsReady:
self.Debug(f"{bar.EndTime} - 4H RSI: {self.rsi_4h.Current.Value:.2f}")
def OnData(self, data):
if self.IsWarmingUp:
return
if not data.ContainsKey(self.symbol):
return
if not (self.sma_daily.IsReady and self.sma_hourly.IsReady):
return
price = self.Securities[self.symbol].Price
sma_daily = self.sma_daily.Current.Value
sma_hourly = self.sma_hourly.Current.Value
# === LOGIQUE MULTI-TIMEFRAME ===
# Regle: Acheter seulement si :
# 1. Prix > SMA Daily (tendance haussiere)
# 2. Prix > SMA Hourly (timing favorable)
# 3. RSI 4H < 50 (pas surachete)
daily_uptrend = price > sma_daily
hourly_uptrend = price > sma_hourly
rsi_favorable = self.rsi_4h.IsReady and self.rsi_4h.Current.Value < 50
if daily_uptrend and hourly_uptrend and rsi_favorable:
if not self.Portfolio[self.symbol].Invested:
self.SetHoldings(self.symbol, 1.0)
self.Debug(f"{self.Time}: BUY - Multi-TF aligned")
elif not daily_uptrend: # Sortie si tendance daily inverse
if self.Portfolio[self.symbol].Invested:
self.Liquidate(self.symbol)
self.Debug(f"{self.Time}: SELL - Daily trend reversed")
print("MultiTimeframeAlgorithm defini")
print("\nMulti-Timeframe Setup:")
print(" 1. TradeBarConsolidator(timedelta(hours=1)) : Agregation horaire")
print(" 2. consolidator.DataConsolidated += handler : Handler de callback")
print(" 3. SubscriptionManager.AddConsolidator(symbol, consolidator)")
print(" 4. RegisterIndicator(symbol, indicator, consolidator)")L’analyse multi-timeframe est cruciale pour obtenir une vue complète du marché et éviter le “bruit” d’une seule timeframe.
Hiérarchie des Timeframes :
| Timeframe | Rôle | Horizon |
|---|---|---|
| Daily | Direction générale de la tendance | Long terme (semaines/mois) |
| Hourly (4H) | Timing des entrées/sorties | Moyen terme (jours) |
| 15-min/5-min | Exécution précise | Court terme (heures) |
Principe de la “Top-Down Analysis” : 1. Daily : Identifier la tendance principale (SMA 200, ADX) 2. Hourly : Attendre un pullback vers la zone de valeur 3. 5-min : Entrer sur le signal de retournement court terme
Architecture des Consolidators dans QuantConnect :
# Créer un consolidateur pour aggréger les données
hourly_consolidator = TradeBarConsolidator(timedelta(hours=1))
# Handler pour les barres consolidées
hourly_consolidator.DataConsolidated += self.OnHourlyBar
# Enregistrer avec le SubscriptionManager
self.SubscriptionManager.AddConsolidator(symbol, hourly_consolidator)
# Enregistrer l'indicateur avec le consolidator
self.RegisterIndicator(symbol, self.sma_hourly, hourly_consolidator)Cycle de vie d’un Consolidator : 1. Les données minute arrivent en temps réel 2. Le consolidator accumule N barres minute pour former 1 barre consolidée 3. Quand la barre est complète → Événement DataConsolidated déclenché 4. Les indicateurs enregistrés avec ce consolidator sont mis à jour
Stratégie Multi-TF Typique : - Entrée : Prix > SMA Daily (trend haussier) ET Prix > SMA Hourly (timing favorable) - Sortie : Prix < SMA Daily (trend inversée) OU RSI 4H > 70 (suracheté)
Avantages de l’analyse multi-timeframe : 1. Réduction du bruit : Le filtre daily élimine les faux signaux court terme 2. Meilleur risk/reward : On entre avec la tendance, pas contre elle 3. Sortie anticipée : Le timeframe court terme prévient le retournement avant le daily
Désavantages : 1. Complexité accrue : Plus de code, plus de bugs potentiels 2. Latence : Les timeframes longs ont moins de signaux (opportunités manquées) 3. Warmup plus long : Besoin de plus de données historiques
Règle d’or : - Jamais lutter contre le trend principal (daily) - Utiliser le TF court pour le timing uniquement - Si conflict entre TFs : Toujours privilégier le TF le plus long
Combinons plusieurs indicateurs pour une stratégie robuste :
| Composant | Indicateur | Rôle |
|---|---|---|
| Trend Filter | SMA 200 | Filtre directionnel |
| Entry Signal | RSI < 30 + MACD bullish cross | Condition d’achat |
| Exit Signal | RSI > 70 OR trailing stop | Condition de vente |
| Confirmation | Volume > SMA volume | Validation |
Règles : 1. Acheter seulement si prix > SMA 200 (uptrend) 2. Signal d’achat : RSI oversold ET MACD bullish cross 3. Confirmation : Volume superieur a la moyenne 4. Exit : RSI overbought OU trailing stop touch
# [REFERENCE QC] Code a copier dans main.py QC Lab (non executable ici)
class MultiIndicatorStrategyAlgorithm(QCAlgorithm):
"""
Strategie complete combinant plusieurs indicateurs.
- Trend Filter: SMA 200
- Entry: RSI oversold + MACD bullish cross
- Exit: RSI overbought OR trailing stop
- Confirmation: Volume > average
"""
def Initialize(self):
self.SetStartDate(2015, 1, 1)
self.SetEndDate(2024, 12, 31)
self.SetCash(100000)
# Configuration
symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
self.symbol = symbol
# === TREND FILTER ===
self.sma_200 = self.SMA(symbol, 200, Resolution.Daily)
# === ENTRY INDICATORS ===
self.rsi = self.RSI(symbol, 14, MovingAverageType.Wilders, Resolution.Daily)
self.macd = self.MACD(symbol, 12, 26, 9, MovingAverageType.Exponential, Resolution.Daily)
# === VOLUME CONFIRMATION ===
# SMA du volume pour comparaison
self.volume_sma = self.SMA(symbol, 20, Resolution.Daily, Field.Volume)
# === ROLLING WINDOWS pour detecter les croisements ===
self.macd_hist_window = RollingWindow[float](3)
self.rsi_window = RollingWindow[float](3)
# === TRAILING STOP ===
self.trailing_stop_pct = 0.05 # 5% trailing stop
self.highest_price_since_entry = 0
self.entry_price = 0
# Parametres RSI
self.rsi_oversold = 30
self.rsi_overbought = 70
self.SetWarmup(200)
def OnData(self, data):
if self.IsWarmingUp:
return
if not data.ContainsKey(self.symbol):
return
# Verifier que tous les indicateurs sont prets
if not (self.sma_200.IsReady and self.rsi.IsReady and
self.macd.IsReady and self.volume_sma.IsReady):
return
# Recuperer les donnees
bar = data[self.symbol]
price = bar.Close
volume = bar.Volume
# Mettre a jour les Rolling Windows
macd_histogram = self.macd.Current.Value - self.macd.Signal.Current.Value
self.macd_hist_window.Add(macd_histogram)
self.rsi_window.Add(self.rsi.Current.Value)
if not self.macd_hist_window.IsReady:
return
# === ANALYSE DES CONDITIONS ===
# 1. Trend Filter
in_uptrend = price > self.sma_200.Current.Value
# 2. RSI Conditions
rsi_current = self.rsi.Current.Value
rsi_oversold = rsi_current < self.rsi_oversold
rsi_overbought = rsi_current > self.rsi_overbought
# 3. MACD Bullish Cross (histogram passe de negatif a positif)
macd_bullish_cross = (self.macd_hist_window[1] < 0 and
self.macd_hist_window[0] > 0)
# 4. Volume Confirmation
volume_confirmed = volume > self.volume_sma.Current.Value
# === LOGIQUE DE TRADING ===
if not self.Portfolio[self.symbol].Invested:
# CONDITIONS D'ENTREE
entry_signal = in_uptrend and rsi_oversold and macd_bullish_cross and volume_confirmed
if entry_signal:
self.SetHoldings(self.symbol, 1.0)
self.entry_price = price
self.highest_price_since_entry = price
self.Debug(f"\n{self.Time.date()} === BUY SIGNAL ===")
self.Debug(f" Price: {price:.2f}")
self.Debug(f" SMA 200: {self.sma_200.Current.Value:.2f}")
self.Debug(f" RSI: {rsi_current:.2f} (oversold)")
self.Debug(f" MACD Histogram: {macd_histogram:.4f} (bullish cross)")
self.Debug(f" Volume: {volume:,.0f} vs Avg: {self.volume_sma.Current.Value:,.0f}")
else:
# Mettre a jour trailing stop
if price > self.highest_price_since_entry:
self.highest_price_since_entry = price
# Calculer le niveau du trailing stop
trailing_stop_level = self.highest_price_since_entry * (1 - self.trailing_stop_pct)
# CONDITIONS DE SORTIE
exit_rsi_overbought = rsi_overbought
exit_trailing_stop = price < trailing_stop_level
exit_trend_reversal = not in_uptrend # Prix passe sous SMA 200
if exit_rsi_overbought or exit_trailing_stop or exit_trend_reversal:
# Determiner la raison de sortie
if exit_rsi_overbought:
exit_reason = f"RSI overbought ({rsi_current:.2f})"
elif exit_trailing_stop:
exit_reason = f"Trailing stop hit (stop: {trailing_stop_level:.2f})"
else:
exit_reason = "Trend reversal (price below SMA 200)"
# Calculer P&L
pnl_pct = ((price - self.entry_price) / self.entry_price) * 100
self.Liquidate(self.symbol)
self.Debug(f"\n{self.Time.date()} === SELL SIGNAL ===")
self.Debug(f" Exit Reason: {exit_reason}")
self.Debug(f" Entry: {self.entry_price:.2f} -> Exit: {price:.2f}")
self.Debug(f" P&L: {pnl_pct:+.2f}%")
def OnEndOfAlgorithm(self):
"""Resume final."""
self.Debug(f"\n{'='*50}")
self.Debug("MULTI-INDICATOR STRATEGY - FINAL RESULTS")
self.Debug(f"{'='*50}")
self.Debug(f"Final Portfolio Value: ${self.Portfolio.TotalPortfolioValue:,.2f}")
self.Debug(f"Total Return: {(self.Portfolio.TotalPortfolioValue / 100000 - 1) * 100:.2f}%")
print("MultiIndicatorStrategyAlgorithm defini")Cette section résume la logique de trading combinée que nous venons d’implémenter. La stratégie utilise 4 couches de filtrage pour augmenter la probabilité de succès :
1. Trend Filter (SMA 200) - Empêche d’acheter dans un marché baissier - Réduit le drawdown en évitant les “falling knives”
2. Entry Signals (RSI + MACD) - RSI < 30 : Le marché est survendu (correction à court terme) - MACD bullish cross : Le momentum s’inverse à la hausse - Confluence : Les deux conditions doivent être vraies simultanément
3. Volume Confirmation - Évite les faux signaux sur volume faible - Les mouvements sans volume sont souvent des “fakeouts”
4. Exit Management - RSI > 70 : Take profit lors des extrêmes - Trailing stop 5% : Protection du capital en temps réel - Trend reversal : Sortie d’urgence si le marché s’effondre
Performance attendue : - Moins de trades qu’une stratégie simple, mais meilleure qualité - Drawdown réduit grâce au trend filter - Profits protégés par le trailing stop
# Resume de la strategie
print("="*60)
print("MULTI-INDICATOR STRATEGY - RESUME")
print("="*60)
print("\n1. TREND FILTER:")
print(" - SMA(200) comme filtre directionnel")
print(" - Acheter seulement si prix > SMA 200")
print("\n2. ENTRY SIGNALS:")
print(" - RSI(14) < 30 : Condition oversold")
print(" - MACD histogram passe de negatif a positif : Bullish cross")
print(" - Les deux conditions doivent etre vraies simultanement")
print("\n3. VOLUME CONFIRMATION:")
print(" - Volume > SMA(20) du volume")
print(" - Filtre les faux signaux a faible volume")
print("\n4. EXIT CONDITIONS:")
print(" - RSI > 70 : Overbought (take profit)")
print(" - Trailing Stop 5% : Protection du capital")
print(" - Prix < SMA 200 : Trend reversal")
print("\n5. RISK MANAGEMENT:")
print(" - Trailing stop dynamique (5%)")
print(" - Position sizing: 100% (peut etre ajuste)")============================================================MULTI-INDICATOR STRATEGY - RESUME============================================================1. TREND FILTER: - SMA(200) comme filtre directionnel - Acheter seulement si prix > SMA 2002. ENTRY SIGNALS: - RSI(14) < 30 : Condition oversold - MACD histogram passe de negatif a positif : Bullish cross - Les deux conditions doivent etre vraies simultanement3. VOLUME CONFIRMATION: - Volume > SMA(20) du volume - Filtre les faux signaux a faible volume4. EXIT CONDITIONS: - RSI > 70 : Overbought (take profit) - Trailing Stop 5% : Protection du capital - Prix < SMA 200 : Trend reversal5. RISK MANAGEMENT: - Trailing stop dynamique (5%) - Position sizing: 100% (peut etre ajuste)Dans ce notebook, nous avons couvert :
| Concept | Point Cle |
|---|---|
| IsReady | Toujours verifier avant d’utiliser un indicateur |
| Warmup | Définir SetWarmup() pour eviter les signaux prematures |
| Rolling Windows | Indispensables pour detecter les croisements |
| Consolidators | Permettent l’analyse multi-timeframe |
| Combinaison | Plusieurs indicateurs > un seul indicateur |
Nous couvrirons : - Exécution et Orders avances - Position sizing et money management - Slippage et couts de transaction
Notebook complete. Pret pour QC-Py-12.