ForecastedFeatureForecaster

yohou.compose.forecasted_feature_forecaster.ForecastedFeatureForecaster

Bases: BaseForecaster

Meta-forecaster that chains feature forecasting into target forecasting.

Fits a feature_forecaster to forecast exogenous features X_actual, then feeds those predicted features into a target_forecaster to predict y.

This is useful when exogenous features are not known in advance at prediction time and must be forecasted first.

Parameters

Name	Type	Description	Default
target_forecaster	BaseForecaster	Forecaster for the target variable y. Receives predicted X_actual at predict time.	required
feature_forecaster	BaseForecaster	Forecaster for exogenous features X_actual. Trained to forecast X_actual as if it were y.	required
strategy	('actual', 'predicted', 'rewind')	Training data strategy for target forecaster: "actual": Train target forecaster on actual X_actual values (perfect foresight). Simpler and uses all data, but may cause train-test mismatch since predict() uses forecasted X_actual. "predicted": Split data and train target forecaster on predicted X_actual values. Requires more data but avoids distribution shift between train and predict. "rewind": Fit feature_forecaster on full data, rewind to observation horizon, then predict X_actual and train target_forecaster on predicted X_actual. Uses all data for feature learning while avoiding distribution shift.	"actual"
split_ratio	float	Fraction of data used to fit feature_forecaster when strategy="predicted". Remaining data used for target_forecaster training with predicted X_actual. Ignored when strategy="actual" or "rewind". Must be in (0, 1).	0.5
panel_strategy	('global', 'multivariate')	How to handle panel data. See BaseForecaster for details.	"global"

Attributes

Name	Type	Description
target_forecaster_	BaseForecaster	Fitted target forecaster.
feature_forecaster_	BaseForecaster	Fitted feature forecaster.
fit_forecasting_horizon_	int	Forecasting horizon used during fit.
interval_	timedelta	Time interval between observations.
groups_	list of str or None	Panel group names if fitted on panel data.

Examples

>>> import polars as pl
>>> from datetime import datetime
>>> from sklearn.linear_model import Ridge
>>> from yohou.compose import ForecastedFeatureForecaster
>>> from yohou.point import PointReductionForecaster
>>>
>>> # Create example time series with exogenous features
>>> time = pl.datetime_range(
...     start=datetime(2020, 1, 1), end=datetime(2020, 3, 31), interval="1d", eager=True
... )
>>> y = pl.DataFrame({"time": time, "sales": range(1, len(time) + 1)})
>>> X_actual = pl.DataFrame({"time": time, "price": [10 + i % 5 for i in range(len(time))]})
>>>
>>> # Create forecaster that predicts price first, then uses it for sales
>>> forecaster = ForecastedFeatureForecaster(
...     target_forecaster=PointReductionForecaster(estimator=Ridge()),
...     feature_forecaster=PointReductionForecaster(estimator=Ridge()),
... )
>>> forecaster.fit(y, X_actual, forecasting_horizon=7)
ForecastedFeatureForecaster(...)
>>> y_pred = forecaster.predict(forecasting_horizon=7)
>>> len(y_pred)
7

Notes

• The feature_forecaster is trained with X_actual as y (forecasting the features)
• The target_forecaster receives forecasted X_actual at predict time
• Use strategy="predicted" when feature forecasts are noisy and you want the target forecaster to learn from similar quality inputs as it will see at prediction time
• At predict time, X_actual can contain known-ahead features (e.g., holidays, promotions) that don't need forecasting. These are merged with the forecasted features before being passed to the target_forecaster.

See Also

• ColumnForecaster : Apply different forecasters to different column subsets.
• DecompositionPipeline : Sequential decomposition into trend + seasonality + residual.

Source Code

View on GitHub

Show/Hide source

