Builtin Exogenous Effects

Effects that define relationships between variables and the target.

BaseEffect

Bases: BaseObject

Base class for effects.

Effects are objects which are responsible for preparing the data and applying a specific effect to the forecast. During preparation of the data (which happens in transform method), the effect receives the exogenous variables dataframe and can use them to prepare the jax arrays that will be used at inference time. During inference time, the predict method is called, and it should output a new component to the additive model of Prophetverse.

Remember that Prophetverse's models are Generalized Additive Models, which are composed of many terms summed together to form the final forecast. Each term is represented by an effect.

Children classes should implement the following methods:

• _fit (optional): This method is called during fit() of the forecasting and should be used to initialize any necessary parameters or data structures. It receives the exogenous variables dataframe X, the series y, and the scale factor scale that was used to scale the timeseries.

• _transform (optional): This method receives the exogenous variables dataframe, and should return an object containing the data needed for the effect. This object will be passed to the predict method as data. By default the columns of the dataframe that match the regex pattern are selected, and the result is converted to a jnp.ndarray.

• _predict (mandatory): This method receives the output of _transform and all previously computed effects. It should return the effect values as a jnp.ndarray

Parameters:

Name	Type	Description	Default
id	str	The id of the effect, by default "". Used to identify the effect in the model.	required
regex	Optional[str]	A regex pattern to match the columns of the exogenous variables dataframe, by default None. If None, and _tags["skip_predict_if_no_match"] is True, the effect will be skipped if no columns are found.	required
effect_mode	EFFECT_APPLICATION_TYPE	The mode of the effect, either "additive" or "multiplicative", by default "multiplicative". If "multiplicative", the effect multiplies the trend values before returning them.	required

Attributes:

Name	Type	Description
input_feature_column_names	List[str]	The names of the input feature columns. This is set during the fit() method.
should_skip_predict	bool	If True, the effect should be skipped during prediction. This is determined by the skip_predict_if_no_match tag and the presence of input feature columns names. If the tag is True and there are no input feature columns names, the effect should be skipped during prediction.

Source code in src/prophetverse/effects/base.py

