budget-variance-analyzer

name	budget-variance-analyzer
description	Analyze budget vs actual cost variances. Identify overruns, forecast final costs, and generate variance reports.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"💰","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":["python3"]}}}

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"
    ON_BUDGET = "on_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