Adaptive engine (Orca Stage 1)

engine

core/engine.py

AdaptiveSignalStabilizer — filtro adattivo per serie temporali e gradienti IA.

Caratteristiche

• Rilevazione di anomalie tramite soglia dinamica basata su volatilità rolling.
• Damping selettivo con preservazione della norma L2 per evitare esplosioni locali.
• Calibrazione macro-contesto su batch 2D (N_scenari × T_steps) con parametri dinamici.
• Implementazione interamente JAX-compatibile, con kernel vmap + scan precompilati.

AdaptiveSignalStabilizer

AdaptiveSignalStabilizer(static_threshold: float = 1e-05, initial_damping: float = 0.1, alpha: float = _SIGMA, anomaly_sigma_mult: float = 2.0, k_anom_min: float = _ALPHA, k_anom_max: float = _K_MAX_ZETA, window_radius: int = 1, smooth_l2_blend: float = 0.5)

Filtro antipanico adattivo per segnali IA caotici.

Configura soglie/gain iniziali e precompila i kernel JAX (scan 1D, batch vmap).

Source code in dense_armor/core/engine.py

def __init__(
    self,
    static_threshold: float = 1e-5,
    initial_damping: float = 0.1,
    alpha: float = _SIGMA,               # <-- Agganciato a PHI
    anomaly_sigma_mult: float = 2.0,
    k_anom_min: float = _ALPHA,          # <-- Agganciato a PHI
    k_anom_max: float = _K_MAX_ZETA,     # <-- costante fissa (vedi sopra); sostituisce il vecchio _ALPHA+_SIGMA
    window_radius: int = 1,
    smooth_l2_blend: float = 0.5,
) -> None:
    """Configura soglie/gain iniziali e precompila i kernel JAX (scan 1D, batch vmap)."""
    # Parametri di configurazione (CPU side)
    self.threshold: float = float(static_threshold)
    self.damping: float = float(initial_damping)
    self.alpha: float = float(alpha)

    # Iperparametri per la gestione delle anomalie
    self.anomaly_sigma_mult: float = float(anomaly_sigma_mult)
    self.k_anom_min: float = float(k_anom_min)
    self.k_anom_max: float = float(k_anom_max)
    self.window_radius: int = int(max(window_radius, 0))
    self.smooth_l2_blend: float = float(np.clip(smooth_l2_blend, 0.0, 1.0))

    # Parametri dinamici (calibrati per-batch a runtime)
    self.dyn_thr: float = self.threshold
    self.dyn_dmp: float = self.damping
    self.dyn_alp: float = self.alpha
    self.noise_scalar: float = 1.0

    # Kernel JAX precompilati
    self._compiled_stream_filter = self._build_stream_filter()
    self._compiled_batch_filter = jax.jit(
        jax.vmap(
            self._process_single_scenario,
            in_axes=(0, None, None, None, None, 0),
        )
    )
    # Kernel 1D usato da filter_data_stream: thr/dmp/alp/noise_scalar
    # passati come ARGOMENTI jit (come gia' fa _compiled_batch_filter
    # sopra), non chiusi su self.* dentro la funzione -- altrimenti
    # ogni jax.jit(lambda...) creato inline ad ogni chiamata sarebbe
    # un oggetto Python nuovo, cache-miss garantito, ricompilazione
    # XLA completa ad OGNI singola chiamata (~80ms fissi anche a
    # regime, misurato: la cache non si scalda mai).
    self._compiled_single_stream_filter = jax.jit(self._run_single_stream_scan)

calibrate_macro_context

calibrate_macro_context(raw_batch: ndarray) -> None

Calibra le soglie dinamiche sul contesto globale del batch.

Analizza la distribuzione dei salti fra campioni per individuare ambienti estremamente turbolenti (shock) rispetto a rumore fisiologico.

Source code in dense_armor/core/engine.py

