scaled_conditional_adstock

class ScaledConditionalAdstock(OneToOneObservationRelatedTransformer):
    """Transform features by reflecting the lasting impact of your variable, scaling the impact by the observed values.

    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 0 and 1.
        adstock_effect_length (int): Maximum window of time to consider for the adstock calculation.
        decay_method (Literal["geometric", "binomial"]): Method to use for the decay.
            Defaults to "geometric".
            - "geometric": The decay is calculated using the geometric series.
            - "binomial": The decay is calculated using the binomial distribution.

        Example:
        >>> serie = np.array([10, 0, 10, 30])
        >>> Example with geometric decay method:
        >>> transformer = ScaledConditionalAdstock(adstock=0.5, adstock_effect_length=3, decay_method="geometric")
        >>> the window_size is adstock_effect_length + 1 = 4
        >>> the weights are [0.5**3, 0.5**2, 0.5**1, 0.5**0] = [0.125, 0.25, 0.5, 1]
        >>> the weights are normalized to have a total sum of 1 -> [0.06666667, 0.13333333, 0.26666667, 0.53333333]
        >>> Elements are backfilled with 0 to match window_size length, so for first time period : [0, 0, 0, 10]
        >>> the result is [0*0.066667 + 0*0.133333 + 0*0.266667 + 10*0.533333] = 5.333333 for the first time period
        >>> the result is [0*0.066667 + 0*0.133333 + 10*0.266667 + 0*0.533333] = 2.666667 for the second time period
        >>> the result is [0*0.066667 + 10*0.133333 + 0*0.266667 + 10*0.533333] = 6.666667 for the third time period
        >>> the result is [10*0.066667 + 0*0.133333 + 10*0.266667 + 30*0.533333] = 19.333333 for the fourth time period
        >>> transformer.fit_transform(serie)
        array([5.333333, 2.666667, 6.666667, 19.333333])
        >>> Example with binomial decay method:
        >>> transformer = ScaledConditionalAdstock(adstock=0.5, adstock_effect_length=3, decay_method="binomial")
        >>> the window_size is max_lag + 1 = 4
        >>> the mapped_alpha_binomial is (1 / 0.5) - 1 = 1
        >>> the weights are 1 - [3/4, 2/4, 1/4, 0/4] ** 1 = [0.25, 0.5, 0.75, 1]
        >>> the weights are normalized to have a total sum of 1 -> [0.1, 0.2, 0.3, 0.4]
        >>> Elements are backfilled with 0 to match window_size length, so for first time period : [0, 0, 0, 10]
        >>> the result is [0*0.1 + 0*0.2 + 0*0.3 + 10*0.4] = 4 for the first time period
        >>> the result is [0*0.1 + 0*0.2 + 10*0.3 + 0*0.4] = 3 for the second time period
        >>> the result is [0*0.1 + 10*0.2 + 0*0.3 + 10*0.4] = 6 for the third time period
        >>> the result is [10*0.1 + 0*0.2 + 10*0.3 + 30*0.4] = 16 for the fourth time period
        >>> transformer.fit_transform(serie)
        array([4.0, 3.0, 6.0, 16.0])
    """

    def __init__(
        self,
        adstock: float,
        adstock_effect_length: int,
        decay_method: Literal["geometric", "binomial"] = "geometric",
    ) -> None:
        self.adstock = adstock
        self.adstock_effect_length = adstock_effect_length
        self.decay_method = decay_method

    @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 sequence defined by the adstock transformation."""
        return serie

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

    @staticmethod
    def _transformer(  # type: ignore
        serie: np.ndarray,
        adstock: float,
        adstock_effect_length: int,
        decay_method: Literal["geometric", "binomial"] = "geometric",
    ) -> np.ndarray:
        """For each datapoint, we consider the adstock_effect_length previous datapoints to compute the adstocked effect.
        The weights are calculated in order to have the higher impact for the most recent datapoints.
        The weights are then normalized to have a total sum of 1."""
        input_array = np.array(serie, dtype=float)
        input_length = input_array.size
        output_length = input_length
        window_length = min(adstock_effect_length + 1, input_length)
        required_length = output_length + window_length - 1

        if input_length > required_length:
            padded_array = input_array[-required_length:]
        elif input_length < required_length:
            padding = np.zeros(required_length - input_length, dtype=float)
            padded_array = np.concatenate([padding, input_array])
        else:
            padded_array = input_array

        l_range = np.arange(window_length, dtype=float)[::-1]
        if decay_method == "geometric":
            decay_weights = adstock**l_range
        elif decay_method == "binomial":
            mapped_alpha_binomial = (1 / adstock) - 1 if adstock != 0 else np.inf
            decay_weights = (1 - l_range / window_length) ** mapped_alpha_binomial

        weight_sum = np.sum(decay_weights)
        normalized_weights = decay_weights / weight_sum

        result = np.empty(output_length, dtype=float)
        for t in range(output_length):
            result[t] = np.dot(padded_array[t : t + window_length], normalized_weights)

        return np.nan_to_num(result)

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

    def check_params(self, serie: np.ndarray):
        """Check if parameters respect their application scope."""
        if self.adstock_effect_length < 0:
            raise ParameterScopeException(f"Parameter max_lag should be positive, not {self.adstock_effect_length}.")
        if self.decay_method not in ["geometric", "binomial"]:
            raise ParameterScopeException(
                f"Parameter decay_method should be 'geometric' or 'binomial', not {self.decay_method}."
            )
        if (1 < self.adstock) or (self.adstock < 0):
            raise ParameterScopeException(
                f"Be aware adstock values should range between [0, 1], check adstock value: {self.adstock} "
                "is correctly selected."
            )