Historical Cost Analyzer for Construction
Overview
Analyze historical construction cost data for benchmarking, escalation tracking, and estimating calibration. Compare similar projects, identify cost drivers, and improve future estimates.
Business Case
Historical cost analysis enables:
- Benchmarking: Compare current estimates to past projects
- Calibration: Improve estimating accuracy using actual data
- Trends: Track cost escalation and market changes
- Risk Assessment: Identify cost drivers and overrun patterns
Technical Implementation
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Tuple
import pandas as pd
import numpy as np
from datetime import datetime
from scipy import stats
@dataclass
class CostBenchmark:
metric_name: str
value: float
unit: str
percentile_25: float
percentile_50: float
percentile_75: float
sample_size: int
project_types: List[str]
@dataclass
class EscalationAnalysis:
from_year: int
to_year: int
annual_rate: float
total_change: float
category: str
confidence: float
@dataclass
class CostDriver:
factor: str
impact_percentage: float
correlation: float
description: str
class HistoricalCostAnalyzer:
"""Analyze historical construction costs."""
# RSMeans City Cost Indexes (sample - would be loaded from database)
LOCATION_FACTORS = {
'New York': 1.32, 'San Francisco': 1.28, 'Los Angeles': 1.15,
'Chicago': 1.12, 'Houston': 0.92, 'Dallas': 0.89,
'Phoenix': 0.93, 'Atlanta': 0.91, 'Denver': 1.02,
'Seattle': 1.08, 'National Average': 1.00
}
# Historical cost indices by year
COST_INDICES = {
2015: 100.0, 2016: 102.1, 2017: 105.3, 2018: 109.2,
2019: 112.5, 2020: 114.8, 2021: 121.4, 2022: 135.6,
2023: 142.3, 2024: 148.7, 2025: 154.2, 2026: 160.0
}
def __init__(self, historical_data: pd.DataFrame = None):
self.data = historical_data
self.benchmarks: Dict[str, CostBenchmark] = {}
def load_data(self, data: pd.DataFrame):
"""Load historical project data."""
self.data = data.copy()
# Normalize data
if 'completion_year' not in self.data.columns and 'completion_date' in self.data.columns:
self.data['completion_year'] = pd.to_datetime(self.data['completion_date']).dt.year
# Calculate key metrics
if 'gross_area' in self.data.columns and 'final_cost' in self.data.columns:
self.data['cost_per_sf'] = self.data['final_cost'] / self.data['gross_area']
if 'original_estimate' in self.data.columns and 'final_cost' in self.data.columns:
self.data['overrun_pct'] = ((self.data['final_cost'] - self.data['original_estimate'])
/ self.data['original_estimate'] * 100)
def normalize_to_year(self, costs: pd.Series, from_years: pd.Series,
to_year: int = 2026) -> pd.Series:
"""Normalize costs to a common year using cost indices."""
normalized = costs.copy()
for i, (cost, year) in enumerate(zip(costs, from_years)):
if pd.notna(cost) and pd.notna(year):
year = int(year)
if year in self.COST_INDICES and to_year in self.COST_INDICES:
factor = self.COST_INDICES[to_year] / self.COST_INDICES[year]
normalized.iloc[i] = cost * factor
return normalized
def normalize_to_location(self, costs: pd.Series, locations: pd.Series,
to_location: str = 'National Average') -> pd.Series:
"""Normalize costs to a common location."""
normalized = costs.copy()
to_factor = self.LOCATION_FACTORS.get(to_location, 1.0)
for i, (cost, loc) in enumerate(zip(costs, locations)):
if pd.notna(cost) and loc in self.LOCATION_FACTORS:
from_factor = self.LOCATION_FACTORS[loc]
normalized.iloc[i] = cost * (to_factor / from_factor)
return normalized
def calculate_benchmarks(self, project_type: str = None,
year_range: Tuple[int, int] = None) -> Dict[str, CostBenchmark]:
"""Calculate cost benchmarks from historical data."""
df = self.data.copy()
# Filter by project type
if project_type and 'project_type' in df.columns:
df = df[df['project_type'] == project_type]
# Filter by year range
if year_range and 'completion_year' in df.columns:
df = df[(df['completion_year'] >= year_range[0]) &
(df['completion_year'] <= year_range[1])]
benchmarks = {}
# Cost per SF
if 'cost_per_sf' in df.columns:
values = df['cost_per_sf'].dropna()
if len(values) > 0:
benchmarks['cost_per_sf'] = CostBenchmark(
metric_name='Cost per SF',
value=values.median(),
unit='$/SF',
percentile_25=values.quantile(0.25),
percentile_50=values.quantile(0.50),
percentile_75=values.quantile(0.75),
sample_size=len(values),
project_types=[project_type] if project_type else df['project_type'].unique().tolist()
)
# Overrun percentage
if 'overrun_pct' in df.columns:
values = df['overrun_pct'].dropna()
if len(values) > 0:
benchmarks['overrun_pct'] = CostBenchmark(
metric_name='Cost Overrun',
value=values.median(),
unit='%',
percentile_25=values.quantile(0.25),
percentile_50=values.quantile(0.50),
percentile_75=values.quantile(0.75),
sample_size=len(values),
project_types=[project_type] if project_type else df['project_type'].unique().tolist()
)
self.benchmarks.update(benchmarks)
return benchmarks
def calculate_escalation(self, category: str = 'overall',
from_year: int = 2020,
to_year: int = 2026) -> EscalationAnalysis:
"""Calculate cost escalation between years."""
if from_year in self.COST_INDICES and to_year in self.COST_INDICES:
from_index = self.COST_INDICES[from_year]
to_index = self.COST_INDICES[to_year]
total_change = (to_index - from_index) / from_index
years = to_year - from_year
annual_rate = (to_index / from_index) ** (1 / years) - 1 if years > 0 else 0
return EscalationAnalysis(
from_year=from_year,
to_year=to_year,
annual_rate=annual_rate,
total_change=total_change,
category=category,
confidence=0.95
)
return None
def identify_cost_drivers(self, target_col: str = 'cost_per_sf') -> List[CostDriver]:
"""Identify factors that drive costs."""
if self.data is None or target_col not in self.data.columns:
return []
drivers = []
target = self.data[target_col].dropna()
# Analyze numeric columns
numeric_cols = self.data.select_dtypes(include=[np.number]).columns
exclude = [target_col, 'final_cost', 'original_estimate']
for col in numeric_cols:
if col not in exclude:
valid_mask = self.data[col].notna() & self.data[target_col].notna()
if valid_mask.sum() > 10:
corr, p_value = stats.pearsonr(
self.data.loc[valid_mask, col],
self.data.loc[valid_mask, target_col]
)
if abs(corr) > 0.3 and p_va