Implementing Delta Checks for Electrolyte Panels in Python

Problem Statement

Electrolyte results change fast and for many benign reasons — fluid shifts, a fresh draw, a switched reagent lot — so a delta engine tuned for stable analytes generates a flood of false holds when pointed at sodium, potassium, chloride, and bicarbonate. The specific failure this page solves is building an analyte-aware delta check that separates physiologically implausible change from pre-analytically explainable change (above all, spurious potassium elevation from in-vitro hemolysis), computes the delta deterministically against a UTC-aligned baseline, and emits an auditable verdict fast enough to sit inline in result posting.

This is the concrete, electrolyte-specific build of the stage specified in Delta Validation & Trend Analysis; it inherits that stage’s stateful history-lookup contract and its “fail toward review” invariant, and only adds the tuning that electrolyte panels demand.

Prerequisites

  • Runtime: Python 3.11+ (for X | None unions and datetime.UTC).
  • Libraries: pydantic>=2.6 for typed rule and result models; pytest and pytest-asyncio for the test suite; hypothesis if you adopt the property tests from the parent stage.
  • Instrument firmware: analyzers must export a serum indices / HIL panel — specifically a hemolysis index (H) on the mg/dL free-hemoglobin scale — alongside the chemistry result. Without an instrument hemolysis flag, potassium delta suppression is not reliable and must fall back to manual specimen inspection.
  • Regulatory baseline: the laboratory has documented delta-check criteria on file (a CLIA §493.1253 results-review requirement) and an append-only audit store, as built in implementing HIPAA-compliant audit trails in LIMS.
  • Upstream contract: results reach this code already parsed and normalized to canonical analyte codes and UCUM units by the HL7 v2 Segment Mapping layer. Delta evaluation never parses raw messages itself.
Electrolyte delta-check data flow A normalized electrolyte result enters the delta engine. The engine reads a prior baseline from a Historical Results store within the analyte's lookback window, then evaluates four per-analyte rules — sodium, chloride, and bicarbonate compute an absolute delta directly, while potassium first passes through a hemolysis gate that suppresses spurious in-vitro rises when the instrument hemolysis index meets its threshold. Every verdict is written to an append-only audit store, and high-priority potassium breaches fan out to critical-value routing. Normalized result canonical code · UCUM UTC · HIL index Delta engine baseline select + rule dispatch Historical Results prior baseline, lookback window read Na · Δ > 5.0 medium Cl · Δ > 5.0 medium CO2 · Δ > 4.0 low K · Δ > 1.0 high hemolysis gate HIL ≥ 50 mg/dL suppress Verdict breached · reason · priority Audit store append-only, every verdict Critical routing page high-priority K breaches

Step-by-Step Implementation

Step 1: Model the analyte rules, not one global threshold

Electrolytes need per-analyte thresholds and lookback windows because their physiological half-lives and clinical stakes differ — a 1.0 mmol/L potassium swing is an emergency, the same absolute change in chloride is noise. Encode each rule as a validated pydantic model so thresholds are version-controlled data, not literals buried in branch logic. Use Decimal throughout to avoid binary-float rounding at the threshold boundary.

python
from __future__ import annotations

from decimal import Decimal
from enum import Enum

from pydantic import BaseModel, Field


class ClinicalPriority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"


class DeltaRule(BaseModel):
    """One analyte's delta criteria. Serialized to config and lot-mapped."""

    model_config = {"frozen": True}

    analyte: str
    threshold: Decimal = Field(gt=0)      # absolute change, in `unit`
    unit: str = "mmol/L"
    lookback_hours: int = Field(gt=0)
    priority: ClinicalPriority
    hemolysis_suppress_at: int | None = None  # HIL index (mg/dL); K only


ELECTROLYTE_RULES: dict[str, DeltaRule] = {
    "Na":  DeltaRule(analyte="Na",  threshold=Decimal("5.0"), lookback_hours=72,
                     priority=ClinicalPriority.MEDIUM),
    "K":   DeltaRule(analyte="K",   threshold=Decimal("1.0"), lookback_hours=72,
                     priority=ClinicalPriority.HIGH, hemolysis_suppress_at=50),
    "Cl":  DeltaRule(analyte="Cl",  threshold=Decimal("5.0"), lookback_hours=72,
                     priority=ClinicalPriority.MEDIUM),
    "CO2": DeltaRule(analyte="CO2", threshold=Decimal("4.0"), lookback_hours=72,
                     priority=ClinicalPriority.LOW),
}

Keep ELECTROLYTE_RULES version-controlled and mapped to instrument reagent lot numbers. Reagent-lot changes shift the analytical baseline, and a static threshold that was correct for the prior lot silently drifts into over- or under-flagging — the calibration feedback that governs these values lives in Threshold Tuning & Calibration.

Step 2: Select the baseline against a UTC-aligned window

