change-order-analysis

name	change-order-analysis
description	Analyze and predict construction change orders using ML. Classify change order types, predict costs and schedule impacts, identify patterns, and optimize approval workflows.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🚀","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

Change Order Analysis

Overview

This skill implements machine learning-based change order analysis for construction projects. Predict change order costs, classify types, identify patterns in historical data, and streamline approval processes.

Capabilities:

• Change order classification
• Cost impact prediction
• Schedule impact analysis
• Pattern identification
• Root cause analysis
• Approval workflow optimization

Quick Start

from dataclasses import dataclass, field
from datetime import date, datetime
from typing import List, Dict, Optional
from enum import Enum

class ChangeOrderType(Enum):
    DESIGN_CHANGE = "design_change"
    OWNER_REQUEST = "owner_request"
    FIELD_CONDITION = "field_condition"
    CODE_COMPLIANCE = "code_compliance"
    VALUE_ENGINEERING = "value_engineering"
    ERROR_OMISSION = "error_omission"
    SCOPE_CHANGE = "scope_change"

class ChangeOrderStatus(Enum):
    DRAFT = "draft"
    SUBMITTED = "submitted"
    UNDER_REVIEW = "under_review"
    APPROVED = "approved"
    REJECTED = "rejected"
    IMPLEMENTED = "implemented"

@dataclass
class ChangeOrder:
    co_number: str
    title: str
    description: str
    co_type: ChangeOrderType
    status: ChangeOrderStatus
    submitted_date: date
    requested_by: str
    cost_impact: float
    schedule_impact_days: int
    affected_elements: List[str] = field(default_factory=list)

def classify_change_order(description: str) -> ChangeOrderType:
    """Simple rule-based classification"""
    description_lower = description.lower()

    if any(word in description_lower for word in ['design', 'drawing', 'specification']):
        return ChangeOrderType.DESIGN_CHANGE
    elif any(word in description_lower for word in ['owner', 'client', 'request']):
        return ChangeOrderType.OWNER_REQUEST
    elif any(word in description_lower for word in ['site', 'field', 'condition', 'unforeseen']):
        return ChangeOrderType.FIELD_CONDITION
    elif any(word in description_lower for word in ['code', 'regulation', 'compliance']):
        return ChangeOrderType.CODE_COMPLIANCE
    elif any(word in description_lower for word in ['value', 'alternative', 'savings']):
        return ChangeOrderType.VALUE_ENGINEERING
    elif any(word in description_lower for word in ['error', 'omission', 'mistake']):
        return ChangeOrderType.ERROR_OMISSION
    else:
        return ChangeOrderType.SCOPE_CHANGE

# Example
co = ChangeOrder(
    co_number="CO-001",
    title="Additional structural reinforcement",
    description="Site conditions revealed weaker soil requiring additional foundation reinforcement",
    co_type=classify_change_order("Site conditions revealed weaker soil"),
    status=ChangeOrderStatus.SUBMITTED,
    submitted_date=date.today(),
    requested_by="Site Engineer",
    cost_impact=50000,
    schedule_impact_days=5
)
print(f"CO Type: {co.co_type.value}")

Comprehensive Change Order System

Change Order Management

from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
from typing import List, Dict, Optional, Tuple
from enum import Enum
import pandas as pd
import numpy as np

class ImpactSeverity(Enum):
    MINOR = "minor"  # < 1% cost, < 1 week schedule
    MODERATE = "moderate"  # 1-5% cost, 1-4 weeks schedule
    MAJOR = "major"  # 5-10% cost, 1-3 months schedule
    CRITICAL = "critical"  # > 10% cost, > 3 months schedule

