Разработка AI-системы анализа качества связи

Разработка AI-системы анализа качества связи

Качество сети (QoS/QoE) — многомерная характеристика: задержка, потери пакетов, джиттер, пропускная способность. ML-системы анализируют эти параметры в реальном времени, обнаруживают деградацию до жалоб пользователей и локализуют проблему в топологии сети.

Метрики качества связи

Иерархия QoS показателей:

network_kpis = {
    # Физический уровень
    'signal_to_noise_ratio_db': 'SNR < 10 dB = деградация',
    'bit_error_rate': 'BER > 1e-6 = проблема',
    'optical_power_dbm': 'для оптики: отклонение ±3 dB от нормы',

    # Транспортный уровень
    'packet_loss_pct': '> 1% = заметно для голоса, > 0.1% критично для видео',
    'latency_ms': 'RTT: < 20 мс = отлично, > 100 мс = деградация голоса',
    'jitter_ms': '> 30 мс = разрыв голосового вызова',
    'throughput_mbps': 'отклонение от SLA > 20% = нарушение',

    # Прикладной уровень (QoE)
    'mos_score': 'Mean Opinion Score 1-5: < 3.5 = жалобы',
    'video_buffering_ratio': '> 2% = заметно пользователю',
    'call_setup_success_rate': '< 99% = проблема в IMS/SS7'
}

Аномалия в временных рядах KPI

EWMA + адаптивные пороги:

import numpy as np
import pandas as pd

class NetworkKPIMonitor:
    def __init__(self, alpha=0.1, sigma_multiplier=3.0):
        self.alpha = alpha
        self.sigma_multiplier = sigma_multiplier
        self.ewma_mean = {}
        self.ewma_var = {}

    def update(self, kpi_name: str, value: float) -> dict:
        if kpi_name not in self.ewma_mean:
            self.ewma_mean[kpi_name] = value
            self.ewma_var[kpi_name] = 0.0
            return {'status': 'initializing'}

        # Обновление EWMA
        prev_mean = self.ewma_mean[kpi_name]
        self.ewma_mean[kpi_name] = (
            self.alpha * value + (1 - self.alpha) * prev_mean
        )
        self.ewma_var[kpi_name] = (
            (1 - self.alpha) * (self.ewma_var[kpi_name] +
             self.alpha * (value - prev_mean)**2)
        )

        std = np.sqrt(self.ewma_var[kpi_name])
        upper_bound = self.ewma_mean[kpi_name] + self.sigma_multiplier * std
        lower_bound = self.ewma_mean[kpi_name] - self.sigma_multiplier * std

        anomaly = value > upper_bound or value < lower_bound

        return {
            'kpi': kpi_name,
            'value': value,
            'expected': self.ewma_mean[kpi_name],
            'upper_bound': upper_bound,
            'anomaly': anomaly,
            'deviation_sigma': (value - self.ewma_mean[kpi_name]) / (std + 1e-9)
        }

Корреляционный анализ деградации

Топологическая локализация проблемы:

import networkx as nx

def localize_network_degradation(anomaly_events: list,
                                  topology_graph: nx.Graph) -> dict:
    """
    Если аномалии одновременно на нескольких сегментах —
    ищем общий upstream узел (root cause).
    """
    # Группируем аномальные узлы
    degraded_nodes = set(e['node_id'] for e in anomaly_events
                         if e['anomaly'] and e['timestamp'] == max(e['timestamp']
                                                                    for e in anomaly_events))

    # Для каждой пары деградированных узлов — найти LCA (Least Common Ancestor)
    suspect_nodes = {}
    for u, v in combinations(degraded_nodes, 2):
        try:
            paths = list(nx.all_simple_paths(topology_graph, u, v, cutoff=5))
            for path in paths:
                for node in path:
                    if node not in degraded_nodes:
                        suspect_nodes[node] = suspect_nodes.get(node, 0) + 1
        except nx.NetworkXNoPath:
            pass

    if suspect_nodes:
        root_cause = max(suspect_nodes, key=suspect_nodes.get)
        return {
            'root_cause_node': root_cause,
            'confidence': suspect_nodes[root_cause] / len(degraded_nodes),
            'affected_downstream': list(degraded_nodes)
        }

    return {'root_cause_node': None, 'affected_downstream': list(degraded_nodes)}

MOS Prediction для Voice/Video

Предсказание пользовательского опыта:

from sklearn.ensemble import GradientBoostingRegressor

def build_mos_prediction_model(network_samples: pd.DataFrame) -> GradientBoostingRegressor:
    """
    Предсказание MOS из сетевых метрик — без субъективного опроса пользователей.
    E-model ITU-T G.107 как baseline, ML улучшает точность.
    """
    features = [
        'packet_loss_pct',
        'latency_ms',
        'jitter_ms',
        'codec_type_encoded',    # G.711=0, G.722=1, Opus=2
        'plc_effectiveness'      # Packet Loss Concealment quality
    ]

    # E-model baseline как дополнительный признак
    network_samples['e_model_r_factor'] = network_samples.apply(
        lambda row: compute_e_model_r_factor(
            row['latency_ms'], row['packet_loss_pct'], row['jitter_ms']
        ), axis=1
    )

    model = GradientBoostingRegressor(n_estimators=200, max_depth=4)
    model.fit(network_samples[features + ['e_model_r_factor']],
              network_samples['mos_score'])
    return model

def compute_e_model_r_factor(latency, loss_pct, jitter):
    """Упрощённая E-model формула ITU-T G.107"""
    r_base = 93.2
    r_latency = 0.024 * latency + 0.11 * max(0, latency - 177.3)
    r_loss = 11 + 40 * np.log(1 + 10 * loss_pct / 100)
    return max(0, r_base - r_latency - r_loss)

SLA Compliance Tracking

Автоматический подсчёт нарушений SLA:

def track_sla_compliance(kpi_history: pd.DataFrame,
                          sla_thresholds: dict,
                          contract_id: str) -> dict:
    violations = {}

    for kpi, threshold in sla_thresholds.items():
        if kpi not in kpi_history.columns:
            continue

        total_minutes = len(kpi_history)
        violation_minutes = len(kpi_history[kpi_history[kpi] > threshold])
        availability = (total_minutes - violation_minutes) / total_minutes * 100

        violations[kpi] = {
            'sla_target': threshold,
            'availability_pct': round(availability, 4),
            'violation_minutes': violation_minutes,
            'sla_breach': availability < 99.9  # стандартный SLA для операторов
        }

    return {
        'contract_id': contract_id,
        'period': f"{kpi_history.index.min()} — {kpi_history.index.max()}",
        'kpi_compliance': violations,
        'overall_compliance': all(not v['sla_breach'] for v in violations.values())
    }

Интеграция с OSS/BSS стеком: Northbound API к Nokia NetAct, Ericsson ENM, Huawei U2000. SNMP/NETCONF для активного оборудования. Grafana + InfluxDB для визуализации, PagerDuty / Zabbix для алертинга.

Сроки: EWMA мониторинг KPI + аномалии + дашборд — 2-3 недели. Топологическая локализация, MOS prediction, SLA compliance tracker, OSS/BSS интеграция — 2-3 месяца.