Run a PyTorch Anomaly Detector in Rust with ONNX and tract

What you'll build: a PyTorch anomaly detector running in Rust with tract

At the end of this tutorial, a Rust binary with no Python inside it scores a minute of pump sensor data in a fraction of a millisecond on the verification machine. Its scores match the PyTorch model it came from with a relative error of 3 parts in 100,000:

valve1-0: 747 windows, 472 alerts, p50 42 us, p99 60 us{  "alert": false,  "device": "valve1-0",  "score": 0.17573332786560059,  "t": "2020-03-09 10:21:31"}747 events, max relative diff vs PyTorch 3.1e-05rust parity ok

Every window becomes one JSON event, and the event carries the raw anomaly score next to the alert flag. The last step needs that score. It replays a proposed threshold change against verdicts that have had time to arrive, and here it refuses the change:

week 16: false alarms in last week's alerts 0.42proposal: threshold 0.2973 -> 0.4460arrived-verdict recall 0.715blocked: matured recall 0.629 < 0.7hindsight recall, weeks 1-16: 0.663

You start in Python. You train a 1D convolutional autoencoder on real pump data, compare it fairly with an Isolation Forest baseline, and find out where its false alarms come from. Then you export it to Open Neural Network Exchange (ONNX) format with a parity check, run it from Rust with the tract inference library, and build a gate that judges threshold raises only on matured verdicts.

The level is intermediate: you should be comfortable with Python, PyTorch basics and running cargo build, but you need no prior ONNX or Rust inference experience. Plan on about 45 minutes of hands-on work. The first Rust build adds anything from several minutes to over half an hour, depending on your machine, and you can read ahead while it runs.

This is the hands-on companion to Edge AI Anomaly Detection on IoT Fleets Learns to Go Silent, which explains why a detector tuned on delayed labels drifts toward silence. You can follow this tutorial without reading it first.

Verified against Python 3.12.2, torch==2.15.0+cpu, onnx==1.23.2, onnxscript==0.7.2, onnxruntime==1.31.1, numpy==2.5.4, scikit-learn==1.9.1, pandas==3.1.1, Rust 1.99.1, tract 0.23.10, ndarray 0.17.2, serde_json 1.0.154 and anyhow 1.0.104, with SKAB at commit b2c0d46, on 2026-11-13.

Prerequisites: Python, PyTorch, ONNX Runtime and Rust

You need:

• Python 3.12. Training runs on a CPU; no GPU is used anywhere. Python 3.12 now receives security fixes only, so python.org no longer builds new Windows installers for it. On Windows, use the last 3.12 installer on python.org's release page, or install it with uv python install 3.12.
• Rust 1.91 or newer, which is the minimum version tract 0.23 supports. Install it with rustup. On Windows, rustup's default target uses the Microsoft Visual C++ (MSVC) linker, so you also need the Visual Studio Build Tools with the "Desktop development with C++" workload.
• Git, to clone the dataset.
• About 2 GB of disk. Measured on the verification machine: 1.0 GB for the virtual environment, 0.8 GB for the Rust build folder and 50 MB for the dataset.
• Free memory for the first Rust build. tract compiles its full default feature set, and Cargo runs one compiler job per CPU core. On a machine with little free RAM, build with cargo build --release -j 2 in Step 7.

No accounts, API keys or paid services are involved.

Create a project folder, clone the Skoltech Anomaly Benchmark (SKAB) dataset into it, and install the pinned packages. The PyTorch wheel comes from PyTorch's CPU-only index, which keeps the download small:

mkdir edge-anomalycd edge-anomalygit clone https://github.com/waico/SKABgit -C SKAB checkout b2c0d46python -m venv .venv

python in that last command must be Python 3.12. If it is another version, or the command does not exist, create the environment another way: py -3.12 -m venv .venv with the Windows python.org install, python3.12 -m venv .venv on Linux or macOS (Debian and Ubuntu also need the python3.12-venv package), or uv venv --python 3.12 .venv if you installed 3.12 with uv. The git checkout line prints a long warning about a "detached HEAD". That is expected, because you are on a fixed commit rather than a branch.

Activate the virtual environment with source .venv/bin/activate on Linux or macOS, .venv\Scripts\activate in PowerShell or cmd.exe, or source .venv/Scripts/activate in Git Bash. Run python --version and confirm it prints 3.12 before you install anything, so the packages land in the environment and not in a global Python. If PowerShell refuses to run the activation script, see "Fix common errors" below. Then install:

pip install torch==2.15.0 --index-url https://download.pytorch.org/whl/cpupip install onnx==1.23.2 onnxscript==0.7.2 onnxruntime==1.31.1 numpy==2.5.4pip install scikit-learn==1.9.1 pandas==3.1.1

The checkout pins SKAB to the commit this tutorial was verified against. If the upstream data ever changes, every count and score below would move. SKAB is licensed under GPL-3.0. Cloning it for your own experiments is fine, but if you later publish a repository, link to SKAB rather than copying its CSV files into yours.

Check that everything is in place before you write any code. Create check_env.py in edge-anomaly:

import sysimport numpy, onnx, onnxruntime, pandas, sklearn, torchprint(sys.version.split()[0], torch.__version__, onnx.__version__,      onnxruntime.__version__, numpy.__version__, sklearn.__version__,      pandas.__version__)

Then run it, and ask Cargo for its version:

python check_env.pycargo --version

Expected output. The first value is your Python version: any 3.12 release works, and 3.12.2 is what this tutorial was verified on. The cargo version can be any 1.91 or newer, and its commit hash and date will differ. On macOS, PyTorch prints 2.15.0 without the +cpu suffix:

3.12.2 2.15.0+cpu 1.23.2 1.31.1 2.5.4 1.9.1 3.1.1cargo 1.99.1 (5f94df478 2026-08-27)

Run every command in the rest of this tutorial from the edge-anomaly folder unless a step says otherwise.

How the Python training and Rust scoring halves connect

There are two halves, and they share exactly two files. The Python half trains the model and decides the alert threshold. The Rust half is what would ship to a device: it reads sensor rows, calibrates itself on the healthy first 400 rows of each recording (SKAB calls a recording a run), and scores every 60-second window. The handover between them is convae.onnx, the trained model, and threshold.json, the alert threshold. Nothing else crosses.

A third piece runs afterwards, back in Python. The gate reads the events the Rust binary wrote and joins them with labels as those labels arrive. This tutorial calls a window's confirmed label, faulty or healthy, its verdict, because on a real fleet it reaches you some time after the alert decision. In a real fleet that join happens on a server weeks later. SKAB already has every label, so here you simulate the delay.

direction: down
classes: {
  py: {style: {fill: "#4A90E2"; stroke: "#2C6FB0"; font-color: "#FFFFFF"}}
  rs: {style: {fill: "#955D37"; stroke: "#6E4426"; font-color: "#FFFFFF"}}
  file: {style: {fill: "#FFD93D"; stroke: "#C9A82A"; font-color: "#2C2C2A"}}
  gate: {style: {fill: "#7B68EE"; stroke: "#5A4BC4"; font-color: "#FFFFFF"}}
}
train: "Python, once: train on healthy SKAB\nwindows and pick the threshold" {class: py}
handover: {
  grid-rows: 1
  label: "The only things that cross"
  label.near: top-center
  onnx: "convae.onnx" {class: file}
  thr: "threshold.json" {class: file}
}
scorer: "Rust, on the device: edge-scorer reads\nSKAB sensor rows, calibrates, scores" {class: rs}
events: "events/*.jsonl: one score per window" {class: file}
gate: "Python, weeks later: matured-verdict gate\naccepts or blocks a threshold raise" {class: gate}
train -> handover: "1. export"
handover -> scorer: "2. load once"
scorer -> events: "3. one JSON line per window"
events -> gate: "4. replay with delayed labels"

Follow the numbered edges. Edge 1 runs once, at training time. Edges 2 and 3 are the device's job, and neither needs Python. Edge 4 is the slow loop. The gate can only judge a threshold change once enough labels have come back, and the shape of that delay is the whole reason the events carry scores.

Step 1: Load SKAB and cut it into calibrated windows

Goal: turn each SKAB run into normalised 60-second windows, split into healthy training windows and test windows.

Why this step. SKAB is 34 runs of a water pump testbed, each about 20 minutes of 8 sensors sampled once per second. The accelerometer columns are already root mean square (RMS) values, so this is slow multivariate sensor data, not a raw vibration waveform. You follow SKAB's own protocol, which treats the first 400 rows of each run as healthy training data.

