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
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