def calibrate_macro_context(self, raw_batch: np.ndarray) -> None:
    """
    Calibra le soglie dinamiche sul contesto globale del batch.

    Analizza la distribuzione dei salti fra campioni per individuare
    ambienti estremamente turbolenti (shock) rispetto a rumore fisiologico.
    """
    if raw_batch.size == 0:
        return

    sample_diff = np.abs(np.diff(raw_batch))
    global_std = float(np.std(raw_batch))

    # Più lo std è alto, più riduciamo l'aggressività del filtro.
    self.noise_scalar = float(1.0 / (1.0 + global_std))

    # Con un solo campione per riga (es. batch di forma (1, 1)) il diff
    # è vuoto: non c'è un "salto" da misurare, quindi non possiamo
    # decidere il panic mode su questa base. Usciamo qui mantenendo
    # noise_scalar già calibrato sopra, invece di far esplodere
    # np.max/np.mean su un array vuoto.
    if sample_diff.size == 0:
        return

    max_jump = float(np.max(sample_diff))
    mean_jump = float(np.mean(sample_diff))

    # Regime “panic mode” per shock estremi
    if max_jump > 3.0 and (max_jump / (mean_jump + 1e-5)) > 5.0:
        self.dyn_thr = 1e-5
        self.dyn_dmp = 1e-3
        self.dyn_alp = 0.999
        return

    # Regime standard: i parametri dinamici scalano con la volatilità globale e la proporzione aurea
    self.dyn_thr = float(0.10 * global_std)
    self.dyn_dmp = float(_SIGMA * (1.0 + self.noise_scalar))
    self.dyn_alp = float(0.75 + (_SIGMA * self.noise_scalar))

filter_data_stream

filter_data_stream(raw_data: ndarray) -> np.ndarray

Filtra una singola serie temporale 1D garantendo la simmetria strutturale.

Source code in dense_armor/core/engine.py

def filter_data_stream(self, raw_data: np.ndarray) -> np.ndarray:
    """
    Filtra una singola serie temporale 1D garantendo la simmetria strutturale.
    """
    if raw_data.size == 0:
        return np.zeros_like(raw_data)

    j_raw = jnp.array(raw_data, dtype=jnp.float64)
    init_val = jnp.float64(j_raw[0])

    # Allineamento definitivo del carry a 4 elementi identico al batch engine
    init_state = (
        init_val,
        jnp.float64(self.damping),
        jnp.float64(0.0),
        init_val,
    )

    _, gated_stream = self._compiled_single_stream_filter(
        init_state,
        j_raw[1:],
        jnp.float64(self.dyn_thr),
        jnp.float64(self.dyn_dmp),
        jnp.float64(self.dyn_alp),
        jnp.float64(self.noise_scalar),
        init_val,
    )

    # Reinserimento del punto fisso iniziale preservando la topologia.
    # np.insert (non jnp.insert): il valore lascia JAX qui comunque, e
    # jnp.insert chiamato fuori da jit ricompila XLA da zero ad ogni
    # chiamata (~25ms fissi anche a regime, stesso problema del kernel
    # scan risolto sopra ma su un'operazione diversa).
    gated_np = np.asarray(gated_stream, dtype=np.float64)
    final_stream = np.insert(gated_np, 0, np.asarray(init_val, dtype=np.float64))
    return final_stream

filter_batch_scenarios

filter_batch_scenarios(raw_batch: ndarray, hard_clamp_mask: Optional[ndarray] = None) -> np.ndarray

Filtra in parallelo (vmap) un batch di scenari strutturati. Accetta tensori 2D nativi o array multidimensionali (3D/4D).

hard_clamp_mask: stessa shape di raw_batch, opzionale (default: nessuna macro-anomalia, comportamento identico a prima). Dove True, attiva lo State Flush dentro _step_kernel per quell'elemento -- vedi la sua docstring. Pensato per essere lo STESSO segnale gia' calcolato a monte da Orca (raw_noise sui dati grezzi > soglia critica), non una sua ricostruzione approssimata qui dentro.

Source code in dense_armor/core/engine.py

