Ваша модель работает идеально. Пока не работает

Представьте: вы запускаете модель для диагностики меланомы по фотографиям родинок. Месяцами она показывает точность 98% на тестовых данных. Потом в клинику приезжает новая камера с другим балансом белого. И ваша модель начинает видеть рак там, где его нет. Или не видеть там, где он есть.

Это не ошибка в коде. Это ковариационный сдвиг - когда распределение входных данных в продакшене отличается от того, на чем модель обучалась. А теперь представьте, что это происходит не с родинками, а с результатами МРТ, анализами крови или ЭКГ.

Главная ошибка: Девять из десяти команд винят данные. "Наши данные плохие", "пациенты странные", "врачи неправильно размечают". На самом деле проблема в системе мониторинга. Или в ее отсутствии.

Как понять, что это ковариационный сдвиг, а не просто шум?

Ковариационный сдвиг - это не просто "данные изменились". Это систематическое изменение распределения признаков. В медицине это выглядит так:

Новый аппарат УЗИ с другим динамическим диапазоном
Изменение референсных значений лаборатории
Сезонные колебания показателей крови (зимой vs летом)
Демографические изменения в популяции пациентов
Обновление протокола забора биоматериала

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

💡

Ковариационный сдвиг отличается от концептуального дрейфа (когда меняются отношения между признаками и целевой переменной). Здесь меняются только входные данные, а правила остаются прежними. Но модель их уже не узнает.

Диагностика: четыре метода, которые работают в реальной жизни

Не доверяйте одной метрике. Используйте несколько методов одновременно. Вот что работает в медицинских проектах:

1 Статистические тесты для каждого признака

Для непрерывных признаков (лабораторные анализы, витальные показатели):

import numpy as np
from scipy import stats
from scipy.spatial.distance import jensenshannon

# KS-тест для сравнения распределений
def detect_feature_drift(train_data, prod_data, feature_name, alpha=0.05):
    """
    Обнаруживает дрейф для одного признака с помощью теста Колмогорова-Смирнова.
    Возвращает True если распределение изменилось.
    """
    statistic, p_value = stats.ks_2samp(
        train_data[feature_name].dropna(),
        prod_data[feature_name].dropna()
    )
    
    # p-value < alpha означает статистически значимое изменение
    is_drift = p_value < alpha
    
    return {
        'feature': feature_name,
        'ks_statistic': statistic,
        'p_value': p_value,
        'is_drift': is_drift,
        'train_mean': train_data[feature_name].mean(),
        'prod_mean': prod_data[feature_name].mean(),
        'relative_change': abs((prod_data[feature_name].mean() - train_data[feature_name].mean()) / train_data[feature_name].mean())
    }

# Пример использования для уровня глюкозы в крови
glucose_drift = detect_feature_drift(
    train_df, 
    production_df, 
    'blood_glucose_mg_dl',
    alpha=0.01  # Более строгий порог для медицинских данных
)

Ошибка новичка: Не проверяйте только средние значения. Изменение дисперсии или формы распределения может быть более опасным. Пациенты могут иметь те же средние показатели, но с большим разбросом.

2 Дистанция между распределениями

Когда нужно оценить изменение всего многомерного пространства:

import pandas as pd
from scipy.spatial.distance import jensenshannon
from sklearn.preprocessing import StandardScaler

# Дистанция Йенсена-Шеннона для категориальных признаков
def calculate_js_distance(train_cats, prod_cats):
    """
    Вычисляет дистанцию Йенсена-Шеннона между распределениями
    категориальных признаков.
    Значения от 0 (идентичные распределения) до 1 (совершенно разные).
    """
    # Собираем все уникальные категории
    all_categories = set(train_cats).union(set(prod_cats))
    
    # Создаем гистограммы
    train_hist = np.zeros(len(all_categories))
    prod_hist = np.zeros(len(all_categories))
    
    category_to_idx = {cat: i for i, cat in enumerate(all_categories)}
    
    for cat in train_cats:
        train_hist[category_to_idx[cat]] += 1
    for cat in prod_cats:
        prod_hist[category_to_idx[cat]] += 1
    
    # Нормализуем
    train_hist = train_hist / np.sum(train_hist)
    prod_hist = prod_hist / np.sum(prod_hist)
    
    # Вычисляем JS дистанцию
    js_dist = jensenshannon(train_hist, prod_hist)
    
    return js_dist

# Пример для типа медицинской страховки
js_distance = calculate_js_distance(
    train_df['insurance_type'],
    prod_df['insurance_type']
)
print(f"JS Distance: {js_distance:.4f}")  # > 0.1 означает серьезные изменения

3 Мониторинг через модель-детектор

