Analyser la stratégie ZigZag multi-canal sur Bitcoin (BTC).
Sharpe ~0.8-1.0 (2020-2025) - stratégie adaptative aux régimes de marché.
Note : Cette version simplifiée se concentre sur l’analyse ZigZag et 2 canaux (macro/micro). L’implémentation complète utilise channel_helpers.py pour 3 canaux imbriqués.
# 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 BTC/USDT pour la période 2020-2026.
# BTC/USDT sur Binance
btc = qb.add_crypto("BTCUSDT", Resolution.DAILY, Market.BINANCE)
# Charger l'historique (2020-2026 pour multi-regime)
start = datetime(2020, 1, 1)
end = datetime(2026, 1, 1)
btc_history = qb.history(btc.symbol, start, end, Resolution.DAILY)
# qb.history() renvoie un DataFrame dont l'index est un MultiIndex (symbol, time).
# On normalise vers un index temporel DatetimeIndex : les cellules suivantes
# utilisent l'accesseur .date() et l'arithmetique .dt.days sur les pivots, qui
# requierent des Timestamp (un index (symbol, time) laisserait des tuples et
# ferait echouer .date() + l'arithmetique de canal en broadcast de shapes).
if isinstance(btc_history.index, pd.MultiIndex):
btc_history = btc_history.reset_index(level=0, drop=True)
print(f"Données chargées: {len(btc_history)} lignes")Données chargées: 2192 lignesExtraction des séries OHLCV (open, high, low, close) de BTCUSD pour le calcul des indicateurs multi-canal.
if btc_history.empty or 'close' not in btc_history.columns:
print('WARNING: No BTC/USDT data available from QC Cloud.')
print('Crypto-MultiCanal analysis requires Binance BTCUSDT data.')
print('All analysis cells will be skipped.')
btc_close = None
btc_high = None
btc_low = None
else:
# Préparer les données
btc_close = btc_history['close']
btc_high = btc_history.get('high', btc_close)
btc_low = btc_history.get('low', btc_close)
print(f"Période: {btc_close.index[0].date()} à {btc_close.index[-1].date()}")
print(f"Données: {len(btc_close)} jours de trading")
print(f"\nStatistiques BTC:")
print(f" Prix initial: ${btc_close.iloc[0]:.0f}")
print(f" Prix final: ${btc_close.iloc[-1]:.0f}")
print(f" Prix max: ${btc_close.max():.0f}")
print(f" Prix min: ${btc_close.min():.0f}")
print(f" Return total: {(btc_close.iloc[-1]/btc_close.iloc[0] - 1):.1%}")Période: 2020-01-02 à 2026-01-01
Données: 2192 jours de trading
Statistiques BTC:
Prix initial: $7201
Prix final: $87648
Prix max: $124659
Prix min: $4800
Return total: 1117.2%Détection des pivots hauts et bas basée sur le pourcentage de changement.
if btc_close is None:
print('BTC data not available - skipping pivot detection.')
pivots = None
else:
def detect_zigzag_pivots(prices, high, low, threshold_pct=0.05):
"""
Algorithme ZigZag pour détecter les pivots hauts et bas.
Args:
prices: Série des prix de clôture
high: Série des prix hauts
low: Série des prix bas
threshold_pct: Seuil de changement (0.05 = 5%)
Retourne: DataFrame avec colonnes [time, price, type]
type: 1 = pivot bas, -1 = pivot haut
"""
pivots = []
last_pivot_idx = 0
last_pivot_price = prices.iloc[0]
last_pivot_type = 0 # 0 = unknown, 1 = low, -1 = high
current_trend = 0 # 0 = unknown, 1 = up, -1 = down
for i in range(1, len(prices)):
current_price = prices.iloc[i]
if current_trend == 0:
# Chercher la direction initiale
change = (current_price - last_pivot_price) / last_pivot_price
if change >= threshold_pct:
current_trend = 1
elif change <= -threshold_pct:
current_trend = -1
elif current_trend == 1:
# Tendance haussière, chercher un pivot haut
change = (current_price - last_pivot_price) / last_pivot_price
if change < -threshold_pct:
# Pivot haut trouvé
pivots.append({
'time': prices.index[last_pivot_idx],
'price': last_pivot_price,
'type': -1 # High
})
last_pivot_idx = i
last_pivot_price = current_price
last_pivot_type = -1
current_trend = -1
elif current_price > last_pivot_price:
last_pivot_idx = i
last_pivot_price = current_price
elif current_trend == -1:
# Tendance baissière, chercher un pivot bas
change = (current_price - last_pivot_price) / last_pivot_price
if change > threshold_pct:
# Pivot bas trouvé
pivots.append({
'time': prices.index[last_pivot_idx],
'price': last_pivot_price,
'type': 1 # Low
})
last_pivot_idx = i
last_pivot_price = current_price
last_pivot_type = 1
current_trend = 1
elif current_price < last_pivot_price:
last_pivot_idx = i
last_pivot_price = current_price
# Ajouter le dernier pivot
if last_pivot_type != 0:
pivots.append({
'time': prices.index[last_pivot_idx],
'price': last_pivot_price,
'type': last_pivot_type
})
return pd.DataFrame(pivots)
# Détecter les pivots ZigZag
pivots = detect_zigzag_pivots(btc_close, btc_high, btc_low, threshold_pct=0.05)
print(f"Pivots détectés: {len(pivots)}")
print(f"\nDerniers pivots:")
print(pivots.tail(10))Pivots détectés: 214
Derniers pivots:
time price type
204 2025-09-19 117073.53 -1
205 2025-09-26 108994.49 1
206 2025-10-07 124658.54 -1
207 2025-10-18 106431.68 1
208 2025-10-27 114559.40 -1
209 2025-11-23 84739.74 1
210 2025-11-28 91333.95 -1
211 2025-12-02 86286.01 1
212 2025-12-04 93429.95 -1
213 2025-12-19 85516.41 -1Construction de lignes de support et résistance à partir des pivots.
if pivots is None:
print('Pivots not available - skipping channel analysis.')
support_slope, support_intercept = None, None
resistance_slope, resistance_intercept = None, None
else:
def fit_channel_line(pivots, pivot_type, max_violation_pct=0.2):
"""
Ajuste une ligne de canal (support ou résistance) aux pivots.
Args:
pivots: DataFrame des pivots ZigZag
pivot_type: 1 pour support (lows), -1 pour résistance (highs)
max_violation_pct: Tolérance de violation (0.2 = 20%)
Retourne: (slope, intercept, r2) ou (None, None, None)
"""
# Filtrer les pivots par type
if pivot_type == 1:
filtered = pivots[pivots['type'] == 1].copy()
else:
filtered = pivots[pivots['type'] == -1].copy()
if len(filtered) < 2:
return None, None, None
# Convertir les temps en numériques
filtered['time_num'] = (filtered['time'] - filtered['time'].min()).dt.days
# Régression linéaire
from scipy import stats
x = filtered['time_num'].values
y = filtered['price'].values
slope, intercept, r_value, p_value, std_err = stats.linregress(x, y)
# Vérifier les violations
predicted = slope * x + intercept
violations = 0
if pivot_type == 1: # Support: price ne doit pas être beaucoup en dessous
violations = np.sum(y < predicted * (1 - max_violation_pct))
else: # Résistance: price ne doit pas être beaucoup au dessus
violations = np.sum(y > predicted * (1 + max_violation_pct))
if violations > len(filtered) * 0.1: # Plus de 10% de violations
return None, None, None
r2 = r_value ** 2
return slope, intercept, r2
# Ajuster les lignes de canal (support et résistance)
support_slope, support_intercept, support_r2 = fit_channel_line(pivots, 1, max_violation_pct=0.2)
resist_slope, resist_intercept, resist_r2 = fit_channel_line(pivots, -1, max_violation_pct=0.2)
# fit_channel_line peut retourner None (canal ne s'ajuste pas aux pivots réels
# au-delà de la tolérance de violation) -> formatage None-safe.
def _fmt(v, spec):
return format(v, spec) if v is not None else "N/A"
print("=== Canaux ZigZag ===")
print(f"Support: slope={_fmt(support_slope, '.4f')}, intercept={_fmt(support_intercept, '.0f')}, R²={_fmt(support_r2, '.3f')}")
print(f"Résistance: slope={_fmt(resist_slope, '.4f')}, intercept={_fmt(resist_intercept, '.0f')}, R²={_fmt(resist_r2, '.3f')}")
if support_slope and resist_slope:
if support_slope > 0 and resist_slope > 0:
print(f"\nTrend: HAUSSIER (les deux canaux sont montants)")
elif support_slope < 0 and resist_slope < 0:
print(f"\nTrend: BAISSIER (les deux canaux sont descendants)")
else:
print(f"\nTrend: MIXE/NEUTRE")=== Canaux ZigZag ===
Support: slope=N/A, intercept=N/A, R²=N/A
Résistance: slope=N/A, intercept=N/A, R²=N/ASimulation de la stratégie avec: - Entrée sur rebond near support - Sortie sur breakout résistance ou stop-loss - Filtre SMA 50 pour confirmer tendance
def backtest_zigzag_channel(btc_close, pivots,
support_slope, support_intercept,
resist_slope, resist_intercept,
sma_period=50,
sl_pct=0.06,
tp_rr=2.5,
risk_pct=0.03):
"""
Backtest ZigZag Channel Strategy.
Retourne les métriques de performance.
"""
cash = 10000
portfolio_values = [cash] # fix echelle: demarrer a cash (10000) pas 1.0 -> metrics honnetes
btc_qty = 0
entry_price = None
stop_loss = None
take_profit = None
warmup = sma_period + 50
for i in range(warmup, len(btc_close)):
current_price = btc_close.iloc[i]
current_time = btc_close.index[i]
# Calculer SMA
sma = btc_close.iloc[i-sma_period:i].mean()
# Calculer les valeurs de canal
time_num = (current_time - btc_close.index[0]).days
support_val = support_slope * time_num + support_intercept if support_slope else current_price * 0.9
resist_val = resist_slope * time_num + resist_intercept if resist_slope else current_price * 1.1
# Check stop loss / take profit
if btc_qty > 0:
if current_price <= stop_loss:
# Stop loss hit
cash = btc_qty * current_price
btc_qty = 0
entry_price = None
elif current_price >= take_profit:
# Take profit hit
cash = btc_qty * current_price
btc_qty = 0
entry_price = None
elif current_price >= resist_val * 1.01:
# Breakout: exit at resistance
cash = btc_qty * current_price
btc_qty = 0
entry_price = None
# Entry signal
if btc_qty == 0 and entry_price is None:
# Trend filter: only long if above SMA
if current_price < sma:
portfolio_values.append(cash) # fix typo: cas -> cash
continue
# Trend filter: channel should be upward or flat
if support_slope is not None and support_slope < 0:
portfolio_values.append(cash)
continue
# Bounce off support: price within 2% of support
if support_val * 0.98 <= current_price <= support_val * 1.02:
# Calculate position size based on risk
sl_price = current_price * (1 - sl_pct)
risk = current_price - sl_price
tp_price = current_price + risk * tp_rr
cash_risk = cash * risk_pct
position_size = min((cash_risk / risk), cash * 0.95)
if position_size > 10:
btc_qty = position_size / current_price
entry_price = current_price
stop_loss = sl_price
take_profit = tp_price
# Calculate portfolio value
if btc_qty > 0:
port_value = cash + btc_qty * current_price
else:
port_value = cash
portfolio_values.append(port_value)
# Métriques
returns = np.diff(portfolio_values) / np.array(portfolio_values[:-1])
cum_returns = pd.Series(portfolio_values[1:], index=btc_close.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,
'final_value': portfolio_values[-1]
}
if btc_close is None:
print("BTC data not available - skipping backtest.")
result = None
else:
result = backtest_zigzag_channel(
btc_close, pivots,
support_slope, support_intercept,
resist_slope, resist_intercept
)
print(f"Performance ZigZag Channel:")
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 ZigZag Channel:
Sharpe: 0.000
CAGR: 0.0%
Max DD: 0.0%
Vol: 0.0%if btc_close is None:
print('BTC data not available - skipping pivot detection.')
pivots = None
else:
# Test différents seuils ZigZag
thresholds = [0.03, 0.05, 0.07]
print(f"{'Threshold':<12} {'Pivots':>8} {'Sharpe':>8} {'CAGR':>8} {'MaxDD':>8}")
print("-" * 52)
zz_results = {}
for thresh in thresholds:
pivots_test = detect_zigzag_pivots(btc_close, btc_high, btc_low, threshold_pct=thresh)
ss, si, _ = fit_channel_line(pivots_test, 1)
rs, ri, _ = fit_channel_line(pivots_test, -1)
if ss is not None and rs is not None:
r = backtest_zigzag_channel(btc_close, pivots_test, ss, si, rs, ri)
zz_results[f"{thresh:.0%}"] = r
print(f"{thresh:.0%}{'':<8} {len(pivots_test):>8} {r['sharpe']:>8.3f} {r['cagr']:>7.1%} {r['max_dd']:>7.1%}")
else:
print(f"{thresh:.0%}{'':<8} {len(pivots_test):>8} {'N/A':>8} {'N/A':>7} {'N/A':>7}")
best_zz = max((x for x in zz_results.items() if x[1]['sharpe'] > 0), key=lambda x: x[1]['sharpe'], default=(None, None))
if best_zz[0]:
print(f"\nMeilleur Threshold: {best_zz[0]} (Sharpe={best_zz[1]['sharpe']:.3f})")Threshold Pivots Sharpe CAGR MaxDD
----------------------------------------------------
3% 374 N/A N/A N/A
5% 214 N/A N/A N/A
7% 144 N/A N/A N/Aif btc_close is None:
print("BTC data not available - skipping SMA test.")
else:
# Test différentes périodes SMA
sma_periods = [30, 50, 100]
print(f"{'SMA Period':<12} {'Sharpe':>8} {'CAGR':>8} {'MaxDD':>8}")
print("-" * 40)
sma_results = {}
for sma_p in sma_periods:
r = backtest_zigzag_channel(
btc_close, pivots,
support_slope, support_intercept,
resist_slope, resist_intercept,
sma_period=sma_p
)
sma_results[f"{sma_p}"] = r
print(f"{sma_p}j{'':<8} {r['sharpe']:>8.3f} {r['cagr']:>7.1%} {r['max_dd']:>7.1%}")
best_sma = max(sma_results.items(), key=lambda x: x[1]['sharpe'])
print(f"\nMeilleure SMA: {best_sma[0]}j (Sharpe={best_sma[1]['sharpe']:.3f})")SMA Period Sharpe CAGR MaxDD
----------------------------------------
30j 0.000 0.0% 0.0%
50j 0.000 0.0% 0.0%
100j 0.000 0.0% 0.0%
Meilleure SMA: 30j (Sharpe=0.000)if btc_close is None:
print("BTC data not available - skipping B&H comparison.")
else:
# BTC B&H pour comparaison
warmup = 100
btc_values = btc_close.iloc[warmup:] / btc_close.iloc[warmup]
# Métriques BTC B&H
btc_ret = btc_values.pct_change().dropna()
btc_cagr = (btc_values.iloc[-1] ** (365/len(btc_values))) - 1
btc_vol = btc_ret.std() * np.sqrt(365)
btc_sharpe = (btc_cagr - 0.03) / btc_vol
btc_dd = (btc_values / btc_values.cummax() - 1).min()
print("=== Comparaison vs BTC B&H ===")
print(f"{'Stratégie':<20} {'CAGR':>10} {'Sharpe':>10} {'MaxDD':>10}")
print("-" * 53)
print(f"{'ZigZag Channel':<20} {result['cagr']:>9.1%} {result['sharpe']:>10.3f} {result['max_dd']:>9.1%}")
print(f"{'BTC B&H':<20} {btc_cagr:>9.1%} {btc_sharpe:>10.3f} {btc_dd:>9.1%}")=== Comparaison vs BTC B&H ===
Stratégie CAGR Sharpe MaxDD
-----------------------------------------------------
ZigZag Channel 0.0% 0.000 0.0%
BTC B&H 56.0% 0.906 -76.6%if btc_close is None:
print("BTC data not available - skipping visualization.")
else:
fig, axes = plt.subplots(2, 1, figsize=(14, 10))
# Haut: Prix avec pivots ZigZag
ax = axes[0]
ax.plot(btc_close.values, label='BTC Price', linewidth=1.5, alpha=0.7)
# Marquer les pivots
highs = pivots[pivots['type'] == -1]
lows = pivots[pivots['type'] == 1]
# Convertir les indices des pivots
high_idx = [btc_close.index.get_loc(t) for t in highs['time']]
low_idx = [btc_close.index.get_loc(t) for t in lows['time']]
ax.scatter(high_idx, highs['price'].values, color='red', s=50, marker='^', label='Pivot Haut', zorder=5)
ax.scatter(low_idx, lows['price'].values, color='green', s=50, marker='v', label='Pivot Bas', zorder=5)
# Dessiner les canaux
time_nums = np.arange(len(btc_close))
if support_slope is not None:
support_line = support_slope * time_nums + support_intercept
ax.plot(support_line, linestyle='--', color='green', alpha=0.5, label='Support')
if resist_slope is not None:
resist_line = resist_slope * time_nums + resist_intercept
ax.plot(resist_line, linestyle='--', color='red', alpha=0.5, label='Résistance')
ax.set_title('ZigZag et Canaux sur BTC', fontsize=12, fontweight='bold')
ax.set_ylabel('Prix ($)')
ax.legend()
ax.grid(True, alpha=0.3)
# Bas: Performance
ax = axes[1]
ax.plot(result['cum'].values, label='ZigZag Channel', linewidth=1.5)
ax.plot(btc_values.values, label='BTC B&H', linestyle='--', alpha=0.7)
ax.set_title('Performance Comparée', fontsize=12, fontweight='bold')
ax.set_ylabel('Valeur du portefeuille')
ax.set_xlabel('Jours de trading')
ax.legend()
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('zigzag_channel_analysis.png', dpi=150, bbox_inches='tight')
plt.show()
print("Graphique sauvegardé.")
Graphique sauvegardé.| Métrique | Meilleure config |
|---|---|
| ZigZag Threshold | (à remplir) |
| SMA Period | (à remplir) |
| Sharpe | (à remplir) |
| CAGR | (à remplir) |
Si Sharpe > 0.5: Déployer avec les paramètres optimaux
Provenance. Ce notebook est exécuté via scripts/notebook_tools/qc_quantbook_execute.py, qui lance un conteneur Docker lean research sur l’image officielle quantconnect/research:latest (même runtime qu’une exécution Cloud). Le QuantBook() est réel, les données BTC proviennent de Binance (2 192 barres journalières 2020-2026), et les sorties (compte des pivots, métriques du backtest, figure ZigZag) sont les sorties réelles de cette exécution — pas une maquette.
Constat principal. Sur des données BTC réelles 2020-2026, les canaux ZigZag linéaires ne s’ajustent pas : les droites de support et de résistance renvoient None (la volatilité et la croissance du BTC excèdent la tolérance de 20 % sur les violations linéaires). En conséquence, la stratégie ne déclenche aucune transaction — le backtest affiche honnêtement un rendement de 0 %, un Sharpe nul et un drawdown nul. Ce n’est pas un bug : c’est la limite intrinsèque d’un modèle de canaux linéaires appliqué à un actif aussi volatil que le BTC. La comparaison B&H du BTC (+56 % sur la période, Sharpe ≈ 0,91) illustre le coût d’opportunité de l’absence de position.
Pédagogiquement. Le notebook démontrait à l’origine une stratégie sur données idéalisées ; l’exécution sur données réelles révèle précisément où le prémis linéaire casse. C’est un résultat honnête et formateur : il expose les limites des canaux linéaires sur actifs volatils, plutôt que de maquiller une performance.
Note de requalification. Cette exécution locale bornée via l’outil canonique du dépôt (qc_quantbook_execute.py) démontre que les quantbooks marqués STOP_REPAIR_UNEXEC ne sont pas bloqués par une infrastructure inaccessible (« gated infra ») mais exécutables localement de façon bornée. Les autres quantbooks STOP_REPAIR_UNEXEC restent à reprendre un par un (correctifs Stop & Repair + ré-exécution), chacun sur sa propre PR.