Robert Helewka
a967f73d09
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.
282 lines
10 KiB
Python
282 lines
10 KiB
Python
"""Pydantic models — every data structure the diagnostic captures or computes.
|
||
|
||
Two families live here:
|
||
|
||
* **Engagement data** — what the workshop records (participants, baseline,
|
||
scores) and what the engine computes (:class:`ValueAtStake`). The
|
||
serialized :class:`Engagement` is the source-of-truth export artifact.
|
||
* **Config data** — the validated shape of ``configs/*.yaml``: the
|
||
competency model (base) and the industry overlay (drivers, unlock costs).
|
||
|
||
Deviations from the build spec (docs/build_spec_v1.md), both additive:
|
||
|
||
* :class:`UnlockMove` carries ``competency_ids`` (plural). When several
|
||
foundational competencies tie at the weakest level, the binding
|
||
constraint *is the set* — a single-competency move would honestly unlock
|
||
nothing. Moves are tier lifts of the whole binding set.
|
||
* :class:`ValueAtStake` also records the driver breakdown, the realization
|
||
band, and guard-rail warnings, so the export explains its own numbers.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date, datetime
|
||
from typing import Literal
|
||
|
||
from pydantic import BaseModel, Field, field_validator, model_validator
|
||
|
||
Function = Literal["cx", "it", "ops", "finance", "other"]
|
||
Confidence = Literal["known", "estimated", "unknown"]
|
||
|
||
CONFIDENCE_ICON: dict[str, str] = {
|
||
"known": "🟢", "estimated": "🟡", "unknown": "🔴",
|
||
}
|
||
|
||
#: The six numeric baseline fields collected in the workshop form.
|
||
BASELINE_FIELDS: tuple[str, ...] = (
|
||
"annual_contact_volume",
|
||
"blended_cost_per_contact",
|
||
"agent_headcount",
|
||
"annual_attrition_rate",
|
||
"current_containment_rate",
|
||
"average_handle_time_seconds",
|
||
)
|
||
|
||
|
||
# ── Engagement data ──────────────────────────────────────────────────
|
||
|
||
|
||
class Participant(BaseModel):
|
||
name: str
|
||
role: str = ""
|
||
function: Function = "other"
|
||
|
||
|
||
class OperationalBaseline(BaseModel):
|
||
"""Contact-center MVP fields; other industries overlay their own."""
|
||
|
||
annual_contact_volume: int = Field(ge=0)
|
||
blended_cost_per_contact: float = Field(ge=0)
|
||
agent_headcount: int = Field(ge=0)
|
||
annual_attrition_rate: float = Field(ge=0, le=1)
|
||
current_containment_rate: float = Field(ge=0, le=1)
|
||
average_handle_time_seconds: int = Field(ge=0)
|
||
csat_baseline: float | None = None
|
||
revenue_at_risk: float | None = None
|
||
#: Per-field confidence flags (🟢 known / 🟡 estimated / 🔴 unknown).
|
||
field_confidence: dict[str, Confidence] = Field(default_factory=dict)
|
||
|
||
@field_validator("field_confidence")
|
||
@classmethod
|
||
def _known_fields_only(cls, v: dict[str, Confidence]) -> dict[str, Confidence]:
|
||
unknown = set(v) - set(BASELINE_FIELDS)
|
||
if unknown:
|
||
raise ValueError(f"confidence flags for unknown fields: {sorted(unknown)}")
|
||
return v
|
||
|
||
def confidence_for(self, field: str) -> Confidence:
|
||
return self.field_confidence.get(field, "estimated")
|
||
|
||
|
||
class CompetencyScore(BaseModel):
|
||
competency_id: str
|
||
dimension: str
|
||
score: int = Field(ge=1, le=5)
|
||
evidence: str = "" # one line: why this score
|
||
scorer_role: str = "facilitator" # MVP: facilitator consensus
|
||
scored_at: datetime
|
||
|
||
|
||
class DriverValue(BaseModel):
|
||
"""One value driver's theoretical annual value, as a range."""
|
||
|
||
driver_id: str
|
||
name: str
|
||
theoretical_low: float
|
||
theoretical_high: float
|
||
|
||
|
||
class UnlockMove(BaseModel):
|
||
"""One move in the unlock sequence: lift the binding set one level.
|
||
|
||
``value_unlocked_*`` is the **annual run-rate** realizable value the
|
||
lift adds (delta of the realization band times theoretical value).
|
||
Costs are ``None`` when the config has no entry for a lift — shown as
|
||
"cost not configured", never guessed.
|
||
"""
|
||
|
||
competency_ids: list[str]
|
||
current_level: int = Field(ge=1, le=4)
|
||
target_level: int = Field(ge=2, le=5)
|
||
est_cost_low: float | None = None
|
||
est_cost_high: float | None = None
|
||
est_weeks: int | None = None
|
||
value_unlocked_low: float
|
||
value_unlocked_high: float
|
||
note: str = ""
|
||
|
||
@model_validator(mode="after")
|
||
def _one_level_lift(self) -> "UnlockMove":
|
||
if self.target_level != self.current_level + 1:
|
||
raise ValueError("unlock moves lift exactly one level")
|
||
return self
|
||
|
||
|
||
class ValueAtStake(BaseModel):
|
||
theoretical_annual_value_low: float
|
||
theoretical_annual_value_high: float
|
||
realizable_18mo_low: float
|
||
realizable_18mo_high: float
|
||
trapped_value_low: float # annual: theoretical − realizable run-rate
|
||
trapped_value_high: float
|
||
binding_constraints: list[str] # competency_ids capping realization
|
||
unlock_sequence: list[UnlockMove]
|
||
# Self-explaining extras (additive to the build spec):
|
||
weakest_foundational_score: int = Field(ge=1, le=5)
|
||
realization_factor_low: float = Field(ge=0, le=1)
|
||
realization_factor_high: float = Field(ge=0, le=1)
|
||
driver_values: list[DriverValue] = Field(default_factory=list)
|
||
warnings: list[str] = Field(default_factory=list)
|
||
|
||
|
||
class Engagement(BaseModel):
|
||
engagement_id: str # e.g. "acme_2026-07-19"
|
||
client_name: str
|
||
industry_config: str # which config was loaded
|
||
facilitator: str
|
||
workshop_date: date
|
||
participants: list[Participant] = Field(default_factory=list)
|
||
operational_baseline: OperationalBaseline
|
||
scores: list[CompetencyScore] = Field(default_factory=list)
|
||
computed_value: ValueAtStake | None = None
|
||
notes: str = ""
|
||
|
||
|
||
# ── Config data (configs/*.yaml) ─────────────────────────────────────
|
||
|
||
|
||
class Dimension(BaseModel):
|
||
id: str
|
||
name: str
|
||
|
||
|
||
class Competency(BaseModel):
|
||
id: str
|
||
dimension: str
|
||
name: str
|
||
description: str
|
||
failure_vignette: str
|
||
level_descriptors: dict[int, str]
|
||
|
||
@field_validator("level_descriptors")
|
||
@classmethod
|
||
def _five_levels(cls, v: dict[int, str]) -> dict[int, str]:
|
||
if set(v) != {1, 2, 3, 4, 5}:
|
||
raise ValueError("level_descriptors must cover exactly levels 1–5")
|
||
return v
|
||
|
||
|
||
class CappingBand(BaseModel):
|
||
realized_low: float = Field(ge=0, le=1)
|
||
realized_high: float = Field(ge=0, le=1)
|
||
|
||
@model_validator(mode="after")
|
||
def _ordered(self) -> "CappingBand":
|
||
if self.realized_low > self.realized_high:
|
||
raise ValueError("realized_low > realized_high")
|
||
return self
|
||
|
||
|
||
class ValueDriver(BaseModel):
|
||
"""A configured value driver. ``kind`` selects the math in
|
||
value_math.py; ``value_formula`` documents it verbatim in exports."""
|
||
|
||
id: str
|
||
name: str
|
||
kind: Literal["containment_lift", "aht_reduction", "attrition_reduction"]
|
||
baseline_field: str
|
||
value_formula: str = ""
|
||
source: str = ""
|
||
# kind-specific parameters (validated in value_math dispatch):
|
||
lift_range_pts_low: float | None = None
|
||
lift_range_pts_high: float | None = None
|
||
reduction_pct_low: float | None = None
|
||
reduction_pct_high: float | None = None
|
||
cost_per_replacement_default: float | None = None
|
||
|
||
|
||
class LiftCost(BaseModel):
|
||
cost_low: float = Field(ge=0)
|
||
cost_high: float = Field(ge=0)
|
||
weeks: int = Field(ge=0)
|
||
|
||
@model_validator(mode="after")
|
||
def _ordered(self) -> "LiftCost":
|
||
if self.cost_low > self.cost_high:
|
||
raise ValueError("cost_low > cost_high")
|
||
return self
|
||
|
||
|
||
class DiagnosticConfig(BaseModel):
|
||
"""base.yaml merged with one industry overlay — what the engine consumes."""
|
||
|
||
version: str
|
||
industry: str
|
||
display_name: str
|
||
dimensions: list[Dimension]
|
||
competencies: list[Competency]
|
||
capping_heuristic: dict[int, CappingBand]
|
||
foundational_competencies: list[str]
|
||
value_drivers: list[ValueDriver] = Field(default_factory=list)
|
||
#: unlock_costs[competency_id]["lift_2_to_3"] -> LiftCost
|
||
unlock_costs: dict[str, dict[str, LiftCost]] = Field(default_factory=dict)
|
||
|
||
@model_validator(mode="after")
|
||
def _consistent(self) -> "DiagnosticConfig":
|
||
comp_ids = [c.id for c in self.competencies]
|
||
if len(comp_ids) != len(set(comp_ids)):
|
||
raise ValueError("duplicate competency ids")
|
||
dim_ids = {d.id for d in self.dimensions}
|
||
for c in self.competencies:
|
||
if c.dimension not in dim_ids:
|
||
raise ValueError(f"competency {c.id}: unknown dimension {c.dimension}")
|
||
missing = set(self.foundational_competencies) - set(comp_ids)
|
||
if missing:
|
||
raise ValueError(f"foundational competencies not defined: {sorted(missing)}")
|
||
if set(self.capping_heuristic) != {1, 2, 3, 4, 5}:
|
||
raise ValueError("capping_heuristic must cover exactly scores 1–5")
|
||
for lo, hi in zip(sorted(self.capping_heuristic), sorted(self.capping_heuristic)[1:]):
|
||
a, b = self.capping_heuristic[lo], self.capping_heuristic[hi]
|
||
if a.realized_low > b.realized_low or a.realized_high > b.realized_high:
|
||
raise ValueError("capping_heuristic must be non-decreasing in score")
|
||
for cid, lifts in self.unlock_costs.items():
|
||
if cid not in comp_ids:
|
||
raise ValueError(f"unlock_costs for unknown competency {cid}")
|
||
for key in lifts:
|
||
if not _valid_lift_key(key):
|
||
raise ValueError(f"unlock_costs[{cid}]: bad lift key {key!r}")
|
||
return self
|
||
|
||
def competency(self, competency_id: str) -> Competency:
|
||
for c in self.competencies:
|
||
if c.id == competency_id:
|
||
return c
|
||
raise KeyError(competency_id)
|
||
|
||
def dimension_name(self, dimension_id: str) -> str:
|
||
for d in self.dimensions:
|
||
if d.id == dimension_id:
|
||
return d.name
|
||
raise KeyError(dimension_id)
|
||
|
||
def lift_cost(self, competency_id: str, from_level: int) -> LiftCost | None:
|
||
return self.unlock_costs.get(competency_id, {}).get(
|
||
f"lift_{from_level}_to_{from_level + 1}")
|
||
|
||
|
||
def _valid_lift_key(key: str) -> bool:
|
||
parts = key.split("_") # "lift", a, "to", b — one-level lifts only
|
||
return (len(parts) == 4 and parts[0] == "lift" and parts[2] == "to"
|
||
and parts[1].isdigit() and parts[3].isdigit()
|
||
and int(parts[3]) == int(parts[1]) + 1 and 1 <= int(parts[1]) <= 4)
|