ml-model-builder

What this skill does

# ML Model Builder

## Business Case

### Problem Statement
Construction prediction challenges:
- Complex relationships between variables
- Limited historical data utilization
- Need for multiple prediction targets
- Model validation and deployment

### Solution
Comprehensive ML model building framework for construction predictions with data preprocessing, model training, evaluation, and export capabilities.

## Technical Implementation

```python
import pandas as pd
import numpy as np
from typing import Dict, Any, List, Optional, Tuple, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import json
import math


class PredictionTarget(Enum):
    COST = "cost"
    DURATION = "duration"
    RISK_SCORE = "risk_score"
    PRODUCTIVITY = "productivity"
    QUALITY = "quality"


class AlgorithmType(Enum):
    LINEAR_REGRESSION = "linear_regression"
    RIDGE_REGRESSION = "ridge_regression"
    KNN = "knn"
    DECISION_TREE = "decision_tree"
    ENSEMBLE = "ensemble"


class FeatureType(Enum):
    NUMERIC = "numeric"
    CATEGORICAL = "categorical"
    BOOLEAN = "boolean"
    DATE = "date"


@dataclass
class Feature:
    name: str
    feature_type: FeatureType
    importance: float = 0.0
    categories: List[str] = field(default_factory=list)


@dataclass
class ModelMetrics:
    mae: float
    mape: float
    rmse: float
    r_squared: float
    samples: int


@dataclass
class TrainedModel:
    model_id: str
    target: PredictionTarget
    algorithm: AlgorithmType
    features: List[Feature]
    metrics: ModelMetrics
    coefficients: Dict[str, float]
    intercept: float
    trained_at: datetime
    training_samples: int


class MLModelBuilder:
    """Build and train ML models for construction predictions."""

    def __init__(self, project_name: str = "Construction ML"):
        self.project_name = project_name
        self.models: Dict[str, TrainedModel] = {}
        self.feature_stats: Dict[str, Dict[str, float]] = {}
        self.categorical_encodings: Dict[str, Dict[str, int]] = {}

    def prepare_data(self, df: pd.DataFrame,
                     target_column: str,
                     feature_columns: List[str],
                     test_size: float = 0.2) -> Tuple[np.ndarray, np.ndarray,
                                                       np.ndarray, np.ndarray]:
        """Prepare and split data for training."""

        # Handle missing values
        df = df.dropna(subset=[target_column] + feature_columns)

        # Encode categorical features
        X_processed = []

        for col in feature_columns:
            if df[col].dtype == 'object':
                # Categorical encoding
                if col not in self.categorical_encodings:
                    unique_vals = df[col].unique()
                    self.categorical_encodings[col] = {v: i for i, v in enumerate(unique_vals)}

                encoded = df[col].map(self.categorical_encodings[col]).fillna(0)
                X_processed.append(encoded.values)
            else:
                # Numeric - normalize
                values = df[col].values
                if col not in self.feature_stats:
                    self.feature_stats[col] = {
                        'mean': np.mean(values),
                        'std': np.std(values) or 1
                    }

                normalized = (values - self.feature_stats[col]['mean']) / self.feature_stats[col]['std']
                X_processed.append(normalized)

        X = np.column_stack(X_processed)
        y = df[target_column].values

        # Train-test split
        n = len(df)
        indices = np.random.permutation(n)
        test_n = int(n * test_size)

        test_indices = indices[:test_n]
        train_indices = indices[test_n:]

        X_train = X[train_indices]
        X_test = X[test_indices]
        y_train = y[train_indices]
        y_test = y[test_indices]

        return X_train, X_test, y_train, y_test

    def train_linear_regression(self, X: np.ndarray, y: np.ndarray,
                                regularization: float = 0.0) -> Tuple[np.ndarray, float]:
        """Train linear regression model."""

        # Add intercept
        X_with_intercept = np.column_stack([np.ones(len(X)), X])

        if regularization > 0:
            # Ridge regression
            n_features = X_with_intercept.shape[1]
            reg_matrix = regularization * np.eye(n_features)
            reg_matrix[0, 0] = 0  # Don't regularize intercept

            XtX = X_with_intercept.T @ X_with_intercept + reg_matrix
        else:
            XtX = X_with_intercept.T @ X_with_intercept

        try:
            XtX_inv = np.linalg.inv(XtX)
            beta = XtX_inv @ X_with_intercept.T @ y
        except np.linalg.LinAlgError:
            # Use pseudoinverse if singular
            beta = np.linalg.pinv(X_with_intercept) @ y

        return beta[1:], beta[0]

    def train_knn_model(self, X_train: np.ndarray, y_train: np.ndarray,
                        k: int = 5) -> Callable:
        """Create k-NN prediction function."""

        def predict(X_new: np.ndarray) -> np.ndarray:
            predictions = []
            for x in X_new:
                distances = np.sqrt(np.sum((X_train - x) ** 2, axis=1))
                nearest_indices = np.argsort(distances)[:k]
                nearest_values = y_train[nearest_indices]
                predictions.append(np.mean(nearest_values))
            return np.array(predictions)

        return predict

    def calculate_metrics(self, y_true: np.ndarray,
                          y_pred: np.ndarray) -> ModelMetrics:
        """Calculate model performance metrics."""

        residuals = y_true - y_pred
        mae = np.mean(np.abs(residuals))
        mape = np.mean(np.abs(residuals / (y_true + 1e-10))) * 100
        rmse = math.sqrt(np.mean(residuals ** 2))

        # R-squared
        ss_res = np.sum(residuals ** 2)
        ss_tot = np.sum((y_true - np.mean(y_true)) ** 2)
        r_squared = 1 - (ss_res / (ss_tot + 1e-10))

        return ModelMetrics(
            mae=round(mae, 2),
            mape=round(mape, 2),
            rmse=round(rmse, 2),
            r_squared=round(r_squared, 4),
            samples=len(y_true)
        )

    def build_model(self, df: pd.DataFrame,
                    target_column: str,
                    feature_columns: List[str],
                    target_type: PredictionTarget,
                    algorithm: AlgorithmType = AlgorithmType.LINEAR_REGRESSION,
                    model_id: str = None,
                    **kwargs) -> TrainedModel:
        """Build and train a prediction model."""

        model_id = model_id or f"{target_type.value}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"

        # Prepare data
        X_train, X_test, y_train, y_test = self.prepare_data(
            df, target_column, feature_columns,
            test_size=kwargs.get('test_size', 0.2)
        )

        # Train model based on algorithm
        if algorithm == AlgorithmType.LINEAR_REGRESSION:
            coefficients, intercept = self.train_linear_regression(X_train, y_train)
            y_pred = X_test @ coefficients + intercept

        elif algorithm == AlgorithmType.RIDGE_REGRESSION:
            coefficients, intercept = self.train_linear_regression(
                X_train, y_train,
                regularization=kwargs.get('alpha', 1.0)
            )
            y_pred = X_test @ coefficients + intercept

        elif algorithm == AlgorithmType.KNN:
            predict_fn = self.train_knn_model(
                X_train, y_train,
                k=kwargs.get('k', 5)
            )
            y_pred = predict_fn(X_test)
            coefficients = np.zeros(len(feature_columns))
            intercept = np.mean(y_train)

        else:
            # Default to linear
            coefficients, intercept = self.train_linear_regression(X_train, y_train)
            y_pred = X_test @ coefficients + intercept