Виявлення аномалій

Виявлення аномалій (anomaly detection) ідентифікує рідкісні, незвичайні патерни, що відрізняються від очікуваної поведінки. На відміну від класифікації, ви часто маєте мало або жодних розмічених аномалій - система має вивчити "нормальне" та флагувати відхилення. Це забезпечує fraud detection, моніторинг інфраструктури, контроль якості та системи безпеки.

Де це використовується

Виявлення аномалій захищає системи та ідентифікує можливості:

• Фінанси: Fraud detection, підозрілі транзакції
• Інфраструктура: Моніторинг серверів, network intrusion
• Виробництво: Прогнозування поломок обладнання, контроль якості
• Медицина: Моніторинг пацієнтів, незвичайні vital signs
• Кібербезпека: Виявлення загроз, поведінкові аномалії
• IoT: Несправність сенсорів, незвичайні патерни

Продукти: Datadog, Splunk, AWS GuardDuty, системи якості на виробництві.

Типові дані

Типи даних: - Time series (метрики, readings сенсорів) - Табличні (транзакції, events) - Зображення (візуальна інспекція) - Текст (log files, повідомлення) - Network дані (трафік, з'єднання)

Що ви робите на практиці

1. Визначте "аномалію" - Що робить щось ненормальним?
2. Зрозумійте нормальне - Як виглядає типова поведінка?
3. Оберіть підхід - Supervised, semi-supervised, або unsupervised
4. Інженерія фіч - Захопити патерни що відрізняють аномалії
5. Оберіть алгоритм - Статистичний, tree-based, або neural
6. Встановіть пороги - Баланс false positives та false negatives
7. Оцініть ретельно - Часто обмежені labeled дані
8. Деплойте з моніторингом - Відстежуйте alert rates та feedback
9. Обробіть feedback - Інкорпоруйте людські мітки
10. Ітеруйте постійно - Нормальна поведінка еволюціонує

Спочатку бейзлайн

Почніть з простих статистичних методів:

import numpy as np
from scipy import stats

# Z-score бейзлайн
def zscore_anomaly(data, threshold=3):
    """Флагувати точки далі ніж threshold std від mean."""
    z_scores = np.abs(stats.zscore(data))
    return z_scores > threshold

# IQR метод
def iqr_anomaly(data, multiplier=1.5):
    """Флагувати outliers використовуючи interquartile range."""
    q1 = np.percentile(data, 25)
    q3 = np.percentile(data, 75)
    iqr = q3 - q1
    lower = q1 - multiplier * iqr
    upper = q3 + multiplier * iqr
    return (data < lower) | (data > upper)

# Moving average для time series
def moving_avg_anomaly(data, window=24, threshold=3):
    """Флагувати відхилення від moving average."""
    rolling_mean = data.rolling(window=window).mean()
    rolling_std = data.rolling(window=window).std()
    z_score = (data - rolling_mean) / rolling_std
    return np.abs(z_score) > threshold

Чому статистичний бейзлайн: - Простий та інтерпретований - Не потребує тренування - Добре працює для univariate даних - Встановлює очікування anomaly rate

Підходи алгоритмів

Isolation Forest:

from sklearn.ensemble import IsolationForest

# Тренування на нормальних даних (або всіх якщо unsupervised)
model = IsolationForest(
    n_estimators=100,
    contamination=0.01,  # Очікуваний anomaly rate
    random_state=42
)
model.fit(X_train)

# Predict: -1 для аномалії, 1 для нормального
predictions = model.predict(X_test)
anomaly_scores = model.decision_function(X_test)  # Нижче = більш аномальне

Autoencoder для виявлення аномалій:

import torch
import torch.nn as nn

class Autoencoder(nn.Module):
    def __init__(self, input_dim, hidden_dim=32):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, 64),
            nn.ReLU(),
            nn.Linear(64, hidden_dim)
        )
        self.decoder = nn.Sequential(
            nn.Linear(hidden_dim, 64),
            nn.ReLU(),
            nn.Linear(64, input_dim)
        )

    def forward(self, x):
        encoded = self.encoder(x)
        decoded = self.decoder(encoded)
        return decoded

# Тренування на нормальних даних
model = Autoencoder(input_dim=features)
criterion = nn.MSELoss()

for epoch in range(epochs):
    for batch in normal_data_loader:
        reconstructed = model(batch)
        loss = criterion(reconstructed, batch)
        loss.backward()
        optimizer.step()

# Anomaly score = reconstruction error
def get_anomaly_score(model, x):
    with torch.no_grad():
        reconstructed = model(x)
        error = ((x - reconstructed) ** 2).mean(dim=1)
    return error

Виявлення аномалій у Time Series

Time series потребує temporal context:

import pandas as pd
from prophet import Prophet

# Prophet для time series decomposition
def prophet_anomaly(df, interval_width=0.99):
    """Використовувати Prophet для виявлення аномалій."""
    model = Prophet(interval_width=interval_width)
    model.fit(df[['ds', 'y']])

    forecast = model.predict(df[['ds']])
    df['yhat'] = forecast['yhat']
    df['yhat_lower'] = forecast['yhat_lower']
    df['yhat_upper'] = forecast['yhat_upper']

    df['anomaly'] = (df['y'] < df['yhat_lower']) | (df['y'] > df['yhat_upper'])
    return df

