unit-price-database-manager

name	unit-price-database-manager
description	Manage construction unit price databases: update prices, track vendors, apply location factors, maintain historical records. Essential for accurate estimating.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🗃️","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

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
 datetime  datetime, date
 decimal  Decimal
 pandas  pd
 json


 :
    code: 
    description: 
    unit: 
    base_price: Decimal
    labor_cost: Decimal
    material_cost: Decimal
    equipment_cost: Decimal
    effective_date: date
    expiration_date: [date] = 
    source:  = 
    vendor:  = 
    location:  = 
    notes:  = 
    tags: [] = field(default_factory=)


 :
    code: 
    old_price: Decimal
    new_price: Decimal
    change_pct: 
    updated_at: datetime
    updated_by: 
    reason: 


 :
    vendor_name: 
    item_code: 
    quoted_price: Decimal
    quote_date: date
    valid_until: date
    quantity_break: [] = 
    notes:  = 

 :
    

    
    LOCATION_FACTORS = {
        : , : , : ,
        : , : , : ,
        : , : ,
        : , : , : ,
        : , : 
    }

     ():
        .prices: [, UnitPrice] = {}
        .price_history: [, [UnitPrice]] = {}
        .vendor_quotes: [, [VendorQuote]] = {}
        .updates: [PriceUpdate] = []
        .db_path = db_path

     () -> :
        
        code = price.code

        
         code  .prices:
             code   .price_history:
                .price_history[code] = []
            .price_history[code].append(.prices[code])

            
            old_price = .prices[code].base_price
             old_price != price.base_price:
                change_pct = ((price.base_price - old_price) / old_price * )
                .updates.append(PriceUpdate(
                    code=code,
                    old_price=old_price,
                    new_price=price.base_price,
                    change_pct=change_pct,
                    updated_at=datetime.now(),
                    updated_by=,
                    reason=
                ))

        .prices[code] = price
         code

     () -> [UnitPrice]:
        
         code   .prices:
             

        price = .prices[code]

        
         as_of_date:
             price.effective_date > as_of_date:
                
                 code  .price_history:
                     hist_price  (.price_history[code]):
                         hist_price.effective_date <= as_of_date:
                             hist_price.expiration_date    hist_price.expiration_date >= as_of_date:
                                price = hist_price
                                

             price.expiration_date  price.expiration_date < as_of_date:
                 

        
         location  location != price.location:
            adjusted = UnitPrice(
                code=price.code,
                description=price.description,
                unit=price.unit,
                base_price=._apply_location_factor(price.base_price, price.location, location),
                labor_cost=._apply_location_factor(price.labor_cost, price.location, location),
                material_cost=price.material_cost,  
                equipment_cost=._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=,
                tags=price.tags
            )
             adjusted

         price

     () -> Decimal:
        
        from_factor = .LOCATION_FACTORS.get(from_loc, )
        to_factor = .LOCATION_FACTORS.get(to_loc, )
         Decimal(((amount) * to_factor / from_factor))

     () -> :
        
         effective_date  :
            effective_date = date.today()

        count = 
        factor = Decimal(( + percentage / ))

         code, price  .prices.items():
             categories   (tag  price.tags  tag  categories):
                

            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=,
                vendor=price.vendor,
                location=price.location,
                tags=price.tags
            )

            .add_price(new_price)
            count += 

         count

     ():
        
        code = quote.item_code
         code   .vendor_quotes:
            .vendor_quotes[code] = []
        .vendor_quotes[code].append(quote)

     () -> []:
        
         code   .vendor_quotes:
             

        valid_quotes = []
        today = date.today()

         quote  .vendor_quotes[code]:
             quote.valid_until >= today:
                 quote.quantity_break    quantity >= quote.quantity_break:
                    valid_quotes.append(quote)

          valid_quotes:
             

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

         {
            : best.vendor_name,
            : best.quoted_price,
            : best.valid_until,
            : [
                {: q.vendor_name, : q.quoted_price}
                 q  (valid_quotes, key= x: x.quoted_price)
            ]
        }

     () -> [UnitPrice]:
        
        results = []

         code, price  .prices.items():
            
             query:
                query_lower = query.lower()
                 (query_lower   code.lower() 
                    query_lower   price.description.lower()):
                    

            
             category  category   price.tags:
                

            
             min_price  (price.base_price) < min_price:
                
             max_price  (price.base_price) > max_price:
                

            results.append(price)

         results

     () -> []:
        
        history = []

         code  .price_history:
             price  .price_history[code]:
                history.append({
                    : price.effective_date,
                    : (price.base_price),
                    : price.source
                })

         code  .prices:
            history.append({
                : .prices[code].effective_date,
                : (.prices[code].base_price),
                : .prices[code].source
            })

         (history, key= x: x[])

     () -> :
        
        history = .get_price_history(code)

         (history) < :
             {: }

        prices = [h[]  h  history]
        dates = [h[]  h  history]

        
        first_price = prices[]
        last_price = prices[-]
        total_change = (last_price - first_price) / first_price * 

        
        days = (dates[-] - dates[]).days
        years = days / 
         years > :
            annual_rate = ((last_price / first_price) ** ( / years) - ) * 
        :
            annual_rate = 

         {
            : code,
            : first_price,
            : last_price,
            : total_change,
            : annual_rate,
            : (history),
            : years,
            :   total_change >     total_change < -  
        }

     () -> :
        
        df = pd.read_csv(file_path)
        count = 

         _, row  df.iterrows():
            price = UnitPrice(
                code=row[],
                description=row[],
                unit=row[],
                base_price=Decimal((row[])),
                labor_cost=Decimal((row.get(, ))),
                material_cost=Decimal((row.get(, ))),
                equipment_cost=Decimal((row.get(, ))),
                effective_date=date.today()     row  pd.to_datetime(row[]).date(),
                source=row.get(, ),
                tags=row.get(, ).split()    row  []
            )
            .add_price(price)
            count += 

         count

     () -> :
        
        data = []

         code, price  .prices.items():
             location:
                price = .get_price(code, location)

            data.append({
                : price.code,
                : price.description,
                : price.unit,
                : (price.base_price),
                : (price.labor_cost),
                : (price.material_cost),
                : (price.equipment_cost),
                : price.location,
                : price.effective_date.isoformat(),
                : price.source,
                : .join(price.tags)
            })

        df = pd.DataFrame(data)
        df.to_csv(file_path, index=)
         (data)

     () -> []:
        
        issues = []

         code, price  .prices.items():
            
             price.expiration_date  price.expiration_date < date.today():
                issues.append({
                    : code,
                    : ,
                    : 
                })

            
            age_days = (date.today() - price.effective_date).days
             age_days > :
                issues.append({
                    : code,
                    : ,
                    : 
                })

            
             price.base_price <= :
                issues.append({
                    : code,
                    : ,
                    : 
                })

            
            total_components = price.labor_cost + price.material_cost + price.equipment_cost
             total_components >   ((price.base_price - total_components)) > :
                issues.append({
                    : code,
                    : ,
                    : 
                })

         issues

     () -> :
        
        lines = [, ]
        lines.append()
        lines.append()
        lines.append()

        
        categories = {}
         price  .prices.values():
             tag  price.tags:
                categories[tag] = categories.get(tag, ) + 

         categories:
            lines.append()
             cat, count  (categories.items(), key= x: -x[]):
                lines.append()
            lines.append()

        
        recent_updates = (.updates, key= x: x.updated_at, reverse=)[:]
         recent_updates:
            lines.append()
             update  recent_updates:
                lines.append()
            lines.append()

        
        issues = .validate_prices()
         issues:
            lines.append()
            lines.append()
             issue  issues[:]:
                lines.append()

         .join(lines)