""" Test des stratégies ultra-agressives V7: Ultra Leverage V8: Pyramiding V9: Strategy Optimizer V10: Combined Aggressive """ import logging import pandas as pd import numpy as np from apollo_client import ApolloDegenClient from strategies.inverse_strategy import AdaptiveInverseBacktester as BaselineBacktester from strategies.inverse_strategy_aggressive import ( InverseStrategyV7_UltraLeverage, InverseStrategyV8_Pyramiding, InverseStrategyV9_StrategyOptimizer, InverseStrategyV10_Combined_Aggressive ) import config logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Désactiver logs verbeux for module in ['InverseStrategy', 'ShortTermSignalGenerator', 'ApolloDegenClient', 'MarketRegimeDetectorAdvanced', 'InverseStrategyV7_UltraLeverage', 'InverseStrategyV8_Pyramiding', 'InverseStrategyV9_StrategyOptimizer', 'InverseStrategyV10_Combined_Aggressive']: logging.getLogger(module).setLevel(logging.WARNING) class AggressiveBacktester: """Backtester pour stratégies agressives""" def __init__(self, strategy, initial_capital: float = 1000): self.strategy = strategy self.initial_capital = initial_capital self.trading_fee = config.TRADING_FEE_TAKER def backtest(self, df: pd.DataFrame, symbol: str) -> dict: capital = self.initial_capital trades = [] i = 50 while i < len(df) - 100: window = df.iloc[:i+1].copy() signal = self.strategy.generate_inverse_signal(window, symbol) if not signal: i += 1 continue # Leverage et risk dynamiques (V7, V10) if hasattr(self.strategy, 'get_leverage_and_risk'): leverage, risk_pct = self.strategy.get_leverage_and_risk( signal.score, len(signal.confluences), signal.strategy ) else: leverage = 200 risk_pct = 0.02 # Position sizing de base base_position_size = self.initial_capital * risk_pct * leverage # Multiplicateur selon stratégie (V9, V10) if hasattr(self.strategy, 'get_position_multiplier'): multiplier = self.strategy.get_position_multiplier(signal.strategy) base_position_size *= multiplier entry_time = df.iloc[i]['timestamp'] entry_price = signal.entry_price sl = signal.stop_loss tp = signal.take_profit # Tracking pour pyramiding total_position_size = base_position_size pyramids_added = 0 pyramid_prices = [] # Chercher sortie exit_found = False for j in range(i + 1, min(i + 100, len(df))): current = df.iloc[j] high = current['high'] low = current['low'] close = current['close'] # Pyramiding (V8, V10) if hasattr(self.strategy, 'should_add_to_position'): should_add, add_pct = self.strategy.should_add_to_position( signal.side, entry_price, close, pyramids_added ) if should_add: add_size = base_position_size * add_pct total_position_size += add_size pyramids_added += 1 pyramid_prices.append(close) # Check SL/TP exit_price = None exit_reason = None if signal.side == 'LONG': if low <= sl: exit_price = sl exit_reason = 'SL' elif high >= tp: exit_price = tp exit_reason = 'TP' else: if high >= sl: exit_price = sl exit_reason = 'SL' elif low <= tp: exit_price = tp exit_reason = 'TP' if exit_price: # Calculer PnL if signal.side == 'LONG': pnl_pct = (exit_price - entry_price) / entry_price else: pnl_pct = (entry_price - exit_price) / entry_price pnl_usd = total_position_size * pnl_pct pnl_usd -= total_position_size * self.trading_fee * 2 capital += pnl_usd # Update performance (V9, V10) if hasattr(self.strategy, 'update_performance'): self.strategy.update_performance(signal.strategy, pnl_usd) trades.append({ 'entry_time': entry_time, 'exit_time': current['timestamp'], 'side': signal.side, 'strategy': signal.strategy, 'pnl_pct': pnl_pct * 100, 'pnl_usd': pnl_usd, 'exit_reason': exit_reason, 'leverage': leverage, 'pyramids': pyramids_added, 'position_size': total_position_size }) i = j + 1 exit_found = True break if not exit_found: i += 1 if not trades: return None winning = [t for t in trades if t['pnl_usd'] > 0] losing = [t for t in trades if t['pnl_usd'] <= 0] total_return = (capital - self.initial_capital) / self.initial_capital * 100 win_rate = len(winning) / len(trades) if trades else 0 total_wins = sum(t['pnl_usd'] for t in winning) if winning else 0 total_losses = abs(sum(t['pnl_usd'] for t in losing)) if losing else 1 profit_factor = total_wins / total_losses if total_losses > 0 else 999 return { 'total_return': total_return, 'win_rate': win_rate, 'profit_factor': profit_factor, 'trades': trades, 'num_trades': len(trades), 'wins': len(winning), 'losses': len(losing), 'avg_win': np.mean([t['pnl_pct'] for t in winning]) if winning else 0, 'avg_loss': np.mean([t['pnl_pct'] for t in losing]) if losing else 0, 'final_capital': capital, 'max_leverage': max([t['leverage'] for t in trades]) if trades else 0, 'avg_pyramids': np.mean([t['pyramids'] for t in trades]) if trades else 0 } if __name__ == '__main__': logger.info("🚀 TEST STRATÉGIES ULTRA-AGRESSIVES\n") logger.info("="*80) # Données client = ApolloDegenClient() df = client.get_klines('BTCUSDT', '1m', limit=1500) logger.info(f"Données: {len(df)} chandeliers BTC 1m\n") results = {} # BASELINE logger.info("Testing BASELINE...") try: baseline_bt = BaselineBacktester(initial_capital=1000, leverage=200) baseline_result = baseline_bt.backtest(df, 'BTCUSDT') if baseline_result: winning = [t for t in baseline_result['trades'] if t['pnl_usd'] > 0] losing = [t for t in baseline_result['trades'] if t['pnl_usd'] <= 0] total_wins = sum(t['pnl_usd'] for t in winning) if winning else 0 total_losses = abs(sum(t['pnl_usd'] for t in losing)) if losing else 1 baseline_result['profit_factor'] = total_wins / total_losses if total_losses > 0 else 999 baseline_result['num_trades'] = len(baseline_result['trades']) baseline_result['wins'] = len(winning) baseline_result['losses'] = len(losing) baseline_result['avg_win'] = np.mean([t['pnl_pct'] for t in winning]) if winning else 0 baseline_result['avg_loss'] = np.mean([t['pnl_pct'] for t in losing]) if losing else 0 baseline_result['max_leverage'] = 200 baseline_result['avg_pyramids'] = 0 results['BASELINE'] = baseline_result except Exception as e: logger.error(f"BASELINE failed: {e}") results['BASELINE'] = None # V7: Ultra Leverage logger.info("Testing V7: Ultra Leverage...") try: v7_strategy = InverseStrategyV7_UltraLeverage() v7_bt = AggressiveBacktester(v7_strategy) results['V7_ULTRA_LEVERAGE'] = v7_bt.backtest(df, 'BTCUSDT') except Exception as e: logger.error(f"V7 failed: {e}") results['V7_ULTRA_LEVERAGE'] = None # V8: Pyramiding logger.info("Testing V8: Pyramiding...") try: v8_strategy = InverseStrategyV8_Pyramiding() v8_bt = AggressiveBacktester(v8_strategy) results['V8_PYRAMIDING'] = v8_bt.backtest(df, 'BTCUSDT') except Exception as e: logger.error(f"V8 failed: {e}") results['V8_PYRAMIDING'] = None # V9: Strategy Optimizer logger.info("Testing V9: Strategy Optimizer...") try: v9_strategy = InverseStrategyV9_StrategyOptimizer() v9_bt = AggressiveBacktester(v9_strategy) results['V9_STRATEGY_OPT'] = v9_bt.backtest(df, 'BTCUSDT') except Exception as e: logger.error(f"V9 failed: {e}") results['V9_STRATEGY_OPT'] = None # V10: Combined Aggressive logger.info("Testing V10: Combined Aggressive...") try: v10_strategy = InverseStrategyV10_Combined_Aggressive() v10_bt = AggressiveBacktester(v10_strategy) results['V10_COMBINED_AGG'] = v10_bt.backtest(df, 'BTCUSDT') except Exception as e: logger.error(f"V10 failed: {e}") results['V10_COMBINED_AGG'] = None # RÉSULTATS logger.info("\n" + "="*80) logger.info("📊 RÉSULTATS COMPARATIFS - STRATÉGIES AGRESSIVES") logger.info("="*80 + "\n") logger.info(f"{'Version':<25} {'Return%':<10} {'WR%':<8} {'Trades':<8} {'MaxLev':<8} {'Pyramids':<10}") logger.info("-"*85) sorted_results = sorted( [(k, v) for k, v in results.items() if v is not None], key=lambda x: x[1]['total_return'], reverse=True ) for version, res in sorted_results: logger.info( f"{version:<25} " f"{res['total_return']:<10.2f} " f"{res['win_rate']*100:<8.0f} " f"{res['num_trades']:<8} " f"{res['max_leverage']:<8.0f} " f"{res['avg_pyramids']:<10.2f}" ) # Gagnant if sorted_results: winner_name, winner = sorted_results[0] baseline_return = results['BASELINE']['total_return'] if results['BASELINE'] else 0 logger.info("\n" + "="*80) logger.info("🏆 MEILLEURE STRATÉGIE") logger.info("="*80) logger.info(f"Version: {winner_name}") logger.info(f"Return: {winner['total_return']:+.2f}%") logger.info(f"Win Rate: {winner['win_rate']*100:.1f}%") logger.info(f"Profit Factor: {winner['profit_factor']:.2f}") logger.info(f"Total Trades: {winner['num_trades']}") logger.info(f"Wins/Losses: {winner['wins']}/{winner['losses']}") logger.info(f"Max Leverage: {winner['max_leverage']:.0f}x") logger.info(f"Avg Pyramids/Trade: {winner['avg_pyramids']:.2f}") logger.info(f"") logger.info(f"💰 Amélioration vs BASELINE: {winner['total_return'] - baseline_return:+.2f} points") logger.info("="*80) if winner['avg_pyramids'] > 0: pyramid_trades = [t for t in winner['trades'] if t['pyramids'] > 0] logger.info(f"\n📈 PYRAMIDING STATS:") logger.info(f" Trades with pyramids: {len(pyramid_trades)}/{winner['num_trades']}") if pyramid_trades: avg_pnl_pyramid = np.mean([t['pnl_pct'] for t in pyramid_trades]) logger.info(f" Avg PnL with pyramids: {avg_pnl_pyramid:+.2f}%") logger.info("\n")