tools-only 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
predicted_delay = self.models['delay'].predict(features_scaled)[0]
# Get prediction interval (using model variance)
tree_predictions = np.array([
tree.predict(features_scaled)[0]
for tree in self.models['delay'].estimators_
])
delay_std = np.std(tree_predictions)
# Calculate dates
planned_end = pd.to_datetime(project_data.get('planned_end'))
predicted_completion = planned_end + timedelta(days=int(predicted_delay))
confidence_low = planned_end + timedelta(days=int(predicted_delay - 1.96 * delay_std))
confidence_high = planned_end + timedelta(days=int(predicted_delay + 1.96 * delay_std))
# Calculate delay probability
delay_probability = 1 / (1 + np.exp(-predicted_delay / 30)) # Sigmoid transform
# Identify risk factors
risk_factors = self._identify_risk_factors(project_data, features_scaled)
# Generate recommendations
recommendations = self._generate_recommendations(
predicted_delay, current_progress, project_data
)
return ScheduleForecast(
project_id=project_data.get('project_id', 'Unknown'),
forecast_date=current_date,
predicted_completion=predicted_completion,
confidence_interval=(confidence_low, confidence_high),
delay_probability=delay_probability,
delay_days=int(predicted_delay),
key_risk_factors=risk_factors,
recommended_actions=recommendations
)
def _identify_risk_factors(self, project_data: Dict, features_scaled: np.ndarray) -> List[str]:
"""Identify key risk factors for the project."""
risk_factors = []
importance = dict(zip(self.feature_columns, self.models['delay'].feature_importances_))
top_features = sorted(importance.items(), key=lambda x: -x[1])[:5]
for feat, imp in top_features:
if imp > 0.1:
value = project_data.get(feat)
if value:
risk_factors.append(f"{feat}: {value} (impact: {imp:.1%})")
# Add context-specific risks
if project_data.get('num_change_orders', 0) > 10:
risk_factors.append("High number of change orders")
if project_data.get('winter_start'):
risk_factors.append("Winter start increases weather risk")
return risk_factors[:5]
def _generate_recommendations(self, predicted_delay: float,
current_progress: float,
project_data: Dict) -> List[str]:
"""Generate actionable recommendations."""
recommendations = []
if predicted_delay > 30:
recommendations.append("Consider schedule compression techniques (crashing/fast-tracking)")
recommendations.append("Evaluate additional resource allocation")
if predicted_delay > 0 and current_progress < 0.5:
recommendations.append("Review critical path activities for optimization")
if project_data.get('spi', 1.0) < 0.9:
recommendations.append("Schedule Performance Index is low - investigate root causes")
if project_data.get('num_change_orders', 0) > 5:
recommendations.append("High change order volume - improve change management process")
if not recommendations:
recommendations.append("Project on track - maintain current pace")
return recommendations
def update_forecast_with_progress(self, project_id: str,
progress_history: List[ProgressSnapshot],
project_data: Dict) -> ScheduleForecast:
"""Update forecast based on current progress trajectory."""
if len(progress_history) < 2:
return self.forecast_completion(project_data, 0, datetime.now())
# Calculate trends
recent = progress_history[-5:]
progress_rates = []
for i in range(1, len(recent)):
days = (recent[i].date - recent[i-1].date).days
if days > 0:
rate = (recent[i].actual_progress - recent[i-1].actual_progress) / days
progress_rates.append(rate)
avg_rate = np.mean(progress_rates) if progress_rates else 0
current_progress = progress_history[-1].actual_progress
# Estimate remaining duration
remaining_progress = 100 - current_progress
if avg_rate > 0:
remaining_days = remaining_progress / avg_rate
else:
remaining_days = 365 # Fallback
# Adjust with SPI
current_spi = progress_history[-1].spi
if current_spi > 0:
adjusted_remaining = remaining_days / current_spi
else:
adjusted_remaining = remaining_days
# Get base forecast
base_forecast = self.forecast_completion(
project_data, current_progress, datetime.now()
)
# Blend predictions
progress_completion = datetime.now() + timedelta(days=int(adjusted_remaining))
# Weight recent progress more heavily
blended_completion = base_forecast.predicted_completion + (
(progress_completion - base_forecast.predicted_completion) * 0.6
)
return ScheduleForecast(
project_id=project_id,
forecast_date=datetime.now(),
predicted_completion=blended_completion,
confidence_interval=base_forecast.confidence_interval,
delay_probability=base_forecast.delay_probability,
delay_days=int((blended_completion - pd.to_datetime(project_data['planned_end'])).days),
key_risk_factors=base_forecast.key_risk_factors + [f"Current SPI: {current_spi:.2f}"],
recommended_actions=base_forecast.recommended_actions
)
def generate_forecast_report(self, forecast: ScheduleForecast, project_name: str) -> str:
"""Generate forecast report."""
lines = ["# Schedule Forecast Report", ""]
lines.append(f"**Project:** {project_name}")
lines.append(f"**Forecast Date:** {forecast.forecast_date.strftime('%Y-%m-%d')}")
lines.append("")
lines.append("## Completion Forecast")
lines.append(f"**Predicted Completion:** {forecast.predicted_completion.strftime('%Y-%m-%d')}")
lines.append(f"**Confidence Interval:** {forecast.confidence_interval[0].strftime('%Y-%m-%d')} to {forecast.confidence_interval[1].strftime('%Y-%m-%d')}")
lines.append(f"**Predicted Delay:** {forecast.delay_days} days")
lines.append(f"**Delay Probability:** {forecast.delay_probability:.1%}")
lines.append("")
lines.append("## Risk Factors")
for risk in forecast.key_risk_factors:
lines.append(f"- ⚠️ {risk}")
lines.append("")
lines.append("## Recommended Actions")
for action in forecast.recommended_actions:
lines.append(f"- {action}")
return "\n".join(lines)
Quick Start
import pandas as pd
from datetime import datetime
# Load historical data
historical = pd.read_excel("historical_projects.xlsx")
# Initialize forecaster
forecaster = ConstructionScheduleForecaster()
# Train model
metrics = forecaster.train_delay_model(historical)
print(f"Model MAE: {metrics['mae']:.1f} days")
# Forecast for new project
new_project = {
'project_id': 'PROJ-001',
'project_type': 'Office',
'gross_area': 50000,
'contract_value': 5000000,
'planned_duration': 365,
'num_subcontractors': 12,
'num_change_orders': 3,
'planned_start': '2026-01-15',
'planned_end': '2027-01-15',
'start_month': 1,
'winter_start': 1
}
forecast = forecaster.forecast_completion(new_project, 25, datetime.now())
print(f"Predicted completion: {forecast.predicted_completion.strftime('%Y-%m-%d')}")
print(f"Delay probability: {forecast.delay_probability:.1%}")
# Generate report
report = forecaster.generate_forecast_report(forecast, "Office Building Project")
print(report)
Dependencies
pip install pandas numpy scikit-learn