r/palladium
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: restructure repo into core/app modules with per-study folders

Browse Source 
Reorganize Palladium codebase into a modular architecture with `core/`
shared logic and `app/` Streamlit UI, separating per-study assets into
`studies/YYYYMM_<Vendor>/` folders containing notebooks, seed data, and
configuration. Update README to reflect new structure, add `.gitignore`
entries for `.env` and study exports, and refresh component documentation.
This commit is contained in:
Robert Helewka
2026-05-20 22:28:12 -04:00
parent a6f3ee3676
commit a2420ed692
52 changed files with 35300 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
core/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""Palladium core — shared TEI client, calculations, export, helpers."""
__version__ = "0.1.0"

31
core/calculations/__init__.py Normal file
View File

@@ -0,0 +1,31 @@
"""Pure-python TEI financial math."""
from core.calculations.npv import (
discount_factor,
npv,
present_value,
present_value_series,
)
from core.calculations.payback import payback_months, payback_years
from core.calculations.roi import roi, roi_percentage
from core.calculations.scenarios import (
SCENARIOS,
apply_scenario,
risk_adjust_benefit,
risk_adjust_cost,
)
__all__ = [
"SCENARIOS",
"apply_scenario",
"discount_factor",
"npv",
"payback_months",
"payback_years",
"present_value",
"present_value_series",
"risk_adjust_benefit",
"risk_adjust_cost",
"roi",
"roi_percentage",
]

78
core/calculations/npv.py Normal file
View File

@@ -0,0 +1,78 @@
"""
Net Present Value and discounting.
Convention (matches the Forrester TEI methodology used in the Amazon Connect
study):
* The *Initial* investment (column "Initial" in TEI tables) is **not**
discounted — it occurs at time zero.
* Year-N cash flows are discounted at the end of the year:
``PV = CF_n / (1 + r) ** n`` for ``n = 1, 2, …``.
That matches the PDF note: *"The initial investment column contains costs
incurred at 'time 0' or at the beginning of Year 1 that are not discounted.
All other cash flows are discounted using the discount rate at the end of
the year."*
"""
from __future__ import annotations
from collections.abc import Iterable
def discount_factor(year: int, discount_rate: float) -> float:
"""Return ``1 / (1 + r) ** year``. Year 0 → 1.0 (no discount)."""
if year < 0:
raise ValueError("year must be >= 0")
return 1.0 / ((1.0 + discount_rate) ** year)
def present_value(amount: float, year: int, discount_rate: float) -> float:
"""Discount ``amount`` from end-of-year ``year`` to present."""
return amount * discount_factor(year, discount_rate)
def present_value_series(
cashflows: Iterable[float],
discount_rate: float,
start_year: int = 1,
) -> float:
"""
Sum the present value of a stream of year-end cashflows.
Args:
cashflows: iterable of year-N amounts (Y1, Y2, …).
discount_rate: e.g. 0.10 for 10%.
start_year: year of the first element. Default 1 (skip year-0).
"""
total = 0.0
for offset, cf in enumerate(cashflows):
total += present_value(cf, start_year + offset, discount_rate)
return total
def npv(
cashflows: Iterable[float],
discount_rate: float,
initial: float = 0.0,
) -> float:
"""
Net Present Value.
Args:
cashflows: year-end cashflows for Year 1, Year 2, …
discount_rate: e.g. 0.10
initial: undiscounted year-0 cashflow (negative for a cost,
positive for a one-off benefit).
Returns:
``initial + Σ CF_n / (1 + r)^n`` for ``n = 1..len(cashflows)``.
Example::
>>> # Amazon Connect total benefits PV ≈ $101.7M
>>> benefits = [27_279_019, 40_333_658, 57_983_794]
>>> round(npv(benefits, 0.10) / 1_000_000, 1)
101.7
"""
return initial + present_value_series(cashflows, discount_rate, start_year=1)

63
core/calculations/payback.py Normal file
View File

@@ -0,0 +1,63 @@
"""
Payback period.
Linear interpolation within the crossing year: returns the moment when the
running net cashflow first turns non-negative.
Inputs are *risk-adjusted, undiscounted* cashflows by convention (TEI shows
"<6 months" payback for the Amazon Connect composite using the risk-adjusted
nominal cashflows from the Cash Flow Analysis table).
"""
from __future__ import annotations
from collections.abc import Iterable
def payback_years(
initial_cost: float,
yearly_net_benefits: Iterable[float],
) -> float | None:
"""
Years until cumulative net benefits cover the initial cost.
Args:
initial_cost: positive number — undiscounted year-0 outlay.
yearly_net_benefits: sequence of (benefits − costs) per year, Y1+.
Returns:
Float number of years, or ``None`` if payback is never reached
within the supplied horizon.
Example::
>>> # Amazon Connect: initial cost $1.196M, Y1 net $19.998M
>>> # → ~0.06 years ≈ 0.7 months — well under 6 months.
>>> round(payback_years(1_196_250, [19_997_953, 31_562_489, 47_443_905]), 3)
0.06
"""
remaining = float(initial_cost)
if remaining <= 0:
return 0.0
cumulative_year = 0
for cf in yearly_net_benefits:
cumulative_year += 1
cf = float(cf)
if cf <= 0:
remaining += -cf # net loss this year increases the gap
continue
if cf >= remaining:
# Crossing happens partway through this year.
fraction = remaining / cf
return (cumulative_year - 1) + fraction
remaining -= cf
return None
def payback_months(
initial_cost: float,
yearly_net_benefits: Iterable[float],
) -> float | None:
"""Same as :func:`payback_years`, expressed in months."""
yrs = payback_years(initial_cost, yearly_net_benefits)
return yrs * 12.0 if yrs is not None else None

27
core/calculations/roi.py Normal file
View File

@@ -0,0 +1,27 @@
"""Return on Investment."""
from __future__ import annotations
def roi(benefits_pv: float, costs_pv: float) -> float:
"""
Return on Investment as a fraction.
``ROI = (Benefits − Costs) / Costs``
Costs here are expressed as a positive present-value amount (the absolute
cost). Returns ``0.0`` when costs are zero (rather than dividing by zero).
Example::
>>> round(roi(101_696_791, 22_983_076), 2) # Amazon Connect: 342%
3.42
"""
if costs_pv <= 0:
return 0.0
return (benefits_pv - costs_pv) / costs_pv
def roi_percentage(benefits_pv: float, costs_pv: float) -> float:
"""ROI as a percentage (e.g. 342.0 for 342%)."""
return roi(benefits_pv, costs_pv) * 100.0

128
core/calculations/scenarios.py Normal file
View File

@@ -0,0 +1,128 @@
"""
Scenario modelling and risk adjustment.
Forrester TEI applies a *downward* risk adjustment to benefits (subtract
``risk_factor × benefit``) and an *upward* adjustment to costs (add
``risk_factor × cost``). Scenarios scale both adoption (cashflow magnitude)
and uncertainty (risk factor).
Default scenario multipliers are sensible starting points. Override per
study by editing the study's ``config.py`` and passing a custom dict to
:func:`apply_scenario`.
"""
from __future__ import annotations
from collections.abc import Iterable
from copy import deepcopy
from typing import Any
#: Default scenario multipliers used by Palladium notebooks.
#:
#: * ``adoption`` scales nominal annual values up or down.
#: * ``risk_delta`` is *added* to the benefit's risk_adjustment factor and
#: *subtracted* from the cost's risk_adjustment factor (conservative
#: = more uncertainty on benefits, less padding on costs).
SCENARIOS: dict[str, dict[str, float]] = {
"conservative": {"adoption": 0.80, "risk_delta": 0.10},
"moderate": {"adoption": 1.00, "risk_delta": 0.00},
"aggressive": {"adoption": 1.15, "risk_delta": -0.05},
}
def risk_adjust_benefit(amount: float, risk_factor: float) -> float:
"""
Apply a downward risk adjustment to a benefit.
``adjusted = amount × (1 − risk_factor)``.
``risk_factor`` is clamped to ``[0, 1]``.
"""
rf = max(0.0, min(1.0, float(risk_factor)))
return amount * (1.0 - rf)
def risk_adjust_cost(amount: float, risk_factor: float) -> float:
"""
Apply an upward risk adjustment to a cost.
``adjusted = amount × (1 + risk_factor)``.
``risk_factor`` is clamped to ``[0, 1]``.
"""
rf = max(0.0, min(1.0, float(risk_factor)))
return amount * (1.0 + rf)
def _scale_yearly(values: Iterable[float], factor: float) -> list[float]:
return [float(v) * factor for v in values]
def apply_scenario(
items: list[dict],
scenario: str = "moderate",
*,
multipliers: dict[str, dict[str, float]] | None = None,
table: str | None = None,
) -> list[dict]:
"""
Return a deep-copied list of value-rows with scenario adjustments applied.
Each item is expected to have:
- ``table`` (``'benefits'`` or ``'costs'``) — required for sign of
risk_delta. If absent, defaults to ``benefits`` unless ``table=``
is passed explicitly.
- ``year_values`` (dict of year-string → float) **or** a scalar
``value``.
- ``risk_adjustment`` (optional float).
- ``initial`` (optional, costs only) — scaled by adoption.
Args:
items: rows shaped like the ``_normalize_value`` output of
:class:`core.tei_client.TEIClient`.
scenario: key into ``multipliers`` (default ``SCENARIOS``).
multipliers: override map. Same shape as ``SCENARIOS``.
table: force a table when items lack one.
Returns:
A new list — original items are not mutated.
"""
cfg = (multipliers or SCENARIOS).get(scenario)
if cfg is None:
raise KeyError(f"Unknown scenario: {scenario!r}")
adoption = float(cfg.get("adoption", 1.0))
risk_delta = float(cfg.get("risk_delta", 0.0))
out: list[dict] = []
for raw in items:
item: dict[str, Any] = deepcopy(raw)
item_table = item.get("table") or table or "benefits"
item["table"] = item_table
# Adoption scaling
if "year_values" in item and isinstance(item["year_values"], dict):
item["year_values"] = {
k: float(v) * adoption for k, v in item["year_values"].items()
}
if "value" in item and item["value"] is not None:
try:
item["value"] = float(item["value"]) * adoption
except (TypeError, ValueError):
pass
if "initial" in item and item["initial"] is not None:
try:
item["initial"] = float(item["initial"]) * adoption
except (TypeError, ValueError):
pass
# Risk adjustment delta
ra = item.get("risk_adjustment")
if ra is None:
ra = 0.0
if item_table == "benefits":
new_ra = float(ra) + risk_delta
else: # costs: adverse scenario should *raise* costs less, so subtract
new_ra = float(ra) - risk_delta
item["risk_adjustment"] = max(0.0, min(1.0, new_ra))
out.append(item)
return out

1
core/cli/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Palladium CLI package — invoked via ``python -m palladium``."""

