weibull

class Weibull(OneToOneInversableDerivableTransformer):
    """Transform series by applying a Weibull saturation function.

    This transformation is typically applied after an adstock transformation
    to model saturation effects in response curves.

    The Weibull formula is adapted to account for scaling and shift parameters,
    allowing additional flexibility in the response curve.

    ```math
        volume_1 + volume_2 \\times \\exp\\left(-\\left(driver_1 \\times serie + driver_2\\right)^{symmetry}\\right)
    ```

    Args:
        volume_1 (float):
            Baseline level of the normalized curve.

        volume_2 (float, optional):
            Amplitude of the exponential component controlling the saturation range.
            Defaults to -volume_1.

        driver_1 (float):
            Scaling coefficient applied to the input series before the Weibull
            exponent. Controls how quickly the response saturates with respect
            to the input. Must be non-zero for invertibility.

        driver_2 (float, optional):
            Horizontal shift applied to the input before exponentiation.
            Defaults to 0.

        symmetry (float):
            Weibull shape parameter controlling the curvature of the response
            curve. Must be strictly positive.
    """

    def __init__(
        self,
        volume_1: float,
        volume_2: Optional[float] = None,
        driver_1: float = 1.0,
        driver_2: Optional[float] = None,
        symmetry: float = 1.0,
    ) -> None:
        self.volume_1 = volume_1
        self.volume_2 = -volume_1 if volume_2 is None else volume_2
        self.driver_1 = driver_1
        self.driver_2 = 0.0 if driver_2 is None else driver_2
        self.symmetry = symmetry

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

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

    def fit(self, serie: np.ndarray, y: Optional[np.ndarray] = None):
        return super().fit(serie, y)

    def transform(self, serie: np.ndarray, copy: bool = False) -> np.ndarray:
        serie = super().transform(serie, copy)

        return self._transformer(serie, self.volume_1, self.volume_2, self.driver_1, self.driver_2, self.symmetry)

    def inverse_transform(self, serie: np.ndarray, copy: bool = False) -> np.ndarray:
        serie = super().inverse_transform(serie, copy)

        return self._inverse_transformer(
            serie, self.volume_1, self.volume_2, self.driver_1, self.driver_2, self.symmetry
        )

    def derivative_transform(self, serie: np.ndarray, copy: bool = False) -> np.ndarray:
        serie = super().derivative_transform(serie, copy)

        return self._derivative_transformer(serie, self.volume_2, self.driver_1, self.driver_2, self.symmetry)

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

    @staticmethod
    def _transformer(
        serie: np.ndarray, volume_1: float, volume_2: float, driver_1: float, driver_2: float, symmetry: float
    ) -> np.ndarray:
        base = driver_1 * serie + driver_2

        with np.errstate(over="ignore", invalid="ignore"):
            transformed_serie = volume_1 + volume_2 * np.exp(-(base**symmetry))

        return np.nan_to_num(transformed_serie)

    @staticmethod
    def _inverse_transformer(
        serie: np.ndarray, volume_1: float, volume_2: float, driver_1: float, driver_2: float, symmetry: float
    ) -> np.ndarray:
        ratio = (serie - volume_1) / volume_2

        if np.any(~np.isfinite(ratio)) or np.any((ratio <= 0) | (ratio > 1)):
            raise ParameterScopeException(
                "Invalid value for inverse transform: (serie - volume_1) / volume_2 must be within the interval (0, 1]."
            )

        with np.errstate(divide="raise", invalid="raise"):
            inverse_serie = ((-np.log(ratio)) ** (1 / symmetry) - driver_2) / driver_1

        return inverse_serie

    @staticmethod
    def _derivative_transformer(
        serie: np.ndarray, volume_2: float, driver_1: float, driver_2: float, symmetry: float
    ) -> np.ndarray:
        base = driver_1 * serie + driver_2

        with np.errstate(over="ignore", invalid="ignore"):
            transformed_serie = -volume_2 * np.exp(-(base**symmetry)) * symmetry * driver_1 * (base ** (symmetry - 1))

        return np.nan_to_num(transformed_serie)

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

    def check_params(self, serie: np.ndarray) -> None:
        """Check if parameters respect their application scope."""
        if self.driver_1 == 0:
            raise ParameterScopeException("Parameter driver_1 must be non-zero for invertibility reasons.")

        if self.symmetry <= 0:
            raise ParameterScopeException(f"Parameter symmetry must be strictly positive, not {self.symmetry}.")

        if self.volume_2 == 0:
            raise ParameterScopeException("Parameter volume_2 must be non-zero for invertibility reasons.")

        if np.any(self.driver_1 * serie + self.driver_2 < 0):
            raise ParameterScopeException("Parameter combination must ensure driver_1 * serie + driver_2 >= 0.")