You normalise each run on its own first 400 rows rather than on one global mean. The same pump on a different day sits at a different pressure and temperature, and a single fleet-wide scaler turns that offset into a false fault. Per-run calibration is also something a device can do by itself: record a healthy baseline after installation, then score against it. The Rust scorer repeats this calculation exactly, which is why it lives in one small function here.

direction: down
classes: {
  healthy: {style: {fill: "#6BCF7F"; stroke: "#3F9E52"; font-color: "#2C2C2A"}}
  test: {style: {fill: "#4A90E2"; stroke: "#2C6FB0"; font-color: "#FFFFFF"}}
  note: {style: {fill: "#95A5A6"; stroke: "#6B7B7C"; font-color: "#FFFFFF"}}
}
run: "One SKAB run, about 1,150 rows" {
  grid-rows: 1
  label.near: top-center
  base: "Rows 0-399\nhealthy baseline:\nmean and std per sensor" {class: healthy}
  rest: "Rows 400 onward\nscored" {class: test}
}
win: "60-row windows, shape (8, 60)" {
  grid-rows: 1
  label.near: top-center
  tw: "Train: windows ending\ninside rows 59-399" {class: healthy}
  sw: "Test: windows ending\nat row 400 or later" {class: test}
}
rule: "A window's label is the label\nof its last row" {class: note}
run -> win: "normalise with the baseline, then slide by 1 row"
win -> rule

Create data.py:

"""Load SKAB runs, normalise each run on its own healthy start, cut windows."""import globimport numpy as npimport pandas as pdWINDOW = 60          # 60 rows = 60 seconds at SKAB's 1 HzCALIB_ROWS = 400     # each run's first 400 rows are its healthy baselineSENSORS = ["Accelerometer1RMS", "Accelerometer2RMS", "Current", "Pressure",           "Temperature", "Thermocouple", "Voltage", "Volume Flow RateRMS"]def run_paths(root="SKAB/data"):    paths = []    for group in ("valve1", "valve2", "other"):        paths += glob.glob(f"{root}/{group}/*.csv")    return sorted(paths)def device_name(path):    """SKAB/data/valve1/0.csv -> valve1-0"""    group, name = path.replace("\\", "/").split("/")[-2:]    return f"{group}-{name[:-4]}"def load_run(path):    df = pd.read_csv(path, sep=";", index_col="datetime", parse_dates=True)    x = df[SENSORS].to_numpy("float32")    y = df["anomaly"].to_numpy("int64")    return df.index, x, ydef normalise(x):    mean = x[:CALIB_ROWS].mean(axis=0)    std = x[:CALIB_ROWS].std(axis=0) + 1e-6    return (x - mean) / stddef windows(z):    """(rows, 8) -> (rows - WINDOW + 1, 8, WINDOW): one window per end row."""    starts = range(len(z) - WINDOW + 1)    return np.stack([z[s:s + WINDOW].T for s in starts]).astype("float32")def split_run(path):    """Train windows sit inside the baseline; test windows end after it."""    times, x, y = load_run(path)    w = windows(normalise(x))    k = CALIB_ROWS - WINDOW + 1    return w[:k], w[k:], y[WINDOW - 1:][k:], times[WINDOW - 1:][k:]def load_all():    parts = [split_run(p) for p in run_paths()]    train = np.concatenate([p[0] for p in parts])    test = np.concatenate([p[1] for p in parts])    labels = np.concatenate([p[2] for p in parts])    return train, test, labelsif __name__ == "__main__":    train, test, labels = load_all()    print(train.shape, test.shape, round(float(labels.mean()), 3))

Run it. The __main__ block at the bottom of data.py loads every run and prints the array shapes and the share of faulty test windows:

python data.py

Expected output:

(11594, 8, 60) (23801, 8, 60) 0.537

What just happened. You have 11,594 healthy training windows and 23,801 test windows from the 34 runs in valve1, valve2 and other. The windows are channels-first, (8, 60), because that is the layout PyTorch's Conv1d expects. SKAB's runs are designed to contain a fault after the baseline, so 53.7% of test windows end on a faulty row. Labels never enter training. run_paths skips SKAB/data/anomaly-free/ on purpose: its one file has no anomaly column, so there is nothing to score it against.

Step 2: Score an Isolation Forest baseline

Goal: measure how well a standard Isolation Forest finds the faults, so the autoencoder has something to be compared with.

Why this step. Without a baseline you cannot tell what a neural network adds. Isolation Forest is the usual first choice for unlabelled anomaly detection. You try it two ways: on each sensor row alone (pointwise), and on the whole flattened 480-value window.

Both detectors, and later the autoencoder, use the same threshold rule. Score the healthy training windows, put the threshold at their 99th percentile, and alert on any test window above it. A deployed system would also have to set its threshold from healthy data only. Four numbers describe the result: recall (the share of faulty windows that alert), false alarm rate (the share of healthy windows that alert), precision, and F1.

direction: down
classes: {
  healthy: {style: {fill: "#6BCF7F"; stroke: "#3F9E52"; font-color: "#2C2C2A"}}
  step: {style: {fill: "#4A90E2"; stroke: "#2C6FB0"; font-color: "#FFFFFF"}}
  decide: {style: {fill: "#7B68EE"; stroke: "#5A4BC4"; font-color: "#FFFFFF"}}
}
a: "Score every healthy training window" {class: healthy}
b: "Threshold = 99th percentile\nof those scores" {class: decide}
c: "Score every test window" {class: step}
d: "Alert if score > threshold" {class: decide}
a -> b: "1"
b -> d: "2"
c -> d: "3"

Create metrics.py, which every later script reuses:

import numpy as npdef report(name, train_scores, test_scores, labels, q=0.99):    threshold = float(np.quantile(train_scores, q))    alert = test_scores > threshold    fault = labels == 1    tp = int((alert & fault).sum())    fp = int((alert & ~fault).sum())    recall = tp / int(fault.sum())    false_alarm_rate = fp / int((~fault).sum())    precision = tp / max(tp + fp, 1)    f1 = 2 * precision * recall / (precision + recall)    print(f"{name:15s} recall={recall:.3f} false_alarm_rate={false_alarm_rate:.3f} "          f"precision={precision:.3f} f1={f1:.3f}")    return threshold

Create baseline.py. Isolation Forest's score_samples returns lower values for more abnormal inputs, so the code negates it to match the autoencoder, where higher means more abnormal:

import numpy as npfrom sklearn.ensemble import IsolationForestfrom data import load_allfrom metrics import reporttrain, test, labels = load_all()print("train", train.shape, "test", test.shape, "anomaly share", round(labels.mean(), 3))flat_train = train.reshape(len(train), -1)flat_test = test.reshape(len(test), -1)forest = IsolationForest(n_estimators=200, random_state=0).fit(flat_train)report("iforest-window", -forest.score_samples(flat_train),       -forest.score_samples(flat_test), labels)point = IsolationForest(n_estimators=200, random_state=0).fit(train[:, :, -1])report("iforest-point", -point.score_samples(train[:, :, -1]),       -point.score_samples(test[:, :, -1]), labels)

Run it:

python baseline.py

Expected output:

train (11594, 8, 60) test (23801, 8, 60) anomaly share 0.537iforest-window  recall=0.905 false_alarm_rate=0.506 precision=0.674 f1=0.773iforest-point   recall=0.420 false_alarm_rate=0.166 precision=0.745 f1=0.537

What just happened. The pointwise forest misses most faults: recall 0.420. The windowed forest reaches an F1 of 0.773, which looks strong until you read its false alarm rate. At 0.506, half of all healthy minutes would page someone. F1 hides that because the test set is more than half faults, so false alarms barely dent precision. Keep the false alarm rate in view from here on, because it decides whether operators keep trusting the alerts.

Step 3: Train a 1D convolutional autoencoder for anomaly detection

Goal: train a small autoencoder whose forward pass returns the anomaly score directly.

Why this step. An autoencoder learns to compress and rebuild healthy windows. It rebuilds a faulty window badly, so the reconstruction error is the anomaly score. Two design choices matter for what comes later.

First, the model is convolutional over time. It learns local shapes across all 8 sensors at once, where the flattened forest treats 480 values as an unordered bag. Second, forward returns the mean squared error per window instead of the reconstruction. That puts the scoring arithmetic inside the exported graph, so the Rust side gets one number back and has no formula to reimplement and get subtly wrong.

