""" Live Trading Engine V10 - ApolloX Degen Bot Exécute les signaux de la stratégie inverse V10 en mode live """ import logging import time import json from typing import Dict, List, Optional from datetime import datetime from pathlib import Path import threading import pandas as pd from apollo_client import ApolloDegenClient from strategies.inverse_strategy_v10 import InverseStrategyV10, InverseBacktesterV10 import config logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class Position: """Classe pour gérer une position ouverte""" def __init__(self, symbol: str, side: str, entry_price: float, quantity: float, leverage: int, strategy_name: str): self.symbol = symbol self.side = side # LONG ou SHORT self.entry_price = entry_price self.quantity = quantity self.leverage = leverage self.strategy_name = strategy_name # Risk management self.stop_loss = None self.take_profit = None self.pyramids_added = 0 # Tracking self.opened_at = datetime.now() self.status = 'OPEN' self.pnl = 0.0 self.pnl_pct = 0.0 def update_pnl(self, current_price: float): """Met à jour le PnL""" if self.side == 'LONG': self.pnl_pct = (current_price - self.entry_price) / self.entry_price else: # SHORT self.pnl_pct = (self.entry_price - current_price) / self.entry_price # PnL avec leverage self.pnl = self.pnl_pct * self.leverage * (self.quantity * self.entry_price) def to_dict(self) -> Dict: """Convertit en dict pour sauvegarde""" return { 'symbol': self.symbol, 'side': self.side, 'entry_price': self.entry_price, 'quantity': self.quantity, 'leverage': self.leverage, 'strategy_name': self.strategy_name, 'stop_loss': self.stop_loss, 'take_profit': self.take_profit, 'pyramids_added': self.pyramids_added, 'opened_at': self.opened_at.isoformat(), 'status': self.status, 'pnl': self.pnl, 'pnl_pct': self.pnl_pct } class LiveTraderV10: """ Live Trading Engine V10 Features: - Génération de signaux inverse V10 - Exécution automatique des trades - Pyramiding intelligent - Stop-loss et take-profit dynamiques - Gestion multi-positions - Tracking performance - Sauvegarde état """ def __init__(self, symbols: List[str] = None, mode: str = 'LIVE'): """ Args: symbols: Liste des symboles à trader (défaut: MAJOR_MARKETS) mode: LIVE ou PAPER (paper trading sans ordres réels) """ self.mode = mode self.symbols = symbols or list(config.MAJOR_MARKETS.values()) # Clients self.client = ApolloDegenClient() self.strategy = InverseStrategyV10() # État self.positions: Dict[str, Position] = {} # symbol -> Position self.is_running = False self.thread = None # Configuration self.check_interval = 10 # Vérifier toutes les 10 secondes self.max_positions = 5 # Maximum 5 positions simultanées self.min_balance = 100 # Balance minimum $100 # Tracking self.trades_history = [] self.stats = { 'total_trades': 0, 'wins': 0, 'losses': 0, 'total_pnl': 0.0, 'win_rate': 0.0 } # Sauvegarde self.data_dir = Path('live_trading_data') self.data_dir.mkdir(exist_ok=True) logger.info(f"🚀 LiveTraderV10 initialisé ({mode} mode)") logger.info(f" Symboles: {', '.join(self.symbols)}") logger.info(f" Max positions: {self.max_positions}") def start(self): """Démarre le trading live""" if self.is_running: logger.warning("Trader déjà en cours") return # Vérifications de sécurité if not self._safety_checks(): logger.error("❌ Safety checks échoués - trading non démarré") return self.is_running = True self.thread = threading.Thread(target=self._run_loop, daemon=True) self.thread.start() logger.info(f"✅ Trading démarré en mode {self.mode}") def stop(self): """Arrête le trading""" self.is_running = False if self.thread: self.thread.join(timeout=10) self._save_state() logger.info("⏹️ Trading arrêté") def _safety_checks(self) -> bool: """Vérifications de sécurité avant démarrage""" logger.info("🔍 Safety checks...") # 1. Vérifier API keys if self.mode == 'LIVE': if not config.APOLLOX_API_KEY or not config.APOLLOX_API_SECRET: logger.error("❌ API keys non configurées") return False logger.info(" ✅ API keys configurées") # 2. Vérifier connexion try: balance = self.client.get_balance() logger.info(f" ✅ Balance: ${balance:.2f}") if balance < self.min_balance: logger.error(f"❌ Balance insuffisante (min ${self.min_balance})") return False except Exception as e: logger.error(f"❌ Erreur connexion: {e}") return False # 3. Vérifier stratégie if not self.strategy: logger.error("❌ Stratégie non initialisée") return False logger.info(" ✅ Stratégie V10 prête") logger.info("✅ Tous les checks passés") return True def _run_loop(self): """Boucle principale de trading""" iteration = 0 while self.is_running: iteration += 1 try: logger.info(f"\n{'='*70}") logger.info(f"🔄 Cycle #{iteration} - {datetime.now().strftime('%H:%M:%S')}") logger.info(f"{'='*70}") # 1. Mettre à jour positions existantes self._update_positions() # 2. Chercher nouveaux signaux if len(self.positions) < self.max_positions: self._check_new_signals() else: logger.info(f"⚠️ Max positions atteint ({self.max_positions})") # 3. Gérer le pyramiding self._manage_pyramiding() # 4. Gérer stop-loss et take-profit self._manage_risk() # 5. Afficher résumé self._print_summary() # 6. Sauvegarder état if iteration % 6 == 0: # Toutes les minutes self._save_state() # Sleep if self.is_running: time.sleep(self.check_interval) except Exception as e: logger.error(f"❌ Erreur dans loop: {e}") time.sleep(30) def _update_positions(self): """Met à jour les positions existantes""" if not self.positions: return logger.info(f"📊 Mise à jour {len(self.positions)} positions...") for symbol, position in list(self.positions.items()): try: # Récupérer prix actuel ticker = self.client.get_ticker(symbol) current_price = float(ticker.get('lastPrice', 0)) if current_price == 0: logger.warning(f"⚠️ Prix invalide pour {symbol}") continue # Mettre à jour PnL position.update_pnl(current_price) logger.info( f" {symbol}: {position.side} @ ${position.entry_price:.2f} " f"→ ${current_price:.2f} | PnL: {position.pnl_pct*100:+.2f}% " f"(${position.pnl:+.2f})" ) except Exception as e: logger.error(f"❌ Erreur update {symbol}: {e}") def _check_new_signals(self): """Vérifie s'il y a de nouveaux signaux à trader""" logger.info("🔍 Recherche nouveaux signaux...") for symbol in self.symbols: # Skip si position déjà ouverte if symbol in self.positions: continue try: # Récupérer données df = self.client.get_klines(symbol, config.TIMEFRAME_PRIMARY, limit=200) if df.empty or len(df) < 100: logger.warning(f"⚠️ Données insuffisantes pour {symbol}") continue # Générer signal signal = self.strategy.generate_inverse_signal(df, symbol) if signal and signal['confidence'] >= config.MIN_SIGNAL_SCORE_AGGRESSIVE: logger.info( f"🎯 SIGNAL: {signal['side']} {symbol} " f"(conf: {signal['confidence']*100:.0f}%, " f"lev: {signal['leverage']}x)" ) # Ouvrir position self._open_position(signal, df) except Exception as e: logger.error(f"❌ Erreur signal {symbol}: {e}") def _open_position(self, signal: Dict, df: pd.DataFrame): """Ouvre une nouvelle position""" symbol = signal['symbol'] side = signal['side'] leverage = signal['leverage'] risk_pct = signal['risk_pct'] try: # 1. Récupérer balance balance = self.client.get_balance() # 2. Calculer taille position current_price = df['close'].iloc[-1] risk_amount = balance * risk_pct # Quantité = (capital à risquer) / (prix * SL%) sl_pct = config.STOP_LOSS_PCT[leverage] quantity = risk_amount / (current_price * sl_pct) # Arrondir à la précision du symbole (0.001 pour BTC, 0.01 pour ETH, etc.) if 'BTC' in symbol: quantity = round(quantity, 3) else: quantity = round(quantity, 2) logger.info( f"💰 Position: ${risk_amount:.2f} risk → {quantity} {symbol} " f"@ ${current_price:.2f}" ) if self.mode == 'PAPER': logger.info("📝 PAPER MODE - Ordre simulé") order_success = True else: # 3. Configurer leverage et margin type self.client.set_margin_type(symbol, 'ISOLATED') self.client.set_leverage(symbol, leverage) # 4. Placer ordre market order_side = 'BUY' if side == 'LONG' else 'SELL' order = self.client.place_market_order(symbol, order_side, quantity) order_success = order is not None if order_success: # Récupérer prix d'exécution current_price = float(order.get('avgPrice', current_price)) if not order_success: logger.error(f"❌ Échec ouverture position {symbol}") return # 5. Créer objet Position position = Position( symbol=symbol, side=side, entry_price=current_price, quantity=quantity, leverage=leverage, strategy_name=signal['strategy'] ) # 6. Calculer SL et TP if side == 'LONG': position.stop_loss = current_price * (1 - sl_pct) position.take_profit = current_price * (1 + sl_pct * config.TAKE_PROFIT_RATIO) else: # SHORT position.stop_loss = current_price * (1 + sl_pct) position.take_profit = current_price * (1 - sl_pct * config.TAKE_PROFIT_RATIO) # 7. Placer ordres SL/TP if self.mode == 'LIVE': sl_side = 'SELL' if side == 'LONG' else 'BUY' self.client.place_stop_loss(symbol, sl_side, quantity, position.stop_loss) self.client.place_take_profit(symbol, sl_side, quantity, position.take_profit) # 8. Sauvegarder position self.positions[symbol] = position self.stats['total_trades'] += 1 logger.info( f"✅ Position ouverte: {side} {quantity} {symbol} @ ${current_price:.2f}" ) logger.info( f" SL: ${position.stop_loss:.2f} | " f"TP: ${position.take_profit:.2f} | " f"Leverage: {leverage}x" ) except Exception as e: logger.error(f"❌ Erreur ouverture position {symbol}: {e}") def _manage_pyramiding(self): """Gère le pyramiding sur positions gagnantes""" if not config.V10_PYRAMID_ENABLED: return for symbol, position in self.positions.items(): if position.status != 'OPEN': continue try: current_price = float(self.client.get_ticker(symbol).get('lastPrice', 0)) # Vérifier si doit ajouter should_add, add_size = self.strategy.should_add_to_position( position.side, position.entry_price, current_price, position.pyramids_added ) if should_add: logger.info( f"🔼 Pyramiding {symbol}: Ajouter {add_size*100:.0f}% " f"(PnL: +{position.pnl_pct*100:.1f}%)" ) # Calculer quantité à ajouter add_quantity = position.quantity * add_size if self.mode == 'LIVE': # Placer ordre order_side = 'BUY' if position.side == 'LONG' else 'SELL' self.client.place_market_order(symbol, order_side, add_quantity) # Annuler ancien SL/TP et replacer avec nouvelle quantité self.client.cancel_all_orders(symbol) new_quantity = position.quantity + add_quantity sl_side = 'SELL' if position.side == 'LONG' else 'BUY' self.client.place_stop_loss(symbol, sl_side, new_quantity, position.stop_loss) self.client.place_take_profit(symbol, sl_side, new_quantity, position.take_profit) # Mettre à jour position position.quantity += add_quantity position.pyramids_added += 1 logger.info( f" ✅ Ajouté {add_quantity:.3f} {symbol} " f"(total: {position.quantity:.3f})" ) except Exception as e: logger.error(f"❌ Erreur pyramiding {symbol}: {e}") def _manage_risk(self): """Gère stop-loss et take-profit""" for symbol, position in list(self.positions.items()): if position.status != 'OPEN': continue try: current_price = float(self.client.get_ticker(symbol).get('lastPrice', 0)) # Vérifier SL if position.side == 'LONG' and current_price <= position.stop_loss: logger.warning(f"🛑 Stop-Loss hit: {symbol} @ ${current_price:.2f}") self._close_position(symbol, 'STOP_LOSS') elif position.side == 'SHORT' and current_price >= position.stop_loss: logger.warning(f"🛑 Stop-Loss hit: {symbol} @ ${current_price:.2f}") self._close_position(symbol, 'STOP_LOSS') # Vérifier TP elif position.side == 'LONG' and current_price >= position.take_profit: logger.info(f"🎯 Take-Profit hit: {symbol} @ ${current_price:.2f}") self._close_position(symbol, 'TAKE_PROFIT') elif position.side == 'SHORT' and current_price <= position.take_profit: logger.info(f"🎯 Take-Profit hit: {symbol} @ ${current_price:.2f}") self._close_position(symbol, 'TAKE_PROFIT') except Exception as e: logger.error(f"❌ Erreur risk management {symbol}: {e}") def _close_position(self, symbol: str, reason: str): """Ferme une position""" position = self.positions.get(symbol) if not position: return try: logger.info(f"🔻 Fermeture {symbol} ({reason})...") if self.mode == 'LIVE': # Annuler ordres pendants self.client.cancel_all_orders(symbol) # Fermer position self.client.close_position(symbol, position.side, position.quantity) # Mettre à jour stats position.status = 'CLOSED' if position.pnl > 0: self.stats['wins'] += 1 else: self.stats['losses'] += 1 self.stats['total_pnl'] += position.pnl if self.stats['total_trades'] > 0: self.stats['win_rate'] = self.stats['wins'] / self.stats['total_trades'] # Sauvegarder dans historique trade_record = position.to_dict() trade_record['closed_at'] = datetime.now().isoformat() trade_record['close_reason'] = reason self.trades_history.append(trade_record) # Retirer des positions actives del self.positions[symbol] logger.info( f"✅ Position fermée: {position.side} {symbol} | " f"PnL: {position.pnl_pct*100:+.2f}% (${position.pnl:+.2f})" ) except Exception as e: logger.error(f"❌ Erreur fermeture {symbol}: {e}") def _print_summary(self): """Affiche résumé du status""" logger.info(f"\n{'='*70}") logger.info("📊 RÉSUMÉ") logger.info(f"{'='*70}") logger.info(f"Positions actives: {len(self.positions)}/{self.max_positions}") logger.info(f"Total trades: {self.stats['total_trades']}") logger.info(f"Win rate: {self.stats['win_rate']*100:.1f}% " f"({self.stats['wins']}W / {self.stats['losses']}L)") logger.info(f"Total PnL: ${self.stats['total_pnl']:+.2f}") if self.positions: logger.info(f"\n💼 POSITIONS ACTIVES:") for symbol, pos in self.positions.items(): logger.info( f" {symbol}: {pos.side} {pos.leverage}x | " f"Entry: ${pos.entry_price:.2f} | " f"PnL: {pos.pnl_pct*100:+.2f}%" ) logger.info(f"{'='*70}\n") def _save_state(self): """Sauvegarde l'état actuel""" state = { 'timestamp': datetime.now().isoformat(), 'mode': self.mode, 'positions': {sym: pos.to_dict() for sym, pos in self.positions.items()}, 'stats': self.stats, 'history': self.trades_history[-100:] # Garder 100 derniers trades } # Sauvegarder état with open(self.data_dir / 'current_state.json', 'w') as f: json.dump(state, f, indent=2) # Sauvegarder historique complet with open(self.data_dir / 'trades_history.json', 'w') as f: json.dump(self.trades_history, f, indent=2) def close_all_positions(self, reason: str = 'MANUAL'): """Ferme toutes les positions (pour arrêt d'urgence)""" logger.warning(f"⚠️ Fermeture de toutes les positions ({reason})...") for symbol in list(self.positions.keys()): self._close_position(symbol, reason) logger.info("✅ Toutes les positions fermées") if __name__ == '__main__': logger.info("="*80) logger.info("🤖 APOLLO DEGEN BOT V10 - LIVE TRADING") logger.info("="*80) logger.info("") # Mode paper trading par défaut pour tests mode = input("Mode (PAPER/LIVE) [PAPER]: ").strip().upper() or 'PAPER' if mode == 'LIVE': confirm = input("⚠️ MODE LIVE - Êtes-vous sûr? (yes/no): ").strip().lower() if confirm != 'yes': logger.info("❌ Annulé") exit(0) # Créer trader trader = LiveTraderV10(mode=mode) # Démarrer trader.start() try: logger.info("⏱️ Bot en cours... (Ctrl+C pour arrêter)") while True: time.sleep(1) except KeyboardInterrupt: logger.info("\n⚠️ Arrêt demandé...") finally: trader.stop() # Afficher résumé final logger.info("\n" + "="*80) logger.info("📊 RÉSUMÉ FINAL") logger.info("="*80) logger.info(f"Total trades: {trader.stats['total_trades']}") logger.info(f"Win rate: {trader.stats['win_rate']*100:.1f}%") logger.info(f"Total PnL: ${trader.stats['total_pnl']:+.2f}") logger.info("\n✅ Bot arrêté")