cwicr-escalation

CWICR 升级计算器

商业案例

问题陈述

建筑成本会随时间变化：

• 通货膨胀会影响所有成本
• 材料价格会波动
• 劳动力费率逐年上涨
• 长期项目需要进行升级

解决方案

使用历史指数、预测费率和特定类别的升级系数，按时间进行成本升级。

业务价值

• 未来定价 - 估算施工时的成本
• 预算规划 - 将通货膨胀纳入考虑
• 合同定价 - 设置升级条款
• 历史分析 - 将过去的成本调整为当前水平

技术实现

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

快速开始

from datetime import date

# Initialize escalation calculator
esc = CWICREscalation()

# Escalate single cost
result = esc.escalate_cost(
    base_cost=1000000,
    base_date=date(2024, 1, 1),
    target_date=date(2026, 6, 1)
)

print(f"Base Cost: ${result.base_cost:,.2f}")
print(f"Escalated: ${result.escalated_cost:,.2f}")
print(f"Escalation: {result.escalation_rate}%")

常见用例

1. 预测未来成本

projections = esc.project_future_costs(
    base_cost=5000000,
    base_date=date.today(),
    years_forward=5
)
print(projections)

2. 升级估算

cwicr = pd.read_parquet("TR_workitems_costs_resources_DDC_CWICR.parquet")
esc = CWICREscalation(cwicr)

items = [
    {'work_item_code': 'CONC-001', 'quantity': 150},
    {'work_item_code': 'STRL-002', 'quantity': 25}
]

escalated = esc.escalate_estimate(
    items,
    base_date=date(2024, 1, 1),
    target_date=date(2025, 12, 1)
)

3. 下调历史成本

base_cost = esc.de_escalate_cost(
    current_cost=1200000,
    current_date=date(2024, 6, 1),
    base_date=date(2020, 1, 1)
)
print(f"2020 equivalent: ${base_cost['base_cost']:,.2f}")

资源

• GitHub：OpenConstructionEstimate-DDC-CWICR
• DDC Book：第 3.1 章 - 成本升级方法