class BaseEffect(BaseObject):
    """Base class for effects.

    Effects are objects which are responsible for preparing the data and applying
    a specific effect to the forecast. During preparation of the data (which happens in
    `transform` method), the effect receives the exogenous variables dataframe and can
    use them to prepare the jax arrays that will be used at inference time. During
    inference time, the `predict` method is called, and it should output a new component
    to the additive model of Prophetverse.

    Remember that Prophetverse's models are Generalized Additive Models, which are
    composed of many terms summed together to form the final forecast. Each term is
    represented by an effect.

    Children classes should implement the following methods:


    * `_fit` (optional): This method is called during fit() of the forecasting  and
    should be used to initialize any necessary parameters or data structures.
    It receives the exogenous variables dataframe X, the series `y`, and the scale
    factor `scale` that was used to scale the timeseries.

    * `_transform` (optional): This method receives the exogenous variables
    dataframe, and should return an object containing the data needed for the
    effect. This object will be passed to the predict method as `data`. By default
    the columns of the dataframe that match the regex pattern are selected, and the
    result is converted to a `jnp.ndarray`.

    * `_predict` (mandatory): This method receives the output of `_transform` and
    all previously computed effects. It should return the effect values as a
    `jnp.ndarray`


    Parameters
    ----------
    id : str, optional
        The id of the effect, by default "". Used to identify the effect in the model.
    regex : Optional[str], optional
        A regex pattern to match the columns of the exogenous variables dataframe,
        by default None. If None, and _tags["skip_predict_if_no_match"] is True, the
        effect will be skipped if no columns are found.
    effect_mode : EFFECT_APPLICATION_TYPE, optional
        The mode of the effect, either "additive" or "multiplicative", by default
        "multiplicative". If "multiplicative", the effect multiplies the trend values
        before returning them.


    Attributes
    ----------
    input_feature_column_names : List[str]
        The names of the input feature columns. This is set during the `fit()` method.
    should_skip_predict : bool
        If True, the effect should be skipped during prediction. This is determined by
        the `skip_predict_if_no_match` tag and the presence of input feature columns
        names. If the tag is True and there are no input feature columns names, the
        effect should be skipped during prediction.
    """

    _tags = {
        "supports_multivariate": False,
        # If no columns are found, should
        # _predict be skipped?
        "skip_predict_if_no_match": True,
        # Should only the indexes related to the forecasting horizon be passed to
        # _transform?
        "filter_indexes_with_forecating_horizon_at_transform": True,
    }

    def __init__(self):
        self._input_feature_column_names: List[str] = []
        self._is_fitted: bool = False

    @property
    def input_feature_column_names(self) -> List[str]:
        """Return the input feature columns names."""
        return self._input_feature_column_names

    @property
    def should_skip_predict(self) -> bool:
        """Return if the effect should be skipped by the forecaster.

        Returns
        -------
        bool
            If the effect should be skipped by the forecaster.
        """
        if not self._input_feature_column_names and self.get_tag(
            "skip_predict_if_no_match", True
        ):
            return True
        return False

    def fit(self, y: pd.DataFrame, X: pd.DataFrame, scale: float = 1.0):
        """Initialize the effect.

        This method is called during `fit()` of the forecasting model.
        It receives the Exogenous variables DataFrame and should be used to initialize
        any necessary parameters or data structures, such as detecting the columns that
        match the regex pattern.

        This method MUST set _input_feature_columns_names to a list of column names

        Parameters
        ----------
        y : pd.DataFrame
            The timeseries dataframe

        X : pd.DataFrame
            The DataFrame to initialize the effect.

        scale : float, optional
            The scale of the timeseries. For multivariate timeseries, this is
            a dataframe. For univariate, it is a simple float.

        Returns
        -------
        None

        Raises
        ------
        ValueError
            If the effect does not support multivariate data and the DataFrame has more
            than one level of index.
        """
        if not self.get_tag("supports_multivariate", False):
            if X is not None and X.index.nlevels > 1:
                raise ValueError(
                    f"The effect {self.__class__.__name__} does not "
                    + "support multivariate data"
                )

        if X is None or X.empty:
            self._input_feature_column_names = []
        else:
            self._input_feature_column_names = X.columns.tolist()

        self._fit(y=y, X=X, scale=scale)
        self._is_fitted = True

    def _fit(self, y: pd.DataFrame, X: pd.DataFrame, scale: float = 1.0):
        """Customize the initialization of the effect.

        This method is called by the `fit()` method and can be overridden by
        subclasses to provide additional initialization logic.

        Parameters
        ----------
        y : pd.DataFrame
            The timeseries dataframe

        X : pd.DataFrame
            The DataFrame to initialize the effect.

        scale : float, optional
            The scale of the timeseries. For multivariate timeseries, this is
            a dataframe. For univariate, it is a simple float.
        """
        pass

    def transform(
        self,
        X: pd.DataFrame,
        fh: pd.Index,
    ) -> Any:
        """Prepare input data to be passed to numpyro model.

        This method receives the Exogenous variables DataFrame and should return a
        the data needed for the effect. Those data will be passed to the `predict`
        method as `data` argument.

        Parameters
        ----------
        X : pd.DataFrame
            The input DataFrame containing the exogenous variables for the training
            time indexes, if passed during fit, or for the forecasting time indexes, if
            passed during predict.

        fh : pd.Index
            The forecasting horizon as a pandas Index.

        Returns
        -------
        Any
            Any object containing the data needed for the effect. The object will be
            passed to `predict` method as `data` argument.

        Raises
        ------
        ValueError
            If the effect has not been fitted.
        """
        if not self._is_fitted:
            raise ValueError("You must call fit() before calling this method")

        # If apply should be skipped, return an empty dictionary
        if self.should_skip_predict:
            return {}

        if self.get_tag("filter_indexes_with_forecating_horizon_at_transform", True):
            # Filter when index level -1 is in fh
            if X is not None:
                X = X.loc[X.index.get_level_values(-1).isin(fh)]

        X = X[self.input_feature_column_names]
        return self._transform(X, fh)

    def _transform(
        self,
        X: pd.DataFrame,
        fh: pd.Index,
    ) -> Any:
        """Prepare input data to be passed to numpyro model.

        This method receives the Exogenous variables DataFrame and should return a
        the data needed for the effect. Those data will be passed to the `predict`
        method as `data` argument.

        Parameters
        ----------
        X : pd.DataFrame
            The input DataFrame containing the exogenous variables for the training
            time indexes, if passed during fit, or for the forecasting time indexes, if
            passed during predict.

        fh : pd.Index
            The forecasting horizon as a pandas Index.

        Returns
        -------
        Any
            Any object containing the data needed for the effect. The object will be
            passed to `predict` method as `data` argument.
        """
        array = series_to_tensor_or_array(X)
        return array

    def predict(
        self,
        data: Dict,
        predicted_effects: Optional[Dict[str, jnp.ndarray]] = None,
    ) -> jnp.ndarray:
        """Apply and return the effect values.

        Parameters
        ----------
        data : Any
            Data obtained from the transformed method.

        predicted_effects : Dict[str, jnp.ndarray], optional
            A dictionary containing the predicted effects, by default None.

        Returns
        -------
        jnp.ndarray
            An array with shape (T,1) for univariate timeseries, or (N, T, 1) for
            multivariate timeseries, where T is the number of timepoints and N is the
            number of series.
        """
        if predicted_effects is None:
            predicted_effects = {}

        x = self._predict(data, predicted_effects)

        return x

    def _predict(
        self,
        data: Dict,
        predicted_effects: Dict[str, jnp.ndarray],
    ) -> jnp.ndarray:
        """Apply and return the effect values.

        Parameters
        ----------
        data : Any
            Data obtained from the transformed method.

        predicted_effects : Dict[str, jnp.ndarray], optional
            A dictionary containing the predicted effects, by default None.

        Returns
        -------
        jnp.ndarray
            An array with shape (T,1) for univariate timeseries, or (N, T, 1) for
            multivariate timeseries, where T is the number of timepoints and N is the
            number of series.
        """
        raise NotImplementedError("Subclasses must implement _predict()")

    def __call__(
        self, data: Dict, predicted_effects: Dict[str, jnp.ndarray]
    ) -> jnp.ndarray:
        """Run the processes to calculate effect as a function."""
        return self.predict(data=data, predicted_effects=predicted_effects)

