retention-tracker

name	retention-tracker
description	Track construction retainage/retention amounts. Monitor held amounts by subcontractor, track release conditions, and manage retainage billing.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"💵","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":["python3"]}}}

Retention Tracker

Overview

Track retainage (retention) amounts held and released throughout construction projects. Monitor amounts by subcontractor, track release milestones, and ensure proper documentation for retention release.

Retainage Flow

┌─────────────────────────────────────────────────────────────────┐
│                    RETAINAGE LIFECYCLE                           │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  Progress Billing    →    Substantial    →    Final Release    │
│  ────────────────         ───────────         ─────────────     │
│  10% withheld            50% released         50% released      │
│  Each pay app            At punch list        At final          │
│  Cumulative              completion           completion        │
│                                                                  │
│  Owner holds from GC  →  GC holds from subs  →  Flow-down      │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Technical Implementation

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime, timedelta
from enum import Enum

class RetentionStatus(Enum):
    HELD = "held"
    PARTIAL_RELEASE = "partial_release"
    PENDING_RELEASE = "pending_release"
    RELEASED = "released"

class ReleaseMilestone(Enum):
    SUBSTANTIAL_COMPLETION = "substantial_completion"
    PUNCH_LIST_COMPLETE = "punch_list_complete"
    FINAL_COMPLETION = "final_completion"
    WARRANTY_EXPIRATION = "warranty_expiration"

@dataclass
class RetentionEntry:
    pay_app_number: int
    billing_date: datetime
    gross_billing: float
    retention_rate: float
    retention_amount: float
    status: RetentionStatus = RetentionStatus.HELD

@dataclass
class RetentionRelease:
    id: str
    release_date: datetime
    milestone: ReleaseMilestone
    amount: float
    remaining_after: float
    approved_by: str
    conditions_met: List[str] = field(default_factory=list)
    lien_waivers_received: bool = False
    consent_of_surety: bool = False

@dataclass
class SubcontractorRetention:
    subcontractor_id: str
    subcontractor_name: str
    trade: str
    contract_value: float
    retention_rate: float
    entries: List[RetentionEntry] = field(default_factory=list)
    releases: List[RetentionRelease] = field(default_factory=list)
    total_billed: float = 0.0
    total_retained: float = 0.0
    total_released: float = 0.0
    balance_held: float = 0.0
    status: RetentionStatus = RetentionStatus.HELD

