BasePointForecaster

yohou.point.base.BasePointForecaster

Bases: BaseForecaster

Base class for point forecasters.

Parameters

Name	Type	Description	Default
target_transformer	instance of `BaseTransformer` or None	Transformer used to transform the target time series into the new target.	None
feature_transformer	instance of `BaseTransformer` or None	Transformer used to transform the target time series into features.	None
target_as_feature	(transformed, raw)	Controls whether the target is included as a feature. "transformed" includes the transformed target, "raw" includes the raw target, and None uses only exogenous features.	"transformed"
panel_strategy	('global', multivariate)	How to handle panel data. See BaseForecaster for details.	"global"

Notes

Subclasses must implement _predict_one to produce point predictions for a single forecast step. The forecaster_type tag is set to POINT.

See Also

• PointReductionForecaster : ML-based point forecaster.
• SeasonalNaive : Simple seasonal naive forecaster.
• BaseIntervalForecaster : Base class for interval forecasters.

Source Code

View on GitHub

Show/Hide source

class BasePointForecaster(BaseForecaster, metaclass=abc.ABCMeta):
    """Base class for point forecasters.

    Parameters
    ----------
    target_transformer : instance of `BaseTransformer` or None, default=None
        Transformer used to transform the target time series into the new target.
    feature_transformer : instance of `BaseTransformer` or None, default=None
        Transformer used to transform the target time series into features.
    target_as_feature : {"transformed", "raw"} or None, default="transformed"
        Controls whether the target is included as a feature.
        ``"transformed"`` includes the transformed target, ``"raw"``
        includes the raw target, and ``None`` uses only exogenous features.
    panel_strategy : {"global", "multivariate"}, default="global"
        How to handle panel data. See `BaseForecaster` for details.

    Notes
    -----
    Subclasses must implement ``_predict_one`` to produce point
    predictions for a single forecast step.  The ``forecaster_type``
    tag is set to ``POINT``.

    See Also
    --------
    - [`PointReductionForecaster`][yohou.point.reduction.PointReductionForecaster] : ML-based point forecaster.
    - [`SeasonalNaive`][yohou.point.naive.SeasonalNaive] : Simple seasonal naive forecaster.
    - [`BaseIntervalForecaster`][yohou.interval.base.BaseIntervalForecaster] : Base class for interval forecasters.

    """

    def __sklearn_tags__(self) -> Tags:
        """Get estimator tags.

        Returns
        -------
        Tags
            Estimator tags with yohou-specific attributes.

        """
        tags = super().__sklearn_tags__()
        assert tags.forecaster_tags is not None
        tags.forecaster_tags.forecaster_type = POINT
        return tags

    @_fit_context(prefer_skip_nested_validation=True)
    def fit(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        forecasting_horizon: StrictInt = 1,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> "BasePointForecaster":
        """Fit the forecaster to historical data.

        Parameters
        ----------
        y : pl.DataFrame
            Target time series with a ``"time"`` column (datetime) and one
            or more numeric value columns.
        X_actual : pl.DataFrame or None, default=None
            Actual feature observations with a ``"time"`` column aligned
            with ``y``. Processed by the feature transformer to produce
            lags, rolling statistics, and other derived features. If
            ``None``, only target-derived features are used.
        forecasting_horizon : int, default=1
            Number of time steps to forecast into the future.
        X_future : pl.DataFrame or None, default=None
            Known future features with a ``"time"`` column. Deterministic
            values available for past and future dates. Bypasses the
            feature transformer.
        X_forecast : pl.DataFrame or None, default=None
            External forecasts with ``"vintage_time"`` and ``"time"``
            columns. Vintage times do not need to align exactly with
            observation times; the latest vintage at or before each
            observation time is selected automatically (as-of matching).
            Bypasses the feature transformer.
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        self
            The fitted forecaster instance.

        Raises
        ------
        ValueError
            If ``forecasting_horizon`` < 1, or if ``y`` / ``X_actual`` have invalid
            structure (e.g., missing ``"time"`` column).

        """
        forecasting_horizon = self._validate_fit_params(forecasting_horizon)

        y_t, X_t = self._pre_fit(
            y=y,
            X_actual=X_actual,
            forecasting_horizon=forecasting_horizon,
            X_future=X_future,
            X_forecast=X_forecast,
        )

        self._fit(y_t, X_t, forecasting_horizon)

        return self

    def _validate_predict_params(self, forecasting_horizon: StrictInt | None) -> StrictInt:
        """Validate and return predict parameters.

        Parameters
        ----------
        forecasting_horizon : int or None
            Forecasting horizon to validate. If None, uses fit_forecasting_horizon_.

        Returns
        -------
        int
            Validated forecasting horizon.

        Raises
        ------
        ValueError
            If forecasting_horizon < 1.

        """
        if forecasting_horizon is None:
            forecasting_horizon = self.fit_forecasting_horizon_
        return self._validate_fit_params(forecasting_horizon)

    def predict(
        self,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        forecasting_horizon: StrictInt | None = None,
        groups: list[str] | None = None,
        predict_transformed: bool = False,
        **params,
    ) -> pl.DataFrame:
        """Generate point forecasts.

        Parameters
        ----------
        X_future : pl.DataFrame or None, default=None
            Known future features override. Re-derives step columns
            without mutating forecaster state.
        X_forecast : pl.DataFrame or None, default=None
            External forecast override with ``"vintage_time"`` and
            ``"time"`` columns. Re-derives step columns using as-of
            matching without mutating forecaster state.
        forecasting_horizon : int or None, default=None
            Number of time steps to forecast into the future.  If ``None``,
            uses the horizon specified at fit time.
        groups : list of str or None, default=None
            Panel group prefixes to operate on.  If ``None``, all groups
            are used.  Ignored when the forecaster was not fitted on panel
            data.
        predict_transformed : bool, default=False
            If ``True``, return predictions in the transformed space without
            applying inverse target transformation.
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        pl.DataFrame
            Point predictions with ``"vintage_time"``, ``"time"``, and one
            column per target variable.

        Raises
        ------
        sklearn.exceptions.NotFittedError
            If the forecaster has not been fitted yet.
        ValueError
            If ``groups`` contains names not seen during fit.

        """
        check_is_fitted(
            self,
            ["local_y_schema_", "local_X_actual_schema_", "shared_X_actual_schema_", "groups_"],
        )

        _, _, groups = validate_forecaster_data(
            self,
            y=None,
            X_actual=None,
            reset=False,
            groups=groups,
            X_future=X_future,
            X_forecast=X_forecast,
        )

        forecasting_horizon = self._validate_predict_params(forecasting_horizon)

        def step_fn(forecaster, groups):
            """Produce one point-prediction block."""
            y_pred_step, y_pred_step_inv = forecaster._predict(groups)
            y_accumulate = y_pred_step if predict_transformed else y_pred_step_inv
            return y_accumulate, y_pred_step_inv

        def derive_observation_fn(forecaster, y_pred_step_inv):
            """Derive observation from inverse-transformed prediction."""
            if self.groups_ is None:
                y = y_pred_step_inv.select(["time"] + list(self.local_y_schema_.keys()))
            else:
                y_columns = ["time"]
                for group_name in self.groups_:
                    y_columns.extend([f"{group_name}__{col}" for col in self.local_y_schema_])
                y = y_pred_step_inv.select(y_columns)
            return y

        def predict_fn():
            """Run recursive predict with step columns."""
            return self._recursive_predict(
                forecasting_horizon=forecasting_horizon,
                groups=groups,
                step_fn=step_fn,
                derive_observation_fn=derive_observation_fn,
            )

        return self._predict_with_step_override(
            X_future=X_future,
            X_forecast=X_forecast,
            predict_fn=predict_fn,
        )

    def observe_predict(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        forecasting_horizon: StrictInt | None = None,
        groups: list[str] | None = None,
        stride: StrictInt | None = None,
        predict_transformed: bool = False,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> pl.DataFrame:
        """Alternate recursive predict and observe.

        Equivalent to calling ``observe(y, X_actual)`` then ``predict()``.
        Returns point predictions.

        Parameters
        ----------
        y : pl.DataFrame
            Target time series with a ``"time"`` column (datetime) and one
            or more numeric value columns.
        X_actual : pl.DataFrame or None, default=None
            Actual feature observations with a ``"time"`` column aligned
            with ``y``. Sliced and observed incrementally at each step of
            the rolling loop.
        forecasting_horizon : int or None, default=None
            Number of time steps to forecast into the future.  If ``None``,
            uses the horizon specified at fit time.
        groups : list of str or None, default=None
            Panel group prefixes to operate on.  If ``None``, all groups
            are used.  Ignored when the forecaster was not fitted on panel
            data.
        stride : int or None, default=None
            Step size for rolling update-predict.  If ``None``, defaults to
            ``forecasting_horizon``.
        predict_transformed : bool, default=False
            If ``True``, return predictions in the transformed space without
            applying inverse target transformation.
        X_future : pl.DataFrame or None, default=None
            Known future features with a ``"time"`` column.
        X_forecast : pl.DataFrame or None, default=None
            External forecasts with ``"vintage_time"`` and ``"time"``
            columns. Vintage times do not need to align exactly with
            observation times; the latest vintage at or before each
            observation time is selected automatically (as-of matching).
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        pl.DataFrame
            Point predictions with ``"vintage_time"``, ``"time"``, and one
            column per target variable.

        Raises
        ------
        sklearn.exceptions.NotFittedError
            If the forecaster has not been fitted yet.
        ValueError
            If ``y`` / ``X_actual`` have invalid structure or ``groups``
            contains names not seen during fit.

        """
        check_is_fitted(
            self,
            ["local_y_schema_", "local_X_actual_schema_", "shared_X_actual_schema_", "groups_"],
        )

        y, X_actual, groups = validate_forecaster_data(
            self,
            y=y,
            X_actual=X_actual,
            reset=False,
            groups=groups,
            X_future=X_future,
            X_forecast=X_forecast,
        )

        forecasting_horizon = self._validate_predict_params(forecasting_horizon)
        if stride is None:
            stride = self.fit_forecasting_horizon_

        return self._observe_predict_loop(
            predict_fn=self.predict,
            y=y,
            X_actual=X_actual,
            X_future=X_future,
            X_forecast=X_forecast,
            groups=groups,
            stride=stride,
            forecasting_horizon=forecasting_horizon,
            predict_transformed=predict_transformed,
            **params,
        )

Methods

__sklearn_tags__()

Get estimator tags.

Returns

Type	Description
Tags	Estimator tags with yohou-specific attributes.

Source Code

View on GitHub

Show/Hide source

def __sklearn_tags__(self) -> Tags:
    """Get estimator tags.

    Returns
    -------
    Tags
        Estimator tags with yohou-specific attributes.

    """
    tags = super().__sklearn_tags__()
    assert tags.forecaster_tags is not None
    tags.forecaster_tags.forecaster_type = POINT
    return tags

fit(y, X_actual=None, forecasting_horizon=1, X_future=None, X_forecast=None, **params)

Fit the forecaster to historical data.

Parameters

Name	Type	Description	Default
y	DataFrame	Target time series with a "time" column (datetime) and one or more numeric value columns.	required
X_actual	DataFrame or None	Actual feature observations with a "time" column aligned with y. Processed by the feature transformer to produce lags, rolling statistics, and other derived features. If None, only target-derived features are used.	None
forecasting_horizon	int	Number of time steps to forecast into the future.	1
X_future	DataFrame or None	Known future features with a "time" column. Deterministic values available for past and future dates. Bypasses the feature transformer.	None
X_forecast	DataFrame or None	External forecasts with "vintage_time" and "time" columns. Vintage times do not need to align exactly with observation times; the latest vintage at or before each observation time is selected automatically (as-of matching). Bypasses the feature transformer.	None
**params	dict	Metadata to route to nested estimators.	{}

Returns

Type	Description
self	The fitted forecaster instance.

Raises

Type	Description
ValueError	If forecasting_horizon < 1, or if y / X_actual have invalid structure (e.g., missing "time" column).

Source Code

View on GitHub

Show/Hide source

@_fit_context(prefer_skip_nested_validation=True)
def fit(
    self,
    y: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
    forecasting_horizon: StrictInt = 1,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> "BasePointForecaster":
    """Fit the forecaster to historical data.

    Parameters
    ----------
    y : pl.DataFrame
        Target time series with a ``"time"`` column (datetime) and one
        or more numeric value columns.
    X_actual : pl.DataFrame or None, default=None
        Actual feature observations with a ``"time"`` column aligned
        with ``y``. Processed by the feature transformer to produce
        lags, rolling statistics, and other derived features. If
        ``None``, only target-derived features are used.
    forecasting_horizon : int, default=1
        Number of time steps to forecast into the future.
    X_future : pl.DataFrame or None, default=None
        Known future features with a ``"time"`` column. Deterministic
        values available for past and future dates. Bypasses the
        feature transformer.
    X_forecast : pl.DataFrame or None, default=None
        External forecasts with ``"vintage_time"`` and ``"time"``
        columns. Vintage times do not need to align exactly with
        observation times; the latest vintage at or before each
        observation time is selected automatically (as-of matching).
        Bypasses the feature transformer.
    **params : dict
        Metadata to route to nested estimators.

    Returns
    -------
    self
        The fitted forecaster instance.

    Raises
    ------
    ValueError
        If ``forecasting_horizon`` < 1, or if ``y`` / ``X_actual`` have invalid
        structure (e.g., missing ``"time"`` column).

    """
    forecasting_horizon = self._validate_fit_params(forecasting_horizon)

    y_t, X_t = self._pre_fit(
        y=y,
        X_actual=X_actual,
        forecasting_horizon=forecasting_horizon,
        X_future=X_future,
        X_forecast=X_forecast,
    )

    self._fit(y_t, X_t, forecasting_horizon)

    return self

predict(X_future=None, X_forecast=None, forecasting_horizon=None, groups=None, predict_transformed=False, **params)

Generate point forecasts.

Parameters

Name	Type	Description	Default
X_future	DataFrame or None	Known future features override. Re-derives step columns without mutating forecaster state.	None
X_forecast	DataFrame or None	External forecast override with "vintage_time" and "time" columns. Re-derives step columns using as-of matching without mutating forecaster state.	None
forecasting_horizon	int or None	Number of time steps to forecast into the future. If None, uses the horizon specified at fit time.	None
groups	list of str or None	Panel group prefixes to operate on. If None, all groups are used. Ignored when the forecaster was not fitted on panel data.	None
predict_transformed	bool	If True, return predictions in the transformed space without applying inverse target transformation.	False
**params	dict	Metadata to route to nested estimators.	{}

Returns

Type	Description
DataFrame	Point predictions with "vintage_time", "time", and one column per target variable.

Raises

Type	Description
NotFittedError	If the forecaster has not been fitted yet.
ValueError	If groups contains names not seen during fit.

Source Code

View on GitHub

Show/Hide source

def predict(
    self,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    forecasting_horizon: StrictInt | None = None,
    groups: list[str] | None = None,
    predict_transformed: bool = False,
    **params,
) -> pl.DataFrame:
    """Generate point forecasts.

    Parameters
    ----------
    X_future : pl.DataFrame or None, default=None
        Known future features override. Re-derives step columns
        without mutating forecaster state.
    X_forecast : pl.DataFrame or None, default=None
        External forecast override with ``"vintage_time"`` and
        ``"time"`` columns. Re-derives step columns using as-of
        matching without mutating forecaster state.
    forecasting_horizon : int or None, default=None
        Number of time steps to forecast into the future.  If ``None``,
        uses the horizon specified at fit time.
    groups : list of str or None, default=None
        Panel group prefixes to operate on.  If ``None``, all groups
        are used.  Ignored when the forecaster was not fitted on panel
        data.
    predict_transformed : bool, default=False
        If ``True``, return predictions in the transformed space without
        applying inverse target transformation.
    **params : dict
        Metadata to route to nested estimators.

    Returns
    -------
    pl.DataFrame
        Point predictions with ``"vintage_time"``, ``"time"``, and one
        column per target variable.

    Raises
    ------
    sklearn.exceptions.NotFittedError
        If the forecaster has not been fitted yet.
    ValueError
        If ``groups`` contains names not seen during fit.

    """
    check_is_fitted(
        self,
        ["local_y_schema_", "local_X_actual_schema_", "shared_X_actual_schema_", "groups_"],
    )

    _, _, groups = validate_forecaster_data(
        self,
        y=None,
        X_actual=None,
        reset=False,
        groups=groups,
        X_future=X_future,
        X_forecast=X_forecast,
    )

    forecasting_horizon = self._validate_predict_params(forecasting_horizon)

    def step_fn(forecaster, groups):
        """Produce one point-prediction block."""
        y_pred_step, y_pred_step_inv = forecaster._predict(groups)
        y_accumulate = y_pred_step if predict_transformed else y_pred_step_inv
        return y_accumulate, y_pred_step_inv

    def derive_observation_fn(forecaster, y_pred_step_inv):
        """Derive observation from inverse-transformed prediction."""
        if self.groups_ is None:
            y = y_pred_step_inv.select(["time"] + list(self.local_y_schema_.keys()))
        else:
            y_columns = ["time"]
            for group_name in self.groups_:
                y_columns.extend([f"{group_name}__{col}" for col in self.local_y_schema_])
            y = y_pred_step_inv.select(y_columns)
        return y

    def predict_fn():
        """Run recursive predict with step columns."""
        return self._recursive_predict(
            forecasting_horizon=forecasting_horizon,
            groups=groups,
            step_fn=step_fn,
            derive_observation_fn=derive_observation_fn,
        )

    return self._predict_with_step_override(
        X_future=X_future,
        X_forecast=X_forecast,
        predict_fn=predict_fn,
    )

observe_predict(y, X_actual=None, forecasting_horizon=None, groups=None, stride=None, predict_transformed=False, X_future=None, X_forecast=None, **params)

Alternate recursive predict and observe.

Equivalent to calling observe(y, X_actual) then predict(). Returns point predictions.

Parameters

Name	Type	Description	Default
y	DataFrame	Target time series with a "time" column (datetime) and one or more numeric value columns.	required
X_actual	DataFrame or None	Actual feature observations with a "time" column aligned with y. Sliced and observed incrementally at each step of the rolling loop.	None
forecasting_horizon	int or None	Number of time steps to forecast into the future. If None, uses the horizon specified at fit time.	None
groups	list of str or None	Panel group prefixes to operate on. If None, all groups are used. Ignored when the forecaster was not fitted on panel data.	None
stride	int or None	Step size for rolling update-predict. If None, defaults to forecasting_horizon.	None
predict_transformed	bool	If True, return predictions in the transformed space without applying inverse target transformation.	False
X_future	DataFrame or None	Known future features with a "time" column.	None
X_forecast	DataFrame or None	External forecasts with "vintage_time" and "time" columns. Vintage times do not need to align exactly with observation times; the latest vintage at or before each observation time is selected automatically (as-of matching).	None
**params	dict	Metadata to route to nested estimators.	{}

Returns

Type	Description
DataFrame	Point predictions with "vintage_time", "time", and one column per target variable.

Raises

Type	Description
NotFittedError	If the forecaster has not been fitted yet.
ValueError	If y / X_actual have invalid structure or groups contains names not seen during fit.

Source Code

View on GitHub

Show/Hide source

def observe_predict(
    self,
    y: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
    forecasting_horizon: StrictInt | None = None,
    groups: list[str] | None = None,
    stride: StrictInt | None = None,
    predict_transformed: bool = False,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> pl.DataFrame:
    """Alternate recursive predict and observe.

    Equivalent to calling ``observe(y, X_actual)`` then ``predict()``.
    Returns point predictions.

    Parameters
    ----------
    y : pl.DataFrame
        Target time series with a ``"time"`` column (datetime) and one
        or more numeric value columns.
    X_actual : pl.DataFrame or None, default=None
        Actual feature observations with a ``"time"`` column aligned
        with ``y``. Sliced and observed incrementally at each step of
        the rolling loop.
    forecasting_horizon : int or None, default=None
        Number of time steps to forecast into the future.  If ``None``,
        uses the horizon specified at fit time.
    groups : list of str or None, default=None
        Panel group prefixes to operate on.  If ``None``, all groups
        are used.  Ignored when the forecaster was not fitted on panel
        data.
    stride : int or None, default=None
        Step size for rolling update-predict.  If ``None``, defaults to
        ``forecasting_horizon``.
    predict_transformed : bool, default=False
        If ``True``, return predictions in the transformed space without
        applying inverse target transformation.
    X_future : pl.DataFrame or None, default=None
        Known future features with a ``"time"`` column.
    X_forecast : pl.DataFrame or None, default=None
        External forecasts with ``"vintage_time"`` and ``"time"``
        columns. Vintage times do not need to align exactly with
        observation times; the latest vintage at or before each
        observation time is selected automatically (as-of matching).
    **params : dict
        Metadata to route to nested estimators.

    Returns
    -------
    pl.DataFrame
        Point predictions with ``"vintage_time"``, ``"time"``, and one
        column per target variable.

    Raises
    ------
    sklearn.exceptions.NotFittedError
        If the forecaster has not been fitted yet.
    ValueError
        If ``y`` / ``X_actual`` have invalid structure or ``groups``
        contains names not seen during fit.

    """
    check_is_fitted(
        self,
        ["local_y_schema_", "local_X_actual_schema_", "shared_X_actual_schema_", "groups_"],
    )

    y, X_actual, groups = validate_forecaster_data(
        self,
        y=y,
        X_actual=X_actual,
        reset=False,
        groups=groups,
        X_future=X_future,
        X_forecast=X_forecast,
    )

    forecasting_horizon = self._validate_predict_params(forecasting_horizon)
    if stride is None:
        stride = self.fit_forecasting_horizon_

    return self._observe_predict_loop(
        predict_fn=self.predict,
        y=y,
        X_actual=X_actual,
        X_future=X_future,
        X_forecast=X_forecast,
        groups=groups,
        stride=stride,
        forecasting_horizon=forecasting_horizon,
        predict_transformed=predict_transformed,
        **params,
    )

Tutorials

The following example notebooks use this component:

• How to Create a Custom Estimator

  ---

  Getting-Started

  Implement a LastValueForecaster from scratch, validate it with the check generator, and use it in a forecast pipeline.

  View · Open in marimo