Armatura (1D series shield)

Armatura cleans one series of numbers: it finds the readings that are obviously broken (a sudden huge value, a missing reading) and replaces them with sensible values. You get back the cleaned series and the list of places it changed.

A loss curve that suddenly jumps to 10⁶. A sensor channel that drops a reading and returns NaN. A token-likelihood stream that goes from 0.9 to 0 for one sample. These are the kinds of single-value glitches that break a downstream pipeline silently: the model does not complain, it just produces wrong numbers from that point on.

Armatura is the shield for a single 1D series. You give it an array; you get back the same array, with the broken samples replaced by the local baseline, plus the list of indices that were replaced.

1. Protect the series

One sensor channel at 100 Hz: noise around 0.5, one absurd value (9999) at sample 100, one NaN at sample 200.

import numpy as np
from dense_armor import Armatura

rng = np.random.default_rng(42)
x = 0.5 + 0.1 * rng.standard_normal(300)
x[100], x[200] = 9999.0, float("nan")
clean, K, anomalies = Armatura(livello_ia=0.0).analizza(x)
print(len(clean), anomalies, round(float(clean[100]), 3), K[100], K[50])

300 [100, 200] 0.463 1.0 0.0

clean has the same shape as x: the two broken samples are replaced by the local baseline (9999 becomes 0.463). Ordinary samples are kept, except those the engine's trigger classifies as static, which are replaced by the local median without being listed in anomalies (35 of the 300 here, see Hybrid engine, step 3). anomalies lists the replaced indices, [100, 200]. K is one flag per sample from the engine's trigger: 1 where the spike rule fired (sample 100), 0 elsewhere; the NaN is cleaned before the trigger, so sample 200 is in anomalies with K = 0.

livello_ia = 0 means "actively filter". livello_ia = 1 means "only mark, do not change the values" — useful when you want to log what the shield would have done without altering the stream.

2. What "local baseline" means

