ml-model-retrainer

What this skill does

# ML Model Retrainer for Construction

## Overview

Automated pipeline for keeping construction ML models up-to-date. Monitor for data drift, trigger retraining when needed, validate performance, and manage model versions.

## Business Case

ML models degrade over time as:
- Market conditions change (material prices, labor rates)
- New construction methods emerge
- Project complexity evolves
- Regional factors shift

Continuous retraining ensures predictions remain accurate.

## Technical Implementation

```python
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Callable
from datetime import datetime, timedelta
import pandas as pd
import numpy as np
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
from sklearn.model_selection import cross_val_score
import pickle
import hashlib
import json
import os

@dataclass
class ModelVersion:
    version_id: str
    model_name: str
    created_at: datetime
    training_samples: int
    metrics: Dict[str, float]
    feature_columns: List[str]
    hyperparameters: Dict[str, Any]
    data_hash: str
    is_active: bool = False

@dataclass
class DriftReport:
    checked_at: datetime
    data_drift_detected: bool
    performance_drift_detected: bool
    drift_score: float
    affected_features: List[str]
    recommendation: str

@dataclass
class RetrainingResult:
    success: bool
    new_version: Optional[ModelVersion]
    old_metrics: Dict[str, float]
    new_metrics: Dict[str, float]
    improvement: Dict[str, float]
    validation_passed: bool
    notes: List[str]

class MLModelRetrainer:
    """Automated ML model retraining for construction predictions."""

    def __init__(self, model_dir: str = "./models"):
        self.model_dir = model_dir
        self.models: Dict[str, Any] = {}
        self.versions: Dict[str, List[ModelVersion]] = {}
        self.active_versions: Dict[str, ModelVersion] = {}
        self.drift_thresholds = {
            'performance_degradation': 0.15,  # 15% degradation triggers retrain
            'data_drift_score': 0.3,
            'min_new_samples': 50,
        }

        os.makedirs(model_dir, exist_ok=True)

    def register_model(self, model_name: str, model: Any,
                       feature_columns: List[str],
                       hyperparameters: Dict = None) -> ModelVersion:
        """Register a new model for management."""
        version = ModelVersion(
            version_id=f"{model_name}-v{datetime.now().strftime('%Y%m%d%H%M%S')}",
            model_name=model_name,
            created_at=datetime.now(),
            training_samples=0,
            metrics={},
            feature_columns=feature_columns,
            hyperparameters=hyperparameters or {},
            data_hash="",
            is_active=True
        )

        self.models[model_name] = model
        if model_name not in self.versions:
            self.versions[model_name] = []
        self.versions[model_name].append(version)
        self.active_versions[model_name] = version

        return version

    def calculate_data_hash(self, data: pd.DataFrame) -> str:
        """Calculate hash of training data for change detection."""
        data_str = data.to_json()
        return hashlib.md5(data_str.encode()).hexdigest()

    def detect_data_drift(self, model_name: str,
                          reference_data: pd.DataFrame,
                          current_data: pd.DataFrame) -> DriftReport:
        """Detect data drift between reference and current data."""

        drift_scores = {}
        affected_features = []

        # Compare distributions for each feature
        for col in reference_data.select_dtypes(include=[np.number]).columns:
            if col in current_data.columns:
                ref_mean = reference_data[col].mean()
                ref_std = reference_data[col].std()
                cur_mean = current_data[col].mean()
                cur_std = current_data[col].std()

                # Normalized difference
                if ref_std > 0:
                    mean_drift = abs(cur_mean - ref_mean) / ref_std
                    std_drift = abs(cur_std - ref_std) / ref_std
                    drift_scores[col] = (mean_drift + std_drift) / 2

                    if drift_scores[col] > 0.5:
                        affected_features.append(f"{col} (drift: {drift_scores[col]:.2f})")

        avg_drift = np.mean(list(drift_scores.values())) if drift_scores else 0
        data_drift_detected = avg_drift > self.drift_thresholds['data_drift_score']

        recommendation = "No action needed"
        if data_drift_detected:
            recommendation = "Data drift detected - consider retraining"
        elif avg_drift > self.drift_thresholds['data_drift_score'] * 0.7:
            recommendation = "Minor drift detected - monitor closely"

        return DriftReport(
            checked_at=datetime.now(),
            data_drift_detected=data_drift_detected,
            performance_drift_detected=False,
            drift_score=avg_drift,
            affected_features=affected_features,
            recommendation=recommendation
        )

    def evaluate_model_performance(self, model_name: str,
                                   test_data: pd.DataFrame,
                                   target_col: str) -> Dict[str, float]:
        """Evaluate current model performance on new data."""

        if model_name not in self.models:
            raise ValueError(f"Model {model_name} not registered")

        model = self.models[model_name]
        version = self.active_versions[model_name]

        # Prepare features
        X = test_data[version.feature_columns].fillna(0)
        y = test_data[target_col]

        # Predict
        y_pred = model.predict(X)

        # Calculate metrics
        metrics = {
            'mae': mean_absolute_error(y, y_pred),
            'rmse': np.sqrt(mean_squared_error(y, y_pred)),
            'r2': r2_score(y, y_pred),
            'mape': np.mean(np.abs((y - y_pred) / y.replace(0, 1))) * 100,
        }

        return metrics

    def check_performance_drift(self, model_name: str,
                                 baseline_metrics: Dict[str, float],
                                 current_metrics: Dict[str, float]) -> DriftReport:
        """Check if model performance has degraded."""

        # Calculate degradation for each metric
        degradation = {}
        for metric in ['mae', 'rmse']:
            if metric in baseline_metrics and metric in current_metrics:
                # Higher is worse for these metrics
                change = (current_metrics[metric] - baseline_metrics[metric]) / baseline_metrics[metric]
                degradation[metric] = change

        for metric in ['r2']:
            if metric in baseline_metrics and metric in current_metrics:
                # Lower is worse for R2
                change = (baseline_metrics[metric] - current_metrics[metric]) / abs(baseline_metrics[metric])
                degradation[metric] = change

        avg_degradation = np.mean(list(degradation.values())) if degradation else 0
        performance_drift = avg_degradation > self.drift_thresholds['performance_degradation']

        affected = [f"{m}: {d:+.1%}" for m, d in degradation.items() if d > 0.1]

        recommendation = "No action needed"
        if performance_drift:
            recommendation = "Performance degraded - retraining recommended"
        elif avg_degradation > self.drift_thresholds['performance_degradation'] * 0.5:
            recommendation = "Performance declining - monitor closely"

        return DriftReport(
            checked_at=datetime.now(),
            data_drift_detected=False,
            performance_drift_detected=performance_drift,
            drift_score=avg_degradation,
            affected_features=affected,
            recommendation=recommendation
        )

    def retrain_model(self, model_name: str,
                      training_data: pd.DataFram