443 lines
16 KiB
Python
443 lines
16 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_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"]]
|