Armatura looks at a window of samples before the one being scored (never after): its baseline is the mean of those previous samples (radius min(20, max(3, n // 3))), and a binary trigger decides whether the sample is an impulse. If it is, it is replaced by the baseline; otherwise it passes through unchanged. The first two samples are never tested (the trigger needs two previous points), so a glitch there must be caught by another check.

The consequence, on the running example:

• The value 9999 at sample 100 is far above the baseline of the previous samples, so it is replaced. Index 100 appears in anomalies.
• The NaN at sample 200 never enters the window (the shield drops non-finite samples before the comparison), so the next finite sample is scored against the window ending at sample 199. Index 200 appears in anomalies, and clean[200] is the baseline.

3. What it is not for

Armatura is a point-level shield. It removes isolated spikes and non-finite values. It does not remove a slow drift, and it does not classify a regime change: those need a different tool (see CUSUM and Arbiter). If your signal is a step change that is genuinely intended, Armatura leaves it alone after the first few samples, because after those samples the step is the new local baseline.

4. The engine inside

Armatura is built on core.hybrid_engine, the binary-trigger engine that generalises the logic already verified in Dense-Evolution's own ia_utils.vector_healing.enhanced_dense_healing_hybrid. See Hybrid engine for the internals.

API reference

armatura

armatura.py — L'Armatura: istanza indossabile dello scudo per una serie 1D. Usa solo i moduli in core/ e utility/ di questo pacchetto.

Uso rapido (CLI): python -m dense_armor 1.2 1.3 9999 1.25 nan 1.3 python -m dense_armor file_con_numeri.txt

Uso da codice

from dense_armor import Armatura a = Armatura(livello_ia=0.0) # 0=IA neonata (FILTRA) ... 1=matura (solo MARCA) pulito, K, anomalie = a.analizza(serie) pulito, K, anomalie = a.analizza(serie_oggi, riferimento=baseline_di_ieri) # anti-deriva

Armatura

Armatura(
    static_threshold: float = 0.15,
    initial_damping: float = 0.85,
    alpha: float = 0.05,
    soglia_anomalia: float | None = None,
    livello_ia: float = 1.0,
)

Scudo indossabile con CLIP DINAMICO universale.

livello_ia (0..1): quanto e' matura l'IA che lo indossa. 0.0 -> clip pieno: lo scudo FILTRA il dato (corazza per IA neonata) 1.0 -> clip zero: lo scudo NON tocca il dato, solo MARCA (lente per IA matura) PAVIMENTO: la marcatura delle anomalie non viene MAI clippata — a nessun livello lo scudo tace (l'anomalia puo' ESSERE la risposta: mai cancellarla in silenzio). Il livello puo' essere misurato, non dichiarato: Armatura.livello_da_entropia(entropia, vocab_size).

MEMORIA: lo scudo e' una funzione PURA, senza stato (leggero per design). La memoria appartiene all'IA che lo indossa: per vedere la DERIVA LENTA passa la tua baseline storica come riferimento.

static_threshold/initial_damping/alpha — mantenuti per compatibilità di firma con le versioni precedenti, non usati dal motore hybrid (che non ha stato/calibrazione: è una funzione pura, vedi CHANGELOG 1.1.0). soglia_anomalia — soglia fissa opzionale (None = calcolata dai dati); livello_ia — 0=neonata (filtra) .. 1=matura (solo marca).

Source code in dense_armor/armatura.py

def __init__(self, static_threshold: float = 0.15, initial_damping: float = 0.85,
             alpha: float = 0.05, soglia_anomalia: float | None = None,
             livello_ia: float = 1.0) -> None:
    """static_threshold/initial_damping/alpha — mantenuti per compatibilità di
    firma con le versioni precedenti, non usati dal motore hybrid (che non ha
    stato/calibrazione: è una funzione pura, vedi CHANGELOG 1.1.0).
    soglia_anomalia — soglia fissa opzionale (None = calcolata dai dati);
    livello_ia — 0=neonata (filtra) .. 1=matura (solo marca)."""
    self.soglia = soglia_anomalia
    self.livello_ia = float(min(1.0, max(0.0, livello_ia)))

set_livello

set_livello(livello: float) -> None

Aggiorna il livello dell'IA (0=debole -> filtra; 1=matura -> marca).

Source code in dense_armor/armatura.py

def set_livello(self, livello: float) -> None:
    """Aggiorna il livello dell'IA (0=debole -> filtra; 1=matura -> marca)."""
    self.livello_ia = float(min(1.0, max(0.0, livello)))

livello_da_entropia staticmethod

livello_da_entropia(
    entropia: float, vocab_size: int
) -> float

Livello MISURATO: 1 - entropia/log(V) (stessa formula dello sfogo dinamico).

Source code in dense_armor/armatura.py

@staticmethod
def livello_da_entropia(entropia: float, vocab_size: int) -> float:
    """Livello MISURATO: 1 - entropia/log(V) (stessa formula dello sfogo dinamico)."""
    import math
    return max(0.0, min(1.0, 1.0 - entropia / (math.log(max(2, vocab_size)) + 1e-12)))

analizza

analizza(
    serie: ndarray, riferimento: ndarray | None = None
) -> tuple[np.ndarray, np.ndarray, list[int]]

serie: 1D (lista/array). Per tensori N-D: passare tensore.ravel() e rifare reshape dopo (lo scudo e' agnostico: la trasposizione basta). Ritorna (pulito, K, indici_anomalie).

Source code in dense_armor/armatura.py

def analizza(
    self, serie: np.ndarray, riferimento: np.ndarray | None = None
) -> tuple[np.ndarray, np.ndarray, list[int]]:
    """serie: 1D (lista/array). Per tensori N-D: passare tensore.ravel()
    e rifare reshape dopo (lo scudo e' agnostico: la trasposizione basta).
    Ritorna (pulito, K, indici_anomalie)."""
    grezza = np.asarray(serie, dtype=np.float64).ravel()

    pulito_pieno, trigger, _meta = hybrid_shield(grezza, riferimento)

    # PERCEZIONE a K pieno (pavimento: mai clippata) — binario per design:
    # il trigger sottostante e' strettamente 0.0/1.0, non una sigmoide continua.
    K = 1.0 - trigger

    # grezza sanificata (solo NaN/Inf -> mediana locale), stessa
    # sanificazione usata internamente da hybrid_shield (vedi
    # _local_nan_fill): e' il "non tocca il dato" reale a clip zero,
    # non un segnale gia' pre-filtrato da uno stadio a monte. Mediana
    # LOCALE, non media dell'intera serie: altrimenti uno spike altrove
    # nella serie distorce il sostituto di ogni NaN, non solo quelli
    # vicini (vedi CHANGELOG 1.1.1).
    radius_fill = max(3 if len(grezza) < 3 else min(20, max(3, len(grezza) // 3)), 3)
    grezza_pulita = _local_nan_fill(np.where(np.isinf(grezza), np.nan, grezza), radius_fill)

    # INTERVENTO clippato dal livello dell'IA
    clip = 1.0 - self.livello_ia
    pulito = grezza_pulita - (K * clip) * (grezza_pulita - pulito_pieno)

    dev = np.abs(grezza_pulita - pulito_pieno)
    soglia = self.soglia if self.soglia is not None else (dev.mean() + 2.0 * dev.std())
    anomalie = {int(i) for i in np.where(dev > max(soglia, 1e-12))[0]}

    # rilevatore robusto (mediana/MAD), indipendente dal motore hybrid —
    # copre anche i primi 2 punti della serie, che il ciclo del trigger
    # non valuta mai (vedi limite noto in hybrid_engine.hybrid_shield)
    finiti = grezza[np.isfinite(grezza)]
    if finiti.size >= 4:
        med = float(np.median(finiti))
        mad = float(np.median(np.abs(finiti - med))) or 1e-12
        z_rob = np.abs((np.nan_to_num(grezza, nan=med) - med) / (1.4826 * mad))
        anomalie |= {int(i) for i in np.where(z_rob > 6.0)[0]}

    anomalie |= {int(i) for i in np.where(~np.isfinite(grezza))[0]}
    return pulito, K, sorted(anomalie)

deriva

deriva(serie: ndarray) -> tuple[float, bool, float]

RILEVATORE DI DERIVA LENTA (usa METRO, come suggerito da Salvatore).

La rana bollita: incrementi infinitesimi (es. +0.004/passo) invisibili allo scudo punto-per-punto. Qui si guarda la TENDENZA degli incrementi: se la loro media e' sistematica rispetto al rumore, c'e' deriva — e METRO (scaler logaritmico) ne risolve la scala anche quando e' sotto la precisione visibile. Funziona SENZA baseline storica (ma la baseline via riferimento in analizza() resta il metodo piu' forte).

Ritorna (tasso_per_passo, significativa: bool, esponente_metro).

Source code in dense_armor/armatura.py

def deriva(self, serie: np.ndarray) -> tuple[float, bool, float]:
    """RILEVATORE DI DERIVA LENTA (usa METRO, come suggerito da Salvatore).

    La rana bollita: incrementi infinitesimi (es. +0.004/passo) invisibili
    allo scudo punto-per-punto. Qui si guarda la TENDENZA degli incrementi:
    se la loro media e' sistematica rispetto al rumore, c'e' deriva — e
    METRO (scaler logaritmico) ne risolve la scala anche quando e' sotto
    la precisione visibile. Funziona SENZA baseline storica (ma la baseline
    via `riferimento` in analizza() resta il metodo piu' forte).

    Ritorna (tasso_per_passo, significativa: bool, esponente_metro).
    """
    v = np.asarray(serie, dtype=np.float64).ravel()
    v = v[np.isfinite(v)]
    if v.size < 9:
        return 0.0, False, 0.0
    # la firma della deriva e' lo SPOSTAMENTO CUMULATIVO: mediana del primo
    # terzo vs mediana dell'ultimo terzo (robusto agli spike), confrontato
    # col rumore della serie. Gli incrementi singoli restano nel rumore —
    # e' la somma che tradisce la rana.
    terzo = v.size // 3
    med_inizio = float(np.median(v[:terzo]))
    med_fine = float(np.median(v[-terzo:]))
    spostamento = med_fine - med_inizio
    tasso = spostamento / max(1, v.size - terzo)
    mad_serie = float(np.median(np.abs(v - np.median(v)))) or 1e-12
    rumore_mediane = 1.4826 * mad_serie / np.sqrt(terzo)
    significativa = abs(spostamento) > 3.0 * rumore_mediane
    # METRO: risolve la scala dell'infinitesimo (leggibile anche a 1e-12)
    _v_new, fact = Metro().enc(abs(tasso))
    esponente = float(np.log10(fact)) - (-4.0) if fact > 0 else 0.0  # = -log10(|tasso|)
    return tasso, bool(significativa), esponente

referto_json

referto_json(
    serie: ndarray,
    nome: str = "serie",
    riferimento: ndarray | None = None,
) -> dict

Referto MACCHINA-LEGGIBILE (dict pronto per json.dumps). Pensato per IA e altri programmi: niente prosa, solo dati. I valori non finiti diventano stringhe ("NaN") per dare JSON valido.

Source code in dense_armor/armatura.py

def referto_json(
    self, serie: np.ndarray, nome: str = "serie", riferimento: np.ndarray | None = None
) -> dict:
    """Referto MACCHINA-LEGGIBILE (dict pronto per json.dumps).
    Pensato per IA e altri programmi: niente prosa, solo dati.
    I valori non finiti diventano stringhe ("NaN") per dare JSON valido."""
    pulito, K, anomalie = self.analizza(serie, riferimento)
    s = np.asarray(serie, dtype=np.float64).ravel()
    tasso, sig, esp = self.deriva(serie)
    def _num(v: float) -> float | str:
        """Converte in float, o in "NaN" (stringa, per restare JSON-valido) se non finito."""
        return float(v) if np.isfinite(v) else "NaN"
    return {
        "nome": nome,
        "punti": int(s.size),
        "livello_ia": self.livello_ia,
        "anomalie": [{"indice": int(i), "grezzo": _num(s[i]),
                      "pulito": _num(pulito[i]), "K": _num(K[i])} for i in anomalie],
        "K_medio": _num(np.mean(K)),
        "deriva": {"tasso_per_passo": _num(tasso), "significativa": bool(sig),
                   "esponente_metro": _num(esp)},
    }

referto

referto(
    serie: ndarray,
    nome: str = "serie",
    riferimento: ndarray | None = None,
) -> list[int]

Analisi + referto leggibile.

Source code in dense_armor/armatura.py

def referto(
    self, serie: np.ndarray, nome: str = "serie", riferimento: np.ndarray | None = None
) -> list[int]:
    """Analisi + referto leggibile."""
    pulito, K, anomalie = self.analizza(serie, riferimento)
    s = np.asarray(serie, dtype=np.float64).ravel()
    print(f"[ARMATURA] {nome}: {len(s)} punti | livello IA {self.livello_ia:.2f} | anomalie: {len(anomalie)}")
    for i in anomalie[:10]:
        print(f"   -> indice {i}: valore {s[i]!r} (pulito: {pulito[i]:.4f}, K={K[i]:.3f})")
    if len(anomalie) > 10:
        print(f"   ... e altre {len(anomalie) - 10}")
    tasso, sig, esp = self.deriva(serie)
    if sig:
        print(f"   ⚠ DERIVA LENTA rilevata (Metro): {tasso:+.2e} per passo (scala 10^-{esp:.1f})")
    return anomalie

main

main() -> None

Entry point CLI: python -m dense_armor [--json] <file_o_numeri...>.

Source code in dense_armor/armatura.py

def main() -> None:
    """Entry point CLI: `python -m dense_armor [--json] <file_o_numeri...>`."""
    import re as _re
    argv = sys.argv[1:]
    come_json = "--json" in argv
    argv = [a for a in argv if a != "--json"]
    if not argv:
        print("USO: python -m dense_armor [--json] <file_con_numeri>  oppure  python -m dense_armor [--json] n1 n2 n3 ...")
        sys.exit(0)
    if len(argv) == 1 and os.path.exists(argv[0]):
        with open(argv[0], encoding="utf-8", errors="ignore") as fh:
            testo = fh.read()
        numeri = [float(x) if x.lower() != "nan" else float("nan")
                  for x in _re.findall(r"[-+]?(?:\d+\.?\d*(?:[eE][-+]?\d+)?|nan|NaN)", testo)]
        nome = os.path.basename(argv[0])
    else:
        numeri = [float(x) if x.lower() != "nan" else float("nan") for x in argv]
        nome = "serie da riga di comando"
    serie = np.asarray(numeri, dtype=np.float64)
    if come_json:
        import json as _json
        print(_json.dumps(Armatura().referto_json(serie, nome), ensure_ascii=False))
    else:
        Armatura().referto(serie, nome)

---

Details

See also: Hybrid engine — the core/hybrid_engine.py module Armatura is built on. Orca — the full input + output shield for an entire model, if you have a model rather than a bare series.