direction: down
classes: {
  t: {style: {fill: "#4A90E2"; stroke: "#2C6FB0"; font-color: "#FFFFFF"}}
  z: {style: {fill: "#7B68EE"; stroke: "#5A4BC4"; font-color: "#FFFFFF"}}
  s: {style: {fill: "#FFA07A"; stroke: "#D9785A"; font-color: "#2C2C2A"}}
}
x: "Input window (8, 60)" {class: t}
enc: "Encoder" {
  grid-rows: 1
  label.near: top-center
  e1: "Conv1d 8 -> 32\nstride 2: (32, 30)" {class: t}
  e2: "Conv1d 32 -> 16\nstride 2: (16, 15)" {class: z}
}
dec: "Decoder" {
  grid-rows: 1
  label.near: top-center
  d1: "ConvTranspose1d 16 -> 32\n(32, 30)" {class: t}
  d2: "ConvTranspose1d 32 -> 8\n(8, 60)" {class: t}
}
score: "Score: mean of (reconstruction - input)\nsquared, one number per window" {class: s}
x -> enc: "1. compress"
enc -> dec: "2. rebuild"
dec -> score: "3. compare with the input"

Create model.py:

from torch import nnclass ConvAE(nn.Module):    """1D convolutional autoencoder whose output is the anomaly score."""    def __init__(self, channels=8):        super().__init__()        self.encoder = nn.Sequential(            nn.Conv1d(channels, 32, 7, stride=2, padding=3), nn.ReLU(),            nn.Conv1d(32, 16, 7, stride=2, padding=3), nn.ReLU())        self.decoder = nn.Sequential(            nn.ConvTranspose1d(16, 32, 7, stride=2, padding=3, output_padding=1),            nn.ReLU(),            nn.ConvTranspose1d(32, channels, 7, stride=2, padding=3, output_padding=1))    def forward(self, x):        recon = self.decoder(self.encoder(x))        return ((recon - x) ** 2).mean(dim=(1, 2))

Create train.py. Training minimises the mean score over healthy windows. The script then reports the same four numbers as the baseline and saves the weights and the threshold:

import jsonimport torchfrom data import load_allfrom metrics import reportfrom model import ConvAEtorch.manual_seed(0)train, test, labels = load_all()model = ConvAE()opt = torch.optim.Adam(model.parameters(), lr=1e-3)x = torch.from_numpy(train)for epoch in range(1, 31):    order = torch.randperm(len(x))    total = 0.0    for i in range(0, len(x), 64):        batch = x[order[i:i + 64]]        loss = model(batch).mean()        opt.zero_grad()        loss.backward()        opt.step()        total += loss.item() * len(batch)    if epoch % 10 == 0:        print(f"epoch {epoch} loss {total / len(x):.4f}")model.eval()with torch.no_grad():    train_scores = model(x).numpy()    test_scores = model(torch.from_numpy(test)).numpy()threshold = report("conv-ae", train_scores, test_scores, labels)torch.save(model.state_dict(), "convae.pt")json.dump({"threshold": threshold}, open("threshold.json", "w"))print(f"saved convae.pt, threshold {threshold:.4f}")

Run it (one to a few minutes on a laptop CPU, longer when memory is tight):

python train.py

Expected output:

epoch 10 loss 0.2005epoch 20 loss 0.1901epoch 30 loss 0.1861conv-ae         recall=0.792 false_alarm_rate=0.317 precision=0.743 f1=0.767saved convae.pt, threshold 0.2973

The seed fixes the run on one machine. Floating-point kernels can differ slightly on a different CPU, so the last digit of these numbers may move. Later steps read the threshold from threshold.json and stay consistent with whatever your run produced. If your output here differs, every later figure that depends on the trained weights will differ slightly too: the scores, the alert counts, and the numbers in Steps 4, 5, 11, 12 and 13. The parity checks in Steps 6 and 9 must pass either way.

What just happened. Under the same 99th-percentile rule, the autoencoder raises a false alarm on 31.7% of healthy windows, against the windowed forest's 50.6%, and its F1 is within 0.006 of the forest's. It also catches fewer faults: recall 0.792 against 0.905. You now have convae.pt, the trained weights, and threshold.json, the alert threshold of 0.2973.

Two things in that result should bother you. The detectors were compared at different recalls, so the lower false alarm rate may only mean the autoencoder alerts less often. And a threshold at the 99th percentile of healthy scores should give about 1% false alarms, not 31.7%. The next step checks both.

Step 4: Compare the detectors at equal recall

Goal: compare the autoencoder and the windowed forest at the same recall.

Why this step. A threshold rule picks one point on each detector's trade-off between catching faults and raising false alarms. Two detectors sitting at two different points cannot be ranked from one row each. The fair comparison fixes recall, moves each detector's threshold until it reaches that recall, and then compares false alarm rates. You do this at both recalls from Step 3, the autoencoder's 0.792 and the forest's 0.905. The script recomputes those recalls from the scores instead of copying them, so it stays correct if your Step 3 numbers differ.

direction: down
classes: {
  rule: {style: {fill: "#4A90E2"; stroke: "#2C6FB0"; font-color: "#FFFFFF"}}
  fair: {style: {fill: "#6BCF7F"; stroke: "#3F9E52"; font-color: "#2C2C2A"}}
}
cmp: "Two ways to compare detectors" {
  grid-rows: 1
  label.near: top-center
  same_rule: "Same threshold rule\neach lands on its own recall:\nnot comparable" {class: rule}
  same_recall: "Same recall\nmove each threshold to it,\nthen compare false alarms" {class: fair}
}

Create compare.py:

import numpy as npimport torchfrom sklearn.ensemble import IsolationForestfrom data import load_allfrom model import ConvAEtrain, test, labels = load_all()fault = labels == 1model = ConvAE()model.load_state_dict(torch.load("convae.pt"))model.eval()with torch.no_grad():    ae_train = model(torch.from_numpy(train)).numpy()    ae = model(torch.from_numpy(test)).numpy()forest = IsolationForest(n_estimators=200, random_state=0)forest.fit(train.reshape(len(train), -1))iso_train = -forest.score_samples(train.reshape(len(train), -1))iso = -forest.score_samples(test.reshape(len(test), -1))# The recall each detector reached under Step 3's 99th-percentile rule.recalls = [(s[fault] > np.quantile(t, 0.99)).mean()           for t, s in ((ae_train, ae), (iso_train, iso))]for recall in recalls:    for name, scores in (("conv-ae", ae), ("iforest-window", iso)):        cut = np.quantile(scores[fault], 1 - recall)        far = (scores[~fault] > cut).mean()        print(f"recall {recall:.3f}: {name:15s} false_alarm_rate={far:.3f}")

np.quantile(scores[fault], 1 - recall) is the threshold at which exactly that share of faulty windows scores above it. The next line counts how many healthy windows also score above that threshold. As a check on the method, the autoencoder's rate at its own recall comes out at 0.317, the same 31.7% Step 3 reported.

Run it:

python compare.py

Expected output:

recall 0.792: conv-ae         false_alarm_rate=0.317recall 0.792: iforest-window  false_alarm_rate=0.304recall 0.905: conv-ae         false_alarm_rate=0.535recall 0.905: iforest-window  false_alarm_rate=0.506

What just happened. At equal recall the autoencoder does not beat the forest. The forest has slightly fewer false alarms at both recalls. Step 3's gap came entirely from where the threshold rule happened to land each detector.

The rest of this tutorial ships the autoencoder anyway. It is a small neural graph that exports to one ONNX file and runs in tract, and that deployment path is what you are here to learn. Do not choose it over a forest for accuracy on data like this.

Step 5: Find where the false alarms come from

Goal: explain why a 99th-percentile threshold gives 31.7% false alarms instead of about 1%.

Why this step. The threshold came from windows inside each run's first 400 rows. Suppose a pump's healthy readings move away from that baseline over the following 20 minutes. Then healthy windows late in a run look abnormal to any detector calibrated at the start. A second effect is in the labels. A window takes the label of its last row, so a window that ends just after a fault is labelled healthy while most of its 60 rows are still faulty. Grouping the healthy windows both ways shows which effect is at work.

direction: down
classes: {
  h: {style: {fill: "#6BCF7F"; stroke: "#3F9E52"; font-color: "#2C2C2A"}}
  time: {style: {fill: "#FFA07A"; stroke: "#D9785A"; font-color: "#2C2C2A"}}
  tail: {style: {fill: "#FFD93D"; stroke: "#C9A82A"; font-color: "#2C2C2A"}}
}
all: "Healthy test windows (last row labelled healthy)" {class: h}
split: {
  grid-rows: 1
  label: "Two ways to group them"
  label.near: top-center
  by_time: "By rows since the\n400-row baseline:\n0-99, 100-299, 300+" {class: time}
  by_tail: "By whether the window\nstill holds faulty rows\nfrom a fault just ended" {class: tail}
}
all -> split: "false alarm rate per group"

