""" Bot de Trading Crypto avec Ordres Automatiques =============================================== ⚠️ ATTENTION: Ce bot peut passer des ordres RÉELS si TESTNET_MODE = False Commencer TOUJOURS en mode TESTNET ! """ import os import sys import asyncio import json import time import hmac import hashlib from datetime import datetime from collections import deque from urllib.parse import urlencode import numpy as np try: import websockets import requests except ImportError: import subprocess subprocess.check_call(['pip', 'install', 'websockets', 'requests']) import websockets import requests from config import * # ═══════════════════════════════════════════════════════════════════════════════ # DYNAMIC SL/TP - Gestion IA des Stop-Loss et Take-Profit # ═══════════════════════════════════════════════════════════════════════════════ DYNAMIC_SLTP_ENABLED = False try: import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from dynamic_sltp import calculate_optimal_sltp, get_dynamic_sltp_calculator from market_regime import get_market_regime_detector DYNAMIC_SLTP_ENABLED = True print("✅ Dynamic SL/TP IA activé") except ImportError as e: print(f"⚠️ Dynamic SL/TP non disponible: {e} - Utilisation des valeurs fixes de config.py") # ═══════════════════════════════════════════════════════════════════════════════ # CONVERSION DEVISES # ═══════════════════════════════════════════════════════════════════════════════ class CurrencyConverter: """Convertisseur USD/EUR avec cache""" def __init__(self): self.rate = 1.0 self.last_update = 0 self.update_interval = 300 # 5 minutes self.symbol = "€" if DISPLAY_CURRENCY == "EUR" else "$" def update_rate(self): """Met à jour le taux EUR/USD""" if DISPLAY_CURRENCY != "EUR": self.rate = 1.0 return if time.time() - self.last_update < self.update_interval: return try: # Utiliser l'API Binance pour obtenir EUR/USDT response = requests.get("https://api.binance.com/api/v3/ticker/price", params={"symbol": "EURUSDT"}, timeout=5) data = response.json() # EUR/USDT = combien de USDT pour 1 EUR, donc rate = 1/prix self.rate = 1 / float(data['price']) self.last_update = time.time() except: # Taux par défaut si l'API échoue self.rate = 0.95 def convert(self, usd_amount): """Convertit USD en devise d'affichage""" self.update_rate() return usd_amount * self.rate def format(self, usd_amount, decimals=3): """Formate un montant en devise d'affichage""" converted = self.convert(usd_amount) if decimals == 0: return f"{converted:,.0f}{self.symbol}" return f"{converted:,.{decimals}f}{self.symbol}" # Instance globale currency = CurrencyConverter() # ═══════════════════════════════════════════════════════════════════════════════ # CLIENT API BINANCE # ═══════════════════════════════════════════════════════════════════════════════ class BinanceClient: """Client pour l'API Binance (Spot)""" def __init__(self, api_key="", api_secret="", testnet=True): self.api_key = api_key self.api_secret = api_secret self.testnet = testnet if testnet: self.base_url = "https://testnet.binance.vision" print(" 📡 Mode TESTNET (argent fictif)") else: self.base_url = "https://api.binance.com" print(" ⚠️ Mode PRODUCTION (argent réel)") def _sign(self, params): """Signe une requête avec HMAC SHA256""" query_string = urlencode(params) signature = hmac.new( self.api_secret.encode('utf-8'), query_string.encode('utf-8'), hashlib.sha256 ).hexdigest() return signature def _request(self, method, endpoint, params=None, signed=False): """Effectue une requête API""" url = f"{self.base_url}{endpoint}" headers = {"X-MBX-APIKEY": self.api_key} if params is None: params = {} if signed: params['timestamp'] = int(time.time() * 1000) params['signature'] = self._sign(params) try: if method == "GET": response = requests.get(url, params=params, headers=headers) elif method == "POST": response = requests.post(url, params=params, headers=headers) elif method == "DELETE": response = requests.delete(url, params=params, headers=headers) data = response.json() if 'code' in data and data['code'] < 0: print(f" ❌ Erreur API: {data['msg']}") return None return data except Exception as e: print(f" ❌ Erreur requête: {e}") return None # ─── Informations compte ─────────────────────────────────────────────── def get_account(self): """Récupère les informations du compte""" return self._request("GET", "/api/v3/account", signed=True) def get_balance(self, asset="USDT"): """Récupère le solde d'un asset""" account = self.get_account() if account: for balance in account.get('balances', []): if balance['asset'] == asset: return { 'free': float(balance['free']), 'locked': float(balance['locked']) } return {'free': 0, 'locked': 0} # ─── Prix et marché ──────────────────────────────────────────────────── def get_price(self, symbol): """Récupère le prix actuel""" data = self._request("GET", "/api/v3/ticker/price", {"symbol": symbol}) if data: return float(data['price']) return None def get_klines(self, symbol, interval="5m", limit=100): """Récupère les bougies historiques""" params = {"symbol": symbol, "interval": interval, "limit": limit} return self._request("GET", "/api/v3/klines", params) # ─── Ordres ──────────────────────────────────────────────────────────── def create_order(self, symbol, side, order_type, quantity=None, quote_quantity=None, price=None, stop_price=None): """ Crée un ordre Args: symbol: Paire (ex: BTCUSDT) side: BUY ou SELL order_type: MARKET, LIMIT, STOP_LOSS_LIMIT, TAKE_PROFIT_LIMIT quantity: Quantité en base asset (ex: 0.001 BTC) quote_quantity: Quantité en quote asset (ex: 100 USDT) price: Prix limite stop_price: Prix de déclenchement pour stop orders """ params = { "symbol": symbol, "side": side, "type": order_type } if quantity: params["quantity"] = f"{quantity:.8f}" elif quote_quantity: params["quoteOrderQty"] = f"{quote_quantity:.2f}" if order_type == "LIMIT": params["timeInForce"] = "GTC" params["price"] = f"{price:.2f}" if stop_price: params["stopPrice"] = f"{stop_price:.2f}" return self._request("POST", "/api/v3/order", params, signed=True) def market_buy(self, symbol, usdt_amount): """Achat au marché avec un montant en USDT""" return self.create_order(symbol, "BUY", "MARKET", quote_quantity=usdt_amount) def market_sell(self, symbol, quantity): """Vente au marché""" return self.create_order(symbol, "SELL", "MARKET", quantity=quantity) def limit_buy(self, symbol, quantity, price): """Achat limite""" return self.create_order(symbol, "BUY", "LIMIT", quantity=quantity, price=price) def limit_sell(self, symbol, quantity, price): """Vente limite""" return self.create_order(symbol, "SELL", "LIMIT", quantity=quantity, price=price) def cancel_order(self, symbol, order_id): """Annule un ordre""" params = {"symbol": symbol, "orderId": order_id} return self._request("DELETE", "/api/v3/order", params, signed=True) def get_open_orders(self, symbol=None): """Récupère les ordres ouverts""" params = {} if symbol: params["symbol"] = symbol return self._request("GET", "/api/v3/openOrders", params, signed=True) # ═══════════════════════════════════════════════════════════════════════════════ # INDICATEURS TECHNIQUES # ═══════════════════════════════════════════════════════════════════════════════ class TechnicalIndicators: @staticmethod def rsi(prices, period=14): if len(prices) < period + 1: return None prices = np.array(prices) deltas = np.diff(prices) gains = np.where(deltas > 0, deltas, 0) losses = np.where(deltas < 0, -deltas, 0) avg_gain = np.mean(gains[-period:]) avg_loss = np.mean(losses[-period:]) if avg_loss == 0: return 100 rs = avg_gain / avg_loss return 100 - (100 / (1 + rs)) @staticmethod def ema(prices, period): if len(prices) < period: return None multiplier = 2 / (period + 1) ema = prices[0] for price in prices[1:]: ema = (price * multiplier) + (ema * (1 - multiplier)) return ema @staticmethod def bollinger(prices, period=20, std_dev=2): if len(prices) < period: return None, None, None prices = np.array(prices[-period:]) sma = np.mean(prices) std = np.std(prices) return sma + (std_dev * std), sma, sma - (std_dev * std) @staticmethod def momentum(prices, period=10): """Calcule le momentum (variation % sur la période)""" if len(prices) < period + 1: return None return ((prices[-1] - prices[-period]) / prices[-period]) * 100 @staticmethod def trend_strength(prices, short_period=5, long_period=20): """Mesure la force de la tendance (0-100)""" if len(prices) < long_period: return None, None # Calcul des EMA ema_short = TechnicalIndicators.ema(prices, short_period) ema_long = TechnicalIndicators.ema(prices, long_period) if not ema_short or not ema_long: return None, None # Direction: positif = haussier, négatif = baissier spread = ((ema_short - ema_long) / ema_long) * 100 # Force basée sur la pente des dernières bougies recent_prices = prices[-5:] if len(recent_prices) >= 5: slope = (recent_prices[-1] - recent_prices[0]) / recent_prices[0] * 100 else: slope = 0 # Combinaison spread + slope pour force totale strength = min(100, abs(spread * 10) + abs(slope * 5)) direction = "bullish" if spread > 0 else "bearish" return strength, direction @staticmethod def pullback_detection(prices, ema_period=9): """Détecte un pullback dans une tendance haussière (opportunité d'entrée)""" if len(prices) < ema_period + 5: return False, 0 ema = TechnicalIndicators.ema(prices, ema_period) if not ema: return False, 0 current_price = prices[-1] prev_price = prices[-2] # Pullback = prix touche ou passe sous l'EMA après être au-dessus price_near_ema = abs(current_price - ema) / ema < 0.005 # Dans 0.5% de l'EMA was_above = prev_price > ema # Distance en % par rapport à l'EMA distance_pct = ((current_price - ema) / ema) * 100 return price_near_ema and was_above, distance_pct # ═══════════════════════════════════════════════════════════════════════════════ # GESTIONNAIRE DE POSITIONS # ═══════════════════════════════════════════════════════════════════════════════ # Répertoire du script (pour les chemins relatifs) SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) class PositionManager: """Gère les positions ouvertes avec stop-loss et take-profit""" POSITIONS_FILE = os.path.join(SCRIPT_DIR, "positions.json") HISTORY_FILE = os.path.join(SCRIPT_DIR, "trade_history.json") def __init__(self, client): self.client = client self.positions = {} # {symbol: {entry_price, quantity, stop_loss, take_profit}} self.trade_history = [] # Liste de tous les trades self._load_data() def _load_data(self): """Charge les positions et l'historique depuis les fichiers JSON""" # Charger les positions ouvertes try: if os.path.exists(self.POSITIONS_FILE): with open(self.POSITIONS_FILE, 'r') as f: data = json.load(f) for symbol, pos in data.items(): pos['timestamp'] = datetime.fromisoformat(pos['timestamp']) self.positions[symbol] = pos print(f" 📂 {len(self.positions)} position(s) restaurée(s)") except Exception as e: print(f" ⚠️ Erreur chargement positions: {e}") # Charger l'historique des trades try: if os.path.exists(self.HISTORY_FILE): with open(self.HISTORY_FILE, 'r') as f: self.trade_history = json.load(f) print(f" 📜 {len(self.trade_history)} trade(s) dans l'historique") except Exception as e: print(f" ⚠️ Erreur chargement historique: {e}") def _save_positions(self): """Sauvegarde les positions ouvertes""" try: data = {} for symbol, pos in self.positions.items(): data[symbol] = { **pos, 'timestamp': pos['timestamp'].isoformat() } with open(self.POSITIONS_FILE, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) print(f" 💾 Positions sauvegardées: {len(data)} position(s)") except Exception as e: print(f" ⚠️ Erreur sauvegarde positions: {e}") import traceback traceback.print_exc() def _save_history(self): """Sauvegarde l'historique des trades""" try: with open(self.HISTORY_FILE, 'w') as f: json.dump(self.trade_history, f, indent=2) except Exception as e: print(f" ⚠️ Erreur sauvegarde historique: {e}") def get_total_pnl(self): """Calcule le P&L total de tous les trades fermés""" total_pnl = sum(t.get('pnl', 0) for t in self.trade_history) total_trades = len(self.trade_history) wins = sum(1 for t in self.trade_history if t.get('pnl', 0) > 0) losses = sum(1 for t in self.trade_history if t.get('pnl', 0) < 0) return { 'total_pnl': total_pnl, 'total_trades': total_trades, 'wins': wins, 'losses': losses, 'win_rate': (wins / total_trades * 100) if total_trades > 0 else 0 } def open_position(self, symbol, side, usdt_amount, stop_loss_pct=STOP_LOSS_PERCENT, take_profit_pct=TAKE_PROFIT_PERCENT): """Ouvre une nouvelle position""" if symbol in self.positions: print(f" ⚠️ Position déjà ouverte sur {symbol}") return None # Vérifier le solde balance = self.client.get_balance("USDT") if balance['free'] < usdt_amount: print(f" ❌ Solde insuffisant: {balance['free']:.2f} USDT") return None # Passer l'ordre order = self.client.market_buy(symbol, usdt_amount) if order and 'orderId' in order: entry_price = float(order.get('fills', [{}])[0].get('price', 0)) quantity = float(order.get('executedQty', 0)) if entry_price == 0: entry_price = self.client.get_price(symbol) # Calculer stop-loss et take-profit stop_loss = entry_price * (1 - stop_loss_pct / 100) take_profit = entry_price * (1 + take_profit_pct / 100) self.positions[symbol] = { 'entry_price': entry_price, 'quantity': quantity, 'stop_loss': stop_loss, 'take_profit': take_profit, 'side': side, 'order_id': order['orderId'], 'timestamp': datetime.now() } # Sauvegarder les positions self._save_positions() print(f"\n ✅ POSITION OUVERTE: {symbol}") print(f" Prix entrée: {currency.format(entry_price)}") print(f" Quantité: {quantity}") print(f" Stop-Loss: {currency.format(stop_loss)} (-{stop_loss_pct}%)") print(f" Take-Profit: {currency.format(take_profit)} (+{take_profit_pct}%)") return order return None def close_position(self, symbol, reason="manual"): """Ferme une position""" if symbol not in self.positions: print(f" ⚠️ Pas de position ouverte sur {symbol}") return None position = self.positions[symbol] # Vendre au marché order = self.client.market_sell(symbol, position['quantity']) if order and 'orderId' in order: exit_price = float(order.get('fills', [{}])[0].get('price', 0)) if exit_price == 0: exit_price = self.client.get_price(symbol) # Calculer le P&L pnl = (exit_price - position['entry_price']) * position['quantity'] pnl_pct = ((exit_price / position['entry_price']) - 1) * 100 # Ajouter à l'historique trade_record = { 'symbol': symbol, 'side': 'BUY', 'entry_price': position['entry_price'], 'exit_price': exit_price, 'quantity': position['quantity'], 'pnl': pnl, 'pnl_pct': pnl_pct, 'reason': reason, 'entry_time': position['timestamp'].isoformat(), 'exit_time': datetime.now().isoformat() } self.trade_history.append(trade_record) self._save_history() print(f"\n 📤 POSITION FERMÉE: {symbol} ({reason})") print(f" Prix sortie: {currency.format(exit_price)}") print(f" P&L: {currency.format(pnl, 2)} ({pnl_pct:+.2f}%)") del self.positions[symbol] self._save_positions() return order return None def check_stop_loss_take_profit(self): """Vérifie les stop-loss et take-profit pour toutes les positions""" for symbol, position in list(self.positions.items()): current_price = self.client.get_price(symbol) if current_price is None: continue if current_price <= position['stop_loss']: print(f"\n 🔴 STOP-LOSS DÉCLENCHÉ: {symbol}") self.close_position(symbol, "stop-loss") elif current_price >= position['take_profit']: print(f"\n 🟢 TAKE-PROFIT ATTEINT: {symbol}") self.close_position(symbol, "take-profit") # ═══════════════════════════════════════════════════════════════════════════════ # BOT DE TRADING PRINCIPAL # ═══════════════════════════════════════════════════════════════════════════════ class TradingBot: """Bot de trading automatique""" SETTINGS_FILE = os.path.join(SCRIPT_DIR, "bot_settings.json") WATCHLIST_FILE = os.path.join(SCRIPT_DIR, "watchlist.json") def __init__(self): self.client = BinanceClient( api_key=BINANCE_API_KEY, api_secret=BINANCE_API_SECRET, testnet=TESTNET_MODE ) # Charger la watchlist dynamique (synchronisée avec le dashboard) self.watch_symbols = self._load_watchlist() self.position_manager = PositionManager(self.client) self.prices = {s: deque(maxlen=100) for s in self.watch_symbols} self.running = False self.last_signal = {} self.signal_cooldown = 30 # 30 secondes entre les signaux (réduit pour plus de réactivité) self.trend_cooldown = 15 # 15 secondes si tendance forte détectée self.settings = self._load_settings() def _load_watchlist(self): """Charge la liste des cryptos depuis watchlist.json (synchronisée avec le dashboard)""" try: if os.path.exists(self.WATCHLIST_FILE): with open(self.WATCHLIST_FILE, 'r', encoding='utf-8') as f: data = json.load(f) symbols = data.get('symbols', WATCH_SYMBOLS) print(f" 📋 Watchlist chargée: {len(symbols)} cryptos") print(f" {', '.join([s.replace('USDT', '') for s in symbols[:10]])}{'...' if len(symbols) > 10 else ''}") return symbols except Exception as e: print(f" ⚠️ Erreur chargement watchlist: {e}") # Utiliser la liste par défaut de config.py print(f" 📋 Watchlist par défaut: {len(WATCH_SYMBOLS)} cryptos") return WATCH_SYMBOLS def reload_watchlist(self): """Recharge la watchlist (pour mise à jour dynamique)""" new_symbols = self._load_watchlist() # Ajouter les nouveaux symboles for symbol in new_symbols: if symbol not in self.watch_symbols: self.watch_symbols.append(symbol) self.prices[symbol] = deque(maxlen=100) print(f" ➕ Ajout {symbol} à la surveillance") # Retirer les symboles supprimés (sauf si position ouverte) for symbol in self.watch_symbols[:]: if symbol not in new_symbols: if symbol not in self.position_manager.positions: self.watch_symbols.remove(symbol) if symbol in self.prices: del self.prices[symbol] print(f" ➖ Retrait {symbol} de la surveillance") else: print(f" ⚠️ {symbol} a une position ouverte, conservé") def _reload_config_module(self): """Recharge dynamiquement le module config.py pour obtenir les dernières valeurs""" import importlib import sys try: # Recharger le module config if 'config' in sys.modules: importlib.reload(sys.modules['config']) # Mettre à jour les variables globales global STOP_LOSS_PERCENT, TAKE_PROFIT_PERCENT, MAX_ORDER_SIZE global RSI_OVERSOLD, RSI_OVERBOUGHT, EMA_SHORT, EMA_LONG global BB_PERIOD, BB_STD, RSI_PERIOD, REQUIRED_SIGNALS from config import ( STOP_LOSS_PERCENT, TAKE_PROFIT_PERCENT, MAX_ORDER_SIZE, RSI_OVERSOLD, RSI_OVERBOUGHT, EMA_SHORT, EMA_LONG, BB_PERIOD, BB_STD, RSI_PERIOD, REQUIRED_SIGNALS ) return True except Exception as e: print(f" ⚠️ Erreur rechargement config.py: {e}") return False def _load_settings(self): """Charge les paramètres depuis config.py (fichier maître pour SL/TP) et bot_settings.json (pour positionSize, autoTrade et maxPositions)""" # Recharger config.py pour obtenir les valeurs les plus récentes self._reload_config_module() # VALEURS MAÎTRES depuis config.py (SL/TP non modifiables depuis le dashboard) settings = { 'stopLoss': STOP_LOSS_PERCENT, # ← Toujours depuis config.py 'takeProfit': TAKE_PROFIT_PERCENT, # ← Toujours depuis config.py 'positionSize': MAX_ORDER_SIZE, # ← Valeur par défaut, peut être override 'autoTrade': True, 'maxPositions': MAX_OPEN_POSITIONS } # Charger autoTrade, maxPositions ET positionSize depuis bot_settings.json try: if os.path.exists(self.SETTINGS_FILE): with open(self.SETTINGS_FILE, 'r', encoding='utf-8') as f: user_settings = json.load(f) # Ne garder que les paramètres autorisés (pas SL/TP) if 'autoTrade' in user_settings: settings['autoTrade'] = user_settings['autoTrade'] if 'maxPositions' in user_settings: settings['maxPositions'] = user_settings['maxPositions'] if 'positionSize' in user_settings: settings['positionSize'] = user_settings['positionSize'] except Exception as e: print(f" ⚠️ Erreur chargement paramètres: {e}") print(f" ⚙️ Paramètres chargés: SL={settings['stopLoss']}% (config.py), TP={settings['takeProfit']}% (config.py), Taille={settings['positionSize']}€, Max={settings['maxPositions']}") return settings def load_historical_data(self): """Charge les données historiques""" print(f"\n📊 Chargement des données historiques pour {len(self.watch_symbols)} cryptos...") print(f"⚙️ Paramètres actifs: RSI={RSI_OVERSOLD}/{RSI_OVERBOUGHT}, EMA={EMA_SHORT}/{EMA_LONG}, SL={STOP_LOSS_PERCENT}%, TP={TAKE_PROFIT_PERCENT}%") for symbol in self.watch_symbols: if symbol not in self.prices: self.prices[symbol] = deque(maxlen=100) klines = self.client.get_klines(symbol, DEFAULT_INTERVAL, 100) if klines: for k in klines: self.prices[symbol].append(float(k[4])) # Close price print(f" ✅ {symbol}: {len(self.prices[symbol])} bougies") def analyze(self, symbol): """Analyse un symbole et retourne un signal avec détection de tendance""" prices = list(self.prices[symbol]) if len(prices) < 26: return "HOLD", {} current_price = prices[-1] # Indicateurs de base rsi = TechnicalIndicators.rsi(prices, RSI_PERIOD) ema_short = TechnicalIndicators.ema(prices, EMA_SHORT) ema_long = TechnicalIndicators.ema(prices, EMA_LONG) bb_upper, bb_mid, bb_lower = TechnicalIndicators.bollinger(prices) # NOUVEAUX indicateurs de momentum/tendance momentum = TechnicalIndicators.momentum(prices, 10) trend_strength, trend_direction = TechnicalIndicators.trend_strength(prices, 5, 20) is_pullback, pullback_distance = TechnicalIndicators.pullback_detection(prices, 9) signals = [] trend_signals = [] # Signaux spécifiques à la tendance # ═══════════════════════════════════════════════════════════════════ # STRATÉGIE 1: SIGNAUX CLASSIQUES (RSI, EMA, Bollinger) # ═══════════════════════════════════════════════════════════════════ # RSI if rsi and rsi < RSI_OVERSOLD: signals.append("BUY") elif rsi and rsi > RSI_OVERBOUGHT: signals.append("SELL") # EMA Cross if ema_short and ema_long: if ema_short > ema_long: signals.append("BUY") else: signals.append("SELL") # Bollinger if bb_lower and current_price < bb_lower: signals.append("BUY") elif bb_upper and current_price > bb_upper: signals.append("SELL") # ═══════════════════════════════════════════════════════════════════ # STRATÉGIE 2: TREND FOLLOWING (NOUVEAU!) # Acheter pendant les tendances haussières fortes # ═══════════════════════════════════════════════════════════════════ if trend_strength and trend_direction: # TENDANCE HAUSSIÈRE FORTE (force > 30) if trend_direction == "bullish" and trend_strength > 30: # Momentum positif = tendance confirmée if momentum and momentum > 1.0: # Hausse > 1% sur 10 bougies trend_signals.append("BUY") # Pullback dans tendance haussière = EXCELLENTE opportunité if is_pullback: trend_signals.append("BUY") trend_signals.append("BUY") # Double signal pour pullback # RSI pas encore suracheté (< 75) = encore de la marge if rsi and 40 < rsi < 75: trend_signals.append("BUY") # TENDANCE BAISSIÈRE FORTE elif trend_direction == "bearish" and trend_strength > 40: if momentum and momentum < -2.0: # Baisse > 2% trend_signals.append("SELL") # ═══════════════════════════════════════════════════════════════════ # DÉCISION FINALE # ═══════════════════════════════════════════════════════════════════ buy_count = signals.count("BUY") + trend_signals.count("BUY") sell_count = signals.count("SELL") + trend_signals.count("SELL") indicators = { 'rsi': rsi, 'ema_short': ema_short, 'ema_long': ema_long, 'bb': (bb_upper, bb_mid, bb_lower), 'price': current_price, 'momentum': momentum, 'trend_strength': trend_strength, 'trend_direction': trend_direction, 'is_pullback': is_pullback } # Ajuster les critères selon le nombre de positions current_positions = len(self.position_manager.positions) # MODE TREND FOLLOWING: Entrer plus facilement si tendance forte has_strong_trend = trend_strength and trend_strength > 40 and trend_direction == "bullish" if current_positions < MAX_OPEN_POSITIONS: # Tendance forte = 1 signal suffit if has_strong_trend and buy_count >= 1: return "BUY", indicators # Mode normal: 1 signal positif suffit elif buy_count > 0: return "BUY", indicators elif sell_count >= 3: return "SELL", indicators else: # Positions pleines: être plus sélectif if buy_count >= 2: return "BUY", indicators elif sell_count >= 2: return "SELL", indicators return "HOLD", indicators def execute_signal(self, symbol, signal, indicators=None): """Exécute un signal de trading avec cooldown dynamique""" # Déterminer le cooldown selon la force de la tendance cooldown = self.signal_cooldown # 30s par défaut if indicators: trend_strength = indicators.get('trend_strength') trend_direction = indicators.get('trend_direction') # Tendance forte haussière = cooldown réduit pour plus de trades if trend_strength and trend_strength > 40 and trend_direction == "bullish": cooldown = self.trend_cooldown # 15s # Vérifier le cooldown if symbol in self.last_signal: elapsed = time.time() - self.last_signal[symbol] if elapsed < cooldown: remaining = int(cooldown - elapsed) print(f" ⏳ {symbol}: Cooldown actif (encore {remaining}s)") return # Recharger les paramètres à chaque signal pour prendre en compte les modifications self.settings = self._load_settings() # Vérifier le nombre maximum de positions current_positions = len(self.position_manager.positions) max_positions = self.settings.get('maxPositions', MAX_OPEN_POSITIONS) if signal == "BUY" and symbol not in self.position_manager.positions: # Vérifier le mode auto-trading if not self.settings.get('autoTrade', True): print(f" ⚠️ {symbol}: Trading automatique désactivé") return # Vérifier si on peut ouvrir une nouvelle position if current_positions >= max_positions: print(f" ⚠️ {symbol}: Max positions atteint ({current_positions}/{max_positions})") return # Calculer le montant balance = self.client.get_balance("USDT") if not balance: print(f" ⚠️ {symbol}: Impossible de récupérer le solde") return position_size = self.settings.get('positionSize', MAX_ORDER_SIZE) max_risk_amount = balance['free'] * (MAX_RISK_PER_TRADE / 100) order_amount = min(max_risk_amount, position_size) print(f" 💰 {symbol}: Solde libre={currency.format(balance['free'])} | Max risque (20%)={currency.format(max_risk_amount)} | Position={currency.format(position_size)} | Ordre={currency.format(order_amount)}") if order_amount >= MIN_ORDER_SIZE: print(f"\n🟢 SIGNAL ACHAT: {symbol} (Position {current_positions+1}/{max_positions})") # ═══════════════════════════════════════════════════════════ # DYNAMIC SL/TP - Calcul IA adaptatif # ═══════════════════════════════════════════════════════════ if DYNAMIC_SLTP_ENABLED: try: # Récupérer le régime de marché actuel detector = get_market_regime_detector() regime_name, regime_config = detector.get_current_regime() # Récupérer les données réelles du symbole depuis bot_analysis.json crypto_data = { 'symbol': symbol, 'volatility_score': 50, 'confidence': 70, 'atr_percent': 2.0 } try: analysis_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'bot_analysis.json') if os.path.exists(analysis_path): with open(analysis_path, 'r', encoding='utf-8') as f: analysis = json.load(f) cryptos = analysis.get('cryptos', []) crypto_info = next((c for c in cryptos if c.get('symbol') == symbol), None) if crypto_info: # Calculer volatility_score basé sur le momentum et RSI momentum = abs(crypto_info.get('momentum', 0) or 0) * 100 rsi = crypto_info.get('rsi', 50) ai_score = crypto_info.get('aiScore', 50) # Volatilité = combinaison momentum + distance RSI des extrêmes rsi_volatility = abs(rsi - 50) * 2 # 0-100 volatility_score = min(100, (momentum * 10) + rsi_volatility) crypto_data = { 'symbol': symbol, 'volatility_score': volatility_score, 'confidence': ai_score, 'atr_percent': momentum * 2 # Approximation ATR } except Exception as e: print(f" ⚠️ Lecture bot_analysis.json: {e}") # Calculer SL/TP optimaux via l'IA stop_loss_pct, take_profit_pct = calculate_optimal_sltp( symbol=symbol, crypto_data=crypto_data, market_regime=regime_name ) print(f" 🤖 IA Dynamic SL/TP: SL={stop_loss_pct}% TP={take_profit_pct}% (Régime: {regime_name}, Vol: {crypto_data['volatility_score']:.0f})") except Exception as e: print(f" ⚠️ Erreur Dynamic SL/TP: {e} - Utilisation valeurs config.py") stop_loss_pct = STOP_LOSS_PERCENT take_profit_pct = TAKE_PROFIT_PERCENT else: # Fallback: utiliser les valeurs fixes de config.py stop_loss_pct = STOP_LOSS_PERCENT take_profit_pct = TAKE_PROFIT_PERCENT print(f" 📋 SL/TP fixes: SL={stop_loss_pct}% TP={take_profit_pct}%") self.position_manager.open_position(symbol, "BUY", order_amount, stop_loss_pct, take_profit_pct) self.last_signal[symbol] = time.time() else: print(f" ⚠️ {symbol}: Montant insuffisant ({currency.format(order_amount)} < {currency.format(MIN_ORDER_SIZE)})") elif signal == "SELL" and symbol in self.position_manager.positions: print(f"\n🔴 SIGNAL VENTE: {symbol}") self.position_manager.close_position(symbol, "signal") self.last_signal[symbol] = time.time() def display_status(self): """Affiche le statut du bot""" print("\033[2J\033[H") print("=" * 70) print(" 🤖 TRADING BOT - ORDRES AUTOMATIQUES") print("=" * 70) print(f" ⏰ {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f" 📡 Mode: {'🧪 TESTNET' if TESTNET_MODE else '⚠️ PRODUCTION'}") print(f" 💱 Devise: {DISPLAY_CURRENCY}") # Solde balance = self.client.get_balance("USDT") print(f" 💰 Solde: {currency.format(balance['free'])} (libre) | {currency.format(balance['locked'])} (bloqué)") print("=" * 70) # Positions ouvertes current_positions = len(self.position_manager.positions) max_positions = self.settings.get('maxPositions', MAX_OPEN_POSITIONS) auto_status = "🟢 AUTO" if self.settings.get('autoTrade', True) else "🔴 MANUEL" print(f"\n📊 POSITIONS: {current_positions}/{max_positions} | {auto_status} | SL: {STOP_LOSS_PERCENT}% | TP: {TAKE_PROFIT_PERCENT}%") if self.position_manager.positions: for symbol, pos in self.position_manager.positions.items(): current = self.client.get_price(symbol) or pos['entry_price'] pnl_pct = ((current / pos['entry_price']) - 1) * 100 icon = "🟢" if pnl_pct > 0 else "🔴" print(f" {icon} {symbol}: {pos['quantity']:.6f} @ {currency.format(pos['entry_price'])} → {currency.format(current)} ({pnl_pct:+.2f}%)") else: print("\n📊 Aucune position ouverte") # Analyse des symboles print("\n📈 ANALYSE EN TEMPS RÉEL:") required_signals = 2 if current_positions >= (max_positions // 2) else 1 print(f" (Signaux requis pour achat: {required_signals}/3)") for symbol in self.watch_symbols: prices = list(self.prices[symbol]) if len(prices) < 20: continue signal, indicators = self.analyze(symbol) price = indicators.get('price', 0) rsi = indicators.get('rsi', 0) ema_short = indicators.get('ema_short', 0) ema_long = indicators.get('ema_long', 0) # Compter les signaux individuels signals_detail = [] if rsi and rsi < RSI_OVERSOLD: signals_detail.append("RSI✓") if ema_short and ema_long and ema_short > ema_long: signals_detail.append("EMA✓") bb = indicators.get('bb', (None, None, None)) if bb[2] and price < bb[2]: signals_detail.append("BB✓") in_position = "📍" if symbol in self.position_manager.positions else "" icon = "🟢" if signal == "BUY" else "🔴" if signal == "SELL" else "⚪" signals_str = f" [{', '.join(signals_detail)}]" if signals_detail else "" print(f" {icon} {symbol}{in_position}: {currency.format(price)} | RSI: {rsi:.1f} | {signal}{signals_str}") # Afficher l'historique des trades stats = self.position_manager.get_total_pnl() if stats['total_trades'] > 0: pnl_icon = "🟢" if stats['total_pnl'] >= 0 else "🔴" print(f"\n📜 HISTORIQUE ({stats['total_trades']} trades):") print(f" {pnl_icon} P&L Total: {currency.format(stats['total_pnl'], 2)}") print(f" 📊 Win Rate: {stats['win_rate']:.1f}% ({stats['wins']}W / {stats['losses']}L)") # Derniers trades recent = self.position_manager.trade_history[-3:] if recent: print(" 📝 Derniers trades:") for t in reversed(recent): t_icon = "🟢" if t['pnl'] >= 0 else "🔴" print(f" {t_icon} {t['symbol']}: {currency.format(t['pnl'], 2)} ({t['pnl_pct']:+.2f}%) - {t['reason']}") print("\n" + "=" * 70) print(" [Ctrl+C pour arrêter]") async def price_updater(self, symbol): """Met à jour les prix via WebSocket""" stream = symbol.lower() + "@trade" url = f"wss://stream.binance.com:9443/ws/{stream}" while self.running: try: async with websockets.connect(url) as ws: async for message in ws: if not self.running: break data = json.loads(message) self.prices[symbol].append(float(data['p'])) except: if self.running: await asyncio.sleep(5) async def trading_loop(self): """Boucle principale de trading""" watchlist_check_counter = 0 while self.running: # Vérifier stop-loss / take-profit self.position_manager.check_stop_loss_take_profit() # Recharger la watchlist toutes les 30 secondes (15 itérations) watchlist_check_counter += 1 if watchlist_check_counter >= 15: self.reload_watchlist() watchlist_check_counter = 0 # Analyser et exécuter les signaux buy_signals = 0 for symbol in self.watch_symbols: if symbol not in self.prices: self.prices[symbol] = deque(maxlen=100) signal, indicators = self.analyze(symbol) if signal == "BUY": buy_signals += 1 # Afficher les infos de tendance si forte if indicators.get('trend_strength') and indicators.get('trend_strength') > 40: print(f" 📈 {symbol}: Tendance {indicators.get('trend_direction')} (force: {indicators.get('trend_strength'):.0f})") if signal != "HOLD": self.execute_signal(symbol, signal, indicators) # Log si signaux d'achat détectés if buy_signals > 0: current_positions = len(self.position_manager.positions) max_positions = self.settings.get('maxPositions', MAX_OPEN_POSITIONS) print(f"\n💡 {buy_signals} signaux BUY détectés | Positions: {current_positions}/{max_positions}") # Afficher le statut self.display_status() await asyncio.sleep(2) # Mise à jour toutes les 2 secondes async def run(self): """Lance le bot""" print("\n🚀 Démarrage du Bot de Trading...") # Vérifier les clés API if not BINANCE_API_KEY or not BINANCE_API_SECRET: print("\n❌ ERREUR: Clés API non configurées!") print(" Édite config.py et ajoute tes clés Binance") print(" Pour le testnet: https://testnet.binance.vision/") return # Test connexion account = self.client.get_account() if not account: print("\n❌ ERREUR: Impossible de se connecter à Binance") return print(" ✅ Connexion réussie!") # Charger l'historique self.load_historical_data() self.running = True # Lancer les tâches tasks = [ asyncio.create_task(self.price_updater(s)) for s in self.watch_symbols ] tasks.append(asyncio.create_task(self.trading_loop())) try: await asyncio.gather(*tasks) except asyncio.CancelledError: pass finally: self.running = False def main(): # Forcer l'encodage UTF-8 pour la console Windows if sys.platform == 'win32': import io sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8') print(""" ╔══════════════════════════════════════════════════════════════════════╗ ║ 🤖 CRYPTO TRADING BOT - ORDRES AUTOMATIQUES 🤖 ║ ╠══════════════════════════════════════════════════════════════════════╣ ║ ⚠️ Ce bot peut passer des ordres RÉELS ! ║ ║ 🧪 Mode TESTNET activé par défaut (argent fictif) ║ ║ 📊 Stratégie: RSI + EMA + Bollinger Bands ║ ╚══════════════════════════════════════════════════════════════════════╝ """) if not TESTNET_MODE: print("⚠️ ATTENTION: Mode PRODUCTION activé!") confirm = input(" Confirmer ? (oui/non): ") if confirm.lower() != "oui": print(" Annulé.") return bot = TradingBot() try: asyncio.run(bot.run()) except KeyboardInterrupt: print("\n\n👋 Bot arrêté.") if __name__ == "__main__": main()