229
core/cli/main.py Normal file
View File

@@ -0,0 +1,229 @@
"""
Palladium CLI implementation.
Subcommands::
palladium test # Verify Athena connectivity
palladium list # List TEI tool instances
palladium reports # List TEI report templates
palladium summary <public_id> # Print a tool's financial summary
palladium calculate <public_id> # Trigger /calculate
palladium export <public_id> -o file # Save export envelope as JSON
The CLI is invoked via ``python -m palladium`` (root shim) which calls
:func:`main` here.
"""
from __future__ import annotations
import argparse
import json
import logging
import sys
from collections.abc import Sequence
from core.export import build_report_data, write_report_data
from core.tei_client import AthenaAPIError, TEIClient
def _configure_logging(verbosity: int) -> None:
level = logging.WARNING
if verbosity == 1:
level = logging.INFO
elif verbosity >= 2:
level = logging.DEBUG
logging.basicConfig(
level=level, format="%(asctime)s %(levelname)s %(name)s: %(message)s"
)
def _build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="palladium",
description="Palladium — TEI Calculator CLI for Athena.",
)
p.add_argument("-v", "--verbose", action="count", default=0)
sub = p.add_subparsers(dest="command", required=True)
sub.add_parser("test", help="Verify Athena API connectivity")
sub.add_parser("list", help="List TEI tool instances")
sub.add_parser("reports", help="List TEI report templates")
s_summary = sub.add_parser("summary", help="Print a tool's financial summary")
s_summary.add_argument("public_id")
s_calc = sub.add_parser("calculate", help="Trigger calculation for a tool")
s_calc.add_argument("public_id")
s_export = sub.add_parser("export", help="Export a tool's report data as JSON")
s_export.add_argument("public_id")
s_export.add_argument(
"-o",
"--output",
default="-",
help="Output path (default: stdout)",
)
s_export.add_argument(
"--no-scenarios",
action="store_true",
help="Skip computing conservative/moderate/aggressive scenarios.",
)
s_export.add_argument(
"--study-slug",
default=None,
help="Optional study identifier to embed in metadata.",
)
return p
def _print_table(rows: list[dict], columns: list[tuple[str, str]]) -> None:
"""Tiny pure-stdlib pretty-printer."""
widths = [len(label) for _, label in columns]
formatted: list[list[str]] = []
for r in rows:
rec = [str(r.get(key, "") or "") for key, _ in columns]
formatted.append(rec)
for i, val in enumerate(rec):
if len(val) > widths[i]:
widths[i] = len(val)
fmt = " ".join(f"{{:<{w}}}" for w in widths)
print(fmt.format(*[label for _, label in columns]))
print(fmt.format(*["-" * w for w in widths]))
for rec in formatted:
print(fmt.format(*rec))
def cmd_test(client: TEIClient, args) -> int:
result = client.test_connection()
print(json.dumps(result, indent=2))
return 0 if result.get("status") == "ok" else 1
def cmd_list(client: TEIClient, args) -> int:
tools = client.list_tools()
if not tools:
print("(no TEI tools)")
return 0
rows = []
for t in tools:
report = t.get("report")
if isinstance(report, dict):
report_name = report.get("name", "")
else:
report_name = report or ""
rows.append(
{
"id": t.get("id", ""),
"name": t.get("name", ""),
"report": report_name,
"status": t.get("status", ""),
"version": t.get("current_version", ""),
"modified": t.get("modified_date", ""),
}
)
_print_table(
rows,
[
("id", "PUBLIC_ID"),
("name", "NAME"),
("report", "REPORT"),
("status", "STATUS"),
("version", "VER"),
("modified", "MODIFIED"),
],
)
return 0
def cmd_reports(client: TEIClient, args) -> int:
reports = client.list_reports()
if not reports:
print("(no reports)")
return 0
rows = [
{
"id": r.get("id", ""),
"name": r.get("name", ""),
"vendor": r.get("vendor", ""),
"version": r.get("version", ""),
"fields": r.get("field_count", 0),
"instances": r.get("instance_count", 0),
"status": r.get("status", ""),
}
for r in reports
]
_print_table(
rows,
[
("id", "PUBLIC_ID"),
("name", "NAME"),
("vendor", "VENDOR"),
("version", "VER"),
("fields", "FIELDS"),
("instances", "TOOLS"),
("status", "STATUS"),
],
)
return 0
def cmd_summary(client: TEIClient, args) -> int:
client.print_summary(args.public_id)
return 0
def cmd_calculate(client: TEIClient, args) -> int:
client.calculate(args.public_id)
print(f"Recalculated {args.public_id}.")
client.print_summary(args.public_id)
return 0
def cmd_export(client: TEIClient, args) -> int:
envelope = build_report_data(
client,
args.public_id,
include_scenarios=not args.no_scenarios,
study_slug=args.study_slug,
)
if args.output in ("-", ""):
json.dump(envelope, sys.stdout, indent=2, default=str)
sys.stdout.write("\n")
else:
path = write_report_data(envelope, args.output)
print(f"Wrote {path}")
return 0
COMMANDS = {
"test": cmd_test,
"list": cmd_list,
"reports": cmd_reports,
"summary": cmd_summary,
"calculate": cmd_calculate,
"export": cmd_export,
}
def main(argv: Sequence[str] | None = None) -> int:
parser = _build_parser()
args = parser.parse_args(argv)
_configure_logging(args.verbose)
try:
client = TEIClient()
except ValueError as e:
print(f"error: {e}", file=sys.stderr)
return 2
handler = COMMANDS[args.command]
try:
return handler(client, args)
except AthenaAPIError as e:
print(f"Athena API error {e.status_code}: {e.detail}", file=sys.stderr)
return 1
if __name__ == "__main__": # pragma: no cover
raise SystemExit(main())

