name: "cwicr-escalation" description: "Apply price escalation to CWICR estimates over time. Calculate inflation adjustments, material price indices, and labor rate increases." homepage: "https://datadrivenconstruction.io" metadata: {"openclaw": {"emoji": "🗄️", "os": ["darwin", "linux", "win32"], "homepage": "https://datadrivenconstruction.io", "requires": {"bins": ["python3"]}}}

CWICR Escalation Calculator

Business Case

Problem Statement

Construction costs change over time:

• Inflation affects all costs
• Material prices fluctuate
• Labor rates increase annually
• Long projects need escalation

Solution

Time-based cost escalation using historical indices, projected rates, and category-specific escalation factors.

Business Value

• Future pricing - Estimate costs at construction time
• Budget planning - Account for inflation
• Contract pricing - Escalation clauses
• Historical analysis - Adjust past costs to current

Technical Implementation

import pandas as pd
import numpy as np
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass
from datetime import datetime, date
from dateutil.relativedelta import relativedelta
from enum import Enum


class EscalationType(Enum):
    """Types of escalation."""
    LABOR = "labor"
    MATERIAL = "material"
    EQUIPMENT = "equipment"
    GENERAL = "general"


@dataclass
class EscalationIndex:
    """Escalation index for a period."""
    period: str  # YYYY-MM
    labor_index: float
    material_index: float
    equipment_index: float
    general_index: float


@dataclass
class EscalationResult:
    """Result of escalation calculation."""
    base_cost: float
    base_date: date
    target_date: date
    months: int
    escalation_rate: float
    escalation_amount: float
    escalated_cost: float
    by_category: Dict[str, Dict[str, float]]


# Historical escalation rates (annual %)
HISTORICAL_RATES = {
    2020: {'labor': 2.5, 'material': 1.8, 'equipment': 1.5, 'general': 2.0},
    2021: {'labor': 3.2, 'material': 8.5, 'equipment': 2.0, 'general': 4.5},
    2022: {'labor': 4.5, 'material': 12.0, 'equipment': 3.5, 'general': 7.0},
    2023: {'labor': 4.0, 'material': 5.0, 'equipment': 3.0, 'general': 4.0},
    2024: {'labor': 3.5, 'material': 3.0, 'equipment': 2.5, 'general': 3.0},
    2025: {'labor': 3.0, 'material': 2.5, 'equipment': 2.0, 'general': 2.5},
}

# Material-specific escalation factors
MATERIAL_ESCALATION = {
    'steel': 1.20,      # Higher volatility
    'lumber': 1.30,     # High volatility
    'concrete': 0.90,   # Lower volatility
    'copper': 1.25,     # Commodity driven
    'aluminum': 1.15,
    'plastic': 1.10,
    'glass': 0.95,
    'default': 1.00
}


