From 7dc2fda524880a0cefa187aa5b86d12c38d08fc1 Mon Sep 17 00:00:00 2001 From: root Date: Fri, 8 May 2026 21:36:50 +0000 Subject: [PATCH] feat(core): compute_adaptive_threshold pure function + tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implementa il calcolo del percentile rolling con warmup, transizione min_days → target_days e floor assoluto. Pure function senza I/O: il caller passa la sequenza pre-filtrata (NULL e fetch_ok=0 esclusi). Tests: warmup, transizione finestra, floor, percentili. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/cerbero_bite/core/adaptive_threshold.py | 77 +++++++++++ tests/unit/test_adaptive_threshold.py | 142 ++++++++++++++++++++ 2 files changed, 219 insertions(+) create mode 100644 src/cerbero_bite/core/adaptive_threshold.py create mode 100644 tests/unit/test_adaptive_threshold.py diff --git a/src/cerbero_bite/core/adaptive_threshold.py b/src/cerbero_bite/core/adaptive_threshold.py new file mode 100644 index 0000000..34e08f8 --- /dev/null +++ b/src/cerbero_bite/core/adaptive_threshold.py @@ -0,0 +1,77 @@ +"""Funzione pura per calcolare la soglia adattiva del gate IV-RV. + +Spec: ``docs/superpowers/specs/2026-05-08-iv-rv-adaptive-gate-design.md``. + +Determinismic, no I/O. La query del repository è effettuata dal caller +(``runtime/entry_cycle``) prima di chiamare questa funzione. +""" + +from __future__ import annotations + +from decimal import Decimal +from typing import Sequence + +__all__ = ["compute_adaptive_threshold"] + +_TICKS_PER_DAY = 96 # cron */15 → 4 tick/h × 24h + + +def compute_adaptive_threshold( + history: Sequence[Decimal], + *, + percentile: Decimal, + absolute_floor: Decimal, + min_days: int, + target_days: int, +) -> Decimal | None: + """Ritorna la soglia adattiva o ``None`` durante il warmup hard. + + Args: + history: Sequenza ordinata ASC dei valori IV-RV (un valore per + ogni tick disponibile, max ``target_days * 96``). NULL e + tick non riusciti devono essere già stati filtrati dal + caller. + percentile: Quantile target nella distribuzione (es. ``0.25``). + absolute_floor: Floor minimo applicato dopo il calcolo del + percentile. La soglia restituita è + ``max(P_q, absolute_floor)``. + min_days: Sotto questa soglia di giorni di storia, la finestra + usata è "tutta la storia disponibile". Sopra, la finestra è + fissa a ``min_days`` finché non si raggiunge ``target_days``. + target_days: Finestra finale stabile. + + Returns: + ``None`` se la storia è < 1 giorno (warmup hard, gate + disabilitato), altrimenti il percentile della finestra, + bounded dal floor. + """ + if not history: + return None + n_ticks = len(history) + if n_ticks < _TICKS_PER_DAY: + return None + if n_ticks >= target_days * _TICKS_PER_DAY: + window = history[-target_days * _TICKS_PER_DAY:] + elif n_ticks >= min_days * _TICKS_PER_DAY: + window = history[-min_days * _TICKS_PER_DAY:] + else: + window = list(history) + return max(_percentile(window, percentile), absolute_floor) + + +def _percentile(values: Sequence[Decimal], q: Decimal) -> Decimal: + """Linear-interpolated percentile, NumPy-compatible (method='linear'). + + Implementato in Decimal puro per evitare dipendenze numpy nel core. + """ + if not values: + raise ValueError("percentile of empty sequence") + sorted_v = sorted(values) + n = len(sorted_v) + k = (Decimal(n) - Decimal(1)) * q + f = int(k) # floor + c = min(f + 1, n - 1) + if f == c: + return sorted_v[f] + frac = k - Decimal(f) + return sorted_v[f] + (sorted_v[c] - sorted_v[f]) * frac diff --git a/tests/unit/test_adaptive_threshold.py b/tests/unit/test_adaptive_threshold.py new file mode 100644 index 0000000..8d127b1 --- /dev/null +++ b/tests/unit/test_adaptive_threshold.py @@ -0,0 +1,142 @@ +"""TDD per :mod:`cerbero_bite.core.adaptive_threshold`. + +Spec: ``docs/superpowers/specs/2026-05-08-iv-rv-adaptive-gate-design.md``. +""" + +from __future__ import annotations + +from decimal import Decimal + +import pytest + +from cerbero_bite.core.adaptive_threshold import compute_adaptive_threshold + + +# --------------------------------------------------------------------------- +# Warmup +# --------------------------------------------------------------------------- + + +def test_empty_history_returns_none() -> None: + out = compute_adaptive_threshold( + history=[], + percentile=Decimal("0.25"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + assert out is None + + +def test_history_under_one_day_returns_none() -> None: + out = compute_adaptive_threshold( + history=[Decimal("3")] * 50, # 50 tick < 96 + percentile=Decimal("0.25"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + assert out is None + + +def test_history_exactly_one_day_returns_percentile() -> None: + history = [Decimal(i) / Decimal("10") for i in range(96)] # 0.0..9.5 + out = compute_adaptive_threshold( + history=history, + percentile=Decimal("0.25"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + # P25 di [0.0..9.5] passo 0.1 con method='linear': k=23.75, val ≈ 2.375 + assert out is not None + assert Decimal("2.3") < out < Decimal("2.5") + + +def _ramp(n: int, base: Decimal = Decimal("1")) -> list[Decimal]: + """Ramp lineare 1, 2, 3, ... per testare in modo predicibile il P25.""" + return [base * Decimal(i + 1) for i in range(n)] + + +def test_below_min_days_uses_full_history() -> None: + # 5 giorni di storia (5*96=480 tick), min_days=30, target=60. + # Window = full history. + history = _ramp(480) + out = compute_adaptive_threshold( + history=history, + percentile=Decimal("0.25"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + # P25 di ramp [1..480] = ~120.75 (k=479*0.25=119.75, ramp[119]=120, ramp[120]=121) + assert out is not None + assert Decimal("120") <= out <= Decimal("121") + + +def test_between_min_and_target_uses_min_window() -> None: + # 50 giorni di storia (4800 tick), min_days=30, target=60. Window = ultimi 30g. + history = _ramp(4800) # values 1..4800 + out = compute_adaptive_threshold( + history=history, + percentile=Decimal("0.25"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + # Window = ultimi 30*96=2880, valori 1921..4800, P25 ≈ 2640 + assert out is not None + assert Decimal("2630") <= out <= Decimal("2650") + + +def test_above_target_uses_target_window() -> None: + # 100 giorni (9600 tick), target=60. Window = ultimi 60g. + history = _ramp(9600) # values 1..9600 + out = compute_adaptive_threshold( + history=history, + percentile=Decimal("0.25"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + # Window = ultimi 5760, valori 3841..9600, P25 ≈ 5280 + assert out is not None + assert Decimal("5270") <= out <= Decimal("5290") + + +def test_floor_binding_overrides_low_percentile() -> None: + history = [Decimal("0.5")] * 200 + out = compute_adaptive_threshold( + history=history, + percentile=Decimal("0.25"), + absolute_floor=Decimal("3"), + min_days=30, + target_days=60, + ) + assert out == Decimal("3") + + +def test_floor_not_binding_returns_percentile() -> None: + history = [Decimal("5")] * 200 + out = compute_adaptive_threshold( + history=history, + percentile=Decimal("0.25"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + assert out == Decimal("5") + + +def test_median_percentile_returns_p50() -> None: + history = _ramp(200) + out = compute_adaptive_threshold( + history=history, + percentile=Decimal("0.5"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + # P50 di [1..200] = (200+1)/2 = 100.5 + assert out is not None + assert Decimal("100") <= out <= Decimal("101")