predictive-analytics-construction

What this skill does

# Predictive Analytics for Construction

## Overview

Use historical project data to predict future outcomes: cost overruns, schedule delays, quality issues, and risks. Apply machine learning models tailored for construction industry patterns.

## Business Case

Predictive analytics enables proactive project management:
- **Early Warning**: Identify projects likely to overrun before it happens
- **Resource Optimization**: Allocate resources based on predicted needs
- **Risk Mitigation**: Focus on high-risk areas early
- **Better Estimates**: Learn from historical accuracy

## Technical Implementation

```python
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Tuple
import pandas as pd
import numpy as np
from datetime import datetime
from sklearn.ensemble import RandomForestRegressor, GradientBoostingClassifier
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import mean_absolute_error, accuracy_score, classification_report
import warnings
warnings.filterwarnings('ignore')

@dataclass
class PredictionResult:
    prediction: float
    confidence: float
    prediction_type: str
    features_used: List[str]
    feature_importance: Dict[str, float]
    comparable_projects: List[str]
    risk_factors: List[str]

@dataclass
class ModelMetrics:
    model_name: str
    accuracy: float
    mae: float
    feature_importance: Dict[str, float]
    training_samples: int
    last_trained: datetime

class ConstructionPredictiveAnalytics:
    """Predictive analytics for construction projects."""

    def __init__(self):
        self.models: Dict[str, Any] = {}
        self.scalers: Dict[str, StandardScaler] = {}
        self.encoders: Dict[str, LabelEncoder] = {}
        self.metrics: Dict[str, ModelMetrics] = {}
        self.feature_columns: Dict[str, List[str]] = {}

    def prepare_features(self, df: pd.DataFrame, target_col: str) -> Tuple[pd.DataFrame, pd.Series]:
        """Prepare features for model training."""
        # Separate numeric and categorical columns
        numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
        categorical_cols = df.select_dtypes(include=['object', 'category']).columns.tolist()

        # Remove target from features
        if target_col in numeric_cols:
            numeric_cols.remove(target_col)
        if target_col in categorical_cols:
            categorical_cols.remove(target_col)

        # Encode categorical variables
        df_encoded = df.copy()
        for col in categorical_cols:
            if col not in self.encoders:
                self.encoders[col] = LabelEncoder()
                df_encoded[col] = self.encoders[col].fit_transform(df[col].astype(str))
            else:
                df_encoded[col] = self.encoders[col].transform(df[col].astype(str))

        feature_cols = numeric_cols + categorical_cols
        X = df_encoded[feature_cols].fillna(0)
        y = df[target_col]

        return X, y, feature_cols

    def train_cost_overrun_model(self, historical_data: pd.DataFrame) -> ModelMetrics:
        """Train model to predict cost overrun percentage."""
        # Expected columns: project_type, original_estimate, gross_area, duration_months,
        # num_change_orders, complexity_score, contractor_experience, final_cost

        required_cols = ['original_estimate', 'final_cost']
        if not all(col in historical_data.columns for col in required_cols):
            raise ValueError(f"Missing required columns: {required_cols}")

        # Calculate overrun percentage
        df = historical_data.copy()
        df['overrun_pct'] = ((df['final_cost'] - df['original_estimate']) / df['original_estimate']) * 100

        # Prepare features
        feature_cols = [col for col in df.columns if col not in ['final_cost', 'overrun_pct', 'project_id', 'project_name']]
        X, y, used_features = self.prepare_features(df[feature_cols + ['overrun_pct']], 'overrun_pct')

        # Split data
        X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

        # Scale features
        scaler = StandardScaler()
        X_train_scaled = scaler.fit_transform(X_train)
        X_test_scaled = scaler.transform(X_test)

        # Train model
        model = GradientBoostingRegressor(n_estimators=100, max_depth=5, random_state=42)
        model.fit(X_train_scaled, y_train)

        # Evaluate
        y_pred = model.predict(X_test_scaled)
        mae = mean_absolute_error(y_test, y_pred)

        # Cross-validation
        cv_scores = cross_val_score(model, X_train_scaled, y_train, cv=5, scoring='neg_mean_absolute_error')

        # Feature importance
        importance = dict(zip(used_features, model.feature_importances_))

        # Store model
        self.models['cost_overrun'] = model
        self.scalers['cost_overrun'] = scaler
        self.feature_columns['cost_overrun'] = used_features

        metrics = ModelMetrics(
            model_name='cost_overrun',
            accuracy=1 - (mae / df['overrun_pct'].std()),
            mae=mae,
            feature_importance=importance,
            training_samples=len(X_train),
            last_trained=datetime.now()
        )
        self.metrics['cost_overrun'] = metrics

        return metrics

    def train_schedule_delay_model(self, historical_data: pd.DataFrame) -> ModelMetrics:
        """Train model to predict schedule delay probability."""
        df = historical_data.copy()

        # Binary classification: was project delayed?
        df['was_delayed'] = (df['actual_duration'] > df['planned_duration']).astype(int)

        feature_cols = [col for col in df.columns
                       if col not in ['actual_duration', 'was_delayed', 'project_id', 'project_name']]

        X, y, used_features = self.prepare_features(df[feature_cols + ['was_delayed']], 'was_delayed')

        X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

        scaler = StandardScaler()
        X_train_scaled = scaler.fit_transform(X_train)
        X_test_scaled = scaler.transform(X_test)

        model = GradientBoostingClassifier(n_estimators=100, max_depth=5, random_state=42)
        model.fit(X_train_scaled, y_train)

        y_pred = model.predict(X_test_scaled)
        accuracy = accuracy_score(y_test, y_pred)

        importance = dict(zip(used_features, model.feature_importances_))

        self.models['schedule_delay'] = model
        self.scalers['schedule_delay'] = scaler
        self.feature_columns['schedule_delay'] = used_features

        metrics = ModelMetrics(
            model_name='schedule_delay',
            accuracy=accuracy,
            mae=0,
            feature_importance=importance,
            training_samples=len(X_train),
            last_trained=datetime.now()
        )
        self.metrics['schedule_delay'] = metrics

        return metrics

    def predict_cost_overrun(self, project_data: Dict) -> PredictionResult:
        """Predict cost overrun for a new project."""
        if 'cost_overrun' not in self.models:
            raise ValueError("Cost overrun model not trained. Call train_cost_overrun_model first.")

        model = self.models['cost_overrun']
        scaler = self.scalers['cost_overrun']
        features = self.feature_columns['cost_overrun']

        # Prepare input
        input_df = pd.DataFrame([project_data])

        # Encode categorical
        for col in input_df.select_dtypes(include=['object']).columns:
            if col in self.encoders:
                input_df[col] = self.encoders[col].transform(input_df[col].astype(str))

        # Ensure all features present
        for feat in features:
            if feat not in input_df.columns:
                input_df[feat] = 0

        X = input_df[features].fillna(0)
        X_scaled = scaler.transform(X)

        prediction = model.predict(X_scaled)[