tools-only Budget Variance Analyzer
Business Case
Problem Statement
Cost overruns surprise project teams:
- Late detection of budget issues
- No systematic variance analysis
- Difficult to forecast final costs
- Unclear root causes
Solution
Systematic budget variance analysis that tracks costs against budget, identifies trends, and forecasts final project costs.
Business Value
- Early warning - Detect overruns early
- Forecasting - Predict final costs
- Accountability - Track variance causes
- Decision support - Informed cost decisions
Technical Implementation
import pandas as pd
from datetime import datetime, date
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
class VarianceStatus(Enum):
"""Variance status."""
UNDER_BUDGET = "under_budget"
OVER_BUDGET = "over_budget"
CRITICAL = "critical"
class CostCategory(Enum):
"""Cost categories."""
LABOR = "labor"
MATERIAL = "material"
EQUIPMENT = "equipment"
SUBCONTRACTOR = "subcontractor"
OVERHEAD = "overhead"
CONTINGENCY = "contingency"
OTHER = "other"
class VarianceCause(Enum):
"""Common variance causes."""
SCOPE_CHANGE = "scope_change"
QUANTITY_CHANGE = "quantity_change"
PRICE_ESCALATION = "price_escalation"
PRODUCTIVITY = "productivity"
REWORK = "rework"
DELAY = "delay"
UNFORESEEN = "unforeseen"
ESTIMATE_ERROR = "estimate_error"
OTHER = "other"
@dataclass
class BudgetItem:
"""Single budget line item."""
item_code: str
description: str
category: CostCategory
original_budget: float
current_budget: float # After approved changes
committed_cost: float # Contracts, POs
actual_cost: float # Paid/invoiced
forecast_cost: float # Estimate at completion
percent_complete: float
notes: str = ""
@property
def variance_amount(self) -> float:
"""Budget variance (negative = over budget)."""
return self.current_budget - self.forecast_cost
@property
def variance_percent(self) -> float:
"""Variance as percentage."""
if self.current_budget == 0:
return 0
return (self.variance_amount / self.current_budget) * 100
@property
def status(self) -> VarianceStatus:
"""Determine variance status."""
pct = self.variance_percent
if pct > 5:
return VarianceStatus.UNDER_BUDGET
elif pct >= -5:
return VarianceStatus.ON_BUDGET
elif pct >= -15:
return VarianceStatus.OVER_BUDGET
else:
return VarianceStatus.CRITICAL
@dataclass
class VarianceRecord:
"""Record of budget variance."""
record_id: str
item_code: str
variance_amount: float
cause: VarianceCause
explanation: str
recorded_date: date
recorded_by: str
approved: bool = False
approval_date: Optional[date] = None
@dataclass
class ForecastScenario:
"""Cost forecast scenario."""
name: str
description: str
adjustments: Dict[str, float] # item_code: adjustment amount
total_forecast: float
variance_from_budget: float
class BudgetVarianceAnalyzer:
"""Analyze budget vs actual cost variances."""
VARIANCE_THRESHOLD_WARNING = -0.05 # -5%
VARIANCE_THRESHOLD_CRITICAL = -0.15 # -15%
def __init__(self, project_name: str, original_budget: float, currency: str = "USD"):
self.project_name = project_name
self.original_budget = original_budget
self.currency = currency
self.items: Dict[str, BudgetItem] = {}
self.variance_records: List[VarianceRecord] = []
self.history: List[Dict[str, Any]] = []
def add_budget_item(self,
item_code: str,
description: str,
category: CostCategory,
budget: float,
committed: float = 0,
actual: float = 0,
percent_complete: float = 0) -> BudgetItem:
"""Add budget line item."""
forecast = max(committed, actual / percent_complete * 100) if percent_complete > 0 else budget
item = BudgetItem(
item_code=item_code,
description=description,
category=category,
original_budget=budget,
current_budget=budget,
committed_cost=committed,
actual_cost=actual,
forecast_cost=forecast,
percent_complete=percent_complete
)
self.items[item_code] = item
return item
def update_costs(self, item_code: str,
committed: float = None,
actual: float = None,
percent_complete: float = None,
forecast: float = None):
"""Update item costs."""
if item_code not in self.items:
raise ValueError(f"Item {item_code} not found")
item = self.items[item_code]
if committed is not None:
item.committed_cost = committed
if actual is not None:
item.actual_cost = actual
if percent_complete is not None:
item.percent_complete = percent_complete
if forecast is not None:
item.forecast_cost = forecast
else:
# Auto-calculate forecast
if item.percent_complete > 0:
item.forecast_cost = item.actual_cost / item.percent_complete * 100
else:
item.forecast_cost = max(item.committed_cost, item.current_budget)
self._record_history()
def adjust_budget(self, item_code: str, amount: float, reason: str):
"""Adjust current budget (approved change)."""
if item_code not in self.items:
raise ValueError(f"Item {item_code} not found")
self.items[item_code].current_budget += amount
self.items[item_code].notes += f"\nBudget adjusted by {amount}: {reason}"
def record_variance(self,
item_code: str,
cause: VarianceCause,
explanation: str,
recorded_by: str) -> VarianceRecord:
"""Record variance explanation."""
item = self.items.get(item_code)
if not item:
raise ValueError(f"Item {item_code} not found")
record_id = f"VAR-{len(self.variance_records) + 1:04d}"
record = VarianceRecord(
record_id=record_id,
item_code=item_code,
variance_amount=item.variance_amount,
cause=cause,
explanation=explanation,
recorded_date=date.today(),
recorded_by=recorded_by
)
self.variance_records.append(record)
return record
def _record_history(self):
"""Record current state to history."""
snapshot = {
'date': date.today().isoformat(),
'total_budget': sum(i.current_budget for i in self.items.values()),
'total_committed': sum(i.committed_cost for i in self.items.values()),
'total_actual': sum(i.actual_cost for i in self.items.values()),
'total_forecast': sum(i.forecast_cost for i in self.items.values())
}
self.history.append(snapshot)
def calculate_summary(self) -> Dict[str, Any]:
"""Calculate overall budget summary."""
total_budget = sum(i.current_budget for i in self.items.values())
total_committed = sum(i.committed_cost for i in self.items.values())
total_actual = sum(i.actual_cost for i in self.items.values())
total_forecast = sum(i.forecast_cost for i in self.items.values())
variance = total_budget - total_forecast
variance_pct = (variance / total_budget * 100) if total_budget > 0 else 0
# By category
by_category = {}
for item in self.items.values():
cat = item.category.value
if cat not in by_category:
by_category[cat] = {
'budget': 0, 'actual': 0, 'forecast': 0, 'variance': 0
}
by_category[cat]['budget'] += item.current_budget
by_category[cat]['actual'] += item.actual_cost
by_category[cat]['forecast'] += item.forecast_cost
by_category[cat]['variance'] += item.variance_amount
# Items needing attention
critical = [i for i in self.items.values() if i.status == VarianceStatus.CRITICAL]
over_budget = [i for i in self.items.values() if i.status == VarianceStatus.OVER_BUDGET]
return {
'project': self.project_name,
'currency': self.currency,
'original_budget': self.original_budget,
'current_budget': total_budget,
'committed': total_committed,
'actual': total_actual,
'forecast': total_forecast,
'variance': variance,
'variance_percent': round(variance_pct, 1),
'status': 'ON_TRACK' if variance >= 0 else 'OVER_BUDGET',
'by_category': by_category,
'critical_items': len(critical),
'over_budget_items': len(over_budget),
'contingency_used': total_budget - self.original_budget
}
def get_critical_items(self) -> List[BudgetItem]:
"""Get items with critical variances."""
return [i for i in self.items.values()
if i.status in [VarianceStatus.CRITICAL, VarianceStatus.OVER_BUDGET]]
def forecast_completion(self,
optimistic_factor: float = 0.95,
pessimistic_factor: float = 1.15) -> Dict[str, ForecastScenario]:
"""Generate forecast scenarios."""
current_forecast = sum(i.forecast_cost for i in self.items.values())
current_budget = sum(i.current_budget for i in self.items.values())
scenarios = {
'optimistic': ForecastScenario(
name="Optimistic",
description="Best case with no further overruns",
adjustments={},
total_forecast=current_forecast * optimistic_factor,
variance_from_budget=current_budget - (current_forecast * optimistic_factor)
),
'most_likely': ForecastScenario(
name="Most Likely",
description="Current trend continues",
adjustments={},
total_forecast=current_forecast,
variance_from_budget=current_budget - current_forecast
),
'pessimistic': ForecastScenario(
name="Pessimistic",
description="Additional overruns expected",
adjustments={},
total_forecast=current_forecast * pessimistic_factor,
variance_from_budget=current_budget - (current_forecast * pessimistic_factor)
)
}
return scenarios
def analyze_trends(self) -> Dict[str, Any]:
"""Analyze cost trends from history."""
if len(self.history) < 2:
return {'trend': 'insufficient_data'}
forecasts = [h['total_forecast'] for h in self.history]
actuals = [h['total_actual'] for h in self.history]
# Calculate trend direction
forecast_trend = forecasts[-1] - forecasts[0]
actual_trend = actuals[-1] - actuals[0]
return {
'forecast_trend': 'increasing' if forecast_trend > 0 else 'decreasing',
'forecast_change': forecast_trend,
'actual_trend': 'increasing' if actual_trend > 0 else 'stable',
'actual_change': actual_trend,
'data_points': len(self.history)
}
def export_variance_report(self, output_path: str):
"""Export detailed variance report to Excel."""
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
# Summary
summary = self.calculate_summary()
summary_df = pd.DataFrame([
{'Metric': k, 'Value': v}
for k, v in summary.items()
if not isinstance(v, dict)
])
summary_df.to_excel(writer, sheet_name='Summary', index=False)
# Line items
items_data = []
for item in self.items.values():
items_data.append({
'Code': item.item_code,
'Description': item.description,
'Category': item.category.value,
'Budget': item.current_budget,
'Committed': item.committed_cost,
'Actual': item.actual_cost,
'Forecast': item.forecast_cost,
'Variance $': item.variance_amount,
'Variance %': round(item.variance_percent, 1),
'Status': item.status.value,
'% Complete': item.percent_complete
})
pd.DataFrame(items_data).to_excel(writer, sheet_name='Line Items', index=False)
# Variance records
if self.variance_records:
records_df = pd.DataFrame([{
'ID': r.record_id,
'Item': r.item_code,
'Amount': r.variance_amount,
'Cause': r.cause.value,
'Explanation': r.explanation,
'Date': r.recorded_date,
'By': r.recorded_by
} for r in self.variance_records])
records_df.to_excel(writer, sheet_name='Variance Records', index=False)
return output_path
Quick Start
# Initialize analyzer
analyzer = BudgetVarianceAnalyzer(
project_name="Office Tower",
original_budget=50000000,
currency="USD"
)
# Add budget items
analyzer.add_budget_item("01-SITE", "Site Work", CostCategory.SUBCONTRACTOR, 2000000)
analyzer.add_budget_item("03-CONC", "Concrete", CostCategory.SUBCONTRACTOR, 8000000)
analyzer.add_budget_item("05-STEEL", "Structural Steel", CostCategory.SUBCONTRACTOR, 6000000)
# Update with actuals
analyzer.update_costs("03-CONC", committed=8500000, actual=4000000, percent_complete=45)
# Get summary
summary = analyzer.calculate_summary()
print(f"Variance: ${summary['variance']:,.0f} ({summary['variance_percent']}%)")
Common Use Cases
1. Monthly Cost Review
summary = analyzer.calculate_summary()
critical = analyzer.get_critical_items()
print(f"Items needing attention: {len(critical)}")
2. Record Variance Cause
analyzer.record_variance(
item_code="03-CONC",
cause=VarianceCause.PRICE_ESCALATION,
explanation="Steel rebar prices increased 15%",
recorded_by="Cost Manager"
)
3. Forecast Scenarios
scenarios = analyzer.forecast_completion()
for name, scenario in scenarios.items():
print(f"{scenario.name}: ${scenario.total_forecast:,.0f}")
Resources
- DDC Book: Chapter 3.1 - Cost Management
- Reference: PMI Cost Management