risk-assessment-ml

name	risk-assessment-ml
description	Apply machine learning for construction project risk assessment. Predict schedule delays, cost overruns, and safety incidents using historical data and project characteristics.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🚀","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

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

        
        categorical_cols = features.select_dtypes(include=[]).columns
         col  categorical_cols:
             col   .encoders:
                .encoders[col] = LabelEncoder()
                features[col] = .encoders[col].fit_transform(features[col].astype())
            :
                features[col] = .encoders[col].transform(features[col].astype())

        
        features = features.fillna(features.median())

         features

     ():
        
        features = .prepare_features(df.drop(columns=[target_col]))
        target = df[target_col]

        
        target_binary = (target > ).astype()

        
        .scalers[] = StandardScaler()
        X_scaled = .scalers[].fit_transform(features)

        
        .models[] = RandomForestClassifier(
            n_estimators=,
            max_depth=,
            random_state=
        )
        .models[].fit(X_scaled, target_binary)

        
        delayed_mask = target > 
         delayed_mask.() > :
            .models[] = GradientBoostingRegressor(
                n_estimators=,
                max_depth=,
                random_state=
            )
            .models[].fit(
                X_scaled[delayed_mask],
                target[delayed_mask]
            )

        
        .feature_importance[] = ((
            features.columns,
            .models[].feature_importances_
        ))

         ._evaluate_model(, X_scaled, target_binary)

     ():
        
        features = .prepare_features(df.drop(columns=[target_col]))
        target = df[target_col]

        
        target_binary = (target > ).astype()

        .scalers[] = StandardScaler()
        X_scaled = .scalers[].fit_transform(features)

        .models[] = RandomForestClassifier(
            n_estimators=,
            max_depth=,
            random_state=
        )
        .models[].fit(X_scaled, target_binary)

        
        overrun_mask = target > 
         overrun_mask.() > :
            .models[] = GradientBoostingRegressor(
                n_estimators=,
                max_depth=,
                random_state=
            )
            .models[].fit(
                X_scaled[overrun_mask],
                target[overrun_mask]
            )

        .feature_importance[] = ((
            features.columns,
            .models[].feature_importances_
        ))

         ._evaluate_model(, X_scaled, target_binary)

     ():
        
        features = .prepare_features(df.drop(columns=[target_col]))
        target = df[target_col]

        .scalers[] = StandardScaler()
        X_scaled = .scalers[].fit_transform(features)

        .models[] = RandomForestClassifier(
            n_estimators=,
            max_depth=,
            class_weight=,  
            random_state=
        )
        .models[].fit(X_scaled, target)

        .feature_importance[] = ((
            features.columns,
            .models[].feature_importances_
        ))

         ._evaluate_model(, X_scaled, target)

     () -> :
        
        model = .models[model_name]
        scores = cross_val_score(model, X, y, cv=, scoring=)

         {
            : model_name,
            : scores.mean(),
            : scores.std()
        }

     () -> [RiskPrediction]:
        
        df = pd.DataFrame([project_data])
        features = .prepare_features(df)
        predictions = []

        
           .models:
            X_delay = .scalers[].transform(features)
            delay_prob = .models[].predict_proba(X_delay)[][]

            delay_days = 
             delay_prob >     .models:
                delay_days = .models[].predict(X_delay)[]

            predictions.append(RiskPrediction(
                category=,
                probability=delay_prob,
                severity=._get_severity(delay_prob),
                impact_days=delay_days,
                impact_cost=delay_days * project_data.get(, )  delay_days  ,
                confidence=,
                contributing_factors=._get_top_factors(, features.iloc[]),
                recommended_actions=._get_delay_actions(delay_prob)
            ))

        
           .models:
            X_cost = .scalers[].transform(features)
            cost_prob = .models[].predict_proba(X_cost)[][]

            overrun_pct = 
             cost_prob >     .models:
                overrun_pct = .models[].predict(X_cost)[]

            predictions.append(RiskPrediction(
                category=,
                probability=cost_prob,
                severity=._get_severity(cost_prob),
                impact_days=,
                impact_cost=project_data.get(, ) * overrun_pct /   overrun_pct  ,
                confidence=,
                contributing_factors=._get_top_factors(, features.iloc[]),
                recommended_actions=._get_cost_actions(cost_prob)
            ))

        
           .models:
            X_safety = .scalers[].transform(features)
            safety_prob = .models[].predict_proba(X_safety)[][]

            predictions.append(RiskPrediction(
                category=,
                probability=safety_prob,
                severity=._get_severity(safety_prob),
                impact_days=,
                impact_cost=,
                confidence=,
                contributing_factors=._get_top_factors(, features.iloc[]),
                recommended_actions=._get_safety_actions(safety_prob)
            ))

         predictions

     () -> :
         probability >= :
             
         probability >= :
             
        :
             

     () -> []:
        
        importance = .feature_importance.get(risk_type, {})
        sorted_factors = (importance.items(), key= x: x[], reverse=)
         [f[]  f  sorted_factors[:]]

     () -> []:
        actions = []
         probability > :
            actions.extend([
                ,
                ,
                
            ])
         probability > :
            actions.extend([
                ,
                
            ])
        :
            actions.append()
         actions

     () -> []:
        actions = []
         probability > :
            actions.extend([
                ,
                ,
                
            ])
         probability > :
            actions.extend([
                ,
                
            ])
        :
            actions.append()
         actions

     () -> []:
        actions = []
         probability > :
            actions.extend([
                ,
                ,
                
            ])
         probability > :
            actions.extend([
                ,
                
            ])
        :
            actions.append()
         actions

     ():
        
        joblib.dump({
            : .models,
            : .scalers,
            : .encoders,
            : .feature_importance
        }, path)

     ():
        
        data = joblib.load(path)
        .models = data[]
        .scalers = data[]
        .encoders = data[]
        .feature_importance = data[]

