cashflow-forecaster

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)