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)