#!/usr/bin/env python3 """ 🔄 TREND REVERSAL DETECTOR Erkennt potenzielle Trendumkehrungen durch Multi-Signal-Analyse SIGNALE: 1. RSI Divergenz (Bullish/Bearish) 2. EMA Slope Change 3. Volume Spike 4. Candlestick Patterns 5. Break of Structure VERWENDUNG: from reversal_detector import ReversalDetector detector = ReversalDetector() result = detector.analyze(df_m15, df_h1, current_trend='uptrend') if result['reversal_score'] >= 60: print("Reversal-Warnung! Trade blockieren.") """ import numpy as np import pandas as pd from typing import Dict, Tuple, Optional, List import logging logger = logging.getLogger(__name__) class ReversalDetector: """ Multi-Signal Trend Reversal Detector Kombiniert mehrere Indikatoren um Trendumkehrungen zu erkennen: - RSI Divergenz - EMA Slope Änderung - Volume Spikes - Candlestick Patterns - Break of Structure """ def __init__(self, rsi_period: int = 14, ema_fast: int = 9, ema_slow: int = 21, volume_spike_mult: float = 2.0, divergence_lookback: int = 10, mode: str = "defensive"): """ Args: rsi_period: RSI Berechnungsperiode ema_fast: Schneller EMA für Slope ema_slow: Langsamer EMA für Slope volume_spike_mult: Multiplikator für Volume-Spike-Erkennung divergence_lookback: Bars für Divergenz-Suche mode: "defensive" (blockiert Trades) oder "info" (nur Logging) """ self.rsi_period = rsi_period self.ema_fast = ema_fast self.ema_slow = ema_slow self.volume_spike_mult = volume_spike_mult self.divergence_lookback = divergence_lookback self.mode = mode # Gewichtung der Signale self.weights = { 'rsi_divergence': 30, # Sehr zuverlässig 'ema_slope_change': 20, # Früher Indikator 'volume_spike': 15, # Bestätigung 'candlestick': 15, # Pattern-basiert 'break_of_structure': 20 # Strukturbruch } logger.info("=" * 60) logger.info("🔄 REVERSAL DETECTOR INITIALIZED") logger.info("=" * 60) logger.info(f" Mode: {mode.upper()}") logger.info(f" RSI Period: {rsi_period}") logger.info(f" EMA Fast/Slow: {ema_fast}/{ema_slow}") logger.info(f" Volume Spike Mult: {volume_spike_mult}x") logger.info(f" Divergence Lookback: {divergence_lookback} bars") logger.info("=" * 60) # ========================================== # MAIN ANALYSIS METHOD # ========================================== def analyze(self, df: pd.DataFrame, current_trend: str, df_higher_tf: Optional[pd.DataFrame] = None) -> Dict: """ Führt vollständige Reversal-Analyse durch Args: df: DataFrame mit OHLCV Daten (primärer Timeframe, z.B. M15) current_trend: 'uptrend' oder 'downtrend' df_higher_tf: Optional höherer Timeframe für Bestätigung (z.B. H1) Returns: Dict mit Reversal-Score und Details """ if df is None or len(df) < 50: return self._empty_result("Insufficient data") # Stelle sicher dass nötige Indikatoren berechnet sind df = self._ensure_indicators(df) signals = {} # 1. RSI Divergenz signals['rsi_divergence'] = self._detect_rsi_divergence(df, current_trend) # 2. EMA Slope Change signals['ema_slope_change'] = self._detect_ema_slope_change(df, current_trend) # 3. Volume Spike signals['volume_spike'] = self._detect_volume_spike(df) # 4. Candlestick Patterns signals['candlestick'] = self._detect_candlestick_patterns(df, current_trend) # 5. Break of Structure signals['break_of_structure'] = self._detect_break_of_structure(df, current_trend) # Berechne Gesamt-Score reversal_score = self._calculate_reversal_score(signals) # Bestimme Reversal-Typ if current_trend == 'uptrend': reversal_type = 'BEARISH' if reversal_score >= 30 else 'NONE' else: reversal_type = 'BULLISH' if reversal_score >= 30 else 'NONE' # Bestimme Aktion action, lot_multiplier = self._determine_action(reversal_score) result = { 'reversal_score': reversal_score, 'reversal_type': reversal_type, 'current_trend': current_trend, 'signals': signals, 'action': action, 'lot_multiplier': lot_multiplier, 'should_trade': action != 'BLOCK', 'mode': self.mode } # Logging self._log_analysis(result) return result # ========================================== # SIGNAL DETECTION METHODS # ========================================== def _detect_rsi_divergence(self, df: pd.DataFrame, trend: str) -> Dict: """ Erkennt RSI Divergenz Bearish Divergenz: Preis Higher High, RSI Lower High Bullish Divergenz: Preis Lower Low, RSI Higher Low """ if 'rsi' not in df.columns: return {'detected': False, 'strength': 0, 'type': 'none', 'description': 'RSI not available'} lookback = self.divergence_lookback recent = df.tail(lookback) if len(recent) < lookback: return {'detected': False, 'strength': 0, 'type': 'none', 'description': 'Not enough data'} prices = recent['close'].values rsi = recent['rsi'].values # Finde lokale Extrema price_highs_idx = self._find_local_extrema(prices, 'high') price_lows_idx = self._find_local_extrema(prices, 'low') rsi_highs_idx = self._find_local_extrema(rsi, 'high') rsi_lows_idx = self._find_local_extrema(rsi, 'low') detected = False div_type = 'none' strength = 0 description = 'No divergence' # Bearish Divergenz prüfen (für Uptrend) if trend == 'uptrend' and len(price_highs_idx) >= 2 and len(rsi_highs_idx) >= 2: # Preis macht Higher High if prices[price_highs_idx[-1]] > prices[price_highs_idx[-2]]: # RSI macht Lower High if rsi[rsi_highs_idx[-1]] < rsi[rsi_highs_idx[-2]]: detected = True div_type = 'bearish' # Stärke basierend auf Differenz price_diff = (prices[price_highs_idx[-1]] - prices[price_highs_idx[-2]]) / prices[price_highs_idx[-2]] rsi_diff = rsi[rsi_highs_idx[-2]] - rsi[rsi_highs_idx[-1]] strength = min(100, int((price_diff * 1000 + rsi_diff) * 2)) description = f"Bearish Divergence: Price HH, RSI LH (RSI diff: {rsi_diff:.1f})" # Bullish Divergenz prüfen (für Downtrend) if trend == 'downtrend' and len(price_lows_idx) >= 2 and len(rsi_lows_idx) >= 2: # Preis macht Lower Low if prices[price_lows_idx[-1]] < prices[price_lows_idx[-2]]: # RSI macht Higher Low if rsi[rsi_lows_idx[-1]] > rsi[rsi_lows_idx[-2]]: detected = True div_type = 'bullish' price_diff = (prices[price_lows_idx[-2]] - prices[price_lows_idx[-1]]) / prices[price_lows_idx[-2]] rsi_diff = rsi[rsi_lows_idx[-1]] - rsi[rsi_lows_idx[-2]] strength = min(100, int((price_diff * 1000 + rsi_diff) * 2)) description = f"Bullish Divergence: Price LL, RSI HL (RSI diff: {rsi_diff:.1f})" return { 'detected': detected, 'strength': strength, 'type': div_type, 'description': description } def _detect_ema_slope_change(self, df: pd.DataFrame, trend: str) -> Dict: """ Erkennt Änderung der EMA-Steigung Warnsignal wenn Slope sich dem Nullpunkt nähert oder Vorzeichen wechselt """ if f'ema_{self.ema_fast}' not in df.columns: # Berechne EMA falls nicht vorhanden df[f'ema_{self.ema_fast}'] = df['close'].ewm(span=self.ema_fast).mean() ema = df[f'ema_{self.ema_fast}'].values if len(ema) < 5: return {'detected': False, 'strength': 0, 'slope': 0, 'description': 'Not enough data'} # Berechne Slopes current_slope = (ema[-1] - ema[-3]) / ema[-3] * 100 # Letzte 3 Bars prev_slope = (ema[-4] - ema[-6]) / ema[-6] * 100 if len(ema) >= 6 else current_slope detected = False strength = 0 description = f"Slope: {current_slope:.4f}%" # Slope-Wechsel erkennen if trend == 'uptrend': # Warnung wenn positiver Slope abflacht oder negativ wird if current_slope < prev_slope * 0.5: # Slope halbiert sich detected = True strength = min(100, int(abs(prev_slope - current_slope) * 500)) description = f"Slope weakening: {prev_slope:.4f}% -> {current_slope:.4f}%" if current_slope < 0 and prev_slope > 0: # Vorzeichenwechsel detected = True strength = 80 description = f"Slope turned negative: {current_slope:.4f}%" elif trend == 'downtrend': # Warnung wenn negativer Slope abflacht oder positiv wird if current_slope > prev_slope * 0.5: # Slope halbiert sich detected = True strength = min(100, int(abs(prev_slope - current_slope) * 500)) description = f"Slope weakening: {prev_slope:.4f}% -> {current_slope:.4f}%" if current_slope > 0 and prev_slope < 0: # Vorzeichenwechsel detected = True strength = 80 description = f"Slope turned positive: {current_slope:.4f}%" return { 'detected': detected, 'strength': strength, 'slope': current_slope, 'prev_slope': prev_slope, 'description': description } def _detect_volume_spike(self, df: pd.DataFrame) -> Dict: """ Erkennt Volume-Spikes bei potenziellen Wendepunkten """ if 'tick_volume' not in df.columns and 'volume' not in df.columns: return {'detected': False, 'strength': 0, 'ratio': 1.0, 'description': 'Volume not available'} vol_col = 'tick_volume' if 'tick_volume' in df.columns else 'volume' # Durchschnittsvolumen der letzten 20 Bars avg_volume = df[vol_col].tail(20).mean() current_volume = df[vol_col].iloc[-1] if avg_volume == 0: return {'detected': False, 'strength': 0, 'ratio': 1.0, 'description': 'No volume data'} ratio = current_volume / avg_volume detected = ratio >= self.volume_spike_mult strength = min(100, int((ratio - 1) * 50)) if detected else 0 return { 'detected': detected, 'strength': strength, 'ratio': ratio, 'current_volume': current_volume, 'avg_volume': avg_volume, 'description': f"Volume: {ratio:.1f}x average" + (" (SPIKE!)" if detected else "") } def _detect_candlestick_patterns(self, df: pd.DataFrame, trend: str) -> Dict: """ Erkennt Reversal-Candlestick-Patterns - Doji (Unentschlossenheit) - Engulfing (Umkehr) - Hammer/Shooting Star - Morning/Evening Star """ if len(df) < 3: return {'detected': False, 'strength': 0, 'pattern': 'none', 'description': 'Not enough data'} # Letzte Kerzen current = df.iloc[-1] prev = df.iloc[-2] prev2 = df.iloc[-3] if len(df) >= 3 else None o, h, l, c = current['open'], current['high'], current['low'], current['close'] body = abs(c - o) upper_wick = h - max(o, c) lower_wick = min(o, c) - l total_range = h - l if h != l else 0.0001 detected = False strength = 0 pattern = 'none' description = 'No reversal pattern' # 1. Doji (Body < 10% der Range) if body / total_range < 0.1: detected = True strength = 40 pattern = 'doji' description = "Doji: Market indecision" # 2. Engulfing Pattern prev_body = abs(prev['close'] - prev['open']) if body > prev_body * 1.5: # Aktuelle Kerze größer if trend == 'uptrend' and c < o and prev['close'] > prev['open']: # Bearish Engulfing detected = True strength = 70 pattern = 'bearish_engulfing' description = "Bearish Engulfing: Strong reversal signal" elif trend == 'downtrend' and c > o and prev['close'] < prev['open']: # Bullish Engulfing detected = True strength = 70 pattern = 'bullish_engulfing' description = "Bullish Engulfing: Strong reversal signal" # 3. Hammer (Bullish) / Shooting Star (Bearish) if lower_wick > body * 2 and upper_wick < body * 0.5: # Hammer-Form if trend == 'downtrend': detected = True strength = 60 pattern = 'hammer' description = "Hammer: Potential bullish reversal" if upper_wick > body * 2 and lower_wick < body * 0.5: # Shooting Star-Form if trend == 'uptrend': detected = True strength = 60 pattern = 'shooting_star' description = "Shooting Star: Potential bearish reversal" # 4. Pin Bar (Long Wick Rejection) if upper_wick > total_range * 0.6 or lower_wick > total_range * 0.6: if trend == 'uptrend' and upper_wick > lower_wick: detected = True strength = max(strength, 55) pattern = 'pin_bar_bearish' if pattern == 'none' else pattern description = "Pin Bar: Upper wick rejection" elif trend == 'downtrend' and lower_wick > upper_wick: detected = True strength = max(strength, 55) pattern = 'pin_bar_bullish' if pattern == 'none' else pattern description = "Pin Bar: Lower wick rejection" return { 'detected': detected, 'strength': strength, 'pattern': pattern, 'description': description } def _detect_break_of_structure(self, df: pd.DataFrame, trend: str) -> Dict: """ Erkennt Break of Structure (BOS) Uptrend BOS: Erstes Lower Low nach Higher Highs Downtrend BOS: Erstes Higher High nach Lower Lows """ lookback = 20 if len(df) < lookback: return {'detected': False, 'strength': 0, 'description': 'Not enough data'} recent = df.tail(lookback) highs = recent['high'].values lows = recent['low'].values detected = False strength = 0 description = 'Structure intact' # Finde Swing Points swing_highs = [] swing_lows = [] for i in range(2, len(highs) - 2): # Swing High: Höher als 2 Bars links und rechts if highs[i] > highs[i-1] and highs[i] > highs[i-2] and highs[i] > highs[i+1] and highs[i] > highs[i+2]: swing_highs.append((i, highs[i])) # Swing Low: Niedriger als 2 Bars links und rechts if lows[i] < lows[i-1] and lows[i] < lows[i-2] and lows[i] < lows[i+1] and lows[i] < lows[i+2]: swing_lows.append((i, lows[i])) if len(swing_highs) >= 2 and len(swing_lows) >= 2: if trend == 'uptrend': # Check for Lower Low (BOS) if swing_lows[-1][1] < swing_lows[-2][1]: detected = True diff_pct = (swing_lows[-2][1] - swing_lows[-1][1]) / swing_lows[-2][1] * 100 strength = min(100, int(diff_pct * 20)) description = f"BOS: Lower Low detected ({diff_pct:.2f}% below prev swing)" elif trend == 'downtrend': # Check for Higher High (BOS) if swing_highs[-1][1] > swing_highs[-2][1]: detected = True diff_pct = (swing_highs[-1][1] - swing_highs[-2][1]) / swing_highs[-2][1] * 100 strength = min(100, int(diff_pct * 20)) description = f"BOS: Higher High detected ({diff_pct:.2f}% above prev swing)" return { 'detected': detected, 'strength': strength, 'swing_highs': len(swing_highs), 'swing_lows': len(swing_lows), 'description': description } # ========================================== # HELPER METHODS # ========================================== def _ensure_indicators(self, df: pd.DataFrame) -> pd.DataFrame: """Stellt sicher dass alle nötigen Indikatoren berechnet sind""" df = df.copy() # RSI if 'rsi' not in df.columns: delta = df['close'].diff() gain = (delta.where(delta > 0, 0)).rolling(window=self.rsi_period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=self.rsi_period).mean() rs = gain / loss df['rsi'] = 100 - (100 / (1 + rs)) # EMAs if f'ema_{self.ema_fast}' not in df.columns: df[f'ema_{self.ema_fast}'] = df['close'].ewm(span=self.ema_fast).mean() if f'ema_{self.ema_slow}' not in df.columns: df[f'ema_{self.ema_slow}'] = df['close'].ewm(span=self.ema_slow).mean() return df def _find_local_extrema(self, data: np.ndarray, extrema_type: str, window: int = 3) -> List[int]: """Findet lokale Hochs/Tiefs in einem Array""" extrema = [] for i in range(window, len(data) - window): if extrema_type == 'high': if all(data[i] >= data[i-j] for j in range(1, window+1)) and \ all(data[i] >= data[i+j] for j in range(1, window+1)): extrema.append(i) else: # low if all(data[i] <= data[i-j] for j in range(1, window+1)) and \ all(data[i] <= data[i+j] for j in range(1, window+1)): extrema.append(i) return extrema def _calculate_reversal_score(self, signals: Dict) -> int: """Berechnet gewichteten Reversal-Score""" score = 0 for signal_name, signal_data in signals.items(): if signal_data.get('detected', False): weight = self.weights.get(signal_name, 10) signal_strength = signal_data.get('strength', 50) / 100 score += weight * signal_strength return min(100, int(score)) def _determine_action(self, score: int) -> Tuple[str, float]: """ Bestimmt Aktion basierend auf Reversal-Score Returns: (action, lot_multiplier) """ if self.mode == "info": # Info-Mode: Nur Logging, keine Änderungen return "ALLOW", 1.0 # Defensive Mode if score < 30: return "ALLOW", 1.0 elif score < 50: return "CAUTION", 0.75 elif score < 70: return "REDUCE", 0.5 else: return "BLOCK", 0.0 def _empty_result(self, reason: str) -> Dict: """Gibt leeres Ergebnis zurück""" return { 'reversal_score': 0, 'reversal_type': 'NONE', 'current_trend': 'unknown', 'signals': {}, 'action': 'ALLOW', 'lot_multiplier': 1.0, 'should_trade': True, 'error': reason, 'mode': self.mode } def _log_analysis(self, result: Dict): """Loggt Analyse-Ergebnis""" score = result['reversal_score'] # Emoji basierend auf Score if score < 30: emoji = "✅" level = "LOW" elif score < 50: emoji = "⚠️" level = "MODERATE" elif score < 70: emoji = "🔶" level = "HIGH" else: emoji = "🔴" level = "CRITICAL" print(f"\n🔄 REVERSAL CHECK ({result['current_trend'].upper()}):") print("-" * 50) for signal_name, signal_data in result['signals'].items(): status = "⚠️" if signal_data.get('detected', False) else "✅" desc = signal_data.get('description', 'N/A') strength = signal_data.get('strength', 0) print(f" {signal_name.replace('_', ' ').title():20s}: {status} {desc}") if signal_data.get('detected'): print(f" {'':20s} Strength: {strength}%") print("-" * 50) print(f" {emoji} Reversal Score: {score}% ({level})") print(f" Action: {result['action']} | Lot Mult: {result['lot_multiplier']:.0%}") if result['action'] == 'BLOCK': print(f" ⛔ TRADE BLOCKED due to high reversal risk!") elif result['action'] == 'REDUCE': print(f" 📉 Lot size reduced to {result['lot_multiplier']:.0%}") # ========================================== # QUICK CHECK METHOD # ========================================== def quick_check(self, df: pd.DataFrame, trend: str) -> Tuple[bool, int, str]: """ Schnelle Reversal-Prüfung Returns: (should_trade, reversal_score, reason) """ result = self.analyze(df, trend) return result['should_trade'], result['reversal_score'], result['action'] # ========================================== # STANDALONE TESTING # ========================================== if __name__ == "__main__": print("=" * 60) print("🔄 REVERSAL DETECTOR TEST") print("=" * 60) # Simuliere Test-Daten import numpy as np np.random.seed(42) # Erzeuge einen Uptrend mit Reversal-Anzeichen n = 100 trend_base = np.linspace(100, 120, n) # Uptrend noise = np.random.normal(0, 1, n) # Füge Reversal-Anzeichen am Ende hinzu trend_base[-10:] = trend_base[-10] - np.linspace(0, 3, 10) # Abflachung prices = trend_base + noise df = pd.DataFrame({ 'open': prices - np.random.uniform(0, 0.5, n), 'high': prices + np.random.uniform(0.5, 1.5, n), 'low': prices - np.random.uniform(0.5, 1.5, n), 'close': prices, 'tick_volume': np.random.randint(100, 1000, n) }) # Volume Spike am Ende df.loc[df.index[-1], 'tick_volume'] = 5000 # Test detector = ReversalDetector(mode="defensive") result = detector.analyze(df, current_trend='uptrend') print(f"\n📊 Test Result:") print(f" Reversal Score: {result['reversal_score']}%") print(f" Should Trade: {result['should_trade']}") print(f" Action: {result['action']}") print(f" Lot Multiplier: {result['lot_multiplier']}")