Reproduire l’analyse exploratoire de research.ipynb avec les données natives QuantConnect.
QuantBook initialise.11 ETFs sectoriels GICS + SPY (benchmark et regime filter).
sector_etfs = ['XLK', 'XLF', 'XLE', 'XLV', 'XLI', 'XLY', 'XLP', 'XLU', 'XLB', 'XLRE', 'XLC']
all_tickers = sector_etfs + ['SPY']
symbols = {}
for ticker in all_tickers:
symbols[ticker] = qb.add_equity(ticker, Resolution.DAILY).symbol
start = datetime(2016, 1, 1)
end = datetime(2026, 1, 1)
history = qb.history(list(symbols.values()), start, end, Resolution.DAILY)
closes = history['close'].unstack(level=0)
symbol_to_ticker = {str(v): k for k, v in symbols.items()}
closes.columns = [symbol_to_ticker.get(str(c), str(c)) for c in closes.columns]
closes = closes.dropna()
print(f"Periode: {closes.index[0].date()} a {closes.index[-1].date()}")
print(f"Donnees: {len(closes)} jours de trading")
print(f"Secteurs: {[t for t in closes.columns if t != 'SPY']}")Periode: 2016-01-04 a 2025-12-31
Donnees: 2514 jours de trading
Secteurs: []Identifier les secteurs les plus/moins performants pour comprendre le comportement de rotation sectorielle.
returns_df = closes.pct_change()
bt_data = closes.loc['2018-01-01':]
print(f"{'Secteur':<8} {'Rend. Total':>12} {'Vol. Ann.':>12} {'Sharpe':>8}")
print("-" * 42)
for etf in sector_etfs + ['SPY']:
if etf not in bt_data.columns:
continue
ret = bt_data[etf].iloc[-1] / bt_data[etf].iloc[0] - 1
vol = returns_df[etf].loc['2018-01-01':].std() * np.sqrt(252)
years = len(bt_data) / 252
cagr = (1 + ret) ** (1 / years) - 1
sharpe = (cagr - 0.03) / vol if vol > 0 else 0
print(f"{etf:<8} {ret:>11.1%} {vol:>11.1%} {sharpe:>7.2f}")Secteur Rend. Total Vol. Ann. Sharpe
------------------------------------------
SPY 56.5% 14.3% 0.19def compute_rsi(prices, period=14):
"""Calcul RSI standard."""
delta = prices.diff()
gain = delta.where(delta > 0, 0)
loss = -delta.where(delta < 0, 0)
avg_gain = gain.rolling(window=period).mean()
avg_loss = loss.rolling(window=period).mean()
rs = avg_gain / avg_loss
return 100 - (100 / (1 + rs))
def compute_bb(prices, period=20, num_std=2):
"""Calcul Bollinger Bands - retourne le %B."""
sma = prices.rolling(period).mean()
std = prices.rolling(period).std()
upper = sma + num_std * std
lower = sma - num_std * std
pct_b = (prices - lower) / (upper - lower)
return pct_b
# Calculer RSI(14) et BB%B(20,2) pour tous les secteurs
rsi_data = {}
bb_data = {}
for etf in sector_etfs:
if etf in closes.columns:
rsi_data[etf] = compute_rsi(closes[etf], 14)
bb_data[etf] = compute_bb(closes[etf], 20, 2)
rsi_df = pd.DataFrame(rsi_data)
bb_df = pd.DataFrame(bb_data)
print(f"RSI moyen par secteur:")
print(rsi_df.mean().round(1).to_string())
print(f"\nJours RSI < 30 par secteur:")
print((rsi_df < 30).sum().to_string())RSI moyen par secteur:
Series([], )
Jours RSI < 30 par secteur:
Series([], )Le nombre de jours ou RSI < 30 indique combien de fois le signal de mean reversion se declenche. Trop peu = cash drag (règle #19). Trop souvent = bruit.
def mean_reversion_backtest(closes, sector_etfs, rsi_threshold=30, rsi_exit=50,
max_positions=3, holding_days=10,
use_sma_filter=False, sma_period=200,
use_bollinger=False, bb_entry=0.0, bb_exit=0.5,
stop_loss=-0.10):
"""Backtest mean reversion sur secteurs."""
returns_df = closes.pct_change()
rsi = {etf: compute_rsi(closes[etf], 14) for etf in sector_etfs if etf in closes.columns}
bb = {etf: compute_bb(closes[etf], 20, 2) for etf in sector_etfs if etf in closes.columns}
sma200 = closes['SPY'].rolling(sma_period).mean() if use_sma_filter else None
bt_start = max(sma_period if use_sma_filter else 20, 20) + 1
positions = {} # {etf: {'entry_price': x, 'entry_day': i, 'days_held': 0}}
portfolio_values = [1.0]
n_trades = 0
wins = 0
for i in range(bt_start, len(closes)):
# Update positions
daily_pnl = 0
to_close = []
for etf, pos in positions.items():
if etf not in returns_df.columns:
continue
ret = returns_df[etf].iloc[i]
daily_pnl += ret / max(len(positions), 1)
pos['days_held'] += 1
current_price = closes[etf].iloc[i]
pos_return = current_price / pos['entry_price'] - 1
# Exit conditions
should_exit = False
if use_bollinger:
if bb[etf].iloc[i] > bb_exit:
should_exit = True
else:
if rsi[etf].iloc[i] > rsi_exit:
should_exit = True
if pos['days_held'] >= holding_days:
should_exit = True
if pos_return <= stop_loss:
should_exit = True
if should_exit:
to_close.append(etf)
n_trades += 1
if pos_return > 0:
wins += 1
for etf in to_close:
del positions[etf]
portfolio_values.append(portfolio_values[-1] * (1 + daily_pnl))
# Entry signals
if len(positions) < max_positions:
if use_sma_filter and closes['SPY'].iloc[i] < sma200.iloc[i]:
continue
candidates = []
for etf in sector_etfs:
if etf in positions or etf not in closes.columns:
continue
if use_bollinger:
if bb[etf].iloc[i] < bb_entry:
candidates.append((etf, bb[etf].iloc[i]))
else:
if rsi[etf].iloc[i] < rsi_threshold:
candidates.append((etf, rsi[etf].iloc[i]))
candidates.sort(key=lambda x: x[1])
for etf, _ in candidates[:max_positions - len(positions)]:
positions[etf] = {'entry_price': closes[etf].iloc[i], 'entry_day': i, 'days_held': 0}
vals = np.array(portfolio_values)
rets = np.diff(vals) / vals[:-1]
total_ret = vals[-1] / vals[0] - 1
years = len(rets) / 252
cagr = (1 + total_ret) ** (1 / years) - 1 if years > 0 else 0
vol = np.std(rets) * np.sqrt(252)
sharpe = (cagr - 0.03) / vol if vol > 0.001 else 0
cum = pd.Series(vals[1:])
max_dd = ((cum - cum.expanding().max()) / cum.expanding().max()).min()
win_rate = wins / n_trades if n_trades > 0 else 0
trades_per_year = n_trades / years if years > 0 else 0
return {'sharpe': sharpe, 'cagr': cagr, 'max_dd': max_dd, 'vol': vol,
'cum': cum, 'n_trades': n_trades, 'win_rate': win_rate,
'trades_per_year': trades_per_year}
print("Fonction de backtest mean reversion definie.")Fonction de backtest mean reversion definie.print(f"{'RSI Seuil':<12} {'Sharpe':>8} {'CAGR':>8} {'MaxDD':>8} {'Trades/an':>10} {'Win%':>6}")
print("-" * 60)
results_rsi = {}
for threshold in [20, 25, 30, 35]:
r = mean_reversion_backtest(closes, sector_etfs, rsi_threshold=threshold,
max_positions=3, holding_days=10, stop_loss=-0.10)
results_rsi[threshold] = r
print(f"RSI < {threshold:<6} {r['sharpe']:>8.3f} {r['cagr']:>7.1%} {r['max_dd']:>7.1%} {r['trades_per_year']:>9.1f} {r['win_rate']:>5.0%}")RSI Seuil Sharpe CAGR MaxDD Trades/an Win%
------------------------------------------------------------
RSI < 20 0.000 0.0% 0.0% 0.0 0%
RSI < 25 0.000 0.0% 0.0% 0.0 0%
RSI < 30 0.000 0.0% 0.0% 0.0 0%
RSI < 35 0.000 0.0% 0.0% 0.0 0%Attention au trade-off frequence vs qualite (règle #19 du backlog): - RSI < 20 = peu de trades, potentiel cash drag - RSI < 35 = beaucoup de trades, signal dilue - Sur 11 secteurs, RSI < 30 devrait donner ~20-30 trades/an (acceptable)
print(f"{'Signal':<20} {'Sharpe':>8} {'CAGR':>8} {'MaxDD':>8} {'Trades/an':>10}")
print("-" * 55)
# Best RSI
best_rsi = max(results_rsi.items(), key=lambda x: x[1]['sharpe'])
print(f"{'RSI < ' + str(best_rsi[0]):<20} {best_rsi[1]['sharpe']:>8.3f} {best_rsi[1]['cagr']:>7.1%} {best_rsi[1]['max_dd']:>7.1%} {best_rsi[1]['trades_per_year']:>9.1f}")
# Bollinger Bands
for bb_entry in [0.0, -0.1, 0.1]:
r = mean_reversion_backtest(closes, sector_etfs, use_bollinger=True,
bb_entry=bb_entry, bb_exit=0.5,
max_positions=3, holding_days=10, stop_loss=-0.10)
print(f"{'BB %B < ' + str(bb_entry):<20} {r['sharpe']:>8.3f} {r['cagr']:>7.1%} {r['max_dd']:>7.1%} {r['trades_per_year']:>9.1f}")Signal Sharpe CAGR MaxDD Trades/an
-------------------------------------------------------
RSI < 20 0.000 0.0% 0.0% 0.0
BB %B < 0.0 0.000 0.0% 0.0% 0.0
BB %B < -0.1 0.000 0.0% 0.0% 0.0
BB %B < 0.1 0.000 0.0% 0.0% 0.0Attention: règle #15 du backlog - SMA100 filter trop restrictif pour mean reversion.
best_thresh = best_rsi[0]
print(f"{'Config':<25} {'Sharpe':>8} {'CAGR':>8} {'MaxDD':>8} {'Trades/an':>10}")
print("-" * 60)
r_no = mean_reversion_backtest(closes, sector_etfs, rsi_threshold=best_thresh,
max_positions=3, use_sma_filter=False)
print(f"{'Sans SMA filter':<25} {r_no['sharpe']:>8.3f} {r_no['cagr']:>7.1%} {r_no['max_dd']:>7.1%} {r_no['trades_per_year']:>9.1f}")
for sma in [100, 200]:
r = mean_reversion_backtest(closes, sector_etfs, rsi_threshold=best_thresh,
max_positions=3, use_sma_filter=True, sma_period=sma)
print(f"{f'SMA{sma} filter':<25} {r['sharpe']:>8.3f} {r['cagr']:>7.1%} {r['max_dd']:>7.1%} {r['trades_per_year']:>9.1f}")
print("\nNote: SMA filter devrait reduire MaxDD mais aussi reduire les trades.")
print("Mean reversion a besoin de volatilite - filtrer les baisses = filtrer les opportunites.")Config Sharpe CAGR MaxDD Trades/an
------------------------------------------------------------
Sans SMA filter 0.000 0.0% 0.0% 0.0
SMA100 filter 0.000 0.0% 0.0% 0.0
SMA200 filter 0.000 0.0% 0.0% 0.0
Note: SMA filter devrait reduire MaxDD mais aussi reduire les trades.
Mean reversion a besoin de volatilite - filtrer les baisses = filtrer les opportunites.| Test | Résultat QuantBook | Coherent avec yfinance? |
|---|---|---|
| Seuil RSI | (a remplir) | (a verifier) |
| RSI vs BB | (a remplir) | (a verifier) |
| SMA filter | (a remplir) | (a verifier) |