#!/usr/bin/env python3 """ Martingale Strategy - Ultra Degen Mode Stratégie de trading martingale avec leverage jusqu'à 1000x AVERTISSEMENT: Système extrêmement risqué! - La martingale double la mise après chaque perte - Avec 1000x leverage, le risque de liquidation est énorme - Utilisez uniquement avec du capital que vous pouvez perdre Principe: 1. Entrée avec signal de base (RSI, MACD, EMA) 2. Si perte: doubler la position size au prochain signal 3. Si gain: retour à la position size de base 4. Leverage adaptatif selon le niveau de martingale """ 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 Dict, Optional, Tuple from dataclasses import dataclass import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class MartingaleSignal: """Signal de trading martingale""" timestamp: pd.Timestamp side: str # 'LONG' ou 'SHORT' entry_price: float stop_loss: float take_profit: float leverage: int position_size_usd: float martingale_level: int # 0 = base, 1 = x2, 2 = x4, etc. signal_score: float # Confiance du signal de base strategy: str class MartingaleStrategy: """ Stratégie Martingale avec signals techniques Paramètres clés: - base_position_size: Taille de position initiale en USD - max_martingale_level: Niveau max de martingale (ex: 5 = jusqu'à x32) - base_leverage: Leverage de base (ex: 100x) - max_leverage: Leverage maximum (ex: 1000x) - reset_on_win: Si True, reset à base après un gain """ def __init__( self, base_position_size: float = 10.0, max_martingale_level: int = 5, base_leverage: int = 100, max_leverage: int = 1000, reset_on_win: bool = True, stop_loss_pct: float = 0.003, # 0.3% SL de base take_profit_ratio: float = 3.0, # 3:1 reward/risk min_signal_score: float = 0.6 # Score minimum pour trade ): self.base_position_size = base_position_size self.max_martingale_level = max_martingale_level self.base_leverage = base_leverage self.max_leverage = max_leverage self.reset_on_win = reset_on_win self.stop_loss_pct = stop_loss_pct self.take_profit_ratio = take_profit_ratio self.min_signal_score = min_signal_score # État de la martingale self.current_level = 0 self.consecutive_losses = 0 logger.info(f"✅ Martingale Strategy initialisée") logger.info(f" Base position: ${base_position_size}") logger.info(f" Max level: {max_martingale_level} (x{2**max_martingale_level})") logger.info(f" Leverage: {base_leverage}x → {max_leverage}x") def calculate_indicators(self, df: pd.DataFrame) -> pd.DataFrame: """Calcule les indicateurs techniques""" df = df.copy() # RSI delta = df['close'].diff() gain = (delta.where(delta > 0, 0)).rolling(window=7).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=7).mean() rs = gain / loss df['rsi'] = 100 - (100 / (1 + rs)) # EMA rapide et lente df['ema_fast'] = df['close'].ewm(span=8, adjust=False).mean() df['ema_slow'] = df['close'].ewm(span=21, adjust=False).mean() # MACD exp1 = df['close'].ewm(span=12, adjust=False).mean() exp2 = df['close'].ewm(span=26, adjust=False).mean() df['macd'] = exp1 - exp2 df['macd_signal'] = df['macd'].ewm(span=9, adjust=False).mean() df['macd_hist'] = df['macd'] - df['macd_signal'] # Volume spike df['volume_ma'] = df['volume'].rolling(window=20).mean() df['volume_spike'] = df['volume'] / df['volume_ma'] return df def generate_base_signal(self, df: pd.DataFrame) -> Optional[Tuple[str, float]]: """ Génère un signal de base (LONG/SHORT) avec score Returns: Tuple (side, score) ou None """ if len(df) < 50: return None last = df.iloc[-1] prev = df.iloc[-2] # Compteurs de signaux long_signals = 0 short_signals = 0 max_signals = 5 # Nombre max de signaux # Signal 1: RSI if last['rsi'] < 30: # Oversold long_signals += 1 elif last['rsi'] > 70: # Overbought short_signals += 1 # Signal 2: EMA Cross if last['ema_fast'] > last['ema_slow'] and prev['ema_fast'] <= prev['ema_slow']: long_signals += 1 elif last['ema_fast'] < last['ema_slow'] and prev['ema_fast'] >= prev['ema_slow']: short_signals += 1 # Signal 3: MACD Cross if last['macd'] > last['macd_signal'] and prev['macd'] <= prev['macd_signal']: long_signals += 1 elif last['macd'] < last['macd_signal'] and prev['macd'] >= prev['macd_signal']: short_signals += 1 # Signal 4: MACD Histogram momentum if last['macd_hist'] > 0 and last['macd_hist'] > prev['macd_hist']: long_signals += 1 elif last['macd_hist'] < 0 and last['macd_hist'] < prev['macd_hist']: short_signals += 1 # Signal 5: Volume spike (confirme le mouvement) if last['volume_spike'] > 1.5: # Renforce le signal dominant if long_signals > short_signals: long_signals += 1 elif short_signals > long_signals: short_signals += 1 # Déterminer side et score if long_signals > short_signals and long_signals >= 2: score = long_signals / max_signals return ('LONG', score) elif short_signals > long_signals and short_signals >= 2: score = short_signals / max_signals return ('SHORT', score) return None def get_martingale_multiplier(self) -> int: """Retourne le multiplicateur selon le niveau de martingale""" return 2 ** self.current_level def get_leverage_for_level(self) -> int: """Retourne le leverage adapté au niveau de martingale""" # Plus le niveau augmente, plus on utilise de leverage # Niveau 0: base_leverage (100x) # Niveau max: max_leverage (1000x) if self.current_level == 0: return self.base_leverage # Interpolation linéaire leverage_range = self.max_leverage - self.base_leverage level_pct = self.current_level / self.max_martingale_level leverage = self.base_leverage + int(leverage_range * level_pct) return min(leverage, self.max_leverage) def generate_signal(self, df: pd.DataFrame) -> Optional[MartingaleSignal]: """ Génère un signal de trading martingale Args: df: DataFrame avec OHLCV Returns: MartingaleSignal ou None """ # Calculer indicateurs df = self.calculate_indicators(df) # Générer signal de base base_signal = self.generate_base_signal(df) if not base_signal: return None side, score = base_signal # Vérifier score minimum if score < self.min_signal_score: return None # Prix d'entrée entry_price = df.iloc[-1]['close'] # Calculer SL/TP adaptatifs au leverage leverage = self.get_leverage_for_level() # SL: plus serré avec leverage élevé # À 1000x, liquidation à ~0.1%, donc SL max 0.05% max_sl_pct = 1.0 / leverage * 0.5 # 50% de la marge de liquidation sl_pct = min(self.stop_loss_pct, max_sl_pct) if side == 'LONG': stop_loss = entry_price * (1 - sl_pct) take_profit = entry_price * (1 + sl_pct * self.take_profit_ratio) else: # SHORT stop_loss = entry_price * (1 + sl_pct) take_profit = entry_price * (1 - sl_pct * self.take_profit_ratio) # Position size avec martingale multiplier = self.get_martingale_multiplier() position_size = self.base_position_size * multiplier # Limite de sécurité: max 50% du capital sur une position # (assumant capital initial de 1000$) max_position_size = 500.0 position_size = min(position_size, max_position_size) signal = MartingaleSignal( timestamp=df.iloc[-1]['timestamp'], side=side, entry_price=entry_price, stop_loss=stop_loss, take_profit=take_profit, leverage=leverage, position_size_usd=position_size, martingale_level=self.current_level, signal_score=score, strategy=f"MARTINGALE_L{self.current_level}" ) logger.info(f"\n📊 SIGNAL MARTINGALE:") logger.info(f" Side: {side} | Score: {score:.2f}") logger.info(f" Entry: ${entry_price:.2f}") logger.info(f" SL: ${stop_loss:.2f} ({sl_pct*100:.3f}%)") logger.info(f" TP: ${take_profit:.2f} ({sl_pct*self.take_profit_ratio*100:.2f}%)") logger.info(f" Leverage: {leverage}x") logger.info(f" Position: ${position_size:.2f} (x{multiplier})") logger.info(f" Level: {self.current_level}/{self.max_martingale_level}") return signal def update_after_trade(self, won: bool): """ Met à jour l'état de la martingale après un trade Args: won: True si trade gagnant """ if won: logger.info(f"✅ WIN! Reset martingale level: {self.current_level} → 0") if self.reset_on_win: self.current_level = 0 self.consecutive_losses = 0 else: logger.info(f"❌ LOSS! Augmentation martingale: {self.current_level} → {self.current_level + 1}") self.consecutive_losses += 1 if self.current_level < self.max_martingale_level: self.current_level += 1 else: logger.warning(f"⚠️ MAX MARTINGALE LEVEL ATTEINT! Level reste à {self.max_martingale_level}") def get_stats(self) -> Dict: """Retourne les statistiques de la stratégie""" return { 'current_level': self.current_level, 'consecutive_losses': self.consecutive_losses, 'current_multiplier': self.get_martingale_multiplier(), 'current_leverage': self.get_leverage_for_level(), 'next_position_size': self.base_position_size * self.get_martingale_multiplier() } if __name__ == '__main__': logger.info("="*80) logger.info("🧪 TEST MARTINGALE STRATEGY") logger.info("="*80) # Créer données de test dates = pd.date_range('2025-01-01', periods=200, freq='1min') base_price = 50000 # Simuler prix avec tendance trend = np.linspace(0, 500, 200) noise = np.random.randn(200) * 100 closes = base_price + trend + noise df = pd.DataFrame({ 'timestamp': dates, 'open': closes + np.random.randn(200) * 10, 'high': closes + abs(np.random.randn(200)) * 20, 'low': closes - abs(np.random.randn(200)) * 20, 'close': closes, 'volume': np.random.randint(1000, 10000, 200) }) # Créer stratégie strategy = MartingaleStrategy( base_position_size=10.0, max_martingale_level=5, base_leverage=100, max_leverage=1000 ) # Générer signal signal = strategy.generate_signal(df) if signal: logger.info("\n✅ Signal généré avec succès!") logger.info(f"\nStats: {strategy.get_stats()}") # Simuler quelques pertes pour tester la martingale logger.info("\n" + "="*80) logger.info("📉 Simulation de pertes consécutives:") logger.info("="*80) for i in range(6): logger.info(f"\n--- Trade {i+1} ---") signal = strategy.generate_signal(df) if signal: # Simuler une perte strategy.update_after_trade(won=False) logger.info(f"Stats: {strategy.get_stats()}") else: logger.info("⚠️ Aucun signal généré") logger.info("\n" + "="*80) logger.info("✅ Test terminé") logger.info("="*80)