cbazza/Place-Order-Trading-Bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add Trend Reversal Detector with multi-signal analysis
Deploy to Windows VPS / deploy (push) Has been cancelled
Details

Browse Source 
New features:
- Reversal Detector with 5 detection signals:
  - RSI Divergence (bearish/bullish)
  - EMA Slope Change detection
  - Volume Spike analysis
  - Candlestick patterns (Doji, Engulfing, Hammer, Pin Bar)
  - Break of Structure detection
- Integrated into enhanced_trading_check_wrapper (SCHRITT 2.5)
- Defensive mode: blocks trades at 70%+ reversal score
- Lot size reduction at 30-69% reversal score
- Enable Overlap session (13:00-16:00 UTC)

Files added:
- reversal_detector.py: Core detection algorithms
- reversal_integration.py: Bot integration wrapper
- REVERSAL_DETECTOR_INTEGRATION.md: Documentation

Modified:
- TradingBot notebook: Added reversal check integration
- session_filter_patch.py: Enabled overlap session

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
cbazza
2026-01-30 09:46:40 +01:00
co-authored by Claude Opus 4.5
parent b5b91224df
commit ce4e961541
5 changed files with 1623 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files
reversal_detector.py
+633
View File
@@ -0,0 +1,633 @@
#!/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']}")