Skip to content

Preprocessing

Raw sensor numbers come in different sizes and units. These tools put them on a common scale and turn them into the numbers that matter (such as speed or power), one reading at a time, so that a model can learn from them.

Raw sensor numbers come in different units and sizes — a joint position in radians, a torque in newton-metres, a word in a sentence. Before a model can learn from them they are put on a common scale, turned into the quantities that matter (speed, power, word weights), filtered, and balanced. Everything here works one sample at a time: the estimator sees a sample, updates its own state, and is ready for the next one, without keeping the whole history in memory.

Nine steps follow. The first four are about numbers, the next three about text, the last two about which features to keep and how to handle rare events.

1. Same scale for every feature

A standard scaler subtracts the running mean and divides by the running standard deviation, so every feature lives around 0 with spread 1.

from dense_armor.utility.preprocessing.scale import StandardScaler

sc = StandardScaler()
for v in [1.0, 2.0, 3.0, 4.0, 5.0]:
    sc.learn_one({"x": v})
print(round(sc.transform_one({"x": 5.0})["x"], 4))

1.2649

The mean of 1, 2, 3, 4, 5 is 3, the standard deviation is $\sqrt{2.5} = 1.581$, so $(5 - 3) / 1.581 = 1.2649$. In general the scaler computes

$$z = \frac{x - \bar x}{s}$$

where $\bar x$ and $s$ are the running mean and standard deviation seen so far. Both are updated one sample at a time, so the scaler never needs to store the past values.

2. One statistic per joint

A robot arm sends its joints as one vector; each joint gets its own mean and spread.

from dense_armor.utility.preprocessing.scale import StandardScaler

sc = StandardScaler()
for q in ([0.0, 10.0], [1.0, 20.0], [2.0, 30.0]):
    sc.learn_one({"q": q})
print([round(v, 3) for v in sc.transform_one({"q": [2.0, 30.0]})["q"]])

[1.0, 1.0]

Both joints are at the top of their own range, so both get the same standardised value, even though the second joint moves ten times more. The scaler keeps one mean and one standard deviation per joint, not one for the whole vector.

3. Robust to spikes

The robust scaler uses the rolling median and the median absolute deviation (MAD, times 1.4826 so it matches the standard deviation on Gaussian data); one spike barely moves it.

from dense_armor.utility.preprocessing.scale import RobustScaler, StandardScaler

rs, ss = RobustScaler(window=20), StandardScaler()
for v in [1.0, 2.0, 3.0, 2.0, 1.0, 2.0, 3.0, 2.0, 100.0]:
    rs.learn_one({"x": v})
    ss.learn_one({"x": v})
print(round(rs.transform_one({"x": 3.0})["x"], 3), round(ss.transform_one({"x": 3.0})["x"], 3))

0.674 -0.303

The same value 3.0 after a spike of 100: the robust scaler sees it as 0.674 spreads above the median (median 2, MAD-based spread 1.483); the standard scaler, pulled by the spike, sees it as below average. The median is the middle value of the sorted window, so a single extreme sample does not move it; the MAD is the median of the distances from the median, so it measures the typical spread without ever squaring the spike. The factor 1.4826 turns the MAD into the same scale as a standard deviation when the data is roughly Gaussian.

4. Speed, acceleration and jerk from positions

The robot reports positions; velocity, acceleration and jerk come from differences between samples divided by the real time between them, even when the samples are not evenly spaced.

from dense_armor.utility.preprocessing.joints import JointDerivatives

jd = JointDerivatives(order=3)
for t in [0.0, 0.3, 1.1]:
    jd.learn_one({"q": [t ** 3]}, t=t)
out = jd.transform_one({"q": [2.0 ** 3]}, t=2.0)
print([round(out[k][0], 6) for k in ("qd", "qdd", "jerk")])

[12.0, 12.0, 6.0]

Positions follow $q = t^3$ at uneven times 0, 0.3, 1.1, 2.0; at $t = 2$ the exact values are $3t^2 = 12$, $6t = 12$, $6$: the method is exact on a cubic. It uses the Newton divided differences of the last four samples; the current sample is included, so the value belongs to the present moment, not to the previous sample. Divided differences are the natural generalisation of "difference divided by time" to several unevenly spaced points: they are built recursively so that the resulting polynomial passes through all the samples, and differentiating it at the last point gives the derivative there.

5. Power

from dense_armor.utility.preprocessing.joints import JointPower

out = JointPower().transform_one({"tau": [1.0, 2.0], "qd": [3.0, 4.0]})
print(out["power"], out["power_total"])

[3.0, 8.0] 11.0

Mechanical power of a joint is torque times angular velocity,

$$P = \tau\,\dot q$$

so the first joint contributes $1 \times 3 = 3$, the second $2 \times 4 = 8$, and the total is $3 + 8 = 11$.

6. Text: words, counts and weights

A sentence becomes a list of words, then counts; TF-IDF weighs each word by how often it appears in this document and how rare it is across all documents seen.

from dense_armor.utility.preprocessing.text import TFIDF, BagOfWords, Tokenizer

tk, bw, tf = Tokenizer(), BagOfWords(), TFIDF()
docs = ["joint three overheats", "joint three ok", "gripper ok"]
for d in docs:
    tf.learn_one(bw.transform_one(tk.transform_one(d)))
out = tf.transform_one(bw.transform_one(tk.transform_one("joint three overheats")))
print({k: round(v, 3) for k, v in out["scores"].items()})

{'joint': 0.135, 'three': 0.135, 'overheats': 0.366}

The formula is the one of Silajev (2026), section 1:

$$\text{TF-IDF}(t, i) = \frac{d_i(t)}{|d_i|}\,\ln\frac{N}{\text{DF}(t)}$$

$d_i(t)$ is how many times the word appears in the document, $|d_i|$ the document length, $N$ the number of documents seen (3), $\text{DF}(t)$ how many contain the word: "overheats" appears in one document, so it weighs most; "joint" and "three" appear in two.

The word "overheats" is rare and informative; "ok" appears in two documents out of three, so its weight is small. The ratio $d_i(t)/|d_i|$ is the frequency of the word inside the document, and the logarithm of $N/\text{DF}(t)$ rewards words that few documents share. Silajev (2026) shows that the same formula is exactly a Kullback-Leibler divergence between two probability models.

