check_rewind_propagates_to_transformers¶

def check_rewind_propagates_to_transformers(
    forecaster,
    y_reset: pl.DataFrame,
    X_actual_reset: pl.DataFrame | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
) -> None:
    """Check rewind() propagates to transformers in forecaster.

    When a forecaster with transformers calls rewind(), the transformers
    should also have their observation buffers reset accordingly.

    Parameters
    ----------
    forecaster : BaseForecaster
        Fitted forecaster instance with transformers
    y_reset : pl.DataFrame
        New data for reset
    X_actual_reset : pl.DataFrame, optional
        Features for reset
    X_future : pl.DataFrame, optional
        Known-future features forwarded to rewind()
    X_forecast : pl.DataFrame, optional
        External forecast features forwarded to rewind()

    Raises
    ------
    AssertionError
        If transformers are not properly reset

    """
    # Check if forecaster has transformers (target_transformer or actual_transformer)
    if not hasattr(forecaster, "target_transformer_") and not hasattr(
        forecaster, "actual_transformer_"
    ):  # pragma: no branch - the generator only yields this check for transformer-using forecasters
        return  # Nothing to check

    # Rewind the forecaster
    forecaster.rewind(y_reset, X_actual=X_actual_reset, X_future=X_future, X_forecast=X_forecast)

    # Check target transformer is reset
    if hasattr(forecaster, "target_transformer_") and forecaster.target_transformer_ is not None:
        if isinstance(forecaster.target_transformer_, dict):
            # Panel data - check each transformer
            for group_name, transformer in forecaster.target_transformer_.items():
                if (
                    hasattr(transformer, "_X_observed")
                    and transformer._X_observed is not None
                    and getattr(transformer, "observation_horizon", 0) > 0
                ):
                    # Transformer should have observation data matching reset data
                    assert len(transformer._X_observed) > 0, (
                        f"Target transformer for group '{group_name}' should have observations after rewind"
                    )
        # Non-panel data
        elif (
            hasattr(forecaster.target_transformer_, "_X_observed")
            and forecaster.target_transformer_._X_observed is not None
            and getattr(forecaster.target_transformer_, "observation_horizon", 0) > 0
        ):
            assert len(forecaster.target_transformer_._X_observed) > 0, (
                "Target transformer should have observations after rewind"
            )

    # Check actual transformer is reset (if exists)
    if hasattr(forecaster, "actual_transformer_") and forecaster.actual_transformer_ is not None:
        if isinstance(forecaster.actual_transformer_, dict):
            # Panel data - check each transformer
            for group_name, transformer in forecaster.actual_transformer_.items():
                if (
                    hasattr(transformer, "_X_observed")
                    and transformer._X_observed is not None
                    and getattr(transformer, "observation_horizon", 0) > 0
                ):
                    assert len(transformer._X_observed) > 0, (
                        f"Actual transformer for group '{group_name}' should have observations after rewind"
                    )
        # Non-panel data
        elif (
            hasattr(forecaster.actual_transformer_, "_X_observed")
            and forecaster.actual_transformer_._X_observed is not None
            and getattr(forecaster.actual_transformer_, "observation_horizon", 0) > 0
        ):
            assert len(forecaster.actual_transformer_._X_observed) > 0, (
                "Actual transformer should have observations after rewind"
            )