#!/usr/bin/env python3 """ Martingale Paper Trading Bot Trade en mode simulation avec données Pyth Network en temps réel ULTRA DEGEN MODE - 1000x LEVERAGE """ import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.append(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'apollo_degen_bot')) import time import pandas as pd import numpy as np from datetime import datetime, timedelta from typing import Optional, List, Dict import json import logging from pyth_price_feed import PythPriceFeed from martingale_strategy import MartingaleStrategy, MartingaleSignal logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class PaperPosition: """Position paper trading""" def __init__( self, signal: MartingaleSignal, entry_time: datetime ): self.signal = signal self.entry_time = entry_time self.entry_price = signal.entry_price self.stop_loss = signal.stop_loss self.take_profit = signal.take_profit self.position_size = signal.position_size_usd self.leverage = signal.leverage self.side = signal.side self.martingale_level = signal.martingale_level # État self.closed = False self.exit_time: Optional[datetime] = None self.exit_price: Optional[float] = None self.exit_reason: Optional[str] = None self.pnl_usd: float = 0.0 self.pnl_pct: float = 0.0 # Stats self.peak_pnl: float = 0.0 self.lowest_pnl: float = 0.0 def update_pnl(self, current_price: float) -> float: """Calcule le PnL actuel""" if self.closed: return self.pnl_usd # Calculer PnL if self.side == 'LONG': pnl_pct = (current_price - self.entry_price) / self.entry_price else: # SHORT pnl_pct = (self.entry_price - current_price) / self.entry_price # Appliquer leverage pnl_usd = self.position_size * pnl_pct * self.leverage # Track peaks self.peak_pnl = max(self.peak_pnl, pnl_usd) self.lowest_pnl = min(self.lowest_pnl, pnl_usd) return pnl_usd def check_exit(self, high: float, low: float, current_time: datetime) -> bool: """ Vérifie si la position doit être fermée Args: high: Prix haut de la bougie low: Prix bas de la bougie current_time: Timestamp actuel Returns: True si position fermée """ if self.closed: return True # Vérifier SL/TP if self.side == 'LONG': # Stop-loss touché if low <= self.stop_loss: self.close_position( exit_price=self.stop_loss, exit_time=current_time, reason='SL' ) return True # Take-profit touché if high >= self.take_profit: self.close_position( exit_price=self.take_profit, exit_time=current_time, reason='TP' ) return True else: # SHORT # Stop-loss touché if high >= self.stop_loss: self.close_position( exit_price=self.stop_loss, exit_time=current_time, reason='SL' ) return True # Take-profit touché if low <= self.take_profit: self.close_position( exit_price=self.take_profit, exit_time=current_time, reason='TP' ) return True # Timeout après 1h duration = (current_time - self.entry_time).total_seconds() / 60 if duration > 60: # Fermer au prix actuel (approximation: moyenne high/low) exit_price = (high + low) / 2 self.close_position( exit_price=exit_price, exit_time=current_time, reason='TIMEOUT' ) return True return False def close_position(self, exit_price: float, exit_time: datetime, reason: str): """Ferme la position""" self.closed = True self.exit_price = exit_price self.exit_time = exit_time self.exit_reason = reason # Calculer PnL final if self.side == 'LONG': self.pnl_pct = (exit_price - self.entry_price) / self.entry_price else: self.pnl_pct = (self.entry_price - exit_price) / self.entry_price # Appliquer leverage self.pnl_usd = self.position_size * self.pnl_pct * self.leverage def to_dict(self) -> Dict: """Export en dict""" return { 'entry_time': str(self.entry_time), 'exit_time': str(self.exit_time) if self.exit_time else None, 'side': self.side, 'entry_price': self.entry_price, 'exit_price': self.exit_price, 'stop_loss': self.stop_loss, 'take_profit': self.take_profit, 'position_size': self.position_size, 'leverage': self.leverage, 'martingale_level': self.martingale_level, 'pnl_usd': self.pnl_usd, 'pnl_pct': self.pnl_pct * 100, 'exit_reason': self.exit_reason, 'closed': self.closed } class MartingalePaperBot: """Bot de paper trading martingale""" def __init__( self, symbol: str = 'BTCUSDT', initial_capital: float = 1000.0, base_position_size: float = 10.0, max_martingale_level: int = 5, base_leverage: int = 100, max_leverage: int = 1000, interval_seconds: int = 60 # Vérifier toutes les 60s ): self.symbol = symbol self.initial_capital = initial_capital self.capital = initial_capital self.interval_seconds = interval_seconds # Pyth price feed self.price_feed = PythPriceFeed() # Stratégie martingale self.strategy = MartingaleStrategy( base_position_size=base_position_size, max_martingale_level=max_martingale_level, base_leverage=base_leverage, max_leverage=max_leverage ) # Positions self.current_position: Optional[PaperPosition] = None self.closed_positions: List[PaperPosition] = [] # Données de prix pour indicateurs self.price_history: List[Dict] = [] self.max_history = 200 # Garder 200 bougies logger.info("\n" + "="*80) logger.info(f"🤖 MARTINGALE PAPER BOT INITIALISÉ") logger.info("="*80) logger.info(f"Symbol: {symbol}") logger.info(f"Capital: ${initial_capital}") logger.info(f"Base position: ${base_position_size}") logger.info(f"Max martingale: {max_martingale_level} (x{2**max_martingale_level})") logger.info(f"Leverage: {base_leverage}x → {max_leverage}x") logger.info(f"Interval: {interval_seconds}s") logger.info("="*80 + "\n") def fetch_current_candle(self) -> Optional[Dict]: """Récupère la bougie actuelle depuis Pyth/Binance""" try: # Utiliser Binance pour OHLCV complet (1 bougie) df = self.price_feed.get_klines(self.symbol, '1m', 1) if df is None or len(df) == 0: return None candle = df.iloc[-1] return { 'timestamp': candle['timestamp'], 'open': candle['open'], 'high': candle['high'], 'low': candle['low'], 'close': candle['close'], 'volume': candle['volume'] } except Exception as e: logger.error(f"❌ Erreur fetch_current_candle: {e}") return None def update_price_history(self, candle: Dict): """Met à jour l'historique des prix""" # Éviter doublons if self.price_history: last_timestamp = self.price_history[-1]['timestamp'] if candle['timestamp'] == last_timestamp: # Mettre à jour la dernière bougie (close peut changer) self.price_history[-1] = candle return # Ajouter nouvelle bougie self.price_history.append(candle) # Limiter taille if len(self.price_history) > self.max_history: self.price_history = self.price_history[-self.max_history:] def get_price_dataframe(self) -> pd.DataFrame: """Convertit l'historique en DataFrame pour les indicateurs""" if not self.price_history: return pd.DataFrame() df = pd.DataFrame(self.price_history) return df def check_current_position(self, candle: Dict): """Vérifie la position actuelle""" if not self.current_position: return # Vérifier sortie closed = self.current_position.check_exit( high=candle['high'], low=candle['low'], current_time=candle['timestamp'] ) if closed: # Position fermée self.capital += self.current_position.pnl_usd # Log emoji = "✅" if self.current_position.pnl_usd > 0 else "❌" logger.info(f"\n{emoji} POSITION FERMÉE:") logger.info(f" Side: {self.current_position.side}") logger.info(f" Entry: ${self.current_position.entry_price:.2f}") logger.info(f" Exit: ${self.current_position.exit_price:.2f}") logger.info(f" Reason: {self.current_position.exit_reason}") logger.info(f" PnL: ${self.current_position.pnl_usd:+.2f} ({self.current_position.pnl_pct*100:+.2f}%)") logger.info(f" Capital: ${self.capital:.2f}") logger.info(f" Level: {self.current_position.martingale_level}") # Mise à jour stratégie won = self.current_position.pnl_usd > 0 self.strategy.update_after_trade(won) # Sauvegarder self.closed_positions.append(self.current_position) self.current_position = None # Sauvegarder état self.save_state() else: # Position toujours ouverte - afficher PnL pnl = self.current_position.update_pnl(candle['close']) logger.debug(f" 📊 Position ouverte - PnL: ${pnl:+.2f}") def try_open_position(self, candle: Dict): """Essaie d'ouvrir une nouvelle position""" # Skip si position déjà ouverte if self.current_position: return # Besoin d'au moins 50 bougies pour indicateurs if len(self.price_history) < 50: logger.debug(f" ⏳ Warmup... {len(self.price_history)}/50 bougies") return # Générer signal df = self.get_price_dataframe() signal = self.strategy.generate_signal(df) if not signal: return # Vérifier capital suffisant if signal.position_size_usd > self.capital: logger.warning(f"⚠️ Capital insuffisant: ${self.capital:.2f} < ${signal.position_size_usd:.2f}") logger.warning(f" GAME OVER!") return # Ouvrir position self.current_position = PaperPosition( signal=signal, entry_time=candle['timestamp'] ) logger.info(f"\n🚀 NOUVELLE POSITION:") logger.info(f" Side: {signal.side}") logger.info(f" Entry: ${signal.entry_price:.2f}") logger.info(f" SL: ${signal.stop_loss:.2f}") logger.info(f" TP: ${signal.take_profit:.2f}") logger.info(f" Size: ${signal.position_size_usd:.2f}") logger.info(f" Leverage: {signal.leverage}x") logger.info(f" Level: {signal.martingale_level}") logger.info(f" Score: {signal.signal_score:.2f}") def get_stats(self) -> Dict: """Statistiques du bot""" total_trades = len(self.closed_positions) if total_trades == 0: return { 'total_trades': 0, 'win_rate': 0, 'total_pnl': 0, 'roi_pct': 0 } winning_trades = [p for p in self.closed_positions if p.pnl_usd > 0] losing_trades = [p for p in self.closed_positions if p.pnl_usd <= 0] win_rate = len(winning_trades) / total_trades total_pnl = sum(p.pnl_usd for p in self.closed_positions) roi_pct = (self.capital - self.initial_capital) / self.initial_capital * 100 return { 'total_trades': total_trades, 'winning_trades': len(winning_trades), 'losing_trades': len(losing_trades), 'win_rate': win_rate, 'total_pnl': total_pnl, 'current_capital': self.capital, 'roi_pct': roi_pct, 'martingale_level': self.strategy.current_level, 'consecutive_losses': self.strategy.consecutive_losses } def print_stats(self): """Affiche les stats""" stats = self.get_stats() logger.info("\n" + "="*80) logger.info("📊 STATISTIQUES BOT") logger.info("="*80) logger.info(f"Capital: ${stats['current_capital']:.2f}") logger.info(f"ROI: {stats['roi_pct']:+.2f}%") logger.info(f"PnL total: ${stats['total_pnl']:+.2f}") logger.info(f"\nTrades: {stats['total_trades']}") logger.info(f"Win rate: {stats['win_rate']*100:.1f}%") logger.info(f"Gagnants: {stats['winning_trades']}") logger.info(f"Perdants: {stats['losing_trades']}") logger.info(f"\nMartingale level: {stats['martingale_level']}") logger.info(f"Pertes consécutives: {stats['consecutive_losses']}") logger.info("="*80 + "\n") def save_state(self): """Sauvegarde l'état du bot""" state = { 'timestamp': str(datetime.now()), 'capital': self.capital, 'stats': self.get_stats(), 'closed_positions': [p.to_dict() for p in self.closed_positions] } with open('paper_bot_state.json', 'w') as f: json.dump(state, f, indent=2) def run(self, duration_minutes: Optional[int] = None): """ Lance le bot Args: duration_minutes: Durée de run en minutes (None = infini) """ logger.info(f"🚀 Démarrage du bot...") start_time = datetime.now() try: while True: # Vérifier durée if duration_minutes: elapsed = (datetime.now() - start_time).total_seconds() / 60 if elapsed >= duration_minutes: logger.info(f"⏱️ Durée atteinte: {duration_minutes} minutes") break # Récupérer bougie actuelle candle = self.fetch_current_candle() if candle: # Mettre à jour historique self.update_price_history(candle) # Vérifier position actuelle self.check_current_position(candle) # Essayer d'ouvrir nouvelle position self.try_open_position(candle) # Attendre prochain cycle time.sleep(self.interval_seconds) except KeyboardInterrupt: logger.info("\n⚠️ Interruption utilisateur (Ctrl+C)") finally: # Stats finales self.print_stats() self.save_state() logger.info("✅ Bot arrêté") if __name__ == '__main__': logger.info("\n" + "="*80) logger.info("🎲 MARTINGALE PAPER TRADING BOT") logger.info("="*80 + "\n") # Créer bot bot = MartingalePaperBot( symbol='BTCUSDT', initial_capital=1000.0, base_position_size=10.0, max_martingale_level=5, base_leverage=100, max_leverage=1000, interval_seconds=60 ) # Lancer (Ctrl+C pour arrêter) bot.run()