Розробка AI-системи аналізу якості зв'язку під ключ

Якість мережі (QoS/QoE) — багатовимірна характеристика: затримка, втрати пакетів, джитер, пропускна здатність. ML-системи, що аналізують часові ряди KPI, виявляють деградацію до скарг користувачів та локалізують проблему в топології мережі. Для оператора зв'язку втрата доходу через незадоволеність клієнтів може становити мільйони на рік. Ми будуємо такі системи під ключ: від збору SNMP-метрик до інтеграції з OSS/BSS вашого оператора. Типові болі — хибні спрацьовування статичних порогів, ручний аналіз логів, довгий пошук кореневої причини. Наші ML-алгоритми знижують false positives на 60% і скорочують час локалізації з годин до хвилин. Економія операційних витрат сягає 30%, а зниження штрафів за порушення SLA — до 40%, що забезпечує значну економію для великого оператора. Наша команда реалізувала 15+ проєктів для мереж 4G/5G. Зв'яжіться з нами, щоб обговорити впровадження у вашій мережі.

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

Метрики якості зв'язку

Ієрархія 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' } 

Чому ML краще статичних порогів?

Статичні пороги (наприклад, latency > 150 мс) дають багато хибних спрацьовувань і не адаптуються до змін мережі. ML моделі, такі як EWMA зі змінним порогом, автоматично підлаштовуються під добові патерни та тренди. У наших проєктах кількість false positives знижується на 60%, а час виявлення деградації скорочується в 5 разів.

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

Передбачаємо користувацький досвід без опитувань — за мережевими метриками. Використовуємо E-model як baseline та градієнтний бустинг для підвищення точності. В результаті MOS передбачається з помилкою менше 0.3 бала. Точне передбачення MOS дозволяє оператору проактивно керувати якістю, не чекаючи скарг. Це особливо критично для преміальних сервісів, де SLA може передбачати штрафи за падіння MOS нижче 4.0.

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

Автоматичний розрахунок доступності за кожним KPI, порівняння з контрактними SLA та генерація звітів. Система підтримує довільні SLA пороги (наприклад, 99.9% для voice, 99.5% для data).

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()) } 

Порівняння методів виявлення аномалій

Що входить в роботу