TFIDF() follows the paper exactly. The option smooth_idf=True replaces the logarithm with $\ln(1 + N/\text{DF})$, a common practical variant that keeps every word with at least a small positive score; it is off by default because the paper's formula does not include it.

7. Text in a fixed size: hashing

from dense_armor.utility.preprocessing.text import FeatureHasher

fh = FeatureHasher(n_features=8, seed=0)
print(fh.transform_one({"counts": {"joint": 1.0, "three": 1.0, "overheats": 1.0}})["hashed"])

{7: -1.0, 2: 1.0, 5: 1.0}

Every word is sent by a hash function to one of 8 slots, with a random sign; the vocabulary can grow forever, the vector stays the same size. Weinberger et al. (2009), Theorem 3 (eq. 4): for a unit vector $x$ with $|x|_\infty \le \epsilon/(18\sqrt{\log(1/\delta)\log(m/\delta)})$ and $m \ge 72\log(1/\delta)/\epsilon^2$, the hashed squared length is within $1 \pm \epsilon$ with probability at least $1 - 2\delta$.

The map used here is the signed hash of the paper: each word gets an index and a random sign, and words that share a slot add their weights with signs that cancel on average. Without the sign, two words in the same slot would always add up, biasing the sum upward. The bound above says how many slots are enough: the error decreases with more slots, the confidence increases with fewer collisions, and the largest single coordinate of the input must not dominate the vector. In practice eight slots already separate the three words of the example, each landing in its own bucket.

8. Keep only the useful features

from dense_armor.utility.preprocessing.select import SelectKBest, VarianceThreshold

vt, sk = VarianceThreshold(threshold=0.1), SelectKBest(k=1)
for i in range(20):
    x = {"a": float(i), "b": float(i % 3), "c": 7.0}
    vt.learn_one(x)
    sk.learn_one(x, y=2.0 * i)
print(sorted(vt.transform_one(x)), sorted(sk.transform_one(x)))

['a', 'b'] ['a']

The variance threshold drops c, which never changes; k-best keeps the feature most correlated with the target, a. The two tests look at different things: variance is about a feature alone, correlation is about a feature and the label together. A constant feature cannot help any model, and a feature uncorrelated with the target is just noise.

9. Rare events: queue resampling

When only 1 % of the samples are the event of interest (a fault), a model sees almost only normal samples and learns to ignore the rare ones. Queue resampling keeps the last few examples of each class and trains on both every time (Malialis et al. 2018, Algorithm 1).

import random
from dense_armor.utility.evaluate import progressive_val_score
from dense_armor.utility.learn.online_classifiers import OnlineGaussianNB
from dense_armor.utility.metrics import Recall
from dense_armor.utility.preprocessing.imbalance import QueueResampler

rng = random.Random(0)
s = []
for _ in range(5000):
    y = 1 if rng.random() < 0.01 else 0
    m = 2.0 if y else -2.0
    s.append(({"x0": rng.gauss(m, 1.0), "x1": rng.gauss(m, 1.0)}, y))
a = progressive_val_score(s, OnlineGaussianNB(), Recall(positive=1))
b = progressive_val_score(s, QueueResampler(OnlineGaussianNB(), queue_size=25), Recall(positive=1))
print(round(a.get(), 4), round(b.get(), 4))

0.8955 0.9403

Recall of the rare class (how many of the real events are caught) without and with queue resampling, on 5000 samples with 1 % events. The resampler keeps two queues of the same length, one for the positive class and one for the negative, and at each step it feeds the classifier the union of the two. The queue length bounds how many old examples are kept, so the training set stays balanced even when the stream is not. The union also acts as a sliding window: very old examples leave the queue, which lets the classifier follow a change in the stream.

API reference

scale

Online per-feature scalers, one sample at a time.

Each scaler is a :class:~dense_armor.roles.Transformer: learn_one updates the running state, transform_one uses the current state. Scalars and vectors are both supported: a scalar keeps one statistic, a vector keeps one statistic per joint. A feature never seen, or with zero spread, passes through unchanged.

The running mean and variance are updated incrementally with the stable one-pass formulas of running moments. The exponentially weighted mean and variance follow the recursive form of exponentially weighted statistics, where recent samples carry more weight. The rolling median and median absolute deviation come from a bounded FIFO window. None of these carries a reference entry: the methods are standard and the code documents them.

StandardScaler 

StandardScaler(
    features: list[str] | None = None, eps: float = 1e-12
)

Bases: Transformer

Per-feature standardization with running mean and std.

z_i = (x_i - mean_i) / std_i for each feature and, when the value is a vector, for each joint. A feature (or a joint) seen fewer than twice, or whose std is below eps, passes through unchanged.

Memory grows with the number of features (and per-feature joints): each feature keeps two floats, so the total is O(d) for d scalar features.

Parameters:

Name Type Description Default
features list[str] | None

explicit feature order. None sorts the first dict's keys.

 None
eps float

std floor below which the feature is left alone.

 1e-12

Examples:

>>> from dense_armor.utility.preprocessing.scale import StandardScaler
>>> sc = StandardScaler()
>>> for v in [1.0, 2.0, 3.0, 4.0, 5.0]:
...     _ = sc.learn_one({"x": v})
>>> round(sc.transform_one({"x": 5.0})["x"], 4)
1.2649
>>> sc2 = StandardScaler()
>>> for q in ([0.0, 1.0], [1.0, 2.0], [2.0, 3.0]):
...     _ = sc2.learn_one({"q": q})
>>> [round(v, 4) for v in sc2.transform_one({"q": [2.0, 3.0]})["q"]]
[1.0, 1.0]
Source code in dense_armor/utility/preprocessing/scale.py 
92
93
94
95
96
97
def __init__(
    self, features: list[str] | None = None, eps: float = 1e-12
) -> None:
    self.features = features
    self.eps = eps
    self.stats_: dict[str, list[RunningMoments]] | None = None

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> StandardScaler

Update the running mean and std for each feature in x.

Source code in dense_armor/utility/preprocessing/scale.py 
130
131
132
133
134
135
def learn_one(
    self, x: Any, y: Any = None, t: float | None = None
) -> "StandardScaler":
    """Update the running mean and std for each feature in ``x``."""
    self._update(_as_dict(x), t)
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> dict

Standardize each feature using the current state.

