import dask as da

from raptor.utils import BaseIndicator, _ema


class RSIIndicator(BaseIndicator):

    def __init__(self, close: da.Series, window: int = 14, fillna: bool = False):
        self._close = close
        self._window = window
        self._fillna = fillna
        self._run()

    def _run(self):
        diff = self._close.diff(1)
        up_direction = diff.where(diff > 0, 0.0)
        down_direction = -diff.where(diff < 0, 0.0)
        min_periods = 0 if self._fillna else self._window
        emaup = up_direction.ewm(
            alpha=1 / self._window, min_periods=min_periods, adjust=False
        ).mean()
        emadn = down_direction.ewm(
            alpha=1 / self._window, min_periods=min_periods, adjust=False
        ).mean()
        relative_strength = emaup / emadn
        self._rsi = da.Series(
            da.where(emadn == 0, 100, 100 - (100 / (1 + relative_strength))),
            index=self._close.index,
        )

    def rsi(self) -> da.Series:
        
        rsi_series = self._check_fillna(self._rsi, value=50)
        return da.Series(rsi_series, name="rsi")








class StochasticOscillator(BaseIndicator):
    

    def __init__(
        self,
        high: da.Series,
        low: da.Series,
        close: da.Series,
        window: int = 14,
        smooth_window: int = 3,
        fillna: bool = False,
    ):
        self._close = close
        self._high = high
        self._low = low
        self._window = window
        self._smooth_window = smooth_window
        self._fillna = fillna
        self._run()

    def _run(self):
        min_periods = 0 if self._fillna else self._window
        smin = self._low.rolling(self._window, min_periods=min_periods).min()
        smax = self._high.rolling(self._window, min_periods=min_periods).max()
        self._stoch_k = 100 * (self._close - smin) / (smax - smin)

    def stoch(self) -> da.Series:
        
        stoch_k = self._check_fillna(self._stoch_k, value=50)
        return da.Series(stoch_k, name="stoch_k")

    def stoch_signal(self) -> da.Series:
        
        min_periods = 0 if self._fillna else self._smooth_window
        stoch_d = self._stoch_k.rolling(
            self._smooth_window, min_periods=min_periods
        ).mean()
        stoch_d = self._check_fillna(stoch_d, value=50)
        return da.Series(stoch_d, name="stoch_k_signal")


class KAMAIndicator(BaseIndicator):
    

    def __init__(
        self,
        close: da.Series,
        window: int = 10,
        pow1: int = 2,
        pow2: int = 30,
        fillna: bool = False,
    ):
        self._close = close
        self._window = window
        self._pow1 = pow1
        self._pow2 = pow2
        self._fillna = fillna
        self._run()

    def _run(self):
        close_values = self._close.values
        vol = da.Series(abs(self._close - da.roll(self._close, 1)))

        min_periods = 0 if self._fillna else self._window
        er_num = abs(close_values - da.roll(close_values, self._window))
        er_den = vol.rolling(self._window, min_periods=min_periods).sum()
        efficiency_ratio = er_num / er_den

        smoothing_constant = (
            (
                efficiency_ratio * (2.0 / (self._pow1 + 1) - 2.0 / (self._pow2 + 1.0))
                + 2 / (self._pow2 + 1.0)
            )
            ** 2.0
        ).values

        self._kama = da.zeros(smoothing_constant.size)
        len_kama = len(self._kama)
        first_value = True

        for i in range(len_kama):
            if da.isnan(smoothing_constant[i]):
                self._kama[i] = da.nan
            elif first_value:
                self._kama[i] = close_values[i]
                first_value = False
            else:
                self._kama[i] = self._kama[i - 1] + smoothing_constant[i] * (
                    close_values[i] - self._kama[i - 1]
                )

    def kama(self) -> da.Series:
        
        kama_series = da.Series(self._kama, index=self._close.index)
        kama_series = self._check_fillna(kama_series, value=self._close)
        return da.Series(kama_series, name="kama")


class ROCIndicator(BaseIndicator):
    

    def __init__(self, close: da.Series, window: int = 12, fillna: bool = False):
        self._close = close
        self._window = window
        self._fillna = fillna
        self._run()

    def _run(self):
        self._roc = (
            (self._close - self._close.shift(self._window))
            / self._close.shift(self._window)
        ) * 100

    def roc(self) -> da.Series:
        roc_series = self._check_fillna(self._roc)
        return da.Series(roc_series, name="roc")




class StochRSIIndicator(BaseIndicator):

    def __init__(
        self,
        close: da.Series,
        window: int = 14,
        smooth1: int = 3,
        smooth2: int = 3,
        fillna: bool = False,
    ):
        self._close = close
        self._window = window
        self._smooth1 = smooth1
        self._smooth2 = smooth2
        self._fillna = fillna
        self._run()

    def _run(self):
        self._rsi = RSIIndicator(
            close=self._close, window=self._window, fillna=self._fillna
        ).rsi()
        lowest_low_rsi = self._rsi.rolling(self._window).min()
        self._stochrsi = (self._rsi - lowest_low_rsi) / (
            self._rsi.rolling(self._window).max() - lowest_low_rsi
        )
        self._stochrsi_k = self._stochrsi.rolling(self._smooth1).mean()

    def stochrsi(self):
        
        stochrsi_series = self._check_fillna(self._stochrsi)
        return da.Series(stochrsi_series, name="stochrsi")

    def stochrsi_k(self):
        
        stochrsi_k_series = self._check_fillna(self._stochrsi_k)
        return da.Series(stochrsi_k_series, name="stochrsi_k")

    def stochrsi_d(self):
        
        stochrsi_d_series = self._stochrsi_k.rolling(self._smooth2).mean()
        stochrsi_d_series = self._check_fillna(stochrsi_d_series)
        return da.Series(stochrsi_d_series, name="stochrsi_d")





