feat: add master notebook library scaffolding and review tooling
Add CLAUDE.md defining the Palladium master notebook conventions and Red Panda Approval criteria, plus a review-notebook slash command for LLM-driven notebook review. Expand .gitignore to block client/engagement documents and generated exports, keeping masters client-clean while allowing text/image sources. Normalize slider widget numeric values from floats to integers in notebook JSON.
This commit is contained in:
228
assessments/CX_AI_Diagnostic/diaglib/value_math.py
Normal file
228
assessments/CX_AI_Diagnostic/diaglib/value_math.py
Normal file
@@ -0,0 +1,228 @@
|
||||
"""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,
|
||||
)
|
||||
Reference in New Issue
Block a user