5
core/export/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Export utilities — build the LLM-ready JSON envelope."""
from core.export.report_data import build_report_data, write_report_data
__all__ = ["build_report_data", "write_report_data"]

224
core/export/report_data.py Normal file
View File

@@ -0,0 +1,224 @@
"""
Build the structured JSON consumed by the report pipeline.
The Athena ``GET /tools/{public_id}/export/`` endpoint already returns most
of what we need; this module:
1. Calls the export endpoint.
2. Optionally augments it with locally computed scenario analysis
(conservative / moderate / aggressive).
3. Stamps Palladium metadata (export timestamp, study slug, generator).
4. Serializes to a stable JSON file that html2docx / Peitho can consume.
"""
from __future__ import annotations
import json
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from core import __version__
from core.calculations import (
apply_scenario,
npv,
payback_months,
risk_adjust_benefit,
risk_adjust_cost,
roi_percentage,
)
def _split_by_table(values: list[dict]) -> tuple[list[dict], list[dict]]:
benefits = [v for v in values if v.get("table") == "benefits"]
costs = [v for v in values if v.get("table") == "costs"]
return benefits, costs
def _yearly_totals(items: list[dict], analysis_years: int) -> list[float]:
totals = [0.0] * analysis_years
for it in items:
yv = it.get("year_values") or {}
for k, v in yv.items():
try:
year = int(k)
except (TypeError, ValueError):
continue
if 1 <= year <= analysis_years:
totals[year - 1] += float(v or 0)
return totals
def _initial_total(items: list[dict]) -> float:
return sum(float(it.get("initial") or 0) for it in items)
def _risk_adjusted(items: list[dict], for_table: str) -> list[dict]:
out: list[dict] = []
for it in items:
rf = float(it.get("risk_adjustment") or 0.0)
adj_year_values: dict[str, float] = {}
for k, v in (it.get("year_values") or {}).items():
v = float(v or 0)
adj_year_values[str(k)] = (
risk_adjust_benefit(v, rf)
if for_table == "benefits"
else risk_adjust_cost(v, rf)
)
adj = dict(it)
adj["year_values"] = adj_year_values
if "initial" in adj and adj["initial"] is not None:
adj["initial"] = (
risk_adjust_cost(float(adj["initial"]), rf)
if for_table == "costs"
else float(adj["initial"])
)
out.append(adj)
return out
def _compute_summary(
benefits: list[dict],
costs: list[dict],
discount_rate: float,
analysis_years: int,
) -> dict[str, Any]:
benefits_ra = _risk_adjusted(benefits, "benefits")
costs_ra = _risk_adjusted(costs, "costs")
benefits_yr = _yearly_totals(benefits_ra, analysis_years)
costs_yr = _yearly_totals(costs_ra, analysis_years)
initial_costs = _initial_total(costs_ra)
benefits_pv = npv(benefits_yr, discount_rate)
costs_pv = npv(costs_yr, discount_rate, initial=initial_costs)
nominal_benefits = sum(benefits_yr)
nominal_costs = sum(costs_yr) + initial_costs
net_yearly = [b - c for b, c in zip(benefits_yr, costs_yr, strict=False)]
pb = payback_months(initial_costs, net_yearly)
return {
"discount_rate": discount_rate,
"analysis_years": analysis_years,
"total_benefits_nominal": nominal_benefits,
"total_benefits_pv": benefits_pv,
"total_costs_nominal": nominal_costs,
"total_costs_pv": costs_pv,
"npv": benefits_pv - costs_pv,
"roi_pct": roi_percentage(benefits_pv, costs_pv),
"payback_months": pb,
"yearly_breakdown": [
{
"year": idx + 1,
"benefits": benefits_yr[idx],
"costs": costs_yr[idx],
"net": net_yearly[idx],
"cumulative_net": sum(net_yearly[: idx + 1]) - initial_costs,
}
for idx in range(analysis_years)
],
"initial_costs": initial_costs,
}
def build_report_data(
client,
public_id: str,
*,
include_scenarios: bool = True,
study_slug: str | None = None,
) -> dict[str, Any]:
"""
Build the full export envelope for a TEI tool.
Args:
client: a :class:`core.tei_client.TEIClient` instance.
public_id: the TEI tool's public_id.
include_scenarios: if True, locally compute conservative / moderate /
aggressive summaries and attach them under ``scenarios``.
study_slug: optional human-friendly study identifier (e.g.
``"202602_AmazonConnect"``) — written into ``metadata``.
Returns:
A dict with keys::
{
"metadata": {...}, # client / opportunity / study / generator
"report": {...}, # report template echo
"values": {benefits, costs},
"summary": {...}, # locally recomputed (mirrors Athena)
"athena_export": {...}, # raw payload from Athena (if available)
"scenarios": {...} # optional
}
"""
# Pull everything we need
bundle = client.get_tool_with_data(public_id)
tool = bundle["tool"]
fields = bundle["fields"]
values = bundle["values"]
report_obj = tool.get("report")
if isinstance(report_obj, str):
report = client.get_report(report_obj)
elif isinstance(report_obj, dict):
report = report_obj
else:
report = {}
discount_rate = float(report.get("discount_rate") or 0.10)
analysis_years = int(report.get("analysis_period_years") or 3)
benefits, costs = _split_by_table(values)
summary = _compute_summary(benefits, costs, discount_rate, analysis_years)
try:
athena_export = client.export(public_id)
except Exception as e: # pragma: no cover – best effort
athena_export = {"error": str(e)}
envelope: dict[str, Any] = {
"metadata": {
"study_slug": study_slug or "",
"tool_public_id": public_id,
"tool_name": tool.get("name", ""),
"report_name": report.get("name", ""),
"report_vendor": report.get("vendor", ""),
"report_version": report.get("version", ""),
"report_public_id": report.get("id", ""),
"proposal": tool.get("proposal") or tool.get("opportunity"),
"engagement": tool.get("engagement"),
"generated_at": datetime.now(UTC).isoformat(),
"generator": f"palladium core {__version__}",
},
"report": report,
"fields": fields,
"values": {"benefits": benefits, "costs": costs},
"summary": summary,
"athena_export": athena_export,
}
if include_scenarios:
scenario_results: dict[str, Any] = {}
for scenario_name in ("conservative", "moderate", "aggressive"):
sb = apply_scenario(benefits, scenario_name, table="benefits")
sc = apply_scenario(costs, scenario_name, table="costs")
scenario_results[scenario_name] = _compute_summary(
sb, sc, discount_rate, analysis_years
)
envelope["scenarios"] = scenario_results
return envelope
def write_report_data(
envelope: dict[str, Any],
output_path: str | Path,
*,
indent: int = 2,
) -> Path:
"""Serialize ``envelope`` to ``output_path`` and return the Path."""
path = Path(output_path)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(envelope, indent=indent, default=str))
return path

5
core/notebook_helpers/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Notebook helpers — pandas tables, plotly charts, IPython display."""
from core.notebook_helpers import charts, display, tables
__all__ = ["charts", "display", "tables"]

