Skip to main content

Predictive maintenance — vibration anomaly detection

At a glance

Effort: ~1–2 hours · What you'll build: a model that learns a machine's healthy vibration and warns you when it starts to drift · Stack: the SDK for data in and out, plus three Python libraries — numpy, scipy, scikit-learn.

New to this?

No machine-learning background needed. If a word here is unfamiliar, the gentle primer explains it in one line. And to run this end-to-end, generate a sandbox first (section A creates the vibration signal used below).

The idea in one paragraph. A failing bearing vibrates differently long before it breaks — but not louder, so a simple "is it above X?" alarm misses it. The difference is in the texture of the vibration: tiny repeated shocks, energy shifting into certain frequencies. We'll let a model look at lots of examples of the machine running healthy, learn what healthy texture looks like, and then raise a flag when new data stops looking like it. Nothing here needs labelled failures — only normal running.

The quick examples raise an alarm on a raw threshold; this is the grown-up version that learns normal first.

Need data to run this?

These steps read a vibration series that already exists. Generate a sandbox first — section A ingests the degrading pump_07_vibration_mm_s this build scores.

Step 1 — Get a stretch of healthy data​

First we need examples of the machine running well. We pull a stretch of vibration readings from a period the machine was known to be healthy, and load them into a plain array of numbers we can do maths on. This data step is identical in every language:

import datahub_sdk, numpy as np, pandas as pd

client = datahub_sdk.DataHubClient.from_env()

def load_series(external_id, start, end):
    rf = datahub_sdk.RetrieveFilter(ts=external_id, start=start, end=end, limit=100_000)
    points = client.timeseries.retrieve_datapoints(rf)[0].get_datapoints()
    idx = pd.to_datetime([p.timestamp for p in points])
    return pd.Series([float(p.value) for p in points], index=idx).sort_index()

healthy = load_series(
    "pump_07_vibration_mm_s",
    pd.Timestamp.now(tz="UTC") - pd.Timedelta(hours=24),
    pd.Timestamp.now(tz="UTC") - pd.Timedelta(hours=6))

Step 2 — Turn the raw wiggle into a few meaningful numbers​

A model can't learn much from one raw vibration value — it has no context. So we chop the signal into short windows (say, two seconds each) and, for each window, calculate a few numbers that capture its texture. These numbers are called features. We use four, and here's what each one means in plain terms:

  • RMS — roughly the overall "loudness" or energy of the window.
  • Kurtosis — how spiky it is. A healthy machine hums smoothly; a failing bearing adds sharp little shocks that push this number up.
  • Crest factor — the biggest peak compared to the average level; another way to catch those shocks.
  • Band energy — how much of the vibration sits in a particular frequency range, found with an FFT (a standard tool that splits a wiggle into the pure tones it's made of). Bearing faults ring at characteristic frequencies, so energy showing up there is a red flag.
from scipy.stats import kurtosis
from scipy.fft import rfft, rfftfreq

SAMPLE_HZ = 1000        # the sensor reports 1000 readings per second
WINDOW = SAMPLE_HZ * 2  # work in 2-second windows

def features(window, fs=SAMPLE_HZ):
    x = window - window.mean()                       # remove the average so we see the wobble
    rms = np.sqrt(np.mean(x**2))                     # overall energy
    crest = np.max(np.abs(x)) / rms if rms else 0.0  # biggest peak vs. average
    spectrum = np.abs(rfft(x))                       # the FFT: how much of each frequency
    freqs = rfftfreq(len(x), 1 / fs)
    band = spectrum[(freqs >= 120) & (freqs <= 180)] # the bearing's fault frequency range
    band_energy = float(np.sum(band**2))
    return [rms, kurtosis(x), crest, band_energy]    # four numbers describe this window

def feature_matrix(series):
    vals = series.to_numpy()
    rows = [features(vals[i:i+WINDOW]) for i in range(0, len(vals) - WINDOW, WINDOW)]
    return np.array(rows)

X_healthy = feature_matrix(healthy)   # one row of four features per window

So each two-second window is now just four numbers. A week of healthy data becomes a big table of "this is what normal looks like."

Step 3 — Let the model learn "normal"​

Now we hand that table to an Isolation Forest. The one-line idea: it learns the shape of the normal cloud of points and can then tell, for any new point, how far outside that cloud it sits. It needs no examples of failure — a big deal, because you rarely have many. (We also scale the features first, so that one feature with big numbers doesn't drown out the others — a routine tidy-up step.)

from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

scaler = StandardScaler().fit(X_healthy)                       # remember each feature's scale
model = IsolationForest(contamination=0.01, random_state=0)    # 0.01 = expect ~1% odd points
model.fit(scaler.transform(X_healthy))                         # learn what healthy looks like

def anomaly_score(X):
    # the model returns "how normal"; we flip the sign so higher = more abnormal
    return -model.decision_function(scaler.transform(X))

Step 4 — Score new data, and write the answer back​

Finally we pull the latest hour, turn it into the same four features, and ask the model how abnormal each window is. We write that score back to the platform as a new series, so it's a normal, chartable signal — and if the score stays high (not just a one-off blip), we raise an event.

recent = load_series("pump_07_vibration_mm_s",
                     pd.Timestamp.now(tz="UTC") - pd.Timedelta(hours=1),
                     pd.Timestamp.now(tz="UTC"))
X_recent = feature_matrix(recent)
scores = anomaly_score(X_recent)
score_index = recent.index[WINDOW::WINDOW][:len(scores)]   # timestamp each window's score

client.timeseries.create([datahub_sdk.TimeSeries(
    external_id="pump_07_vibration_anomaly", name="Pump 07 vibration anomaly score", unit="score", value_type="float")])
client.timeseries.insert_from_lists(
    timestamps=score_index, values=scores, ts="pump_07_vibration_anomaly")

ALARM = 0.04                          # healthy scores sit near/below 0; a sustained positive score is abnormal
if scores[-5:].mean() > ALARM:        # sustained, not a single spike
    client.events.create([datahub_sdk.Event(
        external_id=f"degradation_predicted_pump_07_{int(pd.Timestamp.now().timestamp())}",
        type="degradation_predicted", status="open",
        event_time=score_index[-1],
        metadata={"asset": "pump_07", "score": f"{scores[-1]:.3f}", "model": "iforest_v1"})])

That's the whole loop. The anomaly score is now a first-class series — chart it beside the raw vibration, fold it into a health score, or alert on it like any other signal.

Where to take it further​

  • Name the failure. Once you've collected a few real failures (labels), a classifier can predict the type of failure, not just "something's off."
  • Estimate time-to-failure. Fit a line to the rising anomaly score and extrapolate to the alarm level for a rough "days left."
  • Cover the fleet. When one machine fails, walk the graph to every identical sibling and check them too.

Further reading​

See also​