Create drift.py:

import jsonimport numpy as npimport torchfrom data import CALIB_ROWS, WINDOW, load_run, run_paths, split_runfrom model import ConvAEmodel = ConvAE()model.load_state_dict(torch.load("convae.pt"))model.eval()threshold = json.load(open("threshold.json"))["threshold"]alert, healthy, since, holds_fault = [], [], [], []for path in run_paths():    _, test, labels, _ = split_run(path)    y = load_run(path)[2]    with torch.no_grad():        alert.append(model(torch.from_numpy(test)).numpy() > threshold)    healthy.append(labels == 0)    since.append(np.arange(len(test)))  # rows between the baseline and window end    ends = range(CALIB_ROWS, len(y))    holds_fault.append(np.array([y[e - WINDOW + 1:e + 1].any() for e in ends]))alert, healthy, since, holds_fault = map(    np.concatenate, (alert, healthy, since, holds_fault))for lo, hi in ((0, 100), (100, 300), (300, None)):    group = healthy & (since >= lo) & (since < (hi or since.max() + 1))    rows = f"{lo}-{hi - 1}" if hi else f"{lo}+"    print(f"healthy, {rows:7s} rows after baseline: "          f"false_alarm_rate={alert[group].mean():.3f}")tail = healthy & holds_faultprint(f"healthy, window still holds faulty rows: share={tail.sum() / healthy.sum():.3f} "      f"false_alarm_rate={alert[tail].mean():.3f}")print(f"healthy, no faulty row in window:        "      f"false_alarm_rate={alert[healthy & ~holds_fault].mean():.3f}")

Test window number i of a run ends at row 400 + i, which is why np.arange(len(test)) counts rows since the baseline and range(CALIB_ROWS, len(y)) lists the end rows.

Run it:

python drift.py

Expected output:

healthy, 0-99    rows after baseline: false_alarm_rate=0.062healthy, 100-299 rows after baseline: false_alarm_rate=0.137healthy, 300+    rows after baseline: false_alarm_rate=0.571healthy, window still holds faulty rows: share=0.167 false_alarm_rate=0.632healthy, no faulty row in window:        false_alarm_rate=0.254

What just happened. Both effects are real. In the first 100 rows after the baseline, the autoencoder raises false alarms on 6.2% of healthy windows. That is much closer to the 1% the rule promises, though still six times it. More than 300 rows later it alerts on 57.1% of them. Separately, 16.7% of the healthy windows still contain rows of a fault that just ended, and 63.2% of those alert. Remove them and the rate is still 25.4%, so distance from calibration is the larger cause. On this data, how recently a detector was calibrated matters more than which model it is.

Step 6: Export a PyTorch model to ONNX and check parity

Goal: write the model to a single convae.onnx file and prove that ONNX Runtime computes the same scores as PyTorch.

Why this step. ONNX is a framework-neutral file format for a computation graph and its weights. Once the model is ONNX, any runtime that reads ONNX can execute it, including tract in Rust. Export is also where models quietly change. Check parity immediately, while PyTorch is still there to compare against.

Three details in the export call matter for this model:

• PyTorch 2.15 uses its newer torch.export-based exporter by default (dynamo=True), which needs the onnxscript package you installed.
• external_data=False keeps the weights inside convae.onnx. The default writes them to a separate convae.onnx.data file, which is one more file to ship and to forget.
• The example input is a fixed (1, 8, 60) window and no dynamic shapes are declared. The device always scores one window of that size, and with a fully static graph tract has no shapes left to work out when it loads the model (Step 8 explains that load).

verbose=False silences the exporter's progress messages. On a Windows console that uses the cp1252 code page, those messages contain an emoji that crashes the export (see "Fix common errors" below).

direction: down
classes: {
  t: {style: {fill: "#4A90E2"; stroke: "#2C6FB0"; font-color: "#FFFFFF"}}
  o: {style: {fill: "#955D37"; stroke: "#6E4426"; font-color: "#FFFFFF"}}
  ok: {style: {fill: "#6BCF7F"; stroke: "#3F9E52"; font-color: "#2C2C2A"}}
  w: {style: {fill: "#98D8C8"; stroke: "#5FA898"; font-color: "#2C2C2A"}}
}
sample: "239 test windows\n(every 100th)" {class: w}
paths: {
  grid-rows: 1
  label: "Same windows, two runtimes"
  label.near: top-center
  pt: "PyTorch\nConvAE + convae.pt" {class: t}
  ort: "ONNX Runtime\nconvae.onnx" {class: o}
}
cmp: "max relative difference\nmust stay under 1e-4" {class: ok}
sample -> paths
paths -> cmp

Create export.py:

import numpy as npimport onnximport onnxruntime as ortimport torchfrom data import load_allfrom model import ConvAEmodel = ConvAE()model.load_state_dict(torch.load("convae.pt"))model.eval()example = torch.zeros(1, 8, 60)torch.onnx.export(model, (example,), "convae.onnx", input_names=["window"],                  output_names=["score"], external_data=False, verbose=False)onnx.checker.check_model("convae.onnx")print("opset", onnx.load("convae.onnx").opset_import[0].version)_, test, _ = load_all()session = ort.InferenceSession("convae.onnx", providers=["CPUExecutionProvider"])sample = test[::100]with torch.no_grad():    torch_scores = model(torch.from_numpy(sample)).numpy()onnx_scores = np.array(    [session.run(["score"], {"window": w[None]})[0][0] for w in sample])rel = (np.abs(torch_scores - onnx_scores) / np.abs(torch_scores)).max()print(f"compared {len(sample)} windows, max relative diff {rel:.1e}")assert np.allclose(torch_scores, onnx_scores, rtol=1e-4, atol=1e-5)print("parity ok")

Run it:

python export.py

Expected output. First you see four warning lines on stderr, one per skipped operator, such as W1113 06:42:37.341000 1660 .venv\Lib\site-packages\torch\onnx\_internal\exporter\_registration.py:107] torchvision is not installed. Skipping torchvision::nms. The timestamp, process ID and path differ on your machine. The warnings are harmless, because this model uses no torchvision operators. Then, on stdout:

opset 20compared 239 windows, max relative diff 3.6e-07parity ok

What just happened. convae.onnx now exists, about 58 KB, and it passed the ONNX checker. It declares opset 20. An opset is a numbered version of the ONNX operator set, and a runtime must support the opset a file declares. ONNX Runtime fed each window one at a time, as a device would, and matched PyTorch to within 4 parts in 10 million. If the assertion ever fails, stop here. A model that drifts at export drifts on every device you ship it to.

Step 7: Create the Rust project and compile tract

Goal: create the edge-scorer Rust project, pin its dependencies, and compile tract once.

Why this step. This is the half that would run on the device, and Steps 7 to 9 together form a minimal tract ONNX Rust example. tract is an ONNX inference library written entirely in Rust, maintained by Sonos. There is no C++ runtime to link and no Python, so the binary cross-compiles to ARM boards like any other Rust program. It runs on the CPU only, which suits a model of this size.

The first build of tract is by far the slowest command in this tutorial. You run it now, against the placeholder program cargo new creates, to keep the long compile separate from your own code. If this step fails, the cause is the toolchain or memory, not anything you wrote.

Create the Rust project inside edge-anomaly:

cargo new edge-scorer

Replace the [dependencies] section of edge-scorer/Cargo.toml with these pinned versions. The = prefix pins an exact version, which matters for tract because its 0.22 and 0.23 APIs differ:

[dependencies]anyhow = "=1.0.104"ndarray = "=0.17.2"serde_json = "=1.0.154"tract = "=0.23.10"

Run it. Build in release mode, then run the placeholder program. On a machine with little free memory, use cargo build --release -j 2 in place of the plain build command. The first build compiles tract and its default features from source, which takes anything from several minutes to over half an hour depending on your cores and memory:

cd edge-scorercargo build --releasecargo run --releasecd ..

Expected output. The build ends with a line like Finished `release` profile [optimized] target(s) in .... cargo run then prints Cargo's own status lines on stderr, followed by the placeholder's output:

Hello, world!

If the machine runs out of memory during the build, limit parallel jobs with cargo build --release -j 2.

What just happened. Every dependency, including the whole tract runtime, is now compiled and cached in edge-scorer/target/release/. The binary itself still only prints a greeting. From here on, a rebuild compiles only your own main.rs, which takes seconds.

Step 8: Run the ONNX model in Rust: write the tract scorer in main.rs

Goal: replace the placeholder with a scorer that loads convae.onnx, calibrates on a run's first 400 rows, and prints one JSON event per window with a latency summary.

