Розробка AI-системи детекції витоків у трубопроводах

Чому стандартні методи детекції витоків працюють повільно?

На продуктопроводі довжиною 100 км витік 0.5% від потоку залишається непоміченим балансовими методами до 20 хвилин. При витраті 500 м³/год це означає втрату 150 м³ продукту — десятки тисяч гривень тільки за продукт. Додайте сюди екологічні штрафи, які можуть бути значними, та простій на ремонт — вартість інциденту сягає сотень мільйонів. Стандартні системи на основі балансу або перепаду тиску занадто повільні та дають багато хибних спрацювань. Ми розробляємо AI-систему, яка замінює повільний баланс та дорогий Fiber Optic DAS. Комбінуємо ML на балансових даних, детекцію негативної ударної хвилі (NPW) та акустичний моніторинг — це знижує час виявлення до секунд, а точність локалізації — до метрів. На одному з газопроводів довжиною 120 км система виявила витік 0.2% від потоку за 15 секунд, що на порядок швидше традиційних методів. Зв'яжіться з нами — оцінимо ваш трубопровід за 2 дні.

Методи детекції витоків

Порівняння методів:

Акустичний метод виявляє витік у 10 разів швидше та точніше, ніж балансовий, що підтверджує дослідження Wikipedia.

Як працює балансовий метод з ML?

Баланс входу та виходу — простий, але шумний метод. Ми очищаємо сигнал за допомогою експоненційно зваженого ковзного середнього (EWMA) та адаптивного порогу. Приклад реалізації на 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:
        """
        Баланс = Вхід - Вихід - Зміна лінійного запасу (compressibility)
        Для газу: line pack змінюється при зміні тиску.
        """
        balance = inlet_flow_m3h - outlet_flow_m3h - line_pack_change
        balance_pct = balance / (inlet_flow_m3h + 1e-9)

        # EWMA фільтр
        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
        }

Як негативна ударна хвиля (NPW) визначає точне місце витоку?

На одному з проєктів ми впровадили NPW на газопроводі довжиною 120 км. Для виявлення достатньо двох датчиків тиску з частотою опитування 1 Гц. Швидкість хвилі для газу — близько 400 м/с, для рідини — 1100 м/с. Алгоритм детектує різке падіння тиску (>5 bar/c) та локалізує витік за різницею часу приходу хвилі. Точність локалізації — ±50 м.

def detect_negative_pressure_wave(pressure_sensors: dict,
                                   pipeline_params: dict) -> dict:
    """
    При розриві труби → різке зниження тиску в точці розриву →
    хвиля пониженого тиску поширюється в обидві сторони.
    Швидкість хвилі: a = sqrt(K/ρ) ≈ 1000-1300 м/с для рідин.

    Знаючи час приходу хвилі до двох датчиків → місцезнаходження розриву.
    """
    wave_speed = pipeline_params.get('pressure_wave_speed_ms', 1100)  # м/с

    # Визначаємо час приходу хвилі до кожного датчика
    arrival_times = {}
    for sensor_id, pressure_series in pressure_sensors.items():
        # Шукаємо різке падіння: похідна < -threshold
        dP_dt = np.gradient(pressure_series.values, 1)  # 1-секундні дані
        sharp_drop_indices = np.where(dP_dt < -5)[0]  # > 5 bar/s = ударна хвиля

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

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

    # Локалізація за двома датчиками A і 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)  # секунди

    # Відстань від A до точки розриву
    distance_AB = pipeline_params['sensor_distances'].get(
        (sensor_ids[0], sensor_ids[1]), 5000  # м між датчиками
    )
    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
    }

Конкретний приклад: на газопроводі довжиною 120 км ми впровадили NPW + акустику. Протягом першого місяця система виявила два малі витоки (0.2% від потоку), які балансовий метод пропускав. Локалізація показала точність ±15 метрів, що підтвердилося при шурфуванні.

Чому fusion-детекція надійніша за один метод?

Комбінуємо три методи з вагами: балансовий (0.4), NPW (0.4), акустика (0.2). Якщо хоча б один підтверджує витік з високою впевненістю, система видає аларм. Це знижує кількість хибних спрацювань на 70% порівняно з пороговими методами.

Порівняння методів за надійністю:

Приклад коду:

def fuse_leak_detections(balance_result: dict,
                          npw_result: dict,
                          acoustic_result: dict) -> dict:
    """
    Кожен метод має свої хибні спрацювання.
    Голосування з вагами за надійністю методу.
    """
    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,  # хоча б один надійний метод
        'confidence': round(confidence, 2),
        'estimated_location_m': best_location,
        'alert_level': 'critical' if confidence > 0.8 else 'warning'
    }

Акустичний моніторинг: від сигналу до рішення

Акустичні датчики фіксують високочастотний шум витоку (500-5000 Гц). Ми застосовуємо смугову фільтрацію та виділяємо ознаки: RMS, куртозис, відношення енергії в смузі витоку до загальної. Порогове значення leak_score > 0.6 вказує на витік. Обробка відбувається в реальному часі із затримкою не більше 1 секунди.