Source code in dense_armor/utility/preprocessing/scale.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def transform_one(self, x: Any, t: float | None = None) -> dict:
    """Standardize each feature using the current state."""
    d = _as_dict(x)
    if self.stats_ is None:
        return d
    out = dict(d)
    for k, s_list in self.stats_.items():
        if k not in out:
            continue
        vec, is_scalar = _to_vec(out[k])
        if vec is None or vec.size != len(s_list):
            continue
        res = np.empty(vec.size, dtype=float)
        for i, s in enumerate(s_list):
            if s.count >= 2 and s.std > self.eps:
                res[i] = (vec[i] - s.mean) / s.std
            else:
                res[i] = vec[i]
        out[k] = _write_out(res, is_scalar)
    return out

EWScaler 

EWScaler(
    alpha: float = 0.1,
    features: list[str] | None = None,
    eps: float = 1e-12,
)

Bases: Transformer

Per-feature standardization with exponentially weighted stats.

Same shape as :class:StandardScaler but the mean and variance follow the recursive exponential weighting, so recent samples weigh more and a slow drift is tracked instead of averaged out.

Memory grows with the number of features (O(d) for d scalar features); each feature keeps one mean and one variance.

Parameters:

Name Type Description Default
alpha float

weight of the new sample in (0, 1].

 0.1
features list[str] | None

explicit feature order.

 None
eps float

std floor.

 1e-12

Examples:

>>> from dense_armor.utility.preprocessing.scale import EWScaler
>>> sc = EWScaler(alpha=0.5)
>>> for v in [1.0, 2.0, 3.0]:
...     _ = sc.learn_one({"x": v})
>>> got = sc.transform_one({"x": 3.0})["x"]
>>> round(got, 6)
0.904534
Source code in dense_armor/utility/preprocessing/scale.py 
187
188
189
190
191
192
193
194
195
196
def __init__(
    self,
    alpha: float = 0.1,
    features: list[str] | None = None,
    eps: float = 1e-12,
) -> None:
    self.alpha = alpha
    self.features = features
    self.eps = eps
    self.stats_: dict[str, list[EWStats]] | None = None

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> EWScaler

Update the exponentially weighted stats per feature.

Source code in dense_armor/utility/preprocessing/scale.py 
229
230
231
232
233
234
def learn_one(
    self, x: Any, y: Any = None, t: float | None = None
) -> "EWScaler":
    """Update the exponentially weighted stats per feature."""
    self._update(_as_dict(x), t)
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> dict

Standardize with the current exponentially weighted state.

Source code in dense_armor/utility/preprocessing/scale.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def transform_one(self, x: Any, t: float | None = None) -> dict:
    """Standardize with the current exponentially weighted state."""
    d = _as_dict(x)
    if self.stats_ is None:
        return d
    out = dict(d)
    for k, s_list in self.stats_.items():
        if k not in out:
            continue
        vec, is_scalar = _to_vec(out[k])
        if vec is None or vec.size != len(s_list):
            continue
        res = np.empty(vec.size, dtype=float)
        for i, s in enumerate(s_list):
            if s.count >= 1 and s.std > self.eps:
                res[i] = (vec[i] - s.mean) / s.std
            else:
                res[i] = vec[i]
        out[k] = _write_out(res, is_scalar)
    return out

RobustScaler 

RobustScaler(
    window: int = 20,
    features: list[str] | None = None,
    eps: float = 1e-12,
)

Bases: Transformer

Robust per-feature standardization with rolling median and MAD.

z_i = (x_i - median_i) / MAD_i for each feature and, when the value is a vector, for each joint. A feature with MAD below eps passes through unchanged.

Memory grows with the number of features and the window length: each feature keeps window samples, so the total is O(d*w).

Parameters:

Name Type Description Default
window int

rolling window length in samples.

 20
features list[str] | None

explicit feature order.

 None
eps float

MAD floor.

 1e-12

Examples:

>>> from dense_armor.utility.preprocessing.scale import RobustScaler
>>> sc = RobustScaler(window=5)
>>> for v in [1.0, 2.0, 3.0, 4.0, 5.0]:
...     _ = sc.learn_one({"x": v})
>>> round(sc.transform_one({"x": 5.0})["x"], 4)
1.349
Source code in dense_armor/utility/preprocessing/scale.py 
285
286
287
288
289
290
291
292
293
294
295
def __init__(
    self,
    window: int = 20,
    features: list[str] | None = None,
    eps: float = 1e-12,
) -> None:
    self.window = window
    self.features = features
    self.eps = eps
    self.medians_: dict[str, list[RollingMedian]] | None = None
    self.mads_: dict[str, list[RollingMAD]] | None = None

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> RobustScaler

Push one sample into both rolling windows per feature.

Source code in dense_armor/utility/preprocessing/scale.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
def learn_one(
    self, x: Any, y: Any = None, t: float | None = None
) -> "RobustScaler":
    """Push one sample into both rolling windows per feature."""
    self._time_step(t)
    d = _as_dict(x)
    med, mad = self._ensure(d)
    for k, med_list in med.items():
        if k not in d:
            continue
        vec, _ = _to_vec(d[k])
        if vec is None or vec.size != len(med_list):
            continue
        mad_list = mad[k]
        for i in range(vec.size):
            med_list[i].learn_one({"_v": float(vec[i])}, t=t)
            mad_list[i].learn_one({"_v": float(vec[i])}, t=t)
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> dict

Center by the median and scale by the MAD.

Source code in dense_armor/utility/preprocessing/scale.py 
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
def transform_one(self, x: Any, t: float | None = None) -> dict:
    """Center by the median and scale by the MAD."""
    d = _as_dict(x)
    if self.medians_ is None or self.mads_ is None:
        return d
    out = dict(d)
    for k, med_list in self.medians_.items():
        if k not in out:
            continue
        vec, is_scalar = _to_vec(out[k])
        if vec is None or vec.size != len(med_list):
            continue
        mad_list = self.mads_[k]
        res = np.empty(vec.size, dtype=float)
        for i, med in enumerate(med_list):
            mad = mad_list[i].value
            if med.count >= 1 and mad > self.eps:
                res[i] = (vec[i] - med.value) / mad
            else:
                res[i] = vec[i]
        out[k] = _write_out(res, is_scalar)
    return out

MinMaxScaler 

