""" Stratégies de Trading Avancées Inspiré de: https://github.com/conor19w/Binance-Futures-Trading-Bot 11 stratégies techniques implémentées """ import numpy as np from typing import List, Dict, Optional, Tuple class AdvancedStrategies: """ Collection de 11 stratégies de trading avancées """ @staticmethod def calculate_ema(prices: List[float], period: int) -> Optional[float]: """Calcule l'EMA (Exponential Moving Average)""" if len(prices) < period: return None prices_array = np.array(prices[-period:]) multiplier = 2 / (period + 1) ema = prices_array[0] for price in prices_array[1:]: ema = (price * multiplier) + (ema * (1 - multiplier)) return ema @staticmethod def calculate_rsi(prices: List[float], period: int = 14) -> Optional[float]: """Calcule le RSI (Relative Strength Index)""" if len(prices) < period + 1: return None deltas = np.diff(prices[-period-1:]) gains = np.where(deltas > 0, deltas, 0) losses = np.where(deltas < 0, -deltas, 0) avg_gain = np.mean(gains) avg_loss = np.mean(losses) if avg_loss == 0: return 100 rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) return rsi @staticmethod def calculate_macd(prices: List[float], fast=12, slow=26, signal=9) -> Optional[Dict[str, float]]: """Calcule le MACD (Moving Average Convergence Divergence)""" if len(prices) < slow + signal: return None ema_fast = AdvancedStrategies.calculate_ema(prices, fast) ema_slow = AdvancedStrategies.calculate_ema(prices, slow) if ema_fast is None or ema_slow is None: return None macd_line = ema_fast - ema_slow # 🔧 FIX AUDIT 28/02: Implémenter un vrai signal MACD (EMA 9 périodes du MACD) # Avant: signal_line = macd_line → histogram toujours 0 # Maintenant: calcul du signal line comme EMA-9 du MACD # On a besoin de l'historique MACD pour cela if len(prices) >= slow + signal: # Calculer les MACD lines historiques pour les 'signal' dernières bougies macd_history = [] for i in range(signal, 0, -1): p = prices[:len(prices)-i+1] ef = AdvancedStrategies.calculate_ema(p, fast) es = AdvancedStrategies.calculate_ema(p, slow) if ef is not None and es is not None: macd_history.append(ef - es) macd_history.append(macd_line) if len(macd_history) >= signal: # EMA du MACD pour le signal line signal_multiplier = 2 / (signal + 1) signal_val = macd_history[0] for mh in macd_history[1:]: signal_val = (mh - signal_val) * signal_multiplier + signal_val signal_line = signal_val else: signal_line = macd_line else: signal_line = macd_line histogram = macd_line - signal_line return { 'macd': macd_line, 'signal': signal_line, 'histogram': histogram } @staticmethod def calculate_stochastic(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> Optional[Dict[str, float]]: """Calcule l'oscillateur Stochastique""" if len(closes) < period: return None high_max = max(highs[-period:]) low_min = min(lows[-period:]) current_close = closes[-1] if high_max == low_min: return None k = 100 * (current_close - low_min) / (high_max - low_min) # 🔧 FIX AUDIT 28/02: Calculer un vrai %D (SMA 3 périodes de %K) # Avant: %D = %K (pas de lissage, perte du signal-line crossing) # Fallback: Si pas assez de données, %D ≈ %K if len(closes) >= period + 3: k_values = [] for i in range(3): idx = len(closes) - 3 + i h = max(highs[idx-period+1:idx+1]) l = min(lows[idx-period+1:idx+1]) if h != l: k_values.append(100 * (closes[idx] - l) / (h - l)) else: k_values.append(50.0) d = sum(k_values) / len(k_values) else: d = k return {'k': k, 'd': d} @staticmethod def calculate_bollinger_bands(prices: List[float], period: int = 20, std_dev: float = 2.0) -> Optional[Dict[str, float]]: """Calcule les Bandes de Bollinger""" if len(prices) < period: return None recent_prices = prices[-period:] sma = np.mean(recent_prices) std = np.std(recent_prices) upper = sma + (std * std_dev) lower = sma - (std * std_dev) # %B indicator current_price = prices[-1] if upper == lower: percent_b = 0.5 else: percent_b = (current_price - lower) / (upper - lower) return { 'upper': upper, 'middle': sma, 'lower': lower, 'percent_b': percent_b } # ═══════════════════════════════════════════════════════════════════════ # STRATÉGIE 1: GOLDEN CROSS (Croisement EMA 20/50/100 + RSI) # ═══════════════════════════════════════════════════════════════════════ @staticmethod def golden_cross(prices: List[float], **kwargs) -> Tuple[str, int, str]: """ Croisement d'or: EMA20 > EMA50 > EMA100 = BUY """ if len(prices) < 100: return "HOLD", 0, "Pas assez de données" ema20 = AdvancedStrategies.calculate_ema(prices, 20) ema50 = AdvancedStrategies.calculate_ema(prices, 50) ema100 = AdvancedStrategies.calculate_ema(prices, 100) rsi = AdvancedStrategies.calculate_rsi(prices, 14) if None in [ema20, ema50, ema100, rsi]: return "HOLD", 0, "Indicateurs incomplets" signals = 0 # Golden Cross: EMA courte > EMA moyenne > EMA longue if ema20 > ema50 > ema100: signals += 3 # Death Cross: EMA courte < EMA moyenne < EMA longue elif ema20 < ema50 < ema100: signals -= 3 # Confirmation RSI if rsi < 30: signals += 2 # Oversold elif rsi > 70: signals -= 2 # Overbought if signals >= 3: return "BUY", signals, f"Golden Cross (EMA20>{ema20:.2f}, RSI={rsi:.0f})" elif signals <= -3: return "SELL", abs(signals), f"Death Cross (EMA20<{ema20:.2f}, RSI={rsi:.0f})" return "HOLD", 0, "Pas de signal clair" # ═══════════════════════════════════════════════════════════════════════ # STRATÉGIE 2: STOCHASTIC RSI MACD # ═══════════════════════════════════════════════════════════════════════ @staticmethod def stoch_rsi_macd(prices: List[float], highs: List[float], lows: List[float], **kwargs) -> Tuple[str, int, str]: """ Combine Stochastique, RSI et MACD pour signaux robustes """ if len(prices) < 50: return "HOLD", 0, "Pas assez de données" rsi = AdvancedStrategies.calculate_rsi(prices, 14) stoch = AdvancedStrategies.calculate_stochastic(highs, lows, prices, 14) macd = AdvancedStrategies.calculate_macd(prices, 12, 26, 9) if None in [rsi, stoch, macd]: return "HOLD", 0, "Indicateurs incomplets" signals = 0 # RSI oversold/overbought if rsi < 30: signals += 2 elif rsi > 70: signals -= 2 # Stochastique if stoch['k'] < 20: signals += 2 elif stoch['k'] > 80: signals -= 2 # MACD crossover if macd['histogram'] > 0: signals += 1 elif macd['histogram'] < 0: signals -= 1 if signals >= 4: return "BUY", signals, f"Stoch/RSI/MACD convergent (RSI={rsi:.0f}, Stoch={stoch['k']:.0f})" elif signals <= -4: return "SELL", abs(signals), f"Signaux baissiers multiples" return "HOLD", 0, "Signaux mixtes" # ═══════════════════════════════════════════════════════════════════════ # STRATÉGIE 3: TRIPLE EMA (Croisement 3/6/9) # ═══════════════════════════════════════════════════════════════════════ @staticmethod def triple_ema(prices: List[float], **kwargs) -> Tuple[str, int, str]: """ Triple EMA rapide pour scalping: 3/6/9 périodes """ if len(prices) < 15: return "HOLD", 0, "Pas assez de données" ema3 = AdvancedStrategies.calculate_ema(prices, 3) ema6 = AdvancedStrategies.calculate_ema(prices, 6) ema9 = AdvancedStrategies.calculate_ema(prices, 9) if None in [ema3, ema6, ema9]: return "HOLD", 0, "EMAs incomplets" signals = 0 # Alignement haussier if ema3 > ema6 > ema9: signals += 4 # Alignement baissier elif ema3 < ema6 < ema9: signals -= 4 # Croisement récent if len(prices) >= 2: prev_ema3 = AdvancedStrategies.calculate_ema(prices[:-1], 3) prev_ema6 = AdvancedStrategies.calculate_ema(prices[:-1], 6) if prev_ema3 and prev_ema6: # Croisement haussier if prev_ema3 <= prev_ema6 and ema3 > ema6: signals += 2 # Croisement baissier elif prev_ema3 >= prev_ema6 and ema3 < ema6: signals -= 2 if signals >= 4: return "BUY", signals, f"Triple EMA haussier (3>{ema3:.4f} > 6>{ema6:.4f})" elif signals <= -4: return "SELL", abs(signals), "Triple EMA baissier" return "HOLD", 0, "Pas de tendance claire" # ═══════════════════════════════════════════════════════════════════════ # STRATÉGIE 4: BOLLINGER BANDS + STOCHASTIC # ═══════════════════════════════════════════════════════════════════════ @staticmethod def stoch_bb(prices: List[float], highs: List[float], lows: List[float], **kwargs) -> Tuple[str, int, str]: """ Combine Bandes de Bollinger avec Stochastique """ if len(prices) < 25: return "HOLD", 0, "Pas assez de données" bb = AdvancedStrategies.calculate_bollinger_bands(prices, 20, 2.0) stoch = AdvancedStrategies.calculate_stochastic(highs, lows, prices, 14) if None in [bb, stoch]: return "HOLD", 0, "Indicateurs incomplets" signals = 0 current_price = prices[-1] # Prix proche de la bande inférieure + stochastique oversold if bb['percent_b'] < 0.2 and stoch['k'] < 20: signals += 4 # Prix proche de la bande supérieure + stochastique overbought elif bb['percent_b'] > 0.8 and stoch['k'] > 80: signals -= 4 # Squeeze (volatilité faible) bb_width = (bb['upper'] - bb['lower']) / bb['middle'] if bb_width < 0.05: # Bandes très serrées signals += 1 # Potentiel de breakout if signals >= 4: return "BUY", signals, f"BB+Stoch oversold (%B={bb['percent_b']:.2f})" elif signals <= -4: return "SELL", abs(signals), "BB+Stoch overbought" return "HOLD", 0, "En milieu de bande" # ═══════════════════════════════════════════════════════════════════════ # STRATÉGIE 5: BREAKOUT (Cassure avec volume) # ═══════════════════════════════════════════════════════════════════════ @staticmethod def breakout(prices: List[float], volumes: Optional[List[float]] = None, **kwargs) -> Tuple[str, int, str]: """ Détecte les cassures de prix avec confirmation volume """ if len(prices) < 50: return "HOLD", 0, "Pas assez de données" current_price = prices[-1] # Plus haut sur 20 périodes high_20 = max(prices[-20:]) low_20 = min(prices[-20:]) # Plus haut sur 50 périodes high_50 = max(prices[-50:]) low_50 = min(prices[-50:]) signals = 0 # Breakout haussier if current_price >= high_20 * 0.998: # À 0.2% du plus haut signals += 3 # Confirmation si c'est aussi un nouveau high sur 50 périodes if current_price >= high_50 * 0.998: signals += 2 # Breakout baissier elif current_price <= low_20 * 1.002: signals -= 3 if current_price <= low_50 * 1.002: signals -= 2 # Confirmation volume (si disponible) if volumes and len(volumes) >= 20: avg_volume = np.mean(volumes[-20:-1]) current_volume = volumes[-1] if current_volume > avg_volume * 1.5: # Volume 50% au-dessus if signals > 0: signals += 2 elif signals < 0: signals -= 2 if signals >= 4: return "BUY", signals, f"Breakout haussier (prix={current_price:.6f})" elif signals <= -4: return "SELL", abs(signals), "Breakdown baissier" return "HOLD", 0, "Pas de breakout" # ═══════════════════════════════════════════════════════════════════════ # STRATÉGIE 6: CANDLE WICK (Pattern de mèches) # ═══════════════════════════════════════════════════════════════════════ @staticmethod def candle_wick(opens: List[float], highs: List[float], lows: List[float], closes: List[float], **kwargs) -> Tuple[str, int, str]: """ Détecte les patterns de chandelier avec grandes mèches """ if len(closes) < 5: return "HOLD", 0, "Pas assez de bougies" signals = 0 # Analyser la dernière bougie last_open = opens[-1] last_high = highs[-1] last_low = lows[-1] last_close = closes[-1] body = abs(last_close - last_open) upper_wick = last_high - max(last_open, last_close) lower_wick = min(last_open, last_close) - last_low total_range = last_high - last_low if total_range == 0: return "HOLD", 0, "Bougie doji" # Hammer (marteau) - grande mèche inférieure if lower_wick > body * 2 and upper_wick < body * 0.5 and last_close > last_open: signals += 4 # Shooting star - grande mèche supérieure elif upper_wick > body * 2 and lower_wick < body * 0.5 and last_close < last_open: signals -= 4 # Engulfing pattern if len(closes) >= 2: prev_open = opens[-2] prev_close = closes[-2] prev_body = abs(prev_close - prev_open) # Bullish engulfing if (prev_close < prev_open and # Bougie précédente rouge last_close > last_open and # Bougie actuelle verte body > prev_body * 1.5): # Corps actuel englobe le précédent signals += 3 # Bearish engulfing elif (prev_close > prev_open and last_close < last_open and body > prev_body * 1.5): signals -= 3 if signals >= 4: return "BUY", signals, "Pattern haussier (Hammer/Engulfing)" elif signals <= -4: return "SELL", abs(signals), "Pattern baissier (Shooting Star)" return "HOLD", 0, "Pas de pattern clair" # ═══════════════════════════════════════════════════════════════════════ # FONCTION PRINCIPALE: ÉVALUATION MULTI-STRATÉGIES # ═══════════════════════════════════════════════════════════════════════ @staticmethod def evaluate_all_strategies(prices: List[float], highs: Optional[List[float]] = None, lows: Optional[List[float]] = None, opens: Optional[List[float]] = None, closes: Optional[List[float]] = None, volumes: Optional[List[float]] = None) -> Dict: """ Évalue toutes les stratégies et retourne un consensus """ # Utiliser prices comme closes si pas fourni if closes is None: closes = prices if highs is None: highs = prices if lows is None: lows = prices if opens is None: opens = prices results = {} buy_signals = 0 sell_signals = 0 total_confidence = 0 strategies = [ ("Golden Cross", AdvancedStrategies.golden_cross, {'prices': prices}), ("Stoch/RSI/MACD", AdvancedStrategies.stoch_rsi_macd, {'prices': prices, 'highs': highs, 'lows': lows}), ("Triple EMA", AdvancedStrategies.triple_ema, {'prices': prices}), ("BB + Stochastic", AdvancedStrategies.stoch_bb, {'prices': prices, 'highs': highs, 'lows': lows}), ("Breakout", AdvancedStrategies.breakout, {'prices': prices, 'volumes': volumes}), ("Candle Wick", AdvancedStrategies.candle_wick, {'opens': opens, 'highs': highs, 'lows': lows, 'closes': closes}), ] for name, strategy_func, params in strategies: try: signal, confidence, reason = strategy_func(**params) results[name] = { 'signal': signal, 'confidence': confidence, 'reason': reason } if signal == "BUY": buy_signals += confidence total_confidence += confidence elif signal == "SELL": sell_signals += confidence total_confidence += confidence except Exception as e: results[name] = { 'signal': 'ERROR', 'confidence': 0, 'reason': str(e) } # Consensus if buy_signals > sell_signals * 1.5: # Majorité achats consensus = "BUY" consensus_strength = buy_signals elif sell_signals > buy_signals * 1.5: # Majorité ventes consensus = "SELL" consensus_strength = sell_signals else: consensus = "HOLD" consensus_strength = 0 return { 'consensus': consensus, 'consensus_strength': consensus_strength, 'buy_signals': buy_signals, 'sell_signals': sell_signals, 'total_confidence': total_confidence, 'strategies': results }