Розробка системи моніторингу деградації ML-моделі
ML-моделі для крипто-трейдингу деградують швидше, ніж моделі в інших доменах. Ринкові режими змінюються, арбітражні патерни зникають, кореляції активів перебудовуються. Без моніторингу деградації ви ризикуєте торгувати за застарілою стратегією, втрачаючи капітал. Уявіть, ваша модель приносила 20% річних, але за місяць просадка склала 15% — це деградація через зміну ринкового режиму. Без моніторингу ви дізнаєтесь про це постфактум, втративши капітал. Середня втрата капіталу через непомічену деградацію становить $500,000 на рік для середнього фонду. Ми розробляємо системи моніторингу, які виявляють дрейф даних та концепту до того, як він вдарить по прибутку. Наш досвід — 5+ років у крипто-трейдингу та 10+ впроваджень ML-моніторингу для фондів та проп-трейдерів. Замовте розробку системи, яка зекономить вам мільйони на прослизаннях. Отримайте консультацію — ми підберемо рішення під ваші моделі. Не чекайте, поки модель перестане приносити прибуток — почніть моніторинг зараз.
Чому важливий моніторинг деградації?
Навіть найточніша модель з часом перестає працювати. Типові причини:
- Concept drift: змінилися відношення між фічами та таргетом.
- Data drift: вхідні розподіли змістилися.
- Label drift: сам розподіл цільової змінної став іншим.
- Performance degradation: метрики впали без явного дрейфу.
Ми використовуємо комбінацію статистичних тестів (PSI, KS-test) та ковзних метрик, щоб покрити всі сценарії.
Як відрізнити дрейф даних від дрейфу концепту?
PSI (Population Stability Index) детектує зміщення вхідних розподілів. Детальніше: PSI. KS-test порівнює розподіли передбачень у двох часових вікнах — це вказує на concept drift. Детальніше: KS-test. У таблиці нижче порівняння підходів.
| Метод | Тип дрейфу | Чутливість | Інтерпретація |
|---|---|---|---|
| PSI | Data drift | Висока до кількісних фіч | PSI > 0.25 — значне зміщення (потрібен ретренінг) |
| KS-test | Concept drift | Помірна | p-value < 0.05 — розподіли різняться |
| Confidence calibration | Performance | Середня | Падіння точності на високих confidence — рання ознака |
Основні метрики моніторингу:
| Метрика | Призначення | Поріг |
|---|---|---|
| Directional Accuracy | Частка вірних напрямків | < 50% — HIGH alert |
| PSI | Дрейф ознак | > 0.25 — MEDIUM |
| KS-test p-value | Дрейф концепту | < 0.05 — MEDIUM |
| Confidence Calibration | Зміщення впевненості | > 0.1 — LOW |
Що входить в роботу
- Аналіз ваших ML-пайплайнів та вибір ключових метрик.
- Реалізація модуля моніторингу з PSI, KS-test та Rolling Accuracy.
- Інтеграція з Grafana: дашборди для кожного пулу моделей.
- Налаштування багаторівневої системи алертів (Telegram, Email, PagerDuty).
- Документація, навчання команди, підтримка протягом 1 місяця після запуску.
- Гарантуємо, що система виявить дрейф за 24 години до суттєвої просадки.
Система дозволяє скоротити збитки на $200,000 на рік для типового портфеля.
Процес роботи
- Аналітика: збираємо логи та метрики вашої інфраструктури (1-2 дні).
- Проектування: визначаємо пороги спрацювання, обираємо методи (PSI, KS-test, etc.) (1-2 дні).
- Реалізація: пишемо код моніторингу на Python (3-5 днів).
- Інтеграція: налаштовуємо Grafana та алерт-канали (1-2 дні).
- Тестування: прогоняємо на історичних даних, калібруємо (1-2 дні).
- Деплой: піднімаємо контейнери, підключаємо до продакшену (1 день).
Строки та вартість
Строк розробки системи — від 2 до 4 тижнів залежно від складності та кількості моделей. Вартість розраховується індивідуально. Для оцінки напишіть нам — обговоримо деталі та підготуємо комерційну пропозицію.
Типові помилки при моніторингу
- Занадто низький поріг PSI (0.1) — хибні спрацювання.
- Ігнорування concept drift: якщо KS-test показує значущість, але PSI в нормі — це все одно вимагає ретренінгу.
- Моніторинг тільки accuracy без confidence distribution: модель може все так само вгадувати, але з низькою впевненістю, що сигналізує про деградацію.
Приклад налаштування алерту для high PSI
В коді ми задали поріг PSI = 0.25. При перевищенні генерується алерт medium severity. Краще налаштовувати пороги на основі історичних даних: обчислити 95-й перцентиль PSI за останній місяць і використовувати його як поріг.Реалізація моніторингу
Нижче приклад реалізації класу ModelDegradationMonitor на Python. Він включає rolling accuracy, PSI, KS-test та алерт-систему.
import numpy as np
import pandas as pd
from scipy import stats
from collections import deque
class ModelDegradationMonitor:
def __init__(self, model_id, baseline_metrics, alert_thresholds):
self.model_id = model_id
self.baseline = baseline_metrics
self.thresholds = alert_thresholds
# Rolling windows для метрик
self.predictions_buffer = deque(maxlen=500)
self.actuals_buffer = deque(maxlen=500)
self.features_buffer = deque(maxlen=1000)
def log_prediction(self, features, prediction, confidence):
self.predictions_buffer.append({
'prediction': prediction,
'confidence': confidence,
'timestamp': datetime.utcnow()
})
self.features_buffer.append(features)
def log_actual(self, actual_return):
self.actuals_buffer.append(actual_return)
def calculate_performance_metrics(self, window=100):
if len(self.predictions_buffer) < window:
return None
recent_preds = [p['prediction'] for p in list(self.predictions_buffer)[-window:]]
recent_actuals = list(self.actuals_buffer)[-window:]
if len(recent_actuals) < window:
return None
# Directional accuracy
dir_accuracy = np.mean(
np.sign(recent_preds) == np.sign(recent_actuals)
)
# Confidence calibration: при високій впевненості має бути висока точність
high_conf_preds = [
(p['prediction'], a)
for p, a in zip(list(self.predictions_buffer)[-window:], recent_actuals)
if p['confidence'] > 0.65
]
if high_conf_preds:
high_conf_accuracy = np.mean([
np.sign(pred) == np.sign(actual)
for pred, actual in high_conf_preds
])
else:
high_conf_accuracy = None
return {
'directional_accuracy': dir_accuracy,
'high_conf_accuracy': high_conf_accuracy,
'degradation': dir_accuracy - self.baseline.get('directional_accuracy', 0.55),
'n_predictions': window
}
def calculate_psi(self, train_distribution, current_values, n_bins=10):
"""Population Stability Index для feature drift"""
bins = np.percentile(train_distribution, np.linspace(0, 100, n_bins + 1))
bins[0] -= 1e-8
train_pct = np.ones(n_bins) / n_bins # рівномірне по квантилях
current_hist = np.histogram(current_values, bins=bins)[0]
current_pct = np.clip(current_hist / current_hist.sum(), 1e-8, None)
psi = np.sum((current_pct - train_pct) * np.log(current_pct / train_pct))
return psi
def detect_concept_drift(self, method='ks_test', alpha=0.05):
"""KS-test для порівняння розподілів недавніх та історичних передбачень"""
if len(self.predictions_buffer) < 200:
return False, 1.0
preds = [p['prediction'] for p in self.predictions_buffer]
old_preds = preds[:100]
new_preds = preds[-100:]
if method == 'ks_test':
ks_stat, p_value = stats.ks_2samp(old_preds, new_preds)
return p_value < alpha, p_value
return False, 1.0
def check_all_alerts(self):
alerts = []
# 1. Performance degradation
perf = self.calculate_performance_metrics()
if perf and perf['degradation'] < -self.thresholds.get('max_accuracy_drop', 0.05):
alerts.append({
'type': 'performance_degradation',
'severity': 'HIGH',
'detail': f"Accuracy dropped {perf['degradation']:.3f} from baseline"
})
# 2. Feature drift
recent_features = list(self.features_buffer)[-100:]
if recent_features and self.baseline.get('feature_distributions'):
for feature_name in self.baseline['feature_distributions']:
current_vals = [f.get(feature_name) for f in recent_features if f.get(feature_name) is not None]
if current_vals:
psi = self.calculate_psi(
self.baseline['feature_distributions'][feature_name],
current_vals
)
if psi > 0.25:
alerts.append({
'type': 'feature_drift',
'severity': 'MEDIUM',
'feature': feature_name,
'psi': psi
})
# 3. Concept drift
drifted, p_val = self.detect_concept_drift()
if drifted:
alerts.append({
'type': 'concept_drift',
'severity': 'MEDIUM',
'p_value': p_val
})
return alerts
Алерт система
ALERT_CHANNELS = {
'HIGH': ['telegram', 'email', 'pagerduty'],
'MEDIUM': ['telegram', 'email'],
'LOW': ['telegram']
}
async def send_degradation_alert(alert, model_id):
message = f"""
⚠️ ML Model Degradation Alert
Model: {model_id}
Type: {alert['type']}
Severity: {alert['severity']}
Detail: {alert.get('detail', '')}
Time: {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}
Action recommended: Check model retraining system
"""
channels = ALERT_CHANNELS.get(alert['severity'], ['telegram'])
for channel in channels:
await send_notification(channel, message)
Повний стек: Python 3.10+, scipy, numpy, Grafana, Alertmanager. Система розгортається в Docker та інтегрується з будь-яким бекендом. Замовте розробку системи моніторингу вже сьогодні — ми підготуємо рішення під ваші моделі.