MinMaxScaler(
    features: list[str] | None = None, eps: float = 1e-12
)

Bases: Transformer

Rescale each feature to [0, 1] with running min and max.

z_i = (x_i - min_i) / (max_i - min_i) for each feature and, when the value is a vector, for each joint. A feature with max == min passes through unchanged.

Memory grows with the number of features: each keeps two floats.

Parameters:

Name Type Description Default
features list[str] | None

explicit feature order.

 None
eps float

spread floor below which the feature is left alone.

 1e-12

Examples:

>>> from dense_armor.utility.preprocessing.scale import MinMaxScaler
>>> sc = MinMaxScaler()
>>> for v in [1.0, 2.0, 3.0, 4.0, 5.0]:
...     _ = sc.learn_one({"x": v})
>>> sc.transform_one({"x": 5.0})["x"]
1.0
Source code in dense_armor/utility/preprocessing/scale.py 
394
395
396
397
398
399
def __init__(
    self, features: list[str] | None = None, eps: float = 1e-12
) -> None:
    self.features = features
    self.eps = eps
    self.stats_: dict[str, list[RunningMoments]] | None = None

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> MinMaxScaler

Update the running min and max for each feature.

Source code in dense_armor/utility/preprocessing/scale.py 
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
def learn_one(
    self, x: Any, y: Any = None, t: float | None = None
) -> "MinMaxScaler":
    """Update the running min and max for each feature."""
    self._time_step(t)
    d = _as_dict(x)
    stats = self._ensure(d)
    for k, s_list in stats.items():
        if k not in d:
            continue
        vec, _ = _to_vec(d[k])
        if vec is None or vec.size != len(s_list):
            continue
        for i, s in enumerate(s_list):
            s.learn_one({"_v": float(vec[i])})
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> dict

Shift and rescale each feature to [0, 1].

Source code in dense_armor/utility/preprocessing/scale.py 
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
def transform_one(self, x: Any, t: float | None = None) -> dict:
    """Shift and rescale each feature to ``[0, 1]``."""
    d = _as_dict(x)
    if self.stats_ is None:
        return d
    out = dict(d)
    for k, s_list in self.stats_.items():
        if k not in out:
            continue
        vec, is_scalar = _to_vec(out[k])
        if vec is None or vec.size != len(s_list):
            continue
        res = np.empty(vec.size, dtype=float)
        for i, s in enumerate(s_list):
            lo, hi = s.min, s.max
            if lo is None or hi is None:
                res[i] = vec[i]
                continue
            spread = hi - lo
            if spread > self.eps:
                res[i] = (vec[i] - lo) / spread
            else:
                res[i] = vec[i]
        out[k] = _write_out(res, is_scalar)
    return out

joints

Joint kinematics by finite differences on real timestamps.

JointDerivatives turns joint positions q into velocity, acceleration and jerk. The derivatives are computed on the real timestamps t, so non-uniform sampling is handled exactly. Missing samples (NaN in any element of q or in t) are skipped and counted; the next derivative uses the last valid sample with its own t.

The formulas come from Newton divided differences over the last order + 1 valid points, which are exact on any polynomial of degree at most order. transform_one(x, t) uses the current sample x at time t together with the buffer, so in a pipeline (transform first, then learn) the derivative returned at time t refers to the sample at time t, not to the last stored one. learn_one stores the sample in the buffer.

JointPower gives the mechanical power tau * qd per joint and its total.

JointDerivatives 

JointDerivatives(order: int = 3, q_key: str = 'q')

Bases: Transformer

Velocity, acceleration and jerk from joint positions.

Input: a dict with the joint positions under q_key (a vector) and the timestamp t passed to learn_one and to transform_one. Output: a dict with qd, qdd and jerk keys, each a vector of the same length as q. order selects how many derivatives are produced (1, 2 or 3).

In a pipeline, call transform_one(x, t) first and learn_one(x, t) after: the derivative returned at time t refers to the sample at time t.

Memory grows with the number of joints and the derivative order: the buffer holds order + 1 past samples, one per joint, so the total is O(d * order).

Parameters:

Name Type Description Default
order int

number of derivatives to emit.

 3
q_key str

input key holding the joint positions.

 'q'

Examples:

>>> from dense_armor.utility.preprocessing.joints import JointDerivatives
>>> jd = JointDerivatives(order=3)
>>> for t in [0.0, 1.0, 2.0, 3.0]:
...     _ = jd.learn_one({"q": [t ** 3]}, t=t)
>>> round(jd.transform_one({})["qd"][0], 6)
27.0
>>> round(jd.transform_one({})["qdd"][0], 6)
18.0
Source code in dense_armor/utility/preprocessing/joints.py 
103
104
105
106
107
def __init__(self, order: int = 3, q_key: str = "q") -> None:
    self.order = order
    self.q_key = q_key
    self.buf_: deque[tuple[float, np.ndarray]] = deque(maxlen=order + 1)
    self.n_missing_ = 0

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> JointDerivatives

Append one valid pair of timestamp and positions to the buffer.

Source code in dense_armor/utility/preprocessing/joints.py 
125
126
127
128
129
130
131
132
133
134
135
136
def learn_one(
    self, x: Any, y: Any = None, t: float | None = None
) -> "JointDerivatives":
    """Append one valid pair of timestamp and positions to the buffer."""
    self._time_step(t)
    res = self._extract(_as_dict(x), t)
    if res is None:
        self.n_missing_ += 1
        return self
    tf, qv = res
    self.buf_.append((tf, qv.copy()))
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> dict

Return the derivative vectors at time t.

Uses the buffer together with the current sample x at time t. If t is None, or x lacks the positions, the last stored derivatives are returned.

