""" 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 v10_range_strategy import V10RangeStrategy 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 BinanceDataProvider: """Fournisseur de données OHLCV depuis Binance API (gratuit, public)""" BASE_URL = "https://api.binance.com/api/v3" @staticmethod def get_klines(symbol: str, interval: str = '1m', limit: int = 200) -> pd.DataFrame: """ Récupère les klines depuis Binance Args: symbol: 'BTCUSDT', 'ETHUSDT', etc. interval: '1m', '5m', '15m', etc. limit: nombre de chandeliers Returns: DataFrame avec colonnes: timestamp, open, high, low, close, volume """ try: url = f"{BinanceDataProvider.BASE_URL}/klines" params = { 'symbol': symbol, 'interval': interval, 'limit': limit } response = requests.get(url, params=params, timeout=10) response.raise_for_status() data = response.json() # Convertir en DataFrame df = pd.DataFrame(data, columns=[ 'timestamp', 'open', 'high', 'low', 'close', 'volume', 'close_time', 'quote_volume', 'trades', 'taker_buy_base', 'taker_buy_quote', 'ignore' ]) # Garder colonnes essentielles et convertir types df = df[['timestamp', 'open', 'high', 'low', 'close', 'volume']] df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') for col in ['open', 'high', 'low', 'close', 'volume']: df[col] = df[col].astype(float) return df except Exception as e: logger.error(f"Erreur récupération klines {symbol}: {e}") return pd.DataFrame() 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): 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 # 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 (sur la margin) self.pnl = self.pnl_pct * self.leverage * self.margin_usd 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 } 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 '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) self.strategy = V10RangeStrategy(leverage=150) # STRATÉGIE OPTIMISÉE: +71% PnL, PF 3.85 self.data_provider = BinanceDataProvider() # É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.005 # Minimum BNB pour gas self.min_balance_usdt = 40 # Minimum USDT pour trading # 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) # 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.error(f"❌ BNB insuffisant (min {self.min_balance_bnb})") return False if usdt_balance < self.min_balance_usdt: logger.error(f"❌ USDT insuffisant (min {self.min_balance_usdt})") return False 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. 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é") # 6. 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 Binance""" trading_pair = self.symbol_map.get(symbol) if not trading_pair: return 0.0 try: url = f"{BinanceDataProvider.BASE_URL}/ticker/price" response = requests.get(url, params={'symbol': trading_pair}, timeout=5) response.raise_for_status() data = response.json() return float(data['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) 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 _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}") # Générer signal (V10RangeStrategy utilise 'analyze') signal_dict = self.strategy.analyze(symbol, df) if signal_dict: # Convertir dict en objet pour accès par attribut from types import SimpleNamespace signal = SimpleNamespace(**signal_dict) # Use leverage and risk from strategy (or fallback to calculation) if signal.leverage is not None and signal.risk_pct is not None: leverage = signal.leverage risk_pct = signal.risk_pct logger.info(f" 📊 Utilise paramètres de la stratégie: {leverage}x leverage, {risk_pct*100:.1f}% risk") else: # Fallback for old strategies leverage, risk_pct = self.strategy.get_leverage_and_risk( signal.score, len(signal.confluences), signal.strategy ) # Afficher infos du signal logger.info( f" 🎯 Signal détecté: {signal.side} | Score: {signal.score:.2f} | " f"Leverage: {leverage}x | Confluences: {len(signal.confluences)}" ) # Vérifier si le signal est assez fort if signal.score >= config.MIN_SIGNAL_SCORE_AGGRESSIVE: 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é (V10 filtre en interne) logger.info(f" ⏸️ {symbol}: Pas de signal (V10 attend conditions RANGE spécifiques)") 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: # 1. Récupérer balances usdt_balance = self.client.get_usdt_balance() bnb_balance = self.client.get_balance() # 2. Calculer taille position (margin) # Pour petits comptes: utiliser quasi tout le capital (mode YOLO/Cross) if usdt_balance < 100: # Garder seulement $2 de réserve pour les frais margin_usd = max(usdt_balance - 2, 0) logger.info(f"💎 Petit compte détecté: Mode FULL DEGEN (${margin_usd:.2f} margin)") else: # Comptes plus gros: utiliser le risk % calculé margin_usd = usdt_balance * risk_pct # Minimum $1 de margin if margin_usd < 1: logger.warning(f"⚠️ Margin trop faible: ${margin_usd:.2f} (min $1)") 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. 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: # Use SL/TP from signal (calculated by strategy) is_long = signal.side == 'LONG' stop_loss = signal.stop_loss take_profit = signal.take_profit logger.info(f"📊 Strategy SL/TP: SL=${stop_loss:.2f} TP=${take_profit:.2f}") # 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: logger.error(f"❌ Échec ouverture position {symbol}") return # 6. Créer objet Position 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 ) # 7. Set SL/TP from signal (calculated by strategy) position.stop_loss = signal.stop_loss position.take_profit = signal.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" 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 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 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'] # Update strategy performance self.strategy.update_performance(position.strategy_name, position.pnl) # 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 au démarrage""" try: 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 is_long (positions.json) ou side (ancien format) is_long = pos_dict.get('is_long', True) side = 'LONG' if is_long else 'SHORT' position = Position( symbol=symbol, side=side, entry_price=pos_dict['entry_price'], margin_usd=pos_dict['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) 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é")