AI-Driven Anomaly Detection for Industrial Equipment

Imagine: a critical pump stops at a factory. One hour of downtime costs 500,000 rubles. We build predictive maintenance systems that predict failures 2-3 weeks before breakdown. The architecture revolves around a core priority: minimize missed failures while maintaining an acceptable false positive rate. The foundation is a combination of five methods—from instant thresholds to physics models.

Problems We Solve

Missed failures vs. false alarms. A single detector yields a high FPR—up to 30% on noisy data. Multi-level consensus reduces FPR to 5% while maintaining 95% sensitivity. The system is designed for industrial IoT and uses AI diagnostics at every level.

Noisy data and correlated incidents. One pump failure causes anomalies in pressure, temperature, and flow. Our system groups them into a single incident rather than generating multiple alerts.

How We Build the Anomaly Detection System

Multi-level architecture combines five approaches:

L1/L2 provide instant alerts, L3/L4 early diagnostics, L5 long-term trend. Consensus reduces the false positive rate by 40% compared to a single detector.

Feature Engineering for Equipment

We transform raw sensor signals (vibration, current, pressure) into multi-domain features:

import numpy as np
from scipy import stats, signal

def extract_equipment_features(raw_signal, sampling_rate=1000):
    # Time domain
    features = {
        'rms': np.sqrt(np.mean(raw_signal**2)),
        'peak': np.max(np.abs(raw_signal)),
        'crest_factor': np.max(np.abs(raw_signal)) / np.sqrt(np.mean(raw_signal**2)),
        'kurtosis': stats.kurtosis(raw_signal),
        'skewness': stats.skew(raw_signal),
        'peak_to_peak': np.ptp(raw_signal),
        'shape_factor': np.sqrt(np.mean(raw_signal**2)) / np.mean(np.abs(raw_signal))
    }
    # Frequency domain
    freqs, psd = signal.welch(raw_signal, fs=sampling_rate, nperseg=512)
    total_power = np.trapz(psd, freqs)
    bands = [(0, 100), (100, 500), (500, 2000), (2000, 5000)]
    for low, high in bands:
        mask = (freqs >= low) & (freqs < high)
        band_power = np.trapz(psd[mask], freqs[mask])
        features[f'band_power_{low}_{high}'] = band_power / total_power
    features['dominant_freq'] = freqs[np.argmax(psd)]
    features['spectral_centroid'] = np.sum(freqs * psd) / np.sum(psd)
    return features

What matters is not the absolute value but the deviation from the asset's normal baseline. Hence we compute delta features:

def compute_delta_features(current_features, baseline_features, trend_features_7d):
    deltas = {}
    for key in current_features:
        if key in baseline_features:
            deltas[f'{key}_delta_abs'] = current_features[key] - baseline_features[key]
            if baseline_features[key] != 0:
                deltas[f'{key}_delta_pct'] = ((current_features[key] - baseline_features[key]) / abs(baseline_features[key]) * 100)
        if key in trend_features_7d:
            deltas[f'{key}_trend_7d'] = trend_features_7d[key]
    return deltas

Unsupervised Detection

Isolation Forest with seasonality adaptation is trained on 30+ days of normal operation. Autoencoder (LSTM) captures reconstruction error for multi-sensor windows:

from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import pandas as pd
import numpy as np

class EquipmentAnomalyDetector:
    def __init__(self, contamination=0.02):
        self.scaler = StandardScaler()
        self.model = IsolationForest(contamination=contamination, n_estimators=200, random_state=42)
        self.baseline_built = False

    def fit_baseline(self, normal_operation_features: pd.DataFrame):
        X = self.scaler.fit_transform(normal_operation_features)
        self.model.fit(X)
        self.baseline_built = True
        scores = self.model.score_samples(X)
        self.threshold = np.percentile(scores, 5)

    def detect(self, current_features: dict) -> dict:
        if not self.baseline_built:
            return {'status': 'no_baseline', 'anomaly': False}
        X = self.scaler.transform([list(current_features.values())])
        score = self.model.score_samples(X)[0]
        is_anomaly = score < self.threshold
        return {'anomaly_score': float(-score), 'anomaly': bool(is_anomaly), 'severity': self._classify_severity(-score)}

    def _classify_severity(self, anomaly_score):
        if anomaly_score > 0.8: return 'critical'
        if anomaly_score > 0.6: return 'high'
        if anomaly_score > 0.4: return 'medium'
        return 'low'

import torch
import torch.nn as nn

class SensorAutoencoder(nn.Module):
    def __init__(self, input_dim, hidden_dim=32, latent_dim=8, seq_len=60):
        super().__init__()
        self.encoder = nn.LSTM(input_dim, hidden_dim, batch_first=True)
        self.bottleneck = nn.Linear(hidden_dim, latent_dim)
        self.decoder = nn.LSTM(latent_dim, hidden_dim, batch_first=True)
        self.output_layer = nn.Linear(hidden_dim, input_dim)

    def forward(self, x):
        enc_out, _ = self.encoder(x)
        z = self.bottleneck(enc_out[:, -1, :])
        z_expanded = z.unsqueeze(1).repeat(1, x.shape[1], 1)
        dec_out, _ = self.decoder(z_expanded)
        reconstruction = self.output_layer(dec_out)
        return reconstruction

Alert Consensus

Different detectors catch different anomaly types. For standard equipment, agreement of 2 out of 3 detectors is required; for critical equipment, a single detector suffices. Alerts correlated in time and topology are grouped into one incident.

Why Multi-Level Architecture Is More Effective Than a Single Method?

Each method has blind spots. Isolation Forest poorly catches slow trend drift, while statistical methods miss multivariate patterns. The combination provides full coverage. In our projects, consensus reduced false alarms by 60%.

Model Drift Monitoring

Over time, equipment wear changes baseline features. Weekly KS-test checks for drift:

from scipy.stats import ks_2samp

def detect_model_drift(recent_features, baseline_features, p_threshold=0.01):
    drift_features = []
    for col in recent_features.columns:
        stat, p_value = ks_2samp(baseline_features[col], recent_features[col])
        if p_value < p_threshold:
            drift_features.append(col)
    drift_ratio = len(drift_features) / len(recent_features.columns)
    return {'drift_detected': drift_ratio > 0.3, 'drifted_features': drift_features, 'drift_ratio': drift_ratio}

When drift is detected, retraining is triggered on the last 60 days of data (if they are labeled as normal).

What's Included

1. Equipment audit and data collection — on-site engineer visit, sensor and historical log analysis.
2. Multi-level model development (L1-L5) — threshold setting, ML model training, physics model creation.
3. Integration into infrastructure — SCADA connection, data stream setup, dashboard and alerts.
4. Documentation and staff training — operation instructions, response procedures.
5. Support and retraining — drift monitoring, baseline updates, adaptation for new failures.

Process and Timelines

Total time: from 3 weeks to 3 months depending on complexity.

Typical Implementation Mistakes

• Training on dirty data (maintenance labels not removed)
• Ignoring seasonality (heating, night shifts)
• Lack of baseline for normal mode (need at least 30 days)
• Retraining models without drift check

Example from practice: at a fertilizer plant, the system detected an anomaly in a compressor bearing 18 days before failure. Repair costs were negligible compared to emergency shutdown expenses (about 500 thousand rubles per hour of downtime).

Want to assess potential for your equipment? Order a free audit—our engineers will analyze readiness for implementation. Get a consultation on architecture selection and data volume. Contact us to discuss your project.