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章 - コストエスカレーションの手法