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

Розробники часто стикаються з ситуацією: модель кредитного скорингу раптово перестає адекватно оцінювати ризики, хоча код не змінювали. Причина — дрейф розподілу transaction_amount, який не помітили при ручному контролі. Коли таблиць більше 50, ручний моніторинг неефективний: ви пропустите аномалію, яка зламає ML-модель або звіт. Ми будуємо AI-системи, які ловлять такі аномалії за хвилини: від пропусків і NULL-спайків до зміни схеми та багатовимірних викидів. Стек — Python, scikit-learn, PyTorch, PostgreSQL, ClickHouse, Grafana. За час роботи впровадили рішення для 15+ компаній, заощаджуючи до 40% часу на quality checks. Зв'яжіться з нами — безкоштовно оцінимо ваш датасет і запропонуємо пілот.

Які аномалії ми виявляємо?

Система покриває всі основні типи аномалій, критичні для якості даних. Нижче — класифікація з прикладами з реальних проєктів.

Як працює автоматичне виявлення?

Data Quality Monitoring — наш базовий модуль. Він будує baseline за історичними даними та виконує регулярні перевірки. Ось спрощена реалізація на Python:

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

class DataQualityMonitor:
    def __init__(self, table_name: str, baseline_stats: dict):
        self.table_name = table_name
        self.baseline = baseline_stats

    def run_quality_checks(self, current_df: pd.DataFrame) -> dict:
        results = {'table': self.table_name, 'checks': [], 'issues': []}

        # 1. Перевірка обсягу
        row_count = len(current_df)
        baseline_rows = self.baseline.get('row_count_mean', row_count)
        baseline_rows_std = self.baseline.get('row_count_std', row_count * 0.1)

        volume_z = (row_count - baseline_rows) / (baseline_rows_std + 1e-9)
        if abs(volume_z) > 3:
            results['issues'].append({
                'check': 'volume',
                'severity': 'critical' if abs(volume_z) > 5 else 'warning',
                'current': row_count,
                'expected': int(baseline_rows),
                'z_score': round(volume_z, 2)
            })

        # 2. NULL ratio по колонках
        for col in current_df.columns:
            null_pct = current_df[col].isnull().mean() * 100
            baseline_null = self.baseline.get(f'{col}_null_pct', 0)

            if null_pct > baseline_null + 10:  # >10% зростання
                results['issues'].append({
                    'check': 'null_spike',
                    'column': col,
                    'severity': 'major' if null_pct > 50 else 'warning',
                    'current_null_pct': round(null_pct, 1),
                    'baseline_null_pct': round(baseline_null, 1)
                })

        # 3. Дрейф розподілу (KS-тест)
        for col in current_df.select_dtypes(include=[np.number]).columns:
            if f'{col}_sample' in self.baseline:
                stat, p_value = ks_2samp(
                    self.baseline[f'{col}_sample'],
                    current_df[col].dropna().values
                )
                if p_value < 0.001:
                    results['issues'].append({
                        'check': 'distribution_drift',
                        'column': col,
                        'severity': 'warning',
                        'ks_statistic': round(stat, 3),
                        'p_value': round(p_value, 5)
                    })

        results['passed'] = len(results['issues']) == 0
        return results

Baseline будується за останні 30 днів і оновлюється щотижня. У реальному проєкті ми додали читання схеми з DWH та алерти в Slack при severity 'major'.

Автоматичне профілювання та baseline

Модуль build_data_baseline збирає статистику за числовими та категоріальними колонками, зберігає семпли для KS-тесту. Це дозволяє швидко перераховувати baseline при додаванні нових джерел.

def build_data_baseline(historical_batches: list[pd.DataFrame]) -> dict:
    """
    Baseline = статистика за останні 30 днів (оновлюється щотижня).
    """
    row_counts = [len(df) for df in historical_batches]

    baseline = {
        'row_count_mean': np.mean(row_counts),
        'row_count_std': np.std(row_counts),
        'row_count_min': np.min(row_counts),
        'row_count_max': np.max(row_counts)
    }

    if historical_batches:
        sample_df = pd.concat(historical_batches[-7:])  # останній тиждень

        for col in sample_df.select_dtypes(include=[np.number]).columns:
            col_data = sample_df[col].dropna()
            baseline[f'{col}_mean'] = col_data.mean()
            baseline[f'{col}_std'] = col_data.std()
            baseline[f'{col}_p5'] = col_data.quantile(0.05)
            baseline[f'{col}_p95'] = col_data.quantile(0.95)
            baseline[f'{col}_null_pct'] = sample_df[col].isnull().mean() * 100
            # Зберігаємо 500 семплів для KS-тесту
            baseline[f'{col}_sample'] = col_data.sample(min(500, len(col_data))).values

        for col in sample_df.select_dtypes(include=['object', 'category']).columns:
            baseline[f'{col}_cardinality'] = sample_df[col].nunique()
            baseline[f'{col}_null_pct'] = sample_df[col].isnull().mean() * 100
            baseline[f'{col}_top_values'] = sample_df[col].value_counts().head(20).to_dict()

    return baseline

