feat(core): compute_adaptive_threshold pure function + tests
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) <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
Reference in New Issue
Block a user