Arbiter (per-point routing for Orca)

Routes each point to the right corrector instead of forcing one gate on an entire signal -- grew out of a real benchmark showing hybrid_engine-style rejection wins on isolated impulses while pressure_valve's tolerance wins on a sustained level change, with neither beating the other everywhere. Classifies every point as clean/spike/regime against a wide, causal reference window (only points before it, never after -- a symmetric window straddling a real transition dilutes its own scale right where it needs to detect it), then the run-length and internal coherence of consecutive deviant points decides isolated impulse vs genuine regime change. Wired into Orca.protect_and_forward(..., use_arbiter=True); off by default.

arbiter

utility/arbiter.py

Instrada ogni punto verso lo specialista giusto invece di forzarne uno solo su tutto il segnale -- nato dal confronto reale in test/test_pressure_valve_vs_orca_experiment.py: hybrid_shield (phi_ab) vince nettamente sugli impulsi isolati (A, B, C, G) ma perde su un cambio di regime sostenuto (F), pressure_valve fa l'esatto contrario. Nessuno dei due batte l'altro ovunque -- la vera differenza tra i due casi e' la DURATA della deviazione, non la sua ampiezza.

APPROCCIO: per ogni punto, si confronta con una finestra di riferimento LARGA (raggio*ref_mult, la stessa scala usata dalla "molla" JSD di pressure_valve) -- deliberatamente non con la finestra locale stretta, che nel benchmark si e' vista collassare (mediana/MAD/IQR esattamente 0) quando piu' outlier vicini finiscono nella stessa finestra piccola. Questa finestra e' CAUSALE (_window_causal: solo punti PRIMA di i, mai dopo) -- una finestra simmetrica, usata proprio a cavallo di una transizione di livello, mescolerebbe vecchio e nuovo livello nella propria scala, diluendo la deviazione che dovrebbe far scattare la soglia (era cosi' nella prima versione di questo modulo, corretto dopo un caso reale non rilevato -- vedi cronologia in CHANGELOG.md [1.1.10]). Una volta segnati i punti "devianti", la LUNGHEZZA della sequenza consecutiva di punti devianti decide: corta (<= spike_run_max) = impulso isolato, lunga = cambio di regime genuino.

CASO DEGENERE (baseline esattamente piatta, scala robusta = 0): non e' un problema di stimatore -- vedi il LIMITE NOTO nel docstring di pressure_valve, un M-estimator di Huber converge comunque a scala 0 per lo stesso motivo strutturale. Qui non si divide mai per una scala zero: se la dispersione della finestra di riferimento e' sotto epsilon, QUALSIASI scostamento non nullo dalla mediana e' trattato come deviante senza bisogno di una soglia in sigma (su una baseline davvero costante, "diverso da costante" e' gia' la risposta, non serve altro).