193
core/notebook_helpers/charts.py Normal file
View File

@@ -0,0 +1,193 @@
"""
Plotly charts for TEI analyses.
Each function returns a ``plotly.graph_objects.Figure`` so callers can
``.show()`` (notebook), pass to ``st.plotly_chart`` (Streamlit), or write to
HTML / image. No styling is hard-coded beyond a neutral default palette.
"""
from __future__ import annotations
from collections.abc import Iterable
import plotly.graph_objects as go
PALETTE = {
"benefits": "#2E7D32", # green
"costs": "#C62828", # red
"net_positive": "#1565C0", # blue
"net_negative": "#C62828",
"cumulative": "#616161", # grey
}
def cashflow_chart(
yearly_breakdown: list[dict],
*,
title: str = "Cash Flow Analysis (Risk-Adjusted)",
initial_cost: float = 0.0,
) -> go.Figure:
"""
Stacked bars of benefits & costs by year + cumulative net line.
Mirrors the chart on page 25 of the Forrester Amazon Connect TEI study.
"""
if not yearly_breakdown:
return go.Figure(layout={"title": title})
years = ["Initial"] + [f"Year {row['year']}" for row in yearly_breakdown]
benefits = [0.0] + [float(row.get("benefits", 0)) for row in yearly_breakdown]
costs = [-float(initial_cost)] + [
-float(row.get("costs", 0)) for row in yearly_breakdown
]
# cumulative_net assumes initial cost has already been deducted
cumulative = [-float(initial_cost)] + [
float(row.get("cumulative_net", 0)) for row in yearly_breakdown
]
fig = go.Figure()
fig.add_bar(
name="Total benefits",
x=years,
y=benefits,
marker_color=PALETTE["benefits"],
)
fig.add_bar(
name="Total costs",
x=years,
y=costs,
marker_color=PALETTE["costs"],
)
fig.add_scatter(
name="Cumulative net benefits",
x=years,
y=cumulative,
mode="lines+markers",
line={"color": PALETTE["cumulative"], "width": 3},
)
fig.update_layout(
title=title,
barmode="relative",
yaxis_tickformat="$,.0f",
legend={"orientation": "h", "y": -0.15},
margin={"l": 40, "r": 20, "t": 60, "b": 40},
)
return fig
def benefits_bar(items: list[dict], *, title: str = "Benefits (Three-Year)") -> go.Figure:
"""Horizontal bars of risk-adjusted three-year totals per benefit."""
labels: list[str] = []
totals: list[float] = []
for it in items:
rf = float(it.get("risk_adjustment") or 0.0)
yv = it.get("year_values") or {}
ra_total = sum(float(v or 0) * (1.0 - rf) for v in yv.values())
labels.append(it.get("label", "") or it.get("field_key", ""))
totals.append(ra_total)
fig = go.Figure(
go.Bar(
x=totals,
y=labels,
orientation="h",
marker_color=PALETTE["benefits"],
text=[f"${t/1_000_000:,.1f}M" for t in totals],
textposition="auto",
)
)
fig.update_layout(
title=title,
xaxis_tickformat="$,.0f",
yaxis={"autorange": "reversed"},
margin={"l": 40, "r": 20, "t": 60, "b": 40},
)
return fig
def cost_breakdown_pie(
items: list[dict], *, title: str = "Cost Breakdown (Three-Year, Risk-Adjusted)"
) -> go.Figure:
"""Pie chart of risk-adjusted costs by category/label."""
labels: list[str] = []
values: list[float] = []
for it in items:
rf = float(it.get("risk_adjustment") or 0.0)
yv = it.get("year_values") or {}
initial = float(it.get("initial") or 0.0)
ra_total = (
initial * (1.0 + rf)
+ sum(float(v or 0) * (1.0 + rf) for v in yv.values())
)
labels.append(it.get("label", "") or it.get("field_key", ""))
values.append(ra_total)
fig = go.Figure(go.Pie(labels=labels, values=values, hole=0.35))
fig.update_layout(title=title, margin={"l": 40, "r": 20, "t": 60, "b": 40})
return fig
def scenario_comparison(scenarios: dict) -> go.Figure:
"""Grouped bars comparing NPV and Costs PV across scenarios."""
keys: list[str] = list(scenarios.keys())
if not keys:
return go.Figure()
benefits = [float(scenarios[k].get("total_benefits_pv") or 0) for k in keys]
costs = [float(scenarios[k].get("total_costs_pv") or 0) for k in keys]
npvs = [float(scenarios[k].get("npv") or 0) for k in keys]
fig = go.Figure()
fig.add_bar(name="Benefits PV", x=keys, y=benefits, marker_color=PALETTE["benefits"])
fig.add_bar(name="Costs PV", x=keys, y=costs, marker_color=PALETTE["costs"])
fig.add_bar(name="NPV", x=keys, y=npvs, marker_color=PALETTE["net_positive"])
fig.update_layout(
title="Scenario Comparison",
barmode="group",
yaxis_tickformat="$,.0f",
legend={"orientation": "h", "y": -0.15},
)
return fig
def cumulative_benefits_chart(
yearly_breakdown: list[dict],
*,
title: str = "Cumulative Net Benefits",
) -> go.Figure:
"""Single-line cumulative net benefits trajectory."""
if not yearly_breakdown:
return go.Figure(layout={"title": title})
years = [f"Year {row['year']}" for row in yearly_breakdown]
cumulative = [float(row.get("cumulative_net", 0)) for row in yearly_breakdown]
fig = go.Figure(
go.Scatter(
x=years,
y=cumulative,
mode="lines+markers",
fill="tozeroy",
line={"color": PALETTE["net_positive"], "width": 3},
)
)
fig.update_layout(title=title, yaxis_tickformat="$,.0f")
return fig
def waterfall(values: Iterable[tuple[str, float]], *, title: str = "TEI Waterfall") -> go.Figure:
"""
Generic waterfall (pass tuples of (label, value)).
Used by 03_business_case to show: Benefits PV → Costs PV → NPV.
"""
labels, amounts = zip(*values, strict=True) if values else ([], [])
measures = ["relative"] * (len(labels) - 1) + ["total"] if labels else []
fig = go.Figure(
go.Waterfall(
x=list(labels),
y=list(amounts),
measure=measures,
text=[f"${v/1_000_000:,.1f}M" for v in amounts],
textposition="outside",
)
)
fig.update_layout(title=title, yaxis_tickformat="$,.0f")
return fig

141
core/notebook_helpers/display.py Normal file
View File