Source code in dense_armor/utility/preprocessing/joints.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def transform_one(self, x: Any, t: float | None = None) -> dict:
    """Return the derivative vectors at time ``t``.

    Uses the buffer together with the current sample ``x`` at time
    ``t``. If ``t`` is None, or ``x`` lacks the positions, the
    last stored derivatives are returned.
    """
    out = _as_dict(x)
    res = self._extract(_as_dict(x), t)
    if res is None:
        hist = list(self.buf_)
    else:
        tf, qv = res
        hist = list(self.buf_) + [(tf, qv.copy())]
    if len(hist) < 2:
        return out
    hist = hist[-(self.order + 1):]
    ts = [p[0] for p in hist]
    qs = np.stack([p[1] for p in hist], axis=0)
    _, d = qs.shape
    v = np.zeros(d)
    a = np.zeros(d)
    j = np.zeros(d)
    for k in range(d):
        vk, ak, jk = _derivatives_at_last(ts, qs[:, k].tolist())
        v[k], a[k], j[k] = vk, ak, jk
    if self.order >= 1:
        out["qd"] = v.tolist()
    if self.order >= 2:
        out["qdd"] = a.tolist()
    if self.order >= 3:
        out["jerk"] = j.tolist()
    return out
    hist = hist[-(self.order + 1):]
    ts = [p[0] for p in hist]
    qs = np.stack([p[1] for p in hist], axis=0)
    _, d = qs.shape
    v = np.zeros(d)
    a = np.zeros(d)
    j = np.zeros(d)
    for k in range(d):
        vk, ak, jk = _derivatives_at_last(ts, qs[:, k].tolist())
        v[k], a[k], j[k] = vk, ak, jk
    if self.order >= 1:
        out["qd"] = v.tolist()
    if self.order >= 2:
        out["qdd"] = a.tolist()
    if self.order >= 3:
        out["jerk"] = j.tolist()
    return out

JointPower 

JointPower(tau_key: str = 'tau', qd_key: str = 'qd')

Bases: Transformer

Mechanical power tau * qd per joint and its total.

Memory grows with the number of joints only, since no state is kept.

Parameters:

Name Type Description Default
tau_key str

input key holding the joint torques.

 'tau'
qd_key str

input key holding the joint velocities.

 'qd'

Examples:

>>> from dense_armor.utility.preprocessing.joints import JointPower
>>> jp = JointPower()
>>> out = jp.transform_one({"tau": [1.0, 2.0], "qd": [3.0, 4.0]})
>>> out["power"], out["power_total"]
([3.0, 8.0], 11.0)
Source code in dense_armor/utility/preprocessing/joints.py 
211
212
213
214
def __init__(self, tau_key: str = "tau", qd_key: str = "qd") -> None:
    self.tau_key = tau_key
    self.qd_key = qd_key
    self.n_missing_ = 0

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> JointPower

No state to update; kept for the Transformer interface.

Source code in dense_armor/utility/preprocessing/joints.py 
216
217
218
219
220
221
def learn_one(
    self, x: Any, y: Any = None, t: float | None = None
) -> "JointPower":
    """No state to update; kept for the ``Transformer`` interface."""
    self._time_step(t)
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> dict

Return the per-joint power and the total power.

Source code in dense_armor/utility/preprocessing/joints.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def transform_one(self, x: Any, t: float | None = None) -> dict:
    """Return the per-joint power and the total power."""
    d = _as_dict(x)
    out = dict(d)
    if self.tau_key not in d or self.qd_key not in d:
        self.n_missing_ += 1
        return out
    tau = np.asarray(d[self.tau_key], dtype=float)
    qd = np.asarray(d[self.qd_key], dtype=float)
    if tau.shape != qd.shape:
        self.n_missing_ += 1
        return out
    p = tau * qd
    out["power"] = p.tolist()
    out["power_total"] = float(p.sum())
    return out

text

Online text features: tokenization, bag of words, TF-IDF, hashing.

All four are :class:~dense_armor.roles.Transformer subclasses: they read a string (or a token list) from one key and write their result under another. Input is a plain str or a dict holding it.

Tokenizer 

Tokenizer(
    lowercase: bool = True,
    ngrams: int = 1,
    text_key: str = "text",
    tokens_key: str = "tokens",
)

Bases: Transformer

Lower-case word tokens, with optional n-grams.

Memory is constant; no state is kept between calls.

Parameters:

Name Type Description Default
lowercase bool

convert the text to lower case.

 True
ngrams int

maximum n-gram order (1 = unigrams only).

 1
text_key str

dict key holding the text string.

 'text'
tokens_key str

dict key written with the token list.

 'tokens'

Examples:

>>> from dense_armor.utility.preprocessing.text import Tokenizer
>>> tk = Tokenizer(ngrams=2)
>>> tk.transform_one({"text": "Hello world"})["tokens"]
['hello', 'world', 'hello world']
Source code in dense_armor/utility/preprocessing/text.py 
47
48
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    lowercase: bool = True,
    ngrams: int = 1,
    text_key: str = "text",
    tokens_key: str = "tokens",
) -> None:
    self.lowercase = lowercase
    self.ngrams = ngrams
    self.text_key = text_key
    self.tokens_key = tokens_key

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> Tokenizer

No state to update; kept for the interface.

Source code in dense_armor/utility/preprocessing/text.py 
59
60
61
62
63
64
def learn_one(
    self, x: Any, y: Any = None, t: float | None = None
) -> "Tokenizer":
    """No state to update; kept for the interface."""
    self._time_step(t)
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> dict

Split the text into words (and n-grams) and lower-case them.

Source code in dense_armor/utility/preprocessing/text.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def transform_one(self, x: Any, t: float | None = None) -> dict:
    """Split the text into words (and n-grams) and lower-case them."""
    d = _as_dict(x)
    text = d.get(self.text_key, "")
    if not isinstance(text, str):
        text = str(text)
    if self.lowercase:
        text = text.lower()
    words = _TOKEN_RE.findall(text)
    tokens = list(words)
    for n in range(2, self.ngrams + 1):
        for i in range(len(words) - n + 1):
            tokens.append(" ".join(words[i : i + n]))
    out = dict(d)
    out[self.tokens_key] = tokens
    return out

BagOfWords 

BagOfWords(
    tokens_key: str = "tokens", counts_key: str = "counts"
)

Bases: Transformer

Term counts of one document, from a token list.

Memory grows with the number of distinct tokens in the document only; the counts dict is rebuilt on every call.

Parameters:

Name Type Description Default
tokens_key str

dict key holding the token list.

 'tokens'
counts_key str

dict key written with the {term: count} dict.

 'counts'

Examples:

>>> from dense_armor.utility.preprocessing.text import BagOfWords
>>> bw = BagOfWords()
>>> bw.transform_one({"tokens": ["a", "b", "a"]})["counts"]
{'a': 2, 'b': 1}
Source code in dense_armor/utility/preprocessing/text.py 
104
105
106
107
108
def __init__(
    self, tokens_key: str = "tokens", counts_key: str = "counts"
) -> None:
    self.tokens_key = tokens_key
    self.counts_key = counts_key

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> BagOfWords

