229 lines
10 KiB
Python
229 lines
10 KiB
Python
"""Value-at-stake math: driver values, capability capping, unlock sequence.
|
||
|
||
The pipeline (build spec §6):
|
||
|
||
1. Each configured value driver yields a theoretical annual value range
|
||
from the operational baseline (dispatch on ``driver.kind``).
|
||
2. Theoretical annual value = sum of drivers.
|
||
3. The **weakest foundational competency score** selects a realization
|
||
band from the capping heuristic.
|
||
4. ``realizable_18mo = theoretical × realization_factor × 1.5``
|
||
(18 months of annual run-rate).
|
||
5. Trapped value (annual) = theoretical − realizable run-rate. Range
|
||
pairing is conservative-consistent: the low trapped estimate assumes
|
||
the low theoretical *and* the high realization factor, and vice versa.
|
||
6. Binding constraints = every foundational competency sitting at the
|
||
weakest score.
|
||
7. Unlock sequence = up to three **tier lifts**: raise the whole binding
|
||
set one level, recompute the band, attribute the delta. When several
|
||
competencies tie at the weakest level a single-competency lift would
|
||
honestly unlock nothing — the set is the move (see UnlockMove docs).
|
||
|
||
Guard rails: every output is a range; 🔴-unknown inputs raise warnings on
|
||
the result; money *display* is capped at two significant figures
|
||
(:func:`money`) while raw floats stay exact in exports.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from .models import (
|
||
CompetencyScore,
|
||
DiagnosticConfig,
|
||
DriverValue,
|
||
OperationalBaseline,
|
||
UnlockMove,
|
||
ValueAtStake,
|
||
ValueDriver,
|
||
)
|
||
|
||
#: 18 months expressed in years of annual run-rate.
|
||
MONTHS_18 = 1.5
|
||
|
||
#: How many unlock moves the sequence proposes.
|
||
MAX_UNLOCK_MOVES = 3
|
||
|
||
|
||
# ── Money display (guard rail: ≤ 2 significant figures) ──────────────
|
||
|
||
|
||
def _round_2sf(v: float) -> float:
|
||
if v == 0:
|
||
return 0.0
|
||
from math import floor, log10
|
||
exp = floor(log10(abs(v)))
|
||
return round(v, -exp + 1)
|
||
|
||
|
||
def money(v: float) -> str:
|
||
"""House money format, capped at two significant figures: $2.5M, $950K."""
|
||
sign, a = ("-" if v < 0 else ""), _round_2sf(abs(v))
|
||
if a >= 1e6:
|
||
m = a / 1e6
|
||
return f"{sign}${m:,.1f}M" if m < 10 else f"{sign}${m:,.0f}M"
|
||
if a >= 1e3:
|
||
return f"{sign}${a / 1e3:,.0f}K"
|
||
return f"{sign}${a:,.0f}"
|
||
|
||
|
||
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("$", "$")
|
||
|
||
|
||
# ── Driver math (dispatch on kind) ───────────────────────────────────
|
||
|
||
|
||
def driver_value(driver: ValueDriver, baseline: OperationalBaseline) -> DriverValue:
|
||
"""Theoretical annual value range for one configured driver."""
|
||
if driver.kind == "containment_lift":
|
||
if driver.lift_range_pts_low is None or driver.lift_range_pts_high is None:
|
||
raise ValueError(f"driver {driver.id}: containment_lift needs lift_range_pts_low/high")
|
||
low = baseline.annual_contact_volume * driver.lift_range_pts_low \
|
||
* baseline.blended_cost_per_contact
|
||
high = baseline.annual_contact_volume * driver.lift_range_pts_high \
|
||
* baseline.blended_cost_per_contact
|
||
elif driver.kind == "aht_reduction":
|
||
# volume × (AHT × pct) seconds saved × ($/contact ÷ AHT) per second
|
||
# — the baseline AHT cancels: volume × $/contact × pct.
|
||
if driver.reduction_pct_low is None or driver.reduction_pct_high is None:
|
||
raise ValueError(f"driver {driver.id}: aht_reduction needs reduction_pct_low/high")
|
||
low = baseline.annual_contact_volume * baseline.blended_cost_per_contact \
|
||
* driver.reduction_pct_low
|
||
high = baseline.annual_contact_volume * baseline.blended_cost_per_contact \
|
||
* driver.reduction_pct_high
|
||
elif driver.kind == "attrition_reduction":
|
||
if driver.reduction_pct_low is None or driver.reduction_pct_high is None:
|
||
raise ValueError(f"driver {driver.id}: attrition_reduction needs reduction_pct_low/high")
|
||
cost_per_replacement = driver.cost_per_replacement_default or 0.0
|
||
low = baseline.agent_headcount * baseline.annual_attrition_rate \
|
||
* driver.reduction_pct_low * cost_per_replacement
|
||
high = baseline.agent_headcount * baseline.annual_attrition_rate \
|
||
* driver.reduction_pct_high * cost_per_replacement
|
||
else: # pragma: no cover — Literal already restricts kinds
|
||
raise ValueError(f"driver {driver.id}: unknown kind {driver.kind}")
|
||
return DriverValue(driver_id=driver.id, name=driver.name,
|
||
theoretical_low=low, theoretical_high=high)
|
||
|
||
|
||
# ── Capping ──────────────────────────────────────────────────────────
|
||
|
||
|
||
def weakest_foundational_score(config: DiagnosticConfig,
|
||
scores: list[CompetencyScore]) -> int:
|
||
by_id = {s.competency_id: s.score for s in scores}
|
||
missing = [c for c in config.foundational_competencies if c not in by_id]
|
||
if missing:
|
||
raise ValueError(f"foundational competencies unscored: {missing}")
|
||
return min(by_id[c] for c in config.foundational_competencies)
|
||
|
||
|
||
def binding_constraints(config: DiagnosticConfig,
|
||
scores: list[CompetencyScore]) -> list[str]:
|
||
"""Foundational competencies sitting at the weakest score, config order."""
|
||
weakest = weakest_foundational_score(config, scores)
|
||
by_id = {s.competency_id: s.score for s in scores}
|
||
return [c for c in config.foundational_competencies if by_id[c] == weakest]
|
||
|
||
|
||
# ── Unlock sequence (tier lifts of the binding set) ──────────────────
|
||
|
||
|
||
def _tier_move(config: DiagnosticConfig, level: int, members: list[str],
|
||
th_low: float, th_high: float) -> UnlockMove:
|
||
band_now = config.capping_heuristic[level]
|
||
band_next = config.capping_heuristic[level + 1]
|
||
costs = {m: config.lift_cost(m, level) for m in members}
|
||
missing = [m for m, c in costs.items() if c is None]
|
||
note = ""
|
||
if len(members) > 1:
|
||
note = "joint lift — the tied competencies must move together to shift the cap"
|
||
if missing:
|
||
note = (note + "; " if note else "") + \
|
||
f"cost not configured for: {', '.join(missing)}"
|
||
have_all = not missing
|
||
return UnlockMove(
|
||
competency_ids=members,
|
||
current_level=level,
|
||
target_level=level + 1,
|
||
est_cost_low=sum(c.cost_low for c in costs.values() if c) if have_all else None,
|
||
est_cost_high=sum(c.cost_high for c in costs.values() if c) if have_all else None,
|
||
est_weeks=max((c.weeks for c in costs.values() if c), default=None) if have_all else None,
|
||
value_unlocked_low=th_low * (band_next.realized_low - band_now.realized_low),
|
||
value_unlocked_high=th_high * (band_next.realized_high - band_now.realized_high),
|
||
note=note,
|
||
)
|
||
|
||
|
||
def unlock_sequence(config: DiagnosticConfig, scores: list[CompetencyScore],
|
||
th_low: float, th_high: float,
|
||
max_moves: int = MAX_UNLOCK_MOVES) -> list[UnlockMove]:
|
||
"""Up to ``max_moves`` sequential tier lifts of the binding set.
|
||
|
||
Each move lifts every foundational competency at the current weakest
|
||
level by one level (weeks = the longest workstream, run in parallel;
|
||
costs summed). Value unlocked is the annual realizable delta from the
|
||
capping-band shift. Moves stay in sequence order — each one is the
|
||
prerequisite of the next, so ranking them against each other would be
|
||
meaningless; the ratio walk (value/cost declining) is the story.
|
||
"""
|
||
if th_low == 0 and th_high == 0:
|
||
return []
|
||
current = {s.competency_id: s.score for s in scores
|
||
if s.competency_id in config.foundational_competencies}
|
||
moves: list[UnlockMove] = []
|
||
for _ in range(max_moves):
|
||
level = min(current.values())
|
||
if level >= 5:
|
||
break
|
||
members = [c for c in config.foundational_competencies
|
||
if current[c] == level]
|
||
moves.append(_tier_move(config, level, members, th_low, th_high))
|
||
for m in members:
|
||
current[m] = level + 1
|
||
return moves
|
||
|
||
|
||
# ── The full computation ─────────────────────────────────────────────
|
||
|
||
|
||
def value_at_stake(config: DiagnosticConfig, baseline: OperationalBaseline,
|
||
scores: list[CompetencyScore]) -> ValueAtStake:
|
||
"""Steps 1–8 of the build spec, as one call. See module docstring."""
|
||
drivers = [driver_value(d, baseline) for d in config.value_drivers]
|
||
th_low = sum(d.theoretical_low for d in drivers)
|
||
th_high = sum(d.theoretical_high for d in drivers)
|
||
|
||
weakest = weakest_foundational_score(config, scores)
|
||
band = config.capping_heuristic[weakest]
|
||
|
||
warnings: list[str] = []
|
||
if not config.value_drivers:
|
||
warnings.append(
|
||
f"config '{config.industry}' has no value drivers — "
|
||
"value-at-stake is zero (stub config)")
|
||
used_fields = sorted({d.baseline_field for d in config.value_drivers}
|
||
| ({"annual_contact_volume", "blended_cost_per_contact"}
|
||
if config.value_drivers else set()))
|
||
for f in used_fields:
|
||
if baseline.confidence_for(f) == "unknown":
|
||
warnings.append(
|
||
f"baseline input '{f}' is flagged 🔴 unknown — "
|
||
"the ranges below inherit that uncertainty")
|
||
|
||
return ValueAtStake(
|
||
theoretical_annual_value_low=th_low,
|
||
theoretical_annual_value_high=th_high,
|
||
realizable_18mo_low=th_low * band.realized_low * MONTHS_18,
|
||
realizable_18mo_high=th_high * band.realized_high * MONTHS_18,
|
||
trapped_value_low=th_low * (1 - band.realized_high),
|
||
trapped_value_high=th_high * (1 - band.realized_low),
|
||
binding_constraints=binding_constraints(config, scores),
|
||
unlock_sequence=unlock_sequence(config, scores, th_low, th_high),
|
||
weakest_foundational_score=weakest,
|
||
realization_factor_low=band.realized_low,
|
||
realization_factor_high=band.realized_high,
|
||
driver_values=drivers,
|
||
warnings=warnings,
|
||
)
|