""" Détecteur de tendance pour filtrer les signaux de trading Évite d'ouvrir des positions contre la tendance dominante """ import logging from datetime import datetime, timedelta from typing import Dict, Optional, Tuple import sys import os sys.path.append('..') from hyperliquid_client import HyperliquidClient class TrendDetector: """Détecte la tendance du marché avant d'autoriser les ouvertures de positions""" def __init__(self, hyperliquid_client: HyperliquidClient = None): self.hyperliquid_client = hyperliquid_client self.logger = logging.getLogger("TrendDetector") # Configuration des seuils self.strong_trend_threshold = 1.5 # % pour tendance forte self.neutral_zone = 0.5 # % zone neutre # Cache des tendances (pour éviter trop d'appels API) self.trend_cache = {} self.cache_duration = 180 # 3 minutes def get_market_trend(self, symbol: str) -> Tuple[str, float, Dict]: """ Détermine la tendance actuelle du marché pour un symbole Returns: Tuple: (trend, strength, details) - trend: 'BULLISH', 'BEARISH', 'NEUTRAL' - strength: force de la tendance (0-100) - details: détails de l'analyse """ # Vérifier le cache cache_key = f"{symbol}_trend" if cache_key in self.trend_cache: cached_data, timestamp = self.trend_cache[cache_key] if (datetime.now() - timestamp).seconds < self.cache_duration: self.logger.debug(f"Using cached trend for {symbol}") return cached_data try: # Récupérer les données de marché via l'API Hyperliquid candles = self.fetch_recent_candles(symbol) if not candles: self.logger.warning(f"Pas de données pour {symbol}") return ('NEUTRAL', 0, {'error': 'No data'}) # Analyser plusieurs timeframes analysis = { '5min': self.analyze_timeframe(candles[-12:], '5min'), # 1 heure '15min': self.analyze_timeframe(candles[-16:], '15min'), # 4 heures '1h': self.analyze_timeframe(candles[-24:], '1h'), # 24 heures '4h': self.analyze_timeframe(candles[-28:], '4h'), # 7 jours } # Calculer la tendance globale pondérée weighted_score = 0 weights = {'5min': 0.15, '15min': 0.25, '1h': 0.35, '4h': 0.25} for timeframe, weight in weights.items(): tf_data = analysis[timeframe] if tf_data['trend'] == 'BULLISH': weighted_score += weight * tf_data['strength'] elif tf_data['trend'] == 'BEARISH': weighted_score -= weight * tf_data['strength'] # Déterminer la tendance finale if weighted_score > 30: trend = 'BULLISH' strength = min(100, weighted_score) elif weighted_score < -30: trend = 'BEARISH' strength = min(100, abs(weighted_score)) else: trend = 'NEUTRAL' strength = abs(weighted_score) # Ajouter des indicateurs supplémentaires current_price = candles[-1]['close'] if candles else 0 ma_20 = sum(c['close'] for c in candles[-20:]) / 20 if len(candles) >= 20 else current_price ma_50 = sum(c['close'] for c in candles[-50:]) / 50 if len(candles) >= 50 else current_price details = { 'trend': trend, 'strength': strength, 'current_price': current_price, 'ma_20': ma_20, 'ma_50': ma_50, 'price_vs_ma20': ((current_price - ma_20) / ma_20 * 100) if ma_20 > 0 else 0, 'price_vs_ma50': ((current_price - ma_50) / ma_50 * 100) if ma_50 > 0 else 0, 'timeframes': analysis, 'weighted_score': weighted_score } # Log de l'analyse self.logger.info(f"{symbol} Trend: {trend} (Strength: {strength:.1f}%)") if abs(details['price_vs_ma20']) > 2: self.logger.info(f" Price vs MA20: {details['price_vs_ma20']:.2f}%") # Mettre en cache result = (trend, strength, details) self.trend_cache[cache_key] = (result, datetime.now()) return result except Exception as e: self.logger.error(f"Erreur analyse tendance {symbol}: {e}") return ('NEUTRAL', 0, {'error': str(e)}) def analyze_timeframe(self, candles: list, timeframe: str) -> Dict: """Analyse la tendance sur un timeframe spécifique""" if not candles or len(candles) < 2: return {'trend': 'NEUTRAL', 'strength': 0, 'change': 0} # Calculer le changement de prix first_price = candles[0]['close'] last_price = candles[-1]['close'] price_change = ((last_price - first_price) / first_price) * 100 if first_price > 0 else 0 # Analyser la direction des chandeliers bullish_candles = sum(1 for c in candles if c['close'] > c['open']) bearish_candles = sum(1 for c in candles if c['close'] < c['open']) # Calculer les highs et lows higher_highs = 0 lower_lows = 0 for i in range(1, len(candles)): if candles[i]['high'] > candles[i-1]['high']: higher_highs += 1 if candles[i]['low'] < candles[i-1]['low']: lower_lows += 1 # Score de tendance trend_score = 0 # Basé sur le changement de prix if abs(price_change) > self.strong_trend_threshold: trend_score += 40 * (1 if price_change > 0 else -1) elif abs(price_change) > self.neutral_zone: trend_score += 20 * (1 if price_change > 0 else -1) # Basé sur la direction des chandeliers candle_ratio = (bullish_candles - bearish_candles) / len(candles) trend_score += candle_ratio * 30 # Basé sur les highs/lows if higher_highs > lower_lows: trend_score += 15 elif lower_lows > higher_highs: trend_score -= 15 # Déterminer la tendance if trend_score > 25: trend = 'BULLISH' elif trend_score < -25: trend = 'BEARISH' else: trend = 'NEUTRAL' return { 'trend': trend, 'strength': min(100, abs(trend_score)), 'change': price_change, 'bullish_candles': bullish_candles, 'bearish_candles': bearish_candles, 'higher_highs': higher_highs, 'lower_lows': lower_lows } def fetch_recent_candles(self, symbol: str, limit: int = 100) -> list: """Récupère les chandeliers récents depuis Hyperliquid""" try: if not self.hyperliquid_client: return [] # Utiliser l'API Hyperliquid pour récupérer les candles import requests url = "https://api.hyperliquid.xyz/info" # Récupérer 5min candles end_time = int(datetime.now().timestamp() * 1000) start_time = int((datetime.now() - timedelta(hours=12)).timestamp() * 1000) payload = { "type": "candleSnapshot", "req": { "coin": symbol, "interval": "5m", "startTime": start_time, "endTime": end_time } } response = requests.post(url, json=payload) if response.status_code == 200: data = response.json() # Formatter les candles candles = [] for candle_data in data: candles.append({ 'timestamp': int(candle_data['t']), 'open': float(candle_data['o']), 'high': float(candle_data['h']), 'low': float(candle_data['l']), 'close': float(candle_data['c']), 'volume': float(candle_data['v']) }) return candles[-limit:] if len(candles) > limit else candles return [] except Exception as e: self.logger.error(f"Erreur récupération candles {symbol}: {e}") return [] def should_allow_signal(self, signal: Dict) -> Tuple[bool, str]: """ Vérifie si un signal devrait être exécuté basé sur la tendance Args: signal: Le signal de trading à vérifier Returns: Tuple: (allowed, reason) """ symbol = signal.get('symbol') side = signal.get('side', '').upper() if not symbol or not side: return (False, "Signal invalide") # Récupérer la tendance trend, strength, details = self.get_market_trend(symbol) # RÈGLES ASSOUPLIES: Bloquer seulement les cas extrêmes if side == 'LONG': # Bloquer LONG seulement si tendance très fortement baissière if trend == 'BEARISH' and strength > 70: return (False, f"Tendance BEARISH très forte ({strength:.0f}%), LONG trop risqué") elif trend == 'BEARISH' and strength > 50: # Vérifier si le prix est très au-dessus des moyennes (possible correction) price_vs_ma20 = details.get('price_vs_ma20', 0) if price_vs_ma20 < -3: # Prix déjà bien en dessous MA20 return (False, f"Tendance BEARISH forte + prix sous MA20 ({price_vs_ma20:.1f}%), LONG refusé") else: return (True, f"Tendance BEARISH ({strength:.0f}%) mais prix correct, LONG autorisé") else: return (True, f"Tendance {trend} ({strength:.0f}%), LONG autorisé") elif side == 'SHORT': # Bloquer SHORT seulement si tendance très fortement haussière if trend == 'BULLISH' and strength > 70: return (False, f"Tendance BULLISH très forte ({strength:.0f}%), SHORT trop risqué") elif trend == 'BULLISH' and strength > 50: # Vérifier si le prix est très au-dessus des moyennes (possible correction) price_vs_ma20 = details.get('price_vs_ma20', 0) if price_vs_ma20 > 3: # Prix bien au-dessus MA20, correction possible return (True, f"Tendance BULLISH mais prix survalorisé (+{price_vs_ma20:.1f}% vs MA20), SHORT autorisé") else: return (False, f"Tendance BULLISH forte + prix normal, SHORT refusé") else: return (True, f"Tendance {trend} ({strength:.0f}%), SHORT autorisé") return (True, "Signal autorisé par défaut") def get_trend_recommendation(self, symbol: str) -> str: """Donne une recommandation basée sur la tendance actuelle""" trend, strength, details = self.get_market_trend(symbol) if trend == 'BULLISH': if strength > 60: return "STRONG_BUY - Tendance haussière forte" elif strength > 30: return "BUY - Tendance haussière modérée" else: return "WEAK_BUY - Tendance haussière faible" elif trend == 'BEARISH': if strength > 60: return "STRONG_SELL - Tendance baissière forte" elif strength > 30: return "SELL - Tendance baissière modérée" else: return "WEAK_SELL - Tendance baissière faible" else: return "HOLD - Marché sans tendance claire" def test_trend_detector(): """Test le détecteur de tendance""" print("=== TEST DÉTECTEUR DE TENDANCE ===\n") from hyperliquid_client import HyperliquidClient client = HyperliquidClient() detector = TrendDetector(client) # Tester sur plusieurs symboles symbols = ['BNB', 'BTC', 'ETH', 'SOL'] for symbol in symbols: print(f"\n{symbol}:") print("-" * 40) trend, strength, details = detector.get_market_trend(symbol) print(f"Tendance: {trend}") print(f"Force: {strength:.1f}%") print(f"Recommandation: {detector.get_trend_recommendation(symbol)}") # Tester les signaux long_signal = {'symbol': symbol, 'side': 'LONG'} short_signal = {'symbol': symbol, 'side': 'SHORT'} long_allowed, long_reason = detector.should_allow_signal(long_signal) short_allowed, short_reason = detector.should_allow_signal(short_signal) print(f"\nSignal LONG: {'OK' if long_allowed else 'BLOCKED'} - {long_reason}") print(f"Signal SHORT: {'OK' if short_allowed else 'BLOCKED'} - {short_reason}") if 'timeframes' in details: print("\nAnalyse par timeframe:") for tf, data in details['timeframes'].items(): print(f" {tf}: {data['trend']} ({data['change']:+.2f}%)") if __name__ == "__main__": test_trend_detector()