No state to update; kept for the interface.

Source code in dense_armor/utility/preprocessing/text.py 
110
111
112
113
114
115
def learn_one(
    self, x: Any, y: Any = None, t: float | None = None
) -> "BagOfWords":
    """No state to update; kept for the interface."""
    self._time_step(t)
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> dict

Return the term counts of the document.

Source code in dense_armor/utility/preprocessing/text.py 
117
118
119
120
121
122
123
def transform_one(self, x: Any, t: float | None = None) -> dict:
    """Return the term counts of the document."""
    d = _as_dict(x)
    tokens = d.get(self.tokens_key, [])
    out = dict(d)
    out[self.counts_key] = dict(Counter(tokens))
    return out

TFIDF 

TFIDF(
    smooth_idf: bool = False,
    tokens_key: str = "tokens",
    counts_key: str = "counts",
    scores_key: str = "scores",
)

Bases: Transformer

Online TF-IDF following Silajev 2026, section 1.

The paper defines, for a term t in document i with raw counts d_i and length |d_i|:

.. math::

\mathrm{TF\!\cdot\!IDF}(t, i)
= \frac{d_i(t)}{|d_i|} \ln\!\left(\frac{N}{\mathrm{DF}(t)}\right)

where N is the number of documents seen so far and \mathrm{DF}(t) is the number of documents containing t. The term frequency is the relative frequency in the document (count divided by document length), and the idf is ln(N / DF) with no +1 smoothing. A term present in every document has DF(t) = N, so its idf is ln(1) = 0 and the term vanishes from the score, exactly as the paper states.

smooth_idf=True replaces the idf with ln(1 + N / DF); the smoothed variant is not the paper's formula and is provided only to avoid zero scores for terms that appear in every document.

Memory grows with the vocabulary: one counter per distinct term seen, so the total is O(V) for V distinct terms.

Parameters:

Name Type Description Default
smooth_idf bool

if True, use ln(1 + N / DF). Default False, which is the paper's formula.

 False
tokens_key str

dict key holding the token list.

 'tokens'
counts_key str

optional dict key holding a precomputed count dict; when present it is used instead of recounting.

 'counts'
scores_key str

dict key written with {term: score}.

 'scores'

Examples:

>>> from dense_armor.utility.preprocessing.text import TFIDF
>>> tf = TFIDF()
>>> _ = tf.learn_one({"tokens": ["a", "b"]})
>>> _ = tf.learn_one({"tokens": ["a"]})
>>> out = tf.transform_one({"tokens": ["a", "b"]})
>>> out["scores"]["a"]
0.0
>>> round(out["scores"]["b"], 6)
0.346574
References

Silajev, I. (2026). TF-IDF and BM25 as Kullback-Leibler divergences. arXiv:2609.14016, section 1.

Source code in dense_armor/utility/preprocessing/text.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
def __init__(
    self,
    smooth_idf: bool = False,
    tokens_key: str = "tokens",
    counts_key: str = "counts",
    scores_key: str = "scores",
) -> None:
    self.smooth_idf = smooth_idf
    self.tokens_key = tokens_key
    self.counts_key = counts_key
    self.scores_key = scores_key
    self.n_docs_: int = 0
    self.df_: dict[str, int] = {}

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> TFIDF

Count one document towards N and DF.

Source code in dense_armor/utility/preprocessing/text.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def learn_one(
    self, x: Any, y: Any = None, t: float | None = None
) -> "TFIDF":
    """Count one document towards ``N`` and ``DF``."""
    self._time_step(t)
    d = _as_dict(x)
    if self.counts_key in d:
        terms = set(d[self.counts_key].keys())
    else:
        terms = set(d.get(self.tokens_key, []))
    self.n_docs_ += 1
    for term in terms:
        self.df_[term] = self.df_.get(term, 0) + 1
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> dict

Return the TF-IDF scores of the current document.

Source code in dense_armor/utility/preprocessing/text.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def transform_one(self, x: Any, t: float | None = None) -> dict:
    """Return the TF-IDF scores of the current document."""
    d = _as_dict(x)
    if self.counts_key in d:
        counts = dict(d[self.counts_key])
    else:
        counts = dict(Counter(d.get(self.tokens_key, [])))
    doc_len = sum(counts.values())
    n = max(self.n_docs_, 1)
    scores: dict[str, float] = {}
    for term, c in counts.items():
        tf = (c / doc_len) if doc_len > 0 else 0.0
        df = self.df_.get(term, 1)
        if self.smooth_idf:
            idf = math.log(1.0 + n / df)
        else:
            idf = math.log(n / df)
        scores[term] = float(tf) * idf
    out = dict(d)
    out[self.scores_key] = scores
    return out

FeatureHasher 

FeatureHasher(
    n_features: int = 1024,
    seed: int = 0,
    counts_key: str = "counts",
    hashed_key: str = "hashed",
)

Bases: Transformer

Signed feature hashing (Weinberger et al. 2009, Theorem 3).

For each term j the hasher draws an index h(j) in {1, ..., m} and a sign xi(j) in {-1, +1} from one SHA-256 digest. The hashed feature vector is

.. math::

\phi_i(x) = \sum_{j : h(j) = i} \xi(j) x_j.

The complete statement of Theorem 3 (equation 4 of the paper) is the following. Let epsilon < 1 be a fixed constant and let x be an instance with ||x||_2 = 1. If

.. math::

m \geq \frac{72 \log(1/\delta)}{\epsilon^2}
\quad\text{and}\quad
\|x\|_\infty \leq
\frac{\epsilon}{18\sqrt{\log(1/\delta)\log(m/\delta)}},

then

.. math::

\Pr\!\left[\,
\bigl|\,\|x\|_\phi^2 - 1\,\bigr| \geq \epsilon
\,\right] \leq 2\delta.

The bound is what sizes n_features: for a target distortion epsilon and a failure probability 2\delta, m must grow like log(1/\delta) / \epsilon^2 and the largest coordinate of x must stay below the stated threshold.

Memory is fixed by the hash space: one accumulator slot per bucket in the current document, so the total is O(m).

Parameters:

Name Type Description Default
n_features int

hash space size m.

 1024
seed int

