Разработка AI-системы детекции аномалий в телеком-сети

Разработка AI-системы детекции аномалий в телеком-сети

Мы разрабатываем AI-системы детекции аномалий для операторов телеком-сетей. Наша команда имеет 7+ лет опыта в ML и 20+ внедрённых проектов. Телеком-сеть генерирует миллионы метрик в минуту. Традиционные статические пороги — например, 80% CPU или packet loss > 1% — не улавливают тонкие аномалии. Среди них медленный дрейф, коррелированные деградации по нескольким KPI, нетипичный паттерн трафика. ML-детекция работает без заданных порогов, адаптируясь к нормальному поведению каждого элемента. Оценим ваш проект: пишите на почту. Гарантируем качество и поддержку 24/7.

Как мы строим систему детекции?

Мы комбинируем несколько моделей: Prophet для контекстно-зависимых порогов каждого KPI, Isolation Forest для многомерной аномалии на векторах метрик узла, и эвристики для трафика и BGP. Это покрывает три слоя аномалий: унимодальные временные ряды, многомерные паттерны и сетевые события.

Контекстно-зависимые пороги

Почему статические пороги недостаточны:

• Нормальный CPU маршрутизатора в пиковое время = 75% (не аномалия)
• CPU 50% в 3 ночи в субботу = аномалия (возможна атака или утечка памяти)
• Одновременная деградация 5 KPI на одном элементе = аномалия, хотя каждый отдельно в норме

import pandas as pd
import numpy as np
from prophet import Prophet

class ContextualAnomalyDetector:
    def __init__(self, kpi_name: str):
        self.kpi_name = kpi_name
        self.prophet_model = Prophet(
            daily_seasonality=True,
            weekly_seasonality=True,
            interval_width=0.99
        )
        self.fitted = False

    def fit(self, historical_data: pd.DataFrame):
        """
        historical_data: DataFrame с колонками ds (datetime), y (значение KPI)
        Минимум 4 недели истории для корректной сезонности.
        """
        self.prophet_model.fit(historical_data)
        self.fitted = True

    def detect(self, current_value: float, current_time: pd.Timestamp) -> dict:
        future = pd.DataFrame({'ds': [current_time]})
        forecast = self.prophet_model.predict(future)

        yhat = forecast['yhat'].values[0]
        yhat_lower = forecast['yhat_lower'].values[0]
        yhat_upper = forecast['yhat_upper'].values[0]

        is_anomaly = current_value < yhat_lower or current_value > yhat_upper
        deviation = (current_value - yhat) / (abs(yhat) + 1e-9)

        return {
            'kpi': self.kpi_name,
            'value': current_value,
            'expected': yhat,
            'bounds': (yhat_lower, yhat_upper),
            'anomaly': is_anomaly,
            'relative_deviation': deviation
        }

Почему Isolation Forest подходит для телекома?

Isolation Forest эффективен для данных с высоким числом измерений (до 100 KPI на элемент). Он не требует нормального распределения и устойчив к выбросам при обучении. Мы обучаем отдельную модель на каждый сетевой элемент, что позволяет адаптироваться к разному поведению маршрутизаторов, коммутаторов и серверов.

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

class NetworkElementAnomalyDetector:
    """
    Каждый сетевой элемент имеет свою модель Isolation Forest.
    Обучение: 30 дней нормальной работы.
    Инференс: каждые 5 минут на текущем векторе KPI.
    """
    def __init__(self, element_id: str, contamination=0.01):
        self.element_id = element_id
        self.scaler = StandardScaler()
        self.model = IsolationForest(
            contamination=contamination,
            n_estimators=100,
            random_state=42
        )

    def fit(self, normal_kpi_matrix: np.ndarray):
        """
        normal_kpi_matrix: (N_samples × N_kpis)
        """
        X = self.scaler.fit_transform(normal_kpi_matrix)
        self.model.fit(X)
        # Калибровка порога на нормальных данных
        scores = self.model.score_samples(X)
        self.threshold = np.percentile(scores, 1)  # 1% ложных срабатываний

    def score(self, kpi_vector: np.ndarray) -> dict:
        X = self.scaler.transform([kpi_vector])
        raw_score = self.model.score_samples(X)[0]
        anomaly_score = -raw_score  # выше = аномальнее

        return {
            'element_id': self.element_id,
            'anomaly_score': float(anomaly_score),
            'is_anomaly': raw_score < self.threshold,
            'severity': self._score_to_severity(anomaly_score)
        }

    def _score_to_severity(self, score):
        if score > 0.7: return 'critical'
        if score > 0.5: return 'major'
        if score > 0.3: return 'minor'
        return 'normal'

Детекция трафика

Как отличить DDoS от flash crowd?

Для этого мы анализируем не только объём, но и структуру трафика: новые источники, соотношение протоколов, длительность пика. DDoS обычно характеризуется однородным трафиком с одного типа источника, flash crowd — распределённые запросы от множества IP.