class WeibullUnscale(Weibull):
    """Transform series by applying the Weibull function multiplied by the max parameter.

    ```math
        max \\times \\left(
            volume_1 + volume_2 \\times \\exp\\left(-\\left(driver_1 \\times serie + driver_2\\right)^{symmetry}\\right)
        \\right)
    ```

    Args:
        volume_1 (float):
            Baseline level of the normalized curve.

        volume_2 (float, optional):
            Amplitude of the exponential component controlling the saturation range.
            Defaults to -volume_1.

        driver_1 (float):
            Scaling coefficient applied to the input series before the Weibull
            exponent. Controls how quickly the response saturates with respect
            to the input. Must be non-zero for invertibility.

        driver_2 (float, optional):
            Horizontal shift applied to the input before exponentiation.
            Defaults to 0.

        symmetry (float):
            Weibull shape parameter controlling the curvature of the response
            curve. Must be strictly positive.

        max_training (float, optional):
            Symbolic maximum applied to the output of the transformation.
            Defaults to None and is automatically set during fitting to the maximum
            value of the input training series.
    """

    def __init__(
        self,
        volume_1: float,
        volume_2: Optional[float] = None,
        driver_1: float = 1.0,
        driver_2: Optional[float] = None,
        symmetry: float = 1.0,
        max_training: Optional[float] = None,
    ) -> None:
        super().__init__(volume_1, volume_2, driver_1, driver_2, symmetry)
        self.max_training = max_training

    def fit(self, serie: np.ndarray, y: Optional[np.ndarray] = None) -> None:
        if self.max_training is None:
            if np.any(serie[serie > 0]):
                self.max_training = serie.max()
            else:
                self.max_training = serie.min() or 1

        return super().fit(serie, y)

    @staticmethod
    def _transformer_unscale(
        serie: np.ndarray,
        volume_1: float,
        volume_2: float,
        driver_1: float,
        driver_2: float,
        symmetry: float,
        max_training: float,
    ) -> np.ndarray:
        return max_training * Weibull._transformer(
            serie=serie,
            volume_1=volume_1,
            volume_2=volume_2,
            driver_1=driver_1,
            driver_2=driver_2,
            symmetry=symmetry,
        )

    @staticmethod
    def _inverse_transformer_unscale(
        serie: np.ndarray,
        volume_1: float,
        volume_2: float,
        driver_1: float,
        driver_2: float,
        symmetry: float,
        max_training: float,
    ) -> np.ndarray:
        scaled_serie = serie / max_training
        return Weibull._inverse_transformer(
            serie=scaled_serie,
            volume_1=volume_1,
            volume_2=volume_2,
            driver_1=driver_1,
            driver_2=driver_2,
            symmetry=symmetry,
        )

    @staticmethod
    def _derivative_transformer_unscale(
        serie: np.ndarray, volume_2: float, driver_1: float, driver_2: float, symmetry: float, max_training: float
    ) -> np.ndarray:
        return max_training * Weibull._derivative_transformer(
            serie=serie,
            volume_2=volume_2,
            driver_1=driver_1,
            driver_2=driver_2,
            symmetry=symmetry,
        )

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

    def transform(self, serie: np.ndarray, copy: bool = False) -> np.ndarray:
        if self.max_training is None:
            raise ValueError("max_training is not set. Call fit() before transform().")
        serie = super(Weibull, self).transform(serie, copy)
        return self._transformer_unscale(
            serie, self.volume_1, self.volume_2, self.driver_1, self.driver_2, self.symmetry, self.max_training
        )

    def inverse_transform(self, serie: np.ndarray, copy: bool = False) -> np.ndarray:
        if self.max_training is None:
            raise ValueError("max_training is not set. Call fit() before inverse_transform().")
        serie = super(Weibull, self).inverse_transform(serie, copy)
        return self._inverse_transformer_unscale(
            serie, self.volume_1, self.volume_2, self.driver_1, self.driver_2, self.symmetry, self.max_training
        )

    def derivative_transform(self, serie: np.ndarray, copy: bool = False) -> np.ndarray:
        if self.max_training is None:
            raise ValueError("max_training is not set. Call fit() before derivative_transform().")
        serie = super(Weibull, self).derivative_transform(serie, copy)
        return self._derivative_transformer_unscale(
            serie, self.volume_2, self.driver_1, self.driver_2, self.symmetry, self.max_training
        )

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

    def check_params(self, serie: np.ndarray) -> None:
        """Check if parameters respect their application scope."""
        super().check_params(serie)

        if self.max_training is not None and self.max_training <= 0:
            raise ParameterScopeException(f"Parameter max must be strictly positive, not {self.max_training}.")