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 = None,
                        materials_stored: float = None) -> SOVLineItem:
        """Update line item progress."""
        if app_number not in self.pay_apps:
            raise ValueError(f"Pay app {app_number} not found")

        pay_app = self.pay_apps[app_number]
        item = next((i for i in pay_app.line_items if i.number == line_number), None)

        if not item:
            raise ValueError(f"Line item {line_number} not found")

        if current_completed is not None:
            item.current_completed = current_completed
        if materials_stored is not None:
            item.materials_stored = materials_stored

        item.calculate(pay_app.work_retainage_rate)

        # Recalculate pay app totals
        self._calculate_totals(pay_app)

        return item

    def update_by_percentage(self, app_number: int, line_number: str,
                            percent_complete: float) -> SOVLineItem:
        """Update line item by percentage complete."""
        if app_number not in self.pay_apps:
            raise ValueError(f"Pay app {app_number} not found")

        pay_app = self.pay_apps[app_number]
        item = next((i for i in pay_app.line_items if i.number == line_number), None)

        if not item:
            raise ValueError(f"Line item {line_number} not found")

        # Calculate current completed from percentage
        total_earned = item.scheduled_value * (percent_complete / 100)
        item.current_completed = total_earned - item.previous_completed - item.materials_stored

        if item.current_completed < 0:
            item.current_completed = 0

        item.calculate(pay_app.work_retainage_rate)
        self._calculate_totals(pay_app)

        return item

    def _calculate_totals(self, pay_app: PaymentApplication):
        """Calculate pay app totals."""
        pay_app.total_scheduled = sum(i.scheduled_value for i in pay_app.line_items)
        pay_app.previous_completed = sum(i.previous_completed for i in pay_app.line_items)
        pay_app.current_completed = sum(i.current_completed for i in pay_app.line_items)
        pay_app.materials_stored = sum(i.materials_stored for i in pay_app.line_items)

        pay_app.total_completed = (
            pay_app.previous_completed +
            pay_app.current_completed +
            pay_app.materials_stored
        )

        pay_app.percent_complete = (
            pay_app.total_completed / pay_app.total_scheduled * 100
            if pay_app.total_scheduled > 0 else 0
        )

        pay_app.balance_to_finish = pay_app.contract_sum - pay_app.total_completed

        # Calculate retainage
        work_retainage = (pay_app.previous_completed + pay_app.current_completed) * pay_app.work_retainage_rate
        stored_retainage = pay_app.materials_stored * pay_app.stored_retainage_rate
        pay_app.retainage = work_retainage + stored_retainage

        # Previous certificates
        previous_certs = 0
        if pay_app.application_number > 1:
            prev_app = self.pay_apps.get(pay_app.application_number - 1)
            if prev_app:
                previous_certs = prev_app.total_completed - prev_app.retainage

        # Net amount due
        pay_app.net_amount_due = (
            pay_app.total_completed - pay_app.retainage - previous_certs
        )

    def submit_pay_app(self, app_number: int) -> PaymentApplication:
        """Submit pay app for review."""
        if app_number not in self.pay_apps:
            raise ValueError(f"Pay app {app_number} not found")

        pay_app = self.pay_apps[app_number]
        pay_app.status = PayAppStatus.SUBMITTED
        pay_app.submitted_date = datetime.now()

        return pay_app

    def approve_pay_app(self, app_number: int,
                       approved_amount: float = None) -> PaymentApplication:
        """Approve pay app."""
        if app_number not in self.pay_apps:
            raise ValueError(f"Pay app {app_number} not found")

        pay_app = self.pay_apps[app_number]
        pay_app.status = PayAppStatus.APPROVED
        pay_app.approved_date = datetime.now()
        pay_app.approved_amount = approved_amount if approved_amount else pay_app.net_amount_due

        return pay_app

    def record_payment(self, app_number: int, paid_amount: float,
                      paid_date: datetime = None) -> PaymentApplication:
        """Record payment received."""
        if app_number not in self.pay_apps:
            raise ValueError(f"Pay app {app_number} not found")

        pay_app = self.pay_apps[app_number]
        pay_app.paid_date = paid_date or datetime.now()
        pay_app.paid_amount = paid_amount

        if paid_amount >= pay_app.approved_amount:
            pay_app.status = PayAppStatus.PAID
        else:
            pay_app.status = PayAppStatus.PARTIAL_PAID

        return pay_app

    def generate_g702(self, app_number: int, owner: str,
                     architect: str) -> G702Data:
        """Generate AIA G702 data."""
        if app_number not in self.pay_apps:
            raise ValueError(f"Pay app {app_number} not found")

        pay_app = self.pay_apps[app_number]

        # Previous certificates
        previous_certs = 0
        if app_number > 1:
            prev_app = self.pay_apps.get(app_number - 1)
            if prev_app:
                previous_certs = prev_app.total_completed - prev_app.retainage

        return G702Data(
            application_number=pay_app.application_number,
            period_to=pay_app.period_to,
            project_name=pay_app.project_name,
            owner=owner,
            contractor=pay_app.contractor,
            architect=architect,
            original_contract_sum=pay_app.original_contract,
            net_change_orders=pay_app.change_orders,
            contract_sum_to_date=pay_app.contract_sum,
            total_completed_stored=pay_app.total_completed,
            retainage=pay_app.retainage,
            total_earned_less_retainage=pay_app.total_completed - pay_app.retainage,
            less_previous_certificates=previous_certs,
            current_payment_due=pay_app.net_amount_due,
            balance_to_finish=pay_app.balance_to_finish
        )

    def generate_g703(self, app_number: int) -> str:
        """Generate AIA G703 continuation sheet."""
        if app_number not in self.pay_apps:
            return "Pay app not found"

        pay_app = self.pay_apps[app_number]

        lines = [
            "# AIA G703 - Continuation Sheet",
            "",
            f"**Application Number:** {pay_app.application_number}",
            f"**Period To:** {pay_app.period_to.strftime('%Y-%m-%d')}",
            "",
            "| Item | Description | Scheduled | Previous | This Period | Stored | Total | % | Balance | Retainage |",
            "|------|-------------|-----------|----------|-------------|--------|-------|---|---------|-----------|"
        ]

        for item in pay_app.line_items:
            lines.append(
                f"| {item.number} | {item.description[:20]} | "
                f"${item.scheduled_value:,.0f} | ${item.previous_completed:,.0f} | "
                f"${item.current_completed:,.0f} | ${item.materials_stored:,.0f} | "
                f"${item.previous_completed + item.current_completed + item.materials_stored:,.0f} | "
                f"{item.percent_complete:.0f}% | ${item.balance_to_finish:,.0f} | "
                f"${item.retainage:,.0f} |"
            )

        lines.extend([
            "",
            f"| **TOTALS** | | ${pay_app.total_scheduled:,.0f} | "
            f"${pay_app.previous_completed:,.0f} | ${pay_app.current_completed:,.0f} | "
            f"${pay_app.materials_stored:,.0f} | ${pay_app.total_completed:,.0f} | "
            f"{pay_app.percent_complete:.0f}% | ${pay_app.balance_to_finish:,.0f} | "
            f"${pay_app.retainage:,.0f} |"
        ])

        return "\n".join(lines)

    def get_billing_summary(self) -> Dict:
        """Get billing summary across all pay apps."""
        total_billed = sum(pa.total_completed for pa in self.pay_apps.values())
        total_paid = sum(pa.paid_amount for pa in self.pay_apps.values())
        total_retainage = sum(pa.retainage for pa in self.pay_apps.values())

        pending = [pa for pa in self.pay_apps.values()
                   if pa.status in [PayAppStatus.SUBMITTED, PayAppStatus.UNDER_REVIEW, PayAppStatus.APPROVED]]

        return {
            "contract_sum": self.original_contract + self.change_orders_total,
            "total_billed": total_billed,
            "total_paid": total_paid,
            "total_retainage": total_retainage,
            "outstanding": total_billed - total_paid - total_retainage,
            "pending_apps": len(pending),
            "pending_amount": sum(pa.net_amount_due for pa in pending),
            "percent_billed": total_billed / (self.original_contract + self.change_orders_total) * 100
        }

