Adaptive engine (Orca Stage 1)¶

This tool follows a signal that drifts slowly, keeps updating what it considers normal and pulls the signal gently toward it, so that a slow drift is not mistaken for a glitch.

A signal that drifts slowly is not a spike. It is a sequence of samples, each of them plausible on its own, that together move the average. A purely local detector misses this by construction: each sample is within the local window's amplitude, so nothing fires. Orca's Stage 1 is the module that handles this case: it does not judge each sample against a fixed baseline, it continuously re-estimates the baseline and damps toward it with a smooth curve.

The module is AdaptiveSignalStabilizer, a causal recursive filter written as a jax.lax.scan so it can run inside a JAX pipeline without leaving the JIT.

The examples use a 100 Hz channel with a spike at sample 1500 and a slow ramp from 0.5 to 0.7 starting at sample 2000.

1. Filter the series¶

import jax
jax.config.update("jax_enable_x64", True)
import numpy as np
from dense_armor.core.engine import AdaptiveSignalStabilizer

x = 0.5 + 0.1 * np.random.default_rng(42).standard_normal(3000)
x[1500] = 2.5
x[2000:] += 0.2 * (np.arange(1000) / 1000)

stabilizer = AdaptiveSignalStabilizer()
clean = stabilizer.filter_data_stream(np.asarray(x))
print(clean.shape, float(clean[1500]))

(3000,) 0.759195119486936

clean is the same series with the spike removed and the ramp followed but slightly smoothed — the filter does not fight a sustained change, it damps the transient part of it. At the spike position the value 2.5 comes out as 0.76: the filter damps the spike strongly but does not replace it with the baseline.

2. What "adaptive" means here¶

The filter measures the local noise of the signal (a rolling volatility) and turns it into a damping gain with a sigmoid:

$$k(n) = k_{min} + \frac{k_{max} - k_{min}}{1 + e^{-(n - 2)}}, \qquad n = \mathrm{clip}(\text{local noise},\, 0,\, 5).$$

Low local noise: the gain stays near $k_{min}$ and the signal passes almost untouched.
High local noise: the gain goes towards $k_{max}$ (about 8.78 by default) and the filter purifies strongly.
Samples beyond a dynamic threshold (anomaly_sigma_mult times the rolling volatility) are damped selectively, preserving the L2 norm of the series so the damping cannot create local explosions.

The sigmoid is why the filter is smooth: there is no single value at which it switches from "pass" to "reject".

3. Presets¶

Four calibrated parameter sets ship with the library:

preset	when to use
`balanced_v2`	default, good across signals
`cifar10_best_v1`	tuned for image-grid-like inputs
`pure_1d_time_v1`	reactive; follows fast changes more closely
`cifar10_hardened_lyapunov`	hardened; damps more aggressively

import numpy as np
from dense_armor.core.preset import SIGNAL_STABILIZER_PRESETS
from dense_armor.core.engine import AdaptiveSignalStabilizer

x = 0.5 + 0.1 * np.random.default_rng(42).standard_normal(3000)
x[1500] = 2.5

stabilizer = AdaptiveSignalStabilizer(**SIGNAL_STABILIZER_PRESETS["pure_1d_time_v1"])
clean = stabilizer.filter_data_stream(np.asarray(x))
print(round(float(np.var(np.asarray(x) - clean)), 6))

0.003074

The presets behave differently on the same series: pure_1d_time_v1 leaves a residual variance of 0.0031 against 0.0061 for balanced_v2 — it changes the signal about half as much, following it more closely.

4. Where Orca uses it¶

Orca.protect_and_forward calls the stabilizer twice: once on the input tensor before the model, once on the response after. The four-phase shield described on the Orca page is exactly this: purify, forward, purify, report trust scores. Stage 2 (Orca's Collatz-based gate) sits on top and decides how much of the damped value to use versus how much of the raw value to trust on the input side; on the output side the check is against the model's own response to the clean reference.

If you use AdaptiveSignalStabilizer on its own, you get the same soft damping without the input/output distinction: a single cleaned stream.

5. Batch and multi-scenario use¶

When you have a whole batch of signals (or several scenarios side by side), the stabilizer exposes batch methods that vmap over rows:

import numpy as np
from dense_armor.core.engine import AdaptiveSignalStabilizer

batch = np.random.default_rng(0).standard_normal((4, 500))

s = AdaptiveSignalStabilizer()
clean_batch = s.filter_batch_scenarios(np.asarray(batch))
print(clean_batch.shape)

(4, 500)

filter_batch_scenarios uses a shared calibration for the whole batch. filter_batch_scenarios_independent gives each row its own thresholds — useful when the rows come from different sensors with different amplitudes.

API reference¶

engine ¶

Adaptive signal stabilizer for time series and AI gradients.

Features

Anomaly detection through a dynamic threshold based on rolling volatility.
Selective damping with L2-norm preservation to avoid local blow-ups.
Macro-context calibration on 2D batches (N_scenarios, T_steps) with dynamic parameters.
Fully JAX-compatible, with precompiled vmap + scan kernels.

The numeric constants below are fixed hyper-parameters of the damping curve. They have no physical or mathematical meaning beyond shaping the gain.

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,
)