def filter_batch_scenarios(
    self, raw_batch: np.ndarray, hard_clamp_mask: Optional[np.ndarray] = None
) -> np.ndarray:
    """
    Filtra in parallelo (vmap) un batch di scenari strutturati.
    Accetta tensori 2D nativi o array multidimensionali (3D/4D).

    hard_clamp_mask: stessa shape di raw_batch, opzionale (default: nessuna
    macro-anomalia, comportamento identico a prima). Dove True, attiva lo
    State Flush dentro _step_kernel per quell'elemento -- vedi la sua
    docstring. Pensato per essere lo STESSO segnale gia' calcolato a
    monte da Orca (raw_noise sui dati grezzi > soglia critica), non una
    sua ricostruzione approssimata qui dentro."""
    if raw_batch.size == 0:
        return np.zeros_like(raw_batch)
    if hard_clamp_mask is None:
        hard_clamp_mask = np.zeros_like(raw_batch, dtype=bool)

    original_shape = raw_batch.shape
    n_scenarios = int(original_shape[0])

    # FIX DEFINITIVO SINTASSI: Estrarre il contesto in modo compatibile con l'ALU
    flat_analysis_batch = raw_batch.reshape(n_scenarios, -1)
    self.calibrate_macro_context(flat_analysis_batch)

    # Adattamento geometrico statico degli assi per non rompere la compilazione AOT/XLA
    if len(original_shape) == 2:
        h_dim = 1
        w_dim = int(original_shape[1])
        structured_batch = raw_batch.reshape(n_scenarios, h_dim, w_dim)
        structured_mask = np.asarray(hard_clamp_mask).reshape(n_scenarios, h_dim, w_dim)
    elif len(original_shape) == 3:
        h_dim, w_dim = int(original_shape[1]), int(original_shape[2])
        structured_batch = raw_batch
        structured_mask = np.asarray(hard_clamp_mask)
    elif len(original_shape) == 4:
        c_channels, h_dim, w_dim = int(original_shape[1]), int(original_shape[2]), int(original_shape[3])
        structured_batch = raw_batch.reshape(n_scenarios * c_channels, h_dim, w_dim)
        structured_mask = np.asarray(hard_clamp_mask).reshape(n_scenarios * c_channels, h_dim, w_dim)
    else:
        raise ValueError(f"Geometria del tensore non supportata dall'engine: {original_shape}")

    # Conversione in array JAX esplicito a precisione singola standard per registri GPU/CPU
    j_batch = jnp.array(structured_batch, dtype=jnp.float64)
    j_mask = jnp.array(structured_mask, dtype=jnp.bool_)

    # Esecuzione del kernel vettorizzato precompilato JAX
    filtered_structured = self._compiled_batch_filter(
        j_batch,
        jnp.float64(self.dyn_thr),
        jnp.float64(self.dyn_dmp),
        jnp.float64(self.dyn_alp),
        jnp.float64(self.noise_scalar),
        j_mask,
    )

    # Conversione sicura in NumPy preservando la topologia
    filtered_np = np.array(filtered_structured, dtype=np.float64)

    # Ripristino esatto della shape geometrica originale del chiamante
    return filtered_np.reshape(original_shape)

dynamic_damping_gain

dynamic_damping_gain(local_noise: ndarray, k_min: float = 0.0, k_max: float = _K_MAX_ZETA) -> jnp.ndarray

Curva di damping a sigmoide, funzione del rumore locale: - local_noise basso → gain vicino al minimo (massima trasparenza) - local_noise alto → gain vicino al massimo (massima purificazione)

Source code in dense_armor/core/engine.py

@jax.jit
def dynamic_damping_gain(local_noise: jnp.ndarray,
                         k_min: float = 0.0,         # <-- gain minimo (massima trasparenza)
                         k_max: float = _K_MAX_ZETA  # <-- gain massimo (costante fissa, vedi sopra)
                         ) -> jnp.ndarray:
    """
    Curva di damping a sigmoide, funzione del rumore locale:
    - local_noise basso → gain vicino al minimo (massima trasparenza)
    - local_noise alto  → gain vicino al massimo (massima purificazione)
    """
    x = jnp.clip(local_noise, 0.0, 5.0)
    steering = 1.0 / (1.0 + jnp.exp(-(x - 2.0)))
    return k_min + steering * (k_max - k_min)

---

See also: Orca, which uses AdaptiveSignalStabilizer as Stage 1, followed by a Collatz-based gate (Stage 2) deciding how much to damp toward the clean reference. This is also the engine exercised directly by the adversarial test suite (test/test_boundA-E.py) behind the robustness numbers on the home page.