From c25cac78e0a84dfda5c6084fb6e1ee0a06330768 Mon Sep 17 00:00:00 2001 From: David Fitzsimmons Date: Wed, 30 Sep 2026 15:28:29 -0400 Subject: [PATCH] perf(classifier): defer exact residual deviation until needed --- loopgain/classifier.py | 77 +++++++++++------ tests/test_classifier_lazy_features.py | 111 +++++++++++++++++++++++++ 2 files changed, 161 insertions(+), 27 deletions(-) create mode 100644 tests/test_classifier_lazy_features.py diff --git a/loopgain/classifier.py b/loopgain/classifier.py index 875c27b..854e09f 100644 --- a/loopgain/classifier.py +++ b/loopgain/classifier.py @@ -208,6 +208,22 @@ def extract_features(error_history: Sequence[float]) -> TrajectoryFeatures: linear. This is the standard transformation for any signal that obeys Barkhausen's E_n = Aβ · E_{n−1}. """ + return _extract_features(error_history, with_oscillation=True) + + +def _oscillation_std(log_e: Sequence[float], slope: float) -> float: + """Preserve the exact public residual calculation and stdlib rounding.""" + n = len(log_e) + xs = list(range(n)) + intercept = sum(log_e) / n - slope * (sum(xs) / n) + residuals = [log_e[i] - (intercept + slope * xs[i]) for i in range(n)] + return statistics.pstdev(residuals) + + +def _extract_features( + error_history: Sequence[float], *, with_oscillation: bool +) -> TrajectoryFeatures: + """Compute trend features, optionally deferring the costly final gate.""" n = len(error_history) if n == 0: return TrajectoryFeatures(0.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0) @@ -233,13 +249,9 @@ def extract_features(error_history: Sequence[float]) -> TrajectoryFeatures: log_e = [math.log10(max(e, _EPS)) for e in error_history] slope, p = _ols_slope_and_p(xs, log_e) - # Detrended residual std (sample std). - intercept = sum(log_e) / n - slope * (sum(xs) / n) - residuals = [log_e[i] - (intercept + slope * xs[i]) for i in range(n)] - if n >= 2: - osc_std = statistics.pstdev(residuals) - else: - osc_std = 0.0 + # Population std is only needed by the public feature API or the final + # oscillation gate. Keep statistics.pstdev's exact rounding semantics. + osc_std = _oscillation_std(log_e, slope) if with_oscillation else 0.0 return TrajectoryFeatures( e_current=e_current, @@ -294,7 +306,25 @@ def classify_trajectory( if n < 2: return INIT - f = extract_features(error_history) + # The public classifier historically exposes eager feature errors for + # non-finite inputs. Do not let a decisive ratio bypass those errors. + # Comparisons also avoid overflowing when the sequence contains a large + # finite Python integer that math.isfinite would coerce to float. + if any(not -math.inf < e < math.inf for e in error_history): + extract_features(error_history) + + # With two observations only the cumulative ratio is used below. Avoid + # computing a regression and residual variance that cannot affect the + # verdict (there are no residual degrees of freedom). + if n == 2: + e_ratio = e_current / max(abs(error_history[0]), _EPS) + if e_ratio <= th.e_ratio_fast: + return FAST_CONVERGE + if e_ratio < 1.0: + return CONVERGING + if e_ratio > 1.0 + th.div_margin: + return DIVERGING + return STALLING # Liveness signal: how many iterations since the loop last achieved a new # best (lowest) error. A genuinely converging loop keeps hitting new lows, @@ -304,32 +334,23 @@ def classify_trajectory( # loop has stopped improving, so it can reach STALLING / OSCILLATING and # terminate instead of riding its historical cumulative win forever. See # DEFAULT_STALL_PATIENCE. - hist = list(error_history) - iters_since_best = (n - 1) - hist.index(min(hist)) + # Preserve first-tie semantics without keeping a second history list + # alive while extracting features. + e_min = min(error_history) + best_index = next(i for i, error in enumerate(error_history) if error == e_min) + iters_since_best = (n - 1) - best_index still_improving = iters_since_best < th.stall_patience - # n == 2 special case: with two observations, the slope is well defined - # but its p-value is not (zero residual degrees of freedom). Fall back to - # the sign of the change. This is the same conservatism as a Wilcoxon - # signed-rank test with n=1: insufficient evidence for a significance - # claim, but the *direction* is unambiguous. - if n == 2: - if f.e_ratio <= th.e_ratio_fast: - return FAST_CONVERGE - if f.e_ratio < 1.0: - return CONVERGING - if f.e_ratio > 1.0 + th.div_margin: - return DIVERGING - return STALLING - # Order matters: FAST_CONVERGE precedes CONVERGING; both precede the # remaining gates. Both continue-verdicts are gated on `still_improving`: # a loop that has stopped hitting new lows is no longer "converging" no # matter how large its historical cumulative reduction was, and must be # allowed to fall through to STALLING / OSCILLATING so it can terminate. - if f.e_ratio <= th.e_ratio_fast and still_improving: + e_ratio = e_current / max(abs(error_history[0]), _EPS) + if e_ratio <= th.e_ratio_fast and still_improving: return FAST_CONVERGE + f = _extract_features(error_history, with_oscillation=False) slope_significant = f.slope_p < th.p_sig if ( @@ -342,8 +363,10 @@ def classify_trajectory( if f.slope_log > 0 and slope_significant and f.e_ratio > 1.0 + th.div_margin: return DIVERGING - if f.osc_std >= th.osc_std_threshold and abs(f.slope_log) < th.slope_tol: - return OSCILLATING + if abs(f.slope_log) < th.slope_tol: + log_e = [math.log10(max(e, _EPS)) for e in error_history] + if _oscillation_std(log_e, f.slope_log) >= th.osc_std_threshold: + return OSCILLATING return STALLING diff --git a/tests/test_classifier_lazy_features.py b/tests/test_classifier_lazy_features.py new file mode 100644 index 0000000..41f09d1 --- /dev/null +++ b/tests/test_classifier_lazy_features.py @@ -0,0 +1,111 @@ +"""Optimized gates must retain exact policy and public feature semantics.""" +import math +import statistics +from dataclasses import replace + +import pytest + +import loopgain.classifier as classifier + + +@pytest.mark.parametrize('history,expected', [ + ([10.0, 0.5], classifier.FAST_CONVERGE), + ([10.0, 1.0], classifier.FAST_CONVERGE), + ([10.0, math.nextafter(1.0, math.inf)], classifier.CONVERGING), + ([10.0, 9.0], classifier.CONVERGING), + ([10.0, 10.0], classifier.STALLING), + ([10.0, 11.0], classifier.STALLING), + ([10.0, 10.0 * math.nextafter(1.1, math.inf)], classifier.DIVERGING), + ([0.0, 0.0], classifier.FAST_CONVERGE), + ([0.0, 1.0], classifier.DIVERGING), +]) +def test_two_points_do_not_fit_a_regression(monkeypatch, history, expected): + def unexpected(*args, **kwargs): + raise AssertionError('two-point ratio gate does not require regression') + monkeypatch.setattr(classifier, '_ols_slope_and_p', unexpected) + assert classifier.classify_trajectory(history) == expected + assert classifier.classify_trajectory(tuple(history)) == expected + + +def test_fast_cumulative_gate_needs_no_regression(monkeypatch): + def unexpected(*args, **kwargs): + raise AssertionError('decisive cumulative gate does not require regression') + monkeypatch.setattr(classifier, '_ols_slope_and_p', unexpected) + assert classifier.classify_trajectory([100, 50, 20, 5]) == classifier.FAST_CONVERGE + + +def test_historical_reduction_does_not_bypass_liveness(): + assert classifier.classify_trajectory([100, 5, 5, 5, 5]) == classifier.STALLING + + +@pytest.mark.parametrize('history,expected', [ + ([100 * .95 ** i for i in range(20)], classifier.CONVERGING), + ([100 * 1.05 ** i for i in range(20)], classifier.DIVERGING), +]) +def test_decisive_trend_does_not_compute_variance(monkeypatch, history, expected): + def unexpected(*args, **kwargs): + raise AssertionError('decisive trend does not require residual variance') + monkeypatch.setattr(classifier.statistics, 'pstdev', unexpected) + assert classifier.classify_trajectory(history) == expected + + +def test_flat_trend_still_computes_exact_variance(monkeypatch): + calls = [] + original = statistics.pstdev + def recording(values): + calls.append(tuple(values)) + return original(values) + monkeypatch.setattr(classifier.statistics, 'pstdev', recording) + assert classifier.classify_trajectory([2.0] * 8) == classifier.STALLING + assert len(calls) == 1 + + +@pytest.mark.parametrize('history', [ + [2.0] * 8, + [1, 10, 2, 20, 3, 15], + [1e-300] * 8, + [1e300, 1e-300] * 4, + [10 ** (200 + .002 * i) for i in range(10)], +]) +def test_public_features_preserve_exact_population_std(history): + features = classifier.extract_features(history) + log_e = [math.log10(max(e, 1e-12)) for e in history] + xs = list(range(len(history))) + slope, p = classifier._ols_slope_and_p(xs, log_e) + intercept = sum(log_e) / len(log_e) - slope * (sum(xs) / len(xs)) + residuals = [log_e[i] - (intercept + slope * xs[i]) for i in xs] + assert features.slope_log == slope + assert features.slope_p == p + assert features.osc_std == statistics.pstdev(residuals) + + +def test_exact_custom_oscillation_boundary_is_preserved(): + history = [1, 10, 2, 20, 3, 15] + features = classifier.extract_features(history) + thresholds = classifier.TrajectoryThresholds( + e_ratio_fast=-1, e_ratio_conv=-1, p_sig=-1, + slope_tol=1000, osc_std_threshold=features.osc_std, + ) + assert classifier.classify_trajectory(history, thresholds=thresholds) == classifier.OSCILLATING + above = replace(thresholds, osc_std_threshold=math.nextafter(features.osc_std, math.inf)) + assert classifier.classify_trajectory(history, thresholds=above) == classifier.STALLING + + +@pytest.mark.parametrize('history', [ + [float('inf'), 1.0], [1.0, float('inf')], + [float('nan'), 1.0], [1.0, float('nan')], + [100, float('inf'), 1], [100, float('nan'), 1], +]) +def test_nonfinite_direct_inputs_keep_eager_feature_error(history): + # statistics.pstdev's exception type differs across supported Python + # versions. Match the unchanged eager feature API on this interpreter. + with pytest.raises(Exception) as reference: + classifier.extract_features(history) + with pytest.raises(type(reference.value)) as actual: + classifier.classify_trajectory(history) + assert str(actual.value) == str(reference.value) + + +def test_large_finite_integer_input_does_not_coerce_to_float_for_finiteness(): + value = 10 ** 1000 + assert classifier.classify_trajectory([value, value // 2]) == classifier.CONVERGING