docs: introduce Mercury Notebook Deliverable Pattern
This commit is contained in:
77
studies/202607_CTM_GenesysCX/tokencalc/__init__.py
Normal file
77
studies/202607_CTM_GenesysCX/tokencalc/__init__.py
Normal file
@@ -0,0 +1,77 @@
|
||||
"""
|
||||
tokencalc — Genesys AI token cost & business case calculator core.
|
||||
|
||||
Pure-Python, UI-agnostic. The notebooks (served interactively with
|
||||
Mercury) are thin presentation layers over these functions.
|
||||
"""
|
||||
|
||||
from .benefit_model import calculate_total_benefit
|
||||
from .business_case import build_business_case, npv, payback_years
|
||||
from .cost_model import (
|
||||
calculate_consumption_ai_cost,
|
||||
calculate_per_user_ai_cost,
|
||||
calculate_platform_license_cost,
|
||||
calculate_total_cost,
|
||||
)
|
||||
from .defaults import (
|
||||
CONTRACTED_NAMED_USERS,
|
||||
CTM_DEFAULT_FEATURE_SCOPES,
|
||||
CTM_DEFAULT_ROLLOUT,
|
||||
CTM_DEFAULT_SITES,
|
||||
CTM_DEFAULT_TAKEOUTS,
|
||||
DEFAULT_METERS,
|
||||
DEFAULT_PRICING,
|
||||
PLATFORM_RATE_PER_USER_MONTHLY,
|
||||
)
|
||||
from .rollout import NO_ROLLOUT, RolloutPlan
|
||||
from .exports import (
|
||||
export_csv,
|
||||
export_excel,
|
||||
meters_dataframe,
|
||||
scenario_state_from_json,
|
||||
scenario_state_to_json,
|
||||
sites_dataframe,
|
||||
)
|
||||
from .inputs import CostTakeout, FeatureScope, SiteInput
|
||||
from .meters import Confidence, MeterType, TokenMeter, TokenPricing
|
||||
from .scenarios import BENEFIT_PARAMS, SCENARIOS, Scenario, get_scenario
|
||||
|
||||
__version__ = "0.1.0"
|
||||
|
||||
__all__ = [
|
||||
"BENEFIT_PARAMS",
|
||||
"CONTRACTED_NAMED_USERS",
|
||||
"CTM_DEFAULT_FEATURE_SCOPES",
|
||||
"CTM_DEFAULT_ROLLOUT",
|
||||
"CTM_DEFAULT_SITES",
|
||||
"CTM_DEFAULT_TAKEOUTS",
|
||||
"Confidence",
|
||||
"CostTakeout",
|
||||
"DEFAULT_METERS",
|
||||
"DEFAULT_PRICING",
|
||||
"FeatureScope",
|
||||
"MeterType",
|
||||
"NO_ROLLOUT",
|
||||
"PLATFORM_RATE_PER_USER_MONTHLY",
|
||||
"RolloutPlan",
|
||||
"SCENARIOS",
|
||||
"Scenario",
|
||||
"SiteInput",
|
||||
"TokenMeter",
|
||||
"TokenPricing",
|
||||
"build_business_case",
|
||||
"calculate_consumption_ai_cost",
|
||||
"calculate_per_user_ai_cost",
|
||||
"calculate_platform_license_cost",
|
||||
"calculate_total_benefit",
|
||||
"calculate_total_cost",
|
||||
"export_csv",
|
||||
"export_excel",
|
||||
"get_scenario",
|
||||
"meters_dataframe",
|
||||
"npv",
|
||||
"payback_years",
|
||||
"scenario_state_from_json",
|
||||
"scenario_state_to_json",
|
||||
"sites_dataframe",
|
||||
]
|
||||
577
studies/202607_CTM_GenesysCX/tokencalc/appendix4.py
Normal file
577
studies/202607_CTM_GenesysCX/tokencalc/appendix4.py
Normal file
@@ -0,0 +1,577 @@
|
||||
"""
|
||||
Appendix-4 corrected business case — the Genesys/Broadreach benefits
|
||||
kept verbatim, with the costs the deck omitted: AI Experience token
|
||||
consumption, AI implementation effort (V2 LoE), and double-billing of
|
||||
the existing platforms until their term contracts end.
|
||||
|
||||
Single source of truth behind the deliverable notebook
|
||||
(``notebooks/ctm_business_case_corrected.ipynb``, served with
|
||||
Mercury) — the presentation layer holds no math.
|
||||
|
||||
Sources: ``docs/Appendix 4 - CCaaS Platform Benefit Calculations
|
||||
(Consolidated).pptx`` (verbatim figures, deployment schedule) and
|
||||
``docs/ctm_ai_labour_estimate_V2.md`` (implementation hours).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import datetime as dt
|
||||
import math
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from .business_case import npv, payback_years
|
||||
from .cost_model import calculate_total_cost
|
||||
from .defaults import DEFAULT_METERS
|
||||
from .inputs import FeatureScope, SiteInput
|
||||
from .meters import Confidence, TokenMeter, TokenPricing
|
||||
from .rollout import RolloutPlan
|
||||
from .scenarios import Scenario
|
||||
|
||||
# ── Timeline ─────────────────────────────────────────────────────────
|
||||
|
||||
YEARS = [2026, 2027, 2028] # model years 1..3, contract start Jan 2026
|
||||
YEAR_INDEX = {2026: 1, 2027: 2, 2028: 3}
|
||||
_MONTHS = ["Jan", "Feb", "Mar", "Apr", "May", "Jun",
|
||||
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
|
||||
|
||||
|
||||
def month_label(m: int) -> str:
|
||||
"""Calendar label for a 1-indexed month from Jan 2026 (m=21 → 'Sep 2027')."""
|
||||
return f"{_MONTHS[(m - 1) % 12]} {2026 + (m - 1) // 12}"
|
||||
|
||||
|
||||
# ── Verbatim Appendix 4 figures ──────────────────────────────────────
|
||||
|
||||
REGIONS = ["NA", "ANZ", "EMEA", "ASIA"]
|
||||
CAPABILITIES = ["Agent Copilot", "WFM", "Email", "STA",
|
||||
"Predictive Routing", "Supervisor Copilot"]
|
||||
|
||||
#: (annual_value, three_yr_value) — VERBATIM slides 12-15, do not edit.
|
||||
VERBATIM_BENEFITS: dict[tuple[str, str], tuple[float, float]] = {
|
||||
("NA", "Agent Copilot"): (2_400_000, 3_400_000),
|
||||
("NA", "Email"): (1_900_000, 2_500_000),
|
||||
("NA", "STA"): (294_000, 506_000),
|
||||
("NA", "Supervisor Copilot"): (218_000, 291_000),
|
||||
("NA", "Predictive Routing"): (97_000, 167_000),
|
||||
("NA", "WFM"): (0, 0), # NA excluded — has similar feature
|
||||
("ANZ", "Agent Copilot"): (3_600_000, 3_900_000),
|
||||
("ANZ", "WFM"): (1_300_000, 1_400_000),
|
||||
("ANZ", "Predictive Routing"): (279_000, 302_000),
|
||||
("ANZ", "Email"): (132_000, 143_000),
|
||||
("ANZ", "STA"): (97_000, 105_000),
|
||||
("ANZ", "Supervisor Copilot"): (25_000, 27_000),
|
||||
("ASIA", "WFM"): (1_600_000, 914_000),
|
||||
("ASIA", "Email"): (160_000, 93_000),
|
||||
("ASIA", "STA"): (124_000, 72_000),
|
||||
("ASIA", "Predictive Routing"): (87_000, 51_000),
|
||||
("ASIA", "Agent Copilot"): (0, 0),
|
||||
("ASIA", "Supervisor Copilot"): (0, 0),
|
||||
("EMEA", "WFM"): (824_000, 687_000),
|
||||
("EMEA", "Email"): (282_000, 235_000),
|
||||
("EMEA", "STA"): (157_000, 131_000),
|
||||
("EMEA", "Agent Copilot"): (77_000, 64_000),
|
||||
("EMEA", "Supervisor Copilot"): (59_000, 49_000),
|
||||
("EMEA", "Predictive Routing"): (7_000, 6_000),
|
||||
}
|
||||
|
||||
#: The deck's own (rounded) summary rows — slides 8-9.
|
||||
SLIDE_TOTALS: dict = {
|
||||
"regional_3yr": {"NA": 6_900_000, "ANZ": 5_900_000,
|
||||
"ASIA": 1_100_000, "EMEA": 1_200_000},
|
||||
"capability_3yr": {"Agent Copilot": 7_400_000, "WFM": 3_000_000,
|
||||
"Email": 2_900_000, "STA": 814_000,
|
||||
"Predictive Routing": 526_000,
|
||||
"Supervisor Copilot": 367_000},
|
||||
"total_3yr": 15_000_000,
|
||||
"total_annual": 13_600_000,
|
||||
}
|
||||
|
||||
#: Verbatim TCO anchors — slides 5-6.
|
||||
TCO_VERBATIM: dict[str, float] = {
|
||||
"current_annual": 7_300_000, # current global spend / yr
|
||||
"current_3yr": 22_000_000,
|
||||
"ccaas_annual": 4_300_000, # licence run-rate / yr
|
||||
"ccaas_3yr": 15_400_000, # deck's 3-yr CCaaS investment (no ramp, no AI costs)
|
||||
"prof_services_y1": 2_400_000,
|
||||
"training_y1": 167_000,
|
||||
"npv_discount_rate": 0.135, # deck's benefit-NPV rate
|
||||
}
|
||||
|
||||
#: Actual contracted values where they differ from the deck — 🟢
|
||||
#: contractual. Layered over TCO_VERBATIM, which stays the untouched
|
||||
#: record of what Genesys pitched (the anchor CTM can follow);
|
||||
#: presentation reads through :func:`tco`.
|
||||
TCO_CONTRACTED: dict[str, float] = {
|
||||
"ccaas_annual": 3_200_000, # signed licence run-rate (deck pitched $4.3M/yr)
|
||||
}
|
||||
|
||||
#: NTT professional services — SOW billing milestones (🟢 contractual).
|
||||
#: The deck's verbatim anchor books $2.4M PS in year 1; the signed SOW
|
||||
#: bills $2,025,446.48 in four milestones split 50/50 across 2026-27.
|
||||
PS_MILESTONES: list[dict] = [
|
||||
{"name": "SOW Effective Date", "date": dt.date(2026, 3, 15),
|
||||
"share": 0.30, "amount": 607_633.94},
|
||||
{"name": "Start of client UAT (first region)", "date": dt.date(2026, 9, 30),
|
||||
"share": 0.20, "amount": 405_089.30},
|
||||
{"name": "Start of client UAT (last region)", "date": dt.date(2027, 6, 30),
|
||||
"share": 0.30, "amount": 607_633.94},
|
||||
{"name": "Completion of last go-live migration", "date": dt.date(2027, 9, 30),
|
||||
"share": 0.20, "amount": 405_089.30},
|
||||
]
|
||||
PS_CONTRACTED_TOTAL = sum(m["amount"] for m in PS_MILESTONES)
|
||||
|
||||
#: NTT managed services — commences billing at MCX go-live (🟢 contractual).
|
||||
#: Not in the deck's TCO at all; an ongoing run-rate cost thereafter.
|
||||
MANAGED_SERVICES_ANNUAL = 410_918.40
|
||||
MCX_GO_LIVE = dt.date(2026, 9, 30)
|
||||
|
||||
|
||||
def tco(key: str) -> float:
|
||||
"""Contracted value where one exists, else the deck's verbatim anchor."""
|
||||
return TCO_CONTRACTED.get(key, TCO_VERBATIM[key])
|
||||
|
||||
#: Genesys/Broadreach deployment schedule (slides 17-21), months from
|
||||
#: Jan 2026 inclusive. Benefits realize IMPL + 3 months.
|
||||
IMPL_MONTH = {"NA": 18, "ANZ": 21, "EMEA": 24, "ASIA": 27}
|
||||
BENEFIT_LAG_MONTHS = 3
|
||||
REALIZE_MONTH = {r: m + BENEFIT_LAG_MONTHS for r, m in IMPL_MONTH.items()}
|
||||
#: NA Gantt exception: Email implemented Jan 2027, realizes Apr 2027.
|
||||
NA_EMAIL_IMPL_MONTH = 13
|
||||
|
||||
DEFAULT_RAMP_MONTHS = 6 # Genesys ramp programme (🟢 order form)
|
||||
DEFAULT_TERMINATION = dt.date(2027, 12, 31) # current-platform term contracts
|
||||
|
||||
# ── Region ⇄ site mapping ────────────────────────────────────────────
|
||||
|
||||
|
||||
def site_region(site_name: str) -> str:
|
||||
"""Map a tokencalc site to its Appendix-4 region (APAC * → ASIA)."""
|
||||
return {"NAM": "NA", "AUZ": "ANZ", "EMEA": "EMEA"}.get(site_name, "ASIA")
|
||||
|
||||
|
||||
def region_site_names(sites: list[SiteInput]) -> dict[str, list[str]]:
|
||||
return {r: [s.site_name for s in sites if site_region(s.site_name) == r]
|
||||
for r in REGIONS}
|
||||
|
||||
|
||||
def region_agents(sites: list[SiteInput]) -> dict[str, int]:
|
||||
return {r: sum(s.agents for s in sites if site_region(s.site_name) == r)
|
||||
for r in REGIONS}
|
||||
|
||||
|
||||
# ── Verbatim benefit helpers ─────────────────────────────────────────
|
||||
|
||||
|
||||
def verbatim_dataframe() -> pd.DataFrame:
|
||||
"""Long DataFrame of the verbatim benefits: region, capability, annual, three_yr."""
|
||||
return pd.DataFrame(
|
||||
[{"region": r, "capability": c, "annual": a, "three_yr": t}
|
||||
for (r, c), (a, t) in VERBATIM_BENEFITS.items()]
|
||||
)
|
||||
|
||||
|
||||
def crossfoot_tolerance(value: float) -> float:
|
||||
"""The deck rounds to $0.1M and its own tables cross-foot ±$50-120K."""
|
||||
return max(100_000, 0.015 * value)
|
||||
|
||||
|
||||
# ── Schedules & rollouts ─────────────────────────────────────────────
|
||||
|
||||
|
||||
def build_rollouts(
|
||||
sites: list[SiteInput],
|
||||
na_email_early: bool = True,
|
||||
ramp_months: int = DEFAULT_RAMP_MONTHS,
|
||||
) -> tuple[RolloutPlan, RolloutPlan, RolloutPlan]:
|
||||
"""(token, email_token, benefit) rollout plans on the deck's schedule.
|
||||
|
||||
``RolloutPlan.go_live_month = m`` means active from month m+1; the
|
||||
deck's labels are inclusive (NA "realizes Sep 2027" ⇒ September
|
||||
counts), so keys are set to label − 1. The benefit plan is keyed by
|
||||
region (plus ``NA_EMAIL`` for the NA Gantt exception); the token
|
||||
plans are keyed by site.
|
||||
"""
|
||||
token = RolloutPlan(
|
||||
contract_start="2026-01", build_months=max(IMPL_MONTH.values()),
|
||||
ramp_months=ramp_months,
|
||||
first_year_platform_discount=0.0, # licences are handled verbatim, not by this plan
|
||||
go_live_month={s.site_name: IMPL_MONTH[site_region(s.site_name)] - 1
|
||||
for s in sites},
|
||||
)
|
||||
email = dataclasses.replace(
|
||||
token,
|
||||
go_live_month={**token.go_live_month,
|
||||
"NAM": (NA_EMAIL_IMPL_MONTH - 1) if na_email_early
|
||||
else IMPL_MONTH["NA"] - 1},
|
||||
)
|
||||
benefit = RolloutPlan(
|
||||
first_year_platform_discount=0.0,
|
||||
go_live_month={**{r: REALIZE_MONTH[r] - 1 for r in REGIONS},
|
||||
"NA_EMAIL": (NA_EMAIL_IMPL_MONTH + BENEFIT_LAG_MONTHS - 1)
|
||||
if na_email_early else REALIZE_MONTH["NA"] - 1},
|
||||
)
|
||||
return token, email, benefit
|
||||
|
||||
|
||||
def benefits_by_year(
|
||||
benefit_rollout: RolloutPlan, na_email_early: bool = True
|
||||
) -> pd.DataFrame:
|
||||
"""Phase each verbatim 3-yr value by its region's realization window.
|
||||
|
||||
Scaling is at the finest grain (region × capability), so every
|
||||
verbatim per-region, per-capability, and grand total is reproduced
|
||||
exactly. Long DataFrame: region, capability, year, benefit.
|
||||
"""
|
||||
rows = []
|
||||
for (region, cap), (_annual, three_yr) in VERBATIM_BENEFITS.items():
|
||||
key = ("NA_EMAIL" if (region == "NA" and cap == "Email" and na_email_early)
|
||||
else region)
|
||||
live = [benefit_rollout.live_months_in_year(key, YEAR_INDEX[y]) for y in YEARS]
|
||||
total_live = sum(live)
|
||||
for y, m in zip(YEARS, live):
|
||||
rows.append({"region": region, "capability": cap, "year": y,
|
||||
"benefit": three_yr * m / total_live if total_live else 0.0})
|
||||
return pd.DataFrame(rows)
|
||||
|
||||
|
||||
# ── Base cost lines (verbatim + contract mechanics) ──────────────────
|
||||
|
||||
|
||||
def current_months_in_year(termination: dt.date, cal_year: int) -> int:
|
||||
"""Months a term contract bills in ``cal_year`` (through its termination month)."""
|
||||
if cal_year < termination.year:
|
||||
return 12
|
||||
if cal_year > termination.year:
|
||||
return 0
|
||||
return termination.month
|
||||
|
||||
|
||||
def current_state_inputs(
|
||||
sites: list[SiteInput],
|
||||
total_annual: float | None = None,
|
||||
termination: dt.date = DEFAULT_TERMINATION,
|
||||
) -> pd.DataFrame:
|
||||
"""Per-region current-platform inputs, seeded by agent share of the
|
||||
verbatim global spend. Region-indexed; annual_cost and
|
||||
contract_termination are the editable columns."""
|
||||
total = TCO_VERBATIM["current_annual"] if total_annual is None else total_annual
|
||||
agents = region_agents(sites)
|
||||
total_agents = sum(agents.values())
|
||||
return pd.DataFrame([
|
||||
{"region": r,
|
||||
"agents": agents[r],
|
||||
"share": agents[r] / total_agents,
|
||||
"annual_cost": total * agents[r] / total_agents,
|
||||
"contract_termination": termination,
|
||||
"confidence": "🟡 agent-share allocation of the verbatim total"}
|
||||
for r in REGIONS
|
||||
]).set_index("region")
|
||||
|
||||
|
||||
def current_costs_by_year(current_state: pd.DataFrame) -> dict[int, float]:
|
||||
"""Existing-platform run-off per calendar year (the double-billing line)."""
|
||||
return {
|
||||
y: float(sum(
|
||||
row["annual_cost"]
|
||||
* current_months_in_year(row["contract_termination"], y) / 12
|
||||
for _, row in current_state.iterrows()))
|
||||
for y in YEARS
|
||||
}
|
||||
|
||||
|
||||
def licence_months_in_year(year_index: int, ramp_months: int) -> int:
|
||||
"""Ramp programme: licence billing starts in calendar month ramp_months + 1."""
|
||||
start, end = 12 * (year_index - 1) + 1, 12 * year_index
|
||||
return max(0, end - max(start, ramp_months + 1) + 1)
|
||||
|
||||
|
||||
def licence_costs_by_year(
|
||||
ramp_months: int = DEFAULT_RAMP_MONTHS, annual: float | None = None
|
||||
) -> dict[int, float]:
|
||||
rate = TCO_VERBATIM["ccaas_annual"] if annual is None else annual
|
||||
return {y: rate * licence_months_in_year(YEAR_INDEX[y], ramp_months) / 12
|
||||
for y in YEARS}
|
||||
|
||||
|
||||
def ps_costs_by_year(contracted: bool = False) -> dict[int, float]:
|
||||
"""Base professional services + training.
|
||||
|
||||
Verbatim: the deck's $2.4M PS lump plus training, all in year 1.
|
||||
Contracted: PS phased on the SOW billing milestones (50/50 across
|
||||
2026-27, $2.03M total); training stays the verbatim year-1 line —
|
||||
the SOW milestones don't itemize it separately.
|
||||
"""
|
||||
training = TCO_VERBATIM["training_y1"]
|
||||
if contracted:
|
||||
ps = {y: 0.0 for y in YEARS}
|
||||
for m in PS_MILESTONES:
|
||||
ps[m["date"].year] += m["amount"]
|
||||
return {y: ps[y] + (training if y == 2026 else 0.0) for y in YEARS}
|
||||
return {2026: TCO_VERBATIM["prof_services_y1"] + training,
|
||||
2027: 0.0, 2028: 0.0}
|
||||
|
||||
|
||||
def ps_milestones_dataframe() -> pd.DataFrame:
|
||||
"""The SOW billing milestones as a display table."""
|
||||
return pd.DataFrame(PS_MILESTONES)
|
||||
|
||||
|
||||
def managed_services_by_year(
|
||||
annual: float = MANAGED_SERVICES_ANNUAL, start: dt.date = MCX_GO_LIVE
|
||||
) -> dict[int, float]:
|
||||
"""Managed services bill from the month after go-live (Sep 30 → Oct),
|
||||
then run at the full annual rate — an ongoing cost with no end date
|
||||
inside the model window."""
|
||||
def _months(y: int) -> int:
|
||||
if y < start.year:
|
||||
return 0
|
||||
return 12 if y > start.year else 12 - start.month
|
||||
return {y: annual * _months(y) / 12 for y in YEARS}
|
||||
|
||||
|
||||
# ── Token consumption (missing cost #1) ──────────────────────────────
|
||||
|
||||
|
||||
def claim_scenario(email_auto_respond_rate: float = 0.255) -> Scenario:
|
||||
"""Claim-level scenario: deck parameters, no consumption maturity ramp."""
|
||||
return Scenario(
|
||||
name="genesys-claim",
|
||||
voice_bot_deflection=0.0, voice_bot_avg_minutes=0.0,
|
||||
agentic_va_deflection=0.0,
|
||||
voice_summarization_eligibility=0.0,
|
||||
voice_knowledge_eligibility=0.0, # unused by the Appendix-4 scope set
|
||||
email_auto_respond_rate=email_auto_respond_rate,
|
||||
consumption_cost_realization={1: 1.0, 2: 1.0, 3: 1.0},
|
||||
)
|
||||
|
||||
|
||||
def autorespond_meter(tokens_per_msg: float = 0.05) -> TokenMeter:
|
||||
"""Email Auto-Respond working meter — rate unpublished (🔴→🟡).
|
||||
|
||||
Anchor: ≈1 AI action per generated response; Genesys Cloud Copilot
|
||||
meters 20 AI actions per token.
|
||||
"""
|
||||
return dataclasses.replace(
|
||||
DEFAULT_METERS["Email AI (Auto-Respond)"],
|
||||
units_per_token=1.0 / tokens_per_msg,
|
||||
tokens_per_unit=tokens_per_msg,
|
||||
confidence=Confidence.ESTIMATED,
|
||||
notes="WORKING ASSUMPTION — rate unpublished; ≈1 AI action per generated "
|
||||
"response (Genesys Cloud Copilot meters 20 AI actions/token).",
|
||||
)
|
||||
|
||||
|
||||
def build_scopes(
|
||||
sites: list[SiteInput],
|
||||
copilot_includes_asia: bool = False,
|
||||
pr_eligibility: float = 1.0,
|
||||
ai_translate_eligibility: float = 0.01,
|
||||
) -> tuple[list[FeatureScope], list[FeatureScope]]:
|
||||
"""(core, email) feature scopes mirroring the six deck capabilities.
|
||||
|
||||
No ``adoption_curve`` on any scope — a curve would silently override
|
||||
the claim scenario's flat consumption realization. Email scopes are
|
||||
separate because NA Email implements early (own rollout plan).
|
||||
"""
|
||||
all_names = [s.site_name for s in sites]
|
||||
asia = [n for n in all_names if site_region(n) == "ASIA"]
|
||||
non_asia = [n for n in all_names if site_region(n) != "ASIA"]
|
||||
copilot_sites = non_asia + (asia if copilot_includes_asia else [])
|
||||
core = [
|
||||
FeatureScope("Agent Copilot [named]", copilot_sites, phase=1),
|
||||
FeatureScope("Speech & Text Analytics [named]", all_names, phase=1),
|
||||
FeatureScope("Predictive Routing", all_names, phase=1,
|
||||
eligibility_pct=pr_eligibility),
|
||||
# $0 by Rule 1 (Copilot covers summarization) — kept visible.
|
||||
FeatureScope("AI Summary & Insights", copilot_sites, phase=1),
|
||||
# Supervisor Copilot small-volume proxy.
|
||||
FeatureScope("AI Translate", asia + ["EMEA"], phase=1,
|
||||
eligibility_pct=ai_translate_eligibility),
|
||||
]
|
||||
email = [FeatureScope("Email AI (Auto-Respond)", all_names, phase=1)]
|
||||
return core, email
|
||||
|
||||
|
||||
def token_costs_by_year(
|
||||
sites: list[SiteInput],
|
||||
meters: dict[str, TokenMeter],
|
||||
pricing: dict[str, TokenPricing],
|
||||
scenario: Scenario,
|
||||
core_scopes: list[FeatureScope],
|
||||
email_scopes: list[FeatureScope],
|
||||
token_rollout: RolloutPlan,
|
||||
email_rollout: RolloutPlan,
|
||||
use_contracted: bool = False,
|
||||
) -> pd.DataFrame:
|
||||
"""Engine-computed token costs, rollout-gated, per calendar year.
|
||||
|
||||
Long DataFrame: cost_line, scope, annual_cost, confidence, year.
|
||||
"""
|
||||
frames = []
|
||||
for y in YEARS:
|
||||
for scopes, rollout in ((core_scopes, token_rollout),
|
||||
(email_scopes, email_rollout)):
|
||||
part = calculate_total_cost(
|
||||
sites, scopes, meters, pricing, scenario, YEAR_INDEX[y],
|
||||
include_platform=False, use_contracted=use_contracted,
|
||||
rollout=rollout,
|
||||
)
|
||||
part["year"] = y
|
||||
frames.append(part)
|
||||
return pd.concat(frames, ignore_index=True)
|
||||
|
||||
|
||||
# ── AI implementation effort (missing cost #2, V2 LoE) ───────────────
|
||||
|
||||
#: (low, high) Y1 hours — docs/ctm_ai_labour_estimate_V2.md.
|
||||
AI_IMPL_HOURS: dict[str, tuple[float, float]] = {
|
||||
"Agent Copilot": (1_200, 1_800), # voice + digital incl. email Auto-Suggest
|
||||
"Email Auto-Respond": (800, 1_400), # separate flow; needs SoR integration
|
||||
"STA": (800, 1_200), # topics, programs, tuning × 7 languages
|
||||
"Supervisor Copilot": (200, 400),
|
||||
"Predictive Routing": (400, 700),
|
||||
"Cross-cutting (PM, governance, testing, integration)": (1_000, 1_800),
|
||||
}
|
||||
KB_READINESS_HOURS = (500, 1_500) # prerequisite project — flagged separately
|
||||
STEADY_STATE_HOURS = (500, 900) # absolute h/yr, 2027-2028
|
||||
DEFAULT_BLENDED_RATE = 225.0
|
||||
SMELL_TEST_FLOOR = 0.15 # impl ≥ 15% of benefit claim, or flag
|
||||
|
||||
|
||||
def impl_feature_regions(copilot_includes_asia: bool = False) -> dict[str, list[str]]:
|
||||
"""Which regions each implementation workstream serves."""
|
||||
return {
|
||||
"Agent Copilot": (["NA", "ANZ", "EMEA"]
|
||||
+ (["ASIA"] if copilot_includes_asia else [])),
|
||||
"Email Auto-Respond": list(REGIONS),
|
||||
"STA": list(REGIONS),
|
||||
"Supervisor Copilot": ["NA", "ANZ", "EMEA"], # deck: $0 SupCopilot in ASIA
|
||||
"Predictive Routing": list(REGIONS),
|
||||
"Cross-cutting (PM, governance, testing, integration)": list(REGIONS),
|
||||
}
|
||||
|
||||
|
||||
def hours_pick(rng: tuple[float, float], mode: str) -> float:
|
||||
low, high = rng
|
||||
return {"low": low, "mid": (low + high) / 2, "high": high}[mode]
|
||||
|
||||
|
||||
def impl_year_fractions(impl_month: int) -> list[float]:
|
||||
"""Spend spreads uniformly from contract start (month 0) to the impl month."""
|
||||
prev, fracs = 0, []
|
||||
for yi in (1, 2, 3):
|
||||
cur = min(12 * yi, impl_month)
|
||||
fracs.append((cur - prev) / impl_month)
|
||||
prev = cur
|
||||
return fracs
|
||||
|
||||
|
||||
def region_impl_month(feature: str, region: str, na_email_early: bool = True) -> int:
|
||||
if feature == "Email Auto-Respond" and region == "NA" and na_email_early:
|
||||
return NA_EMAIL_IMPL_MONTH
|
||||
return IMPL_MONTH[region]
|
||||
|
||||
|
||||
def build_impl_costs(
|
||||
sites: list[SiteInput],
|
||||
mode: str = "mid",
|
||||
rate: float = DEFAULT_BLENDED_RATE,
|
||||
include_kb: bool = True,
|
||||
copilot_includes_asia: bool = False,
|
||||
na_email_early: bool = True,
|
||||
) -> tuple[pd.DataFrame, dict[int, float], dict[int, float], dict[int, float]]:
|
||||
"""V2 hours-range × rate model (swap point for the future LoE engine).
|
||||
|
||||
Returns (detail_df, impl_by_year, kb_by_year, steady_by_year).
|
||||
Hours allocate to each workstream's scoped regions by agent share;
|
||||
steady-state is booked program-level in 2027-2028.
|
||||
"""
|
||||
agents = region_agents(sites)
|
||||
feature_regions = impl_feature_regions(copilot_includes_asia)
|
||||
workstreams = dict(AI_IMPL_HOURS)
|
||||
if include_kb:
|
||||
workstreams["KB readiness (prerequisite)"] = KB_READINESS_HOURS
|
||||
rows = []
|
||||
for feature, rng in workstreams.items():
|
||||
regions = feature_regions.get(feature, list(REGIONS))
|
||||
scope_agents = sum(agents[r] for r in regions)
|
||||
for r in regions:
|
||||
hours = hours_pick(rng, mode) * agents[r] / scope_agents
|
||||
fracs = impl_year_fractions(
|
||||
region_impl_month(feature, r, na_email_early))
|
||||
rows.append({"workstream": feature, "region": r, "hours": hours,
|
||||
"cost": hours * rate,
|
||||
**{y: hours * rate * f for y, f in zip(YEARS, fracs)}})
|
||||
df = pd.DataFrame(rows)
|
||||
is_kb = df["workstream"].str.startswith("KB")
|
||||
impl_y = {y: float(df.loc[~is_kb, y].sum()) for y in YEARS}
|
||||
kb_y = {y: float(df.loc[is_kb, y].sum()) for y in YEARS}
|
||||
steady = hours_pick(STEADY_STATE_HOURS, mode) * rate
|
||||
steady_y = {2026: 0.0, 2027: steady, 2028: steady}
|
||||
return df, impl_y, kb_y, steady_y
|
||||
|
||||
|
||||
# ── Business case (baseline-relative frame) ──────────────────────────
|
||||
|
||||
|
||||
def case_flows(
|
||||
total_cost_by_year: dict[int, float],
|
||||
benefit_total_by_year: dict[int, float],
|
||||
baseline_annual: float | None = None,
|
||||
) -> tuple[dict[int, float], dict[int, float]]:
|
||||
"""(incremental cost, net) vs the do-nothing baseline.
|
||||
|
||||
One frame captures both the 2026-27 double-billing penalty and the
|
||||
post-termination cost-avoidance credit.
|
||||
"""
|
||||
base = TCO_VERBATIM["current_annual"] if baseline_annual is None else baseline_annual
|
||||
inc = {y: total_cost_by_year[y] - base for y in YEARS}
|
||||
net = {y: benefit_total_by_year[y] - inc[y] for y in YEARS}
|
||||
return inc, net
|
||||
|
||||
|
||||
def payback_label(net_by_year: dict[int, float]) -> str:
|
||||
pb = payback_years([net_by_year[y] for y in YEARS])
|
||||
if pb is None:
|
||||
return f"beyond {YEARS[-1]}"
|
||||
if pb == 0:
|
||||
return "immediate"
|
||||
m = math.ceil(pb * 12)
|
||||
return f"{m} months (~{month_label(m)})"
|
||||
|
||||
|
||||
def case_kpis(
|
||||
inc: dict[int, float],
|
||||
net: dict[int, float],
|
||||
discount_rate: float | None = None,
|
||||
) -> dict:
|
||||
"""KPIs for one cost frame. Benefits are recoverable as net + inc."""
|
||||
rate = TCO_VERBATIM["npv_discount_rate"] if discount_rate is None else discount_rate
|
||||
net_list = [net[y] for y in YEARS]
|
||||
inc_total = sum(inc.values())
|
||||
net_total = sum(net_list)
|
||||
return {
|
||||
"benefits_3yr": net_total + inc_total,
|
||||
"incremental_cost_3yr": inc_total,
|
||||
"net_3yr": net_total,
|
||||
"roi": (net_total / inc_total) if inc_total > 0 else None,
|
||||
"npv": npv(net_list, rate),
|
||||
"discount_rate": rate,
|
||||
"payback": payback_label(net),
|
||||
}
|
||||
|
||||
|
||||
# ── Display helpers ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
def money(v: float) -> str:
|
||||
sign, a = ("-" if v < 0 else ""), abs(v)
|
||||
return f"{sign}${a/1e6:,.1f}M" if a >= 1e6 else f"{sign}${a/1e3:,.0f}K"
|
||||
|
||||
|
||||
def html_money(v: float) -> str:
|
||||
"""Plotly text with two or more bare ``$`` triggers MathJax math mode —
|
||||
annotations holding several amounts must use the HTML entity instead."""
|
||||
return money(v).replace("$", "$")
|
||||
429
studies/202607_CTM_GenesysCX/tokencalc/benefit_model.py
Normal file
429
studies/202607_CTM_GenesysCX/tokencalc/benefit_model.py
Normal file
@@ -0,0 +1,429 @@
|
||||
"""
|
||||
Benefit calculation engine.
|
||||
|
||||
All benefits convert saved handle-time seconds into dollars via each
|
||||
site's fully-loaded labour rate per working second. Reduction
|
||||
percentages come from :data:`tokencalc.scenarios.BENEFIT_PARAMS` —
|
||||
``realistic`` (pressure-tested) by default; pass ``params="claim"``
|
||||
to reproduce the Genesys ROI-doc figures for side-by-side comparison.
|
||||
|
||||
Every figure scales by the scenario's year realization ramp.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from .inputs import FeatureScope, SiteInput
|
||||
from .meters import Confidence
|
||||
from .rollout import NO_ROLLOUT, RolloutPlan
|
||||
from .scenarios import BENEFIT_PARAMS, Scenario, get_scenario
|
||||
|
||||
MONTHS_PER_YEAR = 12
|
||||
|
||||
|
||||
def _param(name: str, params: str) -> float:
|
||||
return BENEFIT_PARAMS[name][params]
|
||||
|
||||
|
||||
def _scope_for(feature_scopes: list[FeatureScope] | FeatureScope,
|
||||
feature: str) -> FeatureScope | None:
|
||||
if isinstance(feature_scopes, FeatureScope):
|
||||
return feature_scopes if feature_scopes.feature == feature else None
|
||||
return next((s for s in feature_scopes if s.feature == feature), None)
|
||||
|
||||
|
||||
def _df(rows: list[dict]) -> pd.DataFrame:
|
||||
return pd.DataFrame(
|
||||
rows, columns=["benefit_line", "scope", "annual_value", "confidence"]
|
||||
)
|
||||
|
||||
|
||||
def calculate_voice_handle_time_benefit(
|
||||
sites: list[SiteInput],
|
||||
feature_scope: FeatureScope,
|
||||
scenario: str | Scenario,
|
||||
year: int,
|
||||
params: str = "realistic",
|
||||
rollout: RolloutPlan | None = None,
|
||||
) -> pd.DataFrame:
|
||||
"""AHT reduction from knowledge surfacing (Agent Copilot).
|
||||
|
||||
Benefit = volume × eligibility × AHT × reduction% × labour rate.
|
||||
"""
|
||||
sc = get_scenario(scenario) if isinstance(scenario, str) else scenario
|
||||
ro = rollout or NO_ROLLOUT
|
||||
reduction = _param("voice_aht_knowledge_reduction", params)
|
||||
realization = sc.realization(year)
|
||||
rows = []
|
||||
for s in sites:
|
||||
if not feature_scope.active(s.site_name, year):
|
||||
continue
|
||||
eligibility = (
|
||||
feature_scope.eligibility_pct
|
||||
if feature_scope.eligibility_pct is not None
|
||||
else sc.voice_knowledge_eligibility
|
||||
)
|
||||
seconds_saved = (
|
||||
s.voice_volume_monthly * MONTHS_PER_YEAR
|
||||
* eligibility * s.voice_aht_seconds * reduction * realization
|
||||
)
|
||||
rows.append(
|
||||
{
|
||||
"benefit_line": "Voice AHT (knowledge surfacing)",
|
||||
"scope": s.site_name,
|
||||
"annual_value": seconds_saved * s.agent_cost_per_second
|
||||
* ro.fraction_live(s.site_name, year),
|
||||
"confidence": Confidence.ESTIMATED.value,
|
||||
}
|
||||
)
|
||||
return _df(rows)
|
||||
|
||||
|
||||
def calculate_acw_summarization_benefit(
|
||||
sites: list[SiteInput],
|
||||
feature_scope: FeatureScope,
|
||||
scenario: str | Scenario,
|
||||
year: int,
|
||||
params: str = "realistic",
|
||||
rollout: RolloutPlan | None = None,
|
||||
) -> pd.DataFrame:
|
||||
"""ACW eliminated by auto-summarization (Copilot / AI Summary)."""
|
||||
sc = get_scenario(scenario) if isinstance(scenario, str) else scenario
|
||||
ro = rollout or NO_ROLLOUT
|
||||
reduction = _param("voice_acw_reduction", params)
|
||||
realization = sc.realization(year)
|
||||
rows = []
|
||||
for s in sites:
|
||||
if not feature_scope.active(s.site_name, year):
|
||||
continue
|
||||
eligibility = (
|
||||
feature_scope.eligibility_pct
|
||||
if feature_scope.eligibility_pct is not None
|
||||
else sc.voice_summarization_eligibility
|
||||
)
|
||||
seconds_saved = (
|
||||
s.voice_volume_monthly * MONTHS_PER_YEAR
|
||||
* eligibility * s.voice_acw_seconds * reduction * realization
|
||||
)
|
||||
rows.append(
|
||||
{
|
||||
"benefit_line": "Voice ACW (summarization)",
|
||||
"scope": s.site_name,
|
||||
"annual_value": seconds_saved * s.agent_cost_per_second
|
||||
* ro.fraction_live(s.site_name, year),
|
||||
"confidence": Confidence.ESTIMATED.value,
|
||||
}
|
||||
)
|
||||
return _df(rows)
|
||||
|
||||
|
||||
def calculate_email_ai_benefit(
|
||||
sites: list[SiteInput],
|
||||
feature_scope: FeatureScope,
|
||||
scenario: str | Scenario,
|
||||
year: int,
|
||||
params: str = "realistic",
|
||||
rollout: RolloutPlan | None = None,
|
||||
) -> pd.DataFrame:
|
||||
"""Email Auto-Respond — full displacement at the respond rate.
|
||||
|
||||
(Email Auto-Suggest is not a separate benefit line: it is included
|
||||
in Agent Copilot, whose per-user meter carries the drafting help.)"""
|
||||
sc = get_scenario(scenario) if isinstance(scenario, str) else scenario
|
||||
ro = rollout or NO_ROLLOUT
|
||||
realization = sc.realization(year)
|
||||
rows = []
|
||||
for s in sites:
|
||||
if not feature_scope.active(s.site_name, year):
|
||||
continue
|
||||
respond_rate = (
|
||||
feature_scope.deflection_target
|
||||
if feature_scope.deflection_target is not None
|
||||
else sc.email_auto_respond_rate
|
||||
)
|
||||
annual_emails = s.email_volume_monthly * MONTHS_PER_YEAR
|
||||
respond_seconds = (
|
||||
annual_emails * respond_rate * s.email_aht_seconds * realization
|
||||
)
|
||||
rate = s.agent_cost_per_second
|
||||
rows.append(
|
||||
{
|
||||
"benefit_line": "Email Auto-Respond (displaced handling)",
|
||||
"scope": s.site_name,
|
||||
"annual_value": respond_seconds * rate
|
||||
* ro.fraction_live(s.site_name, year),
|
||||
"confidence": Confidence.UNKNOWN.value, # meter rate unsourced
|
||||
}
|
||||
)
|
||||
return _df(rows)
|
||||
|
||||
|
||||
def calculate_sta_benefit(
|
||||
sites: list[SiteInput],
|
||||
feature_scope: FeatureScope,
|
||||
scenario: str | Scenario,
|
||||
year: int,
|
||||
params: str = "realistic",
|
||||
rollout: RolloutPlan | None = None,
|
||||
) -> pd.DataFrame:
|
||||
"""STA reduces AHT *indirectly* via coaching — small reduction with
|
||||
a realistic ramp (default 1.5% vs the 4% claim)."""
|
||||
sc = get_scenario(scenario) if isinstance(scenario, str) else scenario
|
||||
ro = rollout or NO_ROLLOUT
|
||||
reduction = _param("sta_aht_reduction", params)
|
||||
realization = sc.realization(year)
|
||||
rows = []
|
||||
for s in sites:
|
||||
if not feature_scope.active(s.site_name, year):
|
||||
continue
|
||||
seconds_saved = (
|
||||
s.voice_volume_monthly * MONTHS_PER_YEAR
|
||||
* s.voice_aht_seconds * reduction * realization
|
||||
)
|
||||
rows.append(
|
||||
{
|
||||
"benefit_line": "STA coaching (AHT)",
|
||||
"scope": s.site_name,
|
||||
"annual_value": seconds_saved * s.agent_cost_per_second
|
||||
* ro.fraction_live(s.site_name, year),
|
||||
"confidence": Confidence.ESTIMATED.value,
|
||||
}
|
||||
)
|
||||
return _df(rows)
|
||||
|
||||
|
||||
def calculate_va_deflection_benefit(
|
||||
sites: list[SiteInput],
|
||||
feature_scope: FeatureScope,
|
||||
scenario: str | Scenario,
|
||||
year: int,
|
||||
params: str = "realistic",
|
||||
rollout: RolloutPlan | None = None,
|
||||
) -> pd.DataFrame:
|
||||
"""Agent labour avoided on calls deflected to Voice Bot or Agentic VA.
|
||||
|
||||
**Layered (sequential) deflection model** — Voice Bot runs first on
|
||||
the full call pool; Agentic VA handles a share of the *residual*
|
||||
(calls the bot did not deflect). The two mechanisms are substitutes
|
||||
operating on the same call base, not independent additive benefits.
|
||||
|
||||
Effective total deflection:
|
||||
bot_rate + (1 − bot_rate) × va_rate
|
||||
e.g. 35% + 65% × 15% = 44.75% (not 50%)
|
||||
|
||||
**Three realization haircuts** are applied to convert raw deflected
|
||||
volume into realizable labour savings:
|
||||
|
||||
1. ``completion_rate`` — share of "deflected" calls that don't
|
||||
escalate to an agent mid-session (bot/VA fully handles the call).
|
||||
2. ``labour_realization`` — staffing flexibility factor; deflected
|
||||
volume doesn't reduce headcount 1:1 due to minimums, shrinkage,
|
||||
and occupancy ceilings.
|
||||
3. ``callback_discount`` — fraction of deflected calls that re-enter
|
||||
as repeat contacts (poorly-handled deflections drive callbacks).
|
||||
|
||||
Combined realistic factor: 0.70 × 0.80 × (1 − 0.05) ≈ 0.53
|
||||
|
||||
The ``params="claim"`` path sets all three factors to their
|
||||
``claim`` values (1.0 / 1.0 / 0.0) to reproduce the original
|
||||
Genesys ROI-doc figures for side-by-side comparison.
|
||||
"""
|
||||
sc = get_scenario(scenario) if isinstance(scenario, str) else scenario
|
||||
ro = rollout or NO_ROLLOUT
|
||||
realization = sc.realization(year)
|
||||
|
||||
# Realization haircuts — read from BENEFIT_PARAMS so claim/realistic
|
||||
# paths are consistent with all other benefit lines.
|
||||
completion_rate = _param("va_completion_rate", params)
|
||||
labour_real = _param("va_labour_realization", params)
|
||||
callback_disc = _param("va_callback_discount", params)
|
||||
realization_factor = completion_rate * labour_real * (1.0 - callback_disc)
|
||||
|
||||
rows = []
|
||||
for s in sites:
|
||||
if not feature_scope.active(s.site_name, year):
|
||||
continue
|
||||
|
||||
if feature_scope.feature == "Voice Bot":
|
||||
# Bot operates on the full call pool.
|
||||
bot_rate = (
|
||||
feature_scope.deflection_target
|
||||
if feature_scope.deflection_target is not None
|
||||
else sc.voice_bot_deflection
|
||||
)
|
||||
deflected_calls = s.voice_volume_monthly * MONTHS_PER_YEAR * bot_rate
|
||||
|
||||
else: # Agentic Virtual Agent
|
||||
# VA operates on the residual after the bot has deflected its share.
|
||||
# If Voice Bot is not in scope (VA-only deployment), bot_rate = 0
|
||||
# and the VA works on the full pool — still correct.
|
||||
bot_rate = sc.voice_bot_deflection
|
||||
va_rate = (
|
||||
feature_scope.deflection_target
|
||||
if feature_scope.deflection_target is not None
|
||||
else sc.agentic_va_deflection
|
||||
)
|
||||
residual_calls = (
|
||||
s.voice_volume_monthly * MONTHS_PER_YEAR * (1.0 - bot_rate)
|
||||
)
|
||||
deflected_calls = residual_calls * va_rate
|
||||
|
||||
seconds_saved = deflected_calls * s.voice_aht_seconds * realization
|
||||
rows.append(
|
||||
{
|
||||
"benefit_line": f"{feature_scope.feature} deflection (labour avoided)",
|
||||
"scope": s.site_name,
|
||||
"annual_value": (
|
||||
seconds_saved
|
||||
* s.agent_cost_per_second
|
||||
* realization_factor
|
||||
* ro.fraction_live(s.site_name, year)
|
||||
),
|
||||
"confidence": Confidence.ESTIMATED.value,
|
||||
}
|
||||
)
|
||||
return _df(rows)
|
||||
|
||||
|
||||
def calculate_supervisor_copilot_benefit(
|
||||
sites: list[SiteInput],
|
||||
feature_scope: FeatureScope,
|
||||
scenario: str | Scenario,
|
||||
year: int,
|
||||
params: str = "realistic",
|
||||
rollout: RolloutPlan | None = None,
|
||||
) -> pd.DataFrame:
|
||||
"""Supervisor time reclaimed (summaries, QA triage). ESTIMATED."""
|
||||
sc = get_scenario(scenario) if isinstance(scenario, str) else scenario
|
||||
ro = rollout or NO_ROLLOUT
|
||||
saving = _param("supervisor_copilot_time_saving", params)
|
||||
realization = sc.realization(year)
|
||||
rows = []
|
||||
for s in sites:
|
||||
if not feature_scope.active(s.site_name, year):
|
||||
continue
|
||||
rows.append(
|
||||
{
|
||||
"benefit_line": "Supervisor time (AI summaries/insights)",
|
||||
"scope": s.site_name,
|
||||
"annual_value": s.supervisors
|
||||
* s.fully_loaded_supervisor_cost_annual
|
||||
* saving * realization
|
||||
* ro.fraction_live(s.site_name, year),
|
||||
"confidence": Confidence.ESTIMATED.value,
|
||||
}
|
||||
)
|
||||
return _df(rows)
|
||||
|
||||
|
||||
def calculate_predictive_routing_benefit(
|
||||
sites: list[SiteInput],
|
||||
feature_scope: FeatureScope,
|
||||
scenario: str | Scenario,
|
||||
year: int,
|
||||
params: str = "realistic",
|
||||
rollout: RolloutPlan | None = None,
|
||||
) -> pd.DataFrame:
|
||||
"""Predictive routing AHT effect. ESTIMATED; off unless scoped."""
|
||||
sc = get_scenario(scenario) if isinstance(scenario, str) else scenario
|
||||
ro = rollout or NO_ROLLOUT
|
||||
reduction = _param("predictive_routing_aht_reduction", params)
|
||||
realization = sc.realization(year)
|
||||
rows = []
|
||||
for s in sites:
|
||||
if not feature_scope.active(s.site_name, year):
|
||||
continue
|
||||
seconds_saved = (
|
||||
s.voice_volume_monthly * MONTHS_PER_YEAR
|
||||
* s.voice_aht_seconds * reduction * realization
|
||||
)
|
||||
rows.append(
|
||||
{
|
||||
"benefit_line": "Predictive routing (AHT)",
|
||||
"scope": s.site_name,
|
||||
"annual_value": seconds_saved * s.agent_cost_per_second
|
||||
* ro.fraction_live(s.site_name, year),
|
||||
"confidence": Confidence.ESTIMATED.value,
|
||||
}
|
||||
)
|
||||
return _df(rows)
|
||||
|
||||
|
||||
#: Which calculator handles which feature scope.
|
||||
#: Agent Copilot and STA exist in named/concurrent variants — both map
|
||||
#: to the same benefit calculators.
|
||||
#: Voice Bot and Agentic VA both route to calculate_va_deflection_benefit,
|
||||
#: which implements the layered sequential model — VA operates on the
|
||||
#: residual after the bot has deflected its share.
|
||||
_BENEFIT_DISPATCH = {
|
||||
"Agent Copilot [named]": (
|
||||
calculate_voice_handle_time_benefit,
|
||||
calculate_acw_summarization_benefit,
|
||||
),
|
||||
"Agent Copilot [concurrent]": (
|
||||
calculate_voice_handle_time_benefit,
|
||||
calculate_acw_summarization_benefit,
|
||||
),
|
||||
"AI Summary & Insights": (), # benefit carried by Copilot where present
|
||||
"Speech & Text Analytics [named]": (calculate_sta_benefit,),
|
||||
"Speech & Text Analytics [concurrent]": (calculate_sta_benefit,),
|
||||
"Voice Bot": (calculate_va_deflection_benefit,),
|
||||
"Agentic Virtual Agent": (calculate_va_deflection_benefit,),
|
||||
"Predictive Routing": (calculate_predictive_routing_benefit,),
|
||||
}
|
||||
|
||||
_COPILOT_FEATURES = {"Agent Copilot [named]", "Agent Copilot [concurrent]"}
|
||||
|
||||
|
||||
def calculate_total_benefit(
|
||||
sites: list[SiteInput],
|
||||
feature_scopes: list[FeatureScope],
|
||||
scenario: str | Scenario,
|
||||
year: int,
|
||||
params: str = "realistic",
|
||||
include_supervisor_benefit: bool = True,
|
||||
rollout: RolloutPlan | None = None,
|
||||
) -> pd.DataFrame:
|
||||
"""All benefit lines for one scenario-year, aggregated per line.
|
||||
|
||||
Returns DataFrame: benefit_line, scope, annual_value, confidence.
|
||||
|
||||
Voice Bot and Agentic VA deflection benefits use the layered
|
||||
sequential model: the bot deflects from the full call pool; the VA
|
||||
deflects from the residual. The two features are NOT additive on
|
||||
the same base — see :func:`calculate_va_deflection_benefit`.
|
||||
"""
|
||||
sc = get_scenario(scenario) if isinstance(scenario, str) else scenario
|
||||
frames: list[pd.DataFrame] = []
|
||||
# Find whichever Copilot variant is in scope (named or concurrent).
|
||||
copilot_scope = next(
|
||||
(s for s in feature_scopes if s.feature in _COPILOT_FEATURES), None
|
||||
)
|
||||
|
||||
for scope in feature_scopes:
|
||||
for fn in _BENEFIT_DISPATCH.get(scope.feature, ()): # type: ignore[arg-type]
|
||||
frames.append(fn(sites, scope, sc, year, params=params, rollout=rollout))
|
||||
|
||||
if include_supervisor_benefit and copilot_scope is not None:
|
||||
frames.append(
|
||||
calculate_supervisor_copilot_benefit(
|
||||
sites, copilot_scope, sc, year, params=params, rollout=rollout
|
||||
)
|
||||
)
|
||||
|
||||
frames = [f for f in frames if not f.empty]
|
||||
if not frames:
|
||||
return _df([])
|
||||
detail = pd.concat(frames, ignore_index=True)
|
||||
|
||||
grouped = (
|
||||
detail.groupby("benefit_line", sort=False)
|
||||
.agg(
|
||||
scope=("scope", lambda v: ", ".join(sorted(set(v)))),
|
||||
annual_value=("annual_value", "sum"),
|
||||
confidence=("confidence", "first"),
|
||||
)
|
||||
.reset_index()
|
||||
)
|
||||
return grouped[["benefit_line", "scope", "annual_value", "confidence"]]
|
||||
188
studies/202607_CTM_GenesysCX/tokencalc/business_case.py
Normal file
188
studies/202607_CTM_GenesysCX/tokencalc/business_case.py
Normal file
@@ -0,0 +1,188 @@
|
||||
"""
|
||||
Business case — combines costs, benefits, and cost takeouts into a
|
||||
3-year net view with NPV, payback, and ROI.
|
||||
|
||||
Convention: all cashflows are year-end and discounted at
|
||||
``discount_rate`` (default 8%); there is no undiscounted year-0 column
|
||||
— implementation is amortized straight-line across the analysis years
|
||||
(spec §5.6 "Implementation amort." line).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from .benefit_model import calculate_total_benefit
|
||||
from .cost_model import calculate_total_cost
|
||||
from .defaults import (
|
||||
DEFAULT_DISCOUNT_RATE,
|
||||
DEFAULT_IMPLEMENTATION_COST,
|
||||
PLATFORM_RATE_PER_USER_MONTHLY,
|
||||
)
|
||||
from .inputs import CostTakeout, FeatureScope, SiteInput
|
||||
from .meters import Confidence, TokenMeter, TokenPricing
|
||||
from .rollout import RolloutPlan
|
||||
from .scenarios import Scenario, get_scenario
|
||||
|
||||
|
||||
def npv(cashflows_by_year: list[float], discount_rate: float) -> float:
|
||||
"""Year-end-discounted NPV of year-1..N cashflows."""
|
||||
return sum(
|
||||
cf / (1 + discount_rate) ** year
|
||||
for year, cf in enumerate(cashflows_by_year, start=1)
|
||||
)
|
||||
|
||||
|
||||
def payback_years(cashflows_by_year: list[float]) -> float | None:
|
||||
"""First (fractional) year cumulative net turns >= 0; None if never.
|
||||
|
||||
Cashflows are assumed evenly spread within each year.
|
||||
"""
|
||||
cumulative = 0.0
|
||||
for year, cf in enumerate(cashflows_by_year, start=1):
|
||||
if cumulative + cf >= 0 and cf != 0:
|
||||
if cumulative >= 0:
|
||||
return float(year - 1)
|
||||
return (year - 1) + (-cumulative / cf)
|
||||
cumulative += cf
|
||||
return None
|
||||
|
||||
|
||||
def build_business_case(
|
||||
sites: list[SiteInput],
|
||||
feature_scopes: list[FeatureScope],
|
||||
meters: dict[str, TokenMeter],
|
||||
pricing: dict[str, TokenPricing],
|
||||
takeouts: list[CostTakeout],
|
||||
scenario: str | Scenario,
|
||||
years: int = 3,
|
||||
discount_rate: float = DEFAULT_DISCOUNT_RATE,
|
||||
platform_rate: float = PLATFORM_RATE_PER_USER_MONTHLY,
|
||||
implementation_cost: float = DEFAULT_IMPLEMENTATION_COST,
|
||||
use_contracted: bool = False,
|
||||
benefit_params: str = "realistic",
|
||||
rollout: RolloutPlan | None = None,
|
||||
) -> dict:
|
||||
"""Returns the dict described in spec §4.3 (DataFrames + headline
|
||||
metrics). Every number traces to a cost line, benefit line, or
|
||||
takeout row in the per-year detail frames.
|
||||
"""
|
||||
sc = get_scenario(scenario) if isinstance(scenario, str) else scenario
|
||||
year_cols = [f"Y{y}" for y in range(1, years + 1)]
|
||||
|
||||
cost_frames, benefit_frames = {}, {}
|
||||
for y in range(1, years + 1):
|
||||
cost_frames[y] = calculate_total_cost(
|
||||
sites, feature_scopes, meters, pricing, sc, y,
|
||||
platform_rate=platform_rate, use_contracted=use_contracted,
|
||||
rollout=rollout,
|
||||
)
|
||||
benefit_frames[y] = calculate_total_benefit(
|
||||
sites, feature_scopes, sc, y, params=benefit_params,
|
||||
rollout=rollout,
|
||||
)
|
||||
|
||||
# ── cost_by_year: one row per cost line, one column per year ────
|
||||
cost_lines = list(cost_frames[1]["cost_line"])
|
||||
cost_by_year = pd.DataFrame({"line": cost_lines})
|
||||
for y in range(1, years + 1):
|
||||
cost_by_year[f"Y{y}"] = list(cost_frames[y]["annual_cost"])
|
||||
cost_by_year["confidence"] = list(cost_frames[1]["confidence"])
|
||||
if implementation_cost:
|
||||
amort = implementation_cost / years
|
||||
cost_by_year = pd.concat(
|
||||
[
|
||||
cost_by_year,
|
||||
pd.DataFrame(
|
||||
[
|
||||
{
|
||||
"line": "Implementation (amortized)",
|
||||
**{c: amort for c in year_cols},
|
||||
"confidence": Confidence.ESTIMATED.value,
|
||||
}
|
||||
]
|
||||
),
|
||||
],
|
||||
ignore_index=True,
|
||||
)
|
||||
|
||||
# ── benefit_by_year ──────────────────────────────────────────────
|
||||
benefit_lines: list[str] = []
|
||||
for y in range(1, years + 1):
|
||||
for line in benefit_frames[y]["benefit_line"]:
|
||||
if line not in benefit_lines:
|
||||
benefit_lines.append(line)
|
||||
benefit_by_year = pd.DataFrame({"line": benefit_lines})
|
||||
for y in range(1, years + 1):
|
||||
lookup = dict(
|
||||
zip(benefit_frames[y]["benefit_line"], benefit_frames[y]["annual_value"])
|
||||
)
|
||||
benefit_by_year[f"Y{y}"] = [lookup.get(line, 0.0) for line in benefit_lines]
|
||||
conf_lookup: dict[str, str] = {}
|
||||
for y in range(1, years + 1):
|
||||
conf_lookup.update(
|
||||
dict(zip(benefit_frames[y]["benefit_line"], benefit_frames[y]["confidence"]))
|
||||
)
|
||||
benefit_by_year["confidence"] = [
|
||||
conf_lookup.get(line, Confidence.ESTIMATED.value) for line in benefit_lines
|
||||
]
|
||||
|
||||
# ── takeouts_by_year ─────────────────────────────────────────────
|
||||
takeouts_by_year = pd.DataFrame(
|
||||
[
|
||||
{
|
||||
"line": t.name,
|
||||
**{f"Y{y}": t.value_in_year(y) for y in range(1, years + 1)},
|
||||
"confidence": t.confidence.value,
|
||||
}
|
||||
for t in takeouts
|
||||
]
|
||||
)
|
||||
|
||||
# ── net + cumulative ─────────────────────────────────────────────
|
||||
total_costs = [float(cost_by_year[c].sum()) for c in year_cols]
|
||||
total_benefits = [float(benefit_by_year[c].sum()) for c in year_cols]
|
||||
total_takeouts = [
|
||||
float(takeouts_by_year[c].sum()) if not takeouts_by_year.empty else 0.0
|
||||
for c in year_cols
|
||||
]
|
||||
net = [
|
||||
b + t - c for b, t, c in zip(total_benefits, total_takeouts, total_costs)
|
||||
]
|
||||
cumulative = pd.Series(net).cumsum().tolist()
|
||||
|
||||
net_by_year = pd.DataFrame(
|
||||
{
|
||||
"line": [
|
||||
"TOTAL COSTS", "TOTAL TAKEOUTS", "TOTAL BENEFITS",
|
||||
"NET", "Cumulative net",
|
||||
],
|
||||
**{
|
||||
f"Y{y}": [
|
||||
total_costs[y - 1], total_takeouts[y - 1],
|
||||
total_benefits[y - 1], net[y - 1], cumulative[y - 1],
|
||||
]
|
||||
for y in range(1, years + 1)
|
||||
},
|
||||
}
|
||||
)
|
||||
cumulative_net = pd.DataFrame(
|
||||
{"year": list(range(1, years + 1)), "cumulative_net": cumulative}
|
||||
)
|
||||
|
||||
total_cost_sum = sum(total_costs)
|
||||
total_value_sum = sum(total_benefits) + sum(total_takeouts)
|
||||
return {
|
||||
"cost_by_year": cost_by_year,
|
||||
"benefit_by_year": benefit_by_year,
|
||||
"takeouts_by_year": takeouts_by_year,
|
||||
"net_by_year": net_by_year,
|
||||
"cumulative_net": cumulative_net,
|
||||
"npv": npv(net, discount_rate),
|
||||
"payback_period_years": payback_years(net),
|
||||
"roi_3yr": (
|
||||
(total_value_sum - total_cost_sum) / total_cost_sum
|
||||
if total_cost_sum
|
||||
else None
|
||||
),
|
||||
}
|
||||
317
studies/202607_CTM_GenesysCX/tokencalc/cost_model.py
Normal file
317
studies/202607_CTM_GenesysCX/tokencalc/cost_model.py
Normal file
@@ -0,0 +1,317 @@
|
||||
"""
|
||||
Cost calculation engine.
|
||||
|
||||
Correctness rules implemented here (see spec §4.1):
|
||||
|
||||
1. **Agent Copilot covers Supervisor AI Summary.** Where Agent Copilot
|
||||
is enabled at a site, AI Summary & Insights consumption at that site
|
||||
is forced to zero — Copilot's per-user token rate already includes
|
||||
interaction summarization. Source: Genesys Cloud AI Experience
|
||||
token metering,
|
||||
https://help.genesys.cloud/articles/genesys-cloud-tokens-model/
|
||||
2. **Token rounding.** Genesys rounds consumption up at billing —
|
||||
``math.ceil`` is applied to each site's MONTHLY consumption token
|
||||
total before the rate. Per-user totals (users × tokens/user/month)
|
||||
are exact and not rounded.
|
||||
3. **Regional pricing.** Every site resolves its rate through its
|
||||
``region_pricing`` key — never a hardcoded US rate.
|
||||
4. **Adoption ramp.** Consumption features ramp (default Y1 = 70%);
|
||||
per-user licences are paid in full from their phase year.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from .defaults import PLATFORM_RATE_PER_USER_MONTHLY
|
||||
from .inputs import FeatureScope, SiteInput
|
||||
from .meters import Confidence, MeterType, TokenMeter, TokenPricing
|
||||
from .rollout import NO_ROLLOUT, RolloutPlan
|
||||
from .scenarios import Scenario, get_scenario
|
||||
|
||||
MONTHS_PER_YEAR = 12
|
||||
|
||||
|
||||
def _rate(site: SiteInput, pricing: dict[str, TokenPricing],
|
||||
use_contracted: bool = False) -> float:
|
||||
"""Resolve the per-token rate for a site's pricing region."""
|
||||
region = pricing.get(site.region_pricing)
|
||||
if region is None:
|
||||
raise KeyError(
|
||||
f"No TokenPricing for region {site.region_pricing!r} "
|
||||
f"(site {site.site_name})"
|
||||
)
|
||||
return region.effective_rate(use_contracted)
|
||||
|
||||
|
||||
def calculate_platform_license_cost(
|
||||
sites: list[SiteInput],
|
||||
per_user_monthly_rate: float = PLATFORM_RATE_PER_USER_MONTHLY,
|
||||
year: int = 1,
|
||||
rollout: RolloutPlan | None = None,
|
||||
) -> pd.DataFrame:
|
||||
"""Genesys Cloud CX 3 named-user platform licences.
|
||||
|
||||
The commit bills in full from contract start regardless of site
|
||||
go-lives; the vendor ramp credit reduces YEAR 1 only (typical
|
||||
6-month ramp → 50% Y1 discount).
|
||||
Returns DataFrame: site, agents, supervisors, named_users, annual_cost.
|
||||
"""
|
||||
ro = rollout or NO_ROLLOUT
|
||||
factor = ro.platform_factor(year)
|
||||
rows = [
|
||||
{
|
||||
"site": s.site_name,
|
||||
"agents": s.agents,
|
||||
"supervisors": s.supervisors,
|
||||
"named_users": s.named_users,
|
||||
"annual_cost": s.named_users
|
||||
* per_user_monthly_rate
|
||||
* MONTHS_PER_YEAR
|
||||
* factor,
|
||||
}
|
||||
for s in sites
|
||||
]
|
||||
return pd.DataFrame(rows)
|
||||
|
||||
|
||||
def calculate_per_user_ai_cost(
|
||||
sites: list[SiteInput],
|
||||
feature_scope: FeatureScope,
|
||||
meter: TokenMeter,
|
||||
pricing: dict[str, TokenPricing],
|
||||
year: int = 1,
|
||||
use_contracted: bool = False,
|
||||
rollout: RolloutPlan | None = None,
|
||||
) -> pd.DataFrame:
|
||||
"""Per-user-per-month AI features (STA, Agent Copilot, AI Translate).
|
||||
|
||||
No adoption ramp and no rounding (users × tokens/user/month is
|
||||
exact) — but token usage only starts at site go-live, so the year
|
||||
bills for the months the site is live (``rollout``).
|
||||
Returns DataFrame: site, users_in_scope, tokens_monthly, annual_cost.
|
||||
"""
|
||||
if meter.meter_type is not MeterType.PER_USER_PER_MONTH:
|
||||
raise ValueError(f"{meter.feature} is not a per-user meter")
|
||||
ro = rollout or NO_ROLLOUT
|
||||
rows = []
|
||||
for s in sites:
|
||||
in_scope = feature_scope.active(s.site_name, year)
|
||||
users = s.named_users if in_scope else 0
|
||||
live_months = ro.live_months_in_year(s.site_name, year)
|
||||
tokens_monthly = users * meter.tokens_per_unit
|
||||
rows.append(
|
||||
{
|
||||
"site": s.site_name,
|
||||
"users_in_scope": users,
|
||||
"tokens_monthly": tokens_monthly,
|
||||
"annual_cost": tokens_monthly
|
||||
* live_months
|
||||
* _rate(s, pricing, use_contracted),
|
||||
}
|
||||
)
|
||||
return pd.DataFrame(rows)
|
||||
|
||||
|
||||
def _monthly_units(site: SiteInput, feature: str, scope: FeatureScope,
|
||||
scenario: Scenario) -> float:
|
||||
"""Monthly metered units for a consumption feature at one site.
|
||||
|
||||
Explicit ``scope.deflection_target`` / ``scope.eligibility_pct``
|
||||
override the scenario defaults.
|
||||
"""
|
||||
if feature == "Voice Bot":
|
||||
deflection = (
|
||||
scope.deflection_target
|
||||
if scope.deflection_target is not None
|
||||
else scenario.voice_bot_deflection
|
||||
)
|
||||
return (
|
||||
site.voice_volume_monthly * deflection * scenario.voice_bot_avg_minutes
|
||||
) # minutes
|
||||
if feature == "Agentic Virtual Agent":
|
||||
# Layered model: VA operates on the residual volume after the voice bot
|
||||
# has already deflected its share. Cost base = residual × va_rate.
|
||||
# This is consistent with the benefit model and avoids double-counting
|
||||
# the same call pool across both deflection mechanisms.
|
||||
bot_deflection = scenario.voice_bot_deflection
|
||||
va_deflection = (
|
||||
scope.deflection_target
|
||||
if scope.deflection_target is not None
|
||||
else scenario.agentic_va_deflection
|
||||
)
|
||||
residual = site.voice_volume_monthly * (1.0 - bot_deflection)
|
||||
return residual * va_deflection # interactions
|
||||
if feature == "Virtual Agent (legacy)":
|
||||
deflection = scope.deflection_target or 0.0
|
||||
return site.voice_volume_monthly * deflection
|
||||
if feature == "AI Summary & Insights":
|
||||
eligibility = (
|
||||
scope.eligibility_pct
|
||||
if scope.eligibility_pct is not None
|
||||
else scenario.voice_summarization_eligibility
|
||||
)
|
||||
return site.voice_volume_monthly * eligibility # summaries
|
||||
if feature == "Email AI (Auto-Respond)":
|
||||
rate = (
|
||||
scope.deflection_target
|
||||
if scope.deflection_target is not None
|
||||
else scenario.email_auto_respond_rate
|
||||
)
|
||||
return site.email_volume_monthly * rate # messages
|
||||
if feature in ("Direct Messaging", "Social Listening", "Social Responses"):
|
||||
eligibility = scope.eligibility_pct if scope.eligibility_pct is not None else 1.0
|
||||
return (site.chat_volume_monthly + site.sms_volume_monthly) * eligibility
|
||||
if feature == "AI Translate":
|
||||
# Each voice interaction generates one translation; eligibility_pct
|
||||
# can be used to scope to a subset of interactions (e.g. non-English only).
|
||||
eligibility = scope.eligibility_pct if scope.eligibility_pct is not None else 1.0
|
||||
return site.voice_volume_monthly * eligibility # translations
|
||||
if feature == "Predictive Routing":
|
||||
# Every predictively-routed voice interaction consumes the PR meter;
|
||||
# eligibility_pct scopes to the share of volume on PR-enabled queues.
|
||||
eligibility = scope.eligibility_pct if scope.eligibility_pct is not None else 1.0
|
||||
return site.voice_volume_monthly * eligibility # routed interactions
|
||||
raise KeyError(f"No consumption-volume mapping for feature {feature!r}")
|
||||
|
||||
|
||||
def calculate_consumption_ai_cost(
|
||||
sites: list[SiteInput],
|
||||
feature_scope: FeatureScope,
|
||||
meter: TokenMeter,
|
||||
scenario: str | Scenario,
|
||||
pricing: dict[str, TokenPricing],
|
||||
year: int = 1,
|
||||
use_contracted: bool = False,
|
||||
excluded_sites: set[str] | None = None,
|
||||
rollout: RolloutPlan | None = None,
|
||||
) -> pd.DataFrame:
|
||||
"""Consumption-metered AI features (Voice Bots, Agentic VA,
|
||||
Supervisor AI Summary, Email Auto-Respond, messaging meters).
|
||||
|
||||
Applies eligibility/deflection from the scenario (or explicit scope
|
||||
overrides), the adoption ramp, billing-style ``ceil`` rounding on
|
||||
each site's monthly token total, and — with a ``rollout`` — bills
|
||||
only the months the site is live (usage starts at go-live).
|
||||
|
||||
``excluded_sites`` supports the Copilot-covers-Summary rule.
|
||||
Returns DataFrame: site, eligible_volume, tokens_monthly, annual_cost.
|
||||
"""
|
||||
if meter.meter_type is MeterType.PER_USER_PER_MONTH:
|
||||
raise ValueError(f"{meter.feature} is a per-user meter, not consumption")
|
||||
sc = get_scenario(scenario) if isinstance(scenario, str) else scenario
|
||||
excluded = excluded_sites or set()
|
||||
ro = rollout or NO_ROLLOUT
|
||||
|
||||
# Ramp: an explicit adoption curve wins; otherwise the scenario's
|
||||
# default consumption realization (Y1 = 70%). This models usage
|
||||
# maturity; rollout live-months model calendar availability — they
|
||||
# compound (live 6 months × 70% maturity).
|
||||
ramp = (
|
||||
feature_scope.adoption(year)
|
||||
if feature_scope.adoption_curve
|
||||
else sc.cost_realization(year)
|
||||
)
|
||||
|
||||
rows = []
|
||||
for s in sites:
|
||||
active = (
|
||||
feature_scope.active(s.site_name, year)
|
||||
and s.site_name not in excluded
|
||||
)
|
||||
units = _monthly_units(s, meter.feature, feature_scope, sc) if active else 0.0
|
||||
units *= ramp
|
||||
live_months = ro.live_months_in_year(s.site_name, year)
|
||||
# Rule 2: round each site's monthly token total UP (billing).
|
||||
tokens_monthly = math.ceil(units * meter.tokens_per_unit) if units > 0 else 0
|
||||
rows.append(
|
||||
{
|
||||
"site": s.site_name,
|
||||
"eligible_volume": units,
|
||||
"tokens_monthly": tokens_monthly,
|
||||
"annual_cost": tokens_monthly
|
||||
* live_months
|
||||
* _rate(s, pricing, use_contracted),
|
||||
}
|
||||
)
|
||||
return pd.DataFrame(rows)
|
||||
|
||||
|
||||
def calculate_total_cost(
|
||||
sites: list[SiteInput],
|
||||
feature_scopes: list[FeatureScope],
|
||||
meters: dict[str, TokenMeter],
|
||||
pricing: dict[str, TokenPricing],
|
||||
scenario: str | Scenario,
|
||||
year: int,
|
||||
platform_rate: float = PLATFORM_RATE_PER_USER_MONTHLY,
|
||||
use_contracted: bool = False,
|
||||
include_platform: bool = True,
|
||||
rollout: RolloutPlan | None = None,
|
||||
) -> pd.DataFrame:
|
||||
"""All cost lines for one scenario-year.
|
||||
|
||||
Returns DataFrame: cost_line, scope, annual_cost, confidence.
|
||||
"""
|
||||
sc = get_scenario(scenario) if isinstance(scenario, str) else scenario
|
||||
rows: list[dict] = []
|
||||
|
||||
if include_platform:
|
||||
platform = calculate_platform_license_cost(
|
||||
sites, platform_rate, year=year, rollout=rollout
|
||||
)
|
||||
ramped = rollout is not None and rollout.platform_factor(year) < 1.0
|
||||
rows.append(
|
||||
{
|
||||
"cost_line": "Genesys CX 3 platform licences"
|
||||
+ (" (ramp credit applied)" if ramped else ""),
|
||||
"scope": "all sites",
|
||||
"annual_cost": float(platform["annual_cost"].sum()),
|
||||
"confidence": Confidence.CONFIRMED.value,
|
||||
}
|
||||
)
|
||||
|
||||
# Rule 1: Agent Copilot covers Supervisor AI Summary. Sites where
|
||||
# Copilot is active this year are excluded from AI Summary billing —
|
||||
# Copilot's per-user token rate already includes interaction summarization.
|
||||
# https://help.genesys.cloud/articles/genesys-cloud-tokens-model/
|
||||
_COPILOT_FEATURES = {"Agent Copilot [named]", "Agent Copilot [concurrent]"}
|
||||
copilot_sites: set[str] = set()
|
||||
for scope in feature_scopes:
|
||||
if scope.feature in _COPILOT_FEATURES:
|
||||
copilot_sites |= {
|
||||
s.site_name for s in sites if scope.active(s.site_name, year)
|
||||
}
|
||||
|
||||
for scope in feature_scopes:
|
||||
meter = meters.get(scope.feature)
|
||||
if meter is None:
|
||||
raise KeyError(f"No meter defined for feature {scope.feature!r}")
|
||||
if meter.meter_type is MeterType.PER_USER_PER_MONTH:
|
||||
df = calculate_per_user_ai_cost(
|
||||
sites, scope, meter, pricing, year=year,
|
||||
use_contracted=use_contracted, rollout=rollout,
|
||||
)
|
||||
in_scope = df[df["users_in_scope"] > 0]["site"].tolist()
|
||||
else:
|
||||
excluded = (
|
||||
copilot_sites if scope.feature == "AI Summary & Insights" else None
|
||||
)
|
||||
df = calculate_consumption_ai_cost(
|
||||
sites, scope, meter, sc, pricing, year=year,
|
||||
use_contracted=use_contracted, excluded_sites=excluded,
|
||||
rollout=rollout,
|
||||
)
|
||||
in_scope = df[df["annual_cost"] > 0]["site"].tolist()
|
||||
rows.append(
|
||||
{
|
||||
"cost_line": scope.feature,
|
||||
"scope": ", ".join(in_scope) if in_scope else "—",
|
||||
"annual_cost": float(df["annual_cost"].sum()),
|
||||
"confidence": meter.confidence.value,
|
||||
}
|
||||
)
|
||||
|
||||
return pd.DataFrame(rows)
|
||||
423
studies/202607_CTM_GenesysCX/tokencalc/defaults.py
Normal file
423
studies/202607_CTM_GenesysCX/tokencalc/defaults.py
Normal file
@@ -0,0 +1,423 @@
|
||||
"""
|
||||
CTM default inputs and the Genesys meter catalogue.
|
||||
|
||||
⚠️ Site volumes/AHTs/costs outside NAM are PLACEHOLDERS flagged
|
||||
ESTIMATED — confirm with CTM data before client use. NAM volumes are
|
||||
from the CTM discovery pack. Named users across all sites total the
|
||||
contracted licence count (2,088).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from .inputs import CostTakeout, FeatureScope, SiteInput
|
||||
from .meters import Confidence, MeterType, TokenMeter, TokenPricing
|
||||
from .rollout import RolloutPlan
|
||||
|
||||
# ── Platform ─────────────────────────────────────────────────────────
|
||||
|
||||
#: Genesys Cloud CX 3 named-user list rate, USD/user/month.
|
||||
#: Source: Genesys Cloud public pricing (CX 3 tier), planning figure.
|
||||
PLATFORM_RATE_PER_USER_MONTHLY = 111.28
|
||||
|
||||
#: CTM contracted named-user count — UI warns when site totals diverge.
|
||||
CONTRACTED_NAMED_USERS = 2_088
|
||||
|
||||
#: Business-case discount rate (CTM treasury planning assumption).
|
||||
DEFAULT_DISCOUNT_RATE = 0.08
|
||||
|
||||
#: One-off implementation estimate, amortized straight-line over the
|
||||
#: analysis horizon in the P&L. ESTIMATED — confirm with delivery team.
|
||||
DEFAULT_IMPLEMENTATION_COST = 0.0
|
||||
|
||||
_GENESYS_TOKEN_METERS = (
|
||||
"https://help.genesys.cloud/articles/genesys-cloud-tokens-model/"
|
||||
)
|
||||
|
||||
# ── Token meters ─────────────────────────────────────────────────────
|
||||
# Rates per the published Genesys AI Experience token tables unless
|
||||
# flagged otherwise. UNKNOWN meters carry working defaults (clearly
|
||||
# labelled) so the model still produces a range.
|
||||
|
||||
DEFAULT_METERS: dict[str, TokenMeter] = {
|
||||
m.feature: m
|
||||
for m in [
|
||||
# ── Voice / Bot ───────────────────────────────────────────────
|
||||
TokenMeter(
|
||||
feature="Voice Bot",
|
||||
meter_type=MeterType.PER_MINUTE,
|
||||
units_per_token=17.0,
|
||||
tokens_per_unit=1 / 17, # 0.0588
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes="IVR self-service voice bot minutes; 17 min per token.",
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
TokenMeter(
|
||||
feature="Digital Bot",
|
||||
meter_type=MeterType.PER_INTERACTION,
|
||||
units_per_token=51.0,
|
||||
tokens_per_unit=1 / 51, # 0.0196
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes="Digital (non-voice) bot sessions; 51 sessions per token.",
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
# ── Virtual Agent ─────────────────────────────────────────────
|
||||
TokenMeter(
|
||||
feature="Virtual Agent (legacy)",
|
||||
meter_type=MeterType.PER_INTERACTION,
|
||||
units_per_token=2.0,
|
||||
tokens_per_unit=0.5,
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes="Legacy (non-agentic) virtual agent; 0.5 tokens per interaction.",
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
TokenMeter(
|
||||
feature="Agentic Virtual Agent",
|
||||
meter_type=MeterType.PER_INTERACTION,
|
||||
units_per_token=0.833,
|
||||
tokens_per_unit=1.2,
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes="Agentic VA; 1.2 tokens per interaction.",
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
# ── Agent Copilot (named vs concurrent) ───────────────────────
|
||||
TokenMeter(
|
||||
feature="Agent Copilot [named]",
|
||||
meter_type=MeterType.PER_USER_PER_MONTH,
|
||||
units_per_token=0.0,
|
||||
tokens_per_unit=40.0,
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes=(
|
||||
"40 tokens per named user per month. Includes interaction "
|
||||
"summarization (covers AI Summary & Insights)."
|
||||
),
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
TokenMeter(
|
||||
feature="Agent Copilot [concurrent]",
|
||||
meter_type=MeterType.PER_USER_PER_MONTH,
|
||||
units_per_token=0.0,
|
||||
tokens_per_unit=60.0,
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes=(
|
||||
"60 tokens per concurrent user per month. Includes interaction "
|
||||
"summarization (covers AI Summary & Insights)."
|
||||
),
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
# ── AI Quality / Analytics ────────────────────────────────────
|
||||
TokenMeter(
|
||||
feature="AI Scoring",
|
||||
meter_type=MeterType.PER_INTERACTION,
|
||||
units_per_token=20.0,
|
||||
tokens_per_unit=0.05,
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes="AI-scored quality evaluations; 20 evaluations per token.",
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
TokenMeter(
|
||||
feature="AI Summary & Insights",
|
||||
meter_type=MeterType.PER_SUMMARY,
|
||||
units_per_token=50.0,
|
||||
tokens_per_unit=0.02,
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes=(
|
||||
"Supervisor standalone summarization; 50 summaries per token. "
|
||||
"NOT metered where Agent Copilot is assigned — see cost model."
|
||||
),
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
# ── Speech & Text Analytics (named vs concurrent) ─────────────
|
||||
TokenMeter(
|
||||
feature="Speech & Text Analytics [named]",
|
||||
meter_type=MeterType.PER_USER_PER_MONTH,
|
||||
units_per_token=0.0,
|
||||
tokens_per_unit=30.0,
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes="STA named licence; 30 tokens per named user per month.",
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
TokenMeter(
|
||||
feature="Speech & Text Analytics [concurrent]",
|
||||
meter_type=MeterType.PER_USER_PER_MONTH,
|
||||
units_per_token=0.0,
|
||||
tokens_per_unit=45.0,
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes="STA concurrent licence; 45 tokens per concurrent user per month.",
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
# ── Routing / Engagement ──────────────────────────────────────
|
||||
TokenMeter(
|
||||
feature="Predictive Routing",
|
||||
meter_type=MeterType.PER_INTERACTION,
|
||||
units_per_token=17.0,
|
||||
tokens_per_unit=1 / 17, # 0.0588
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes="Predictive routing; 17 routes per token.",
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
# ── Messaging ─────────────────────────────────────────────────
|
||||
TokenMeter(
|
||||
feature="Direct Messaging",
|
||||
meter_type=MeterType.PER_MESSAGE,
|
||||
units_per_token=400.0,
|
||||
tokens_per_unit=0.0025,
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes=(
|
||||
"Apple Messages for Business, Facebook Messenger, Instagram DM, "
|
||||
"WhatsApp, and X (Twitter) DM; 400 inbound or outbound messages "
|
||||
"per token. Additional carrier charges apply for WhatsApp and X."
|
||||
),
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
TokenMeter(
|
||||
feature="Social Listening",
|
||||
meter_type=MeterType.PER_MESSAGE,
|
||||
units_per_token=400.0,
|
||||
tokens_per_unit=0.0025,
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes="Genesys Cloud Social; 400 social post ingestions per channel per token.",
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
TokenMeter(
|
||||
feature="Social Responses",
|
||||
meter_type=MeterType.PER_MESSAGE,
|
||||
units_per_token=400.0,
|
||||
tokens_per_unit=0.0025,
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes="Social Post Responses; 400 outbound messages per channel per token.",
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
# ── Language / Translation ────────────────────────────────────
|
||||
TokenMeter(
|
||||
feature="AI Translate",
|
||||
meter_type=MeterType.PER_INTERACTION,
|
||||
units_per_token=2.0,
|
||||
tokens_per_unit=0.5,
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes="AI translation; 2 translations per token.",
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
# ── Genesys Cloud Copilot ─────────────────────────────────────
|
||||
TokenMeter(
|
||||
feature="Genesys Cloud Copilot",
|
||||
meter_type=MeterType.PER_INTERACTION,
|
||||
units_per_token=20.0,
|
||||
tokens_per_unit=0.05,
|
||||
confidence=Confidence.CONFIRMED,
|
||||
notes=(
|
||||
"20 AI actions per token; Genesys Cloud knowledge queries "
|
||||
"are not charged."
|
||||
),
|
||||
source_url=_GENESYS_TOKEN_METERS,
|
||||
),
|
||||
# ── Email AI (rate not yet published) ─────────────────────────
|
||||
# Email Auto-Suggest is included in Agent Copilot's per-user
|
||||
# meter (no standalone SKU) — only Auto-Respond is listed here.
|
||||
TokenMeter(
|
||||
feature="Email AI (Auto-Respond)",
|
||||
meter_type=MeterType.PER_MESSAGE,
|
||||
units_per_token=0.0,
|
||||
tokens_per_unit=0.0,
|
||||
confidence=Confidence.UNKNOWN,
|
||||
notes="Feature not yet available; rate TBD.",
|
||||
),
|
||||
]
|
||||
}
|
||||
|
||||
#: Features metered per named user per month.
|
||||
PER_USER_FEATURES = [
|
||||
f for f, m in DEFAULT_METERS.items()
|
||||
if m.meter_type is MeterType.PER_USER_PER_MONTH
|
||||
]
|
||||
|
||||
# ── Token pricing ────────────────────────────────────────────────────
|
||||
# $1/token US list confirmed; other regions default to the same list
|
||||
# rate until regional figures are sourced (override in UI).
|
||||
|
||||
DEFAULT_PRICING: dict[str, TokenPricing] = {
|
||||
"US": TokenPricing(region="US", list_rate_per_token=1.0),
|
||||
"EU": TokenPricing(region="EU", list_rate_per_token=1.0), # TBD — assumed US list
|
||||
"AU": TokenPricing(region="AU", list_rate_per_token=1.0), # TBD — assumed US list
|
||||
"APAC": TokenPricing(region="APAC", list_rate_per_token=1.0), # TBD
|
||||
}
|
||||
|
||||
# ── CTM sites ────────────────────────────────────────────────────────
|
||||
# NAM figures from CTM discovery. ALL OTHER SITES + every AHT/ACW and
|
||||
# labour-cost figure are ESTIMATED placeholders — confirm with CTM.
|
||||
# Named users sum to CONTRACTED_NAMED_USERS (2,088).
|
||||
|
||||
_COMMON = {
|
||||
"voice_aht_seconds": 300, # placeholder — flag as estimate
|
||||
"email_aht_seconds": 600,
|
||||
"chat_aht_seconds": 480,
|
||||
"voice_acw_seconds": 60,
|
||||
}
|
||||
|
||||
CTM_DEFAULT_SITES: list[SiteInput] = [
|
||||
SiteInput(
|
||||
"NAM", "US", agents=890, supervisors=60, # split TBD
|
||||
voice_volume_monthly=1_214_358,
|
||||
email_volume_monthly=275_800,
|
||||
chat_volume_monthly=110,
|
||||
sms_volume_monthly=1_040,
|
||||
fully_loaded_agent_cost_annual=65_000, # placeholder
|
||||
fully_loaded_supervisor_cost_annual=95_000,
|
||||
languages=["English", "French", "Spanish"],
|
||||
**_COMMON,
|
||||
),
|
||||
SiteInput(
|
||||
"EMEA", "EU", agents=320, supervisors=25,
|
||||
voice_volume_monthly=420_000,
|
||||
email_volume_monthly=95_000,
|
||||
chat_volume_monthly=40,
|
||||
sms_volume_monthly=400,
|
||||
fully_loaded_agent_cost_annual=60_000,
|
||||
fully_loaded_supervisor_cost_annual=88_000,
|
||||
languages=["English", "French", "German", "Italian", "Spanish"],
|
||||
**_COMMON,
|
||||
),
|
||||
SiteInput(
|
||||
"AUZ", "AU", agents=180, supervisors=15,
|
||||
voice_volume_monthly=250_000,
|
||||
email_volume_monthly=56_000,
|
||||
chat_volume_monthly=25,
|
||||
sms_volume_monthly=250,
|
||||
fully_loaded_agent_cost_annual=70_000,
|
||||
fully_loaded_supervisor_cost_annual=100_000,
|
||||
languages=["English"],
|
||||
**_COMMON,
|
||||
),
|
||||
SiteInput(
|
||||
"APAC HK", "APAC", agents=120, supervisors=10,
|
||||
voice_volume_monthly=160_000,
|
||||
email_volume_monthly=38_000,
|
||||
chat_volume_monthly=15,
|
||||
sms_volume_monthly=150,
|
||||
fully_loaded_agent_cost_annual=55_000,
|
||||
fully_loaded_supervisor_cost_annual=80_000,
|
||||
languages=["English", "Cantonese", "Mandarin"],
|
||||
**_COMMON,
|
||||
),
|
||||
SiteInput(
|
||||
"APAC SG", "APAC", agents=110, supervisors=10,
|
||||
voice_volume_monthly=150_000,
|
||||
email_volume_monthly=34_000,
|
||||
chat_volume_monthly=15,
|
||||
sms_volume_monthly=120,
|
||||
fully_loaded_agent_cost_annual=55_000,
|
||||
fully_loaded_supervisor_cost_annual=80_000,
|
||||
languages=["English", "Mandarin", "Malay"],
|
||||
**_COMMON,
|
||||
),
|
||||
SiteInput(
|
||||
"APAC SH", "APAC", agents=130, supervisors=10,
|
||||
voice_volume_monthly=175_000,
|
||||
email_volume_monthly=40_000,
|
||||
chat_volume_monthly=15,
|
||||
sms_volume_monthly=130,
|
||||
fully_loaded_agent_cost_annual=35_000,
|
||||
fully_loaded_supervisor_cost_annual=55_000,
|
||||
languages=["Mandarin"],
|
||||
**_COMMON,
|
||||
),
|
||||
SiteInput(
|
||||
"APAC GZ", "APAC", agents=90, supervisors=8,
|
||||
voice_volume_monthly=120_000,
|
||||
email_volume_monthly=28_000,
|
||||
chat_volume_monthly=10,
|
||||
sms_volume_monthly=100,
|
||||
fully_loaded_agent_cost_annual=35_000,
|
||||
fully_loaded_supervisor_cost_annual=55_000,
|
||||
languages=["Mandarin", "Cantonese"],
|
||||
**_COMMON,
|
||||
),
|
||||
SiteInput(
|
||||
"APAC JP", "APAC", agents=60, supervisors=6,
|
||||
voice_volume_monthly=80_000,
|
||||
email_volume_monthly=19_000,
|
||||
chat_volume_monthly=8,
|
||||
sms_volume_monthly=80,
|
||||
fully_loaded_agent_cost_annual=60_000,
|
||||
fully_loaded_supervisor_cost_annual=85_000,
|
||||
languages=["Japanese"],
|
||||
**_COMMON,
|
||||
),
|
||||
SiteInput(
|
||||
"APAC TW", "APAC", agents=40, supervisors=4,
|
||||
voice_volume_monthly=54_000,
|
||||
email_volume_monthly=12_000,
|
||||
chat_volume_monthly=5,
|
||||
sms_volume_monthly=50,
|
||||
fully_loaded_agent_cost_annual=40_000,
|
||||
fully_loaded_supervisor_cost_annual=60_000,
|
||||
languages=["Mandarin"],
|
||||
**_COMMON,
|
||||
),
|
||||
]
|
||||
|
||||
ALL_SITE_NAMES = [s.site_name for s in CTM_DEFAULT_SITES]
|
||||
|
||||
# ── Cost takeouts ────────────────────────────────────────────────────
|
||||
|
||||
CTM_DEFAULT_TAKEOUTS: list[CostTakeout] = [
|
||||
CostTakeout(
|
||||
"NICE IEX (NAM)",
|
||||
annual_cost=1_300_000,
|
||||
start_year=1,
|
||||
start_month=7, # can only switch off after NAM go-live (month 6)
|
||||
confidence=Confidence.ESTIMATED,
|
||||
notes="Mid-band estimate; needs CTM contract confirmation.",
|
||||
),
|
||||
CostTakeout(
|
||||
"Legacy CC platform",
|
||||
annual_cost=0,
|
||||
start_year=2,
|
||||
confidence=Confidence.UNKNOWN,
|
||||
notes="Placeholder — populate once retirement scope is confirmed.",
|
||||
),
|
||||
]
|
||||
|
||||
# ── Default rollout & ramp ───────────────────────────────────────────
|
||||
# 12-month build. Genesys bills the licence commit from contract start;
|
||||
# the 6-month ramp gives a 50% first-year credit on the platform commit.
|
||||
# AI token usage (and benefits) start only when each region goes live.
|
||||
|
||||
CTM_DEFAULT_ROLLOUT = RolloutPlan(
|
||||
contract_start=None, # set when known — "Date Genesys starts billing"
|
||||
build_months=12,
|
||||
ramp_months=6,
|
||||
first_year_platform_discount=0.50,
|
||||
go_live_month={
|
||||
"NAM": 6,
|
||||
"EMEA": 9,
|
||||
"AUZ": 12,
|
||||
"APAC HK": 12,
|
||||
"APAC SG": 12,
|
||||
"APAC SH": 12,
|
||||
"APAC GZ": 12,
|
||||
"APAC JP": 12,
|
||||
"APAC TW": 12,
|
||||
},
|
||||
)
|
||||
|
||||
# ── Default feature scoping / phasing ────────────────────────────────
|
||||
# Phase = model year the feature switches on. Consumption features ramp
|
||||
# via adoption_curve; per-user licences are paid in full from the phase
|
||||
# year.
|
||||
|
||||
_RAMP = {1: 0.70, 2: 1.0, 3: 1.0}
|
||||
|
||||
CTM_DEFAULT_FEATURE_SCOPES: list[FeatureScope] = [
|
||||
FeatureScope("Voice Bot", ALL_SITE_NAMES, phase=1, adoption_curve=_RAMP),
|
||||
FeatureScope("Agentic Virtual Agent", ["NAM", "EMEA"], phase=2,
|
||||
adoption_curve={2: 0.70, 3: 1.0}),
|
||||
# CTM has named licences — use the [named] variant for both STA and Copilot.
|
||||
FeatureScope("Speech & Text Analytics [named]", ALL_SITE_NAMES, phase=1),
|
||||
FeatureScope("Agent Copilot [named]", ALL_SITE_NAMES, phase=1),
|
||||
FeatureScope("AI Summary & Insights", ALL_SITE_NAMES, phase=1,
|
||||
adoption_curve=_RAMP),
|
||||
FeatureScope("Direct Messaging", ALL_SITE_NAMES, phase=1, adoption_curve=_RAMP),
|
||||
FeatureScope("AI Translate",
|
||||
["APAC HK", "APAC SG", "APAC SH", "APAC GZ", "APAC JP", "APAC TW"],
|
||||
phase=3),
|
||||
]
|
||||
131
studies/202607_CTM_GenesysCX/tokencalc/exports.py
Normal file
131
studies/202607_CTM_GenesysCX/tokencalc/exports.py
Normal file
@@ -0,0 +1,131 @@
|
||||
"""
|
||||
Excel / CSV / JSON export.
|
||||
|
||||
Excel uses openpyxl via pandas — multi-sheet workbooks readable in
|
||||
Excel 2019+. JSON round-trips the full input state (sites, takeouts,
|
||||
feature scopes) so a scenario can be saved and reloaded.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from .inputs import CostTakeout, FeatureScope, SiteInput
|
||||
from .meters import Confidence, TokenMeter
|
||||
from .rollout import RolloutPlan
|
||||
|
||||
|
||||
def meters_dataframe(meters: dict[str, TokenMeter]) -> pd.DataFrame:
|
||||
"""Meter catalogue as a display/export-ready DataFrame."""
|
||||
return pd.DataFrame(
|
||||
[
|
||||
{
|
||||
"feature": m.feature,
|
||||
"meter_type": m.meter_type.value,
|
||||
"units_per_token": m.units_per_token or None,
|
||||
"tokens_per_unit": m.tokens_per_unit,
|
||||
"confidence": f"{m.confidence.icon} {m.confidence.value}",
|
||||
"notes": m.notes,
|
||||
"source": m.source_url or "",
|
||||
}
|
||||
for m in meters.values()
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def sites_dataframe(sites: list[SiteInput]) -> pd.DataFrame:
|
||||
rows = []
|
||||
for s in sites:
|
||||
d = dataclasses.asdict(s)
|
||||
d["languages"] = ", ".join(d["languages"])
|
||||
rows.append(d)
|
||||
return pd.DataFrame(rows)
|
||||
|
||||
|
||||
def export_excel(
|
||||
sheets: dict[str, pd.DataFrame],
|
||||
path: str | Path,
|
||||
) -> Path:
|
||||
"""Write a multi-sheet Excel workbook. Sheet names are truncated to
|
||||
Excel's 31-character limit."""
|
||||
path = Path(path)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with pd.ExcelWriter(path, engine="openpyxl") as writer:
|
||||
for name, df in sheets.items():
|
||||
df.to_excel(writer, sheet_name=name[:31], index=False)
|
||||
return path
|
||||
|
||||
|
||||
def export_csv(df: pd.DataFrame, path: str | Path) -> Path:
|
||||
path = Path(path)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
df.to_csv(path, index=False)
|
||||
return path
|
||||
|
||||
|
||||
# ── JSON scenario save / load ────────────────────────────────────────
|
||||
|
||||
def scenario_state_to_json(
|
||||
sites: list[SiteInput],
|
||||
takeouts: list[CostTakeout],
|
||||
feature_scopes: list[FeatureScope],
|
||||
path: str | Path | None = None,
|
||||
rollout: RolloutPlan | None = None,
|
||||
) -> str:
|
||||
"""Serialize the full input state; optionally write to ``path``."""
|
||||
state = {
|
||||
"sites": [dataclasses.asdict(s) for s in sites],
|
||||
"takeouts": [
|
||||
{**dataclasses.asdict(t), "confidence": t.confidence.value}
|
||||
for t in takeouts
|
||||
],
|
||||
"feature_scopes": [
|
||||
{
|
||||
**dataclasses.asdict(f),
|
||||
"adoption_curve": {str(k): v for k, v in f.adoption_curve.items()},
|
||||
}
|
||||
for f in feature_scopes
|
||||
],
|
||||
}
|
||||
if rollout is not None:
|
||||
state["rollout"] = dataclasses.asdict(rollout)
|
||||
text = json.dumps(state, indent=2)
|
||||
if path is not None:
|
||||
Path(path).write_text(text)
|
||||
return text
|
||||
|
||||
|
||||
def scenario_state_from_json(
|
||||
source: str | Path,
|
||||
) -> tuple[list[SiteInput], list[CostTakeout], list[FeatureScope], RolloutPlan | None]:
|
||||
"""Inverse of :func:`scenario_state_to_json`. ``source`` is a JSON
|
||||
string or a file path. The fourth element is None for legacy files
|
||||
saved without a rollout plan."""
|
||||
raw = (
|
||||
Path(source).read_text()
|
||||
if isinstance(source, Path) or (isinstance(source, str) and source.strip().endswith(".json"))
|
||||
else str(source)
|
||||
)
|
||||
state = json.loads(raw)
|
||||
sites = [SiteInput(**s) for s in state["sites"]]
|
||||
takeouts = [
|
||||
CostTakeout(**{**t, "confidence": Confidence(t["confidence"])})
|
||||
for t in state["takeouts"]
|
||||
]
|
||||
scopes = [
|
||||
FeatureScope(
|
||||
**{
|
||||
**f,
|
||||
"adoption_curve": {int(k): v for k, v in f["adoption_curve"].items()},
|
||||
}
|
||||
)
|
||||
for f in state["feature_scopes"]
|
||||
]
|
||||
rollout = (
|
||||
RolloutPlan(**state["rollout"]) if "rollout" in state else None
|
||||
)
|
||||
return sites, takeouts, scopes, rollout
|
||||
155
studies/202607_CTM_GenesysCX/tokencalc/inputs.py
Normal file
155
studies/202607_CTM_GenesysCX/tokencalc/inputs.py
Normal file
@@ -0,0 +1,155 @@
|
||||
"""
|
||||
Input bundles — validated dataclasses, no untyped dicts.
|
||||
|
||||
All volumes are MONTHLY; all AHT/ACW figures are SECONDS; all labour
|
||||
costs are ANNUAL fully-loaded USD.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from .meters import Confidence
|
||||
|
||||
#: Sanity bounds for handle times (seconds).
|
||||
AHT_MIN_SECONDS = 10
|
||||
AHT_MAX_SECONDS = 3600
|
||||
|
||||
#: Working hours per FTE-year used to derive per-second labour rates.
|
||||
WORKING_HOURS_PER_YEAR = 2_080
|
||||
WORKING_SECONDS_PER_YEAR = WORKING_HOURS_PER_YEAR * 3600
|
||||
|
||||
|
||||
@dataclass
|
||||
class SiteInput:
|
||||
site_name: str # "NAM", "EMEA", "AUZ", "APAC HK", …
|
||||
region_pricing: str # "US", "AU", "EU", "APAC"
|
||||
agents: int # excluding supervisors
|
||||
supervisors: int
|
||||
voice_volume_monthly: int
|
||||
email_volume_monthly: int
|
||||
chat_volume_monthly: int
|
||||
sms_volume_monthly: int
|
||||
voice_aht_seconds: int
|
||||
email_aht_seconds: int
|
||||
chat_aht_seconds: int
|
||||
voice_acw_seconds: int
|
||||
fully_loaded_agent_cost_annual: float
|
||||
fully_loaded_supervisor_cost_annual: float
|
||||
licence_type: str = "named" # "named" | "concurrent"
|
||||
languages: list[str] = field(default_factory=list)
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.licence_type not in ("named", "concurrent"):
|
||||
raise ValueError(
|
||||
f"{self.site_name}: licence_type must be 'named' or 'concurrent', "
|
||||
f"got {self.licence_type!r}"
|
||||
)
|
||||
if self.agents < 0 or self.supervisors < 0:
|
||||
raise ValueError(f"{self.site_name}: agent/supervisor counts must be >= 0")
|
||||
for name in (
|
||||
"voice_volume_monthly",
|
||||
"email_volume_monthly",
|
||||
"chat_volume_monthly",
|
||||
"sms_volume_monthly",
|
||||
):
|
||||
if getattr(self, name) < 0:
|
||||
raise ValueError(f"{self.site_name}: {name} must be >= 0")
|
||||
for name in ("voice_aht_seconds", "email_aht_seconds", "chat_aht_seconds"):
|
||||
v = getattr(self, name)
|
||||
if v and not AHT_MIN_SECONDS <= v <= AHT_MAX_SECONDS:
|
||||
raise ValueError(
|
||||
f"{self.site_name}: {name}={v}s outside sensible bounds "
|
||||
f"({AHT_MIN_SECONDS}-{AHT_MAX_SECONDS}s)"
|
||||
)
|
||||
if self.voice_acw_seconds < 0:
|
||||
raise ValueError(f"{self.site_name}: voice_acw_seconds must be >= 0")
|
||||
|
||||
@property
|
||||
def named_users(self) -> int:
|
||||
return self.agents + self.supervisors
|
||||
|
||||
@property
|
||||
def agent_cost_per_second(self) -> float:
|
||||
"""Fully-loaded agent labour rate per working second (DBZ-safe)."""
|
||||
return self.fully_loaded_agent_cost_annual / WORKING_SECONDS_PER_YEAR
|
||||
|
||||
@property
|
||||
def supervisor_cost_per_second(self) -> float:
|
||||
return self.fully_loaded_supervisor_cost_annual / WORKING_SECONDS_PER_YEAR
|
||||
|
||||
|
||||
@dataclass
|
||||
class FeatureScope:
|
||||
"""Which feature is enabled at which sites, in which phase.
|
||||
|
||||
``phase`` is the model year (1-3) the feature switches on;
|
||||
``adoption_curve`` maps model year -> adoption fraction (0.0-1.0)
|
||||
applied to consumption-metered features (per-user licenses are paid
|
||||
in full from the phase year onward).
|
||||
"""
|
||||
|
||||
feature: str
|
||||
enabled_sites: list[str]
|
||||
phase: int = 1
|
||||
adoption_curve: dict[int, float] = field(default_factory=dict)
|
||||
deflection_target: float | None = None
|
||||
eligibility_pct: float | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.phase < 1:
|
||||
raise ValueError(f"{self.feature}: phase must be >= 1")
|
||||
for year, pct in self.adoption_curve.items():
|
||||
if not 0.0 <= pct <= 1.0:
|
||||
raise ValueError(
|
||||
f"{self.feature}: adoption_curve[{year}]={pct} outside 0-1"
|
||||
)
|
||||
for name in ("deflection_target", "eligibility_pct"):
|
||||
v = getattr(self, name)
|
||||
if v is not None and not 0.0 <= v <= 1.0:
|
||||
raise ValueError(f"{self.feature}: {name}={v} outside 0-1")
|
||||
|
||||
def active(self, site_name: str, year: int) -> bool:
|
||||
return site_name in self.enabled_sites and year >= self.phase
|
||||
|
||||
def adoption(self, year: int) -> float:
|
||||
"""Adoption fraction for ``year`` (1.0 when no curve given)."""
|
||||
if not self.adoption_curve:
|
||||
return 1.0
|
||||
if year in self.adoption_curve:
|
||||
return self.adoption_curve[year]
|
||||
# Past the last defined year → hold the last value.
|
||||
last = max(self.adoption_curve)
|
||||
return self.adoption_curve[last] if year > last else 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class CostTakeout:
|
||||
"""A retired platform/licence whose cost the programme reclaims.
|
||||
|
||||
``start_month`` (1-12, within ``start_year``) prorates the first
|
||||
active year — e.g. NICE IEX can only be switched off once NAM is
|
||||
live, so start_year=1, start_month=7 reclaims 6/12 of Y1.
|
||||
"""
|
||||
|
||||
name: str # "NICE IEX (NAM)", "Legacy CC platform", …
|
||||
annual_cost: float
|
||||
start_year: int = 1
|
||||
confidence: Confidence = Confidence.ESTIMATED
|
||||
notes: str = ""
|
||||
start_month: int = 1
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.annual_cost < 0:
|
||||
raise ValueError(f"{self.name}: annual_cost must be >= 0")
|
||||
if self.start_year < 1:
|
||||
raise ValueError(f"{self.name}: start_year must be >= 1")
|
||||
if not 1 <= self.start_month <= 12:
|
||||
raise ValueError(f"{self.name}: start_month must be 1-12")
|
||||
|
||||
def value_in_year(self, year: int) -> float:
|
||||
if year < self.start_year:
|
||||
return 0.0
|
||||
if year == self.start_year:
|
||||
return self.annual_cost * (12 - (self.start_month - 1)) / 12
|
||||
return self.annual_cost
|
||||
87
studies/202607_CTM_GenesysCX/tokencalc/meters.py
Normal file
87
studies/202607_CTM_GenesysCX/tokencalc/meters.py
Normal file
@@ -0,0 +1,87 @@
|
||||
"""
|
||||
Genesys AI Experience token meters and pricing.
|
||||
|
||||
Every meter carries a :class:`Confidence` flag so the UI can distinguish
|
||||
published Genesys rates from estimates and unknowns. Rates here are
|
||||
*planning inputs* — this tool explicitly does not replace contractual
|
||||
pricing (see README, Non-Goals).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class MeterType(Enum):
|
||||
PER_USER_PER_MONTH = "per_user_per_month"
|
||||
PER_INTERACTION = "per_interaction"
|
||||
PER_MINUTE = "per_minute"
|
||||
PER_MESSAGE = "per_message"
|
||||
PER_SUMMARY = "per_summary"
|
||||
|
||||
|
||||
class Confidence(Enum):
|
||||
CONFIRMED = "confirmed" # published Genesys rate
|
||||
ESTIMATED = "estimated" # reasonable industry assumption
|
||||
UNKNOWN = "unknown" # rate not yet sourced
|
||||
|
||||
@property
|
||||
def icon(self) -> str:
|
||||
return {"confirmed": "🟢", "estimated": "🟡", "unknown": "🔴"}[self.value]
|
||||
|
||||
|
||||
@dataclass
|
||||
class TokenMeter:
|
||||
"""One Genesys AI feature's token meter.
|
||||
|
||||
``units_per_token`` and ``tokens_per_unit`` are inverses; both are
|
||||
stored because the UI shows whichever reads more naturally (e.g.
|
||||
"17 minutes per token" vs "0.0588 tokens per minute"). For
|
||||
PER_USER_PER_MONTH meters ``units_per_token`` is 0.0 (n/a) and
|
||||
``tokens_per_unit`` is the flat tokens/user/month figure.
|
||||
"""
|
||||
|
||||
feature: str
|
||||
meter_type: MeterType
|
||||
units_per_token: float
|
||||
tokens_per_unit: float
|
||||
confidence: Confidence
|
||||
notes: str
|
||||
source_url: str | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.tokens_per_unit < 0:
|
||||
raise ValueError(f"{self.feature}: tokens_per_unit must be >= 0")
|
||||
if (
|
||||
self.meter_type is not MeterType.PER_USER_PER_MONTH
|
||||
and self.units_per_token > 0
|
||||
and self.tokens_per_unit > 0
|
||||
):
|
||||
product = self.units_per_token * self.tokens_per_unit
|
||||
if not 0.95 <= product <= 1.05:
|
||||
raise ValueError(
|
||||
f"{self.feature}: units_per_token ({self.units_per_token}) and "
|
||||
f"tokens_per_unit ({self.tokens_per_unit}) are not inverses"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class TokenPricing:
|
||||
"""Per-region token pricing. Default is US list at $1/token."""
|
||||
|
||||
region: str # "US", "AU", "EU", "APAC"
|
||||
list_rate_per_token: float = 1.0
|
||||
contracted_rate_per_token: float | None = None
|
||||
prepay_commit_tokens: int | None = None
|
||||
overage_rate_per_token: float | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.list_rate_per_token < 0:
|
||||
raise ValueError(f"{self.region}: list rate must be >= 0")
|
||||
|
||||
def effective_rate(self, use_contracted: bool = False) -> float:
|
||||
"""Contracted rate when requested and known, else list rate."""
|
||||
if use_contracted and self.contracted_rate_per_token is not None:
|
||||
return self.contracted_rate_per_token
|
||||
return self.list_rate_per_token
|
||||
88
studies/202607_CTM_GenesysCX/tokencalc/migration_wfm.py
Normal file
88
studies/202607_CTM_GenesysCX/tokencalc/migration_wfm.py
Normal file
@@ -0,0 +1,88 @@
|
||||
"""
|
||||
Migration + WFM scenario — the no-AI business case.
|
||||
|
||||
Current state → Genesys Cloud CX migration, NA users migrated onto
|
||||
Genesys WFM, WFM implemented for APAC (ANZ + ASIA). Migration and WFM
|
||||
are included in the base implementation price (the verbatim PS +
|
||||
training), so the only cost lines are the existing-platform run-off,
|
||||
the licence ramp, and base PS — no token consumption, no AI
|
||||
implementation labour. The only benefits kept are the deck's verbatim
|
||||
WFM lines for the regions in WFM scope.
|
||||
|
||||
Single source of truth behind ``notebooks/ctm_migration_wfm.ipynb``
|
||||
(served with Mercury) — the presentation layer holds no math. All
|
||||
primitives come from :mod:`tokencalc.appendix4`; this module only
|
||||
scopes and extrapolates.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from . import appendix4 as a4
|
||||
|
||||
#: WFM scope in this scenario. NA is a migration off its existing
|
||||
#: WFM-like tool (verbatim benefit $0 — "has similar feature"); ANZ and
|
||||
#: ASIA are new implementations ("APAC"); EMEA is out of scope, so the
|
||||
#: deck's EMEA WFM benefit line is dropped.
|
||||
DEFAULT_WFM_REGIONS = ["NA", "ANZ", "ASIA"]
|
||||
|
||||
|
||||
def wfm_benefits_by_year(
|
||||
benefit_rollout, regions: list[str] | None = None
|
||||
) -> pd.DataFrame:
|
||||
"""Verbatim WFM benefits for the scoped regions, phased on the
|
||||
deck's deployment schedule (realize = impl + 3 months, inclusive).
|
||||
|
||||
Long DataFrame: region, capability, year, benefit — the WFM slice
|
||||
of :func:`tokencalc.appendix4.benefits_by_year`.
|
||||
"""
|
||||
scope = DEFAULT_WFM_REGIONS if regions is None else regions
|
||||
df = a4.benefits_by_year(benefit_rollout)
|
||||
return df[(df["capability"] == "WFM")
|
||||
& (df["region"].isin(scope))].reset_index(drop=True)
|
||||
|
||||
|
||||
def wfm_annual_runrate(regions: list[str] | None = None) -> float:
|
||||
"""Sum of the verbatim WFM *annual* values for the scoped regions."""
|
||||
scope = DEFAULT_WFM_REGIONS if regions is None else regions
|
||||
return float(sum(annual for (r, c), (annual, _t) in
|
||||
a4.VERBATIM_BENEFITS.items()
|
||||
if c == "WFM" and r in scope))
|
||||
|
||||
|
||||
def runrate_saving_annual(
|
||||
licence_annual: float | None = None,
|
||||
regions: list[str] | None = None,
|
||||
baseline_annual: float | None = None,
|
||||
managed_annual: float = 0.0,
|
||||
) -> float:
|
||||
"""Steady-state annual saving once term contracts end and the ramp
|
||||
is over: (baseline − licence run-rate − managed services) + scoped
|
||||
WFM annual values.
|
||||
|
||||
Defaults are the deck frame (deck licence rate, no managed
|
||||
services); the contracted frame passes ``a4.MANAGED_SERVICES_ANNUAL``.
|
||||
"""
|
||||
lic = a4.TCO_VERBATIM["ccaas_annual"] if licence_annual is None else licence_annual
|
||||
base = a4.TCO_VERBATIM["current_annual"] if baseline_annual is None else baseline_annual
|
||||
return (base - lic - managed_annual) + wfm_annual_runrate(regions)
|
||||
|
||||
|
||||
def runrate_breakeven_label(
|
||||
net_by_year: dict[int, float], runrate_annual: float
|
||||
) -> str:
|
||||
"""Payback label, extrapolated past the model window at a run-rate.
|
||||
|
||||
Inside 2026-28 this defers to :func:`tokencalc.appendix4.payback_label`;
|
||||
a deficit at end-2028 fills at ``runrate_annual`` per year.
|
||||
"""
|
||||
deficit = -sum(net_by_year[y] for y in a4.YEARS)
|
||||
if deficit <= 0:
|
||||
return a4.payback_label(net_by_year)
|
||||
if runrate_annual <= 0:
|
||||
return "never at current run-rate"
|
||||
m = 12 * len(a4.YEARS) + math.ceil(12 * deficit / runrate_annual)
|
||||
return f"{m} months (~{a4.month_label(m)}, extrapolated)"
|
||||
81
studies/202607_CTM_GenesysCX/tokencalc/rollout.py
Normal file
81
studies/202607_CTM_GenesysCX/tokencalc/rollout.py
Normal file
@@ -0,0 +1,81 @@
|
||||
"""
|
||||
Implementation rollout & ramp model.
|
||||
|
||||
Captures the gap between **when Genesys starts billing** (contract
|
||||
start) and **when each region actually goes live**:
|
||||
|
||||
- The platform licence commit bills in full from contract start; the
|
||||
vendor's *ramp period* compensates with a first-year credit
|
||||
(typical: 6-month ramp → 50% Y1 discount on the platform commit).
|
||||
- AI token usage (per-user and consumption meters) starts only when a
|
||||
site goes live, and bills for the months the site is live in each
|
||||
model year.
|
||||
- Benefits likewise accrue only from go-live (the scenario realization
|
||||
curve then models adoption maturity *within* the live period).
|
||||
|
||||
A site with ``go_live_month = m`` is live for ``12*year − m`` months of
|
||||
the first ``year`` years (clamped to 0..12 per year). So NAM at month 6
|
||||
is live 6 months of Y1; EMEA at month 9 → 3 months; AUZ/APAC at month
|
||||
12 → 0 months in Y1 and fully live from Y2.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
MONTHS_PER_YEAR = 12
|
||||
|
||||
|
||||
@dataclass
|
||||
class RolloutPlan:
|
||||
#: ISO date Genesys starts billing the licence commit (informational,
|
||||
#: surfaced in UI/exports; the model works in months-from-start).
|
||||
contract_start: str | None = None
|
||||
|
||||
#: Total build duration, months (informational).
|
||||
build_months: int = 12
|
||||
|
||||
#: Vendor ramp period, months. Documentation for the Y1 credit below.
|
||||
ramp_months: int = 6
|
||||
|
||||
#: First-year credit on the platform licence commit. Typical
|
||||
#: 6-month ramp = 50% discount in year 1; years 2+ bill in full.
|
||||
first_year_platform_discount: float = 0.5
|
||||
|
||||
#: site_name -> go-live month (months after contract start).
|
||||
#: Sites absent from the map are treated as live from day 0.
|
||||
go_live_month: dict[str, int] = field(default_factory=dict)
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not 0.0 <= self.first_year_platform_discount <= 1.0:
|
||||
raise ValueError("first_year_platform_discount must be within 0-1")
|
||||
if self.ramp_months < 0 or self.build_months < 0:
|
||||
raise ValueError("ramp_months/build_months must be >= 0")
|
||||
for site, m in self.go_live_month.items():
|
||||
if m < 0:
|
||||
raise ValueError(f"{site}: go_live_month must be >= 0")
|
||||
|
||||
# ── Availability ────────────────────────────────────────────────
|
||||
|
||||
def live_months_in_year(self, site_name: str, year: int) -> int:
|
||||
"""Months ``site_name`` is live during model year ``year`` (1-based)."""
|
||||
go_live = self.go_live_month.get(site_name, 0)
|
||||
live_by_year_end = max(0, MONTHS_PER_YEAR * year - go_live)
|
||||
live_by_prev_year_end = max(0, MONTHS_PER_YEAR * (year - 1) - go_live)
|
||||
return min(MONTHS_PER_YEAR, live_by_year_end - live_by_prev_year_end)
|
||||
|
||||
def fraction_live(self, site_name: str, year: int) -> float:
|
||||
return self.live_months_in_year(site_name, year) / MONTHS_PER_YEAR
|
||||
|
||||
# ── Billing ─────────────────────────────────────────────────────
|
||||
|
||||
def platform_factor(self, year: int) -> float:
|
||||
"""Fraction of the full platform commit billed in ``year``."""
|
||||
return 1.0 - self.first_year_platform_discount if year == 1 else 1.0
|
||||
|
||||
|
||||
#: Behaviour identical to the pre-rollout model: everything live from
|
||||
#: day 0, no ramp credit.
|
||||
NO_ROLLOUT = RolloutPlan(
|
||||
build_months=0, ramp_months=0, first_year_platform_discount=0.0
|
||||
)
|
||||
144
studies/202607_CTM_GenesysCX/tokencalc/scenarios.py
Normal file
144
studies/202607_CTM_GenesysCX/tokencalc/scenarios.py
Normal file
@@ -0,0 +1,144 @@
|
||||
"""
|
||||
Scenario definitions — Floor / Realistic / Stretch.
|
||||
|
||||
Every scenario parameter the cost and benefit engines read lives here;
|
||||
no magic numbers in the calculation modules. Ships with the spec
|
||||
defaults; callers may construct custom :class:`Scenario` objects.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
|
||||
@dataclass
|
||||
class Scenario:
|
||||
name: str
|
||||
|
||||
# ── Cost-side drivers ───────────────────────────────────────────
|
||||
voice_bot_deflection: float # share of voice volume deflected to bot
|
||||
voice_bot_avg_minutes: float # bot minutes per deflected call
|
||||
# Agentic VA deflection is INCREMENTAL — applied to the residual volume
|
||||
# after the voice bot has already handled its share (layered model).
|
||||
# Effective total deflection = bot_rate + (1 − bot_rate) × va_rate.
|
||||
agentic_va_deflection: float # share of RESIDUAL voice volume to agentic VA
|
||||
voice_summarization_eligibility: float
|
||||
voice_knowledge_eligibility: float
|
||||
email_auto_respond_rate: float # share of email auto-responded
|
||||
|
||||
# ── Virtual Agent benefit realization factors ───────────────────
|
||||
# Applied to both Voice Bot and Agentic VA deflection benefits.
|
||||
# completion_rate — share of "deflected" calls that don't escalate to an agent
|
||||
# mid-session (bot/VA fully handles the interaction).
|
||||
# labour_realization — staffing flexibility: deflected volume doesn't reduce
|
||||
# headcount 1:1 due to minimums, shrinkage, occupancy ceilings.
|
||||
# callback_discount — fraction of deflected calls that re-enter as repeat contacts
|
||||
# (poorly-handled deflections drive callbacks).
|
||||
# Combined realistic factor: 0.70 × 0.80 × (1 − 0.05) ≈ 0.53
|
||||
va_completion_rate: float = 0.70
|
||||
va_labour_realization: float = 0.80
|
||||
va_callback_discount: float = 0.05
|
||||
|
||||
# year -> fraction of full benefit realized
|
||||
benefit_realization: dict[int, float] = field(default_factory=dict)
|
||||
|
||||
# year -> fraction of steady-state consumption cost incurred.
|
||||
# Per-user licenses are paid in full from day 1; consumption meters
|
||||
# ramp with usage (default Y1 = 70%).
|
||||
consumption_cost_realization: dict[int, float] = field(
|
||||
default_factory=lambda: {1: 0.70, 2: 1.0, 3: 1.0}
|
||||
)
|
||||
|
||||
def realization(self, year: int) -> float:
|
||||
if year in self.benefit_realization:
|
||||
return self.benefit_realization[year]
|
||||
last = max(self.benefit_realization, default=0)
|
||||
return self.benefit_realization.get(last, 1.0) if year > last else 0.0
|
||||
|
||||
def cost_realization(self, year: int) -> float:
|
||||
if year in self.consumption_cost_realization:
|
||||
return self.consumption_cost_realization[year]
|
||||
last = max(self.consumption_cost_realization, default=0)
|
||||
return (
|
||||
self.consumption_cost_realization.get(last, 1.0) if year > last else 0.0
|
||||
)
|
||||
|
||||
|
||||
#: Benefit reduction parameters. ``claim`` = Genesys ROI-doc figure;
|
||||
#: ``realistic`` = pressure-tested midpoint of the spec's Y1 range.
|
||||
#: The benefit engine uses ``realistic`` by default; ``claim`` powers
|
||||
#: the side-by-side comparison view.
|
||||
BENEFIT_PARAMS: dict[str, dict[str, float]] = {
|
||||
"voice_aht_knowledge_reduction": {"claim": 0.094, "realistic": 0.055}, # 4-7% Y1
|
||||
"voice_acw_reduction": {"claim": 1.00, "realistic": 0.40}, # 30-50% Y1
|
||||
"digital_aht_reduction": {"claim": 0.18, "realistic": 0.085}, # 5-12% Y1
|
||||
"digital_acw_reduction": {"claim": 1.00, "realistic": 0.40}, # 30-50% Y1
|
||||
"sta_aht_reduction": {"claim": 0.04, "realistic": 0.015}, # 1-2% Y1
|
||||
# ESTIMATED lines (no Genesys claim published):
|
||||
"supervisor_copilot_time_saving": {"claim": 0.10, "realistic": 0.05},
|
||||
"predictive_routing_aht_reduction": {"claim": 0.04, "realistic": 0.02},
|
||||
# Virtual Agent realization factors.
|
||||
# ``claim`` = 100% realization (original model assumption — no haircuts).
|
||||
# ``realistic`` = production-calibrated midpoints per the spec analysis.
|
||||
"va_completion_rate": {"claim": 1.00, "realistic": 0.70}, # 60-75% voice bot; 50-70% agentic VA Y1
|
||||
"va_labour_realization": {"claim": 1.00, "realistic": 0.80}, # 70-85% staffing flexibility
|
||||
"va_callback_discount": {"claim": 0.00, "realistic": 0.05}, # 5-10% deflected re-enter as repeat contacts
|
||||
}
|
||||
|
||||
|
||||
SCENARIOS: dict[str, Scenario] = {
|
||||
"floor": Scenario(
|
||||
name="floor",
|
||||
voice_bot_deflection=0.20,
|
||||
voice_bot_avg_minutes=1.0,
|
||||
agentic_va_deflection=0.05,
|
||||
voice_summarization_eligibility=0.50,
|
||||
voice_knowledge_eligibility=0.40,
|
||||
email_auto_respond_rate=0.10,
|
||||
# VA realization: conservative — low completion, limited staffing flex
|
||||
# Combined: 0.60 × 0.70 × (1 − 0.05) ≈ 0.40
|
||||
va_completion_rate=0.60,
|
||||
va_labour_realization=0.70,
|
||||
va_callback_discount=0.05,
|
||||
benefit_realization={1: 0.30, 2: 0.60, 3: 0.80},
|
||||
),
|
||||
"realistic": Scenario(
|
||||
name="realistic",
|
||||
voice_bot_deflection=0.35,
|
||||
voice_bot_avg_minutes=1.5,
|
||||
agentic_va_deflection=0.15,
|
||||
voice_summarization_eligibility=0.70,
|
||||
voice_knowledge_eligibility=0.60,
|
||||
email_auto_respond_rate=0.20,
|
||||
# VA realization: production midpoints per spec analysis
|
||||
# Combined: 0.70 × 0.80 × (1 − 0.05) ≈ 0.53
|
||||
va_completion_rate=0.70,
|
||||
va_labour_realization=0.80,
|
||||
va_callback_discount=0.05,
|
||||
benefit_realization={1: 0.50, 2: 0.80, 3: 0.95},
|
||||
),
|
||||
"stretch": Scenario(
|
||||
name="stretch",
|
||||
voice_bot_deflection=0.50,
|
||||
voice_bot_avg_minutes=2.0,
|
||||
agentic_va_deflection=0.25,
|
||||
voice_summarization_eligibility=0.90,
|
||||
voice_knowledge_eligibility=0.80,
|
||||
email_auto_respond_rate=0.50,
|
||||
# VA realization: optimistic — high completion, good staffing flexibility
|
||||
# Combined: 0.75 × 0.85 × (1 − 0.03) ≈ 0.62
|
||||
va_completion_rate=0.75,
|
||||
va_labour_realization=0.85,
|
||||
va_callback_discount=0.03,
|
||||
benefit_realization={1: 0.75, 2: 0.95, 3: 1.00},
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def get_scenario(name: str) -> Scenario:
|
||||
try:
|
||||
return SCENARIOS[name.lower()]
|
||||
except KeyError as e:
|
||||
raise KeyError(
|
||||
f"Unknown scenario {name!r}. Valid: {sorted(SCENARIOS)}"
|
||||
) from e
|
||||
29
studies/202607_CTM_GenesysCX/tokencalc/staging.py
Normal file
29
studies/202607_CTM_GenesysCX/tokencalc/staging.py
Normal file
@@ -0,0 +1,29 @@
|
||||
"""
|
||||
Stage vs backstage — is this notebook render stakeholder-facing?
|
||||
|
||||
The Mercury CLI (``mercury --working-dir …``) exports ``MERCURY_CONFIG_DIR``
|
||||
into the server process so the widget library can locate ``config.toml``
|
||||
(see ``mercury/config.py``); every kernel that server spawns inherits it.
|
||||
JupyterLab and nbconvert kernels don't have it. That makes the variable a
|
||||
reliable signal for "the audience is looking" (the stage) versus an
|
||||
analyst session or a headless export run (backstage).
|
||||
|
||||
Diagnostics routed through :func:`backstage` stay visible in JupyterLab
|
||||
and land in the nbconvert exports (where the machine-readable appendix
|
||||
must appear for LLM consumption) but never render in the Mercury app.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
|
||||
def on_stage() -> bool:
|
||||
"""True when running under the Mercury app (stakeholder-facing)."""
|
||||
return os.getenv("MERCURY_CONFIG_DIR") is not None
|
||||
|
||||
|
||||
def backstage(*args, **kwargs) -> None:
|
||||
"""``print`` that renders only backstage (JupyterLab, nbconvert)."""
|
||||
if not on_stage():
|
||||
print(*args, **kwargs)
|
||||
Reference in New Issue
Block a user