def rsi(close, window=14, fillna=False) -> da.Series:
    return RSIIndicator(close=close, window=window, fillna=fillna).rsi()




def stoch(high, low, close, window=14, smooth_window=3, fillna=False) -> da.Series:

    return StochasticOscillator(
        high=high,
        low=low,
        close=close,
        window=window,
        smooth_window=smooth_window,
        fillna=fillna,
    ).stoch()


def stoch_signal(
    high, low, close, window=14, smooth_window=3, fillna=False
) -> da.Series:
    return StochasticOscillator(
        high=high,
        low=low,
        close=close,
        window=window,
        smooth_window=smooth_window,
        fillna=fillna,
    ).stoch_signal()





def roc(close: da.Series, window: int = 12, fillna: bool = False) -> da.Series:
    return ROCIndicator(close=close, window=window, fillna=fillna).roc()


def stochrsi(
    close: da.Series,
    window: int = 14,
    smooth1: int = 3,
    smooth2: int = 3,
    fillna: bool = False,
) -> da.Series:
    
    return StochRSIIndicator(
        close=close, window=window, smooth1=smooth1, smooth2=smooth2, fillna=fillna
    ).stochrsi()


def stochrsi_k(
    close: da.Series,
    window: int = 14,
    smooth1: int = 3,
    smooth2: int = 3,
    fillna: bool = False,
) -> da.Series:
    
    return StochRSIIndicator(
        close=close, window=window, smooth1=smooth1, smooth2=smooth2, fillna=fillna
    ).stochrsi_k()


def stochrsi_d(
    close: da.Series,
    window: int = 14,
    smooth1: int = 3,
    smooth2: int = 3,
    fillna: bool = False,
) -> da.Series:
    return StochRSIIndicator(
        close=close, window=window, smooth1=smooth1, smooth2=smooth2, fillna=fillna
    ).stochrsi_d()

















import dask as da

from raptor.utils import BaseIndicator, _ema


class RSIIndicator(BaseIndicator):

    def __init__(self, close: da.Series, window: int = 14, fillna: bool = False):
        self._close = close
        self._window = window
        self._fillna = fillna
        self._run()

    def _run(self):
        diff = self._close.diff(1)
        up_direction = diff.where(diff > 0, 0.0)
        down_direction = -diff.where(diff < 0, 0.0)
        min_periods = 0 if self._fillna else self._window
        emaup = up_direction.ewm(
            alpha=1 / self._window, min_periods=min_periods, adjust=False
        ).mean()
        emadn = down_direction.ewm(
            alpha=1 / self._window, min_periods=min_periods, adjust=False
        ).mean()
        relative_strength = emaup / emadn
        self._rsi = da.Series(
            da.where(emadn == 0, 100, 100 - (100 / (1 + relative_strength))),
            index=self._close.index,
        )

    def rsi(self) -> da.Series:
        
        rsi_series = self._check_fillna(self._rsi, value=50)
        return da.Series(rsi_series, name="rsi")








class StochasticOscillator(BaseIndicator):
    

    def __init__(
        self,
        high: da.Series,
        low: da.Series,
        close: da.Series,
        window: int = 14,
        smooth_window: int = 3,
        fillna: bool = False,
    ):
        self._close = close
        self._high = high
        self._low = low
        self._window = window
        self._smooth_window = smooth_window
        self._fillna = fillna
        self._run()

    def _run(self):
        min_periods = 0 if self._fillna else self._window
        smin = self._low.rolling(self._window, min_periods=min_periods).min()
        smax = self._high.rolling(self._window, min_periods=min_periods).max()
        self._stoch_k = 100 * (self._close - smin) / (smax - smin)

    def stoch(self) -> da.Series:
        
        stoch_k = self._check_fillna(self._stoch_k, value=50)
        return da.Series(stoch_k, name="stoch_k")

    def stoch_signal(self) -> da.Series:
        
        min_periods = 0 if self._fillna else self._smooth_window
        stoch_d = self._stoch_k.rolling(
            self._smooth_window, min_periods=min_periods
        ).mean()
        stoch_d = self._check_fillna(stoch_d, value=50)
        return da.Series(stoch_d, name="stoch_k_signal")