input_feature_column_names: List[str] property

Return the input feature columns names.

should_skip_predict: bool property

Return if the effect should be skipped by the forecaster.

Returns:

Type	Description
bool	If the effect should be skipped by the forecaster.

__call__(data, predicted_effects)

Run the processes to calculate effect as a function.

Source code in src/prophetverse/effects/base.py

def __call__(
    self, data: Dict, predicted_effects: Dict[str, jnp.ndarray]
) -> jnp.ndarray:
    """Run the processes to calculate effect as a function."""
    return self.predict(data=data, predicted_effects=predicted_effects)

fit(y, X, scale=1.0)

Initialize the effect.

This method is called during fit() of the forecasting model. It receives the Exogenous variables DataFrame and should be used to initialize any necessary parameters or data structures, such as detecting the columns that match the regex pattern.

This method MUST set _input_feature_columns_names to a list of column names

Parameters:

Name	Type	Description	Default
y	DataFrame	The timeseries dataframe	required
X	DataFrame	The DataFrame to initialize the effect.	required
scale	float	The scale of the timeseries. For multivariate timeseries, this is a dataframe. For univariate, it is a simple float.	1.0

Returns:

Type	Description
None

Raises:

Type	Description
ValueError	If the effect does not support multivariate data and the DataFrame has more than one level of index.

Source code in src/prophetverse/effects/base.py

def fit(self, y: pd.DataFrame, X: pd.DataFrame, scale: float = 1.0):
    """Initialize the effect.

    This method is called during `fit()` of the forecasting model.
    It receives the Exogenous variables DataFrame and should be used to initialize
    any necessary parameters or data structures, such as detecting the columns that
    match the regex pattern.

    This method MUST set _input_feature_columns_names to a list of column names

    Parameters
    ----------
    y : pd.DataFrame
        The timeseries dataframe

    X : pd.DataFrame
        The DataFrame to initialize the effect.

    scale : float, optional
        The scale of the timeseries. For multivariate timeseries, this is
        a dataframe. For univariate, it is a simple float.

    Returns
    -------
    None

    Raises
    ------
    ValueError
        If the effect does not support multivariate data and the DataFrame has more
        than one level of index.
    """
    if not self.get_tag("supports_multivariate", False):
        if X is not None and X.index.nlevels > 1:
            raise ValueError(
                f"The effect {self.__class__.__name__} does not "
                + "support multivariate data"
            )

    if X is None or X.empty:
        self._input_feature_column_names = []
    else:
        self._input_feature_column_names = X.columns.tolist()

    self._fit(y=y, X=X, scale=scale)
    self._is_fitted = True

