adstock

class Adstock(OneToOneObservationRelatedTransformer):
    """Transform features by reflecting the lasting impact of your variable.

    Args:
        adstock (float): 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).
            Expected value between 1 and 100.
    """

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

    @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 adstock transformation,
        its value plus the adstocked effect in the infinite future.
        It represents the total sum of the geometric sequence defined by the adstock transformation."""
        return serie * (1 / (1 - (self.adstock / 100)))

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

    @staticmethod
    def _transformer(serie: np.ndarray, adstock: float) -> np.ndarray:  # type: ignore
        if abs(adstock) < FLOAT_TOLERANCE_THRESHOLD:
            return serie
        serie = np.array(serie, dtype=float)
        normalized_adstock = adstock / 100

        return lfilter([1], [1, -normalized_adstock], serie)

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

    def check_params(self, serie: np.ndarray):
        """Check if parameters respect their application scope."""
        if self.adstock < 0 or self.adstock >= 100:
            raise ParameterScopeException(
                f"Values range of parameter adstock are between [0, 100], not {self.adstock}."
            )
        if 0 < self.adstock <= 1:
            logger.warning(
                f"Be aware adstock values should range between [0, 100], check adstock value: {self.adstock} "
                "is correctly selected."
            )

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

        Args:
            - alpha (float): 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 adstock transformation,
        its value plus the adstocked effect in the finite future.
        It represents the total sum of the sequence defined by the conditional adstock transformation."""

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

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

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

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

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

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

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

        adstock_weights = adstock_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], adstock_weights)[: len(serie)] for i in range(serie.shape[1])]
            )
        else:
            new_serie = np.convolve(serie, adstock_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}.")