Change Order Manager
Overview
Manage the complete change order lifecycle from potential change identification through approval and payment. Track cost and schedule impacts, maintain documentation, and provide analytics for project control.
Change Order Workflow
┌─────────────────────────────────────────────────────────────────┐
│ CHANGE ORDER WORKFLOW │
├─────────────────────────────────────────────────────────────────┤
│ │
│ Identify → Document → Price → Negotiate → Execute │
│ ──────── ──────── ───── ───────── ─────── │
│ 📋 PCO 📝 RFP 💰 Quote 🤝 Review ✅ Approve │
│ 🔍 Review 📸 Photos ⏰ Time 📧 Submit 📄 Sign │
│ 📧 Notify 📄 Backup 📊 Impact 💬 Discuss 💵 Pay │
│ │
└─────────────────────────────────────────────────────────────────┘Technical Implementation
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime, timedelta
from enum import Enum
import json
class ChangeOrderStatus(Enum):
DRAFT = "draft"
SUBMITTED = "submitted"
UNDER_REVIEW = "under_review"
PRICING = "pricing"
NEGOTIATING = "negotiating"
APPROVED = "approved"
REJECTED = "rejected"
EXECUTED = "executed"
VOID = "void"
class ChangeType(Enum):
OWNER_DIRECTED = "owner_directed"
DESIGN_ERROR = "design_error"
FIELD_CONDITION = "field_condition"
CODE_CHANGE = "code_change"
VALUE_ENGINEERING = "value_engineering"
SCHEDULE_ACCELERATION = "schedule_acceleration"
SCOPE_REDUCTION = "scope_reduction"
class PricingMethod(Enum):
LUMP_SUM = "lump_sum"
UNIT_PRICE = "unit_price"
TIME_AND_MATERIALS = "time_and_materials"
COST_PLUS = "cost_plus"
@dataclass
class CostBreakdown:
labor: float = 0.0
materials: float = 0.0
equipment: float = 0.0
subcontractor: float = 0.0
overhead: float = 0.0
profit: float = 0.0
bond: float = 0.0
@property
def direct_cost(self) -> float:
return self.labor + self.materials + self.equipment + self.subcontractor
@property
def total(self) -> float:
return self.direct_cost + self.overhead + self.profit + self.bond
@dataclass
class ChangeOrderItem:
id: str
description: str
quantity: float
unit: str
unit_price: float
total_price: float
spec_section: str = ""
csi_code: str = ""
@dataclass
class ChangeOrder:
id: str
number: int
title: str
description: str
change_type: ChangeType
status: ChangeOrderStatus
# Dates
identified_date: datetime
submitted_date: Optional[datetime] = None
approved_date: Optional[datetime] = None
executed_date: Optional[datetime] = None
# Pricing
pricing_method: PricingMethod = PricingMethod.LUMP_SUM
proposed_amount: float = 0.0
approved_amount: float = 0.0
cost_breakdown: CostBreakdown = field(default_factory=CostBreakdown)
line_items: List[ChangeOrderItem] = field(default_factory=list)
# Schedule
proposed_time_days: int = 0
approved_time_days: int = 0
impacts_critical_path: bool = False
# Documentation
rfi_references: List[str] = field(default_factory=list)
drawing_references: List[str] = field(default_factory=list)
photo_attachments: List[str] = field(default_factory=list)
backup_documents: List[str] = field(default_factory=list)
# Tracking
created_by: str = ""
assigned_to: str = ""
notes: List[Dict] = field(default_factory=list)
@dataclass
class ChangeOrderLog:
project_id: str
project_name: str
original_contract: float
change_orders: List[ChangeOrder]
total_approved: float
total_pending: float
revised_contract: float
class ChangeOrderManager:
"""Manage construction change orders."""
# Default markup rates
DEFAULT_MARKUPS = {
"overhead": 0.10, # 10%
"profit": 0.10, # 10%
"bond": 0.01, # 1%
}
def __init__(self, project_id: str, project_name: str,
original_contract: float):
self.project_id = project_id
self.project_name = project_name
self.original_contract = original_contract
self.change_orders: Dict[str, ChangeOrder] = {}
self.next_number = 1
self.markup_rates = dict(self.DEFAULT_MARKUPS)
def set_markup_rates(self, overhead: float = None, profit: float = None,
bond: float = None):
"""Set markup rates for cost calculations."""
if overhead is not None:
self.markup_rates["overhead"] = overhead
if profit is not None:
self.markup_rates["profit"] = profit
if bond is not None:
self.markup_rates["bond"] = bond
def create_change_order(self, title: str, description: str,
change_type: ChangeType,
created_by: str = "") -> ChangeOrder:
"""Create new change order."""
co_id = f"CO-{self.project_id}-{self.next_number:04d}"
co = ChangeOrder(
id=co_id,
number=self.next_number,
title=title,
description=description,
change_type=change_type,
status=ChangeOrderStatus.DRAFT,
identified_date=datetime.now(),
created_by=created_by
)
self.change_orders[co_id] = co
self.next_number += 1
return co
def add_line_item(self, co_id: str, description: str,
quantity: float, unit: str, unit_price: float,
spec_section: str = "", csi_code: str = "") -> ChangeOrderItem:
"""Add line item to change order."""
if co_id not in self.change_orders:
raise ValueError(f"Change order {co_id} not found")
co = self.change_orders[co_id]
item_id = f"{co_id}-{len(co.line_items)+1:03d}"
item = ChangeOrderItem(
id=item_id,
description=description,
quantity=quantity,
unit=unit,
unit_price=unit_price,
total_price=quantity * unit_price,
spec_section=spec_section,
csi_code=csi_code
)
co.line_items.append(item)
# Update totals
self._recalculate_totals(co)
return item
def set_cost_breakdown(self, co_id: str, labor: float = 0,
materials: float = 0, equipment: float = 0,
subcontractor: float = 0) -> CostBreakdown:
"""Set cost breakdown and calculate markups."""
if co_id not in self.change_orders:
raise ValueError(f"Change order {co_id} not found")
co = self.change_orders[co_id]
direct = labor + materials + equipment + subcontractor
co.cost_breakdown = CostBreakdown(
labor=labor,
materials=materials,
equipment=equipment,
subcontractor=subcontractor,
overhead=direct * self.markup_rates["overhead"],
profit=direct * self.markup_rates["profit"],
bond=direct * self.markup_rates["bond"]
)
co.proposed_amount = co.cost_breakdown.total
return co.cost_breakdown
def _recalculate_totals(self, co: ChangeOrder):
"""Recalculate change order totals from line items."""
if co.line_items:
direct_cost = sum(item.total_price for item in co.line_items)
co.cost_breakdown.labor = direct_cost * 0.4 # Estimate
co.cost_breakdown.materials = direct_cost * 0.4
co.cost_breakdown.equipment = direct_cost * 0.1
co.cost_breakdown.subcontractor = direct_cost * 0.1
co.cost_breakdown.overhead = direct_cost * self.markup_rates["overhead"]
co.cost_breakdown.profit = direct_cost * self.markup_rates["profit"]
co.cost_breakdown.bond = direct_cost * self.markup_rates["bond"]
co.proposed_amount = co.cost_breakdown.total
def submit_change_order(self, co_id: str) -> ChangeOrder:
"""Submit change order for review."""
if co_id not in self.change_orders:
raise ValueError(f"Change order {co_id} not found")
co = self.change_orders[co_id]
co.status = ChangeOrderStatus.SUBMITTED
co.submitted_date = datetime.now()
self._add_note(co, "Submitted for review")
return co
def approve_change_order(self, co_id: str, approved_amount: float,
approved_time: int = 0) -> ChangeOrder:
"""Approve change order."""
if co_id not in self.change_orders:
raise ValueError(f"Change order {co_id} not found")
co = self.change_orders[co_id]
co.status = ChangeOrderStatus.APPROVED
co.approved_date = datetime.now()
co.approved_amount = approved_amount
co.approved_time_days = approved_time
self._add_note(co, f"Approved: ${approved_amount:,.2f}, {approved_time} days")
return co
def reject_change_order(self, co_id: str, reason: str) -> ChangeOrder:
"""Reject change order."""
if co_id not in self.change_orders:
raise ValueError(f"Change order {co_id} not found")
co = self.change_orders[co_id]
co.status = ChangeOrderStatus.REJECTED
self._add_note(co, f"Rejected: {reason}")
return co
def execute_change_order(self, co_id: str) -> ChangeOrder:
"""Mark change order as executed."""
if co_id not in self.change_orders:
raise ValueError(f"Change order {co_id} not found")
co = self.change_orders[co_id]
if co.status != ChangeOrderStatus.APPROVED:
raise ValueError("Change order must be approved before execution")
co.status = ChangeOrderStatus.EXECUTED
co.executed_date = datetime.now()
self._add_note(co, "Executed")
return co
def _add_note(self, co: ChangeOrder, text: str):
"""Add note to change order."""
co.notes.append({
"timestamp": datetime.now().isoformat(),
"text": text
})
def add_reference(self, co_id: str, ref_type: str, reference: str):
"""Add reference document to change order."""
if co_id not in self.change_orders:
raise ValueError(f"Change order {co_id} not found")
co = self.change_orders[co_id]
if ref_type == "rfi":
co.rfi_references.append(reference)
elif ref_type == "drawing":
co.drawing_references.append(reference)
elif ref_type == "photo":
co.photo_attachments.append(reference)
elif ref_type == "backup":
co.backup_documents.append(reference)
def get_summary(self) -> Dict:
"""Get change order summary statistics."""
total_approved = sum(
co.approved_amount for co in self.change_orders.values()
if co.status in [ChangeOrderStatus.APPROVED, ChangeOrderStatus.EXECUTED]
)
total_pending = sum(
co.proposed_amount for co in self.change_orders.values()
if co.status in [ChangeOrderStatus.SUBMITTED, ChangeOrderStatus.UNDER_REVIEW,
ChangeOrderStatus.PRICING, ChangeOrderStatus.NEGOTIATING]
)
by_type = {}
for co in self.change_orders.values():
t = co.change_type.value
by_type[t] = by_type.get(t, 0) + (co.approved_amount or co.proposed_amount)
by_status = {}
for co in self.change_orders.values():
s = co.status.value
by_status[s] = by_status.get(s, 0) + 1
return {
"original_contract": self.original_contract,
"total_approved": total_approved,
"total_pending": total_pending,
"revised_contract": self.original_contract + total_approved,
"change_order_count": len(self.change_orders),
"change_percentage": (total_approved / self.original_contract * 100) if self.original_contract else 0,
"by_type": by_type,
"by_status": by_status
}
def generate_log(self) -> str:
"""Generate change order log."""
summary = self.get_summary()
lines = [
"# Change Order Log",
"",
f"**Project:** {self.project_name}",
f"**Date:** {datetime.now().strftime('%Y-%m-%d')}",
"",
"## Summary",
"",
f"| Metric | Amount |",
f"|--------|--------|",
f"| Original Contract | ${summary['original_contract']:,.2f} |",
f"| Approved Changes | ${summary['total_approved']:,.2f} |",
f"| Pending Changes | ${summary['total_pending']:,.2f} |",
f"| **Revised Contract** | **${summary['revised_contract']:,.2f}** |",
f"| Change % | {summary['change_percentage']:.1f}% |",
"",
"## Change Orders",
"",
"| # | Title | Type | Status | Proposed | Approved | Time |",
"|---|-------|------|--------|----------|----------|------|"
]
for co in sorted(self.change_orders.values(), key=lambda x: x.number):
lines.append(
f"| {co.number} | {co.title[:30]} | {co.change_type.value} | "
f"{co.status.value} | ${co.proposed_amount:,.0f} | "
f"${co.approved_amount:,.0f} | {co.approved_time_days}d |"
)
return "\n".join(lines)
def generate_co_document(self, co_id: str) -> str:
"""Generate formal change order document."""
if co_id not in self.change_orders:
return "Change order not found"
co = self.change_orders[co_id]
lines = [
f"# CHANGE ORDER NO. {co.number}",
"",
f"**Project:** {self.project_name}",
f"**Change Order ID:** {co.id}",
f"**Date:** {co.submitted_date.strftime('%Y-%m-%d') if co.submitted_date else 'Draft'}",
"",
"---",
"",
f"## Description of Change",
"",
co.description,
"",
f"**Type:** {co.change_type.value.replace('_', ' ').title()}",
"",
]
if co.line_items:
lines.extend([
"## Schedule of Values",
"",
"| Item | Description | Qty | Unit | Unit Price | Total |",
"|------|-------------|-----|------|------------|-------|"
])
for item in co.line_items:
lines.append(
f"| {item.id} | {item.description} | {item.quantity} | "
f"{item.unit} | ${item.unit_price:,.2f} | ${item.total_price:,.2f} |"
)
lines.append("")
lines.extend([
"## Cost Summary",
"",
f"| Category | Amount |",
f"|----------|--------|",
f"| Labor | ${co.cost_breakdown.labor:,.2f} |",
f"| Materials | ${co.cost_breakdown.materials:,.2f} |",
f"| Equipment | ${co.cost_breakdown.equipment:,.2f} |",
f"| Subcontractor | ${co.cost_breakdown.subcontractor:,.2f} |",
f"| Overhead | ${co.cost_breakdown.overhead:,.2f} |",
f"| Profit | ${co.cost_breakdown.profit:,.2f} |",
f"| Bond | ${co.cost_breakdown.bond:,.2f} |",
f"| **Total** | **${co.cost_breakdown.total:,.2f}** |",
"",
f"## Time Impact",
"",
f"Proposed Extension: **{co.proposed_time_days} days**",
f"Critical Path Impact: {'Yes' if co.impacts_critical_path else 'No'}",
"",
])
if co.rfi_references:
lines.extend([
"## References",
"",
f"- RFIs: {', '.join(co.rfi_references)}",
f"- Drawings: {', '.join(co.drawing_references)}" if co.drawing_references else "",
])
return "\n".join(lines)Quick Start
# Initialize manager
manager = ChangeOrderManager(
project_id="PRJ-001",
project_name="Office Tower",
original_contract=5000000.0
)
# Create change order
co = manager.create_change_order(
title="Additional Structural Steel",
description="Add steel reinforcement at Level 5 per RFI-042",
change_type=ChangeType.DESIGN_ERROR,
created_by="Project Manager"
)
# Add line items
manager.add_line_item(
co.id,
"W12x26 Steel Beam",
quantity=450,
unit="LF",
unit_price=85.00,
csi_code="05 12 00"
)
# Or set cost breakdown directly
manager.set_cost_breakdown(
co.id,
labor=15000,
materials=25000,
equipment=2000,
subcontractor=5000
)
# Add references
manager.add_reference(co.id, "rfi", "RFI-042")
manager.add_reference(co.id, "drawing", "S-501 Rev 2")
# Submit for approval
manager.submit_change_order(co.id)
# Approve (with negotiated amount)
manager.approve_change_order(co.id, approved_amount=50000, approved_time=5)
# Generate documents
print(manager.generate_co_document(co.id))
print(manager.generate_log())Requirements
pip install (no external dependencies)