Adaptive anti-panic filter for chaotic AI signals.

Parameters:

Name	Type	Description	Default
`static_threshold`	`float`	base detection threshold.	`1e-05`
`initial_damping`	`float`	initial damping coefficient.	`0.1`
`alpha`	`float`	EMA coefficient for the rolling volatility.	`_SIGMA`
`anomaly_sigma_mult`	`float`	multiplier on the rolling volatility used to build the dynamic threshold.	`2.0`
`k_anom_min`	`float`	minimum gain on the anomalous branch.	`_ALPHA`
`k_anom_max`	`float`	maximum gain on the anomalous branch.	`_K_MAX_ZETA`
`window_radius`	`int`	radius of the local mean window.	`1`
`smooth_l2_blend`	`float`	blend between the raw smoothed value and its L2-rescaled version, in `[0.0, 1.0]`.	`0.5`

Configure the thresholds and precompile the JAX kernels.

Source code in dense_armor/core/engine.py

def __init__(
    self,
    static_threshold: float = 1e-5,
    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,
) -> None:
    """Configure the thresholds and precompile the JAX kernels."""
    self.threshold: float = float(static_threshold)
    self.damping: float = float(initial_damping)
    self.alpha: float = float(alpha)

    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))

    self.dyn_thr: float = self.threshold
    self.dyn_dmp: float = self.damping
    self.dyn_alp: float = self.alpha
    self.noise_scalar: float = 1.0

    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),
        )
    )
    self._compiled_batch_filter_independent = jax.jit(
        jax.vmap(
            self._process_single_scenario,
            in_axes=(0, 0, 0, 0, 0, 0),
        )
    )
    self._compiled_single_stream_filter = jax.jit(self._run_single_stream_scan)

calibrate_macro_context ¶

calibrate_macro_context(raw_batch: ndarray) -> None

Calibrate the dynamic thresholds on the global context of the batch.

Analyses the distribution of jumps between samples to tell an extremely turbulent environment (a shock) from physiological noise. The parameters are stored on the instance (dyn_thr, dyn_dmp, dyn_alp, noise_scalar); the method is not pure by design and leaves existing callers unchanged.

Parameters:

Name	Type	Description	Default
`raw_batch`	`ndarray`	the batch of raw signals.	required

Source code in dense_armor/core/engine.py

def calibrate_macro_context(self, raw_batch: np.ndarray) -> None:
    """Calibrate the dynamic thresholds on the global context of the batch.

    Analyses the distribution of jumps between samples to tell an
    extremely turbulent environment (a shock) from physiological noise.
    The parameters are stored on the instance (``dyn_thr``, ``dyn_dmp``,
    ``dyn_alp``, ``noise_scalar``); the method is not pure by design and
    leaves existing callers unchanged.

    Args:
        raw_batch: the batch of raw signals.
    """
    if raw_batch.size == 0:
        return

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

    self.noise_scalar = float(1.0 / (1.0 + global_std))

    if sample_diff.size == 0:
        return

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

    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

    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))

calibrate_macro_context_independent ¶

calibrate_macro_context_independent(
    raw_batch: ndarray,
) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]

Same formulas as :meth:calibrate_macro_context, computed per row.

The parameters are computed along axis=-1 instead of on the whole flattened batch. The method is pure and does not mutate self, so it is safe to use inside vmap downstream. The existing batch-shared :meth:calibrate_macro_context is kept unchanged for its callers (test_boundA.py and the adversarial suite use the shared calibration on purpose: correlated scenarios share one context).

Parameters:

Name	Type	Description	Default
`raw_batch`	`ndarray`	shape `(n_scenarios, F)`. When `F < 2` no row has a "jump" to measure and the method returns the `__init__` defaults (`self.threshold`/`damping`/`alpha`) for every row, instead of trying to replicate the stateful early-return of :meth:`calibrate_macro_context`.	required

Returns:

Type	Description
`ndarray`	A tuple `(dyn_thr, dyn_dmp, dyn_alp, noise_scalar)`, each of
`ndarray`	shape `(n_scenarios,)`.

Source code in dense_armor/core/engine.py

