cashflow-forecaster

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

# 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)