class ForecastedFeatureForecaster(BaseForecaster):
    """Meta-forecaster that chains feature forecasting into target forecasting.

    Fits a `feature_forecaster` to forecast exogenous features X_actual, then
    feeds those predicted features into a `target_forecaster` to predict y.

    This is useful when exogenous features are not known in advance at
    prediction time and must be forecasted first.

    Parameters
    ----------
    target_forecaster : BaseForecaster
        Forecaster for the target variable y. Receives predicted X_actual at predict time.
    feature_forecaster : BaseForecaster
        Forecaster for exogenous features X_actual. Trained to forecast X_actual as if it were y.
    strategy : {"actual", "predicted", "rewind"}, default="actual"
        Training data strategy for target forecaster:

        - "actual": Train target forecaster on actual X_actual values (perfect foresight).
          Simpler and uses all data, but may cause train-test mismatch since
          predict() uses forecasted X_actual.
        - "predicted": Split data and train target forecaster on predicted X_actual values.
          Requires more data but avoids distribution shift between train and predict.
        - "rewind": Fit feature_forecaster on full data, rewind to observation horizon,
          then predict X_actual and train target_forecaster on predicted X_actual. Uses all data
          for feature learning while avoiding distribution shift.
    split_ratio : float, default=0.5
        Fraction of data used to fit feature_forecaster when strategy="predicted".
        Remaining data used for target_forecaster training with predicted X_actual.
        Ignored when strategy="actual" or "rewind". Must be in (0, 1).
    panel_strategy : {"global", "multivariate"}, default="global"
        How to handle panel data. See `BaseForecaster` for details.

    Attributes
    ----------
    target_forecaster_ : BaseForecaster
        Fitted target forecaster.
    feature_forecaster_ : BaseForecaster
        Fitted feature forecaster.
    fit_forecasting_horizon_ : int
        Forecasting horizon used during fit.
    interval_ : timedelta
        Time interval between observations.
    groups_ : list of str or None
        Panel group names if fitted on panel data.

    Examples
    --------
    >>> import polars as pl
    >>> from datetime import datetime
    >>> from sklearn.linear_model import Ridge
    >>> from yohou.compose import ForecastedFeatureForecaster
    >>> from yohou.point import PointReductionForecaster
    >>>
    >>> # Create example time series with exogenous features
    >>> time = pl.datetime_range(
    ...     start=datetime(2020, 1, 1), end=datetime(2020, 3, 31), interval="1d", eager=True
    ... )
    >>> y = pl.DataFrame({"time": time, "sales": range(1, len(time) + 1)})
    >>> X_actual = pl.DataFrame({"time": time, "price": [10 + i % 5 for i in range(len(time))]})
    >>>
    >>> # Create forecaster that predicts price first, then uses it for sales
    >>> forecaster = ForecastedFeatureForecaster(
    ...     target_forecaster=PointReductionForecaster(estimator=Ridge()),
    ...     feature_forecaster=PointReductionForecaster(estimator=Ridge()),
    ... )
    >>> forecaster.fit(y, X_actual, forecasting_horizon=7)  # doctest: +ELLIPSIS
    ForecastedFeatureForecaster(...)
    >>> y_pred = forecaster.predict(forecasting_horizon=7)
    >>> len(y_pred)
    7

    Notes
    -----
    - The feature_forecaster is trained with X_actual as y (forecasting the features)
    - The target_forecaster receives forecasted X_actual at predict time
    - Use strategy="predicted" when feature forecasts are noisy and you want
      the target forecaster to learn from similar quality inputs as it will see
      at prediction time
    - At predict time, X_actual can contain known-ahead features (e.g., holidays,
      promotions) that don't need forecasting. These are merged with the
      forecasted features before being passed to the target_forecaster.

    See Also
    --------
    - [`ColumnForecaster`][yohou.compose.column_forecaster.ColumnForecaster] : Apply different forecasters to different column subsets.
    - [`DecompositionPipeline`][yohou.compose.decomposition_pipeline.DecompositionPipeline] : Sequential decomposition into trend + seasonality + residual.

    """

    _parameter_constraints: dict = {
        **BaseForecaster._parameter_constraints,
        "target_forecaster": [BaseForecaster],
        "feature_forecaster": [BaseForecaster],
        "strategy": [StrOptions({"actual", "predicted", "rewind"})],
        "split_ratio": [Interval(numbers.Real, 0.0, 1.0, closed="neither")],
    }

    def __init__(
        self,
        target_forecaster: BaseForecaster,
        feature_forecaster: BaseForecaster,
        *,
        strategy: Literal["actual", "predicted", "rewind"] = "actual",
        split_ratio: float = 0.5,
        panel_strategy: Literal["global", "multivariate"] = "global",
    ):
        super().__init__(panel_strategy=panel_strategy)
        self.target_forecaster = target_forecaster
        self.feature_forecaster = feature_forecaster
        self.strategy = strategy
        self.split_ratio = split_ratio

    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

        # Aggregate forecaster_type from target (determines output type)
        target_tags = self.target_forecaster.__sklearn_tags__()
        if target_tags.forecaster_tags:
            tags.forecaster_tags.forecaster_type = target_tags.forecaster_tags.forecaster_type

        # Aggregate stateful from both
        feature_tags = self.feature_forecaster.__sklearn_tags__()
        target_stateful = target_tags.forecaster_tags.stateful if target_tags.forecaster_tags else False
        feature_stateful = feature_tags.forecaster_tags.stateful if feature_tags.forecaster_tags else False
        tags.forecaster_tags.stateful = target_stateful or feature_stateful

        # Aggregate other tags
        # Note: uses_reduction is False since this meta-forecaster doesn't have an `estimator`
        # attribute directly - child forecasters may use reduction, but that's their internal detail
        tags.forecaster_tags.uses_reduction = False
        tags.forecaster_tags.supports_panel_data = getattr(
            target_tags.forecaster_tags, "supports_panel_data", True
        ) and getattr(feature_tags.forecaster_tags, "supports_panel_data", True)

        # Delegates observation tracking to child forecasters
        tags.forecaster_tags.tracks_observations = False

        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,
    ) -> ForecastedFeatureForecaster:
        """Fit feature and target forecasters.

        Parameters
        ----------
        y : pl.DataFrame
            Target time series with "time" column.
        X_actual : pl.DataFrame
            Actual feature observations with a ``"time"`` column aligned
            with ``y``. Required. The feature forecaster uses these as
            its target variable.
        forecasting_horizon : int, default=1
            Number of steps ahead to forecast.
        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. Bypasses the feature transformer.
        **params : dict
            Metadata routing parameters.

        Returns
        -------
        self
            Fitted forecaster.

        Raises
        ------
        ValueError
            If X_actual is None (exogenous features are required).

        """
        if X_actual is None:
            raise ValueError(
                "ForecastedFeatureForecaster requires X_actual (exogenous features). "
                "Pass exogenous features that need to be forecasted."
            )

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

        # Validate params before routing
        _raise_for_params(params, self, "fit")

        # Process metadata routing
        routed_params = process_routing(self, "fit", **params)

        # Clone forecasters
        self.feature_forecaster_ = clone(self.feature_forecaster)
        self.target_forecaster_ = clone(self.target_forecaster)

        if self.strategy == "actual":
            # Fit feature forecaster: X_actual is treated as y (what to forecast)
            self.feature_forecaster_.fit(
                y=X_actual,
                X_actual=None,
                forecasting_horizon=forecasting_horizon,
                **routed_params.feature_forecaster.fit,
            )

            # Fit target forecaster with actual X_actual
            self.target_forecaster_.fit(
                y=y,
                X_actual=X_actual,
                forecasting_horizon=forecasting_horizon,
                X_future=X_future,
                X_forecast=X_forecast,
                **routed_params.target_forecaster.fit,
            )

        elif self.strategy == "rewind":
            # Fit feature forecaster on full data
            self.feature_forecaster_.fit(
                y=X_actual,
                X_actual=None,
                forecasting_horizon=forecasting_horizon,
                **routed_params.feature_forecaster.fit,
            )

            # Rewind feature forecaster to minimal observation horizon
            obs_horizon = self.feature_forecaster_.observation_horizon
            # Need obs_horizon + 1 rows: obs_horizon for transformer memory + 1 for transform
            rewind_size = obs_horizon + 1
            if obs_horizon <= 0 or len(X_actual) <= rewind_size:
                raise ValueError(
                    f"Cannot use strategy='rewind': observation_horizon={obs_horizon} "
                    f"but len(X_actual)={len(X_actual)}. Need len(X_actual) > observation_horizon + 1 to rewind."
                )
            self.feature_forecaster_.rewind(y=X_actual[:rewind_size], X_actual=None)

            # Predict X_actual from rewind point to end
            X_pred = self.feature_forecaster_.predict(
                forecasting_horizon=len(X_actual) - rewind_size,
            )

            # Prepare X_actual for target forecaster (drop vintage_time, keep time)
            X_for_target = X_pred.drop(["vintage_time"], strict=False)

            # Fit target forecaster with predicted X_actual
            self.target_forecaster_.fit(
                y=y[rewind_size:],
                X_actual=X_for_target,
                forecasting_horizon=forecasting_horizon,
                X_future=X_future,
                X_forecast=X_forecast,
                **routed_params.target_forecaster.fit,
            )

            # Sync feature_forecaster to same observation time as target_forecaster
            self.feature_forecaster_.observe(y=X_actual[rewind_size:], X_actual=None)

        else:  # strategy == "predicted"
            n_split = int(len(y) * self.split_ratio)

            if n_split < 2:
                raise ValueError(
                    f"split_ratio={self.split_ratio} results in n_split={n_split}. "
                    f"Increase split_ratio or provide more data (len(y)={len(y)})."
                )
            if len(y) - n_split < 2:
                raise ValueError(
                    f"split_ratio={self.split_ratio} results in n_split={n_split}, "
                    f"leaving only {len(y) - n_split} rows for target forecaster. "
                    f"Decrease split_ratio to leave at least 2 rows."
                )

            # Fit feature forecaster on first portion
            self.feature_forecaster_.fit(
                y=X_actual[:n_split],
                X_actual=None,
                forecasting_horizon=len(y) - n_split,
                **routed_params.feature_forecaster.fit,
            )

            # Predict X_actual for second portion
            X_pred = self.feature_forecaster_.predict(
                forecasting_horizon=len(y) - n_split,
            )

            # Prepare X_actual for target forecaster (drop time columns)
            X_for_target = X_pred.drop(["vintage_time"], strict=False)

            # Fit target forecaster on second portion with predicted X_actual
            self.target_forecaster_.fit(
                y=y[n_split:],
                X_actual=X_for_target,
                forecasting_horizon=forecasting_horizon,
                X_future=X_future,
                X_forecast=X_forecast,
                **routed_params.target_forecaster.fit,
            )

            # Sync feature_forecaster to the end of training data so both forecasters
            # share the same observed_time_. We feed actual X_actual (not predicted) because:
            # 1. observe() updates the observation buffer/clock, not the model weights.
            # 2. At predict time and during rolling observe_predict, the feature
            #    forecaster should always base its state on actuals.
            # 3. The "predicted" strategy only governs what X_actual the target_forecaster was
            #    *trained* on: the feature_forecaster's runtime state should track reality.
            # TODO: This might be more complicated than just predict+observe. It depends on
            # how predictions are going to be made (e.g., rolling vs single-shot, whether
            # actual X becomes available after prediction, etc.)
            self.feature_forecaster_.observe(y=X_actual[n_split:], X_actual=None)

        # Store standard fitted attributes
        self.fit_forecasting_horizon_ = forecasting_horizon
        self.interval_ = self.target_forecaster_.interval_
        self.groups_ = self.target_forecaster_.groups_
        if hasattr(self.target_forecaster_, "local_y_schema_"):
            self.local_y_schema_ = self.target_forecaster_.local_y_schema_
        if hasattr(self.target_forecaster_, "local_X_actual_schema_"):
            self.local_X_actual_schema_ = self.target_forecaster_.local_X_actual_schema_
        else:
            # Build local_X_actual_schema_ from X_actual columns (excluding time)
            self.local_X_actual_schema_ = {col: X_actual.schema[col] for col in X_actual.columns if col != "time"}
        self.shared_X_actual_schema_ = None  # No shared X_actual features for this forecaster

        # Set transformed schema attributes (no transformation for meta-forecaster)
        self.local_y_t_schema_ = self.local_y_schema_
        self.local_X_t_schema_ = self.local_X_actual_schema_

        # Set observation buffers for observe/rewind
        self._y_observed = y
        self._X_t_observed = X_actual

        return self

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

        Forecasts X_actual using feature_forecaster, then uses those predictions
        as exogenous features for target_forecaster to predict y.

        Parameters
        ----------
        forecasting_horizon : int, optional
            Number of steps ahead to forecast. If None, uses value from fit().
        groups : list of str or None, default=None
            Group prefixes for panel data prediction.
        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 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 without mutating
            forecaster state.
        **params : dict
            Metadata routing parameters.

        Returns
        -------
        pl.DataFrame
            Predictions with "vintage_time", "time", and target columns.

        """
        check_is_fitted(self, ["target_forecaster_", "feature_forecaster_"])

        if forecasting_horizon is None:
            forecasting_horizon = self.fit_forecasting_horizon_

        # Validate params before routing
        _raise_for_params(params, self, "predict")

        # Process metadata routing
        routed_params = process_routing(self, "predict", **params)

        # Predict y using target forecaster (feature predictions flow
        # through observe; at predict time target_forecaster uses its
        # observation window set during fit/observe)
        return self.target_forecaster_.predict(
            forecasting_horizon=forecasting_horizon,
            groups=groups,
            predict_transformed=predict_transformed,
            X_future=X_future,
            X_forecast=X_forecast,
            **routed_params.target_forecaster.predict,
        )

    @available_if(_target_forecaster_has("predict_interval"))
    def predict_interval(
        self,
        forecasting_horizon: StrictInt | None = None,
        coverage_rates: list[float] | None = None,
        groups: list[str] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> pl.DataFrame:
        """Generate interval forecasts.

        Only available if target_forecaster supports interval predictions.
        Feature forecaster always produces point predictions for X_actual.

        Parameters
        ----------
        forecasting_horizon : int, optional
            Number of steps ahead to forecast. If None, uses value from fit().
        coverage_rates : list of float, optional
            Coverage levels for prediction intervals (e.g., [0.9, 0.95]).
        groups : list of str or None, default=None
            Group prefixes for panel data prediction.
        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 without mutating
            forecaster state.
        **params : dict
            Metadata routing parameters.

        Returns
        -------
        pl.DataFrame
            Interval predictions with lower/upper bounds.

        """
        check_is_fitted(self, ["target_forecaster_", "feature_forecaster_"])

        if forecasting_horizon is None:
            forecasting_horizon = self.fit_forecasting_horizon_

        # Validate params before routing
        _raise_for_params(params, self, "predict_interval")

        # Process metadata routing
        routed_params = process_routing(self, "predict_interval", **params)

        return self.target_forecaster_.predict_interval(
            forecasting_horizon=forecasting_horizon,
            coverage_rates=coverage_rates,
            groups=groups,
            X_future=X_future,
            X_forecast=X_forecast,
            **routed_params.target_forecaster.predict_interval,
        )

    def observe(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        groups: list[str] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
    ) -> ForecastedFeatureForecaster:
        """Observe new data for both forecasters.

        Parameters
        ----------
        y : pl.DataFrame
            New target observations with "time" column.
        X_actual : pl.DataFrame or None, default=None
            New actual feature observations with a ``"time"`` column
            aligned with ``y``. Forwarded to both the feature forecaster
            and target forecaster.
        groups : list of str or None, default=None
            Group prefixes for panel data.
        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.

        Returns
        -------
        self
            Forecaster with new observations incorporated.

        """
        check_is_fitted(self, ["target_forecaster_", "feature_forecaster_"])

        # Observe new X_actual for feature forecaster (X_actual is y for feature forecaster)
        if X_actual is not None:
            self.feature_forecaster_.observe(
                y=X_actual,
                X_actual=None,
                groups=groups,
            )

        # Observe new y and X_actual for target forecaster
        self.target_forecaster_.observe(
            y=y,
            X_actual=X_actual,
            groups=groups,
            X_future=X_future,
            X_forecast=X_forecast,
        )

        return self

    def rewind(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        groups: list[str] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
    ) -> ForecastedFeatureForecaster:
        """Rewind both forecasters to last observation_horizon rows.

        Parameters
        ----------
        y : pl.DataFrame
            Target data to rewind to (last observation_horizon rows kept).
        X_actual : pl.DataFrame or None, default=None
            Actual feature observations to restore the observation
            state to. Must align with ``y``.
        groups : list of str or None, default=None
            Group prefixes for panel data.
        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.

        Returns
        -------
        self
            Rewound forecaster.

        """
        check_is_fitted(self, ["target_forecaster_", "feature_forecaster_"])

        # Rewind feature forecaster (X_actual is y for feature forecaster)
        if X_actual is not None:
            self.feature_forecaster_.rewind(
                y=X_actual,
                X_actual=None,
                groups=groups,
            )

        # Rewind target forecaster
        self.target_forecaster_.rewind(
            y=y,
            X_actual=X_actual,
            groups=groups,
            X_future=X_future,
            X_forecast=X_forecast,
        )

        return self

    @available_if(_target_forecaster_has("predict"))
    def observe_predict(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        groups: list[str] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> pl.DataFrame:
        """Observe new data and generate point forecasts.

        Parameters
        ----------
        y : pl.DataFrame
            New target observations with "time" column.
        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.
        groups : list of str or None, default=None
            Group prefixes for panel data.
        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 routing parameters.

        Returns
        -------
        pl.DataFrame
            Point predictions with "vintage_time", "time", and target columns.

        """
        self.observe(y=y, X_actual=X_actual, groups=groups, X_future=X_future, X_forecast=X_forecast)
        return self.predict(groups=groups, X_future=X_future, X_forecast=X_forecast, **params)

    @available_if(_target_forecaster_has("predict_interval"))
    def observe_predict_interval(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        coverage_rates: list[float] | None = None,
        groups: list[str] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> pl.DataFrame:
        """Observe new data and generate interval forecasts.

        Parameters
        ----------
        y : pl.DataFrame
            New target observations with "time" column.
        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.
        coverage_rates : list of float, optional
            Coverage levels for prediction intervals.
        groups : list of str or None, default=None
            Group prefixes for panel data.
        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 routing parameters.

        Returns
        -------
        pl.DataFrame
            Interval predictions with lower/upper bounds.

        """
        self.observe(y=y, X_actual=X_actual, groups=groups, X_future=X_future, X_forecast=X_forecast)
        return self.predict_interval(
            coverage_rates=coverage_rates,
            groups=groups,
            X_future=X_future,
            X_forecast=X_forecast,
            **params,
        )

    @available_if(_target_forecaster_has("predict_class_proba"))
    def predict_class_proba(
        self,
        forecasting_horizon: StrictInt | None = None,
        groups: list[str] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> pl.DataFrame:
        """Generate class-probability forecasts.

        Only available if target_forecaster supports class-probability predictions.
        Feature forecaster always produces point predictions for X_actual.

        Parameters
        ----------
        forecasting_horizon : int, optional
            Number of steps ahead to forecast. If None, uses value from fit().
        groups : list of str or None, default=None
            Group prefixes for panel data prediction.
        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 without mutating
            forecaster state.
        **params : dict
            Metadata routing parameters.

        Returns
        -------
        pl.DataFrame
            Class-probability predictions with "vintage_time", "time", and
            probability columns.

        """
        check_is_fitted(self, ["target_forecaster_", "feature_forecaster_"])

        if forecasting_horizon is None:
            forecasting_horizon = self.fit_forecasting_horizon_

        _raise_for_params(params, self, "predict_class_proba")
        routed_params = process_routing(self, "predict_class_proba", **params)

        return self.target_forecaster_.predict_class_proba(
            forecasting_horizon=forecasting_horizon,
            groups=groups,
            X_future=X_future,
            X_forecast=X_forecast,
            **routed_params.target_forecaster.predict_class_proba,
        )

    @available_if(_target_forecaster_has("predict_class_proba"))
    def observe_predict_class_proba(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        groups: list[str] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> pl.DataFrame:
        """Observe new data and generate class-probability forecasts.

        Parameters
        ----------
        y : pl.DataFrame
            New target observations with "time" column.
        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.
        groups : list of str or None, default=None
            Group prefixes for panel data.
        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 routing parameters.

        Returns
        -------
        pl.DataFrame
            Class-probability predictions.

        """
        self.observe(y=y, X_actual=X_actual, groups=groups, X_future=X_future, X_forecast=X_forecast)
        return self.predict_class_proba(
            groups=groups,
            X_future=X_future,
            X_forecast=X_forecast,
            **params,
        )

    def get_metadata_routing(self):
        """Get metadata routing for both forecasters.

        Returns
        -------
        MetadataRouter
            Router with mappings for target_forecaster and feature_forecaster.

        """
        router = MetadataRouter(owner=self.__class__.__name__)

        router.add(
            target_forecaster=self.target_forecaster,
            method_mapping=MethodMapping()
            .add(caller="fit", callee="fit")
            .add(caller="predict", callee="predict")
            .add(caller="predict_interval", callee="predict_interval")
            .add(caller="predict_class_proba", callee="predict_class_proba")
            .add(caller="observe_predict", callee="observe_predict")
            .add(caller="observe_predict_interval", callee="observe_predict_interval")
            .add(caller="observe_predict_class_proba", callee="observe_predict_class_proba"),
        )

        router.add(
            feature_forecaster=self.feature_forecaster,
            method_mapping=MethodMapping()
            .add(caller="fit", callee="fit")
            .add(caller="predict", callee="predict")
            .add(caller="predict_interval", callee="predict")
            .add(caller="predict_class_proba", callee="predict")
            .add(caller="observe_predict", callee="observe_predict")
            .add(caller="observe_predict_interval", callee="observe_predict")
            .add(caller="observe_predict_class_proba", callee="observe_predict"),
        )

        return router

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

    # Aggregate forecaster_type from target (determines output type)
    target_tags = self.target_forecaster.__sklearn_tags__()
    if target_tags.forecaster_tags:
        tags.forecaster_tags.forecaster_type = target_tags.forecaster_tags.forecaster_type

    # Aggregate stateful from both
    feature_tags = self.feature_forecaster.__sklearn_tags__()
    target_stateful = target_tags.forecaster_tags.stateful if target_tags.forecaster_tags else False
    feature_stateful = feature_tags.forecaster_tags.stateful if feature_tags.forecaster_tags else False
    tags.forecaster_tags.stateful = target_stateful or feature_stateful

    # Aggregate other tags
    # Note: uses_reduction is False since this meta-forecaster doesn't have an `estimator`
    # attribute directly - child forecasters may use reduction, but that's their internal detail
    tags.forecaster_tags.uses_reduction = False
    tags.forecaster_tags.supports_panel_data = getattr(
        target_tags.forecaster_tags, "supports_panel_data", True
    ) and getattr(feature_tags.forecaster_tags, "supports_panel_data", True)

    # Delegates observation tracking to child forecasters
    tags.forecaster_tags.tracks_observations = False

    return tags

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

Fit feature and target forecasters.

Parameters

Name	Type	Description	Default
y	DataFrame	Target time series with "time" column.	required
X_actual	DataFrame	Actual feature observations with a "time" column aligned with y. Required. The feature forecaster uses these as its target variable.	None
forecasting_horizon	int	Number of steps ahead to forecast.	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. Bypasses the feature transformer.	None
**params	dict	Metadata routing parameters.	{}

Returns

Type	Description
self	Fitted forecaster.

Raises

Type	Description
ValueError	If X_actual is None (exogenous features are required).

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,
) -> ForecastedFeatureForecaster:
    """Fit feature and target forecasters.

    Parameters
    ----------
    y : pl.DataFrame
        Target time series with "time" column.
    X_actual : pl.DataFrame
        Actual feature observations with a ``"time"`` column aligned
        with ``y``. Required. The feature forecaster uses these as
        its target variable.
    forecasting_horizon : int, default=1
        Number of steps ahead to forecast.
    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. Bypasses the feature transformer.
    **params : dict
        Metadata routing parameters.

    Returns
    -------
    self
        Fitted forecaster.

    Raises
    ------
    ValueError
        If X_actual is None (exogenous features are required).

    """
    if X_actual is None:
        raise ValueError(
            "ForecastedFeatureForecaster requires X_actual (exogenous features). "
            "Pass exogenous features that need to be forecasted."
        )

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

    # Validate params before routing
    _raise_for_params(params, self, "fit")

    # Process metadata routing
    routed_params = process_routing(self, "fit", **params)

    # Clone forecasters
    self.feature_forecaster_ = clone(self.feature_forecaster)
    self.target_forecaster_ = clone(self.target_forecaster)

    if self.strategy == "actual":
        # Fit feature forecaster: X_actual is treated as y (what to forecast)
        self.feature_forecaster_.fit(
            y=X_actual,
            X_actual=None,
            forecasting_horizon=forecasting_horizon,
            **routed_params.feature_forecaster.fit,
        )

        # Fit target forecaster with actual X_actual
        self.target_forecaster_.fit(
            y=y,
            X_actual=X_actual,
            forecasting_horizon=forecasting_horizon,
            X_future=X_future,
            X_forecast=X_forecast,
            **routed_params.target_forecaster.fit,
        )

    elif self.strategy == "rewind":
        # Fit feature forecaster on full data
        self.feature_forecaster_.fit(
            y=X_actual,
            X_actual=None,
            forecasting_horizon=forecasting_horizon,
            **routed_params.feature_forecaster.fit,
        )

        # Rewind feature forecaster to minimal observation horizon
        obs_horizon = self.feature_forecaster_.observation_horizon
        # Need obs_horizon + 1 rows: obs_horizon for transformer memory + 1 for transform
        rewind_size = obs_horizon + 1
        if obs_horizon <= 0 or len(X_actual) <= rewind_size:
            raise ValueError(
                f"Cannot use strategy='rewind': observation_horizon={obs_horizon} "
                f"but len(X_actual)={len(X_actual)}. Need len(X_actual) > observation_horizon + 1 to rewind."
            )
        self.feature_forecaster_.rewind(y=X_actual[:rewind_size], X_actual=None)

        # Predict X_actual from rewind point to end
        X_pred = self.feature_forecaster_.predict(
            forecasting_horizon=len(X_actual) - rewind_size,
        )

        # Prepare X_actual for target forecaster (drop vintage_time, keep time)
        X_for_target = X_pred.drop(["vintage_time"], strict=False)

        # Fit target forecaster with predicted X_actual
        self.target_forecaster_.fit(
            y=y[rewind_size:],
            X_actual=X_for_target,
            forecasting_horizon=forecasting_horizon,
            X_future=X_future,
            X_forecast=X_forecast,
            **routed_params.target_forecaster.fit,
        )

        # Sync feature_forecaster to same observation time as target_forecaster
        self.feature_forecaster_.observe(y=X_actual[rewind_size:], X_actual=None)

    else:  # strategy == "predicted"
        n_split = int(len(y) * self.split_ratio)

        if n_split < 2:
            raise ValueError(
                f"split_ratio={self.split_ratio} results in n_split={n_split}. "
                f"Increase split_ratio or provide more data (len(y)={len(y)})."
            )
        if len(y) - n_split < 2:
            raise ValueError(
                f"split_ratio={self.split_ratio} results in n_split={n_split}, "
                f"leaving only {len(y) - n_split} rows for target forecaster. "
                f"Decrease split_ratio to leave at least 2 rows."
            )

        # Fit feature forecaster on first portion
        self.feature_forecaster_.fit(
            y=X_actual[:n_split],
            X_actual=None,
            forecasting_horizon=len(y) - n_split,
            **routed_params.feature_forecaster.fit,
        )

        # Predict X_actual for second portion
        X_pred = self.feature_forecaster_.predict(
            forecasting_horizon=len(y) - n_split,
        )

        # Prepare X_actual for target forecaster (drop time columns)
        X_for_target = X_pred.drop(["vintage_time"], strict=False)

        # Fit target forecaster on second portion with predicted X_actual
        self.target_forecaster_.fit(
            y=y[n_split:],
            X_actual=X_for_target,
            forecasting_horizon=forecasting_horizon,
            X_future=X_future,
            X_forecast=X_forecast,
            **routed_params.target_forecaster.fit,
        )

        # Sync feature_forecaster to the end of training data so both forecasters
        # share the same observed_time_. We feed actual X_actual (not predicted) because:
        # 1. observe() updates the observation buffer/clock, not the model weights.
        # 2. At predict time and during rolling observe_predict, the feature
        #    forecaster should always base its state on actuals.
        # 3. The "predicted" strategy only governs what X_actual the target_forecaster was
        #    *trained* on: the feature_forecaster's runtime state should track reality.
        # TODO: This might be more complicated than just predict+observe. It depends on
        # how predictions are going to be made (e.g., rolling vs single-shot, whether
        # actual X becomes available after prediction, etc.)
        self.feature_forecaster_.observe(y=X_actual[n_split:], X_actual=None)

    # Store standard fitted attributes
    self.fit_forecasting_horizon_ = forecasting_horizon
    self.interval_ = self.target_forecaster_.interval_
    self.groups_ = self.target_forecaster_.groups_
    if hasattr(self.target_forecaster_, "local_y_schema_"):
        self.local_y_schema_ = self.target_forecaster_.local_y_schema_
    if hasattr(self.target_forecaster_, "local_X_actual_schema_"):
        self.local_X_actual_schema_ = self.target_forecaster_.local_X_actual_schema_
    else:
        # Build local_X_actual_schema_ from X_actual columns (excluding time)
        self.local_X_actual_schema_ = {col: X_actual.schema[col] for col in X_actual.columns if col != "time"}
    self.shared_X_actual_schema_ = None  # No shared X_actual features for this forecaster

    # Set transformed schema attributes (no transformation for meta-forecaster)
    self.local_y_t_schema_ = self.local_y_schema_
    self.local_X_t_schema_ = self.local_X_actual_schema_

    # Set observation buffers for observe/rewind
    self._y_observed = y
    self._X_t_observed = X_actual

    return self

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

Generate point forecasts.

Forecasts X_actual using feature_forecaster, then uses those predictions as exogenous features for target_forecaster to predict y.

Parameters

Name	Type	Description	Default
forecasting_horizon	int	Number of steps ahead to forecast. If None, uses value from fit().	None
groups	list of str or None	Group prefixes for panel data prediction.	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 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 without mutating forecaster state.	None
**params	dict	Metadata routing parameters.	{}

Returns

Type	Description
DataFrame	Predictions with "vintage_time", "time", and target columns.

Source Code

View on GitHub

Show/Hide source

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

    Forecasts X_actual using feature_forecaster, then uses those predictions
    as exogenous features for target_forecaster to predict y.

    Parameters
    ----------
    forecasting_horizon : int, optional
        Number of steps ahead to forecast. If None, uses value from fit().
    groups : list of str or None, default=None
        Group prefixes for panel data prediction.
    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 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 without mutating
        forecaster state.
    **params : dict
        Metadata routing parameters.

    Returns
    -------
    pl.DataFrame
        Predictions with "vintage_time", "time", and target columns.

    """
    check_is_fitted(self, ["target_forecaster_", "feature_forecaster_"])

    if forecasting_horizon is None:
        forecasting_horizon = self.fit_forecasting_horizon_

    # Validate params before routing
    _raise_for_params(params, self, "predict")

    # Process metadata routing
    routed_params = process_routing(self, "predict", **params)

    # Predict y using target forecaster (feature predictions flow
    # through observe; at predict time target_forecaster uses its
    # observation window set during fit/observe)
    return self.target_forecaster_.predict(
        forecasting_horizon=forecasting_horizon,
        groups=groups,
        predict_transformed=predict_transformed,
        X_future=X_future,
        X_forecast=X_forecast,
        **routed_params.target_forecaster.predict,
    )

