Budget Variance Analyzer Agent Skill

┌─────────────────────────────────────────────────────────────────┐
│                  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                                   │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

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:
            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)

# 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))

pip install (no external dependencies)