Why this step. The binary has to repeat Step 1's preprocessing exactly: the same 8 columns, the same 400-row baseline, the same population standard deviation (NumPy's std default, which divides by the row count), the same 60-row window ending at the scored row. Any difference changes the scores even though the model itself is fine. That is why Step 9 compares Rust against PyTorch, not only against ONNX Runtime.

One coupling is easy to miss. Python selects the sensor columns by name, while the Rust code below reads them by position. That works because SKAB's column order at the pinned commit matches SENSORS in data.py. If you point the scorer at other data, check the header first.

direction: down
classes: {
  rs: {style: {fill: "#955D37"; stroke: "#6E4426"; font-color: "#FFFFFF"}}
  once: {style: {fill: "#98D8C8"; stroke: "#5FA898"; font-color: "#2C2C2A"}}
  out: {style: {fill: "#FFD93D"; stroke: "#C9A82A"; font-color: "#2C2C2A"}}
}
load: "Load convae.onnx once\ninto a runnable plan" {class: once}
read: "Read the run CSV\nmean and std from rows 0-399" {class: once}
loop: "For each row from 400 to the end" {
  grid-columns: 1
  label.near: top-center
  w: "1. Build the (1, 8, 60) window\nending at this row" {class: rs}
  r: "2. model.run, timed" {class: rs}
  j: "3. Print a JSON line:\ndevice, t, score, alert" {class: rs}
}
sum: "stderr: window count, alerts,\np50 and p99 latency" {class: out}
load -> read -> loop -> sum

Replace edge-scorer/src/main.rs with:

use std::{env, fs, time::Instant};use anyhow::{bail, Context, Result};use ndarray::Array3;use serde_json::json;use tract::prelude::*;tract::impl_ndarray_interop!();const WINDOW: usize = 60;const CALIB_ROWS: usize = 400;const SENSORS: usize = 8;fn read_run(path: &str) -> Result<(Vec<String>, Vec<[f32; SENSORS]>)> {    let text = fs::read_to_string(path).with_context(|| format!("reading {path}"))?;    let mut times = Vec::new();    let mut rows = Vec::new();    for line in text.lines().skip(1) {        let cols: Vec<&str> = line.split(';').collect();        if cols.len() < 1 + SENSORS {            bail!("short row in {path}: {line}");        }        let mut row = [0f32; SENSORS];        for (i, value) in cols[1..=SENSORS].iter().enumerate() {            row[i] = value.parse().with_context(|| format!("bad number {value}"))?;        }        times.push(cols[0].to_string());        rows.push(row);    }    Ok((times, rows))}fn normalise(rows: &[[f32; SENSORS]]) -> Vec<[f32; SENSORS]> {    let base = &rows[..CALIB_ROWS];    let n = CALIB_ROWS as f32;    let mut mean = [0f32; SENSORS];    let mut std = [0f32; SENSORS];    for c in 0..SENSORS {        mean[c] = base.iter().map(|r| r[c]).sum::<f32>() / n;        let var = base.iter().map(|r| (r[c] - mean[c]).powi(2)).sum::<f32>() / n;        std[c] = var.sqrt() + 1e-6;    }    rows.iter()        .map(|r| std::array::from_fn(|c| (r[c] - mean[c]) / std[c]))        .collect()}fn percentile(sorted: &[f64], p: f64) -> f64 {    sorted[((sorted.len() - 1) as f64 * p).round() as usize]}fn main() -> Result<()> {    let args: Vec<String> = env::args().collect();    if args.len() != 5 {        bail!("usage: edge-scorer <model.onnx> <run.csv> <threshold> <device>");    }    let threshold: f32 = args[3].parse()?;    let device = &args[4];    let model = tract::onnx()?.load(&args[1])?.into_model()?.into_runnable()?;    let (times, rows) = read_run(&args[2])?;    let z = normalise(&rows);    let mut micros = Vec::new();    let mut alerts = 0;    for end in CALIB_ROWS..rows.len() {        let start = end + 1 - WINDOW;        let window =            Array3::from_shape_fn((1, SENSORS, WINDOW), |(_, c, t)| z[start + t][c]);        let began = Instant::now();        let outputs = model.run([window.tract()?])?;        micros.push(began.elapsed().as_secs_f64() * 1e6);        let score = outputs[0].as_slice::<f32>()?[0];        let alert = score > threshold;        alerts += alert as usize;        let event = json!({            "device": device, "t": times[end], "score": score, "alert": alert        });        println!("{event}");    }    micros.sort_by(|a, b| a.partial_cmp(b).unwrap());    eprintln!(        "{device}: {} windows, {alerts} alerts, p50 {:.0} us, p99 {:.0} us",        micros.len(),        percentile(&micros, 0.50),        percentile(&micros, 0.99)    );    Ok(())}

Three calls load the model. tract::onnx()?.load(...) parses the ONNX file into a graph. into_model() gives every tensor a concrete type and shape and simplifies the graph, which works completely here because the input is fixed at (1, 8, 60). into_runnable() hands that typed graph to tract's default runtime, which prepares it as a plan: a fixed order of operations that every later model.run call reuses, so no per-window setup happens. model.run returns a list of output tensors. This graph has one output, a single 32-bit float, so outputs[0].as_slice::<f32>()?[0] reads the score.

tract::impl_ndarray_interop!() generates the conversions between tract tensors and your crate's own ndarray version, which is what makes window.tract()? work. The timer wraps only model.run, so the latency figures measure inference, not CSV parsing.

Run it. Rebuild, then run the binary with no arguments. It should refuse and print its usage line, which proves the new code is the one that runs:

cd edge-scorercargo build --releasecargo run --releasecd ..

Expected output. After Cargo's status lines, the program exits with an error. Cargo then adds its own line saying the process did not exit successfully, which is expected here:

Error: usage: edge-scorer <model.onnx> <run.csv> <threshold> <device>

What just happened. edge-scorer/target/release/ now holds the real edge-scorer binary (edge-scorer.exe on Windows). It contains the whole tract runtime, so it needs no Python and no separate ONNX runtime library next to it. It has not scored anything yet. A binary that compiles can still compute different scores, so the next step checks it against PyTorch.

Step 9: Check the Rust scores against PyTorch

Goal: run the binary on one SKAB run and confirm that its scores match PyTorch's.

Why this step. Step 6 proved that the ONNX file computes the same scores as PyTorch. This step proves the same about the whole device path: the CSV parsing, the calibration, the window slicing and the tract plan. The reference is PyTorch rather than ONNX Runtime, because every threshold came from PyTorch.

Create check_rust.py in edge-anomaly. It calls the binary through subprocess, which works the same in every shell. The path has no .exe, and that is fine: on Windows, starting a program without its extension still finds edge-scorer.exe:

import jsonimport osimport subprocessimport numpy as npimport torchfrom data import split_runfrom model import ConvAEEXE = os.path.join("edge-scorer", "target", "release", "edge-scorer")RUN = "SKAB/data/valve1/0.csv"threshold = str(json.load(open("threshold.json"))["threshold"])result = subprocess.run([EXE, "convae.onnx", RUN, threshold, "valve1-0"],                        capture_output=True, text=True, check=True)events = [json.loads(line) for line in result.stdout.splitlines()]print(result.stderr.strip())print(json.dumps(events[0], indent=2))model = ConvAE()model.load_state_dict(torch.load("convae.pt"))model.eval()_, test, _, _ = split_run(RUN)with torch.no_grad():    expected = model(torch.from_numpy(test)).numpy()rust = np.array([e["score"] for e in events])rel = (np.abs(rust - expected) / np.abs(expected)).max()print(f"{len(events)} events, max relative diff vs PyTorch {rel:.1e}")assert rel < 1e-4, "Rust preprocessing differs from data.py"print("rust parity ok")

Run it:

python check_rust.py

Expected output. Your latency figures depend on your CPU and will differ. If your Step 3 output matched exactly, every other line should match too. If it did not, the score and the alert count may differ slightly, but rust parity ok must still print:

valve1-0: 747 windows, 472 alerts, p50 42 us, p99 60 us{  "alert": false,  "device": "valve1-0",  "score": 0.17573332786560059,  "t": "2020-03-09 10:21:31"}747 events, max relative diff vs PyTorch 3.1e-05rust parity ok

What just happened. The Rust binary scored all 747 windows of valve1/0.csv that end after its baseline. The first event's timestamp, 10:21:31, is the 401st row of the file, which confirms the windows line up with Python's.