The single most common source of false electrolyte deltas is timestamp handling: a timezone-naive record or a millisecond-off comparison selects the wrong prior result. Normalize every timestamp to UTC at ingestion, model the result as a frozen pydantic type, and make baseline selection a pure function of a history port so it is trivially testable.

python
from datetime import datetime, timedelta
from typing import Protocol


class ElectrolyteResult(BaseModel):
    model_config = {"frozen": True}

    result_id: str
    patient_id: str
    analyte: str
    value: Decimal
    unit: str
    collected_at: datetime          # MUST be timezone-aware UTC
    hemolysis_index: int = 0         # instrument HIL flag, mg/dL scale


class HistoryPort(Protocol):
    async def recent(
        self, patient_id: str, analyte: str, *, since: datetime, before: datetime
    ) -> list[ElectrolyteResult]: ...


async def select_baseline(
    current: ElectrolyteResult, rule: DeltaRule, history: HistoryPort
) -> ElectrolyteResult | None:
    """Most recent prior result for the same patient+analyte in the window."""
    if current.collected_at.tzinfo is None:
        raise ValueError("collected_at must be timezone-aware (UTC required).")

    since = current.collected_at - timedelta(hours=rule.lookback_hours)
    prior = await history.recent(
        current.patient_id, current.analyte,
        since=since, before=current.collected_at,
    )
    valid = [r for r in prior if r.collected_at < current.collected_at]
    if not valid:
        return None
    return max(valid, key=lambda r: r.collected_at)

The before bound is strict (<), never <=, so a result never selects itself as its own baseline on reprocessing. This mirrors the deterministic baseline-selection guarantee described in the parent stage and keeps reprocessing idempotent.

Step 3: Evaluate the delta with a potassium hemolysis gate

This is the electrolyte-specific core. In-vitro hemolysis leaks intracellular potassium into serum, so a hemolyzed specimen reads a spuriously high potassium that a naive delta engine flags as a critical rise. Gate potassium on the instrument hemolysis index before computing the delta: above the suppression threshold, the result is a specimen-integrity problem, not a physiological change, and the correct action is a redraw comment — not a delta hold that a clinician would act on.

python
class DeltaVerdict(BaseModel):
    analyte: str
    breached: bool
    delta: Decimal | None
    baseline_id: str | None
    reason: str
    priority: ClinicalPriority


def evaluate_delta(
    current: ElectrolyteResult,
    baseline: ElectrolyteResult | None,
    rule: DeltaRule,
) -> DeltaVerdict:
    # Potassium: suppress spurious in-vitro-hemolysis rise before any delta math.
    if (
        rule.hemolysis_suppress_at is not None
        and current.hemolysis_index >= rule.hemolysis_suppress_at
    ):
        return DeltaVerdict(analyte=rule.analyte, breached=False, delta=None,
                            baseline_id=None, reason="hemolysis_suppressed",
                            priority=rule.priority)

    if baseline is None:
        return DeltaVerdict(analyte=rule.analyte, breached=False, delta=None,
                            baseline_id=None, reason="no_baseline_in_window",
                            priority=rule.priority)

    delta = abs(current.value - baseline.value)
    breached = delta > rule.threshold
    return DeltaVerdict(
        analyte=rule.analyte,
        breached=breached,
        delta=delta,
        baseline_id=baseline.result_id,
        reason="delta_exceeded" if breached else "within_threshold",
        priority=rule.priority,
    )

A no_baseline_in_window verdict is a pass, not an error — a first-ever electrolyte panel has nothing to compare against and must post. This is the same static-limit boundary the stateless Reference Range Check Implementation already covered upstream, so a first result is not left unguarded.

Step 4: Emit an immutable audit record and route by priority

Every evaluation — pass, breach, or suppression — must produce a traceable record, and clinically significant breaches must escalate on the analyte’s priority tier. Write the audit record inside the same transactional boundary as the verdict so a partial failure can never leave a decision without its provenance, then route high-priority breaches to paging.

python
import uuid
from datetime import UTC


class DeltaAudit(BaseModel):
    audit_id: str
    patient_id: str
    result_id: str
    verdict: DeltaVerdict
    rule_version: str
    decided_at: datetime


async def run_panel(
    results: list[ElectrolyteResult],
    history: HistoryPort,
    audit_sink,   # append-only store; awaited write
    router,       # critical-value dispatcher
    rule_version: str = "2026.07",
) -> list[DeltaVerdict]:
    verdicts: list[DeltaVerdict] = []
    for current in results:
        rule = ELECTROLYTE_RULES.get(current.analyte)
        if rule is None:
            continue  # not an electrolyte we delta-check
        baseline = await select_baseline(current, rule, history)
        verdict = evaluate_delta(current, baseline, rule)

        await audit_sink.append(DeltaAudit(
            audit_id=str(uuid.uuid4()),
            patient_id=current.patient_id,
            result_id=current.result_id,
            verdict=verdict,
            rule_version=rule_version,
            decided_at=datetime.now(UTC),
        ))
        if verdict.breached and verdict.priority is ClinicalPriority.HIGH:
            await router.page(current, verdict)   # e.g. potassium
        verdicts.append(verdict)
    return verdicts