class RetentionTracker:
    """Track construction retainage amounts."""

    # Default release schedule
    DEFAULT_RELEASE_SCHEDULE = {
        ReleaseMilestone.SUBSTANTIAL_COMPLETION: 0.50,  # 50% at substantial
        ReleaseMilestone.FINAL_COMPLETION: 0.50,        # 50% at final
    }

    def __init__(self, project_name: str, default_rate: float = 0.10):
        self.project_name = project_name
        self.default_rate = default_rate
        self.subcontractors: Dict[str, SubcontractorRetention] = {}
        self.release_schedule = dict(self.DEFAULT_RELEASE_SCHEDULE)

        # Project-level retention (from owner)
        self.owner_retention = SubcontractorRetention(
            subcontractor_id="OWNER",
            subcontractor_name="Owner Retention",
            trade="GC",
            contract_value=0,
            retention_rate=default_rate
        )

    def set_release_schedule(self, schedule: Dict[ReleaseMilestone, float]):
        """Set custom release schedule."""
        self.release_schedule = schedule

    def add_subcontractor(self, id: str, name: str, trade: str,
                         contract_value: float,
                         retention_rate: float = None) -> SubcontractorRetention:
        """Add subcontractor to track."""
        sub = SubcontractorRetention(
            subcontractor_id=id,
            subcontractor_name=name,
            trade=trade,
            contract_value=contract_value,
            retention_rate=retention_rate if retention_rate else self.default_rate
        )
        self.subcontractors[id] = sub
        return sub

    def record_billing(self, subcontractor_id: str, pay_app_number: int,
                      billing_date: datetime, gross_billing: float) -> RetentionEntry:
        """Record billing and calculate retention."""
        if subcontractor_id not in self.subcontractors:
            raise ValueError(f"Subcontractor {subcontractor_id} not found")

        sub = self.subcontractors[subcontractor_id]

        retention_amount = gross_billing * sub.retention_rate

        entry = RetentionEntry(
            pay_app_number=pay_app_number,
            billing_date=billing_date,
            gross_billing=gross_billing,
            retention_rate=sub.retention_rate,
            retention_amount=retention_amount
        )

        sub.entries.append(entry)

        # Update totals
        sub.total_billed += gross_billing
        sub.total_retained += retention_amount
        sub.balance_held = sub.total_retained - sub.total_released

        return entry

    def record_owner_billing(self, pay_app_number: int, billing_date: datetime,
                            gross_billing: float) -> RetentionEntry:
        """Record owner-level retention."""
        retention_amount = gross_billing * self.owner_retention.retention_rate

        entry = RetentionEntry(
            pay_app_number=pay_app_number,
            billing_date=billing_date,
            gross_billing=gross_billing,
            retention_rate=self.owner_retention.retention_rate,
            retention_amount=retention_amount
        )

        self.owner_retention.entries.append(entry)
        self.owner_retention.total_billed += gross_billing
        self.owner_retention.total_retained += retention_amount
        self.owner_retention.balance_held = (
            self.owner_retention.total_retained - self.owner_retention.total_released
        )

        return entry

    def release_retention(self, subcontractor_id: str, milestone: ReleaseMilestone,
                         amount: float = None, approved_by: str = "",
                         conditions: List[str] = None,
                         lien_waivers: bool = False,
                         consent_of_surety: bool = False) -> RetentionRelease:
        """Release retention for subcontractor."""
        if subcontractor_id not in self.subcontractors:
            raise ValueError(f"Subcontractor {subcontractor_id} not found")

        sub = self.subcontractors[subcontractor_id]

        # Calculate release amount if not specified
        if amount is None:
            release_pct = self.release_schedule.get(milestone, 0.5)
            amount = sub.balance_held * release_pct

        if amount > sub.balance_held:
            amount = sub.balance_held

        release_id = f"REL-{subcontractor_id}-{len(sub.releases)+1:03d}"

        release = RetentionRelease(
            id=release_id,
            release_date=datetime.now(),
            milestone=milestone,
            amount=amount,
            remaining_after=sub.balance_held - amount,
            approved_by=approved_by,
            conditions_met=conditions or [],
            lien_waivers_received=lien_waivers,
            consent_of_surety=consent_of_surety
        )

        sub.releases.append(release)
        sub.total_released += amount
        sub.balance_held -= amount

        if sub.balance_held <= 0:
            sub.status = RetentionStatus.RELEASED
        elif sub.total_released > 0:
            sub.status = RetentionStatus.PARTIAL_RELEASE

        return release

    def release_owner_retention(self, milestone: ReleaseMilestone,
                               amount: float = None,
                               approved_by: str = "") -> RetentionRelease:
        """Release owner-level retention."""
        if amount is None:
            release_pct = self.release_schedule.get(milestone, 0.5)
            amount = self.owner_retention.balance_held * release_pct

        if amount > self.owner_retention.balance_held:
            amount = self.owner_retention.balance_held

        release_id = f"REL-OWNER-{len(self.owner_retention.releases)+1:03d}"

        release = RetentionRelease(
            id=release_id,
            release_date=datetime.now(),
            milestone=milestone,
            amount=amount,
            remaining_after=self.owner_retention.balance_held - amount,
            approved_by=approved_by
        )

        self.owner_retention.releases.append(release)
        self.owner_retention.total_released += amount
        self.owner_retention.balance_held -= amount

        return release

    def check_release_conditions(self, subcontractor_id: str) -> Dict:
        """Check conditions required for retention release."""
        if subcontractor_id not in self.subcontractors:
            raise ValueError(f"Subcontractor {subcontractor_id} not found")

        sub = self.subcontractors[subcontractor_id]

        conditions = {
            "punch_list_complete": False,  # Check externally
            "final_lien_waiver": False,
            "consent_of_surety": False,
            "as_builts_submitted": False,
            "warranty_documents": False,
            "training_complete": False,
            "closeout_docs": False
        }

        # Check last release for received items
        if sub.releases:
            last_release = sub.releases[-1]
            conditions["final_lien_waiver"] = last_release.lien_waivers_received
            conditions["consent_of_surety"] = last_release.consent_of_surety

        missing = [k for k, v in conditions.items() if not v]

        return {
            "subcontractor": sub.subcontractor_name,
            "balance_held": sub.balance_held,
            "conditions_status": conditions,
            "missing_conditions": missing,
            "ready_for_release": len(missing) == 0
        }

    def get_retention_summary(self) -> Dict:
        """Get overall retention summary."""
        total_retained = sum(s.total_retained for s in self.subcontractors.values())
        total_released = sum(s.total_released for s in self.subcontractors.values())
        total_held = sum(s.balance_held for s in self.subcontractors.values())

        by_status = {}
        for sub in self.subcontractors.values():
            status = sub.status.value
            by_status[status] = by_status.get(status, 0) + 1

        return {
            "owner_retained": self.owner_retention.total_retained,
            "owner_released": self.owner_retention.total_released,
            "owner_balance": self.owner_retention.balance_held,
            "sub_total_retained": total_retained,
            "sub_total_released": total_released,
            "sub_balance_held": total_held,
            "net_retention_position": self.owner_retention.balance_held - total_held,
            "subcontractors_count": len(self.subcontractors),
            "by_status": by_status
        }

    def get_aged_retention(self) -> List[Dict]:
        """Get aged retention report."""
        aged = []

        for sub in self.subcontractors.values():
            if sub.balance_held <= 0:
                continue

            # Find oldest unreleased entry
            oldest_date = None
            for entry in sub.entries:
                if entry.status == RetentionStatus.HELD:
                    if oldest_date is None or entry.billing_date < oldest_date:
                        oldest_date = entry.billing_date

            days_held = (datetime.now() - oldest_date).days if oldest_date else 0

            aged.append({
                "subcontractor_id": sub.subcontractor_id,
                "subcontractor_name": sub.subcontractor_name,
                "trade": sub.trade,
                "balance_held": sub.balance_held,
                "oldest_entry_date": oldest_date,
                "days_held": days_held
            })

        return sorted(aged, key=lambda x: -x["days_held"])

    def forecast_releases(self, substantial_date: datetime,
                         final_date: datetime) -> List[Dict]:
        """Forecast retention releases."""
        forecasts = []

        for sub in self.subcontractors.values():
            if sub.balance_held <= 0:
                continue

            # Substantial completion release
            sub_release = sub.balance_held * self.release_schedule.get(
                ReleaseMilestone.SUBSTANTIAL_COMPLETION, 0.5
            )
            forecasts.append({
                "date": substantial_date,
                "subcontractor": sub.subcontractor_name,
                "milestone": "Substantial Completion",
                "amount": sub_release
            })

            # Final completion release
            final_release = sub.balance_held - sub_release
            forecasts.append({
                "date": final_date,
                "subcontractor": sub.subcontractor_name,
                "milestone": "Final Completion",
                "amount": final_release
            })

        return sorted(forecasts, key=lambda x: x["date"])

    def generate_report(self) -> str:
        """Generate retention status report."""
        summary = self.get_retention_summary()

        lines = [
            "# Retention Status Report",
            "",
            f"**Project:** {self.project_name}",
            f"**Date:** {datetime.now().strftime('%Y-%m-%d')}",
            "",
            "## Summary",
            "",
            "### Owner Retention",
            "",
            f"| Metric | Amount |",
            f"|--------|--------|",
            f"| Total Retained | ${summary['owner_retained']:,.0f} |",
            f"| Total Released | ${summary['owner_released']:,.0f} |",
            f"| **Balance Held** | **${summary['owner_balance']:,.0f}** |",
            "",
            "### Subcontractor Retention",
            "",
            f"| Metric | Amount |",
            f"|--------|--------|",
            f"| Total Retained | ${summary['sub_total_retained']:,.0f} |",
            f"| Total Released | ${summary['sub_total_released']:,.0f} |",
            f"| **Balance Held** | **${summary['sub_balance_held']:,.0f}** |",
            "",
            f"**Net Position:** ${summary['net_retention_position']:,.0f}",
            "",
            "## By Subcontractor",
            "",
            "| Subcontractor | Trade | Contract | Retained | Released | Held | Status |",
            "|---------------|-------|----------|----------|----------|------|--------|"
        ]

        for sub in sorted(self.subcontractors.values(),
                         key=lambda s: -s.balance_held):
            lines.append(
                f"| {sub.subcontractor_name} | {sub.trade} | "
                f"${sub.contract_value:,.0f} | ${sub.total_retained:,.0f} | "
                f"${sub.total_released:,.0f} | ${sub.balance_held:,.0f} | "
                f"{sub.status.value} |"
            )

        # Aged retention
        aged = self.get_aged_retention()
        if aged:
            lines.extend([
                "",
                "## Aged Retention",
                "",
                "| Subcontractor | Balance | Days Held |",
                "|---------------|---------|-----------|"
            ])
            for item in aged[:10]:
                lines.append(
                    f"| {item['subcontractor_name']} | "
                    f"${item['balance_held']:,.0f} | {item['days_held']} |"
                )

        return "\n".join(lines)

