check_observe_predict_with_step_columns

yohou.testing.forecaster.check_observe_predict_with_step_columns(forecaster, y_train, X_actual_train, y_test, X_actual_test=None, X_future=None, X_forecast=None, forecasting_horizon=3)

Check observe_predict works with step columns (lightweight).

Runs observe_predict with stride=len(y_test)//2 (2 iterations) and validates output structure.

Parameters

Name	Type	Description	Default
forecaster	BaseForecaster	Unfitted forecaster instance.	required
y_train	DataFrame	Training target data.	required
X_actual_train	DataFrame or None	Training features.	required
y_test	DataFrame	Test target data (at least 10 rows).	required
X_actual_test	DataFrame or None	Test features.	None
X_future	DataFrame or None	Known-future features.	None
X_forecast	DataFrame or None	External forecasts.	None
forecasting_horizon	int	Number of steps ahead.	3

Source Code

View on GitHub

Show/Hide source

def check_observe_predict_with_step_columns(
    forecaster,
    y_train: pl.DataFrame,
    X_actual_train: pl.DataFrame | None,
    y_test: pl.DataFrame,
    X_actual_test: pl.DataFrame | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    forecasting_horizon: int = 3,
) -> None:
    """Check observe_predict works with step columns (lightweight).

    Runs observe_predict with stride=len(y_test)//2 (2 iterations) and
    validates output structure.

    Parameters
    ----------
    forecaster : BaseForecaster
        Unfitted forecaster instance.
    y_train : pl.DataFrame
        Training target data.
    X_actual_train : pl.DataFrame or None
        Training features.
    y_test : pl.DataFrame
        Test target data (at least 10 rows).
    X_actual_test : pl.DataFrame or None
        Test features.
    X_future : pl.DataFrame or None
        Known-future features.
    X_forecast : pl.DataFrame or None
        External forecasts.
    forecasting_horizon : int, default=3
        Number of steps ahead.

    """
    forecaster_clone = clone(forecaster)
    forecaster_clone.fit(
        y_train,
        X_actual_train,
        forecasting_horizon=forecasting_horizon,
        X_future=X_future,
        X_forecast=X_forecast,
    )

    stride = max(1, len(y_test) // 2)
    y_pred = forecaster_clone.observe_predict(
        y_test,
        X_actual=X_actual_test,
        forecasting_horizon=forecasting_horizon,
        stride=stride,
        X_future=X_future,
        X_forecast=X_forecast,
    )

    assert isinstance(y_pred, pl.DataFrame), f"observe_predict() must return pl.DataFrame, got {type(y_pred).__name__}"
    assert "time" in y_pred.columns, "observe_predict() output must contain 'time' column"
    assert "vintage_time" in y_pred.columns, "observe_predict() output must contain 'vintage_time' column"
    assert len(y_pred) > 0, "observe_predict() must return non-empty DataFrame"