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  , , , , 
 dataclasses  dataclass, field
 enum  Enum


 ():
    
    UNDER_BUDGET = 
    ON_BUDGET = 
    OVER_BUDGET = 
    CRITICAL = 


 ():
    
    LABOR = 
    MATERIAL = 
    EQUIPMENT = 
    SUBCONTRACTOR = 
    OVERHEAD = 
    CONTINGENCY = 
    OTHER = 


 ():
    
    SCOPE_CHANGE = 
    QUANTITY_CHANGE = 
    PRICE_ESCALATION = 
    PRODUCTIVITY = 
    REWORK = 
    DELAY = 
    UNFORESEEN = 
    ESTIMATE_ERROR = 
    OTHER = 



 :
    
    item_code: 
    description: 
    category: CostCategory
    original_budget: 
    current_budget:   
    committed_cost:   
    actual_cost:      
    forecast_cost:    
    percent_complete: 
    notes:  = 


     () -> :
        
         .current_budget - .forecast_cost


     () -> :
        
         .current_budget == :
             
         (.variance_amount / .current_budget) * 


     () -> VarianceStatus:
        
        pct = .variance_percent
         pct > :
             VarianceStatus.UNDER_BUDGET
         pct >= -:
             VarianceStatus.ON_BUDGET
         pct >= -:
             VarianceStatus.OVER_BUDGET
        :
             VarianceStatus.CRITICAL



 :
    
    record_id: 
    item_code: 
    variance_amount: 
    cause: VarianceCause
    explanation: 
    recorded_date: date
    recorded_by: 
    approved:  = 
    approval_date: [date] = 



 :
    
    name: 
    description: 
    adjustments: [, ]  
    total_forecast: 
    variance_from_budget: 


 :
    

    VARIANCE_THRESHOLD_WARNING = -  
    VARIANCE_THRESHOLD_CRITICAL = -  

     ():
        .project_name = project_name
        .original_budget = original_budget
        .currency = currency
        .items: [, BudgetItem] = {}
        .variance_records: [VarianceRecord] = []
        .history: [[, ]] = []

     () -> BudgetItem:
        
        forecast = (committed, actual / percent_complete * )  percent_complete >   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
        )

        .items[item_code] = item
         item

     ():
        
         item_code   .items:
             ValueError()

        item = .items[item_code]

         committed   :
            item.committed_cost = committed
         actual   :
            item.actual_cost = actual
         percent_complete   :
            item.percent_complete = percent_complete
         forecast   :
            item.forecast_cost = forecast
        :
            
             item.percent_complete > :
                item.forecast_cost = item.actual_cost / item.percent_complete * 
            :
                item.forecast_cost = (item.committed_cost, item.current_budget)

        ._record_history()

     ():
        
         item_code   .items:
             ValueError()

        .items[item_code].current_budget += amount
        .items[item_code].notes += 

     () -> VarianceRecord:
        
        item = .items.get(item_code)
          item:
             ValueError()

        record_id = 

        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
        )

        .variance_records.append(record)
         record

     ():
        
        snapshot = {
            : date.today().isoformat(),
            : (i.current_budget  i  .items.values()),
            : (i.committed_cost  i  .items.values()),
            : (i.actual_cost  i  .items.values()),
            : (i.forecast_cost  i  .items.values())
        }
        .history.append(snapshot)

     () -> [, ]:
        
        total_budget = (i.current_budget  i  .items.values())
        total_committed = (i.committed_cost  i  .items.values())
        total_actual = (i.actual_cost  i  .items.values())
        total_forecast = (i.forecast_cost  i  .items.values())

        variance = total_budget - total_forecast
        variance_pct = (variance / total_budget * )  total_budget >   

        
        by_category = {}
         item  .items.values():
            cat = item.category.value
             cat   by_category:
                by_category[cat] = {
                    : , : , : , : 
                }
            by_category[cat][] += item.current_budget
            by_category[cat][] += item.actual_cost
            by_category[cat][] += item.forecast_cost
            by_category[cat][] += item.variance_amount

        
        critical = [i  i  .items.values()  i.status == VarianceStatus.CRITICAL]
        over_budget = [i  i  .items.values()  i.status == VarianceStatus.OVER_BUDGET]

         {
            : .project_name,
            : .currency,
            : .original_budget,
            : total_budget,
            : total_committed,
            : total_actual,
            : total_forecast,
            : variance,
            : (variance_pct, ),
            :   variance >=   ,
            : by_category,
            : (critical),
            : (over_budget),
            : total_budget - .original_budget
        }

     () -> [BudgetItem]:
        
         [i  i  .items.values()
                 i.status  [VarianceStatus.CRITICAL, VarianceStatus.OVER_BUDGET]]

     () -> [, ForecastScenario]:
        
        current_forecast = (i.forecast_cost  i  .items.values())
        current_budget = (i.current_budget  i  .items.values())

        scenarios = {
            : ForecastScenario(
                name=,
                description=,
                adjustments={},
                total_forecast=current_forecast * optimistic_factor,
                variance_from_budget=current_budget - (current_forecast * optimistic_factor)
            ),
            : ForecastScenario(
                name=,
                description=,
                adjustments={},
                total_forecast=current_forecast,
                variance_from_budget=current_budget - current_forecast
            ),
            : ForecastScenario(
                name=,
                description=,
                adjustments={},
                total_forecast=current_forecast * pessimistic_factor,
                variance_from_budget=current_budget - (current_forecast * pessimistic_factor)
            )
        }

         scenarios

     () -> [, ]:
        
         (.history) < :
             {: }

        forecasts = [h[]  h  .history]
        actuals = [h[]  h  .history]

        
        forecast_trend = forecasts[-] - forecasts[]
        actual_trend = actuals[-] - actuals[]

         {
            :   forecast_trend >   ,
            : forecast_trend,
            :   actual_trend >   ,
            : actual_trend,
            : (.history)
        }

     ():
        
         pd.ExcelWriter(output_path, engine=)  writer:
            
            summary = .calculate_summary()
            summary_df = pd.DataFrame([
                {: k, : v}
                 k, v  summary.items()
                  (v, )
            ])
            summary_df.to_excel(writer, sheet_name=, index=)

            
            items_data = []
             item  .items.values():
                items_data.append({
                    : item.item_code,
                    : item.description,
                    : item.category.value,
                    : item.current_budget,
                    : item.committed_cost,
                    : item.actual_cost,
                    : item.forecast_cost,
                    : item.variance_amount,
                    : (item.variance_percent, ),
                    : item.status.value,
                    : item.percent_complete
                })

            pd.DataFrame(items_data).to_excel(writer, sheet_name=, index=)

            
             .variance_records:
                records_df = pd.DataFrame([{
                    : r.record_id,
                    : r.item_code,
                    : r.variance_amount,
                    : r.cause.value,
                    : r.explanation,
                    : r.recorded_date,
                    : r.recorded_by
                }  r  .variance_records])
                records_df.to_excel(writer, sheet_name=, index=)

         output_path