#!/usr/bin/env python3 """ Backtest Stratégie Hybride sur BNB/USD Test de la stratégie hybride optimisée sur BNB (potentiellement plus facile pour scalping) """ import pandas as pd import numpy as np import requests import json import os from typing import Dict, List from datetime import datetime from hybrid_strategy import HybridStrategy from v10_range_strategy import V10RangeStrategy class HybridBacktest: """Moteur de backtest pour stratégie hybride""" def __init__(self, df: pd.DataFrame, initial_capital: float = 500.0): self.df = df self.initial_capital = initial_capital def _calculate_pnl(self, position: Dict, current_price: float) -> float: """Calcule PnL% avec leverage""" entry = position['entry_price'] leverage = position['leverage'] if position['side'] == 'LONG': price_change_pct = ((current_price - entry) / entry) * 100 else: # SHORT price_change_pct = ((entry - current_price) / entry) * 100 return price_change_pct * leverage def run(self, strategy_name: str, strategy) -> Dict: """ Execute backtest Args: strategy_name: Nom de la stratégie strategy: Instance de stratégie (HybridStrategy ou V10RangeStrategy) Returns: Résultats du backtest """ capital = self.initial_capital trades = [] open_position = None print(f"\n🔄 Backtest {strategy_name}...") # Parcourir données for i in range(100, len(self.df)): window = self.df.iloc[:i+1].copy() # 1. Gérer position ouverte if open_position: current_price = window['close'].iloc[-1] pnl_pct = self._calculate_pnl(open_position, current_price) # Check SL/TP hit_sl = ( (open_position['side'] == 'LONG' and current_price <= open_position['stop_loss']) or (open_position['side'] == 'SHORT' and current_price >= open_position['stop_loss']) ) hit_tp = ( (open_position['side'] == 'LONG' and current_price >= open_position['take_profit']) or (open_position['side'] == 'SHORT' and current_price <= open_position['take_profit']) ) if hit_sl or hit_tp: # Fermer position exit_reason = 'TP' if hit_tp else 'SL' exit_price = open_position['take_profit'] if hit_tp else open_position['stop_loss'] # Position size: 20% du capital (gestion risque) position_size = capital * 0.20 pnl_usd = position_size * (pnl_pct / 100) capital += pnl_usd trades.append({ 'entry_time': open_position['entry_time'], 'exit_time': window['timestamp'].iloc[-1], 'side': open_position['side'], 'entry_price': open_position['entry_price'], 'exit_price': exit_price, 'leverage': open_position['leverage'], 'pnl_pct': pnl_pct, 'pnl_usd': pnl_usd, 'exit_reason': exit_reason, 'strategy_used': open_position.get('strategy_used', 'UNKNOWN'), 'regime': open_position.get('regime', 'UNKNOWN') }) open_position = None # 2. Chercher nouveau signal si pas de position if not open_position: signal = strategy.analyze('BNB', window) if signal: # Convertir signal selon type de stratégie if isinstance(strategy, HybridStrategy): # HybridSignal (dataclass) open_position = { 'entry_time': window['timestamp'].iloc[-1], 'side': signal.side, 'entry_price': signal.entry_price, 'stop_loss': signal.stop_loss, 'take_profit': signal.take_profit, 'leverage': signal.leverage, 'strategy_used': signal.strategy_used, 'regime': signal.market_regime } else: # V10RangeStrategy (dict) open_position = { 'entry_time': window['timestamp'].iloc[-1], 'side': signal['side'], 'entry_price': signal['entry_price'], 'stop_loss': signal['stop_loss'], 'take_profit': signal['take_profit'], 'leverage': signal.get('leverage', 250), 'strategy_used': 'RANGE', 'regime': 'RANGE' } # Résultats finaux final_capital = capital if len(trades) == 0: print(" ⚠️ Aucun trade généré") return { 'strategy': strategy_name, 'total_trades': 0, 'winning_trades': 0, 'losing_trades': 0, 'win_rate': 0, 'profit_factor': 0, 'total_pnl_usd': 0, 'total_pnl_pct': 0, 'final_capital': final_capital, 'avg_pnl_per_trade': 0, 'max_drawdown': 0, 'trades_by_strategy': {}, 'trades_by_regime': {}, 'trades': [] } winning_trades = [t for t in trades if t['pnl_pct'] > 0] losing_trades = [t for t in trades if t['pnl_pct'] <= 0] total_win = sum(t['pnl_usd'] for t in winning_trades) if winning_trades else 0 total_loss = abs(sum(t['pnl_usd'] for t in losing_trades)) if losing_trades else 0 profit_factor = (total_win / total_loss) if total_loss > 0 else (float('inf') if total_win > 0 else 0) # PnL total total_pnl = final_capital - self.initial_capital total_pnl_pct = (total_pnl / self.initial_capital) * 100 # Drawdown capital_curve = [self.initial_capital] running_capital = self.initial_capital for t in trades: running_capital += t['pnl_usd'] capital_curve.append(running_capital) peak = capital_curve[0] max_dd = 0 for cap in capital_curve: if cap > peak: peak = cap dd = ((peak - cap) / peak) * 100 if dd > max_dd: max_dd = dd # Stats par stratégie (pour hybrid) trades_by_strategy = {} trades_by_regime = {} if isinstance(strategy, HybridStrategy): for t in trades: strat = t.get('strategy_used', 'UNKNOWN') regime = t.get('regime', 'UNKNOWN') if strat not in trades_by_strategy: trades_by_strategy[strat] = 0 trades_by_strategy[strat] += 1 if regime not in trades_by_regime: trades_by_regime[regime] = 0 trades_by_regime[regime] += 1 return { 'strategy': strategy_name, 'total_trades': len(trades), 'winning_trades': len(winning_trades), 'losing_trades': len(losing_trades), 'win_rate': (len(winning_trades) / len(trades)) * 100, 'profit_factor': profit_factor, 'total_pnl_usd': total_pnl, 'total_pnl_pct': total_pnl_pct, 'final_capital': final_capital, 'avg_pnl_per_trade': total_pnl / len(trades), 'max_drawdown': max_dd, 'trades_by_strategy': trades_by_strategy, 'trades_by_regime': trades_by_regime, 'trades': trades } def get_bnb_historical_data(days: int = 7, interval: str = '5m') -> pd.DataFrame: """Récupère données historiques BNB/USDT""" print(f"📊 Récupération données BNB ({days} jours, {interval})...") # Calcul nombre de bougies selon interval if interval == '1m': limit = min(1000, days * 1440) elif interval == '5m': limit = min(1000, days * 288) elif interval == '15m': limit = min(1000, days * 96) else: limit = 1000 url = "https://api.binance.com/api/v3/klines" params = { 'symbol': 'BNBUSDT', 'interval': interval, 'limit': limit } response = requests.get(url, params=params, timeout=10) response.raise_for_status() data = response.json() df = pd.DataFrame(data, columns=[ 'timestamp', 'open', 'high', 'low', 'close', 'volume', 'close_time', 'quote_volume', 'trades', 'taker_buy_base', 'taker_buy_quote', 'ignore' ]) # Convertir types df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') for col in ['open', 'high', 'low', 'close', 'volume']: df[col] = df[col].astype(float) print(f" ✅ {len(df)} bougies chargées") print(f" 📅 {df['timestamp'].iloc[0]} → {df['timestamp'].iloc[-1]}") print(f" 💰 Prix: ${df['close'].iloc[0]:.2f} → ${df['close'].iloc[-1]:.2f}") return df def print_comparison(results_v10: Dict, results_hybrid: Dict): """Affiche comparaison visuelle des résultats""" print("\n" + "=" * 90) print("📊 COMPARAISON V10 RANGE vs HYBRID") print("=" * 90) metrics = [ ('Total Trades', 'total_trades', ''), ('Trades gagnants', 'winning_trades', ''), ('Win Rate', 'win_rate', '%'), ('Profit Factor', 'profit_factor', ''), ('PnL Total', 'total_pnl_pct', '%'), ('PnL USD', 'total_pnl_usd', '$'), ('Capital Final', 'final_capital', '$'), ('PnL moyen/trade', 'avg_pnl_per_trade', '$'), ('Max Drawdown', 'max_drawdown', '%'), ] print(f"\n{'Métrique':<30} | {'V10 Range':<20} | {'Hybrid':<20} | {'Diff':<10} | Gagnant") print("-" * 90) winner_count = 0 for label, key, unit in metrics: v10_val = results_v10[key] hybrid_val = results_hybrid[key] # Inversion pour drawdown (plus petit = mieux) if key == 'max_drawdown': diff = v10_val - hybrid_val # V10 moins Hybrid (négatif = Hybrid meilleur) else: diff = hybrid_val - v10_val # Déterminer gagnant if abs(diff) < 0.01: winner = "=" else: winner = "🟢 Hybrid" if diff > 0 else ("🔴 V10" if diff < 0 else "=") if diff > 0: winner_count += 1 v10_str = f"{v10_val:.2f}{unit}" if unit else f"{v10_val:.2f}" hybrid_str = f"{hybrid_val:.2f}{unit}" if unit else f"{hybrid_val:.2f}" diff_str = f"{diff:+.2f}{unit}" if unit else f"{diff:+.2f}" print(f"{label:<30} | {v10_str:<20} | {hybrid_str:<20} | {diff_str:<10} {winner}") print("-" * 90) print(f"\n🏆 GAGNANT: {'HYBRID' if winner_count > len(metrics)/2 else 'V10 RANGE'} ({winner_count}/{len(metrics)} métriques)") # Stats hybride if results_hybrid['trades_by_strategy']: print("\n📊 RÉPARTITION TRADES HYBRID:") for strat, count in results_hybrid['trades_by_strategy'].items(): pct = (count / results_hybrid['total_trades']) * 100 print(f" {strat}: {count} trades ({pct:.1f}%)") if results_hybrid['trades_by_regime']: print("\n🌍 RÉGIMES DÉTECTÉS:") for regime, count in results_hybrid['trades_by_regime'].items(): pct = (count / results_hybrid['total_trades']) * 100 print(f" {regime}: {count} trades ({pct:.1f}%)") def main(): print("=" * 80) print("🎯 BACKTEST BNB: V10 RANGE vs HYBRID") print("=" * 80) # Charger paramètres optimaux params_path = os.path.join(os.path.dirname(__file__), 'optimal_hybrid_params.json') with open(params_path) as f: optimal_params = json.load(f) # Charger données BNB (3 jours, 5min = ~864 bougies) df = get_bnb_historical_data(days=3, interval='5m') # Backtest V10 Range avec params optimaux print("\n📊 Configuration V10 Range (params optimaux BTC):") range_params = optimal_params['range_strategy']['params'] for k, v in range_params.items(): print(f" {k}: {v}") v10_strategy = V10RangeStrategy(leverage=range_params['leverage']) for key, value in range_params.items(): if hasattr(v10_strategy, 'params'): v10_strategy.params[key] = value backtest_v10 = HybridBacktest(df, initial_capital=500.0) results_v10 = backtest_v10.run("V10 Range", v10_strategy) # Backtest Hybrid avec params optimaux print("\n📊 Configuration Hybrid (BULL/BEAR/RANGE optimisés):") print(f" RANGE: {optimal_params['range_strategy']['performance']}") print(f" BULL: {optimal_params['bull_strategy']['performance']}") print(f" BEAR: Params conservateurs") hybrid_strategy = HybridStrategy(leverage=200) # Appliquer params optimaux aux sous-stratégies for key, value in range_params.items(): if hasattr(hybrid_strategy.range_strategy, 'params'): hybrid_strategy.range_strategy.params[key] = value bull_params = optimal_params['bull_strategy']['params'] hybrid_strategy.bull_strategy.min_trend_strength = bull_params['min_trend_strength'] hybrid_strategy.bull_strategy.rsi_pullback_max = bull_params['rsi_pullback_max'] hybrid_strategy.bull_strategy.rsi_breakout_min = bull_params['rsi_breakout_min'] hybrid_strategy.bull_strategy.volume_spike_threshold = bull_params['volume_spike_threshold'] hybrid_strategy.bull_strategy.sl_atr_multiplier = bull_params['sl_atr_multiplier'] hybrid_strategy.bull_strategy.tp_atr_multiplier = bull_params['tp_atr_multiplier'] hybrid_strategy.bull_strategy.trailing_pct = bull_params['trailing_pct'] hybrid_strategy.bull_strategy.min_confidence = bull_params['min_confidence'] hybrid_strategy.bull_strategy.leverage = bull_params['leverage'] bear_params = optimal_params['bear_strategy']['params'] hybrid_strategy.bear_strategy.min_trend_strength = bear_params['min_trend_strength'] hybrid_strategy.bear_strategy.rsi_rally_min = bear_params['rsi_rally_min'] hybrid_strategy.bear_strategy.rsi_overbought = bear_params['rsi_overbought'] hybrid_strategy.bear_strategy.volume_spike_threshold = bear_params['volume_spike_threshold'] hybrid_strategy.bear_strategy.sl_atr_multiplier = bear_params['sl_atr_multiplier'] hybrid_strategy.bear_strategy.tp_atr_multiplier = bear_params['tp_atr_multiplier'] hybrid_strategy.bear_strategy.trailing_pct = bear_params['trailing_pct'] hybrid_strategy.bear_strategy.min_confidence = bear_params['min_confidence'] hybrid_strategy.bear_strategy.leverage = bear_params['leverage'] backtest_hybrid = HybridBacktest(df, initial_capital=500.0) results_hybrid = backtest_hybrid.run("Hybrid", hybrid_strategy) # Comparaison print_comparison(results_v10, results_hybrid) # Sauvegarder résultats results = { 'asset': 'BNBUSDT', 'period': f"{df['timestamp'].iloc[0]} to {df['timestamp'].iloc[-1]}", 'interval': '5m', 'v10_results': results_v10, 'hybrid_results': results_hybrid } output_file = 'backtest_bnb_results.json' with open(output_file, 'w') as f: json.dump(results, f, indent=2, default=str) print(f"\n💾 Résultats sauvegardés: {output_file}") print("\n" + "=" * 80) print("✅ BACKTEST BNB TERMINÉ") print("=" * 80) if __name__ == '__main__': main()