predict_interval(forecasting_horizon=None, coverage_rates=None, groups=None, X_future=None, X_forecast=None, **params)

Generate interval forecasts.

Only available if target_forecaster supports interval predictions. Feature forecaster always produces point predictions for X_actual.

Parameters

Name	Type	Description	Default
forecasting_horizon	int	Number of steps ahead to forecast. If None, uses value from fit().	None
coverage_rates	list of float	Coverage levels for prediction intervals (e.g., [0.9, 0.95]).	None
groups	list of str or None	Group prefixes for panel data prediction.	None
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 without mutating forecaster state.	None
**params	dict	Metadata routing parameters.	{}

Returns

Type	Description
DataFrame	Interval predictions with lower/upper bounds.

Source Code

View on GitHub

Show/Hide source

@available_if(_target_forecaster_has("predict_interval"))
def predict_interval(
    self,
    forecasting_horizon: StrictInt | None = None,
    coverage_rates: list[float] | None = None,
    groups: list[str] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> pl.DataFrame:
    """Generate interval forecasts.

    Only available if target_forecaster supports interval predictions.
    Feature forecaster always produces point predictions for X_actual.

    Parameters
    ----------
    forecasting_horizon : int, optional
        Number of steps ahead to forecast. If None, uses value from fit().
    coverage_rates : list of float, optional
        Coverage levels for prediction intervals (e.g., [0.9, 0.95]).
    groups : list of str or None, default=None
        Group prefixes for panel data prediction.
    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 without mutating
        forecaster state.
    **params : dict
        Metadata routing parameters.

    Returns
    -------
    pl.DataFrame
        Interval predictions with lower/upper bounds.

    """
    check_is_fitted(self, ["target_forecaster_", "feature_forecaster_"])

    if forecasting_horizon is None:
        forecasting_horizon = self.fit_forecasting_horizon_

    # Validate params before routing
    _raise_for_params(params, self, "predict_interval")

    # Process metadata routing
    routed_params = process_routing(self, "predict_interval", **params)

    return self.target_forecaster_.predict_interval(
        forecasting_horizon=forecasting_horizon,
        coverage_rates=coverage_rates,
        groups=groups,
        X_future=X_future,
        X_forecast=X_forecast,
        **routed_params.target_forecaster.predict_interval,
    )

observe(y, X_actual=None, groups=None, X_future=None, X_forecast=None)

Observe new data for both forecasters.

Parameters

Name	Type	Description	Default
y	DataFrame	New target observations with "time" column.	required
X_actual	DataFrame or None	New actual feature observations with a "time" column aligned with y. Forwarded to both the feature forecaster and target forecaster.	None
groups	list of str or None	Group prefixes for panel data.	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

Returns

Type	Description
self	Forecaster with new observations incorporated.

Source Code

View on GitHub

Show/Hide source

def observe(
    self,
    y: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
    groups: list[str] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
) -> ForecastedFeatureForecaster:
    """Observe new data for both forecasters.

    Parameters
    ----------
    y : pl.DataFrame
        New target observations with "time" column.
    X_actual : pl.DataFrame or None, default=None
        New actual feature observations with a ``"time"`` column
        aligned with ``y``. Forwarded to both the feature forecaster
        and target forecaster.
    groups : list of str or None, default=None
        Group prefixes for panel data.
    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.

    Returns
    -------
    self
        Forecaster with new observations incorporated.

    """
    check_is_fitted(self, ["target_forecaster_", "feature_forecaster_"])

    # Observe new X_actual for feature forecaster (X_actual is y for feature forecaster)
    if X_actual is not None:
        self.feature_forecaster_.observe(
            y=X_actual,
            X_actual=None,
            groups=groups,
        )

    # Observe new y and X_actual for target forecaster
    self.target_forecaster_.observe(
        y=y,
        X_actual=X_actual,
        groups=groups,
        X_future=X_future,
        X_forecast=X_forecast,
    )

    return self

rewind(y, X_actual=None, groups=None, X_future=None, X_forecast=None)

Rewind both forecasters to last observation_horizon rows.

Parameters

Name	Type	Description	Default
y	DataFrame	Target data to rewind to (last observation_horizon rows kept).	required
X_actual	DataFrame or None	Actual feature observations to restore the observation state to. Must align with y.	None
groups	list of str or None	Group prefixes for panel data.	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

Returns

Type	Description
self	Rewound forecaster.

Source Code

View on GitHub

Show/Hide source

def rewind(
    self,
    y: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
    groups: list[str] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
) -> ForecastedFeatureForecaster:
    """Rewind both forecasters to last observation_horizon rows.

    Parameters
    ----------
    y : pl.DataFrame
        Target data to rewind to (last observation_horizon rows kept).
    X_actual : pl.DataFrame or None, default=None
        Actual feature observations to restore the observation
        state to. Must align with ``y``.
    groups : list of str or None, default=None
        Group prefixes for panel data.
    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.

    Returns
    -------
    self
        Rewound forecaster.

    """
    check_is_fitted(self, ["target_forecaster_", "feature_forecaster_"])

    # Rewind feature forecaster (X_actual is y for feature forecaster)
    if X_actual is not None:
        self.feature_forecaster_.rewind(
            y=X_actual,
            X_actual=None,
            groups=groups,
        )

    # Rewind target forecaster
    self.target_forecaster_.rewind(
        y=y,
        X_actual=X_actual,
        groups=groups,
        X_future=X_future,
        X_forecast=X_forecast,
    )

    return self

observe_predict(y, X_actual=None, groups=None, X_future=None, X_forecast=None, **params)

Observe new data and generate point forecasts.

Parameters

Name	Type	Description	Default
y	DataFrame	New target observations with "time" column.	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
groups	list of str or None	Group prefixes for panel data.	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 routing parameters.	{}

Returns

Type	Description
DataFrame	Point predictions with "vintage_time", "time", and target columns.

Source Code

View on GitHub

Show/Hide source

@available_if(_target_forecaster_has("predict"))
def observe_predict(
    self,
    y: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
    groups: list[str] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> pl.DataFrame:
    """Observe new data and generate point forecasts.

    Parameters
    ----------
    y : pl.DataFrame
        New target observations with "time" column.
    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.
    groups : list of str or None, default=None
        Group prefixes for panel data.
    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 routing parameters.

    Returns
    -------
    pl.DataFrame
        Point predictions with "vintage_time", "time", and target columns.

    """
    self.observe(y=y, X_actual=X_actual, groups=groups, X_future=X_future, X_forecast=X_forecast)
    return self.predict(groups=groups, X_future=X_future, X_forecast=X_forecast, **params)

observe_predict_interval(y, X_actual=None, coverage_rates=None, groups=None, X_future=None, X_forecast=None, **params)

Observe new data and generate interval forecasts.

Parameters

Name	Type	Description	Default
y	DataFrame	New target observations with "time" column.	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
coverage_rates	list of float	Coverage levels for prediction intervals.	None
groups	list of str or None	Group prefixes for panel data.	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 routing parameters.	{}

Returns

Type	Description
DataFrame	Interval predictions with lower/upper bounds.

Source Code

View on GitHub

Show/Hide source

@available_if(_target_forecaster_has("predict_interval"))
def observe_predict_interval(
    self,
    y: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
    coverage_rates: list[float] | None = None,
    groups: list[str] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> pl.DataFrame:
    """Observe new data and generate interval forecasts.

    Parameters
    ----------
    y : pl.DataFrame
        New target observations with "time" column.
    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.
    coverage_rates : list of float, optional
        Coverage levels for prediction intervals.
    groups : list of str or None, default=None
        Group prefixes for panel data.
    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 routing parameters.

    Returns
    -------
    pl.DataFrame
        Interval predictions with lower/upper bounds.

    """
    self.observe(y=y, X_actual=X_actual, groups=groups, X_future=X_future, X_forecast=X_forecast)
    return self.predict_interval(
        coverage_rates=coverage_rates,
        groups=groups,
        X_future=X_future,
        X_forecast=X_forecast,
        **params,
    )

predict_class_proba(forecasting_horizon=None, groups=None, X_future=None, X_forecast=None, **params)

Generate class-probability forecasts.

Only available if target_forecaster supports class-probability predictions. Feature forecaster always produces point predictions for X_actual.

Parameters

Name	Type	Description	Default
forecasting_horizon	int	Number of steps ahead to forecast. If None, uses value from fit().	None
groups	list of str or None	Group prefixes for panel data prediction.	None
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 without mutating forecaster state.	None
**params	dict	Metadata routing parameters.	{}

Returns

Type	Description
DataFrame	Class-probability predictions with "vintage_time", "time", and probability columns.

Source Code

View on GitHub

Show/Hide source

@available_if(_target_forecaster_has("predict_class_proba"))
def predict_class_proba(
    self,
    forecasting_horizon: StrictInt | None = None,
    groups: list[str] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> pl.DataFrame:
    """Generate class-probability forecasts.

    Only available if target_forecaster supports class-probability predictions.
    Feature forecaster always produces point predictions for X_actual.

    Parameters
    ----------
    forecasting_horizon : int, optional
        Number of steps ahead to forecast. If None, uses value from fit().
    groups : list of str or None, default=None
        Group prefixes for panel data prediction.
    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 without mutating
        forecaster state.
    **params : dict
        Metadata routing parameters.

    Returns
    -------
    pl.DataFrame
        Class-probability predictions with "vintage_time", "time", and
        probability columns.

    """
    check_is_fitted(self, ["target_forecaster_", "feature_forecaster_"])

    if forecasting_horizon is None:
        forecasting_horizon = self.fit_forecasting_horizon_

    _raise_for_params(params, self, "predict_class_proba")
    routed_params = process_routing(self, "predict_class_proba", **params)

    return self.target_forecaster_.predict_class_proba(
        forecasting_horizon=forecasting_horizon,
        groups=groups,
        X_future=X_future,
        X_forecast=X_forecast,
        **routed_params.target_forecaster.predict_class_proba,
    )

observe_predict_class_proba(y, X_actual=None, groups=None, X_future=None, X_forecast=None, **params)

Observe new data and generate class-probability forecasts.

Parameters

Name	Type	Description	Default
y	DataFrame	New target observations with "time" column.	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
groups	list of str or None	Group prefixes for panel data.	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 routing parameters.	{}

Returns

Type	Description
DataFrame	Class-probability predictions.

Source Code

View on GitHub

Show/Hide source

@available_if(_target_forecaster_has("predict_class_proba"))
def observe_predict_class_proba(
    self,
    y: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
    groups: list[str] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> pl.DataFrame:
    """Observe new data and generate class-probability forecasts.

    Parameters
    ----------
    y : pl.DataFrame
        New target observations with "time" column.
    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.
    groups : list of str or None, default=None
        Group prefixes for panel data.
    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 routing parameters.

    Returns
    -------
    pl.DataFrame
        Class-probability predictions.

    """
    self.observe(y=y, X_actual=X_actual, groups=groups, X_future=X_future, X_forecast=X_forecast)
    return self.predict_class_proba(
        groups=groups,
        X_future=X_future,
        X_forecast=X_forecast,
        **params,
    )

get_metadata_routing()

Get metadata routing for both forecasters.

Returns

Type	Description
MetadataRouter	Router with mappings for target_forecaster and feature_forecaster.

Source Code

View on GitHub

Show/Hide source

def get_metadata_routing(self):
    """Get metadata routing for both forecasters.

    Returns
    -------
    MetadataRouter
        Router with mappings for target_forecaster and feature_forecaster.

    """
    router = MetadataRouter(owner=self.__class__.__name__)

    router.add(
        target_forecaster=self.target_forecaster,
        method_mapping=MethodMapping()
        .add(caller="fit", callee="fit")
        .add(caller="predict", callee="predict")
        .add(caller="predict_interval", callee="predict_interval")
        .add(caller="predict_class_proba", callee="predict_class_proba")
        .add(caller="observe_predict", callee="observe_predict")
        .add(caller="observe_predict_interval", callee="observe_predict_interval")
        .add(caller="observe_predict_class_proba", callee="observe_predict_class_proba"),
    )

    router.add(
        feature_forecaster=self.feature_forecaster,
        method_mapping=MethodMapping()
        .add(caller="fit", callee="fit")
        .add(caller="predict", callee="predict")
        .add(caller="predict_interval", callee="predict")
        .add(caller="predict_class_proba", callee="predict")
        .add(caller="observe_predict", callee="observe_predict")
        .add(caller="observe_predict_interval", callee="observe_predict")
        .add(caller="observe_predict_class_proba", callee="observe_predict"),
    )

    return router

Tutorials

The following example notebooks use this component:

• How to Build a Lag-Feature Forecaster

  ---

  Forecasting-Models

  Chain feature and target forecasters with ForecastedFeatureForecaster when exogenous variables are unknown at prediction time and must be forecasted.

  View · Open in marimo

• How to Use Lagged Forecasts as Features

  ---

  Forecasting-Models

  Compare ForecastedFeatureForecaster strategies (actual, predicted, rewind) and split ratio tuning for chaining feature and target forecasters.

  View · Open in marimo