314 lines
12 KiB
Python
314 lines
12 KiB
Python
"""
|
||
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,
|
||
Email Auto-Suggest).
|
||
|
||
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
|
||
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)
|