predict(data, predicted_effects=None)

Apply and return the effect values.

Parameters:

Name	Type	Description	Default
data	Any	Data obtained from the transformed method.	required
predicted_effects	Dict[str, ndarray]	A dictionary containing the predicted effects, by default None.	None

Returns:

Type	Description
ndarray	An array with shape (T,1) for univariate timeseries, or (N, T, 1) for multivariate timeseries, where T is the number of timepoints and N is the number of series.

Source code in src/prophetverse/effects/base.py

def predict(
    self,
    data: Dict,
    predicted_effects: Optional[Dict[str, jnp.ndarray]] = None,
) -> jnp.ndarray:
    """Apply and return the effect values.

    Parameters
    ----------
    data : Any
        Data obtained from the transformed method.

    predicted_effects : Dict[str, jnp.ndarray], optional
        A dictionary containing the predicted effects, by default None.

    Returns
    -------
    jnp.ndarray
        An array with shape (T,1) for univariate timeseries, or (N, T, 1) for
        multivariate timeseries, where T is the number of timepoints and N is the
        number of series.
    """
    if predicted_effects is None:
        predicted_effects = {}

    x = self._predict(data, predicted_effects)

    return x

transform(X, fh)

Prepare input data to be passed to numpyro model.

This method receives the Exogenous variables DataFrame and should return a the data needed for the effect. Those data will be passed to the predict method as data argument.

Parameters:

Name	Type	Description	Default
X	DataFrame	The input DataFrame containing the exogenous variables for the training time indexes, if passed during fit, or for the forecasting time indexes, if passed during predict.	required
fh	Index	The forecasting horizon as a pandas Index.	required

Returns:

Type	Description
Any	Any object containing the data needed for the effect. The object will be passed to predict method as data argument.

Raises:

Type	Description
ValueError	If the effect has not been fitted.

Source code in src/prophetverse/effects/base.py

def transform(
    self,
    X: pd.DataFrame,
    fh: pd.Index,
) -> Any:
    """Prepare input data to be passed to numpyro model.

    This method receives the Exogenous variables DataFrame and should return a
    the data needed for the effect. Those data will be passed to the `predict`
    method as `data` argument.

    Parameters
    ----------
    X : pd.DataFrame
        The input DataFrame containing the exogenous variables for the training
        time indexes, if passed during fit, or for the forecasting time indexes, if
        passed during predict.

    fh : pd.Index
        The forecasting horizon as a pandas Index.

    Returns
    -------
    Any
        Any object containing the data needed for the effect. The object will be
        passed to `predict` method as `data` argument.

    Raises
    ------
    ValueError
        If the effect has not been fitted.
    """
    if not self._is_fitted:
        raise ValueError("You must call fit() before calling this method")

    # If apply should be skipped, return an empty dictionary
    if self.should_skip_predict:
        return {}

    if self.get_tag("filter_indexes_with_forecating_horizon_at_transform", True):
        # Filter when index level -1 is in fh
        if X is not None:
            X = X.loc[X.index.get_level_values(-1).isin(fh)]

    X = X[self.input_feature_column_names]
    return self._transform(X, fh)

HillEffect

Bases: BaseAdditiveOrMultiplicativeEffect

Represents a Hill effect in a time series model.

Parameters:

Name	Type	Description	Default
half_max_prior	Distribution	Prior distribution for the half-maximum parameter	None
slope_prior	Distribution	Prior distribution for the slope parameter	None
max_effect_prior	Distribution	Prior distribution for the maximum effect parameter	None
effect_mode	effects_application	Mode of the effect (either "additive" or "multiplicative")	'multiplicative'

Source code in src/prophetverse/effects/hill.py