@@ -0,0 +1,141 @@
"""
IPython display helpers — KPI cards, formatted summary blocks, alerts.
Functions are notebook-safe: they fall back to plain ``print`` when running
outside Jupyter / when IPython is not available.
"""
from __future__ import annotations
from typing import Any
try: # pragma: no cover – IPython is a soft dep
from IPython.display import HTML, display
_IPY = True
except Exception: # pragma: no cover
_IPY = False
def _money(value: Any, default: str = "—") -> str:
try:
v = float(value)
except (TypeError, ValueError):
return default
if abs(v) >= 1_000_000_000:
return f"${v/1_000_000_000:,.1f}B"
if abs(v) >= 1_000_000:
return f"${v/1_000_000:,.1f}M"
if abs(v) >= 1_000:
return f"${v/1_000:,.1f}K"
return f"${v:,.0f}"
def _pct(value: Any, default: str = "—") -> str:
try:
v = float(value)
except (TypeError, ValueError):
return default
return f"{v:,.0f}%"
def _months(value: Any, default: str = "N/A") -> str:
if value is None:
return default
try:
v = float(value)
except (TypeError, ValueError):
return default
if v < 6:
return f"<6 months ({v:.1f})"
return f"{v:.1f} months"
def kpi_cards(summary: dict, *, title: str | None = None) -> Any:
"""
Render a row of KPI cards (NPV, ROI, Payback, Benefits PV).
In notebooks, returns/displays inline HTML. Outside IPython, prints a
plain text version.
"""
npv = _money(summary.get("npv"))
roi = _pct(summary.get("roi") or summary.get("roi_pct"))
payback = _months(summary.get("payback_months"))
benefits_pv = _money(summary.get("total_benefits_pv"))
costs_pv = _money(summary.get("total_costs_pv"))
if not _IPY: # pragma: no cover
print(title or "TEI Summary")
print(f" NPV: {npv} ROI: {roi} Payback: {payback}")
print(f" Benefits PV: {benefits_pv} Costs PV: {costs_pv}")
return None
title_html = (
f'<div style="font-size:1.1em;font-weight:600;margin-bottom:6px;color:#444;">'
f"{title}</div>"
if title
else ""
)
card_style = (
"flex:1;min-width:140px;padding:14px 18px;margin:4px;border-radius:8px;"
"background:#f7f9fc;border:1px solid #e3e8ee;"
)
label_style = "font-size:0.78em;color:#6b7480;text-transform:uppercase;letter-spacing:0.04em;"
value_style = "font-size:1.6em;font-weight:600;color:#1a2540;margin-top:4px;"
cards = [
("NPV", npv),
("ROI", roi),
("Payback", payback),
("Benefits PV", benefits_pv),
("Costs PV", costs_pv),
]
cards_html = "".join(
f'<div style="{card_style}">'
f'<div style="{label_style}">{label}</div>'
f'<div style="{value_style}">{value}</div>'
f"</div>"
for label, value in cards
)
html = (
f'<div>{title_html}'
f'<div style="display:flex;flex-wrap:wrap;align-items:stretch;">{cards_html}</div>'
f"</div>"
)
return display(HTML(html))
def summary_panel(summary: dict, *, title: str = "TEI Financial Summary") -> None:
"""Plain-text bordered summary block (mirrors the PDF Cash Flow Analysis)."""
width = 60
print("═" * width)
print(f" {title}")
print("═" * width)
print(f" Benefits PV : {_money(summary.get('total_benefits_pv')):>20}")
print(f" Costs PV : {_money(summary.get('total_costs_pv')):>20}")
print("─" * width)
print(f" NPV : {_money(summary.get('npv')):>20}")
roi_val = summary.get("roi") or summary.get("roi_pct")
print(f" ROI : {_pct(roi_val):>20}")
print(f" Payback : {_months(summary.get('payback_months')):>20}")
print("═" * width)
def alert(text: str, kind: str = "info") -> Any:
"""Coloured alert box for notebooks ('info', 'success', 'warning', 'error')."""
colors = {
"info": ("#0277bd", "#e1f5fe"),
"success": ("#2e7d32", "#e8f5e9"),
"warning": ("#ef6c00", "#fff3e0"),
"error": ("#c62828", "#ffebee"),
}
fg, bg = colors.get(kind, colors["info"])
if not _IPY: # pragma: no cover
print(f"[{kind.upper()}] {text}")
return None
html = (
f'<div style="padding:10px 14px;border-left:4px solid {fg};'
f'background:{bg};color:#1a1a1a;border-radius:4px;margin:6px 0;">'
f"{text}</div>"
)
return display(HTML(html))

127
core/notebook_helpers/tables.py Normal file
View File

@@ -0,0 +1,127 @@
"""
Pandas dataframe builders for benefit / cost / summary tables.
Each builder accepts the value-row dicts produced by
``core.tei_client.TEIClient._normalize_value`` and returns a
nicely-formatted DataFrame for display in notebooks.
"""
from __future__ import annotations
from collections.abc import Iterable
from typing import Any
import pandas as pd
from core.calculations import risk_adjust_benefit, risk_adjust_cost
def _years_in_data(items: Iterable[dict]) -> list[int]:
years: set[int] = set()
for it in items:
for k in (it.get("year_values") or {}):
try:
years.add(int(k))
except (TypeError, ValueError):
continue
return sorted(years)
def benefits_table(items: list[dict]) -> pd.DataFrame:
"""Tidy benefits dataframe with one row per benefit, year columns, totals."""
if not items:
return pd.DataFrame(
columns=["field_key", "label", "category", "risk_adjustment"]
)
years = _years_in_data(items)
rows: list[dict[str, Any]] = []
for it in items:
rf = float(it.get("risk_adjustment") or 0.0)
yv = it.get("year_values") or {}
row = {
"field_key": it.get("field_key", ""),
"label": it.get("label", "") or it.get("field_key", ""),
"category": it.get("category", ""),
"risk_adjustment": rf,
}
nominal_total = 0.0
ra_total = 0.0
for y in years:
v = float(yv.get(str(y)) or 0.0)
ra = risk_adjust_benefit(v, rf)
row[f"Year {y}"] = v
row[f"Year {y} (RA)"] = ra
nominal_total += v
ra_total += ra
row["Total"] = nominal_total
row["Total (RA)"] = ra_total
rows.append(row)
return pd.DataFrame(rows)
def costs_table(items: list[dict]) -> pd.DataFrame:
"""Tidy costs dataframe — adds an Initial column when present."""
if not items:
return pd.DataFrame(
columns=["field_key", "label", "category", "risk_adjustment", "Initial"]
)
years = _years_in_data(items)
rows: list[dict[str, Any]] = []
for it in items:
rf = float(it.get("risk_adjustment") or 0.0)
yv = it.get("year_values") or {}
initial = float(it.get("initial") or 0.0)
row = {
"field_key": it.get("field_key", ""),
"label": it.get("label", "") or it.get("field_key", ""),
"category": it.get("category", ""),
"risk_adjustment": rf,
"Initial": initial,
"Initial (RA)": risk_adjust_cost(initial, rf),
}
nominal_total = initial
ra_total = risk_adjust_cost(initial, rf)
for y in years:
v = float(yv.get(str(y)) or 0.0)
ra = risk_adjust_cost(v, rf)
row[f"Year {y}"] = v
row[f"Year {y} (RA)"] = ra
nominal_total += v
ra_total += ra
row["Total"] = nominal_total
row["Total (RA)"] = ra_total
rows.append(row)
return pd.DataFrame(rows)
def summary_table(summary: dict) -> pd.DataFrame:
"""Single-row summary dataframe of headline KPIs."""
pb = summary.get("payback_months")
pb_str = f"{float(pb):.1f} months" if pb not in (None, "") else "N/A"
data = {
"NPV": [float(summary.get("npv") or 0)],
"ROI %": [float(summary.get("roi") or summary.get("roi_pct") or 0)],
"Payback": [pb_str],
"Benefits PV": [float(summary.get("total_benefits_pv") or 0)],
"Costs PV": [float(summary.get("total_costs_pv") or 0)],
"Discount rate": [float(summary.get("discount_rate") or 0)],
"Analysis years": [int(summary.get("analysis_years") or 0)],
}
return pd.DataFrame(data)
def cashflow_table(summary: dict) -> pd.DataFrame:
"""Per-year cashflow dataframe from a summary's ``yearly_breakdown``."""
yb = summary.get("yearly_breakdown") or []
if not yb:
return pd.DataFrame(columns=["Year", "Benefits", "Costs", "Net", "Cumulative"])
df = pd.DataFrame(yb)
rename = {
"year": "Year",
"benefits": "Benefits",
"costs": "Costs",
"net": "Net",
"cumulative_net": "Cumulative",
}
df = df.rename(columns=rename)
return df

