""" 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 import logging # ═══════════════════════════════════════════════════════════════════════════════ # CONFIGURATION DU LOGGING # ═══════════════════════════════════════════════════════════════════════════════ SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) LOG_FILE = os.path.join(SCRIPT_DIR, 'trading_bot.log') # Configurer le logging pour écrire dans le fichier class FileLogger: """Redirige stdout vers fichier uniquement (évite les erreurs Unicode Windows)""" def __init__(self, filename): self.log = open(filename, 'w', encoding='utf-8', buffering=1) # Line buffered def write(self, message): try: self.log.write(message) self.log.flush() except: pass def flush(self): try: self.log.flush() except: pass # Activer le logging fichier (sans console pour éviter les erreurs Unicode Windows) sys.stdout = FileLogger(LOG_FILE) sys.stderr = sys.stdout try: import websockets import requests except ImportError: import subprocess subprocess.check_call(['pip', 'install', 'websockets', 'requests']) import websockets import requests from config import * # Import du module IA de prédiction try: from ai_predictor import get_ai_predictor, get_surveillance_service, AIPredictor AI_PREDICTOR_AVAILABLE = True except ImportError: AI_PREDICTOR_AVAILABLE = False print("⚠️ Module ai_predictor non disponible") # ═══════════════════════════════════════════════════════════════════════════════ # 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)""" # Décalage d'horloge global (synchronisé avec le serveur) TIME_OFFSET = 0 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)") # Synchroniser l'horloge avec le serveur au démarrage self._sync_server_time() def _sync_server_time(self): """Synchronise l'horloge locale avec le serveur Binance""" try: response = requests.get(f"{self.base_url}/api/v3/time", timeout=5) server_time = response.json()['serverTime'] local_time = int(time.time() * 1000) BinanceClient.TIME_OFFSET = server_time - local_time if abs(BinanceClient.TIME_OFFSET) > 1000: print(f" ⏰ Horloge synchronisée (décalage: {BinanceClient.TIME_OFFSET}ms)") except Exception as e: print(f" ⚠️ Impossible de synchroniser l'horloge: {e}") BinanceClient.TIME_OFFSET = 0 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: # Utiliser le timestamp synchronisé avec le serveur params['timestamp'] = int(time.time() * 1000) + BinanceClient.TIME_OFFSET 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_all_prices(self): """Récupère tous les prix en une seule requête (beaucoup plus rapide)""" data = self._request("GET", "/api/v3/ticker/price") if data: return {item['symbol']: float(item['price']) for item in data} return {} 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) def get_klines_production(self, symbol, interval="5m", limit=100, use_cache=True): """ Récupère les klines depuis l'API PRODUCTION (publique). Le testnet n'a souvent pas assez de données de trading. Utilise un cache de 30 secondes pour éviter trop d'appels API. """ import time as _time # Initialiser le cache si besoin if not hasattr(self, '_klines_cache'): self._klines_cache = {} self._klines_cache_time = {} cache_key = f"{symbol}_{interval}_{limit}" # Vérifier le cache (valide 30 secondes) if use_cache and cache_key in self._klines_cache: cache_age = _time.time() - self._klines_cache_time.get(cache_key, 0) if cache_age < 30: # Cache valide 30 secondes return self._klines_cache[cache_key] try: import requests as req url = f"https://api.binance.com/api/v3/klines" params = {"symbol": symbol, "interval": interval, "limit": limit} resp = req.get(url, params=params, timeout=10) if resp.status_code == 200: data = resp.json() if isinstance(data, list) and len(data) > 0: # Stocker dans le cache self._klines_cache[cache_key] = data self._klines_cache_time[cache_key] = _time.time() return data except Exception as e: pass # Silencieux # Retourner le cache expiré plutôt que None if cache_key in self._klines_cache: return self._klines_cache[cache_key] return None def get_symbol_info(self, symbol): """Récupère les infos d'un symbole (précision, etc.)""" data = self._request("GET", "/api/v3/exchangeInfo", {"symbol": symbol}) if data and 'symbols' in data and len(data['symbols']) > 0: return data['symbols'][0] return None def get_quantity_precision(self, symbol): """Récupère la précision de quantité pour un symbole""" info = self.get_symbol_info(symbol) if info: for f in info.get('filters', []): if f['filterType'] == 'LOT_SIZE': step_size = f['stepSize'] # Calculer le nombre de décimales à partir de stepSize if '.' in step_size: decimals = len(step_size.rstrip('0').split('.')[1]) return decimals return 0 # Valeur par défaut sûre return 0 def format_quantity(self, symbol, quantity): """Formate la quantité selon la précision du symbole""" precision = self.get_quantity_precision(symbol) # Arrondir vers le bas pour éviter les dépassements factor = 10 ** precision formatted = int(quantity * factor) / factor # Si précision 0, retourner un entier if precision == 0: return int(formatted) return formatted # ─── 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: # Formatter selon le type (entier ou float) if isinstance(quantity, int): params["quantity"] = str(quantity) else: params["quantity"] = f"{quantity:.8f}".rstrip('0').rstrip('.') 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 (AMÉLIORÉS) # ═══════════════════════════════════════════════════════════════════════════════ 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:]) # Protection: si pas de mouvement, retourner RSI neutre (50) if avg_loss == 0 and avg_gain == 0: return 50 # Marché stagnant = RSI neutre if avg_loss == 0: return 95 # Tendance très haussière mais pas 100 (valeur extrême réservée) if avg_gain == 0: return 5 # Tendance très baissière mais pas 0 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 @staticmethod def bollinger_trend(prices, period=20, std_dev=2): """Analyse la tendance des bandes de Bollinger Retourne: direction ('up', 'down', 'flat'), bandwidth expansion """ if len(prices) < period + 5: return None, None, None # Calculer BB actuel et précédent bb_upper, bb_mid, bb_lower = TechnicalIndicators.bollinger(prices, period, std_dev) bb_upper_prev, bb_mid_prev, bb_lower_prev = TechnicalIndicators.bollinger(prices[:-3], period, std_dev) if not all([bb_upper, bb_mid, bb_lower, bb_mid_prev]): return None, None, None # Direction de la bande centrale mid_slope = ((bb_mid - bb_mid_prev) / bb_mid_prev) * 100 # Bandwidth expansion/contraction bandwidth = (bb_upper - bb_lower) / bb_mid * 100 bandwidth_prev = (bb_upper_prev - bb_lower_prev) / bb_mid_prev * 100 if bb_mid_prev else bandwidth if mid_slope > 0.3: direction = 'up' elif mid_slope < -0.3: direction = 'down' else: direction = 'flat' expansion = bandwidth > bandwidth_prev * 1.05 # 5% plus large return direction, expansion, mid_slope @staticmethod def bollinger_squeeze(prices, period=20, std_dev=2, squeeze_threshold=3.0): """Détecte un Bollinger Squeeze (compression des bandes = faible volatilité) Un squeeze précède souvent une explosion du prix (hausse ou baisse forte) Retourne: is_squeeze, bandwidth, squeeze_strength, breakout_direction - is_squeeze: True si les bandes sont compressées - bandwidth: largeur actuelle des bandes en % - squeeze_strength: 0-100, plus c'est haut plus le squeeze est serré - breakout_direction: 'up' si le prix commence à casser vers le haut, 'down' sinon, None si pas de breakout """ if len(prices) < period + 10: return False, 0, 0, None # Calculer BB actuel bb_upper, bb_mid, bb_lower = TechnicalIndicators.bollinger(prices, period, std_dev) if not all([bb_upper, bb_mid, bb_lower]) or bb_mid == 0: return False, 0, 0, None # Bandwidth actuel (largeur des bandes en %) bandwidth = (bb_upper - bb_lower) / bb_mid * 100 # Calculer le bandwidth moyen sur les 20 dernières bougies pour comparaison bandwidths = [] for i in range(10, min(30, len(prices))): bb_u, bb_m, bb_l = TechnicalIndicators.bollinger(prices[:-i] if i > 0 else prices, period, std_dev) if bb_m and bb_m > 0: bw = (bb_u - bb_l) / bb_m * 100 bandwidths.append(bw) if not bandwidths: return False, bandwidth, 0, None avg_bandwidth = sum(bandwidths) / len(bandwidths) # Un squeeze est quand le bandwidth est significativement inférieur à la moyenne is_squeeze = bandwidth < squeeze_threshold or bandwidth < avg_bandwidth * 0.7 # Force du squeeze (0-100) if avg_bandwidth > 0: squeeze_strength = max(0, min(100, (1 - bandwidth / avg_bandwidth) * 100)) else: squeeze_strength = 0 # Détecter la direction du breakout (si le prix commence à sortir) current_price = prices[-1] prev_price = prices[-2] if len(prices) > 1 else current_price breakout_direction = None if is_squeeze or squeeze_strength > 30: # Prix qui sort vers le haut if current_price > bb_mid and current_price > prev_price: breakout_direction = 'up' # Prix qui sort vers le bas elif current_price < bb_mid and current_price < prev_price: breakout_direction = 'down' return is_squeeze, bandwidth, squeeze_strength, breakout_direction @staticmethod def ema_trend(prices, short=9, mid=21, long=50): """Analyse la configuration EMA Retourne: alignement ('bullish', 'bearish', 'mixed'), force, pente """ ema_s = TechnicalIndicators.ema(prices, short) ema_m = TechnicalIndicators.ema(prices, mid) ema_l = TechnicalIndicators.ema(prices, long) if len(prices) >= long else None if not all([ema_s, ema_m]): return None, 0, 0 # Calculer la pente des EMAs (direction du mouvement) # On compare l'EMA actuelle à celle d'il y a 3 périodes lookback = 3 if len(prices) >= short + lookback: ema_s_prev = TechnicalIndicators.ema(prices[:-lookback], short) ema_m_prev = TechnicalIndicators.ema(prices[:-lookback], mid) # Pente en pourcentage ema_s_slope = ((ema_s - ema_s_prev) / ema_s_prev * 100) if ema_s_prev else 0 ema_m_slope = ((ema_m - ema_m_prev) / ema_m_prev * 100) if ema_m_prev else 0 avg_slope = (ema_s_slope + ema_m_slope) / 2 else: avg_slope = 0 # Vérifier l'alignement EMA (golden cross configuration) if ema_l: if ema_s > ema_m > ema_l: # Alignement bullish MAIS vérifier si pente négative if avg_slope < -0.3: # EMAs en chute malgré alignement return 'bearish', 75, avg_slope return 'bullish', 100, avg_slope # Alignement parfait haussier elif ema_s < ema_m < ema_l: return 'bearish', 100, avg_slope # Alignement parfait baissier elif ema_s > ema_m: if avg_slope < -0.5: # Forte pente négative return 'bearish', 60, avg_slope return 'bullish', 50, avg_slope else: return 'bearish', 50, avg_slope else: if ema_s > ema_m: if avg_slope < -0.5: # Forte pente négative return 'bearish', 60, avg_slope return 'bullish', 50, avg_slope else: return 'bearish', 50, avg_slope return 'mixed', 25, avg_slope @staticmethod def breakout_detection(prices, period=20): """Détecte un breakout au-dessus de la résistance Un breakout est quand le prix dépasse le plus haut des N dernières bougies """ if len(prices) < period + 1: return False, 0 # Exclure la bougie actuelle pour le calcul du range historical_high = max(prices[-period-1:-1]) historical_low = min(prices[-period-1:-1]) current = prices[-1] prev = prices[-2] # Breakout haussier if current > historical_high and prev <= historical_high: strength = ((current - historical_high) / historical_high) * 100 return True, strength return False, 0 @staticmethod def atr(prices, period=14): """Average True Range - mesure de la volatilité pour trailing stop""" if len(prices) < period + 1: return None trs = [] for i in range(-period, 0): high = prices[i] low = prices[i] prev_close = prices[i-1] if i > -len(prices) else prices[i] tr = max(high - low, abs(high - prev_close), abs(low - prev_close)) trs.append(tr) return np.mean(trs) # ═══════════════════════════════════════════════════════════════════════════════ # 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] # Formater la quantité selon la précision du symbole quantity = self.client.format_quantity(symbol, position['quantity']) if quantity <= 0: print(f" ⚠️ Quantité trop petite pour {symbol}: {position['quantity']}") # Supprimer quand même la position car elle est inutilisable del self.positions[symbol] self._save_positions() return None print(f" 📊 Vente de {quantity} {symbol} (original: {position['quantity']})") # Vendre au marché order = self.client.market_sell(symbol, 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_technical_exit(self, symbol, prices_data): """ Vérifie les indicateurs techniques pour une sortie anticipée. Appelée à chaque cycle pour les positions ouvertes. Retourne: (should_exit, reason) """ if symbol not in self.positions or len(prices_data) < 21: return False, "" position = self.positions[symbol] entry_price = position['entry_price'] current_price = prices_data[-1] if prices_data else self.client.get_price(symbol) if current_price is None: return False, "" prices_list = list(prices_data) # Calculer EMA ema9 = TechnicalIndicators.ema(prices_list, 9) ema21 = TechnicalIndicators.ema(prices_list, 21) if not ema9 or not ema21: return False, "" # Calculer momentum sur 3 et 5 bougies momentum_3 = ((current_price - prices_list[-3]) / prices_list[-3]) * 100 if len(prices_list) >= 3 else 0 momentum_5 = ((current_price - prices_list[-5]) / prices_list[-5]) * 100 if len(prices_list) >= 5 else 0 # P&L actuel current_pnl_pct = ((current_price / entry_price) - 1) * 100 exit_reasons = [] exit_score = 0 # ═══════════════════════════════════════════════════════════════════ # STRATÉGIE "BUY THE DIP" - QUICK EXIT adapté # On achète quand EMA9 < EMA21, donc on VEND quand EMA9 > EMA21 # ═══════════════════════════════════════════════════════════════════ # 1. GOLDEN CROSS : EMA9 repasse AU-DESSUS de EMA21 = TAKE PROFIT! ema_gap_pct = ((ema9 - ema21) / ema21) * 100 if ema_gap_pct > 0.15: # EMA9 est 0.15% AU-DESSUS de EMA21 = rebond confirmé! exit_score += 3 # Signal fort de prise de profit exit_reasons.append(f"REBOND! EMA9 > EMA21 (+{ema_gap_pct:.2f}%)") # 2. RSI en surachat = prendre les profits rsi = TechnicalIndicators.rsi(prices_list, RSI_PERIOD) if rsi and rsi > 65: exit_score += 2 exit_reasons.append(f"RSI surachat ({rsi:.0f})") # 3. Prix touche la bande Bollinger haute = zone de vente bb_upper = TechnicalIndicators.bollinger(prices_list)[0] if len(prices_list) >= 20 else None if bb_upper and current_price > bb_upper * 0.995: exit_score += 2 exit_reasons.append("Prix proche BB haute") # 4. Momentum très positif = prendre les gains if momentum_3 > 1.5: # Grosse hausse rapide exit_score += 1 exit_reasons.append(f"Momentum fort: +{momentum_3:.2f}%") # 5. Stop-loss si chute trop forte (protection) max_price = position.get('max_price', entry_price) drop_from_max = ((current_price - max_price) / max_price) * 100 if drop_from_max < -3.0: # Chute de 3% depuis le max exit_score += 3 exit_reasons.append(f"Stop: {drop_from_max:.1f}% depuis max") # ═══════════════════════════════════════════════════════════════════ # DÉCISION DE SORTIE - PRENDRE LES PROFITS SUR LE REBOND # ═══════════════════════════════════════════════════════════════════ should_exit = False # ═══════════════════════════════════════════════════════════════════ # DÉTECTION IA DE VENTE ANTICIPÉE (Premiers signes de faiblesse) # ═══════════════════════════════════════════════════════════════════ if AI_PREDICTOR_AVAILABLE and current_pnl_pct > 0.3: try: ai_predictor = get_ai_predictor() if ai_predictor: ai_should_sell, ai_reason = ai_predictor.should_sell_early( symbol, prices_list, entry_price, current_pnl_pct ) if ai_should_sell: exit_score += 3 exit_reasons.append(ai_reason) except Exception as e: pass # Silencieux si erreur IA # Score >= 3 = prendre les profits (rebond détecté) if exit_score >= 3: should_exit = True # Profit de 1%+ avec signal de rebond elif exit_score >= 2 and current_pnl_pct > 1.0: should_exit = True if should_exit: reason = f"PROFIT/Exit (score={exit_score}): {', '.join(exit_reasons[:3])}" return True, reason return False, "" def check_stop_loss_take_profit(self): """ Vérifie les stop-loss et take-profit avec TRAILING STOP DYNAMIQUE Le trailing stop permet de: - Remonter le stop-loss quand le prix monte - Ne pas vendre trop tôt quand une crypto continue à monter - Protéger les gains acquis """ if len(self.positions) == 0: return for symbol, position in list(self.positions.items()): current_price = self.client.get_price(symbol) if current_price is None: print(f" ⚠️ {symbol}: Prix non disponible") continue entry_price = position['entry_price'] current_pnl_pct = ((current_price / entry_price) - 1) * 100 # ═══════════════════════════════════════════════════════════════ # TRAILING STOP DYNAMIQUE # ═══════════════════════════════════════════════════════════════ # Récupérer ou initialiser le prix max atteint if 'max_price' not in position: position['max_price'] = current_price elif current_price > position['max_price']: position['max_price'] = current_price # Recalculer le trailing stop si on est en profit if current_pnl_pct > 1: # Plus de 1% de profit # Trailing stop à 40% du profit (protège 60% des gains) trailing_distance_pct = STOP_LOSS_PERCENT # Utiliser le même % que le SL initial new_stop = current_price * (1 - trailing_distance_pct / 100) # Ne jamais baisser le stop-loss if new_stop > position['stop_loss']: old_sl = position['stop_loss'] position['stop_loss'] = new_stop print(f" 📈 {symbol}: Trailing stop remonté {currency.format(old_sl)} → {currency.format(new_stop)} (+{current_pnl_pct:.1f}%)") self._save_positions() # ═══════════════════════════════════════════════════════════════ # GESTION DES PALIERS DE PROFIT # ═══════════════════════════════════════════════════════════════ # Palier 1: +50% du TP → Mettre stop au break-even tp_pct = ((position['take_profit'] / entry_price) - 1) * 100 if current_pnl_pct >= tp_pct * 0.5: # 50% du take profit atteint breakeven_stop = entry_price * 1.001 # +0.1% pour couvrir les frais if position['stop_loss'] < breakeven_stop: position['stop_loss'] = breakeven_stop print(f" 🛡️ {symbol}: Stop au break-even (50% du TP atteint)") self._save_positions() # Palier 2: +75% du TP → Protéger 50% des gains if current_pnl_pct >= tp_pct * 0.75: protect_stop = entry_price * (1 + (current_pnl_pct * 0.5) / 100) if position['stop_loss'] < protect_stop: position['stop_loss'] = protect_stop print(f" 💰 {symbol}: Protection 50% des gains (75% du TP atteint)") self._save_positions() # ═══════════════════════════════════════════════════════════════ # VÉRIFICATION STOP-LOSS / TAKE-PROFIT # ═══════════════════════════════════════════════════════════════ # Stop-Loss if current_price <= position['stop_loss']: if current_pnl_pct > 0: print(f"\n 🟡 TRAILING STOP: {symbol} (Prix: {currency.format(current_price)} | P&L: +{current_pnl_pct:.2f}%)") self.close_position(symbol, "trailing-stop") else: print(f"\n 🔴 STOP-LOSS: {symbol} (Prix: {currency.format(current_price)} <= SL: {currency.format(position['stop_loss'])})") self.close_position(symbol, "stop-loss") # Take-Profit (mais considérer de laisser courir si momentum fort) elif current_price >= position['take_profit']: print(f"\n 🟢 TAKE-PROFIT: {symbol} (Prix: {currency.format(current_price)} >= TP: {currency.format(position['take_profit'])})") 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.last_trade = {} # Cooldown par symbole après un trade self.signal_cooldown = 60 # 1 minute entre les signaux (réactif) self.trend_cooldown = 30 # 30 secondes si tendance forte self.trade_cooldown = 120 # 2 minutes après un trade sur le même symbole self.settings = self._load_settings() # === SERVICE DE SURVEILLANCE IA === self.ai_predictor = None self.surveillance_service = None self.ai_watchlist = {} # Symboles sous surveillance IA if AI_PREDICTOR_AVAILABLE: try: self.ai_predictor = get_ai_predictor() self.surveillance_service = get_surveillance_service() # Configurer le service self.surveillance_service.set_klines_fetcher(self._fetch_klines_for_ai) self.surveillance_service.set_on_signal(self._on_ai_signal) self.surveillance_service.set_symbols(self.watch_symbols) print(" 🤖 Service de surveillance IA initialisé") except Exception as e: print(f" ⚠️ Erreur init surveillance IA: {e}") def _fetch_klines_for_ai(self, symbol: str, interval: str, limit: int): """Fetcher de klines pour le service IA""" return self.client.get_klines_production(symbol, interval, limit) def _on_ai_signal(self, symbol: str, item): """Callback quand l'IA détecte un signal d'achat""" print(f" 🤖 Signal IA: {symbol} - Score={item.score} Pattern={item.pattern}") # Le signal sera traité dans la boucle principale 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 load_ai_scoring(self): """Charge le cache IA avec les scores d'opportunités""" cache_file = os.path.join(SCRIPT_DIR, "crypto_cache", "crypto_data.json") try: if os.path.exists(cache_file): with open(cache_file, 'r', encoding='utf-8') as f: self.ai_cache = json.load(f) print(f" 🧠 Cache IA chargé: {len(self.ai_cache.get('symbols', {}))} cryptos") return True except Exception as e: print(f" ⚠️ Erreur chargement cache IA: {e}") self.ai_cache = {} return False def get_ai_score(self, symbol): """Récupère le score IA d'une crypto - utilise le service IA si disponible""" # PRIORITÉ 1: Service de surveillance IA (temps réel avec modèle PyTorch) if self.surveillance_service and self.ai_predictor: try: status = self.surveillance_service.get_surveillance_status() # Chercher dans TOUS les signaux, pas seulement ready_signals all_signals = status.get('ready_signals', []) + status.get('watching_signals', []) # Aussi chercher via la watchlist complète if hasattr(self.surveillance_service, 'watchlist'): for sym, item in self.surveillance_service.watchlist.items(): if sym == symbol: return { 'score': item.score if hasattr(item, 'score') else 50, 'rsi': item.features.get('rsi', 50) if hasattr(item, 'features') else 50, 'trend': 'bullish' if (item.score if hasattr(item, 'score') else 0) >= 60 else 'bearish' if (item.score if hasattr(item, 'score') else 0) < 40 else 'neutral', 'pattern': item.pattern if hasattr(item, 'pattern') else 'NEUTRAL', 'confidence': item.confidence if hasattr(item, 'confidence') else 0, 'reason': item.reason if hasattr(item, 'reason') else '', 'price': item.features.get('price_current', 0) if hasattr(item, 'features') else 0, 'from_ai_service': True } for item in all_signals: if item.get('symbol') == symbol: return { 'score': item.get('score', 50), 'rsi': item.get('features', {}).get('rsi', 50), 'trend': 'bullish' if item.get('score', 0) >= 60 else 'bearish' if item.get('score', 0) < 40 else 'neutral', 'pattern': item.get('pattern', 'NEUTRAL'), 'confidence': item.get('confidence', 0), 'reason': item.get('reason', ''), 'price': item.get('features', {}).get('price_current', 0), 'from_ai_service': True } except Exception as e: pass # Fallback au cache # PRIORITÉ 2: Cache IA (crypto_cache/crypto_data.json) if not hasattr(self, 'ai_cache') or not self.ai_cache: self.load_ai_scoring() symbols = self.ai_cache.get('symbols', {}) if symbol not in symbols: return {'score': 50, 'rsi': 50, 'trend': 'neutral', 'from_ai_service': False} data = symbols[symbol] indicators = data.get('indicators', {}) signals_cache = data.get('signals', {}) # Calculer un score amélioré basé sur les indicateurs du cache score = 50 # Base neutre # RSI du cache (pondération forte) rsi = indicators.get('rsi', 50) if rsi < 25: score += 25 # RSI très survendu = forte opportunité elif rsi < 30: score += 18 elif rsi < 40: score += 10 elif rsi > 75: score -= 20 elif rsi > 70: score -= 12 elif rsi > 60: score -= 5 # Trend du cache trend = signals_cache.get('trend', 'neutral') if trend == 'bullish': score += 12 elif trend == 'bearish': score -= 12 # MACD macd = indicators.get('macd', {}) histogram = macd.get('histogram', 0) if histogram > 0: score += 8 elif histogram < -0.5: score -= 8 # Volatilité RSI rsi_signal = signals_cache.get('rsi_signal', 'neutral') if rsi_signal == 'oversold': score += 12 elif rsi_signal == 'overbought': score -= 12 # Bollinger position bb_position = indicators.get('bb_position', 0.5) if bb_position < 0.2: # Proche bande basse score += 10 elif bb_position > 0.8: # Proche bande haute score -= 10 return { 'score': min(100, max(0, score)), 'rsi': rsi, 'trend': trend, 'macd_histogram': histogram, 'price': data.get('price', 0), 'change_24h': data.get('priceChangePercent', 0), 'from_ai_service': False } def get_surveillance_status(self): """Retourne le statut de la surveillance IA pour le dashboard""" if not self.surveillance_service or not self.ai_predictor: return { 'is_running': False, 'total_symbols': 0, 'analyzed': 0, 'ready_to_buy': 0, 'watching': 0, 'top_opportunities': [], 'ready_signals': [], 'ai_available': False } status = self.surveillance_service.get_surveillance_status() status['ai_available'] = True return status def get_ai_watchlist(self): """Retourne la watchlist IA triée par score""" if not self.ai_predictor: return [] return self.ai_predictor.get_watchlist() def analyze(self, symbol): """ ANALYSE AMÉLIORÉE v2.0 - Ne jamais acheter en tendance baissière! Règles strictes: 1. JAMAIS acheter si EMA/Bollinger orientés à la baisse 2. Détecter les breakouts pour réagir aux hausses 3. Utiliser le scoring IA du cache 4. Trailing stop dynamique pour ne pas vendre trop tôt """ # Utiliser les VRAIES klines de PRODUCTION pour une analyse précise # (le testnet n'a pas assez de données de trading) try: klines = self.client.get_klines_production(symbol, DEFAULT_INTERVAL, 100) if klines and len(klines) >= 50: prices = [float(k[4]) for k in klines] # Prix de clôture des bougies else: prices = list(self.prices[symbol]) except: prices = list(self.prices[symbol]) if len(prices) < 50: return "HOLD", {}, "Pas assez de données" 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) momentum = TechnicalIndicators.momentum(prices, 10) # ═══════════════════════════════════════════════════════════════════ # INDICATEURS DE TENDANCE AVANCÉS # ═══════════════════════════════════════════════════════════════════ bb_direction, bb_expanding, bb_slope = TechnicalIndicators.bollinger_trend(prices) ema_alignment, ema_strength, ema_slope = TechnicalIndicators.ema_trend(prices, EMA_SHORT, EMA_LONG, 50) is_breakout, breakout_strength = TechnicalIndicators.breakout_detection(prices, 20) is_pullback, pullback_distance = TechnicalIndicators.pullback_detection(prices, 9) # NOUVEAU: Détection du Bollinger Squeeze (compression avant explosion) is_squeeze, bb_bandwidth, squeeze_strength, squeeze_breakout = TechnicalIndicators.bollinger_squeeze(prices) # NOUVEAU: Détection du GOLDEN CROSS récent (EMA9 vient de croiser EMA21 à la hausse) golden_cross_recent = False if len(prices) >= 5 and ema_short and ema_long: # Calculer EMA9 et EMA21 il y a 2-3 bougies ema9_prev = TechnicalIndicators.ema(prices[:-2], EMA_SHORT) ema21_prev = TechnicalIndicators.ema(prices[:-2], EMA_LONG) # Golden Cross = EMA9 était SOUS EMA21, maintenant EMA9 est AU-DESSUS if ema9_prev and ema21_prev: was_below = ema9_prev < ema21_prev is_above = ema_short > ema_long if was_below and is_above: golden_cross_recent = True # Score IA (service temps réel ou cache) ai_data = self.get_ai_score(symbol) ai_score = ai_data.get('score', 50) ai_from_service = ai_data.get('from_ai_service', False) ai_pattern = ai_data.get('pattern', 'NEUTRAL') if ai_from_service else None # ═══════════════════════════════════════════════════════════════════ # RÈGLE #1: BLOQUER LES ACHATS EN TENDANCE BAISSIÈRE # ═══════════════════════════════════════════════════════════════════ is_bearish_trend = False bearish_reasons = [] # ══════════════════════════════════════════════════════════════════════ # CRITÈRES DE BLOCAGE: Uniquement les CHUTES ACTIVES, pas les creux stables # La stratégie "Buy the Dip" achète quand EMA9 < EMA21, donc on ne bloque PAS # simplement parce que EMA9 < EMA21 - on bloque seulement si CHUTE ACTIVE # ══════════════════════════════════════════════════════════════════════ # Critère 1: Bollinger très orientées à la baisse (chute forte) if bb_direction == 'down' and ema_slope and ema_slope < -0.3: is_bearish_trend = True bearish_reasons.append("BB + EMAs en chute") # Critère 2: Momentum TRÈS négatif (chute en cours) if momentum and momentum < -1.5: is_bearish_trend = True bearish_reasons.append(f"Momentum tres negatif ({momentum:.1f}%)") # Critère 3: Prix en chute forte sur 3 bougies (> 1.5%) if len(prices) >= 3: price_3_ago = prices[-3] price_change_3 = ((current_price - price_3_ago) / price_3_ago) * 100 if price_change_3 < -1.5: is_bearish_trend = True bearish_reasons.append(f"Chute forte ({price_change_3:.2f}%)") # Critère 4: Prix en chute très forte sur 5 bougies (> 3%) if len(prices) >= 5: price_5_ago = prices[-5] price_change_5 = ((current_price - price_5_ago) / price_5_ago) * 100 if price_change_5 < -3.0: is_bearish_trend = True bearish_reasons.append(f"Chute prolongee ({price_change_5:.2f}%)") # Critère 5: RSI très bas + momentum négatif = danger if rsi and rsi < 25 and momentum and momentum < -0.5: is_bearish_trend = True bearish_reasons.append("RSI extreme + Momentum negatif") # Critère 6: TENDANCE BAISSIÈRE CONTINUE - Prix < EMA21 ET EMA21 en pente descendante # C'est le cas typique où le prix descend lentement mais sûrement if ema_long and current_price < ema_long: # Calculer la pente de EMA21 sur les dernières bougies if len(prices) >= 10: # Comparer EMA21 estimée il y a 5 bougies vs maintenant price_avg_5_ago = sum(prices[-10:-5]) / 5 price_avg_now = sum(prices[-5:]) / 5 ema21_falling = price_avg_now < price_avg_5_ago if ema21_falling and ema_alignment == 'bearish': is_bearish_trend = True bearish_reasons.append("Prix < EMA21 descendante (tendance baissière continue)") # Critère 7: Score IA faible = pas d'achat if ai_score and ai_score < 50: is_bearish_trend = True bearish_reasons.append(f"Score IA trop faible ({ai_score})") # ═══════════════════════════════════════════════════════════════════ # STRATÉGIE D'ACHAT: "BUY THE DIP" + "BOLLINGER SQUEEZE" # ═══════════════════════════════════════════════════════════════════ buy_signals = 0 buy_reasons = [] # STRATÉGIE 1: CREUX EMA (Buy the Dip) # EMA9 < EMA21 = creux = opportunité d'achat if ema_alignment == 'bearish' and ema_short and ema_long: ema_gap = ((ema_short - ema_long) / ema_long) * 100 buy_signals += 2 # Signal fort buy_reasons.append(f"CREUX EMA ({ema_gap:.2f}%)") # RSI pas trop haut (< 60) = bon pour acheter if rsi and rsi < 60: buy_signals += 1 buy_reasons.append(f"RSI OK ({rsi:.1f})") # Prix proche bande basse = encore mieux if bb_lower and current_price < bb_lower * 1.03: buy_signals += 1 buy_reasons.append("Proche BB basse") # Début de stabilisation/rebond (momentum pas trop négatif) if momentum and momentum > -1.5: buy_signals += 1 buy_reasons.append("Stabilisation") # STRATÉGIE 2: BOLLINGER SQUEEZE (compression avant explosion) # Bandes serrées + EMA bullish ou neutre + breakout vers le haut elif is_squeeze or squeeze_strength > 40: # Squeeze détecté - faible volatilité, explosion imminente if squeeze_breakout == 'up': # Breakout vers le haut confirmé! buy_signals += 3 # Signal très fort buy_reasons.append(f"SQUEEZE BREAKOUT UP (force={squeeze_strength:.0f}%)") # EMA en bonne disposition (pas fortement bearish) if ema_alignment == 'bullish' or (ema_slope and ema_slope > -0.3): buy_signals += 2 buy_reasons.append("EMA favorable") # RSI pas en surachat = encore de la marge if rsi and rsi < 65: buy_signals += 1 buy_reasons.append(f"RSI marge ({rsi:.0f})") elif squeeze_strength > 50 and ema_alignment == 'bullish': # Squeeze fort avec EMA bullish = prêt à exploser buy_signals += 2 buy_reasons.append(f"SQUEEZE TENSION (force={squeeze_strength:.0f}%)") # Momentum positif = direction confirmée if momentum and momentum > 0: buy_signals += 1 buy_reasons.append("Momentum+") # STRATÉGIE 3: PULLBACK sur EMA bullish (légère correction dans tendance haussière) elif ema_alignment == 'bullish' and is_pullback and pullback_distance: if pullback_distance < 1.5: # Correction légère (< 1.5%) buy_signals += 2 buy_reasons.append(f"PULLBACK EMA ({pullback_distance:.1f}%)") if rsi and rsi < 55: buy_signals += 1 buy_reasons.append(f"RSI cool ({rsi:.0f})") # STRATÉGIE 4: TREND FOLLOWING (achat sur cassure haussière) # Conditions : EMA9 > EMA21, momentum positif, RSI < 75, score IA >= 50 elif ema_alignment == 'bullish' and ema_short and ema_long and ema_short > ema_long: if momentum and momentum > 0: if rsi and rsi < 75: if ai_score and ai_score >= 50: # Critère volume : bonus si volume > moyenne des 10 dernières bougies vol_ok = False if 'volumes' in locals() or 'volumes' in globals(): vols = volumes if 'volumes' in locals() else globals().get('volumes', []) if vols and len(vols) >= 10: avg_vol = sum(vols[-10:]) / 10 if volumes[-1] > avg_vol * 1.2: vol_ok = True if vol_ok: buy_signals += 2 buy_reasons.append("Volume break: volume > moyenne x1.2") buy_signals += 2 buy_reasons.append(f"TREND FOLLOWING: EMA9>EMA21, momentum+, RSI={rsi:.1f}, AI={ai_score}") # ═══════════════════════════════════════════════════════════════════ # RÈGLE #3: SIGNAUX DE VENTE (PROTÉGER LE CAPITAL) # ═══════════════════════════════════════════════════════════════════ sell_signals = 0 sell_reasons = [] # RSI surachat extrême if rsi and rsi > 80: sell_signals += 2 sell_reasons.append(f"RSI surachat extreme ({rsi:.1f})") elif rsi and rsi > RSI_OVERBOUGHT: sell_signals += 1 sell_reasons.append(f"RSI surachat ({rsi:.1f})") # Prix au-dessus bande haute Bollinger if bb_upper and current_price > bb_upper: sell_signals += 1 sell_reasons.append("Prix > BB haute") # Tendance baissière confirmée (pour positions ouvertes) if is_bearish_trend and len(bearish_reasons) >= 2: sell_signals += 2 sell_reasons.append("Tendance baissiere confirmee") # Momentum très négatif (chute rapide) if momentum and momentum < -3: sell_signals += 2 sell_reasons.append(f"Chute rapide ({momentum:.1f}%)") elif momentum and momentum < -1.5: sell_signals += 1 sell_reasons.append(f"Momentum negatif ({momentum:.1f}%)") # EMA en croisement baissier (death cross) if ema_short and ema_long and ema_short < ema_long: sell_signals += 1 sell_reasons.append("EMA death cross") # Prix sous les deux EMA = signal faible if ema_short and ema_long and current_price < ema_short and current_price < ema_long: sell_signals += 1 sell_reasons.append("Prix sous EMAs") # ═══════════════════════════════════════════════════════════════════ # DÉCISION FINALE # ═══════════════════════════════════════════════════════════════════ indicators = { 'rsi': rsi, 'ema_short': ema_short, 'ema_long': ema_long, 'bb': (bb_upper, bb_mid, bb_lower), 'bb_direction': bb_direction, 'bb_bandwidth': bb_bandwidth, # Largeur des bandes 'ema_alignment': ema_alignment, 'ema_slope': ema_slope, 'price': current_price, 'momentum': momentum, 'is_breakout': is_breakout, 'breakout_strength': breakout_strength, 'is_pullback': is_pullback, 'is_squeeze': is_squeeze, # NOUVEAU: Bollinger Squeeze 'squeeze_strength': squeeze_strength, # Force du squeeze 'squeeze_breakout': squeeze_breakout, # Direction du breakout 'ai_score': ai_score, 'is_bearish': is_bearish_trend, 'bearish_reasons': bearish_reasons, 'buy_signals': buy_signals, 'sell_signals': sell_signals } # Nombre de positions ouvertes current_positions = len(self.position_manager.positions) # ═══════════════════════════════════════════════════════════════════ # LOGIQUE "BUY THE DIP" - Acheter les creux, vendre les rebonds # ═══════════════════════════════════════════════════════════════════ # NOTE: La logique de vente est maintenant à la FIN de la fonction # pour s'assurer qu'on ne vend PAS en perte sauf stop-loss # 2. Vérifications avant achat can_buy = True block_reasons = [] # ═══════════════════════════════════════════════════════════════════ # BLOCAGE PRIORITAIRE: TENDANCE BAISSIÈRE = PAS D'ACHAT # ═══════════════════════════════════════════════════════════════════ if is_bearish_trend: can_buy = False block_reasons.extend(bearish_reasons) # ═══════════════════════════════════════════════════════════════════ # STRATÉGIE "BUY THE DIP" - Acheter le creux, vendre le rebond # ═══════════════════════════════════════════════════════════════════ # LOGIQUE: Acheter quand EMA9 < EMA21 (creux), vendre quand EMA9 > EMA21 (rebond) # CRITÈRE 1: EMA9 DOIT être EN DESSOUS de EMA21 (on achète le creux!) if not (ema_short and ema_long and ema_short < ema_long): can_buy = False block_reasons.append("EMA9 >= EMA21 (attendre le creux)") # CRITÈRE 2: Le creux doit être significatif (EMA9 au moins 0.1% sous EMA21) if ema_short and ema_long: ema_gap = ((ema_short - ema_long) / ema_long) * 100 # Le creux doit être assez profond pour être significatif if ema_gap > -0.1: # Si EMA9 n'est pas au moins 0.1% sous EMA21 can_buy = False block_reasons.append(f"Creux EMA trop faible ({ema_gap:.2f}%)") # CRITÈRE 3: RSI pas en surachat (< 60 acceptable pour un creux) if rsi and rsi > 60: can_buy = False block_reasons.append(f"RSI trop haut ({rsi:.0f})") # CRITÈRE 4: SUPPRIMÉ - BB basse trop restrictif # Le creux EMA est suffisant, pas besoin d'être sur BB basse # CRITÈRE 5: Le prix montre des signes de stabilisation ou rebond # Pas forcément en hausse, mais pas en chute libre if len(prices) >= 3: last_3_change = ((current_price - prices[-3]) / prices[-3]) * 100 if last_3_change < -0.5: # Tolérer très petite baisse seulement can_buy = False block_reasons.append(f"Encore en baisse ({last_3_change:.2f}%)") # CRITÈRE 5b: Momentum 10 bougies doit être positif ou neutre if len(prices) >= 10: last_10_change = ((current_price - prices[-10]) / prices[-10]) * 100 if last_10_change < -1.0: # Baisse sur 10 bougies can_buy = False block_reasons.append(f"Momentum 10 négatif ({last_10_change:.2f}%)") # CRITÈRE 6: Score IA pour confirmation # Bloquer si l'IA détecte un pattern fortement négatif if ai_score < 35: can_buy = False block_reasons.append(f"Score IA trop bas ({ai_score})") elif ai_score >= 60: # Bonus de confiance si IA positive buy_signals += 1 buy_reasons.append(f"IA: Score {ai_score}") # CRITÈRE 7: Pas en chute libre (seuil assoupli) if len(prices) >= 5: last_5_change = ((current_price - prices[-5]) / prices[-5]) * 100 if last_5_change < -3: # Chute de 3% = trop risqué can_buy = False block_reasons.append(f"Chute trop forte ({last_5_change:.2f}%)") # CRITÈRE 8: TENDANCE GLOBALE sur 20 bougies - LE PLUS IMPORTANT # Si le prix a baissé de plus de 0.8% sur 20 bougies, c'est une tendance baissière # On NE DOIT PAS acheter dans une tendance baissière prolongée if len(prices) >= 20: last_20_change = ((current_price - prices[-20]) / prices[-20]) * 100 if last_20_change < -0.8: # Baisse globale > 0.8% can_buy = False block_reasons.append(f"Tendance 20min baissière ({last_20_change:.2f}%)") # Récupérer max_positions depuis settings max_positions = self.settings.get('maxPositions', MAX_OPEN_POSITIONS) # ═══════════════════════════════════════════════════════════════════ # 3a. STRATÉGIE PRIORITAIRE: GOLDEN CROSS (croisement EMA très récent) # Détecter le croisement EMA9 > EMA21 dans les 2-3 dernières bougies # C'est le signal le plus réactif pour capturer le début d'une hausse # ═══════════════════════════════════════════════════════════════════ if current_positions < max_positions and golden_cross_recent: # Golden Cross détecté - conditions minimales pour réactivité maximale # On vérifie juste qu'on n'est pas en chute libre last_3_ok = True if len(prices) >= 3: last_3_change = ((current_price - prices[-3]) / prices[-3]) * 100 last_3_ok = last_3_change > -1.0 # Tolérance if last_3_ok and rsi and rsi < 75: ema_gap_cross = ((ema_short - ema_long) / ema_long) * 100 reason = f"⚡ GOLDEN CROSS: EMA9 croise EMA21 (+{ema_gap_cross:.2f}%), RSI={rsi:.0f}" return "BUY", indicators, reason # 3b. Signal d'achat si EMA en creux + pas de blocage + IA favorable # PRIORISÉ par score IA pour des achats plus intelligents if can_buy and current_positions < max_positions: reason = f"BUY THE DIP: EMA9 EMA21 (tendance haussière) # - Momentum positif # - RSI pas en surachat extrême (< 75) # - Score IA >= 50 # - Pas en chute sur les 3 dernières bougies if ema_short and ema_long and ema_short > ema_long: ema_gap_bullish = ((ema_short - ema_long) / ema_long) * 100 last_3_ok = True if len(prices) >= 3: last_3_change = ((current_price - prices[-3]) / prices[-3]) * 100 last_3_ok = last_3_change > -0.5 # Tolérance légère baisse if momentum and momentum > 0 and rsi and rsi < 75 and ai_score >= 50 and last_3_ok: reason = f"TREND FOLLOWING: EMA9>EMA21 (+{ema_gap_bullish:.2f}%), Momentum+, RSI={rsi:.0f}, AI={ai_score}" return "BUY", indicators, reason # ═══════════════════════════════════════════════════════════════════ # 4. Signal de vente - VENDRE QUAND ON A FAIT UN PROFIT SIGNIFICATIF # ═══════════════════════════════════════════════════════════════════ if symbol in self.position_manager.positions: position = self.position_manager.positions.get(symbol) if position: entry_price = position.get('entry_price', current_price) pnl_pct = ((current_price - entry_price) / entry_price) * 100 # CONDITION 1: Take Profit atteint (+2% minimum) if pnl_pct >= 2.0: reason = f"TAKE PROFIT: +{pnl_pct:.2f}%" return "SELL", indicators, reason # CONDITION 2: Golden Cross SIGNIFICATIF (EMA9 > EMA21 de +0.3% minimum) # ET on est en profit if ema_short and ema_long and ema_short > ema_long: ema_gap = ((ema_short - ema_long) / ema_long) * 100 if ema_gap > 0.3 and pnl_pct > 0: # Gap significatif ET profit reason = f"SELL REBOND: EMA9>EMA21 (+{ema_gap:.2f}%), P&L: +{pnl_pct:.2f}%" return "SELL", indicators, reason # CONDITION 3: RSI très élevé (surachat) = vendre si en profit if rsi and rsi > 70 and pnl_pct > 0: reason = f"SELL SURACHAT: RSI={rsi:.0f}, P&L: +{pnl_pct:.2f}%" return "SELL", indicators, reason # CONDITION 4: Stop Loss technique (baisse de 2% depuis l'entrée) if pnl_pct < -2.0: reason = f"STOP LOSS: {pnl_pct:.2f}%" return "SELL", indicators, reason # 5. Défaut = HOLD return "HOLD", indicators, "Conditions non remplies" def execute_signal(self, symbol, signal, indicators=None, reason=""): """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" Cooldown signal {symbol}: encore {remaining}s") return # Vérifier le cooldown de trade (5 min après chaque trade) if symbol in self.last_trade: elapsed = time.time() - self.last_trade[symbol] if elapsed < self.trade_cooldown: remaining = int(self.trade_cooldown - elapsed) print(f" Cooldown trade {symbol}: 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 ({MAX_RISK_PER_TRADE}%)={currency.format(max_risk_amount)} | Position={currency.format(position_size)} | Ordre={currency.format(order_amount)}") if order_amount >= MIN_ORDER_SIZE: print(f"\nSIGNAL ACHAT: {symbol} (Position {current_positions+1}/{max_positions})") # TOUJOURS utiliser les valeurs de config.py (fichier maître) stop_loss_pct = STOP_LOSS_PERCENT take_profit_pct = TAKE_PROFIT_PERCENT self.position_manager.open_position(symbol, "BUY", order_amount, stop_loss_pct, take_profit_pct) self.last_signal[symbol] = time.time() self.last_trade[symbol] = time.time() # Cooldown de trade 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"\nSIGNAL VENTE: {symbol}") self.position_manager.close_position(symbol, "signal") self.last_signal[symbol] = time.time() self.last_trade[symbol] = time.time() # Cooldown de trade 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, reason = self.analyze(symbol) price = indicators.get('price', 0) rsi = indicators.get('rsi', 0) or 50 ai_score = indicators.get('ai_score', 50) bb_direction = indicators.get('bb_direction', 'flat') ema_alignment = indicators.get('ema_alignment', 'mixed') buy_signals_count = indicators.get('buy_signals', 0) is_bearish = indicators.get('is_bearish', False) # Indicateurs visuels trend_icon = "📈" if ema_alignment == 'bullish' else "📉" if ema_alignment == 'bearish' else "➡️" bb_icon = "⬆️" if bb_direction == 'up' else "⬇️" if bb_direction == 'down' else "➖" in_position = "📍" if symbol in self.position_manager.positions else "" icon = "🟢" if signal == "BUY" else "🔴" if signal == "SELL" else "🚫" if is_bearish else "⚪" # Affichage compact mais informatif print(f" {icon} {symbol}{in_position}: {currency.format(price)} | RSI:{rsi:.0f} AI:{ai_score} | {trend_icon}EMA {bb_icon}BB | Sig:{buy_signals_count}") # 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):") # P&L déjà en EUR, pas besoin de conversion print(f" {pnl_icon} P&L Total: {stats['total_pnl']:+.2f}€") 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 "🔴" # P&L déjà en EUR, pas besoin de conversion print(f" {t_icon} {t['symbol']}: {t['pnl']:+.2f}€ ({t['pnl_pct']:+.2f}%) - {t['reason']}") print("\n" + "=" * 70) print(" [Ctrl+C pour arrêter]") # Sauvegarder les données d'analyse pour le dashboard self.save_analysis_data() def save_analysis_data(self): """Sauvegarde les données d'analyse en temps réel pour le dashboard""" try: current_positions = len(self.position_manager.positions) max_positions = self.settings.get('maxPositions', MAX_OPEN_POSITIONS) stats = self.position_manager.get_total_pnl() # Données des cryptos analysées - utiliser les VRAIES klines de PRODUCTION pour RSI précis cryptos_data = [] for symbol in self.watch_symbols: # Récupérer les vraies klines de PRODUCTION (pas testnet qui manque de données) try: klines = self.client.get_klines_production(symbol, DEFAULT_INTERVAL, 50) if klines and len(klines) >= 20: klines_prices = [float(k[4]) for k in klines] # Prix de clôture # Calculer le RSI avec les vrais prix de bougies real_rsi = TechnicalIndicators.rsi(klines_prices, RSI_PERIOD) else: real_rsi = None except: real_rsi = None prices = list(self.prices.get(symbol, [])) if len(prices) < 20: continue signal, indicators, reason = self.analyze(symbol) price = indicators.get('price', 0) # Utiliser le RSI des klines si disponible, sinon celui de l'analyse rsi = real_rsi if real_rsi is not None else (indicators.get('rsi', 0) or 50) ai_score = indicators.get('ai_score', 50) bb_direction = indicators.get('bb_direction', 'flat') ema_alignment = indicators.get('ema_alignment', 'mixed') buy_signals_count = indicators.get('buy_signals', 0) is_bearish = indicators.get('is_bearish', False) momentum = indicators.get('momentum', 0) or 0 in_position = symbol in self.position_manager.positions cryptos_data.append({ 'symbol': symbol, 'price': price, 'rsi': round(rsi, 1), 'aiScore': ai_score, 'emaAlignment': ema_alignment, 'bbDirection': bb_direction, 'buySignals': buy_signals_count, 'signal': signal, 'isBearish': is_bearish, 'inPosition': in_position, 'momentum': round(momentum, 2), 'reason': reason }) # Positions ouvertes avec détails positions_data = [] 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 positions_data.append({ 'symbol': symbol, 'quantity': pos['quantity'], 'entryPrice': pos['entry_price'], 'currentPrice': current, 'pnlPct': round(pnl_pct, 2) }) # Logs récents (derniers trades) logs = [] for t in self.position_manager.trade_history[-10:]: log_type = 'success' if t['pnl'] >= 0 else 'error' logs.append({ 'type': log_type, 'message': f"{t['symbol']}: {t['pnl']:+.2f}€ ({t['pnl_pct']:+.2f}%) - {t['reason']}", 'timestamp': t.get('exit_time', datetime.now().isoformat()) }) # Compter les signaux buy_signals_total = sum(1 for c in cryptos_data if c['signal'] == 'BUY') sell_signals_total = sum(1 for c in cryptos_data if c['signal'] == 'SELL') # Analyse du marché - pourquoi le bot n'achète pas total_cryptos = len(cryptos_data) bearish_count = sum(1 for c in cryptos_data if c['isBearish']) bullish_count = total_cryptos - bearish_count rsi_oversold = [c for c in cryptos_data if c['rsi'] < RSI_OVERSOLD] rsi_overbought = [c for c in cryptos_data if c['rsi'] > RSI_OVERBOUGHT] high_ai = [c for c in cryptos_data if c['aiScore'] >= MIN_AI_SCORE_FOR_BUY] with_signals = [c for c in cryptos_data if c['buySignals'] >= MIN_BUY_SIGNALS] # Identifier les blocages blockers = [] if bearish_count > total_cryptos * 0.7: blockers.append(f"Marché baissier: {bearish_count}/{total_cryptos} cryptos en tendance baissière ({bearish_count*100//total_cryptos}%)") if len(with_signals) == 0: blockers.append(f"Aucune crypto n'atteint {MIN_BUY_SIGNALS} signaux d'achat (paramètre MIN_BUY_SIGNALS)") if len(high_ai) == 0: blockers.append(f"Aucune crypto n'a un AI Score >= {MIN_AI_SCORE_FOR_BUY} (paramètre MIN_AI_SCORE)") if len(rsi_oversold) == 0: blockers.append(f"Aucune crypto en survente (RSI < {RSI_OVERSOLD})") # Top opportunités (même si bloquées) top_opportunities = sorted(cryptos_data, key=lambda x: (-x['buySignals'], -x['aiScore']))[:5] market_analysis = { 'totalCryptos': total_cryptos, 'bearishCount': bearish_count, 'bullishCount': bullish_count, 'bearishPercent': round(bearish_count * 100 / total_cryptos, 1) if total_cryptos > 0 else 0, 'rsiOversold': len(rsi_oversold), 'rsiOverbought': len(rsi_overbought), 'highAiCount': len(high_ai), 'withSignalsCount': len(with_signals), 'blockers': blockers, 'topOpportunities': [{ 'symbol': c['symbol'], 'buySignals': c['buySignals'], 'aiScore': c['aiScore'], 'rsi': c['rsi'], 'isBearish': c['isBearish'], 'reason': c['reason'] } for c in top_opportunities], 'requirements': { 'minBuySignals': MIN_BUY_SIGNALS, 'minAiScore': MIN_AI_SCORE_FOR_BUY, 'rsiOversold': RSI_OVERSOLD, 'rsiOverbought': RSI_OVERBOUGHT } } # Données complètes analysis_data = { 'timestamp': datetime.now().isoformat(), 'stats': { 'positions': current_positions, 'maxPositions': max_positions, 'winRate': round(stats.get('win_rate', 0), 1), 'pnl': round(stats.get('total_pnl', 0), 2), 'totalTrades': stats.get('total_trades', 0), 'wins': stats.get('wins', 0), 'losses': stats.get('losses', 0) }, 'signals': { 'buy': buy_signals_total, 'sell': sell_signals_total }, 'marketAnalysis': market_analysis, 'positions': positions_data, 'cryptos': cryptos_data, 'logs': logs, 'settings': { 'autoTrade': self.settings.get('autoTrade', True), 'stopLoss': STOP_LOSS_PERCENT, 'takeProfit': TAKE_PROFIT_PERCENT, 'testnet': TESTNET_MODE } } # Sauvegarder dans le fichier analysis_file = os.path.join(SCRIPT_DIR, 'bot_analysis.json') with open(analysis_file, 'w', encoding='utf-8') as f: json.dump(analysis_data, f, ensure_ascii=False, indent=2) except Exception as e: # Silencieux pour ne pas perturber le bot pass async def price_updater(self, symbol): """Met à jour les prix via WebSocket (ou API en testnet)""" # En mode testnet, pas de WebSocket disponible, utiliser l'API REST if TESTNET_MODE: while self.running: try: price = self.client.get_price(symbol) if price: self.prices[symbol].append(price) await asyncio.sleep(2) # Mise à jour toutes les 2 secondes except Exception as e: if self.running: await asyncio.sleep(5) else: # Mode production: utiliser 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 klines_refresh_counter = 0 while self.running: # Rafraîchir les klines toutes les 60 itérations (~2 min) pour RSI fiable klines_refresh_counter += 1 if klines_refresh_counter >= 60: try: for symbol in self.watch_symbols[:20]: # Limiter pour performance klines = self.client.get_klines(symbol, DEFAULT_INTERVAL, 50) if klines: # Remplacer les données par les vrais prix de clôture des bougies self.prices[symbol] = deque([float(k[4]) for k in klines], maxlen=100) except Exception as e: pass # Silencieux klines_refresh_counter = 0 # En mode testnet, ajouter le dernier prix pour réactivité if TESTNET_MODE: try: all_prices = self.client.get_all_prices() for symbol in self.watch_symbols: if symbol in all_prices: self.prices[symbol].append(all_prices[symbol]) except Exception as e: print(f" Erreur récupération prix: {e}") # Vérifier stop-loss / take-profit self.position_manager.check_stop_loss_take_profit() # ═══════════════════════════════════════════════════════════════ # VÉRIFICATION DU SIGNAL "VENDRE TOUT" depuis le Dashboard # ═══════════════════════════════════════════════════════════════ sell_all_file = os.path.join(SCRIPT_DIR, 'sell_all_signal.json') if os.path.exists(sell_all_file): try: with open(sell_all_file, 'r') as f: signal = json.load(f) if signal.get('action') == 'SELL_ALL': print(f"\n🚨 SIGNAL VENDRE TOUT reçu du Dashboard!") positions_to_sell = list(self.position_manager.positions.keys()) if positions_to_sell: print(f" Vente de {len(positions_to_sell)} positions: {positions_to_sell}") for symbol in positions_to_sell: try: result = self.position_manager.close_position(symbol, "manual_sell_all") if result: print(f" ✅ {symbol} vendu") else: print(f" ❌ {symbol} échec de vente") except Exception as e: print(f" ❌ {symbol} erreur: {e}") print(f" 📊 Vente terminée!") else: print(f" ℹ️ Aucune position ouverte à vendre") # Supprimer le fichier signal après traitement os.remove(sell_all_file) print(f" 🗑️ Signal traité et supprimé") except Exception as e: print(f" ⚠️ Erreur traitement sell_all: {e}") # Supprimer le fichier même en cas d'erreur pour éviter boucle try: os.remove(sell_all_file) except: pass # ═══════════════════════════════════════════════════════════════ # QUICK EXIT: Vérification technique anticipée pour les positions # ═══════════════════════════════════════════════════════════════ for symbol in list(self.position_manager.positions.keys()): if symbol in self.prices and len(self.prices[symbol]) >= 21: should_exit, exit_reason = self.position_manager.check_technical_exit(symbol, self.prices[symbol]) if should_exit: print(f"\n ⚡ {symbol}: {exit_reason}") self.position_manager.close_position(symbol, "quick-exit") self.last_trade[symbol] = time.time() # 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 # ═══════════════════════════════════════════════════════════════ # PRIORITÉ 1: SIGNAUX IA "READY" (Score >= 70, détectés par surveillance) # ═══════════════════════════════════════════════════════════════ ai_buy_candidates = [] current_positions = len(self.position_manager.positions) max_positions = self.settings.get('maxPositions', MAX_OPEN_POSITIONS) if self.surveillance_service and current_positions < max_positions: try: ai_status = self.surveillance_service.get_surveillance_status() ready_signals = ai_status.get('ready_signals', []) for sig in ready_signals: symbol = sig.get('symbol') score = sig.get('score', 0) status = sig.get('status', '') pattern = sig.get('pattern', 'NEUTRAL') # Signal IA "ready" avec score >= 70 = ACHAT PRIORITAIRE if status == 'ready' and score >= 70 and symbol not in self.position_manager.positions: # Vérifier le cooldown if symbol in self.last_trade: elapsed = time.time() - self.last_trade[symbol] if elapsed < self.trade_cooldown: continue # ══════════════════════════════════════════════════════ # VÉRIFICATION OBLIGATOIRE: EMA9 < EMA21 (Buy the Dip) # ══════════════════════════════════════════════════════ prices = list(self.prices.get(symbol, [])) if len(prices) >= 20: # Calculer les EMAs ema_short = TechnicalIndicators.ema(prices, EMA_SHORT) ema_long = TechnicalIndicators.ema(prices, EMA_LONG) # RÈGLE FONDAMENTALE: On achète SEULEMENT le creux (EMA9 < EMA21) if ema_short and ema_long: if ema_short >= ema_long: # EMA9 >= EMA21 = PAS un creux = NE PAS ACHETER continue # Vérifier aussi qu'on n'est pas en chute libre tech_signal, indicators, tech_reason = self.analyze(symbol, []) if tech_signal != 'BUY': continue features = sig.get('features', {}) rsi = features.get('rsi', 50) reason = sig.get('reason', f'IA: {pattern}') ai_buy_candidates.append({ 'symbol': symbol, 'signal': 'BUY', 'indicators': {'ai_score': score, 'rsi': rsi, 'pattern': pattern}, 'reason': f"🤖 IA SIGNAL: {pattern} (Score={score}, RSI={rsi:.0f})", 'ai_score': score, 'rsi': rsi, 'priority': score + (30 if rsi < 30 else 15 if rsi < 40 else 0), 'from_ai_service': True }) except Exception as e: pass # Silencieux si erreur # Traiter les signaux IA en PRIORITÉ - ACHATS MULTIPLES if ai_buy_candidates: ai_buy_candidates.sort(key=lambda x: x['priority'], reverse=True) print(f"\n🤖 SIGNAUX IA PRÊTS ({len(ai_buy_candidates)}):") for i, cand in enumerate(ai_buy_candidates[:5]): icon = "🔥" if i == 0 else "⚡" if i == 1 else "✨" print(f" {icon} {cand['symbol']}: Score={cand['ai_score']} Pattern={cand['indicators'].get('pattern')} RSI={cand['rsi']:.0f}") # Exécuter jusqu'à 3 signaux IA par cycle (accélération) max_buys_per_cycle = 3 buys_done = 0 for best_ai in ai_buy_candidates: if current_positions < max_positions and buys_done < max_buys_per_cycle: if best_ai['symbol'] not in self.position_manager.positions: print(f"\n🤖 ACHAT IA: {best_ai['symbol']} (Score: {best_ai['priority']})") self.execute_signal(best_ai['symbol'], 'BUY', best_ai['indicators'], best_ai['reason']) buys_done += 1 current_positions += 1 # ═══════════════════════════════════════════════════════════════ # PRIORITÉ 2: ANALYSE TECHNIQUE (EMA + RSI + Bollinger) # ═══════════════════════════════════════════════════════════════ buy_candidates = [] sell_signals_list = [] # Symboles déjà traités par IA (éviter double traitement) ai_processed_symbols = {c['symbol'] for c in ai_buy_candidates} if ai_buy_candidates else set() for symbol in self.watch_symbols: if symbol not in self.prices: self.prices[symbol] = deque(maxlen=100) signal, indicators, reason = self.analyze(symbol) if signal == "BUY": # Ignorer si déjà traité par signaux IA if symbol in ai_processed_symbols: continue ai_score = indicators.get('ai_score', 50) rsi = indicators.get('rsi', 50) or 50 buy_candidates.append({ 'symbol': symbol, 'signal': signal, 'indicators': indicators, 'reason': reason, 'ai_score': ai_score, 'rsi': rsi, # Score combiné: AI + bonus RSI oversold 'priority': ai_score + (20 if rsi < 30 else 10 if rsi < 40 else 0) }) elif signal == "SELL" and symbol in self.position_manager.positions: print(f" ⚠️ {symbol}: {reason}") sell_signals_list.append((symbol, signal, indicators, reason)) # Exécuter d'abord les VENTES (protéger le capital) for symbol, signal, indicators, reason in sell_signals_list: self.execute_signal(symbol, signal, indicators, reason) # Trier les candidats d'achat par priorité (score IA + RSI) buy_candidates.sort(key=lambda x: x['priority'], reverse=True) # Afficher les top opportunités if buy_candidates: print(f"\n🎯 TOP Opportunités IA ({len(buy_candidates)} signaux):") for i, cand in enumerate(buy_candidates[:5]): icon = "🥇" if i == 0 else "🥈" if i == 1 else "🥉" if i == 2 else "📍" print(f" {icon} {cand['symbol']}: AI={cand['ai_score']} RSI={cand['rsi']:.0f} | {cand['reason']}") # Exécuter UNIQUEMENT le meilleur candidat (éviter de disperser le capital) current_positions = len(self.position_manager.positions) max_positions = self.settings.get('maxPositions', MAX_OPEN_POSITIONS) # Acheter aussi via analyse technique (jusqu'à 2 de plus) if buy_candidates and current_positions < max_positions: tech_buys = 0 for best in buy_candidates: if current_positions >= max_positions or tech_buys >= 2: break if best['symbol'] in self.position_manager.positions: continue if best['symbol'] in ai_processed_symbols: continue # Score IA >= 40 ou RSI très bas if best['ai_score'] >= 40 or best['rsi'] < 25: print(f"\n💎 SÉLECTION TECH: {best['symbol']} (Score: {best['priority']})") self.execute_signal(best['symbol'], best['signal'], best['indicators'], best['reason']) tech_buys += 1 current_positions += 1 else: print(f"\n⏳ Attente: {best['symbol']} (AI={best['ai_score']}) sous seuil") break # Arrêter si le meilleur n'est pas assez bon # Log récapitulatif if buy_candidates: print(f"\n💡 {len(buy_candidates)} signaux BUY | 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() # Démarrer le service de surveillance IA if self.surveillance_service: self.surveillance_service.start() print(" 🤖 Surveillance IA démarrée") self.running = True # En mode testnet, optimiser: une seule tâche pour tous les prix if TESTNET_MODE: print(" ⚡ Mode TESTNET optimisé (API REST au lieu de WebSocket)") tasks = [ asyncio.create_task(self.trading_loop()) ] else: # Mode production: WebSocket pour chaque crypto tasks = [ asyncio.create_task(self.price_updater(s)) for s in self.watch_symbols ] tasks.append(asyncio.create_task(self.trading_loop())) try: print(f" ▶️ Lancement de {len(tasks)} tâche(s) asynchrone(s)...") await asyncio.gather(*tasks) except asyncio.CancelledError: print(" ⚠️ Tâches annulées") pass except Exception as e: print(f" ❌ Erreur dans la boucle principale: {e}") import traceback traceback.print_exc() finally: self.running = False def main(): # Le logging fichier est déjà configuré en début de fichier (FileLogger) # Ne pas toucher à sys.stdout ici 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 # Sauvegarder le PID pour permettre l'arrêt/redémarrage pid = os.getpid() pid_file = os.path.join(SCRIPT_DIR, "bot.pid") try: with open(pid_file, 'w') as f: f.write(str(pid)) print(f" 📝 PID sauvegardé: {pid}") except Exception as e: print(f" ⚠️ Impossible de sauvegarder le PID: {e}") bot = TradingBot() try: asyncio.run(bot.run()) except KeyboardInterrupt: print("\n\n👋 Bot arrêté.") finally: # Nettoyer le fichier PID à l'arrêt try: if os.path.exists(pid_file): os.remove(pid_file) print(" 🗑️ PID nettoyé") except: pass if __name__ == "__main__": main()