技能名称：historical-cost-analyzer

施工历史成本分析器

概述

分析历史施工成本数据，用于基准测试、成本上涨追踪和估算校准。比较类似项目，识别成本驱动因素，并改进未来估算。

商业案例

历史成本分析能够实现：

• - 基准测试：将当前估算与过往项目进行比较
• 校准：利用实际数据提高估算准确性
• 趋势分析：追踪成本上涨和市场变化
• 风险评估：识别成本驱动因素和超支模式

技术实现

python
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:
分析历史施工成本。

# RSMeans城市成本指数（示例——将从数据库加载）
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
}

# 按年份划分的历史成本指数
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):
加载历史项目数据。
self.data = data.copy()

# 标准化数据
if completionyear not in self.data.columns and completiondate in self.data.columns:
self.data[completionyear] = pd.todatetime(self.data[completion_date]).dt.year

# 计算关键指标
if grossarea in self.data.columns and finalcost in self.data.columns:
self.data[costpersf] = self.data[finalcost] / self.data[grossarea]

if originalestimate in self.data.columns and finalcost in self.data.columns:
self.data[overrunpct] = ((self.data[finalcost] - self.data[original_estimate])
/ self.data[original_estimate] * 100)

def normalizetoyear(self, costs: pd.Series, from_years: pd.Series,
to_year: int = 2026) -> pd.Series:
使用成本指数将成本标准化到同一年份。
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.COSTINDICES and toyear in self.COST_INDICES:
factor = self.COSTINDICES[toyear] / self.COST_INDICES[year]
normalized.iloc[i] = cost * factor

return normalized

def normalizetolocation(self, costs: pd.Series, locations: pd.Series,
to_location: str = National Average) -> pd.Series:
将成本标准化到同一地点。
normalized = costs.copy()
tofactor = self.LOCATIONFACTORS.get(to_location, 1.0)

for i, (cost, loc) in enumerate(zip(costs, locations)):
if pd.notna(cost) and loc in self.LOCATION_FACTORS:
fromfactor = self.LOCATIONFACTORS[loc]
normalized.iloc[i] = cost * (tofactor / fromfactor)

return normalized

def calculatebenchmarks(self, projecttype: str = None,
year_range: Tuple[int, int] = None) -> Dict[str, CostBenchmark]:
根据历史数据计算成本基准。
df = self.data.copy()

# 按项目类型筛选
if projecttype and projecttype in df.columns:
df = df[df[projecttype] == projecttype]

# 按年份范围筛选
if yearrange and completionyear in df.columns:
df = df[(df[completionyear] >= yearrange[0]) &
(df[completionyear] <= yearrange[1])]

benchmarks = {}

# 每平方英尺成本
if costpersf in df.columns:
values = df[costpersf].dropna()
if len(values) > 0:
benchmarks[costpersf] = 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),
projecttypes=[projecttype] if projecttype else df[projecttype].unique().tolist()
)

# 超支百分比
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),
projecttypes=[projecttype] if projecttype else df[projecttype].unique().tolist()
)

self.benchmarks.update(benchmarks)
return benchmarks

def calculate_escalation(self, category: str = overall,
from_year: int = 2020,
to_year: int = 2026) -> EscalationAnalysis:
计算年份之间的成本上涨。
if fromyear in self.COSTINDICES and toyear in self.COSTINDICES:
fromindex = self.COSTINDICES[from_year]
toindex = self.COSTINDICES[to_year]

totalchange = (toindex - fromindex) / fromindex
years = toyear - fromyear
annualrate = (toindex / from_index) (1 / years) - 1 if years > 0 else 0

return EscalationAnalysis(
fromyear=fromyear,
toyear=toyear,
annualrate=annualrate,
totalchange=totalchange,
category=category,
confidence=0.95
)

return None

def identifycostdrivers(self, targetcol: str = costper_sf) -> List[CostDriver]:
识别驱动成本的因素。
if self.data is None or target_col not in self.data.columns:
return []

drivers = []
target = self.data[target_col].dropna()

# 分析数值列
numericcols = self.data.selectdtypes(include=[np.number]).columns
exclude = [targetcol, finalcost, original_estimate]

for col in numeric_cols:
if col not in exclude:
validmask = self.data[col].notna() & self.data[targetcol].notna()
if valid_mask.sum() > 10:
corr, p_value = stats.pearsonr(
self.data.loc[valid_mask, col],
self.data.loc[validmask, targetcol]
)

if abs(corr) > 0.3 and p_value < 0.05:
impact = corr self.data[col].std() / target.std() 100

drivers.append(CostDriver(
factor=