check_search_rewind_delegates¶

def check_search_rewind_delegates(
    search_cv,
    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() delegates to best_forecaster_.rewind() correctly.

    Parameters
    ----------
    search_cv : BaseSearchCV
        Fitted search CV instance
    y_reset : pl.DataFrame
        Reset target data
    X_actual_reset : pl.DataFrame, optional
        Reset features
    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 rewind() doesn't delegate correctly

    """
    # Capture observed_time before the rewind so we can assert it moved in the
    # direction the rewind actually performs.
    initial_observed_time = search_cv.best_forecaster_.observed_time_

    # Rewind via search CV. rewind() sets observed state to the reset data, so
    # the new observed_time must equal the last time in y_reset regardless of
    # whether that is earlier or later than the pre-rewind state.
    target_time = y_reset["time"][-1]
    search_cv.rewind(y_reset, X_actual_reset, X_future=X_future, X_forecast=X_forecast)

    # Check that best_forecaster_ was rewound to the reset target.
    reset_observed_time = search_cv.best_forecaster_.observed_time_

    if isinstance(initial_observed_time, dict):
        # Panel data case
        for group_name in initial_observed_time:
            assert reset_observed_time[group_name] != initial_observed_time[group_name], (
                f"observed_time for group {group_name} should change after rewind"
            )
            assert reset_observed_time[group_name] == target_time, (
                f"observed_time for group {group_name} should be reset to the last time in y_reset "
                f"({target_time}), got {reset_observed_time[group_name]}"
            )
    else:
        # Non-panel case
        assert reset_observed_time != initial_observed_time, "observed_time should change after rewind"
        assert reset_observed_time == target_time, (
            f"observed_time should be reset to the last time in y_reset ({target_time}), got {reset_observed_time}"
        )