BaseIntervalForecaster

yohou.interval.base.BaseIntervalForecaster

Bases: BaseForecaster

Base class for interval forecasters.

Parameters

Name	Type	Description	Default
feature_transformer	instance of `BaseTransformer` or None	Transformer used to transform the feature 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"

Attributes

Name	Type	Description
fit_coverage_rates_	list of float	Coverage rates used during fit.

Notes

Interval forecasters produce prediction intervals at specified coverage rates. The forecaster_type tag is INTERVAL (or POINT_INTERVAL if point predictions are also available).

See Also

• SplitConformalForecaster : Conformal interval forecaster.
• IntervalReductionForecaster : ML-based interval forecaster.
• BasePointForecaster : Base class for point forecasters.

Source Code

View on GitHub

Show/Hide source

class BaseIntervalForecaster(BaseForecaster, metaclass=abc.ABCMeta):
    """Base class for interval forecasters.

    Parameters
    ----------
    feature_transformer : instance of `BaseTransformer` or None, default=None
        Transformer used to transform the feature 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.

    Attributes
    ----------
    fit_coverage_rates_ : list of float
        Coverage rates used during fit.

    Notes
    -----
    Interval forecasters produce prediction intervals at specified
    coverage rates.  The ``forecaster_type`` tag is ``INTERVAL``
    (or ``POINT_INTERVAL`` if point predictions are also available).

    See Also
    --------
    - [`SplitConformalForecaster`][yohou.interval.split_conformal.SplitConformalForecaster] : Conformal interval forecaster.
    - [`IntervalReductionForecaster`][yohou.interval.reduction.IntervalReductionForecaster] : ML-based interval forecaster.
    - [`BasePointForecaster`][yohou.point.base.BasePointForecaster] : Base class for point forecasters.

    """

    _tags: dict = {"forecaster_type": INTERVAL}

    _parameter_constraints: dict = {}

    def __init__(
        self,
        feature_transformer: BaseTransformer | None = None,
        target_as_feature: Literal["transformed", "raw"] | None = "transformed",
        panel_strategy: Literal["global", "multivariate"] = "global",
    ) -> None:
        super().__init__(
            feature_transformer=feature_transformer,
            target_transformer=None,
            target_as_feature=target_as_feature,
            panel_strategy=panel_strategy,
        )

    @_fit_context(prefer_skip_nested_validation=True)
    def fit(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        forecasting_horizon: StrictInt = 1,
        *,
        coverage_rates: list[float] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> "BaseIntervalForecaster":
        """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``.  If ``None``, no exogenous features are used.
        forecasting_horizon : int, default=1
            Number of time steps to forecast into the future.
        coverage_rates : list of float or None, default=None
            Coverage levels for prediction intervals (e.g., ``[0.9, 0.95]``
            for 90 % and 95 % intervals).  If ``None``, defaults to
            ``[0.95]``.
        X_future : pl.DataFrame or None, default=None
            Known future features with a ``"time"`` column. Deterministic
            values available for past and future dates.
        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, ``coverage_rates`` not in (0, 1],
            or if ``y`` / ``X_actual`` have invalid structure.

        """
        forecasting_horizon, coverage_rates = self._validate_interval_fit_params(forecasting_horizon, coverage_rates)
        self.fit_coverage_rates_ = coverage_rates

        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_interval_fit_params(
        self,
        forecasting_horizon: StrictInt,
        coverage_rates: list[StrictFloat] | None = None,
    ) -> tuple[StrictInt, list[StrictFloat]]:
        """Validate fit parameters.

        Parameters
        ----------
        forecasting_horizon : int
            Forecasting horizon to validate.
        coverage_rates : list of float or None
            Coverage rates to validate. If None, uses [0.95].

        Returns
        -------
        tuple of (int, list of float)
            Validated forecasting horizon and coverage rates.

        Raises
        ------
        ValueError
            If forecasting_horizon < 1 or coverage_rates not in [0, 1].

        """
        if forecasting_horizon < 1:
            raise ValueError(f"forecasting_horizon must be >= 1, got {forecasting_horizon}")

        if coverage_rates is None:
            coverage_rates = [0.95]

        # Validate coverage rates
        for rate in coverage_rates:
            if not (0 <= rate <= 1):
                raise ValueError(f"All coverage_rates must be in [0, 1], got {rate}")

        return forecasting_horizon, coverage_rates

    def _validate_predict_params(
        self,
        forecasting_horizon: StrictInt | None,
        coverage_rates: list[StrictFloat] | None = None,
    ) -> tuple[StrictInt, list[StrictFloat]]:
        """Validate and return predict parameters.

        Parameters
        ----------
        forecasting_horizon : int or None
            Forecasting horizon to validate. If None, uses fit_forecasting_horizon_.
        coverage_rates : list of float or None
            Coverage rates to validate. If None, uses fit_coverage_rates_.

        Returns
        -------
        tuple of (int, list of float)
            Validated forecasting horizon and coverage rates.

        Raises
        ------
        ValueError
            If forecasting_horizon < 1 or coverage_rates not in [0, 1].

        """
        if forecasting_horizon is None:
            forecasting_horizon = self.fit_forecasting_horizon_
        if coverage_rates is None:
            # fit_coverage_rates_ is set by concrete subclasses during fit().
            coverage_rates = self.fit_coverage_rates_
        return self._validate_interval_fit_params(forecasting_horizon, coverage_rates)

    def predict_interval(
        self,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        forecasting_horizon: StrictInt | None = None,
        coverage_rates: list[float] | None = None,
        strategy: Literal["mean", "median", "point"] | None = None,
        groups: list[str] | None = None,
        **params,
    ) -> pl.DataFrame:
        """Generate interval 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.
        coverage_rates : list of float or None, default=None
            Coverage levels for prediction intervals (e.g., ``[0.9, 0.95]``
            for 90 % and 95 % intervals).  If ``None``, defaults to the rates
            used at fit time.
        strategy : {"mean", "median", "point"} or None, default=None
            Strategy for deriving point predictions from prediction intervals
            during recursive multi-step forecasting:

            - ``"mean"``: use the mean of the interval bounds
            - ``"median"``: use the median of the interval bounds
            - ``"point"``: use the point forecast directly (if available)

            If ``None``, defaults to ``"mean"``.
        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.
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        pl.DataFrame
            Interval predictions with ``"vintage_time"``, ``"time"``, and
            lower/upper bound columns for each target at each coverage rate.

        Raises
        ------
        sklearn.exceptions.NotFittedError
            If the forecaster has not been fitted yet.
        ValueError
            If ``coverage_rates`` not in [0, 1],
            or ``groups`` contains names not seen during fit.

        """
        check_is_fitted(self, ["groups_", "local_y_schema_", "fit_forecasting_horizon_"])
        _, _, groups = validate_forecaster_data(
            self,
            y=None,
            X_actual=None,
            reset=False,
            groups=groups,
            X_future=X_future,
            X_forecast=X_forecast,
        )

        forecasting_horizon, coverage_rates = self._validate_predict_params(forecasting_horizon, coverage_rates)

        y_columns = list(self.local_y_schema_.keys())
        if groups is not None:
            y_columns = [f"{panel_group}__{col}" for panel_group in groups for col in self.local_y_schema_]

        def step_fn(forecaster, groups):
            """Produce one interval-prediction block."""
            y_pred_step, y_pred_step_inv = forecaster._predict(groups, coverage_rates=coverage_rates)
            return y_pred_step_inv, y_pred_step_inv

        def derive_observation_fn(forecaster, y_pred_step_inv):
            """Derive observation from interval bounds."""
            time = y_pred_step_inv.select(cs.by_name("time"))

            y_data: dict[str, Any] = {"time": time["time"]}
            for col in y_columns:
                lower_cols = [c for c in y_pred_step_inv.columns if c.startswith(f"{col}_lower_")]
                upper_cols = [c for c in y_pred_step_inv.columns if c.startswith(f"{col}_upper_")]

                all_bound_cols = lower_cols + upper_cols

                if strategy == "point" and col in y_pred_step_inv.columns:
                    y_data[col] = y_pred_step_inv[col]
                elif strategy == "median":
                    y_data[col] = y_pred_step_inv.select(
                        pl.median_horizontal(all_bound_cols)  # ty: ignore[unresolved-attribute]
                    ).to_series()
                else:
                    y_data[col] = y_pred_step_inv.select(pl.mean_horizontal(all_bound_cols)).to_series()

            y = pl.DataFrame(y_data)

            if groups is not None:
                cast_schema = {}
                for group_name in groups:
                    for col_name, dtype in forecaster.local_y_schema_.items():
                        cast_schema[f"{group_name}__{col_name}"] = dtype
            else:
                cast_schema = forecaster.local_y_schema_

            y = cast(y.select(~cs.by_name("time")), cast_schema)
            y = pl.concat([y_data["time"].to_frame(), y], how="horizontal")
            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_interval(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        forecasting_horizon: StrictInt | None = None,
        coverage_rates: list[float] | None = None,
        strategy: Literal["mean", "median", "point"] | None = None,
        groups: list[str] | None = None,
        stride: StrictInt | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> pl.DataFrame:
        """Alternate recursive predict_interval and observe.

        Equivalent to calling ``observe(y, X_actual)`` then
        ``predict_interval()``.  Returns interval 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.
        coverage_rates : list of float or None, default=None
            Coverage levels for prediction intervals (e.g., ``[0.9, 0.95]``
            for 90 % and 95 % intervals).  If ``None``, defaults to the rates
            used at fit time.
        strategy : {"mean", "median", "point"} or None, default=None
            Strategy for deriving point predictions from prediction intervals
            during recursive multi-step forecasting:

            - ``"mean"``: use the mean of the interval bounds
            - ``"median"``: use the median of the interval bounds
            - ``"point"``: use the point forecast directly (if available)

            If ``None``, defaults to ``"mean"``.
        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``.
        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.
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        pl.DataFrame
            Interval predictions with ``"vintage_time"``, ``"time"``, and
            lower/upper bound columns for each target at each coverage rate.

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

        """
        check_is_fitted(self, ["groups_", "local_y_schema_", "fit_forecasting_horizon_"])
        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, coverage_rates)

        if stride is None:
            stride = self.fit_forecasting_horizon_

        return self._observe_predict_loop(
            predict_fn=self.predict_interval,
            y=y,
            X_actual=X_actual,
            X_future=X_future,
            X_forecast=X_forecast,
            groups=groups,
            stride=stride,
            forecasting_horizon=forecasting_horizon,
            coverage_rates=coverage_rates,
            strategy=strategy,
            **params,
        )

    def _predict_one(
        self,
        groups: list[str],
        coverage_rates: list[StrictFloat] | None = None,
        **params,
    ) -> pl.DataFrame:
        """Predicts `_fit_forecasting_horizon` steps from the observation horizon.

        Parameters
        ----------
        groups : list of str
            Panel group names to predict for.
        coverage_rates : list of float
            Coverage rates for the prediction intervals.

        Returns
        -------
        pl.DataFrame
            Predicted time series.

        """
        raise NotImplementedError()

Methods

fit(y, X_actual=None, forecasting_horizon=1, *, coverage_rates=None, 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. If None, no exogenous features are used.	None
forecasting_horizon	int	Number of time steps to forecast into the future.	1
coverage_rates	list of float or None	Coverage levels for prediction intervals (e.g., [0.9, 0.95] for 90 % and 95 % intervals). If None, defaults to [0.95].	None
X_future	DataFrame or None	Known future features with a "time" column. Deterministic values available for past and future dates.	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, coverage_rates not in (0, 1], or if y / X_actual have invalid structure.

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,
    *,
    coverage_rates: list[float] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> "BaseIntervalForecaster":
    """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``.  If ``None``, no exogenous features are used.
    forecasting_horizon : int, default=1
        Number of time steps to forecast into the future.
    coverage_rates : list of float or None, default=None
        Coverage levels for prediction intervals (e.g., ``[0.9, 0.95]``
        for 90 % and 95 % intervals).  If ``None``, defaults to
        ``[0.95]``.
    X_future : pl.DataFrame or None, default=None
        Known future features with a ``"time"`` column. Deterministic
        values available for past and future dates.
    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, ``coverage_rates`` not in (0, 1],
        or if ``y`` / ``X_actual`` have invalid structure.

    """
    forecasting_horizon, coverage_rates = self._validate_interval_fit_params(forecasting_horizon, coverage_rates)
    self.fit_coverage_rates_ = coverage_rates

    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_interval(X_future=None, X_forecast=None, forecasting_horizon=None, coverage_rates=None, strategy=None, groups=None, **params)

Generate interval 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
coverage_rates	list of float or None	Coverage levels for prediction intervals (e.g., [0.9, 0.95] for 90 % and 95 % intervals). If None, defaults to the rates used at fit time.	None
strategy	(mean, median, point)	Strategy for deriving point predictions from prediction intervals during recursive multi-step forecasting: "mean": use the mean of the interval bounds "median": use the median of the interval bounds "point": use the point forecast directly (if available) If None, defaults to "mean".	"mean"
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
**params	dict	Metadata to route to nested estimators.	{}

Returns

Type	Description
DataFrame	Interval predictions with "vintage_time", "time", and lower/upper bound columns for each target at each coverage rate.

Raises

Type	Description
NotFittedError	If the forecaster has not been fitted yet.
ValueError	If coverage_rates not in [0, 1], or groups contains names not seen during fit.

Source Code

View on GitHub

Show/Hide source

def predict_interval(
    self,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    forecasting_horizon: StrictInt | None = None,
    coverage_rates: list[float] | None = None,
    strategy: Literal["mean", "median", "point"] | None = None,
    groups: list[str] | None = None,
    **params,
) -> pl.DataFrame:
    """Generate interval 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.
    coverage_rates : list of float or None, default=None
        Coverage levels for prediction intervals (e.g., ``[0.9, 0.95]``
        for 90 % and 95 % intervals).  If ``None``, defaults to the rates
        used at fit time.
    strategy : {"mean", "median", "point"} or None, default=None
        Strategy for deriving point predictions from prediction intervals
        during recursive multi-step forecasting:

        - ``"mean"``: use the mean of the interval bounds
        - ``"median"``: use the median of the interval bounds
        - ``"point"``: use the point forecast directly (if available)

        If ``None``, defaults to ``"mean"``.
    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.
    **params : dict
        Metadata to route to nested estimators.

    Returns
    -------
    pl.DataFrame
        Interval predictions with ``"vintage_time"``, ``"time"``, and
        lower/upper bound columns for each target at each coverage rate.

    Raises
    ------
    sklearn.exceptions.NotFittedError
        If the forecaster has not been fitted yet.
    ValueError
        If ``coverage_rates`` not in [0, 1],
        or ``groups`` contains names not seen during fit.

    """
    check_is_fitted(self, ["groups_", "local_y_schema_", "fit_forecasting_horizon_"])
    _, _, groups = validate_forecaster_data(
        self,
        y=None,
        X_actual=None,
        reset=False,
        groups=groups,
        X_future=X_future,
        X_forecast=X_forecast,
    )

    forecasting_horizon, coverage_rates = self._validate_predict_params(forecasting_horizon, coverage_rates)

    y_columns = list(self.local_y_schema_.keys())
    if groups is not None:
        y_columns = [f"{panel_group}__{col}" for panel_group in groups for col in self.local_y_schema_]

    def step_fn(forecaster, groups):
        """Produce one interval-prediction block."""
        y_pred_step, y_pred_step_inv = forecaster._predict(groups, coverage_rates=coverage_rates)
        return y_pred_step_inv, y_pred_step_inv

    def derive_observation_fn(forecaster, y_pred_step_inv):
        """Derive observation from interval bounds."""
        time = y_pred_step_inv.select(cs.by_name("time"))

        y_data: dict[str, Any] = {"time": time["time"]}
        for col in y_columns:
            lower_cols = [c for c in y_pred_step_inv.columns if c.startswith(f"{col}_lower_")]
            upper_cols = [c for c in y_pred_step_inv.columns if c.startswith(f"{col}_upper_")]

            all_bound_cols = lower_cols + upper_cols

            if strategy == "point" and col in y_pred_step_inv.columns:
                y_data[col] = y_pred_step_inv[col]
            elif strategy == "median":
                y_data[col] = y_pred_step_inv.select(
                    pl.median_horizontal(all_bound_cols)  # ty: ignore[unresolved-attribute]
                ).to_series()
            else:
                y_data[col] = y_pred_step_inv.select(pl.mean_horizontal(all_bound_cols)).to_series()

        y = pl.DataFrame(y_data)

        if groups is not None:
            cast_schema = {}
            for group_name in groups:
                for col_name, dtype in forecaster.local_y_schema_.items():
                    cast_schema[f"{group_name}__{col_name}"] = dtype
        else:
            cast_schema = forecaster.local_y_schema_

        y = cast(y.select(~cs.by_name("time")), cast_schema)
        y = pl.concat([y_data["time"].to_frame(), y], how="horizontal")
        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_interval(y, X_actual=None, forecasting_horizon=None, coverage_rates=None, strategy=None, groups=None, stride=None, X_future=None, X_forecast=None, **params)

Alternate recursive predict_interval and observe.

Equivalent to calling observe(y, X_actual) then predict_interval(). Returns interval 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
coverage_rates	list of float or None	Coverage levels for prediction intervals (e.g., [0.9, 0.95] for 90 % and 95 % intervals). If None, defaults to the rates used at fit time.	None
strategy	(mean, median, point)	Strategy for deriving point predictions from prediction intervals during recursive multi-step forecasting: "mean": use the mean of the interval bounds "median": use the median of the interval bounds "point": use the point forecast directly (if available) If None, defaults to "mean".	"mean"
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
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.	None
**params	dict	Metadata to route to nested estimators.	{}

Returns

Type	Description
DataFrame	Interval predictions with "vintage_time", "time", and lower/upper bound columns for each target at each coverage rate.

Raises

Type	Description
NotFittedError	If the forecaster has not been fitted yet.
ValueError	If y / X_actual have invalid structure, coverage_rates not in (0, 1], or groups contains names not seen during fit.

Source Code

View on GitHub

Show/Hide source

def observe_predict_interval(
    self,
    y: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
    forecasting_horizon: StrictInt | None = None,
    coverage_rates: list[float] | None = None,
    strategy: Literal["mean", "median", "point"] | None = None,
    groups: list[str] | None = None,
    stride: StrictInt | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> pl.DataFrame:
    """Alternate recursive predict_interval and observe.

    Equivalent to calling ``observe(y, X_actual)`` then
    ``predict_interval()``.  Returns interval 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.
    coverage_rates : list of float or None, default=None
        Coverage levels for prediction intervals (e.g., ``[0.9, 0.95]``
        for 90 % and 95 % intervals).  If ``None``, defaults to the rates
        used at fit time.
    strategy : {"mean", "median", "point"} or None, default=None
        Strategy for deriving point predictions from prediction intervals
        during recursive multi-step forecasting:

        - ``"mean"``: use the mean of the interval bounds
        - ``"median"``: use the median of the interval bounds
        - ``"point"``: use the point forecast directly (if available)

        If ``None``, defaults to ``"mean"``.
    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``.
    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.
    **params : dict
        Metadata to route to nested estimators.

    Returns
    -------
    pl.DataFrame
        Interval predictions with ``"vintage_time"``, ``"time"``, and
        lower/upper bound columns for each target at each coverage rate.

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

    """
    check_is_fitted(self, ["groups_", "local_y_schema_", "fit_forecasting_horizon_"])
    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, coverage_rates)

    if stride is None:
        stride = self.fit_forecasting_horizon_

    return self._observe_predict_loop(
        predict_fn=self.predict_interval,
        y=y,
        X_actual=X_actual,
        X_future=X_future,
        X_forecast=X_forecast,
        groups=groups,
        stride=stride,
        forecasting_horizon=forecasting_horizon,
        coverage_rates=coverage_rates,
        strategy=strategy,
        **params,
    )

Tutorials

The following example notebooks use this component:

• How to Create a Custom Interval Forecaster

  ---

  Getting-Started

  Implement a NaiveIntervalForecaster from scratch, validate it with the check generator, and compare it against SplitConformalForecaster.

  View · Open in marimo