""" Script pour lancer les backtests sur données réelles ApolloX """ import logging import sys from datetime import datetime, timedelta import pandas as pd from apollo_client import ApolloDegenClient from backtesting.backtester import DegenBacktester import config def setup_logging(): """Configure le logging""" logging.basicConfig( level=logging.INFO, format='%(asctime)s - [%(name)s] - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(config.LOG_FILE), logging.StreamHandler(sys.stdout) ] ) def download_historical_data(client: ApolloDegenClient, symbol: str, interval: str = '1m', days: int = 7) -> pd.DataFrame: """ Télécharge les données historiques Args: symbol: Symbole (BTCUSDT) interval: Timeframe (1m, 5m, etc.) days: Nombre de jours à télécharger Returns: DataFrame complet """ logger = logging.getLogger('DataDownloader') logger.info(f"📥 Téléchargement de {symbol} {interval} ({days} jours)...") # ApolloX limite à 1500 chandeliers par requête max_candles_per_request = 1500 # Calculer nombre de requêtes nécessaires if interval == '1m': candles_per_day = 1440 elif interval == '5m': candles_per_day = 288 elif interval == '15m': candles_per_day = 96 elif interval == '1h': candles_per_day = 24 else: candles_per_day = 1440 total_candles = days * candles_per_day num_requests = (total_candles // max_candles_per_request) + 1 logger.info(f"📊 {total_candles} chandeliers requis → {num_requests} requêtes") all_data = [] for i in range(num_requests): try: df = client.get_klines(symbol, interval, limit=max_candles_per_request) if df.empty: logger.warning(f"⚠️ Requête {i+1}/{num_requests} vide") continue all_data.append(df) logger.info(f"✅ Requête {i+1}/{num_requests}: {len(df)} chandeliers récupérés") # Si on a moins que la limite, on a tout if len(df) < max_candles_per_request: break except Exception as e: logger.error(f"❌ Erreur requête {i+1}: {e}") break if not all_data: logger.error("❌ Aucune donnée récupérée") return pd.DataFrame() # Combiner et dédupliquer df_combined = pd.concat(all_data, ignore_index=True) df_combined = df_combined.drop_duplicates(subset=['timestamp'], keep='first') df_combined = df_combined.sort_values('timestamp').reset_index(drop=True) logger.info(f"✅ Total: {len(df_combined)} chandeliers ({df_combined.iloc[0]['timestamp']} → {df_combined.iloc[-1]['timestamp']})") return df_combined def run_single_backtest(symbol: str, leverage: int = 250, days: int = 7): """Lance un backtest sur un symbole""" logger = logging.getLogger('BacktestRunner') logger.info(f"\n{'='*80}") logger.info(f"🚀 BACKTEST: {symbol} | Leverage: {leverage}x | Période: {days} jours") logger.info(f"{'='*80}\n") # Client client = ApolloDegenClient() # Télécharger données df = download_historical_data(client, symbol, interval=config.TIMEFRAME_PRIMARY, days=days) if df.empty: logger.error(f"❌ Pas de données pour {symbol}") return None # Backtester backtester = DegenBacktester( initial_capital=config.BACKTEST_INITIAL_CAPITAL, leverage=leverage ) # Comparer toutes les stratégies results = backtester.compare_strategies(df, symbol) # Sauvegarder timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f'backtest_{symbol}_{leverage}x_{timestamp}.json' backtester.save_results(results, filename) return results def run_multi_symbol_backtest(symbols: list, leverage: int = 250, days: int = 7): """Lance des backtests sur plusieurs symboles""" logger = logging.getLogger('MultiBacktest') all_results = {} for symbol in symbols: try: results = run_single_backtest(symbol, leverage=leverage, days=days) if results: all_results[symbol] = results except Exception as e: logger.error(f"❌ Erreur backtest {symbol}: {e}") # Résumé final logger.info(f"\n{'='*80}") logger.info("🏆 RÉSUMÉ MULTI-SYMBOLES") logger.info(f"{'='*80}\n") best_overall = None best_return = float('-inf') for symbol, results in all_results.items(): logger.info(f"\n📊 {symbol}:") sorted_strategies = sorted(results.items(), key=lambda x: x[1].total_return_pct, reverse=True) for strategy_name, result in sorted_strategies: logger.info( f" {strategy_name:<25} Return: {result.total_return_pct:>8.2f}% | " f"Trades: {result.total_trades:>4} | Win Rate: {result.win_rate*100:>5.1f}%" ) # Track meilleure stratégie globale if result.total_return_pct > best_return: best_return = result.total_return_pct best_overall = (symbol, strategy_name, result) if best_overall: symbol, strategy, result = best_overall logger.info(f"\n{'='*80}") logger.info(f"🥇 MEILLEURE CONFIGURATION GLOBALE:") logger.info(f" Symbole: {symbol}") logger.info(f" Stratégie: {strategy}") logger.info(f" Return: {result.total_return_pct:+.2f}%") logger.info(f" Win Rate: {result.win_rate*100:.1f}%") logger.info(f" Trades: {result.total_trades}") logger.info(f" Sharpe: {result.sharpe_ratio:.2f}") logger.info(f"{'='*80}\n") return all_results if __name__ == '__main__': setup_logging() logger = logging.getLogger(__name__) logger.info("🤖 Apollo Degen Bot - Backtester\n") # Test sur les majors print("Sélection du mode de backtest:") print("1. Test rapide (BTC seulement, 3 jours)") print("2. Test complet majors (BTC, ETH, SOL, BNB, 7 jours)") print("3. Test single symbole custom") choice = input("\nChoix (1-3): ").strip() if choice == '1': # Test rapide logger.info("🚀 Mode: Test rapide\n") run_single_backtest('BTCUSDT', leverage=250, days=3) elif choice == '2': # Test complet logger.info("🚀 Mode: Test complet majors\n") symbols = [ config.MAJOR_MARKETS['BTC'], config.MAJOR_MARKETS['ETH'], config.MAJOR_MARKETS['SOL'], config.MAJOR_MARKETS['BNB'] ] run_multi_symbol_backtest(symbols, leverage=250, days=7) elif choice == '3': # Custom symbol = input("Symbole (ex: BTCUSDT): ").strip().upper() leverage = int(input("Leverage (ex: 250): ").strip()) days = int(input("Jours (ex: 7): ").strip()) run_single_backtest(symbol, leverage=leverage, days=days) else: logger.error("❌ Choix invalide") logger.info("\n✅ Backtest terminé!")