#!/usr/bin/env python3 """ Martingale Backtester Backtest la stratégie martingale avec données réelles de Binance/Pyth """ import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import pandas as pd import numpy as np from typing import List, Dict, Optional from dataclasses import dataclass, asdict import logging import requests import json from datetime import datetime from martingale_strategy import MartingaleStrategy, MartingaleSignal logging.basicConfig(level=logging.INFO, format='%(levelname)s - %(message)s') logger = logging.getLogger(__name__) @dataclass class MartingaleTrade: """Trade exécuté dans le backtest""" entry_time: str exit_time: str side: str entry_price: float exit_price: float stop_loss: float take_profit: float position_size_usd: float leverage: int martingale_level: int pnl_pct: float pnl_usd: float exit_reason: str # 'TP', 'SL', 'TIMEOUT' signal_score: float duration_minutes: int @dataclass class BacktestResults: """Résultats du backtest""" symbol: str timeframe: str start_date: str end_date: str initial_capital: float final_capital: float total_return_pct: float total_trades: int winning_trades: int losing_trades: int win_rate: float avg_win_pct: float avg_loss_pct: float largest_win_pct: float largest_loss_pct: float profit_factor: float max_drawdown_pct: float max_consecutive_losses: int max_martingale_level_reached: int total_fees_usd: float trades: List[MartingaleTrade] class MartingaleBacktester: """Backtester pour stratégie martingale""" def __init__( self, initial_capital: float = 1000.0, trading_fee_pct: float = 0.0005, # 0.05% frais slippage_pct: float = 0.0005, # 0.05% slippage timeout_minutes: int = 60 # Timeout si pas de SL/TP après 60min ): self.initial_capital = initial_capital self.trading_fee_pct = trading_fee_pct self.slippage_pct = slippage_pct self.timeout_minutes = timeout_minutes logger.info(f"✅ Backtester initialisé") logger.info(f" Capital: ${initial_capital}") logger.info(f" Frais: {trading_fee_pct*100}%") logger.info(f" Slippage: {slippage_pct*100}%") def fetch_binance_data( self, symbol: str = "BTCUSDT", interval: str = "1m", limit: int = 1000 ) -> Optional[pd.DataFrame]: """ Récupère les données depuis Binance API Args: symbol: Paire (BTCUSDT, ETHUSDT, etc.) interval: Timeframe (1m, 5m, 15m, 1h, etc.) limit: Nombre de bougies (max 1000) Returns: DataFrame avec OHLCV """ try: logger.info(f"📥 Téléchargement données Binance: {symbol} {interval} ({limit} bougies)...") url = "https://api.binance.com/api/v3/klines" params = { 'symbol': symbol, 'interval': interval, 'limit': limit } response = requests.get(url, params=params, timeout=30) response.raise_for_status() klines = response.json() # Parser les données df = pd.DataFrame(klines, 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) # Garder colonnes importantes df = df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].copy() logger.info(f"✅ {len(df)} bougies récupérées") logger.info(f" Période: {df.iloc[0]['timestamp']} → {df.iloc[-1]['timestamp']}") logger.info(f" Prix: ${df.iloc[0]['close']:.2f} → ${df.iloc[-1]['close']:.2f}") return df except Exception as e: logger.error(f"❌ Erreur téléchargement Binance: {e}") return None def simulate_trade( self, signal: MartingaleSignal, df: pd.DataFrame, start_idx: int, capital: float ) -> Optional[MartingaleTrade]: """ Simule un trade depuis un signal Args: signal: Signal martingale df: DataFrame avec OHLCV start_idx: Index de départ capital: Capital disponible Returns: MartingaleTrade ou None """ # Vérifier capital suffisant if signal.position_size_usd > capital: logger.warning(f"⚠️ Capital insuffisant: ${capital:.2f} < ${signal.position_size_usd:.2f}") return None entry_time = df.iloc[start_idx]['timestamp'] entry_price = signal.entry_price # Appliquer slippage sur entrée if signal.side == 'LONG': entry_price *= (1 + self.slippage_pct) else: entry_price *= (1 - self.slippage_pct) # Frais d'entrée entry_fee = signal.position_size_usd * self.trading_fee_pct # Simuler le trade for i in range(start_idx + 1, min(start_idx + self.timeout_minutes + 1, len(df))): current = df.iloc[i] high = current['high'] low = current['low'] close = current['close'] # Vérifier SL/TP if signal.side == 'LONG': # Check stop-loss if low <= signal.stop_loss: exit_price = signal.stop_loss * (1 - self.slippage_pct) exit_reason = 'SL' exit_time = current['timestamp'] break # Check take-profit if high >= signal.take_profit: exit_price = signal.take_profit * (1 - self.slippage_pct) exit_reason = 'TP' exit_time = current['timestamp'] break else: # SHORT # Check stop-loss if high >= signal.stop_loss: exit_price = signal.stop_loss * (1 + self.slippage_pct) exit_reason = 'SL' exit_time = current['timestamp'] break # Check take-profit if low <= signal.take_profit: exit_price = signal.take_profit * (1 + self.slippage_pct) exit_reason = 'TP' exit_time = current['timestamp'] break # Timeout if i - start_idx >= self.timeout_minutes: exit_price = close exit_reason = 'TIMEOUT' exit_time = current['timestamp'] break else: # Fin des données exit_price = df.iloc[-1]['close'] exit_reason = 'EOD' exit_time = df.iloc[-1]['timestamp'] # Calculer PnL if signal.side == 'LONG': pnl_pct = (exit_price - entry_price) / entry_price else: pnl_pct = (entry_price - exit_price) / entry_price # PnL en USD avec leverage pnl_usd = signal.position_size_usd * pnl_pct * signal.leverage # Frais de sortie exit_fee = signal.position_size_usd * self.trading_fee_pct # PnL net pnl_usd -= (entry_fee + exit_fee) pnl_pct = pnl_usd / signal.position_size_usd # % du capital risqué # Duration duration = int((pd.to_datetime(exit_time) - pd.to_datetime(entry_time)).total_seconds() / 60) return MartingaleTrade( entry_time=str(entry_time), exit_time=str(exit_time), side=signal.side, entry_price=entry_price, exit_price=exit_price, stop_loss=signal.stop_loss, take_profit=signal.take_profit, position_size_usd=signal.position_size_usd, leverage=signal.leverage, martingale_level=signal.martingale_level, pnl_pct=pnl_pct * 100, pnl_usd=pnl_usd, exit_reason=exit_reason, signal_score=signal.signal_score, duration_minutes=duration ) def run_backtest( self, df: pd.DataFrame, symbol: str, strategy: MartingaleStrategy ) -> BacktestResults: """ Execute le backtest Args: df: DataFrame avec données OHLCV symbol: Symbole tradé strategy: Stratégie martingale Returns: BacktestResults """ logger.info("\n" + "="*80) logger.info(f"🚀 DÉMARRAGE BACKTEST: {symbol}") logger.info("="*80) capital = self.initial_capital trades = [] equity_curve = [capital] consecutive_losses = 0 max_consecutive_losses = 0 max_martingale_level = 0 total_fees = 0 # Parcourir les données i = 50 # Warmup pour indicateurs cooldown = 0 # Cooldown après un trade while i < len(df) - self.timeout_minutes: # Cooldown if cooldown > 0: cooldown -= 1 i += 1 continue # Générer signal window = df.iloc[:i+1].copy() signal = strategy.generate_signal(window) if not signal: i += 1 continue # Vérifier capital suffisant if signal.position_size_usd > capital: logger.warning(f"⚠️ Capital insuffisant: ${capital:.2f} < ${signal.position_size_usd:.2f}") logger.warning(f" GAME OVER - Capital épuisé!") break # Simuler trade trade = self.simulate_trade(signal, df, i, capital) if trade: trades.append(trade) capital += trade.pnl_usd equity_curve.append(capital) # Frais totaux total_fees += signal.position_size_usd * self.trading_fee_pct * 2 # Mise à jour stratégie won = trade.pnl_usd > 0 strategy.update_after_trade(won) # Stats if won: consecutive_losses = 0 else: consecutive_losses += 1 max_consecutive_losses = max(max_consecutive_losses, consecutive_losses) max_martingale_level = max(max_martingale_level, trade.martingale_level) logger.info( f"\n[{len(trades)}] {trade.side} @ ${trade.entry_price:.2f} | " f"{'✅' if won else '❌'} {trade.exit_reason} | " f"PnL: {trade.pnl_pct:+.2f}% (${trade.pnl_usd:+.2f}) | " f"Level: {trade.martingale_level} | " f"Capital: ${capital:.2f}" ) # Avancer après le trade exit_idx = df[df['timestamp'] == pd.to_datetime(trade.exit_time)].index[0] i = exit_idx + 1 cooldown = 5 # 5 minutes de cooldown # Stop si capital trop bas if capital < self.initial_capital * 0.1: # 10% du capital initial logger.warning(f"⚠️ Capital critique: ${capital:.2f}") logger.warning(f" ARRÊT DU BACKTEST") break else: i += 1 # Calculer métriques finales if not trades: logger.warning("⚠️ Aucun trade exécuté") return None winning_trades = [t for t in trades if t.pnl_usd > 0] losing_trades = [t for t in trades if t.pnl_usd <= 0] win_rate = len(winning_trades) / len(trades) total_wins = sum(t.pnl_usd for t in winning_trades) total_losses = abs(sum(t.pnl_usd for t in losing_trades)) avg_win = np.mean([t.pnl_pct for t in winning_trades]) if winning_trades else 0 avg_loss = np.mean([t.pnl_pct for t in losing_trades]) if losing_trades else 0 largest_win = max([t.pnl_pct for t in winning_trades]) if winning_trades else 0 largest_loss = min([t.pnl_pct for t in losing_trades]) if losing_trades else 0 profit_factor = total_wins / total_losses if total_losses > 0 else float('inf') # Max drawdown equity_series = pd.Series(equity_curve) running_max = equity_series.expanding().max() drawdown = (equity_series - running_max) / running_max * 100 max_drawdown = drawdown.min() total_return = (capital - self.initial_capital) / self.initial_capital * 100 results = BacktestResults( symbol=symbol, timeframe='1m', start_date=str(df.iloc[0]['timestamp']), end_date=str(df.iloc[-1]['timestamp']), initial_capital=self.initial_capital, final_capital=capital, total_return_pct=total_return, total_trades=len(trades), winning_trades=len(winning_trades), losing_trades=len(losing_trades), win_rate=win_rate, avg_win_pct=avg_win, avg_loss_pct=avg_loss, largest_win_pct=largest_win, largest_loss_pct=largest_loss, profit_factor=profit_factor, max_drawdown_pct=max_drawdown, max_consecutive_losses=max_consecutive_losses, max_martingale_level_reached=max_martingale_level, total_fees_usd=total_fees, trades=trades ) self.print_results(results) return results def print_results(self, results: BacktestResults): """Affiche les résultats""" logger.info("\n" + "="*80) logger.info(f"📊 RÉSULTATS BACKTEST MARTINGALE") logger.info("="*80) logger.info(f"\n💰 PERFORMANCE:") logger.info(f" Capital initial: ${results.initial_capital:,.2f}") logger.info(f" Capital final: ${results.final_capital:,.2f}") logger.info(f" Retour total: {results.total_return_pct:+.2f}%") logger.info(f" Frais totaux: ${results.total_fees_usd:.2f}") logger.info(f"\n📈 TRADES:") logger.info(f" Total: {results.total_trades}") logger.info(f" Gagnants: {results.winning_trades} ({results.win_rate*100:.1f}%)") logger.info(f" Perdants: {results.losing_trades}") logger.info(f"\n💵 PNL:") logger.info(f" Gain moyen: {results.avg_win_pct:+.2f}%") logger.info(f" Perte moyenne: {results.avg_loss_pct:+.2f}%") logger.info(f" Plus gros gain: {results.largest_win_pct:+.2f}%") logger.info(f" Plus grosse perte: {results.largest_loss_pct:+.2f}%") logger.info(f" Profit factor: {results.profit_factor:.2f}") logger.info(f"\n⚠️ RISQUE:") logger.info(f" Max drawdown: {results.max_drawdown_pct:.2f}%") logger.info(f" Max pertes consec.: {results.max_consecutive_losses}") logger.info(f" Max martingale level: {results.max_martingale_level_reached}") logger.info("\n" + "="*80) # Résumé selon résultat if results.total_return_pct > 0: logger.info(f"✅ STRATÉGIE PROFITABLE: +{results.total_return_pct:.2f}%") else: logger.info(f"❌ STRATÉGIE PERDANTE: {results.total_return_pct:.2f}%") logger.info("="*80 + "\n") def save_results(self, results: BacktestResults, filename: str = 'martingale_results.json'): """Sauvegarde les résultats en JSON""" output = asdict(results) with open(filename, 'w') as f: json.dump(output, f, indent=2) logger.info(f"💾 Résultats sauvegardés: {filename}") if __name__ == '__main__': logger.info("="*80) logger.info("🎲 MARTINGALE BACKTESTER - DEGEN MODE") logger.info("="*80) # Créer backtester backtester = MartingaleBacktester( initial_capital=1000.0, trading_fee_pct=0.0005, timeout_minutes=60 ) # Télécharger données Binance df = backtester.fetch_binance_data( symbol='BTCUSDT', interval='1m', limit=1000 # ~16h de données ) if df is not None: # Créer stratégie martingale strategy = MartingaleStrategy( base_position_size=10.0, max_martingale_level=5, base_leverage=100, max_leverage=1000, stop_loss_pct=0.003, take_profit_ratio=3.0 ) # Backtest results = backtester.run_backtest(df, 'BTCUSDT', strategy) if results: # Sauvegarder backtester.save_results(results) logger.info("\n✅ Backtest terminé!")