""" Détecteur de tops pour shitcoins Stratégie: Shorter après le premier top d'un pump ⚠️ ULTRA RISQUÉ - Possibilité de pump supplémentaire Utiliser seulement avec leverage modéré (50-100x) et stop-loss ultra serré """ import pandas as pd import numpy as np from typing import Dict, List, Optional, Tuple import logging from dataclasses import dataclass from datetime import datetime, timedelta import sys sys.path.append('..') import config from signals.short_term_signals import TechnicalIndicators, Signal @dataclass class TopDetection: """Détection d'un top de shitcoin""" symbol: str detection_time: pd.Timestamp current_price: float pump_start_price: float pump_percent: float time_since_pump_start: float # heures volume_spike_ratio: float rsi: float has_bearish_divergence: bool rejection_wick_percent: float confidence_score: float suggested_entry: float suggested_stop_loss: float suggested_take_profit: float class ShitcoinTopDetector: """Détecte les tops de pumps pour shorter""" def __init__(self): self.logger = logging.getLogger(self.__class__.__name__) self.params = config.SHITCOIN_TOP_PARAMS def detect_pump(self, df: pd.DataFrame) -> Optional[Tuple[float, float, int]]: """ Détecte un pump récent Returns: (start_price, pump_percent, candles_ago) si pump détecté, None sinon """ if len(df) < 100: return None current_price = df.iloc[-1]['close'] # Chercher le point bas dans les dernières 24h # (24h = 1440 chandeliers sur 1m, on limite à 100 pour rester court terme) lookback = min(100, len(df)) window = df.iloc[-lookback:] lowest_idx = window['low'].idxmin() lowest_price = window.loc[lowest_idx, 'low'] lowest_time = window.loc[lowest_idx, 'timestamp'] # Calculer le pump pump_percent = (current_price - lowest_price) / lowest_price * 100 # Distance en temps candles_since_low = len(window) - (window.index.get_loc(lowest_idx) + 1) if pump_percent >= self.params['min_pump_percent']: return (lowest_price, pump_percent, candles_since_low) return None def detect_bearish_divergence(self, df: pd.DataFrame, lookback: int = 20) -> bool: """ Détecte une divergence baissière (price higher high, RSI lower high) Args: df: DataFrame avec RSI déjà calculé lookback: Nombre de périodes à analyser Returns: True si divergence détectée """ if len(df) < lookback + 10: return False window = df.iloc[-lookback:] # Trouver les highs de price price_peaks = [] for i in range(1, len(window) - 1): if window.iloc[i]['high'] > window.iloc[i-1]['high'] and \ window.iloc[i]['high'] > window.iloc[i+1]['high']: price_peaks.append((i, window.iloc[i]['high'])) # Trouver les highs de RSI rsi_peaks = [] for i in range(1, len(window) - 1): if window.iloc[i]['rsi'] > window.iloc[i-1]['rsi'] and \ window.iloc[i]['rsi'] > window.iloc[i+1]['rsi']: rsi_peaks.append((i, window.iloc[i]['rsi'])) # Il faut au moins 2 peaks de chaque if len(price_peaks) < 2 or len(rsi_peaks) < 2: return False # Prendre les 2 derniers de chaque price_peak1, price_val1 = price_peaks[-2] price_peak2, price_val2 = price_peaks[-1] # Trouver les RSI peaks les plus proches dans le temps rsi_near_peak1 = min(rsi_peaks, key=lambda x: abs(x[0] - price_peak1)) rsi_near_peak2 = min(rsi_peaks, key=lambda x: abs(x[0] - price_peak2)) _, rsi_val1 = rsi_near_peak1 _, rsi_val2 = rsi_near_peak2 # Divergence: price fait higher high, RSI fait lower high price_higher_high = price_val2 > price_val1 rsi_lower_high = rsi_val2 < rsi_val1 return price_higher_high and rsi_lower_high def calculate_rejection_wick(self, candle: pd.Series) -> float: """ Calcule le pourcentage de la mèche haute (rejection wick) Args: candle: Série avec open, high, low, close Returns: Pourcentage de rejection wick """ high = candle['high'] close = candle['close'] open_price = candle['open'] # Top du body body_top = max(open_price, close) # Wick = distance entre high et top du body wick = high - body_top # Pourcentage par rapport au body body_size = abs(close - open_price) if body_size == 0: body_size = high - candle['low'] # Fallback: total range if body_size == 0: return 0.0 wick_percent = (wick / body_size) * 100 return wick_percent def analyze_for_top(self, df: pd.DataFrame, symbol: str) -> Optional[TopDetection]: """ Analyse complète pour détecter un top de shitcoin Args: df: DataFrame OHLCV symbol: Symbole du token Returns: TopDetection si top détecté, None sinon """ if len(df) < 100: return None # 1. Détecter le pump pump_info = self.detect_pump(df) if not pump_info: return None start_price, pump_percent, candles_since_low = pump_info self.logger.debug(f"{symbol}: Pump détecté +{pump_percent:.1f}% depuis {candles_since_low} chandeliers") # 2. Calculer indicateurs df = df.copy() df['rsi'] = TechnicalIndicators.rsi(df['close'], period=config.RSI_PERIOD) df['volume_ma'] = TechnicalIndicators.volume_profile(df['volume'], period=20) df['volume_ratio'] = df['volume'] / df['volume_ma'] current = df.iloc[-1] current_price = current['close'] rsi = current['rsi'] volume_ratio = current['volume_ratio'] # 3. Vérifier les critères de top confidence_points = [] score = 0.0 # Critère 1: Pump suffisant if pump_percent >= self.params['min_pump_percent']: confidence_points.append(f"PUMP_{pump_percent:.0f}%") score += 0.25 # Critère 2: Volume spike if volume_ratio >= self.params['volume_spike_min']: confidence_points.append(f"VOLUME_{volume_ratio:.1f}x") score += 0.20 # Critère 3: RSI overbought if rsi >= self.params['rsi_overbought']: confidence_points.append(f"RSI_{rsi:.0f}") score += 0.20 # Critère 4: Divergence baissière has_divergence = False if self.params['bearish_divergence']: has_divergence = self.detect_bearish_divergence(df) if has_divergence: confidence_points.append("DIVERGENCE") score += 0.20 # Critère 5: Rejection wick rejection_wick = self.calculate_rejection_wick(current) if rejection_wick >= self.params['rejection_wick_min'] * 100: confidence_points.append(f"WICK_{rejection_wick:.0f}%") score += 0.15 # Score minimum requis: 0.60 (3 critères sur 5) if score < 0.60: return None # Calculer temps depuis début pump (en heures) hours_since_pump = candles_since_low / 60 # 1m timeframe # Vérifier timeframe du pump (ne pas shorter trop tôt) if hours_since_pump > self.params['pump_timeframe_hours']: self.logger.debug(f"{symbol}: Pump trop ancien ({hours_since_pump:.1f}h)") return None # 4. Calculer niveaux d'entrée, SL, TP # Entry: Attendre un pull-back de 2% depuis le high récent recent_high = df.iloc[-20:]['high'].max() suggested_entry = recent_high * 0.98 # Stop-loss ultra serré: +1% au-dessus du high récent (risque de continuation) suggested_sl = recent_high * 1.01 # Take-profit: Retour vers le milieu du pump (50% retracement) pump_mid = start_price + (recent_high - start_price) * 0.5 suggested_tp = pump_mid return TopDetection( symbol=symbol, detection_time=current['timestamp'], current_price=current_price, pump_start_price=start_price, pump_percent=pump_percent, time_since_pump_start=hours_since_pump, volume_spike_ratio=volume_ratio, rsi=rsi, has_bearish_divergence=has_divergence, rejection_wick_percent=rejection_wick, confidence_score=score, suggested_entry=suggested_entry, suggested_stop_loss=suggested_sl, suggested_take_profit=suggested_tp ) def generate_short_signal(self, detection: TopDetection) -> Signal: """ Convertit une détection de top en signal SHORT Args: detection: TopDetection Returns: Signal SHORT """ confluences = [ f"PUMP_{detection.pump_percent:.0f}%", f"RSI_{detection.rsi:.0f}", f"VOLUME_{detection.volume_spike_ratio:.1f}x" ] if detection.has_bearish_divergence: confluences.append("DIVERGENCE") if detection.rejection_wick_percent >= 3: confluences.append(f"REJECTION_WICK_{detection.rejection_wick_percent:.0f}%") return Signal( symbol=detection.symbol, side='SHORT', score=detection.confidence_score, strategy='SHITCOIN_TOP', entry_price=detection.suggested_entry, stop_loss=detection.suggested_stop_loss, take_profit=detection.suggested_take_profit, confluences=confluences, timeframe=config.TIMEFRAME_PRIMARY, timestamp=detection.detection_time ) class ShitcoinScanner: """Scanner pour trouver des shitcoins en pump sur ApolloX""" def __init__(self, apollo_client): self.logger = logging.getLogger(self.__class__.__name__) self.client = apollo_client self.detector = ShitcoinTopDetector() def scan_for_pumps(self, min_volume_usd: float = 100000) -> List[TopDetection]: """ Scanne tous les marchés pour trouver des pumps Args: min_volume_usd: Volume minimum 24h en USD Returns: Liste de TopDetection trouvées """ self.logger.info("🔍 Scan des marchés pour détecter des tops...") # Récupérer tous les tickers all_tickers = self.client.get_all_tickers() if not all_tickers: self.logger.warning("⚠️ Aucun ticker récupéré") return [] # Filtrer USDT perps usdt_pairs = [ ticker for ticker in all_tickers if ticker['symbol'].endswith('USDT') ] self.logger.info(f"📊 {len(usdt_pairs)} paires USDT trouvées") # Récupérer données 24h pour chaque pumping_tokens = [] for ticker_data in usdt_pairs: symbol = ticker_data['symbol'] try: # Récupérer ticker 24h ticker = self.client.get_ticker(symbol) if not ticker: continue # Vérifier changement 24h change_24h = float(ticker.get('priceChangePercent', 0)) volume_usd = float(ticker.get('quoteVolume', 0)) # Filtres rapides if change_24h < self.detector.params['min_pump_percent']: continue if volume_usd < min_volume_usd: continue self.logger.info(f"🚀 {symbol}: +{change_24h:.1f}% (vol: ${volume_usd:,.0f})") # Récupérer klines pour analyse détaillée df = self.client.get_klines(symbol, config.TIMEFRAME_PRIMARY, limit=100) if df.empty: continue # Analyser pour top detection = self.detector.analyze_for_top(df, symbol) if detection: pumping_tokens.append(detection) self.logger.info( f"✅ TOP DÉTECTÉ: {symbol} | " f"Pump: +{detection.pump_percent:.1f}% | " f"Score: {detection.confidence_score:.2f}" ) except Exception as e: self.logger.debug(f"Erreur analyse {symbol}: {e}") continue self.logger.info(f"\n✅ Scan terminé: {len(pumping_tokens)} tops détectés") return pumping_tokens def get_shortable_signals(self, min_confidence: float = 0.70) -> List[Signal]: """ Retourne les signaux SHORT pour les tops détectés Args: min_confidence: Score minimum Returns: Liste de signaux SHORT """ detections = self.scan_for_pumps() signals = [] for detection in detections: if detection.confidence_score >= min_confidence: signal = self.detector.generate_short_signal(detection) signals.append(signal) return signals if __name__ == '__main__': # Test logging.basicConfig(level=logging.INFO) # Simuler un pump avec top dates = pd.date_range('2025-01-01', periods=100, freq='1min') # Créer un pump puis rejet prices = [] base = 1.0 # Phase pump (70 chandeliers) for i in range(70): base *= 1.01 # +1% par chandelierr prices.append(base) # Top avec wick de rejet prices.append(base * 1.05) # Spike # Rejet for i in range(29): base *= 0.99 # -1% prices.append(base) df = pd.DataFrame({ 'timestamp': dates, 'open': prices, 'high': [p * 1.01 for p in prices], 'low': [p * 0.99 for p in prices], 'close': prices, 'volume': [1000] * 70 + [10000] + [5000] * 29 # Volume spike au top }) # Ajouter un gros wick au top df.loc[70, 'high'] = df.loc[70, 'close'] * 1.10 detector = ShitcoinTopDetector() detection = detector.analyze_for_top(df, 'SHITUSDT') if detection: print("\n✅ TOP DÉTECTÉ!") print(f"Pump: +{detection.pump_percent:.1f}%") print(f"RSI: {detection.rsi:.0f}") print(f"Volume spike: {detection.volume_spike_ratio:.1f}x") print(f"Divergence: {detection.has_bearish_divergence}") print(f"Rejection wick: {detection.rejection_wick_percent:.1f}%") print(f"Score: {detection.confidence_score:.2%}") print(f"\nEntry: ${detection.suggested_entry:.4f}") print(f"SL: ${detection.suggested_stop_loss:.4f}") print(f"TP: ${detection.suggested_take_profit:.4f}") signal = detector.generate_short_signal(detection) print(f"\n📉 Signal: {signal.side} {signal.symbol}") print(f"Confluences: {', '.join(signal.confluences)}") else: print("❌ Pas de top détecté")