← Files Investment BankingARCHIVED FILE
skills/dcf-model-builder/scripts/build_banker_formula_workbook_runtime
29 KB · Oct 5, 2026 · 18:28 UTC
#!/usr/bin/env python3
"""Materialize the bundled DCF banker formula workbook template from plan.json.
This is a template materializer: it preserves the shipped workbook's formulas,
styles, tabs, and checks, then writes plan-derived inputs, source notes, and
posture warnings into the control areas.
"""
from __future__ import annotations
import argparse
import json
import math
import sys
import zipfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple
from xml.etree import ElementTree as ET
SCRIPT_DIR = Path(__file__).resolve().parent
SKILL_ROOT = SCRIPT_DIR.parent
PLUGIN_ROOT = SCRIPT_DIR.parents[2]
sys.path.insert(0, str(SCRIPT_DIR))
if str(PLUGIN_ROOT) not in sys.path:
sys.path.insert(0, str(PLUGIN_ROOT))
from shared.model_artifacts import write_model_manifest # noqa: E402
from shared.model_citations import write_model_citations_for_workbook # noqa: E402
from skill_core import load_json, validate_plan_structure # noqa: E402
NS_MAIN = "http://schemas.openxmlformats.org/spreadsheetml/2006/main"
NS_REL = "http://schemas.openxmlformats.org/officeDocument/2006/relationships"
ET.register_namespace("", NS_MAIN)
ET.register_namespace("r", NS_REL)
DEFAULT_TEMPLATE = SKILL_ROOT / "assets" / "templates" / "banker_formula_workbook_template.xlsx"
OUTPUT_WORKBOOK = "banker_formula_workbook.xlsx"
OUTPUT_RUN_LOG = "banker_formula_workbook_run_log.json"
OUTPUT_MANIFEST = "manifest.json"
PARTIAL_CONTEXT_WARNING = "Screen-grade only; placeholder assumptions used."
TEMPLATE_FORECAST_PERIODS = 6
FORECAST_COLS = ["B", "C", "D", "E", "F", "G"]
REQUIRED_SHEETS = [
"Cover",
"Executive Summary",
"Control Panel",
"Historical Financials",
"Revenue Build",
"Margin Cost Build",
"Working Capital",
"Capex D&A",
"Tax Schedule",
"Unlevered FCF",
"WACC",
"Terminal Value",
"DCF Valuation",
"Sensitivities",
"Checks",
"Source Notes",
]
def _file_record(path: Path, role: str, description: str) -> Dict[str, Any]:
return {
"path": str(path),
"role": role,
"description": description,
"exists": True if path.name == OUTPUT_MANIFEST else path.exists(),
}
def write_output_manifest(output_dir: Path, run_log: Dict[str, Any]) -> Dict[str, Any]:
workbook_path = output_dir / OUTPUT_WORKBOOK
support_paths = [(output_dir / OUTPUT_RUN_LOG, "formula workbook run log")]
if (output_dir / "model_citations.json").exists():
support_paths.append((output_dir / "model_citations.json", "workbook cell/range citation ledger for model outputs"))
return write_model_manifest(
output_dir,
"dcf-model-builder",
"banker_formula_workbook",
workbook_path,
str(run_log.get("model_status", "screen-grade")),
support_paths,
run_log.get("hard_failures", []),
run_log.get("warnings", []),
)
manifest_path = output_dir / OUTPUT_MANIFEST
workbook_path = output_dir / OUTPUT_WORKBOOK
run_log_path = output_dir / OUTPUT_RUN_LOG
manifest = {
"manifest_version": "1.0",
"generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
"skill": "dcf-model-builder",
"artifact_mode": "banker_formula_workbook",
"model_status": run_log.get("model_status"),
"output_dir": str(output_dir),
"primary_human_deliverable": str(workbook_path),
"human_deliverables": [_file_record(workbook_path, "human_deliverable", "banker formula workbook")],
"agent_artifacts": [
_file_record(run_log_path, "agent_artifact", "formula workbook run log"),
_file_record(manifest_path, "agent_artifact", "agent-facing output manifest"),
],
"hard_failure_count": len(run_log.get("hard_failures", [])),
"warning_count": len(run_log.get("warnings", [])),
"discipline_note": "Use the workbook as the main deliverable; manifest/run_log files are agent-facing support artifacts.",
}
manifest_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
return manifest
def qname(tag: str) -> str:
return f"{{{NS_MAIN}}}{tag}"
def col_to_num(col: str) -> int:
value = 0
for char in col:
value = value * 26 + ord(char.upper()) - 64
return value
def cell_ref_parts(ref: str) -> Tuple[str, int]:
import re
match = re.fullmatch(r"([A-Z]+)([0-9]+)", ref)
if not match:
raise ValueError(f"Invalid cell reference: {ref}")
return match.group(1), int(match.group(2))
def cell_sort_key(ref: str) -> Tuple[int, int]:
col, row = cell_ref_parts(ref)
return row, col_to_num(col)
def is_number(value: Any) -> bool:
return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(float(value))
def source_for_topic(plan: Dict[str, Any], topic: str) -> Optional[Dict[str, Any]]:
for source in plan.get("source_basis", []):
if isinstance(source, dict) and source.get("topic") == topic:
return source
return None
def source_row(source: Optional[Dict[str, Any]], fallback_note: str) -> Tuple[str, str, str, str, str]:
if not source:
return ("Missing source", "placeholder", "", "Low", fallback_note)
return (
str(source.get("source_name", source.get("id", "source_basis"))),
str(source.get("label", "placeholder")),
str(source.get("as_of_date", "")),
str(source.get("confidence", "low")).title(),
str(source.get("notes", fallback_note)),
)
def screen_grade_required(plan: Dict[str, Any]) -> bool:
weak_labels = {"placeholder", "analyst_estimate"}
for source in plan.get("source_basis", []):
label = str(source.get("label", "")).lower()
confidence = str(source.get("confidence", "")).lower()
if label in weak_labels or confidence == "low":
return True
return False
def forecast_period_labels(plan: Dict[str, Any], warnings: List[str]) -> List[str]:
start_year = int(plan.get("timeline", {}).get("start_year", 2025)) + 1
horizon = int(plan.get("timeline", {}).get("horizon_years", TEMPLATE_FORECAST_PERIODS))
if horizon != TEMPLATE_FORECAST_PERIODS:
warnings.append(
f"Formula template has {TEMPLATE_FORECAST_PERIODS} forecast columns; plan has {horizon}. "
"Values were extended or truncated for template compatibility."
)
return [f"{start_year + i}E" for i in range(TEMPLATE_FORECAST_PERIODS)]
def extend_values(value: Any, length: int, field_name: str, warnings: List[str], default: Optional[float] = None) -> List[Optional[float]]:
if isinstance(value, list):
values = [float(v) if is_number(v) else default for v in value]
elif is_number(value):
values = [float(value)] * length
else:
values = [default] * length
if not values:
values = [default] * length
if len(values) < length:
warnings.append(f"{field_name} supplied {len(values)} values; final value repeated to fill template forecast columns.")
fill = values[-1]
values = values + [fill] * (length - len(values))
if len(values) > length:
warnings.append(f"{field_name} supplied {len(values)} values; truncated to template forecast columns.")
values = values[:length]
return values
def scenario_values(plan: Dict[str, Any], scenario_name: str, field: str, warnings: List[str], default: Optional[float] = None) -> List[Optional[float]]:
scenario = plan.get("scenarios", {}).get(scenario_name, {})
return extend_values(scenario.get(field), TEMPLATE_FORECAST_PERIODS, f"scenarios.{scenario_name}.{field}", warnings, default)
def update_series(updates: Dict[str, Any], row: int, values: List[Any], cols: List[str] = FORECAST_COLS) -> None:
for col, value in zip(cols, values):
if value is not None:
updates[f"{col}{row}"] = value
def ebitda_margin_values(plan: Dict[str, Any], scenario_name: str, warnings: List[str]) -> List[Optional[float]]:
ebit = scenario_values(plan, scenario_name, "ebit_margin", warnings, 0.0)
da = scenario_values(plan, scenario_name, "da_percent_revenue", warnings, 0.0)
return [(e or 0.0) + (d or 0.0) for e, d in zip(ebit, da)]
def wacc_value(plan: Dict[str, Any]) -> float:
wacc = plan.get("wacc", {})
tax = float(wacc.get("marginal_tax_rate", 0.0) or 0.0)
cost_equity = (
float(wacc.get("risk_free_rate", 0.0) or 0.0)
+ float(wacc.get("beta", 0.0) or 0.0) * float(wacc.get("equity_risk_premium", 0.0) or 0.0)
+ float(wacc.get("size_premium", 0.0) or 0.0)
+ float(wacc.get("company_specific_premium", 0.0) or 0.0)
+ float(wacc.get("country_risk_premium", 0.0) or 0.0)
)
after_tax_debt = float(wacc.get("pre_tax_cost_of_debt", 0.0) or 0.0) * (1.0 - tax)
debt_pct = float(wacc.get("target_debt_pct", 0.0) or 0.0)
equity_pct = float(wacc.get("target_equity_pct", 1.0) or 1.0)
preferred_pct = float(wacc.get("preferred_pct", 0.0) or 0.0)
preferred_cost = float(wacc.get("pre_tax_cost_of_preferred", 0.0) or 0.0)
base = cost_equity * equity_pct + after_tax_debt * debt_pct + preferred_cost * preferred_pct
return base + float(plan.get("scenarios", {}).get("base", {}).get("wacc_adjustment", 0.0) or 0.0)
def net_debt_value(plan: Dict[str, Any]) -> float:
bridge = plan.get("ev_to_equity_bridge", {})
debt_like = (
float(bridge.get("debt", 0.0) or 0.0)
+ float(bridge.get("leases", 0.0) or 0.0)
+ float(bridge.get("pensions", 0.0) or 0.0)
+ float(bridge.get("other_debt_like_items", 0.0) or 0.0)
)
return debt_like - float(bridge.get("cash", 0.0) or 0.0)
def build_control_panel_updates(plan: Dict[str, Any], missing_inputs: List[str], warnings: List[str]) -> Dict[str, Any]:
meta = plan.get("meta", {})
bridge = plan.get("ev_to_equity_bridge", {})
wacc = plan.get("wacc", {})
terminal = plan.get("terminal_value", {})
periods = forecast_period_labels(plan, warnings)
screen_grade = screen_grade_required(plan)
current_share_price = (
bridge.get("current_share_price")
or plan.get("market", {}).get("current_share_price")
or meta.get("current_share_price")
)
if not is_number(current_share_price):
missing_inputs.append("current_share_price")
current_share_price = None
updates: Dict[str, Any] = {
"B4": meta.get("company", "Company"),
"B5": meta.get("currency", "USD"),
"B6": meta.get("units", "$mm except per-share data"),
"B7": meta.get("valuation_date", meta.get("as_of_date", "")),
"B8": "screen-grade" if screen_grade else "senior-review-ready",
"B9": "banker_formula_workbook",
"B10": "Base",
"B11": "Yes" if plan.get("forecast", {}).get("mid_year_convention") else "No",
"B12": current_share_price if current_share_price is not None else 0.0,
"B13": PARTIAL_CONTEXT_WARNING if screen_grade else "Source-backed DCF; review checks before circulation.",
"B14": "Gordon Growth" if terminal.get("method") == "perpetual_growth" else "Exit Multiple",
"B15": 0.0,
"B69": wacc.get("risk_free_rate", 0.0),
"B70": wacc.get("beta", 0.0),
"B71": wacc.get("equity_risk_premium", 0.0),
"B72": wacc.get("size_premium", 0.0),
"B73": wacc.get("pre_tax_cost_of_debt", 0.0),
"B74": wacc.get("target_debt_pct", 0.0),
"B76": "No",
"B77": wacc_value(plan),
"B79": "No external links",
"B83": plan.get("scenarios", {}).get("base", {}).get("terminal_growth_rate", terminal.get("perpetual_growth_rate", 0.0)),
"B84": terminal.get("exit_ebitda_multiple", 0.0),
"B85": net_debt_value(plan),
"B86": bridge.get("diluted_shares", 0.0),
"B87": bridge.get("minorities", 0.0),
"B88": bridge.get("preferred_stock", 0.0),
"B89": float(bridge.get("non_operating_assets", 0.0) or 0.0) + float(bridge.get("associates", 0.0) or 0.0),
"B90": -float(bridge.get("options", 0.0) or 0.0),
}
update_series(updates, 18, periods)
row_map = {"base": 21, "downside": 22, "upside": 23}
for scenario_name, row in row_map.items():
update_series(updates, row, scenario_values(plan, scenario_name, "revenue_growth", warnings, 0.0))
row_map = {"base": 26, "downside": 27, "upside": 28}
for scenario_name, row in row_map.items():
update_series(updates, row, ebitda_margin_values(plan, scenario_name, warnings))
row_map = {"base": 31, "downside": 32, "upside": 33}
for scenario_name, row in row_map.items():
update_series(updates, row, scenario_values(plan, scenario_name, "ebit_margin", warnings, 0.0))
row_map = {"base": 36, "downside": 37, "upside": 38}
for scenario_name, row in row_map.items():
update_series(updates, row, scenario_values(plan, scenario_name, "da_percent_revenue", warnings, 0.0))
row_map = {"base": 41, "downside": 42, "upside": 43}
for scenario_name, row in row_map.items():
update_series(updates, row, scenario_values(plan, scenario_name, "capex_percent_revenue", warnings, 0.0))
row_map = {"base": 46, "downside": 47, "upside": 48}
for scenario_name, row in row_map.items():
update_series(updates, row, scenario_values(plan, scenario_name, "nwc_percent_revenue", warnings, 0.0))
row_map = {"base": 51, "downside": 52, "upside": 53}
for scenario_name, row in row_map.items():
update_series(updates, row, scenario_values(plan, scenario_name, "tax_rate", warnings, 0.0))
row_map = {"base": 56, "downside": 57, "upside": 58}
for scenario_name, row in row_map.items():
update_series(updates, row, scenario_values(plan, scenario_name, "tax_rate", warnings, 0.0))
# The DCF template uses component working-capital drivers; the plan schema
# currently provides total operating NWC % revenue. Map that total to AR and
# zero the remaining components so model NWC still equals the plan driver.
base_nwc = scenario_values(plan, "base", "nwc_percent_revenue", warnings, 0.0)
update_series(updates, 61, base_nwc)
for row in [62, 63, 64, 65]:
update_series(updates, row, [0.0] * TEMPLATE_FORECAST_PERIODS)
missing_inputs.append("working_capital_component_split")
if "gross_profit_margin" not in plan.get("historicals", {}) and "gross_margin" not in plan.get("historicals", {}):
missing_inputs.append("gross_profit_margin_or_gross_profit_history")
if current_share_price is None:
warnings.append("Current share price was not provided; formula workbook uses 0.0 placeholder for upside/downside until supplied.")
warnings.append("Total operating NWC % revenue was mapped to the AR bucket because no AR/inventory/AP/OCA/OCL split is present in plan.json.")
return updates
def build_executive_summary_updates(plan: Dict[str, Any]) -> Dict[str, Any]:
screen_grade = screen_grade_required(plan)
return {
"A3": "Posture warning",
"B3": PARTIAL_CONTEXT_WARNING if screen_grade else "Source-backed DCF; review checks before circulation.",
}
def build_historical_updates(plan: Dict[str, Any], missing_inputs: List[str], warnings: List[str]) -> Dict[str, Any]:
hist = plan.get("historicals", {})
source = source_for_topic(plan, "historicals")
label = str(source.get("label", hist.get("source_id", "placeholder"))) if source else "placeholder"
note = "Latest historical year populated from plan.json; add multi-year sourced history before circulation."
latest_year = hist.get("latest_year")
updates: Dict[str, Any] = {}
if latest_year:
updates["D5"] = f"{latest_year}A"
if is_number(hist.get("revenue")):
updates["D6"] = hist["revenue"]
updates["D10"] = hist["revenue"]
if is_number(hist.get("ebitda")):
updates["D10"] = hist["ebitda"]
if is_number(hist.get("da")):
updates["D12"] = hist["da"]
if is_number(hist.get("ebit")):
updates["D13"] = hist["ebit"]
if is_number(hist.get("cash_taxes")):
updates["D15"] = hist["cash_taxes"]
if is_number(hist.get("capex")):
updates["D17"] = hist["capex"]
if is_number(hist.get("net_working_capital")):
updates["D18"] = hist["net_working_capital"]
if is_number(hist.get("unlevered_fcf")):
updates["D20"] = hist["unlevered_fcf"]
for row in [6, 8, 10, 12, 13, 15, 16, 17, 18, 20, 21]:
updates[f"K{row}"] = label
updates[f"L{row}"] = note
missing_inputs.append("three_year_historical_financials")
warnings.append("Only latest historical year is available in the DCF plan schema; template historical columns B/C remain unchanged.")
return updates
def build_revenue_updates(plan: Dict[str, Any]) -> Dict[str, Any]:
hist = plan.get("historicals", {})
revenue = float(hist.get("revenue", 0.0) or 0.0)
return {
"D10": revenue,
"D11": 0.0,
}
def build_source_notes_updates(plan: Dict[str, Any]) -> Dict[str, Any]:
rows = {
5: ("historicals", "Replace with filings, audit reports, or normalized financials."),
6: ("forecast", "Replace with sourced operating forecast, management case, or approved analyst case."),
7: ("wacc", "Replace with market data as of valuation date."),
8: ("terminal_value", "Confirm long-run growth, exit multiple, and terminal ROIC support."),
9: ("net_debt", "Replace with latest balance sheet and debt-like item bridge."),
10: ("share_count", "Replace with latest diluted share count or treasury-stock-method schedule."),
}
updates: Dict[str, Any] = {}
for row, (topic, fallback_note) in rows.items():
source, label, as_of, confidence, note = source_row(source_for_topic(plan, topic), fallback_note)
updates[f"B{row}"] = source
updates[f"C{row}"] = label
updates[f"D{row}"] = as_of
updates[f"E{row}"] = confidence
updates[f"F{row}"] = note
updates[f"G{row}"] = "Model owner"
updates[f"H{row}"] = "Open" if label in {"placeholder", "analyst_estimate"} or confidence.lower() == "low" else "Reviewed"
valuation_date = plan.get("meta", {}).get("valuation_date", "")
updates.update(
{
"B11": "Missing source",
"C11": "placeholder",
"D11": valuation_date,
"E11": "Low",
"F11": "Current share price was not provided in plan.json unless supplied in market.current_share_price or ev_to_equity_bridge.current_share_price.",
"G11": "Model owner",
"H11": "Open",
"B12": "Plan schema",
"C12": "derived",
"D12": valuation_date,
"E12": "Medium",
"F12": "No segment-level revenue split provided; forecast starts from total historical revenue.",
"G12": "Model owner",
"H12": "Open",
}
)
return updates
def xml_sheet_paths(xlsx_path: Path) -> Dict[str, str]:
with zipfile.ZipFile(xlsx_path) as zf:
workbook = ET.fromstring(zf.read("xl/workbook.xml"))
rels = ET.fromstring(zf.read("xl/_rels/workbook.xml.rels"))
relmap = {rel.attrib["Id"]: rel.attrib["Target"] for rel in rels}
sheets_el = workbook.find(qname("sheets"))
if sheets_el is None:
return {}
paths: Dict[str, str] = {}
for sheet in sheets_el:
rid = sheet.attrib[f"{{{NS_REL}}}id"]
target = relmap[rid].lstrip("/")
paths[sheet.attrib["name"]] = target if target.startswith("xl/") else f"xl/{target}"
return paths
def get_or_create_row(sheet_root: ET.Element, row_num: int) -> ET.Element:
sheet_data = sheet_root.find(qname("sheetData"))
if sheet_data is None:
sheet_data = ET.SubElement(sheet_root, qname("sheetData"))
for row in sheet_data.findall(qname("row")):
if int(row.attrib.get("r", "0")) == row_num:
return row
row = ET.Element(qname("row"), {"r": str(row_num)})
inserted = False
for idx, existing in enumerate(list(sheet_data)):
if existing.tag == qname("row") and int(existing.attrib.get("r", "0")) > row_num:
sheet_data.insert(idx, row)
inserted = True
break
if not inserted:
sheet_data.append(row)
return row
def get_or_create_cell(row: ET.Element, ref: str) -> ET.Element:
for cell in row.findall(qname("c")):
if cell.attrib.get("r") == ref:
return cell
cell = ET.Element(qname("c"), {"r": ref})
new_col = col_to_num(cell_ref_parts(ref)[0])
inserted = False
for idx, existing in enumerate(list(row)):
if existing.tag == qname("c") and col_to_num(cell_ref_parts(existing.attrib.get("r", "A1"))[0]) > new_col:
row.insert(idx, cell)
inserted = True
break
if not inserted:
row.append(cell)
return cell
def set_cell_value(cell: ET.Element, value: Any) -> None:
ref = cell.attrib.get("r", "")
style = cell.attrib.get("s")
cell.attrib.clear()
cell.attrib["r"] = ref
if style is not None:
cell.attrib["s"] = style
for child in list(cell):
cell.remove(child)
if is_number(value):
v = ET.SubElement(cell, qname("v"))
v.text = str(float(value))
else:
cell.attrib["t"] = "inlineStr"
is_el = ET.SubElement(cell, qname("is"))
t = ET.SubElement(is_el, qname("t"))
t.text = "" if value is None else str(value)
def patch_sheet_xml(xml_bytes: bytes, updates: Dict[str, Any]) -> bytes:
root = ET.fromstring(xml_bytes)
for ref, value in sorted(updates.items(), key=lambda item: cell_sort_key(item[0])):
_, row_num = cell_ref_parts(ref)
row = get_or_create_row(root, row_num)
cell = get_or_create_cell(row, ref)
cell.attrib["r"] = ref
set_cell_value(cell, value)
return ET.tostring(root, encoding="utf-8", xml_declaration=True)
def materialize_workbook(template: Path, output: Path, updates_by_sheet: Dict[str, Dict[str, Any]]) -> None:
sheet_paths = xml_sheet_paths(template)
missing = [sheet for sheet in updates_by_sheet if sheet not in sheet_paths]
if missing:
raise ValueError(f"Template missing sheets required for updates: {missing}")
with zipfile.ZipFile(template, "r") as zin, zipfile.ZipFile(output, "w", zipfile.ZIP_DEFLATED) as zout:
for item in zin.infolist():
data = zin.read(item.filename)
for sheet, updates in updates_by_sheet.items():
if item.filename == sheet_paths[sheet]:
data = patch_sheet_xml(data, updates)
break
zout.writestr(item, data)
def inspect_workbook(path: Path) -> Dict[str, Any]:
with zipfile.ZipFile(path) as zf:
names = set(zf.namelist())
sheet_paths = xml_sheet_paths(path)
missing = [sheet for sheet in REQUIRED_SHEETS if sheet not in sheet_paths]
formula_count = 0
formula_sheets = 0
for sheet_path in sheet_paths.values():
root = ET.fromstring(zf.read(sheet_path))
count = len(root.findall(f".//{{{NS_MAIN}}}f"))
formula_count += count
if count:
formula_sheets += 1
external_links = [name for name in names if name.startswith("xl/externalLinks/")]
has_styles = "xl/styles.xml" in names and len(zf.read("xl/styles.xml")) > 100
return {
"sheet_names": list(sheet_paths.keys()),
"required_sheets_present": not missing,
"missing_required_sheets": missing,
"formula_count": formula_count,
"formula_sheets": formula_sheets,
"has_styles": has_styles,
"external_links": external_links,
"no_external_links": not external_links,
}
def build(plan_path: Path, output_dir: Path, template: Path) -> Dict[str, Any]:
plan = load_json(plan_path)
validation_errors = validate_plan_structure(plan)
if validation_errors:
raise ValueError("Plan validation failed before workbook materialization: " + "; ".join(validation_errors))
output_dir.mkdir(parents=True, exist_ok=True)
workbook_path = output_dir / OUTPUT_WORKBOOK
run_log_path = output_dir / OUTPUT_RUN_LOG
warnings: List[str] = []
missing_inputs: List[str] = []
updates_by_sheet = {
"Executive Summary": build_executive_summary_updates(plan),
"Control Panel": build_control_panel_updates(plan, missing_inputs, warnings),
"Historical Financials": build_historical_updates(plan, missing_inputs, warnings),
"Revenue Build": build_revenue_updates(plan),
"Source Notes": build_source_notes_updates(plan),
}
materialize_workbook(template, workbook_path, updates_by_sheet)
inspection = inspect_workbook(workbook_path)
hard_failures: List[str] = []
if not inspection["required_sheets_present"]:
hard_failures.append("banker_formula_workbook_missing_required_sheets")
if inspection["formula_count"] < 100:
hard_failures.append("banker_formula_workbook_formula_count_below_threshold")
if not inspection["has_styles"]:
hard_failures.append("banker_formula_workbook_styles_missing")
if not inspection["no_external_links"]:
hard_failures.append("banker_formula_workbook_contains_external_links")
model_status = "not-decision-ready" if hard_failures else ("screen-grade" if screen_grade_required(plan) else "senior-review-ready")
run_log = {
"model_status": model_status,
"workbook_mode": "banker_formula_workbook",
"artifact_level": "banker_formula_workbook_template_materializer",
"template_path": str(template),
"output_paths": {
"banker_formula_workbook": str(workbook_path),
"model_citations_json": str(output_dir / "model_citations.json"),
"banker_formula_workbook_run_log": str(run_log_path),
"manifest": str(output_dir / OUTPUT_MANIFEST),
},
"partial_context_warning": PARTIAL_CONTEXT_WARNING if screen_grade_required(plan) else None,
"missing_inputs": sorted(set(missing_inputs)),
"warnings": sorted(set(warnings)),
"hard_failures": hard_failures,
"workbook_inspection": inspection,
"formula_mode_limitations": [
"Preserves and populates the bundled DCF formula workbook template.",
"Does not synthesize new tabs, formulas, or workbook architecture beyond the shipped template.",
"The template has six forecast columns; plan vectors are extended or truncated for formula-mode compatibility.",
"Excel or compatible spreadsheet software should recalculate formulas after opening.",
],
}
model_citations = write_model_citations_for_workbook(output_dir / "model_citations.json", workbook_path)
run_log["model_citations_path"] = str(output_dir / "model_citations.json")
run_log["model_citation_count"] = len(model_citations)
run_log_path.write_text(json.dumps(run_log, indent=2), encoding="utf-8")
write_output_manifest(output_dir, run_log)
return run_log
def main() -> int:
parser = argparse.ArgumentParser(description="Materialize the DCF banker formula workbook template from plan.json.")
parser.add_argument("plan_json", type=Path)
parser.add_argument("--output-dir", type=Path, default=None, help="Output directory for workbook, run log, and manifest. Defaults to ./output.")
parser.add_argument("--template", type=Path, default=DEFAULT_TEMPLATE)
parser.add_argument("--json-run-log", "--json", dest="json_run_log", action="store_true", help="Print machine-readable run summary to stdout. Default stdout is human-readable.")
parser.add_argument("--quiet-human-output", action="store_true", help="Suppress default human-readable stdout.")
args = parser.parse_args()
output_dir = args.output_dir if args.output_dir else Path.cwd() / "output"
try:
run_log = build(args.plan_json, output_dir, args.template)
except Exception as exc:
print(f"banker_formula_workbook materialization FAILED: {exc}", file=sys.stderr)
return 1
summary = {
"model_status": run_log["model_status"],
"workbook_mode": run_log["workbook_mode"],
"output_paths": run_log["output_paths"],
"hard_failures": run_log["hard_failures"],
"warnings": run_log["warnings"],
"missing_inputs": run_log["missing_inputs"],
"model_citations_path": run_log.get("model_citations_path"),
"model_citation_count": run_log.get("model_citation_count", 0),
}
if args.json_run_log:
print(json.dumps(summary, indent=2))
elif not args.quiet_human_output:
print(f"DCF formula workbook complete: {run_log['model_status']}")
print(f"Open workbook: {run_log['output_paths']['banker_formula_workbook']}")
print(f"Manifest: {run_log['output_paths']['manifest']}")
print(f"Model citations: {run_log.get('model_citations_path')} ({run_log.get('model_citation_count', 0)} records)")
if run_log["warnings"]:
print(f"Warnings: {len(run_log['warnings'])}")
if run_log["hard_failures"]:
print(f"Hard failures: {len(run_log['hard_failures'])}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
SHA-256: d3d6cb3ccfd140249c6b66fd6f5e322e50244d45ef636e894b15b6d066d95d0a