decay

class ConditionalDecay(OneToOneObservationRelatedTransformer):
    def __init__(self, alpha: float, theta: int = 0, n: int = 1, L: int = 13) -> None:
        """Conditional Decay, using formula from the Adstock:
        https://storage.googleapis.com/pub-tools-public-publication-data/pdf/6995d00a49a332b63e924dbcd42b37782d4ff498.pdf

        Args:
            - alpha (float): Inverse of the retention rate.
            - theta (float): Delay of peak effect (dependant of your TS freq). Defaults to 0.
            - n (int, optional): Power. Defaults to 1.
            - L (int): Maximum effect of carry effect. Defaults to 13.
        """
        self.alpha = alpha
        self.L = L
        self.theta = theta
        self.n = n

    @property
    def parameters(self) -> Dict[str, float]:
        return self.__dict__

    @property
    def is_linear(self) -> bool:
        return True

    # ------- METHODS -------

    def indexed_cumulative_sum(self, serie: np.ndarray) -> np.ndarray:
        """Return for each data point the total impact of the point from the conditional decay transformation,
        its value plus the adstocked effect in the finite future.
        It represents the total sum of the sequence defined by the conditional decay transformation."""

        def decay_weights(alpha, theta, n, L):
            def weight(lag, alpha, theta, n):
                return (1 - alpha) ** ((lag - theta) ** n)

            return [weight(lag, alpha, theta, n) for lag in range(L + 1)]

        return serie * sum(decay_weights(self.alpha, self.theta, self.n, self.L))

    # ------- TRANSFORMERS -------

    @staticmethod
    def _transformer(serie: np.ndarray, alpha, theta, n, L) -> np.ndarray:  # type: ignore
        serie = np.array(serie, dtype="float")

        def decay_weights(alpha, theta, n, L):
            def weight(lag, alpha, theta, n):
                return (1 - alpha) ** ((lag - theta) ** n)

            return [weight(lag, alpha, theta, n) for lag in range(L + 1)]

        decay_weights = decay_weights(alpha, theta, n, L)

        # Handle 2D arrays by applying convolution to each column
        if serie.ndim == 2:
            new_serie = np.column_stack(
                [np.convolve(serie[:, i], decay_weights)[: len(serie)] for i in range(serie.shape[1])]
            )
        else:
            new_serie = np.convolve(serie, decay_weights)[: len(serie)]

        return np.nan_to_num(new_serie)

    # ------- CHECKERS -------

    def check_params(self, serie: np.ndarray):
        """Check if parameters respect their application scope."""
        if self.alpha <= 0 or self.alpha > 1:
            raise ParameterScopeException(f"Values range of parameter alpha are between ]0, 1], not {self.alpha}.")
        if self.theta < 0:
            raise ParameterScopeException(f"Parameter theta should be positive, not {self.theta}.")
        if self.n <= 0:
            raise ParameterScopeException(f"Parameter n should be strictly positive, not {self.n}.")
        if self.L <= 0:
            raise ParameterScopeException(f"Parameter L should be strictly positive, not {self.L}.")

class Decay(OneToOneObservationRelatedTransformer):
    """Transform features by applying a decay to your variable (decay = 100% - adstock).

    Args:
        decay (float): Percentage such as 100%-decay (=adstock) is the percentage of the impact from an activity in a
                       certain datapoint (timestep e.g. week in a time serie) that is designated to the following
                       datapoints (timesteps e.g. weeks in a time serie).
    """

    def __init__(self, decay: float) -> None:
        self.decay = decay

    @property
    def parameters(self) -> Dict[str, float]:
        return self.__dict__

    @property
    def is_linear(self) -> bool:
        return True

    # ------- METHODS -------
    def indexed_cumulative_sum(self, serie: np.ndarray) -> np.ndarray:
        """Return for each data point the total impact of the point from the decay transformation,
        its value plus the adstocked effect in the infinite future.
        It represents the total sum of the geometric sequence defined by the decay transformation."""
        return serie * (1 / (self.decay / 100))

    # ------- TRANSFORMERS -------

    @staticmethod
    def _transformer(serie: np.ndarray, decay: float) -> np.ndarray:  # type: ignore
        if abs(decay - 1) < FLOAT_TOLERANCE_THRESHOLD:
            return serie
        serie = np.array(serie, dtype="float")
        normalized_decay = decay / 100
        for i in range(1, len(serie)):
            serie[i] += serie[i - 1] * (1 - normalized_decay)
        return np.nan_to_num(serie)

    # ------- CHECKERS -------

    def check_params(self, serie: np.ndarray):
        """Check if parameters respect their application scope."""
        if self.decay <= 0 or self.decay > 100:
            raise ParameterScopeException(f"Values range of parameter decay are between ]0, 100], not {self.decay}.")
        if self.decay <= 1:
            logger.warning("Values range of parameter decay are between ]0, 100] and not between ]0, 1].")