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)
else:
current_date += timedelta(days=7)
# Add retention release at project end
if item.retention_percent > 0:
retention_amount = item.total_amount * item.retention_percent
self._payment_counter += 1
retention_payment = PaymentSchedule(
payment_id=f"PAY-{self._payment_counter:05d}",
item_id=item.item_id,
description=f"{item.description} - Retention Release",
amount=retention_amount,
due_date=self.project_end + timedelta(days=60),
payment_type=flow_type,
is_retention=True
)
self.payments.append(retention_payment)
def generate_cash_flow_forecast(self, period_type: str = "monthly") -> List[CashFlowPeriod]:
"""Generate cash flow forecast."""
if not self.payments:
self.generate_payment_schedule(period_type)
# Group payments by period
periods = []
current_date = self.project_start
cumulative = 0
balance = self.initial_balance
while current_date <= self.project_end + timedelta(days=90):
# Calculate period end
if period_type == "monthly":
if current_date.month == 12:
period_end = date(current_date.year + 1, 1, 1) - timedelta(days=1)
else:
period_end = date(current_date.year, current_date.month + 1, 1) - timedelta(days=1)
else:
period_end = current_date + timedelta(days=6)
# Filter payments for this period
period_payments = [p for p in self.payments
if current_date <= p.due_date <= period_end]
inflows = sum(p.amount for p in period_payments
if p.payment_type == CashFlowType.INFLOW)
outflows = sum(p.amount for p in period_payments
if p.payment_type == CashFlowType.OUTFLOW)
net = inflows - outflows
cumulative += net
period = CashFlowPeriod(
period_start=current_date,
period_end=period_end,
inflows=inflows,
outflows=outflows,
net_cash_flow=net,
cumulative_cash_flow=cumulative,
opening_balance=balance,
closing_balance=balance + net
)
periods.append(period)
balance = period.closing_balance
# Move to next period
current_date = period_end + timedelta(days=1)
return periods
def generate_s_curve(self) -> pd.DataFrame:
"""Generate S-curve data (cumulative costs over time)."""
forecast = self.generate_cash_flow_forecast()
# Costs only (outflows)
data = []
cumulative_cost = 0
total_cost = sum(item.total_amount for item in self.cost_items)
for period in forecast:
cumulative_cost += period.outflows
percent_complete = (cumulative_cost / total_cost * 100) if total_cost > 0 else 0
data.append({
'date': period.period_end,
'period_cost': period.outflows,
'cumulative_cost': cumulative_cost,
'percent_complete': round(percent_complete, 1),
'total_budget': total_cost
})
return pd.DataFrame(data)
def get_funding_requirements(self, buffer_percent: float = 0.10) -> Dict[str, Any]:
"""Calculate funding requirements."""
forecast = self.generate_cash_flow_forecast()
# Find peak negative cash flow
min_balance = min(p.closing_balance for p in forecast)
peak_funding = abs(min(0, min_balance))
# Add buffer
required_funding = peak_funding * (1 + buffer_percent)
# Monthly funding needs
monthly_needs = []
for period in forecast:
if period.net_cash_flow < 0:
monthly_needs.append({
'month': period.period_start.strftime('%Y-%m'),
'funding_needed': abs(period.net_cash_flow)
})
return {
'peak_funding_required': round(required_funding, 2),
'peak_funding_month': min(forecast, key=lambda x: x.closing_balance).period_start.strftime('%Y-%m'),
'total_outflows': sum(p.outflows for p in forecast),
'total_inflows': sum(p.inflows for p in forecast),
'monthly_funding_needs': monthly_needs,
'buffer_percent': buffer_percent
}
def export_forecast(self, output_path: str):
"""Export cash flow forecast to Excel."""
forecast = self.generate_cash_flow_forecast()
s_curve = self.generate_s_curve()
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
# Cash flow forecast
forecast_df = pd.DataFrame([{
'Period Start': p.period_start,
'Period End': p.period_end,
'Inflows': p.inflows,
'Outflows': p.outflows,
'Net Cash Flow': p.net_cash_flow,
'Cumulative': p.cumulative_cash_flow,
'Opening Balance': p.opening_balance,
'Closing Balance': p.closing_balance
} for p in forecast])
forecast_df.to_excel(writer, sheet_name='Cash Flow', index=False)
# S-curve
s_curve.to_excel(writer, sheet_name='S-Curve', index=False)
# Payment schedule
payments_df = pd.DataFrame([{
'ID': p.payment_id,
'Item': p.item_id,
'Description': p.description,
'Amount': p.amount,
'Due Date': p.due_date,
'Type': p.payment_type.value,
'Retention': p.is_retention
} for p in self.payments])
payments_df.to_excel(writer, sheet_name='Payments', index=False)
return output_path
Quick Start
from datetime import date
Initialize forecaster
forecaster = CashFlowForecaster( project_name="Office Tower", project_start=date(2024, 1, 1), project_end=date(2025, 12, 31), initial_balance=5000000 )
Add costs
forecaster.add_cost_item("CONC", "Concrete Work", 8000000, date(2024, 3, 1), date(2024, 9, 30), distribution="s_curve")
Add revenue
forecaster.add_revenue_item("DRAW", "Owner Draws", 50000000, date(2024, 1, 1), date(2025, 12, 31), distribution="s_curve")
Generate forecast
forecast = forecaster.generate_cash_flow_forecast() print(f"Peak cash requirement: ${min(p.closing_balance for p in forecast):,.0f}")
Common Use Cases
- S-Curve Analysis
s_curve = forecaster.generate_s_curve()
Plot cumulative cost over time
- Funding Requirements
funding = forecaster.get_funding_requirements(buffer_percent=0.15) print(f"Required funding: ${funding['peak_funding_required']:,.0f}")
- Export Report
forecaster.export_forecast("cash_flow_forecast.xlsx")
Resources
-
DDC Book: Chapter 3.1 - Cost Management
-
Reference: Project Financial Management