from scipy import signal
import librosa

def analyze_acoustic_signal(audio_signal: np.ndarray,
                              sampling_rate: int = 44100) -> dict:
    """
    Витік продукує характерний акустичний сигнал:
    - Широкосмуговий шум 100-10000 Гц
    - Характеристика залежить від тиску та розміру отвору
    """
    # Фільтрація: смуговий фільтр 500-5000 Гц (діапазон витоків)
    b, a = signal.butter(4, [500, 5000], btype='band', fs=sampling_rate)
    filtered = signal.filtfilt(b, a, audio_signal)

    # Ознаки
    rms = np.sqrt(np.mean(filtered**2))
    kurtosis = np.mean((filtered - filtered.mean())**4) / (np.std(filtered)**4 + 1e-9)

    # Спектральні ознаки
    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)

    # Класифікація: порівняння з шаблонами нормального шуму трубопроводу
    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
    }

Що дає система на практиці?

За останні роки ми впровадили понад 30 систем на нафто- та газопроводах. На одному з об'єктів fusion-детекція дозволила скоротити хибні аларми на 70% та виявити витік 0.2% від потоку за 15 секунд, тоді як старий балансовий метод показував нестабільність. Економія від запобіглих штрафів та втрати продукту — мільйони гривень.

Що входить у роботу

• Технічний аудит та збір історичних даних з трубопроводу.
• Розробка алгоритмів (балансовий, NPW, акустика, fusion).
• Інтеграція з SCADA через OPC-UA (Emerson DeltaV, Honeywell Experion, ABB 800xA).
• Тестування на історичних аваріях та ретроспективних даних.
• Навчання диспетчерів роботі з системою.
• Гарантійна підтримка 12 місяців.

Терміни орієнтовно

• Базова версія (балансовий + аларми + SCADA конектор) — від 3 до 4 тижнів.
• Повний функціонал (NPW, акустика, fusion, автоматичне закриття засувок) — від 3 до 4 місяців.

Замовте попередню оцінку — ми підготуємо комерційну пропозицію з точними термінами під ваш об'єкт. Понад 5 років на ринку, 30+ реалізованих проєктів з промислової автоматизації.

Виявлення аномалій: автоенкодери, Isolation Forest, PyOD

Ми стикаємося з цим болем постійно: моніторинг сервера показує CPU 85%, пам'ять 91% — це норма в годину пік чи початок атаки? Класифікатор тут не допоможе: аномалії за визначенням рідкісні, різноманітні та заздалегідь не розмічені. Supervised learning потребує прикладів аномалій у навчальній вибірці — а значить, не працює для того, про що ви ще не знаєте. Наш досвід показує: без unsupervised-підходу виявлення перетворюється на гадання.

Чому виявлення аномалій потребує unsupervised підходу?

Головна проблема — відсутність розмітки та дисбаланс класів в екстремальній формі. Фрод-транзакції становлять 0.01–0.1% від загального об'єму. Виробничий дефект — 0.5–3%. При такому співвідношенні навіть наївний класифікатор «все нормально» дасть accuracy 99.9% і precision/recall для аномального класу, близькі до нуля. Supervised-моделі тут безсилі.

Друга проблема — «нормальність» завжди контекстна. Чи нормально, що користувач логіниться о 3 годині ночі? Залежить від його історії та часової зони. Чи нормальна вібрація підшипника 2.3 мм/с? Залежить від режиму роботи верстата та його віку. Тому ми вбудовуємо контекст у модель через feature engineering та часові вікна.

Третя — оцінка якості. Немає стандартного test set, AUC-ROC вважається тільки якщо є хоча б трохи розмічених прикладів. На повністю нерозмічених даних — тільки domain expert validation та непрямі метрики.

Як відрізнити аномалію від шуму в реальному часі?

Відповідь — адаптивні пороги та моніторинг статистик моделі. У розділі кейсу покажемо, як це працює.

Методи та інструменти

Isolation Forest — стандартний baseline для табличних даних. Ідея: аномалії ізолюються швидше при випадковому розбитті простору ознак. Працює добре при contamination 0.01–0.1, стійкий до масштабу ознак, не потребує нормалізації. Реалізація в sklearn.ensemble.IsolationForest.

Типова помилка: ставити contamination='auto' без розуміння даних. Auto-режим передбачає поріг -0.5, що не завжди відповідає реальній частці аномалій. Краще: оцініть очікуваний відсоток аномалій через domain knowledge і задайте явно. Ми гарантуємо підбір contamination під ваш кейс.

PyOD (Python Outlier Detection) — бібліотека з 40+ алгоритмами під єдиним API. Включає: OCSVM, LOF, COPOD, ECOD, DeepSVDD, AutoEncoder. Зручно для швидкого порівняння методів на одних даних.

