check_rewind_updates_memory¶

def check_rewind_updates_memory(transformer, X: pl.DataFrame, y: pl.DataFrame | None = None) -> None:
    """Check rewind(X) sets _X_observed to X.tail(observation_horizon).

    The rewind() method should update the transformer's memory to contain
    only the last observation_horizon rows of the provided data.

    Parameters
    ----------
    transformer : BaseActualTransformer
        Unfitted transformer
    X : pl.DataFrame
        Training data (should have at least observation_horizon rows)
    y : pl.DataFrame, optional
        Target data

    Raises
    ------
    AssertionError
        If _X_observed is not properly updated

    Notes
    -----
    For stateless transformers (observation_horizon == 0) the expected memory
    is an empty frame, so the assertions hold trivially; no memory behavior is
    exercised. The check is also a no-op when len(X) < observation_horizon.

    """
    transformer_clone = clone(transformer)
    transformer_clone.fit(X, y)

    horizon = transformer_clone.observation_horizon

    if len(X) < horizon:  # pragma: no cover - the check suite always supplies enough rows for the horizon
        # Precondition cannot be met with this data; skip like the other checks.
        return

    # Create new data to reset with
    X_new = X.head(horizon + 5) if len(X) >= horizon + 5 else X
    transformer_clone.rewind(X_new)

    # Check _X_observed has correct length
    assert len(transformer_clone._X_observed) == min(horizon, len(X_new)), (
        f"_X_observed length should be {min(horizon, len(X_new))}, got {len(transformer_clone._X_observed)}"
    )

    # Check _X_observed contains last horizon rows
    expected = X_new.tail(horizon)
    assert_frame_equal(transformer_clone._X_observed, expected)