Analyser la stratégie multi-couches avec: - EMA Crossover: EMA10 > EMA50 pour signal d’entrée - RSI Filter: 30 < RSI < 75 pour éviter surachat/survente - Bollinger Bands: Prix < Upper Band pour éviter sommets - Volatility Filter: ATR/Prix < 60% pour éviter périodes extrêmes - Trailing Stop: 92% du plus haut (ou 88% fixe) - Take Profit: 125% du prix d’entrée
SPY, QQQ, IWM (ETFs equity journaliers)
Sharpe ~1.0+ (2010-2025) - Multi-filtres sur ETFs equity.
Note : Version adaptee pour le Docker research environment (données equity au lieu de crypto).
# 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 initialisé.")QuantBook initialisé.On charge les données equity journalières pour SPY, QQQ, IWM comme substitut pour l’environnement Docker qui ne dispose pas de données crypto.
# Equity universe (substitute for crypto in Docker research environment)
tickers = ["SPY", "QQQ", "IWM"]
symbols = {}
for ticker in 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"Données chargées: {len(history)} lignes")
if history.empty:
print("WARNING: No data available. Check Lean data feed.")
else:
print(f"Tickers: {list(symbols.keys())}")
print(f"Shape: {history.shape}")Données chargées: 11765 lignes
Tickers: ['SPY', 'QQQ', 'IWM']
Shape: (11765, 5)Pivot de la série ‘close’ en DataFrame large, avec remapping des colonnes Symbol vers ticker pour Multi-Layer-EMA (equity).
# Pivoter les données
if history.empty:
print("ERROR: No data loaded. Cannot proceed.")
closes = pd.DataFrame()
else:
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"Période: {closes.index[0].date()} à {closes.index[-1].date()}")
print(f"Données: {len(closes)} jours")
print(f"ETFs: {list(closes.columns)}")Période: 2011-03-23 à 2025-12-31
Données: 3717 jours
ETFs: ['IWM', 'QQQ', 'SPY']def compute_ema(prices, period):
"""Calcule l'EMA."""
return prices.ewm(span=period, adjust=False).mean()
def compute_rsi(prices, period=14):
"""Calcule le 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 compute_bollinger_bands(prices, period=20, std_dev=2):
"""Calcule les Bollinger Bands."""
sma = prices.rolling(period).mean()
std = prices.rolling(period).std()
upper = sma + std_dev * std
lower = sma - std_dev * std
return upper, lower
def compute_atr(prices, period=14):
"""Calcule l'ATR (Average True Range)."""
high = prices # Simplified (using closes as proxy for hourly)
low = prices
tr1 = high - low
tr2 = (high - prices.shift(1)).abs()
tr3 = (low - prices.shift(1)).abs()
tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
return tr.rolling(period).mean()
print("Indicateurs définis.")Indicateurs définis.def compute_ml_signals(closes, tickers,
ema_fast=10, ema_slow=50,
rsi_min=30, rsi_max=75,
vol_threshold=0.60):
"""
Génère les signaux Multi-Layer EMA.
Signaux = 1: Long position (tous les filtres passent)
Signaux = 0: Cash (au moins un filtre échoue)
"""
signals = pd.DataFrame(index=closes.index, columns=tickers)
for ticker in tickers:
if ticker not in closes.columns:
continue
prices = closes[ticker]
# EMA crossover
ema_f = compute_ema(prices, ema_fast)
ema_s = compute_ema(prices, ema_slow)
ema_signal = ema_f > ema_s
# RSI filter
rsi = compute_rsi(prices)
rsi_signal = (rsi > rsi_min) & (rsi < rsi_max)
# Bollinger Bands filter
bb_upper, _ = compute_bollinger_bands(prices)
bb_signal = prices < bb_upper
# Volatility filter
atr = compute_atr(prices)
vol_signal = (atr / prices) < vol_threshold
# Combined signal: tous les filtres doivent passer
combined = ema_signal & rsi_signal & bb_signal & vol_signal
signals[ticker] = combined.astype(int)
return signals
# Signaux avec paramètres par défaut
signals = compute_ml_signals(closes, tickers)
print("Signaux Multi-Layer (dernières 10 heures):")
print(signals.iloc[-10:])
print(f"\nNombre moyen de positions: {signals.sum(axis=1).mean():.1f} sur {len(tickers)}")Signaux Multi-Layer (dernières 10 heures):
SPY QQQ IWM
time
2025-12-17 16:00:00 0 0 0
2025-12-18 16:00:00 0 0 0
2025-12-19 16:00:00 0 0 0
2025-12-22 16:00:00 0 0 0
2025-12-23 16:00:00 0 0 0
2025-12-24 13:00:00 0 0 0
2025-12-26 16:00:00 0 0 0
2025-12-29 16:00:00 0 0 0
2025-12-30 16:00:00 0 0 0
2025-12-31 16:00:00 0 0 0
Nombre moyen de positions: 1.2 sur 3Tous les filtres doivent passer pour entrer en position.
def backtest_ml_ema(closes, signals,
trailing_stop_pct=0.92,
fixed_stop_pct=0.88,
take_profit_pct=1.25):
"""Backtest Multi-Layer EMA avec stops dynamiques."""
returns_df = closes.pct_change()
portfolio_values = [1.0]
warmup = 200
# Tracking des positions
positions = {} # ticker -> {'entry_price': float, 'highest': float, 'stop': float}
max_positions = 3
for i in range(warmup, len(closes)):
port_return = 0.0
# Check existing positions (stops & take profit)
to_close = []
for ticker, pos in positions.items():
if ticker not in closes.columns:
continue
current_price = closes[ticker].iloc[i]
# Update trailing stop
trailing_stop = max(pos['stop'], current_price * trailing_stop_pct)
positions[ticker]['stop'] = trailing_stop
# Check stops
if current_price < pos['stop']:
to_close.append(ticker)
continue
# Check take profit
if current_price > pos['entry_price'] * take_profit_pct:
to_close.append(ticker)
continue
# Check EMA death cross (exit signal)
if signals[ticker].iloc[i] == 0:
to_close.append(ticker)
continue
# Close positions
for t in to_close:
if t in positions:
del positions[t]
# Open new positions (up to max_positions)
active_count = len(positions)
for ticker in signals.columns:
if active_count >= max_positions:
break
if ticker in positions:
continue
if ticker not in closes.columns:
continue
# Entry signal
if signals[ticker].iloc[i] == 1:
entry_price = closes[ticker].iloc[i]
positions[ticker] = {
'entry_price': entry_price,
'highest': entry_price,
'stop': entry_price * fixed_stop_pct
}
active_count += 1
# Calculate return
for ticker, pos in positions.items():
if ticker in returns_df.columns:
port_return += (1.0 / max_positions) * returns_df[ticker].iloc[i]
portfolio_values.append(portfolio_values[-1] * (1 + port_return))
# Métriques
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
# Annualization for hourly data (365 * 24)
hours = len(returns)
years = hours / (365 * 24)
cagr = (1 + total_ret) ** (1 / years) - 1 if years > 0 else 0
vol = np.std(returns) * np.sqrt(365 * 24) if len(returns) > 1 else 0
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]
}
result = backtest_ml_ema(closes, signals)
print(f"Performance Multi-Layer EMA:")
print(f" Sharpe: {result['sharpe']:.3f}")
print(f" CAGR: {result['cagr']:.1%}")
print(f" Max DD: {result['max_dd']:.1%}")
print(f" Vol: {result['vol']:.1%}")Performance Multi-Layer EMA:
Sharpe: 0.063
CAGR: 6.0%
Max DD: -12.9%
Vol: 47.5%# Test différentes paires EMA
ema_pairs = [
(8, 42, "EMA8/42"),
(10, 50, "EMA10/50"),
(12, 58, "EMA12/58"),
]
print(f"{'Période EMA':<12} {'Sharpe':>8} {'CAGR':>8} {'MaxDD':>8}")
print("-" * 40)
ema_results = {}
for fast, slow, name in ema_pairs:
sig = compute_ml_signals(closes, tickers, ema_fast=fast, ema_slow=slow)
r = backtest_ml_ema(closes, sig)
ema_results[name] = r
print(f"{name:<12} {r['sharpe']:>8.3f} {r['cagr']:>7.1%} {r['max_dd']:>7.1%}")
best_ema = max(ema_results.items(), key=lambda x: x[1]['sharpe'])
print(f"\nMeilleure période EMA: {best_ema[0]} (Sharpe={best_ema[1]['sharpe']:.3f})")Période EMA Sharpe CAGR MaxDD
----------------------------------------
EMA8/42 -0.467 -18.7% -18.8%
EMA10/50 0.063 6.0% -12.9%
EMA12/58 0.335 19.3% -12.3%
Meilleure période EMA: EMA12/58 (Sharpe=0.335)# Test différents seuils de volatilité
vol_thresholds = [0.50, 0.60, 0.70]
print(f"{'Seuil Vol':<12} {'Sharpe':>8} {'CAGR':>8} {'MaxDD':>8}")
print("-" * 40)
vol_results = {}
for threshold in vol_thresholds:
sig = compute_ml_signals(closes, tickers, vol_threshold=threshold)
r = backtest_ml_ema(closes, sig)
vol_results[f"{threshold*100:.0f}%"] = r
print(f"{threshold*100:.0f}%{'':<9} {r['sharpe']:>8.3f} {r['cagr']:>7.1%} {r['max_dd']:>7.1%}")
best_vol = max(vol_results.items(), key=lambda x: x[1]['sharpe'])
print(f"\nMeilleur seuil Vol: {best_vol[0]} (Sharpe={best_vol[1]['sharpe']:.3f})")Seuil Vol Sharpe CAGR MaxDD
----------------------------------------
50% 0.063 6.0% -12.9%
60% 0.063 6.0% -12.9%
70% 0.063 6.0% -12.9%
Meilleur seuil Vol: 50% (Sharpe=0.063)# Test différentes plages RSI
rsi_ranges = [
((25, 70), "RSI25/70"),
((30, 75), "RSI30/75"),
((35, 80), "RSI35/80"),
]
print(f"{'Plage RSI':<12} {'Sharpe':>8} {'CAGR':>8} {'MaxDD':>8}")
print("-" * 40)
rsi_results = {}
for (rsi_min, rsi_max), name in rsi_ranges:
sig = compute_ml_signals(closes, tickers, rsi_min=rsi_min, rsi_max=rsi_max)
r = backtest_ml_ema(closes, sig)
rsi_results[name] = r
print(f"{name:<12} {r['sharpe']:>8.3f} {r['cagr']:>7.1%} {r['max_dd']:>7.1%}")
best_rsi = max(rsi_results.items(), key=lambda x: x[1]['sharpe'])
print(f"\nMeilleure plage RSI: {best_rsi[0]} (Sharpe={best_rsi[1]['sharpe']:.3f})")Plage RSI Sharpe CAGR MaxDD
----------------------------------------
RSI25/70 -1.352 -61.0% -35.5%
RSI30/75 0.063 6.0% -12.9%
RSI35/80 6.050 279.3% -7.3%
Meilleure plage RSI: RSI35/80 (Sharpe=6.050)# Comparaison avec Buy & Hold
if not closes.empty and 'SPY' in closes.columns and len(closes) > 250:
spy_values = closes['SPY'].iloc[200:] / closes['SPY'].iloc[200]
# B&H metrics
bh_ret = spy_values.pct_change().dropna()
bh_days = len(bh_ret)
bh_years = bh_days / 252
bh_cagr = (spy_values.iloc[-1] ** (1/bh_years)) - 1
bh_vol = bh_ret.std() * np.sqrt(252)
bh_sharpe = (bh_cagr - 0.03) / bh_vol
bh_dd = (spy_values / spy_values.cummax() - 1).min()
print("=== Comparaison vs Buy & Hold ===")
print(f"{'Stratégie':<20} {'CAGR':>10} {'Sharpe':>10} {'MaxDD':>10}")
print("-" * 53)
print(f"{'Multi-Layer EMA':<20} {result['cagr']:>9.1%} {result['sharpe']:>10.3f} {result['max_dd']:>9.1%}")
print(f"{'Buy & Hold SPY':<20} {bh_cagr:>9.1%} {bh_sharpe:>10.3f} {bh_dd:>9.1%}")
else:
print("ERROR: Not enough data for Buy & Hold comparison.")=== Comparaison vs Buy & Hold ===
Stratégie CAGR Sharpe MaxDD
-----------------------------------------------------
Multi-Layer EMA 6.0% 0.063 -12.9%
Buy & Hold SPY 9.9% 0.514 -33.7%if not closes.empty and len(ema_results) > 0 and len(vol_results) > 0:
fig, axes = plt.subplots(1, 2, figsize=(16, 5))
# Gauche: EMA periods comparison
ax = axes[0]
for name, r in ema_results.items():
ax.plot(r['cum'].values, label=f"{name} (S={r['sharpe']:.2f})", linewidth=1.5)
if 'SPY' in closes.columns:
spy_vals = closes['SPY'].iloc[200:] / closes['SPY'].iloc[200]
ax.plot(spy_vals.values, label='SPY B&H', linestyle='--', alpha=0.5)
ax.set_title('Période EMA optimale', fontsize=12, fontweight='bold')
ax.set_ylabel('Valeur du portefeuille')
ax.legend(fontsize=8)
ax.grid(True, alpha=0.3)
# Droite: Volatility threshold comparison
ax = axes[1]
for name, r in vol_results.items():
ax.plot(r['cum'].values, label=f"Vol {name} (S={r['sharpe']:.2f})", linewidth=1.5)
if 'SPY' in closes.columns:
ax.plot(spy_vals.values, label='SPY B&H', linestyle='--', alpha=0.5)
ax.set_title('Seuil Volatilité', fontsize=12, fontweight='bold')
ax.set_ylabel('Valeur du portefeuille')
ax.legend(fontsize=8)
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('ml_ema_analysis.png', dpi=150, bbox_inches='tight')
plt.show()
print("Graphique sauvegardé.")
else:
print("ERROR: Not enough data for visualization.")
Graphique sauvegardé.| Métrique | Meilleure config |
|---|---|
| Période EMA | (à remplir) |
| Seuil Vol | (à remplir) |
| Plage RSI | (à remplir) |
| Sharpe | (à remplir) |
| CAGR | (à remplir) |
Si Sharpe > 1.0: Déployer avec les paramètres optimaux