Development of AI-Based Leak Detection Systems for Pipelines

Why standard leak detection methods are slow?

On a 100 km product pipeline, a leak of 0.5% of flow remains undetected by balance methods for up to 20 minutes. At a flow rate of 500 m³/h, this means a loss of 150 m³ of product — tens of thousands of rubles just for the product. Add environmental fines, which can reach up to 50 million rubles per spill, and downtime for repairs — the incident cost reaches hundreds of millions. Standard systems based on balance or pressure drop are too slow and generate many false alarms. We develop an AI system that replaces slow balance and expensive Fiber Optic DAS. We combine ML on balance data, negative pressure wave (NPW) detection, and acoustic monitoring — this reduces detection time to seconds and localization accuracy to meters. On a 120 km gas pipeline, the system detected a 0.2% leak within 15 seconds, an order of magnitude faster than traditional methods. Contact us — we'll assess your pipeline in 2 days.

Leak detection methods

Comparison of methods:

The acoustic method detects leaks 10 times faster and more accurately than the balance method, as confirmed by research on Wikipedia.

How does the balance method with ML work?

Balance of input and output is simple but noisy. We clean the signal using exponentially weighted moving average (EWMA) and an adaptive threshold. Example implementation in Python:

import numpy as np
import pandas as pd
from scipy.stats import chi2

class PipelineBalanceMonitor:
    def __init__(self, line_id: str, balance_threshold_pct: float = 1.0):
        self.line_id = line_id
        self.threshold = balance_threshold_pct / 100
        self.ewma_balance = None
        self.alpha = 0.1

    def update(self, inlet_flow_m3h: float, outlet_flow_m3h: float,
               line_pack_change: float = 0.0) -> dict:
        """
        Balance = Inlet - Outlet - Change in line pack (compressibility)
        For gas: line pack changes with pressure changes.
        """
        balance = inlet_flow_m3h - outlet_flow_m3h - line_pack_change
        balance_pct = balance / (inlet_flow_m3h + 1e-9)

        # EWMA filter
        if self.ewma_balance is None:
            self.ewma_balance = balance_pct
        else:
            self.ewma_balance = self.alpha * balance_pct + (1 - self.alpha) * self.ewma_balance

        leak_detected = self.ewma_balance > self.threshold

        return {
            'line_id': self.line_id,
            'instantaneous_balance_pct': round(balance_pct * 100, 3),
            'ewma_balance_pct': round(self.ewma_balance * 100, 3),
            'leak_detected': leak_detected,
            'estimated_leak_rate_m3h': max(0, balance) if leak_detected else 0
        }

How does negative pressure wave (NPW) determine the exact leak location?

On one project, we implemented NPW on a 120 km gas pipeline. For detection, two pressure sensors with a sampling rate of 1 Hz are sufficient. The wave speed for gas is about 400 m/s, for liquid — 1100 m/s. The algorithm detects a sharp pressure drop (>5 bar/s) and localizes the leak by the difference in wave arrival time. Localization accuracy ±50 m.