Чому варто використовувати Isolation Forest для багатовимірних аномалій?

Для пошуку аномальних записів цілком використовуємо Isolation Forest. Він ефективніший за One-Class SVM у 2-3 рази на датасетах від 1M рядків. Нижче — приклад для транзакційних даних:

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

def detect_row_level_anomalies(df: pd.DataFrame,
                                 contamination: float = 0.02) -> pd.DataFrame:
    """
    Виявлення аномальних записів (не тільки окремих значень).
    Корисно для: транзакційні дані, логи, CRM записи.
    """
    # Препроцесинг
    df_processed = df.copy()

    for col in df_processed.select_dtypes(include=['object']).columns:
        le = LabelEncoder()
        df_processed[col] = le.fit_transform(df_processed[col].astype(str))

    df_numeric = df_processed.select_dtypes(include=[np.number]).fillna(-999)

    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(df_numeric)

    model = IsolationForest(contamination=contamination, random_state=42)
    anomaly_labels = model.fit_predict(X_scaled)
    anomaly_scores = -model.score_samples(X_scaled)

    df['is_anomaly'] = anomaly_labels == -1
    df['anomaly_score'] = anomaly_scores

    # Пояснення: які ознаки найбільш аномальні
    df_anomalies = df[df['is_anomaly']].copy()
    return df_anomalies.sort_values('anomaly_score', ascending=False)

В одному проєкті алгоритм засік дрейф розподілу transaction_amount за 10 хвилин до того, як модель кредитного скорингу почала видавати помилки. Ми зупинили пайплайн, перенавчили модель та запобігли збитку в ~ $50K.

Schema Drift Detection

Зміна схеми джерела — часта причина мовчазних помилок. Модуль порівнює поточну схему з baseline та видає критичні попередження:

def detect_schema_drift(current_schema: dict, baseline_schema: dict) -> dict:
    """
    Порівнюємо схему поточних даних зі схемою з baseline.
    Критично для ETL пайплайнів: зміна upstream source ламає downstream.
    """
    issues = []

    # Зниклі стовпці
    missing_cols = set(baseline_schema.keys()) - set(current_schema.keys())
    for col in missing_cols:
        issues.append({
            'type': 'column_dropped',
            'column': col,
            'severity': 'critical',
            'action': 'check_upstream_source'
        })

    # Нові стовпці
    new_cols = set(current_schema.keys()) - set(baseline_schema.keys())
    for col in new_cols:
        issues.append({
            'type': 'column_added',
            'column': col,
            'severity': 'info',
            'action': 'review_and_update_documentation'
        })

    # Зміна типів
    for col in set(baseline_schema.keys()) & set(current_schema.keys()):
        if baseline_schema[col] != current_schema[col]:
            issues.append({
                'type': 'type_changed',
                'column': col,
                'from': baseline_schema[col],
                'to': current_schema[col],
                'severity': 'major',
                'action': 'validate_downstream_compatibility'
            })

    return {
        'schema_drift_detected': len(issues) > 0,
        'critical_issues': [i for i in issues if i['severity'] == 'critical'],
        'all_issues': issues
    }

Інтеграція з Great Expectations для декларативних тестів, dbt tests для трансформацій, Apache Atlas / Datahub для data lineage. Алерти в Slack, PagerDuty, email при severity >= 'major'. Дашборд у Grafana з history quality score по кожній таблиці.

Як ми будуємо процес?

1. Аналітика: Вивчаємо джерела даних, бізнес-контекст, типові failure patterns.
2. Проєктування baseline: Визначаємо метрики, пороги, частоту перевірок.
3. Реалізація модуля: Пишемо код детекції, інтегруємо з DWH/S3/API.
4. Тестування: Проганяємо на історичних даних, налаштовуємо точність (precision/recall).
5. Деплой: Розгортаємо на вашій інфраструктурі (Kubernetes, Airflow, bare metal).
6. Дашборди та алерти: Налаштовуємо Grafana, канали оповіщення, SLAs.

Строки та що входить у роботу

Ми гарантуємо SLA за швидкістю детекції: аномалії фіксуються не пізніше ніж через 5 хвилин після появи даних. Наші інженери мають сертифікати з ML та DWH. Замовте консультацію — оцінимо ваш проєкт за 2 дні.

Необхідність автоматизації моніторингу якості даних

Ручний контроль не масштабується: при 50+ таблицях ви пропустите аномалію, яка зламає ML-модель або звіт. Наша система детектує проблеми за хвилини, заощаджуючи до 40% часу Data Engineers та запобігаючи дорогим інцидентам. Отримайте розрахунок вартості впровадження — зв'яжіться з нами.

Виявлення аномалій: автоенкодери, 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-моніторингу. Отримайте консультацію — розкажемо, як вирішити вашу задачу.