tools-only 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):
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:
return VarianceStatus.CRITICAL
def analyze_by_category(self) -> Dict[str, Dict]:
"""Analyze variances by cost category."""
by_category = {}
for category in CostCategory:
codes = [cc for cc in self.cost_codes.values()
if cc.category == category]
if not codes:
continue
budget = sum(cc.budget for cc in codes)
actual = sum(cc.actual for cc in codes)
committed = sum(cc.committed for cc in codes)
forecast = sum(cc.forecast for cc in codes)
variance = budget - actual
variance_pct = (variance / budget * 100) if budget else 0
by_category[category.value] = {
"budget": budget,
"actual": actual,
"committed": committed,
"forecast": forecast,
"variance": variance,
"variance_percent": variance_pct,
"status": self.get_variance_status(variance_pct / 100).value
}
return by_category
def identify_critical_items(self, threshold: float = -0.10) -> List[CostCode]:
"""Identify cost codes with critical variance."""
critical = []
for cc in self.cost_codes.values():
variance_pct = cc.variance_percent / 100
if variance_pct < threshold:
critical.append(cc)
return sorted(critical, key=lambda x: x.variance_percent)
def calculate_cpi(self) -> float:
"""Calculate Cost Performance Index."""
total_budget = sum(cc.budget * cc.percent_complete / 100
for cc in self.cost_codes.values())
total_actual = sum(cc.actual for cc in self.cost_codes.values())
if total_actual == 0:
return 1.0
return total_budget / total_actual
def forecast_eac(self, method: str = "cpi") -> float:
"""Forecast Estimate at Completion."""
total_budget = sum(cc.budget for cc in self.cost_codes.values())
total_actual = sum(cc.actual for cc in self.cost_codes.values())
if method == "cpi":
cpi = self.calculate_cpi()
if cpi == 0:
return total_budget
return total_actual + (total_budget - total_actual * cpi) / cpi
elif method == "forecast":
return sum(cc.forecast for cc in self.cost_codes.values())
elif method == "committed":
return sum(max(cc.committed, cc.actual) for cc in self.cost_codes.values())
return total_budget
def analyze_trend(self, code: str = None) -> Dict:
"""Analyze variance trend over time."""
if code:
snapshots = [s for s in self.snapshots if s.cost_code == code]
else:
snapshots = self.snapshots
if len(snapshots) < 2:
return {"trend": "insufficient_data", "slope": 0}
# Group by week
weekly_actuals = {}
for s in snapshots:
week = s.date.isocalendar()[1]
if week not in weekly_actuals:
weekly_actuals[week] = []
weekly_actuals[week].append(s.actual)
# Calculate trend
weeks = sorted(weekly_actuals.keys())
if len(weeks) < 2:
return {"trend": "insufficient_data", "slope": 0}
week_avgs = [statistics.mean(weekly_actuals[w]) for w in weeks]
# Simple slope calculation
n = len(weeks)
x_mean = sum(range(n)) / n
y_mean = sum(week_avgs) / n
numerator = sum((i - x_mean) * (week_avgs[i] - y_mean) for i in range(n))
denominator = sum((i - x_mean) ** 2 for i in range(n))
slope = numerator / denominator if denominator else 0
if slope > 1000:
trend = "increasing_rapidly"
elif slope > 0:
trend = "increasing"
elif slope < -1000:
trend = "decreasing_rapidly"
elif slope < 0:
trend = "decreasing"
else:
trend = "stable"
return {"trend": trend, "slope": slope, "data_points": len(snapshots)}
def generate_variance_report(self) -> VarianceReport:
"""Generate comprehensive variance report."""
total_budget = sum(cc.budget for cc in self.cost_codes.values())
total_actual = sum(cc.actual for cc in self.cost_codes.values())
total_committed = sum(cc.committed for cc in self.cost_codes.values())
total_forecast = sum(cc.forecast for cc in self.cost_codes.values())
variance = total_budget - total_actual
variance_pct = (variance / total_budget * 100) if total_budget else 0
eac = self.forecast_eac()
etc = eac - total_actual
vac = total_budget - eac
trend_analysis = self.analyze_trend()
return VarianceReport(
report_date=datetime.now(),
project_name=self.project_name,
total_budget=total_budget,
total_actual=total_actual,
total_committed=total_committed,
total_forecast=total_forecast,
variance=variance,
variance_percent=variance_pct,
eac=eac,
etc=etc,
vac=vac,
variances_by_category=self.analyze_by_category(),
critical_items=self.identify_critical_items(),
trend=trend_analysis["trend"]
)
def generate_report_markdown(self, report: VarianceReport) -> str:
"""Generate markdown report."""
lines = [
"# Budget Variance Report",
"",
f"**Project:** {report.project_name}",
f"**Report Date:** {report.report_date.strftime('%Y-%m-%d')}",
"",
"## Executive Summary",
"",
f"| Metric | Amount |",
f"|--------|--------|",
f"| Total Budget | ${report.total_budget:,.0f} |",
f"| Actual to Date | ${report.total_actual:,.0f} |",
f"| Committed | ${report.total_committed:,.0f} |",
f"| **Variance** | **${report.variance:,.0f} ({report.variance_percent:.1f}%)** |",
"",
"## Forecast",
"",
f"| Metric | Amount |",
f"|--------|--------|",
f"| Estimate at Completion (EAC) | ${report.eac:,.0f} |",
f"| Estimate to Complete (ETC) | ${report.etc:,.0f} |",
f"| Variance at Completion (VAC) | ${report.vac:,.0f} |",
f"| CPI | {self.calculate_cpi():.2f} |",
f"| Trend | {report.trend} |",
"",
"## By Category",
"",
"| Category | Budget | Actual | Variance | Status |",
"|----------|--------|--------|----------|--------|"
]
for cat, data in report.variances_by_category.items():
status_icon = "🟢" if data["status"] == "under_budget" else "🟡" if data["status"] == "on_budget" else "🔴"
lines.append(
f"| {cat} | ${data['budget']:,.0f} | ${data['actual']:,.0f} | "
f"${data['variance']:,.0f} ({data['variance_percent']:.1f}%) | {status_icon} |"
)
if report.critical_items:
lines.extend([
"",
"## Critical Items (>10% Over Budget)",
"",
"| Code | Description | Budget | Actual | Variance |",
"|------|-------------|--------|--------|----------|"
])
for cc in report.critical_items[:10]:
lines.append(
f"| {cc.code} | {cc.description[:25]} | ${cc.budget:,.0f} | "
f"${cc.actual:,.0f} | ${cc.variance:,.0f} ({cc.variance_percent:.1f}%) |"
)
return "\n".join(lines)
Quick Start
# Initialize analyzer
analyzer = BudgetVarianceAnalyzer("Office Tower", 5000000)
# Add cost codes
analyzer.add_cost_code("01-100", "Project Management", CostCategory.GENERAL_CONDITIONS, 150000)
analyzer.add_cost_code("03-100", "Concrete", CostCategory.MATERIALS, 400000)
analyzer.add_cost_code("05-100", "Structural Steel", CostCategory.SUBCONTRACTOR, 800000)
analyzer.add_cost_code("15-100", "Mechanical", CostCategory.SUBCONTRACTOR, 600000)
analyzer.add_cost_code("16-100", "Electrical", CostCategory.SUBCONTRACTOR, 450000)
# Update actuals
analyzer.update_actual("01-100", 120000)
analyzer.update_actual("03-100", 380000)
analyzer.update_actual("05-100", 850000) # Over budget
analyzer.update_actual("15-100", 300000)
analyzer.update_actual("16-100", 200000)
# Update committed
analyzer.update_committed("05-100", 900000)
# Update forecast
analyzer.update_forecast("05-100", 920000, percent_complete=85)
# Analyze
cpi = analyzer.calculate_cpi()
print(f"Cost Performance Index: {cpi:.2f}")
critical = analyzer.identify_critical_items()
print(f"Critical items: {len(critical)}")
# Generate report
report = analyzer.generate_variance_report()
print(analyzer.generate_report_markdown(report))
Requirements
pip install (no external dependencies)