# -*- 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 from market_regime_detector import MarketRegimeDetector @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 base self.backoff_duration = 3600 # 1h si rate limited self.last_update = 0 self.rate_limited_until = 0 # Timestamp jusqu'à quand on est rate limited 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) avec backoff automatique si 429""" now = time.time() # Si on est en backoff suite à un 429, attendre if now < self.rate_limited_until: return self.market_sentiment # Retourner la dernière valeur cachée # Sinon update normalement si cache expiré 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 via Binance ticker""" try: # API Binance 24h ticker (pas de rate limit strict) url = "https://api.binance.com/api/v3/ticker/24hr" response = requests.get(url, timeout=10) if response.status_code == 200: data = response.json() # Mapper les symboles majors major_symbols = { 'BTCUSDT': 'bitcoin', 'ETHUSDT': 'ethereum', 'SOLUSDT': 'solana', 'BNBUSDT': 'binancecoin' } # Calculer sentiment basé sur les variations 24h changes = [] for symbol in major_symbols.keys(): ticker = next((t for t in data if t['symbol'] == symbol), None) if ticker and 'priceChangePercent' in ticker: changes.append(float(ticker['priceChangePercent'])) 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 Binance 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 # Analyseur de sentiment de marché self.sentiment_analyzer = MarketSentimentAnalyzer() # Détecteur de régime de marché pour adaptation dynamique self.regime_detector = MarketRegimeDetector(logger=self.logger) # 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 avec paramètres standards""" 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 calculate_stochastic_rsi(self, prices: List[float], period: int = 14) -> float: """Calcule le Stochastic RSI""" if len(prices) < period * 2: return 50.0 # Calculer RSI sur période étendue rsi_values = [] for i in range(period, len(prices)): rsi_val = self.calculate_rsi(prices[:i+1], period) rsi_values.append(rsi_val) if len(rsi_values) < period: return 50.0 # Appliquer formule stochastic sur les valeurs RSI recent_rsi = rsi_values[-period:] current_rsi = rsi_values[-1] min_rsi = min(recent_rsi) max_rsi = max(recent_rsi) if max_rsi == min_rsi: return 50.0 stoch_rsi = (current_rsi - min_rsi) / (max_rsi - min_rsi) * 100 return stoch_rsi def calculate_volume_ratio(self, volumes: List[float], period: int = 20) -> float: """Calcule le ratio volume actuel / moyenne""" if len(volumes) < period + 1: return 1.0 current_volume = volumes[-1] avg_volume = np.mean(volumes[-period-1:-1]) # Exclure volume actuel if avg_volume == 0: return 1.0 return current_volume / avg_volume def calculate_balanced_score(self, prices: List[float], volumes: List[float], current_price: float) -> Dict: """ NOUVEAU SYSTÈME DE SCORING ÉQUILIBRÉ Élimine la redondance et privilégie momentum + volume """ if len(prices) < 50 or len(volumes) < 20: return {"score": 0.5, "components": {}, "side": None, "strength": 0.5} # ========== COMPOSANTS NON REDONDANTS ========== # 1. TREND COMPONENT (25%) - Simplifié EMA 9 vs 21 ema9 = self.calculate_ema(prices, 9) ema21 = self.calculate_ema(prices, 21) trend_strength = abs(ema9 - ema21) / ema21 if ema21 > 0 else 0 trend_score = min(trend_strength * 10, 1.0) trend_direction = "BULLISH" if ema9 > ema21 else "BEARISH" # 2. MOMENTUM COMPONENT (35%) - RSI + Stoch RSI rsi = self.calculate_rsi(prices, 14) stoch_rsi = self.calculate_stochastic_rsi(prices, 14) # RSI signals (seuils crypto-optimisés: 25/75) rsi_signal = 0 if rsi < 25: # Oversold rsi_signal = (25 - rsi) / 25 # Score 0-1 elif rsi > 75: # Overbought rsi_signal = -(rsi - 75) / 25 # Score négatif pour SHORT # Stoch RSI confirmation stoch_signal = 0 if stoch_rsi < 20: stoch_signal = (20 - stoch_rsi) / 20 elif stoch_rsi > 80: stoch_signal = -(stoch_rsi - 80) / 20 momentum_score = abs((rsi_signal + stoch_signal) / 2) momentum_side = "LONG" if (rsi_signal + stoch_signal) > 0 else "SHORT" # 3. VOLATILITY COMPONENT (25%) - Bollinger optimisé upper_bb, middle_bb, lower_bb = self.calculate_bollinger_bands(prices) bb_position = (current_price - lower_bb) / (upper_bb - lower_bb) if upper_bb != lower_bb else 0.5 bb_squeeze = 1 - ((upper_bb - lower_bb) / middle_bb) if middle_bb > 0 else 0 # Score basé sur position + potentiel breakout volatility_score = 0 volatility_side = None if bb_position < 0.2 and bb_squeeze > 0.8: # Bas + squeeze = breakout haussier volatility_score = 0.8 volatility_side = "LONG" elif bb_position > 0.8 and bb_squeeze > 0.8: # Haut + squeeze = breakout baissier volatility_score = 0.8 volatility_side = "SHORT" else: volatility_score = abs(bb_position - 0.5) * 2 # Distance du milieu # 4. VOLUME COMPONENT (15%) - Confirmation volume_ratio = self.calculate_volume_ratio(volumes) volume_score = min(volume_ratio / 2.0, 1.0) # 2x volume = score max # ========== SCORING FINAL ========== components = { "trend": {"score": trend_score, "direction": trend_direction, "weight": 0.25}, "momentum": {"score": momentum_score, "side": momentum_side, "weight": 0.35}, "volatility": {"score": volatility_score, "side": volatility_side, "weight": 0.25}, "volume": {"score": volume_score, "weight": 0.15} } # Score composite pondéré final_score = (trend_score * 0.25 + momentum_score * 0.35 + volatility_score * 0.25 + volume_score * 0.15) # Déterminer le côté dominant side_votes = {"LONG": 0, "SHORT": 0} if momentum_side: side_votes[momentum_side] += 0.35 if volatility_side: side_votes[volatility_side] += 0.25 if trend_direction == "BULLISH": side_votes["LONG"] += 0.25 else: side_votes["SHORT"] += 0.25 dominant_side = "LONG" if side_votes["LONG"] > side_votes["SHORT"] else "SHORT" return { "score": final_score, "components": components, "side": dominant_side, "strength": final_score, "rsi": rsi, "stoch_rsi": stoch_rsi, "volume_ratio": volume_ratio, "trend_direction": trend_direction } def get_btc_macro_trend(self) -> Dict: """Analyse la tendance macro de BTC pour filtrer les signaux contre-tendance""" try: # Récupérer les données BTC btc_data = self.get_market_data('BTC') if not btc_data: return {"trend": "NEUTRAL", "strength": 0.0} closes = btc_data['closes'] if len(closes) < 50: return {"trend": "NEUTRAL", "strength": 0.0} # EMAs sur BTC pour tendance moyen terme ema9 = self.calculate_ema(closes, 9) ema21 = self.calculate_ema(closes, 21) ema50 = self.calculate_ema(closes, 50) # Force de la tendance trend_strength = 0.0 trend_direction = "NEUTRAL" # Bullish: EMA9 > EMA21 > EMA50 if ema9 > ema21 > ema50: trend_direction = "BULLISH" trend_strength = min(((ema9 - ema50) / ema50) * 100, 1.0) # % écart normalisé # Bearish: EMA9 < EMA21 < EMA50 elif ema9 < ema21 < ema50: trend_direction = "BEARISH" trend_strength = min(((ema50 - ema9) / ema50) * 100, 1.0) # Mitigé else: trend_direction = "NEUTRAL" trend_strength = 0.0 self.logger.info(f"🔍 BTC Macro: {trend_direction} (force: {trend_strength:.2f})") return { "trend": trend_direction, "strength": trend_strength, "ema9": ema9, "ema21": ema21, "ema50": ema50 } except Exception as e: self.logger.warning(f"Erreur analyse BTC macro: {e}") return {"trend": "NEUTRAL", "strength": 0.0} def analyze_symbol(self, symbol: str) -> Optional[TradingSignal]: """Analyse un symbole avec le NOUVEAU SYSTÈME DE SCORING ÉQUILIBRÉ""" 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'] volumes = data.get('volumes', [1] * len(closes)) # Fallback si pas de volume current_price = data['current_price'] # ========== NOUVEAU SYSTÈME DE SCORING ========== scoring_result = self.calculate_balanced_score(closes, volumes, current_price) if scoring_result['score'] < 0.35: # Seuil minimum 35% (marché très flat) self.logger.info(f"📊 {symbol}: Score trop faible {scoring_result['score']:.3f}") return None # Extraire les composants components = scoring_result['components'] side = scoring_result['side'] strength = scoring_result['strength'] self.logger.info(f"📊 {symbol}: Score {scoring_result['score']:.3f} " f"(trend: {components['trend']['score']:.2f}, " f"momentum: {components['momentum']['score']:.2f}, " f"volatility: {components['volatility']['score']:.2f}, " f"volume: {components['volume']['score']:.2f}) → {side}") # ========== DÉTECTION RÉGIME DE MARCHÉ & ADAPTATION ========== # Détecter le régime spécifique à ce symbole regime, regime_params = self.regime_detector.detect_regime(closes, volumes, current_price) # Logger la détection self.regime_detector.log_regime_detection(symbol, regime, regime_params) # Vérifier si on peut trader dans ce régime selon le side (DUMP = SHORT only) if not self.regime_detector.should_trade_in_regime(regime, side): self.logger.warning(f"⚠️ {symbol}: REJETÉ - {side} interdit en régime {regime}") return None # Ajuster le score selon le régime ET le side signal_with_regime = {'symbol': symbol, 'side': side, 'score': scoring_result['score']} adjusted_score = self.regime_detector.adjust_signal_score(signal_with_regime, regime, regime_params) # Mettre à jour le score dans scoring_result score_change = adjusted_score - scoring_result['score'] if abs(score_change) > 0.01: self.logger.info(f"🔄 {symbol}: Score ajusté par régime: {scoring_result['score']:.3f} → {adjusted_score:.3f} ({score_change:+.3f})") scoring_result['score'] = adjusted_score # Stocker le régime dans les indicateurs pour logging futur regime_info = { 'regime': regime, 'regime_change_24h': regime_params.get('change_24h', 0), 'regime_volatility': regime_params.get('volatility_pct', 0) } # ========== FILTRE MACRO BTC (ASSOUPLI) ========== # Pénaliser les contre-tendances mais ne pas tout bloquer btc_macro = self.get_btc_macro_trend() # Si BTC bullish et signal SHORT → pénaliser if btc_macro['trend'] == "BULLISH" and side == "SHORT": if btc_macro['strength'] > 0.7: # Tendance très forte scoring_result['score'] *= 0.75 # -25% self.logger.info(f"⚠️ {symbol}: SHORT pénalisé 25% - BTC bullish fort {btc_macro['strength']:.2f} (score → {scoring_result['score']:.3f})") elif btc_macro['strength'] > 0.4: scoring_result['score'] *= 0.85 # -15% self.logger.info(f"⚠️ {symbol}: SHORT pénalisé 15% - BTC bullish modéré (score → {scoring_result['score']:.3f})") # Si BTC bearish et signal LONG → pénaliser elif btc_macro['trend'] == "BEARISH" and side == "LONG": if btc_macro['strength'] > 0.7: # Tendance très forte scoring_result['score'] *= 0.75 # -25% self.logger.info(f"⚠️ {symbol}: LONG pénalisé 25% - BTC bearish fort {btc_macro['strength']:.2f} (score → {scoring_result['score']:.3f})") elif btc_macro['strength'] > 0.4: scoring_result['score'] *= 0.85 # -15% self.logger.info(f"⚠️ {symbol}: LONG pénalisé 15% - BTC bearish modéré (score → {scoring_result['score']:.3f})") # Revérifier le seuil après pénalité (assoupli pour marché très plat) if scoring_result['score'] < 0.25: self.logger.info(f"📊 {symbol}: Score insuffisant après filtre macro {scoring_result['score']:.3f}") return None # ========== GÉNÉRATION DU SIGNAL OPTIMISÉ ========== # Calculer leverage basé sur le score et la volatilité volatility = components['volatility']['score'] leverage = self._calculate_smart_leverage(strength, volatility * 10, symbol) # Score qualité basé sur consensus des composants quality_score = min(scoring_result['score'], 0.95) # Raison détaillée avec composants dominants dominant_components = [] for comp_name, comp_data in components.items(): if comp_data['score'] > 0.7: dominant_components.append(f"{comp_name}({comp_data['score']:.2f})") reason = f"Nouveau scoring: {' + '.join(dominant_components[:3])}" # Calculer target price (optionnel avec trailing stops) target_price, target_pnl = self._calculate_target_price(current_price, side, leverage, strength) # Créer le signal optimisé signal = TradingSignal( symbol=symbol, side=side, strength=strength, current_price=current_price, leverage=leverage, quality_score=quality_score, reason=reason, timestamp=datetime.now(), target_price=target_price, target_pnl_percent=target_pnl, indicators={ 'rsi': scoring_result['rsi'], 'stoch_rsi': scoring_result['stoch_rsi'], 'volume_ratio': scoring_result['volume_ratio'], 'trend_direction': scoring_result['trend_direction'], 'scoring_components': components, 'final_score': scoring_result['score'], # Informations de régime pour logging et analyse 'regime': regime_info['regime'], 'regime_change_24h': regime_info['regime_change_24h'], 'regime_volatility': regime_info['regime_volatility'] } ) # Vérification finale des règles suitable_signals = [] if self.is_signal_suitable_for_rules(signal)[0]: suitable_signals.append(signal) self.logger.info(f"✅ {symbol} {side}: Signal généré avec score {quality_score:.3f}") else: self.logger.info(f"❌ {symbol}: Signal rejeté par règles de filtrage") return suitable_signals[0] if suitable_signals else None except Exception as e: self.logger.error(f"Erreur analyse {symbol}: {e}") return None def _calculate_smart_leverage(self, strength: float, volatility: float, symbol: str = None) -> int: """Calcule le leverage intelligent basé sur la force du signal et volatilité""" # Import du cache de leverage max try: from leverage_limits import get_max_leverage max_leverage_for_symbol = get_max_leverage(symbol) if symbol else 20 except ImportError: # Fallback si leverage_limits n'existe pas max_leverage_for_symbol = 20 # Import config try: from trading_config import LEVERAGE_SAFETY_FACTOR, MAX_LEVERAGE, MIN_LEVERAGE_THRESHOLD safety_factor = LEVERAGE_SAFETY_FACTOR global_max = MAX_LEVERAGE min_leverage = MIN_LEVERAGE_THRESHOLD except: safety_factor = 0.95 global_max = 30 min_leverage = 8 # Appliquer sécurité safe_max = int(min(max_leverage_for_symbol, global_max) * safety_factor) # Calcul selon force signal if strength >= 0.85: base_leverage = int(safe_max * 1.0) elif strength >= 0.75: base_leverage = int(safe_max * 0.9) elif strength >= 0.65: base_leverage = int(safe_max * 0.75) elif strength >= 0.55: base_leverage = int(safe_max * 0.6) else: base_leverage = int(safe_max * 0.5) # Ajuster volatilité if volatility > 8: leverage = max(5, int(base_leverage * 0.65)) elif volatility > 5: leverage = max(8, int(base_leverage * 0.85)) elif volatility < 2: leverage = min(max_leverage_for_symbol, int(base_leverage * 1.3)) else: leverage = base_leverage # Appliquer limites leverage = max(min_leverage, min(leverage, max_leverage_for_symbol)) return leverage def _calculate_target_price(self, entry_price: float, side: str, leverage: int, signal_strength: float) -> tuple: """Calcule le prix cible et PnL target""" # Base TP selon leverage if leverage >= 20: base_tp_percent = 0.35 elif leverage >= 10: base_tp_percent = 0.5 elif leverage >= 5: base_tp_percent = 0.8 else: base_tp_percent = 1.2 # Ajuster selon force signal tp_percent = base_tp_percent * (0.8 + 0.4 * signal_strength) if side == 'LONG': target_price = entry_price * (1 + tp_percent / 100) else: target_price = entry_price * (1 - tp_percent / 100) target_pnl = tp_percent * leverage return target_price, target_pnl def is_signal_suitable_for_rules(self, signal: TradingSignal) -> tuple: """Vérifie si le signal est adapté aux règles (simplifié pour trailing stop)""" indicators = signal.indicators or {} volatility = indicators.get('volatility', 3) # Rejeter seulement cas extrêmes if volatility > 15: return False, "Volatilité extrême > 15%" return True, "OK pour trailing stop" def _apply_market_sentiment_filter(self, signal: TradingSignal, market_sentiment: float): """Filtre les signaux selon sentiment (simplifié)""" # Pour l'instant, accepter tous les signaux # Le sentiment est déjà pris en compte dans le scoring return signal def _apply_shitcoin_bias(self, signal: TradingSignal, symbol: str): """Applique un biais shitcoin (simplifié)""" # Pour l'instant, accepter tous les signaux # La qualité est déjà gérée par le système de scoring return signal def _update_market_classification_if_needed(self): """Met à jour la classification des marchés si nécessaire""" try: current_time = time.time() if current_time - self.last_market_update < self.market_update_interval: return # Code de mise à jour ici si nécessaire self.last_market_update = current_time 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""" self.logger.info(f"🔄 generate_signals() appelé - {len(self.symbols)} symboles à analyser") 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() self.logger.info(f"🎯 Début analyse - sentiment marché: {market_sentiment:.3f}") 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, 'score': signal.quality_score, # score pour le filtre 'quality_score': signal.quality_score, 'confidence': int(signal.quality_score * 100), # confidence en % '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') signal_dict['components'] = signal.indicators.get('scoring_components', {}) # Ajouter les informations de régime pour le filtre adaptatif signal_dict['regime'] = signal.indicators.get('regime', 'CONSOLIDATION') signal_dict['regime_change_24h'] = signal.indicators.get('regime_change_24h', 0) signal_dict['regime_volatility'] = signal.indicators.get('regime_volatility', 0) 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