Schedule Forecaster for Construction

Overview

Predict project completion dates using machine learning models trained on historical data. Forecast delays based on current progress, weather patterns, resource availability, and project characteristics.

Business Case

Accurate schedule forecasting enables:

• Early Warning: Identify potential delays before they impact milestones
• Resource Planning: Adjust staffing based on predicted needs
• Client Communication: Provide reliable completion estimates
• Risk Management: Proactively address schedule risks

Technical Implementation

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, timedelta
from sklearn.ensemble import GradientBoostingRegressor, RandomForestRegressor
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split, TimeSeriesSplit
from sklearn.metrics import mean_absolute_error, mean_squared_error
import warnings
warnings.filterwarnings('ignore')

@dataclass
class ScheduleForecast:
    project_id: str
    forecast_date: datetime
    predicted_completion: datetime
    confidence_interval: Tuple[datetime, datetime]
    delay_probability: float
    delay_days: int
    key_risk_factors: List[str]
    recommended_actions: List[str]

@dataclass
class ProgressSnapshot:
    date: datetime
    planned_progress: float
    actual_progress: float
    earned_value: float
    planned_value: float
    spi: float  # Schedule Performance Index
    cpi: float  # Cost Performance Index

class ConstructionScheduleForecaster:
    """ML-based schedule forecasting for construction projects."""

    def __init__(self):
        self.models: Dict[str, Any] = {}
        self.scalers: Dict[str, StandardScaler] = {}
        self.feature_columns: List[str] = []
        self.is_trained = False

    def prepare_training_data(self, historical_projects: pd.DataFrame) -> Tuple[pd.DataFrame, pd.Series]:
        """Prepare features from historical project data."""

        df = historical_projects.copy()

        # Calculate target: actual delay in days
        df['planned_duration'] = (pd.to_datetime(df['planned_end']) - pd.to_datetime(df['planned_start'])).dt.days
        df['actual_duration'] = (pd.to_datetime(df['actual_end']) - pd.to_datetime(df['actual_start'])).dt.days
        df['delay_days'] = df['actual_duration'] - df['planned_duration']

        # Feature engineering
        features = pd.DataFrame()

        # Project characteristics
        if 'project_type' in df.columns:
            features = pd.concat([features, pd.get_dummies(df['project_type'], prefix='type')], axis=1)

        if 'gross_area' in df.columns:
            features['gross_area'] = df['gross_area']
            features['log_area'] = np.log1p(df['gross_area'])

        if 'contract_value' in df.columns:
            features['contract_value'] = df['contract_value']
            features['value_per_sf'] = df['contract_value'] / df['gross_area'].replace(0, 1)

        if 'planned_duration' in df.columns:
            features['planned_duration'] = df['planned_duration']

        # Complexity indicators
        if 'num_subcontractors' in df.columns:
            features['num_subcontractors'] = df['num_subcontractors']

        if 'num_change_orders' in df.columns:
            features['num_change_orders'] = df['num_change_orders']

        # Historical performance
        if 'contractor_avg_delay' in df.columns:
            features['contractor_avg_delay'] = df['contractor_avg_delay']

        # Seasonal factors
        if 'planned_start' in df.columns:
            start_dates = pd.to_datetime(df['planned_start'])
            features['start_month'] = start_dates.dt.month
            features['start_quarter'] = start_dates.dt.quarter
            features['winter_start'] = ((start_dates.dt.month >= 11) | (start_dates.dt.month <= 2)).astype(int)

        # Location factors
        if 'location_factor' in df.columns:
            features['location_factor'] = df['location_factor']

        self.feature_columns = features.columns.tolist()

        return features.fillna(0), df['delay_days']

    def train_delay_model(self, historical_projects: pd.DataFrame) -> Dict[str, float]:
        """Train model to predict schedule delays."""

        X, y = self.prepare_training_data(historical_projects)

        # 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,
            learning_rate=0.1,
            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)
        rmse = np.sqrt(mean_squared_error(y_test, y_pred))

        # Store model
        self.models['delay'] = model
        self.scalers['delay'] = scaler
        self.is_trained = True

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

        return {
            'mae': mae,
            'rmse': rmse,
            'training_samples': len(X_train),
            'feature_importance': importance
        }

    def train_progress_model(self, progress_data: pd.DataFrame) -> Dict[str, float]:
        """Train model to predict progress based on current trajectory."""

        df = progress_data.copy()

        # Features: current progress, SPI, historical trend
        features = []
        targets = []

        for project_id in df['project_id'].unique():
            project_data = df[df['project_id'] == project_id].sort_values('date')

            for i in range(len(project_data) - 1):
                current = project_data.iloc[i]
                final = project_data.iloc[-1]

                feature = {
                    'current_progress': current['actual_progress'],
                    'planned_progress': current['planned_progress'],
                    'progress_variance': current['actual_progress'] - current['planned_progress'],
                    'spi': current.get('spi', 1.0),
                    'cpi': current.get('cpi', 1.0),
                    'days_elapsed': i,
                    'days_remaining_planned': len(project_data) - i - 1,
                }
                features.append(feature)
                targets.append(final['actual_progress'] - current['actual_progress'])

        X = pd.DataFrame(features)
        y = pd.Series(targets)

        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 = RandomForestRegressor(n_estimators=100, max_depth=10, random_state=42)
        model.fit(X_train_scaled, y_train)

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

        self.models['progress'] = model
        self.scalers['progress'] = scaler

        return {'mae': mae, 'training_samples': len(X_train)}

    def forecast_completion(self, project_data: Dict,
                            current_progress: float,
                            current_date: datetime) -> ScheduleForecast:
        """Forecast project completion date."""

        if not self.is_trained:
            raise ValueError("Model not trained. Call train_delay_model first.")

        # Prepare features
        features = pd.DataFrame([project_data])[self.feature_columns].fillna(0)
        features_scaled = self.scalers['delay'].transform(features)

        # Predict delay
        pre