RSI Divergence Signal

import pandas as pd
import numpy as np
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from datetime import datetime
import talib
from scipy.signal import argrelextrema

@dataclass
class RSIDivergenceSignal:
    """RSI Divergence Signal Data Class"""
    timestamp: datetime
    symbol: str
    divergence_type: str  # 'bullish' or 'bearish'
    regular_hidden: str   # 'regular' or 'hidden'
    rsi_value: float
    price: float
    pivot_points: List[Tuple[datetime, float]]
    confidence: float
    strength: str  # 'weak', 'moderate', 'strong'

class RSIDivergenceDetector:
    """RSI Divergence Detection Algorithm"""
    
    def __init__(self, rsi_period: int = 14, lookback: int = 20):
        self.rsi_period = rsi_period
        self.lookback = lookback
        self.oversold_threshold = 30
        self.overbought_threshold = 70
        
    def calculate_rsi(self, prices: pd.Series) -> pd.Series:
        """Calculate RSI indicator"""
        return talib.RSI(prices, timeperiod=self.rsi_period)
    
    def find_pivots(self, data: pd.Series, order: int = 5) -> pd.Series:
        """Find pivot highs and lows"""
        highs = argrelextrema(data.values, np.greater, order=order)[0]
        lows = argrelextrema(data.values, np.less, order=order)[0]
        return highs, lows
    
    def detect_divergence(self, prices: pd.Series, rsi: pd.Series) -> List[RSIDivergenceSignal]:
        """Detect RSI divergences"""
        signals = []
        
        # Find price and RSI pivots
        price_highs, price_lows = self.find_pivots(prices)
        rsi_highs, rsi_lows = self.find_pivots(rsi)
        
        # Check for bullish divergence (price makes lower low, RSI makes higher low)
        for i in range(1, len(price_lows)):
            if price_lows[i] in rsi_lows:
                curr_price_low = prices.iloc[price_lows[i]]
                prev_price_low = prices.iloc[price_lows[i-1]]
                curr_rsi_low = rsi.iloc[price_lows[i]]
                prev_rsi_low = rsi.iloc[price_lows[i-1]]
                
                # Regular bullish divergence
                if curr_price_low < prev_price_low and curr_rsi_low > prev_rsi_low:
                    signals.append(self._create_signal(
                        prices.index[price_lows[i]],
                        'bullish',
                        'regular',
                        curr_rsi_low,
                        curr_price_low,
                        [(prices.index[price_lows[i-1]], prev_price_low),
                         (prices.index[price_lows[i]], curr_price_low)]
                    ))
        
        # Check for bearish divergence (price makes higher high, RSI makes lower high)
        for i in range(1, len(price_highs)):
            if price_highs[i] in rsi_highs:
                curr_price_high = prices.iloc[price_highs[i]]
                prev_price_high = prices.iloc[price_highs[i-1]]
                curr_rsi_high = rsi.iloc[price_highs[i]]
                prev_rsi_high = rsi.iloc[price_highs[i-1]]
                
                # Regular bearish divergence
                if curr_price_high > prev_price_high and curr_rsi_high < prev_rsi_high:
                    signals.append(self._create_signal(
                        prices.index[price_highs[i]],
                        'bearish',
                        'regular',
                        curr_rsi_high,
                        curr_price_high,
                        [(prices.index[price_highs[i-1]], prev_price_high),
                         (prices.index[price_highs[i]], curr_price_high)]
                    ))
        
        return signals