# -*- coding: utf-8 -*- """ Système de Signaux Pragmatique pour Trading Simplifié Génère des signaux de trading basés sur des indicateurs techniques simples et fiables """ import logging import time from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple from dataclasses import dataclass import numpy as np import requests import sys import os # Ajouter le chemin parent pour importer binance_data_provider parent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.append(parent_dir) from binance_data_provider import BinanceDataProvider from hyperliquid_leverages_cache import get_max_leverage @dataclass class TradingSignal: """Structure d'un signal de trading""" symbol: str side: str # 'LONG' ou 'SHORT' strength: float # 0.0 à 1.0 current_price: float leverage: int quality_score: float # 0.0 à 1.0 reason: str timestamp: datetime indicators: Dict = None target_price: float = None # Prix cible pour take profit target_pnl_percent: float = None # % de PnL cible (avec leverage) class MarketSentimentAnalyzer: """Analyse le sentiment global du marché pour ajuster les signaux""" def __init__(self): self.cache_duration = 300 # Cache de 5 minutes self.last_update = 0 self.market_sentiment = 0.0 # -1 (très bearish) à +1 (très bullish) self.major_coins = ['bitcoin', 'ethereum', 'solana', 'binancecoin'] def get_market_sentiment(self) -> float: """Retourne le sentiment du marché (-1 à +1)""" now = time.time() if now - self.last_update > self.cache_duration: self._update_sentiment() self.last_update = now return self.market_sentiment def _update_sentiment(self): """Met à jour le sentiment basé sur les variations 24h des majors""" try: # API CoinGecko pour les majors coins_str = ','.join(self.major_coins) url = f"https://api.coingecko.com/api/v3/simple/price?ids={coins_str}&vs_currencies=usd&include_24hr_change=true" response = requests.get(url, timeout=10) if response.status_code == 200: data = response.json() # Calculer sentiment basé sur les variations 24h changes = [] for coin in self.major_coins: if coin in data and 'usd_24h_change' in data[coin]: changes.append(data[coin]['usd_24h_change']) if changes: avg_change = np.mean(changes) # Normaliser entre -1 et +1 (seuils à -10% et +10%) self.market_sentiment = np.clip(avg_change / 10.0, -1.0, 1.0) print(f"📊 Market sentiment: {self.market_sentiment:.3f} (avg 24h: {avg_change:.2f}%)") else: self.market_sentiment = 0.0 else: print(f"⚠ Erreur sentiment API: {response.status_code}") self.market_sentiment = 0.0 except Exception as e: print(f"⚠ Erreur calcul sentiment: {e}") self.market_sentiment = 0.0 class PragmaticSignalGenerator: """ Générateur de signaux pragmatique basé sur des règles simples mais efficaces """ def __init__(self, hyperliquid_client=None, logger=None): self.hyperliquid_client = hyperliquid_client self.logger = logger or logging.getLogger(__name__) # Initialiser le fournisseur de données Binance pour les candles historiques self.binance_provider = BinanceDataProvider() self.logger.info("BinanceDataProvider initialisé pour les candles historiques") # Cache des leverages max pour éviter trop d'appels API self.leverage_cache = {} self.leverage_cache_time = None self._update_leverage_cache() # Analyseur de sentiment de marché self.sentiment_analyzer = MarketSentimentAnalyzer() # Mise à jour dynamique des marchés (toutes les 6 heures) self.last_market_update = 0 self.market_update_interval = 6 * 3600 # 6 heures # Configuration DYNAMIQUE des symboles depuis Hyperliquid self.symbols = self._get_tradable_symbols() self.symbols_last_update = time.time() self.symbols_update_interval = 3600 # Mise à jour toutes les heures self.min_volume_24h = 500000 # Volume minimum réduit pour couvrir plus de marchés # RSI plus réactifs self.rsi_oversold = 35 # Plus haut pour réagir plus vite self.rsi_overbought = 65 # Plus bas pour réagir plus vite self.rsi_extreme_oversold = 25 # Signal fort self.rsi_extreme_overbought = 75 # Signal fort # EMAs pour tendance rapide self.ema_fast = 5 # Plus rapide self.ema_short = 9 self.ema_long = 21 # Bollinger Bands ajustées self.bb_period = 20 self.bb_std = 2 # MACD plus sensible self.macd_fast = 8 self.macd_slow = 17 self.macd_signal = 9 # Cache des données pour éviter trop d'appels API self.price_cache = {} self.cache_duration = 10 # secondes self.logger.info("PragmaticSignalGenerator initialisé avec paramètres optimisés") def get_market_data(self, symbol: str) -> Optional[Dict]: """Récupère les données de marché avec cache - utilise Binance pour les candles historiques""" try: # Vérifier le cache if symbol in self.price_cache: cached_data, timestamp = self.price_cache[symbol] if time.time() - timestamp < self.cache_duration: return cached_data if not self.hyperliquid_client: # Mode simulation avec données fictives return self._get_mock_data(symbol) # Essayer Binance d'abord, puis fallback Hyperliquid candles = self.binance_provider.get_candles(symbol, interval='1h', limit=100) data_source = "Binance" if not candles or len(candles) < 50: # Fallback vers Hyperliquid pour les symboles non supportés par Binance candles = self.hyperliquid_client.get_candles_data(symbol, interval='1h', lookback_hours=100) data_source = "Hyperliquid" if not candles or len(candles) < 50: self.logger.debug(f"Pas assez de données pour {symbol} (Binance + Hyperliquid)") return None # Log de la source utilisée pour le debug if self.logger.isEnabledFor(logging.DEBUG): self.logger.debug(f"Données {symbol} récupérées depuis {data_source}") elif len(self.price_cache) % 20 == 0: # Log périodique self.logger.info(f"Données récupérées depuis {data_source} pour certains symboles") # Extraire les prix (format standardisé) closes = [float(c['close']) for c in candles] highs = [float(c['high']) for c in candles] lows = [float(c['low']) for c in candles] volumes = [float(c['volume']) for c in candles] current_price = closes[-1] volume_24h = sum(volumes[-24:]) * current_price if len(volumes) >= 24 else 0 data = { 'symbol': symbol, 'current_price': current_price, 'closes': closes, 'highs': highs, 'lows': lows, 'volume_24h': volume_24h, 'timestamp': datetime.now() } # Mettre en cache self.price_cache[symbol] = (data, time.time()) return data except Exception as e: self.logger.error(f"Erreur récupération données {symbol}: {e}") return None def _get_tradable_symbols(self) -> List[str]: """Récupère dynamiquement TOUS les symboles tradables disponibles sur Hyperliquid""" try: if self.hyperliquid_client: # Récupérer les symboles depuis Hyperliquid meta_info = self.hyperliquid_client.get_meta_info() if meta_info and 'universe' in meta_info: # Récupérer TOUS les symboles tradables all_symbols = [] for token in meta_info['universe']: symbol = token.get('name') # Vérifier que le token est réellement tradable if symbol and token.get('maxLeverage', 0) > 0: all_symbols.append(symbol) # Trier par priorité : majors en premier, puis alphabétique priority_symbols = ['BTC', 'ETH', 'SOL', 'BNB', 'AVAX', 'ARB', 'OP', 'LINK', 'UNI', 'DOT'] # Séparer les prioritaires des autres prioritized = [s for s in priority_symbols if s in all_symbols] others = sorted([s for s in all_symbols if s not in priority_symbols]) final_symbols = prioritized + others self.logger.info(f"Chargé {len(final_symbols)} symboles tradables depuis Hyperliquid") self.logger.debug(f"Symboles: {', '.join(final_symbols[:20])}{'...' if len(final_symbols) > 20 else ''}") return final_symbols # Fallback si pas de client self.logger.warning("Client Hyperliquid non disponible, utilisation symboles par défaut") return ['BTC', 'ETH', 'SOL', 'BNB', 'AVAX', 'ARB', 'OP', 'LINK', 'UNI', 'DOT'] except Exception as e: self.logger.error(f"Erreur récupération symboles: {e}") # Fallback minimal return ['BTC', 'ETH', 'SOL', 'BNB'] def _update_symbols_if_needed(self): """Met à jour la liste des symboles si nécessaire""" now = time.time() if now - self.symbols_last_update > self.symbols_update_interval: self.logger.info("Mise à jour de la liste des symboles...") new_symbols = self._get_tradable_symbols() if new_symbols: self.symbols = new_symbols self.symbols_last_update = now self.logger.info(f"Liste mise à jour: {len(self.symbols)} symboles") def _get_mock_data(self, symbol: str) -> Dict: """Génère des données fictives pour les tests""" import random base_prices = { 'BTC': 65000, 'ETH': 3500, 'SOL': 150, 'BNB': 600, 'AVAX': 40, 'ARB': 1.2, 'OP': 2.5 } base_price = base_prices.get(symbol, 100) volatility = 0.02 # Générer une série de prix closes = [] highs = [] lows = [] for i in range(100): change = random.gauss(0, volatility) price = base_price * (1 + change) high = price * (1 + abs(random.gauss(0, volatility/2))) low = price * (1 - abs(random.gauss(0, volatility/2))) closes.append(price) highs.append(high) lows.append(low) return { 'symbol': symbol, 'current_price': closes[-1], 'closes': closes, 'highs': highs, 'lows': lows, 'volume_24h': random.uniform(1000000, 10000000), 'timestamp': datetime.now() } def calculate_rsi(self, prices: List[float], period: int = 14) -> float: """Calcule le RSI""" if len(prices) < period + 1: return 50.0 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.0 rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) return rsi def calculate_ema(self, prices: List[float], period: int) -> float: """Calcule l'EMA""" if len(prices) < period: return prices[-1] if prices else 0 multiplier = 2 / (period + 1) ema = prices[-period] # Commence avec SMA for price in prices[-period+1:]: ema = (price * multiplier) + (ema * (1 - multiplier)) return ema def calculate_bollinger_bands(self, prices: List[float], period: int = 20, std_dev: int = 2) -> Tuple[float, float, float]: """Calcule les Bandes de Bollinger""" if len(prices) < period: current = prices[-1] if prices else 0 return current, current, current recent_prices = prices[-period:] middle = np.mean(recent_prices) std = np.std(recent_prices) upper = middle + (std_dev * std) lower = middle - (std_dev * std) return upper, middle, lower def calculate_macd(self, prices: List[float]) -> Tuple[float, float]: """Calcule le MACD""" if len(prices) < 26: return 0, 0 ema12 = self.calculate_ema(prices, 12) ema26 = self.calculate_ema(prices, 26) macd_line = ema12 - ema26 # Signal line (EMA 9 du MACD) macd_values = [] for i in range(26, len(prices)): ema12_temp = self.calculate_ema(prices[:i+1], 12) ema26_temp = self.calculate_ema(prices[:i+1], 26) macd_values.append(ema12_temp - ema26_temp) signal_line = self.calculate_ema(macd_values, 9) if len(macd_values) >= 9 else macd_line return macd_line, signal_line def analyze_symbol(self, symbol: str) -> Optional[TradingSignal]: """Analyse un symbole et génère un signal si opportunité""" try: # Récupérer les données data = self.get_market_data(symbol) if not data: self.logger.info(f"📊 {symbol}: REJETÉ (données indisponibles)") return None # Vérifier le volume minimum volume_24h = data['volume_24h'] if volume_24h < self.min_volume_24h: self.logger.info(f"📊 {symbol}: REJETÉ (volume trop faible: ${volume_24h:,.0f} < ${self.min_volume_24h:,.0f})") return None self.logger.info(f"📊 {symbol}: ANALYSE (volume OK: ${volume_24h:,.0f}, prix: ${data['current_price']:.6f})") closes = data['closes'] current_price = data['current_price'] # Calculer les indicateurs rsi = self.calculate_rsi(closes, period=14) rsi_fast = self.calculate_rsi(closes, period=7) # RSI rapide pour réactivité ema_fast = self.calculate_ema(closes, self.ema_fast) ema_short = self.calculate_ema(closes, self.ema_short) ema_long = self.calculate_ema(closes, self.ema_long) upper_bb, middle_bb, lower_bb = self.calculate_bollinger_bands(closes) macd, signal = self.calculate_macd(closes) # Analyse de tendance améliorée multi-timeframe trend, trend_strength, trend_quality = self._analyze_enhanced_trend(closes, ema_fast, ema_short, ema_long) # Vérifier la volatilité pour ajuster les paramètres volatility = (upper_bb - lower_bb) / middle_bb * 100 is_high_volatility = volatility > 4 # Plus de 4% de volatilité # Générer les signaux signals = [] # === SIGNAL LONG RÉACTIF === # Conditions multiples pour plus de réactivité long_conditions = [ (rsi < self.rsi_oversold, "RSI oversold"), (rsi_fast < 30, "RSI rapide oversold"), (current_price < lower_bb * 1.01, "Prix proche BB inférieure"), # Plus permissif (trend == "BULLISH" and trend_strength > 0.15, "Tendance haussière"), # Moins strict (macd > signal and macd > 0, "MACD bullish"), # Nouvelle condition: rebond depuis un support (len(closes) >= 2 and current_price > closes[-2] * 1.005, "Rebond détecté"), ] # Conditions renforcées pour plus de confiance strong_long_conditions = [ (rsi < self.rsi_extreme_oversold, "RSI extrême oversold"), (current_price < lower_bb * 0.98, "Prix bien sous BB"), (trend == "BULLISH" and trend_strength > 0.4, "Forte tendance haussière"), ] # Compter les conditions long_score = sum(1 for cond, _ in long_conditions if cond) strong_long_score = sum(1 for cond, _ in strong_long_conditions if cond) # Signal LONG si au moins 2 conditions normales OU 1 condition forte # MAIS seulement si la qualité de tendance est acceptable if (long_score >= 2 or strong_long_score >= 1) and trend_quality > 0.3: # Calculer la force et qualité basée sur le nombre de conditions strength = min(long_score / 5, 1.0) quality = 0.6 + (0.1 * long_score) # Intégrer la qualité de tendance dans le score final quality = quality * (0.7 + 0.3 * trend_quality) # Pondération trend_quality # Signal plus fort si RSI extrême ET bonne tendance if rsi < self.rsi_extreme_oversold and trend_quality > 0.5: strength = min(strength * 1.3, 1.0) quality = min(quality + 0.1, 1.0) # Pénalité pour signaux contre-tendance if trend == "BEARISH" and trend_quality > 0.6: quality *= 0.8 # Réduire qualité LONG sur forte tendance bearish # Raisons du signal reasons = [desc for cond, desc in long_conditions if cond] # Calculer le leverage et le target price leverage = self._calculate_smart_leverage(strength, volatility, symbol) target_price, target_pnl = self._calculate_target_price(current_price, 'LONG', leverage, strength) signals.append(TradingSignal( symbol=symbol, side='LONG', strength=strength, current_price=current_price, leverage=leverage, quality_score=min(quality, 0.95), reason=" + ".join(reasons[:2]), # Top 2 raisons timestamp=datetime.now(), target_price=target_price, target_pnl_percent=target_pnl, indicators={ 'rsi': rsi, 'rsi_fast': rsi_fast, 'ema_fast': ema_fast, 'ema_short': ema_short, 'ema_long': ema_long, 'bb_upper': upper_bb, 'bb_lower': lower_bb, 'trend': trend, 'volatility': volatility } )) # === SIGNAL SHORT RÉACTIF === short_conditions = [ (rsi > self.rsi_overbought, "RSI overbought"), (rsi_fast > 70, "RSI rapide overbought"), (current_price > upper_bb * 0.99, "Prix proche BB supérieure"), # Plus permissif (trend == "BEARISH" and trend_strength > 0.15, "Tendance baissière"), # Moins strict (macd < signal and macd < 0, "MACD bearish"), # Nouvelle condition: rejet depuis une résistance (len(closes) >= 2 and current_price < closes[-2] * 0.995, "Rejet détecté"), ] # Conditions renforcées pour plus de confiance strong_short_conditions = [ (rsi > self.rsi_extreme_overbought, "RSI extrême overbought"), (current_price > upper_bb * 1.02, "Prix bien au-dessus BB"), (trend == "BEARISH" and trend_strength > 0.4, "Forte tendance baissière"), ] # Compter les conditions short_score = sum(1 for cond, _ in short_conditions if cond) strong_short_score = sum(1 for cond, _ in strong_short_conditions if cond) # Signal SHORT si au moins 2 conditions normales OU 1 condition forte # MAIS seulement si la qualité de tendance est acceptable if (short_score >= 2 or strong_short_score >= 1) and trend_quality > 0.3: strength = min(short_score / 5, 1.0) quality = 0.6 + (0.1 * short_score) # Intégrer la qualité de tendance dans le score final quality = quality * (0.7 + 0.3 * trend_quality) # Pondération trend_quality # Signal plus fort si RSI extrême ET bonne tendance if rsi > self.rsi_extreme_overbought and trend_quality > 0.5: strength = min(strength * 1.3, 1.0) quality = min(quality + 0.1, 1.0) # Pénalité pour signaux contre-tendance if trend == "BULLISH" and trend_quality > 0.6: quality *= 0.8 # Réduire qualité SHORT sur forte tendance bullish reasons = [desc for cond, desc in short_conditions if cond] # Calculer le leverage et le target price leverage = self._calculate_smart_leverage(strength, volatility, symbol) target_price, target_pnl = self._calculate_target_price(current_price, 'SHORT', leverage, strength) signals.append(TradingSignal( symbol=symbol, side='SHORT', strength=strength, current_price=current_price, leverage=leverage, quality_score=min(quality, 0.95), reason=" + ".join(reasons[:2]), # Top 2 raisons timestamp=datetime.now(), target_price=target_price, target_pnl_percent=target_pnl, indicators={ 'rsi': rsi, 'rsi_fast': rsi_fast, 'ema_fast': ema_fast, 'ema_short': ema_short, 'ema_long': ema_long, 'bb_upper': upper_bb, 'bb_lower': lower_bb, 'trend': trend, 'volatility': volatility } )) # === SIGNAL MOMENTUM (très réactif) === # Détecter les mouvements rapides de prix if len(closes) >= 3: price_change_3bars = (current_price - closes[-3]) / closes[-3] * 100 # Signal momentum LONG if price_change_3bars < -2 and rsi_fast < 40 and trend != "BEARISH": signals.append(TradingSignal( symbol=symbol, side='LONG', strength=0.7, current_price=current_price, leverage=self._calculate_smart_leverage(0.7, volatility, symbol), quality_score=0.72, reason=f"Momentum: chute rapide {price_change_3bars:.1f}% + RSI bas", timestamp=datetime.now(), indicators={'price_change': price_change_3bars, 'rsi_fast': rsi_fast} )) # Signal momentum SHORT elif price_change_3bars > 2 and rsi_fast > 60 and trend != "BULLISH": signals.append(TradingSignal( symbol=symbol, side='SHORT', strength=0.7, current_price=current_price, leverage=self._calculate_smart_leverage(0.7, volatility, symbol), quality_score=0.72, reason=f"Momentum: hausse rapide {price_change_3bars:.1f}% + RSI haut", timestamp=datetime.now(), indicators={'price_change': price_change_3bars, 'rsi_fast': rsi_fast} )) # Filtrer les signaux selon leur adaptabilité aux règles suitable_signals = [] for sig in signals: is_suitable, reason = self.is_signal_suitable_for_rules(sig) if is_suitable: suitable_signals.append(sig) else: self.logger.debug(f"Signal {symbol} {sig.side} rejeté: {reason}") # Retourner le meilleur signal adapté if suitable_signals: best_signal = max(suitable_signals, key=lambda s: s.quality_score) self.logger.info(f"📊 {symbol}: ✅ SIGNAL GÉNÉRÉ - {best_signal.side} (qualité: {best_signal.quality_score:.2f}, leverage: {best_signal.leverage}x) - {best_signal.reason}") return best_signal # Aucun signal généré - log avec détails if len(signals) > 0: rejected_reasons = [f"{s.side}:{s.quality_score:.2f}" for s in signals] self.logger.info(f"📊 {symbol}: ❌ SIGNAUX REJETÉS (qualité insuffisante) - Candidats: {', '.join(rejected_reasons)}") else: self.logger.info(f"📊 {symbol}: ❌ AUCUN SIGNAL (conditions techniques non remplies)") return None except Exception as e: self.logger.error(f"Erreur analyse {symbol}: {e}") return None def _calculate_leverage(self, strength: float, symbol: str = None) -> int: """Calcule le leverage basé sur la force du signal et le max disponible""" # Obtenir le leverage max pour ce symbole max_leverage = 5 # Fallback default if symbol and symbol in self.leverage_cache: max_leverage = self.leverage_cache[symbol] # Facteur de sécurité depuis config from trading_config import LEVERAGE_SAFETY_FACTOR safe_max = int(max_leverage * LEVERAGE_SAFETY_FACTOR) # Calcul intelligent basé sur la force ET le max disponible if strength > 0.85: # Signal très fort target = min(safe_max, max(8, safe_max // 2)) # Au moins 8x ou la moitié du max elif strength > 0.75: # Signal fort target = min(safe_max, max(6, safe_max // 3)) # Au moins 6x ou 1/3 du max elif strength > 0.65: # Signal moyen-fort target = min(safe_max, max(4, safe_max // 4)) # Au moins 4x ou 1/4 du max elif strength > 0.5: # Signal moyen target = min(safe_max, max(3, safe_max // 5)) # Au moins 3x ou 1/5 du max else: # Signal faible target = min(safe_max, 2) # 2x max pour signaux faibles return max(1, target) # Jamais en dessous de 1x def _update_leverage_cache(self): """Met à jour le cache des leverages max depuis Hyperliquid""" if self.hyperliquid_client: try: self.leverage_cache = self.hyperliquid_client.get_available_leverages() self.leverage_cache_time = time.time() self.logger.info(f"Cache des leverages mis à jour: {len(self.leverage_cache)} tokens") except Exception as e: self.logger.error(f"Erreur mise à jour cache leverages: {e}") def _calculate_target_price(self, entry_price: float, side: str, leverage: int, signal_strength: float) -> Tuple[float, float]: """ Calcule le prix cible (TP) basé sur le leverage et la force du signal Returns: tuple: (target_price, target_pnl_percent) """ # Base TP% selon le leverage - MODÉRÉ pour laisser place au trailing stop # Viser 6-7% PnL initial, puis trailing stop pour les gros gains if leverage >= 20: base_tp_percent = 0.35 # 0.35% mouvement = 7% PnL avec 20x elif leverage >= 10: base_tp_percent = 0.6 # 0.6% mouvement = 6% PnL avec 10x elif leverage >= 8: base_tp_percent = 0.75 # 0.75% mouvement = 6% PnL avec 8x elif leverage >= 5: base_tp_percent = 1.2 # 1.2% mouvement = 6% PnL avec 5x elif leverage >= 3: base_tp_percent = 2.0 # 2.0% mouvement = 6% PnL avec 3x else: base_tp_percent = 3.0 # 3.0% mouvement = 6% PnL avec 2x # Ajuster selon la force du signal (plus fort = plus ambitieux) tp_percent = base_tp_percent * (1 + signal_strength * 0.5) # jusqu'à +50% si signal très fort # Calculer le prix cible if side == 'LONG': target_price = entry_price * (1 + tp_percent / 100) else: # SHORT target_price = entry_price * (1 - tp_percent / 100) # PnL% avec leverage target_pnl_percent = tp_percent * leverage return target_price, target_pnl_percent def _calculate_smart_leverage(self, strength: float, volatility: float, symbol: str = None) -> int: """ Calcule le leverage intelligent basé sur la force du signal ET la volatilité Plus la volatilité est haute, moins on prend de leverage pour contrôler le risque Utilise les vrais maximums depuis Hyperliquid """ # Récupérer le leverage max depuis le cache statique (plus fiable) max_leverage_for_symbol = get_max_leverage(symbol) if symbol else 20 # Essayer le cache dynamique en fallback si disponible if symbol and self.leverage_cache and max_leverage_for_symbol <= 5: clean_symbol = symbol.replace('USDT', '') if clean_symbol in self.leverage_cache: dynamic_leverage = self.leverage_cache[clean_symbol] if dynamic_leverage > max_leverage_for_symbol: max_leverage_for_symbol = dynamic_leverage # Rafraîchir le cache s'il est trop vieux (> 1h) if self.leverage_cache_time and (time.time() - self.leverage_cache_time > 3600): self._update_leverage_cache() # Import du facteur de sécurité depuis la config try: from trading_config import LEVERAGE_SAFETY_FACTOR, MAX_LEVERAGE safety_factor = LEVERAGE_SAFETY_FACTOR global_max = MAX_LEVERAGE except: safety_factor = 0.95 # Défaut amélioré (était 0.8) global_max = 30 # Appliquer le facteur de sécurité au max du token safe_max = int(min(max_leverage_for_symbol, global_max) * safety_factor) # Calcul du leverage basé sur la force du signal (PLUS AGRESSIF) if strength >= 0.85: base_leverage = int(safe_max * 1.0) # 100% du max sécurisé pour signaux très forts elif strength >= 0.75: base_leverage = int(safe_max * 0.9) # 90% du max sécurisé pour signaux forts elif strength >= 0.65: base_leverage = int(safe_max * 0.75) # 75% du max sécurisé pour signaux moyens-forts elif strength >= 0.55: base_leverage = int(safe_max * 0.6) # 60% du max sécurisé pour signaux moyens else: base_leverage = int(safe_max * 0.5) # 50% du max sécurisé pour signaux faibles # Ajuster selon la volatilité (MOINS PÉNALISANT) if volatility > 8: # Très haute volatilité extrême leverage = max(5, int(base_leverage * 0.65)) # Réduire de 35% seulement elif volatility > 5: # Haute volatilité leverage = max(8, int(base_leverage * 0.85)) # Réduire de 15% seulement elif volatility < 2: # Faible volatilité, on peut être plus agressif leverage = min(max_leverage_for_symbol, int(base_leverage * 1.3)) # Augmenter de 30% else: # Volatilité normale leverage = base_leverage # S'assurer qu'on reste dans les limites avec MINIMUM ÉLEVÉ try: from trading_config import MIN_LEVERAGE_THRESHOLD, LEVERAGE_EXCEPTIONS min_leverage = MIN_LEVERAGE_THRESHOLD leverage_exceptions = LEVERAGE_EXCEPTIONS except ImportError: min_leverage = 8 # Défaut: minimum 8x leverage_exceptions = {} # Vérifier si ce token a une exception de leverage if symbol in leverage_exceptions: # Pour les tokens avec leverage max < minimum global, utiliser leur maximum effective_min_leverage = min(min_leverage, max_leverage_for_symbol) self.logger.info(f"Exception leverage pour {symbol}: min effectif {effective_min_leverage}x (max disponible: {max_leverage_for_symbol}x)") else: effective_min_leverage = min_leverage leverage = max(effective_min_leverage, min(leverage, max_leverage_for_symbol)) # Si le leverage calculé est trop faible (hors exceptions), rejeter le signal if leverage < effective_min_leverage: self.logger.debug(f"Signal rejeté: leverage trop faible {leverage}x < {effective_min_leverage}x minimum") return None self.logger.debug(f"Leverage calculé: {leverage}x (max: {max_leverage_for_symbol}x, force: {strength:.2f}, volatilité: {volatility:.1f})") return leverage def is_signal_suitable_for_rules(self, signal: TradingSignal) -> Tuple[bool, str]: """ Vérifie si le signal est adapté aux règles de trading (simplifié pour trailing stop uniquement) Retourne (suitable, reason) """ indicators = signal.indicators or {} volatility = indicators.get('volatility', 3) # RÈGLES SIMPLIFIÉES car on utilise que trailing stop maintenant # Rejeter seulement les cas extrêmes qui pourraient causer des problèmes if volatility > 15: # Très volatil = trop risqué return False, "Volatilité extrême > 15%" # Accepter tous les autres signaux - le trailing stop gèrera le reste return True, "OK pour trailing stop" def _analyze_enhanced_trend(self, closes: List[float], ema_fast: float, ema_short: float, ema_long: float) -> Tuple[str, float, float]: """Analyse de tendance améliorée avec détection de contre-courants""" trend = "NEUTRAL" trend_strength = 0.0 trend_quality = 0.5 # Qualité de la tendance (0-1) if len(closes) < 20: return trend, trend_strength, trend_quality current_price = closes[-1] # 1. ANALYSE MULTI-EMA (court, moyen, long terme) ema_alignment_score = 0 # Vérifier l'alignement des EMAs if ema_fast > ema_short > ema_long: ema_alignment_score = 1 # Parfait alignement haussier trend = "BULLISH" elif ema_fast < ema_short < ema_long: ema_alignment_score = 1 # Parfait alignement baissier trend = "BEARISH" elif ema_fast > ema_short: ema_alignment_score = 0.5 # Tendance haussière partielle trend = "BULLISH" elif ema_fast < ema_short: ema_alignment_score = 0.5 # Tendance baissière partielle trend = "BEARISH" # 2. ANALYSE DE LA MOMENTUM (force de la tendance) if trend == "BULLISH": trend_strength = (ema_fast - ema_long) / ema_long * 100 elif trend == "BEARISH": trend_strength = (ema_long - ema_fast) / ema_long * 100 trend_strength = max(0, min(trend_strength, 10)) # Limiter à 10% # 3. DÉTECTION DE CONTRE-COURANTS (patterns de retournement) recent_prices = closes[-10:] # 10 dernières bougies # Détection de divergences (prix vs tendance EMA) divergence_penalty = 0 if trend == "BULLISH": # Si prix fait des plus bas alors que EMA monte = divergence bearish recent_low = min(recent_prices[-5:]) older_low = min(recent_prices[:5]) if recent_low < older_low and ema_fast > ema_short: divergence_penalty = 0.3 elif trend == "BEARISH": # Si prix fait des plus hauts alors que EMA descend = divergence bullish recent_high = max(recent_prices[-5:]) older_high = max(recent_prices[:5]) if recent_high > older_high and ema_fast < ema_short: divergence_penalty = 0.3 # 4. CALCUL DE LA QUALITÉ FINALE # Base : alignement des EMAs trend_quality = ema_alignment_score # Bonus pour momentum fort if trend_strength > 2: trend_quality += 0.2 elif trend_strength > 1: trend_quality += 0.1 # Pénalité pour divergences trend_quality -= divergence_penalty # Pénalité pour volatilité excessive volatility = np.std(recent_prices) / np.mean(recent_prices) * 100 if volatility > 8: # Plus de 8% de volatilité trend_quality -= 0.2 # Limiter entre 0 et 1 trend_quality = max(0.0, min(1.0, trend_quality)) return trend, trend_strength, trend_quality def _apply_market_sentiment_filter(self, signal: TradingSignal, market_sentiment: float) -> Optional[TradingSignal]: """Filtre les signaux selon le sentiment de marché global""" # Seuils de filtrage bearish_threshold = -0.3 # Marché bearish si sentiment < -30% bullish_threshold = 0.3 # Marché bullish si sentiment > +30% original_reason = signal.reason # MARCHÉ BEARISH : Réduire les LONG, favoriser les SHORT if market_sentiment <= bearish_threshold: if signal.side == 'LONG': # Augmenter le seuil qualité pour les LONG sur marché bearish if signal.quality_score < 0.85: return None # Rejeter LONG de qualité moyenne # Réduire légèrement le score et ajuster la raison signal.quality_score *= 0.9 signal.reason = f"⚠ {original_reason} (marché bearish)" elif signal.side == 'SHORT': # Favoriser les SHORT sur marché bearish signal.quality_score = min(signal.quality_score * 1.1, 0.95) signal.reason = f"📈 {original_reason} + marché bearish" # MARCHÉ BULLISH : Favoriser les LONG, réduire les SHORT elif market_sentiment >= bullish_threshold: if signal.side == 'SHORT': # Augmenter le seuil qualité pour les SHORT sur marché bullish if signal.quality_score < 0.85: return None # Rejeter SHORT de qualité moyenne signal.quality_score *= 0.9 signal.reason = f"⚠ {original_reason} (marché bullish)" elif signal.side == 'LONG': # Favoriser les LONG sur marché bullish signal.quality_score = min(signal.quality_score * 1.1, 0.95) signal.reason = f"📈 {original_reason} + marché bullish" # MARCHÉ NEUTRE : Pas de modification return signal def _apply_shitcoin_bias(self, signal: 'TradingSignal', symbol: str) -> Optional['TradingSignal']: """ Applique un biais SHORT aux shitcoins selon la configuration Args: signal: Signal de trading original symbol: Symbole du crypto Returns: Signal modifié ou None si filtré """ try: from trading_config import SHITCOIN_STRATEGY, SHITCOIN_SHORT_BIAS_STRENGTH, SHITCOIN_MAX_POSITIONS quality = self._get_crypto_quality_category(symbol) # Appliquer le biais uniquement aux shitcoins if quality == 'shitcoin' and SHITCOIN_STRATEGY == "SHORT_BIAS": original_side = signal.side # Si signal LONG sur un shitcoin, transformer en SHORT avec probabilité basée sur bias_strength if signal.side == 'LONG': # Force du biais : 0.8 = 80% de chance de transformer LONG en SHORT import random if random.random() < SHITCOIN_SHORT_BIAS_STRENGTH: signal.side = 'SHORT' signal.quality_score *= 1.2 # Augmenter la confiance pour les shorts shitcoin signal.reason = f"🔻 SHORT bias shitcoin: {signal.reason}" self.logger.info(f"SHITCOIN BIAS: {symbol} {original_side} → {signal.side}") else: # Rejeter le signal LONG sur shitcoin self.logger.debug(f"SHITCOIN BIAS: {symbol} LONG rejeté") return None # Si signal SHORT sur shitcoin, augmenter la confiance elif signal.side == 'SHORT': signal.quality_score *= 1.3 # Bonus pour short shitcoin signal.reason = f"🎯 {signal.reason} + shitcoin SHORT" self.logger.debug(f"SHITCOIN BIAS: {symbol} SHORT renforcé") return signal except ImportError: return signal def _update_market_classification_if_needed(self): """Met à jour la classification des marchés si nécessaire""" import time current_time = time.time() if current_time - self.last_market_update > self.market_update_interval: try: self.logger.info("🔄 Mise à jour automatique de la classification des marchés...") # Forcer la mise à jour du cache from dynamic_market_classifier import DynamicMarketClassifier classifier = DynamicMarketClassifier(cache_duration_hours=0) # Force refresh high_quality, medium_quality = classifier.get_updated_classification() # Mettre à jour les listes globales en important trading_config import trading_config trading_config.HIGH_QUALITY_CRYPTOS = list(high_quality) trading_config.MEDIUM_QUALITY_CRYPTOS = list(medium_quality) self.last_market_update = current_time self.logger.info(f"✅ Classification mise à jour: {len(high_quality)} high, {len(medium_quality)} medium") except Exception as e: self.logger.error(f"Erreur mise à jour marchés: {e}") def _get_crypto_quality_category(self, symbol: str) -> str: """ Détermine la catégorie de qualité d'un crypto Args: symbol: Symbole du crypto Returns: str: 'high', 'medium', 'shitcoin' """ try: from trading_config import HIGH_QUALITY_CRYPTOS, MEDIUM_QUALITY_CRYPTOS # Nettoyer le symbole clean_symbol = symbol.replace('USDT', '').upper() # Cryptos de haute qualité if clean_symbol in HIGH_QUALITY_CRYPTOS: return 'high' # Cryptos de qualité moyenne if clean_symbol in MEDIUM_QUALITY_CRYPTOS: return 'medium' # Tout le reste = shitcoin return 'shitcoin' except ImportError: return 'high' # Par défaut, traiter comme haute qualité def _should_process_crypto(self, symbol: str, current_positions: int = 0) -> bool: """ Vérifie si un crypto doit être traité selon la stratégie configurée Args: symbol: Symbole du crypto current_positions: Nombre de positions actuelles Returns: bool: True si le crypto doit être traité """ try: from trading_config import ( USE_QUALITY_FILTER, SHITCOIN_STRATEGY, MAX_MEDIUM_QUALITY_RATIO, MAX_SIMULTANEOUS_POSITIONS ) # Si le filtre n'est pas activé, autoriser tout if not USE_QUALITY_FILTER: self.logger.info(f"🔍 {symbol}: AUTORISÉ (filtre qualité désactivé)") return True quality = self._get_crypto_quality_category(symbol) # Cryptos de haute qualité : toujours autorisées if quality == 'high': self.logger.info(f"🔍 {symbol}: AUTORISÉ (haute qualité)") return True # Cryptos de qualité moyenne : vérifier la limite if quality == 'medium': max_medium_positions = int(MAX_SIMULTANEOUS_POSITIONS * MAX_MEDIUM_QUALITY_RATIO) allowed = current_positions < max_medium_positions status = "AUTORISÉ" if allowed else "REJETÉ" self.logger.info(f"🔍 {symbol}: {status} (qualité moyenne, {current_positions}/{max_medium_positions})") return allowed # Shitcoins : selon la stratégie avec limite stricte if quality == 'shitcoin': if SHITCOIN_STRATEGY == "EXCLUDE": self.logger.info(f"🔍 {symbol}: REJETÉ (shitcoin exclu par stratégie)") return False elif SHITCOIN_STRATEGY == "SHORT_BIAS": # Vérifier la limite de shitcoins (très restrictif) from trading_config import SHITCOIN_MAX_POSITIONS # Compter approximativement les shitcoins dans les signaux actuels shitcoin_count = sum(1 for s in range(current_positions) if self._get_crypto_quality_category(f"DUMMY_{s}") == 'shitcoin') if shitcoin_count >= SHITCOIN_MAX_POSITIONS: self.logger.info(f"🔍 {symbol}: REJETÉ (limite shitcoins atteinte {shitcoin_count}/{SHITCOIN_MAX_POSITIONS})") return False self.logger.info(f"🔍 {symbol}: AUTORISÉ (shitcoin avec SHORT bias)") return True # Autoriser pour appliquer le biais SHORT elif SHITCOIN_STRATEGY == "ALLOW": self.logger.info(f"🔍 {symbol}: AUTORISÉ (shitcoin autorisé par stratégie)") return True return True except ImportError: return True def generate_signals(self) -> List[Dict]: """Génère des signaux pour tous les symboles surveillés""" signals = [] # Mettre à jour la liste des symboles si nécessaire (toutes les heures) self._update_symbols_if_needed() # Mettre à jour la classification des marchés si nécessaire (toutes les 6 heures) self._update_market_classification_if_needed() # Obtenir le sentiment de marché global market_sentiment = self.sentiment_analyzer.get_market_sentiment() filtered_count = 0 shitcoin_signals = 0 for i, symbol in enumerate(self.symbols): # Vérifier si le crypto doit être traité if not self._should_process_crypto(symbol, len(signals)): filtered_count += 1 continue # Ajouter un petit délai tous les 5 symboles pour éviter rate limit if i > 0 and i % 5 == 0: import time time.sleep(0.5) signal = self.analyze_symbol(symbol) if signal: # Appliquer le filtre de sentiment de marché signal = self._apply_market_sentiment_filter(signal, market_sentiment) # Appliquer le biais shitcoin si nécessaire if signal: signal = self._apply_shitcoin_bias(signal, symbol) # Compter les signaux shitcoin traités if signal and self._get_crypto_quality_category(symbol) == 'shitcoin': shitcoin_signals += 1 if signal: # Signal toujours valide après filtrage # Convertir en format attendu par SimpleTradingEngine signal_dict = { 'symbol': signal.symbol, 'side': signal.side, 'current_price': signal.current_price, 'leverage': signal.leverage, 'quality_score': signal.quality_score, 'reason': signal.reason, 'timestamp': signal.timestamp.isoformat() } # Ajouter les indicateurs si disponibles pour debug if signal.indicators: signal_dict['volatility'] = signal.indicators.get('volatility', 0) signal_dict['rsi'] = signal.indicators.get('rsi', 50) signal_dict['trend'] = signal.indicators.get('trend', 'NEUTRAL') signals.append(signal_dict) # Log du filtrage et biais if filtered_count > 0 or shitcoin_signals > 0: self.logger.info(f"🔍 Filtre: {filtered_count} exclus, {shitcoin_signals} shitcoins SHORT bias, {len(signals)} signaux générés") return signals def get_signal_summary(self) -> Dict: """Retourne un résumé des signaux actuels""" summary = { 'timestamp': datetime.now().isoformat(), 'symbols_monitored': len(self.symbols), 'signals': [] } for symbol in self.symbols: signal = self.analyze_symbol(symbol) if signal: summary['signals'].append({ 'symbol': signal.symbol, 'side': signal.side, 'quality': signal.quality_score, 'reason': signal.reason }) summary['total_signals'] = len(summary['signals']) return summary # Fonction helper pour l'intégration def create_signal_provider(hyperliquid_client=None, logger=None): """Crée une instance du générateur de signaux""" generator = PragmaticSignalGenerator(hyperliquid_client, logger) return generator.generate_signals