Price API for Construction Materials

Overview

Material prices fluctuate constantly. This skill fetches prices from open sources, tracks trends, and updates cost databases with current market data.

Python Implementation

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):
    """Construction material categories."""
    CONCRETE = "concrete"
    STEEL = "steel"
    LUMBER = "lumber"
    COPPER = "copper"
    ALUMINUM = "aluminum"
    CEMENT = "cement"
    AGGREGATES = "aggregates"
    ASPHALT = "asphalt"


@dataclass
class MaterialPrice:
    """Material price point."""
    material: str
    price: float
    unit: str
    currency: str
    source: str
    date: datetime
    region: str = ""


@dataclass
class PriceTrend:
    """Price trend analysis."""
    material: str
    current_price: float
    week_change: float
    month_change: float
    year_change: float
    trend_direction: str  # 'up', 'down', 'stable'


class OpenPriceAPI:
    """Client for open material price APIs."""

    # Commodity price sources
    FRED_BASE = "https://api.stlouisfed.org/fred/series/observations"

    # FRED Series IDs for construction commodities
    FRED_SERIES = {
        'steel': 'WPU101',
        'lumber': 'WPS0811',
        'concrete': 'WPU133',
        'copper': 'PCOPPUSDM',
        'aluminum': 'PALUMUSDM'
    }

    def __init__(self, fred_api_key: Optional[str] = None):
        self.fred_api_key = fred_api_key

    def get_fred_prices(self, material: str,
                        start_date: str = None,
                        end_date: str = None) -> List[MaterialPrice]:
        """Get prices from FRED API."""

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

        series_id = self.FRED_SERIES[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 = {
            'series_id': series_id,
            'observation_start': start_date,
            'observation_end': end_date,
            'file_type': 'json'
        }

        if self.fred_api_key:
            params['api_key'] = self.fred_api_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,...