A breached potassium delta bypasses the standard review queue and pages immediately, while medium- and low-priority breaches route to a secondary validation workbench — the escalation policy is owned by Critical Value Alert Routing, which consumes these verdicts.

Verification & Testing

Confirm the two behaviors that matter most: the potassium hemolysis gate fires before delta math, and a breach produces the exact verdict fields the audit record depends on. Pin them with pytest.

python
import pytest
from decimal import Decimal
from datetime import datetime, timedelta, UTC


def _result(analyte, value, *, hil=0, hours_ago=0, rid="r"):
    return ElectrolyteResult(
        result_id=rid, patient_id="P1", analyte=analyte,
        value=Decimal(value), unit="mmol/L",
        collected_at=datetime.now(UTC) - timedelta(hours=hours_ago),
        hemolysis_index=hil,
    )


def test_hemolyzed_potassium_is_suppressed_not_flagged():
    current = _result("K", "6.2", hil=120)          # spuriously high, hemolyzed
    baseline = _result("K", "4.1", hours_ago=24, rid="b")
    verdict = evaluate_delta(current, baseline, ELECTROLYTE_RULES["K"])
    assert verdict.breached is False
    assert verdict.reason == "hemolysis_suppressed"


def test_true_potassium_rise_breaches_with_baseline_id():
    current = _result("K", "6.2", hil=5)            # clean specimen
    baseline = _result("K", "4.1", hours_ago=24, rid="b")
    verdict = evaluate_delta(current, baseline, ELECTROLYTE_RULES["K"])
    assert verdict.breached is True
    assert verdict.delta == Decimal("2.1")
    assert verdict.baseline_id == "b"

Expected results: both tests pass, the first returning reason="hemolysis_suppressed" with delta=None, the second returning delta=Decimal("2.1") and breached=True. Extend the suite with golden-file fixtures — (current, baseline) → expected verdict cases per analyte, including the boundary where delta == threshold exactly (which must not breach, since the comparison is strict >) and where the baseline sits exactly on the lookback edge.

Compliance Note

Delta checks are a recognized mechanism for the results-review obligation in CLIA §493.1253(b)(2), which requires the laboratory to establish, verify, and follow procedures for reviewing results before release. The hold-to-review path here, plus the documented per-analyte criteria in ELECTROLYTE_RULES, is the auditable evidence that clinically significant change is not auto-released. Because each DeltaAudit record is attributable, timestamped in UTC, and written immutably in the verdict’s transaction, it also satisfies the electronic-record requirements of 21 CFR Part 11.10(e) for a tamper-evident review trail; the history read stays inside the CLIA/CAP data boundaries that segment protected patient data from downstream interfaces.

Troubleshooting

Every potassium result on a hemolyzed run is flagging as a critical delta

The hemolysis gate is not seeing an instrument index. Confirm the analyzer exports its serum-indices (HIL) panel and that hemolysis_index is populated on the canonical ElectrolyteResult — a default of 0 means “no hemolysis” and disables suppression. Verify the mapping in the HL7 v2 Segment Mapping layer carries the H index OBX segment through, and that hemolysis_suppress_at is set on the potassium rule.

A patient’s first-ever panel is being held instead of posting

select_baseline correctly returns None when no prior result falls in the window, and evaluate_delta turns that into a no_baseline_in_window pass. If the record is being held, the hold is coming from a different stage — check that your disposition logic treats breached=False verdicts as releasable rather than treating any non-empty verdict as a flag.

Deltas flip between held and released when I reprocess the same message

This is a timestamp-equality bug. The before bound must be strict (collected_at < current.collected_at); a <= comparison lets a result select itself as its own baseline on the second pass, yielding a delta of zero and a different verdict. Confirm the strict comparison and that all timestamps are timezone-aware UTC, not local time.

Thresholds look right in code but flagging rates jumped after a reagent change

The static rule set drifted relative to the new reagent lot. Map ELECTROLYTE_RULES (and its rule_version) to instrument lot numbers and reload on lot change, then let the population-based feedback in handling out-of-range flags without manual intervention recalibrate the thresholds against observed false-positive rates.

The delta engine is timing out under load and blocking result posting

The history lookup is on the synchronous path. Keep run_panel fully async, put HistoryPort.recent behind a bounded connection pool with a per-call timeout, and short-circuit to a history_unavailable hold (never a pass) when the store is slow — the same circuit-breaker discipline the parent stage specifies. Cache recent baselines so repeat draws on the same admission do not re-query.

Part of: Delta Validation & Trend Analysis