predictive-analytics-construction

建設業界向け予測分析

概要

過去のプロジェクトデータを使用して、コスト超過、スケジュールの遅延、品質問題、リスクなど、将来の結果を予測します。建設業界特有の傾向に合わせた機械学習モデルを適用します。

ビジネス上の価値

予測分析により、プロアクティブなプロジェクト管理が可能になります。

• 早期警告：コスト超過が発生する前に、超過の可能性が高いプロジェクトを特定します
• リソースの最適化：予測されたニーズに基づいてリソースを配分します
• リスクの軽減：リスクの高い領域に早期に注力します
• 見積もりの改善：過去の精度から学習します

技術的な実装

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
from sklearn.ensemble import RandomForestRegressor, GradientBoostingClassifier
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import mean_absolute_error, accuracy_score, classification_report
import warnings
warnings.filterwarnings('ignore')

@dataclass
class PredictionResult:
    prediction: float
    confidence: float
    prediction_type: str
    features_used: List[str]
    feature_importance: Dict[str, float]
    comparable_projects: List[str]
    risk_factors: List[str]

@dataclass
class ModelMetrics:
    model_name: str
    accuracy: float
    mae: float
    feature_importance: Dict[str, float]
    training_samples: int
    last_trained: datetime

class ConstructionPredictiveAnalytics:
    """Predictive analytics for construction projects."""

    def __init__(self):
        self.models: Dict[str, Any] = {}
        self.scalers: Dict[str, StandardScaler] = {}
        self.encoders: Dict[str, LabelEncoder] = {}
        self.metrics: Dict[str, ModelMetrics] = {}
        self.feature_columns: Dict[str, List[str]] = {}

    def prepare_features(self, df: pd.DataFrame, target_col: str) -> Tuple[pd.DataFrame, pd.Series]:
        """Prepare features for model training."""
        # Separate numeric and categorical columns
        numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
        categorical_cols = df.select_dtypes(include=['object', 'category']).columns.tolist()

        # Remove target from features
        if target_col in numeric_cols:
            numeric_cols.remove(target_col)
        if target_col in categorical_cols:
            categorical_cols.remove(target_col)

        # Encode categorical variables
        df_encoded = df.copy()
        for col in categorical_cols:
            if col not in self.encoders:
                self.encoders[col] = LabelEncoder()
                df_encoded[col] = self.encoders[col].fit_transform(df[col].astype(str))
            else:
                df_encoded[col] = self.encoders[col].transform(df[col].astype(str))

        feature_cols = numeric_cols + categorical_cols
        X = df_encoded[feature_cols].fillna(0)
        y = df[target_col]

        return X, y, feature_cols

    def train_cost_overrun_model(self, historical_data: pd.DataFrame) -> ModelMetrics:
        """Train model to predict cost overrun percentage."""
        # Expected columns: project_type, original_estimate, gross_area, duration_months,
        # num_change_orders, complexity_score, contractor_experience, final_cost

        required_cols = ['original_estimate', 'final_cost']
        if not all(col in historical_data.columns for col in required_cols):
            raise ValueError(f"Missing required columns: {required_cols}")

        # Calculate overrun percentage
        df = historical_data.copy()
        df['overrun_pct'] = ((df['final_cost'] - df['original_estimate']) / df['original_estimate']) * 100

        # Prepare features
        feature_cols = [col for col in df.columns if col not in ['final_cost', 'overrun_pct', 'project_id', 'project_name']]
        X, y, used_features = self.prepare_features(df[feature_cols + ['overrun_pct']], 'overrun_pct')

        # 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, 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)

        # Cross-validation
        cv_scores = cross_val_score(model, X_train_scaled, y_train, cv=5, scoring='neg_mean_absolute_error')

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

        # Store model
        self.models['cost_overrun'] = model
        self.scalers['cost_overrun'] = scaler
        self.feature_columns['cost_overrun'] = used_features

        metrics = ModelMetrics(
            model_name='cost_overrun',
            accuracy=1 - (mae / df['overrun_pct'].std()),
            mae=mae,
            feature_importance=importance,
            training_samples=len(X_train),
            last_trained=datetime.now()
        )
        self.metrics['cost_overrun'] = metrics

        return metrics

    def train_schedule_delay_model(self, historical_data: pd.DataFrame) -> ModelMetrics:
        """Train model to predict schedule delay probability."""
        df = historical_data.copy()

        # Binary classification: was project delayed?
        df['was_delayed'] = (df['actual_duration'] > df['planned_duration']).astype(int)

        feature_cols = [col for col in df.columns
                       if col not in ['actual_duration', 'was_delayed', 'project_id', 'project_name']]

        X, y, used_features = self.prepare_features(df[feature_cols + ['was_delayed']], 'was_delayed')

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

        y_pred = model.predict(X_test_scaled)
        accuracy = accuracy_score(y_test, y_pred)

        importance = dict(zip(used_features, model.feature_importances_))

        self.models['schedule_delay'] = model
        self.scalers['schedule_delay'] = scaler
        self.feature_columns['schedule_delay'] = used_features

        metrics = ModelMetrics(
            model_name='schedule_delay',
            accuracy=accuracy,
            mae=0,
            feature_importance=importance,
            training_samples=len(X_train),
            last_trained=datetime.now()
        )
        self.metrics['schedule_delay'] = metrics

        return metrics

    def predict_cost_overrun(self, project_data: Dict) -> PredictionResult:
        """Predict cost overrun for a new project."""
        if 'cost_overrun' not in self.models:
            raise ValueError("Cost overrun model not trained. Call train_cost_overrun_model first.")

        model = self.models['cost_overrun']
        scaler = self.scalers['cost_overrun']
        features = self.feature_columns['cost_overrun']

        # Prepare input
        input_df = pd.DataFrame([project_data])

        # Encode categorical
        for col in input_df.select_dtypes(include=['object']).columns:
            if col in self.encoders:
                input_df[col] = self.encoders[col].transform(input_df[col].astype(str))

        # Ensure all features present
        for feat in features:
            if feat not in input_df.columns:
                input_df[feat] = 0

        X = input_df[features].fillna(0)
        X_scaled = scaler.transform(X)

        prediction = model.predict(X_scaled)[0]

        # Get feature importance for this prediction
        importance = dict(zip(features, model.feature_importances_))
        top_features = sorted(importance.items(), key=lambda x: -x[1])[:5]

        # Identify risk factors
        risk_factors = []
        if prediction > 10:
            risk_factors.append(f"High overrun risk: {prediction:.1f}%")
        for feat, imp in top_features[:3]:
            risk_factors.append(f"Key factor: {feat} (importance: {imp:.2%})")

        return PredictionResult(
            prediction=prediction,
            confidence=0.8,  # Could calculate from model uncertainty
            prediction_type='cost_overrun_percentage',
            features_used=features,
            feature_importance=dict(top_features),
            comparable_projects=[],
            risk_factors=risk_factors
        )

    def predict_delay_probability(self, project_data: Dict) -> PredictionResult:
        """Predict probability of schedule delay."""
        if 'schedule_delay' not in self.models:
            raise ValueError("Schedule delay model not trained.")

        model = self.models['schedule_delay']
        scaler = self.scalers['schedule_delay']
        features = self.feature_columns['schedule_delay']

        input_df = pd.DataFrame([project_data])

        for col in input_df.select_dtypes(include=['object']).columns:
            if col in self.encoders:
                input_df[col] = self.encoders[col].transform(input_df[col].astype(str))

        for feat in features:
            if feat not in input_df.columns:
                input_df[feat] = 0

        X = input_df[features].fillna(0)
        X_scaled = scaler.transform(X)

        probability = model.predict_proba(X_scaled)[0][1]
        prediction = model.predict(X_scaled)[0]

        importance = dict(zip(features, model.feature_importances_))
        top_features = sorted(importance.items(), key=lambda x: -x[1])[:5]

        risk_factors = []
        if probability > 0.7:
            risk_factors.append(f"High delay probability: {probability:.1%}")
        elif probability > 0.4:
            risk_factors.append(f"Moderate delay probability: {probability:.1%}")

        return PredictionResult(
            prediction=probability,
            confidence=probability if prediction == 1 else 1 - probability,
            prediction_type='delay_probability',
            features_used=features,
            feature_importance=dict(top_features),
            comparable_projects=[],
            risk_factors=risk_factors
        )

    def find_similar_projects(self, project_data: Dict, historical_data: pd.DataFrame,
                             n: int = 5) -> pd.DataFrame:
        """Find similar projects from historical data."""
        from sklearn.neighbors import NearestNeighbors

        numeric_cols = historical_data.select_dtypes(include=[np.number]).columns.tolist()
        exclude = ['final_cost', 'actual_duration', 'overrun_pct']
        feature_cols = [c for c in numeric_cols if c not in exclude]

        X = historical_data[feature_cols].fillna(0)

        scaler = StandardScaler()
        X_scaled = scaler.fit_transform(X)

        # Prepare new project
        new_project = pd.DataFrame([project_data])[feature_cols].fillna(0)
        new_scaled = scaler.transform(new_project)

        # Find neighbors
        nn = NearestNeighbors(n_neighbors=min(n, len(X)), metric='euclidean')
        nn.fit(X_scaled)
        distances, indices = nn.kneighbors(new_scaled)

        similar = historical_data.iloc[indices[0]].copy()
        similar['similarity_score'] = 1 / (1 + distances[0])

        return similar

    def generate_prediction_report(self, project_data: Dict, historical_data: pd.DataFrame) -> str:
        """Generate comprehensive prediction report."""
        lines = ["# Project Prediction Report", ""]
        lines.append(f"**Generated:** {datetime.now().strftime('%Y-%m-%d %H:%M')}")
        lines.append(f"**Project:** {project_data.get('project_name', 'New Project')}")
        lines.append("")

        # Cost prediction
        if 'cost_overrun' in self.models:
            cost_pred = self.predict_cost_overrun(project_data)
            lines.append("## Cost Overrun Prediction")
            lines.append(f"**Predicted Overrun:** {cost_pred.prediction:.1f}%")
            lines.append(f"**Confidence:** {cost_pred.confidence:.1%}")
            lines.append("")
            lines.append("**Key Factors:**")
            for feat, imp in list(cost_pred.feature_importance.items())[:5]:
                lines.append(f"- {feat}: {imp:.2%}")
            lines.append("")

        # Schedule prediction
        if 'schedule_delay' in self.models:
            delay_pred = self.predict_delay_probability(project_data)
            lines.append("## Schedule Delay Prediction")
            lines.append(f"**Delay Probability:** {delay_pred.prediction:.1%}")
            lines.append("")

        # Similar projects
        lines.append("## Similar Historical Projects")
        similar = self.find_similar_projects(project_data, historical_data, n=5)
        for _, row in similar.iterrows():
            name = row.get('project_name', 'Project')
            overrun = row.get('overrun_pct', 0)
            similarity = row.get('similarity_score', 0)
            lines.append(f"- **{name}**: {overrun:.1f}% overrun (similarity: {similarity:.1%})")

        # Risk summary
        lines.append("")
        lines.append("## Risk Summary")
        all_risks = []
        if 'cost_overrun' in self.models:
            all_risks.extend(cost_pred.risk_factors)
        if 'schedule_delay' in self.models:
            all_risks.extend(delay_pred.risk_factors)

        for risk in all_risks:
            lines.append(f"- ⚠️ {risk}")

        return "\n".join(lines)

クイックスタート

import pandas as pd

# Load historical data
historical = pd.read_excel("historical_projects.xlsx")

# Initialize analytics
analytics = ConstructionPredictiveAnalytics()

# Train models
cost_metrics = analytics.train_cost_overrun_model(historical)
print(f"Cost model MAE: {cost_metrics.mae:.2f}%")

delay_metrics = analytics.train_schedule_delay_model(historical)
print(f"Delay model accuracy: {delay_metrics.accuracy:.1%}")

# Predict for new project
new_project = {
    'project_type': 'Office',
    'original_estimate': 5000000,
    'gross_area': 50000,
    'duration_months': 18,
    'complexity_score': 7,
    'contractor_experience': 15
}

cost_prediction = analytics.predict_cost_overrun(new_project)
print(f"Predicted overrun: {cost_prediction.prediction:.1f}%")

delay_prediction = analytics.predict_delay_probability(new_project)
print(f"Delay probability: {delay_prediction.prediction:.1%}")

# Generate report
report = analytics.generate_prediction_report(new_project, historical)
print(report)

依存関係

pip install pandas numpy scikit-learn

リソース

• 建設業界向け機械学習：予測モデルに関する研究
• 特徴量エンジニアリング：建設業界に特化した特徴量