kbarbel640-del 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
total_retainage = self.contract_value * self.retainage_rate
final_date = self.end_date + timedelta(days=30)
self.income_items.append(IncomeItem(
id="INC-RET",
description="Retainage Release",
amount=total_retainage,
billing_date=final_date,
expected_payment_date=final_date + timedelta(days=self.payment_terms_income.value)
))
def generate_forecast(self, period_days: int = 30,
opening_balance: float = 0) -> CashFlowForecast:
"""Generate cash flow forecast."""
if not self.cost_items:
self.generate_cost_distribution()
if not self.income_items:
self.generate_billing_schedule()
periods = []
current_date = self.start_date
balance = opening_balance
cumulative_income = 0
cumulative_expenses = 0
peak_deficit = 0
peak_deficit_date = current_date
breakeven_date = None
# Extend forecast beyond project end
forecast_end = self.end_date + timedelta(days=90)
while current_date < forecast_end:
period_end = current_date + timedelta(days=period_days)
# Calculate expenses for period (when paid, not when incurred)
period_expenses = sum(
c.amount for c in self.cost_items
if current_date <= c.scheduled_date + timedelta(days=c.payment_terms_days) < period_end
)
# Calculate income for period (when received)
period_income = sum(
i.amount for i in self.income_items
if current_date <= i.expected_payment_date < period_end
)
net_cashflow = period_income - period_expenses
closing_balance = balance + net_cashflow
cumulative_income += period_income
cumulative_expenses += period_expenses
period = CashFlowPeriod(
period_start=current_date,
period_end=period_end,
opening_balance=balance,
income=period_income,
expenses=period_expenses,
net_cashflow=net_cashflow,
closing_balance=closing_balance,
cumulative_income=cumulative_income,
cumulative_expenses=cumulative_expenses
)
periods.append(period)
# Track peak deficit
if closing_balance < peak_deficit:
peak_deficit = closing_balance
peak_deficit_date = current_date
# Track breakeven
if breakeven_date is None and closing_balance > 0 and balance <= 0:
breakeven_date = current_date
balance = closing_balance
current_date = period_end
financing_required = abs(peak_deficit) if peak_deficit < 0 else 0
return CashFlowForecast(
project_name=self.project_name,
forecast_date=datetime.now(),
total_contract=self.contract_value,
total_costs=self.estimated_cost,
periods=periods,
peak_deficit=peak_deficit,
peak_deficit_date=peak_deficit_date,
breakeven_date=breakeven_date,
financing_required=financing_required
)
def analyze_scenarios(self) -> Dict[str, CashFlowForecast]:
"""Analyze different payment scenarios."""
scenarios = {}
# Base case
scenarios["base"] = self.generate_forecast()
# Optimistic - faster payments
original_terms = self.payment_terms_income
self.payment_terms_income = PaymentTerms.NET_30
scenarios["optimistic"] = self.generate_forecast()
# Pessimistic - slower payments
self.payment_terms_income = PaymentTerms.NET_60
scenarios["pessimistic"] = self.generate_forecast()
self.payment_terms_income = original_terms
return scenarios
def calculate_financing_cost(self, forecast: CashFlowForecast,
annual_rate: float = 0.08) -> Dict:
"""Calculate cost of financing the cash deficit."""
if forecast.financing_required == 0:
return {"financing_needed": False, "cost": 0}
# Calculate weighted average deficit duration
total_deficit_days = 0
weighted_deficit = 0
for period in forecast.periods:
if period.closing_balance < 0:
deficit = abs(period.closing_balance)
days = (period.period_end - period.period_start).days
total_deficit_days += days
weighted_deficit += deficit * days
avg_deficit = weighted_deficit / total_deficit_days if total_deficit_days else 0
# Calculate interest cost
daily_rate = annual_rate / 365
interest_cost = weighted_deficit * daily_rate
return {
"financing_needed": True,
"peak_deficit": forecast.peak_deficit,
"deficit_days": total_deficit_days,
"average_deficit": avg_deficit,
"annual_rate": annual_rate,
"estimated_interest": interest_cost,
"recommendation": f"Line of credit needed: ${forecast.financing_required:,.0f}"
}
def generate_report(self, forecast: CashFlowForecast) -> str:
"""Generate cash flow forecast report."""
lines = [
"# Cash Flow Forecast Report",
"",
f"**Project:** {forecast.project_name}",
f"**Forecast Date:** {forecast.forecast_date.strftime('%Y-%m-%d')}",
"",
"## Summary",
"",
f"| Metric | Value |",
f"|--------|-------|",
f"| Contract Value | ${forecast.total_contract:,.0f} |",
f"| Estimated Cost | ${forecast.total_costs:,.0f} |",
f"| Gross Margin | ${forecast.total_contract - forecast.total_costs:,.0f} ({(forecast.total_contract - forecast.total_costs)/forecast.total_contract*100:.1f}%) |",
f"| Peak Cash Deficit | ${forecast.peak_deficit:,.0f} |",
f"| Peak Deficit Date | {forecast.peak_deficit_date.strftime('%Y-%m-%d')} |",
f"| Financing Required | ${forecast.financing_required:,.0f} |",
"",
"## Monthly Cash Flow",
"",
"| Period | Income | Expenses | Net | Balance |",
"|--------|--------|----------|-----|---------|"
]
for period in forecast.periods:
if period.income > 0 or period.expenses > 0:
lines.append(
f"| {period.period_start.strftime('%Y-%m')} | "
f"${period.income:,.0f} | ${period.expenses:,.0f} | "
f"${period.net_cashflow:,.0f} | ${period.closing_balance:,.0f} |"
)
# Financing analysis
financing = self.calculate_financing_cost(forecast)
if financing["financing_needed"]:
lines.extend([
"",
"## Financing Analysis",
"",
f"- Peak Deficit: ${financing['peak_deficit']:,.0f}",
f"- Days in Deficit: {financing['deficit_days']}",
f"- Estimated Interest Cost: ${financing['estimated_interest']:,.0f}",
f"- **{financing['recommendation']}**"
])
return "\n".join(lines)
Quick Start
from datetime import datetime
# Initialize forecaster
forecaster = CashFlowForecaster(
project_name="Office Tower",
contract_value=5000000,
estimated_cost=4200000,
start_date=datetime(2024, 1, 1),
duration_months=12
)
# Set payment terms
forecaster.set_payment_terms(
income_terms=PaymentTerms.NET_45,
retainage_rate=0.10
)
# Generate forecast
forecast = forecaster.generate_forecast(opening_balance=100000)
print(f"Peak Cash Deficit: ${forecast.peak_deficit:,.0f}")
print(f"Financing Required: ${forecast.financing_required:,.0f}")
# Analyze scenarios
scenarios = forecaster.analyze_scenarios()
for name, scen in scenarios.items():
print(f"{name}: Peak deficit ${scen.peak_deficit:,.0f}")
# Generate report
print(forecaster.generate_report(forecast))
Requirements
pip install (no external dependencies)