#!/usr/bin/env python3 """ Backtest Stratégie Hybride Compare V10 Range seule vs Hybrid (BULL/BEAR/RANGE adaptatif) """ import pandas as pd import numpy as np import requests 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 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 = None if isinstance(strategy, HybridStrategy): signal = strategy.analyze('BTCUSD', window) else: # V10RangeStrategy v10_signal = strategy.analyze('BTCUSD', window) if v10_signal: # Convertir format V10 en format unifié signal = type('Signal', (), { 'side': v10_signal['side'], 'entry_price': v10_signal['entry_price'], 'stop_loss': v10_signal['stop_loss'], 'take_profit': v10_signal['take_profit'], 'leverage': v10_signal.get('leverage', 200), 'strategy_used': 'RANGE', 'market_regime': 'RANGE' })() if signal: open_position = { 'entry_time': window['timestamp'].iloc[-1], 'entry_price': signal.entry_price, 'side': signal.side, 'stop_loss': signal.stop_loss, 'take_profit': signal.take_profit, 'leverage': signal.leverage, 'strategy_used': getattr(signal, 'strategy_used', 'UNKNOWN'), 'regime': getattr(signal, 'market_regime', 'UNKNOWN') } # Résultats return self._calculate_results(trades, capital, strategy_name, strategy) def _calculate_pnl(self, position: Dict, current_price: float) -> float: """Calcule PnL % d'une position""" 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 _calculate_results(self, trades: List[Dict], final_capital: float, strategy_name: str, strategy) -> Dict: """Calcule métriques de performance""" if not trades: return { 'strategy': strategy_name, 'total_trades': 0, 'win_rate': 0, 'profit_factor': 0, 'total_pnl_pct': 0, 'final_capital': final_capital, 'avg_pnl_per_trade': 0, 'max_drawdown': 0, 'trades_by_strategy': {}, 'trades_by_regime': {} } 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_btc_historical_data(days: int = 7, interval: str = '1m') -> pd.DataFrame: """Récupère données historiques BTC""" print(f"📊 Récupération données BTC ({days} jours, {interval})...") # Binance limite à 1000 bougies par requête limit = min(1000, days * 1440) if interval == '1m' else 1000 url = "https://api.binance.com/api/v3/klines" params = { 'symbol': 'BTCUSDT', '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' ]) df = df[['timestamp', 'open', 'high', 'low', 'close', 'volume']] 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") return df def print_comparison(results_v10: Dict, results_hybrid: Dict): """Affiche comparaison des résultats""" print("\n" + "=" * 80) print("📊 COMPARAISON V10 RANGE vs HYBRID") print("=" * 80) print(f"\n{'Métrique':<30} | {'V10 Range':<20} | {'Hybrid':<20} | {'Diff':<15}") print("-" * 90) metrics = [ ('Total trades', 'total_trades', ''), ('Trades gagnants', 'winning_trades', ''), ('Trades perdants', 'losing_trades', ''), ('Win Rate', 'win_rate', '%'), ('Profit Factor', 'profit_factor', ''), ('PnL Total', 'total_pnl_pct', '%'), ('PnL Moyen/Trade', 'avg_pnl_per_trade', '$'), ('Max Drawdown', 'max_drawdown', '%'), ('Capital Final', 'final_capital', '$'), ] winner_count = 0 for label, key, unit in metrics: v10_val = results_v10.get(key, 0) hybrid_val = results_hybrid.get(key, 0) if key == 'max_drawdown': diff = v10_val - hybrid_val # Moins de DD = mieux winner = "🟢 Hybrid" if diff > 0 else ("🔴 V10" if diff < 0 else "=") if diff > 0: winner_count += 1 else: diff = hybrid_val - v10_val 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 COMPARATIF: V10 RANGE vs HYBRID") print("=" * 80) # Charger données df = get_btc_historical_data(days=1, interval='1m') # Backtest V10 Range v10_strategy = V10RangeStrategy(leverage=250) v10_strategy.params['rsi_oversold'] = 35 v10_strategy.params['rsi_overbought'] = 78 v10_strategy.params['range_atr_threshold'] = 0.32 v10_strategy.params['sl_atr_multiplier'] = 1.64 backtest_v10 = HybridBacktest(df, initial_capital=500.0) results_v10 = backtest_v10.run("V10 Range", v10_strategy) # Backtest Hybrid hybrid_strategy = HybridStrategy(leverage=250) backtest_hybrid = HybridBacktest(df, initial_capital=500.0) results_hybrid = backtest_hybrid.run("Hybrid", hybrid_strategy) # Comparaison print_comparison(results_v10, results_hybrid) print("\n" + "=" * 80) print("✅ BACKTEST TERMINÉ") print("=" * 80) if __name__ == '__main__': main()