Виявлення аномалій у часових рядах

Виявлення аномалій у часових рядах ідентифікує незвичні точки або патерни в послідовних даних. Це критично для моніторингу систем, виявлення шахрайства, predictive maintenance та quality control. На відміну від статичного виявлення аномалій, потрібно враховувати часову динаміку - тренди, сезонність та автокореляції.

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

Виявлення аномалій застосовується скрізь де є часові дані:

• IT Operations: Моніторинг серверів, мережевий трафік
• Фінанси: Виявлення шахрайства, market anomalies
• Manufacturing: Quality control, predictive maintenance
• IoT: Sensor monitoring, equipment failures
• Ecommerce: Traffic spikes, conversion anomalies
• Cybersecurity: Intrusion detection, unusual patterns
• Healthcare: Patient vitals, equipment monitoring

Продукти: Datadog, Splunk, AWS CloudWatch, Azure Monitor.

Типи аномалій

Типові дані

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

1. Зрозумійте дані - Частота, сезонність, тренди
2. Визначте що є аномалією - Бізнес контекст
3. Обробіть missing values - Інтерполяція, fill strategies
4. Видаліть/змоделюйте тренд та сезонність - Декомпозиція
5. Оберіть метод - Statistical, ML-based, deep learning
6. Встановіть thresholds - Static або dynamic
7. Валідуйте на labeled даних - Якщо доступні
8. Імплементуйте alerting - Severity levels, routing
9. Зменшіть alert fatigue - Агрегація, suppression
10. Моніторте та ітеруйте - False positive/negative rates

Спочатку базова лінія: Z-Score

Найпростіший підхід для univariate рядів:

import numpy as np
import pandas as pd

class ZScoreDetector:
    def __init__(self, window_size=30, threshold=3.0):
        self.window_size = window_size
        self.threshold = threshold

    def detect(self, series):
        rolling_mean = series.rolling(self.window_size).mean()
        rolling_std = series.rolling(self.window_size).std()

        z_scores = (series - rolling_mean) / rolling_std
        anomalies = np.abs(z_scores) > self.threshold

        return anomalies, z_scores

# Використання
detector = ZScoreDetector(window_size=24, threshold=3.0)
anomalies, scores = detector.detect(df['metric'])

Методи виявлення

STL Decomposition:

from statsmodels.tsa.seasonal import STL

def stl_anomaly_detection(series, period, threshold=3.0):
    stl = STL(series, period=period, robust=True)
    result = stl.fit()

    residuals = result.resid
    threshold_value = threshold * np.std(residuals)
    anomalies = np.abs(residuals) > threshold_value

    return anomalies, result

Prophet-based Detection:

from prophet import Prophet
import pandas as pd

def prophet_anomaly_detection(df, interval_width=0.99):
    model = Prophet(interval_width=interval_width)
    model.fit(df[['ds', 'y']])

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

    # Точки за межами prediction interval
    df['anomaly'] = (df['y'] < forecast['yhat_lower']) | (df['y'] > forecast['yhat_upper'])

    return df['anomaly'], forecast

LSTM Autoencoder:

import torch
import torch.nn as nn

class LSTMAutoencoder(nn.Module):
    def __init__(self, input_dim, hidden_dim=64, n_layers=2):
        super().__init__()
        self.encoder = nn.LSTM(input_dim, hidden_dim, n_layers, batch_first=True)
        self.decoder = nn.LSTM(hidden_dim, input_dim, n_layers, batch_first=True)

    def forward(self, x):
        # Encode
        _, (hidden, cell) = self.encoder(x)

        # Decode
        seq_len = x.shape[1]
        decoder_input = torch.zeros_like(x)
        output, _ = self.decoder(decoder_input, (hidden, cell))

        return output

def train_autoencoder(model, train_loader, epochs=50):
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
    criterion = nn.MSELoss()

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

def detect_with_autoencoder(model, data, threshold_percentile=95):
    model.eval()
    with torch.no_grad():
        reconstructed = model(data)
        errors = torch.mean((data - reconstructed) ** 2, dim=(1, 2))

    threshold = np.percentile(errors.numpy(), threshold_percentile)
    anomalies = errors > threshold

    return anomalies, errors

Dynamic Thresholds

class DynamicThreshold:
    def __init__(self, window_size=100, sensitivity=3.0):
        self.window_size = window_size
        self.sensitivity = sensitivity

    def calculate(self, scores):
        rolling_mean = pd.Series(scores).rolling(self.window_size).mean()
        rolling_std = pd.Series(scores).rolling(self.window_size).std()

        upper = rolling_mean + self.sensitivity * rolling_std
        lower = rolling_mean - self.sensitivity * rolling_std

        return upper, lower

Метрики оцінки

Типові помилки та чек-лист

• [ ] Ігнорування сезонності: False positives на регулярних піках
• [ ] Static thresholds: Не адаптуються до змін
• [ ] Overfitting до історії: Пропуск нових типів аномалій
• [ ] Alert fatigue: Занадто багато alerts
• [ ] Відсутність context: Немає інформації про причину

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

Перед деплоєм: - [ ] Валідація на historical anomalies - [ ] Tune thresholds для acceptable FPR - [ ] Alerting інтеграція

Після деплою: - [ ] Моніторинг false positive rate - [ ] Feedback loop від operators - [ ] Регулярне перенавчання

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

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

Сценарій: Виявлення аномалій в server metrics.

Підхід: - STL decomposition для CPU/memory - LSTM autoencoder для multivariate - Dynamic thresholds

Результати: - 85% recall на known incidents - 5% false positive rate - Mean time to detect: 3 хвилини

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

• Індекс: ML Field Guide
• Прогнозування часових рядів
• Фінансові часові ряди

Додаткові матеріали

• PyOD Library
• ADTK Documentation
• Facebook Prophet

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

Дослідіть пов'язану модель на практиці: відкрийте Time Series Anomaly Detection Lab — Catch the Spike That Matters і змінюйте параметри під час роботи. Усе лишається у вашому браузері.