| name | cashflow-forecaster |
| description | Forecast construction project cash flow. Project income and expenses, identify funding gaps, and optimize payment timing for improved financial management. |
| homepage | https://datadrivenconstruction.io |
| metadata | {"openclaw":{"emoji":"💵","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}} |
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:
billing_date: datetime
expected_payment_date: datetime
received: =
received_date: [datetime] =
received_amount: =
:
period_start: datetime
period_end: datetime
opening_balance:
income:
expenses:
net_cashflow:
closing_balance:
cumulative_income:
cumulative_expenses:
:
project_name:
forecast_date: datetime
total_contract:
total_costs:
periods: [CashFlowPeriod]
peak_deficit:
peak_deficit_date: datetime
breakeven_date: [datetime]
financing_required:
:
S_CURVE = [, , , , , , , ]
():
.project_name = project_name
.contract_value = contract_value
.estimated_cost = estimated_cost
.start_date = start_date
.duration_months = duration_months
.end_date = start_date + timedelta(days=duration_months * )
.cost_items: [CostItem] = []
.income_items: [IncomeItem] = []
.retainage_rate =
.payment_terms_income = PaymentTerms.NET_30
.billing_frequency =
():
.payment_terms_income = income_terms
.retainage_rate = retainage_rate
() -> CostItem:
item = CostItem(
=,
description=description,
category=category,
amount=amount,
scheduled_date=scheduled_date,
payment_terms_days=payment_terms_days
)
.cost_items.append(item)
item
():
cost_breakdown :
cost_breakdown = {
CostCategory.LABOR: .estimated_cost * ,
CostCategory.MATERIALS: .estimated_cost * ,
CostCategory.SUBCONTRACTOR: .estimated_cost * ,
CostCategory.EQUIPMENT: .estimated_cost * ,
CostCategory.GENERAL_CONDITIONS: .estimated_cost * ,
CostCategory.OVERHEAD: .estimated_cost * ,
}
months = .duration_months
curve_months = (.S_CURVE)
category, total cost_breakdown.items():
month (months):
curve_idx = (month / months * curve_months)
curve_idx = (curve_idx, curve_months - )
monthly_pct = .S_CURVE[curve_idx]
adjustment = months / curve_months
amount = total * monthly_pct / adjustment
cost_date = .start_date + timedelta(days=month * )
payment_days =
category == CostCategory.SUBCONTRACTOR:
payment_days =
category == CostCategory.MATERIALS:
payment_days =
.add_cost_item(
,
category,
amount,
cost_date,
payment_days
)
():
months = .duration_months
month (months):
curve_months = (.S_CURVE)
curve_idx = (month / months * curve_months)
curve_idx = (curve_idx, curve_months - )
monthly_pct = .S_CURVE[curve_idx]
adjustment = months / curve_months
billing_amount = .contract_value * monthly_pct / adjustment
retainage = billing_amount * .retainage_rate
net_billing = billing_amount - retainage
billing_date = .start_date + timedelta(days=(month + ) * )
payment_date = billing_date + timedelta(days=.payment_terms_income.value)
.income_items.append(IncomeItem(
=,
description=,
amount=net_billing,
billing_date=billing_date,
expected_payment_date=payment_date
))
total_retainage = .contract_value * .retainage_rate
final_date = .end_date + timedelta(days=)
.income_items.append(IncomeItem(
=,
description=,
amount=total_retainage,
billing_date=final_date,
expected_payment_date=final_date + timedelta(days=.payment_terms_income.value)
))
() -> CashFlowForecast:
.cost_items:
.generate_cost_distribution()
.income_items:
.generate_billing_schedule()
periods = []
current_date = .start_date
balance = opening_balance
cumulative_income =
cumulative_expenses =
peak_deficit =
peak_deficit_date = current_date
breakeven_date =
forecast_end = .end_date + timedelta(days=)
current_date < forecast_end:
period_end = current_date + timedelta(days=period_days)
period_expenses = (
c.amount c .cost_items
current_date <= c.scheduled_date + timedelta(days=c.payment_terms_days) < period_end
)
period_income = (
i.amount i .income_items
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)
closing_balance < peak_deficit:
peak_deficit = closing_balance
peak_deficit_date = current_date
breakeven_date closing_balance > balance <= :
breakeven_date = current_date
balance = closing_balance
current_date = period_end
financing_required = (peak_deficit) peak_deficit <
CashFlowForecast(
project_name=.project_name,
forecast_date=datetime.now(),
total_contract=.contract_value,
total_costs=.estimated_cost,
periods=periods,
peak_deficit=peak_deficit,
peak_deficit_date=peak_deficit_date,
breakeven_date=breakeven_date,
financing_required=financing_required
)
() -> [, CashFlowForecast]:
scenarios = {}
scenarios[] = .generate_forecast()
original_terms = .payment_terms_income
.payment_terms_income = PaymentTerms.NET_30
scenarios[] = .generate_forecast()
.payment_terms_income = PaymentTerms.NET_60
scenarios[] = .generate_forecast()
.payment_terms_income = original_terms
scenarios
() -> :
forecast.financing_required == :
{: , : }
total_deficit_days =
weighted_deficit =
period forecast.periods:
period.closing_balance < :
deficit = (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 total_deficit_days
daily_rate = annual_rate /
interest_cost = weighted_deficit * daily_rate
{
: ,
: forecast.peak_deficit,
: total_deficit_days,
: avg_deficit,
: annual_rate,
: interest_cost,
:
}
() -> :
lines = [
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
,
]
period forecast.periods:
period.income > period.expenses > :
lines.append(
)
financing = .calculate_financing_cost(forecast)
financing[]:
lines.extend([
,
,
,
,
,
,
])
.join(lines)
Quick Start
from datetime import datetime
forecaster = CashFlowForecaster(
project_name="Office Tower",
contract_value=5000000,
estimated_cost=4200000,
start_date=datetime(2024, 1, 1),
duration_months=12
)
forecaster.set_payment_terms(
income_terms=PaymentTerms.NET_45,
retainage_rate=0.10
)
forecast = forecaster.generate_forecast(opening_balance=100000)
print(f"Peak Cash Deficit: ${forecast.peak_deficit:,.0f}")
print(f"Financing Required: ${forecast.financing_required:,.0f}")
scenarios = forecaster.analyze_scenarios()
for name, scen in scenarios.items():
print(f"{name}: Peak deficit ${scen.peak_deficit:,.0f}")
print(forecaster.generate_report(forecast))
Requirements
pip install (no external dependencies)
