#!/usr/bin/env python3 """ 🔄 REVERSAL DETECTOR INTEGRATION Integriert den Reversal Detector in den Trading Bot VERWENDUNG IM NOTEBOOK: # Am Anfang importieren (nach den anderen Imports): from reversal_integration import reversal_detector, check_reversal_risk # In enhanced_trading_check_wrapper nach Signal Analysis: reversal_result = check_reversal_risk(signal_info, symbol) if not reversal_result['should_trade']: print(f"⛔ TRADE BLOCKIERT durch Reversal Detector") return None # Lot Multiplier anpassen: lot_multiplier *= reversal_result['lot_multiplier'] """ import logging from typing import Dict, Optional import pandas as pd # Import the reversal detector from reversal_detector import ReversalDetector logger = logging.getLogger(__name__) # ========================================== # GLOBAL REVERSAL DETECTOR INSTANCE # ========================================== # Initialize with defensive mode (blocks/reduces trades at high reversal risk) reversal_detector = ReversalDetector( rsi_period=14, ema_fast=9, ema_slow=21, volume_spike_mult=2.0, divergence_lookback=10, mode="defensive" # "defensive" or "info" ) # ========================================== # CONFIGURATION # ========================================== REVERSAL_CONFIG = { 'enabled': True, # Reversal Check aktiviert 'block_threshold': 70, # Score >= 70 = Trade blockieren 'reduce_threshold': 50, # Score >= 50 = Lot reduzieren 'caution_threshold': 30, # Score >= 30 = Warnung 'use_m15_data': True, # M15 für Analyse verwenden 'use_h1_confirmation': True, # H1 für Bestätigung } # ========================================== # INTEGRATION FUNCTION # ========================================== def check_reversal_risk(signal_info: Dict, symbol: str = "XAUUSD") -> Dict: """ Prüft Reversal-Risiko basierend auf Signal-Info Args: signal_info: Dict von extended_top_down_v2_adaptive() symbol: Trading Symbol Returns: Dict mit: - should_trade: bool - reversal_score: int (0-100) - lot_multiplier: float (1.0, 0.75, 0.5, 0.0) - action: str (ALLOW, CAUTION, REDUCE, BLOCK) - reason: str """ if not REVERSAL_CONFIG['enabled']: return { 'should_trade': True, 'reversal_score': 0, 'lot_multiplier': 1.0, 'action': 'ALLOW', 'reason': 'Reversal check disabled' } if signal_info is None: return { 'should_trade': True, 'reversal_score': 0, 'lot_multiplier': 1.0, 'action': 'ALLOW', 'reason': 'No signal info available' } try: # Bestimme aktuellen Trend entry_signal = signal_info.get('entry_signal', 0) standard_trend = signal_info.get('standard_trend', 'sideways') # Trend aus entry_signal ableiten falls standard_trend nicht verfügbar if standard_trend == 'sideways' and entry_signal != 0: current_trend = 'uptrend' if entry_signal == 1 else 'downtrend' else: current_trend = standard_trend # Hole M15 DataFrame aus trend_info trend_info = signal_info.get('trend_info', {}) df_m15 = None df_h1 = None # Versuche M15 Daten zu holen if REVERSAL_CONFIG['use_m15_data'] and 'M15' in trend_info: m15_info = trend_info['M15'] if isinstance(m15_info, dict) and 'df' in m15_info: df_m15 = m15_info['df'] elif hasattr(m15_info, 'df'): df_m15 = m15_info.df # Versuche H1 Daten zu holen if REVERSAL_CONFIG['use_h1_confirmation'] and 'H1' in trend_info: h1_info = trend_info['H1'] if isinstance(h1_info, dict) and 'df' in h1_info: df_h1 = h1_info['df'] elif hasattr(h1_info, 'df'): df_h1 = h1_info.df # Falls keine DataFrame verfügbar, versuche get_rates if df_m15 is None: try: # Import get_rates falls verfügbar import sys if 'get_rates' in dir(sys.modules.get('__main__', {})): from __main__ import get_rates df_m15 = get_rates('m15', 100, symbol) except Exception as e: logger.debug(f"Could not get M15 data: {e}") if df_m15 is None or len(df_m15) < 50: print(f"⚠️ Reversal Check: Not enough M15 data, skipping") return { 'should_trade': True, 'reversal_score': 0, 'lot_multiplier': 1.0, 'action': 'ALLOW', 'reason': 'Insufficient data for reversal check' } # Führe Reversal-Analyse durch result = reversal_detector.analyze( df=df_m15, current_trend=current_trend, df_higher_tf=df_h1 ) reversal_score = result['reversal_score'] # Bestimme Aktion basierend auf Config-Thresholds if reversal_score >= REVERSAL_CONFIG['block_threshold']: action = 'BLOCK' lot_multiplier = 0.0 should_trade = False reason = f"High reversal risk ({reversal_score}% >= {REVERSAL_CONFIG['block_threshold']}%)" elif reversal_score >= REVERSAL_CONFIG['reduce_threshold']: action = 'REDUCE' lot_multiplier = 0.5 should_trade = True reason = f"Moderate reversal risk ({reversal_score}%), lot reduced to 50%" elif reversal_score >= REVERSAL_CONFIG['caution_threshold']: action = 'CAUTION' lot_multiplier = 0.75 should_trade = True reason = f"Low reversal risk ({reversal_score}%), lot reduced to 75%" else: action = 'ALLOW' lot_multiplier = 1.0 should_trade = True reason = f"No significant reversal risk ({reversal_score}%)" return { 'should_trade': should_trade, 'reversal_score': reversal_score, 'lot_multiplier': lot_multiplier, 'action': action, 'reason': reason, 'reversal_type': result.get('reversal_type', 'NONE'), 'signals': result.get('signals', {}) } except Exception as e: logger.error(f"Reversal check error: {e}") print(f"⚠️ Reversal Check Error: {e}") return { 'should_trade': True, 'reversal_score': 0, 'lot_multiplier': 1.0, 'action': 'ALLOW', 'reason': f'Error: {e}' } def get_reversal_status() -> str: """Gibt aktuellen Reversal Detector Status zurück""" status = [] status.append("=" * 60) status.append("🔄 REVERSAL DETECTOR STATUS") status.append("=" * 60) status.append(f" Enabled: {REVERSAL_CONFIG['enabled']}") status.append(f" Mode: {reversal_detector.mode}") status.append(f" Block Threshold: {REVERSAL_CONFIG['block_threshold']}%") status.append(f" Reduce Threshold: {REVERSAL_CONFIG['reduce_threshold']}%") status.append(f" Caution Threshold: {REVERSAL_CONFIG['caution_threshold']}%") status.append("=" * 60) return "\n".join(status) def enable_reversal_check(): """Aktiviert den Reversal Check""" REVERSAL_CONFIG['enabled'] = True print("✅ Reversal Check ENABLED") def disable_reversal_check(): """Deaktiviert den Reversal Check""" REVERSAL_CONFIG['enabled'] = False print("⚠️ Reversal Check DISABLED") def set_reversal_mode(mode: str): """ Setzt den Reversal Detector Mode Args: mode: "defensive" (blockiert Trades) oder "info" (nur Logging) """ global reversal_detector reversal_detector = ReversalDetector( rsi_period=14, ema_fast=9, ema_slow=21, volume_spike_mult=2.0, divergence_lookback=10, mode=mode ) print(f"✅ Reversal Detector mode set to: {mode.upper()}") # ========================================== # STANDALONE TEST # ========================================== if __name__ == "__main__": print(get_reversal_status()) # Test mit simulierten Daten import numpy as np np.random.seed(42) n = 100 prices = np.linspace(100, 110, n) + np.random.normal(0, 0.5, n) df = pd.DataFrame({ 'open': prices - np.random.uniform(0, 0.3, n), 'high': prices + np.random.uniform(0.3, 0.8, n), 'low': prices - np.random.uniform(0.3, 0.8, n), 'close': prices, 'tick_volume': np.random.randint(100, 500, n) }) # Simuliere signal_info mock_signal_info = { 'entry_signal': 1, 'standard_trend': 'uptrend', 'trend_info': { 'M15': {'df': df} } } print("\n🧪 Testing Reversal Check...") result = check_reversal_risk(mock_signal_info, "XAUUSD") print(f"\n📊 Result:") print(f" Should Trade: {result['should_trade']}") print(f" Reversal Score: {result['reversal_score']}%") print(f" Action: {result['action']}") print(f" Lot Multiplier: {result['lot_multiplier']}") print(f" Reason: {result['reason']}")