Feature Engineering

Project Risk Features

def engineer_risk_features(project_data: pd.DataFrame) -> pd.DataFrame:
    """Create features for risk prediction"""
    df = project_data.copy()

    # Size and complexity metrics
    df['cost_per_sqm'] = df['budget'] / df['area_sqm']
    df['duration_per_sqm'] = df['duration_days'] / df['area_sqm']

    # Team metrics
    df['workers_per_1000sqm'] = df['peak_workers'] / (df['area_sqm'] / 1000)

    # Weather exposure
    df['outdoor_work_pct'] = df['outdoor_activities'] / df['total_activities']

    # Contract complexity
    df['subcontractor_ratio'] = df['num_subcontractors'] / df['total_contractors']

    # Experience factors
    df['team_avg_experience'] = df['total_team_years'] / df['team_size']

    # Season risk
    df['winter_months'] = df.apply(
        lambda r: count_winter_months(r['start_date'], r['end_date']), axis=1
    )

    # Location risk
    df['urban_complexity'] = df['location_type'].map({
        'rural': 1, 'suburban': 2, 'urban': 3, 'downtown': 4
    })

     df

 ():
    
    winter_months = [, , , , ]
    months = pd.date_range(start_date, end_date, freq=)
     (  m  months  m.month  winter_months)

Risk Report Generation

def generate_risk_report(assessor: ConstructionRiskAssessor,
                        project_data: Dict,
                        output_path: str):
    """Generate comprehensive risk assessment report"""
    predictions = assessor.predict_risks(project_data)

    with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
        # Summary
        summary = pd.DataFrame([{
            'Category': p.category,
            'Risk Level': p.severity,
            'Probability': f"{p.probability:.1%}",
            'Impact Days': p.impact_days,
            'Impact Cost': p.impact_cost,
            'Top Factor': p.contributing_factors[0] if p.contributing_factors else ''
        } for p in predictions])
        summary.to_excel(writer, sheet_name='Summary', index=False)

        # Detailed recommendations
        actions = []
        for p in predictions:
            for action in p.recommended_actions:
                actions.append({
                    'Category': p.category,
                    'Risk Level': p.severity,
                    'Recommended Action': action
                })
        pd.DataFrame(actions).to_excel(writer, sheet_name='Actions', index=False)

    return output_path

Quick Reference

Risk Type	Key Features	Model Type
Schedule	Duration, complexity, weather	Random Forest
Cost	Budget, scope changes, market	Gradient Boosting
Safety	Work type, team experience	Random Forest
Quality	Supervision, materials	Logistic Regression

Resources

• Scikit-learn: https://scikit-learn.org
• DDC Website: https://datadrivenconstruction.io

Next Steps

• See cost-prediction for detailed cost modeling
• See 4d-simulation for schedule analysis
• See data-visualization for risk dashboards