Its scores agree with PyTorch to within 3 parts in 100,000. That is looser than ONNX Runtime's match in Step 6. Both sides compute in 32-bit floats, but they add numbers up in a different order: NumPy's mean uses pairwise summation where the Rust loop sums in sequence, and tract's convolution kernels are not PyTorch's. The difference is far too small for an alert threshold to notice, and the assertion fails loudly if a preprocessing bug ever makes it larger.

The median latency was 0.04 ms per window on the machine used here. In the same session, Step 10's batch script timed the same file at a median of 0.15 ms. The binary and the file were identical, so the gap reflects the machine: other work running, and how warm the caches were. The timer also includes the first model.run call, with no warm-up. Treat the latency lines as a rough guide that depends on your CPU and on what else is running.

Step 10: Score all SKAB runs with the Rust binary

Goal: run the scorer on all 34 runs and keep the events, one JSONL file per run.

Why this step. The gate in Step 12 needs a history: many runs, each with every window's score. Those scores must come from the device binary, not from PyTorch. The gate has to judge what the fleet actually emitted, including any difference the Rust preprocessing introduces. In a fleet, this is what each device would upload. A Python script runs the loop, so the same command works in Bash and PowerShell.

Create score_all.py:

import jsonimport osimport subprocessimport sysfrom data import device_name, run_pathsexe = os.path.join("edge-scorer", "target", "release", "edge-scorer")threshold = str(json.load(open("threshold.json"))["threshold"])os.makedirs("events", exist_ok=True)total = 0for path in run_paths():    device = device_name(path)    with open(f"events/{device}.jsonl", "w") as out:        result = subprocess.run([exe, "convae.onnx", path, threshold, device],                                stdout=out, stderr=subprocess.PIPE, text=True)    if result.returncode != 0:        sys.exit(result.stderr)    print(result.stderr.strip())    with open(f"events/{device}.jsonl") as f:        total += sum(1 for _ in f)print(f"{len(run_paths())} event files, {total} events")

Run it:

python score_all.py

Expected output. 34 summary lines, one per run, in file-name order, then a total. The first four and the last line look like this, with your own latency figures:

other-1: 345 windows, 141 alerts, p50 39 us, p99 114 usother-10: 927 windows, 713 alerts, p50 72 us, p99 164 usother-11: 790 windows, 615 alerts, p50 126 us, p99 367 usother-12: 648 windows, 412 alerts, p50 76 us, p99 290 us...34 event files, 23801 events

What just happened. The events folder now holds 34 files, such as events/valve1-0.jsonl, with 23,801 events between them. That is the same count as Step 1's test windows, because these are those windows, scored by the device binary. Each line carries the raw score but no verdict yet. Verdicts arrive later, and Step 11 simulates when. Each file keeps the window order of its CSV, which is what lets Step 11 pair every event with its row's label.

Step 11: Build a ledger of delayed verdicts

Goal: pair every event with its label and record the week in which that label would reach you.

Why this step. A detector's alerts get checked quickly: a technician looks, and within days you know whether the alert was real. A miss gives no such signal. A missed fault surfaces only when the part fails in the field, weeks later. So in any recent week, the verdicts that have arrived are mostly about alerts. If someone measures recall on whatever has arrived, the misses that have not surfaced yet are absent, and the threshold looks safer to raise than it is. The companion article calls this delay pattern verdict skew.

SKAB has every label already, so this step simulates when each verdict would arrive, and Step 12 uses the result:

• Each of the 34 runs becomes one "week" of a fleet. The runs are assigned to weeks in a seeded random order, so that every week is a similar mix of SKAB's conditions. That keeps this demonstration about the delay alone. A real fleet also drifts from week to week, and matured evidence then describes the fleet as it was, which is a separate problem.
• An alerted window gets its verdict one week later.
• A healthy window that did not alert is also confirmed one week later.
• A missed fault gets its verdict MISS_WINDOW = 8 weeks later, when it fails.

direction: down
classes: {
  mat: {style: {fill: "#6BCF7F"; stroke: "#3F9E52"; font-color: "#2C2C2A"}}
  part: {style: {fill: "#FFD93D"; stroke: "#C9A82A"; font-color: "#2C2C2A"}}
  now: {style: {fill: "#7B68EE"; stroke: "#5A4BC4"; font-color: "#FFFFFF"}}
}
time: "Fleet weeks, as of week 16 (MISS_WINDOW = 8)" {
  grid-rows: 1
  label.near: top-center
  old: "Weeks 1-8: matured\nalerts and misses\nhave all arrived" {class: mat}
  recent: "Weeks 9-15: alert\nverdicts in, misses\nstill out" {class: part}
  now: "Week 16: now\nits own alerts\nnot yet checked" {class: now}
}

Create ledger.py:

"""Pair the device's events with SKAB labels and simulate when verdicts arrive."""import jsonimport numpy as npimport pandas as pdfrom data import CALIB_ROWS, device_name, load_run, run_pathsMISS_WINDOW = 8   # weeks before a missed fault surfaces as a field failuredef build_ledger(seed=0):    rows = []    for path in run_paths():        device = device_name(path)        _, _, labels = load_run(path)        with open(f"events/{device}.jsonl") as f:            events = [json.loads(line) for line in f]        assert len(events) == len(labels) - CALIB_ROWS, device        for event, fault in zip(events, labels[CALIB_ROWS:]):            rows.append((device, event["score"], event["alert"], int(fault)))    ledger = pd.DataFrame(rows, columns=["device", "score", "alert", "fault"])    devices = sorted(ledger.device.unique())    np.random.default_rng(seed).shuffle(devices)    ledger["week"] = ledger.device.map({d: i + 1 for i, d in enumerate(devices)})    # Alerts get a verdict next week; a missed fault only when it fails.    delay = np.where(ledger.alert, 1, MISS_WINDOW)    ledger["verdict_week"] = np.where(ledger.fault == 1, ledger.week + delay,                                      ledger.week + 1)    return ledgerif __name__ == "__main__":    ledger = build_ledger()    now = 16    seen = ledger[(ledger.fault == 1) & (ledger.week <= now)]    came = seen.verdict_week <= now    print(f"{len(ledger)} windows over {ledger.week.max()} weeks")    print(f"week {now}: {len(seen)} faulty windows scored so far")    print(f"  verdict arrived: {(came & seen.alert).sum()} alerted, "          f"{(came & ~seen.alert).sum()} missed")    print(f"  still pending:   {(~came & seen.alert).sum()} alerted, "          f"{(~came & ~seen.alert).sum()} missed")

The event files are in the same order as the CSV rows from row 400 onward, which is why zip against labels[CALIB_ROWS:] pairs each score with its row's label. The assert stops the run if a file is short, because zip would otherwise drop the extra labels silently.

Run it:

python ledger.py

Expected output:

23801 windows over 34 weeksweek 16: 5649 faulty windows scored so far  verdict arrived: 4286 alerted, 782 missed  still pending:   141 alerted, 440 missed

What just happened. By week 16, the fleet has scored 5,649 faulty windows. The verdicts that have arrived cover 4,286 of the 4,427 alerted ones, 97%, but only 782 of the 1,222 missed ones, 64%. The 141 pending alerts are week 16's own, which nobody has checked yet. So the arrived verdicts over-represent alerts, and any recall computed on them comes out too high. The ledger is now a table with one row per window and the columns device, score, alert, fault, week and verdict_week. The gate in the next step only reads it.

Step 12: Gate a threshold raise on matured verdicts (delayed labels)

Goal: decide whether a proposed threshold raise may ship, using only evidence old enough to include the faults the raise would miss.

Why this step. The fix for the skew Step 11 showed is to judge a raise only on windows that have matured: old enough that, if one was a miss, its failure would have surfaced by now. At week 16 with MISS_WINDOW = 8, those are weeks 1-8. The gate replays the proposed threshold against the stored scores of those matured windows, which it can do only because every event carries its score as well as its alert flag. Two more rules make it a gate: it requires at least MIN_RUNS = 6 matured runs, and it blocks any raise whose matured recall falls below FLOOR = 0.70.

direction: down
classes: {
  mat: {style: {fill: "#6BCF7F"; stroke: "#3F9E52"; font-color: "#2C2C2A"}}
  bad: {style: {fill: "#E74C3C"; stroke: "#B03A2E"; font-color: "#FFFFFF"}}
  in: {style: {fill: "#4A90E2"; stroke: "#2C6FB0"; font-color: "#FFFFFF"}}
}
prop: "Proposal: raise the threshold, replayed\nagainst the stored scores" {class: in}
est: {
  grid-rows: 1
  label: "Two recall estimates for the same raise"
  label.near: top-center
  arrived: "Arrived-verdict recall\nall weeks, misses missing\n0.715: looks safe" {class: bad}
  matured: "Matured recall\nweeks 1-8 only, complete\n0.629: blocked" {class: mat}
}
prop -> est

