Cash Flow Forecaster

Business Case

Problem Statement

Poor cash flow management causes issues:

• Insufficient funds for payments
• Missed early payment discounts
• Inaccurate financial projections
• Difficulty in financing negotiations

Solution

Generate cash flow forecasts from schedule and cost data, including S-curve projections and payment timing analysis.

Business Value

• Financial planning - Accurate funding requirements
• Vendor relations - Timely payments
• Financing - Support loan draw schedules
• Decision support - Cash position awareness

Technical Implementation

import pandas as pd
import numpy as np
from datetime import datetime, date, timedelta
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum


class CashFlowType(Enum):
    """Cash flow types."""
    INFLOW = "inflow"
    OUTFLOW = "outflow"


class PaymentTerms(Enum):
    """Standard payment terms."""
    NET_30 = 30
    NET_45 = 45
    NET_60 = 60
    NET_90 = 90
    MILESTONE = 0
    PROGRESS = 0


@dataclass
class CostItem:
    """Cost item for cash flow."""
    item_id: str
    description: str
    total_amount: float
    start_date: date
    end_date: date
    payment_terms: PaymentTerms
    distribution: str = "linear"  # linear, front_loaded, back_loaded, s_curve
    retention_percent: float = 0.10
    category: str = ""


@dataclass
class PaymentSchedule:
    """Scheduled payment."""
    payment_id: str
    item_id: str
    description: str
    amount: float
    due_date: date
    payment_type: CashFlowType
    is_retention: bool = False
    paid: bool = False
    paid_date: Optional[date] = None


@dataclass
class CashFlowPeriod:
    """Cash flow for a period."""
    period_start: date
    period_end: date
    inflows: float
    outflows: float
    net_cash_flow: float
    cumulative_cash_flow: float
    opening_balance: float
    closing_balance: float


class CashFlowForecaster:
    """Forecast project cash flow."""

    def __init__(self, project_name: str, project_start: date, project_end: date,
                 initial_balance: float = 0, currency: str = "USD"):
        self.project_name = project_name
        self.project_start = project_start
        self.project_end = project_end
        self.initial_balance = initial_balance
        self.currency = currency
        self.cost_items: List[CostItem] = []
        self.revenue_items: List[CostItem] = []
        self.payments: List[PaymentSchedule] = []
        self._payment_counter = 0

    def add_cost_item(self, item_id: str, description: str, total_amount: float,
                     start_date: date, end_date: date,
                     payment_terms: PaymentTerms = PaymentTerms.NET_30,
                     distribution: str = "linear",
                     retention: float = 0.10,
                     category: str = "") -> CostItem:
        """Add cost item (outflow)."""
        item = CostItem(
            item_id=item_id,
            description=description,
            total_amount=total_amount,
            start_date=start_date,
            end_date=end_date,
            payment_terms=payment_terms,
            distribution=distribution,
            retention_percent=retention,
            category=category
        )
        self.cost_items.append(item)
        return item

    def add_revenue_item(self, item_id: str, description: str, total_amount: float,
                        start_date: date, end_date: date,
                        payment_terms: PaymentTerms = PaymentTerms.NET_30,
                        distribution: str = "linear",
                        retention: float = 0.10) -> CostItem:
        """Add revenue item (inflow)."""
        item = CostItem(
            item_id=item_id,
            description=description,
            total_amount=total_amount,
            start_date=start_date,
            end_date=end_date,
            payment_terms=payment_terms,
            distribution=distribution,
            retention_percent=retention
        )
        self.revenue_items.append(item)
        return item

    def _distribute_amount(self, total: float, start: date, end: date,
                          distribution: str, periods: int) -> List[float]:
        """Distribute amount over periods based on distribution type."""
        if periods <= 0:
            return [total]

        if distribution == "linear":
            return [total / periods] * periods
        elif distribution == "front_loaded":
            # More at the beginning
            weights = [periods - i for i in range(periods)]
            total_weight = sum(weights)
            return [total * w / total_weight for w in weights]
        elif distribution == "back_loaded":
            # More at the end
            weights = [i + 1 for i in range(periods)]
            total_weight = sum(weights)
            return [total * w / total_weight for w in weights]
        elif distribution == "s_curve":
            # S-curve distribution
            x = np.linspace(-3, 3, periods)
            weights = 1 / (1 + np.exp(-x))
            weights = weights / weights.sum()
            return [total * w for w in weights]
        else:
            return [total / periods] * periods

    def generate_payment_schedule(self, period_type: str = "monthly") -> List[PaymentSchedule]:
        """Generate payment schedule from cost items."""
        self.payments = []

        # Process cost items (outflows)
        for item in self.cost_items:
            self._generate_item_payments(item, CashFlowType.OUTFLOW, period_type)

        # Process revenue items (inflows)
        for item in self.revenue_items:
            self._generate_item_payments(item, CashFlowType.INFLOW, period_type)

        return sorted(self.payments, key=lambda x: x.due_date)

    def _generate_item_payments(self, item: CostItem, flow_type: CashFlowType,
                               period_type: str):
        """Generate payments for a single item."""
        # Calculate number of periods
        if period_type == "monthly":
            months = (item.end_date.year - item.start_date.year) * 12 + \
                    (item.end_date.month - item.start_date.month) + 1
            periods = max(1, months)
        else:  # weekly
            days = (item.end_date - item.start_date).days
            periods = max(1, days // 7)

        # Distribute amount
        net_amount = item.total_amount * (1 - item.retention_percent)
        amounts = self._distribute_amount(net_amount, item.start_date, item.end_date,
                                         item.distribution, periods)

        # Create payments
        current_date = item.start_date
        for i, amount in enumerate(amounts):
            # Calculate payment due date based on terms
            if item.payment_terms == PaymentTerms.MILESTONE:
                due_date = current_date
            else:
                due_date = current_date + timedelta(days=item.payment_terms.value)

            self._payment_counter += 1
            payment = PaymentSchedule(
                payment_id=f"PAY-{self._payment_counter:05d}",
                item_id=item.item_id,
                description=f"{item.description} - Period {i+1}",
                amount=amount,
                due_date=due_date,
                payment_type=flow_type
            )
            self.payments.append(payment)

            # Move to next period
            if period_type == "monthly":
                if current_date.month == 12:
                    current_date = date(current_date.year + 1, 1, current_date.day)
                else:
                    try:
                        current_date = date(current_date.year, current_date.month + 1, current_date.day)
                    except ValueError:
                        # Handle months with fewer days
                        current_date = date(current_date.year, current_date.month + 1, 28)