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_value < 0.05:
impact = corr * self.data[col].std() / target.std() * 100
drivers.append(CostDriver(
factor=col,
impact_percentage=abs(impact),
correlation=corr,
description=f"{'Positive' if corr > 0 else 'Negative'} correlation with {target_col}"
))
# Analyze categorical columns
categorical_cols = self.data.select_dtypes(include=['object', 'category']).columns
for col in categorical_cols:
if col not in ['project_id', 'project_name']:
groups = self.data.groupby(col)[target_col].mean()
if len(groups) > 1:
variance = groups.var()
overall_var = target.var()
if variance / overall_var > 0.1:
drivers.append(CostDriver(
factor=col,
impact_percentage=variance / overall_var * 100,
correlation=0,
description=f"Categorical factor with significant cost variation"
))
return sorted(drivers, key=lambda x: -x.impact_percentage)
def compare_to_benchmark(self, estimate: Dict, project_type: str = None) -> Dict:
"""Compare an estimate to historical benchmarks."""
if project_type:
self.calculate_benchmarks(project_type)
comparison = {}
# Cost per SF comparison
if 'cost_per_sf' in estimate and 'cost_per_sf' in self.benchmarks:
benchmark = self.benchmarks['cost_per_sf']
value = estimate['cost_per_sf']
percentile = stats.percentileofscore(
self.data['cost_per_sf'].dropna(), value
)
comparison['cost_per_sf'] = {
'estimate': value,
'benchmark_median': benchmark.value,
'benchmark_range': (benchmark.percentile_25, benchmark.percentile_75),
'percentile': percentile,
'status': 'within_range' if benchmark.percentile_25 <= value <= benchmark.percentile_75 else 'outside_range'
}
return comparison
def find_similar_projects(self, criteria: Dict, n: int = 10) -> pd.DataFrame:
"""Find similar historical projects."""
df = self.data.copy()
# Filter by criteria
if 'project_type' in criteria:
df = df[df['project_type'] == criteria['project_type']]
if 'gross_area' in criteria:
target = criteria['gross_area']
tolerance = criteria.get('area_tolerance', 0.3)
df = df[(df['gross_area'] >= target * (1 - tolerance)) &
(df['gross_area'] <= target * (1 + tolerance))]
if 'location' in criteria and 'location' in df.columns:
df = df[df['location'] == criteria['location']]
if 'year_range' in criteria:
df = df[(df['completion_year'] >= criteria['year_range'][0]) &
(df['completion_year'] <= criteria['year_range'][1])]
# Sort by similarity (simple: by area difference)
if 'gross_area' in criteria and 'gross_area' in df.columns:
df['similarity'] = 1 - abs(df['gross_area'] - criteria['gross_area']) / criteria['gross_area']
df = df.sort_values('similarity', ascending=False)
return df.head(n)
def analyze_overrun_patterns(self) -> Dict:
"""Analyze patterns in cost overruns."""
if 'overrun_pct' not in self.data.columns:
return {}
analysis = {}
# Overall statistics
overruns = self.data['overrun_pct'].dropna()
analysis['overall'] = {
'mean': overruns.mean(),
'median': overruns.median(),
'std': overruns.std(),
'projects_over_budget': (overruns > 0).sum(),
'projects_under_budget': (overruns < 0).sum(),
'pct_over_budget': (overruns > 0).mean() * 100
}
# By project type
if 'project_type' in self.data.columns:
by_type = self.data.groupby('project_type')['overrun_pct'].agg(['mean', 'std', 'count'])
analysis['by_type'] = by_type.to_dict('index')
# By size category
if 'gross_area' in self.data.columns:
self.data['size_category'] = pd.cut(
self.data['gross_area'],
bins=[0, 10000, 50000, 100000, np.inf],
labels=['Small (<10k SF)', 'Medium (10-50k SF)', 'Large (50-100k SF)', 'Very Large (>100k SF)']
)
by_size = self.data.groupby('size_category')['overrun_pct'].agg(['mean', 'std', 'count'])
analysis['by_size'] = by_size.to_dict('index')
return analysis
def generate_report(self, project_type: str = None) -> str:
"""Generate comprehensive cost analysis report."""
lines = ["# Historical Cost Analysis Report", ""]
lines.append(f"**Generated:** {datetime.now().strftime('%Y-%m-%d')}")
lines.append(f"**Projects Analyzed:** {len(self.data):,}")
if project_type:
lines.append(f"**Project Type:** {project_type}")
lines.append("")
# Benchmarks
benchmarks = self.calculate_benchmarks(project_type)
if benchmarks:
lines.append("## Cost Benchmarks")
for name, bm in benchmarks.items():
lines.append(f"\n### {bm.metric_name}")
lines.append(f"- **Median:** {bm.value:.2f} {bm.unit}")
lines.append(f"- **25th Percentile:** {bm.percentile_25:.2f} {bm.unit}")
lines.append(f"- **75th Percentile:** {bm.percentile_75:.2f} {bm.unit}")
lines.append(f"- **Sample Size:** {bm.sample_size}")
# Escalation
lines.append("\n## Cost Escalation")
esc = self.calculate_escalation(from_year=2020, to_year=2026)
if esc:
lines.append(f"- **Period:** {esc.from_year} to {esc.to_year}")
lines.append(f"- **Annual Rate:** {esc.annual_rate:.1%}")
lines.append(f"- **Total Change:** {esc.total_change:.1%}")
# Cost Drivers
drivers = self.identify_cost_drivers()
if drivers:
lines.append("\n## Key Cost Drivers")
for driver in drivers[:5]:
lines.append(f"- **{driver.factor}:** {driver.impact_percentage:.1f}% impact (r={driver.correlation:.2f})")
# Overrun Analysis
overrun_analysis = self.analyze_overrun_patterns()
if 'overall' in overrun_analysis:
lines.append("\n## Overrun Analysis")
overall = overrun_analysis['overall']
lines.append(f"- **Average Overrun:** {overall['mean']:.1f}%")
lines.append(f"- **Projects Over Budget:** {overall['pct_over_budget']:.1f}%")
return "\n".join(lines)Quick Start
import pandas as pd
# Load historical data
historical = pd.read_excel("historical_projects.xlsx")
# Initialize analyzer
analyzer = HistoricalCostAnalyzer()
analyzer.load_data(historical)
# Calculate benchmarks for office buildings
benchmarks = analyzer.calculate_benchmarks(project_type='Office')
print(f"Office median cost: ${benchmarks['cost_per_sf'].value:.2f}/SF")
# Calculate escalation
escalation = analyzer.calculate_escalation(from_year=2020, to_year=2026)
print(f"Annual escalation: {escalation.annual_rate:.1%}")
# Find similar projects
similar = analyzer.find_similar_projects({
'project_type': 'Office',
'gross_area': 50000,
'year_range': (2020, 2025)
})
print(f"Found {len(similar)} similar projects")
# Compare estimate to benchmark
comparison = analyzer.compare_to_benchmark({'cost_per_sf': 250}, 'Office')
print(f"Estimate percentile: {comparison['cost_per_sf']['percentile']:.0f}th")
# Generate report
report = analyzer.generate_report('Office')
print(report)Dependencies
pip install pandas numpy scipy