134 lines
4.7 KiB
Python
134 lines
4.7 KiB
Python
"""One-at-a-time sweeps and tornado data.
|
|
|
|
Data only — the notebook draws the figures. Drivers are addressed by a
|
|
short dotted path so the notebook can name them without reaching into the
|
|
dataclasses itself.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
from dataclasses import replace
|
|
from typing import Any
|
|
|
|
from .model import CalculatorInputs, CalculatorResult, calculate
|
|
|
|
#: Driver name → (container attribute on CalculatorInputs, field on it).
|
|
#: ``None`` container means the field sits directly on CalculatorInputs.
|
|
DRIVERS: dict[str, tuple[str | None, str]] = {
|
|
"token_price": ("price", "list_rate"),
|
|
"concession_pct": ("price", "concession_pct"),
|
|
"users": (None, "users"),
|
|
"voice_volume": ("volumes", "voice_inbound_monthly"),
|
|
"bot_share": ("mix", "bot_only_share"),
|
|
"virtual_agent_share": ("mix", "virtual_agent_share"),
|
|
"agentic_va_share": ("mix", "agentic_va_share"),
|
|
"bot_seconds": ("voice_bot", "avg_bot_seconds_per_call"),
|
|
"tts_chars_per_call": ("tts", "chars_per_call"),
|
|
}
|
|
|
|
|
|
def driver_value(inputs: CalculatorInputs, driver: str) -> float:
|
|
"""The current value of ``driver`` in ``inputs``."""
|
|
container, attr = _resolve(driver)
|
|
target = inputs if container is None else getattr(inputs, container)
|
|
if target is None:
|
|
raise ValueError(f"{driver} is not active in this scenario")
|
|
return float(getattr(target, attr))
|
|
|
|
|
|
def set_driver(inputs: CalculatorInputs, driver: str, value: float) -> CalculatorInputs:
|
|
"""A copy of ``inputs`` with ``driver`` set to ``value``."""
|
|
container, attr = _resolve(driver)
|
|
if container is None:
|
|
return inputs.with_(**{attr: _coerce(attr, value)})
|
|
current = getattr(inputs, container)
|
|
if current is None:
|
|
raise ValueError(f"{driver} is not active in this scenario")
|
|
return inputs.with_(**{container: replace(current, **{attr: _coerce(attr, value)})})
|
|
|
|
|
|
def _resolve(driver: str) -> tuple[str | None, str]:
|
|
try:
|
|
return DRIVERS[driver]
|
|
except KeyError:
|
|
raise KeyError(
|
|
f"{driver!r} is not a known driver. Known: {', '.join(sorted(DRIVERS))}"
|
|
) from None
|
|
|
|
|
|
def _coerce(attr: str, value: float) -> Any:
|
|
"""``users`` and ``chars_per_call`` are integers by contract."""
|
|
if attr in ("users", "chars_per_call"):
|
|
return int(round(value))
|
|
return float(value)
|
|
|
|
|
|
def sweep(
|
|
base: CalculatorInputs, driver: str, values: Sequence[float]
|
|
) -> list[dict[str, float]]:
|
|
"""Grand total across a range of one driver, everything else held."""
|
|
rows: list[dict[str, float]] = []
|
|
for value in values:
|
|
result: CalculatorResult = calculate(set_driver(base, driver, value))
|
|
rows.append(
|
|
{
|
|
driver: float(value),
|
|
"token_cost_annual": result.totals.token_cost_annual,
|
|
"tts_cost_annual": result.tts_cost_annual,
|
|
"grand_total_annual": result.grand_total_annual,
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
def tornado(
|
|
base: CalculatorInputs, drivers: Sequence[str], delta: float = 0.25
|
|
) -> list[dict[str, Any]]:
|
|
"""±``delta`` on each driver in turn, sorted by the swing it causes.
|
|
|
|
Drivers that are not active in the scenario (no TTS, no voice bot) are
|
|
skipped rather than raising — a tornado over an inactive lever is noise.
|
|
Shares are clamped to keep the highest-tier partition legal; a driver
|
|
whose low and high both clamp to the same value contributes no swing.
|
|
"""
|
|
if not 0.0 < delta < 1.0:
|
|
raise ValueError("delta must be in (0, 1)")
|
|
|
|
baseline = calculate(base).grand_total_annual
|
|
rows: list[dict[str, Any]] = []
|
|
|
|
for driver in drivers:
|
|
try:
|
|
current = driver_value(base, driver)
|
|
except (ValueError, KeyError):
|
|
continue
|
|
|
|
low_value = current * (1.0 - delta)
|
|
high_value = current * (1.0 + delta)
|
|
if driver.endswith("_share"):
|
|
low_value = max(0.0, min(1.0, low_value))
|
|
high_value = max(0.0, min(1.0, high_value))
|
|
|
|
try:
|
|
low = calculate(set_driver(base, driver, low_value)).grand_total_annual
|
|
high = calculate(set_driver(base, driver, high_value)).grand_total_annual
|
|
except ValueError:
|
|
# e.g. raising a share would break the partition — skip the lever
|
|
continue
|
|
|
|
rows.append(
|
|
{
|
|
"driver": driver,
|
|
"baseline_value": current,
|
|
"low_value": low_value,
|
|
"high_value": high_value,
|
|
"low": low,
|
|
"high": high,
|
|
"baseline": baseline,
|
|
"swing": abs(high - low),
|
|
}
|
|
)
|
|
|
|
return sorted(rows, key=lambda r: float(r["swing"]), reverse=True)
|