class HillEffect(BaseAdditiveOrMultiplicativeEffect):
    """Represents a Hill effect in a time series model.

    Parameters
    ----------
    half_max_prior : Distribution, optional
        Prior distribution for the half-maximum parameter
    slope_prior : Distribution, optional
        Prior distribution for the slope parameter
    max_effect_prior : Distribution, optional
        Prior distribution for the maximum effect parameter
    effect_mode : effects_application, optional
        Mode of the effect (either "additive" or "multiplicative")
    """

    def __init__(
        self,
        effect_mode: EFFECT_APPLICATION_TYPE = "multiplicative",
        half_max_prior: Optional[Distribution] = None,
        slope_prior: Optional[Distribution] = None,
        max_effect_prior: Optional[Distribution] = None,
    ):
        self.half_max_prior = half_max_prior or dist.Gamma(1, 1)
        self.slope_prior = slope_prior or dist.HalfNormal(10)
        self.max_effect_prior = max_effect_prior or dist.Gamma(1, 1)

        super().__init__(effect_mode=effect_mode)

    def _predict(
        self,
        data: Dict[str, jnp.ndarray],
        predicted_effects: Dict[str, jnp.ndarray],
    ) -> jnp.ndarray:
        """Apply and return the effect values.

        Parameters
        ----------
        data : Any
            Data obtained from the transformed method.

        predicted_effects : Dict[str, jnp.ndarray]
            A dictionary containing the predicted effects

        Returns
        -------
        jnp.ndarray
            An array with shape (T,1) for univariate timeseries.
        """
        half_max = numpyro.sample("half_max", self.half_max_prior)
        slope = numpyro.sample("slope", self.slope_prior)
        max_effect = numpyro.sample("max_effect", self.max_effect_prior)

        x = _exponent_safe(data / half_max, -slope)
        effect = max_effect / (1 + x)
        return effect

LiftExperimentLikelihood

Bases: BaseEffect

Wrap an effect and applies a normal likelihood to its output.

This class uses an input as a reference for the effect, and applies a normal likelihood to the output of the effect.

Parameters:

Name	Type	Description	Default
effect	BaseEffect	The effect to wrap.	required
lift_test_results	DataFrame	A dataframe with the lift test results. Should be in sktime format, and must have the same index as the input data.	required
prior_scale	float	The scale of the prior distribution for the likelihood.	required

Source code in src/prophetverse/effects/lift_experiment.py

class LiftExperimentLikelihood(BaseEffect):
    """Wrap an effect and applies a normal likelihood to its output.

    This class uses an input as a reference for the effect, and applies a normal
    likelihood to the output of the effect.

    Parameters
    ----------
    effect : BaseEffect
        The effect to wrap.
    lift_test_results : pd.DataFrame
        A dataframe with the lift test results. Should be in sktime format, and must
        have the same index as the input data.
    prior_scale : float
        The scale of the prior distribution for the likelihood.
    """

    _tags = {"skip_predict_if_no_match": False, "supports_multivariate": False}

    def __init__(
        self,
        effect: BaseEffect,
        lift_test_results: pd.DataFrame,
        prior_scale: float,
    ):

        self.effect = effect
        self.lift_test_results = lift_test_results
        self.prior_scale = prior_scale

        assert self.prior_scale > 0, "prior_scale must be greater than 0"

        super().__init__()

    def fit(self, y: pd.DataFrame, X: pd.DataFrame, scale: float = 1):
        """Initialize the effect.

        This method is called during `fit()` of the forecasting model.
        It receives the Exogenous variables DataFrame and should be used to initialize
        any necessary parameters or data structures, such as detecting the columns that
        match the regex pattern.

        This method MUST set _input_feature_columns_names to a list of column names

        Parameters
        ----------
        y : pd.DataFrame
            The timeseries dataframe

        X : pd.DataFrame
            The DataFrame to initialize the effect.

        scale : float, optional
            The scale of the timeseries. For multivariate timeseries, this is
            a dataframe. For univariate, it is a simple float.

        Returns
        -------
        None
        """
        self.effect.fit(X=X, y=y, scale=scale)
        self.timeseries_scale = scale
        super().fit(X=X, y=y, scale=scale)

    def _transform(self, X: pd.DataFrame, fh: pd.Index) -> Dict[str, Any]:
        """Prepare input data to be passed to numpyro model.

        Returns a dictionary with the data for the lift and for the inner effect.

        Parameters
        ----------
        X : pd.DataFrame
            The input DataFrame containing the exogenous variables for the training
            time indexes, if passed during fit, or for the forecasting time indexes, if
            passed during predict.

        fh : pd.Index
            The forecasting horizon as a pandas Index.

        Returns
        -------
        Dict[str, Any]
            Dictionary with data for the lift and for the inner effect
        """
        data_dict = {}
        data_dict["inner_effect_data"] = self.effect._transform(X, fh=fh)

        X_lift = self.lift_test_results.reindex(fh, fill_value=jnp.nan)
        lift_array = series_to_tensor_or_array(X_lift)
        data_dict["observed_lift"] = lift_array / self.timeseries_scale
        data_dict["obs_mask"] = ~jnp.isnan(data_dict["observed_lift"])

        return data_dict

    def _predict(
        self, data: Dict, predicted_effects: Dict[str, jnp.ndarray]
    ) -> jnp.ndarray:
        """Apply and return the effect values.

        Parameters
        ----------
        data : Any
            Data obtained from the transformed method.

        predicted_effects : Dict[str, jnp.ndarray], optional
            A dictionary containing the predicted effects, by default None.

        Returns
        -------
        jnp.ndarray
            An array with shape (T,1) for univariate timeseries.
        """
        observed_lift = data["observed_lift"]
        obs_mask = data["obs_mask"]

        x = self.effect.predict(
            data=data["inner_effect_data"], predicted_effects=predicted_effects
        )

        numpyro.sample(
            "lift_experiment",
            dist.Normal(x, self.prior_scale),
            obs=observed_lift,
            obs_mask=obs_mask,
        )

        return x

    @property
    def input_feature_column_names(self) -> List[str]:
        """Return the input feature columns names."""
        return self.effect._input_feature_column_names

