Place-Order-Trading-Bot/reversal_detector.py

#!/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']}")