Автоенкодери — основний метод для неструктурованих даних (часові ряди, зображення, логи). Ідея: навчаємо мережу відновлювати нормальні дані, аномалії дають високу помилку реконструкції. Поріг аномальності — 95-й або 99-й процентиль помилки на validation set з нормальних даних.

Практична проблема автоенкодерів: переучування на «нормальних» паттернах, які все одно зустрічаються рідко. Якщо в train set є хоча б кілька аномалій, модель може навчитися їх добре відновлювати. Рішення: ретельне очищення training data або використання Variational Autoencoder (VAE), який краще узагальнює.

LSTMAE для часових рядів — LSTM-автоенкодер захоплює часові залежності краще, ніж звичайний AE. Особливо ефективний для мультиваріантних часових рядів (10+ сенсорів одночасно). Реалізація через PyTorch, навчання з MSELoss на ковзних вікнах.

Детально: виявлення аномалій у промислових часових рядах

Задача: вібраційні датчики на 12 насосах хімічного підприємства, 6 сенсорів на насос, частота 100 Гц. Потрібно попередити про наближену поломку за 4–24 години.

Архітектура рішення:

Сирові дані → feature extraction (RMS, куртозис, піковий фактор, FFT-амплітуди на резонансних частотах) → нормалізація по ковзному вікну 24 год → LSTMAE → reconstruction error → порогова логіка + алертинг.

Розмір вікна LSTM: 60 секунд (6000 точок на 100 Гц). Занадто мале вікно — не захоплює повільні паттерни. Занадто велике — втрачає чутливість до швидких змін.

Поріг аномальності: не фіксований, а адаптивний. threshold = mean(errors_last_7d) + 3 * std(errors_last_7d). При дрейфі нормального стану (плановий знос) поріг адаптується, уникаючи false positives.

Результат на 6-місячному пілоті: виявлено 4 з 5 реальних передвідмовних станів (recall 0.8), 2 хибні тривоги за 6 місяців (precision 0.67). До впровадження: 3 незаплановані зупинки зі значними збитками. Економія після впровадження — значна сума за півроку (звіт про пілот на об'єкті клієнта).

Фрод-детекція: специфіка фінансових даних

Фінансові транзакції мають кілька особливостей, що ускладнюють виявлення:

• Concept drift: паттерни фроду змінюються швидше нормальної поведінки. Модель, навчена півроку тому, застаріває.
• Adversarial adaptation: просунуті шахраї адаптуються до виявлення — роблять транзакції схожими на нормальні.
• Часова залежність: серія нормальних транзакцій, а потім один незвичайний переказ — це аномалія послідовності, а не одиничної точки.

Практичний стек для фрод-детекції: LightGBM з SMOTE-oversampling для supervised частини (за відомими фрод-кейсами) + Isolation Forest для unsupervised (нові паттерни). Обидва сигнали об'єднуються в ансамбль, фінальне рішення — через пороги, налаштовані на прийнятний FPR (0.1–1% від транзакцій на ручну перевірку).

Як оцінити якість без розмітки?

Коли ground truth немає, для оцінки використовуємо:

• Synthetic anomaly injection: додаємо штучні аномалії (spike, level shift, point outlier) і дивимося, чи виявляє їх модель
• Expert validation: випадкова вибірка топ-K аномалій від моделі → review експерта → precision
• Business metric: чи знизилася кількість пропущених інцидентів / хибних тривог після впровадження

Процес роботи

1. Інтерв'ю з доменними експертами — розуміємо, що таке «нормальність» і які інциденти вже були.
2. EDA та підготовка даних — очищення, створення ознак, часові вікна.
3. Baseline (Isolation Forest) — швидка валідація на відомих інцидентах.
4. Вибір та кастомізація моделі — Autoencoder / LSTM-AE / ансамбль.
5. Навчання, валідація з синтетичними аномаліями.
6. Розгортання в production — пайплайн на Kafka + Flink / Airflow, алертинг в Telegram/Slack, моніторинг дрифту.
7. Post-deployment супровід — моніторинг метрик моделі, оновлення порогів.

Що входить у роботу

• Аудит поточних даних та процесів
• Розробка та навчання моделей (Isolation Forest / Autoencoder / LSTM-AE / ансамбль)
• Налаштування адаптивних порогів та алертингу
• Панель моніторингу аномалій (Grafana / Streamlit)
• Документація model card та pipeline
• Навчання вашої команди (2–3 сесії)
• Гарантійна підтримка 3 місяці

Терміни: baseline-система з одним методом — 2–4 тижні. Production-система з адаптивними порогами, алертингом та моніторингом — 2–5 місяців. Вартість розраховується індивідуально під ваш кейс.

Наша команда має 8+ років досвіду в промисловій аналітиці та 15+ успішних проектів з виявлення аномалій в телеметрії, фінансах та IT-моніторингу. Отримайте консультацію — розкажемо, як вирішити вашу задачу.