input_feature_column_names: List[str] property

Return the input feature columns names.

fit(y, X, scale=1)

Initialize the effect.

This method is called during fit() of the forecasting model. It receives the Exogenous variables DataFrame and should be used to initialize any necessary parameters or data structures, such as detecting the columns that match the regex pattern.

This method MUST set _input_feature_columns_names to a list of column names

Parameters:

Name	Type	Description	Default
y	DataFrame	The timeseries dataframe	required
X	DataFrame	The DataFrame to initialize the effect.	required
scale	float	The scale of the timeseries. For multivariate timeseries, this is a dataframe. For univariate, it is a simple float.	1

Returns:

Type	Description
None

Source code in src/prophetverse/effects/lift_experiment.py

def fit(self, y: pd.DataFrame, X: pd.DataFrame, scale: float = 1):
    """Initialize the effect.

    This method is called during `fit()` of the forecasting model.
    It receives the Exogenous variables DataFrame and should be used to initialize
    any necessary parameters or data structures, such as detecting the columns that
    match the regex pattern.

    This method MUST set _input_feature_columns_names to a list of column names

    Parameters
    ----------
    y : pd.DataFrame
        The timeseries dataframe

    X : pd.DataFrame
        The DataFrame to initialize the effect.

    scale : float, optional
        The scale of the timeseries. For multivariate timeseries, this is
        a dataframe. For univariate, it is a simple float.

    Returns
    -------
    None
    """
    self.effect.fit(X=X, y=y, scale=scale)
    self.timeseries_scale = scale
    super().fit(X=X, y=y, scale=scale)

LinearEffect

Bases: BaseAdditiveOrMultiplicativeEffect

Represents a linear effect in a hierarchical prophet model.

Parameters:

Name	Type	Description	Default
prior	Distribution	A numpyro distribution to use as prior. Defaults to dist.Normal(0, 1)	None
effect_mode	effects_application	Either "multiplicative" or "additive" by default "multiplicative".	'multiplicative'

Source code in src/prophetverse/effects/linear.py

