""" Live Trading Engine V10 - ApolloX V2 On-Chain Exécute les signaux de la stratégie inverse V10 en mode on-chain (Web3) """ 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 import requests from dotenv import load_dotenv import os import argparse from apollox_v2_client import ApolloXV2Client, PAIRS from hybrid_strategy import HybridStrategy from dynamic_params_adapter import DynamicParameterAdapter from position_sync import PositionSync from pyth_price_feed import PythPriceFeed # Pyth Network oracle (même source qu'ApolloX on-chain) import config # Load env load_dotenv() 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 on-chain""" def __init__(self, symbol: str, side: str, entry_price: float, margin_usd: float, leverage: int, strategy_name: str, trade_hash: str = None, trailing_pct: float = 0.05): self.symbol = symbol self.side = side # LONG ou SHORT self.entry_price = entry_price self.margin_usd = margin_usd # Margin en USD self.leverage = leverage self.strategy_name = strategy_name self.trade_hash = trade_hash # Hash de la transaction on-chain # Risk management self.stop_loss = None self.take_profit = None self.pyramids_added = 0 # Trailing stop (5% par défaut - meilleur config trouvée) self.trailing_pct = trailing_pct self.max_pnl_pct = 0.0 # Track le PnL max atteint self.trailing_stop_active = False # 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 et le trailing stop""" 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 (sur la margin) self.pnl = self.pnl_pct * self.leverage * self.margin_usd # Trailing stop: Track le max PnL atteint if self.pnl_pct > self.max_pnl_pct: self.max_pnl_pct = self.pnl_pct if self.max_pnl_pct > 0: self.trailing_stop_active = True def should_close_trailing(self) -> bool: """Vérifie si le trailing stop est touché""" if not self.trailing_stop_active or self.max_pnl_pct <= 0: return False # Si le PnL a reculé de trailing_pct% depuis le max pullback = (self.max_pnl_pct - self.pnl_pct) / self.max_pnl_pct return pullback >= self.trailing_pct def to_dict(self) -> Dict: """Convertit en dict pour sauvegarde""" return { 'symbol': self.symbol, 'side': self.side, 'entry_price': self.entry_price, 'margin_usd': self.margin_usd, 'leverage': self.leverage, 'strategy_name': self.strategy_name, 'trade_hash': self.trade_hash, '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, 'trailing_pct': self.trailing_pct, 'max_pnl_pct': self.max_pnl_pct, 'trailing_stop_active': self.trailing_stop_active } class LiveTraderV10OnChain: """ Live Trading Engine V10 - On-Chain Version Features: - Trading on-chain via smart contracts (Web3) - Génération de signaux inverse V10 - Exécution automatique des trades - Stop-loss et take-profit gérés en code - Gestion multi-positions (future) - Tracking performance """ def __init__(self, symbols: List[str] = None, mode: str = 'LIVE'): """ Args: symbols: Liste des symboles à trader (ex: ['BTCUSD_DEGEN']) mode: LIVE ou PAPER (paper trading sans ordres réels) """ self.mode = mode self.symbols = symbols or ['BTCUSD_DEGEN'] # Par défaut BTCUSD Degen Mode # Map symbol to trading pair for price data self.symbol_map = { 'BTC': 'BTCUSDT', 'BTCUSD_DEGEN': 'BTCUSDT', # Degen Mode utilise prix BTC Binance '500BTC/USD_DEGEN': 'BTCUSDT', # Degen Mode avec leverage prefix 'ETH': 'ETHUSDT', 'SOL': 'SOLUSDT', 'BNB': 'BNBUSDT' } # Clients private_key = os.getenv('WALLET_PRIVATE_KEY') if not private_key: raise ValueError("WALLET_PRIVATE_KEY non définie dans .env") self.client = ApolloXV2Client(private_key=private_key) # ========== PROTECTION CONTRE SPAM TX INVALIDES ========== self.failed_attempts = {} # {symbol: count} self.consecutive_failures = 0 # Échecs consécutifs globaux self.MIN_BNB_REQUIRED = 0.0001 # Minimum très bas self.MAX_FAILED_ATTEMPTS_PER_SYMBOL = 2 # Max 2 essais par symbol self.MAX_CONSECUTIVE_FAILURES = 3 # Stop après 3 échecs consécutifs # ========== STRATÉGIE HYBRIDE OPTIMISÉE ========== # Charge paramètres optimisés depuis optimal_hybrid_params.json import json params_path = os.path.join(os.path.dirname(__file__), 'optimal_hybrid_params.json') with open(params_path) as f: optimal_params = json.load(f) logger.info("📊 Chargement stratégie hybride optimisée:") logger.info(f" • RANGE: {optimal_params['range_strategy']['performance']}") logger.info(f" • BULL: {optimal_params['bull_strategy']['performance']}") logger.info(f" • BEAR: {optimal_params['bear_strategy']['name']}") # Initialiser stratégie hybride (détection automatique BULL/BEAR/RANGE) self.strategy = HybridStrategy(leverage=200, use_llm=False) # Appliquer paramètres optimisés à chaque sous-stratégie # 1. V10 Range Strategy (déjà optimale) range_params = optimal_params['range_strategy']['params'] for key, value in range_params.items(): if hasattr(self.strategy.range_strategy, 'params'): self.strategy.range_strategy.params[key] = value else: setattr(self.strategy.range_strategy, key, value) # 2. Bull Strategy (optimisée sur 14j) bull_params = optimal_params['bull_strategy']['params'] self.strategy.bull_strategy.min_trend_strength = bull_params['min_trend_strength'] self.strategy.bull_strategy.rsi_pullback_max = bull_params['rsi_pullback_max'] self.strategy.bull_strategy.rsi_breakout_min = bull_params['rsi_breakout_min'] self.strategy.bull_strategy.volume_spike_threshold = bull_params['volume_spike_threshold'] self.strategy.bull_strategy.sl_atr_multiplier = bull_params['sl_atr_multiplier'] self.strategy.bull_strategy.tp_atr_multiplier = bull_params['tp_atr_multiplier'] self.strategy.bull_strategy.trailing_pct = bull_params['trailing_pct'] self.strategy.bull_strategy.min_confidence = bull_params['min_confidence'] self.strategy.bull_strategy.leverage = bull_params['leverage'] # 3. Bear Strategy (params conservateurs) bear_params = optimal_params['bear_strategy']['params'] self.strategy.bear_strategy.min_trend_strength = bear_params['min_trend_strength'] self.strategy.bear_strategy.rsi_rally_min = bear_params['rsi_rally_min'] self.strategy.bear_strategy.rsi_overbought = bear_params['rsi_overbought'] self.strategy.bear_strategy.volume_spike_threshold = bear_params['volume_spike_threshold'] self.strategy.bear_strategy.sl_atr_multiplier = bear_params['sl_atr_multiplier'] self.strategy.bear_strategy.tp_atr_multiplier = bear_params['tp_atr_multiplier'] self.strategy.bear_strategy.trailing_pct = bear_params['trailing_pct'] self.strategy.bear_strategy.min_confidence = bear_params['min_confidence'] self.strategy.bear_strategy.leverage = bear_params['leverage'] logger.info("✅ Stratégie hybride initialisée avec paramètres optimaux") # Adaptation dynamique ATR base_params = { 'trailing_pct': 0.05, # 5% trailing stop (base) 'rsi_oversold': 35, 'rsi_overbought': 78, 'sl_atr_mult': 1.64 } self.dynamic_adapter = DynamicParameterAdapter(base_params) self.data_provider = PythPriceFeed() # Pyth Network (même oracle qu'ApolloX on-chain) # État self.positions: Dict[str, Position] = {} # symbol -> Position self.is_running = False self.thread = None # Configuration self.check_interval = 30 # Vérifier toutes les 30 secondes (gas cost) self.max_positions = 1 # 1 position à la fois pour commencer self.min_balance_bnb = 0.00001 # Minimum très bas self.min_balance_usdt = 0.1 # Minimum très bas # Tracking self.trades_history = [] self.stats = { 'total_trades': 0, 'wins': 0, 'losses': 0, 'total_pnl': 0.0, 'win_rate': 0.0 } # Retournement de position (flip) self.position_flips = {} # symbol -> {'last_flip_time': timestamp, 'flip_count_today': int} self.max_flips_per_day = 3 # Max 3 retournements par jour par symbole self.min_flip_interval = 300 # 5 minutes minimum entre retournements (secondes) # Sauvegarde self.data_dir = Path('live_trading_data') self.data_dir.mkdir(exist_ok=True) # Position Sync - Monitoring on-chain self.position_sync = PositionSync() # Charger les positions ouvertes depuis positions.json au démarrage self._load_open_positions() logger.info(f"🚀 LiveTraderV10OnChain initialisé ({mode} mode)") logger.info(f" Wallet: {self.client.account.address}") logger.info(f" Symboles: {', '.join(self.symbols)}") logger.info(f" Max positions: {self.max_positions}") logger.info(f" Positions ouvertes chargées: {len(self.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é self.can_open_positions = self._safety_checks() if not self.can_open_positions: if len(self.positions) > 0: logger.warning("⚠️ Safety checks échoués - Mode CLOSE-ONLY (peut fermer positions mais pas en ouvrir)") else: logger.error("❌ Safety checks échoués et aucune position - bot en attente") self.is_running = True self.thread = threading.Thread(target=self._run_loop, daemon=True) self.thread.start() if self.can_open_positions: logger.info(f"✅ Trading démarré en mode {self.mode}") else: logger.info(f"⚠️ Bot démarré en mode CLOSE-ONLY") 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 wallet try: bnb_balance = self.client.get_balance() usdt_balance = self.client.get_usdt_balance() logger.info(f" ✅ Wallet: {self.client.account.address}") logger.info(f" ✅ BNB: {bnb_balance:.6f}") logger.info(f" ✅ USDT: {usdt_balance:.2f}") if bnb_balance < self.min_balance_bnb: logger.warning(f"⚠️ BNB faible (min {self.min_balance_bnb}) - Continue quand même") if usdt_balance < self.min_balance_usdt: logger.warning(f"⚠️ USDT faible (min {self.min_balance_usdt}) - Continue quand même") except Exception as e: logger.error(f"❌ Erreur wallet: {e}") return False # 2. Vérifier stratégie if not self.strategy: logger.error("❌ Stratégie non initialisée") return False logger.info(" ✅ Stratégie V10 prête") # 3. Vérifier data provider try: test_df = self.data_provider.get_klines('BTCUSDT', '1m', limit=10) if test_df.empty: logger.error("❌ Impossible de récupérer données de prix") return False logger.info(" ✅ Data provider OK") except Exception as e: logger.error(f"❌ Erreur data provider: {e}") return False 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 (seulement si safety checks OK) if self.can_open_positions: if len(self.positions) < self.max_positions: self._check_new_signals() else: logger.info(f"⚠️ Max positions atteint ({self.max_positions})") else: logger.info(f"⏸️ Mode CLOSE-ONLY: Pas de recherche de nouveaux signaux") # 3. Gérer stop-loss et take-profit self._manage_risk() # 4. Vérifier signaux de fermeture manuelle (depuis web interface) self._check_manual_close_signals() # 5. Synchroniser positions on-chain à CHAQUE cycle (détection instantanée positions manuelles) logger.info("🔍 Synchronisation positions on-chain...") self.position_sync.sync_positions() # Recharger positions après sync self.positions.clear() self._load_open_positions() # 6. Re-vérifier balance toutes les 10 itérations (~5 minutes) if iteration % 10 == 0: previous_status = self.can_open_positions self.can_open_positions = self._safety_checks() if not previous_status and self.can_open_positions: logger.info("✅ Balance suffisante - Mode trading activé!") elif previous_status and not self.can_open_positions: logger.warning("⚠️ Balance insuffisante - Mode CLOSE-ONLY activé") # 7. Afficher résumé self._print_summary() # 7. Sauvegarder état if iteration % 2 == 0: # Toutes les 2 itérations self._save_state() # Sleep if self.is_running: time.sleep(self.check_interval) except Exception as e: logger.error(f"❌ Erreur dans loop: {e}", exc_info=True) time.sleep(60) def _get_current_price(self, symbol: str) -> float: """Récupère le prix actuel depuis Pyth Network (oracle on-chain)""" trading_pair = self.symbol_map.get(symbol) if not trading_pair: return 0.0 try: price = self.data_provider.get_current_price(trading_pair) if price is None: logger.error(f"Erreur récupération prix {symbol} depuis Pyth") return 0.0 return price except Exception as e: logger.error(f"Erreur récupération prix {symbol}: {e}") return 0.0 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 current_price = self._get_current_price(symbol) if current_price == 0: logger.warning(f"⚠️ Prix invalide pour {symbol}") continue # Mettre à jour PnL position.update_pnl(current_price) # Afficher info avec trailing stop status trailing_info = f" | Max: {position.max_pnl_pct*100:+.2f}%" if position.trailing_stop_active else "" logger.info( f" {symbol}: {position.side} @ ${position.entry_price:.2f} " f"→ ${current_price:.2f} | PnL: {position.pnl_pct*100:+.2f}% " f"(${position.pnl:+.2f}){trailing_info}" ) # ⚡ TRAILING STOP CHECK if position.should_close_trailing(): logger.info(f" 🎯 TRAILING STOP DÉCLENCHÉ!") logger.info(f" Max PnL: {position.max_pnl_pct*100:+.2f}%") logger.info(f" PnL actuel: {position.pnl_pct*100:+.2f}%") logger.info(f" Pullback: {((position.max_pnl_pct - position.pnl_pct)/position.max_pnl_pct)*100:.1f}% ≥ {position.trailing_pct*100:.0f}%") self._close_position(symbol, reason='TRAILING_STOP') continue except Exception as e: logger.error(f"❌ Erreur update {symbol}: {e}") def _can_flip_position(self, symbol: str) -> bool: """Vérifie si on peut retourner la position (anti-loop)""" now = datetime.now() if symbol not in self.position_flips: return True flip_data = self.position_flips[symbol] # Reset compteur si nouveau jour if flip_data.get('last_reset_day') != now.date(): flip_data['flip_count_today'] = 0 flip_data['last_reset_day'] = now.date() # Vérifier limite journalière if flip_data.get('flip_count_today', 0) >= self.max_flips_per_day: logger.warning(f"⚠️ {symbol}: Limite de {self.max_flips_per_day} retournements/jour atteinte") return False # Vérifier intervalle minimum last_flip = flip_data.get('last_flip_time') if last_flip: elapsed = (now - last_flip).total_seconds() if elapsed < self.min_flip_interval: remaining = int(self.min_flip_interval - elapsed) logger.info(f" ⏱️ {symbol}: Cooldown retournement ({remaining}s restants)") return False return True def _record_flip(self, symbol: str): """Enregistre un retournement de position""" now = datetime.now() if symbol not in self.position_flips: self.position_flips[symbol] = { 'flip_count_today': 0, 'last_reset_day': now.date() } self.position_flips[symbol]['last_flip_time'] = now self.position_flips[symbol]['flip_count_today'] += 1 logger.info(f"📊 {symbol}: Retournement {self.position_flips[symbol]['flip_count_today']}/{self.max_flips_per_day} aujourd'hui") def _check_new_signals(self): """Vérifie s'il y a de nouveaux signaux à trader (+ retournement si position existante)""" logger.info("🔍 Recherche nouveaux signaux...") for symbol in self.symbols: try: # Récupérer données depuis Binance trading_pair = self.symbol_map.get(symbol) if not trading_pair: logger.warning(f"⚠️ Pas de mapping pour {symbol}") continue # Récupérer prix actuel pour info current_price = self._get_current_price(symbol) df = self.data_provider.get_klines( trading_pair, config.TIMEFRAME_PRIMARY, limit=200 ) if df.empty or len(df) < 100: logger.warning(f"⚠️ Données insuffisantes pour {symbol}") continue # Log prix actuel (la stratégie V10 détecte le régime en interne) logger.info(f" 📊 {symbol}: Prix ${current_price:,.2f}") # ⚡ ADAPTATION DYNAMIQUE ATR - Ajuster params selon volatilité adapted_params = self.dynamic_adapter.adapt_params(df) # Appliquer adaptation dynamique à la sous-stratégie range if hasattr(self.strategy.range_strategy, 'params'): self.strategy.range_strategy.params['rsi_oversold'] = adapted_params['rsi_oversold'] self.strategy.range_strategy.params['rsi_overbought'] = adapted_params['rsi_overbought'] self.strategy.range_strategy.params['sl_atr_multiplier'] = adapted_params['sl_atr_mult'] # Stocker trailing adapté pour l'utiliser lors de l'ouverture de position self.current_trailing_pct = adapted_params['trailing_pct'] logger.info(f" ⚡ ATR: {adapted_params['atr_pct']:.2f}% | Regime: {adapted_params['volatility_regime']} | Trail: {adapted_params['trailing_pct']:.2%}") # Générer signal (HybridStrategy retourne HybridSignal) hybrid_signal = self.strategy.analyze(symbol, df) if hybrid_signal: # HybridSignal est un dataclass avec tous les attributs nécessaires # On l'adapte au format attendu par le reste du code from types import SimpleNamespace # Adapter HybridSignal au format legacy signal = SimpleNamespace( side=hybrid_signal.side, entry_price=hybrid_signal.entry_price, stop_loss=hybrid_signal.stop_loss, take_profit=hybrid_signal.take_profit, leverage=hybrid_signal.leverage, risk_pct=None, # Non utilisé par HybridStrategy score=hybrid_signal.confidence, # confidence = score confluences=[hybrid_signal.reason], # reason comme confluence unique strategy=hybrid_signal.strategy_used, market_regime=hybrid_signal.market_regime ) # HybridStrategy fournit déjà leverage optimisé leverage = signal.leverage risk_pct = 0.20 # 20% de la balance par position # Afficher infos du signal logger.info( f" 🎯 Signal détecté: {signal.side} | Confidence: {signal.score:.0f}% | " f"Leverage: {leverage}x | Stratégie: {signal.strategy}" ) logger.info(f" 📈 Régime: {signal.market_regime} | {hybrid_signal.reason}") # Vérifier si le signal est assez fort (60% confidence minimum) if signal.score >= 60: logger.info(f" ✅ Signal validé (score ≥ {config.MIN_SIGNAL_SCORE_AGGRESSIVE})") logger.info(f" 📝 Stratégie: {signal.strategy}") # RETOURNEMENT DE POSITION: Vérifier si on a déjà une position contraire if symbol in self.positions: current_position = self.positions[symbol] # Signal contraire? is_contrary = (current_position.side == 'LONG' and signal.side == 'SHORT') or \ (current_position.side == 'SHORT' and signal.side == 'LONG') if is_contrary: # Calculer PnL actuel current_price = self._get_current_price(symbol) current_position.update_pnl(current_price) # PnL positif? if current_position.pnl > 0: # Peut-on retourner? (anti-loop) if self._can_flip_position(symbol): logger.info(f"🔄 RETOURNEMENT DE POSITION DÉTECTÉ!") logger.info(f" Position actuelle: {current_position.side} @ ${current_position.entry_price:.2f}") logger.info(f" PnL actuel: +${current_position.pnl:.2f} ({current_position.pnl_pct*100:+.2f}%)") logger.info(f" Nouveau signal: {signal.side}") logger.info(f" → Fermeture + Ouverture inverse") # Enregistrer le flip self._record_flip(symbol) # Fermer position actuelle self._close_position(symbol, reason='FLIP_PROFIT') # Ouvrir nouvelle position inverse self._open_position(symbol, signal, leverage, risk_pct) else: logger.info(f" ⏸️ Signal contraire mais cooldown actif, position gardée") else: logger.info(f" ⏸️ Signal contraire mais PnL négatif (${current_position.pnl:.2f}), position gardée") else: logger.info(f" ⏸️ Position {current_position.side} déjà ouverte, signal {signal.side} ignoré") else: # Pas de position existante, ouvrir normalement self._open_position(symbol, signal, leverage, risk_pct) else: logger.info( f" ❌ Signal rejeté: Score {signal.score:.2f} < {config.MIN_SIGNAL_SCORE_AGGRESSIVE} " f"(seuil minimum)" ) else: # Pas de signal généré par stratégie hybride logger.info(f" ⏸️ {symbol}: Pas de signal (stratégie hybride attend conditions favorables)") except Exception as e: logger.error(f"❌ Erreur signal {symbol}: {e}", exc_info=True) def _open_position(self, symbol: str, signal, leverage: int, risk_pct: float): """Ouvre une nouvelle position on-chain""" try: # ========== PROTECTIONS ANTI-SPAM TX ========== # 1. Vérifier si trop d'échecs consécutifs globaux if self.consecutive_failures >= self.MAX_CONSECUTIVE_FAILURES: logger.error(f"🛑 STOP: {self.consecutive_failures} échecs consécutifs - BOT DÉSACTIVÉ") logger.error(f" → Vérifiez les logs et corrigez avant de redémarrer") self.is_running = False return # 2. Vérifier si ce symbol a déjà échoué trop de fois if self.failed_attempts.get(symbol, 0) >= self.MAX_FAILED_ATTEMPTS_PER_SYMBOL: logger.warning(f"⏭️ {symbol}: Skip (déjà {self.failed_attempts[symbol]} échecs)") return # 3. Vérifier BNB minimum AVANT toute tentative bnb_balance = self.client.get_balance() if bnb_balance < self.MIN_BNB_REQUIRED: logger.warning(f"⚠️ BNB faible: {bnb_balance:.6f} < {self.MIN_BNB_REQUIRED}") # Ne pas stopper, juste warning pass # ========== FIN PROTECTIONS ========== # 4. Récupérer balances usdt_balance = self.client.get_usdt_balance() # 5. Calculer taille position: TOUJOURS 20% de la balance POSITION_SIZE_PCT = 0.20 # 20% de la balance par position margin_usd = usdt_balance * POSITION_SIZE_PCT logger.info(f"💰 Calcul margin: ${usdt_balance:.2f} × {POSITION_SIZE_PCT*100:.0f}% = ${margin_usd:.2f}") # Minimum $1 de margin if margin_usd < 1: logger.warning(f"⚠️ Margin trop faible: ${margin_usd:.2f} (min $1)") logger.warning(f" Balance USDT: ${usdt_balance:.2f}") return # 3. Prix d'entrée current_price = self._get_current_price(symbol) if current_price == 0: logger.error(f"❌ Prix invalide pour {symbol}") return logger.info( f"💰 Position: ${margin_usd:.2f} margin | {leverage}x leverage | " f"@ ${current_price:.2f}" ) # 4. Vérifier gas if bnb_balance < 0.002: logger.warning(f"⚠️ BNB faible pour gas: {bnb_balance:.6f}") # 5. Calculer SL/TP - HYBRIDE: Signal + Limites Capital is_long = signal.side == 'LONG' # ========== ÉTAPE 1: UTILISER SL/TP DU SIGNAL ========== # Le signal fournit des SL/TP basés sur ATR (volatilité du marché) if hasattr(signal, 'stop_loss') and hasattr(signal, 'take_profit') and signal.stop_loss and signal.take_profit: # Utiliser les valeurs du signal (stratégie BULL/BEAR/RANGE) stop_loss_signal = signal.stop_loss take_profit_signal = signal.take_profit # Calculer % mouvement prix sl_pct_signal = abs((stop_loss_signal - current_price) / current_price) tp_pct_signal = abs((take_profit_signal - current_price) / current_price) logger.info(f"📊 SL/TP du signal (ATR-based):") logger.info(f" SL: ${stop_loss_signal:.2f} ({sl_pct_signal*100:.3f}% prix)") logger.info(f" TP: ${take_profit_signal:.2f} ({tp_pct_signal*100:.3f}% prix)") else: # Fallback: Pas de SL/TP dans signal, calculer par défaut CAPITAL_RISK_PCT = 0.05 price_risk_pct = CAPITAL_RISK_PCT / leverage sl_pct_signal = price_risk_pct tp_pct_signal = max(price_risk_pct * 2, 0.003) stop_loss_signal = current_price * (1 - sl_pct_signal) if is_long else current_price * (1 + sl_pct_signal) take_profit_signal = current_price * (1 + tp_pct_signal) if is_long else current_price * (1 - tp_pct_signal) logger.warning(f"⚠️ Signal sans SL/TP, utilisation valeurs par défaut") # ========== ÉTAPE 2: APPLIQUER LIMITES CAPITAL ========== # Calculer impact en % capital sl_capital_pct = sl_pct_signal * leverage tp_capital_pct = tp_pct_signal * leverage # Limites de sécurité MAX_SL_CAPITAL = 0.10 # 10% capital max de perte MAX_TP_CAPITAL = 0.30 # 30% capital max de gain MIN_TP_PRICE = 0.003 # 0.3% prix minimum (protection fees) # Ajuster SL si trop risqué if sl_capital_pct > MAX_SL_CAPITAL: logger.warning(f"⚠️ SL capital trop élevé: {sl_capital_pct*100:.1f}% > {MAX_SL_CAPITAL*100:.0f}%") sl_pct_adjusted = MAX_SL_CAPITAL / leverage stop_loss = current_price * (1 - sl_pct_adjusted) if is_long else current_price * (1 + sl_pct_adjusted) logger.warning(f" → Réduit à {sl_pct_adjusted*100:.3f}% prix ({MAX_SL_CAPITAL*100:.0f}% capital)") else: stop_loss = stop_loss_signal sl_pct_adjusted = sl_pct_signal # Ajuster TP si trop élevé ou trop bas if tp_capital_pct > MAX_TP_CAPITAL: logger.warning(f"⚠️ TP capital trop élevé: {tp_capital_pct*100:.1f}% > {MAX_TP_CAPITAL*100:.0f}%") tp_pct_adjusted = MAX_TP_CAPITAL / leverage take_profit = current_price * (1 + tp_pct_adjusted) if is_long else current_price * (1 - tp_pct_adjusted) logger.warning(f" → Réduit à {tp_pct_adjusted*100:.3f}% prix ({MAX_TP_CAPITAL*100:.0f}% capital)") elif tp_pct_signal < MIN_TP_PRICE: logger.warning(f"⚠️ TP trop bas: {tp_pct_signal*100:.3f}% < {MIN_TP_PRICE*100:.2f}%") tp_pct_adjusted = MIN_TP_PRICE take_profit = current_price * (1 + tp_pct_adjusted) if is_long else current_price * (1 - tp_pct_adjusted) logger.warning(f" → Augmenté à {tp_pct_adjusted*100:.2f}% prix minimum") else: take_profit = take_profit_signal tp_pct_adjusted = tp_pct_signal # ========== ÉTAPE 3: PROTECTION ANTI-FEES ========== estimated_open_fee = 0.50 estimated_close_fee = 0.75 total_fees_usd = estimated_open_fee + estimated_close_fee # ~1.25 USDT # Calculer profit potentiel en USDT profit_pct_with_leverage = tp_pct_adjusted * leverage potential_profit_usd = margin_usd * profit_pct_with_leverage # Profit minimum = 3x les fees min_profit_needed = total_fees_usd * 3 # Vérifier si le trade vaut le coup if potential_profit_usd < min_profit_needed: logger.warning(f"⚠️ TRADE SKIPPÉ - Profit insuffisant vs fees:") logger.warning(f" Profit potentiel: ${potential_profit_usd:.2f}") logger.warning(f" Fees estimés: ${total_fees_usd:.2f}") logger.warning(f" Minimum requis (3x fees): ${min_profit_needed:.2f}") logger.warning(f" → Augmentez la marge ou attendez un meilleur signal") return # ========== RÉSUMÉ FINAL ========== logger.info(f"") logger.info(f"{'='*70}") logger.info(f"📊 SL/TP FINAL (Leverage {leverage}x)") logger.info(f"{'='*70}") logger.info(f"Entrée: ${current_price:.2f}") logger.info(f"SL: ${stop_loss:.2f} ({sl_pct_adjusted*100:.3f}% prix | {sl_pct_adjusted*leverage*100:.1f}% capital)") logger.info(f"TP: ${take_profit:.2f} ({tp_pct_adjusted*100:.3f}% prix | {tp_pct_adjusted*leverage*100:.1f}% capital)") logger.info(f"Ratio: {tp_pct_adjusted/sl_pct_adjusted:.1f}:1 (reward:risk)") logger.info(f"") logger.info(f"💰 Profit potentiel: ${potential_profit_usd:.2f}") logger.info(f"💸 Fees estimés: ${total_fees_usd:.2f}") logger.info(f"📈 Ratio profit/fees: {potential_profit_usd/total_fees_usd:.1f}x") logger.info(f"✅ Trade validé (profit > {min_profit_needed:.2f})") logger.info(f"{'='*70}") logger.info(f"") # 6. Ouvrir position on-chain if self.mode == 'PAPER': logger.info("📝 PAPER MODE - Ordre simulé") trade_hash = 'PAPER_' + datetime.now().strftime('%Y%m%d%H%M%S') else: # Ouvrir position via smart contract logger.info(f"📡 Envoi transaction on-chain...") trade_hash = self.client.open_position( pair=symbol, is_long=is_long, margin_usd=margin_usd, leverage=leverage, price=current_price, stop_loss=stop_loss, take_profit=take_profit ) if not trade_hash: # ÉCHEC: Incrémenter compteurs self.failed_attempts[symbol] = self.failed_attempts.get(symbol, 0) + 1 self.consecutive_failures += 1 logger.error(f"❌ Échec ouverture position {symbol}") logger.error(f" Tentatives {symbol}: {self.failed_attempts[symbol]}/{self.MAX_FAILED_ATTEMPTS_PER_SYMBOL}") logger.error(f" Échecs consécutifs: {self.consecutive_failures}/{self.MAX_CONSECUTIVE_FAILURES}") return # SUCCÈS: Reset compteurs self.consecutive_failures = 0 if symbol in self.failed_attempts: del self.failed_attempts[symbol] logger.info(f"✅ Position ouverte avec succès (échecs reset)") # 6. Créer objet Position avec trailing stop dynamique trailing_pct = getattr(self, 'current_trailing_pct', 0.05) # Utiliser trailing adapté ou 5% par défaut position = Position( symbol=symbol, side=signal.side, entry_price=current_price, margin_usd=margin_usd, leverage=leverage, strategy_name=signal.strategy, trade_hash=trade_hash, trailing_pct=trailing_pct ) # 7. Set SL/TP recalculés (ajustés pour leverage) position.stop_loss = stop_loss position.take_profit = take_profit # 8. Sauvegarder position self.positions[symbol] = position self.stats['total_trades'] += 1 # Calculer les % pour l'affichage (depuis le signal) sl_pct_display = abs(position.stop_loss - current_price) / current_price * 100 tp_pct_display = abs(position.take_profit - current_price) / current_price * 100 logger.info( f"✅ Position ouverte: {signal.side} {symbol} " f"@ ${current_price:.2f} | {leverage}x" ) logger.info( f" Margin: ${margin_usd:.2f}" ) logger.info( f" SL: ${position.stop_loss:.2f} ({sl_pct_display:.3f}%)" ) logger.info( f" TP: ${position.take_profit:.2f} ({tp_pct_display:.3f}%)" ) logger.info( f" ⚡ Trailing Stop: {trailing_pct*100:.0f}% (adapté dynamiquement)" ) logger.info( f" Trade Hash: {trade_hash}" ) except Exception as e: logger.error(f"❌ Erreur ouverture position {symbol}: {e}", exc_info=True) def _check_manual_close_signals(self): """Vérifie et traite les signaux de fermeture manuelle depuis la web interface""" signal_file = Path('close_position_signal.json') if not signal_file.exists(): return try: # Lire le signal with open(signal_file, 'r') as f: signal = json.load(f) symbol = signal.get('symbol') if not symbol: logger.warning("⚠️ Signal de fermeture invalide: pas de symbol") signal_file.unlink() # Supprimer le fichier invalide return # Vérifier que la position existe if symbol not in self.positions: logger.warning(f"⚠️ Aucune position ouverte pour {symbol}") signal_file.unlink() return # Fermer la position logger.info(f"🖱️ FERMETURE MANUELLE demandée pour {symbol} (depuis web interface)") self._close_position(symbol, reason='MANUAL_WEB') # Supprimer le fichier de signal après traitement signal_file.unlink() logger.info(f"✅ Signal de fermeture traité et supprimé") except Exception as e: logger.error(f"❌ Erreur traitement signal de fermeture: {e}") # En cas d'erreur, supprimer quand même le fichier pour éviter la boucle if signal_file.exists(): signal_file.unlink() 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 = self._get_current_price(symbol) if current_price == 0: continue # Vérifier SL (seulement si défini) if position.stop_loss is not None: 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 (seulement si défini) if position.take_profit is not None: if 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 on-chain""" position = self.positions.get(symbol) if not position: return try: logger.info(f"🔻 Fermeture {symbol} ({reason})...") if self.mode == 'LIVE': # Fermer position via smart contract logger.info(f"📡 Envoi transaction close...") success = self.client.close_position(position.trade_hash) if not success: logger.error(f"❌ Échec fermeture position {symbol}") # CRITICAL FIX: Incrémenter compteur d'échecs if not hasattr(position, 'close_failures'): position.close_failures = 0 position.close_failures += 1 # Après 3 échecs, retirer la position pour éviter spam if position.close_failures >= 3: logger.error(f"🛑 {symbol}: 3 échecs de fermeture - ABANDON de la position") logger.error(f" Position probablement déjà fermée on-chain ou invalide") del self.positions[symbol] # RETIRER POUR ÉVITER BOUCLE INFINIE return # 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'] # Note: HybridStrategy tracks its own stats via self.strategy.stats # update_performance() n'existe pas dans HybridStrategy # 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}", exc_info=True) def _print_summary(self): """Affiche résumé du status""" # Récupérer balances try: bnb = self.client.get_balance() usdt = self.client.get_usdt_balance() except: bnb, usdt = 0, 0 logger.info(f"\n{'='*70}") logger.info("📊 RÉSUMÉ") logger.info(f"{'='*70}") logger.info(f"Wallet: BNB {bnb:.6f} | USDT ${usdt:.2f}") 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}% (${pos.pnl:+.2f})" ) logger.info(f"{'='*70}\n") def _load_open_positions(self): """Charge les positions ouvertes depuis positions.json""" try: # NE PAS sync ici - déjà fait dans la boucle principale if not Path('positions.json').exists(): logger.info("📂 Aucun fichier positions.json trouvé") return with open('positions.json', 'r') as f: data = json.load(f) # Charger seulement les positions OUVERTES for trade_hash, pos_dict in data.get('positions', {}).items(): if pos_dict.get('status') == 'OPEN': # Créer objet Position depuis le dict symbol = pos_dict['symbol'] # Gérer les 2 formats: on-chain (is_long, margin) vs bot (side, margin_usd) if 'side' in pos_dict: # Format bot side = pos_dict['side'] margin = pos_dict.get('margin_usd', 0) else: # Format on-chain is_long = pos_dict.get('is_long', True) side = 'LONG' if is_long else 'SHORT' margin = pos_dict.get('margin', 0) position = Position( symbol=symbol, side=side, entry_price=pos_dict['entry_price'], margin_usd=margin, leverage=pos_dict['leverage'], strategy_name=pos_dict.get('strategy_name', 'UNKNOWN'), trade_hash=trade_hash ) position.stop_loss = pos_dict.get('stop_loss') position.take_profit = pos_dict.get('take_profit') position.opened_at = datetime.fromisoformat(pos_dict['opened_at']) self.positions[symbol] = position logger.info(f"📥 Position ouverte chargée: {symbol} {position.side}") except Exception as e: logger.error(f"❌ Erreur chargement positions: {e}") def _save_state(self): """Sauvegarde l'état actuel""" state = { 'timestamp': datetime.now().isoformat(), 'mode': self.mode, 'wallet': self.client.account.address, '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_onchain.json', 'w') as f: json.dump(state, f, indent=2) # Sauvegarder historique complet with open(self.data_dir / 'trades_history_onchain.json', 'w') as f: json.dump(self.trades_history, f, indent=2) # Sauvegarder aussi dans positions.json pour le web monitor positions_data = { 'wallet': self.client.account.address, 'last_updated': datetime.now().isoformat(), 'positions': {pos.trade_hash: pos.to_dict() for sym, pos in self.positions.items()} } with open('positions.json', 'w') as f: json.dump(positions_data, 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__': # Parse command-line arguments parser = argparse.ArgumentParser(description='Apollo Degen Bot V10 - Live Trading On-Chain') parser.add_argument('--mode', type=str, default='PAPER', choices=['PAPER', 'LIVE'], help='Trading mode: PAPER (simulation) or LIVE (real on-chain trades)') parser.add_argument('--symbols', type=str, default='BTC', help='Comma-separated list of symbols to trade (e.g., BTC,ETH,SOL)') parser.add_argument('--confirm-live', action='store_true', help='Required flag to confirm LIVE mode (safety check)') args = parser.parse_args() logger.info("="*80) logger.info("🤖 APOLLO DEGEN BOT V10 - LIVE TRADING ON-CHAIN") logger.info("="*80) logger.info("") # Parse mode mode = args.mode.upper() # Safety check for LIVE mode if mode == 'LIVE' and not args.confirm_live: logger.error("❌ LIVE mode requires --confirm-live flag for safety") logger.error(" Usage: python3 live_trader_v10_onchain.py --mode LIVE --confirm-live") exit(1) # Parse symbols symbols = [s.strip().upper() for s in args.symbols.split(',')] logger.info(f"📊 Mode: {mode}") logger.info(f"📈 Symbols: {', '.join(symbols)}") logger.info("") # Créer trader trader = LiveTraderV10OnChain(symbols=symbols, 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é")