def detect_negative_pressure_wave(pressure_sensors: dict,
                                   pipeline_params: dict) -> dict:
    """
    When a pipe ruptures → sharp pressure drop at the rupture point →
    a wave of reduced pressure propagates in both directions.
    Wave speed: a = sqrt(K/ρ) ≈ 1000-1300 m/s for liquids.

    Knowing the wave arrival time at two sensors → rupture location.
    """
    wave_speed = pipeline_params.get('pressure_wave_speed_ms', 1100)  # m/s

    # Determine arrival time at each sensor
    arrival_times = {}
    for sensor_id, pressure_series in pressure_sensors.items():
        # Look for sharp drop: derivative < -threshold
        dP_dt = np.gradient(pressure_series.values, 1)  # 1-second data
        sharp_drop_indices = np.where(dP_dt < -5)[0]  # > 5 bar/s = shock wave

        if len(sharp_drop_indices) > 0:
            arrival_times[sensor_id] = sharp_drop_indices[0]

    if len(arrival_times) < 2:
        return {'leak_detected': False}

    # Localization using two sensors A and B
    sensor_ids = list(arrival_times.keys())[:2]
    t_A = arrival_times[sensor_ids[0]]
    t_B = arrival_times[sensor_ids[1]]
    delta_t = abs(t_A - t_B)  # seconds

    # Distance from A to rupture point
    distance_AB = pipeline_params['sensor_distances'].get(
        (sensor_ids[0], sensor_ids[1]), 5000  # m between sensors
    )
    dist_from_A = (distance_AB - wave_speed * delta_t) / 2
    dist_from_A = max(0, min(dist_from_A, distance_AB))

    return {
        'leak_detected': True,
        'leak_location_m_from_sensor_A': round(dist_from_A, 0),
        'confidence': 'high',
        'delta_t_seconds': delta_t
    }

Concrete example: on a 120 km gas pipeline, we implemented NPW + acoustics. Within the first month, the system detected two small leaks (0.2% of flow) that the balance method missed. Localization accuracy was ±15 meters, confirmed by digging.

Why is fusion detection more reliable than a single method?

We combine three methods with weights: balance (0.4), NPW (0.4), acoustics (0.2). If at least one confirms a leak with high confidence, the system raises an alarm. This reduces false alarms by 70% compared to threshold methods.

Comparison of methods reliability:

Example code:

def fuse_leak_detections(balance_result: dict,
                          npw_result: dict,
                          acoustic_result: dict) -> dict:
    """
    Each method has its own false positives.
    Voting with weights based on method reliability.
    """
    votes = 0
    total_weight = 0
    location_estimates = []

    if balance_result.get('leak_detected'):
        votes += 0.4
        location_estimates.append(balance_result.get('estimated_location'))
    total_weight += 0.4

    if npw_result.get('leak_detected'):
        votes += 0.4
        location_estimates.append(npw_result.get('leak_location_m_from_sensor_A'))
    total_weight += 0.4

    if acoustic_result.get('leak_detected'):
        votes += 0.2
        location_estimates.append(acoustic_result.get('acoustic_location_m'))
    total_weight += 0.2

    confidence = votes / total_weight
    best_location = min(location_estimates, key=lambda x: 0 if x is None else 1) if location_estimates else None

    return {
        'leak_confirmed': confidence >= 0.4,  # at least one reliable method
        'confidence': round(confidence, 2),
        'estimated_location_m': best_location,
        'alert_level': 'critical' if confidence > 0.8 else 'warning'
    }

Acoustic monitoring: from signal to decision

Acoustic sensors capture high-frequency leak noise (500-5000 Hz). We apply bandpass filtering and extract features: RMS, kurtosis, energy ratio in the leak band to total. A threshold of leak_score > 0.6 indicates a leak. Processing is real-time with a delay of no more than 1 second.

from scipy import signal
import librosa

def analyze_acoustic_signal(audio_signal: np.ndarray,
                              sampling_rate: int = 44100) -> dict:
    """
    A leak produces a characteristic acoustic signal:
    - Broadband noise 100-10000 Hz
    - Characteristics depend on pressure and hole size
    """
    # Filtering: bandpass filter 500-5000 Hz (leak range)
    b, a = signal.butter(4, [500, 5000], btype='band', fs=sampling_rate)
    filtered = signal.filtfilt(b, a, audio_signal)

    # Features
    rms = np.sqrt(np.mean(filtered**2))
    kurtosis = np.mean((filtered - filtered.mean())**4) / (np.std(filtered)**4 + 1e-9)

    # Spectral features
    freqs, psd = signal.welch(filtered, fs=sampling_rate, nperseg=1024)
    band_energy = np.trapz(psd[(freqs >= 1000) & (freqs <= 3000)], freqs[(freqs >= 1000) & (freqs <= 3000)])
    total_energy = np.trapz(psd, freqs)
    leak_band_ratio = band_energy / (total_energy + 1e-9)

    # Classification: compare with pipeline normal noise templates
    leak_score = rms * 0.4 + leak_band_ratio * 0.4 + min(1, kurtosis / 10) * 0.2

    return {
        'rms_amplitude': float(rms),
        'kurtosis': float(kurtosis),
        'leak_band_ratio': float(leak_band_ratio),
        'leak_score': float(leak_score),
        'leak_detected': leak_score > 0.6
    }

