check_interval_bounds¶

def check_interval_bounds(forecaster, y_test: pl.DataFrame) -> None:
    """Check upper >= lower for all coverage rates and time steps.

    Parameters
    ----------
    forecaster : BaseIntervalForecaster
        Fitted interval forecaster instance
    y_test : pl.DataFrame
        Test target data

    Raises
    ------
    AssertionError
        If upper bounds are less than lower bounds

    """
    forecasting_horizon = min(3, len(y_test))
    y_pred = forecaster.predict_interval(forecasting_horizon=forecasting_horizon)

    coverage_rates = forecaster.fit_coverage_rates_

    # Check if we have panel data (columns with __ separator)
    _, y_panel_groups = inspect_panel(y_test)

    if len(y_panel_groups) > 0:
        # For panel data, interval columns use __ separator (e.g. "stores__store_0").
        fields = [field for group_prefix in y_panel_groups for field in y_panel_groups[group_prefix]]
    else:
        fields = list(forecaster.local_y_schema_.keys())

    # Build one violation flag per (field, rate) pair and evaluate in a single pass.
    violation_exprs = [
        (pl.col(f"{field}_lower_{rate}") > pl.col(f"{field}_upper_{rate}")).any().alias(f"{field}_lower_{rate}")
        for rate in coverage_rates
        for field in fields
    ]
    violations = y_pred.select(violation_exprs)
    offenders = [name for name, flagged in violations.row(0, named=True).items() if flagged]
    if offenders:
        raise AssertionError(f"Found bounds where lower > upper: {offenders}")