Mapping Excursion Severity to CAPA Priority in Python

The severity-to-priority mapping is the single most scrutinized function in a cold chain quality system, because it is where a physical measurement becomes a regulatory decision. This guide builds that mapping as a deterministic, change-controlled Python function that consumes an excursion’s Mean Kinetic Temperature (MKT), its cumulative out-of-band duration, and its product stability class, and emits a corrective and preventive action (CAPA) priority with an audit-logged decision record. It is the classification core that the stateful engine in CAPA routing automation for temperature excursions wraps with owner resolution and escalation; here the focus is the decision itself, and proving which logic produced it.

Regulatory hook

Two clauses govern this function directly. ICH Q9 requires the formality and documentation of a quality action to be commensurate with quantified risk, which is exactly what a severity-to-priority map encodes — a small, brief drift on a robust product and a large, sustained excursion on a biologic must not receive the same response. 21 CFR Part 11 §11.10(k) requires change control over the procedures that operate the system, so the mapping table is a controlled artifact: it must be versioned, its version bound to every decision, and any edit re-validated. The decision record itself is a §11.10(e) audit-trail entry — attributable, time-stamped, and tamper-evident.

The reason this function attracts more inspection attention than any other in the disposition stack is that it is the point where quantitative risk becomes a categorical regulatory obligation. Everything upstream produces numbers — a Mean Kinetic Temperature, an exposure duration, a normalized score — and numbers are auditable but not, in themselves, decisions. The moment the mapping assigns a priority, it commits the organization to a defined response: a P1 opens a critical investigation with a 24-hour clock, a P4 logs and moves on. An inspector who disagrees with a disposition will trace it back to this table and ask two questions — was this the approved logic at the time, and was it applied deterministically — and the answer to both must be recoverable from the record alone. That is why the version fingerprint is not an optimization but a compliance requirement.

Prerequisites

  • Python 3.11 or newer (uses tuple[...] generics and match where convenient).
  • Dependencies: the mapping uses only the standard library — hashlib, json, decimal, datetime, enum. Install nothing beyond a clean interpreter:
bash
python3.11 -m venv .venv && source .venv/bin/activate
python -m pip install --upgrade pip
  • Inputs assumed upstream: each excursion arrives with a computed MKT (°C), a cumulative out-of-band duration (minutes), and a validated stability class. MKT and duration come from the scoring layer described in duration-based scoring for temperature excursions.
  • Access control: the mapping table is a controlled document. Only change-management-approved edits reach the deployed _MATRIX, and the deployment pins the approved fingerprint.

Step-by-Step Implementation

Step 1 — Model the inputs and the priority enum

Represent the decision inputs explicitly so the function signature is the contract. Use Decimal for MKT and duration to keep the classification bit-reproducible across hosts.

python
from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal
from enum import Enum


class CapaPriority(str, Enum):
    # ICH Q9: priority scales with quantified risk, not with reviewer intuition.
    P1_CRITICAL = "P1_CRITICAL"
    P2_MAJOR = "P2_MAJOR"
    P3_MINOR = "P3_MINOR"
    P4_LOG_ONLY = "P4_LOG_ONLY"


@dataclass(frozen=True)
class SeverityInput:
    excursion_id: str
    stability_class: str          # e.g. "biologic_2_8", "small_molecule_ambient"
    mkt_c: Decimal                # Mean Kinetic Temperature over the event
    cumulative_minutes: Decimal   # out-of-band exposure time
    spec_high_c: Decimal          # upper validated storage limit for the product

Step 2 — Encode the severity matrix as controlled data

The mapping is data, not branching logic scattered through code, so it can be serialized, fingerprinted, and reviewed as a controlled document. Each rule states the minimum MKT margin above spec and the minimum duration that qualifies for a priority.

python
# Rule = (min_margin_c_above_spec, min_cumulative_minutes, priority).
# Ordered most-severe first; the first satisfied rule wins. This whole table is
# a controlled artifact under 21 CFR Part 11 §11.10(k) — version it, review it.
_MATRIX: tuple[tuple[Decimal, Decimal, CapaPriority], ...] = (
    (Decimal("3.0"), Decimal("120"), CapaPriority.P1_CRITICAL),
    (Decimal("3.0"), Decimal("0"),   CapaPriority.P2_MAJOR),
    (Decimal("1.0"), Decimal("60"),  CapaPriority.P2_MAJOR),
    (Decimal("0.5"), Decimal("0"),   CapaPriority.P3_MINOR),
    (Decimal("0"),   Decimal("0"),   CapaPriority.P4_LOG_ONLY),
)

# Stability classes that may never be logged-only or minor on real exposure.
_COLD_SENSITIVE = {"biologic_2_8", "vaccine_2_8", "cell_therapy_cryo"}


