unit-price-database-manager

What this skill does

# Unit Price Database Manager for Construction

## Overview

Manage and maintain construction unit price databases. Update prices from vendors, apply location and time adjustments, track price history, and ensure estimating accuracy.

## Business Case

Accurate unit prices are critical for:
- **Competitive Bids**: Win work with accurate pricing
- **Cost Control**: Avoid budget surprises
- **Vendor Management**: Track supplier pricing
- **Historical Analysis**: Understand price trends

## Technical Implementation

```python
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
from datetime import datetime, date
from decimal import Decimal
import pandas as pd
import json

@dataclass
class UnitPrice:
    code: str
    description: str
    unit: str
    base_price: Decimal
    labor_cost: Decimal
    material_cost: Decimal
    equipment_cost: Decimal
    effective_date: date
    expiration_date: Optional[date] = None
    source: str = ""
    vendor: str = ""
    location: str = "National Average"
    notes: str = ""
    tags: List[str] = field(default_factory=list)

@dataclass
class PriceUpdate:
    code: str
    old_price: Decimal
    new_price: Decimal
    change_pct: float
    updated_at: datetime
    updated_by: str
    reason: str

@dataclass
class VendorQuote:
    vendor_name: str
    item_code: str
    quoted_price: Decimal
    quote_date: date
    valid_until: date
    quantity_break: Optional[int] = None
    notes: str = ""

class UnitPriceDatabaseManager:
    """Manage construction unit price databases."""

    # Location adjustment factors
    LOCATION_FACTORS = {
        'New York': 1.32, 'San Francisco': 1.28, 'Los Angeles': 1.15,
        'Chicago': 1.12, 'Boston': 1.18, 'Seattle': 1.08,
        'Denver': 1.02, 'National Average': 1.00,
        'Houston': 0.92, 'Dallas': 0.89, 'Phoenix': 0.93,
        'Atlanta': 0.91, 'Miami': 0.95
    }

    def __init__(self, db_path: str = None):
        self.prices: Dict[str, UnitPrice] = {}
        self.price_history: Dict[str, List[UnitPrice]] = {}
        self.vendor_quotes: Dict[str, List[VendorQuote]] = {}
        self.updates: List[PriceUpdate] = []
        self.db_path = db_path

    def add_price(self, price: UnitPrice) -> str:
        """Add or update a unit price."""
        code = price.code

        # Track history
        if code in self.prices:
            if code not in self.price_history:
                self.price_history[code] = []
            self.price_history[code].append(self.prices[code])

            # Record update
            old_price = self.prices[code].base_price
            if old_price != price.base_price:
                change_pct = float((price.base_price - old_price) / old_price * 100)
                self.updates.append(PriceUpdate(
                    code=code,
                    old_price=old_price,
                    new_price=price.base_price,
                    change_pct=change_pct,
                    updated_at=datetime.now(),
                    updated_by="system",
                    reason="Price update"
                ))

        self.prices[code] = price
        return code

    def get_price(self, code: str, location: str = None,
                  as_of_date: date = None) -> Optional[UnitPrice]:
        """Get unit price with optional location adjustment."""
        if code not in self.prices:
            return None

        price = self.prices[code]

        # Check date validity
        if as_of_date:
            if price.effective_date > as_of_date:
                # Look in history
                if code in self.price_history:
                    for hist_price in reversed(self.price_history[code]):
                        if hist_price.effective_date <= as_of_date:
                            if hist_price.expiration_date is None or hist_price.expiration_date >= as_of_date:
                                price = hist_price
                                break

            if price.expiration_date and price.expiration_date < as_of_date:
                return None

        # Apply location factor
        if location and location != price.location:
            adjusted = UnitPrice(
                code=price.code,
                description=price.description,
                unit=price.unit,
                base_price=self._apply_location_factor(price.base_price, price.location, location),
                labor_cost=self._apply_location_factor(price.labor_cost, price.location, location),
                material_cost=price.material_cost,  # Materials less location-sensitive
                equipment_cost=self._apply_location_factor(price.equipment_cost, price.location, location),
                effective_date=price.effective_date,
                expiration_date=price.expiration_date,
                source=price.source,
                vendor=price.vendor,
                location=location,
                notes=f"Adjusted from {price.location}",
                tags=price.tags
            )
            return adjusted

        return price

    def _apply_location_factor(self, amount: Decimal, from_loc: str, to_loc: str) -> Decimal:
        """Apply location adjustment factor."""
        from_factor = self.LOCATION_FACTORS.get(from_loc, 1.0)
        to_factor = self.LOCATION_FACTORS.get(to_loc, 1.0)
        return Decimal(str(float(amount) * to_factor / from_factor))

    def apply_escalation(self, percentage: float, categories: List[str] = None,
                         effective_date: date = None) -> int:
        """Apply escalation to prices."""
        if effective_date is None:
            effective_date = date.today()

        count = 0
        factor = Decimal(str(1 + percentage / 100))

        for code, price in self.prices.items():
            if categories and not any(tag in price.tags for tag in categories):
                continue

            old_price = price.base_price
            new_price = UnitPrice(
                code=price.code,
                description=price.description,
                unit=price.unit,
                base_price=price.base_price * factor,
                labor_cost=price.labor_cost * factor,
                material_cost=price.material_cost * factor,
                equipment_cost=price.equipment_cost * factor,
                effective_date=effective_date,
                source=f"Escalated {percentage}% from {price.source}",
                vendor=price.vendor,
                location=price.location,
                tags=price.tags
            )

            self.add_price(new_price)
            count += 1

        return count

    def add_vendor_quote(self, quote: VendorQuote):
        """Add a vendor quote."""
        code = quote.item_code
        if code not in self.vendor_quotes:
            self.vendor_quotes[code] = []
        self.vendor_quotes[code].append(quote)

    def get_best_price(self, code: str, quantity: int = 1) -> Optional[Dict]:
        """Get best available price from vendors."""
        if code not in self.vendor_quotes:
            return None

        valid_quotes = []
        today = date.today()

        for quote in self.vendor_quotes[code]:
            if quote.valid_until >= today:
                if quote.quantity_break is None or quantity >= quote.quantity_break:
                    valid_quotes.append(quote)

        if not valid_quotes:
            return None

        best = min(valid_quotes, key=lambda q: q.quoted_price)

        return {
            'vendor': best.vendor_name,
            'price': best.quoted_price,
            'valid_until': best.valid_until,
            'all_quotes': [
                {'vendor': q.vendor_name, 'price': q.quoted_price}
                for q in sorted(valid_quotes, key=lambda x: x.quoted_price)
            ]
        }

    def search_prices(self, query: str = None, category: str = None,
                       min_price: float = None, max_price: float = None) -> List[UnitPrice]: