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 + self.equipment + self.subcontractor + self.overhead + self.profit

@dataclass class ScheduleImpact: direct_days: int ripple_days: int critical_path_affected: bool affected_activities: List[str] = field(default_factory=list)

@property
def total_days(self) -> int:
    return self.direct_days + self.ripple_days

@dataclass class ChangeOrderDetail: co_id: str co_number: str title: str description: str justification: str

# Classification
co_type: ChangeOrderType
initiated_by: str  # owner, contractor, designer, etc.
responsibility: str  # who pays

# Status
status: ChangeOrderStatus
submitted_date: date
approved_date: Optional[date] = None
implemented_date: Optional[date] = None

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

# Affected scope
affected_elements: List[str] = field(default_factory=list)
affected_drawings: List[str] = field(default_factory=list)
affected_specs: List[str] = field(default_factory=list)

# Supporting documents
attachments: List[str] = field(default_factory=list)
related_rfis: List[str] = field(default_factory=list)
related_cos: List[str] = field(default_factory=list)

# Approval
approvals: List[Dict] = field(default_factory=list)
comments: List[Dict] = field(default_factory=list)

class ChangeOrderManager: """Manage project change orders"""

def __init__(self, project_id: str, contract_value: float):
    self.project_id = project_id
    self.contract_value = contract_value
    self.change_orders: Dict[str, ChangeOrderDetail] = {}
    self.co_counter = 0

def create_change_order(self, title: str, description: str,
                       co_type: ChangeOrderType,
                       initiated_by: str) -> ChangeOrderDetail:
    """Create new change order"""
    self.co_counter += 1
    co_id = f"CO-{self.project_id}-{self.co_counter:04d}"

    co = ChangeOrderDetail(
        co_id=co_id,
        co_number=f"CO-{self.co_counter:04d}",
        title=title,
        description=description,
        justification="",
        co_type=co_type,
        initiated_by=initiated_by,
        responsibility="TBD",
        status=ChangeOrderStatus.DRAFT,
        submitted_date=date.today()
    )

    self.change_orders[co_id] = co
    return co

def update_cost(self, co_id: str, cost_breakdown: CostBreakdown):
    """Update change order cost"""
    co = self.change_orders.get(co_id)
    if co:
        co.cost_breakdown = cost_breakdown
        co.severity = self._calculate_severity(co)

def update_schedule_impact(self, co_id: str, impact: ScheduleImpact):
    """Update schedule impact"""
    co = self.change_orders.get(co_id)
    if co:
        co.schedule_impact = impact
        co.severity = self._calculate_severity(co)

def _calculate_severity(self, co: ChangeOrderDetail) -> ImpactSeverity:
    """Calculate change order severity"""
    cost_pct = co.cost_breakdown.total / self.contract_value * 100
    schedule_days = co.schedule_impact.total_days if co.schedule_impact else 0

    if cost_pct > 10 or schedule_days > 90:
        return ImpactSeverity.CRITICAL
    elif cost_pct > 5 or schedule_days > 30:
        return ImpactSeverity.MAJOR
    elif cost_pct > 1 or schedule_days > 7:
        return ImpactSeverity.MODERATE
    else:
        return ImpactSeverity.MINOR

def submit_for_approval(self, co_id: str):
    """Submit change order for approval"""
    co = self.change_orders.get(co_id)
    if co and co.status == ChangeOrderStatus.DRAFT:
        co.status = ChangeOrderStatus.SUBMITTED
        co.submitted_date = date.today()

def approve(self, co_id: str, approver: str, comments: str = ""):
    """Approve change order"""
    co = self.change_orders.get(co_id)
    if co:
        co.approvals.append({
            'approver': approver,
            'action': 'approved',
            'date': date.today().isoformat(),
            'comments': comments
        })
        co.status = ChangeOrderStatus.APPROVED
        co.approved_date = date.today()

def get_summary(self) -> Dict:
    """Get change order summary"""
    if not self.change_orders:
        return {'message': 'No change orders'}

    total_cost = sum(co.cost_breakdown.total for co in self.change_orders.values())
    total_schedule = sum(
        co.schedule_impact.total_days if co.schedule_impact else 0
        for co in self.change_orders.values()
    )

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

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

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

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

    return {
        'total_change_orders': len(self.change_orders),
        'total_cost_impact': total_cost,
        'cost_impact_pct': total_cost / self.contract_value * 100,
        'total_schedule_impact_days': total_schedule,
        'by_type': by_type,
        'by_status': by_status,
        'by_severity': 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['description'])

    # Prepare numeric features
    numeric_features = historical_data[[
        'contract_value', 'affected_elements_count'
    ]].values

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

    # Train type classifier
    y_type = self.type_encoder.fit_transform(historical_data['co_type'])
    X_train, X_test, y_train, y_test = train_test_split(X, y_type, test_size=0.2)

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

    type_accuracy = self.type_classifier.score(X_test, y_test)

    # Train cost predictor
    y_cost = historical_data['cost_impact'].values
    self.cost_predictor = GradientBoostingRegressor(n_estimators=100, random_state=42)
    self.cost_predictor.fit(X, y_cost)

    # Train schedule predictor
    y_schedule = historical_data['schedule_impact'].values
    self.schedule_predictor = GradientBoostingRegressor(n_estimators=100, random_state=42)
    self.schedule_predictor.fit(X, y_schedule)

    self.is_trained = True

    return {
        'type_classifier_accuracy': type_accuracy,
        'models_trained': True
    }