13
core/tei_client/__init__.py Normal file
View File

@@ -0,0 +1,13 @@
"""TEI Client — Athena API wrapper for Palladium."""
from core.tei_client.client import AthenaAPIError, TEIClient
from core.tei_client.models import TEIField, TEIReport, TEISummary, TEIValue
__all__ = [
"AthenaAPIError",
"TEIClient",
"TEIField",
"TEIReport",
"TEISummary",
"TEIValue",
]

563
core/tei_client/client.py Normal file
View File

@@ -0,0 +1,563 @@
"""
TEI Client — Athena API wrapper for Palladium.
Endpoints (per Athena API.yaml, all under ``/api/v1/tei/``):
Reports (templates)
GET /reports/ list_reports
GET /reports/{public_id}/ get_report
GET /reports/{public_id}/fields/ list_fields
PATCH /reports/{public_id}/fields/reorder/ reorder_fields
Tools (instances)
GET /tools/ list_tools
POST /tools/ create_tool
GET /tools/{public_id}/ get_tool
PATCH /tools/{public_id}/ update_tool
DELETE /tools/{public_id}/ delete_tool
Values (data entry)
GET /tools/{public_id}/values/ get_values
PUT /tools/{public_id}/values/ update_values (bulk)
PATCH /tools/{public_id}/values/{field_key}/ patch_value (single)
Calculation & summary
POST /tools/{public_id}/calculate/ calculate
GET /tools/{public_id}/summary/ get_summary
GET /summary/ aggregate_summary
Versions
GET /tools/{public_id}/versions/ list_versions
POST /tools/{public_id}/versions/ save_version
GET /tools/{public_id}/versions/{n}/ get_version
Export
GET /tools/{public_id}/export/ export
Authentication uses the ``Authorization: Api-Key {key}`` header.
"""
from __future__ import annotations
import json
import logging
import os
from datetime import datetime
from typing import Any
import requests
from dotenv import load_dotenv
load_dotenv()
logger = logging.getLogger(__name__)
API_PREFIX = "/api/v1/tei"
class AthenaAPIError(Exception):
"""Raised when Athena returns a non-success response."""
def __init__(self, status_code: int, detail: str, url: str):
self.status_code = status_code
self.detail = detail
self.url = url
super().__init__(f"Athena API {status_code} at {url}: {detail}")
class TEIClient:
"""
Client for Athena's TEI Calculator API.
Wraps every TEI endpoint and provides a few convenience helpers used by
the Palladium notebooks and Streamlit app.
Environment variables (read via python-dotenv):
ATHENA_BASE_URL e.g. https://athena.nttdata.com
ATHENA_API_KEY Api-Key value (admin-issued)
Example::
from core.tei_client import TEIClient
client = TEIClient()
client.test_connection()
for r in client.list_reports():
print(r["name"])
"""
def __init__(
self,
base_url: str | None = None,
api_key: str | None = None,
timeout: int = 30,
):
self.base_url = (base_url or os.getenv("ATHENA_BASE_URL", "")).rstrip("/")
self.api_key = api_key or os.getenv("ATHENA_API_KEY", "")
self.timeout = timeout
if not self.base_url:
raise ValueError(
"ATHENA_BASE_URL is required. Set it in .env or pass base_url."
)
if not self.api_key:
raise ValueError(
"ATHENA_API_KEY is required. Set it in .env or pass api_key."
)
self.session = requests.Session()
self.session.headers.update(
{
"Authorization": f"Api-Key {self.api_key}",
"Content-Type": "application/json",
"Accept": "application/json",
}
)
logger.info("TEIClient initialised for %s", self.base_url)
# ─────────────────────────────────────────────
# Internal HTTP helpers
# ─────────────────────────────────────────────
def _url(self, path: str) -> str:
if not path.startswith("/"):
path = f"/{path}"
return f"{self.base_url}{path}"
def _request(
self,
method: str,
path: str,
params: dict | None = None,
json_data: Any | None = None,
) -> Any:
url = self._url(path)
logger.debug("%s %s", method.upper(), url)
try:
response = self.session.request(
method=method,
url=url,
params=params,
json=json_data,
timeout=self.timeout,
)
except requests.ConnectionError as e:
raise AthenaAPIError(0, f"Connection failed: {e}", url) from e
except requests.Timeout as e:
raise AthenaAPIError(408, "Request timed out", url) from e
if response.status_code >= 400:
try:
payload = response.json()
detail = payload.get("detail") or json.dumps(payload)
except (json.JSONDecodeError, ValueError, AttributeError):
detail = response.text
raise AthenaAPIError(response.status_code, detail, url)
if response.status_code == 204 or not response.content:
return {}
return response.json()
def _get(self, path: str, params: dict | None = None) -> Any:
return self._request("GET", path, params=params)
def _post(self, path: str, data: Any | None = None) -> Any:
return self._request("POST", path, json_data=data)
def _put(self, path: str, data: Any | None = None) -> Any:
return self._request("PUT", path, json_data=data)
def _patch(self, path: str, data: Any | None = None) -> Any:
return self._request("PATCH", path, json_data=data)
def _delete(self, path: str) -> Any:
return self._request("DELETE", path)
def _paginated(self, path: str, params: dict | None = None) -> list[dict]:
"""
Fetch all pages of a paginated list endpoint.
Athena uses the standard DRF page/results envelope::
{"count": N, "next": url|None, "previous": ..., "results": [...]}
"""
out: list[dict] = []
result = self._get(path, params=params)
while True:
if isinstance(result, list):
out.extend(result)
return out
if not isinstance(result, dict):
return out
out.extend(result.get("results", []) or [])
next_url = result.get("next")
if not next_url:
return out
# Follow absolute next URL
try:
result = self.session.get(next_url, timeout=self.timeout).json()
except Exception: # pragma: no cover – defensive
return out
# ─────────────────────────────────────────────
# Connection test
# ─────────────────────────────────────────────
def test_connection(self) -> dict:
"""Verify API connectivity and authentication."""
try:
result = self._get(f"{API_PREFIX}/reports/")
count = (
result.get("count", len(result.get("results", [])))
if isinstance(result, dict)
else len(result)
)
return {
"status": "ok",
"base_url": self.base_url,
"authenticated": True,
"reports_found": count,
"timestamp": datetime.now().isoformat(),
}
except AthenaAPIError as e:
return {
"status": "error",
"base_url": self.base_url,
"authenticated": e.status_code != 401,
"error_code": e.status_code,
"detail": e.detail,
"timestamp": datetime.now().isoformat(),
}
# ─────────────────────────────────────────────
# Reports (templates)
# ─────────────────────────────────────────────
def list_reports(self) -> list[dict]:
"""List all TEI report templates (auto-paginated)."""
return self._paginated(f"{API_PREFIX}/reports/")
def get_report(self, public_id: str) -> dict:
"""Get a TEI report template by its public_id."""
return self._get(f"{API_PREFIX}/reports/{public_id}/")
def list_fields(
self,
report_public_id: str,
table: str | None = None,
) -> list[dict]:
"""
Get field definitions for a report.
Args:
report_public_id: The report template's public_id (12-char short UUID).
table: Optional filter — ``'benefits'`` or ``'costs'``.
Returns a list of field-definition dicts. See ``TEIField.from_dict``
for the expected shape.
"""
params = {"table": table} if table else None
rows = self._paginated(
f"{API_PREFIX}/reports/{report_public_id}/fields/", params=params
)
# Defensive — server-side filter may not be implemented; filter locally.
if table:
rows = [r for r in rows if r.get("table") == table]
rows.sort(key=lambda r: (r.get("table", ""), r.get("sort_order") or 0))
return rows
def create_field(self, report_public_id: str, field: dict) -> dict:
"""Create a new field definition under a report (admin only)."""
return self._post(
f"{API_PREFIX}/reports/{report_public_id}/fields/", data=field
)
def update_field(self, report_public_id: str, field_id: int, **changes) -> dict:
"""Patch one field definition by its integer id."""
return self._patch(
f"{API_PREFIX}/reports/{report_public_id}/fields/{field_id}/",
data=changes,
)
def delete_field(self, report_public_id: str, field_id: int) -> dict:
return self._delete(
f"{API_PREFIX}/reports/{report_public_id}/fields/{field_id}/"
)
def reorder_fields(self, report_public_id: str, field_ids: list[int]) -> dict:
"""Bulk-reorder fields. Spec: PATCH /reports/{id}/fields/reorder/."""
return self._patch(
f"{API_PREFIX}/reports/{report_public_id}/fields/reorder/",
data={"field_ids": field_ids},
)
# ─────────────────────────────────────────────
# Tools (instances)
# ─────────────────────────────────────────────
def list_tools(self) -> list[dict]:
"""List TEI tool instances owned by the current API key."""
return self._paginated(f"{API_PREFIX}/tools/")
def get_tool(self, public_id: str) -> dict:
"""Get a TEI tool instance by public_id."""
return self._get(f"{API_PREFIX}/tools/{public_id}/")
def create_tool(
self,
report_public_id: str,
proposal: int | None = None,
engagement: int | None = None,
name: str | None = None,
status: str = "draft",
) -> dict:
"""
Create a new TEI tool instance from a report template.
Athena scopes a TEI tool to a *Proposal* (which itself belongs to an
Opportunity) and/or an *Engagement*. Pass the integer PK of either or
both to link the tool.
"""
data: dict[str, Any] = {"report": report_public_id, "status": status}
if proposal is not None:
data["proposal"] = proposal
if engagement is not None:
data["engagement"] = engagement
if name:
data["name"] = name
return self._post(f"{API_PREFIX}/tools/", data=data)
def update_tool(
self,
public_id: str,
name: str | None = None,
status: str | None = None,
) -> dict:
"""Update tool metadata. Only ``name`` and ``status`` are mutable."""
data: dict[str, Any] = {}
if name is not None:
data["name"] = name
if status is not None:
data["status"] = status
return self._patch(f"{API_PREFIX}/tools/{public_id}/", data=data)
def delete_tool(self, public_id: str) -> dict:
return self._delete(f"{API_PREFIX}/tools/{public_id}/")
# ─────────────────────────────────────────────
# Values (data entry)
# ─────────────────────────────────────────────
@staticmethod
def _normalize_value(value: dict) -> dict:
"""
Normalize a value-row dict into the shape the API expects.
Accepts any of the following input forms and produces a uniform
wire-format dict::
# annual fields
{"field_key": "A1", "year_1": 100, "year_2": 200, "year_3": 300, ...}
{"field_key": "A1", "year_values": {"1": 100, "2": 200, "3": 300}, ...}
# non-annual scalars
{"field_key": "rate", "value": 0.10, ...}
Returns a dict like::
{"field_key": "A1",
"year_values": {"1": 100.0, "2": 200.0, "3": 300.0},
"risk_adjustment": 0.15,
"notes": "…"}
"""
out: dict[str, Any] = {}
if "field_key" in value:
out["field_key"] = value["field_key"]
elif "field" in value:
out["field_key"] = value["field"]
# Collect annual year_N keys into year_values
year_values: dict[str, float] = {}
if "year_values" in value and isinstance(value["year_values"], dict):
for k, v in value["year_values"].items():
year_values[str(k)] = float(v) if v is not None else 0.0
for key, raw in value.items():
if key.startswith("year_"):
try:
n = int(key.split("_", 1)[1])
except ValueError:
continue
year_values[str(n)] = float(raw) if raw is not None else 0.0
if year_values:
out["year_values"] = year_values
if "value" in value and value["value"] is not None and not year_values:
out["value"] = value["value"]
if value.get("initial") is not None:
out["initial"] = float(value["initial"])
if value.get("risk_adjustment") is not None:
out["risk_adjustment"] = float(value["risk_adjustment"])
if value.get("notes"):
out["notes"] = str(value["notes"])
return out
def get_values(self, public_id: str) -> list[dict]:
"""Get all current field values for a TEI tool instance."""
result = self._get(f"{API_PREFIX}/tools/{public_id}/values/")
if isinstance(result, dict):
# Could be {"values": [...]} envelope, the TEITool wrapper, or a page
if "values" in result and isinstance(result["values"], list):
return result["values"]
if "results" in result and isinstance(result["results"], list):
return result["results"]
return []
if isinstance(result, list):
return result
return []
def update_values(self, public_id: str, values: list[dict]) -> dict:
"""
Bulk-update field values. See ``_normalize_value`` for accepted shapes.
"""
payload = {"values": [self._normalize_value(v) for v in values]}
return self._put(f"{API_PREFIX}/tools/{public_id}/values/", data=payload)
def patch_value(self, public_id: str, field_key: str, **changes) -> dict:
"""
Patch a single field value by its ``field_key``.
Accepts the same shorthand as ``update_values`` (``year_1=…``, etc).
"""
body = self._normalize_value({"field_key": field_key, **changes})
body.pop("field_key", None) # carried in URL
return self._patch(
f"{API_PREFIX}/tools/{public_id}/values/{field_key}/", data=body
)
# ─────────────────────────────────────────────
# Calculation & summary
# ─────────────────────────────────────────────
def calculate(self, public_id: str) -> dict:
"""Trigger server-side calculation; returns the updated summary."""
return self._post(f"{API_PREFIX}/tools/{public_id}/calculate/")
def get_summary(self, public_id: str) -> dict:
"""Return the most-recent summary (404 if never calculated)."""
return self._get(f"{API_PREFIX}/tools/{public_id}/summary/")
def aggregate_summary(self) -> dict:
"""Aggregate NPV across all tools owned by the current API key."""
return self._get(f"{API_PREFIX}/summary/")
# ─────────────────────────────────────────────
# Versions
# ─────────────────────────────────────────────
def list_versions(self, public_id: str) -> list[dict]:
"""List all saved version snapshots for a TEI tool."""
result = self._get(f"{API_PREFIX}/tools/{public_id}/versions/")
if isinstance(result, list):
return result
if isinstance(result, dict):
if "results" in result and isinstance(result["results"], list):
return result["results"]
if "versions" in result and isinstance(result["versions"], list):
return result["versions"]
return []
def save_version(self, public_id: str, note: str = "") -> dict:
"""Snapshot current values + summary as a new version."""
return self._post(
f"{API_PREFIX}/tools/{public_id}/versions/",
data={"note": note},
)
def get_version(self, public_id: str, version_number: int) -> dict:
"""Get a single version's full snapshot."""
return self._get(
f"{API_PREFIX}/tools/{public_id}/versions/{int(version_number)}/"
)
# ─────────────────────────────────────────────
# Export
# ─────────────────────────────────────────────
def export(self, public_id: str) -> dict:
"""
Return the LLM-ready export payload for the report pipeline.
The shape is determined by Athena and consumed by Peitho /
html2docx; Palladium's ``core.export.report_data`` builds on this.
"""
return self._get(f"{API_PREFIX}/tools/{public_id}/export/")
# ─────────────────────────────────────────────
# Convenience
# ─────────────────────────────────────────────
def get_benefits(self, public_id: str) -> list[dict]:
"""Return only benefit-table values (table='benefits')."""
return [v for v in self.get_values(public_id) if v.get("table") == "benefits"]
def get_costs(self, public_id: str) -> list[dict]:
"""Return only cost-table values (table='costs')."""
return [v for v in self.get_values(public_id) if v.get("table") == "costs"]
def get_tool_with_data(self, public_id: str) -> dict:
"""
Bundle a tool, its field definitions, current values, and summary.
Convenience for notebook initialisation. The summary is allowed to
404 (returned as ``None``) when the tool has never been calculated.
"""
tool = self.get_tool(public_id)
report_pid = tool.get("report")
if isinstance(report_pid, dict):
report_pid = report_pid.get("id") or report_pid.get("public_id")
fields = self.list_fields(report_pid) if report_pid else []
values = self.get_values(public_id)
try:
summary = self.get_summary(public_id)
except AthenaAPIError as e:
if e.status_code == 404:
summary = None
else:
raise
return {
"tool": tool,
"fields": fields,
"values": values,
"summary": summary,
}
# ─────────────────────────────────────────────
# Display
# ─────────────────────────────────────────────
def __repr__(self) -> str: # pragma: no cover
return f"TEIClient(base_url='{self.base_url}')"
def print_summary(self, public_id: str) -> None:
"""Pretty-print a financial summary block for notebooks/REPL."""
s = self.get_summary(public_id)
def _f(v: Any, default: float = 0.0) -> float:
try:
return float(v) if v is not None else default
except (TypeError, ValueError):
return default
print("═" * 56)
print(" TEI Financial Summary")
print("═" * 56)
print(f" Total Benefits (PV): ${_f(s.get('total_benefits_pv')):>16,.0f}")
print(f" Total Costs (PV): ${_f(s.get('total_costs_pv')):>16,.0f}")
print("─" * 56)
print(f" Net Present Value: ${_f(s.get('npv')):>16,.0f}")
print(f" ROI: {_f(s.get('roi')):>15,.0f}%")
payback = s.get("payback_months")
payback_str = f"{_f(payback):.1f} months" if payback is not None else "N/A"
print(f" Payback: {payback_str:>17}")
print("═" * 56)