def calibrate_macro_context_independent(
    self, raw_batch: np.ndarray
) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
    """Same formulas as :meth:`calibrate_macro_context`, computed per row.

    The parameters are computed along ``axis=-1`` instead of on the whole
    flattened batch. The method is pure and does not mutate ``self``, so
    it is safe to use inside ``vmap`` downstream. The existing
    batch-shared :meth:`calibrate_macro_context` is kept unchanged for
    its callers (``test_boundA.py`` and the adversarial suite use the
    shared calibration on purpose: correlated scenarios share one
    context).

    Args:
        raw_batch: shape ``(n_scenarios, F)``. When ``F < 2`` no row has
            a "jump" to measure and the method returns the ``__init__``
            defaults (``self.threshold``/``damping``/``alpha``) for
            every row, instead of trying to replicate the stateful
            early-return of :meth:`calibrate_macro_context`.

    Returns:
        A tuple ``(dyn_thr, dyn_dmp, dyn_alp, noise_scalar)``, each of
        shape ``(n_scenarios,)``.
    """
    n_scenarios = int(raw_batch.shape[0])
    per_row_std = np.std(raw_batch, axis=-1)
    noise_scalar = 1.0 / (1.0 + per_row_std)
    F = raw_batch.shape[-1] if raw_batch.ndim > 1 else 0
    if F < 2:
        return (
            np.full(n_scenarios, self.threshold),
            np.full(n_scenarios, self.damping),
            np.full(n_scenarios, self.alpha),
            noise_scalar,
        )
    sample_diff = np.abs(np.diff(raw_batch, axis=-1))
    max_jump = np.max(sample_diff, axis=-1)
    mean_jump = np.mean(sample_diff, axis=-1)
    panic = (max_jump > 3.0) & ((max_jump / (mean_jump + 1e-5)) > 5.0)
    dyn_thr = np.where(panic, 1e-5, 0.10 * per_row_std)
    dyn_dmp = np.where(panic, 1e-3, _SIGMA * (1.0 + noise_scalar))
    dyn_alp = np.where(panic, 0.999, 0.75 + (_SIGMA * noise_scalar))
    return dyn_thr, dyn_dmp, dyn_alp, noise_scalar

filter_batch_scenarios_independent ¶

filter_batch_scenarios_independent(
    raw_batch: ndarray,
    hard_clamp_mask: ndarray | None = None,
) -> np.ndarray

Like :meth:filter_batch_scenarios with per-scenario calibration.

Every scenario receives its own calibration from :meth:calibrate_macro_context_independent instead of one shared on the whole batch. Numerically equivalent to calling :meth:filter_batch_scenarios once per scenario (n_scenarios=1 each), but in a single vmap call instead of N separate JAX dispatches. Added for :meth:Orca.protect_and_forward: the library has never exercised n_scenarios > 1 on independent scenarios before (the only existing multi-scenario caller, the adversarial suite in test_boundA.py, wants the shared calibration on purpose).

Parameters:

Name	Type	Description	Default
`raw_batch`	`ndarray`	the batch of raw signals, shape `(N, ...)`.	required
`hard_clamp_mask`	`ndarray \| None`	optional boolean mask, same shape as `raw_batch`; `True` where the caller (Orca) has already flagged a macro anomaly. Activates the State Flush in :meth:`_step_kernel`.	`None`

Returns:

Type	Description
`ndarray`	The filtered batch, same shape as `raw_batch`.

Source code in dense_armor/core/engine.py

def filter_batch_scenarios_independent(
    self, raw_batch: np.ndarray, hard_clamp_mask: np.ndarray | None = None
) -> np.ndarray:
    """Like :meth:`filter_batch_scenarios` with per-scenario calibration.

    Every scenario receives its own calibration from
    :meth:`calibrate_macro_context_independent` instead of one shared on
    the whole batch. Numerically equivalent to calling
    :meth:`filter_batch_scenarios` once per scenario
    (``n_scenarios=1`` each), but in a single ``vmap`` call instead of N
    separate JAX dispatches. Added for
    :meth:`Orca.protect_and_forward`: the library has never exercised
    ``n_scenarios > 1`` on independent scenarios before (the only
    existing multi-scenario caller, the adversarial suite in
    ``test_boundA.py``, wants the shared calibration on purpose).

    Args:
        raw_batch: the batch of raw signals, shape ``(N, ...)``.
        hard_clamp_mask: optional boolean mask, same shape as
            ``raw_batch``; ``True`` where the caller (Orca) has already
            flagged a macro anomaly. Activates the State Flush in
            :meth:`_step_kernel`.

    Returns:
        The filtered batch, same shape as ``raw_batch``.
    """
    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])

    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"tensor geometry not supported by the engine: {original_shape}"
        )

    flat_per_unit = structured_batch.reshape(structured_batch.shape[0], -1)
    dyn_thr, dyn_dmp, dyn_alp, noise_scalar = (
        self.calibrate_macro_context_independent(flat_per_unit)
    )

    j_batch = jnp.array(structured_batch, dtype=jnp.float64)
    j_mask = jnp.array(structured_mask, dtype=jnp.bool_)

    filtered_structured = self._compiled_batch_filter_independent(
        j_batch,
        jnp.array(dyn_thr, dtype=jnp.float64),
        jnp.array(dyn_dmp, dtype=jnp.float64),
        jnp.array(dyn_alp, dtype=jnp.float64),
        jnp.array(noise_scalar, dtype=jnp.float64),
        j_mask,
    )

    filtered_np = np.array(filtered_structured, dtype=jnp.float64)
    return filtered_np.reshape(original_shape)