@dataclass
class CostBreakdown:
    labor: float = 0
    materials: float = 0
    equipment: float = 0
    subcontractor: float = 0
    overhead: float = 0
    profit: float = 0

    @property
    def total(self) -> float:
        return self.labor + self.materials + .equipment + .subcontractor + .overhead + .profit


 :
    direct_days: 
    ripple_days: 
    critical_path_affected: 
    affected_activities: [] = field(default_factory=)


     () -> :
         .direct_days + .ripple_days


 :
    co_id: 
    co_number: 
    title: 
    description: 
    justification: 

    
    co_type: ChangeOrderType
    initiated_by:   
    responsibility:   

    
    status: ChangeOrderStatus
    submitted_date: date
    approved_date: [date] = 
    implemented_date: [date] = 

    
    cost_breakdown: CostBreakdown = field(default_factory=CostBreakdown)
    schedule_impact: ScheduleImpact = 
    severity: ImpactSeverity = ImpactSeverity.MINOR

    
    affected_elements: [] = field(default_factory=)
    affected_drawings: [] = field(default_factory=)
    affected_specs: [] = field(default_factory=)

    
    attachments: [] = field(default_factory=)
    related_rfis: [] = field(default_factory=)
    related_cos: [] = field(default_factory=)

    
    approvals: [] = field(default_factory=)
    comments: [] = field(default_factory=)

 :
    

     ():
        .project_id = project_id
        .contract_value = contract_value
        .change_orders: [, ChangeOrderDetail] = {}
        .co_counter = 

     () -> ChangeOrderDetail:
        
        .co_counter += 
        co_id = 

        co = ChangeOrderDetail(
            co_id=co_id,
            co_number=,
            title=title,
            description=description,
            justification=,
            co_type=co_type,
            initiated_by=initiated_by,
            responsibility=,
            status=ChangeOrderStatus.DRAFT,
            submitted_date=date.today()
        )

        .change_orders[co_id] = co
         co

     ():
        
        co = .change_orders.get(co_id)
         co:
            co.cost_breakdown = cost_breakdown
            co.severity = ._calculate_severity(co)

     ():
        
        co = .change_orders.get(co_id)
         co:
            co.schedule_impact = impact
            co.severity = ._calculate_severity(co)

     () -> ImpactSeverity:
        
        cost_pct = co.cost_breakdown.total / .contract_value * 
        schedule_days = co.schedule_impact.total_days  co.schedule_impact  

         cost_pct >   schedule_days > :
             ImpactSeverity.CRITICAL
         cost_pct >   schedule_days > :
             ImpactSeverity.MAJOR
         cost_pct >   schedule_days > :
             ImpactSeverity.MODERATE
        :
             ImpactSeverity.MINOR

     ():
        
        co = .change_orders.get(co_id)
         co  co.status == ChangeOrderStatus.DRAFT:
            co.status = ChangeOrderStatus.SUBMITTED
            co.submitted_date = date.today()

     ():
        
        co = .change_orders.get(co_id)
         co:
            co.approvals.append({
                : approver,
                : ,
                : date.today().isoformat(),
                : comments
            })
            co.status = ChangeOrderStatus.APPROVED
            co.approved_date = date.today()

     () -> :
        
          .change_orders:
             {: }

        total_cost = (co.cost_breakdown.total  co  .change_orders.values())
        total_schedule = (
            co.schedule_impact.total_days  co.schedule_impact  
             co  .change_orders.values()
        )

        by_type = {}
        by_status = {}
        by_severity = {}

         co  .change_orders.values():
            t = co.co_type.value
            by_type[t] = by_type.get(t, ) + co.cost_breakdown.total

            s = co.status.value
            by_status[s] = by_status.get(s, ) + 

            sev = co.severity.value
            by_severity[sev] = by_severity.get(sev, ) + 

         {
            : (.change_orders),
            : total_cost,
            : total_cost / .contract_value * ,
            : total_schedule,
            : by_type,
            : by_status,
            : by_severity
        }

ML Classification and Prediction

from sklearn.ensemble import RandomForestClassifier, GradientBoostingRegressor
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
import pandas as pd
import numpy as np
import joblib

class ChangeOrderPredictor:
    """ML-based change order classification and cost prediction"""

    def __init__(self):
        self.type_classifier = None
        self.cost_predictor = None
        self.schedule_predictor = None
        self.vectorizer = TfidfVectorizer(max_features=500, ngram_range=(1, 2))
        self.type_encoder = LabelEncoder()
        self.is_trained = False

    def train(self, historical_data: pd.DataFrame):
        """Train models on historical change order data

        Expected columns:
        - description: text description
        - co_type: change order type
        - cost_impact: cost in dollars
        - schedule_impact: days of delay
        - contract_value: original contract value
        - project_phase: phase when CO was raised
        - affected_elements_count: number of affected elements
        """
        # Prepare text features
        text_features = self.vectorizer.fit_transform(historical_data[])

        
        numeric_features = historical_data[[
            , 
        ]].values

        
        X = np.hstack([text_features.toarray(), numeric_features])

        
        y_type = .type_encoder.fit_transform(historical_data[])
        X_train, X_test, y_train, y_test = train_test_split(X, y_type, test_size=)

        .type_classifier = RandomForestClassifier(n_estimators=, random_state=)
        .type_classifier.fit(X_train, y_train)

        type_accuracy = .type_classifier.score(X_test, y_test)

        
        y_cost = historical_data[].values
        .cost_predictor = GradientBoostingRegressor(n_estimators=, random_state=)
        .cost_predictor.fit(X, y_cost)

        
        y_schedule = historical_data[].values
        .schedule_predictor = GradientBoostingRegressor(n_estimators=, random_state=)
        .schedule_predictor.fit(X, y_schedule)

        .is_trained = 

         {
            : type_accuracy,
            : 
        }

     () -> :
        
          .is_trained:
             {: }

        
        text_features = .vectorizer.transform([description])
        numeric_features = np.array([[contract_value, affected_elements_count]])
        X = np.hstack([text_features.toarray(), numeric_features])

        
        type_probs = .type_classifier.predict_proba(X)[]
        type_idx = np.argmax(type_probs)
        predicted_type = .type_encoder.inverse_transform([type_idx])[]

        
        predicted_cost = .cost_predictor.predict(X)[]

        
        predicted_schedule = .schedule_predictor.predict(X)[]

         {
            : predicted_type,
            : (type_probs[type_idx]),
            : {
                .type_encoder.inverse_transform([i])[]: (p)
                 i, p  (type_probs)
            },
            : ((, predicted_cost)),
            : ((, predicted_schedule)),
            : (predicted_cost / contract_value * )
        }

     ():
        
        joblib.dump({
            : .type_classifier,
            : .cost_predictor,
            : .schedule_predictor,
            : .vectorizer,
            : .type_encoder
        }, path)

     ():
        
        data = joblib.load(path)
        .type_classifier = data[]
        .cost_predictor = data[]
        .schedule_predictor = data[]
        .vectorizer = data[]
        .type_encoder = data[]
        .is_trained = 

