Analyser la combinaison de deux stratégies complementaires: - SectorMomentum: Dual momentum SPY/IEF/GLD avec lookbacks multiples (1/3/6/12 mois) - RegimeSwitching: Momentum en bull, mean-reversion en bear/sideways
# Setup QuantBook
from AlgorithmImports import *
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import warnings
warnings.filterwarnings('ignore')
plt.style.use('seaborn-v0_8-darkgrid')
plt.rcParams['figure.figsize'] = (14, 5)
qb = QuantBook()
print("QuantBook initialise.")QuantBook initialise.On charge les 4 assets de l’univers combine: SPY, QQQ, IEF, GLD.
# Univers combine
all_tickers = ["SPY", "QQQ", "IEF", "GLD"]
symbols = {}
for ticker in all_tickers:
symbols[ticker] = qb.add_equity(ticker, Resolution.DAILY).symbol
# Charger l'historique (2010-2026)
start = datetime(2010, 1, 1)
end = datetime(2026, 1, 1)
history = qb.history(list(symbols.values()), start, end, Resolution.DAILY)
print(f"Donnees chargees: {len(history)} lignes")Donnees chargees: 7741 lignesPivot de la série ‘close’ en DataFrame large, avec remapping des colonnes Symbol → ticker pour Framework_Composite_MomentumRegime.
# Pivoter les donnees
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")Periode: 2011-03-23 a 2025-12-31
Donnees: 3717 jours de tradingdef compute_sector_momentum_signals(closes, tickers, lookbacks=[21, 63, 126, 252], weights=[0.4, 0.2, 0.2, 0.2]):
"""Signaux SectorMomentum: composite momentum multi-lookbacks."""
signals = pd.DataFrame(index=closes.index, columns=tickers)
scores = pd.DataFrame(index=closes.index, columns=tickers)
for ticker in tickers:
if ticker not in closes.columns:
continue
prices = closes[ticker]
for i in range(max(lookbacks), len(prices)):
composite_score = 0.0
valid = True
for lb, w in zip(lookbacks, weights):
past_price = prices.iloc[i - lb]
if past_price <= 0:
valid = False
break
momentum = (prices.iloc[i] / past_price) - 1
composite_score += w * momentum
if valid:
scores[ticker].iloc[i] = composite_score
# SMA200 filter
sma200 = closes['SPY'].rolling(200).mean() if 'SPY' in closes.columns else None
for i in range(max(lookbacks), len(closes)):
current_scores = {}
for t in tickers:
if t in scores.columns:
score = scores[t].iloc[i]
if pd.notna(score):
current_scores[t] = score
if not current_scores:
continue
best = max(current_scores, key=current_scores.get)
# SPY only if positive momentum AND above SMA200
if best == "SPY":
if sma200 is not None:
spy_price = closes['SPY'].iloc[i]
spy_sma = sma200.iloc[i]
if pd.notna(spy_sma) and (spy_price <= spy_sma or current_scores[best] <= 0):
# Fallback to defensive
defensive = {k: v for k, v in current_scores.items() if k in ["IEF", "GLD"]}
if defensive:
best = max(defensive, key=defensive.get)
for t in tickers:
signals[t].iloc[i] = 1 if t == best else 0
return signals
def compute_rsi(prices, period=14):
"""Calcule RSI."""
delta = prices.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
return rsi
def detect_regime(closes):
"""Detecte le regime: bull, bear, sideways."""
if 'SPY' not in closes.columns:
return pd.Series(index=closes.index, data='unknown')
prices = closes['SPY']
sma50 = prices.rolling(50).mean()
sma200 = prices.rolling(200).mean()
regimes = []
for i in range(len(closes)):
if pd.isna(sma50.iloc[i]) or pd.isna(sma200.iloc[i]):
regimes.append('unknown')
continue
price = prices.iloc[i]
s50 = sma50.iloc[i]
s200 = sma200.iloc[i]
if price > s200 and s50 > s200:
regimes.append('bull')
elif price < s200 and s50 < s200:
regimes.append('bear')
else:
regimes.append('sideways')
return pd.Series(regimes, index=closes.index)
def compute_regime_switching_signals(closes, tickers, regimes):
"""Signaux RegimeSwitching: momentum en bull, mean-reversion en bear/sideways."""
signals = pd.DataFrame(index=closes.index, columns=tickers)
# RSI pour mean-reversion
rsi_values = {}
for t in ["SPY", "QQQ"]:
if t in closes.columns:
rsi_values[t] = compute_rsi(closes[t])
for i in range(252, len(closes)):
regime = regimes.iloc[i]
if regime == 'bull':
# Momentum: SPY 70%, QQQ 30%
for t in tickers:
if t == "SPY":
signals[t].iloc[i] = 0.7
elif t == "QQQ":
signals[t].iloc[i] = 0.3
else:
signals[t].iloc[i] = 0
elif regime == 'bear':
# Defensive: GLD 50%, IEF 50%
for t in tickers:
if t == "GLD":
signals[t].iloc[i] = 0.5
elif t == "IEF":
signals[t].iloc[i] = 0.5
else:
signals[t].iloc[i] = 0
else: # sideways
# Check RSI oversold
oversold = []
for t in ["SPY", "QQQ"]:
if t in rsi_values and i < len(rsi_values[t]):
rsi = rsi_values[t].iloc[i]
if pd.notna(rsi) and rsi < 30:
oversold.append(t)
if oversold:
# Mean-reversion: oversold 30%, GLD 35%, IEF 35%
for t in tickers:
if t in oversold:
signals[t].iloc[i] = 0.30 / len(oversold)
elif t == "GLD":
signals[t].iloc[i] = 0.35
elif t == "IEF":
signals[t].iloc[i] = 0.35
else:
signals[t].iloc[i] = 0
else:
# Reduced equity: SPY 30%, GLD 35%, IEF 35%
for t in tickers:
if t == "SPY":
signals[t].iloc[i] = 0.30
elif t == "GLD":
signals[t].iloc[i] = 0.35
elif t == "IEF":
signals[t].iloc[i] = 0.35
else:
signals[t].iloc[i] = 0
return signals
# Generer les signaux
mom_tickers = ["SPY", "IEF", "GLD"]
regime_tickers = ["SPY", "QQQ", "IEF", "GLD"]
mom_signals = compute_sector_momentum_signals(closes, mom_tickers)
regimes = detect_regime(closes)
regime_signals = compute_regime_switching_signals(closes, regime_tickers, regimes)
print("Signaux SectorMomentum (derniers 5 jours):")
print(mom_signals.iloc[-5:])
print(f"\nRegimes (derniers 5 jours):")
print(regimes.iloc[-5:])Signaux SectorMomentum (derniers 5 jours):
SPY IEF GLD
time
2025-12-24 13:00:00 1 0 0
2025-12-26 16:00:00 1 0 0
2025-12-29 16:00:00 1 0 0
2025-12-30 16:00:00 1 0 0
2025-12-31 16:00:00 1 0 0
Regimes (derniers 5 jours):
time
2025-12-24 13:00:00 sideways
2025-12-26 16:00:00 sideways
2025-12-29 16:00:00 sideways
2025-12-30 16:00:00 sideways
2025-12-31 16:00:00 sideways
dtype: objectLes deux stratégies utilisent SPY/IEF/GLD, creant une synergie potentielle.
Fonction de backtest combinant SectorMomentum + RegimeSwitching.
def backtest_composite(closes, mom_signals, regime_signals,
mom_alloc=0.60, regime_alloc=0.40,
rebal_freq=21):
"""
Backtest du composite Momentum + RegimeSwitching.
Args:
mom_alloc: Allocation SectorMomentum (ex: 0.60 = 60%)
regime_alloc: Allocation RegimeSwitching (ex: 0.40 = 40%)
rebal_freq: Frequence rebalance (21 = mensuel ~)
"""
returns_df = closes.pct_change()
portfolio_values = [1.0]
warmup = 252
counter = 0
for i in range(warmup, len(closes)):
# Rebalance mensuel
counter += 1
if counter >= rebal_freq:
counter = 0
# Calcul du return
port_return = 0.0
# Contribution SectorMomentum
for t in mom_tickers:
if t in mom_signals.columns:
signal = mom_signals[t].iloc[i]
if signal > 0 and t in returns_df.columns:
port_return += mom_alloc * signal * returns_df[t].iloc[i]
# Contribution RegimeSwitching
for t in regime_tickers:
if t in regime_signals.columns:
signal = regime_signals[t].iloc[i]
if signal > 0 and t in returns_df.columns:
port_return += regime_alloc * signal * returns_df[t].iloc[i]
portfolio_values.append(portfolio_values[-1] * (1 + port_return))
# Metriques
returns = np.diff(portfolio_values) / np.array(portfolio_values[:-1])
cum_returns = pd.Series(portfolio_values[1:], index=closes.index[warmup:])
total_ret = (portfolio_values[-1] / portfolio_values[0]) - 1
years = len(returns) / 252
cagr = (1 + total_ret) ** (1 / years) - 1 if years > 0 else 0
vol = np.std(returns) * np.sqrt(252)
sharpe = (cagr - 0.03) / vol if vol > 0.001 else 0
running_max = cum_returns.expanding().max()
drawdown = (cum_returns - running_max) / running_max
max_dd = drawdown.min()
return {
'cum': cum_returns,
'sharpe': sharpe,
'cagr': cagr,
'max_dd': max_dd,
'vol': vol,
'final_value': portfolio_values[-1]
}
print("Fonction de backtest composite definie.")Fonction de backtest composite definie.On teste différentes allocations Trend/RegimeSwitching.
# Test differentes allocations
allocations = [
(0.55, 0.45, "T55/RS45"),
(0.60, 0.40, "T60/RS40"),
(0.65, 0.35, "T65/RS35"),
]
results = {}
print(f"{'Allocation':<15} {'Sharpe':>8} {'CAGR':>8} {'MaxDD':>8} {'Vol':>8}")
print("-" * 50)
for mom_alloc, regime_alloc, name in allocations:
r = backtest_composite(closes, mom_signals, regime_signals,
mom_alloc=mom_alloc, regime_alloc=regime_alloc)
results[name] = r
print(f"{name:<15} {r['sharpe']:>8.3f} {r['cagr']:>7.1%} {r['max_dd']:>7.1%} {r['vol']:>7.1%}")
best_alloc = max(results.items(), key=lambda x: x[1]['sharpe'])
print(f"\nMeilleure allocation: {best_alloc[0]} (Sharpe={best_alloc[1]['sharpe']:.3f})")Allocation Sharpe CAGR MaxDD Vol
--------------------------------------------------
T55/RS45 0.613 10.0% -24.0% 11.4%
T60/RS40 0.596 9.9% -25.1% 11.6%
T65/RS35 0.580 9.8% -26.3% 11.8%
Meilleure allocation: T55/RS45 (Sharpe=0.613)# Distribution des regimes
regime_counts = regimes.value_counts()
regime_pct = regimes.value_counts(normalize=True) * 100
print("Distribution des regimes (2010-2026):")
print(f"{'Regime':<12} {'Jours':>10} {'Pourcentage':>12}")
print("-" * 35)
for regime in ['bull', 'bear', 'sideways']:
count = regime_counts.get(regime, 0)
pct = regime_pct.get(regime, 0)
print(f"{regime:<12} {count:>10} {pct:>11.1f}%")
print(f"\nInterpretation: Le passe decompose le temps en regimes distincts.\n"
"La strategy RegimeSwitching adapte son comportement a chaque regime.")Distribution des regimes (2010-2026):
Regime Jours Pourcentage
-----------------------------------
bull 2156 58.0%
bear 150 4.0%
sideways 1212 32.6%
Interpretation: Le passe decompose le temps en regimes distincts.
La strategy RegimeSwitching adapte son comportement a chaque regime.fig, axes = plt.subplots(1, 2, figsize=(16, 5))
# Gauche: Comparaison des allocations
ax = axes[0]
for name, r in results.items():
ax.plot(r['cum'].values, label=f"{name} (S={r['sharpe']:.2f})", linewidth=1.5)
ax.set_title('Allocation Trend/RegimeSwitching', fontsize=12, fontweight='bold')
ax.set_ylabel('Valeur du portefeuille')
ax.legend(fontsize=9)
ax.grid(True, alpha=0.3)
# Droite: SPY vs Composite
ax = axes[1]
# Buy and hold SPY
spy_values = closes['SPY'].iloc[252:] / closes['SPY'].iloc[252]
ax.plot(spy_values.values, label=f"SPY B&H", linewidth=1.5, alpha=0.7)
ax.plot(best_alloc[1]['cum'].values,
label=f"Composite {best_alloc[0]} (S={best_alloc[1]['sharpe']:.2f})", linewidth=1.5)
ax.set_title(f'Composite vs SPY', fontsize=12, fontweight='bold')
ax.set_ylabel('Valeur du portefeuille (normalisee)')
ax.legend(fontsize=9)
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('composite_momentum_regime.png', dpi=150, bbox_inches='tight')
plt.show()
print("Graphique sauvegarde.")
Graphique sauvegarde.| Metrique | SectorMomentum | RegimeSwitching | Composite optimal |
|---|---|---|---|
| Sharpe | (a remplir) | (a remplir) | (a remplir) |
| CAGR | (a remplir) | (a remplir) | (a remplir) |
| Max DD | (a remplir) | (a remplir) | (a remplir) |
Allocation: [a remplir]