Analyser la stratégie EMA Crossover sur un univers d’ETFs large-cap avec filtre SMA200 et trailing stop.
Sharpe ~0.8-1.0 (2010-2025) avec SMA200 filter sur ETFs.
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 (SPY, QQQ, IWM) pour la période 2010-2026 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"]
equities = {}
for ticker in tickers:
equities[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(equities.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(equities.keys())}")
print(f"Shape: {history.shape}")Données chargées: 11765 lignes
Tickers: ['SPY', 'QQQ', 'IWM']
Shape: (11765, 5)Extraction de la série de clôture et remapping des colonnes Symbol vers ticker pour EMA-Cross-Crypto (equity).
# Préparer les données
if history.empty:
print("ERROR: No data loaded. Cannot proceed.")
closes = pd.Series(dtype=float)
else:
closes_df = history['close'].unstack(level=0)
symbol_to_ticker = {str(v): k for k, v in equities.items()}
closes_df.columns = [symbol_to_ticker.get(str(c), str(c)) for c in closes_df.columns]
# Use first ticker as primary series for single-asset analysis
primary = tickers[0]
closes = closes_df[primary].dropna()
print(f"Période: {closes.index[0].date()} à {closes.index[-1].date()}")
print(f"Données: {len(closes)} jours")
print(f"\nStatistiques {primary}:")
print(f" Prix initial: ${closes.iloc[0]:.2f}")
print(f" Prix final: ${closes.iloc[-1]:.2f}")
print(f" Return total: {(closes.iloc[-1]/closes.iloc[0] - 1):.1%}")Période: 2010-01-04 à 2025-12-31
Données: 4024 jours
Statistiques SPY:
Prix initial: $90.75
Prix final: $396.33
Return total: 336.7%def compute_ema(closes, period):
"""Calcule l'EMA."""
multiplier = 2.0 / (period + 1)
ema = closes.iloc[:period].iloc[0]
ema_values = []
for i in range(len(closes)):
if i < period:
ema_values.append(np.nan)
elif i == period:
ema = closes.iloc[:period+1].mean()
ema_values.append(ema)
else:
ema = (closes.iloc[i] - ema) * multiplier + ema
ema_values.append(ema)
return pd.Series(ema_values, index=closes.index)
def compute_sma(closes, period):
"""Calcule la SMA."""
return closes.rolling(period).mean()
# Calculer les indicateurs avec paramètres par défaut
ema_fast = compute_ema(closes, 20)
ema_slow = compute_ema(closes, 50)
sma200 = compute_sma(closes, 200)
# Signaux
ema_cross_signal = ema_fast > ema_slow
sma_filter = closes > sma200
print("Indicateurs calculés (derniers 5 jours):")
print(pd.DataFrame({
'Close': closes.iloc[-5:],
'EMA20': ema_fast.iloc[-5:],
'EMA50': ema_slow.iloc[-5:],
'SMA200': sma200.iloc[-5:]
}).round(2))Indicateurs calculés (derniers 5 jours):
Close EMA20 EMA50 SMA200
time
2025-12-24 13:00:00 396.33 396.33 396.33 396.33
2025-12-26 16:00:00 396.33 396.33 396.33 396.33
2025-12-29 16:00:00 396.33 396.33 396.33 396.33
2025-12-30 16:00:00 396.33 396.33 396.33 396.33
2025-12-31 16:00:00 396.33 396.33 396.33 396.33Simulation avec: - EMA crossover pour entry/exit - SMA200 filter (bull market only) - Trailing stop 10%
def backtest_ema_cross_crypto(closes, ema_fast, ema_slow, sma200,
trailing_stop_pct=0.10,
position_size=0.80):
"""
Backtest EMA Cross Crypto avec trailing stop.
"""
portfolio_values = [1.0]
invested = False
entry_price = None
peak_price = None
warmup = 250
# Stats
trades = 0
trailing_stop_exits = 0
ema_cross_exits = 0
for i in range(warmup, len(closes)):
current_price = closes.iloc[i]
if pd.isna(ema_fast.iloc[i]) or pd.isna(ema_slow.iloc[i]) or pd.isna(sma200.iloc[i]):
portfolio_values.append(portfolio_values[-1])
continue
port_return = 0.0
if invested and entry_price is not None:
# Update trailing stop peak
if peak_price is None or current_price > peak_price:
peak_price = current_price
# Check trailing stop
drawdown_from_peak = (current_price - peak_price) / peak_price if peak_price > 0 else 0
if drawdown_from_peak <= -trailing_stop_pct:
invested = False
entry_price = None
peak_price = None
trailing_stop_exits += 1
port_return = drawdown_from_peak
# Check EMA cross exit
elif ema_fast.iloc[i] < ema_slow.iloc[i]:
invested = False
entry_price = None
peak_price = None
ema_cross_exits += 1
pnl_pct = (current_price - entry_price) / entry_price if entry_price else 0
port_return = pnl_pct
else:
# Hold
port_return = (current_price - closes.iloc[i-1]) / closes.iloc[i-1]
# Entry signal
elif not invested:
# Double confirmation: EMA cross + SMA200 filter
if ema_fast.iloc[i] > ema_slow.iloc[i] and current_price > sma200.iloc[i]:
invested = True
entry_price = current_price
peak_price = current_price
trades += 1
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
years = len(returns) / 365
cagr = (1 + total_ret) ** (1 / years) - 1 if years > 0 else 0
vol = np.std(returns) * np.sqrt(365) 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,
'trades': trades,
'trailing_stop_exits': trailing_stop_exits,
'ema_cross_exits': ema_cross_exits
}
print("Fonction de backtest définie.")Fonction de backtest définie.# Test différentes paires EMA
ema_pairs = [
(15, 45, "EMA15/45"),
(20, 50, "EMA20/50"),
(25, 55, "EMA25/55"),
]
print(f"{'Période EMA':<12} {'Sharpe':>8} {'CAGR':>8} {'MaxDD':>8} {'Trades':>8}")
print("-" * 52)
ema_results = {}
for fast, slow, name in ema_pairs:
emaf = compute_ema(closes, fast)
emas = compute_ema(closes, slow)
sma200 = compute_sma(closes, 200)
r = backtest_ema_cross_crypto(closes, emaf, emas, sma200)
ema_results[name] = r
print(f"{name:<12} {r['sharpe']:>8.3f} {r['cagr']:>7.1%} {r['max_dd']:>7.1%} {r['trades']:>8}")
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 Trades
----------------------------------------------------
EMA15/45 0.329 6.9% -20.3% 27
EMA20/50 0.377 7.6% -22.1% 18
EMA25/55 0.436 8.4% -19.0% 17
Meilleure période EMA: EMA25/55 (Sharpe=0.436)# Test différents trailing stops
ts_values = [0.05, 0.10, 0.15]
ema_fast = compute_ema(closes, 20)
ema_slow = compute_ema(closes, 50)
sma200 = compute_sma(closes, 200)
print(f"{'Trailing Stop':<14} {'Sharpe':>8} {'CAGR':>8} {'MaxDD':>8} {'TS Exits':>10}")
print("-" * 58)
ts_results = {}
for ts in ts_values:
r = backtest_ema_cross_crypto(closes, ema_fast, ema_slow, sma200, trailing_stop_pct=ts)
ts_results[f"{ts*100:.0f}%"] = r
print(f"{ts*100:.0f}%{'':<12} {r['sharpe']:>8.3f} {r['cagr']:>7.1%} {r['max_dd']:>7.1%} {r['trailing_stop_exits']:>10}")
best_ts = max(ts_results.items(), key=lambda x: x[1]['sharpe'])
print(f"\nMeilleur Trailing Stop: {best_ts[0]} (Sharpe={best_ts[1]['sharpe']:.3f})")Trailing Stop Sharpe CAGR MaxDD TS Exits
----------------------------------------------------------
5% -0.285 -0.9% -39.5% 20
10% 0.377 7.6% -22.1% 2
15% 0.562 9.3% -22.1% 0
Meilleur Trailing Stop: 15% (Sharpe=0.562)# Test différentes position sizes
ps_values = [0.60, 0.80, 1.00]
print(f"{'Position Size':<14} {'Sharpe':>8} {'CAGR':>8} {'MaxDD':>8}")
print("-" * 46)
ps_results = {}
for ps in ps_values:
r = backtest_ema_cross_crypto(closes, ema_fast, ema_slow, sma200, position_size=ps)
ps_results[f"{ps*100:.0f}%"] = r
print(f"{ps*100:.0f}%{'':<12} {r['sharpe']:>8.3f} {r['cagr']:>7.1%} {r['max_dd']:>7.1%}")
best_ps = max(ps_results.items(), key=lambda x: x[1]['sharpe'])
print(f"\nMeilleure Position Size: {best_ps[0]} (Sharpe={best_ps[1]['sharpe']:.3f})")Position Size Sharpe CAGR MaxDD
----------------------------------------------
60% 0.377 7.6% -22.1%
80% 0.377 7.6% -22.1%
100% 0.377 7.6% -22.1%
Meilleure Position Size: 60% (Sharpe=0.377)# Comparer avec et sans SMA200 filter
def backtest_ema_cross_no_sma(closes, ema_fast, ema_slow, trailing_stop_pct=0.10):
"""Backtest sans filtre SMA200."""
portfolio_values = [1.0]
invested = False
entry_price = None
peak_price = None
warmup = 250
for i in range(warmup, len(closes)):
current_price = closes.iloc[i]
if pd.isna(ema_fast.iloc[i]) or pd.isna(ema_slow.iloc[i]):
portfolio_values.append(portfolio_values[-1])
continue
port_return = 0.0
if invested and entry_price is not None:
if peak_price is None or current_price > peak_price:
peak_price = current_price
dd = (current_price - peak_price) / peak_price if peak_price > 0 else 0
if dd <= -trailing_stop_pct:
invested, entry_price, peak_price = False, None, None
port_return = dd
elif ema_fast.iloc[i] < ema_slow.iloc[i]:
invested, entry_price, peak_price = False, None, None
port_return = (current_price - entry_price) / entry_price if entry_price else 0
else:
port_return = (current_price - closes.iloc[i-1]) / closes.iloc[i-1]
elif not invested and ema_fast.iloc[i] > ema_slow.iloc[i]:
invested, entry_price, peak_price = True, current_price, current_price
portfolio_values.append(portfolio_values[-1] * (1 + port_return))
returns = np.diff(portfolio_values) / np.array(portfolio_values[:-1])
cum_returns = pd.Series(portfolio_values[1:], index=closes.index[warmup:])
cagr = (portfolio_values[-1] / portfolio_values[0]) ** (1 / (len(returns)/252)) - 1
vol = np.std(returns) * np.sqrt(252)
sharpe = (cagr - 0.03) / vol if vol > 0.001 else 0
max_dd = (cum_returns / cum_returns.cummax() - 1).min()
return {'sharpe': sharpe, 'cagr': cagr, 'max_dd': max_dd, 'cum': cum_returns}
if len(closes) > 300:
# Avec et sans SMA200
r_with_sma = backtest_ema_cross_crypto(closes, ema_fast, ema_slow, sma200)
r_without_sma = backtest_ema_cross_no_sma(closes, ema_fast, ema_slow)
print("=== Impact du filtre SMA200 ===")
print(f"{'Version':<20} {'Sharpe':>10} {'CAGR':>10} {'MaxDD':>10}")
print("-" * 53)
print(f"{'Avec SMA200':<20} {r_with_sma['sharpe']:>10.3f} {r_with_sma['cagr']:>9.1%} {r_with_sma['max_dd']:>9.1%}")
print(f"{'Sans SMA200':<20} {r_without_sma['sharpe']:>10.3f} {r_without_sma['cagr']:>9.1%} {r_without_sma['max_dd']:>9.1%}")
print(f"\nAmélioration Sharpe: {(r_with_sma['sharpe'] - r_without_sma['sharpe']):.3f}")
print(f"Réduction MaxDD: {(r_with_sma['max_dd'] - r_without_sma['max_dd']):.1%}")
else:
print("ERROR: Not enough data for SMA200 comparison.")=== Impact du filtre SMA200 ===
Version Sharpe CAGR MaxDD
-----------------------------------------------------
Avec SMA200 0.377 7.6% -22.1%
Sans SMA200 0.274 5.8% -22.0%
Amélioration Sharpe: 0.104
Réduction MaxDD: -0.1%if len(closes) > 300:
# EMA Cross avec paramètres optimaux
ema_result = backtest_ema_cross_crypto(closes, ema_fast, ema_slow, sma200)
# Equity B&H
bh_values = closes.iloc[250:] / closes.iloc[250]
# Métriques B&H
bh_ret = bh_values.pct_change().dropna()
bh_cagr = (bh_values.iloc[-1] ** (252/len(bh_values))) - 1
bh_vol = bh_ret.std() * np.sqrt(252)
bh_sharpe = (bh_cagr - 0.03) / bh_vol
bh_dd = (bh_values / bh_values.cummax() - 1).min()
print("=== Comparaison vs Buy & Hold ===")
print(f"{'Stratégie':<20} {'CAGR':>10} {'Sharpe':>10} {'MaxDD':>10}")
print("-" * 53)
print(f"{'EMA Cross Equity':<20} {ema_result['cagr']:>9.1%} {ema_result['sharpe']:>10.3f} {ema_result['max_dd']:>9.1%}")
print(f"{'Buy & Hold':<20} {bh_cagr:>9.1%} {bh_sharpe:>10.3f} {bh_dd:>9.1%}")
print(f"\n=== Statistiques Trading ===")
print(f"Trades totaux: {ema_result['trades']}")
print(f"Exits Trailing Stop: {ema_result['trailing_stop_exits']}")
print(f"Exits EMA Cross: {ema_result['ema_cross_exits']}")
else:
print("ERROR: Not enough data for Buy & Hold comparison.")=== Comparaison vs Buy & Hold ===
Stratégie CAGR Sharpe MaxDD
-----------------------------------------------------
EMA Cross Equity 7.6% 0.377 -22.1%
Buy & Hold 9.4% 0.453 -33.7%
=== Statistiques Trading ===
Trades totaux: 18
Exits Trailing Stop: 2
Exits EMA Cross: 15if len(closes) > 300 and len(ema_results) > 0 and len(ts_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)
ax.plot(bh_values.values, label='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: Trailing stop comparison
ax = axes[1]
for name, r in ts_results.items():
ax.plot(r['cum'].values, label=f"TS {name} (S={r['sharpe']:.2f})", linewidth=1.5)
ax.plot(bh_values.values, label='B&H', linestyle='--', alpha=0.5)
ax.set_title('Trailing Stop', 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('ema_cross_equity_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) |
| Trailing Stop | (à remplir) |
| Position Size | (à remplir) |
| Sharpe | (à remplir) |
| CAGR | (à remplir) |
Si Sharpe > 0.8: Déployer avec les paramètres optimaux