class LinearEffect(BaseAdditiveOrMultiplicativeEffect):
    """Represents a linear effect in a hierarchical prophet model.

    Parameters
    ----------
    prior : Distribution, optional
        A numpyro distribution to use as prior. Defaults to dist.Normal(0, 1)
    effect_mode : effects_application, optional
        Either "multiplicative" or "additive" by default "multiplicative".
    """

    _tags = {
        "supports_multivariate": True,
    }

    def __init__(
        self,
        effect_mode: EFFECT_APPLICATION_TYPE = "multiplicative",
        prior: Optional[Distribution] = None,
    ):
        self.prior = prior or dist.Normal(0, 0.1)

        super().__init__(effect_mode=effect_mode)

    def _predict(
        self,
        data: Any,
        predicted_effects: Optional[Dict[str, jnp.ndarray]] = None,
    ) -> jnp.ndarray:
        """Apply and return the effect values.

        Parameters
        ----------
        data : Any
            Data obtained from the transformed method.

        predicted_effects : Dict[str, jnp.ndarray], optional
            A dictionary containing the predicted effects, by default None.

        Returns
        -------
        jnp.ndarray
            An array with shape (T,1) for univariate timeseries, or (N, T, 1) for
            multivariate timeseries, where T is the number of timepoints and N is the
            number of series.
        """
        n_features = data.shape[-1]

        with numpyro.plate("features_plate", n_features, dim=-1):
            coefficients = numpyro.sample("coefs", self.prior)

        if coefficients.ndim == 1:
            coefficients = jnp.expand_dims(coefficients, axis=-1)

        if data.ndim == 3 and coefficients.ndim == 2:
            coefficients = jnp.expand_dims(coefficients, axis=0)

        return matrix_multiplication(data, coefficients)

LinearFourierSeasonality

Bases: BaseEffect

Linear Fourier Seasonality effect.

Compute the linear seasonality using Fourier features.

Parameters:

Name	Type	Description	Default
sp_list	List[float]	List of seasonal periods.	required
fourier_terms_list	List[int]	List of number of Fourier terms to use for each seasonal period.	required
freq	str	Frequency of the time series. Example: "D" for daily, "W" for weekly, etc.	required
prior_scale	float	Scale of the prior distribution for the effect, by default 1.0.	1.0
effect_mode	str	Either "multiplicative" or "additive" by default "additive".	'additive'

Source code in src/prophetverse/effects/fourier.py

class LinearFourierSeasonality(BaseEffect):
    """Linear Fourier Seasonality effect.

    Compute the linear seasonality using Fourier features.

    Parameters
    ----------
    sp_list : List[float]
        List of seasonal periods.
    fourier_terms_list : List[int]
        List of number of Fourier terms to use for each seasonal period.
    freq : str
        Frequency of the time series. Example: "D" for daily, "W" for weekly, etc.
    prior_scale : float, optional
        Scale of the prior distribution for the effect, by default 1.0.
    effect_mode : str, optional
        Either "multiplicative" or "additive" by default "additive".
    """

    _tags = {
        # Supports multivariate data? Can this
        # Effect be used with Multiariate prophet?
        "supports_multivariate": True,
        # If no columns are found, should
        # _predict be skipped?
        "skip_predict_if_no_match": False,
    }

    def __init__(
        self,
        sp_list: List[float],
        fourier_terms_list: List[int],
        freq: Union[str, None],
        prior_scale: float = 1.0,
        effect_mode: EFFECT_APPLICATION_TYPE = "additive",
    ):
        self.sp_list = sp_list
        self.fourier_terms_list = fourier_terms_list
        self.freq = freq
        self.prior_scale = prior_scale
        self.effect_mode = effect_mode
        self.expand_column_per_level_ = None  # type: Union[None,ExpandColumnPerLevel]

    def _fit(self, y: pd.DataFrame, X: pd.DataFrame, scale: float = 1.0):
        """Customize the initialization of the effect.

        Fit the fourier feature transformer and the linear effect.

        Parameters
        ----------
        y : pd.DataFrame
            The timeseries dataframe

        X : pd.DataFrame
            The DataFrame to initialize the effect.

        scale: float, optional
            The scale of the timeseries, by default 1.0.
        """
        self.fourier_features_ = FourierFeatures(
            sp_list=self.sp_list,
            fourier_terms_list=self.fourier_terms_list,
            freq=self.freq,
            keep_original_columns=False,
        )

        self.fourier_features_.fit(X=X)
        X = self.fourier_features_.transform(X)

        if X.index.nlevels > 1 and X.index.droplevel(-1).nunique() > 1:
            self.expand_column_per_level_ = ExpandColumnPerLevel([".*"]).fit(X=X)
            X = self.expand_column_per_level_.transform(X)  # type: ignore

        self.linear_effect_ = LinearEffect(
            prior=dist.Normal(0, self.prior_scale), effect_mode=self.effect_mode
        )

        self.linear_effect_.fit(X=X, y=y, scale=scale)

    def _transform(self, X: pd.DataFrame, fh: pd.Index) -> jnp.ndarray:
        """Prepare input data to be passed to numpyro model.

        This method return a jnp.ndarray of sines and cosines of the given
        frequencies.

        Parameters
        ----------
        X : pd.DataFrame
            The input DataFrame containing the exogenous variables for the training
            time indexes, if passed during fit, or for the forecasting time indexes, if
            passed during predict.

        fh : pd.Index
            The forecasting horizon as a pandas Index.

        Returns
        -------
        jnp.ndarray
            Any object containing the data needed for the effect. The object will be
            passed to `predict` method as `data` argument.
        """
        X = self.fourier_features_.transform(X)

        if self.expand_column_per_level_ is not None:
            X = self.expand_column_per_level_.transform(X)

        array = self.linear_effect_.transform(X, fh)

        return array

    def _predict(
        self,
        data: Dict,
        predicted_effects: Dict[str, jnp.ndarray],
    ) -> jnp.ndarray:
        """Apply and return the effect values.

        Parameters
        ----------
        data : Any
            Data obtained from the transformed method.

        predicted_effects : Dict[str, jnp.ndarray], optional
            A dictionary containing the predicted effects, by default None.

        Returns
        -------
        jnp.ndarray
            An array with shape (T,1) for univariate timeseries, or (N, T, 1) for
            multivariate timeseries, where T is the number of timepoints and N is the
            number of series.
        """
        return self.linear_effect_.predict(
            data=data, predicted_effects=predicted_effects
        )