# Sliding window фічі
def create_window_features(series, window_sizes=[6, 12, 24]):
    """Створити фічі з sliding windows."""
    features = pd.DataFrame()

    for w in window_sizes:
        features[f'mean_{w}'] = series.rolling(w).mean()
        features[f'std_{w}'] = series.rolling(w).std()
        features[f'min_{w}'] = series.rolling(w).min()
        features[f'max_{w}'] = series.rolling(w).max()
        features[f'diff_{w}'] = series - series.shift(w)

    return features.dropna()

Вибір порогу

Вибір правильного порогу балансує precision та recall:

def optimize_threshold(scores, labels, optimize_for='f1'):
    """Знайти поріг що оптимізує задану метрику."""
    thresholds = np.percentile(scores, np.arange(90, 100, 0.5))
    best_score = 0
    best_threshold = thresholds[0]

    for thresh in thresholds:
        predictions = scores > thresh

        if optimize_for == 'f1':
            score = f1_score(labels, predictions)
        elif optimize_for == 'precision':
            score = precision_score(labels, predictions)
        elif optimize_for == 'recall':
            score = recall_score(labels, predictions)

        if score > best_score:
            best_score = score
            best_threshold = thresh

    return best_threshold, best_score

# Business-driven поріг
def cost_based_threshold(scores, labels, cost_fp=1, cost_fn=10):
    """Знайти поріг що мінімізує бізнес-витрати."""
    thresholds = np.percentile(scores, np.arange(80, 100, 0.5))
    costs = []

    for thresh in thresholds:
        predictions = scores > thresh
        fp = ((predictions == 1) & (labels == 0)).sum()
        fn = ((predictions == 0) & (labels == 1)).sum()
        total_cost = fp * cost_fp + fn * cost_fn
        costs.append((thresh, total_cost))

    return min(costs, key=lambda x: x[1])

Оцінка та метрики

Оцінка без міток: - Domain expert review флагнутих аномалій - Порівняння з відомими інцидентами - A/B testing (чи корелювали alerts з проблемами?) - Manual labeling семплу

Режими відмов та чеклист налагодження

• [ ] Concept drift: Нормальна поведінка змінюється з часом
• [ ] Seasonal patterns: Регулярні варіації флагуються як аномалії
• [ ] Feature leakage: Anomaly-related фічі в тренуванні
• [ ] Contaminated training: Аномалії в "нормальних" тренувальних даних
• [ ] Threshold sensitivity: Малі зміни викликають великі alert swings
• [ ] Alert fatigue: Занадто багато false positives
• [ ] Missing context: Аномалії без бізнес-контексту

Чеклист для продакшну

Перед деплоєм: - [ ] Базовий anomaly rate встановлено - [ ] Поріг налаштовано на validation даних з мітками - [ ] Alert volume прийнятний (не занадто багато false positives) - [ ] Escalation path визначено - [ ] Feedback mechanism для збору міток

Після деплою: - [ ] Alert rate моніториться - [ ] Precision відстежується через feedback - [ ] Модель ретренується на нових labeled даних - [ ] Seasonal patterns враховані - [ ] Drift detection на input features

Операціоналізація alerts: - [ ] Severity levels визначені (low/medium/high) - [ ] Routing rules до відповідних команд - [ ] Playbooks для поширених типів аномалій - [ ] Feedback loop для покращення моделі

Обробка Drift

Нормальна поведінка змінюється - моделі мають адаптуватись:

def detect_drift(current_data, reference_data, threshold=0.05):
    """Виявити distribution drift використовуючи KS test."""
    from scipy.stats import ks_2samp

    drift_detected = {}
    for column in current_data.columns:
        stat, p_value = ks_2samp(
            reference_data[column], 
            current_data[column]
        )
        drift_detected[column] = p_value < threshold

    return drift_detected

# Адаптивний поріг
class AdaptiveThreshold:
    def __init__(self, window_size=1000, percentile=99):
        self.window_size = window_size
        self.percentile = percentile
        self.scores = []

    def update(self, score):
        self.scores.append(score)
        if len(self.scores) > self.window_size:
            self.scores.pop(0)

    def get_threshold(self):
        return np.percentile(self.scores, self.percentile)

Інструменти та стек

Приклад міні-кейсу

Сценарій: E-commerce платформа потребує виявлення шахрайських транзакцій у real-time.

Підхід: 1. Фічі: Сума транзакції, частота, локація, device, time patterns 2. Бейзлайн: Z-score на суму по користувачу - Ловить 40% fraud 3. Модель: Isolation Forest натренований на нормальних транзакціях 4. Поріг: Налаштований для 1% false positive rate 5. Real-time: Score обчислюється на транзакцію, alert якщо вище порогу

Результати: - Fraud detection rate: 78% - False positive rate: 0.8% - Alert volume: 500/день (manageable для review team) - Latency: 15ms на транзакцію

Ключові уроки: - User-level фічі критичні (відхилення від персонального baseline) - Time-of-day patterns важливі для threshold adjustment - Feedback loop від fraud team покращив precision на 20%

Пов'язані сторінки

• Індекс: ML Field Guide
• Fraud Detection
• Моніторинг моделей
• Прогнозування часових рядів

Додаткове читання

• PyOD: Python Outlier Detection
• Isolation Forest Paper
• Anomaly Detection Survey
• Time Series Anomaly Detection with MERLION

Спробуйте наживо

Дослідіть пов'язану модель на практиці: відкрийте AI Anomaly Detection Lab — A Manufacturing Sensor Case Study і змінюйте параметри під час роботи. Усе лишається у вашому браузері.