class CWICREscalation:
    """Calculate cost escalation over time."""

    def __init__(self,
                 cwicr_data: pd.DataFrame = None,
                 custom_rates: Dict[int, Dict[str, float]] = None):
        self.cost_data = cwicr_data
        self.rates = custom_rates or HISTORICAL_RATES
        if cwicr_data is not None:
            self._index_data()

    def _index_data(self):
        """Index cost data."""
        if 'work_item_code' in self.cost_data.columns:
            self._code_index = self.cost_data.set_index('work_item_code')
        else:
            self._code_index = None

    def get_rate(self,
                  year: int,
                  category: EscalationType = EscalationType.GENERAL) -> float:
        """Get escalation rate for year and category."""
        year_rates = self.rates.get(year, self.rates.get(max(self.rates.keys())))
        return year_rates.get(category.value, year_rates.get('general', 3.0))

    def calculate_compound_factor(self,
                                   base_date: date,
                                   target_date: date,
                                   category: EscalationType = EscalationType.GENERAL) -> float:
        """Calculate compound escalation factor between dates."""

        if target_date <= base_date:
            return 1.0

        factor = 1.0
        current = base_date

        while current < target_date:
            year = current.year
            annual_rate = self.get_rate(year, category) / 100

            # Calculate months in this year
            year_end = date(year + 1, 1, 1)
            if target_date < year_end:
                months = (target_date.year - current.year) * 12 + target_date.month - current.month
            else:
                months = (year_end.year - current.year) * 12 + year_end.month - current.month

            # Apply monthly compound rate
            monthly_rate = (1 + annual_rate) ** (1/12) - 1
            factor *= (1 + monthly_rate) ** months

            current = year_end

        return factor

    def escalate_cost(self,
                       base_cost: float,
                       base_date: date,
                       target_date: date,
                       cost_breakdown: Dict[str, float] = None) -> EscalationResult:
        """Escalate cost from base date to target date."""

        if cost_breakdown is None:
            cost_breakdown = {
                'labor': base_cost * 0.40,
                'material': base_cost * 0.45,
                'equipment': base_cost * 0.15
            }

        months = (target_date.year - base_date.year) * 12 + target_date.month - base_date.month

        # Escalate each category
        by_category = {}
        total_escalated = 0

        for category, amount in cost_breakdown.items():
            esc_type = EscalationType.LABOR if category == 'labor' else \
                       EscalationType.MATERIAL if category == 'material' else \
                       EscalationType.EQUIPMENT if category == 'equipment' else \
                       EscalationType.GENERAL

            factor = self.calculate_compound_factor(base_date, target_date, esc_type)
            escalated = amount * factor
            escalation = escalated - amount

            by_category[category] = {
                'base': round(amount, 2),
                'factor': round(factor, 4),
                'escalated': round(escalated, 2),
                'escalation': round(escalation, 2)
            }

            total_escalated += escalated

        total_escalation = total_escalated - base_cost
        esc_rate = (total_escalation / base_cost * 100) if base_cost > 0 else 0

        return EscalationResult(
            base_cost=round(base_cost, 2),
            base_date=base_date,
            target_date=target_date,
            months=months,
            escalation_rate=round(esc_rate, 2),
            escalation_amount=round(total_escalation, 2),
            escalated_cost=round(total_escalated, 2),
            by_category=by_category
        )

    def escalate_estimate(self,
                           items: List[Dict[str, Any]],
                           base_date: date,
                           target_date: date) -> Dict[str, Any]:
        """Escalate entire estimate."""

        escalated_items = []
        total_base = 0
        total_escalated = 0

        for item in items:
            code = item.get('work_item_code', item.get('code'))
            qty = item.get('quantity', 0)

            # Get costs from CWICR
            labor = 0
            material = 0
            equipment = 0

            if self._code_index is not None and code in self._code_index.index:
                wi = self._code_index.loc[code]
                labor = float(wi.get('labor_cost', 0) or 0) * qty
                material = float(wi.get('material_cost', 0) or 0) * qty
                equipment = float(wi.get('equipment_cost', 0) or 0) * qty

            base = labor + material + equipment
            breakdown = {'labor': labor, 'material': material, 'equipment': equipment}

            result = self.escalate_cost(base, base_date, target_date, breakdown)

            escalated_items.append({
                'code': code,
                'base_cost': result.base_cost,
                'escalated_cost': result.escalated_cost,
                'escalation': result.escalation_amount
            })

            total_base += base
            total_escalated += result.escalated_cost

        return {
            'items': escalated_items,
            'total_base': round(total_base, 2),
            'total_escalated': round(total_escalated, 2),
            'total_escalation': round(total_escalated - total_base, 2),
            'escalation_rate': round((total_escalated - total_base) / total_base * 100, 2) if total_base > 0 else 0,
            'base_date': base_date,
            'target_date': target_date
        }

    def project_future_costs(self,
                              base_cost: float,
                              base_date: date,
                              years_forward: int = 5,
                              annual_rate: float = None) -> pd.DataFrame:
        """Project costs for multiple future years."""

        projections = []
        current = base_cost

        for i in range(years_forward + 1):
            target = base_date + relativedelta(years=i)
            year = target.year

            if annual_rate is None:
                rate = self.get_rate(year)
            else:
                rate = annual_rate

            if i > 0:
                current = current * (1 + rate / 100)

            projections.append({
                'Year': year,
                'Date': target,
                'Annual Rate': f"{rate}%",
                'Projected Cost': round(current, 2),
                'Cumulative Escalation': round(current - base_cost, 2),
                'Cumulative %': round((current - base_cost) / base_cost * 100, 1)
            })

        return pd.DataFrame(projections)

    def de_escalate_cost(self,
                          current_cost: float,
                          current_date: date,
                          base_date: date,
                          category: EscalationType = EscalationType.GENERAL) -> Dict[str, Any]:
        """De-escalate current cost back to base date."""

        factor = self.calculate_compound_factor(base_date, current_date, category)
        base_cost = current_cost / factor

        return {
            'current_cost': round(current_cost, 2),
            'current_date': current_date,
            'base_date': base_date,
            'de_escalation_factor': round(1 / factor, 4),
            'base_cost': round(base_cost, 2),
            'category': category.value
        }

    def export_escalation(self,
                          result: EscalationResult,
                          output_path: str) -> str:
        """Export escalation to Excel."""

        with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
            # Summary
            summary_df = pd.DataFrame([{
                'Base Cost': result.base_cost,
                'Base Date': result.base_date,
                'Target Date': result.target_date,
                'Months': result.months,
                'Escalation Rate': f"{result.escalation_rate}%",
                'Escalation Amount': result.escalation_amount,
                'Escalated Cost': result.escalated_cost
            }])
            summary_df.to_excel(writer, sheet_name='Summary', index=False)

            # By Category
            cat_df = pd.DataFrame([
                {
                    'Category': cat,
                    'Base': data['base'],
                    'Factor': data['factor'],
                    'Escalated': data['escalated'],
                    'Escalation': data['escalation']
                }
                for cat, data in result.by_category.items()
            ])
            cat_df.to_excel(writer, sheet_name='By Category', index=False)

        return output_path

Quick Start

from datetime import date

# Initialize escalation calculator
esc = CWICREscalation()

# Escalate single cost
result = esc.escalate_cost(