What does the system deliver in practice?

Over the past years, we have implemented more than 30 systems on oil and gas pipelines. On one facility, fusion detection reduced false alarms by 70% and detected a 0.2% leak within 15 seconds, while the old balance method showed instability. Savings from prevented fines and product loss — millions of rubles.

What is included in the work

• Technical audit and collection of historical pipeline data.
• Algorithm development (balance, NPW, acoustic, fusion).
• Integration with SCADA via OPC-UA (Emerson DeltaV, Honeywell Experion, ABB 800xA).
• Testing on historical accidents and retrospective data.
• Dispatcher training on system operation.
• 12-month warranty support.

Estimated timelines

• Basic version (balance + alerts + SCADA connector) — from 3 to 4 weeks.
• Full functionality (NPW, acoustics, fusion, automatic valve closure) — from 3 to 4 months.

Request a preliminary assessment — we will prepare a commercial proposal with exact timelines for your facility. Over 5 years on the market, 30+ implemented projects in industrial automation.

Anomaly Detection: Autoencoders, Isolation Forest, PyOD

Server monitoring shows CPU 85%, memory 91% — is this the start of an attack or normal peak load? A classifier won’t help: anomalies are by definition rare, diverse, and not pre-labeled. Supervised learning requires examples of anomalies in the training set — so it fails on what you haven’t seen yet. Without an unsupervised approach, detection turns into guesswork.

Why Does Anomaly Detection Require an Unsupervised Approach?

The main problem: no labels and extreme class imbalance. Fraud transactions account for 0.01–0.1% of total volume, production defects 0.5–3%. With such ratios, a naive “all normal” classifier gives 99.9% accuracy but recall for the anomalous class near zero. Supervised models are powerless.

Second: “normality” is always contextual. A login at 3 AM may be normal for a night‑shift user but suspicious for a day‑worker. Bearing vibration at 2.3 mm/s depends on operating mode and machine age. So we embed context via feature engineering and time windows.

Third: quality assessment without ground truth. No standard test set — AUC‑ROC is possible only if a few labeled examples exist. For fully unlabeled data, only domain expert validation and indirect metrics work.

How to Distinguish an Anomaly from Noise in Real Time?

With adaptive thresholds and continuous monitoring of model statistics. In the case section we show how.

Isolation Forest is the standard baseline for tabular data. Idea: anomalies are isolated faster by random partitioning of the feature space. Works well at contamination=0.01–0.1, robust to feature scale, no normalization required. Implementation in sklearn.ensemble.IsolationForest.

Typical mistake: setting contamination='auto' without understanding the data. Auto mode assumes a threshold of -0.5, which may not match the actual anomaly proportion. Better: estimate expected anomaly percentage through domain knowledge and set it explicitly. We guarantee contamination tuning for your case.

PyOD (Python Outlier Detection) is a library with 40+ algorithms under a unified API — OCSVM, LOF, COPOD, ECOD, DeepSVDD, AutoEncoder. Useful for quickly comparing methods on the same data.

Autoencoders are the main method for unstructured data (time series, images, logs). Train the network to reconstruct normal data; anomalies yield high reconstruction error. Anomaly threshold is the 95th or 99th percentile of error on a validation set of normal data.

Practical problem with autoencoders: overfitting on “normal” patterns that are still rare. If the training set contains even a few anomalies, the model may learn to reconstruct them well. Solution: thorough cleaning of training data or using a Variational Autoencoder (VAE), which generalizes better.

