技能名称: price-api

详细描述:

建筑材料价格API

概述

Python实现

python
import requests
import pandas as pd
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import json

class MaterialCategory(Enum):
建筑材料类别。
CONCRETE = concrete
STEEL = steel
LUMBER = lumber
COPPER = copper
ALUMINUM = aluminum
CEMENT = cement
AGGREGATES = aggregates
ASPHALT = asphalt

@dataclass
class MaterialPrice:
材料价格点。
material: str
price: float
unit: str
currency: str
source: str
date: datetime
region: str =

@dataclass
class PriceTrend:
价格趋势分析。
material: str
current_price: float
week_change: float
month_change: float
year_change: float
trend_direction: str # up, down, stable

class OpenPriceAPI:
开放材料价格API客户端。

# 商品价格来源
FRED_BASE = https://api.stlouisfed.org/fred/series/observations

# 建筑商品的FRED系列ID
FRED_SERIES = {
steel: WPU101,
lumber: WPS0811,
concrete: WPU133,
copper: PCOPPUSDM,
aluminum: PALUMUSDM
}

def init(self, fredapikey: Optional[str] = None):
self.fredapikey = fredapikey

def getfredprices(self, material: str,
start_date: str = None,
end_date: str = None) -> List[MaterialPrice]:
从FRED API获取价格。

if material.lower() not in self.FRED_SERIES:
return []

seriesid = self.FREDSERIES[material.lower()]

if start_date is None:
start_date = (datetime.now() - timedelta(days=365)).strftime(%Y-%m-%d)
if end_date is None:
end_date = datetime.now().strftime(%Y-%m-%d)

params = {
seriesid: seriesid,
observationstart: startdate,
observationend: enddate,
file_type: json
}

if self.fredapikey:
params[apikey] = self.fredapi_key

try:
response = requests.get(self.FRED_BASE, params=params)
if response.status_code != 200:
return []

data = response.json()
observations = data.get(observations, [])

prices = []
for obs in observations:
try:
price = float(obs[value])
prices.append(MaterialPrice(
material=material,
price=price,
unit=index,
currency=USD,
source=FRED,
date=datetime.strptime(obs[date], %Y-%m-%d),
region=US
))
except (ValueError, KeyError):
continue

return prices

except Exception as e:
print(f获取FRED数据时出错: {e})
return []

def to_dataframe(self, prices: List[MaterialPrice]) -> pd.DataFrame:
将价格转换为DataFrame。
data = [{
material: p.material,
price: p.price,
unit: p.unit,
currency: p.currency,
source: p.source,
date: p.date,
region: p.region
} for p in prices]
return pd.DataFrame(data)

class ConstructionPriceTracker:
追踪和分析建筑材料价格。

# 默认区域因子
REGIONAL_FACTORS = {
US_National: 1.0,
US_Northeast: 1.15,
US_Southeast: 0.95,
US_Midwest: 0.92,
US_West: 1.10,
Germany: 1.25,
UK: 1.20,
France: 1.18
}

def init(self):
self.price_cache: Dict[str, pd.DataFrame] = {}

def calculate_trend(self, prices: pd.DataFrame) -> PriceTrend:
从历史数据计算价格趋势。

if prices.empty or price not in prices.columns:
return None

prices = prices.sort_values(date)
current = prices[price].iloc[-1]

# 计算变化
weekagoidx = len(prices) - 7 if len(prices) >= 7 else 0
monthagoidx = len(prices) - 30 if len(prices) >= 30 else 0
yearagoidx = len(prices) - 365 if len(prices) >= 365 else 0

weekprice = prices[price].iloc[weekago_idx]
monthprice = prices[price].iloc[monthago_idx]
yearprice = prices[price].iloc[yearago_idx]

weekchange = ((current - weekprice) / weekprice * 100) if weekprice else 0
monthchange = ((current - monthprice) / monthprice * 100) if monthprice else 0
yearchange = ((current - yearprice) / yearprice * 100) if yearprice else 0

# 确定趋势
if month_change > 5:
trend = up
elif month_change < -5:
trend = down
else:
trend = stable

return PriceTrend(
material=prices[material].iloc[0],
current_price=current,
weekchange=round(weekchange, 2),
monthchange=round(monthchange, 2),
yearchange=round(yearchange, 2),
trend_direction=trend
)

def applyregionalfactor(self, base_price: float,
region: str) -> float:
应用区域价格因子。
factor = self.REGIONAL_FACTORS.get(region, 1.0)
return base_price * factor

def updatecostdatabase(self, cost_df: pd.DataFrame,
price_updates: Dict[str, float],
datecolumn: str = lastupdated) -> pd.DataFrame:
用新价格更新成本数据库。
updated = cost_df.copy()

for material, price in price_updates.items():
# 查找包含此材料的行
mask = updated[material].str.lower() == material.lower()
if mask.any():
# 计算调整因子
oldprice = updated.loc[mask, unitprice].mean()
factor = price / oldprice if oldprice > 0 else 1

# 更新价格
updated.loc[mask, unit_price] *= factor
updated.loc[mask, date_column] = datetime.now()

return updated

class MaterialPriceEstimator:
当API数据不可用时估算材料价格。

# 参考价格（美元/单位，截至2024年）
REFERENCE_PRICES = {
concrete_m3: 120,
rebar_ton: 800,
structuralsteelton: 1200,
lumber_mbf: 450,
copperwirekg: 12,
brick_1000: 550,
cement_ton: 130,
sand_m3: 35,
gravel_m3: 40,
drywall_m2: 8,
insulation_m2: 25
}

def estimate_price(self, material: str,
region: str = US_National,
inflation_adjustment: float = 0) -> float:
估算材料的当前价格。
baseprice = self.REFERENCEPRICES.get(material, 0)

if base_price == 0:
return 0

# 应用通胀调整
adjusted = baseprice * (1 + inflationadjustment)

# 应用区域因子
tracker = ConstructionPriceTracker()
return tracker.applyregionalfactor(adjusted, region)

def bulk_estimate(self, materials: List[str],
region: str = US_National) -> pd.DataFrame:
估算多种材料的价格。
estimates = []
for material in materials:
price = self.estimate_price(material, region)
estimates.append({
material: material,
estimated_price: price,
region: region,
source: estimate,
date: datetime