← Files Investment BankingARCHIVED FILE
skills/three-statement-model-builder/scripts/skill_core_runtime
62 KB · Oct 2, 2026 · 00:27 UTC
"""Core deterministic 3-statement operating model engine.
Design goals:
- No network calls, no hidden randomness, standard library only.
- Preserve source/assumption labels in outputs.
- Produce a long-format model table suitable for deterministic export.
- Keep workbook rendering separate from source ingestion and model judgment.
"""
from __future__ import annotations
import copy
import json
import math
import re
import zipfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple
from xml.sax.saxutils import escape
ALLOWED_PERIODICITIES = {"annual", "quarterly"}
ALLOWED_SCENARIOS = ("base", "downside", "upside")
MODEL_TOLERANCE = 0.05
@dataclass
class Period:
index: int
label: str
year: int
periodicity: str
time_factor: float
def deep_merge(base: Any, override: Any) -> Any:
"""Recursively merge override into base without mutating either argument."""
if override is None:
return copy.deepcopy(base)
if isinstance(base, dict) and isinstance(override, dict):
out = copy.deepcopy(base)
for key, value in override.items():
out[key] = deep_merge(out.get(key), value)
return out
return copy.deepcopy(override)
def coalesce(*values: Any, default: Any = None) -> Any:
for value in values:
if value is not None:
return value
return default
def is_number(value: Any) -> bool:
return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(float(value))
def period_sort_key(label: str) -> Tuple[int, int]:
text = str(label)
year_match = re.search(r"(19|20|21)\d{2}", text)
year = int(year_match.group(0)) if year_match else 0
q_match = re.search(r"q([1-4])", text.lower())
quarter = int(q_match.group(1)) if q_match else 4
return (year, quarter)
def latest_period_key(mapping: Dict[str, Any]) -> Optional[str]:
if not isinstance(mapping, dict) or not mapping:
return None
return sorted(mapping.keys(), key=period_sort_key)[-1]
def normalize_plan(plan: Dict[str, Any], skill_root: Path) -> Dict[str, Any]:
"""Apply non-conclusion-changing defaults and canonical metadata."""
normalized = copy.deepcopy(plan)
normalized.setdefault("meta", {})
normalized["meta"].setdefault("currency", "USD")
normalized["meta"].setdefault("units", "USD_mm")
normalized["meta"].setdefault("accounting_basis", "unspecified")
normalized.setdefault("source_basis", [])
normalized.setdefault("scenarios", {"base": {"overrides": {}}, "downside": {"overrides": {}}, "upside": {"overrides": {}}})
normalized.setdefault("sensitivities", {})
normalized.setdefault("other_balance_sheet", {})
normalized["other_balance_sheet"].setdefault("other_assets", last_bs_value(normalized, "other_assets", 0.0))
normalized["other_balance_sheet"].setdefault("other_liabilities", last_bs_value(normalized, "other_liabilities", 0.0))
normalized["workbook_mode"] = "deterministic_export"
normalized["artifact_level"] = "deterministic_export"
return normalized
def build_timeline(start_year: int, horizon_periods: int, periodicity: str) -> List[Period]:
if periodicity not in ALLOWED_PERIODICITIES:
raise ValueError(f"Unsupported periodicity: {periodicity}")
if horizon_periods <= 0:
raise ValueError("horizon_periods must be positive")
periods: List[Period] = []
if periodicity == "annual":
for i in range(horizon_periods):
year = start_year + i
periods.append(Period(i, f"FY{year}", year, periodicity, 1.0))
else:
total_quarters = horizon_periods
for i in range(total_quarters):
year = start_year + (i // 4)
quarter = (i % 4) + 1
periods.append(Period(i, f"Q{quarter}-FY{year}", year, periodicity, 0.25))
return periods
def get_assumption(value: Any, period: Period, default: float = 0.0) -> float:
"""Fetch scalar or period map value with prior-period fallback."""
if value is None:
return float(default)
if is_number(value):
return float(value)
if not isinstance(value, dict) or not value:
return float(default)
exact_keys = [period.label, f"FY{period.year}", str(period.year)]
for key in exact_keys:
if key in value and value[key] is not None:
return float(value[key])
keyed: List[Tuple[Tuple[int, int], str]] = []
for key in value.keys():
keyed.append((period_sort_key(str(key)), str(key)))
keyed.sort()
target = period_sort_key(period.label)
prior = [key for key_sort, key in keyed if key_sort <= target]
if prior:
return float(value[prior[-1]])
return float(value[keyed[0][1]])
def periodicize_annual_rate(annual_rate: float, time_factor: float) -> float:
if time_factor >= 1.0:
return annual_rate
if annual_rate <= -0.99:
return -0.99
return (1.0 + annual_rate) ** time_factor - 1.0
def last_bs(plan: Dict[str, Any]) -> Dict[str, Any]:
bs = plan.get("historicals", {}).get("balance_sheet", {})
key = latest_period_key(bs)
return copy.deepcopy(bs.get(key, {})) if key else {}
def last_bs_value(plan: Dict[str, Any], key: str, default: float = 0.0) -> float:
value = last_bs(plan).get(key, default)
return float(value or 0.0)
def last_nwc(plan: Dict[str, Any]) -> float:
wc_hist = plan.get("historicals", {}).get("working_capital", {})
key = latest_period_key(wc_hist)
if key and is_number(wc_hist[key].get("nwc")):
return float(wc_hist[key]["nwc"])
bs = last_bs(plan)
return (
float(bs.get("ar", 0.0) or 0.0)
+ float(bs.get("inventory", 0.0) or 0.0)
+ float(bs.get("other_current_assets", 0.0) or 0.0)
- float(bs.get("ap", 0.0) or 0.0)
- float(bs.get("accrued_expenses", 0.0) or 0.0)
- float(bs.get("deferred_revenue", 0.0) or 0.0)
)
def last_hist_is(plan: Dict[str, Any]) -> Dict[str, Any]:
hist = plan.get("historicals", {}).get("income_statement", {})
key = latest_period_key(hist)
return copy.deepcopy(hist.get(key, {})) if key else {}
def compute_revenue(plan: Dict[str, Any], periods: List[Period]) -> Dict[str, Any]:
revenue_plan = plan.get("revenue", {})
model = revenue_plan.get("model", "total_growth")
total: List[float] = []
segment_rows: Dict[str, List[float]] = {}
if model == "segments":
segments = revenue_plan.get("segments", {})
annualized: Dict[str, float] = {}
for name, cfg in segments.items():
annualized[name] = float(cfg.get("base_revenue", 0.0) or 0.0)
segment_rows[name] = []
for period in periods:
period_total = 0.0
for name, cfg in segments.items():
growth = get_assumption(cfg.get("growth_rates"), period, 0.0)
growth = periodicize_annual_rate(growth, period.time_factor)
annualized[name] *= 1.0 + growth
value = annualized[name] * period.time_factor
segment_rows[name].append(value)
period_total += value
total.append(period_total)
return {"revenue": total, "segments": segment_rows, "source_id": revenue_plan.get("source_id"), "evidence_label": revenue_plan.get("evidence_label", "model_calculated")}
if model == "volume_price":
units = float(revenue_plan.get("base_units", 0.0) or 0.0)
price = float(revenue_plan.get("base_price", 0.0) or 0.0)
segment_rows["volume_price_revenue"] = []
for period in periods:
ug = periodicize_annual_rate(get_assumption(revenue_plan.get("unit_growth_rates"), period, 0.0), period.time_factor)
pg = periodicize_annual_rate(get_assumption(revenue_plan.get("price_growth_rates"), period, 0.0), period.time_factor)
units *= 1.0 + ug
price *= 1.0 + pg
value = units * price * period.time_factor
total.append(value)
segment_rows["volume_price_revenue"].append(value)
return {"revenue": total, "segments": segment_rows, "source_id": revenue_plan.get("source_id"), "evidence_label": revenue_plan.get("evidence_label", "model_calculated")}
base_revenue = float(coalesce(revenue_plan.get("base_revenue"), last_hist_is(plan).get("revenue"), default=0.0) or 0.0)
annualized_revenue = base_revenue
segment_rows["total_revenue"] = []
for period in periods:
growth = periodicize_annual_rate(get_assumption(revenue_plan.get("growth_rates"), period, 0.0), period.time_factor)
annualized_revenue *= 1.0 + growth
value = annualized_revenue * period.time_factor
total.append(value)
segment_rows["total_revenue"].append(value)
return {"revenue": total, "segments": segment_rows, "source_id": revenue_plan.get("source_id"), "evidence_label": revenue_plan.get("evidence_label", "model_calculated")}
def compute_income_statement(
plan: Dict[str, Any],
periods: List[Period],
revenue_result: Dict[str, Any],
ppe_result: Optional[Dict[str, List[float]]] = None,
interest: Optional[List[float]] = None,
) -> Dict[str, List[float]]:
costs = plan.get("costs", {})
cogs_plan = costs.get("cogs", {})
opex_plan = costs.get("opex", {})
revenue = revenue_result["revenue"]
da = ppe_result.get("depreciation", [0.0] * len(periods)) if ppe_result else [0.0] * len(periods)
interest_values = interest if interest is not None else [0.0] * len(periods)
out: Dict[str, List[float]] = {k: [] for k in ["revenue", "cogs", "gross_profit", "opex", "ebitda", "da", "ebit", "interest", "ebt", "book_taxes", "cash_taxes", "net_income", "deferred_tax", "nol_used", "ending_nol"]}
nol = float(plan.get("tax", {}).get("nol_balance", 0.0) or 0.0)
book_tax_rate = float(plan.get("tax", {}).get("book_tax_rate", 0.0) or 0.0)
cash_tax_rate = float(plan.get("tax", {}).get("cash_tax_rate", book_tax_rate) or 0.0)
for i, period in enumerate(periods):
rev = revenue[i]
if cogs_plan.get("method", "gross_margin") == "pct_revenue":
cogs = rev * get_assumption(cogs_plan.get("pct_revenue"), period, 0.0)
else:
gm = get_assumption(cogs_plan.get("gross_margin"), period, 0.0)
cogs = rev * (1.0 - gm)
if opex_plan.get("method", "pct_revenue") == "amount":
opex = get_assumption(opex_plan.get("amount"), period, 0.0)
else:
opex = rev * get_assumption(opex_plan.get("pct_revenue"), period, 0.0)
gp = rev - cogs
ebitda = gp - opex
ebit = ebitda - da[i]
ebt = ebit - interest_values[i]
if ebt > 0:
nol_used = min(nol, ebt)
taxable_income = max(0.0, ebt - nol_used)
book_tax = ebt * book_tax_rate
cash_tax = taxable_income * cash_tax_rate
nol = max(0.0, nol - nol_used)
else:
nol_used = 0.0
book_tax = 0.0
cash_tax = 0.0
nol += abs(ebt)
net_income = ebt - book_tax
deferred_tax = book_tax - cash_tax
values = {
"revenue": rev,
"cogs": cogs,
"gross_profit": gp,
"opex": opex,
"ebitda": ebitda,
"da": da[i],
"ebit": ebit,
"interest": interest_values[i],
"ebt": ebt,
"book_taxes": book_tax,
"cash_taxes": cash_tax,
"net_income": net_income,
"deferred_tax": deferred_tax,
"nol_used": nol_used,
"ending_nol": nol,
}
for key, value in values.items():
out[key].append(value)
return out
def compute_ppe_and_da(plan: Dict[str, Any], periods: List[Period], revenue: List[float]) -> Dict[str, List[float]]:
ppe_plan = plan.get("ppe", {})
begin_ppe = float(coalesce(last_bs_value(plan, "ppe_net", 0.0), default=0.0) or 0.0)
out = {"beginning_ppe": [], "capex": [], "depreciation": [], "disposals": [], "ending_ppe": []}
for i, period in enumerate(periods):
capex_method = ppe_plan.get("capex_method", "pct_revenue")
if capex_method == "amount":
capex = get_assumption(ppe_plan.get("capex_amount"), period, 0.0)
else:
capex = revenue[i] * get_assumption(ppe_plan.get("capex_pct_revenue"), period, 0.0)
dep_method = ppe_plan.get("depreciation_method", "pct_beginning_ppe")
if dep_method == "pct_revenue":
depreciation = revenue[i] * get_assumption(ppe_plan.get("depreciation_pct_revenue"), period, 0.0)
elif dep_method == "amount":
depreciation = get_assumption(ppe_plan.get("depreciation_amount"), period, 0.0)
else:
annual_rate = get_assumption(ppe_plan.get("depreciation_pct_beginning_ppe"), period, 0.0)
depreciation = (begin_ppe + 0.5 * capex) * annual_rate * period.time_factor
disposals = get_assumption(ppe_plan.get("disposals"), period, 0.0)
ending_ppe = max(0.0, begin_ppe + capex - depreciation - disposals)
out["beginning_ppe"].append(begin_ppe)
out["capex"].append(capex)
out["depreciation"].append(depreciation)
out["disposals"].append(disposals)
out["ending_ppe"].append(ending_ppe)
begin_ppe = ending_ppe
return out
def compute_working_capital(plan: Dict[str, Any], periods: List[Period], revenue: List[float], cogs: List[float]) -> Dict[str, List[float]]:
wc_plan = plan.get("working_capital", {})
prev_nwc = last_nwc(plan)
out = {"ar_days": [], "inventory_days": [], "ap_days": [], "ar": [], "inventory": [], "other_current_assets": [], "ap": [], "accrued_expenses": [], "deferred_revenue": [], "nwc": [], "change_nwc": []}
for i, period in enumerate(periods):
annual_revenue = revenue[i] / period.time_factor if period.time_factor else revenue[i]
annual_cogs = cogs[i] / period.time_factor if period.time_factor else cogs[i]
ar_days = get_assumption(wc_plan.get("ar_days"), period, 0.0)
inv_days = get_assumption(wc_plan.get("inventory_days"), period, 0.0)
ap_days = get_assumption(wc_plan.get("ap_days"), period, 0.0)
ar = annual_revenue * ar_days / 365.0
inventory = annual_cogs * inv_days / 365.0
oca = annual_revenue * get_assumption(wc_plan.get("other_current_assets_pct_revenue"), period, 0.0)
ap = annual_cogs * ap_days / 365.0
accrued = annual_revenue * get_assumption(wc_plan.get("accrued_expenses_pct_revenue"), period, 0.0)
deferred = annual_revenue * get_assumption(wc_plan.get("deferred_revenue_pct_revenue"), period, 0.0)
nwc = ar + inventory + oca - ap - accrued - deferred
change_nwc = nwc - prev_nwc
values = {
"ar_days": ar_days,
"inventory_days": inv_days,
"ap_days": ap_days,
"ar": ar,
"inventory": inventory,
"other_current_assets": oca,
"ap": ap,
"accrued_expenses": accrued,
"deferred_revenue": deferred,
"nwc": nwc,
"change_nwc": change_nwc,
}
for key, value in values.items():
out[key].append(value)
prev_nwc = nwc
return out
def compute_debt_and_interest(plan: Dict[str, Any], periods: List[Period], integrated: Dict[str, List[float]]) -> Dict[str, List[float]]:
"""Return debt schedule and cash sweep using already computed operating cash flow inputs."""
debt_plan = plan.get("debt", {})
sweep_plan = debt_plan.get("cash_sweep", {})
beginning_debt = float(debt_plan.get("beginning_debt", last_bs_value(plan, "debt", 0.0)) or 0.0)
beginning_cash = last_bs_value(plan, "cash", 0.0)
beginning_revolver = float(debt_plan.get("beginning_revolver_drawn", 0.0) or 0.0)
revolver_commitment = float(debt_plan.get("revolver_commitment", 0.0) or 0.0)
min_cash = float(sweep_plan.get("min_cash", 0.0) or 0.0)
sweep_pct = float(sweep_plan.get("sweep_pct", 0.0) if sweep_plan.get("enabled", True) else 0.0)
out = {"beginning_cash": [], "beginning_debt": [], "beginning_revolver_drawn": [], "scheduled_draws": [], "required_draws": [], "total_draws": [], "mandatory_repayment": [], "optional_repayment": [], "total_repayments": [], "interest_rate": [], "interest": [], "ending_debt": [], "ending_revolver_drawn": [], "revolver_availability": [], "cash_before_sweep": [], "cash_change": [], "ending_cash": [], "minimum_cash": []}
beg_debt = beginning_debt
beg_cash = beginning_cash
beg_revolver = beginning_revolver
for i, period in enumerate(periods):
rate = get_assumption(debt_plan.get("interest_rate"), period, 0.0)
scheduled_draw = get_assumption(debt_plan.get("optional_draws"), period, 0.0)
available_before_draw = max(0.0, revolver_commitment - beg_revolver)
scheduled_draw = min(max(0.0, scheduled_draw), available_before_draw) if revolver_commitment > 0 else max(0.0, scheduled_draw)
mandatory_repay = min(max(0.0, get_assumption(debt_plan.get("mandatory_amortization"), period, 0.0)), beg_debt + scheduled_draw)
interest_base = max(0.0, beg_debt + 0.5 * scheduled_draw - 0.5 * mandatory_repay)
interest = interest_base * rate * period.time_factor
cfo = integrated["cash_flow_from_operations"][i]
capex = integrated["capex"][i]
dividends = integrated["dividends"][i]
buybacks = integrated["buybacks"][i]
issuance = integrated["issuance"][i]
cash_before_sweep = beg_cash + cfo - capex - dividends - buybacks + issuance + scheduled_draw - mandatory_repay
required_draw = 0.0
revolver_after_scheduled = beg_revolver + scheduled_draw
debt_after_scheduled = beg_debt + scheduled_draw - mandatory_repay
if cash_before_sweep < min_cash:
need = min_cash - cash_before_sweep
availability = max(0.0, revolver_commitment - revolver_after_scheduled)
required_draw = min(need, availability) if revolver_commitment > 0 else need
cash_before_sweep += required_draw
revolver_after_scheduled += required_draw
debt_after_scheduled += required_draw
optional_repay = 0.0
if sweep_pct > 0.0 and cash_before_sweep > min_cash and debt_after_scheduled > 0.0:
optional_repay = min((cash_before_sweep - min_cash) * sweep_pct, debt_after_scheduled)
ending_cash = cash_before_sweep - optional_repay
ending_debt = max(0.0, debt_after_scheduled - optional_repay)
revolver_repay = min(optional_repay, revolver_after_scheduled)
ending_revolver = max(0.0, revolver_after_scheduled - revolver_repay)
availability = max(0.0, revolver_commitment - ending_revolver) if revolver_commitment > 0 else 0.0
total_draws = scheduled_draw + required_draw
total_repayments = mandatory_repay + optional_repay
cash_change = ending_cash - beg_cash
values = {
"beginning_cash": beg_cash,
"beginning_debt": beg_debt,
"beginning_revolver_drawn": beg_revolver,
"scheduled_draws": scheduled_draw,
"required_draws": required_draw,
"total_draws": total_draws,
"mandatory_repayment": mandatory_repay,
"optional_repayment": optional_repay,
"total_repayments": total_repayments,
"interest_rate": rate,
"interest": interest,
"ending_debt": ending_debt,
"ending_revolver_drawn": ending_revolver,
"revolver_availability": availability,
"cash_before_sweep": cash_before_sweep,
"cash_change": cash_change,
"ending_cash": ending_cash,
"minimum_cash": min_cash,
}
for key, value in values.items():
out[key].append(value)
beg_cash = ending_cash
beg_debt = ending_debt
beg_revolver = ending_revolver
return out
def compute_cash_flow_statement(income_statement: Dict[str, List[float]], wc: Dict[str, List[float]], ppe: Dict[str, List[float]], debt: Dict[str, List[float]], equity_flows: Dict[str, List[float]]) -> Dict[str, List[float]]:
out = {"net_income": [], "da": [], "deferred_tax": [], "change_nwc": [], "cash_flow_from_operations": [], "capex": [], "cash_flow_from_investing": [], "debt_draws": [], "debt_repayments": [], "dividends": [], "buybacks": [], "issuance": [], "cash_flow_from_financing": [], "cash_change": [], "ending_cash": []}
n = len(income_statement["net_income"])
for i in range(n):
cfo = income_statement["net_income"][i] + income_statement["da"][i] + income_statement["deferred_tax"][i] - wc["change_nwc"][i]
cfi = -ppe["capex"][i]
cff = debt["total_draws"][i] - debt["total_repayments"][i] - equity_flows["dividends"][i] - equity_flows["buybacks"][i] + equity_flows["issuance"][i]
values = {
"net_income": income_statement["net_income"][i],
"da": income_statement["da"][i],
"deferred_tax": income_statement["deferred_tax"][i],
"change_nwc": wc["change_nwc"][i],
"cash_flow_from_operations": cfo,
"capex": ppe["capex"][i],
"cash_flow_from_investing": cfi,
"debt_draws": debt["total_draws"][i],
"debt_repayments": debt["total_repayments"][i],
"dividends": equity_flows["dividends"][i],
"buybacks": equity_flows["buybacks"][i],
"issuance": equity_flows["issuance"][i],
"cash_flow_from_financing": cff,
"cash_change": debt["cash_change"][i],
"ending_cash": debt["ending_cash"][i],
}
for key, value in values.items():
out[key].append(value)
return out
def compute_balance_sheet(plan: Dict[str, Any], periods: List[Period], income_statement: Dict[str, List[float]], wc: Dict[str, List[float]], ppe: Dict[str, List[float]], debt: Dict[str, List[float]], equity_flows: Dict[str, List[float]]) -> Dict[str, List[float]]:
out = {"cash": [], "ar": [], "inventory": [], "other_current_assets": [], "ppe_net": [], "other_assets": [], "total_assets": [], "ap": [], "accrued_expenses": [], "deferred_revenue": [], "debt": [], "other_liabilities": [], "common_equity": [], "retained_earnings": [], "total_liabilities_equity": [], "balance_check": []}
other_assets = float(plan.get("other_balance_sheet", {}).get("other_assets", last_bs_value(plan, "other_assets", 0.0)) or 0.0)
other_liabilities = float(plan.get("other_balance_sheet", {}).get("other_liabilities", last_bs_value(plan, "other_liabilities", 0.0)) or 0.0)
common_equity = float(plan.get("equity", {}).get("common_equity", last_bs_value(plan, "common_equity", 0.0)) or 0.0)
retained_earnings = last_bs_value(plan, "retained_earnings", 0.0)
cumulative_deferred_tax_liability = 0.0
for i, period in enumerate(periods):
common_equity += equity_flows["issuance"][i] - equity_flows["buybacks"][i]
retained_earnings += income_statement["net_income"][i] - equity_flows["dividends"][i]
cumulative_deferred_tax_liability += income_statement.get("deferred_tax", [0.0] * len(periods))[i]
current_other_liabilities = other_liabilities + cumulative_deferred_tax_liability
total_assets = debt["ending_cash"][i] + wc["ar"][i] + wc["inventory"][i] + wc["other_current_assets"][i] + ppe["ending_ppe"][i] + other_assets
total_liab_eq = wc["ap"][i] + wc["accrued_expenses"][i] + wc["deferred_revenue"][i] + debt["ending_debt"][i] + current_other_liabilities + common_equity + retained_earnings
values = {
"cash": debt["ending_cash"][i],
"ar": wc["ar"][i],
"inventory": wc["inventory"][i],
"other_current_assets": wc["other_current_assets"][i],
"ppe_net": ppe["ending_ppe"][i],
"other_assets": other_assets,
"total_assets": total_assets,
"ap": wc["ap"][i],
"accrued_expenses": wc["accrued_expenses"][i],
"deferred_revenue": wc["deferred_revenue"][i],
"debt": debt["ending_debt"][i],
"other_liabilities": current_other_liabilities,
"common_equity": common_equity,
"retained_earnings": retained_earnings,
"total_liabilities_equity": total_liab_eq,
"balance_check": total_assets - total_liab_eq,
}
for key, value in values.items():
out[key].append(value)
return out
def compute_covenants_or_liquidity(plan: Dict[str, Any], periods: List[Period], income_statement: Dict[str, List[float]], debt: Dict[str, List[float]]) -> Dict[str, List[float]]:
cov = plan.get("debt", {}).get("covenants", {})
out = {"net_debt": [], "liquidity": [], "net_leverage": [], "interest_coverage": [], "min_liquidity": [], "max_net_leverage": [], "min_interest_coverage": [], "liquidity_headroom": [], "net_leverage_headroom": [], "interest_coverage_headroom": [], "covenant_breach_flag": []}
for i, period in enumerate(periods):
cash = debt["ending_cash"][i]
ending_debt = debt["ending_debt"][i]
liquidity = cash + debt["revolver_availability"][i]
ebitda = income_statement["ebitda"][i]
interest = income_statement["interest"][i]
net_debt = ending_debt - cash
net_lev = net_debt / ebitda if ebitda > 0 else float("inf")
icr = ebitda / interest if interest > 0 else float("inf")
min_liq = float(cov.get("min_liquidity", 0.0) or 0.0)
max_lev = float(cov.get("max_net_leverage", 1e9) or 1e9)
min_icr = float(cov.get("min_interest_coverage", 0.0) or 0.0)
liquidity_headroom = liquidity - min_liq
net_lev_headroom = max_lev - net_lev if math.isfinite(net_lev) else -1e9
icr_headroom = icr - min_icr if math.isfinite(icr) else 1e9
breach = 1.0 if liquidity_headroom < -MODEL_TOLERANCE or net_lev_headroom < -0.01 or icr_headroom < -0.01 else 0.0
values = {
"net_debt": net_debt,
"liquidity": liquidity,
"net_leverage": net_lev,
"interest_coverage": icr,
"min_liquidity": min_liq,
"max_net_leverage": max_lev,
"min_interest_coverage": min_icr,
"liquidity_headroom": liquidity_headroom,
"net_leverage_headroom": net_lev_headroom,
"interest_coverage_headroom": icr_headroom,
"covenant_breach_flag": breach,
}
for key, value in values.items():
out[key].append(value)
return out
def equity_flows(plan: Dict[str, Any], periods: List[Period]) -> Dict[str, List[float]]:
eq = plan.get("equity", {})
out = {"dividends": [], "buybacks": [], "issuance": []}
for period in periods:
out["dividends"].append(get_assumption(eq.get("dividends"), period, 0.0))
out["buybacks"].append(get_assumption(eq.get("buybacks"), period, 0.0))
out["issuance"].append(get_assumption(eq.get("issuance"), period, 0.0))
return out
def build_operating_scaffold(plan: Dict[str, Any], periods: List[Period]) -> Tuple[Dict[str, Any], Dict[str, List[float]], Dict[str, List[float]], Dict[str, List[float]]]:
revenue_result = compute_revenue(plan, periods)
# First pass IS without D&A/interest gives COGS for WC and revenue for PP&E.
temp_is = compute_income_statement(plan, periods, revenue_result, ppe_result={"depreciation": [0.0] * len(periods)}, interest=[0.0] * len(periods))
ppe = compute_ppe_and_da(plan, periods, revenue_result["revenue"])
wc = compute_working_capital(plan, periods, revenue_result["revenue"], temp_is["cogs"])
return revenue_result, temp_is, ppe, wc
def run_integrated_model(plan: Dict[str, Any], scenario_name: str = "base") -> Dict[str, Any]:
timeline = plan.get("timeline", {})
periods = build_timeline(int(timeline["start_year"]), int(timeline["horizon_periods"]), timeline.get("periodicity", "annual"))
revenue_result, temp_is, ppe, wc = build_operating_scaffold(plan, periods)
eq = equity_flows(plan, periods)
# Resolve the mild interest/cash-sweep circularity by fixed-point iteration.
interest = [0.0] * len(periods)
integrated_for_debt = {
"cash_flow_from_operations": [0.0] * len(periods),
"capex": ppe["capex"],
"dividends": eq["dividends"],
"buybacks": eq["buybacks"],
"issuance": eq["issuance"],
}
debt = None
final_is = None
for _ in range(20):
final_is = compute_income_statement(plan, periods, revenue_result, ppe, interest=interest)
final_cfo = [final_is["net_income"][i] + final_is["da"][i] + final_is["deferred_tax"][i] - wc["change_nwc"][i] for i in range(len(periods))]
integrated_for_debt["cash_flow_from_operations"] = final_cfo
debt = compute_debt_and_interest(plan, periods, integrated_for_debt)
new_interest = debt["interest"]
if max(abs(new_interest[i] - interest[i]) for i in range(len(periods))) < 1e-8:
interest = new_interest
break
interest = new_interest
final_is = compute_income_statement(plan, periods, revenue_result, ppe, interest=interest)
final_cfo = [final_is["net_income"][i] + final_is["da"][i] + final_is["deferred_tax"][i] - wc["change_nwc"][i] for i in range(len(periods))]
integrated_for_debt["cash_flow_from_operations"] = final_cfo
debt = compute_debt_and_interest(plan, periods, integrated_for_debt)
final_is = compute_income_statement(plan, periods, revenue_result, ppe, interest=debt["interest"])
cf = compute_cash_flow_statement(final_is, wc, ppe, debt, eq)
bs = compute_balance_sheet(plan, periods, final_is, wc, ppe, debt, eq)
cov = compute_covenants_or_liquidity(plan, periods, final_is, debt)
result = {
"scenario": scenario_name,
"periods": periods,
"revenue_detail": revenue_result,
"income_statement": final_is,
"working_capital": wc,
"ppe": ppe,
"debt": debt,
"cash_flow_statement": cf,
"balance_sheet": bs,
"covenants_liquidity": cov,
"equity_flows": eq,
}
result["checks"] = compute_checks(plan, result)
return result
def apply_scenario(plan: Dict[str, Any], scenario_name: str) -> Dict[str, Any]:
scenario = plan.get("scenarios", {}).get(scenario_name, {})
return deep_merge(plan, scenario.get("overrides", {}))
def run_scenarios(plan: Dict[str, Any]) -> Dict[str, Dict[str, Any]]:
outputs: Dict[str, Dict[str, Any]] = {}
for scenario in ALLOWED_SCENARIOS:
scenario_plan = apply_scenario(plan, scenario)
outputs[scenario] = run_integrated_model(scenario_plan, scenario)
return outputs
def shock_growth_rates(plan: Dict[str, Any], delta: float) -> Dict[str, Any]:
out = copy.deepcopy(plan)
rev = out.get("revenue", {})
if rev.get("model") == "segments":
for cfg in rev.get("segments", {}).values():
rates = cfg.get("growth_rates", {})
for key in list(rates.keys()):
rates[key] = float(rates[key]) + delta
else:
rates = rev.setdefault("growth_rates", {})
for key in list(rates.keys()):
rates[key] = float(rates[key]) + delta
return out
def shock_map(plan: Dict[str, Any], path: List[str], delta: float) -> Dict[str, Any]:
out = copy.deepcopy(plan)
node = out
for key in path[:-1]:
node = node.setdefault(key, {})
leaf = node.get(path[-1], {})
if isinstance(leaf, dict):
for key in list(leaf.keys()):
if is_number(leaf[key]):
leaf[key] = float(leaf[key]) + delta
elif is_number(leaf):
node[path[-1]] = float(leaf) + delta
return out
def run_sensitivities(plan: Dict[str, Any]) -> List[Dict[str, Any]]:
sens = plan.get("sensitivities", {})
rows: List[Dict[str, Any]] = []
cases: List[Tuple[str, float, Dict[str, Any]]] = []
for delta in sens.get("revenue_growth_shocks", []):
cases.append(("revenue_growth", float(delta), shock_growth_rates(plan, float(delta))))
for delta in sens.get("gross_margin_shocks", []):
cases.append(("gross_margin", float(delta), shock_map(plan, ["costs", "cogs", "gross_margin"], float(delta))))
for delta in sens.get("dso_day_shocks", []):
cases.append(("dso_days", float(delta), shock_map(plan, ["working_capital", "ar_days"], float(delta))))
for delta in sens.get("capex_pct_revenue_shocks", []):
cases.append(("capex_pct_revenue", float(delta), shock_map(plan, ["ppe", "capex_pct_revenue"], float(delta))))
for delta in sens.get("interest_rate_shocks", []):
cases.append(("interest_rate", float(delta), shock_map(plan, ["debt", "interest_rate"], float(delta))))
for driver, delta, case_plan in cases:
result = run_integrated_model(case_plan, f"sensitivity_{driver}_{delta:+.4f}")
summary = summarize_result(result)
rows.append({
"case": f"{driver}_{delta:+.4f}",
"driver": driver,
"shock": delta,
"final_revenue": summary["final_revenue"],
"final_ebitda": summary["final_ebitda"],
"final_fcf": summary["final_fcf"],
"ending_cash": summary["ending_cash"],
"liquidity_trough": summary["liquidity_trough"],
"peak_net_leverage": summary["peak_net_leverage"],
})
return rows
def compute_checks(plan: Dict[str, Any], result: Dict[str, Any]) -> Dict[str, Any]:
periods = result["periods"]
bs = result["balance_sheet"]
cf = result["cash_flow_statement"]
debt = result["debt"]
ppe = result["ppe"]
wc = result["working_capital"]
is_ = result["income_statement"]
eq = result["equity_flows"]
checks: Dict[str, Any] = {
"period_checks": [],
"max_balance_sheet_abs_error": 0.0,
"max_cash_tie_abs_error": 0.0,
"max_retained_earnings_abs_error": 0.0,
"max_debt_rollforward_abs_error": 0.0,
"max_ppe_rollforward_abs_error": 0.0,
"max_nwc_rollforward_abs_error": 0.0,
}
prior_re = last_bs_value(plan, "retained_earnings", 0.0)
prior_nwc = last_nwc(plan)
for i, period in enumerate(periods):
balance_error = bs["balance_check"][i]
cash_tie = cf["ending_cash"][i] - bs["cash"][i]
expected_re = prior_re + is_["net_income"][i] - eq["dividends"][i]
re_error = expected_re - bs["retained_earnings"][i]
debt_error = debt["beginning_debt"][i] + debt["total_draws"][i] - debt["total_repayments"][i] - debt["ending_debt"][i]
ppe_error = ppe["beginning_ppe"][i] + ppe["capex"][i] - ppe["depreciation"][i] - ppe["disposals"][i] - ppe["ending_ppe"][i]
nwc_error = prior_nwc + wc["change_nwc"][i] - wc["nwc"][i]
checks["period_checks"].append({
"period": period.label,
"balance_sheet_error": balance_error,
"cash_tie_error": cash_tie,
"retained_earnings_error": re_error,
"debt_rollforward_error": debt_error,
"ppe_rollforward_error": ppe_error,
"nwc_rollforward_error": nwc_error,
})
checks["max_balance_sheet_abs_error"] = max(checks["max_balance_sheet_abs_error"], abs(balance_error))
checks["max_cash_tie_abs_error"] = max(checks["max_cash_tie_abs_error"], abs(cash_tie))
checks["max_retained_earnings_abs_error"] = max(checks["max_retained_earnings_abs_error"], abs(re_error))
checks["max_debt_rollforward_abs_error"] = max(checks["max_debt_rollforward_abs_error"], abs(debt_error))
checks["max_ppe_rollforward_abs_error"] = max(checks["max_ppe_rollforward_abs_error"], abs(ppe_error))
checks["max_nwc_rollforward_abs_error"] = max(checks["max_nwc_rollforward_abs_error"], abs(nwc_error))
prior_re = bs["retained_earnings"][i]
prior_nwc = wc["nwc"][i]
return checks
def summarize_result(result: Dict[str, Any]) -> Dict[str, float]:
is_ = result["income_statement"]
cf = result["cash_flow_statement"]
cov = result["covenants_liquidity"]
debt = result["debt"]
periods = result["periods"]
fcf = [cf["cash_flow_from_operations"][i] - cf["capex"][i] for i in range(len(periods))]
final_idx = len(periods) - 1
peak_net_leverage = max([x for x in cov["net_leverage"] if math.isfinite(x)] or [0.0])
return {
"final_revenue": is_["revenue"][final_idx],
"final_ebitda": is_["ebitda"][final_idx],
"final_ebitda_margin": is_["ebitda"][final_idx] / is_["revenue"][final_idx] if is_["revenue"][final_idx] else 0.0,
"final_fcf": fcf[final_idx],
"ending_cash": debt["ending_cash"][final_idx],
"ending_debt": debt["ending_debt"][final_idx],
"liquidity_trough": min(cov["liquidity"]),
"peak_net_leverage": peak_net_leverage,
"min_interest_coverage": min([x for x in cov["interest_coverage"] if math.isfinite(x)] or [999.0]),
"final_period": periods[final_idx].label,
}
def evaluate_hard_failures_and_warnings(plan: Dict[str, Any], scenario_outputs: Dict[str, Dict[str, Any]]) -> Tuple[List[Dict[str, str]], List[Dict[str, str]], Dict[str, Any]]:
hard_failures: List[Dict[str, str]] = []
warnings: List[Dict[str, str]] = []
checks_summary: Dict[str, Any] = {}
for scenario, result in scenario_outputs.items():
checks = result["checks"]
checks_summary[scenario] = {k: v for k, v in checks.items() if k != "period_checks"}
if checks["max_balance_sheet_abs_error"] > MODEL_TOLERANCE:
hard_failures.append({"code": "BALANCE_SHEET_DOES_NOT_BALANCE", "message": f"{scenario}: balance sheet error exceeds tolerance."})
if checks["max_cash_tie_abs_error"] > MODEL_TOLERANCE:
hard_failures.append({"code": "CASH_DOES_NOT_TIE", "message": f"{scenario}: ending cash does not tie to cash flow statement."})
if checks["max_retained_earnings_abs_error"] > MODEL_TOLERANCE:
hard_failures.append({"code": "RETAINED_EARNINGS_ROLLFORWARD_FAIL", "message": f"{scenario}: retained earnings roll-forward fails."})
if checks["max_debt_rollforward_abs_error"] > MODEL_TOLERANCE:
hard_failures.append({"code": "DEBT_ROLLFORWARD_FAIL", "message": f"{scenario}: debt roll-forward fails."})
if checks["max_ppe_rollforward_abs_error"] > MODEL_TOLERANCE:
hard_failures.append({"code": "PPE_ROLLFORWARD_FAIL", "message": f"{scenario}: PP&E roll-forward fails."})
if checks["max_nwc_rollforward_abs_error"] > MODEL_TOLERANCE:
hard_failures.append({"code": "NWC_ROLLFORWARD_FAIL", "message": f"{scenario}: working capital roll-forward fails."})
cov = result["covenants_liquidity"]
if max(cov["covenant_breach_flag"] or [0.0]) > 0:
warnings.append({"code": "LIQUIDITY_OR_COVENANT_ISSUE", "message": f"{scenario}: liquidity or covenant headroom issue appears in forecast."})
if min(result["debt"]["ending_cash"] or [0.0]) < -MODEL_TOLERANCE:
warnings.append({"code": "NEGATIVE_CASH", "message": f"{scenario}: ending cash falls below zero."})
base = scenario_outputs.get("base")
downside = scenario_outputs.get("downside")
upside = scenario_outputs.get("upside")
if base and downside and upside:
b = summarize_result(base)
d = summarize_result(downside)
u = summarize_result(upside)
if abs(b["final_revenue"] - d["final_revenue"]) < MODEL_TOLERANCE and abs(b["final_revenue"] - u["final_revenue"]) < MODEL_TOLERANCE:
hard_failures.append({"code": "SCENARIO_SWITCH_NO_OUTPUT_CHANGE", "message": "scenario switch changes labels but not material model outputs."})
source_basis = plan.get("source_basis", [])
if not source_basis:
hard_failures.append({"code": "SOURCE_BASIS_MISSING", "message": "source_basis is missing for material historicals or forecast drivers."})
else:
covers = {cover for src in source_basis for cover in src.get("covers", [])}
if "historicals" not in covers:
hard_failures.append({"code": "HISTORICAL_SOURCE_MISSING", "message": "source_basis does not cover historical financials."})
if not ({"revenue", "forecast", "costs"} & covers):
hard_failures.append({"code": "FORECAST_SOURCE_MISSING", "message": "source_basis does not cover material forecast drivers."})
if any(src.get("evidence_label") == "placeholder" for src in source_basis):
warnings.append({"code": "PLACEHOLDER_ASSUMPTIONS_ACTIVE", "message": "placeholder source basis remains active; model should be treated as screen-grade only."})
# Senior judgment warnings.
hist_is = last_hist_is(plan)
hist_revenue = float(hist_is.get("revenue", 0.0) or 0.0)
if base and hist_revenue > 0:
first_revenue = base["income_statement"]["revenue"][0]
final_revenue = base["income_statement"]["revenue"][-1]
periods_count = max(1, len(base["periods"]))
annualized_cagr = (final_revenue / hist_revenue) ** (1.0 / periods_count) - 1.0 if final_revenue > 0 else -1.0
if annualized_cagr > 0.20:
warnings.append({"code": "AGGRESSIVE_REVENUE_RAMP", "message": "base case revenue CAGR exceeds 20%; verify capacity, market share, sales productivity, and demand evidence."})
if first_revenue > hist_revenue * 1.25:
warnings.append({"code": "STEP_UP_REVENUE", "message": "first forecast period revenue steps up more than 25% from latest historical period."})
if base:
gm_first = base["income_statement"]["gross_profit"][0] / base["income_statement"]["revenue"][0]
gm_last = base["income_statement"]["gross_profit"][-1] / base["income_statement"]["revenue"][-1]
if gm_last - gm_first > 0.05:
warnings.append({"code": "UNSUPPORTED_MARGIN_EXPANSION", "message": "gross margin expands by more than 500 bps; verify mix, pricing, utilization, input costs, and sourcing evidence."})
capex = base["ppe"]["capex"][-1]
da = base["ppe"]["depreciation"][-1]
rev_growth = (base["income_statement"]["revenue"][-1] / base["income_statement"]["revenue"][0] - 1.0) if base["income_statement"]["revenue"][0] else 0.0
if rev_growth > 0.10 and capex < da * 0.75:
warnings.append({"code": "CAPEX_TOO_LOW_FOR_GROWTH", "message": "capex is below 75% of D&A despite forecast revenue growth; verify maintenance and growth capex."})
hist_nwc_ratio = last_nwc(plan) / hist_revenue if hist_revenue else 0.0
forecast_nwc_ratio = base["working_capital"]["nwc"][-1] / base["income_statement"]["revenue"][-1] if base["income_statement"]["revenue"][-1] else 0.0
if hist_nwc_ratio - forecast_nwc_ratio > 0.05:
warnings.append({"code": "WORKING_CAPITAL_RELEASE_INCONSISTENT_WITH_HISTORY", "message": "forecast working capital intensity improves by more than 500 bps vs. history; verify DSO/DIO/DPO assumptions."})
debt = plan.get("debt", {})
cash_sweep = debt.get("cash_sweep", {})
covenants = debt.get("covenants", {})
def has_nonzero_value(value: Any) -> bool:
if isinstance(value, dict):
return any(has_nonzero_value(item) for item in value.values())
if isinstance(value, list):
return any(has_nonzero_value(item) for item in value)
return is_number(value) and abs(float(value)) > MODEL_TOLERANCE
def has_validated_source(cover: str) -> bool:
return any(
cover in src.get("covers", [])
and src.get("evidence_label") not in {"placeholder", "assumption"}
and src.get("confidence") != "low"
for src in source_basis
)
modeled_debt = has_nonzero_value({
"beginning_debt": debt.get("beginning_debt"),
"beginning_revolver_drawn": debt.get("beginning_revolver_drawn"),
"revolver_commitment": debt.get("revolver_commitment"),
"mandatory_amortization": debt.get("mandatory_amortization"),
"optional_draws": debt.get("optional_draws"),
})
modeled_liquidity = has_nonzero_value({
"beginning_revolver_drawn": debt.get("beginning_revolver_drawn"),
"revolver_commitment": debt.get("revolver_commitment"),
"minimum_cash": cash_sweep.get("min_cash"),
"cash_sweep_pct": cash_sweep.get("sweep_pct") if cash_sweep.get("enabled", True) else 0.0,
})
debt_support_missing = modeled_debt and not has_validated_source("debt")
liquidity_support_missing = modeled_liquidity and not (
has_validated_source("debt") or has_validated_source("liquidity")
)
covenant_support_missing = bool(covenants) and not has_validated_source("covenants")
if debt_support_missing or liquidity_support_missing or covenant_support_missing:
warnings.append({
"code": "LIQUIDITY_COVENANT_SUPPORT_UNVALIDATED",
"message": "debt draws, cash sweeps and covenant capacity are illustrative until debt, liquidity and any modeled covenant support is provided.",
})
return hard_failures, warnings, checks_summary
def model_status(hard_failures: List[Dict[str, str]], warnings: List[Dict[str, str]], plan: Dict[str, Any]) -> str:
if hard_failures:
return "not-decision-ready"
if any(
w.get("code") in {"PLACEHOLDER_ASSUMPTIONS_ACTIVE", "LIQUIDITY_COVENANT_SUPPORT_UNVALIDATED"}
for w in warnings
):
return "screen-grade"
if warnings:
return "senior-review-ready"
labels = {src.get("evidence_label") for src in plan.get("source_basis", [])}
if labels and labels <= {"source_reported", "connector_sourced", "public_filing", "web_verified", "management_guidance", "company_provided"}:
return "decision-grade"
return "senior-review-ready"
def fmt(value: Any) -> str:
if value is None:
return ""
if isinstance(value, float):
if math.isinf(value):
return "n/m"
return f"{value:,.1f}"
return str(value)
def to_model_rows(plan: Dict[str, Any], result: Dict[str, Any]) -> List[Dict[str, Any]]:
rows: List[Dict[str, Any]] = []
scenario = result["scenario"]
periods = result["periods"]
unit = plan.get("meta", {}).get("units", "")
def add(statement: str, section: str, line_item: str, series: List[float], formula_basis: str, evidence_label: str = "model_calculated", source_id: str = "", notes: str = "") -> None:
for period, value in zip(periods, series):
rows.append({
"scenario": scenario,
"statement": statement,
"section": section,
"line_item": line_item,
"period": period.label,
"value": value,
"unit": unit,
"evidence_label": evidence_label,
"source_id": source_id,
"formula_basis": formula_basis,
"notes": notes,
})
is_ = result["income_statement"]
bs = result["balance_sheet"]
cf = result["cash_flow_statement"]
debt = result["debt"]
wc = result["working_capital"]
ppe = result["ppe"]
cov = result["covenants_liquidity"]
add("IS", "Revenue", "Revenue", is_["revenue"], "driver-based revenue forecast", result["revenue_detail"].get("evidence_label", "model_calculated"), result["revenue_detail"].get("source_id", ""))
add("IS", "COGS", "COGS", is_["cogs"], "revenue x (1 - gross margin)")
add("IS", "Gross Profit", "Gross Profit", is_["gross_profit"], "revenue - COGS")
add("IS", "Opex", "Operating Expense", is_["opex"], "revenue x opex percent or amount")
add("IS", "Profitability", "EBITDA", is_["ebitda"], "gross profit - opex")
add("IS", "Profitability", "D&A", is_["da"], "from PP&E schedule")
add("IS", "Profitability", "EBIT", is_["ebit"], "EBITDA - D&A")
add("IS", "Financing", "Interest Expense", is_["interest"], "average debt x interest rate")
add("IS", "Taxes", "EBT", is_["ebt"], "EBIT - interest")
add("IS", "Taxes", "Book Taxes", is_["book_taxes"], "positive EBT x book tax rate")
add("IS", "Taxes", "Cash Taxes", is_["cash_taxes"], "taxable income after NOL x cash tax rate")
add("IS", "Net Income", "Net Income", is_["net_income"], "EBT - book taxes")
for line in ["cash", "ar", "inventory", "other_current_assets", "ppe_net", "other_assets", "total_assets", "ap", "accrued_expenses", "deferred_revenue", "debt", "other_liabilities", "common_equity", "retained_earnings", "total_liabilities_equity", "balance_check"]:
add("BS", "Balance Sheet", line.replace("_", " ").title(), bs[line], "linked balance sheet schedule")
for line in ["net_income", "da", "deferred_tax", "change_nwc", "cash_flow_from_operations", "capex", "cash_flow_from_investing", "debt_draws", "debt_repayments", "dividends", "buybacks", "issuance", "cash_flow_from_financing", "cash_change", "ending_cash"]:
add("CF", "Cash Flow", line.replace("_", " ").title(), cf[line], "cash flow statement roll-forward")
for line in ["beginning_debt", "scheduled_draws", "required_draws", "total_draws", "mandatory_repayment", "optional_repayment", "total_repayments", "interest_rate", "interest", "ending_debt", "ending_revolver_drawn", "revolver_availability", "minimum_cash"]:
add("DEBT", "Debt", line.replace("_", " ").title(), debt[line], "debt roll-forward and cash sweep")
for line in ["ar_days", "inventory_days", "ap_days", "ar", "inventory", "other_current_assets", "ap", "accrued_expenses", "deferred_revenue", "nwc", "change_nwc"]:
add("WORKING_CAPITAL", "Working Capital", line.replace("_", " ").title(), wc[line], "days/percent-driven working capital")
for line in ["beginning_ppe", "capex", "depreciation", "disposals", "ending_ppe"]:
add("PPE", "PP&E", line.replace("_", " ").title(), ppe[line], "PP&E roll-forward")
for line in ["net_debt", "liquidity", "net_leverage", "interest_coverage", "liquidity_headroom", "net_leverage_headroom", "interest_coverage_headroom", "covenant_breach_flag"]:
add("COVENANTS_LIQUIDITY", "Liquidity", line.replace("_", " ").title(), cov[line], "liquidity and covenant metrics")
for pc in result["checks"]["period_checks"]:
for key, value in pc.items():
if key == "period":
continue
rows.append({
"scenario": scenario,
"statement": "CHECKS",
"section": "QA",
"line_item": key.replace("_", " ").title(),
"period": pc["period"],
"value": value,
"unit": unit,
"evidence_label": "model_calculated",
"source_id": "",
"formula_basis": "machine-computed tie-out check",
"notes": "hard failure if above tolerance",
})
return rows
def assumption_rows(plan: Dict[str, Any]) -> List[Dict[str, Any]]:
rows: List[Dict[str, Any]] = []
def walk(prefix: str, node: Any, source_id: str = "", evidence_label: str = "") -> None:
if isinstance(node, dict):
sid = node.get("source_id", source_id)
ev = node.get("evidence_label", evidence_label)
for key, value in node.items():
if key in {"source_id", "evidence_label", "description", "notes"}:
continue
walk(f"{prefix}.{key}" if prefix else key, value, sid, ev)
elif isinstance(node, list):
rows.append({"assumption_path": prefix, "period_or_key": "list", "value": json.dumps(node), "evidence_label": evidence_label, "source_id": source_id, "notes": "list assumption"})
else:
bits = prefix.split(".")
period_or_key = bits[-1] if bits else ""
category = ".".join(bits[:-1]) if len(bits) > 1 else prefix
rows.append({"assumption_path": category, "period_or_key": period_or_key, "value": node, "evidence_label": evidence_label, "source_id": source_id, "notes": ""})
for top in ["revenue", "costs", "working_capital", "ppe", "debt", "tax", "equity", "scenarios", "sensitivities"]:
if top in plan:
walk(top, plan[top])
return rows
def summary_rows(plan: Dict[str, Any], scenario_outputs: Dict[str, Dict[str, Any]]) -> List[Dict[str, Any]]:
rows: List[Dict[str, Any]] = []
for scenario, result in scenario_outputs.items():
s = summarize_result(result)
for key, value in s.items():
rows.append({"scenario": scenario, "metric": key, "value": value, "unit": plan.get("meta", {}).get("units", ""), "notes": "scenario summary"})
return rows
def check_rows(scenario_outputs: Dict[str, Dict[str, Any]]) -> List[Dict[str, Any]]:
rows: List[Dict[str, Any]] = []
for scenario, result in scenario_outputs.items():
for key, value in result["checks"].items():
if key == "period_checks":
continue
rows.append({"scenario": scenario, "check": key, "value": value, "pass": abs(float(value)) <= MODEL_TOLERANCE, "tolerance": MODEL_TOLERANCE})
return rows
def source_rows(plan: Dict[str, Any]) -> List[Dict[str, Any]]:
return [dict(src) for src in plan.get("source_basis", [])]
def p0_handoff(plan: Dict[str, Any], scenario_outputs: Dict[str, Dict[str, Any]], hard_failures: List[Dict[str, str]], warnings: List[Dict[str, str]], status: str, output_dir: Path, include_report_md: bool = True) -> Dict[str, Any]:
scenarios = {name: summarize_result(result) for name, result in scenario_outputs.items()}
base = scenarios.get("base", {})
all_liquidity = [summary.get("liquidity_trough", 0.0) for summary in scenarios.values()]
paths = {
"workbook": str(output_dir / "model.xlsx"),
"plan": str(output_dir / "plan.json"),
"run_log": str(output_dir / "run_log.json"),
}
if include_report_md:
paths["report"] = str(output_dir / "report.md")
return {
"operating_forecast_summary": {
"company_name": plan.get("meta", {}).get("company_name"),
"industry": plan.get("meta", {}).get("industry"),
"currency": plan.get("meta", {}).get("currency"),
"units": plan.get("meta", {}).get("units"),
"final_period": base.get("final_period"),
},
"scenario_outputs": scenarios,
"base_downside_upside_revenue_ebitda_fcf_cash": {
name: {
"final_revenue": s.get("final_revenue"),
"final_ebitda": s.get("final_ebitda"),
"final_fcf": s.get("final_fcf"),
"ending_cash": s.get("ending_cash"),
}
for name, s in scenarios.items()
},
"liquidity_trough": min(all_liquidity) if all_liquidity else None,
"key_operating_drivers": ["revenue growth", "gross margin", "opex leverage", "DSO/DIO/DPO", "capex intensity", "interest rate", "cash sweep"],
"checks_passed_failed": {
"hard_failure_count": len(hard_failures),
"warning_count": len(warnings),
},
"model_status": status,
"paths": paths,
}
def _xlsx_col_ref(col_1idx: int) -> str:
s = ""
col = col_1idx
while col:
col, rem = divmod(col - 1, 26)
s = chr(65 + rem) + s
return s
def _xlsx_cell_ref(row_1idx: int, col_1idx: int) -> str:
return f"{_xlsx_col_ref(col_1idx)}{row_1idx}"
def _xlsx_shared_string(text: str, shared_map: Dict[str, int], shared_list: List[str]) -> int:
if text not in shared_map:
shared_map[text] = len(shared_list)
shared_list.append(text)
return shared_map[text]
def sanitize_sheet_name(name: str) -> str:
cleaned = re.sub(r"[][\\/*?:]", "_", str(name))[:31]
return cleaned or "Sheet"
def write_xlsx(path: Path, sheets: Dict[str, List[Dict[str, Any]]], sheet_name: str = "Model") -> None:
"""Write a minimal multi-sheet .xlsx workbook using only stdlib zip/xml."""
path.parent.mkdir(parents=True, exist_ok=True)
if not sheets:
raise ValueError("No sheets to write")
if isinstance(sheets, list): # compatibility with single-sheet callers
sheets = {sheet_name: sheets} # type: ignore[assignment]
shared_map: Dict[str, int] = {}
shared_list: List[str] = []
sheet_xmls: List[Tuple[str, str]] = []
def emit_cell(row_1idx: int, col_1idx: int, value: Any) -> str:
if value is None or value == "":
return ""
ref = _xlsx_cell_ref(row_1idx, col_1idx)
if isinstance(value, bool):
return f'<c r="{ref}" t="b"><v>{1 if value else 0}</v></c>'
if is_number(value):
return f'<c r="{ref}"><v>{float(value)}</v></c>'
text = str(value)
idx = _xlsx_shared_string(text, shared_map, shared_list)
return f'<c r="{ref}" t="s"><v>{idx}</v></c>'
used_names: set[str] = set()
for raw_name, rows in sheets.items():
name = sanitize_sheet_name(raw_name)
base_name = name
suffix = 1
while name in used_names:
suffix += 1
name = sanitize_sheet_name(f"{base_name[:28]}{suffix}")
used_names.add(name)
if rows:
headers = list(rows[0].keys())
else:
headers = ["message"]
rows = [{"message": "no rows"}]
sheet_rows: List[str] = []
values = headers
cells = "".join(emit_cell(1, c, v) for c, v in enumerate(values, start=1))
sheet_rows.append(f'<row r="1">{cells}</row>')
for row_num, row in enumerate(rows, start=2):
cells = "".join(emit_cell(row_num, c, row.get(h, "")) for c, h in enumerate(headers, start=1))
sheet_rows.append(f'<row r="{row_num}">{cells}</row>')
dim = f"A1:{_xlsx_cell_ref(len(rows) + 1, len(headers))}"
sheet_xml = (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<worksheet xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main" '
'xmlns:r="http://schemas.openxmlformats.org/officeDocument/2006/relationships">'
f'<dimension ref="{dim}"/>'
'<sheetViews><sheetView workbookViewId="0"><pane ySplit="1" topLeftCell="A2" activePane="bottomLeft" state="frozen"/></sheetView></sheetViews>'
'<sheetData>' + "".join(sheet_rows) + '</sheetData>'
'</worksheet>'
)
sheet_xmls.append((name, sheet_xml))
def si(text: str) -> str:
needs_preserve = text[:1].isspace() or text[-1:].isspace()
attr = ' xml:space="preserve"' if needs_preserve else ""
return f"<si><t{attr}>{escape(text)}</t></si>"
sst_xml = (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<sst xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main" '
f'count="{len(shared_list)}" uniqueCount="{len(shared_list)}">'
+ "".join(si(s) for s in shared_list)
+ '</sst>'
)
styles_xml = '''<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<styleSheet xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main">
<fonts count="1"><font><sz val="11"/><color theme="1"/><name val="Calibri"/><family val="2"/></font></fonts>
<fills count="2"><fill><patternFill patternType="none"/></fill><fill><patternFill patternType="gray125"/></fill></fills>
<borders count="1"><border><left/><right/><top/><bottom/><diagonal/></border></borders>
<cellStyleXfs count="1"><xf numFmtId="0" fontId="0" fillId="0" borderId="0"/></cellStyleXfs>
<cellXfs count="1"><xf numFmtId="0" fontId="0" fillId="0" borderId="0" xfId="0"/></cellXfs>
<cellStyles count="1"><cellStyle name="Normal" xfId="0" builtinId="0"/></cellStyles>
</styleSheet>
'''
workbook_sheets = []
rels = []
content_overrides = []
for idx, (name, _) in enumerate(sheet_xmls, start=1):
workbook_sheets.append(f'<sheet name="{escape(name)}" sheetId="{idx}" r:id="rId{idx}"/>')
rels.append(f'<Relationship Id="rId{idx}" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/worksheet" Target="worksheets/sheet{idx}.xml"/>')
content_overrides.append(f'<Override PartName="/xl/worksheets/sheet{idx}.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.worksheet+xml"/>')
style_rid = len(sheet_xmls) + 1
sst_rid = len(sheet_xmls) + 2
rels.append(f'<Relationship Id="rId{style_rid}" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/styles" Target="styles.xml"/>')
rels.append(f'<Relationship Id="rId{sst_rid}" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/sharedStrings" Target="sharedStrings.xml"/>')
workbook_xml = (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<workbook xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main" '
'xmlns:r="http://schemas.openxmlformats.org/officeDocument/2006/relationships"><sheets>'
+ "".join(workbook_sheets)
+ '</sheets></workbook>'
)
workbook_rels_xml = (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
+ "".join(rels)
+ '</Relationships>'
)
root_rels_xml = '''<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">
<Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/officeDocument" Target="xl/workbook.xml"/>
</Relationships>
'''
content_types_xml = (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types">'
'<Default Extension="rels" ContentType="application/vnd.openxmlformats-package.relationships+xml"/>'
'<Default Extension="xml" ContentType="application/xml"/>'
'<Override PartName="/xl/workbook.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet.main+xml"/>'
+ "".join(content_overrides)
+ '<Override PartName="/xl/styles.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.styles+xml"/>'
+ '<Override PartName="/xl/sharedStrings.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.sharedStrings+xml"/>'
+ '</Types>'
)
with zipfile.ZipFile(path, "w", compression=zipfile.ZIP_DEFLATED) as z:
z.writestr("[Content_Types].xml", content_types_xml)
z.writestr("_rels/.rels", root_rels_xml)
z.writestr("xl/workbook.xml", workbook_xml)
z.writestr("xl/_rels/workbook.xml.rels", workbook_rels_xml)
for idx, (_, sheet_xml) in enumerate(sheet_xmls, start=1):
z.writestr(f"xl/worksheets/sheet{idx}.xml", sheet_xml)
z.writestr("xl/sharedStrings.xml", sst_xml)
z.writestr("xl/styles.xml", styles_xml)
SHA-256: f314f25d50fffb298c124857d4d45f704bd8337ee56bbc4aa7d1db6b0b8ce9f5