Failure risk and anomaly score from sensor time series with scikit-learn and FastAPI

Solution for the use case Predictive Maintenance for Manufacturing Equipment.

AI tools: scikit-learn, pandas, FastAPI, joblib
Language: Python
Published: 2026-10-26

Approach

A small, complete predictive-maintenance pipeline for machines that report sensor readings once per cycle (or per minute, per shift). It turns raw readings into rolling-window features and trains three scikit-learn models on run-to-failure history:

Failure classifier (HistGradientBoostingClassifier): probability that the machine fails within the next N cycles (default 30).

Remaining useful life (HistGradientBoostingRegressor): a rough estimate of cycles left, capped at 150.

Anomaly score (IsolationForest): trained only on healthy windows, so it also reacts to failure patterns that were never labeled.

A FastAPI endpoint scores the latest readings of one machine and returns the probability, the RUL estimate, the anomaly score and an alert flag. A synthetic data generator (wear that raises temperature and vibration) makes the example run without real data. A loader for the public NASA C-MAPSS turbofan files is included so you can try it on a real benchmark.

The split is by machine, never by row, so the test score is not inflated by neighbouring cycles of the same machine.

Implementation steps

Collect run-to-failure history: one row per machine and cycle with the sensor values; the RUL (cycles until failure) is known after the fact for machines that failed or were replaced.
Generate synthetic data with generatefleet() or load C-MAPSS with loadcmapss() to test the pipeline before connecting real data.
Build rolling-window features per machine and sensor (last value, mean, standard deviation, slope over 10 cycles) that only use past readings.
Label each row 1 if the machine fails within the next 30 cycles, and split train and test by machine with GroupShuffleSplit.
Train a HistGradientBoostingClassifier for failure risk and a HistGradientBoostingRegressor for capped RUL; report ROC AUC, average precision and MAE on the unseen machines.
Fit an IsolationForest on healthy windows only (RUL 100) and set the anomaly threshold at the 99th percentile of their scores.
Save everything as one joblib file and serve it with FastAPI: POST /score takes the last readings of a machine and returns probability, RUL estimate, anomaly score and alert.
Tune the alert probability with maintenance planners against the cost of a missed failure versus an unnecessary inspection, then monitor alerts and retrain when new failures are recorded.

Code (Python)

"""Train and apply the three models: failure classifier, RUL regressor, anomaly detector."""
from __future__ import annotations

from dataclasses import dataclass, field

import joblib
import numpy as np
import pandas as pd
from sklearn.ensemble import HistGradientBoostingClassifier, HistGradientBoostingRegressor, IsolationForest
from sklearn.metrics import average_precision_score, mean_absolute_error, roc_auc_score
from sklearn.model_selection import GroupShuffleSplit

from .data import add_failure_label
from .features import feature_columns, rolling_features


@dataclass
class MaintenanceModel:
    sensors: list[str]
    window: int
    horizon: int
    classifier: HistGradientBoostingClassifier
    regressor: HistGradientBoostingRegressor
    detector: IsolationForest
    anomaly_threshold: float  # 99th percentile of anomaly scores on healthy training windows
    alert_probability: float = 0.5
    metrics: dict = field(default_factory=dict)

    @property
    def features(self) -> list[str]:
        return feature_columns(self.sensors)

    def score(self, readings: pd.DataFrame) -> pd.DataFrame:
        """Score every cycle that has a full window of history."""
        X = rolling_features(readings, self.sensors, self.window)
        out = X[["machine_id", "cycle"]].copy()
        feats = X[self.features]
        out["failure_probability"] = self.classifier.predict_proba(feats)[:, 1]
        out["rul_estimate"] = np.clip(self.regressor.predict(feats), 0, None)
        out["anomaly_score"] = -self.detector.score_samples(feats)  # higher = more unusual
        out["anomaly"] = out["anomaly_score"] > self.anomaly_threshold
        out["alert"] = (out["failure_probability"] >= self.alert_probability) | out["anomaly"]
        return out

    def save(self, path: str) -> None:
        joblib.dump(self, path)

    @staticmethod
    def load(path: str) -> MaintenanceModel:
        return joblib.load(path)


def train(df: pd.DataFrame, sensors: list[str], window: int = 10, horizon: int = 30,
          healthy_rul: int = 100, test_size: float = 0.25, seed: int = 0) -> MaintenanceModel:
    """Train on run-to-failure data with a `rul` column.

    The split is by machine, so test machines are never seen during training.
    The anomaly detector only sees windows far from failure (RUL > healthy_rul).
    """
    data = add_failure_label(rolling_features(df, sensors, window), horizon)
    cols = feature_columns(sensors)
    split = GroupShuffleSplit(n_splits=1, test_size=test_size, random_state=seed)
    train_idx, test_idx = next(split.split(data, groups=data["machine_id"]))
    tr, te = data.iloc[train_idx], data.iloc[test_idx]

    clf = HistGradientBoostingClassifier(max_iter=200, learning_rate=0.1, random_state=seed)
    clf.fit(tr[cols], tr["fails_soon"])
    # RUL is capped: far from failure the sensors carry little information.
    reg = HistGradientBoostingRegressor(max_iter=200, random_state=seed)
    reg.fit(tr[cols], tr["rul"].clip(upper=150))

    healthy = tr[tr["rul"] > healthy_rul][cols]
    det = IsolationForest(n_estimators=200, random_state=seed).fit(healthy)
    threshold = float(np.quantile(-det.score_samples(healthy), 0.99))

    proba = clf.predict_proba(te[cols])[:, 1]
    metrics = {
        "test_machines": int(te["machine_id"].nunique()),
        "roc_auc": float(roc_auc_score(te["fails_soon"], proba)),
        "average_precision": float(average_precision_score(te["fails_soon"], proba)),
        "rul_mae_capped": float(mean_absolute_error(te["rul"].clip(upper=150), reg.predict(te[cols]))),
    }
    return MaintenanceModel(sensors, window, horizon, clf, reg, det, threshold, metrics=metrics)

Performance

Targets, not measured results. On the synthetic data the tests require ROC AUC above 0.9 on unseen machines; that only shows the pipeline works, because the synthetic wear is easy to detect.

Target for real equipment: at least 80 % of failures flagged 30 or more cycles ahead, at an alert rate the maintenance team can handle (agree a number, e.g. at most 2 false alerts per machine and month).

Training on 60 synthetic machines (about 15,000 rows) takes seconds on a laptop CPU; scoring one machine takes milliseconds.

Results depend on how many real failures you have. With fewer than 10 to 20 recorded failures per failure mode, rely more on the anomaly score than on the classifier.

Lessons learned

A maintenance planner or a dashboard calls POST /score with the latest readings of a machine and gets back JSON like {"failureprobability": 0.82, "rulestimate": 18.4, "anomaly_score": 0.61, "anomaly": true, "alert": true}. Alerts can be turned into work orders in the CMMS. Training prints the test metrics so the team can see how well the model does on machines it has not seen.