hash seed; different seeds give independent hashes.

 0
counts_key str

dict key holding the term counts or scores.

 'counts'
hashed_key str

dict key written with the hashed vector as a dict.

 'hashed'

Examples:

>>> from dense_armor.utility.preprocessing.text import FeatureHasher
>>> fh = FeatureHasher(n_features=8, seed=0)
>>> out = fh.transform_one({"counts": {"a": 1.0}})
>>> len(out["hashed"])
1
References

Weinberger, K., Dasgupta, A., Attenberg, J., Langford, J., Smola, A. (2009). Feature hashing for large scale multitask learning. ICML, Theorem 3 and equation 4.

Source code in dense_armor/utility/preprocessing/text.py 
301
302
303
304
305
306
307
308
309
310
311
def __init__(
    self,
    n_features: int = 1024,
    seed: int = 0,
    counts_key: str = "counts",
    hashed_key: str = "hashed",
) -> None:
    self.n_features = n_features
    self.seed = seed
    self.counts_key = counts_key
    self.hashed_key = hashed_key

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> FeatureHasher

No state to update; kept for the interface.

Source code in dense_armor/utility/preprocessing/text.py 
313
314
315
316
317
318
def learn_one(
    self, x: Any, y: Any = None, t: float | None = None
) -> "FeatureHasher":
    """No state to update; kept for the interface."""
    self._time_step(t)
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> dict

Hash the term weights into a lower-dimensional sparse vector.

Source code in dense_armor/utility/preprocessing/text.py 
320
321
322
323
324
325
326
327
328
329
330
def transform_one(self, x: Any, t: float | None = None) -> dict:
    """Hash the term weights into a lower-dimensional sparse vector."""
    d = _as_dict(x)
    source = d.get(self.counts_key, {})
    acc: dict[int, float] = {}
    for term, weight in source.items():
        idx, sign = _hash_token(str(term), self.seed, self.n_features)
        acc[idx] = acc.get(idx, 0.0) + sign * float(weight)
    out = dict(d)
    out[self.hashed_key] = acc
    return out

select

Online feature selection.

VarianceThreshold drops features whose running variance is below a threshold. SelectKBest keeps the k features with the highest absolute running Pearson correlation with the target; the ranking is recomputed on every transform_one from the current statistics, so the kept set changes as the stream evolves.

VarianceThreshold 

VarianceThreshold(
    threshold: float = 0.0,
    features: list[str] | None = None,
)

Bases: Transformer

Drop features whose running variance is at or below threshold.

The variance is the unbiased running variance. A feature never seen passes through unchanged, so the first sample is never dropped by accident.

Memory grows with the number of features (one accumulator per feature); the total is O(d) for d features.

Parameters:

Name Type Description Default
threshold float

variance floor; features with var <= threshold are dropped from the output.

 0.0
features list[str] | None

explicit feature order.

 None

Examples:

>>> from dense_armor.utility.preprocessing.select import VarianceThreshold
>>> vt = VarianceThreshold(threshold=0.1)
>>> for v in [1.0, 2.0, 3.0, 4.0, 5.0]:
...     _ = vt.learn_one({"x": v, "c": 7.0})
>>> sorted(vt.transform_one({"x": 3.0, "c": 7.0}).keys())
['x']
Source code in dense_armor/utility/preprocessing/select.py 
80
81
82
83
84
85
def __init__(
    self, threshold: float = 0.0, features: list[str] | None = None
) -> None:
    self.threshold = threshold
    self.features = features
    self.stats_: dict[str, RunningMoments] | None = None

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> VarianceThreshold

Update the running variance of each feature.

Source code in dense_armor/utility/preprocessing/select.py 
 97
 98
 99
100
101
102
103
104
105
106
def learn_one(
    self, x: Any, y: Any = None, t: float | None = None
) -> "VarianceThreshold":
    """Update the running variance of each feature."""
    self._time_step(t)
    d = _as_dict(x)
    for k, s in self._ensure(d).items():
        if k in d:
            s.learn_one(d)
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> dict

Keep only the features above the variance threshold.

Source code in dense_armor/utility/preprocessing/select.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
def transform_one(self, x: Any, t: float | None = None) -> dict:
    """Keep only the features above the variance threshold."""
    d = _as_dict(x)
    if self.stats_ is None:
        return d
    out: dict = {}
    for k, v in d.items():
        if k in self.stats_:
            if self.stats_[k].var > self.threshold:
                out[k] = v
        else:
            out[k] = v
    return out

SelectKBest 

SelectKBest(k: int = 10, features: list[str] | None = None)

Bases: Transformer

Keep the k features with the highest absolute running correlation.

learn_one takes the target y and updates, for each feature, a running Pearson correlation with y. transform_one ranks the features by |corr| and keeps the top k. Features never seen pass through unchanged.

Memory grows with the number of features (one accumulator per feature); the total is O(d) for d features.

Parameters:

Name Type Description Default
k int

number of features to keep.

 10
features list[str] | None

explicit feature order.

 None

Examples:

>>> from dense_armor.utility.preprocessing.select import SelectKBest
>>> sk = SelectKBest(k=1)
>>> for i in range(20):
...     _ = sk.learn_one({"a": float(i), "b": 1.0}, y=2.0 * i)
>>> sorted(sk.transform_one({"a": 1.0, "b": 1.0}).keys())
['a']
Source code in dense_armor/utility/preprocessing/select.py 
150
151
152
153
154
155
def __init__(
    self, k: int = 10, features: list[str] | None = None
) -> None:
    self.k = k
    self.features = features
    self.stats_: dict[str, _CorrStats] | None = None

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> SelectKBest

Update the running correlation of each feature with y.

Source code in dense_armor/utility/preprocessing/select.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def learn_one(
    self, x: Any, y: Any = None, t: float | None = None
) -> "SelectKBest":
    """Update the running correlation of each feature with ``y``."""
    self._time_step(t)
    if y is None:
        return self
    d = _as_dict(x)
    stats = self._ensure(d)
    for k, s in stats.items():
        v = d.get(k)
        if isinstance(v, (int, float)):
            s.update(float(v), float(y))
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> dict

Keep the k features with the highest absolute correlation.