Quick Start

from datetime import datetime

# Initialize tracker
tracker = RetentionTracker("Office Tower", default_rate=0.10)

# Add subcontractors
tracker.add_subcontractor("SUB-001", "ABC Mechanical", "HVAC", 500000)
tracker.add_subcontractor("SUB-002", "XYZ Electric", "Electrical", 350000)

# Record billings
tracker.record_billing("SUB-001", 1, datetime(2024, 1, 31), 50000)
tracker.record_billing("SUB-001", 2, datetime(2024, 2, 29), 75000)
tracker.record_billing("SUB-002", 1, datetime(2024, 1, 31), 35000)

# Record owner-level retention
tracker.record_owner_billing(1, datetime(2024, 1, 31), 200000)

# Check release conditions
conditions = tracker.check_release_conditions("SUB-001")
print(f"Missing conditions: {conditions['missing_conditions']}")

# Release retention at substantial completion
tracker.release_retention(
    "SUB-001",
    ReleaseMilestone.SUBSTANTIAL_COMPLETION,
    approved_by="Project Manager",
    lien_waivers=True
)

# Generate report
print(tracker.generate_report())

# Forecast releases
forecasts = tracker.forecast_releases(
    datetime(2024, 12, 1),
    datetime(2025, 1, 15)
)
for f in forecasts[:5]:
    print(f"{f['date'].strftime('%Y-%m-%d')}: {f['subcontractor']} - ${f['amount']:,.0f}")

Requirements

pip install (no external dependencies)