Как отслеживать качество нейросети?

Отслеживание качества нейросети

Отслеживание качества нейросети в production — это критический процесс, включающий мониторинг технических метрик, business метрик и детектирование дрейфов данных.

1. Основные метрики качества

Метрики точности (по типу задачи):

from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, roc_auc_score

# Для классификации
accuracy = accuracy_score(y_true, y_pred)
precision = precision_score(y_true, y_pred, average="weighted")
recall = recall_score(y_true, y_pred, average="weighted")
f1 = f1_score(y_true, y_pred, average="weighted")
auc_roc = roc_auc_score(y_true, y_pred_proba)

# Для регрессии
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score

mae = mean_absolute_error(y_true, y_pred)
rmse = mean_squared_error(y_true, y_pred, squared=False)
r2 = r2_score(y_true, y_pred)

2. Детектирование дрейфа данных (Data Drift)

Дрейф данных — это изменение распределения входящих данных, что ведёт к снижению качества предсказаний.

from scipy.stats import ks_2samp, chi2_contingency
import numpy as np

def detect_drift_continuous(baseline_data, current_data, threshold=0.05):
    """Kolmogorov-Smirnov тест для численных признаков"""
    statistic, p_value = ks_2samp(baseline_data, current_data)
    is_drift = p_value < threshold
    return is_drift, p_value

def detect_drift_categorical(baseline_cats, current_cats, threshold=0.05):
    """Chi-square тест для категориальных признаков"""
    contingency = np.array([
        np.bincount(baseline_cats, minlength=max(baseline_cats)+1),
        np.bincount(current_cats, minlength=max(current_cats)+1)
    ])
    chi2, p_value, dof, expected = chi2_contingency(contingency)
    is_drift = p_value < threshold
    return is_drift, p_value

3. Детектирование дрейфа предсказаний (Prediction Drift)

Изменение распределения самих предсказаний модели.

def monitor_prediction_distribution(baseline_predictions, current_predictions):
    """Проверка смещения среднего значения предсказаний"""
    baseline_mean = np.mean(baseline_predictions)
    current_mean = np.mean(current_predictions)
    baseline_std = np.std(baseline_predictions)
    
    # z-score для выявления аномалий
    z_score = (current_mean - baseline_mean) / baseline_std
    
    if abs(z_score) > 3:  # 3-sigma rule
        print(f"Prediction drift detected: z-score = {z_score}")
    
    return z_score

4. Отслеживание с помощью времени жизни данных

Для моделей, где ground truth известен с задержкой (fraud detection, loan defaults):

import pandas as pd
from datetime import datetime, timedelta

def track_model_performance_over_time(predictions_df, ground_truth_df):
    """Отслеживание метрик с задержкой ground truth"""
    
    # Объединяем по ID
    merged = predictions_df.merge(
        ground_truth_df,
        left_on="prediction_id",
        right_on="truth_id",
        suffixes=("_pred", "_truth")
    )
    
    # Группируем по дате предсказания
    daily_metrics = merged.groupby("prediction_date").apply(
        lambda x: {
            "accuracy": (x["prediction"] == x["actual"]).mean(),
            "precision": (x[x["prediction"]==1]["actual"]==1).sum() / (x["prediction"]==1).sum(),
            "auc": roc_auc_score(x["actual"], x["pred_proba"])
        }
    ).reset_index()
    
    return daily_metrics

5. Логирование и мониторинг в production

import logging
from dataclasses import dataclass
from datetime import datetime

@dataclass
class ModelQualityLog:
    timestamp: datetime
    model_version: str
    accuracy: float
    precision: float
    recall: float
    data_drift_detected: bool
    prediction_drift_z_score: float
    inference_time_ms: float
    samples_processed: int

class ModelMonitor:
    def __init__(self, model_name: str):
        self.model_name = model_name
        self.logger = logging.getLogger(model_name)
    
    def log_batch_prediction(self, log: ModelQualityLog):
        """Логирование метрик предсказания"""
        self.logger.info(
            f"Model={self.model_name} | "
            f"Accuracy={log.accuracy:.4f} | "
            f"DataDrift={log.data_drift_detected} | "
            f"LatencyMs={log.inference_time_ms}"
        )
        
        # Отправить в систему мониторинга (Prometheus, Grafana, etc.)

6. Пороги алертов (Alerting)

7. Автоматическая переподготовка (AutoML Retraining)

def should_retrain_model(current_metrics: dict, baseline_metrics: dict, 
                        accuracy_threshold=0.95, drift_threshold=0.01):
    """Решение о необходимости переподготовки"""
    
    accuracy_ratio = current_metrics["accuracy"] / baseline_metrics["accuracy"]
    data_drift_p = current_metrics["drift_p_value"]
    
    if accuracy_ratio < accuracy_threshold:
        return True, "Accuracy degrade"
    
    if data_drift_p < drift_threshold:
        return True, "Data drift detected"
    
    return False, "No retraining needed"

8. Best Practices

1. Установите базовые метрики на дату развёртывания модели
2. Отслеживайте несколько сигналов одновременно — не только accuracy
3. Используйте временные окна (часовые, дневные, недельные графики)
4. Проверяйте корреляции между дрейфом данных и снижением точности
5. Ведите версионирование всех моделей и их метрик
6. Сохраняйте все предсказания для post-hoc анализа с задержкой
7. Интегрируйте с системой alerting (PagerDuty, Slack, Email)

Вывод: Эффективный мониторинг требует комбинации метрик точности, детектирования дрейфов и бизнес-метрик с автоматическими алертами и pipeline для переподготовки.