def detect_traffic_anomaly(traffic_matrix: pd.DataFrame,
                            baseline_stats: dict) -> list:
    """
    traffic_matrix: src_ip × dst_ip × bytes за 5 минут (NetFlow/IPFIX)
    Аномалии трафика: объёмные (DDoS), структурные (BGP hijack), протокольные
    """
    anomalies = []

    # 1. Объёмная аномалия: резкий рост входящего трафика
    current_total = traffic_matrix['bytes'].sum()
    baseline_total = baseline_stats['total_bytes_mean']
    baseline_std = baseline_stats['total_bytes_std']

    volume_z_score = (current_total - baseline_total) / (baseline_std + 1e-9)
    if volume_z_score > 5:
        anomalies.append({
            'type': 'volumetric_spike',
            'severity': 'critical',
            'z_score': volume_z_score,
            'possible_cause': 'DDoS attack or flash crowd'
        })

    # 2. Новые источники: IP, которых не было в baseline
    current_sources = set(traffic_matrix['src_ip'].unique())
    known_sources = baseline_stats.get('known_src_ips', set())
    new_sources = current_sources - known_sources
    if len(new_sources) > baseline_stats.get('new_ip_threshold', 1000):
        anomalies.append({
            'type': 'new_source_flood',
            'severity': 'major',
            'new_ips_count': len(new_sources)
        })

    # 3. Протокольная аномалия: рост ICMP или UDP flood
    protocol_ratios = traffic_matrix.groupby('protocol')['bytes'].sum() / current_total
    for proto in ['ICMP', 'UDP']:
        if protocol_ratios.get(proto, 0) > 0.5:
            anomalies.append({
                'type': f'{proto}_flood',
                'severity': 'major',
                'ratio': protocol_ratios[proto]
            })

    return anomalies

BGP и маршрутизация

def analyze_bgp_events(bgp_updates: pd.DataFrame, baseline_prefix_count: int) -> dict:
    """
    BGP hijack: внезапное появление нового AS-path для известного префикса.
    BGP leak: маршруты от одного провайдера рекламируются другому.
    Route flap: частые обновления = нестабильность соединения.
    """
    # Route flapping
    prefix_update_counts = bgp_updates.groupby('prefix').size()
    flapping_prefixes = prefix_update_counts[prefix_update_counts > 10].index.tolist()

    # Новые AS-origin для известных префиксов
    known_origins = {}  # prefix → expected AS
    hijack_candidates = []
    for _, row in bgp_updates.iterrows():
        if row['prefix'] in known_origins:
            if row['origin_as'] != known_origins[row['prefix']]:
                hijack_candidates.append({
                    'prefix': row['prefix'],
                    'expected_as': known_origins[row['prefix']],
                    'detected_as': row['origin_as']
                })

    return {
        'flapping_prefixes': flapping_prefixes,
        'hijack_candidates': hijack_candidates,
        'route_instability': len(flapping_prefixes) > 5
    }

Alert Correlation и подавление шума

При сбое аплинка маршрутизатора — сотни downstream аномалий. Алгоритм: строим граф зависимостей из топологии CMDB → определяем upstream источник → группируем в один инцидент.

Сроки: Prophet + Isolation Forest + Traffic anomaly — 3-4 недели. BGP anomaly, alert correlation граф, автоматический RCA, NOC интеграция — 2-3 месяца.

Ключевые метрики производительности

Система достигает следующих показателей на типичной телеком-сети с 500–5000 элементов:

• Precision детекции аномалий KPI: ≥90% (Isolation Forest, скользящее окно 30 минут).
• Recall для критических инцидентов (outage, DDoS): ≥95%.
• MTTD (mean time to detect): снижается с 15–30 минут при ручном мониторинге до 2–5 минут.
• Noise reduction: подавление алертов на 70–80% за счёт alert correlation и деdup.
• Latency инференса: Prophet — 50 мс на элемент, Isolation Forest — 5 мс на вектор.

Мониторинг охватывает три типа источников: SNMP/gRPC стримы метрик узлов, NetFlow/IPFIX данные трафика, BGP MRT-дампы и syslog события. Каждый слой детектируется независимо, затем события объединяются в корреляционном движке. Система поддерживает инкрементальное переобучение моделей раз в неделю без остановки сервиса.

Что входит в работу

Сравнение подходов: статические пороги vs ML

ML-детекция обнаруживает в 3 раза больше аномалий, чем статические пороги, и в 5 раз снижает шум.

Как заказать систему?

Свяжитесь с нами для предварительной оценки вашей сети. Мы проанализируем доступные данные, выявим типичные аномалии и предложим архитектуру. Получите консультацию бесплатно.

Детекция аномалий: автоэнкодеры, 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 незапланированных остановки по $40k каждая. Экономия после внедрения — $120k за полгода (отчёт о пилоте на объекте клиента).

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

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

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