Pattern Analysis

from collections import defaultdict
from typing import List, Dict
import pandas as pd

class ChangeOrderAnalyzer:
    """Analyze patterns in change orders"""

    def __init__(self, change_orders: List[ChangeOrderDetail]):
        self.cos = change_orders
        self.df = self._to_dataframe()

    def _to_dataframe(self) -> pd.DataFrame:
        """Convert change orders to DataFrame"""
        data = []
        for co in self.cos:
            data.append({
                'co_id': co.co_id,
                'co_type': co.co_type.value,
                'initiated_by': co.initiated_by,
                'cost': co.cost_breakdown.total,
                'schedule_days': co.schedule_impact.total_days if co.schedule_impact else 0,
                'submitted_date': co.submitted_date,
                'affected_elements': len(co.affected_elements),
                'severity': co.severity.value
            })
        return pd.DataFrame(data)

    def analyze_by_type(self) -> Dict:
        """Analyze change orders by type"""
        if .df.empty:
             {}

        analysis = {}
         co_type  .df[].unique():
            type_df = .df[.df[] == co_type]
            analysis[co_type] = {
                : (type_df),
                : type_df[].(),
                : type_df[].mean(),
                : type_df[].(),
                : type_df[].mean()
            }

         analysis

     () -> :
        
         .df.empty:
             {}

        .df[] = pd.to_datetime(.df[]).dt.to_period()

        monthly = .df.groupby().agg({
            : ,
            : ,
            : 
        }).rename(columns={: })

         {
            : monthly.to_dict(),
            : monthly[].idxmax().strftime(),
            :   monthly[].is_monotonic_increasing 
                                 monthly[].is_monotonic_decreasing  
        }

     () -> []:
        
         .df.empty:
             []

        
        causes = .df.groupby([, ]).agg({
            : ,
            : 
        }).reset_index()

        causes = causes.sort_values(, ascending=)

         [
            {
                : row[],
                : row[],
                : row[],
                : row[],
                : ._get_recommendation(row[], row[])
            }
             _, row  causes.head().iterrows()
        ]

     () -> :
        
        recommendations = {
            (, ): ,
            (, ): ,
            (, ): ,
            (, ): ,
            (, ): ,
            (, ): 
        }

         recommendations.get(
            (initiator.lower(), co_type),
            
        )

     () -> :
        
         .df.empty:
             

        
        
        
        
        

        co_rate = (.df) /   
        avg_cost_impact = .df[].mean()
        avg_schedule_impact = .df[].mean()

        
        freq_score = (, co_rate / ) *   
        cost_score = (, avg_cost_impact / ) *   
        schedule_score = (, avg_schedule_impact / ) *   

         freq_score + cost_score + schedule_score

     () -> :
        
         pd.ExcelWriter(output_path, engine=)  writer:
            
            summary = pd.DataFrame([{
                : (.cos),
                : .df[].(),
                : .df[].(),
                : .calculate_risk_score()
            }])
            summary.to_excel(writer, sheet_name=, index=)

            
            pd.DataFrame(.analyze_by_type()).T.to_excel(
                writer, sheet_name=
            )

            
            pd.DataFrame(.identify_root_causes()).to_excel(
                writer, sheet_name=, index=
            )

            
            .df.to_excel(writer, sheet_name=, index=)

         output_path

Quick Reference

CO Type	Typical Cause	Prevention Strategy
Design Change	Incomplete design	BIM coordination, design reviews
Owner Request	Changing requirements	Clear scope definition
Field Condition	Unforeseen site issues	Thorough site investigation
Code Compliance	Regulation changes	Early code review
Value Engineering	Cost savings opportunity	VE workshops
Error/Omission	Design mistakes	QA/QC processes

Resources

• AIA A201: General Conditions of the Contract
• AGC ConsensusDocs: Change order management
• DDC Website: https://datadrivenconstruction.io

Next Steps

• See document-classification-nlp for CO document processing
• See risk-assessment-ml for project risk analysis
• See cost-prediction for cost estimation