Quick Start

from datetime import datetime

# Initialize processor
processor = PaymentApplicationProcessor(
    project_name="Office Tower",
    contractor="ABC Construction",
    original_contract=5000000
)

# Import schedule of values
processor.import_sov([
    {"number": "01", "description": "General Conditions", "value": 250000},
    {"number": "02", "description": "Sitework", "value": 300000},
    {"number": "03", "description": "Concrete", "value": 800000},
    {"number": "04", "description": "Structural Steel", "value": 1200000},
    {"number": "05", "description": "MEP", "value": 1500000},
    {"number": "06", "description": "Finishes", "value": 950000},
])

# Create pay application
pay_app = processor.create_pay_app(
    period_from=datetime(2024, 1, 1),
    period_to=datetime(2024, 1, 31)
)

# Update progress
processor.update_by_percentage(pay_app.application_number, "01", 100)  # GC 100%
processor.update_by_percentage(pay_app.application_number, "02", 75)   # Site 75%
processor.update_by_percentage(pay_app.application_number, "03", 50)   # Concrete 50%
processor.update_line_item(pay_app.application_number, "04",
                          current_completed=200000, materials_stored=300000)

# Submit and approve
processor.submit_pay_app(pay_app.application_number)
processor.approve_pay_app(pay_app.application_number)

# Generate G703
print(processor.generate_g703(pay_app.application_number))

# Get summary
summary = processor.get_billing_summary()
print(f"Percent Billed: {summary['percent_billed']:.1f}%")

Requirements

pip install (no external dependencies)