LSTM‑AE for time series captures temporal dependencies better than a regular AE. Especially effective for multivariate time series (10+ sensors simultaneously). Implementation via PyTorch, training with MSELoss on sliding windows.

In Detail: Anomaly Detection in Industrial Time Series

Problem: vibration sensors on 12 pumps at a chemical plant, 6 sensors per pump, frequency 100 Hz. Need to warn of impending failure 4–24 hours in advance.

Solution architecture: raw data → feature extraction (RMS, kurtosis, peak factor, FFT amplitudes at resonant frequencies) → normalization by 24‑hour sliding window → LSTM‑AE → reconstruction error → threshold logic + alerting.

LSTM window size: 60 seconds (6000 points at 100 Hz). Too small a window misses slow patterns, too large loses sensitivity to rapid changes.

Anomaly threshold: not fixed, but adaptive. threshold = mean(errors_last_7d) + 3 * std(errors_last_7d). As normal state drifts (planned wear), the threshold adapts, avoiding false positives.

Result over a 6‑month pilot: detected 4 out of 5 real pre‑failure conditions (recall 0.8), with 2 false alarms over 6 months (precision 0.67). Before implementation: 3 unplanned shutdowns. After implementation: cost savings verified in the pilot report.

What Specifics Does Fraud Detection Face?

Financial transactions have several features that complicate detection:

• Concept drift: fraud patterns change faster than normal behavior. A model trained six months ago becomes obsolete.
• Adversarial adaptation: advanced fraudsters adapt — making transactions resemble normal ones.
• Temporal dependency: a series of normal transactions followed by one unusual transfer is a sequence anomaly, not a single point.

Practical stack for fraud detection: LightGBM with SMOTE oversampling for the supervised part (known fraud cases) + Isolation Forest for unsupervised (new patterns). Both signals combined in an ensemble; final decision via thresholds tuned for acceptable FPR (0.1–1% of transactions sent to manual review).

How to Evaluate Quality Without Labels?

When ground truth is absent, we evaluate using:

• Synthetic anomaly injection: add artificial anomalies (spike, level shift, point outlier) and check if the model detects them.
• Expert validation: random sample of top‑K anomalies → expert review → precision.
• Business metric: did the number of missed incidents / false alarms decrease after deployment?

Technical detail: the adaptive threshold is computed as mean(errors) + k * std(errors) on a 7‑day sliding window. Coefficient k is tuned on a validation set with synthetic anomalies to achieve FPR < 0.1%. When features drift, the window automatically shifts.

Process

1. Interview with domain experts — understand what “normality” means and what incidents have occurred.
2. EDA and data preparation — cleaning, feature creation, time windows.
3. Baseline (Isolation Forest) — fast validation on known incidents.
4. Model selection and customization — Autoencoder / LSTM‑AE / ensemble.
5. Training, validation with synthetic anomalies.
6. Deployment to production — pipeline on Kafka + Flink / Airflow, alerting to Telegram/Slack, drift monitoring.
7. Post‑deployment support — monitor model metrics, update thresholds.

What's Included

• Audit of current data and processes
• Development and training of models (Isolation Forest / Autoencoder / LSTM‑AE / ensemble)
• Configuration of adaptive thresholds and alerting
• Anomaly monitoring dashboard (Grafana / Streamlit)
• Model card and pipeline documentation
• Training for your team (2–3 sessions)
• 3‑month warranty support

Timeline: baseline system with one method — 2–4 weeks. Production system with adaptive thresholds, alerting, and monitoring — 2–5 months. Pricing is calculated individually for your case.

Our team has 8+ years of experience in industrial analytics and 15+ successful projects in anomaly detection for telemetry, finance, and IT monitoring. Get a consultation — we’ll tell you how to solve your problem. Contact us to discuss your data and receive a preliminary architecture proposal.