Source code in dense_armor/utility/preprocessing/select.py 
182
183
184
185
186
187
188
189
190
191
def transform_one(self, x: Any, t: float | None = None) -> dict:
    """Keep the ``k`` features with the highest absolute correlation."""
    d = _as_dict(x)
    if self.stats_ is None:
        return d
    ranked = sorted(
        self.stats_.items(), key=lambda kv: -abs(kv[1].corr)
    )
    keep = {k for k, _ in ranked[: self.k]}
    return {k: v for k, v in d.items() if k in keep or k not in self.stats_}

imbalance

Queue-based resampling for online class imbalance.

The algorithm keeps two fixed-length queues, one per class, and at each time step trains the wrapped classifier on the union of the two. The per-class queue bounds the effective class ratio: after warm-up the union is balanced (L positives and L negatives), even if the raw stream is 1 % positive. A queue that is not yet full simply does not contribute, so the first samples are learned as they arrive.

Algorithm 1 of Malialis, Panayiotou, Polycarpou (2018), p. 5, maps to this class as follows:

  1. input L: constructor argument queue_size.
  2. loop over time steps: learn_one is called once per step.
  3. receive example x^t: outside learn_one (the caller has it).
  4. predict class y_hat^t: outside learn_one (predict_one / predict_proba_one, delegated to the wrapped classifier).
  5. receive true label y^t: the second argument of learn_one.
  6. z^t = (x^t, y^t): built inside learn_one.
  7. if y^t == 0: append to the negative queue.
  8. else: append to the positive queue.
  9. q^t = q_p^t union q_n^t: the union of the two queues.
  10. calculate cost on q^t: delegated to the wrapped classifier, which trains on each element of the union.
  11. update classifier: learn_one on the wrapped classifier, once per element of the union.

The wrapper adds no state beyond the two queues, so memory is O(L) per class, fixed by queue_size.

References

Malialis, K., Panayiotou, C., Polycarpou, M. M. (2018). Queue-based resampling for online class imbalance learning. ICANN, section 3 and Algorithm 1 on p. 5.

QueueResampler 

QueueResampler(
    classifier: Any, queue_size: int = 25, positive: Any = 1
)

Bases: Classifier

Resampling wrapper for imbalanced online streams.

learn_one appends the labelled sample to the queue of its class, then trains the wrapped classifier once per element of the union of the two queues. predict_one and predict_proba_one delegate straight to the wrapped classifier.

Parameters:

Name Type Description Default
classifier Any

the wrapped classifier.

 required
queue_size int

length L of each per-class queue.

 25
positive Any

label of the positive class.

 1

Examples:

>>> from dense_armor.utility.preprocessing.imbalance import QueueResampler
>>> from dense_armor.roles import Classifier
>>> class Counter(Classifier):
...     def __init__(self):
...         self.n_seen_ = 0
...     def learn_one(self, x, y, t=None):
...         self.n_seen_ += 1
...         return self
...     def predict_one(self, x, t=None):
...         return 0
>>> qr = QueueResampler(Counter(), queue_size=3)
>>> for i in range(10):
...     _ = qr.learn_one({"x": i}, y=1 if i == 4 else 0)
>>> qr.classifier.n_seen_ > 10
True

Cost: 25 microseconds per call (median of 9600 calls, one Windows CPU thread). budget_s is 5 ms because check_p99_within_budget takes the p99 of 24 calls, which is their maximum: over 400 runs of that check the value was 27 microseconds at the median and 166 microseconds at the 99th percentile, with one run at 2.4 ms and one CI run at 2.75 ms, a pause of the runner and not the cost of the code.

References

Malialis, K., Panayiotou, C., Polycarpou, M. M. (2018). ICANN, Algorithm 1 on p. 5.

Source code in dense_armor/utility/preprocessing/imbalance.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(
    self,
    classifier: Any,
    queue_size: int = 25,
    positive: Any = 1,
) -> None:
    self.classifier = classifier
    self.queue_size = queue_size
    self.positive = positive
    self.q_positive_: deque = deque(maxlen=queue_size)
    self.q_negative_: deque = deque(maxlen=queue_size)
    self.n_seen_ = 0

learn_one 

learn_one(
    x: Any, y: Any, t: float | None = None
) -> QueueResampler

Append the sample to its queue and train on the union.

Source code in dense_armor/utility/preprocessing/imbalance.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def learn_one(
    self, x: Any, y: Any, t: float | None = None
) -> "QueueResampler":
    """Append the sample to its queue and train on the union."""
    self._time_step(t)
    z = (_as_dict(x), y)
    if y == self.positive:
        self.q_positive_.append(z)
    else:
        self.q_negative_.append(z)
    for xi, yi in list(self.q_positive_) + list(self.q_negative_):
        call_with_t(self.classifier.learn_one, xi, yi, t=t)
    self.n_seen_ += 1
    return self

predict_one 

predict_one(x: Any, t: float | None = None) -> Any

Delegate to the wrapped classifier.

Source code in dense_armor/utility/preprocessing/imbalance.py 
124
125
126
127
128
def predict_one(self, x: Any, t: float | None = None) -> Any:
    """Delegate to the wrapped classifier."""
    return call_with_t(
        self.classifier.predict_one, _as_dict(x), t=t
    )

predict_proba_one 

predict_proba_one(x: Any, t: float | None = None) -> dict

Delegate to the wrapped classifier.

Source code in dense_armor/utility/preprocessing/imbalance.py 
130
131
132
133
134
def predict_proba_one(self, x: Any, t: float | None = None) -> dict:
    """Delegate to the wrapped classifier."""
    return call_with_t(
        self.classifier.predict_proba_one, _as_dict(x), t=t
    )

Details

  • Scalers keep one statistic per feature and per joint; a feature never seen, or with zero spread, passes unchanged.
  • TFIDF(smooth_idf=True) uses $\ln(1 + N/\text{DF})$; this smoothing is not in Silajev (2026) and is off by default.
  • JointDerivatives keeps the last order + 1 valid samples; missing samples (NaN) are skipped and counted in n_missing.
  • Sources: Silajev, I. (2026), "TF-IDF and BM25 are exact KL divergences", arXiv:2609.14016. Weinberger, K. et al. (2009), "Feature hashing for large scale multitask learning", ICML, arXiv:0902.2206. Malialis, K., Panayiotou, C., Polycarpou, M. (2018), "Queue-based resampling for online class imbalance learning", ICANN, arXiv:1809.10388. ````