Cashflow Forecaster
Overview
Forecast construction project cash flow based on schedule, billing cycles, and payment terms. Identify potential cash shortfalls, optimize payment timing, and support project financing decisions.
Cash Flow Curve
┌─────────────────────────────────────────────────────────────────┐
│ CONSTRUCTION CASH FLOW │
├─────────────────────────────────────────────────────────────────┤
│ │
│ $ Income (payments received) │
│ │ ╱──────────╲ │
│ │ ╱ ╲ Positive cash │
│ │ ╱ ╲ position │
│ │ ╱ ╲ │
│ │ ╱ Cash Gap ╲ │
│ ├───────────────────────────────────────────────────── │
│ │╲ │
│ │ ╲ Expenses (costs incurred) │
│ │ ╲──────────╱ │
│ │ │
│ └────────────────────────────────────────────────────────── │
│ Time → │
│ │
└─────────────────────────────────────────────────────────────────┘Technical Implementation
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from datetime import datetime, timedelta
from enum import Enum
import math
class CostCategory(Enum):
LABOR = "labor"
MATERIALS = "materials"
EQUIPMENT = "equipment"
SUBCONTRACTOR = "subcontractor"
GENERAL_CONDITIONS = "general_conditions"
OVERHEAD = "overhead"
OTHER = "other"
class PaymentTerms(Enum):
NET_30 = 30
NET_45 = 45
NET_60 = 60
NET_90 = 90
@dataclass
class CostItem:
id: str
description: str
category: CostCategory
amount: float
scheduled_date: datetime
payment_terms_days: int = 30
paid: bool = False
paid_date: Optional[datetime] = None
@dataclass
class IncomeItem:
id: str
description: str
amount: float
billing_date: datetime
expected_payment_date: datetime
received: bool = False
received_date: Optional[datetime] = None
received_amount: float = 0.0
@dataclass
class CashFlowPeriod:
period_start: datetime
period_end: datetime
opening_balance: float
income: float
expenses: float
net_cashflow: float
closing_balance: float
cumulative_income: float
cumulative_expenses: float
@dataclass
class CashFlowForecast:
project_name: str
forecast_date: datetime
total_contract: float
total_costs: float
periods: List[CashFlowPeriod]
peak_deficit: float
peak_deficit_date: datetime
breakeven_date: Optional[datetime]
financing_required: float
class CashFlowForecaster:
"""Forecast construction project cash flow."""
# Typical cost distribution curve (S-curve)
S_CURVE = [0.05, 0.10, 0.15, 0.20, 0.20, 0.15, 0.10, 0.05]
def __init__(self, project_name: str, contract_value: float,
estimated_cost: float, start_date: datetime,
duration_months: int):
self.project_name = project_name
self.contract_value = contract_value
self.estimated_cost = estimated_cost
self.start_date = start_date
self.duration_months = duration_months
self.end_date = start_date + timedelta(days=duration_months * 30)
self.cost_items: List[CostItem] = []
self.income_items: List[IncomeItem] = []
self.retainage_rate = 0.10 # 10%
self.payment_terms_income = PaymentTerms.NET_30
self.billing_frequency = 30 # Monthly
def set_payment_terms(self, income_terms: PaymentTerms,
retainage_rate: float = 0.10):
"""Set payment terms for income."""
self.payment_terms_income = income_terms
self.retainage_rate = retainage_rate
def add_cost_item(self, description: str, category: CostCategory,
amount: float, scheduled_date: datetime,
payment_terms_days: int = 30) -> CostItem:
"""Add cost item to forecast."""
item = CostItem(
id=f"COST-{len(self.cost_items)+1:04d}",
description=description,
category=category,
amount=amount,
scheduled_date=scheduled_date,
payment_terms_days=payment_terms_days
)
self.cost_items.append(item)
return item
def generate_cost_distribution(self, cost_breakdown: Dict[CostCategory, float] = None):
"""Generate cost items based on S-curve distribution."""
if cost_breakdown is None:
# Default breakdown
cost_breakdown = {
CostCategory.LABOR: self.estimated_cost * 0.35,
CostCategory.MATERIALS: self.estimated_cost * 0.30,
CostCategory.SUBCONTRACTOR: self.estimated_cost * 0.20,
CostCategory.EQUIPMENT: self.estimated_cost * 0.05,
CostCategory.GENERAL_CONDITIONS: self.estimated_cost * 0.07,
CostCategory.OVERHEAD: self.estimated_cost * 0.03,
}
# Distribute costs over project duration using S-curve
months = self.duration_months
curve_months = len(self.S_CURVE)
for category, total in cost_breakdown.items():
for month in range(months):
# Map to S-curve
curve_idx = int(month / months * curve_months)
curve_idx = min(curve_idx, curve_months - 1)
monthly_pct = self.S_CURVE[curve_idx]
# Adjust for number of months
adjustment = months / curve_months
amount = total * monthly_pct / adjustment
cost_date = self.start_date + timedelta(days=month * 30)
# Payment terms vary by category
payment_days = 30
if category == CostCategory.SUBCONTRACTOR:
payment_days = 45
elif category == CostCategory.MATERIALS:
payment_days = 30
self.add_cost_item(
f"{category.value} - Month {month+1}",
category,
amount,
cost_date,
payment_days
)
def generate_billing_schedule(self):
"""Generate income items based on billing schedule."""
# Monthly billing based on progress
months = self.duration_months
for month in range(months):
# Map to S-curve for progress
curve_months = len(self.S_CURVE)
curve_idx = int(month / months * curve_months)
curve_idx = min(curve_idx, curve_months - 1)
monthly_pct = self.S_CURVE[curve_idx]
# Adjust for number of months
adjustment = months / curve_months
billing_amount = self.contract_value * monthly_pct / adjustment
# Apply retainage
retainage = billing_amount * self.retainage_rate
net_billing = billing_amount - retainage
billing_date = self.start_date + timedelta(days=(month + 1) * 30)
payment_date = billing_date + timedelta(days=self.payment_terms_income.value)
self.income_items.append(IncomeItem(
id=f"INC-{month+1:04d}",
description=f"Progress Billing #{month+1}",
amount=net_billing,
billing_date=billing_date,
expected_payment_date=payment_date
))
# Retainage release at end