Payment Application Processor
Overview
Process construction payment applications from creation through approval. Validate against schedule of values, calculate retainage, generate AIA G702/G703 forms, and track payment status.
Payment Application Flow
┌─────────────────────────────────────────────────────────────────┐
│ PAYMENT APPLICATION FLOW │
├─────────────────────────────────────────────────────────────────┤
│ │
│ Schedule of → Draft → Review → Approve │
│ Values Pay App & Verify & Pay │
│ ────────── ─────── ──────── ──────── │
│ 📋 SOV items 📝 Enter % ✓ Verify ✅ Approve │
│ 💰 Line values 📊 Calculate 📸 Field 💵 Release │
│ 📅 Billing plan 📄 G702/703 📧 Submit 📄 Record │
│ │
└─────────────────────────────────────────────────────────────────┘
Technical Implementation
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime, timedelta
from enum import Enum
class PayAppStatus(Enum):
DRAFT = "draft"
SUBMITTED = "submitted"
UNDER_REVIEW = "under_review"
APPROVED = "approved"
REJECTED = "rejected"
PAID = "paid"
PARTIAL_PAID = "partial_paid"
class LineItemStatus(Enum):
NOT_STARTED = "not_started"
IN_PROGRESS = "in_progress"
COMPLETE = "complete"
@dataclass
class SOVLineItem:
number: str
description: str
scheduled_value: float
previous_completed: float = 0.0
current_completed: float = 0.0
materials_stored: float = 0.0
percent_complete: float = 0.0
balance_to_finish: float = 0.0
retainage: float = 0.0
def calculate(self, retainage_rate: float = 0.10):
"""Calculate line item values."""
self.percent_complete = (
(self.previous_completed + self.current_completed + self.materials_stored)
/ self.scheduled_value * 100
if self.scheduled_value > 0 else 0
)
self.balance_to_finish = (
self.scheduled_value - self.previous_completed -
self.current_completed - self.materials_stored
)
total_completed = self.previous_completed + self.current_completed + self.materials_stored
self.retainage = total_completed * retainage_rate
@dataclass
class PaymentApplication:
application_number: int
project_name: str
contractor: str
period_from: datetime
period_to: datetime
# Contract info
original_contract: float
change_orders: float = 0.0
contract_sum: float = 0.0
# Line items
line_items: List[SOVLineItem] = field(default_factory=list)
# Totals
total_scheduled: float = 0.0
previous_completed: float = 0.0
current_completed: float = 0.0
materials_stored: float = 0.0
total_completed: float = 0.0
percent_complete: float = 0.0
balance_to_finish: float = 0.0
retainage: float = 0.0
net_amount_due: float = 0.0
# Status
status: PayAppStatus = PayAppStatus.DRAFT
submitted_date: Optional[datetime] = None
approved_date: Optional[datetime] = None
approved_amount: float = 0.0
paid_date: Optional[datetime] = None
paid_amount: float = 0.0
# Retainage rates
work_retainage_rate: float = 0.10
stored_retainage_rate: float = 0.10
@dataclass
class G702Data:
"""AIA G702 Application and Certificate for Payment data."""
application_number: int
period_to: datetime
project_name: str
owner: str
contractor: str
architect: str
original_contract_sum: float
net_change_orders: float
contract_sum_to_date: float
total_completed_stored: float
retainage: float
total_earned_less_retainage: float
less_previous_certificates: float
current_payment_due: float
balance_to_finish: float
class PaymentApplicationProcessor:
"""Process construction payment applications."""
def __init__(self, project_name: str, contractor: str,
original_contract: float):
self.project_name = project_name
self.contractor = contractor
self.original_contract = original_contract
self.change_orders_total = 0.0
self.schedule_of_values: List[SOVLineItem] = []
self.pay_apps: Dict[int, PaymentApplication] = {}
self.next_app_number = 1
self.work_retainage_rate = 0.10
self.stored_retainage_rate = 0.10
def set_retainage_rates(self, work_rate: float, stored_rate: float = None):
"""Set retainage rates."""
self.work_retainage_rate = work_rate
self.stored_retainage_rate = stored_rate if stored_rate else work_rate
def add_sov_item(self, number: str, description: str,
scheduled_value: float) -> SOVLineItem:
"""Add line item to schedule of values."""
item = SOVLineItem(
number=number,
description=description,
scheduled_value=scheduled_value
)
self.schedule_of_values.append(item)
return item
def import_sov(self, items: List[Dict]) -> int:
"""Import schedule of values from list."""
count = 0
for item in items:
self.add_sov_item(
item['number'],
item['description'],
item['value']
)
count += 1
return count
def add_change_order(self, amount: float, description: str = ""):
"""Add approved change order to contract."""
self.change_orders_total += amount
# Add as new SOV item
co_number = f"CO-{len([i for i in self.schedule_of_values if 'CO-' in i.number])+1:03d}"
self.add_sov_item(co_number, description or f"Change Order", amount)
def create_pay_app(self, period_from: datetime,
period_to: datetime) -> PaymentApplication:
"""Create new payment application."""
app_number = self.next_app_number
# Copy SOV with previous values
line_items = []
for sov_item in self.schedule_of_values:
# Get previous completed from last pay app
prev_completed = 0.0
if app_number > 1:
prev_app = self.pay_apps.get(app_number - 1)
if prev_app:
prev_item = next(
(i for i in prev_app.line_items if i.number == sov_item.number),
None
)
if prev_item:
prev_completed = (prev_item.previous_completed +
prev_item.current_completed +
prev_item.materials_stored)
line_items.append(SOVLineItem(
number=sov_item.number,
description=sov_item.description,
scheduled_value=sov_item.scheduled_value,
previous_completed=prev_completed
))
pay_app = PaymentApplication(
application_number=app_number,
project_name=self.project_name,
contractor=self.contractor,
period_from=period_from,
period_to=period_to,
original_contract=self.original_contract,
change_orders=self.change_orders_total,
contract_sum=self.original_contract + self.change_orders_total,
line_items=line_items,
work_retainage_rate=self.work_retainage_rate,
stored_retainage_rate=self.stored_retainage_rate
)
self.pay_apps[app_number] = pay_app
self.next_app_number += 1
return pay_app
def update_line_item(self, app_number: int, line_number: str,
current_completed: float