Hybrid engine (Armatura's core)

This is the part inside Armatura that decides, reading by reading, whether a value is genuine or broken, and what to put in place of a broken one: the middle value of the previous readings.

Armatura is the front door for a 1D series: it takes an array, returns a cleaned array. Behind it sits hybrid_engine, the module that actually decides, sample by sample, whether a value is a genuine reading or a bad one — and if bad, what to put in its place.

Each sample is in one of two states: dynamic (a genuine movement of the signal, kept as measured) or static (replaced by the median of the previous window). A binary trigger decides, from how the sample relates to the recent past.

1. Call the engine

import numpy as np
from dense_armor.core.hybrid_engine import hybrid_shield

rng = np.random.default_rng(42)
x = 0.5 + 0.1 * rng.standard_normal(300)
x[100], x[200] = 9999.0, float("nan")
clean, trigger, info = hybrid_shield(x)
print(round(float(clean[100]), 3), trigger[100], round(float(trigger.mean()), 2), info)

0.463 0.0 0.88 {'fallback_triggered': True, 'adaptive_radius_used': 20}

The engine returns the cleaned series, the trigger per sample (1 = dynamic, kept; 0 = static, replaced) and a small dict: whether any sample fell back, and the window radius used, min(20, max(3, n // 3)) (20 here). The spike at 100 has trigger 0 and becomes 0.463.

2. The trigger, symbol by symbol

For sample $x_i$, with $b_i$ the mean of the previous R samples (or riferimento[i] if a reference series is given), $s_i$ the local scale (standard deviation of the differences in the window) and $g_i$ the sign of the previous step $x_{i-1} - x_{i-2}$:

$$\Phi_i = \mathrm{clip}\Big(0.6\,\tfrac{1 + \mathrm{sign}(x_i - b_i)\,g_i}{2} + 0.4\,\big(1 - \tfrac{|x_i - b_i|}{s_i}\big),\; 0,\; 1\Big),$$

$$v_i = 5\,\mathrm{clip}\big(\ln(|x_i| / |b_i|),\, -5,\, 5\big)\,\Phi_i, \qquad \text{trigger}_i = \big[\,|v_i| > 0.01\,\big].$$

• $\Phi_i$ mixes alignment (does the change continue the direction of the last step?) and coherence (is the change small compared with the local scale?).
• $v_i$, the dynamic vector, is the logarithmic change of magnitude weighted by $\Phi_i$.
• A huge spike has $|x_i - b_i| \gg s_i$, so the coherence term drives $\Phi_i$ to 0, $v_i = 0$, trigger 0: the sample is replaced by the window median.

The first two samples are never tested (the step direction needs two previous points).

3. What the trigger also replaces

On the example the trigger is 1 on 88 % of the samples. Besides the spike and the NaN, it also classifies as static ordinary samples that hardly differ from the baseline ($v_i$ below 0.01), and replaces them with the window median:

import numpy as np
from dense_armor.core.hybrid_engine import hybrid_shield

rng = np.random.default_rng(42)
x = 0.5 + 0.1 * rng.standard_normal(300)
x[100], x[200] = 9999.0, float("nan")
clean, trigger, info = hybrid_shield(x)
d = np.abs(clean - x)[np.isfinite(x) & (np.arange(300) != 100)]
print(int((d > 1e-12).sum()), round(float(d.max()), 4))

35 0.2935

35 ordinary samples out of 298 change, by up to 0.29 (about three times the noise level), and they are not counted as anomalies by Armatura. Keep this in mind on smooth signals.

4. Non-finite values

Inf is turned into NaN and every NaN is filled from its local neighbourhood before the trigger runs, so the windows never contain a non-finite value.

API reference

hybrid_engine

Binary-trigger healing engine for Armatura.

Replaces the earlier AdaptiveSignalStabilizer + ABCollatz stage. The ABCollatz gate was mathematically non-discriminating: it was verified with a numerical sweep and never demonstrably fixed.

The pure functions below are vendored from Dense-Evolution/dense_evolution/healing.py (calculate_phi_ab, calculate_vettore_dinamico, evaluate_phi_trigger, GLOBAL_CONSTANTS), tested in that repository and already in production (dense-evolution >= 8.1.9 on PyPI). They are vendored rather than imported at runtime so the core stays on numpy + jax + psutil; dense-evolution remains an optional extra ([quantum]).

Deviation from upstream. calculate_phi_ab originally normalized the state_A / state_B distance against a fixed constant (MAX_SEMANTIC_DISTANCE = sqrt(2)), correct when state_A and state_B are already normalized vectors. Here they are raw scalars of arbitrary scale (a loss, a sensor reading, a metric). With the fixed constant any real jump saturated coherence_component to very negative values, so phi_ab was clipped to 0 for any genuine change larger than ~1.4: the trigger stayed "static" forever even in front of a sustained step that never isolated. Normalizing on the absolute scale of state_A / state_B fixed the step but made a huge jump look proportionally coherent, so an isolated spike passed untouched. The final rule normalizes the distance on the local volatility of the recent window (scaled MAD, robust to a single residual outlier). A huge jump with respect to the usual oscillation of the signal stays suspicious (spike filtered); a sustained step is judged by how anomalous it is relative to the normal noise, not by the magnitude of the number.

hybrid_shield generalizes the two-state logic of ia_utils.vector_healing.enhanced_dense_healing_hybrid (a sequence of vectors) to a 1D scalar series. The two states are strict: for every point the trigger decided by evaluate_phi_trigger is 0.0 (static/noise) or 1.0 (dynamic/genuine), with no intermediate branch.

calculate_phi_ab

calculate_phi_ab(
    state_A: ndarray,
    state_B: ndarray,
    ipg_vector: ndarray,
    scale: ndarray,
) -> jnp.ndarray

Alignment and spatial coherence factor Phi_AB.

scale is the local volatility of the recent window (scaled MAD, computed by the caller), not a fixed constant and not the absolute magnitude of state_A / state_B. See the module docstring for the deviation from upstream.

Parameters:

Name	Type	Description	Default
state_A	ndarray	the reference state (baseline).	required
state_B	ndarray	the current state (raw sample).	required
ipg_vector	ndarray	the unit direction of the recent raw changes.	required
scale	ndarray	the local volatility of the recent window.	required

Returns:

Type	Description
ndarray	phi_ab, clipped to [0.0, 1.0].

Source code in dense_armor/core/hybrid_engine.py

@jax.jit
def calculate_phi_ab(
    state_A: jnp.ndarray,
    state_B: jnp.ndarray,
    ipg_vector: jnp.ndarray,
    scale: jnp.ndarray,
) -> jnp.ndarray:
    """Alignment and spatial coherence factor Phi_AB.

    ``scale`` is the local volatility of the recent window (scaled MAD,
    computed by the caller), not a fixed constant and not the absolute
    magnitude of ``state_A`` / ``state_B``. See the module docstring for
    the deviation from upstream.

    Args:
        state_A: the reference state (baseline).
        state_B: the current state (raw sample).
        ipg_vector: the unit direction of the recent raw changes.
        scale: the local volatility of the recent window.

    Returns:
        ``phi_ab``, clipped to ``[0.0, 1.0]``.
    """
    semantic_change = state_B - state_A
    norm_change = jnp.linalg.norm(semantic_change)
    norm_ipg = jnp.linalg.norm(ipg_vector)

    alignment = jnp.where(
        (norm_change > 1e-12) & (norm_ipg > 1e-12),
        jnp.dot(semantic_change, ipg_vector) / (norm_change * norm_ipg),
        0.0,
    )
    semantic_alignment = (alignment + 1.0) / 2.0

    distance_A_B = jnp.linalg.norm(state_A - state_B)
    coherence_component = 1.0 - (distance_A_B / scale)

    phi_ab = (
        semantic_alignment * GLOBAL_CONSTANTS["WEIGHT_SEMANTIC"]
        + coherence_component * GLOBAL_CONSTANTS["WEIGHT_COHERENCE"]
    )
    return jnp.clip(phi_ab, 0.0, 1.0)

calculate_vettore_dinamico

calculate_vettore_dinamico(
    E_A: ndarray, E_B: ndarray, Phi_AB: ndarray
) -> jnp.ndarray

Dynamic vector (V_dinamic): differential logarithmic energy change.

Parameters:

Name	Type	Description	Default
E_A	ndarray	reference energy.	required
E_B	ndarray	current energy.	required
Phi_AB	ndarray	the alignment factor from :func:calculate_phi_ab.	required

Returns:

Type	Description
ndarray	The dynamic vector value, or 0.0 when either energy is below the
ndarray	numerical floor.

Source code in dense_armor/core/hybrid_engine.py

@jax.jit
def calculate_vettore_dinamico(
    E_A: jnp.ndarray, E_B: jnp.ndarray, Phi_AB: jnp.ndarray
) -> jnp.ndarray:
    """Dynamic vector (V_dinamic): differential logarithmic energy change.

    Args:
        E_A: reference energy.
        E_B: current energy.
        Phi_AB: the alignment factor from :func:`calculate_phi_ab`.

    Returns:
        The dynamic vector value, or 0.0 when either energy is below the
        numerical floor.
    """
    valid_inputs = (E_A > 1e-12) & (E_B > 1e-12)
    ratio = jnp.where(valid_inputs, E_B / E_A, 1.0)
    log_ratio_clamped = jnp.clip(jnp.log(ratio), -5.0, 5.0)
    v_vita = GLOBAL_CONSTANTS["V_DINAMIC_K_COEFF"] * log_ratio_clamped * Phi_AB
    return jnp.where(valid_inputs, v_vita, 0.0)

evaluate_phi_trigger

evaluate_phi_trigger(
    deterministic_dq_dt_a: ndarray,
) -> jnp.ndarray

Binary trigger: 1.0 (dynamic) or 0.0 (static).

Parameters:

Name	Type	Description	Default
deterministic_dq_dt_a	ndarray	the dynamic vector value.	required

Returns:

Type	Description
ndarray	1.0 when the magnitude of the input exceeds NON_STATIC_THRESHOLD_A,
ndarray	0.0 otherwise.

Source code in dense_armor/core/hybrid_engine.py

@jax.jit
def evaluate_phi_trigger(deterministic_dq_dt_a: jnp.ndarray) -> jnp.ndarray:
    """Binary trigger: 1.0 (dynamic) or 0.0 (static).

    Args:
        deterministic_dq_dt_a: the dynamic vector value.

    Returns:
        1.0 when the magnitude of the input exceeds ``NON_STATIC_THRESHOLD_A``,
        0.0 otherwise.
    """
    magnitude_change_a = jnp.abs(deterministic_dq_dt_a)
    trigger_active = magnitude_change_a > GLOBAL_CONSTANTS["NON_STATIC_THRESHOLD_A"]
    return jnp.where(trigger_active, 1.0, 0.0)

hybrid_shield

hybrid_shield(
    serie: ndarray,
    riferimento: ndarray | None = None,
    radius_baseline: int | None = None,
) -> tuple[np.ndarray, np.ndarray, dict]

Apply the two-state scheme (pass-through / fallback to baseline) point by point.

Windows are always taken from processed (the raw, sanitized series), never from the previous output out. Using out protects from a past outlier but introduces a worse bug on a genuine sustained step: the first one or two points after the transition are inevitably rejected (no causal detector can distinguish a true step from a spike on the very first sample), those rejected points end up in out, and the window of the next points feeds them back into the calculation, producing a "halfway" baseline that also rejects the genuinely new points. The result is a stable but wrong equilibrium that never recovers (verified: 30 real points after a step from 1.0 to 5.0 stayed stuck at ~4.0 forever, never reaching 5.0). The median fallback is robust to a single spike inside the window without the self-contamination risk.

The IPG is also taken from processed, not from out: the IPG must see whether the recent raw values are actually moving in a new direction. Computing it from out creates a self-feeding block: a rejected point is replaced by the baseline, so out never shows evidence of the new value, the IPG stays at zero, and no real step can ever be recognized no matter how long it lasts (verified: with IPG from out a 10-point step stayed flattened at 100%, no escape).

Known limitation. The first two points (i = 0 and i = 1) never go through the trigger loop (an IPG from the two previous points is required), so they always pass through with trigger = 1.0, regardless of their value. The behaviour is inherited unchanged from enhanced_dense_healing_hybrid and is negligible on realistic series; any NaN or spike on those two points must be caught by independent upstream checks (the robust filters in Armatura.analizza).

Parameters:

Name	Type	Description	Default
serie	ndarray	a 1D array of floats. NaN and Inf are accepted and sanitized.	required
riferimento	ndarray | None	when given, the baseline for point i is riferimento[i] (anti-drift mode). When None (blind mode), the baseline is the local mean of the previous points of the same series, with adaptive radius min(20, max(3, n // 3)).	None
radius_baseline	int | None	optional fixed radius for the blind-mode window.	None

Returns:

Type	Description
ndarray	A tuple (pulito, trigger, metadata) where pulito is the
ndarray	output series (raw value where trigger == 1, baseline where
dict	trigger == 0), trigger is the 0.0 / 1.0 indicator array, and
tuple[ndarray, ndarray, dict]	metadata holds fallback_triggered and
tuple[ndarray, ndarray, dict]	adaptive_radius_used.

Source code in dense_armor/core/hybrid_engine.py

def hybrid_shield(
    serie: np.ndarray,
    riferimento: np.ndarray | None = None,
    radius_baseline: int | None = None,
) -> tuple[np.ndarray, np.ndarray, dict]:
    """Apply the two-state scheme (pass-through / fallback to baseline) point by point.

    Windows are always taken from ``processed`` (the raw, sanitized series),
    never from the previous output ``out``. Using ``out`` protects from a
    past outlier but introduces a worse bug on a genuine sustained step: the
    first one or two points after the transition are inevitably rejected (no
    causal detector can distinguish a true step from a spike on the very
    first sample), those rejected points end up in ``out``, and the window
    of the next points feeds them back into the calculation, producing a
    "halfway" baseline that also rejects the genuinely new points. The
    result is a stable but wrong equilibrium that never recovers (verified:
    30 real points after a step from 1.0 to 5.0 stayed stuck at ~4.0
    forever, never reaching 5.0). The median fallback is robust to a single
    spike inside the window without the self-contamination risk.

    The IPG is also taken from ``processed``, not from ``out``: the IPG must
    see whether the recent *raw* values are actually moving in a new
    direction. Computing it from ``out`` creates a self-feeding block: a
    rejected point is replaced by the baseline, so ``out`` never shows
    evidence of the new value, the IPG stays at zero, and no real step can
    ever be recognized no matter how long it lasts (verified: with IPG from
    ``out`` a 10-point step stayed flattened at 100%, no escape).

    Known limitation. The first two points (``i = 0`` and ``i = 1``) never
    go through the trigger loop (an IPG from the two previous points is
    required), so they always pass through with ``trigger = 1.0``,
    regardless of their value. The behaviour is inherited unchanged from
    ``enhanced_dense_healing_hybrid`` and is negligible on realistic
    series; any NaN or spike on those two points must be caught by
    independent upstream checks (the robust filters in
    ``Armatura.analizza``).

    Args:
        serie: a 1D array of floats. NaN and Inf are accepted and sanitized.
        riferimento: when given, the baseline for point ``i`` is
            ``riferimento[i]`` (anti-drift mode). When ``None`` (blind
            mode), the baseline is the local mean of the previous points of
            the same series, with adaptive radius
            ``min(20, max(3, n // 3))``.
        radius_baseline: optional fixed radius for the blind-mode window.

    Returns:
        A tuple ``(pulito, trigger, metadata)`` where ``pulito`` is the
        output series (raw value where ``trigger == 1``, baseline where
        ``trigger == 0``), ``trigger`` is the 0.0 / 1.0 indicator array, and
        ``metadata`` holds ``fallback_triggered`` and
        ``adaptive_radius_used``.
    """
    s = np.asarray(serie, dtype=np.float64).ravel()
    n = s.size

    if n == 0:
        return (
            np.empty(0),
            np.empty(0),
            {"fallback_triggered": False, "adaptive_radius_used": 0},
        )

    if radius_baseline is None:
        adaptive_radius_used = 0 if n < 3 else min(20, max(3, n // 3))
    else:
        adaptive_radius_used = radius_baseline
    fill_radius = max(adaptive_radius_used, 3)

    processed = np.copy(s)
    processed[np.isinf(processed)] = np.nan
    processed = _local_nan_fill(processed, fill_radius)

    rif = None
    if riferimento is not None:
        rif = np.asarray(riferimento, dtype=np.float64).ravel()
        if rif.size != n:
            raise ValueError(f"reference has {rif.size} points, series has {n}")
        rif = np.where(np.isinf(rif), np.nan, rif)
        rif = _local_nan_fill(rif, fill_radius)

    if n <= 2:
        return (
            np.copy(processed),
            np.ones(n),
            {"fallback_triggered": False, "adaptive_radius_used": adaptive_radius_used},
        )

    R = adaptive_radius_used
    processed_j = jnp.asarray(processed)
    padded = jnp.concatenate([jnp.full(R, jnp.nan), processed_j])

    def _get_window(i):
        return jax.lax.dynamic_slice(padded, (i,), (R,))

    idx = jnp.arange(n)
    windows = jax.vmap(_get_window)(idx)
    count_valid = jnp.minimum(idx, R)
    valid_mask = jnp.arange(R)[None, :] >= (R - count_valid[:, None])

    def _masked_mean(row, mask):
        return jnp.sum(jnp.where(mask, row, 0.0)) / jnp.maximum(jnp.sum(mask), 1)

    def _masked_median(row, mask, k):
        sorted_vals = jnp.sort(jnp.where(mask, row, jnp.inf))
        mid = k // 2
        return jnp.where(
            k % 2 == 0,
            (sorted_vals[mid - 1] + sorted_vals[mid]) / 2.0,
            sorted_vals[mid],
        )

    def _masked_std_of_diff(row, mask, k):
        """Local volatility: standard deviation of the successive differences.

        Deliberately not the MAD of the window values. On a window that
        straddles a true transition, as soon as the majority of the points
        reach the new level the median of the values collapses onto the new
        level and the MAD drops to ~0 (the pre-transition minority
        disappears from the metric), causing an oscillating rejection right
        after a correct acceptance. The standard deviation of the deltas
        does not collapse: a single large difference (the transition itself)
        raises the std instead of vanishing behind the median.

        Known trade-off: an isolated spike temporarily raises this
        volatility (two large differences, in and out of the spike) for
        ``radius`` steps, making the engine more permissive in that window.
        It self-corrects when the spike leaves the window; it is not a
        permanent block.
        """
        diffs = row[1:] - row[:-1]
        diff_mask = mask[1:] & mask[:-1]
        cnt = jnp.maximum(jnp.sum(diff_mask), 1)
        mean_d = jnp.sum(jnp.where(diff_mask, diffs, 0.0)) / cnt
        var_d = jnp.sum(jnp.where(diff_mask, (diffs - mean_d) ** 2, 0.0)) / cnt
        return jnp.where(k >= 3, jnp.sqrt(var_d), 0.0)

    baseline_mean = jax.vmap(_masked_mean)(windows, valid_mask)
    median_vals = jax.vmap(_masked_median)(windows, valid_mask, count_valid)
    local_scale = jax.vmap(_masked_std_of_diff)(windows, valid_mask, count_valid)
    scale = jnp.maximum(local_scale, 1e-6)

    baseline = jnp.asarray(rif) if rif is not None else baseline_mean
    state_A = baseline
    state_B = processed_j

    d = jnp.diff(processed_j)
    ipg_raw = jnp.concatenate([jnp.zeros(2), d[:-1]])
    norm_ipg_raw = jnp.abs(ipg_raw)
    ipg_vector = jnp.where(
        norm_ipg_raw > 1e-9,
        ipg_raw / jnp.maximum(norm_ipg_raw, 1e-30),
        ipg_raw,
    )

    semantic_change = state_B - state_A
    norm_change = jnp.abs(semantic_change)
    norm_ipg = jnp.abs(ipg_vector)
    alignment = jnp.where(
        (norm_change > 1e-12) & (norm_ipg > 1e-12),
        (semantic_change * ipg_vector) / (norm_change * norm_ipg),
        0.0,
    )
    semantic_alignment = (alignment + 1.0) / 2.0
    distance_A_B = jnp.abs(state_A - state_B)
    coherence_component = 1.0 - (distance_A_B / scale)
    phi_ab = jnp.clip(
        semantic_alignment * GLOBAL_CONSTANTS["WEIGHT_SEMANTIC"]
        + coherence_component * GLOBAL_CONSTANTS["WEIGHT_COHERENCE"],
        0.0,
        1.0,
    )

    E_A = jnp.abs(state_A)
    E_B = jnp.abs(state_B)
    valid_energy = (E_A > 1e-12) & (E_B > 1e-12)
    ratio = jnp.where(valid_energy, E_B / jnp.where(E_A > 1e-12, E_A, 1.0), 1.0)
    log_ratio_clamped = jnp.clip(jnp.log(ratio), -5.0, 5.0)
    v_dinamic = jnp.where(
        valid_energy,
        GLOBAL_CONSTANTS["V_DINAMIC_K_COEFF"] * log_ratio_clamped * phi_ab,
        0.0,
    )

    trigger_full = jnp.where(
        jnp.abs(v_dinamic) > GLOBAL_CONSTANTS["NON_STATIC_THRESHOLD_A"],
        1.0,
        0.0,
    )
    fallback_val = median_vals if rif is None else baseline
    out_full = jnp.where(
        trigger_full > GLOBAL_CONSTANTS["NON_STATIC_THRESHOLD_A"],
        processed_j,
        fallback_val,
    )

    out = np.array(out_full)
    trigger_arr = np.array(trigger_full)
    out[0], out[1] = processed[0], processed[1]
    trigger_arr[0], trigger_arr[1] = 1.0, 1.0

    fallback_triggered_at_all = bool(
        np.any(trigger_arr[2:] < GLOBAL_CONSTANTS["NON_STATIC_THRESHOLD_A"])
    )
    metadata = {
        "fallback_triggered": fallback_triggered_at_all,
        "adaptive_radius_used": adaptive_radius_used,
    }
    return out, trigger_arr, metadata

---

Details

The engine generalises the logic already verified in Dense-Evolution's own ia_utils.vector_healing.enhanced_dense_healing_hybrid. The trigger and the NaN handling are the same as there; this version computes all samples in one vectorised call (jax.vmap).

See also: Armatura — the wearable class built on this engine.