def _matrix_fingerprint() -> str:
    # §11.10(k): bind the exact decision table to every record it produces.
    payload = json.dumps(
        [[str(m), str(d), p.value] for m, d, p in _MATRIX],
        separators=(",", ":"),
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

Step 3 — Implement the deterministic mapping

The function computes the MKT margin above the product’s validated limit, walks the matrix in severity order, and applies a stability-class floor so a cold-sensitive product cannot be downgraded below Major on genuine exposure.

python
def map_severity(inp: SeverityInput) -> CapaPriority:
    """Deterministic MKT+duration -> CAPA priority (ICH Q9 proportionality)."""
    margin = inp.mkt_c - inp.spec_high_c          # thermal severity above spec
    priority = CapaPriority.P4_LOG_ONLY
    for min_margin, min_minutes, band in _MATRIX:
        if margin >= min_margin and inp.cumulative_minutes >= min_minutes:
            priority = band
            break
    # Stability-class floor: cold-sensitive product never below Major on
    # real out-of-band exposure — a documented ICH Q9 risk override.
    if inp.stability_class in _COLD_SENSITIVE and margin > 0:
        if priority in (CapaPriority.P4_LOG_ONLY, CapaPriority.P3_MINOR):
            priority = CapaPriority.P2_MAJOR
    return priority

Two design choices in this function are worth defending explicitly. First, the MKT margin is computed against the product’s validated upper limit rather than a facility-wide constant, so the same absolute temperature can be Minor for an ambient-stable tablet and Critical for a 2–8 °C biologic — the severity is relative to the stability data, as ICH Q9 risk assessment requires. Second, both the margin and the duration must clear a rule’s thresholds for it to fire, which is what stops a single high spike with no dwell time from routing as Critical: a momentary probe artifact has magnitude but effectively zero duration, so it falls through the high-severity rules to a lower band, exactly as a duration-weighted risk model should behave.

Verify the boundary and the override with an assertion:

python
# A 4 °C, 3-hour biologic excursion must be Critical; a 0.6 °C, 0-minute
# small-molecule blip must be Minor. Boundaries are deterministic under ICH Q9.
crit = SeverityInput("EXC-A", "biologic_2_8", Decimal("12.0"),
                     Decimal("180"), Decimal("8.0"))
minor = SeverityInput("EXC-B", "small_molecule_ambient", Decimal("25.6"),
                      Decimal("0"), Decimal("25.0"))
assert map_severity(crit) == CapaPriority.P1_CRITICAL
assert map_severity(minor) == CapaPriority.P3_MINOR

Step 4 — Emit an audit-logged decision record

The priority alone is not defensible; the decision — its inputs, the resulting band, and the version of the logic that produced it — must be captured as a tamper-evident record. Chain each record to its predecessor.

python
def build_decision_record(inp: SeverityInput, prev_hash: str = "0" * 64) -> dict:
    priority = map_severity(inp)
    record = {
        "excursion_id": inp.excursion_id,
        "stability_class": inp.stability_class,
        "mkt_c": str(inp.mkt_c),
        "cumulative_minutes": str(inp.cumulative_minutes),
        "margin_c": str(inp.mkt_c - inp.spec_high_c),
        "priority": priority.value,
        "matrix_version": _matrix_fingerprint(),          # §11.10(k) change control
        "decided_at": datetime.now(timezone.utc).isoformat(),  # §11.10(e) contemporaneous
        "prev_hash": prev_hash,
    }
    canonical = json.dumps(record, sort_keys=True, separators=(",", ":"))
    # §11.10(e): tamper-evident link — any later edit breaks the chain.
    record["record_hash"] = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
    return record

Confirm the decision is reproducible and self-verifying:

python
rec = build_decision_record(crit)
again = build_decision_record(crit, prev_hash=rec["prev_hash"])
# §11.10(a): identical inputs yield identical priority and logic fingerprint.
assert rec["priority"] == again["priority"]
assert rec["matrix_version"] == again["matrix_version"]

Step 5 — Pin the approved matrix version at deploy time

A validated mapping is only validated if the running code is the approved code. Assert the deployed fingerprint against the change-controlled value on startup and refuse to run on a mismatch.

python
import os


def assert_approved_matrix() -> None:
    approved = os.environ.get("CAPA_MATRIX_FINGERPRINT")
    if not approved:
        raise SystemExit("CAPA_MATRIX_FINGERPRINT is required (change-control gate)")
    if _matrix_fingerprint() != approved:
        # §11.10(k): running logic must equal the approved, validated logic.
        raise SystemExit("severity matrix differs from approved version — halting")
bash
# The approved fingerprint is issued by change management and set in the environment.
CAPA_MATRIX_FINGERPRINT="$(python -c 'import app; print(app._matrix_fingerprint())')" \
  python -c 'import app; app.assert_approved_matrix(); print("matrix approved")'

Compliance Validation Checklist

Run this as part of computerized-system validation; each item is independently confirmable by an auditor.

Troubleshooting

Symptom Root cause Fix
Same excursion yields different priority across hosts Float arithmetic on MKT/margin Use Decimal end to end; never mix float into the comparison
Cold-sensitive product routed as log-only Stability class missing from _COLD_SENSITIVE Add the class to the controlled set and re-validate the override
matrix_version differs between staging and prod Unapproved edit reached one environment Pin CAPA_MATRIX_FINGERPRINT; block the deploy on mismatch
Boundary case routes one band too low Using > instead of >= at rule thresholds Use inclusive >= so the documented boundary belongs to the higher band
Decision record verifies but priority looks wrong Matrix rules out of severity order Keep _MATRIX ordered most-severe first; the first match must win

For architectural context, see CAPA routing automation for temperature excursions, part of the broader QMS integration, audit trails & automated batch disposition section.