class KAMAIndicator(BaseIndicator):
    

    def __init__(
        self,
        close: da.Series,
        window: int = 10,
        pow1: int = 2,
        pow2: int = 30,
        fillna: bool = False,
    ):
        self._close = close
        self._window = window
        self._pow1 = pow1
        self._pow2 = pow2
        self._fillna = fillna
        self._run()

    def _run(self):
        close_values = self._close.values
        vol = da.Series(abs(self._close - da.roll(self._close, 1)))

        min_periods = 0 if self._fillna else self._window
        er_num = abs(close_values - da.roll(close_values, self._window))
        er_den = vol.rolling(self._window, min_periods=min_periods).sum()
        efficiency_ratio = er_num / er_den

        smoothing_constant = (
            (
                efficiency_ratio * (2.0 / (self._pow1 + 1) - 2.0 / (self._pow2 + 1.0))
                + 2 / (self._pow2 + 1.0)
            )
            ** 2.0
        ).values

        self._kama = da.zeros(smoothing_constant.size)
        len_kama = len(self._kama)
        first_value = True

        for i in range(len_kama):
            if da.isnan(smoothing_constant[i]):
                self._kama[i] = da.nan
            elif first_value:
                self._kama[i] = close_values[i]
                first_value = False
            else:
                self._kama[i] = self._kama[i - 1] + smoothing_constant[i] * (
                    close_values[i] - self._kama[i - 1]
                )

    def kama(self) -> da.Series:
        
        kama_series = da.Series(self._kama, index=self._close.index)
        kama_series = self._check_fillna(kama_series, value=self._close)
        return da.Series(kama_series, name="kama")


class ROCIndicator(BaseIndicator):
    

    def __init__(self, close: da.Series, window: int = 12, fillna: bool = False):
        self._close = close
        self._window = window
        self._fillna = fillna
        self._run()

    def _run(self):
        self._roc = (
            (self._close - self._close.shift(self._window))
            / self._close.shift(self._window)
        ) * 100

    def roc(self) -> da.Series:
        roc_series = self._check_fillna(self._roc)
        return da.Series(roc_series, name="roc")




class StochRSIIndicator(BaseIndicator):

    def __init__(
        self,
        close: da.Series,
        window: int = 14,
        smooth1: int = 3,
        smooth2: int = 3,
        fillna: bool = False,
    ):
        self._close = close
        self._window = window
        self._smooth1 = smooth1
        self._smooth2 = smooth2
        self._fillna = fillna
        self._run()

    def _run(self):
        self._rsi = RSIIndicator(
            close=self._close, window=self._window, fillna=self._fillna
        ).rsi()
        lowest_low_rsi = self._rsi.rolling(self._window).min()
        self._stochrsi = (self._rsi - lowest_low_rsi) / (
            self._rsi.rolling(self._window).max() - lowest_low_rsi
        )
        self._stochrsi_k = self._stochrsi.rolling(self._smooth1).mean()

    def stochrsi(self):
        
        stochrsi_series = self._check_fillna(self._stochrsi)
        return da.Series(stochrsi_series, name="stochrsi")

    def stochrsi_k(self):
        
        stochrsi_k_series = self._check_fillna(self._stochrsi_k)
        return da.Series(stochrsi_k_series, name="stochrsi_k")

    def stochrsi_d(self):
        
        stochrsi_d_series = self._stochrsi_k.rolling(self._smooth2).mean()
        stochrsi_d_series = self._check_fillna(stochrsi_d_series)
        return da.Series(stochrsi_d_series, name="stochrsi_d")





def rsi(close, window=14, fillna=False) -> da.Series:
    return RSIIndicator(close=close, window=window, fillna=fillna).rsi()




def stoch(high, low, close, window=14, smooth_window=3, fillna=False) -> da.Series:

    return StochasticOscillator(
        high=high,
        low=low,
        close=close,
        window=window,
        smooth_window=smooth_window,
        fillna=fillna,
    ).stoch()


def stoch_signal(
    high, low, close, window=14, smooth_window=3, fillna=False
) -> da.Series:
    return StochasticOscillator(
        high=high,
        low=low,
        close=close,
        window=window,
        smooth_window=smooth_window,
        fillna=fillna,
    ).stoch_signal()





def roc(close: da.Series, window: int = 12, fillna: bool = False) -> da.Series:
    return ROCIndicator(close=close, window=window, fillna=fillna).roc()


def stochrsi(
    close: da.Series,
    window: int = 14,
    smooth1: int = 3,
    smooth2: int = 3,
    fillna: bool = False,
) -> da.Series:
    
    return StochRSIIndicator(
        close=close, window=window, smooth1=smooth1, smooth2=smooth2, fillna=fillna
    ).stochrsi()


def stochrsi_k(
    close: da.Series,
    window: int = 14,
    smooth1: int = 3,
    smooth2: int = 3,
    fillna: bool = False,
) -> da.Series:
    
    return StochRSIIndicator(
        close=close, window=window, smooth1=smooth1, smooth2=smooth2, fillna=fillna
    ).stochrsi_k()


def stochrsi_d(
    close: da.Series,
    window: int = 14,
    smooth1: int = 3,
    smooth2: int = 3,
    fillna: bool = False,
) -> da.Series:
    return StochRSIIndicator(
        close=close, window=window, smooth1=smooth1, smooth2=smooth2, fillna=fillna
    ).stochrsi_d()
