def predict(self, description: str, contract_value: float,
           affected_elements_count: int = 1) -> Dict:
    """Predict change order type and impacts"""
    if not self.is_trained:
        return {'error': 'Models not trained'}

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

    # Predict type
    type_probs = self.type_classifier.predict_proba(X)[0]
    type_idx = np.argmax(type_probs)
    predicted_type = self.type_encoder.inverse_transform([type_idx])[0]

    # Predict cost
    predicted_cost = self.cost_predictor.predict(X)[0]

    # Predict schedule
    predicted_schedule = self.schedule_predictor.predict(X)[0]

    return {
        'predicted_type': predicted_type,
        'type_confidence': float(type_probs[type_idx]),
        'type_probabilities': {
            self.type_encoder.inverse_transform([i])[0]: float(p)
            for i, p in enumerate(type_probs)
        },
        'predicted_cost': float(max(0, predicted_cost)),
        'predicted_schedule_days': int(max(0, predicted_schedule)),
        'cost_as_pct_contract': float(predicted_cost / contract_value * 100)
    }

def save_models(self, path: str):
    """Save trained models"""
    joblib.dump({
        'type_classifier': self.type_classifier,
        'cost_predictor': self.cost_predictor,
        'schedule_predictor': self.schedule_predictor,
        'vectorizer': self.vectorizer,
        'type_encoder': self.type_encoder
    }, path)

def load_models(self, path: str):
    """Load trained models"""
    data = joblib.load(path)
    self.type_classifier = data['type_classifier']
    self.cost_predictor = data['cost_predictor']
    self.schedule_predictor = data['schedule_predictor']
    self.vectorizer = data['vectorizer']
    self.type_encoder = data['type_encoder']
    self.is_trained = True

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 self.df.empty:
        return {}

    analysis = {}
    for co_type in self.df['co_type'].unique():
        type_df = self.df[self.df['co_type'] == co_type]
        analysis[co_type] = {
            'count': len(type_df),
            'total_cost': type_df['cost'].sum(),
            'avg_cost': type_df['cost'].mean(),
            'total_schedule_days': type_df['schedule_days'].sum(),
            'avg_schedule_days': type_df['schedule_days'].mean()
        }

    return analysis

def analyze_trends(self) -> Dict:
    """Analyze trends over time"""
    if self.df.empty:
        return {}

    self.df['month'] = pd.to_datetime(self.df['submitted_date']).dt.to_period('M')

    monthly = self.df.groupby('month').agg({
        'co_id': 'count',
        'cost': 'sum',
        'schedule_days': 'sum'
    }).rename(columns={'co_id': 'count'})

    return {
        'monthly_trend': monthly.to_dict(),
        'peak_month': monthly['count'].idxmax().strftime('%Y-%m'),
        'total_cost_trend': 'increasing' if monthly['cost'].is_monotonic_increasing else
                           'decreasing' if monthly['cost'].is_monotonic_decreasing else 'variable'
    }

def identify_root_causes(self) -> List[Dict]:
    """Identify common root causes"""
    if self.df.empty:
        return []

    # Analyze by initiator and type combination
    causes = self.df.groupby(['initiated_by', 'co_type']).agg({
        'co_id': 'count',
        'cost': 'sum'
    }).reset_index()

    causes = causes.sort_values('cost', ascending=False)

    return [
        {
            'initiator': row['initiated_by'],
            'type': row['co_type'],
            'frequency': row['co_id'],
            'total_cost': row['cost'],
            'recommendation': self._get_recommendation(row['initiated_by'], row['co_type'])
        }
        for _, row in causes.head(10).iterrows()
    ]

def _get_recommendation(self, initiator: str, co_type: str) -> str:
    """Generate recommendation based on pattern"""
    recommendations = {
        ('designer', 'design_change'): 'Improve design review process and BIM coordination',
        ('designer', 'error_omission'): 'Implement design quality checks and clash detection',
        ('owner', 'owner_request'): 'Define scope more clearly during planning phase',
        ('owner', 'scope_change'): 'Conduct thorough requirements gathering',
        ('contractor', 'field_condition'): 'Enhance site investigation before construction',
        ('contractor', 'value_engineering'): 'Include VE sessions earlier in project'
    }

    return recommendations.get(
        (initiator.lower(), co_type),
        'Review process and implement preventive measures'
    )

def calculate_risk_score(self) -> float:
    """Calculate overall change order risk score"""
    if self.df.empty:
        return 0

    # Factors:
    # - Frequency of COs
    # - Cost impact severity
    # - Schedule impact severity
    # - Trend direction

    co_rate = len(self.df) / 12  # COs per month (assuming 12 month project)
    avg_cost_impact = self.df['cost'].mean()
    avg_schedule_impact = self.df['schedule_days'].mean()

    # Normalize and weight
    freq_score = min(1, co_rate / 10) * 30  # Up to 30 points
    cost_score = min(1, avg_cost_impact / 50000) * 40  # Up to 40 points
    schedule_score = min(1, avg_schedule_impact / 30) * 30  # Up to 30 points

    return freq_score + cost_score + schedule_score

def generate_report(self, output_path: str) -> str:
    """Generate comprehensive analysis report"""
    with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
        # Summary
        summary = pd.DataFrame([{
            'Total COs': len(self.cos),
            'Total Cost Impact': self.df['cost'].sum(),
            'Total Schedule Impact (days)': self.df['schedule_days'].sum(),
            'Risk Score': self.calculate_risk_score()
        }])
        summary.to_excel(writer, sheet_name='Summary', index=False)

        # By type
        pd.DataFrame(self.analyze_by_type()).T.to_excel(
            writer, sheet_name='By_Type'
        )

        # Root causes
        pd.DataFrame(self.identify_root_causes()).to_excel(
            writer, sheet_name='Root_Causes', index=False
        )

        # All COs
        self.df.to_excel(writer, sheet_name='All_COs', index=False)

    return 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