From c8079b2a10fbfb97e333c89948eb234fc2381e9d Mon Sep 17 00:00:00 2001 From: bartzbeielstein <32470350+bartzbeielstein@users.noreply.github.com> Date: Fri, 12 Jun 2026 02:50:19 +0200 Subject: [PATCH 1/2] feat(stats): PACF lag selection + search-space boundary report - stats/autocorrelation.py: add select_pacf_lags(), ported from bart26k-lecture/scripts/team4_4zones_submit.py (select_key_lags). Computes PACF up to n_lags, returns top_k significant lags sorted ascending; falls back to caller-supplied list or raises ValueError on degenerate series. - model_selection/boundary.py (new): three search-space boundary helpers promoted from bart26k-lecture/14_team_4_submission.qmd and team4_4zones_submit.py (KB entry 2026-06-08-hyperparameter-boundary- management). report_boundary_positions() logs and returns flagged dims; boundary_report() returns a DataFrame; suggest_bounds() returns a widened search space ready for a round-2 run_task_spotoptim() call. - tests/test_stats_select_pacf_lags.py: 15 tests (AR(1), AR(24), degenerate/constant, fallback, determinism, top_k bounds). - tests/test_model_selection_boundary.py: 27 tests (linear/log10 dims, prefix stripping, bool skip, categorical skip, missing key, logger injection, suggest_bounds widening for float/log/int). - _quarto.yml: register stats.select_pacf_lags and model_selection.boundary in sidebar and quartodoc sections. Co-Authored-By: Claude Fable 5 --- _quarto.yml | 8 +- src/spotforecast2/model_selection/__init__.py | 4 + src/spotforecast2/model_selection/boundary.py | 376 ++++++++++++++++++ src/spotforecast2/stats/__init__.py | 3 +- src/spotforecast2/stats/autocorrelation.py | 131 ++++++ tests/test_model_selection_boundary.py | 284 +++++++++++++ tests/test_stats_select_pacf_lags.py | 159 ++++++++ uv.lock | 2 +- 8 files changed, 964 insertions(+), 3 deletions(-) create mode 100644 src/spotforecast2/model_selection/boundary.py create mode 100644 tests/test_model_selection_boundary.py create mode 100644 tests/test_stats_select_pacf_lags.py diff --git a/_quarto.yml b/_quarto.yml index 63015e8e..298f21f1 100644 --- a/_quarto.yml +++ b/_quarto.yml @@ -106,6 +106,8 @@ website: contents: - text: "bayesian_search" file: docs/reference/model_selection.bayesian_search.qmd + - text: "boundary" + file: docs/reference/model_selection.boundary.qmd - text: "grid_search" file: docs/reference/model_selection.grid_search.qmd - text: "random_search" @@ -126,6 +128,8 @@ website: contents: - text: "autocorrelation" file: docs/reference/stats.autocorrelation.qmd + - text: "select_pacf_lags" + file: docs/reference/stats.select_pacf_lags.qmd - section: "Tasks" contents: - text: "task_demo" @@ -222,9 +226,10 @@ quartodoc: - plots.time_series_visualization - title: "Model Selection" - desc: "Search algorithms and cross-validation tools." + desc: "Search algorithms, cross-validation tools, and boundary management." contents: - model_selection.bayesian_search + - model_selection.boundary - model_selection.grid_search - model_selection.random_search - model_selection.spotoptim_search @@ -256,6 +261,7 @@ quartodoc: desc: "Statistical analysis tools." contents: - stats.autocorrelation + - stats.select_pacf_lags - title: "Tasks" desc: "Demonstration and predefined forecasting tasks." diff --git a/src/spotforecast2/model_selection/__init__.py b/src/spotforecast2/model_selection/__init__.py index 5f133488..69102ef1 100644 --- a/src/spotforecast2/model_selection/__init__.py +++ b/src/spotforecast2/model_selection/__init__.py @@ -2,6 +2,7 @@ # SPDX-License-Identifier: AGPL-3.0-or-later from .bayesian_search import bayesian_search_forecaster +from .boundary import boundary_report, report_boundary_positions, suggest_bounds from .grid_search import grid_search_forecaster from .random_search import random_search_forecaster from .spotoptim_search import build_warm_start_x0, spotoptim_search_forecaster @@ -12,4 +13,7 @@ "bayesian_search_forecaster", "spotoptim_search_forecaster", "build_warm_start_x0", + "report_boundary_positions", + "boundary_report", + "suggest_bounds", ] diff --git a/src/spotforecast2/model_selection/boundary.py b/src/spotforecast2/model_selection/boundary.py new file mode 100644 index 00000000..3cd46835 --- /dev/null +++ b/src/spotforecast2/model_selection/boundary.py @@ -0,0 +1,376 @@ +# SPDX-FileCopyrightText: 2026 bartzbeielstein +# SPDX-License-Identifier: AGPL-3.0-or-later + +"""Search-space boundary management helpers for hyperparameter tuning. + +After a SpotOptim (or any optimizer) run, the tuned optimum may press against +a search-space boundary — meaning the optimizer wanted to go further but was +constrained. These helpers make that visible and actionable. + +Motivation: KB entry 2026-06-08-hyperparameter-boundary-management documents an +operational case where ``reg_alpha`` pinned at its old ceiling (98.9 % of the +linear range), inflating L1 regularization and flattening the live forecast. +Widening that bound and re-running resolved the issue. The helpers below +systematize that diagnostic loop. + +Three functions are provided: + +- `report_boundary_positions` — logs each dimension's position and returns a + list of flagged names. Decoupled from ``MultiTask``: the caller extracts + params (e.g. ``estimator.get_params()``). Primary operational diagnostic. +- `boundary_report` — returns a ``pd.DataFrame`` with one row per numeric + dimension: position, scale, flag. Companion to `suggest_bounds`. +- `suggest_bounds` — returns a copy of the search space with flagged bounds + widened on the pressed side; pass it straight back to + ``run_task_spotoptim(search_space=...)``. + +Ported from: + +- ``report_boundary_positions``: ``bart26k-lecture/scripts/team4_4zones_submit.py`` + (``report_boundary_positions`` function, section ``@sec-team4-boundary-diagnostics``). +- ``boundary_report`` / ``suggest_bounds``: ``bart26k-lecture/14_team_4_submission.qmd`` + (cell ``team4-boundary-helpers``, section ``@sec-team4-boundary-management``). +""" + +from __future__ import annotations + +import logging +import math +from collections.abc import Mapping +from typing import Any + +import pandas as pd + +_logger = logging.getLogger(__name__) + + +def report_boundary_positions( + params: Mapping[str, float | int], + search_space: Mapping[str, Any], + *, + warn_frac: float = 0.10, + logger: logging.Logger | None = None, +) -> list[str]: + """Log where each tuned value sits inside its numeric search-space interval. + + For each entry in `search_space` that is a 2- or 3-tuple numeric dimension + ``(low, high)`` or ``(low, high, "log10")``, the function: + + - strips the ``"estimator__"`` prefix from the key when looking up the + corresponding entry in `params`; + - skips non-numeric or boolean values; + - computes the position in the dimension's own scale (log10 for log dims, + guarding against ``val <= 0`` or ``low <= 0``); + - flags ``"> upper"`` when ``pos > 1 - warn_frac`` and ``"< lower"`` when + ``pos < warn_frac``; + - logs each dimension at INFO level in a columnar format; + - returns the list of flagged strings (e.g. ``["reg_alpha > upper"]``). + + Categorical dimensions (list-valued entries) and unreadable entries are + skipped. The function never raises — it is a diagnostic and returns an + empty list on any unexpected error. + + Args: + params: Flat dict of parameter names to numeric values, as returned by + ``estimator.get_params()`` or equivalent. Keys should NOT carry the + ``"estimator__"`` prefix (it is stripped from `search_space` keys, + not from `params` keys). + search_space: Dict mapping search-space keys (potentially with + ``"estimator__"`` prefix) to dimension specs: ``(low, high)``, + ``(low, high, "log10")``, or a list of categories. + warn_frac: Fraction of the range (in the dimension's own scale) that + defines the "near-boundary" zone at each end. Default is 0.10. + logger: Logger to use for INFO/WARNING messages. Defaults to the + module-level ``logging.getLogger(__name__)`` logger. + + Returns: + List of flagged dimension strings, e.g. ``["reg_alpha > upper", + "learning_rate < lower"]``. Empty if all dimensions are interior. + + Examples: + Interior optimum — no flags returned: + + ```{python} + from spotforecast2.model_selection.boundary import report_boundary_positions + + params = {"num_leaves": 300, "learning_rate": 0.05} + space = { + "estimator__num_leaves": (8, 1024), + "estimator__learning_rate": (0.005, 0.3, "log10"), + } + flagged = report_boundary_positions(params, space) + print("flagged:", flagged) + assert flagged == [] + ``` + + Near-upper-boundary — flag is returned: + + ```{python} + from spotforecast2.model_selection.boundary import report_boundary_positions + + params = {"reg_alpha": 9.9} + space = {"estimator__reg_alpha": (0.001, 10.0)} + flagged = report_boundary_positions(params, space) + print("flagged:", flagged) + assert flagged == ["reg_alpha > upper"] + ``` + + """ + log = logger if logger is not None else _logger + flagged: list[str] = [] + + for name, spec in search_space.items(): + if not (isinstance(spec, tuple) and len(spec) in (2, 3)): + continue # categorical dim (e.g. "lags") + key = name.split("estimator__", 1)[-1] + try: + val = params.get(key) # type: ignore[arg-type] + if val is None: + continue + if not isinstance(val, (int, float)) or isinstance(val, bool): + continue + low, high = float(spec[0]), float(spec[1]) + is_log = len(spec) == 3 and spec[2] == "log10" + if is_log and (val <= 0 or low <= 0): + continue + if is_log: + pos = (math.log10(val) - math.log10(low)) / ( + math.log10(high) - math.log10(low) + ) + else: + pos = (float(val) - low) / (high - low) + flag = ( + "> upper" + if pos > 1 - warn_frac + else ("< lower" if pos < warn_frac else "") + ) + log.info( + " bound %-18s = %-11.5g in [%g, %g]%s pos=%.2f%s", + key, + val, + low, + high, + " log10" if is_log else "", + pos, + f" <-- {flag}" if flag else "", + ) + if flag: + flagged.append(f"{key} {flag}") + except Exception as exc: # noqa: BLE001 + log.warning("boundary check: skipping %r (unreadable entry: %s)", name, exc) + + if flagged: + log.warning( + "boundary check: %d tuned dim(s) near a bound (%s) -- consider " + "widening that side and re-running.", + len(flagged), + ", ".join(flagged), + ) + else: + log.info("boundary check: all tuned dimensions sit interior.") + + return flagged + + +def boundary_report( + best_params: Mapping[str, float | int], + search_space: Mapping[str, Any], + *, + warn_frac: float = 0.10, +) -> pd.DataFrame: + """Tabulate each tuned value's position inside its search-space bound. + + Returns a DataFrame sorted by descending position, with one row per + numeric dimension. Categorical dimensions are skipped. ``flag`` is one of + ``"> upper"``, ``"< lower"``, or ``""`` (interior). + + This function uses the `search_space` keys as-is (including any + ``"estimator__"`` prefix) to look up matching entries in `best_params`. + The returned ``param`` column strips the ``"estimator__"`` prefix for + readability. + + Ported from ``bart26k-lecture/14_team_4_submission.qmd``, cell + ``team4-boundary-helpers``. + + Args: + best_params: Flat dict of parameter names to values, keyed with the + same names as `search_space` (including any ``"estimator__"`` + prefix). + search_space: Dict mapping parameter names to dimension specs: + ``(low, high)``, ``(low, high, "log10")``, or a list of + categories (skipped). + warn_frac: Fraction of the range (in the dimension's own scale) + defining the "near-boundary" zone. Default is 0.10. + + Returns: + DataFrame with columns ``param``, ``low``, ``high``, ``value``, + ``scale``, ``position``, ``flag``, sorted by ``position`` descending. + + Examples: + Report on a near-upper-boundary value: + + ```{python} + from spotforecast2.model_selection.boundary import boundary_report + + best = { + "estimator__reg_alpha": 9.89, + "estimator__learning_rate": 0.069, + } + space = { + "estimator__reg_alpha": (0.001, 10.0), + "estimator__learning_rate": (0.005, 0.3, "log10"), + } + df = boundary_report(best, space) + print(df.to_string(index=False)) + assert "reg_alpha" in df["param"].values + flagged = df[df["flag"] == "> upper"]["param"].tolist() + assert "reg_alpha" in flagged + ``` + + """ + rows = [] + for name, spec in search_space.items(): + if not (isinstance(spec, tuple) and len(spec) in (2, 3)): + continue # categorical dim + low, high = float(spec[0]), float(spec[1]) + is_log = len(spec) == 3 and spec[2] == "log10" + val = best_params.get(name) # type: ignore[arg-type] + if val is None: + continue + if is_log: + if val <= 0 or low <= 0: + continue + pos = (math.log10(val) - math.log10(low)) / ( + math.log10(high) - math.log10(low) + ) + else: + pos = (float(val) - low) / (high - low) + flag = ( + "> upper" + if pos > 1 - warn_frac + else ("< lower" if pos < warn_frac else "") + ) + rows.append( + { + "param": name.replace("estimator__", ""), + "low": low, + "high": high, + "value": val, + "scale": "log10" if is_log else "linear", + "position": round(pos, 3), + "flag": flag, + } + ) + df = pd.DataFrame(rows) + if df.empty: + return pd.DataFrame(columns=["param", "low", "high", "value", "scale", "position", "flag"]) + return df.sort_values("position", ascending=False).reset_index(drop=True) + + +def suggest_bounds( + best_params: Mapping[str, float | int], + search_space: Mapping[str, Any], + *, + warn_frac: float = 0.10, + widen_factor: float = 10.0, +) -> dict[str, Any]: + """Return a copy of `search_space` with flagged bounds widened. + + Upper-pinned dimensions grow upward (``high * widen_factor`` for float/log, + ``high + (high - low)`` for integer); lower-pinned dimensions grow downward + (``low / widen_factor`` for float/log, ``max(1, low - (high - low))`` for + integer). Interior and categorical dimensions are copied unchanged. + + Pass the result straight back to + ``run_task_spotoptim(search_space=suggest_bounds(...))``. + + Ported from ``bart26k-lecture/14_team_4_submission.qmd``, cell + ``team4-boundary-helpers`` (parameter ``widen`` renamed to ``widen_factor`` + for clarity). + + Args: + best_params: Flat dict of parameter names to values, keyed with the + same names as `search_space` (including any ``"estimator__"`` + prefix). + search_space: Dict mapping parameter names to dimension specs: + ``(low, high)``, ``(low, high, "log10")``, or a list of + categories (returned unchanged). + warn_frac: Fraction of the range (in the dimension's own scale) + defining the "near-boundary" zone. Default is 0.10. + widen_factor: Multiplicative factor for widening float/log bounds. + Integer bounds use an additive span instead. Default is 10.0. + + Returns: + New search-space dict with the same keys as `search_space` but with + boundary-pressed bounds extended on the pressed side. + + Examples: + Upper-pinned float bound is multiplied by widen_factor: + + ```{python} + from spotforecast2.model_selection.boundary import suggest_bounds + + best = {"estimator__reg_alpha": 9.89} + space = {"estimator__reg_alpha": (0.001, 10.0)} + new_space = suggest_bounds(best, space, widen_factor=10.0) + print(new_space) + assert new_space["estimator__reg_alpha"][1] == 100.0 + ``` + + Log-scale upper-pinned bound is also multiplied: + + ```{python} + from spotforecast2.model_selection.boundary import suggest_bounds + + best = {"estimator__reg_alpha": 9.89} + space = {"estimator__reg_alpha": (0.001, 10.0, "log10")} + new_space = suggest_bounds(best, space, widen_factor=10.0) + print(new_space) + assert new_space["estimator__reg_alpha"][1] == 100.0 + ``` + + Integer bound grows additively: + + ```{python} + from spotforecast2.model_selection.boundary import suggest_bounds + + best = {"estimator__n_estimators": 4950} + space = {"estimator__n_estimators": (100, 5000)} + new_space = suggest_bounds(best, space, widen_factor=10.0) + print(new_space) + assert new_space["estimator__n_estimators"][1] == 5000 + (5000 - 100) + ``` + + Interior dimension is returned unchanged: + + ```{python} + from spotforecast2.model_selection.boundary import suggest_bounds + + best = {"estimator__num_leaves": 300} + space = {"estimator__num_leaves": (8, 1024)} + new_space = suggest_bounds(best, space) + assert new_space["estimator__num_leaves"] == (8, 1024) + ``` + + """ + report = boundary_report(best_params, search_space, warn_frac=warn_frac) + flagged = { + row["param"]: row["flag"] + for _, row in report.iterrows() + if row["flag"] + } + out: dict[str, Any] = {} + for name, spec in search_space.items(): + if not (isinstance(spec, tuple) and len(spec) in (2, 3)): + out[name] = spec + continue + low, high, rest = spec[0], spec[1], spec[2:] + is_int = isinstance(low, int) and isinstance(high, int) and not rest + short_name = name.replace("estimator__", "") + flag = flagged.get(short_name, "") + if flag == "> upper": + high = int(high + (high - low)) if is_int else high * widen_factor + elif flag == "< lower": + low = max(1, int(low - (high - low))) if is_int else low / widen_factor + out[name] = (low, high, *rest) + return out diff --git a/src/spotforecast2/stats/__init__.py b/src/spotforecast2/stats/__init__.py index 5ae619be..9ee8a065 100644 --- a/src/spotforecast2/stats/__init__.py +++ b/src/spotforecast2/stats/__init__.py @@ -13,10 +13,11 @@ compute_periodogram, ) -from .autocorrelation import calculate_lag_autocorrelation +from .autocorrelation import calculate_lag_autocorrelation, select_pacf_lags __all__ = [ "augmented_dickey_fuller", "calculate_lag_autocorrelation", "compute_periodogram", + "select_pacf_lags", ] diff --git a/src/spotforecast2/stats/autocorrelation.py b/src/spotforecast2/stats/autocorrelation.py index 241c2d95..e89a6ed4 100644 --- a/src/spotforecast2/stats/autocorrelation.py +++ b/src/spotforecast2/stats/autocorrelation.py @@ -3,11 +3,14 @@ from __future__ import annotations +import logging from importlib.util import find_spec import numpy as np import pandas as pd +_logger = logging.getLogger(__name__) + def calculate_lag_autocorrelation( data: pd.Series | pd.DataFrame, @@ -192,3 +195,131 @@ def calculate_lag_autocorrelation( ) return results + + +def select_pacf_lags( + series: pd.Series, + *, + n_lags: int = 200, + top_k: int = 8, + fallback: list[int] | None = None, +) -> list[int]: + """Select the most informative lags using the partial autocorrelation function. + + Computes the PACF up to `n_lags` via `calculate_lag_autocorrelation`, then + returns the `top_k` lags whose `|PACF|` exceeds the 95 % significance band + (1.96 / sqrt(N)), sorted in ascending order. + + This is a pure-compute helper ported from the operational team-4 pipeline + (`select_key_lags` in ``bart26k-lecture/scripts/team4_4zones_submit.py``). + No plotting, no side effects beyond an optional DEBUG log line. + + Args: + series: The time series from which to estimate lags. Must contain at + least `2 * n_lags + 1` observations for statsmodels PACF to run + without truncating `n_lags`. + n_lags: Number of lags passed to `calculate_lag_autocorrelation`. + Default is 200. + top_k: Maximum number of lags to return (the `top_k` significant lags + ranked by descending `|PACF|`). Default is 8. + fallback: Lag list returned when no lag exceeds the significance band + (degenerate series: too short, nearly constant, or `n_lags` too + large). If `None`, a `ValueError` is raised instead. + + Returns: + Sorted list of lag integers (ascending). Length is at most `top_k`; + may be shorter if fewer than `top_k` significant lags exist. + + Raises: + ValueError: If no lag exceeds the significance band and `fallback` is + `None`. Pass `fallback=[1, 2, 24, 168]` (or any operator-chosen + constant) to suppress the error and use a safe default instead. + + Examples: + Select significant lags from a synthetic AR(1) series: + + ```{python} + import numpy as np + import pandas as pd + from spotforecast2.stats.autocorrelation import select_pacf_lags + + rng = np.random.default_rng(42) + n = 24 * 120 # 120 days of hourly data + ar = np.zeros(n) + for t in range(1, n): + ar[t] = 0.7 * ar[t - 1] + rng.standard_normal() + series = pd.Series(ar) + lags = select_pacf_lags(series, n_lags=50, top_k=8) + print("selected lags:", lags) + assert isinstance(lags, list) + assert all(isinstance(x, int) for x in lags) + assert lags == sorted(lags) + assert len(lags) <= 8 + assert 1 in lags, "lag-1 AR component should be selected" + ``` + + Select significant lags from an AR(24) process — lag 24 dominates: + + ```{python} + import numpy as np + import pandas as pd + from spotforecast2.stats.autocorrelation import select_pacf_lags + + rng = np.random.default_rng(42) + n = 24 * 200 + ar = np.zeros(n) + for t in range(24, n): + ar[t] = 0.8 * ar[t - 24] + rng.standard_normal() + series = pd.Series(ar) + lags = select_pacf_lags(series, n_lags=50, top_k=8) + print("selected lags:", lags) + assert 24 in lags, f"lag 24 expected in {lags}" + ``` + + Degenerate (constant) series — fallback is returned when provided: + + ```{python} + import pandas as pd + from spotforecast2.stats.autocorrelation import select_pacf_lags + + series = pd.Series([1.0] * 50) + result = select_pacf_lags(series, n_lags=10, fallback=[1, 2, 24]) + print("fallback lags:", result) + assert result == [1, 2, 24] + ``` + + Degenerate series with no fallback raises ValueError: + + ```{python} + import pytest + import pandas as pd + from spotforecast2.stats.autocorrelation import select_pacf_lags + + series = pd.Series([1.0] * 50) + with pytest.raises(ValueError, match="no significant"): + select_pacf_lags(series, n_lags=10, fallback=None) + ``` + + """ + acf_df = calculate_lag_autocorrelation(series, n_lags=n_lags) + conf = 1.96 / np.sqrt(len(series)) + significant = acf_df[acf_df["partial_autocorrelation_abs"] > conf] + top = significant.nlargest(top_k, "partial_autocorrelation_abs") + key_lags = sorted(top["lag"].astype(int).tolist()) + _logger.debug( + "select_pacf_lags: N=%d band=+/-%.4f significant=%d top_%d=%s", + len(series), + conf, + len(significant), + top_k, + key_lags, + ) + if not key_lags: + if fallback is not None: + return list(fallback) + raise ValueError( + f"select_pacf_lags: no significant PACF lags found " + f"(N={len(series)}, band={conf:.4f}, n_lags={n_lags}). " + "Pass fallback= to use a safe default instead of raising." + ) + return key_lags diff --git a/tests/test_model_selection_boundary.py b/tests/test_model_selection_boundary.py new file mode 100644 index 00000000..8318d483 --- /dev/null +++ b/tests/test_model_selection_boundary.py @@ -0,0 +1,284 @@ +# SPDX-FileCopyrightText: 2026 bartzbeielstein +# SPDX-License-Identifier: AGPL-3.0-or-later + +"""Tests for spotforecast2.model_selection.boundary.""" + +from __future__ import annotations + +import logging + +from spotforecast2.model_selection.boundary import ( + boundary_report, + report_boundary_positions, + suggest_bounds, +) + + +# --------------------------------------------------------------------------- +# Shared fixtures / helpers +# --------------------------------------------------------------------------- + +LINEAR_SPACE = { + "estimator__num_leaves": (8, 1024), + "estimator__n_estimators": (100, 5000), + "estimator__bagging_fraction": (0.5, 1.0), + "estimator__reg_alpha": (0.001, 10.0), + "lags": ["[1, 2, 24]", "[1, 2, 24, 168]"], # categorical, must be skipped +} + +LOG_SPACE = { + "estimator__learning_rate": (0.005, 0.3, "log10"), + "estimator__reg_alpha": (0.001, 100.0, "log10"), +} + +MIXED_SPACE = {**LINEAR_SPACE, **LOG_SPACE} + + +# --------------------------------------------------------------------------- +# report_boundary_positions +# --------------------------------------------------------------------------- + +class TestReportBoundaryPositions: + def test_interior_no_flag(self): + """An optimum well inside all bounds returns an empty list.""" + params = { + "num_leaves": 300, + "n_estimators": 2000, + "bagging_fraction": 0.75, + "reg_alpha": 1.0, + "learning_rate": 0.05, + } + flagged = report_boundary_positions(params, MIXED_SPACE) + assert flagged == [] + + def test_near_upper_linear_flagged(self): + """A value at 98 % of the linear range is flagged as '> upper'.""" + # reg_alpha = 9.8 in (0.001, 10.0): pos = (9.8 - 0.001) / (10.0 - 0.001) ≈ 0.980 + params = {"reg_alpha": 9.8} + space = {"estimator__reg_alpha": (0.001, 10.0)} + flagged = report_boundary_positions(params, space) + assert flagged == ["reg_alpha > upper"] + + def test_near_lower_linear_flagged(self): + """A value at 2 % of the linear range is flagged as '< lower'.""" + # n_estimators = 190 in (100, 5000): pos = (190 - 100) / (5000 - 100) ≈ 0.018 + params = {"n_estimators": 190} + space = {"estimator__n_estimators": (100, 5000)} + flagged = report_boundary_positions(params, space) + assert flagged == ["n_estimators < lower"] + + def test_log10_dim_near_lower(self): + """In log10 space, a value near the lower decade boundary is flagged.""" + # learning_rate = 0.006 in (0.005, 0.3) log10: + # pos = (log10(0.006) - log10(0.005)) / (log10(0.3) - log10(0.005)) + import math + low, high, val = 0.005, 0.3, 0.006 + pos = (math.log10(val) - math.log10(low)) / (math.log10(high) - math.log10(low)) + assert pos < 0.10 # confirm this is a near-lower case + params = {"learning_rate": val} + space = {"estimator__learning_rate": (low, high, "log10")} + flagged = report_boundary_positions(params, space) + assert flagged == ["learning_rate < lower"] + + def test_estimator_prefix_stripped(self): + """The 'estimator__' prefix is stripped before looking up params.""" + params = {"reg_alpha": 9.9} + space = {"estimator__reg_alpha": (0.001, 10.0)} + flagged = report_boundary_positions(params, space) + assert any("reg_alpha" in f for f in flagged) + + def test_categorical_lags_skipped(self): + """List-valued entries in the search space are silently skipped.""" + params = {} + space = {"lags": ["[1, 2, 24]", "[1, 24, 168]"]} + flagged = report_boundary_positions(params, space) + assert flagged == [] + + def test_bool_param_skipped(self): + """Boolean values are skipped (isinstance(True, int) is True in Python).""" + params = {"verbose": True} + space = {"estimator__verbose": (0, 1)} + flagged = report_boundary_positions(params, space) + assert flagged == [] + + def test_missing_param_key_skipped_with_warning(self, caplog): + """A param key not found in params is skipped; no exception is raised.""" + params = {} # empty — key is missing + space = {"estimator__num_leaves": (8, 1024)} + with caplog.at_level(logging.DEBUG): + flagged = report_boundary_positions(params, space) + assert flagged == [] + # No warning should be emitted for a simply missing key (it is None, + # so we skip via the `val is None` check, not the except branch) + + def test_returned_list_exact_contents(self): + """The returned list contains exactly the flagged strings in order.""" + params = {"reg_alpha": 9.9, "n_estimators": 150} + space = { + "estimator__reg_alpha": (0.001, 10.0), + "estimator__n_estimators": (100, 5000), + } + flagged = report_boundary_positions(params, space) + # reg_alpha is near upper; n_estimators 150 in (100, 5000) = pos 0.01 → < lower + assert "reg_alpha > upper" in flagged + assert "n_estimators < lower" in flagged + + def test_logger_injection(self, caplog): + """A custom logger passed via the logger= argument is used for messages.""" + custom_logger = logging.getLogger("test_custom_boundary_logger") + params = {"num_leaves": 300} + space = {"estimator__num_leaves": (8, 1024)} + with caplog.at_level(logging.INFO, logger="test_custom_boundary_logger"): + report_boundary_positions(params, space, logger=custom_logger) + # The custom logger should have emitted at least one INFO message + assert any( + r.name == "test_custom_boundary_logger" for r in caplog.records + ) + + def test_multiple_flags(self): + """Multiple near-boundary dims all appear in the returned list.""" + params = {"reg_alpha": 9.9, "learning_rate": 0.29} + space = { + "estimator__reg_alpha": (0.001, 10.0), + "estimator__learning_rate": (0.005, 0.3, "log10"), + } + flagged = report_boundary_positions(params, space) + assert len(flagged) == 2 + + def test_log_invalid_val_skipped(self): + """A zero or negative value in a log10 dim is silently skipped.""" + params = {"reg_alpha": 0.0} + space = {"estimator__reg_alpha": (0.001, 10.0, "log10")} + flagged = report_boundary_positions(params, space) + assert flagged == [] + + +# --------------------------------------------------------------------------- +# boundary_report (DataFrame form) +# --------------------------------------------------------------------------- + +class TestBoundaryReport: + def test_returns_dataframe(self): + best = {"estimator__reg_alpha": 9.89, "estimator__learning_rate": 0.069} + space = { + "estimator__reg_alpha": (0.001, 10.0), + "estimator__learning_rate": (0.005, 0.3, "log10"), + } + df = boundary_report(best, space) + assert set(df.columns) == {"param", "low", "high", "value", "scale", "position", "flag"} + + def test_flagged_near_upper(self): + best = {"estimator__reg_alpha": 9.89} + space = {"estimator__reg_alpha": (0.001, 10.0)} + df = boundary_report(best, space) + row = df[df["param"] == "reg_alpha"].iloc[0] + assert row["flag"] == "> upper" + assert row["scale"] == "linear" + + def test_log10_scale(self): + best = {"estimator__learning_rate": 0.069} + space = {"estimator__learning_rate": (0.005, 0.3, "log10")} + df = boundary_report(best, space) + row = df[df["param"] == "learning_rate"].iloc[0] + assert row["scale"] == "log10" + assert 0.0 < row["position"] < 1.0 + + def test_categorical_skipped(self): + best = {} + space = {"lags": ["[1, 2, 24]"]} + df = boundary_report(best, space) + assert df.empty + + def test_sorted_descending_position(self): + best = { + "estimator__reg_alpha": 9.89, # near upper → high position + "estimator__num_leaves": 300, # interior + } + space = { + "estimator__reg_alpha": (0.001, 10.0), + "estimator__num_leaves": (8, 1024), + } + df = boundary_report(best, space) + assert df["position"].iloc[0] >= df["position"].iloc[-1] + + def test_prefix_stripped_in_param_column(self): + best = {"estimator__num_leaves": 512} + space = {"estimator__num_leaves": (8, 1024)} + df = boundary_report(best, space) + assert "num_leaves" in df["param"].values + assert "estimator__num_leaves" not in df["param"].values + + +# --------------------------------------------------------------------------- +# suggest_bounds +# --------------------------------------------------------------------------- + +class TestSuggestBounds: + def test_interior_unchanged(self): + best = {"estimator__num_leaves": 300} + space = {"estimator__num_leaves": (8, 1024)} + new_space = suggest_bounds(best, space) + assert new_space["estimator__num_leaves"] == (8, 1024) + + def test_upper_pinned_float_multiplied(self): + """Upper-pinned float bound: high * widen_factor.""" + best = {"estimator__reg_alpha": 9.89} + space = {"estimator__reg_alpha": (0.001, 10.0)} + new_space = suggest_bounds(best, space, widen_factor=10.0) + assert new_space["estimator__reg_alpha"][1] == 100.0 + + def test_upper_pinned_log_multiplied(self): + """Upper-pinned log10 bound: high * widen_factor.""" + best = {"estimator__reg_alpha": 9.89} + space = {"estimator__reg_alpha": (0.001, 10.0, "log10")} + new_space = suggest_bounds(best, space, widen_factor=10.0) + assert new_space["estimator__reg_alpha"][1] == 100.0 + assert new_space["estimator__reg_alpha"][2] == "log10" + + def test_upper_pinned_int_additive(self): + """Upper-pinned integer bound: high + (high - low).""" + best = {"estimator__n_estimators": 4950} + space = {"estimator__n_estimators": (100, 5000)} + new_space = suggest_bounds(best, space, widen_factor=10.0) + assert new_space["estimator__n_estimators"][1] == 5000 + (5000 - 100) + + def test_lower_pinned_float_divided(self): + """Lower-pinned float bound: low / widen_factor.""" + best = {"estimator__learning_rate": 0.006} + space = {"estimator__learning_rate": (0.005, 0.3, "log10")} + new_space = suggest_bounds(best, space, widen_factor=10.0) + assert abs(new_space["estimator__learning_rate"][0] - 0.0005) < 1e-10 + + def test_categorical_unchanged(self): + """Categorical (list) entries pass through unchanged.""" + best = {} + space = {"lags": ["[1, 2, 24]", "[1, 24, 168]"]} + new_space = suggest_bounds(best, space) + assert new_space["lags"] == space["lags"] + + def test_all_keys_preserved(self): + """The returned dict has exactly the same keys as search_space.""" + best = {"num_leaves": 512, "learning_rate": 0.05} + space = { + "estimator__num_leaves": (8, 1024), + "estimator__learning_rate": (0.005, 0.3, "log10"), + "lags": ["[1, 2, 24]"], + } + new_space = suggest_bounds(best, space) + assert set(new_space.keys()) == set(space.keys()) + + def test_widen_factor_parameter(self): + """Different widen_factor values produce different bounds.""" + best = {"estimator__reg_alpha": 9.89} + space = {"estimator__reg_alpha": (0.001, 10.0)} + new5 = suggest_bounds(best, space, widen_factor=5.0) + new10 = suggest_bounds(best, space, widen_factor=10.0) + assert new10["estimator__reg_alpha"][1] > new5["estimator__reg_alpha"][1] + + def test_lower_pinned_int_additive_floor_1(self): + """Lower-pinned integer bound floors at 1: max(1, low - (high - low)).""" + best = {"estimator__n_estimators": 105} # near lower end of (100, 5000) + space = {"estimator__n_estimators": (100, 5000)} + new_space = suggest_bounds(best, space, widen_factor=10.0) + # new low = max(1, 100 - (5000 - 100)) = max(1, -4800) = 1 + assert new_space["estimator__n_estimators"][0] >= 1 diff --git a/tests/test_stats_select_pacf_lags.py b/tests/test_stats_select_pacf_lags.py new file mode 100644 index 00000000..95abc872 --- /dev/null +++ b/tests/test_stats_select_pacf_lags.py @@ -0,0 +1,159 @@ +# SPDX-FileCopyrightText: 2026 bartzbeielstein +# SPDX-License-Identifier: AGPL-3.0-or-later + +"""Tests for spotforecast2.stats.autocorrelation.select_pacf_lags.""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from spotforecast2.stats.autocorrelation import select_pacf_lags + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_daily_ar_series(n_days: int = 120, seed: int = 42) -> pd.Series: + """AR(1) with moderate coefficient for general lag selection tests.""" + rng = np.random.default_rng(seed) + n = n_days * 24 + ar = np.zeros(n) + for t in range(1, n): + ar[t] = 0.7 * ar[t - 1] + rng.standard_normal() + return pd.Series(ar) + + +def _make_ar24_series(n_days: int = 200, seed: int = 42) -> pd.Series: + """AR(24) process — lag 24 dominates the PACF (strong daily periodicity).""" + rng = np.random.default_rng(seed) + n = n_days * 24 + ar = np.zeros(n) + for t in range(24, n): + ar[t] = 0.8 * ar[t - 24] + rng.standard_normal() + return pd.Series(ar) + + +# --------------------------------------------------------------------------- +# Happy-path tests +# --------------------------------------------------------------------------- + +class TestSelectPacfLagsHappyPath: + def test_returns_list_of_ints(self): + series = _make_daily_ar_series() + lags = select_pacf_lags(series, n_lags=50, top_k=8) + assert isinstance(lags, list) + assert all(isinstance(x, int) for x in lags) + + def test_sorted_ascending(self): + series = _make_daily_ar_series() + lags = select_pacf_lags(series, n_lags=50, top_k=8) + assert lags == sorted(lags) + + def test_top_k_respected(self): + series = _make_daily_ar_series() + for k in (1, 4, 8): + lags = select_pacf_lags(series, n_lags=50, top_k=k) + assert len(lags) <= k + + def test_lag_1_selected_ar1_component(self): + """AR(1) coefficient 0.7 should produce a strongly significant lag-1 PACF.""" + series = _make_daily_ar_series() + lags = select_pacf_lags(series, n_lags=50, top_k=8) + assert 1 in lags, f"expected lag 1 in {lags}" + + def test_daily_lag_24_in_top_k(self): + """An AR(24) process must have lag 24 in the top-k selected lags.""" + series = _make_ar24_series(n_days=200) + lags = select_pacf_lags(series, n_lags=50, top_k=8) + assert 24 in lags, f"expected lag 24 in {lags} (AR(24) process)" + + def test_determinism(self): + """Same series, same parameters → same result every call.""" + series = _make_daily_ar_series(seed=0) + r1 = select_pacf_lags(series, n_lags=50, top_k=6) + r2 = select_pacf_lags(series, n_lags=50, top_k=6) + assert r1 == r2 + + def test_n_lags_limits_search(self): + """Returned lags must not exceed n_lags.""" + series = _make_daily_ar_series() + n_lags = 30 + lags = select_pacf_lags(series, n_lags=n_lags, top_k=8) + assert all(lag <= n_lags for lag in lags) + + +# --------------------------------------------------------------------------- +# Degenerate / edge-case tests +# --------------------------------------------------------------------------- + +class TestSelectPacfLagsDegenerate: + def test_constant_series_fallback_returned(self): + series = pd.Series([1.0] * 50) + result = select_pacf_lags(series, n_lags=10, fallback=[1, 2, 24]) + assert result == [1, 2, 24] + + def test_constant_series_no_fallback_raises(self): + series = pd.Series([1.0] * 50) + with pytest.raises(ValueError, match="no significant"): + select_pacf_lags(series, n_lags=10, fallback=None) + + def test_very_short_series_fallback(self): + """Series shorter than 2*n_lags+1; PACF falls back on statsmodels truncation. + + No lag should pass the wide confidence band for a near-white-noise short series. + We do not mandate whether ValueError or fallback is triggered; only that either + the fallback list is returned or a ValueError is raised. + """ + rng = np.random.default_rng(7) + series = pd.Series(rng.standard_normal(20)) + try: + result = select_pacf_lags(series, n_lags=5, top_k=2, fallback=[1, 2]) + # if no significant lags, fallback should be returned + assert isinstance(result, list) + except ValueError: + pass # also acceptable + + def test_fallback_list_copied(self): + """The returned list must be a copy, not the same object as fallback.""" + series = pd.Series([0.0] * 50) + fb = [1, 2, 24] + result = select_pacf_lags(series, n_lags=10, fallback=fb) + assert result is not fb + + def test_fallback_none_is_default(self): + """fallback=None is the default; constant series raises ValueError.""" + series = pd.Series([3.14] * 50) + with pytest.raises(ValueError): + select_pacf_lags(series, n_lags=10) + + +# --------------------------------------------------------------------------- +# Parameter variation +# --------------------------------------------------------------------------- + +class TestSelectPacfLagsParams: + def test_default_n_lags_200_top_k_8(self): + """Default parameters work on a long series without error.""" + series = _make_daily_ar_series(n_days=500) + lags = select_pacf_lags(series) + assert len(lags) <= 8 + assert lags == sorted(lags) + + def test_top_k_1(self): + series = _make_daily_ar_series() + lags = select_pacf_lags(series, n_lags=50, top_k=1) + assert len(lags) == 1 + + def test_large_top_k_bounded_by_significant(self): + """If fewer than top_k significant lags exist, return only those.""" + series = _make_daily_ar_series() + # Use a large top_k; returned list must still be <= n_sig + lags_k3 = select_pacf_lags(series, n_lags=5, top_k=3) + lags_k100 = select_pacf_lags(series, n_lags=5, top_k=100) + # Both share the same significant set (for n_lags=5); the larger top_k + # cannot return MORE lags, only the same. + assert set(lags_k3).issubset(set(lags_k100)) + assert len(lags_k100) >= len(lags_k3) diff --git a/uv.lock b/uv.lock index ddac7dee..6e344191 100644 --- a/uv.lock +++ b/uv.lock @@ -3491,7 +3491,7 @@ wheels = [ [[package]] name = "spotforecast2" -version = "8.0.0rc1" +version = "8.0.0" source = { editable = "." } dependencies = [ { name = "astral" }, From 62af0dddc84de20cbf4145dbf961852d404a3643 Mon Sep 17 00:00:00 2001 From: bartzbeielstein <32470350+bartzbeielstein@users.noreply.github.com> Date: Fri, 12 Jun 2026 03:13:41 +0200 Subject: [PATCH 2/2] fix(stats,model_selection): code-review fixes for PACF lag selection and boundary helpers FIX 1 (_quarto.yml): remove invalid `stats.select_pacf_lags` quartodoc contents entry and matching sidebar item; select_pacf_lags is a function inside stats/autocorrelation, already rendered as a section of stats.autocorrelation.qmd. Delete stale docs/reference/stats.select_pacf_lags.qmd. FIX 2 (boundary.py): expand module docstring with a prominent paragraph contrasting the two key-prefix conventions (report_boundary_positions strips "estimator__", boundary_report/suggest_bounds use full keys). Add logger.warning in boundary_report when result is empty but search_space contains numeric tuple dims, suggesting a key-convention mismatch. FIX 3 (uv.lock): restore uv.lock to develop's version; branch-specific lock hunk dropped. FIX 4 (test_model_selection_boundary.py): change test_all_keys_preserved to use "estimator__"-prefixed best_params so the flagged/widened code path is actually exercised; add asserts that the near-upper-boundary integer dim was widened and the interior log dim is unchanged. FIX 5 (test_stats_select_pacf_lags.py): remove dead `except ValueError: pass` arm from test_very_short_series_fallback (never raises with fallback given); add separate test_very_short_series_no_fallback_raises asserting ValueError on constant series without fallback. FIX 6 (boundary.py): add `if isinstance(val, bool): continue` guard in boundary_report mirroring report_boundary_positions; note boolean skip in docstring. FIX 7 (autocorrelation.py): coerce fallback elements to int via `[int(x) for x in fallback]`; note the coercion in the fallback param docstring. Co-Authored-By: Claude Fable 5 --- _quarto.yml | 3 -- src/spotforecast2/model_selection/boundary.py | 39 ++++++++++++++++++- src/spotforecast2/stats/autocorrelation.py | 5 ++- tests/test_model_selection_boundary.py | 16 +++++++- tests/test_stats_select_pacf_lags.py | 25 +++++++----- uv.lock | 2 +- 6 files changed, 70 insertions(+), 20 deletions(-) diff --git a/_quarto.yml b/_quarto.yml index 298f21f1..e596b288 100644 --- a/_quarto.yml +++ b/_quarto.yml @@ -128,8 +128,6 @@ website: contents: - text: "autocorrelation" file: docs/reference/stats.autocorrelation.qmd - - text: "select_pacf_lags" - file: docs/reference/stats.select_pacf_lags.qmd - section: "Tasks" contents: - text: "task_demo" @@ -261,7 +259,6 @@ quartodoc: desc: "Statistical analysis tools." contents: - stats.autocorrelation - - stats.select_pacf_lags - title: "Tasks" desc: "Demonstration and predefined forecasting tasks." diff --git a/src/spotforecast2/model_selection/boundary.py b/src/spotforecast2/model_selection/boundary.py index 3cd46835..bd40373e 100644 --- a/src/spotforecast2/model_selection/boundary.py +++ b/src/spotforecast2/model_selection/boundary.py @@ -24,6 +24,25 @@ widened on the pressed side; pass it straight back to ``run_task_spotoptim(search_space=...)``. +**Key convention difference — prefix handling:** + +``report_boundary_positions`` strips the ``"estimator__"`` prefix from each +search-space key before looking up the value in ``params``. The ``params`` +dict is expected to come from ``estimator.get_params()`` (scikit-learn style), +which returns UN-prefixed keys such as ``"reg_alpha"`` — not +``"estimator__reg_alpha"``. + +``boundary_report`` and ``suggest_bounds`` look up ``best_params`` using the +search-space key **as-is**, including any ``"estimator__"`` prefix. The +``best_params`` dict is expected to come from a SpotOptim result, which stores +FULL search-space keys (e.g. ``"estimator__reg_alpha"``). + +Mixing conventions (passing ``get_params()``-style keys to ``boundary_report``, +or SpotOptim result keys to ``report_boundary_positions``) will silently produce +an empty result because no key matches. If ``boundary_report`` returns an empty +DataFrame for a space with numeric dimensions, a key-convention mismatch is the +most likely cause. + Ported from: - ``report_boundary_positions``: ``bart26k-lecture/scripts/team4_4zones_submit.py`` @@ -181,8 +200,8 @@ def boundary_report( """Tabulate each tuned value's position inside its search-space bound. Returns a DataFrame sorted by descending position, with one row per - numeric dimension. Categorical dimensions are skipped. ``flag`` is one of - ``"> upper"``, ``"< lower"``, or ``""`` (interior). + numeric dimension. Categorical and boolean-valued dimensions are skipped. + ``flag`` is one of ``"> upper"``, ``"< lower"``, or ``""`` (interior). This function uses the `search_space` keys as-is (including any ``"estimator__"`` prefix) to look up matching entries in `best_params`. @@ -237,6 +256,8 @@ def boundary_report( val = best_params.get(name) # type: ignore[arg-type] if val is None: continue + if isinstance(val, bool): + continue # bool is a subclass of int; skip to avoid false positions if is_log: if val <= 0 or low <= 0: continue @@ -263,6 +284,20 @@ def boundary_report( ) df = pd.DataFrame(rows) if df.empty: + numeric_dims = sum( + 1 + for spec in search_space.values() + if isinstance(spec, tuple) and len(spec) in (2, 3) + ) + if numeric_dims > 0: + _logger.warning( + "boundary_report: result is empty but search_space has %d numeric " + "dimension(s). Check key-convention: best_params must use the FULL " + "search-space keys (e.g. 'estimator__reg_alpha'), not the unprefixed " + "get_params() style ('reg_alpha'). " + "Use report_boundary_positions() instead if you have unprefixed keys.", + numeric_dims, + ) return pd.DataFrame(columns=["param", "low", "high", "value", "scale", "position", "flag"]) return df.sort_values("position", ascending=False).reset_index(drop=True) diff --git a/src/spotforecast2/stats/autocorrelation.py b/src/spotforecast2/stats/autocorrelation.py index e89a6ed4..9b4027c3 100644 --- a/src/spotforecast2/stats/autocorrelation.py +++ b/src/spotforecast2/stats/autocorrelation.py @@ -224,7 +224,8 @@ def select_pacf_lags( ranked by descending `|PACF|`). Default is 8. fallback: Lag list returned when no lag exceeds the significance band (degenerate series: too short, nearly constant, or `n_lags` too - large). If `None`, a `ValueError` is raised instead. + large). Elements are coerced to `int` before returning. If `None`, + a `ValueError` is raised instead. Returns: Sorted list of lag integers (ascending). Length is at most `top_k`; @@ -316,7 +317,7 @@ def select_pacf_lags( ) if not key_lags: if fallback is not None: - return list(fallback) + return [int(x) for x in fallback] raise ValueError( f"select_pacf_lags: no significant PACF lags found " f"(N={len(series)}, band={conf:.4f}, n_lags={n_lags}). " diff --git a/tests/test_model_selection_boundary.py b/tests/test_model_selection_boundary.py index 8318d483..1386bdc3 100644 --- a/tests/test_model_selection_boundary.py +++ b/tests/test_model_selection_boundary.py @@ -257,8 +257,16 @@ def test_categorical_unchanged(self): assert new_space["lags"] == space["lags"] def test_all_keys_preserved(self): - """The returned dict has exactly the same keys as search_space.""" - best = {"num_leaves": 512, "learning_rate": 0.05} + """The returned dict has exactly the same keys as search_space. + + Uses prefixed keys (SpotOptim-result style) so the flagged/widened path + is exercised for at least one dimension. + """ + # num_leaves=1000 is near upper of (8, 1024) — should be widened + best = { + "estimator__num_leaves": 1000, + "estimator__learning_rate": 0.05, + } space = { "estimator__num_leaves": (8, 1024), "estimator__learning_rate": (0.005, 0.3, "log10"), @@ -266,6 +274,10 @@ def test_all_keys_preserved(self): } new_space = suggest_bounds(best, space) assert set(new_space.keys()) == set(space.keys()) + # The near-upper-boundary integer dim must have been widened upward + assert new_space["estimator__num_leaves"][1] > 1024 + # Interior log10 dim is unchanged + assert new_space["estimator__learning_rate"] == space["estimator__learning_rate"] def test_widen_factor_parameter(self): """Different widen_factor values produce different bounds.""" diff --git a/tests/test_stats_select_pacf_lags.py b/tests/test_stats_select_pacf_lags.py index 95abc872..7cf687d3 100644 --- a/tests/test_stats_select_pacf_lags.py +++ b/tests/test_stats_select_pacf_lags.py @@ -101,20 +101,25 @@ def test_constant_series_no_fallback_raises(self): select_pacf_lags(series, n_lags=10, fallback=None) def test_very_short_series_fallback(self): - """Series shorter than 2*n_lags+1; PACF falls back on statsmodels truncation. + """Series shorter than 2*n_lags+1; statsmodels truncates n_lags internally. - No lag should pass the wide confidence band for a near-white-noise short series. - We do not mandate whether ValueError or fallback is triggered; only that either - the fallback list is returned or a ValueError is raised. + With fallback provided the function must return either the fallback list + (no significant lags found) or a valid list of ints (some lags pass the + wide confidence band for a very short series). It must never raise. """ rng = np.random.default_rng(7) series = pd.Series(rng.standard_normal(20)) - try: - result = select_pacf_lags(series, n_lags=5, top_k=2, fallback=[1, 2]) - # if no significant lags, fallback should be returned - assert isinstance(result, list) - except ValueError: - pass # also acceptable + result = select_pacf_lags(series, n_lags=5, top_k=2, fallback=[1, 2]) + assert isinstance(result, list) + assert all(isinstance(x, int) for x in result) + + def test_very_short_series_no_fallback_raises(self): + """Same short series without fallback: ValueError is raised when no + significant lags are found (constant-like after statsmodels truncation). + """ + series = pd.Series([1.0] * 12) # constant → no significant lags, no fallback + with pytest.raises(ValueError, match="no significant"): + select_pacf_lags(series, n_lags=5, top_k=2, fallback=None) def test_fallback_list_copied(self): """The returned list must be a copy, not the same object as fallback.""" diff --git a/uv.lock b/uv.lock index 6e344191..ddac7dee 100644 --- a/uv.lock +++ b/uv.lock @@ -3491,7 +3491,7 @@ wheels = [ [[package]] name = "spotforecast2" -version = "8.0.0" +version = "8.0.0rc1" source = { editable = "." } dependencies = [ { name = "astral" },