CRONOLOGIA (limite trovato e poi risolto nella stessa sessione): la prima versione usava una finestra di riferimento SIMMETRICA, e lo scenario F (rottura strutturale, salto di livello permanente) non veniva rilevato come 'regime' -- deviazione[i] restava 0 su tutta la sequenza attorno alla transizione, perche' la finestra simmetrica, usata proprio a cavallo del salto, conteneva meta' vecchio livello e meta' nuovo, gonfiando la propria scala (MAD) invece di restare piccola. Un primo test aveva scambiato questo per un successo (RMSE 0.0 con route_and_correct standalone), ma era un artefatto: i dati grezzi di quello scenario coincidono GIA' esattamente col target (la "corruzione" e' il salto stesso, non rumore aggiunto), quindi "passa tutto grezzo" da' RMSE 0 indipendentemente da come viene etichettato ogni punto -- il vero banco di prova era la pipeline integrata di Orca (utility/orca.py, _applica_arbitro), dove la mancata rilevazione si traduceva in comportamento SICURO ma non migliorativo (9.4007 identico con o senza l'arbitro). Passando alla finestra CAUSALE (sopra), la rilevazione diventa reale e verificabile: sullo scenario F, gli indici 60-74 (esatti, verificato -- la transizione vera e' a 60) vengono etichettati 'regime', e attraverso la pipeline reale di Orca l'RMSE scende da 9.4007 (default) a 6.0449 (use_arbiter=True) -- un miglioramento reale, non piu' un pareggio. Vedi test/test_arbiter_orca_integration.py per la verifica completa sui 7 scenari con questo fix.

classify_segments

classify_segments(x: ndarray, radius: int = 10, ref_mult: int = 3, n_sigmas: float = 3.0, spike_run_max: int = 2, eps: float = 1e-09) -> Tuple[np.ndarray, np.ndarray, np.ndarray]

Ritorna (etichette, deviazione, incertezza): etichette -- array di stringhe 'clean'/'spike'/'regime', una per punto. deviazione -- |x[i] - riferimento| / scala (o solo |x[i]-riferimento| se la scala e' degenere), stessa unita' per ogni punto. incertezza -- 0..1, quanto il punto e' vicino al confine di decisione (soglia deviante/pulito, o soglia spike/regime sulla lunghezza della sequenza) -- non quanto e' "corretto", ma quanto la classificazione stessa e' precaria.

Source code in dense_armor/utility/arbiter.py

def classify_segments(
    x: np.ndarray, radius: int = 10, ref_mult: int = 3,
    n_sigmas: float = 3.0, spike_run_max: int = 2, eps: float = 1e-9,
) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
    """
    Ritorna (etichette, deviazione, incertezza):
      etichette   -- array di stringhe 'clean'/'spike'/'regime', una per punto.
      deviazione  -- |x[i] - riferimento| / scala (o solo |x[i]-riferimento|
                     se la scala e' degenere), stessa unita' per ogni punto.
      incertezza  -- 0..1, quanto il punto e' vicino al confine di decisione
                     (soglia deviante/pulito, o soglia spike/regime sulla
                     lunghezza della sequenza) -- non quanto e' "corretto",
                     ma quanto la classificazione stessa e' precaria.
    """
    x = np.asarray(x, dtype=np.float64).ravel()
    n = x.size
    deviante = np.zeros(n, dtype=bool)
    deviazione = np.zeros(n, dtype=np.float64)
    scarto_normalizzato = np.zeros(n, dtype=np.float64)  # |dev|/soglia, per l'incertezza
    med_rif = np.zeros(n, dtype=np.float64)  # mediana della finestra di riferimento, per punto

    for i in range(n):
        w_rif = _window_causal(x, i, radius * ref_mult)
        if w_rif.size < 4:
            continue
        med, scala = _robust_center_scale(w_rif)
        med_rif[i] = med
        scarto = abs(x[i] - med)
        if scala < eps:
            # Baseline degenere: nessuna divisione, "diverso da costante"
            # e' gia' la risposta. scarto_normalizzato satura a un valore
            # alto/basso netto (mai vicino al confine per costruzione),
            # cosi' l'incertezza su questi punti resta bassa -- il caso
            # degenere e' l'UNICO dove la decisione e' davvero inequivocabile.
            deviazione[i] = scarto
            deviante[i] = scarto > eps
            scarto_normalizzato[i] = 0.0 if deviante[i] else 10.0
        else:
            deviazione[i] = scarto / scala
            deviante[i] = deviazione[i] > n_sigmas
            scarto_normalizzato[i] = deviazione[i] / n_sigmas

    etichette = np.full(n, "clean", dtype=object)
    incertezza = np.zeros(n, dtype=np.float64)

    # Incertezza sulla soglia deviante/pulito: vicino a scarto_normalizzato==1
    # (cioe' esattamente al confine n_sigmas) -> massima ambiguita'.
    incertezza_soglia = np.exp(-4.0 * (scarto_normalizzato - 1.0) ** 2)

    i = 0
    while i < n:
        if not deviante[i]:
            incertezza[i] = incertezza_soglia[i]
            i += 1
            continue
        j = i
        while j < n and deviante[j]:
            j += 1
        run_len = j - i
        if run_len <= spike_run_max:
            label = "spike"
        else:
            # Una sequenza lunga e' un vero cambio di regime solo se i suoi
            # punti si ASSESTANO su un nuovo livello coerente -- non basta
            # che siano "diversi dal vecchio", devono anche essere simili
            # TRA LORO. Verificato: 3 impulsi alternati (+500,-500,+500,
            # scenario A) sono una sequenza di lunghezza 3 (> spike_run_max
            # di default 2), ma la loro dispersione INTERNA e' enorme
            # (std~471) -- senza questo controllo venivano scambiati per un
            # regime e passati grezzi, RMSE 86.6 invece di ~0. Una sequenza
            # che si assesta davvero (scenario F) ha dispersione interna
            # piccola rispetto alla dimensione del salto stesso.
            run_vals = x[i:j]
            run_spread = float(np.std(run_vals))
            run_jump = float(abs(np.median(run_vals) - med_rif[i]))
            coerente = run_spread < 0.5 * max(run_jump, eps)

            # Coerenza interna non basta: un collasso temporaneo che poi
            # TORNA al vecchio livello (scenario C, 3 zeri poi ritorno a
            # 120) e' internamente coerente quanto un vero regime (spread
            # interno zero in entrambi i casi) -- la differenza vera si
            # vede SOLO guardando cosa succede subito dopo la sequenza:
            # un regime vero (scenario F) resta vicino al NUOVO livello,
            # un collasso torna verso quello VECCHIO. Senza questo
            # controllo scenario C restava RMSE 20.8 invece di ~0
            # (scambiato per regime, passato grezzo invece di respinto).
            post_lo, post_hi = j, min(n, j + radius)
            if post_hi > post_lo:
                post_med = float(np.median(x[post_lo:post_hi]))
                run_med = float(np.median(run_vals))
                persiste = abs(post_med - run_med) < abs(post_med - med_rif[i])
            else:
                persiste = True  # fine serie: nessun "dopo" da controllare

            label = "regime" if (coerente and persiste) else "spike"
        etichette[i:j] = label
        # Incertezza sulla soglia spike/regime: vicino a run_len==spike_run_max
        # -> massima ambiguita' sulla lunghezza; combinata (media) con quella
        # sulla soglia deviante/pulito dei singoli punti della sequenza.
        incertezza_run = float(np.exp(-1.0 * (run_len - spike_run_max) ** 2))
        incertezza[i:j] = 0.5 * incertezza_soglia[i:j] + 0.5 * incertezza_run
        i = j

    return etichette, deviazione, incertezza

route_and_correct

route_and_correct(x: ndarray, radius: int = 10, ref_mult: int = 3, n_sigmas: float = 3.0, spike_run_max: int = 2) -> Tuple[np.ndarray, np.ndarray, np.ndarray]

Applica la classificazione e corregge ogni punto secondo l'etichetta: 'clean'/'regime' -> passa il valore grezzo (un cambio di regime genuino non va respinto, stessa filosofia del gate costante di Orca); 'spike' -> sostituito con la mediana della finestra di riferimento larga (robusta, non contaminata dal singolo impulso breve).

Ritorna (pulito, etichette, incertezza).

Source code in dense_armor/utility/arbiter.py

def route_and_correct(
    x: np.ndarray, radius: int = 10, ref_mult: int = 3,
    n_sigmas: float = 3.0, spike_run_max: int = 2,
) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
    """
    Applica la classificazione e corregge ogni punto secondo l'etichetta:
      'clean'/'regime' -> passa il valore grezzo (un cambio di regime
                          genuino non va respinto, stessa filosofia del
                          gate costante di Orca);
      'spike'          -> sostituito con la mediana della finestra di
                          riferimento larga (robusta, non contaminata
                          dal singolo impulso breve).

    Ritorna (pulito, etichette, incertezza)."""
    x = np.asarray(x, dtype=np.float64).ravel()
    n = x.size
    etichette, _, incertezza = classify_segments(x, radius, ref_mult, n_sigmas, spike_run_max)

    pulito = np.copy(x)
    for i in range(n):
        if etichette[i] == "spike":
            w_rif = _window(x, i, radius * ref_mult)
            pulito[i] = float(np.median(w_rif))

    return pulito, etichette, incertezza

---

See also: Orca -- use_arbiter=True is where this module is actually used; Robust filters -- pressure_valve's JSD-adaptive threshold inspired the wide reference window here.