Сколько автообучались модели на твоём проекте?

Автоматическое переобучение моделей

Автоматическое переобучение (retraining) - это критический аспект ML в production. Без него модели деградируют со временем.

Проблема: Model Drift

Модель, обученная на старых данных, теряет точность когда распределение меняется:

from datetime import datetime
from sklearn.metrics import accuracy_score

accuracy_by_date = {
    "2024-01-01": 0.87,
    "2024-02-01": 0.82,
    "2024-03-01": 0.75,
    "2024-04-01": 0.68,
}

# Видим деградацию - нужно переобучение

1. Когда срабатывает переобучение

Вариант 1: По времени (Schedule-based)

Переобучаем каждый день/неделю/месяц:

from apscheduler.schedulers.background import BackgroundScheduler
import logging

scheduler = BackgroundScheduler()

def retrain_model():
    logging.info("Starting scheduled retraining...")
    X_new, y_new = load_production_data()
    model = train_model(X_new, y_new)
    save_model(model, "model_latest.pkl")
    logging.info("Retraining complete")

scheduler.add_job(retrain_model, 'cron', hour=2, minute=0)
scheduler.start()

Плюсы: простая реализация Минусы: может быть избыточным

Вариант 2: По метрикам (Metric-based)

Переобучаем когда точность упала ниже порога:

from sklearn.metrics import accuracy_score

class ModelMonitor:
    def __init__(self, baseline_accuracy=0.87, drop_threshold=0.05):
        self.baseline = baseline_accuracy
        self.threshold = drop_threshold
    
    def check_and_retrain(self, y_true, y_pred):
        current_accuracy = accuracy_score(y_true, y_pred)
        
        if current_accuracy < self.baseline - self.threshold:
            logging.info(f"Alert! Accuracy dropped to {current_accuracy:.3f}")
            self.retrain()
        
        return current_accuracy
    
    def retrain(self):
        logging.info("Triggering retraining...")
        X_new, y_new = load_recent_data(days=30)
        model = train_model(X_new, y_new)
        save_model(model)

monitor = ModelMonitor(baseline_accuracy=0.87)
accuracy = monitor.check_and_retrain(y_true, y_pred)

Плюсы: срабатывает когда нужно Минусы: нужен хороший мониторинг

Вариант 3: По дрейфу данных (Data Drift Detection)

from scipy.stats import ks_2samp

class DataDriftDetector:
    def __init__(self, reference_data):
        self.reference = reference_data
    
    def detect_drift(self, new_data, p_threshold=0.05):
        drifts = {}
        for col in self.reference.columns:
            statistic, p_value = ks_2samp(
                self.reference[col],
                new_data[col]
            )
            
            if p_value < p_threshold:
                drifts[col] = p_value
                logging.info(f"Drift detected in {col}")
        
        return drifts

detector = DataDriftDetector(X_train)
drifts = detector.detect_drift(X_production)

if drifts:
    logging.info("Data drift detected! Retraining...")
    retrain_model()

2. Полный pipeline автоматического переобучения

import logging
from datetime import datetime
import joblib
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

logging.basicConfig(level=logging.INFO)

class AutoRetrainingPipeline:
    def __init__(self, model_path="model.pkl", threshold=0.05):
        self.model_path = model_path
        self.threshold = threshold
        self.model = joblib.load(model_path)
        self.baseline_accuracy = self.get_baseline()
        self.last_retrain = datetime.now()
    
    def get_baseline(self):
        X_val, y_val = load_validation_data()
        y_pred = self.model.predict(X_val)
        return accuracy_score(y_val, y_pred)
    
    def predict(self, X):
        y_pred = self.model.predict(X)
        self._check_performance_async(X, y_pred)
        return y_pred
    
    def _check_performance_async(self, X, y_pred):
        if (datetime.now() - self.last_retrain).seconds < 3600:
            return
        
        y_true = get_ground_truth(X)
        if y_true is not None:
            current_accuracy = accuracy_score(y_true, y_pred)
            
            if current_accuracy < self.baseline_accuracy - self.threshold:
                logging.warning(f"Accuracy drop: {current_accuracy:.3f}")
                self._trigger_retraining()
    
    def _trigger_retraining(self):
        logging.info("Starting automated retraining...")
        
        try:
            X_new, y_new = self._collect_new_data()
            
            if len(X_new) < 1000:
                logging.warning("Not enough data for retraining")
                return
            
            new_model = self._train_new_model(X_new, y_new)
            X_val, y_val = load_validation_data()
            
            new_accuracy = accuracy_score(y_val, new_model.predict(X_val))
            old_accuracy = accuracy_score(y_val, self.model.predict(X_val))
            
            if new_accuracy > old_accuracy:
                logging.info(f"New model better: {new_accuracy:.3f} > {old_accuracy:.3f}")
                self._deploy_new_model(new_model)
            else:
                logging.info(f"New model worse: {new_accuracy:.3f} < {old_accuracy:.3f}")
        
        except Exception as e:
            logging.error(f"Retraining failed: {e}")
        
        finally:
            self.last_retrain = datetime.now()
    
    def _collect_new_data(self):
        df = load_recent_data(days=30)
        X = df.drop(columns=["target"])
        y = df["target"]
        return X, y
    
    def _train_new_model(self, X, y):
        logging.info(f"Training on {len(X)} new samples...")
        model = RandomForestClassifier(n_estimators=100, n_jobs=-1, random_state=42)
        model.fit(X, y)
        return model
    
    def _deploy_new_model(self, new_model):
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_path = f"models/model_backup_{timestamp}.pkl"
        joblib.dump(self.model, backup_path)
        joblib.dump(new_model, self.model_path)
        self.model = new_model

pipeline = AutoRetrainingPipeline(model_path="models/current_model.pkl", threshold=0.05)
y_pred = pipeline.predict(X_production)

3. Incremental Learning

from sklearn.linear_model import SGDClassifier

class IncrementalLearningModel:
    def __init__(self):
        self.model = SGDClassifier(loss="log_loss", warm_start=True)
    
    def partial_fit(self, X_batch, y_batch):
        self.model.partial_fit(X_batch, y_batch, classes=[0, 1])
    
    def predict(self, X):
        return self.model.predict(X)

incremental_model = IncrementalLearningModel()
X_hour, y_hour = load_data(hours=1)
incremental_model.partial_fit(X_hour, y_hour)

4. Ключевые стратегии

• Schedule-based: Переобучаем по расписанию (простой)
• Metric-based: Переобучаем когда метрики падают
• Data drift detection: Переобучаем когда распределение меняется
• Incremental learning: Обновляем модель постепенно
• Ensemble retraining: Держим несколько моделей, выбираем лучшую

5. Best Practices

• Комбинируй schedule + metric-based подход
• Всегда делай A/B тестирование новой модели
• Бэкапируй старые версии
• Мониторь дрейф данных
• Логируй все переобучения
• Имей возможность откатиться на предыдущую версию

Итог

Автоматическое переобучение критично для production ML потому что данные меняются, модели деградируют. Лучший подход - комбинация мониторинга метрик, детекции дрейфа и регулярного переобучения.