194
core/tei_client/models.py Normal file
View File

@@ -0,0 +1,194 @@
"""
Lightweight dataclasses for TEI API responses.
These are *optional* — the client returns raw dicts. Use these when you want
attribute access or IDE help in notebooks.
>>> from core.tei_client import TEIClient, TEIReport
>>> raw = TEIClient().get_report("abc123")
>>> report = TEIReport.from_dict(raw)
>>> report.discount_rate
0.10
"""
from __future__ import annotations
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Any
def _as_float(value: Any, default: float = 0.0) -> float:
"""Coerce Decimal/str/None into float."""
if value is None or value == "":
return default
if isinstance(value, Decimal):
return float(value)
try:
return float(value)
except (TypeError, ValueError):
return default
@dataclass
class TEIReport:
"""A TEI report template (model definition)."""
id: str # public_id
name: str
vendor: str
version: str
description: str = ""
analysis_period_years: int = 3
discount_rate: float = 0.10
status: str = "active"
field_count: int = 0
instance_count: int = 0
created_at: str = ""
updated_at: str = ""
@classmethod
def from_dict(cls, data: dict) -> TEIReport:
return cls(
id=str(data.get("id", "")),
name=data.get("name", ""),
vendor=data.get("vendor", ""),
version=data.get("version", ""),
description=data.get("description", "") or "",
analysis_period_years=int(data.get("analysis_period_years") or 3),
discount_rate=_as_float(data.get("discount_rate"), 0.10),
status=data.get("status", "active"),
field_count=int(data.get("field_count") or 0),
instance_count=int(data.get("instance_count") or 0),
created_at=data.get("created_at", "") or "",
updated_at=data.get("updated_at", "") or "",
)
@dataclass
class TEIField:
"""A field definition belonging to a TEI report template."""
id: int
table: str # 'benefits' | 'costs'
field_key: str
label: str
field_type: str # currency | percentage | integer | decimal | text
description: str = ""
category: str = ""
default_value: str = ""
is_annual: bool = True
risk_adjustment: float | None = None
sort_order: int = 0
is_required: bool = False
source_notes: str = ""
@classmethod
def from_dict(cls, data: dict) -> TEIField:
ra = data.get("risk_adjustment")
return cls(
id=int(data.get("id") or 0),
table=data.get("table", "benefits"),
field_key=data.get("field_key", ""),
label=data.get("label", ""),
field_type=data.get("field_type", "decimal"),
description=data.get("description", "") or "",
category=data.get("category", "") or "",
default_value=data.get("default_value", "") or "",
is_annual=bool(data.get("is_annual", True)),
risk_adjustment=_as_float(ra) if ra is not None else None,
sort_order=int(data.get("sort_order") or 0),
is_required=bool(data.get("is_required", False)),
source_notes=data.get("source_notes", "") or "",
)
@dataclass
class TEIValue:
"""
A field value for a specific TEI tool instance.
The exact wire format is not fully pinned in the OpenAPI spec; we use a
convention that the client `_normalize_value` helper builds:
- annual fields: {field_key, year_values: {"1": ..., "2": ...},
risk_adjustment, notes}
- non-annual scalar: {field_key, value, risk_adjustment, notes}
"""
field_key: str
year_values: dict[str, float] = field(default_factory=dict)
value: float | None = None
risk_adjustment: float | None = None
notes: str = ""
@classmethod
def from_dict(cls, data: dict) -> TEIValue:
ra = data.get("risk_adjustment")
yv_raw = data.get("year_values") or {}
year_values = {str(k): _as_float(v) for k, v in yv_raw.items()}
v = data.get("value")
return cls(
field_key=data.get("field_key", ""),
year_values=year_values,
value=_as_float(v) if v is not None else None,
risk_adjustment=_as_float(ra) if ra is not None else None,
notes=data.get("notes", "") or "",
)
def to_dict(self) -> dict:
out: dict[str, Any] = {"field_key": self.field_key}
if self.year_values:
out["year_values"] = self.year_values
if self.value is not None:
out["value"] = self.value
if self.risk_adjustment is not None:
out["risk_adjustment"] = self.risk_adjustment
if self.notes:
out["notes"] = self.notes
return out
@dataclass
class TEISummary:
"""Calculated financial summary for a TEI tool instance."""
npv: float = 0.0
roi: float = 0.0
payback_months: float | None = None
discount_rate: float = 0.10
analysis_years: int = 3
total_benefits_nominal: float = 0.0
total_benefits_risk_adjusted: float = 0.0
total_benefits_pv: float = 0.0
total_costs_nominal: float = 0.0
total_costs_risk_adjusted: float = 0.0
total_costs_pv: float = 0.0
yearly_breakdown: list[dict] = field(default_factory=list)
category_breakdown: list[dict] = field(default_factory=list)
raw: dict = field(default_factory=dict)
@classmethod
def from_dict(cls, data: dict) -> TEISummary:
return cls(
npv=_as_float(data.get("npv")),
roi=_as_float(data.get("roi")),
payback_months=(
_as_float(data.get("payback_months"))
if data.get("payback_months") is not None
else None
),
discount_rate=_as_float(data.get("discount_rate"), 0.10),
analysis_years=int(data.get("analysis_years") or 3),
total_benefits_nominal=_as_float(data.get("total_benefits_nominal")),
total_benefits_risk_adjusted=_as_float(
data.get("total_benefits_risk_adjusted")
),
total_benefits_pv=_as_float(data.get("total_benefits_pv")),
total_costs_nominal=_as_float(data.get("total_costs_nominal")),
total_costs_risk_adjusted=_as_float(data.get("total_costs_risk_adjusted")),
total_costs_pv=_as_float(data.get("total_costs_pv")),
yearly_breakdown=list(data.get("yearly_breakdown") or []),
category_breakdown=list(data.get("category_breakdown") or []),
raw=dict(data),
)