Обучите простую модель отличать тренировочные данные от продакшен-данных:

from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_auc_score
import warnings
warnings.filterwarnings('ignore')

def train_drift_detector(train_data, prod_data):
    """
    Обучает детектор дрейфа. Если модель может легко отличить
    тренировочные данные от продакшенных, значит есть дрейф.
    """
    # Подготавливаем данные
    train_data['is_prod'] = 0
    prod_data['is_prod'] = 1
    
    combined = pd.concat([train_data, prod_data], ignore_index=True)
    
    # Удаляем целевую переменную если есть
    if 'target' in combined.columns:
        X = combined.drop(['is_prod', 'target'], axis=1)
    else:
        X = combined.drop('is_prod', axis=1)
        
    y = combined['is_prod']
    
    # Разделяем на тренировочную и валидационную выборки
    X_train, X_val, y_train, y_val = train_test_split(
        X, y, test_size=0.3, random_state=42, stratify=y
    )
    
    # Обучаем простой классификатор
    clf = RandomForestClassifier(n_estimators=50, max_depth=5, random_state=42)
    clf.fit(X_train, y_train)
    
    # Оцениваем качество
    y_pred_proba = clf.predict_proba(X_val)[:, 1]
    auc_score = roc_auc_score(y_val, y_pred_proba)
    
    # Интерпретируем результат
    if auc_score > 0.7:
        print(f"ВНИМАНИЕ: Сильный дрейф обнаружен (AUC = {auc_score:.3f})")
        # Анализируем важность признаков
        feature_importance = pd.DataFrame({
            'feature': X.columns,
            'importance': clf.feature_importances_
        }).sort_values('importance', ascending=False)
        print("\nСамые важные признаки для обнаружения дрейфа:")
        print(feature_importance.head(10).to_string())
    elif auc_score > 0.6:
        print(f"Предупреждение: Умеренный дрейф (AUC = {auc_score:.3f})")
    else:
        print(f"Дрейф не обнаружен (AUC = {auc_score:.3f})")
    
    return clf, auc_score

💡

Если AUC > 0.7, ваша модель-детектор слишком хорошо отличает продакшен-данные от тренировочных. Это красный флаг. Особенно если важные признаки для детектора - это те же признаки, которые использует ваша основная модель.

4 Контрольные группы и канонические данные

Самый надежный, но часто игнорируемый метод. Создайте "золотой стандарт" - набор данных с известным распределением:

import json
from datetime import datetime

class CanonicalDataMonitor:
    def __init__(self, canonical_data_path):
        """
        Мониторинг через сравнение с каноническими данными.
        """
        with open(canonical_data_path, 'r') as f:
            self.canonical_stats = json.load(f)
        
        self.drift_history = []
    
    def check_drift(self, current_batch_stats, threshold=0.15):
        """
        Сравнивает статистики текущего батча с каноническими.
        threshold - допустимое относительное изменение (15% по умолчанию).
        """
        alerts = []
        
        for feature, canon_stats in self.canonical_stats.items():
            if feature in current_batch_stats:
                current_mean = current_batch_stats[feature]['mean']
                canon_mean = canon_stats['mean']
                
                # Проверяем изменение среднего
                rel_change = abs((current_mean - canon_mean) / canon_mean)
                
                if rel_change > threshold:
                    alert = {
                        'timestamp': datetime.now().isoformat(),
                        'feature': feature,
                        'canonical_mean': canon_mean,
                        'current_mean': current_mean,
                        'relative_change': rel_change,
                        'threshold': threshold,
                        'severity': 'HIGH' if rel_change > 0.3 else 'MEDIUM'
                    }
                    alerts.append(alert)
        
        self.drift_history.extend(alerts)
        return alerts
    
    def generate_report(self, last_n_days=30):
        """
        Генерирует отчет по дрейфу за последние N дней.
        """
        recent_alerts = [
            a for a in self.drift_history 
            if datetime.fromisoformat(a['timestamp']) > 
            datetime.now() - timedelta(days=last_n_days)
        ]
        
        if not recent_alerts:
            return "За последние {} дней дрейф не обнаружен.".format(last_n_days)
        
        report_lines = ["Отчет по дрейфу данных:", "=" * 50]
        
        for alert in recent_alerts:
            report_lines.append(
                f"{alert['timestamp']} - {alert['feature']}: "
                f"{alert['relative_change']:.1%} изменение "
                f"({alert['severity']})"
            )
        
        return "\n".join(report_lines)

Лечение: что делать, когда дрейф найден

Обнаружили проблему? Отлично. Теперь самое сложное - что с этим делать. Вот четыре стратегии, от простой к сложной:

Стратегия	Когда использовать	Сложность	Эффективность
Ретрэнинг на новых данных	Дрейф сильный, новых данных много	Высокая	Высокая
Адаптивная нормализация	Изменения только в масштабе/сдвиге	Низкая	Средняя
Доменная адаптация	Нет разметки для новых данных	Очень высокая	Зависит от задачи
Ансамблирование моделей	Дрейф постепенный, нужно плавное обновление	Средняя	Высокая

Адаптивная нормализация: быстрый фикс

Когда дрейф вызван изменением масштаба (например, новая лаборатория использует другие единицы измерения):

import numpy as np
from scipy import stats

class AdaptiveNormalizer:
    """
    Адаптивная нормализация для онлайн-коррекции ковариационного сдвига.
    Работает по принципу: "приводим новые данные к старому распределению".
    """
    def __init__(self, reference_data, features_to_normalize):
        self.reference_stats = {}
        self.features = features_to_normalize
        
        # Вычисляем статистики по референсным данным
        for feature in features_to_normalize:
            feature_data = reference_data[feature].dropna()
            self.reference_stats[feature] = {
                'mean': np.mean(feature_data),
                'std': np.std(feature_data),
                'q5': np.percentile(feature_data, 5),
                'q95': np.percentile(feature_data, 95)
            }
    
    def normalize_batch(self, batch_data, update_stats=False, window_size=1000):
        """
        Нормализует батч данных.
        Если update_stats=True, обновляет статистики скользящим окном.
        """
        normalized_batch = batch_data.copy()
        
        for feature in self.features:
            if feature not in batch_data.columns:
                continue
                
            current_data = batch_data[feature].dropna()
            
            if len(current_data) == 0:
                continue
            
            # Вычисляем текущие статистики
            current_mean = np.mean(current_data)
            current_std = np.std(current_data)
            
            ref_mean = self.reference_stats[feature]['mean']
            ref_std = self.reference_stats[feature]['std']
            
            # Z-нормализация с учетом референсного распределения
            normalized_values = (current_data - current_mean) / current_std
            normalized_values = normalized_values * ref_std + ref_mean
            
            normalized_batch[feature] = normalized_values
            
            # Обновляем статистики если нужно
            if update_stats:
                # Простое экспоненциальное сглаживание
                alpha = 2 / (window_size + 1)
                self.reference_stats[feature]['mean'] = \
                    alpha * current_mean + (1 - alpha) * self.reference_stats[feature]['mean']
                self.reference_stats[feature]['std'] = \
                    alpha * current_std + (1 - alpha) * self.reference_stats[feature]['std']
        
        return normalized_batch

Важно: Адаптивная нормализация не исправит фундаментальные изменения в отношениях между признаками. Если новый аппарат МРТ не только меняет контраст, но и обнаруживает новые паттерны - нужен ретрэнинг.

Реальный кейс: почему модель для диагностики COVID-19 перестала работать через 3 месяца

В 2020 году одна больница развернула модель для определения тяжести COVID-19 по КТ легких. Точность на валидации - 94%. Через три месяца врачи начали жаловаться: модель пропускает тяжелые случаи.

Что произошло:

Больница получила новые КТ-сканеры с другим протоколом реконструкции изображений
Контрастность и разрешение изменились на 30%
Модель обучалась на старых сканах и не знала про новые артефакты
Никто не мониторил распределение пиксельных значений входящих данных

Решение было простым (после того как его нашли):

# 1. Добавили мониторинг распределения интенсивности пикселей
class CTIntensityMonitor:
    def __init__(self, baseline_stats):
        self.baseline = baseline_stats
        
    def check_ct_scan(self, ct_scan_array):
        """Проверяет КТ-скан на соответствие базовому распределению"""
        current_stats = {
            'mean_intensity': np.mean(ct_scan_array),
            'std_intensity': np.std(ct_scan_array),
            'percentile_95': np.percentile(ct_scan_array, 95),
            'percentile_5': np.percentile(ct_scan_array, 5)
        }
        
        # Проверяем отклонения
        deviations = {}
        for key in self.baseline:
            rel_change = abs(current_stats[key] - self.baseline[key]) / self.baseline[key]
            if rel_change > 0.15:  # Более 15% изменение
                deviations[key] = rel_change
        
        return deviations

# 2. Реализовали адаптивную предобработку
class AdaptiveCTPreprocessor:
    def __init__(self, target_mean=0, target_std=1):
        self.target_mean = target_mean
        self.target_std = target_std
        self.history = []
        
    def preprocess(self, ct_scan):
        """Приводит КТ-скан к целевому распределению"""
        current_mean = np.mean(ct_scan)
        current_std = np.std(ct_scan)
        
        # Сохраняем статистики для мониторинга
        self.history.append({
            'timestamp': datetime.now(),
            'mean': current_mean,
            'std': current_std
        })
        
        # Нормализуем
        normalized = (ct_scan - current_mean) / current_std
        normalized = normalized * self.target_std + self.target_mean
        
        return normalized

