CWICR Rate Updater
Business Case
Problem Statement
Resource rates become outdated:
- Material prices fluctuate with market
- Labor rates change annually
- Equipment costs vary by region
- Historical rates need adjustment
Solution
Systematic rate updates integrating market data, inflation indices, and regional factors while maintaining audit trail.
Business Value
- Accuracy - Current market pricing
- Flexibility - Update specific resources or categories
- Audit trail - Track rate changes over time
- Automation - Integrate with price APIs
Technical Implementation
import pandas as pd
import numpy as np
from typing import Dict, Any, List, Optional, Tuple, Callable
from dataclasses import dataclass, field
from datetime import datetime, date
from enum import Enum
import json
class RateType(Enum):
"""Types of rates."""
LABOR = "labor"
MATERIAL = "material"
EQUIPMENT = "equipment"
SUBCONTRACT = "subcontract"
class AdjustmentMethod(Enum):
"""Methods for rate adjustment."""
FIXED_AMOUNT = "fixed_amount"
PERCENTAGE = "percentage"
MULTIPLIER = "multiplier"
REPLACEMENT = "replacement"
@dataclass
class RateChange:
"""Record of rate change."""
resource_code: str
rate_type: RateType
old_rate: float
new_rate: float
change_percent: float
change_date: datetime
reason: str
source: str
@dataclass
class RateUpdateResult:
"""Result of rate update operation."""
total_items: int
updated: int
unchanged: int
errors: int
changes: List[RateChange]
summary: Dict[str, Any]
class CWICRRateUpdater:
"""Update resource rates in CWICR data."""
def __init__(self, cwicr_data: pd.DataFrame):
self.data = cwicr_data.copy()
self.change_log: List[RateChange] = []
self.original_data = cwicr_data.copy()
def get_current_rates(self,
rate_type: RateType = None,
category: str = None) -> pd.DataFrame:
"""Get current rates, optionally filtered."""
df = self.data.copy()
# Filter by category if specified
if category and 'category' in df.columns:
df = df[df['category'].str.contains(category, case=False, na=False)]
# Select relevant columns based on rate type
rate_columns = {
RateType.LABOR: ['work_item_code', 'description', 'labor_rate', 'labor_cost'],
RateType.MATERIAL: ['work_item_code', 'description', 'material_cost'],
RateType.EQUIPMENT: ['work_item_code', 'description', 'equipment_cost', 'equipment_rate']
}
if rate_type and rate_type in rate_columns:
cols = [c for c in rate_columns[rate_type] if c in df.columns]
return df[cols]
return df
def update_rate(self,
work_item_code: str,
rate_type: RateType,
new_rate: float,
reason: str = "Manual update",
source: str = "User") -> Optional[RateChange]:
"""Update single rate."""
rate_column = self._get_rate_column(rate_type)
if rate_column not in self.data.columns:
return None
mask = self.data['work_item_code'] == work_item_code
if not mask.any():
return None
old_rate = float(self.data.loc[mask, rate_column].iloc[0])
self.data.loc[mask, rate_column] = new_rate
change_percent = ((new_rate - old_rate) / old_rate * 100) if old_rate > 0 else 0
change = RateChange(
resource_code=work_item_code,
rate_type=rate_type,
old_rate=old_rate,
new_rate=new_rate,
change_percent=round(change_percent, 2),
change_date=datetime.now(),
reason=reason,
source=source
)
self.change_log.append(change)
return change
def _get_rate_column(self, rate_type: RateType) -> str:
"""Get column name for rate type."""
mapping = {
RateType.LABOR: 'labor_rate',
RateType.MATERIAL: 'material_cost',
RateType.EQUIPMENT: 'equipment_cost',
RateType.SUBCONTRACT: 'subcontract_cost'
}
return mapping.get(rate_type, 'labor_rate')
def apply_percentage_adjustment(self,
rate_type: RateType,
percentage: float,
category: str = None,
reason: str = "Percentage adjustment") -> RateUpdateResult:
"""Apply percentage adjustment to rates."""
rate_column = self._get_rate_column(rate_type)
if rate_column not in self.data.columns:
return RateUpdateResult(0, 0, 0, 1, [], {})
# Build mask
mask = pd.Series([True] * len(self.data))
if category and 'category' in self.data.columns:
mask = self.data['category'].str.contains(category, case=False, na=False)
# Store old values
old_values = self.data.loc[mask, rate_column].copy()
# Apply adjustment
multiplier = 1 + (percentage / 100)
self.data.loc[mask, rate_column] = old_values * multiplier
# Record changes
changes = []
for idx in self.data[mask].index:
old_rate = float(old_values.loc[idx])
new_rate = float(self.data.loc[idx, rate_column])
if old_rate != new_rate:
change = RateChange(
resource_code=str(self.data.loc[idx, 'work_item_code']),
rate_type=rate_type,
old_rate=old_rate,
new_rate=new_rate,
change_percent=percentage,
change_date=datetime.now(),
reason=reason,
source=f"Bulk {percentage}%"
)
changes.append(change)
self.change_log.append(change)
return RateUpdateResult(
total_items=len(self.data[mask]),
updated=len(changes),
unchanged=len(self.data[mask]) - len(changes),
errors=0,
changes=changes,
summary={
'rate_type': rate_type.value,
'adjustment_percent': percentage,
'category': category,
'average_new_rate': self.data.loc[mask, rate_column].mean()
}
)
def apply_inflation_index(self,
base_year: int,
current_year: int,
inflation_rates: Dict[int, float],
rate_types: List[RateType] = None) -> RateUpdateResult:
"""Apply inflation index from base year to current."""
if rate_types is None:
rate_types = [RateType.LABOR, RateType.MATERIAL, RateType.EQUIPMENT]
# Calculate cumulative multiplier
cumulative_multiplier = 1.0
for year in range(base_year, current_year):
rate = inflation_rates.get(year, 0.02) # Default 2%
cumulative_multiplier *= (1 + rate)
total_changes = []
for rate_type in rate_types:
result = self.apply_percentage_adjustment(
rate_type=rate_type,
percentage=(cumulative_multiplier - 1) * 100,
reason=f"Inflation {base_year}-{current_year}"
)
total_changes.extend(result.changes)
return RateUpdateResult(
total_items=len(self.data),
updated=len(total_changes),
unchanged=len(self.data) - len(total_changes),
errors=0,
changes=total_changes,
summary={
'base_year': base_year,
'current_year': current_year,
'cumulative_multiplier': round(cumu