Budget Variance Analyzer
Overview
Analyze construction budget variances between estimated and actual costs. Track cost performance by category, identify concerning trends, forecast final costs, and provide actionable insights for cost control.
Variance Analysis Framework
┌─────────────────────────────────────────────────────────────────┐
│ BUDGET VARIANCE ANALYSIS │
├─────────────────────────────────────────────────────────────────┤
│ │
│ Budget vs Actual = Variance → Forecast │
│ ────── ────── ──────── ──────── │
│ 📋 Original 💰 Spent 📊 Over/Under 🔮 EAC │
│ 📝 Revised 📈 Committed 📉 Trend 📋 ETC │
│ 🎯 Baseline 🧾 Invoiced ⚠️ Alerts 📊 VAC │
│ │
│ EAC = Estimate at Completion │
│ ETC = Estimate to Complete │
│ VAC = Variance at Completion │
│ │
└─────────────────────────────────────────────────────────────────┘Technical Implementation
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from datetime import datetime, timedelta
from enum import Enum
import statistics
class CostCategory(Enum):
LABOR = "labor"
MATERIALS = "materials"
EQUIPMENT = "equipment"
SUBCONTRACTOR = "subcontractor"
GENERAL_CONDITIONS = "general_conditions"
OVERHEAD = "overhead"
CONTINGENCY = "contingency"
FEE = "fee"
class VarianceStatus(Enum):
ON_BUDGET = "on_budget"
UNDER_BUDGET = "under_budget"
OVER_BUDGET = "over_budget"
CRITICAL = "critical"
@dataclass
class CostCode:
code: str
description: str
category: CostCategory
original_budget: float
revised_budget: float = 0.0
committed: float = 0.0
actual: float = 0.0
forecast: float = 0.0
percent_complete: float = 0.0
@property
def budget(self) -> float:
return self.revised_budget if self.revised_budget else self.original_budget
@property
def variance(self) -> float:
return self.budget - self.actual
@property
def variance_percent(self) -> float:
return (self.variance / self.budget * 100) if self.budget else 0
@property
def committed_variance(self) -> float:
return self.budget - self.committed
@property
def estimate_to_complete(self) -> float:
if self.percent_complete >= 100:
return 0
return self.forecast - self.actual
@property
def variance_at_completion(self) -> float:
return self.budget - self.forecast
@dataclass
class CostSnapshot:
date: datetime
cost_code: str
actual: float
committed: float
forecast: float
@dataclass
class VarianceReport:
report_date: datetime
project_name: str
total_budget: float
total_actual: float
total_committed: float
total_forecast: float
variance: float
variance_percent: float
eac: float
etc: float
vac: float
variances_by_category: Dict[str, Dict]
critical_items: List[CostCode]
trend: str
class BudgetVarianceAnalyzer:
"""Analyze construction budget variances."""
# Thresholds for variance status
VARIANCE_THRESHOLDS = {
"under_budget": 0.05, # 5% under
"on_budget": 0.00,
"over_budget": -0.05, # 5% over
"critical": -0.10 # 10% over
}
def __init__(self, project_name: str, original_budget: float):
self.project_name = project_name
self.original_budget = original_budget
self.cost_codes: Dict[str, CostCode] = {}
self.snapshots: List[CostSnapshot] = []
self.contingency_used = 0.0
def add_cost_code(self, code: str, description: str,
category: CostCategory, budget: float) -> CostCode:
"""Add cost code to budget."""
cost_code = CostCode(
code=code,
description=description,
category=category,
original_budget=budget,
revised_budget=budget,
forecast=budget
)
self.cost_codes[code] = cost_code
return cost_code
def import_budget(self, items: List[Dict]) -> int:
"""Import budget from list of items."""
count = 0
for item in items:
self.add_cost_code(
item['code'],
item['description'],
CostCategory(item.get('category', 'other')),
item['budget']
)
count += 1
return count
def update_actual(self, code: str, actual: float) -> CostCode:
"""Update actual cost for cost code."""
if code not in self.cost_codes:
raise ValueError(f"Cost code {code} not found")
cost_code = self.cost_codes[code]
cost_code.actual = actual
# Record snapshot
self._record_snapshot(code)
return cost_code
def update_committed(self, code: str, committed: float) -> CostCode:
"""Update committed cost (contracts, POs)."""
if code not in self.cost_codes:
raise ValueError(f"Cost code {code} not found")
cost_code = self.cost_codes[code]
cost_code.committed = committed
# Update forecast if committed exceeds current forecast
if committed > cost_code.forecast:
cost_code.forecast = committed
self._record_snapshot(code)
return cost_code
def update_forecast(self, code: str, forecast: float,
percent_complete: float = None) -> CostCode:
"""Update forecast for cost code."""
if code not in self.cost_codes:
raise ValueError(f"Cost code {code} not found")
cost_code = self.cost_codes[code]
cost_code.forecast = forecast
if percent_complete is not None:
cost_code.percent_complete = percent_complete
self._record_snapshot(code)
return cost_code
def revise_budget(self, code: str, new_budget: float,
reason: str = "") -> CostCode:
"""Revise budget for cost code."""
if code not in self.cost_codes:
raise ValueError(f"Cost code {code} not found")
cost_code = self.cost_codes[code]
cost_code.revised_budget = new_budget
cost_code.forecast = new_budget
return cost_code
def use_contingency(self, amount: float, target_code: str,
reason: str = "") -> float:
"""Use contingency to cover variance."""
self.contingency_used += amount
if target_code in self.cost_codes:
cc = self.cost_codes[target_code]
cc.revised_budget += amount
return self.contingency_used
def _record_snapshot(self, code: str):
"""Record cost snapshot for trending."""
if code not in self.cost_codes:
return
cc = self.cost_codes[code]
snapshot = CostSnapshot(
date=datetime.now(),
cost_code=code,
actual=cc.actual,
committed=cc.committed,
forecast=cc.forecast
)
self.snapshots.append(snapshot)
def get_variance_status(self, variance_percent: float) -> VarianceStatus:
"""Determine variance status."""
if variance_percent >= self.VARIANCE_THRESHOLDS["under_budget"]:
return VarianceStatus.UNDER_BUDGET
elif variance_percent >= self.VARIANCE_THRESHOLDS["over_budget"]:
return VarianceStatus.ON_BUDGET
elif variance_percent >= self.VARIANCE_THRESHOLDS["critical"]:
return VarianceStatus.OVER_BUDGET
else: