cashflow-forecaster

What this skill does

# 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

```python
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