schedule-forecaster

建筑项目进度预测

概述

使用基于历史数据训练的机器学习模型预测项目完工日期。根据当前进度、天气模式、资源可用性和项目特征预测延期情况。

业务案例

准确的进度预测有助于：

• 提前预警：在潜在延期影响里程碑之前识别问题
• 资源规划：根据预测需求调整人员配置
• 客户沟通：提供可靠的完工时间估算
• 风险管理：主动应对进度风险

技术实现

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)

快速开始

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)

依赖项

pip install pandas numpy scikit-learn