Create gate.py:

"""Matured-verdict gate over the scores the Rust scorer emitted."""import jsonfrom ledger import MISS_WINDOW, build_ledgerFLOOR = 0.70      # lowest recall the fleet acceptsMIN_RUNS = 6      # matured runs needed before any verdictdef arrived_recall(ledger, now, threshold):    faults = ledger[(ledger.fault == 1) & (ledger.verdict_week <= now)]    return float((faults.score > threshold).mean())def matured_recall(ledger, now, threshold):    faults = ledger[(ledger.fault == 1) & (ledger.week <= now - MISS_WINDOW)]    return float((faults.score > threshold).mean()), faults.device.nunique()def gate(ledger, now, proposed):    recall, runs = matured_recall(ledger, now, proposed)    if runs < MIN_RUNS:        return f"blocked: only {runs} matured runs"    if recall < FLOOR:        return f"blocked: matured recall {recall:.3f} < {FLOOR}"    return f"accepted: matured recall {recall:.3f}"if __name__ == "__main__":    current = json.load(open("threshold.json"))["threshold"]    proposed = round(current * 1.5, 4)    ledger = build_ledger()    now = 16    last = ledger[(ledger.week == now - 1) & ledger.alert]    share = (last.fault == 0).mean()    print(f"week {now}: false alarms in last week's alerts {share:.2f}")    print(f"proposal: threshold {current:.4f} -> {proposed:.4f}")    print(f"arrived-verdict recall {arrived_recall(ledger, now, proposed):.3f}")    print(gate(ledger, now, proposed))    seen = ledger[(ledger.fault == 1) & (ledger.week <= now)]    print(f"hindsight recall, weeks 1-{now}: {(seen.score > proposed).mean():.3f}")

Run it:

python gate.py

Expected output:

week 16: false alarms in last week's alerts 0.42proposal: threshold 0.2973 -> 0.4460arrived-verdict recall 0.715blocked: matured recall 0.629 < 0.7hindsight recall, weeks 1-16: 0.663

What just happened. At week 16, 42% of the previous week's alerts were false alarms. A number like that makes a team propose raising the threshold by half. Measured on arrived verdicts, recall at the new threshold is 0.715, above the 0.70 floor, so a naive check would ship it. The gate uses only the 8 matured weeks, measures 0.629, and blocks it. The last line uses every label, which no real fleet has at week 16, and shows the true recall over those weeks was 0.663. The block was correct.

Step 13: Backtest the gate across 20 week orders

Goal: measure how often the gate makes the right call, against simpler rules, over many week orders and two proposals.

Why this step. One week order is one example, not evidence. The arrived-verdict estimate is biased upward by construction, since the misses it leaves out are exactly the windows the raise would lose. The matured estimate has no such bias, but it rests on 8 runs, so it is noisy. A gate also has to beat simple fixed rules. On a raise that should almost never ship, a rule that always blocks is hard to beat, so the test also needs a raise that usually should ship. Here that is a 1.1x raise next to Step 12's 1.5x.

direction: down
classes: {
  in: {style: {fill: "#4A90E2"; stroke: "#2C6FB0"; font-color: "#FFFFFF"}}
  rule: {style: {fill: "#7B68EE"; stroke: "#5A4BC4"; font-color: "#FFFFFF"}}
  truth: {style: {fill: "#6BCF7F"; stroke: "#3F9E52"; font-color: "#2C2C2A"}}
}
cases: "20 week orders x 2 proposals (1.1x, 1.5x)" {class: in}
rules: {
  grid-rows: 1
  label: "Four ways to decide"
  label.near: top-center
  g: "Matured-verdict\ngate" {class: rule}
  a: "Arrived-verdict\nrecall >= floor" {class: rule}
  b: "Always\nblock" {class: rule}
  s: "Always\nship" {class: rule}
}
truth: "Scored against hindsight recall >= 0.70" {class: truth}
cases -> rules -> truth

Create sweep.py:

"""Replay two week-16 proposals under 20 different week orders."""import jsonfrom gate import FLOOR, arrived_recall, gatefrom ledger import build_ledgerNOW = 16SEEDS = range(20)current = json.load(open("threshold.json"))["threshold"]print("raise  should ship  wrong calls: gate  arrived  always-block  always-ship")for factor in (1.1, 1.5):    proposed = round(current * factor, 4)    ship = gate_wrong = naive_wrong = optimistic = 0    for seed in SEEDS:        ledger = build_ledger(seed)        seen = ledger[(ledger.fault == 1) & (ledger.week <= NOW)]        truth = (seen.score > proposed).mean()        arrived = arrived_recall(ledger, NOW, proposed)        accepted = gate(ledger, NOW, proposed).startswith("accepted")        ship += truth >= FLOOR        gate_wrong += accepted != (truth >= FLOOR)        naive_wrong += (arrived >= FLOOR) != (truth >= FLOOR)        optimistic += arrived > truth    blocked_wrong, shipped_wrong = ship, len(SEEDS) - ship    print(f"x{factor:<4}  {ship:>11}  {gate_wrong:>17}  {naive_wrong:>7}  "          f"{blocked_wrong:>12}  {shipped_wrong:>11}")    print(f"       arrived-verdict recall above hindsight in {optimistic} of 20 orders")

A call is wrong when it ships a raise whose hindsight recall is below the floor, or blocks one whose hindsight recall is at or above it. Always-block is wrong exactly when the raise should ship, and always-ship exactly when it should not, so their columns need no separate loop.

Run it:

python sweep.py

Expected output:

raise  should ship  wrong calls: gate  arrived  always-block  always-shipx1.1            14                  3        6            14            6       arrived-verdict recall above hindsight in 20 of 20 ordersx1.5             1                  3        9             1           19       arrived-verdict recall above hindsight in 20 of 20 orders

What just happened. In all 40 cases, recall measured on arrived verdicts came out higher than the true recall. The direction of that bias follows from the simulated delays, since misses arrive last by construction. What the sweep measures is how often the bias flips the decision: 6 and 9 times out of 20.

On the 1.1x raise, which should ship in 14 of 20 orders, the gate made 3 wrong calls. The arrived-verdict check made 6, always-ship also made 6, and always-block made 14. On the 1.5x raise, which should ship in only 1 order, the arrived-verdict check made 9 wrong calls and the gate again made 3. Always-block made just 1 there, because almost nothing should ship. One proposal is not enough to judge a gate.

This is the gate's best case: its maturity window equals the simulated miss delay, 8 weeks in both. On a real fleet, MISS_WINDOW is an estimate, and setting it too short lets unsurfaced misses back into the matured set. The gate's 3 errors are its noise. On the 1.1x raise, it accepted once with matured recall 0.775 when the true recall was 0.696, and blocked twice with 0.686 and 0.677 when the truth was 0.781 and 0.725. sweep.py prints only the counts. These per-order figures come from also printing the seed, the gate's message and truth inside the loop whenever the gate's call is wrong. Recall measured on only 8 matured runs can miss the truth by about 0.1 in either direction. On a real fleet, size MIN_RUNS so that this noise is smaller than the usual distance between a proposal and the floor.

Fix common errors: ONNX export, tract and the Rust build

The cp1252, UTF-16, Git Bash and two missing-file rows were reproduced on the machine used to verify this tutorial. The out-of-memory row is the message Cargo prints when Linux kills a compiler job, and it was not reproduced here. The others are the messages these tools print, and the References list the documentation and issue trackers behind each cause and fix. Long messages are shortened to the part you can search for.

The finished edge anomaly detection pipeline

The diagram shows the finished system, labelled with the names from your code. The Python training side and the Rust device side meet only at two files, and the gate reads the device's output with a deliberate delay.

direction: down
classes: {
  py: {style: {fill: "#4A90E2"; stroke: "#2C6FB0"; font-color: "#FFFFFF"}}
  rs: {style: {fill: "#955D37"; stroke: "#6E4426"; font-color: "#FFFFFF"}}
  file: {style: {fill: "#FFD93D"; stroke: "#C9A82A"; font-color: "#2C2C2A"}}
  gate: {style: {fill: "#7B68EE"; stroke: "#5A4BC4"; font-color: "#FFFFFF"}}
}
training: "Training side (Python)" {
  grid-rows: 1
  label.near: top-center
  data: "data.py\nnormalise, windows" {class: py}
  train: "train.py\nConvAE, 30 epochs" {class: py}
  export: "export.py\nONNX + parity" {class: py}
}
art: "Artifacts" {
  grid-rows: 1
  label.near: top-center
  onnx: "convae.onnx\n58 KB, opset 20" {class: file}
  thr: "threshold.json\n0.2973" {class: file}
}
scorer: "edge-scorer (Rust): tract plan,\nper-run calibration, timer" {class: rs}
ev: "events/*.jsonl: device, t, score, alert" {class: file}
gate: "ledger.py + gate.py: delayed verdicts,\nmatured recall, FLOOR 0.70" {class: gate}
training -> art: "export"
art -> scorer: "load"
scorer -> ev: "score_all.py, 34 runs"
ev -> gate: "delayed labels"

