CX Discovery Notebook

studies/202607_CX_AI_Diagnostic/diaglib/models.py

@@ -0,0 +1,281 @@
+"""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)