filter_data_stream ¶

filter_data_stream(raw_data: ndarray) -> np.ndarray

Filter a single 1D time series.

Parameters:

Name	Type	Description	Default
`raw_data`	`ndarray`	the input series.	required

Returns:

Type	Description
`ndarray`	The filtered series, same length as `raw_data`.

Source code in dense_armor/core/engine.py

def filter_data_stream(self, raw_data: np.ndarray) -> np.ndarray:
    """Filter a single 1D time series.

    Args:
        raw_data: the input series.

    Returns:
        The filtered series, same length as ``raw_data``.
    """
    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])

    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,
    )

    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: ndarray | None = None,
) -> np.ndarray

Filter a batch of structured scenarios in parallel (vmap).

Accepts native 2D tensors or multidimensional arrays (3D/4D).

Parameters:

Name	Type	Description	Default
`raw_batch`	`ndarray`	the batch of raw signals, shape `(N, ...)`.	required
`hard_clamp_mask`	`ndarray \| None`	optional boolean mask, same shape as `raw_batch`. Where `True`, activates the State Flush inside :meth:`_step_kernel` for that element. Meant to be the same signal already computed upstream by Orca (`raw_noise` on the original raw data above the critical threshold), not an approximation computed here.	`None`

Returns:

Type	Description
`ndarray`	The filtered batch, same shape as `raw_batch`.

Source code in dense_armor/core/engine.py

def filter_batch_scenarios(
    self, raw_batch: np.ndarray, hard_clamp_mask: np.ndarray | None = None
) -> np.ndarray:
    """Filter a batch of structured scenarios in parallel (``vmap``).

    Accepts native 2D tensors or multidimensional arrays (3D/4D).

    Args:
        raw_batch: the batch of raw signals, shape ``(N, ...)``.
        hard_clamp_mask: optional boolean mask, same shape as
            ``raw_batch``. Where ``True``, activates the State Flush
            inside :meth:`_step_kernel` for that element. Meant to be
            the same signal already computed upstream by Orca
            (``raw_noise`` on the original raw data above the critical
            threshold), not an approximation computed here.

    Returns:
        The filtered batch, same shape as ``raw_batch``.
    """
    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])

    flat_analysis_batch = raw_batch.reshape(n_scenarios, -1)
    self.calibrate_macro_context(flat_analysis_batch)

    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"tensor geometry not supported by the engine: {original_shape}"
        )

    j_batch = jnp.array(structured_batch, dtype=jnp.float64)
    j_mask = jnp.array(structured_mask, dtype=jnp.bool_)

    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,
    )

    filtered_np = np.array(filtered_structured, dtype=jnp.float64)

    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

Sigmoid damping curve as a function of the local noise.

A low local_noise gives a gain close to k_min (maximum transparency); a high local_noise gives a gain close to k_max (maximum purification).

Parameters:

Name	Type	Description	Default
`local_noise`	`ndarray`	the local noise estimate.	required
`k_min`	`float`	minimum gain (maximum transparency).	`0.0`
`k_max`	`float`	maximum gain (maximum purification).	`_K_MAX_ZETA`

Returns:

Type	Description
`ndarray`	The gain value in `[k_min, k_max]`.

Source code in dense_armor/core/engine.py

@jax.jit
def dynamic_damping_gain(
    local_noise: jnp.ndarray,
    k_min: float = 0.0,
    k_max: float = _K_MAX_ZETA,
) -> jnp.ndarray:
    """Sigmoid damping curve as a function of the local noise.

    A low ``local_noise`` gives a gain close to ``k_min`` (maximum
    transparency); a high ``local_noise`` gives a gain close to ``k_max``
    (maximum purification).

    Args:
        local_noise: the local noise estimate.
        k_min: minimum gain (maximum transparency).
        k_max: maximum gain (maximum purification).

    Returns:
        The gain value in ``[k_min, k_max]``.
    """
    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)

Details¶

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.