← Files Investment BankingARCHIVED FILE
skills/dcf-model-builder/scripts/skill_core_runtime
80.1 KB · Oct 2, 2026 · 00:27 UTC
#!/usr/bin/env python3
"""Deterministic DCF model engine for dcf-model-builder.
Standard library only. No network calls. No hidden randomness.
"""
from __future__ import annotations
import copy
import json
import math
import os
import zipfile
from datetime import datetime
from pathlib import Path
from typing import Any
from xml.sax.saxutils import escape
ALLOWED_LABELS = {
"reported",
"company_guidance",
"consensus",
"management_case",
"user_provided",
"connected_app",
"web_research",
"analyst_estimate",
"placeholder",
"derived",
}
REQUIRED_SOURCE_TOPICS = {
"historicals",
"forecast",
"wacc",
"terminal_value",
"share_count",
"net_debt",
}
MODEL_STATUS_VALUES = {
"decision-grade",
"senior-review-ready",
"screen-grade",
"not-decision-ready",
"blocked",
}
REQUIRED_TOP_LEVEL = [
"meta",
"source_basis",
"timeline",
"historicals",
"forecast",
"wacc",
"terminal_value",
"ev_to_equity_bridge",
"scenarios",
"sensitivities",
]
SCENARIOS = ["base", "downside", "upside"]
def load_json(path: str | os.PathLike[str]) -> dict[str, Any]:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
def write_json(path: str | os.PathLike[str], obj: Any) -> None:
Path(path).parent.mkdir(parents=True, exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
json.dump(obj, f, indent=2, sort_keys=False)
f.write("\n")
def _is_number(value: Any) -> bool:
return (
isinstance(value, (int, float))
and not isinstance(value, bool)
and math.isfinite(float(value))
)
def _date_ok(value: Any) -> bool:
if not isinstance(value, str):
return False
try:
datetime.strptime(value, "%Y-%m-%d")
return True
except ValueError:
return False
def _date(value: str) -> datetime | None:
try:
return datetime.strptime(value, "%Y-%m-%d")
except Exception:
return None
def _num(value: Any, default: float = 0.0) -> float:
if _is_number(value):
return float(value)
return float(default)
def _get(plan: dict[str, Any], dotted: str, default: Any = None) -> Any:
node: Any = plan
for part in dotted.split("."):
if not isinstance(node, dict) or part not in node:
return default
node = node[part]
return node
def _source_map(plan: dict[str, Any]) -> dict[str, dict[str, Any]]:
return {str(s.get("id")): s for s in plan.get("source_basis", []) if isinstance(s, dict)}
def _source_label(plan: dict[str, Any], source_id: str | None) -> str:
if not source_id:
return ""
src = _source_map(plan).get(source_id)
if not src:
return ""
return str(src.get("label", ""))
def _source_for_topic(plan: dict[str, Any], topic: str) -> dict[str, Any] | None:
for src in plan.get("source_basis", []):
if isinstance(src, dict) and src.get("topic") == topic:
return src
return None
def _vector(value: Any, horizon: int, field_name: str) -> list[float]:
if isinstance(value, list):
return [float(v) for v in value]
if _is_number(value):
return [float(value)] * horizon
raise ValueError(f"{field_name} must be a number or list of numbers")
def _validate_vector(
errors: list[str],
value: Any,
path: str,
horizon: int,
min_value: float | None = None,
max_value: float | None = None,
allow_scalar: bool = True,
) -> None:
if isinstance(value, list):
if len(value) != horizon:
errors.append(f"{path} must contain exactly {horizon} values; found {len(value)}")
return
values = value
elif allow_scalar and _is_number(value):
values = [value]
else:
errors.append(f"{path} must be a number or list of numbers")
return
for idx, item in enumerate(values):
loc = f"{path}[{idx}]" if isinstance(value, list) else path
if not _is_number(item):
errors.append(f"{loc} must be numeric")
continue
f = float(item)
if min_value is not None and f < min_value:
errors.append(f"{loc} must be >= {min_value}")
if max_value is not None and f > max_value:
errors.append(f"{loc} must be <= {max_value}")
def validate_plan_structure(plan: dict[str, Any]) -> list[str]:
"""Return actionable validation errors. Does not compute the model."""
errors: list[str] = []
if not isinstance(plan, dict):
return ["plan must be a JSON object"]
for key in REQUIRED_TOP_LEVEL:
if key not in plan:
errors.append(f"missing top-level field: {key}")
if errors:
return errors
meta = plan.get("meta", {})
for field in [
"company",
"industry",
"currency",
"units",
"valuation_date",
"as_of_date",
"accounting_basis",
"valuation_purpose",
"model_type",
]:
if not meta.get(field):
errors.append(f"meta.{field} is required")
if meta.get("model_type") not in {"fcff", "fcfe"}:
errors.append("meta.model_type must be 'fcff' or 'fcfe'")
for field in ["valuation_date", "as_of_date"]:
if meta.get(field) and not _date_ok(meta[field]):
errors.append(f"meta.{field} must be YYYY-MM-DD")
source_basis = plan.get("source_basis")
if not isinstance(source_basis, list) or not source_basis:
errors.append("source_basis must be a non-empty list")
else:
seen_ids = set()
topics = set()
for idx, src in enumerate(source_basis):
if not isinstance(src, dict):
errors.append(f"source_basis[{idx}] must be an object")
continue
for field in [
"id",
"topic",
"label",
"source_name",
"source_type",
"as_of_date",
"confidence",
"notes",
]:
if field not in src or src.get(field) in (None, ""):
errors.append(f"source_basis[{idx}].{field} is required")
sid = str(src.get("id", ""))
if sid:
if sid in seen_ids:
errors.append(f"source_basis id '{sid}' is duplicated")
seen_ids.add(sid)
if src.get("topic"):
topics.add(str(src.get("topic")))
if src.get("label") and src.get("label") not in ALLOWED_LABELS:
errors.append(f"source_basis[{idx}].label '{src.get('label')}' is not allowed")
if src.get("confidence") and src.get("confidence") not in {"high", "medium", "low"}:
errors.append(f"source_basis[{idx}].confidence must be high, medium, or low")
if src.get("as_of_date") and not _date_ok(src["as_of_date"]):
errors.append(f"source_basis[{idx}].as_of_date must be YYYY-MM-DD")
missing_topics = sorted(REQUIRED_SOURCE_TOPICS - topics)
if missing_topics:
errors.append(
"source_basis missing required material topics: " + ", ".join(missing_topics)
)
timeline = plan.get("timeline", {})
horizon = timeline.get("horizon_years")
if not isinstance(timeline.get("start_year"), int):
errors.append("timeline.start_year must be an integer")
if not isinstance(horizon, int) or horizon < 1 or horizon > 15:
errors.append("timeline.horizon_years must be an integer from 1 to 15")
horizon = 0
if timeline.get("periodicity") not in {"annual", "quarterly"}:
errors.append("timeline.periodicity must be 'annual' or 'quarterly'")
historicals = plan.get("historicals", {})
hist_required = [
"latest_year",
"revenue",
"ebitda",
"ebit",
"cash_taxes",
"da",
"capex",
"change_nwc",
"net_working_capital",
"unlevered_fcf",
"source_id",
]
for field in hist_required:
if field not in historicals:
errors.append(f"historicals.{field} is required")
for field in [
"revenue",
"ebitda",
"ebit",
"cash_taxes",
"da",
"capex",
"change_nwc",
"net_working_capital",
"unlevered_fcf",
]:
if field in historicals and not _is_number(historicals[field]):
errors.append(f"historicals.{field} must be numeric")
if (
"revenue" in historicals
and _is_number(historicals.get("revenue"))
and historicals["revenue"] <= 0
):
errors.append("historicals.revenue must be greater than zero")
if meta.get("model_type") == "fcfe":
if "net_income" not in historicals:
errors.append("historicals.net_income is required for FCFE models")
source_ids = (
{str(s.get("id")) for s in source_basis if isinstance(s, dict)}
if isinstance(source_basis, list)
else set()
)
for path in [
"historicals.source_id",
"forecast.source_id",
"wacc.source_id",
"terminal_value.source_id",
]:
sid = _get(plan, path)
if sid and str(sid) not in source_ids:
errors.append(f"{path} references unknown source id '{sid}'")
forecast = plan.get("forecast", {})
if forecast.get("cash_flow_basis") not in {"fcff", "fcfe"}:
errors.append("forecast.cash_flow_basis must be 'fcff' or 'fcfe'")
if (
forecast.get("cash_flow_basis")
and meta.get("model_type")
and forecast.get("cash_flow_basis") != meta.get("model_type")
):
errors.append("forecast.cash_flow_basis must match meta.model_type")
if not isinstance(forecast.get("mid_year_convention"), bool):
errors.append("forecast.mid_year_convention must be true or false")
wacc = plan.get("wacc", {})
wacc_required = [
"risk_free_rate",
"beta",
"equity_risk_premium",
"size_premium",
"pre_tax_cost_of_debt",
"marginal_tax_rate",
"target_debt_pct",
"target_equity_pct",
"source_id",
]
for field in wacc_required:
if field not in wacc:
errors.append(f"wacc.{field} is required")
range_checks = {
"risk_free_rate": (-0.02, 0.20),
"beta": (0.0, 5.0),
"equity_risk_premium": (0.0, 0.20),
"size_premium": (0.0, 0.15),
"company_specific_premium": (-0.05, 0.25),
"country_risk_premium": (0.0, 0.30),
"pre_tax_cost_of_debt": (0.0, 0.40),
"marginal_tax_rate": (0.0, 0.60),
"target_debt_pct": (0.0, 1.0),
"target_equity_pct": (0.0, 1.0),
}
for field, (low, high) in range_checks.items():
if field in wacc:
if not _is_number(wacc[field]):
errors.append(f"wacc.{field} must be numeric")
else:
value = float(wacc[field])
if value < low or value > high:
errors.append(f"wacc.{field} must be between {low} and {high}")
if _is_number(wacc.get("target_debt_pct")) and _is_number(wacc.get("target_equity_pct")):
preferred = _num(wacc.get("preferred_pct"), 0.0)
total_weight = float(wacc["target_debt_pct"]) + float(wacc["target_equity_pct"]) + preferred
if abs(total_weight - 1.0) > 0.02:
errors.append(
f"wacc capital structure weights must sum to approximately 1.0; found {total_weight:.3f}"
)
terminal = plan.get("terminal_value", {})
method = terminal.get("method")
if method not in {"perpetual_growth", "exit_multiple"}:
errors.append("terminal_value.method must be 'perpetual_growth' or 'exit_multiple'")
if method == "perpetual_growth":
if not _is_number(terminal.get("perpetual_growth_rate")):
errors.append(
"terminal_value.perpetual_growth_rate is required and numeric for perpetual growth"
)
else:
g = float(terminal["perpetual_growth_rate"])
if g < -0.05 or g > 0.08:
errors.append("terminal_value.perpetual_growth_rate must be between -5% and 8%")
if method == "exit_multiple":
if not _is_number(terminal.get("exit_ebitda_multiple")):
errors.append(
"terminal_value.exit_ebitda_multiple is required and numeric for exit multiple"
)
elif float(terminal["exit_ebitda_multiple"]) <= 0:
errors.append("terminal_value.exit_ebitda_multiple must be positive")
if "exit_ebitda_multiple" in terminal and _is_number(terminal.get("exit_ebitda_multiple")):
if (
float(terminal["exit_ebitda_multiple"]) <= 0
or float(terminal["exit_ebitda_multiple"]) > 100
):
errors.append(
"terminal_value.exit_ebitda_multiple must be positive and less than or equal to 100x"
)
bridge = plan.get("ev_to_equity_bridge", {})
bridge_required = [
"cash",
"debt",
"leases",
"minorities",
"associates",
"pensions",
"preferred_stock",
"non_operating_assets",
"options",
"other_debt_like_items",
"diluted_shares",
"net_debt_source_id",
"share_count_source_id",
]
for field in bridge_required:
if field not in bridge:
errors.append(f"ev_to_equity_bridge.{field} is required")
for field in bridge_required:
if field.endswith("source_id"):
continue
if field in bridge and not _is_number(bridge[field]):
errors.append(f"ev_to_equity_bridge.{field} must be numeric")
if _is_number(bridge.get("diluted_shares")) and float(bridge["diluted_shares"]) <= 0:
errors.append("ev_to_equity_bridge.diluted_shares must be greater than zero")
for path in [
"ev_to_equity_bridge.net_debt_source_id",
"ev_to_equity_bridge.share_count_source_id",
]:
sid = _get(plan, path)
if sid and str(sid) not in source_ids:
errors.append(f"{path} references unknown source id '{sid}'")
scenarios = plan.get("scenarios", {})
if not isinstance(scenarios, dict):
errors.append("scenarios must be an object")
else:
for name in SCENARIOS:
if name not in scenarios:
errors.append(f"scenarios.{name} is required")
continue
scenario = scenarios[name]
if not isinstance(scenario, dict):
errors.append(f"scenarios.{name} must be an object")
continue
if not scenario.get("description"):
errors.append(f"scenarios.{name}.description is required")
if horizon:
_validate_vector(
errors,
scenario.get("revenue_growth"),
f"scenarios.{name}.revenue_growth",
horizon,
-0.80,
2.00,
)
_validate_vector(
errors,
scenario.get("ebit_margin"),
f"scenarios.{name}.ebit_margin",
horizon,
-0.50,
0.80,
)
_validate_vector(
errors,
scenario.get("tax_rate"),
f"scenarios.{name}.tax_rate",
horizon,
0.0,
0.60,
)
_validate_vector(
errors,
scenario.get("da_percent_revenue"),
f"scenarios.{name}.da_percent_revenue",
horizon,
0.0,
1.00,
)
_validate_vector(
errors,
scenario.get("capex_percent_revenue"),
f"scenarios.{name}.capex_percent_revenue",
horizon,
0.0,
1.50,
)
_validate_vector(
errors,
scenario.get("nwc_percent_revenue"),
f"scenarios.{name}.nwc_percent_revenue",
horizon,
-1.0,
1.0,
)
if forecast.get("cash_flow_basis") == "fcfe":
_validate_vector(
errors,
scenario.get("net_income_margin"),
f"scenarios.{name}.net_income_margin",
horizon,
-0.50,
0.80,
)
_validate_vector(
errors,
scenario.get("net_borrowing", 0.0),
f"scenarios.{name}.net_borrowing",
horizon,
-1e9,
1e9,
)
if not _is_number(scenario.get("wacc_adjustment")):
errors.append(f"scenarios.{name}.wacc_adjustment must be numeric")
if method == "perpetual_growth":
if not _is_number(scenario.get("terminal_growth_rate")):
errors.append(f"scenarios.{name}.terminal_growth_rate is required and numeric")
else:
tg = float(scenario["terminal_growth_rate"])
if tg < -0.05 or tg > 0.08:
errors.append(
f"scenarios.{name}.terminal_growth_rate must be between -5% and 8%"
)
if scenario.get("exit_ebitda_multiple") is not None and not _is_number(
scenario.get("exit_ebitda_multiple")
):
errors.append(
f"scenarios.{name}.exit_ebitda_multiple must be numeric when supplied"
)
sensitivities = plan.get("sensitivities", {})
for field in [
"wacc_delta",
"terminal_growth_delta",
"exit_multiple_delta",
"revenue_growth_delta",
"ebit_margin_delta",
]:
value = sensitivities.get(field)
if not isinstance(value, list) or not value:
errors.append(f"sensitivities.{field} must be a non-empty list")
continue
if 0.0 not in [float(v) for v in value if _is_number(v)]:
errors.append(f"sensitivities.{field} should include 0.0")
for idx, item in enumerate(value):
if not _is_number(item):
errors.append(f"sensitivities.{field}[{idx}] must be numeric")
if not errors:
try:
wacc_info = compute_wacc(plan, "base")
discount_rate = (
wacc_info["wacc"]
if plan["forecast"].get("cash_flow_basis") == "fcff"
else wacc_info["cost_of_equity"]
)
if discount_rate <= 0:
errors.append("computed discount rate must be greater than zero")
terminal_growth = _num(
plan["scenarios"]["base"].get(
"terminal_growth_rate", plan["terminal_value"].get("perpetual_growth_rate")
),
0.0,
)
if (
plan["terminal_value"].get("method") == "perpetual_growth"
and discount_rate <= terminal_growth
):
errors.append("computed discount rate must exceed terminal growth rate")
except Exception as exc:
errors.append(f"could not compute WACC/terminal validation: {exc}")
return errors
def normalize_plan(
plan: dict[str, Any], skill_root: str | os.PathLike[str] | None = None
) -> dict[str, Any]:
"""Return a normalized copy without changing conclusion-driving assumptions."""
normalized = copy.deepcopy(plan)
notes: list[str] = []
wacc = normalized.setdefault("wacc", {})
for field in [
"company_specific_premium",
"country_risk_premium",
"preferred_pct",
"pre_tax_cost_of_preferred",
]:
if field not in wacc:
wacc[field] = 0.0
notes.append(f"wacc.{field} was absent and was set to 0.0 for calculation transparency")
bridge = normalized.setdefault("ev_to_equity_bridge", {})
for field in [
"cash",
"debt",
"leases",
"minorities",
"associates",
"pensions",
"preferred_stock",
"non_operating_assets",
"options",
"other_debt_like_items",
]:
if field not in bridge:
bridge[field] = 0.0
notes.append(f"ev_to_equity_bridge.{field} was absent and was set to 0.0")
normalized["_normalization_notes"] = notes
if skill_root is not None:
normalized["_skill_root"] = str(skill_root)
return normalized
def build_timeline(plan: dict[str, Any]) -> list[str]:
start = int(plan["timeline"]["start_year"])
horizon = int(plan["timeline"]["horizon_years"])
periodicity = plan["timeline"].get("periodicity", "annual")
if periodicity == "annual":
return [str(start + i) for i in range(1, horizon + 1)]
# Quarterly support uses Q1... labels from start year.
labels: list[str] = []
year = start
quarter = 1
for _ in range(horizon):
labels.append(f"{year}Q{quarter}")
quarter += 1
if quarter > 4:
year += 1
quarter = 1
return labels
def compute_wacc(
plan: dict[str, Any], scenario_name: str | dict[str, Any] = "base"
) -> dict[str, float]:
wacc = plan["wacc"]
if isinstance(scenario_name, dict):
scenario = scenario_name
else:
scenario = plan.get("scenarios", {}).get(str(scenario_name), {})
risk_free_rate = float(wacc["risk_free_rate"])
beta = float(wacc["beta"])
erp = float(wacc["equity_risk_premium"])
size = float(wacc.get("size_premium", 0.0))
country = float(wacc.get("country_risk_premium", 0.0))
company_specific = float(wacc.get("company_specific_premium", 0.0))
cost_of_equity = risk_free_rate + beta * erp + size + country + company_specific
marginal_tax_rate = float(wacc["marginal_tax_rate"])
pre_tax_cost_of_debt = float(wacc["pre_tax_cost_of_debt"])
after_tax_cost_of_debt = pre_tax_cost_of_debt * (1.0 - marginal_tax_rate)
pre_tax_cost_of_preferred = float(wacc.get("pre_tax_cost_of_preferred", 0.0))
preferred_pct = float(wacc.get("preferred_pct", 0.0))
debt_pct = float(wacc["target_debt_pct"])
equity_pct = float(wacc["target_equity_pct"])
base_wacc = (
equity_pct * cost_of_equity
+ debt_pct * after_tax_cost_of_debt
+ preferred_pct * pre_tax_cost_of_preferred
)
scenario_adjustment = float(scenario.get("wacc_adjustment", 0.0))
final_wacc = base_wacc + scenario_adjustment
return {
"risk_free_rate": risk_free_rate,
"beta": beta,
"equity_risk_premium": erp,
"size_premium": size,
"country_risk_premium": country,
"company_specific_premium": company_specific,
"cost_of_equity": cost_of_equity,
"pre_tax_cost_of_debt": pre_tax_cost_of_debt,
"after_tax_cost_of_debt": after_tax_cost_of_debt,
"marginal_tax_rate": marginal_tax_rate,
"target_debt_pct": debt_pct,
"target_equity_pct": equity_pct,
"preferred_pct": preferred_pct,
"base_wacc": base_wacc,
"scenario_wacc_adjustment": scenario_adjustment,
"wacc": final_wacc,
}
def compute_unlevered_fcf(
plan: dict[str, Any], scenario_name: str = "base"
) -> list[dict[str, float | str]]:
scenario = plan["scenarios"][scenario_name]
horizon = int(plan["timeline"]["horizon_years"])
periods = build_timeline(plan)
revenue_growth = _vector(scenario["revenue_growth"], horizon, "revenue_growth")
ebit_margin = _vector(scenario["ebit_margin"], horizon, "ebit_margin")
tax_rate = _vector(scenario["tax_rate"], horizon, "tax_rate")
da_pct = _vector(scenario["da_percent_revenue"], horizon, "da_percent_revenue")
capex_pct = _vector(scenario["capex_percent_revenue"], horizon, "capex_percent_revenue")
nwc_pct = _vector(scenario["nwc_percent_revenue"], horizon, "nwc_percent_revenue")
previous_revenue = float(plan["historicals"]["revenue"])
previous_nwc = float(
plan["historicals"].get("net_working_capital", previous_revenue * nwc_pct[0])
)
rows: list[dict[str, float | str]] = []
for i, period in enumerate(periods):
revenue = previous_revenue * (1.0 + revenue_growth[i])
ebit = revenue * ebit_margin[i]
taxes = max(ebit * tax_rate[i], 0.0)
nopat = ebit - taxes
da = revenue * da_pct[i]
capex = revenue * capex_pct[i]
nwc = revenue * nwc_pct[i]
change_nwc = nwc - previous_nwc
unlevered_fcf = nopat + da - capex - change_nwc
ebitda = ebit + da
rows.append(
{
"period": period,
"revenue": revenue,
"revenue_growth": revenue_growth[i],
"ebitda": ebitda,
"ebit": ebit,
"ebit_margin": ebit_margin[i],
"tax_rate": tax_rate[i],
"cash_taxes": taxes,
"nopat": nopat,
"da": da,
"capex": capex,
"nwc": nwc,
"change_nwc": change_nwc,
"unlevered_fcf": unlevered_fcf,
"cash_flow": unlevered_fcf,
}
)
previous_revenue = revenue
previous_nwc = nwc
return rows
def compute_equity_cash_flows(
plan: dict[str, Any], scenario_name: str = "base"
) -> list[dict[str, float | str]]:
scenario = plan["scenarios"][scenario_name]
horizon = int(plan["timeline"]["horizon_years"])
periods = build_timeline(plan)
revenue_growth = _vector(scenario["revenue_growth"], horizon, "revenue_growth")
ebit_margin = _vector(scenario["ebit_margin"], horizon, "ebit_margin")
net_income_margin = _vector(scenario["net_income_margin"], horizon, "net_income_margin")
tax_rate = _vector(scenario["tax_rate"], horizon, "tax_rate")
da_pct = _vector(scenario["da_percent_revenue"], horizon, "da_percent_revenue")
capex_pct = _vector(scenario["capex_percent_revenue"], horizon, "capex_percent_revenue")
nwc_pct = _vector(scenario["nwc_percent_revenue"], horizon, "nwc_percent_revenue")
net_borrowing = _vector(scenario.get("net_borrowing", 0.0), horizon, "net_borrowing")
previous_revenue = float(plan["historicals"]["revenue"])
previous_nwc = float(
plan["historicals"].get("net_working_capital", previous_revenue * nwc_pct[0])
)
rows: list[dict[str, float | str]] = []
for i, period in enumerate(periods):
revenue = previous_revenue * (1.0 + revenue_growth[i])
ebit = revenue * ebit_margin[i]
net_income = revenue * net_income_margin[i]
da = revenue * da_pct[i]
capex = revenue * capex_pct[i]
nwc = revenue * nwc_pct[i]
change_nwc = nwc - previous_nwc
fcfe = net_income + da - capex - change_nwc + net_borrowing[i]
ebitda = ebit + da
rows.append(
{
"period": period,
"revenue": revenue,
"revenue_growth": revenue_growth[i],
"ebitda": ebitda,
"ebit": ebit,
"ebit_margin": ebit_margin[i],
"net_income": net_income,
"net_income_margin": net_income_margin[i],
"tax_rate": tax_rate[i],
"da": da,
"capex": capex,
"nwc": nwc,
"change_nwc": change_nwc,
"net_borrowing": net_borrowing[i],
"fcfe": fcfe,
"cash_flow": fcfe,
}
)
previous_revenue = revenue
previous_nwc = nwc
return rows
def compute_terminal_value(
plan: dict[str, Any],
forecast_rows: list[dict[str, Any]],
discount_rate: float,
scenario_name: str = "base",
method_override: str | None = None,
) -> dict[str, float | str]:
if not forecast_rows:
raise ValueError("forecast_rows cannot be empty")
scenario = plan["scenarios"][scenario_name]
terminal = plan["terminal_value"]
method = method_override or terminal.get("method", "perpetual_growth")
final = forecast_rows[-1]
final_fcf = float(final["cash_flow"])
final_ebitda = float(final.get("ebitda", 0.0))
if method == "perpetual_growth":
g = float(scenario.get("terminal_growth_rate", terminal.get("perpetual_growth_rate", 0.0)))
if discount_rate <= g:
raise ValueError(
f"discount rate {discount_rate:.4f} must exceed terminal growth {g:.4f}"
)
terminal_fcf = final_fcf * (1.0 + g)
terminal_value = terminal_fcf / (discount_rate - g)
implied_exit_multiple = terminal_value / final_ebitda if final_ebitda else None
implied_fcf_yield = terminal_fcf / terminal_value if terminal_value else None
return {
"method": method,
"terminal_growth_rate": g,
"terminal_fcf": terminal_fcf,
"terminal_value": terminal_value,
"exit_ebitda_multiple": float(terminal.get("exit_ebitda_multiple", 0.0)),
"implied_exit_ebitda_multiple": implied_exit_multiple
if implied_exit_multiple is not None
else 0.0,
"implied_fcf_yield": implied_fcf_yield if implied_fcf_yield is not None else 0.0,
}
if method == "exit_multiple":
multiple = float(scenario.get("exit_ebitda_multiple", terminal.get("exit_ebitda_multiple")))
if multiple <= 0:
raise ValueError("exit EBITDA multiple must be positive")
terminal_value = final_ebitda * multiple
implied_fcf_yield = final_fcf / terminal_value if terminal_value else None
implied_growth_rate = None
# Rearranged Gordon growth is approximate because final FCF is used rather than next-year FCF.
if terminal_value and final_fcf:
implied_growth_rate = (discount_rate * terminal_value - final_fcf) / (
terminal_value + final_fcf
)
return {
"method": method,
"terminal_growth_rate": float(
scenario.get("terminal_growth_rate", terminal.get("perpetual_growth_rate", 0.0))
),
"terminal_fcf": final_fcf,
"terminal_value": terminal_value,
"exit_ebitda_multiple": multiple,
"implied_exit_ebitda_multiple": multiple,
"implied_fcf_yield": implied_fcf_yield if implied_fcf_yield is not None else 0.0,
"implied_growth_rate": implied_growth_rate if implied_growth_rate is not None else 0.0,
}
raise ValueError(f"unsupported terminal value method: {method}")
def discount_cash_flows(
forecast_rows: list[dict[str, Any]],
terminal_value: float,
discount_rate: float,
mid_year_convention: bool = True,
) -> dict[str, Any]:
if discount_rate <= -1.0:
raise ValueError("discount rate must be greater than -100%")
pv_rows: list[dict[str, Any]] = []
for i, row in enumerate(forecast_rows, start=1):
period_power = i - 0.5 if mid_year_convention else i
pv = float(row["cash_flow"]) / ((1.0 + discount_rate) ** period_power)
pv_rows.append(
{
"period": row["period"],
"cash_flow": row["cash_flow"],
"discount_period": period_power,
"pv_cash_flow": pv,
}
)
n = len(forecast_rows)
pv_terminal = terminal_value / ((1.0 + discount_rate) ** n)
pv_fcf = sum(float(r["pv_cash_flow"]) for r in pv_rows)
return {
"pv_rows": pv_rows,
"pv_fcf": pv_fcf,
"pv_terminal_value": pv_terminal,
"total_pv": pv_fcf + pv_terminal,
}
def compute_enterprise_value(pv_fcf: float, pv_terminal_value: float) -> float:
return float(pv_fcf) + float(pv_terminal_value)
def compute_equity_value(plan: dict[str, Any], enterprise_value: float) -> dict[str, float]:
bridge = plan["ev_to_equity_bridge"]
add_backs = (
float(bridge.get("cash", 0.0))
+ float(bridge.get("non_operating_assets", 0.0))
+ float(bridge.get("associates", 0.0))
)
deductions = (
float(bridge.get("debt", 0.0))
+ float(bridge.get("leases", 0.0))
+ float(bridge.get("minorities", 0.0))
+ float(bridge.get("pensions", 0.0))
+ float(bridge.get("preferred_stock", 0.0))
+ float(bridge.get("options", 0.0))
+ float(bridge.get("other_debt_like_items", 0.0))
)
equity_value = enterprise_value + add_backs - deductions
return {"add_backs": add_backs, "deductions": deductions, "equity_value": equity_value}
def compute_enterprise_value_from_equity(plan: dict[str, Any], equity_value: float) -> float:
bridge = plan["ev_to_equity_bridge"]
add_backs = (
float(bridge.get("cash", 0.0))
+ float(bridge.get("non_operating_assets", 0.0))
+ float(bridge.get("associates", 0.0))
)
deductions = (
float(bridge.get("debt", 0.0))
+ float(bridge.get("leases", 0.0))
+ float(bridge.get("minorities", 0.0))
+ float(bridge.get("pensions", 0.0))
+ float(bridge.get("preferred_stock", 0.0))
+ float(bridge.get("options", 0.0))
+ float(bridge.get("other_debt_like_items", 0.0))
)
return equity_value - add_backs + deductions
def compute_value_per_share(plan: dict[str, Any], equity_value: float) -> float:
shares = float(plan["ev_to_equity_bridge"]["diluted_shares"])
if shares <= 0:
raise ValueError("diluted shares must be greater than zero")
return equity_value / shares
def run_scenario(
plan: dict[str, Any], scenario_name: str = "base", terminal_method_override: str | None = None
) -> dict[str, Any]:
cash_flow_basis = plan["forecast"].get(
"cash_flow_basis", plan["meta"].get("model_type", "fcff")
)
wacc_info = compute_wacc(plan, scenario_name)
discount_rate = (
wacc_info["wacc"]
if cash_flow_basis == "fcff"
else wacc_info["cost_of_equity"]
+ float(plan["scenarios"][scenario_name].get("wacc_adjustment", 0.0))
)
forecast_rows = (
compute_unlevered_fcf(plan, scenario_name)
if cash_flow_basis == "fcff"
else compute_equity_cash_flows(plan, scenario_name)
)
terminal_info = compute_terminal_value(
plan, forecast_rows, discount_rate, scenario_name, method_override=terminal_method_override
)
pv_info = discount_cash_flows(
forecast_rows,
float(terminal_info["terminal_value"]),
discount_rate,
bool(plan["forecast"].get("mid_year_convention", True)),
)
if cash_flow_basis == "fcff":
enterprise_value = compute_enterprise_value(
float(pv_info["pv_fcf"]), float(pv_info["pv_terminal_value"])
)
equity_bridge = compute_equity_value(plan, enterprise_value)
equity_value = equity_bridge["equity_value"]
else:
equity_value = compute_enterprise_value(
float(pv_info["pv_fcf"]), float(pv_info["pv_terminal_value"])
)
enterprise_value = compute_enterprise_value_from_equity(plan, equity_value)
equity_bridge = {"add_backs": 0.0, "deductions": 0.0, "equity_value": equity_value}
value_per_share = compute_value_per_share(plan, equity_value)
tv_percent_ev = (
float(pv_info["pv_terminal_value"]) / enterprise_value if enterprise_value else 0.0
)
return {
"scenario": scenario_name,
"cash_flow_basis": cash_flow_basis,
"forecast_rows": forecast_rows,
"wacc": wacc_info,
"discount_rate": discount_rate,
"terminal": terminal_info,
"pv": pv_info,
"enterprise_value": enterprise_value,
"equity_bridge": equity_bridge,
"equity_value": equity_value,
"value_per_share": value_per_share,
"tv_percent_ev": tv_percent_ev,
}
def _adjust_vector(
value: Any, horizon: int, delta: float, floor: float | None = None, cap: float | None = None
) -> list[float]:
vals = _vector(value, horizon, "adjust_vector")
out = []
for v in vals:
new_v = v + delta
if floor is not None:
new_v = max(floor, new_v)
if cap is not None:
new_v = min(cap, new_v)
out.append(new_v)
return out
def _safe_run_value_per_share(
plan: dict[str, Any], method: str | None = None
) -> tuple[float | None, str | None]:
try:
result = run_scenario(plan, "base", terminal_method_override=method)
value = float(result["value_per_share"])
if not math.isfinite(value):
return None, "non-finite value per share"
return value, None
except Exception as exc:
return None, str(exc)
def run_sensitivities(plan: dict[str, Any]) -> dict[str, Any]:
rows: list[dict[str, Any]] = []
horizon = int(plan["timeline"]["horizon_years"])
base_scenario = plan["scenarios"]["base"]
base_tg = float(
base_scenario.get(
"terminal_growth_rate", plan["terminal_value"].get("perpetual_growth_rate", 0.0)
)
)
base_exit = float(
base_scenario.get(
"exit_ebitda_multiple", plan["terminal_value"].get("exit_ebitda_multiple", 0.0)
)
)
sens = plan.get("sensitivities", {})
for w_delta in sens.get("wacc_delta", [0.0]):
for tg_delta in sens.get("terminal_growth_delta", [0.0]):
p = copy.deepcopy(plan)
p["terminal_value"]["method"] = "perpetual_growth"
p["scenarios"]["base"]["wacc_adjustment"] = float(
base_scenario.get("wacc_adjustment", 0.0)
) + float(w_delta)
p["scenarios"]["base"]["terminal_growth_rate"] = base_tg + float(tg_delta)
value, error = _safe_run_value_per_share(p, method="perpetual_growth")
rows.append(
{
"sensitivity": "WACC / Terminal Growth",
"x_axis": "wacc_delta",
"x_value": float(w_delta),
"y_axis": "terminal_growth_delta",
"y_value": float(tg_delta),
"metric": "value_per_share",
"value": value,
"units": plan["meta"].get("currency", "") + "/share",
"notes": error or "",
}
)
if base_exit > 0:
for w_delta in sens.get("wacc_delta", [0.0]):
for mult_delta in sens.get("exit_multiple_delta", [0.0]):
p = copy.deepcopy(plan)
p["terminal_value"]["method"] = "exit_multiple"
p["terminal_value"]["exit_ebitda_multiple"] = max(
0.1, base_exit + float(mult_delta)
)
p["scenarios"]["base"]["exit_ebitda_multiple"] = max(
0.1, base_exit + float(mult_delta)
)
p["scenarios"]["base"]["wacc_adjustment"] = float(
base_scenario.get("wacc_adjustment", 0.0)
) + float(w_delta)
value, error = _safe_run_value_per_share(p, method="exit_multiple")
rows.append(
{
"sensitivity": "WACC / Exit Multiple",
"x_axis": "wacc_delta",
"x_value": float(w_delta),
"y_axis": "exit_multiple_delta",
"y_value": float(mult_delta),
"metric": "value_per_share",
"value": value,
"units": plan["meta"].get("currency", "") + "/share",
"notes": error or "",
}
)
for rev_delta in sens.get("revenue_growth_delta", [0.0]):
for margin_delta in sens.get("ebit_margin_delta", [0.0]):
p = copy.deepcopy(plan)
p["scenarios"]["base"]["revenue_growth"] = _adjust_vector(
base_scenario["revenue_growth"], horizon, float(rev_delta), floor=-0.9
)
p["scenarios"]["base"]["ebit_margin"] = _adjust_vector(
base_scenario["ebit_margin"], horizon, float(margin_delta), floor=-0.8, cap=0.95
)
value, error = _safe_run_value_per_share(p, method=plan["terminal_value"].get("method"))
rows.append(
{
"sensitivity": "Revenue Growth / EBIT Margin",
"x_axis": "revenue_growth_delta",
"x_value": float(rev_delta),
"y_axis": "ebit_margin_delta",
"y_value": float(margin_delta),
"metric": "value_per_share",
"value": value,
"units": plan["meta"].get("currency", "") + "/share",
"notes": error or "",
}
)
directionality = compute_sensitivity_directionality(rows)
return {"rows": rows, "directionality": directionality}
def _value_at(
rows: list[dict[str, Any]], sensitivity: str, x_value: float, y_value: float
) -> float | None:
candidates = [
r
for r in rows
if r.get("sensitivity") == sensitivity
and abs(float(r.get("x_value", 999.0)) - x_value) < 1e-9
and abs(float(r.get("y_value", 999.0)) - y_value) < 1e-9
and r.get("value") is not None
]
if not candidates:
return None
return float(candidates[0]["value"])
def compute_sensitivity_directionality(rows: list[dict[str, Any]]) -> dict[str, Any]:
checks: dict[str, Any] = {"passed": True, "details": []}
def add_check(name: str, passed: bool, detail: str) -> None:
checks["details"].append({"check": name, "passed": bool(passed), "detail": detail})
if not passed:
checks["passed"] = False
def check_axis(
sensitivity: str, axis: str, fixed_axis_zero: str, should_increase_with_axis: bool
) -> None:
subset = [
r for r in rows if r.get("sensitivity") == sensitivity and r.get("value") is not None
]
if not subset:
add_check(f"{sensitivity} {axis}", False, "no successful rows")
return
if axis == "x":
fixed_rows = [r for r in subset if abs(float(r.get("y_value", 0.0))) < 1e-9]
key = "x_value"
else:
fixed_rows = [r for r in subset if abs(float(r.get("x_value", 0.0))) < 1e-9]
key = "y_value"
if len(fixed_rows) < 2:
add_check(f"{sensitivity} {axis}", False, "not enough zero-axis rows")
return
low = min(fixed_rows, key=lambda r: float(r[key]))
high = max(fixed_rows, key=lambda r: float(r[key]))
low_v = float(low["value"])
high_v = float(high["value"])
if should_increase_with_axis:
passed = high_v >= low_v
detail = f"value at high axis {high_v:.2f} vs low axis {low_v:.2f}"
else:
passed = high_v <= low_v
detail = f"value at high axis {high_v:.2f} vs low axis {low_v:.2f}"
add_check(f"{sensitivity} {axis} direction", passed, detail)
check_axis(
"WACC / Terminal Growth", "x", "terminal_growth_delta", should_increase_with_axis=False
)
check_axis("WACC / Terminal Growth", "y", "wacc_delta", should_increase_with_axis=True)
check_axis("WACC / Exit Multiple", "x", "exit_multiple_delta", should_increase_with_axis=False)
check_axis("WACC / Exit Multiple", "y", "wacc_delta", should_increase_with_axis=True)
check_axis(
"Revenue Growth / EBIT Margin", "x", "ebit_margin_delta", should_increase_with_axis=True
)
check_axis(
"Revenue Growth / EBIT Margin", "y", "revenue_growth_delta", should_increase_with_axis=True
)
return checks
def _source_warnings(plan: dict[str, Any]) -> list[str]:
warnings: list[str] = []
labels = [
str(src.get("label")) for src in plan.get("source_basis", []) if isinstance(src, dict)
]
if "placeholder" in labels:
warnings.append(
"One or more material source entries use placeholder evidence labels; output is screen-grade at best."
)
weak_topics = [
str(src.get("topic"))
for src in plan.get("source_basis", [])
if isinstance(src, dict)
and src.get("label") in {"analyst_estimate", "placeholder"}
and src.get("topic") in {"forecast", "wacc", "terminal_value"}
]
if weak_topics:
warnings.append(
"Material valuation inputs rely on analyst estimates/placeholders: "
+ ", ".join(sorted(set(weak_topics)))
)
valuation_date = _date(str(plan.get("meta", {}).get("valuation_date", "")))
if valuation_date:
dated = []
for src in plan.get("source_basis", []):
if not isinstance(src, dict):
continue
d = _date(str(src.get("as_of_date", "")))
if d:
dated.append((src.get("topic"), d))
age = abs((valuation_date - d).days)
if (
src.get("topic") in {"wacc", "share_count", "net_debt", "market_data"}
and age > 120
):
warnings.append(
f"Source for {src.get('topic')} is {age} days from valuation date; refresh market/bridge data for decision-grade use."
)
if dated:
min_date = min(d for _, d in dated)
max_date = max(d for _, d in dated)
spread = (max_date - min_date).days
if spread > 365:
warnings.append(
f"Source dates span {spread} days; confirm market data, historicals, and share count are contemporaneous."
)
return warnings
def compute_checks(
plan: dict[str, Any],
scenario_results: dict[str, dict[str, Any]],
sensitivity_result: dict[str, Any],
) -> dict[str, Any]:
hard_failures: list[str] = []
warnings: list[str] = []
informational: list[str] = []
validation_errors = validate_plan_structure(plan)
if validation_errors:
hard_failures.extend(validation_errors)
required_topics = REQUIRED_SOURCE_TOPICS
topics = {
str(src.get("topic")) for src in plan.get("source_basis", []) if isinstance(src, dict)
}
missing_topics = sorted(required_topics - topics)
if missing_topics:
hard_failures.append(
"source basis missing for material valuation inputs: " + ", ".join(missing_topics)
)
base = scenario_results.get("base")
if not base:
hard_failures.append("base scenario did not run")
else:
forecast_rows = base.get("forecast_rows", [])
fcfs = [row.get("cash_flow") for row in forecast_rows]
if not fcfs or all(not _is_number(v) for v in fcfs):
hard_failures.append("no forecast FCF/FCFE was produced")
wacc_value = float(base.get("discount_rate", 0.0))
if wacc_value <= 0 or not math.isfinite(wacc_value):
hard_failures.append("WACC/cost of equity is non-positive or invalid")
terminal = base.get("terminal", {})
if not terminal or not _is_number(terminal.get("terminal_value")):
hard_failures.append("terminal value missing or invalid")
elif terminal.get("method") == "perpetual_growth" and wacc_value <= float(
terminal.get("terminal_growth_rate", 0.0)
):
hard_failures.append("discount rate must exceed terminal growth")
for metric in ["enterprise_value", "equity_value", "value_per_share"]:
value = base.get(metric)
if not _is_number(value):
hard_failures.append(f"discounting math failed for {metric}")
bridge = plan.get("ev_to_equity_bridge", {})
if (
not _is_number(bridge.get("diluted_shares"))
or float(bridge.get("diluted_shares", 0.0)) <= 0
):
hard_failures.append("EV-to-equity bridge missing valid diluted shares")
tv_pct = float(base.get("tv_percent_ev", 0.0))
if tv_pct > 0.85:
warnings.append(
f"Terminal value is {tv_pct:.1%} of enterprise value, which is very high."
)
elif tv_pct > 0.75:
warnings.append(
f"Terminal value is {tv_pct:.1%} of enterprise value; senior review should pressure-test terminal assumptions."
)
terminal_growth = float(terminal.get("terminal_growth_rate", 0.0))
if terminal.get("method") == "perpetual_growth" and terminal_growth > 0.04:
warnings.append(
f"Terminal growth of {terminal_growth:.1%} exceeds a typical mature-company long-run threshold."
)
hist_margin = float(plan["historicals"].get("ebit", 0.0)) / float(
plan["historicals"].get("revenue", 1.0)
)
final_margin = float(forecast_rows[-1].get("ebit_margin", 0.0)) if forecast_rows else 0.0
if final_margin - hist_margin > 0.05:
forecast_src = _source_for_topic(plan, "forecast") or {}
if forecast_src.get("label") not in {
"company_guidance",
"management_case",
"consensus",
"reported",
"user_provided",
}:
warnings.append(
f"Forecast EBIT margin expands by {(final_margin - hist_margin):.1%} versus latest historical margin without strong source support."
)
if not sensitivity_result.get("directionality", {}).get("passed", False):
hard_failures.append("sensitivity outputs failed directionality checks")
warnings.extend(_source_warnings(plan))
warnings.extend(plan.get("_normalization_notes", []))
# Deduplicate preserving order.
hard_failures = list(dict.fromkeys(hard_failures))
warnings = list(dict.fromkeys(warnings))
checks = {
"hard_failures": hard_failures,
"warnings": warnings,
"informational": informational,
"sensitivity_directionality": sensitivity_result.get("directionality", {}),
}
if base:
checks.update(
{
"base_tv_percent_ev": base.get("tv_percent_ev"),
"base_discount_rate": base.get("discount_rate"),
"base_enterprise_value": base.get("enterprise_value"),
"base_equity_value": base.get("equity_value"),
"base_value_per_share": base.get("value_per_share"),
}
)
return checks
def determine_model_status(
plan: dict[str, Any], hard_failures: list[str], warnings: list[str]
) -> str:
if hard_failures:
return "not-decision-ready"
labels = [src.get("label") for src in plan.get("source_basis", []) if isinstance(src, dict)]
if "placeholder" in labels:
return "screen-grade"
if any(label == "analyst_estimate" for label in labels):
return "screen-grade"
if warnings:
return "senior-review-ready"
return "decision-grade"
def _fmt_pct(x: float | None) -> str:
if x is None or not math.isfinite(float(x)):
return "n/a"
return f"{float(x) * 100:.1f}%"
def _fmt_num(x: float | None) -> str:
if x is None or not math.isfinite(float(x)):
return "n/a"
return f"{float(x):,.1f}"
def _fmt_price(x: float | None) -> str:
if x is None or not math.isfinite(float(x)):
return "n/a"
return f"{float(x):,.2f}"
def build_p0_handoff(
plan: dict[str, Any],
scenario_results: dict[str, dict[str, Any]],
sensitivity_result: dict[str, Any],
model_status: str,
warnings: list[str],
output_dir: str | os.PathLike[str],
include_report_md: bool = True,
) -> dict[str, Any]:
scenario_summary: dict[str, Any] = {}
values = []
for name in SCENARIOS:
r = scenario_results.get(name)
if not r:
continue
scenario_summary[name] = {
"enterprise_value": r.get("enterprise_value"),
"equity_value": r.get("equity_value"),
"value_per_share": r.get("value_per_share"),
"discount_rate": r.get("discount_rate"),
"terminal_value": r.get("terminal", {}).get("terminal_value"),
"terminal_method": r.get("terminal", {}).get("method"),
"tv_percent_ev": r.get("tv_percent_ev"),
}
if _is_number(r.get("value_per_share")):
values.append(float(r["value_per_share"]))
selected_range = {
"low_value_per_share": min(values) if values else None,
"high_value_per_share": max(values) if values else None,
"basis": "downside/base/upside scenario range",
}
key_drivers = identify_key_value_drivers(sensitivity_result.get("rows", []))
base = scenario_results.get("base", {})
output = Path(output_dir)
paths = {
"workbook": str(output / "model.xlsx"),
"plan": str(output / "plan.json"),
"run_log": str(output / "run_log.json"),
}
if include_report_md:
paths["report"] = str(output / "report.md")
return {
"selected_valuation_range": selected_range,
"scenarios": scenario_summary,
"wacc_and_terminal_assumptions": {
"base_discount_rate": base.get("discount_rate"),
"base_wacc": base.get("wacc", {}).get("wacc"),
"base_cost_of_equity": base.get("wacc", {}).get("cost_of_equity"),
"terminal_method": base.get("terminal", {}).get("method"),
"terminal_growth_rate": base.get("terminal", {}).get("terminal_growth_rate"),
"exit_ebitda_multiple": base.get("terminal", {}).get("exit_ebitda_multiple"),
},
"key_value_drivers": key_drivers,
"major_caveats": warnings[:10],
"model_status": model_status,
"paths": paths,
}
def identify_key_value_drivers(sensitivity_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
drivers: list[dict[str, Any]] = []
for sensitivity in sorted({str(r.get("sensitivity")) for r in sensitivity_rows}):
vals = [
float(r["value"])
for r in sensitivity_rows
if r.get("sensitivity") == sensitivity and _is_number(r.get("value"))
]
if vals:
drivers.append(
{
"driver": sensitivity,
"value_per_share_range": max(vals) - min(vals),
"low": min(vals),
"high": max(vals),
}
)
drivers.sort(key=lambda d: d["value_per_share_range"], reverse=True)
return drivers[:5]
def to_model_rows(
plan: dict[str, Any],
scenario_results: dict[str, dict[str, Any]],
sensitivity_rows: list[dict[str, Any]] | None = None,
checks: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
forecast_source = str(plan.get("forecast", {}).get("source_id", ""))
forecast_label = _source_label(plan, forecast_source)
wacc_source = str(plan.get("wacc", {}).get("source_id", ""))
wacc_label = _source_label(plan, wacc_source)
terminal_source = str(plan.get("terminal_value", {}).get("source_id", ""))
terminal_label = _source_label(plan, terminal_source)
bridge = plan.get("ev_to_equity_bridge", {})
bridge_source = str(bridge.get("net_debt_source_id", ""))
bridge_label = _source_label(plan, bridge_source)
share_source = str(bridge.get("share_count_source_id", ""))
share_label = _source_label(plan, share_source)
units = plan.get("meta", {}).get("units", "")
for scenario_name, result in scenario_results.items():
for f in result.get("forecast_rows", []):
period = f["period"]
for section, line_item, key in [
("IS", "Revenue", "revenue"),
("IS", "EBITDA", "ebitda"),
("IS", "EBIT", "ebit"),
("IS", "Cash Taxes", "cash_taxes"),
("FCF", "NOPAT", "nopat"),
("FCF", "D&A", "da"),
("FCF", "Capex", "capex"),
("FCF", "Change in NWC", "change_nwc"),
("FCF", "Unlevered FCF", "unlevered_fcf"),
("FCF", "FCFE", "fcfe"),
]:
if key not in f:
continue
rows.append(
{
"section": section,
"line_item": line_item,
"scenario": scenario_name,
"period": period,
"value": f[key],
"units": units,
"evidence_label": forecast_label,
"source_id": forecast_source,
"notes": "forecast output",
}
)
w = result.get("wacc", {})
for line_item, key in [
("Cost of Equity", "cost_of_equity"),
("Pre-tax Cost of Debt", "pre_tax_cost_of_debt"),
("After-tax Cost of Debt", "after_tax_cost_of_debt"),
("Marginal Tax Rate", "marginal_tax_rate"),
("Target Debt %", "target_debt_pct"),
("Target Equity %", "target_equity_pct"),
("WACC", "wacc"),
]:
rows.append(
{
"section": "WACC",
"line_item": line_item,
"scenario": scenario_name,
"period": "valuation_date",
"value": w.get(key),
"units": "%"
if "Rate" in line_item
or "%" in line_item
or line_item
in {"WACC", "Cost of Equity", "Pre-tax Cost of Debt", "After-tax Cost of Debt"}
else "x",
"evidence_label": wacc_label,
"source_id": wacc_source,
"notes": "cost of capital output",
}
)
terminal = result.get("terminal", {})
for line_item, value in [
("Terminal Value", terminal.get("terminal_value")),
("PV of Terminal Value", result.get("pv", {}).get("pv_terminal_value")),
("TV Percent of EV", result.get("tv_percent_ev")),
("Implied Exit EBITDA Multiple", terminal.get("implied_exit_ebitda_multiple")),
("Implied FCF Yield", terminal.get("implied_fcf_yield")),
]:
rows.append(
{
"section": "TV",
"line_item": line_item,
"scenario": scenario_name,
"period": "terminal",
"value": value,
"units": "%" if "Percent" in line_item or "Yield" in line_item else units,
"evidence_label": terminal_label,
"source_id": terminal_source,
"notes": str(terminal.get("method", "")),
}
)
for line_item, value, label, source_id in [
("PV of FCF", result.get("pv", {}).get("pv_fcf"), forecast_label, forecast_source),
("Enterprise Value", result.get("enterprise_value"), bridge_label, bridge_source),
("Equity Value", result.get("equity_value"), bridge_label, bridge_source),
("Diluted Shares", bridge.get("diluted_shares"), share_label, share_source),
("Value per Share", result.get("value_per_share"), share_label, share_source),
]:
rows.append(
{
"section": "VALUATION",
"line_item": line_item,
"scenario": scenario_name,
"period": "valuation_date",
"value": value,
"units": plan.get("meta", {}).get("currency", "")
+ ("/share" if "Share" in line_item else "mm"),
"evidence_label": label,
"source_id": source_id,
"notes": "valuation output",
}
)
if checks:
for item in checks.get("hard_failures", []):
rows.append(
{
"section": "CHECKS",
"line_item": "Hard Failure",
"scenario": "all",
"period": "run",
"value": "",
"units": "",
"evidence_label": "derived",
"source_id": "",
"notes": item,
}
)
for item in checks.get("warnings", []):
rows.append(
{
"section": "CHECKS",
"line_item": "Warning",
"scenario": "all",
"period": "run",
"value": "",
"units": "",
"evidence_label": "derived",
"source_id": "",
"notes": item,
}
)
for src in plan.get("source_basis", []):
rows.append(
{
"section": "ASSUMPTIONS",
"line_item": f"Source Basis - {src.get('topic')}",
"scenario": "all",
"period": str(src.get("as_of_date", "")),
"value": "",
"units": "",
"evidence_label": src.get("label", ""),
"source_id": src.get("id", ""),
"notes": f"{src.get('source_name', '')}: {src.get('notes', '')}",
}
)
return rows
def _rows_to_table(rows: list[dict[str, Any]], columns: list[str] | None = None) -> list[list[Any]]:
if not rows:
return [columns or []]
if columns is None:
columns = []
for row in rows:
for key in row.keys():
if key not in columns:
columns.append(key)
return [columns] + [[row.get(col, "") for col in columns] for row in rows]
def build_workbook_sheets(
plan: dict[str, Any],
scenario_results: dict[str, dict[str, Any]],
sensitivity_result: dict[str, Any],
checks: dict[str, Any],
run_log: dict[str, Any],
) -> dict[str, list[list[Any]]]:
summary_rows: list[dict[str, Any]] = []
for name in SCENARIOS:
result = scenario_results.get(name)
if not result:
continue
summary_rows.append(
{
"scenario": name,
"enterprise_value": result.get("enterprise_value"),
"equity_value": result.get("equity_value"),
"value_per_share": result.get("value_per_share"),
"discount_rate": result.get("discount_rate"),
"pv_fcf": result.get("pv", {}).get("pv_fcf"),
"pv_terminal_value": result.get("pv", {}).get("pv_terminal_value"),
"terminal_value": result.get("terminal", {}).get("terminal_value"),
"tv_percent_ev": result.get("tv_percent_ev"),
"terminal_method": result.get("terminal", {}).get("method"),
}
)
model_rows = to_model_rows(plan, scenario_results, sensitivity_result.get("rows", []), checks)
sensitivity_rows = sensitivity_result.get("rows", [])
check_rows = []
for item in checks.get("hard_failures", []):
check_rows.append({"severity": "hard_failure", "check": item, "status": "fail"})
for item in checks.get("warnings", []):
check_rows.append({"severity": "warning", "check": item, "status": "review"})
for detail in checks.get("sensitivity_directionality", {}).get("details", []):
check_rows.append(
{
"severity": "check",
"check": detail.get("check"),
"status": "pass" if detail.get("passed") else "fail",
"detail": detail.get("detail"),
}
)
assumptions_rows = []
for src in plan.get("source_basis", []):
assumptions_rows.append(
{
"id": src.get("id"),
"topic": src.get("topic"),
"label": src.get("label"),
"source_name": src.get("source_name"),
"source_type": src.get("source_type"),
"as_of_date": src.get("as_of_date"),
"confidence": src.get("confidence"),
"notes": src.get("notes"),
}
)
assumptions_rows.extend(
[
{
"id": "meta.company",
"topic": "meta",
"label": "derived",
"source_name": plan.get("meta", {}).get("company"),
"source_type": "plan",
"as_of_date": plan.get("meta", {}).get("as_of_date"),
"confidence": "",
"notes": "company modeled",
},
{
"id": "meta.model_type",
"topic": "meta",
"label": "derived",
"source_name": plan.get("meta", {}).get("model_type"),
"source_type": "plan",
"as_of_date": plan.get("meta", {}).get("valuation_date"),
"confidence": "",
"notes": "cash flow basis",
},
]
)
run_rows = [
["field", "value"],
["model_status", run_log.get("model_status")],
["workbook_mode", run_log.get("workbook_mode")],
["hard_failure_count", len(run_log.get("hard_failures", []))],
["warning_count", len(run_log.get("warnings", []))],
["company", plan.get("meta", {}).get("company")],
["valuation_date", plan.get("meta", {}).get("valuation_date")],
]
for idx, warning in enumerate(run_log.get("warnings", [])[:20], start=1):
run_rows.append([f"warning_{idx}", warning])
for idx, failure in enumerate(run_log.get("hard_failures", [])[:20], start=1):
run_rows.append([f"hard_failure_{idx}", failure])
return {
"Executive Summary": _rows_to_table(summary_rows),
"Model": _rows_to_table(
model_rows,
[
"section",
"line_item",
"scenario",
"period",
"value",
"units",
"evidence_label",
"source_id",
"notes",
],
),
"Sensitivities": _rows_to_table(
sensitivity_rows,
[
"sensitivity",
"x_axis",
"x_value",
"y_axis",
"y_value",
"metric",
"value",
"units",
"notes",
],
),
"Checks": _rows_to_table(check_rows),
"Assumptions": _rows_to_table(assumptions_rows),
"Run Log": run_rows,
}
def render_report(
plan: dict[str, Any],
scenario_results: dict[str, dict[str, Any]],
checks: dict[str, Any],
run_log: dict[str, Any],
) -> str:
meta = plan.get("meta", {})
lines: list[str] = []
lines.append(f"# DCF Valuation Report: {meta.get('company', 'Company')}")
lines.append("")
lines.append(f"**Model status:** `{run_log.get('model_status')}` ")
lines.append(f"**Workbook mode:** `{run_log.get('workbook_mode')}` ")
lines.append(f"**Valuation date:** {meta.get('valuation_date')} ")
lines.append(f"**Currency / units:** {meta.get('currency')} / {meta.get('units')} ")
lines.append("")
lines.append("## Valuation range")
handoff = run_log.get("p0_handoff", {})
value_range = handoff.get("selected_valuation_range", {})
lines.append(
f"- Scenario value per share range: **{_fmt_price(value_range.get('low_value_per_share'))} to {_fmt_price(value_range.get('high_value_per_share'))}**"
)
lines.append(f"- Basis: {value_range.get('basis', 'scenario range')}")
lines.append("")
lines.append("| Scenario | EV | Equity value | Value / share | Discount rate | TV % EV |")
lines.append("|---|---:|---:|---:|---:|---:|")
for name in SCENARIOS:
r = scenario_results.get(name)
if not r:
continue
lines.append(
f"| {name.title()} | {_fmt_num(r.get('enterprise_value'))} | {_fmt_num(r.get('equity_value'))} | {_fmt_price(r.get('value_per_share'))} | {_fmt_pct(r.get('discount_rate'))} | {_fmt_pct(r.get('tv_percent_ev'))} |"
)
lines.append("")
base = scenario_results.get("base", {})
if base:
terminal = base.get("terminal", {})
wacc = base.get("wacc", {})
lines.append("## Base-case DCF bridge")
lines.append(
f"- PV of explicit cash flows: **{_fmt_num(base.get('pv', {}).get('pv_fcf'))}**"
)
lines.append(
f"- PV of terminal value: **{_fmt_num(base.get('pv', {}).get('pv_terminal_value'))}**"
)
lines.append(f"- Enterprise value: **{_fmt_num(base.get('enterprise_value'))}**")
lines.append(f"- Equity value: **{_fmt_num(base.get('equity_value'))}**")
lines.append(f"- Value per share: **{_fmt_price(base.get('value_per_share'))}**")
lines.append("")
lines.append("## WACC and terminal value")
lines.append(f"- Cost of equity: **{_fmt_pct(wacc.get('cost_of_equity'))}**")
lines.append(f"- WACC: **{_fmt_pct(wacc.get('wacc'))}**")
lines.append(f"- Terminal method: **{terminal.get('method')}**")
lines.append(f"- Terminal growth: **{_fmt_pct(terminal.get('terminal_growth_rate'))}**")
lines.append(
f"- Implied exit EBITDA multiple: **{_fmt_num(terminal.get('implied_exit_ebitda_multiple'))}x**"
)
lines.append("")
lines.append("## Key value drivers")
drivers = handoff.get("key_value_drivers", [])
if drivers:
lines.append("| Driver | Value/share range | Low | High |")
lines.append("|---|---:|---:|---:|")
for d in drivers:
lines.append(
f"| {d.get('driver')} | {_fmt_price(d.get('value_per_share_range'))} | {_fmt_price(d.get('low'))} | {_fmt_price(d.get('high'))} |"
)
else:
lines.append("No sensitivity drivers were available.")
lines.append("")
lines.append("## QA checks")
hard = checks.get("hard_failures", [])
warns = checks.get("warnings", [])
if hard:
lines.append("**Hard failures:**")
for item in hard:
lines.append(f"- {item}")
else:
lines.append("No hard failures detected.")
if warns:
lines.append("")
lines.append("**Warnings / senior-review items:**")
for item in warns:
lines.append(f"- {item}")
else:
lines.append("")
lines.append("No warnings detected.")
lines.append("")
lines.append("## Source basis")
lines.append("| Topic | Label | Source | As of | Confidence |")
lines.append("|---|---|---|---|---|")
for src in plan.get("source_basis", []):
lines.append(
f"| {src.get('topic')} | {src.get('label')} | {src.get('source_name')} | {src.get('as_of_date')} | {src.get('confidence')} |"
)
lines.append("")
lines.append("## Generated artifacts")
for label, path in handoff.get("paths", {}).items():
lines.append(f"- {label}: `{path}`")
lines.append("")
lines.append(
"This deterministic export is value-based. It is not a fully linked banker formula workbook."
)
return "\n".join(lines).rstrip() + "\n"
# ---- Minimal XLSX writer -------------------------------------------------
def _col_letter(n: int) -> str:
result = ""
while n:
n, rem = divmod(n - 1, 26)
result = chr(65 + rem) + result
return result
def _cell_ref(row_idx: int, col_idx: int) -> str:
return f"{_col_letter(col_idx)}{row_idx}"
def _xml_text(value: Any) -> str:
return escape(str(value), {'"': """})
def _sheet_name(name: str, used: set[str]) -> str:
invalid = set("[]:*?/\\")
clean = "".join("_" if c in invalid else c for c in name).strip() or "Sheet"
clean = clean[:31]
original = clean
i = 2
while clean in used:
suffix = f"_{i}"
clean = (original[: 31 - len(suffix)] + suffix)[:31]
i += 1
used.add(clean)
return clean
def _sheet_xml(rows: list[list[Any]]) -> str:
xml_rows: list[str] = []
for r_idx, row in enumerate(rows, start=1):
cells: list[str] = []
for c_idx, value in enumerate(row, start=1):
if value is None:
continue
ref = _cell_ref(r_idx, c_idx)
if isinstance(value, bool):
cells.append(f'<c r="{ref}" t="b"><v>{1 if value else 0}</v></c>')
elif _is_number(value):
num = float(value)
cells.append(f'<c r="{ref}"><v>{num:.12g}</v></c>')
else:
text = _xml_text(value)
cells.append(f'<c r="{ref}" t="inlineStr"><is><t>{text}</t></is></c>')
xml_rows.append(f'<row r="{r_idx}">' + "".join(cells) + "</row>")
dimension = "A1"
if rows:
max_cols = max((len(r) for r in rows), default=1)
dimension = f"A1:{_cell_ref(len(rows), max_cols)}"
cols = "".join(
f'<col min="{i}" max="{i}" width="18" customWidth="1"/>'
for i in range(1, max((len(r) for r in rows), default=1) + 1)
)
return (
'<?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="{dimension}"/>'
'<sheetViews><sheetView showGridLines="0" workbookViewId="0"><pane ySplit="1" topLeftCell="A2" activePane="bottomLeft" state="frozen"/></sheetView></sheetViews>'
f"<cols>{cols}</cols>"
"<sheetData>" + "".join(xml_rows) + "</sheetData>"
"</worksheet>"
)
def _content_types(sheet_count: int) -> str:
overrides = [
'<Override PartName="/xl/workbook.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet.main+xml"/>',
'<Override PartName="/xl/styles.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.styles+xml"/>',
'<Override PartName="/docProps/core.xml" ContentType="application/vnd.openxmlformats-package.core-properties+xml"/>',
'<Override PartName="/docProps/app.xml" ContentType="application/vnd.openxmlformats-officedocument.extended-properties+xml"/>',
]
for i in range(1, sheet_count + 1):
overrides.append(
f'<Override PartName="/xl/worksheets/sheet{i}.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.worksheet+xml"/>'
)
return (
'<?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"/>' + "".join(overrides) + "</Types>"
)
def _root_rels() -> str:
return (
'<?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"/>'
'<Relationship Id="rId2" Type="http://schemas.openxmlformats.org/package/2006/relationships/metadata/core-properties" Target="docProps/core.xml"/>'
'<Relationship Id="rId3" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/extended-properties" Target="docProps/app.xml"/>'
"</Relationships>"
)
def _workbook_xml(sheet_names: list[str]) -> str:
sheets = []
for idx, name in enumerate(sheet_names, start=1):
sheets.append(f'<sheet name="{_xml_text(name)}" sheetId="{idx}" r:id="rId{idx}"/>')
return (
'<?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">'
"<bookViews><workbookView/></bookViews>"
"<sheets>" + "".join(sheets) + "</sheets>"
"</workbook>"
)
def _workbook_rels(sheet_count: int) -> str:
rels = []
for idx in range(1, sheet_count + 1):
rels.append(
f'<Relationship Id="rId{idx}" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/worksheet" Target="worksheets/sheet{idx}.xml"/>'
)
rels.append(
f'<Relationship Id="rId{sheet_count + 1}" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/styles" Target="styles.xml"/>'
)
return (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
+ "".join(rels)
+ "</Relationships>"
)
def _styles_xml() -> str:
return (
'<?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>"
)
def _core_xml() -> str:
now = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
return (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<cp:coreProperties xmlns:cp="http://schemas.openxmlformats.org/package/2006/metadata/core-properties" '
'xmlns:dc="http://purl.org/dc/elements/1.1/" '
'xmlns:dcterms="http://purl.org/dc/terms/" '
'xmlns:dcmitype="http://purl.org/dc/dcmitype/" '
'xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">'
"<dc:title>DCF Model Export</dc:title>"
"<dc:creator>dcf-model-builder</dc:creator>"
f'<dcterms:created xsi:type="dcterms:W3CDTF">{now}</dcterms:created>'
f'<dcterms:modified xsi:type="dcterms:W3CDTF">{now}</dcterms:modified>'
"</cp:coreProperties>"
)
def _app_xml(sheet_names: list[str]) -> str:
names = "".join(f"<vt:lpstr>{_xml_text(name)}</vt:lpstr>" for name in sheet_names)
return (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<Properties xmlns="http://schemas.openxmlformats.org/officeDocument/2006/extended-properties" '
'xmlns:vt="http://schemas.openxmlformats.org/officeDocument/2006/docPropsVTypes">'
"<Application>dcf-model-builder</Application>"
'<HeadingPairs><vt:vector size="2" baseType="variant"><vt:variant><vt:lpstr>Worksheets</vt:lpstr></vt:variant><vt:variant><vt:i4>'
+ str(len(sheet_names))
+ "</vt:i4></vt:variant></vt:vector></HeadingPairs>"
'<TitlesOfParts><vt:vector size="'
+ str(len(sheet_names))
+ '" baseType="lpstr">'
+ names
+ "</vt:vector></TitlesOfParts>"
"</Properties>"
)
def write_xlsx(path: str | os.PathLike[str], rows: Any, sheet_name: str = "Model") -> None:
"""Write a minimal valid XLSX. `rows` may be a table or dict of sheet tables."""
if isinstance(rows, dict):
sheets_input = rows
else:
sheets_input = {sheet_name: rows}
used: set[str] = set()
sheet_names = []
sheet_tables = []
for raw_name, table in sheets_input.items():
name = _sheet_name(str(raw_name), used)
sheet_names.append(name)
sheet_tables.append(table)
Path(path).parent.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
zf.writestr("[Content_Types].xml", _content_types(len(sheet_names)))
zf.writestr("_rels/.rels", _root_rels())
zf.writestr("xl/workbook.xml", _workbook_xml(sheet_names))
zf.writestr("xl/_rels/workbook.xml.rels", _workbook_rels(len(sheet_names)))
zf.writestr("xl/styles.xml", _styles_xml())
zf.writestr("docProps/core.xml", _core_xml())
zf.writestr("docProps/app.xml", _app_xml(sheet_names))
for idx, table in enumerate(sheet_tables, start=1):
zf.writestr(f"xl/worksheets/sheet{idx}.xml", _sheet_xml(table))
SHA-256: b0cc3db62f4d4d143cae8c3c2c4f3374edaab2ce9054e9031c7001360a2beabd