LogEffect

Bases: BaseAdditiveOrMultiplicativeEffect

Represents a log effect as effect = scale * log(rate * data + 1).

Parameters:

Name	Type	Description	Default
scale_prior	Optional[Distribution]	The prior distribution for the scale parameter., by default Gamma	None
rate_prior	Optional[Distribution]	The prior distribution for the rate parameter., by default Gamma	None
effect_mode	effects_application	Either "additive" or "multiplicative", by default "multiplicative"	'multiplicative'

Source code in src/prophetverse/effects/log.py

class LogEffect(BaseAdditiveOrMultiplicativeEffect):
    """Represents a log effect as effect = scale * log(rate * data + 1).

    Parameters
    ----------
    scale_prior : Optional[Distribution], optional
        The prior distribution for the scale parameter., by default Gamma
    rate_prior : Optional[Distribution], optional
        The prior distribution for the rate parameter., by default Gamma
    effect_mode : effects_application, optional
        Either "additive" or "multiplicative", by default "multiplicative"
    """

    def __init__(
        self,
        effect_mode: EFFECT_APPLICATION_TYPE = "multiplicative",
        scale_prior: Optional[Distribution] = None,
        rate_prior: Optional[Distribution] = None,
    ):
        self.scale_prior = scale_prior or dist.Gamma(1, 1)
        self.rate_prior = rate_prior or dist.Gamma(1, 1)
        super().__init__(effect_mode=effect_mode)

    def _predict(  # type: ignore[override]
        self,
        data: jnp.ndarray,
        predicted_effects: Optional[Dict[str, jnp.ndarray]] = None,
    ) -> jnp.ndarray:
        """Apply and return the effect values.

        Parameters
        ----------
        data : Any
            Data obtained from the transformed method.

        predicted_effects : Dict[str, jnp.ndarray], optional
            A dictionary containing the predicted effects, by default None.

        Returns
        -------
        jnp.ndarray
            An array with shape (T,1) for univariate timeseries, or (N, T, 1) for
            multivariate timeseries, where T is the number of timepoints and N is the
            number of series.
        """
        scale = numpyro.sample("log_scale", self.scale_prior)
        rate = numpyro.sample("log_rate", self.rate_prior)
        effect = scale * jnp.log(jnp.clip(rate * data + 1, 1e-8, None))

        return effect