Robert Helewka
64fb83257d
- Add 202512_GenesysCX TEI study (config, seed data, notebooks, README) with NPV $10.8M / ROI 266% including AI-token cost line - Add explicit `key` parameter to all chart wrappers in app/components to prevent StreamlitDuplicateElementId errors when the same figure type renders across Summary/Benefits/Costs tabs - Render benefits bar and cost pie charts on their respective tabs - Add benefits_vs_costs_by_year chart wrapper
380 lines
13 KiB
Python
380 lines
13 KiB
Python
"""
|
||
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) plus
|
||
Auto-Suggest (time saving × acceptance on the remainder)."""
|
||
sc = get_scenario(scenario) if isinstance(scenario, str) else scenario
|
||
ro = rollout or NO_ROLLOUT
|
||
suggest_saving = _param("email_auto_suggest_time_saving", params)
|
||
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
|
||
)
|
||
suggest_seconds = (
|
||
annual_emails * (1 - respond_rate)
|
||
* sc.email_auto_suggest_acceptance * s.email_aht_seconds
|
||
* suggest_saving * 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
|
||
}
|
||
)
|
||
rows.append(
|
||
{
|
||
"benefit_line": "Email Auto-Suggest (drafting time)",
|
||
"scope": s.site_name,
|
||
"annual_value": suggest_seconds * rate
|
||
* ro.fraction_live(s.site_name, year),
|
||
"confidence": Confidence.UNKNOWN.value,
|
||
}
|
||
)
|
||
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_bot_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 / Agentic VA.
|
||
|
||
Not in the original function list but required for a complete net
|
||
case — deflected volume never reaches an agent, so the full AHT is
|
||
avoided.
|
||
"""
|
||
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
|
||
if feature_scope.feature == "Voice Bot":
|
||
deflection = (
|
||
feature_scope.deflection_target
|
||
if feature_scope.deflection_target is not None
|
||
else sc.voice_bot_deflection
|
||
)
|
||
else: # Agentic Virtual Agent
|
||
deflection = (
|
||
feature_scope.deflection_target
|
||
if feature_scope.deflection_target is not None
|
||
else sc.agentic_va_deflection
|
||
)
|
||
seconds_saved = (
|
||
s.voice_volume_monthly * MONTHS_PER_YEAR
|
||
* deflection * 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
|
||
* 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.
|
||
_BENEFIT_DISPATCH = {
|
||
"Agent Copilot": (
|
||
calculate_voice_handle_time_benefit,
|
||
calculate_acw_summarization_benefit,
|
||
),
|
||
"AI Summary & Insights": (), # benefit carried by Copilot where present
|
||
"Speech & Text Analytics": (calculate_sta_benefit,),
|
||
"Voice Bot": (calculate_bot_deflection_benefit,),
|
||
"Agentic Virtual Agent": (calculate_bot_deflection_benefit,),
|
||
"Email AI (Auto-Respond)": (calculate_email_ai_benefit,),
|
||
"Predictive Routing": (calculate_predictive_routing_benefit,),
|
||
}
|
||
|
||
|
||
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.
|
||
"""
|
||
sc = get_scenario(scenario) if isinstance(scenario, str) else scenario
|
||
frames: list[pd.DataFrame] = []
|
||
copilot_scope = _scope_for(feature_scopes, "Agent Copilot")
|
||
|
||
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"]]
|