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
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]: