Risk Assessment with Machine Learning

Overview

This skill implements ML-based risk assessment for construction projects. Predict potential risks before they occur and prioritize mitigation strategies based on data-driven insights.

Risk Categories:

• Schedule Risk: Delays, critical path impacts
• Cost Risk: Budget overruns, change orders
• Safety Risk: Incident probability, hazard identification
• Quality Risk: Defects, rework probability

Quick Start

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier, GradientBoostingRegressor
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

# Load historical project data
projects = pd.read_csv("project_history.csv")

# Features for risk prediction
features = ['project_size_m2', 'budget_usd', 'duration_days',
            'complexity_score', 'team_size', 'similar_projects_exp']

X = projects[features]
y_delay = projects['had_delay']  # Binary: 1=delay, 0=on-time

# Train risk model
X_train, X_test, y_train, y_test = train_test_split(X, y_delay, test_size=0.2)

model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)

# Predict risk for new project
new_project = [[5000, 2000000, 365, 3, 50, 5]]
risk_probability = model.predict_proba(new_project)[0][1]
print(f"Delay Risk: {risk_probability:.1%}")

Comprehensive Risk Model

Risk Assessment Framework

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier, GradientBoostingRegressor
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.model_selection import cross_val_score
from dataclasses import dataclass
from typing import Dict, List, Optional
import joblib

@dataclass
class RiskPrediction:
    category: str
    probability: float
    severity: str
    impact_days: Optional[float]
    impact_cost: Optional[float]
    confidence: float
    contributing_factors: List[str]
    recommended_actions: List[str]

class ConstructionRiskAssessor:
    """ML-based construction risk assessment"""

    def __init__(self):
        self.models = {}
        self.scalers = {}
        self.encoders = {}
        self.feature_importance = {}

    def prepare_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Prepare features for training/prediction"""
        features = df.copy()

        # Encode categorical variables
        categorical_cols = features.select_dtypes(include=['object']).columns
        for col in categorical_cols:
            if col not in self.encoders:
                self.encoders[col] = LabelEncoder()
                features[col] = self.encoders[col].fit_transform(features[col].astype(str))
            else:
                features[col] = self.encoders[col].transform(features[col].astype(str))

        # Handle missing values
        features = features.fillna(features.median())

        return features

    def train_delay_model(self, df: pd.DataFrame, target_col: str = 'delay_days'):
        """Train schedule delay prediction model"""
        features = self.prepare_features(df.drop(columns=[target_col]))
        target = df[target_col]

        # Binary classification: delay or not
        target_binary = (target > 0).astype(int)

        # Scale features
        self.scalers['delay'] = StandardScaler()
        X_scaled = self.scalers['delay'].fit_transform(features)

        # Train model
        self.models['delay_classifier'] = RandomForestClassifier(
            n_estimators=100,
            max_depth=10,
            random_state=42
        )
        self.models['delay_classifier'].fit(X_scaled, target_binary)

        # Train regression for delay magnitude
        delayed_mask = target > 0
        if delayed_mask.sum() > 10:
            self.models['delay_regressor'] = GradientBoostingRegressor(
                n_estimators=100,
                max_depth=5,
                random_state=42
            )
            self.models['delay_regressor'].fit(
                X_scaled[delayed_mask],
                target[delayed_mask]
            )

        # Store feature importance
        self.feature_importance['delay'] = dict(zip(
            features.columns,
            self.models['delay_classifier'].feature_importances_
        ))

        return self._evaluate_model('delay_classifier', X_scaled, target_binary)

    def train_cost_overrun_model(self, df: pd.DataFrame,
                                  target_col: str = 'cost_overrun_pct'):
        """Train cost overrun prediction model"""
        features = self.prepare_features(df.drop(columns=[target_col]))
        target = df[target_col]

        # Binary: overrun or not
        target_binary = (target > 0).astype(int)

        self.scalers['cost'] = StandardScaler()
        X_scaled = self.scalers['cost'].fit_transform(features)

        self.models['cost_classifier'] = RandomForestClassifier(
            n_estimators=100,
            max_depth=10,
            random_state=42
        )
        self.models['cost_classifier'].fit(X_scaled, target_binary)

        # Regression for magnitude
        overrun_mask = target > 0
        if overrun_mask.sum() > 10:
            self.models['cost_regressor'] = GradientBoostingRegressor(
                n_estimators=100,
                max_depth=5,
                random_state=42
            )
            self.models['cost_regressor'].fit(
                X_scaled[overrun_mask],
                target[overrun_mask]
            )

        self.feature_importance['cost'] = dict(zip(
            features.columns,
            self.models['cost_classifier'].feature_importances_
        ))

        return self._evaluate_model('cost_classifier', X_scaled, target_binary)

    def train_safety_model(self, df: pd.DataFrame,
                           target_col: str = 'incident_occurred'):
        """Train safety incident prediction model"""
        features = self.prepare_features(df.drop(columns=[target_col]))
        target = df[target_col]

        self.scalers['safety'] = StandardScaler()
        X_scaled = self.scalers['safety'].fit_transform(features)

        self.models['safety_classifier'] = RandomForestClassifier(
            n_estimators=100,
            max_depth=8,
            class_weight='balanced',  # Handle imbalanced data
            random_state=42
        )
        self.models['safety_classifier'].fit(X_scaled, target)

        self.feature_importance['safety'] = dict(zip(
            features.columns,
            self.models['safety_classifier'].feature_importances_
        ))

        return self._evaluate_model('safety_classifier', X_scaled, target)

    def _evaluate_model(self, model_name: str, X: np.ndarray, y: np.ndarray) -> Dict:
        """Evaluate model with cross-validation"""
        model = self.models[model_name]
        scores = cross_val_score(model, X, y, cv=5, scoring='accuracy')

        return {
            'model': model_name,
            'accuracy_mean': scores.mean(),
            'accuracy_std': scores.std()
        }

    def predict_risks(self, project_data: Dict) -> List[RiskPrediction]:
        """Predict all risks for a project"""
        df = pd.DataFrame([project_data])
        features = self.prepare_features(df)
        predictions = []

        # Schedule risk
        if 'delay_classifier' in self.models:
            X_delay = self.scalers['delay'].transform(features)
            delay_prob = self.models['delay_classifier'].predict_proba(X_delay)[0][1]

            delay_days = None
            if delay_prob > 0.5 and 'delay_regressor' in self.models:
                delay_days = self.models['delay_regressor'].predict(X_delay)[0]

            predictions.append(RiskPrediction(
                category='Schedule',
                probability=delay_prob,
                severity=self._get_severity(delay_prob),
                impact_days=delay_days,
                i