The device box contains no Python and no model-specific arithmetic. The score formula is inside convae.onnx, and the binary only reproduces the input calibration. Upgrading the model means replacing one 58 KB file, as long as the window shape stays (1, 8, 60). The gate judges only threshold raises, because a raise costs missed faults, and those are the last verdicts to arrive.

Complete project file tree and run order

If anything diverged along the way, compare your folder with this tree. Every file appears in full in the step named next to it.

edge-anomaly/  .venv/                     Prerequisites  SKAB/                      git clone, not modified  check_env.py               Prerequisites  data.py                    Step 1  metrics.py                 Step 2  baseline.py                Step 2  model.py                   Step 3  train.py                   Step 3  convae.pt                  created by train.py  threshold.json             created by train.py  compare.py                 Step 4  drift.py                   Step 5  export.py                  Step 6  convae.onnx                created by export.py  check_rust.py              Step 9  score_all.py               Step 10  events/                    created by score_all.py, 34 .jsonl files  ledger.py                  Step 11  gate.py                    Step 12  sweep.py                   Step 13  __pycache__/               created by Python on import  edge-scorer/    .gitignore, .git/        created by cargo new, unless edge-anomaly is in a git repo    Cargo.toml               Step 7, [dependencies] section    Cargo.lock               created by cargo build    src/main.rs              Step 8    target/release/          created by cargo build --release

Run order from a clean folder: python baseline.py (optional), python train.py, then python compare.py and python drift.py (both optional), python export.py, cargo build --release inside edge-scorer (add -j 2 if memory is tight), python check_rust.py, python score_all.py, python ledger.py (optional), python gate.py, python sweep.py.

Where to go next: ARM boards, censored verdicts and model rollouts

• Run the scorer on an ARM board and measure it under sustained load. Because tract is pure Rust, cargo build --release --target aarch64-unknown-linux-gnu with a cross linker produces a binary for a Raspberry Pi-class board. Once convae.onnx sits on a device, anyone with the device can copy it. The attack surface of on-device AI covers what that exposes. Score the same runs in a loop for ten minutes, not one, and compare the p99 from the first minute with the last. A short benchmark measures the board before it throttles. Chapter 12 of my book Right-Sized AI, "Edge and Embedded Deployment," works through that gap between peak and sustained throughput, and the energy budget of a device that runs on a battery.
• Add censoring to the gate. Here every true alarm gets a verdict. On a real fleet, a repair after a true alarm often removes the evidence, so some true alarms are never recorded as faults. Add a record_rate below 1.0 to build_ledger, drop that share of true-alarm verdicts, and watch which way matured_recall moves. The companion article shows why that rate has to be measured, not assumed.
• Ship a second model to part of the fleet. Narrow the bottleneck from 16 channels to 8, retrain, export convae-v2.onnx, and score half the runs with each model. The gate then compares two models instead of two thresholds, and the rollback question becomes concrete: how long would it take to move every device back? Chapter 13 of Right-Sized AI, "Fleet Operations and On-Device Lifecycle," computes rollback duration across a real update channel and explains why a fleet's version tail never reaches zero.

References

Dataset and method:

• Katser, I. D. and Kozitsin, V. O. (2020). Skoltech Anomaly Benchmark (SKAB). Kaggle, DOI 10.34740/KAGGLE/DSV/1693952. Data and code, GPL-3.0, commit b2c0d46 (2024-08-11). github.com/waico/SKAB
• Liu, F. T., Ting, K. M. and Zhou, Z.-H. (2008). Isolation Forest. IEEE International Conference on Data Mining. doi.org/10.1109/ICDM.2008.17

Python and its tooling:

• Python Software Foundation (2026). Status of Python versions, where 3.12 is listed in its security-fix phase, which ships source-only releases. devguide.python.org/versions
• Astral (2026). uv: Installing Python, for uv python install 3.12. docs.astral.sh/uv/guides/install-python
• PyTorch (2026). Get started locally, including the CPU-only wheel index. pytorch.org/get-started/locally
• PyTorch (2026). torch.onnx export, version 2.15, including the dynamo and external_data defaults. docs.pytorch.org/docs/2.15/onnx_export.html
• ONNX (2026). ONNX operator reference (the live page tracks the newest ONNX release; this tutorial pins 1.23). onnx.ai/onnx/operators
• Microsoft (2026). onnxscript, version 0.7.2. github.com/microsoft/onnxscript
• Microsoft (2026). ONNX Runtime Python API (this tutorial pins 1.31). onnxruntime.ai/docs/api/python
• Microsoft (2026). ONNX Runtime: Common errors, the source of the two INVALID_ARGUMENT rows. onnxruntime.ai/docs/api/python/auto_examples/plot_common_errors.html
• NumPy (2026). numpy.sum, version 2.5, whose notes describe the pairwise summation behind mean. numpy.org/doc/2.5/reference/generated/numpy.sum.html
• pandas (2026). pandas documentation, version 3.1. pandas.pydata.org/docs
• scikit-learn (2026). IsolationForest, version 1.9. scikit-learn.org/stable/modules/generated/sklearn.ensemble.IsolationForest.html

Rust crates and toolchain:

• crates.io (2026). tract 0.23.10, whose metadata declares the minimum Rust version, 1.91. crates.io/crates/tract/0.23.10
• Sonos (2026). tract, version 0.23.10. github.com/sonos/tract and docs.rs/tract/0.23.10
• ndarray (2026). ndarray, version 0.17.2. docs.rs/ndarray/0.17.2
• serde_json (2026). serde_json, version 1.0.154. docs.rs/serde_json/1.0.154
• anyhow (2026). anyhow, version 1.0.104. docs.rs/anyhow/1.0.104
• The Rust Project (2026). Cargo build: the --jobs option. doc.rust-lang.org/cargo/commands/cargo-build.html
• Microsoft (2026). Visual Studio Build Tools. visualstudio.microsoft.com/visual-cpp-build-tools
• The Rust Project (2026). rustup on Windows with the MSVC toolchain. rust-lang.github.io/rustup/installation/windows-msvc.html
• Microsoft (2026). about_Execution_Policies. learn.microsoft.com/powershell/module/microsoft.powershell.core/about/about_execution_policies
• Microsoft (2026). Out-File, Windows PowerShell 5.1, whose default -Encoding is unicode (UTF-16 little-endian); the > redirection operator behaves like Out-File without extra parameters. learn.microsoft.com/powershell/module/microsoft.powershell.utility/out-file?view=powershell-5.1

Error reports behind the troubleshooting table:

• PyTorch Forums (2025). Issues with exporting PyTorch model as ONNX, the missing onnxscript error. discuss.pytorch.org/t/issues-with-exporting-pytorch-model-as-onnx/216394
• marcoslucianops (2025). DeepStream-Yolo issue #668, the same error in another project. github.com/marcoslucianops/DeepStream-Yolo/issues/668
• Rust Users Forum (2025). link.exe not found error when building Rust on Windows 11 MSVC target. users.rust-lang.org/t/link-exe-not-found-error-when-building-rust-on-windows-11-msvc-target/133232
• scikit-learn (2023). Issue #25844, the IsolationForest feature-name warning. github.com/scikit-learn/scikit-learn/issues/25844

Related reading:

• Kumar, R. (2026). Edge AI Anomaly Detection on IoT Fleets Learns to Go Silent. /edge-ai-iot-fleet-anomaly-detection-verdict-skew
• Kumar, R. (2026). Right-Sized AI: Small Language Models in Production. Chapter 12, "Edge and Embedded Deployment," and Chapter 13, "Fleet Operations and On-Device Lifecycle." right-sized-ai.ranjankumar.in

Related Articles

AI Engineering

• How to Write a Chat Template for a Fine-Tuned Model and Test It
• Claude Code Stop Hook with Hypothesis Property Tests
• How to Detect Chat Template Backdoors and Forged Tokens

Iot Edge Computing

• Edge AI Anomaly Detection on IoT Fleets Learns to Go Silent