Через месяц после внедрения мониторинга система обнаружила, что 40% новых сканов выходят за допустимые границы. Инженеры добавили эти данные в тренировочный набор, дообучили модель, и точность вернулась к 92%.

Как не надо делать: пять ошибок, которые совершают все

Ждать, пока пользователи пожалуются
К тому времени модель уже месяц делает ошибки. В медицине это могут быть человеческие жизни.
Проверять только accuracy или F1-score
Метрики могут оставаться высокими, пока модель делает систематические ошибки на определенных подгруппах.
Использовать один метод обнаружения
KS-тест пропустит ковариационный сдвиг в категориальных признаках. Модель-детектор может быть слишком чувствительной. Нужна комбинация методов.
Не документировать версии данных
Когда появляется дрейф, вы должны знать: что изменилось? Новый аппарат? Новая лаборатория? Другая популяция пациентов?
Думать, что это разовая проблема
Ковариационный сдвиг - это не баг, это фича реального мира. Данные всегда меняются. Система мониторинга должна работать постоянно.

Инструменты, которые реально экономят время

Не пишите всё с нуля. Вот что работает в 2024:

Инструмент	Для чего	Плюсы	Минусы
Evidently AI	Мониторинг дрейфа в реальном времени	Готовые метрики, красивые дашборды	Тяжеловесный для простых задач
Alibi Detect	Обнаружение аномалий и дрейфа	Много алгоритмов, работает с изображениями	Сложная настройка
River	Онлайн-машинное обучение	Идеально для постепенного дрейфа	Ограниченный набор моделей
Простые скрипты на pandas + scipy	Быстрый прототип	Полный контроль, нет зависимостей	Нужно писать всё самому

Мой совет: начинайте с простых скриптов как в этой статье. Поймите, какие метрики важны именно для вашей задачи. Потом переходите на Evidently или Alibi Detect.

Что делать прямо сейчас: чеклист на неделю

День 1-2: Соберите статистики по тренировочным данным
Для каждого признака: среднее, стандартное отклонение, квартили. Сохраните в JSON. Это ваш baseline.
День 3: Напишите скрипт для сравнения с новыми данными
Используйте KS-тест для непрерывных признаков, дистанцию Йенсена-Шеннона для категориальных.
День 4: Настройте алерты
Когда p-value < 0.01 или JS-дистанция > 0.1 - отправляйте уведомление в Slack/Telegram.
День 5: Протестируйте на исторических данных
Если у вас есть данные за последние месяцы - запустите детектор на них. Увидите ли вы известные изменения?
День 6-7: Автоматизируйте и документируйте
Сделайте скрипт частью вашего пайплайна. Добавьте в документацию: "Вот как мы обнаруживаем дрейф данных".

💡

Самая частая ошибка на этом этапе - сделать систему мониторинга, но не сделать систему реагирования. Что делать, когда алерт пришел? Кто отвечает? Сколько времени на реакцию? Пропишите это в runbook.

Когда всё сломается (и это нормально)

Ковариационный сдвиг - это не авария. Это нормальное состояние production-системы. Данные меняются всегда. Пациенты стареют. Технологии развиваются. Протоколы обновляются.

Ваша задача - не предотвратить изменения, а обнаружить их раньше, чем они навредят. Не идеальная модель, а устойчивая система. Не единичная точность, а стабильная работа.

Помните историю про модель для COVID-19? После того как они внедрили мониторинг, они обнаружили еще три сдвига за полгода:

Изменение референсных значений в лаборатории
Новый протокол подготовки пациентов к КТ
Сезонные колебания в анализах крови

Каждый раз они адаптировали модель. Каждый раз это занимало дни, а не недели. Потому что система мониторинга уже была на месте.

Ковариационный сдвиг - это не технический долг. Это техническая гигиена. Как мыть руки перед операцией. Как стерилизовать инструменты. Как проверять срок годности лекарств.

Не ждите, пока модель ослепнет. Начните мониторить сегодня. Даже если это просто скрипт на 50 строк. Даже если он будет проверять только три самых важных признака. Даже если алерты будут приходить вам в личный Telegram.

Потому что завтра в вашу клинику привезут новый аппарат. Или поменяют реактивы в лаборатории. Или изменится демография пациентов. И ваша модель этого не заметит.

А вы - заметите.

Ковариационный сдвиг: когда ваша модель внезапно слепнет