Type Coercion for Clinical Data Types

Type coercion is the transformation step where a string like 98.6, a partial date like 2015-03, or a coded value like 2345-7^Glucose^LN becomes a typed, canonical value your warehouse and decision-support engines can trust. In clinical data engineering it is not a syntactic convenience — it is a deterministic, auditable discipline. Implicit casts and ad-hoc parsing introduce silent corruption: a truncated lab decimal becomes a wrong reference-range flag, a dropped timezone offset inverts a medication-administration timeline, and an invalid string coerced to zero becomes a clinically dangerous result. Within the broader Clinical Data Parsing & Transformation Workflows pipeline, coercion sits at the boundary between raw ingestion and canonical normalization — after schema validation, before the typed values are projected into the target store. This page covers the concrete engineering of that boundary: a versioned coercion registry, precision-preserving numeric handling, strict temporal normalization, explicit null semantics, and the immutable audit trail that keeps the whole operation defensible under regulatory scrutiny.

The reader who needs this page is usually debugging a specific failure — a FLOAT64 rounding artifact in a dosage column, a partial date that silently became midnight UTC, or a deprecated code that slipped into an analytical table. The patterns below are written to be lifted into a production Python pipeline and tested in isolation.

Prerequisites & Context

These items are load-bearing for the implementation that follows. Confirm each before wiring coercion into your pipeline.

• Python 3.11+ with the standard library decimal and datetime modules, plus pydantic v2 for typed validation.
• A validated FHIR layer — this page assumes the Pydantic-backed models from using fhir.resources for Python ETL for FHIR R4 payloads.
• HL7 v2 segment extraction in place — the field-access patterns from the HL7 Python library integration guide are used for the OBX examples below.
• Access to a FHIR terminology server for validating coded values against active value sets — coercion of code/Coding types must never resolve against a hardcoded lookup.
• A quarantine sink (DLQ + encrypted object store) for payloads that fail coercion, so malformed records are isolated rather than dropped or guessed.
• A target schema with explicit numeric scale/precision (NUMERIC(p,s) / DECIMAL) — not a default FLOAT64 column — for any clinical decimal.

Coercion runs after structural validation. Invalid payloads should never reach the coercion stage; they bypass it and route to error handling, exactly as in the async batch worker control flow.

Concept & Spec Detail: Deterministic Coercion Rules

A correct coercion layer is defined by three non-negotiable properties, then a rule table that maps every source representation to a canonical typed output.

1. Idempotency. Re-running coercion against the same raw payload must yield byte-identical canonical output. This requires stateless transformation functions, a versioned rule registry, and deterministic fallback hierarchies. Ambiguous states map to explicit semantic values rather than being collapsed.

2. Precision preservation. Clinical decimals (lab results, dosages, vitals) and temporal types cannot tolerate IEEE-754 floating-point drift or implicit timezone resolution. Coercion uses fixed-precision arithmetic (decimal.Decimal) and strict ISO 8601 parsing with explicit offset retention.

3. Explicit failure modes. Coercion never silently truncates, coerces an invalid string to zero, or guesses a missing unit. Fail-fast validation routes malformed payloads to quarantine, logged with a cryptographic hash of the source.

Coercion rules by clinical type

The registry is the single source of truth for how each source type becomes a canonical value. Treat this table as the contract the implementation must satisfy.

The last row is where most pipelines quietly fail. Regulatory frameworks distinguish NULL (data not collected) from UNKNOWN (collected but unverified) and NOT_APPLICABLE (clinically irrelevant). A registry that collapses all three into a single SQL NULL destroys information an auditor or analyst depends on.

Implementation

The following steps build a deterministic coercion layer incrementally. Each step is independently testable.

Step 1: A versioned coercion registry

Centralize coercion logic behind a registry keyed by clinical type. Versioning the registry is what makes coercion auditable: every transformed field records which rule version produced it.

from decimal import Decimal, InvalidOperation
from datetime import datetime, timezone
from dataclasses import dataclass
from typing import Callable, Any

REGISTRY_VERSION = "2024.1.0"

@dataclass
class CoercionResult:
    status: str          # "coerced" | "explicit_null" | "quarantined"
    value: Any = None
    canonical_null: str | None = None   # NULL | UNKNOWN | NOT_APPLICABLE
    rule_version: str = REGISTRY_VERSION
    error: str | None = None

NULL_SENTINELS = {
    "": "NULL", "NA": "NULL", "NULL": "NULL",
    "UNK": "UNKNOWN", "ASKU": "UNKNOWN",
    "NAV": "NOT_APPLICABLE", "NASK": "NOT_APPLICABLE",
}

def classify_null(raw: str) -> str | None:
    """Map a source sentinel to a canonical null state, or None if it is real data."""
    return NULL_SENTINELS.get(raw.strip().upper())

Validation — the null classifier must keep the three states distinct:

assert classify_null("UNK") == "UNKNOWN"
assert classify_null("") == "NULL"
assert classify_null("NASK") == "NOT_APPLICABLE"
assert classify_null("98.5") is None

Step 2: Precision-preserving numeric coercion

Numeric coercion is where FLOAT64 drift originates. Use Decimal, quantize to a declared scale, and strip locale grouping separators before parsing.

from decimal import Decimal, InvalidOperation, ROUND_HALF_UP

def coerce_numeric(raw: str, scale: str = "0.01") -> CoercionResult:
    """Coerce a clinical numeric string to a fixed-scale Decimal."""
    null_state = classify_null(raw)
    if null_state is not None:
        return CoercionResult(status="explicit_null", canonical_null=null_state)
    try:
        cleaned = raw.strip().replace(",", "")        # strip grouping separators
        if "e" in cleaned.lower():
            raise InvalidOperation("scientific notation rejected")
        value = Decimal(cleaned).quantize(Decimal(scale), rounding=ROUND_HALF_UP)
        return CoercionResult(status="coerced", value=value)
    except InvalidOperation as exc:
        return CoercionResult(status="quarantined", error=f"non-numeric: {exc}")

Validation — assert no precision loss and explicit rejection of bad input:

assert coerce_numeric("1,250.5", "0.1").value == Decimal("1250.5")
assert coerce_numeric("UNK").canonical_null == "UNKNOWN"
assert coerce_numeric("abc").status == "quarantined"
assert coerce_numeric("1.005", "0.01").value == Decimal("1.01")  # ROUND_HALF_UP

Step 3: Strict temporal normalization

Temporal coercion must retain the source offset and reject naive datetimes. Partial dates are not errors — they are mapped to a period range rather than an implied midnight. The full set of offset-drift cases is covered in debugging timezone mismatches in clinical timestamps.

import re

PARTIAL_DATE = re.compile(r"^\d{4}(-\d{2})?$")  # "2020" or "2020-03"

def coerce_temporal(raw: str) -> CoercionResult:
    """Coerce a clinical timestamp to an offset-aware UTC datetime, or a period range."""
    null_state = classify_null(raw)
    if null_state is not None:
        return CoercionResult(status="explicit_null", canonical_null=null_state)

    raw = raw.strip()
    if PARTIAL_DATE.match(raw):
        # Map a partial date to an explicit range; never assume midnight.
        return CoercionResult(status="coerced", value={"partial": raw})

    try:
        parsed = datetime.fromisoformat(raw)
        if parsed.tzinfo is None:
            return CoercionResult(status="quarantined",
                                  error="naive datetime rejected (no offset)")
        return CoercionResult(status="coerced", value=parsed.astimezone(timezone.utc))
    except ValueError as exc:
        return CoercionResult(status="quarantined", error=f"invalid timestamp: {exc}")

Validation:

r = coerce_temporal("2024-01-15T08:30:00-05:00")
assert r.value.isoformat() == "2024-01-15T13:30:00+00:00"   # offset preserved → UTC
assert coerce_temporal("2015-03").value == {"partial": "2015-03"}
assert coerce_temporal("2024-01-15T08:30:00").status == "quarantined"  # naive rejected

Step 4: FHIR resource coercion

When parsing FHIR resources, a validated object model prevents silent schema drift. Coerce valueQuantity and effectiveDateTime through the registry rather than reading raw JSON.

from fhir.resources.observation import Observation
from pydantic import ValidationError

def coerce_observation(obs_json: dict) -> dict:
    """Deterministic coercion of a FHIR Observation's value and effective time."""
    try:
        obs = Observation.model_validate(obs_json)   # structural validation first
    except ValidationError as exc:
        return {"status": "quarantined", "error": str(exc)}

    if not obs.valueQuantity:
        return {"status": "explicit_null", "canonical_null": "NULL"}

    num = coerce_numeric(str(obs.valueQuantity.value), scale="0.0001")
    if num.status == "quarantined":
        return {"status": "quarantined", "error": num.error}

    effective = None
    if obs.effectiveDateTime:
        t = coerce_temporal(str(obs.effectiveDateTime))
        if t.status == "quarantined":
            return {"status": "quarantined", "error": t.error}
        effective = t.value.isoformat() if hasattr(t.value, "isoformat") else t.value

    return {
        "status": num.status,
        "value": float(num.value) if num.value is not None else None,
        "unit": obs.valueQuantity.unit,        # normalize to UCUM downstream
        "effective_utc": effective,
        "rule_version": REGISTRY_VERSION,
    }

Step 5: HL7 v2 OBX coercion

HL7 v2 relies on delimited strings and implicit typing. Extract the OBX-5 value via the library, then route it through the same numeric coercion so FHIR and v2 paths share one rule set. For the segment-to-resource mapping, see converting HL7 v2 OBX segments to FHIR Observation.

from hl7apy.parser import parse_message

def coerce_obx_numeric(raw_obx_segment: str) -> CoercionResult:
    """Parse an HL7 v2 OBX segment and coerce OBX-5 to a fixed-scale Decimal.

    Expects one OBX line, e.g.:
      'OBX|1|NM|2345-7^Glucose^LN||98.5|mg/dL|...'
    """
    wrapper = (
        "MSH|^~\\&|LAB|FAC|RCV|FAC|20240101120000||ORU^R01|1|P|2.5\r"
        "OBR|1\r"
        f"{raw_obx_segment}\r"
    )
    try:
        msg = parse_message(wrapper, find_groups=False)
        obx_list = [c for c in msg.children if c.name == "OBX"]
        if not obx_list:
            return CoercionResult(status="quarantined", error="no OBX segment")
        raw_value = obx_list[0].obx_5.value.strip() if obx_list[0].obx_5 else ""
    except Exception as exc:  # hl7apy raises a range of parse exceptions
        return CoercionResult(status="quarantined", error=f"parse failure: {exc}")

    return coerce_numeric(raw_value, scale="0.01")

Validation:

ok = coerce_obx_numeric("OBX|1|NM|2345-7^Glucose^LN||98.5|mg/dL|")
assert ok.status == "coerced" and ok.value == Decimal("98.50")

bad = coerce_obx_numeric("OBX|1|NM|2345-7^Glucose^LN||positive|mg/dL|")
assert bad.status == "quarantined"

Edge Cases & Vendor Deviations

Clinical data rarely conforms to textbook schemas. Handle these explicitly rather than letting them surface as silent data loss.

The comparator case deserves emphasis: coercing <5 by stripping the < and storing 5 is a clinically dangerous data loss. The comparator carries meaning (the true value is below the detection limit) and must be preserved in a dedicated field.

Compliance Note: Provenance for Coerced Values

Coercion is a transformation of a clinical record, so under the HIPAA Security Rule, 21 CFR Part 11, and GDPR data-minimization it must be reconstructable. The audit requirement that applies specifically to this stage is lineage: every coerced field must let an auditor reconstruct the exact path from raw payload to analytical column.

• Immutable provenance logging. For each coercion event, append (never overwrite) source_payload_hash, rule_version, input_type, output_type, and coercion_timestamp. The rule_version from the registry is what lets you replay historical coercions exactly.
• Hash the source, not the PHI. Log a SHA-256 hash of the raw value for traceability; never write the raw clinical value or patient identifier into the coercion log.
• Preserve null semantics end to end. Carry the canonical null state (NULL / UNKNOWN / NOT_APPLICABLE) through to the target schema. Collapsing them at load time is both an analytical defect and a compliance gap, because “not collected” and “patient declined” are legally distinct.
• Quarantine without leaking. A payload that fails coercion still contains PHI. Route it to an encrypted DLQ with a hashed reference, exactly as the async batch worker does — never inline the raw value in the error record.

Troubleshooting

Related

• Debugging timezone mismatches in clinical timestamps
• Using fhir.resources for Python ETL
• HL7 Python library integration guide
• FHIR terminology server integration
• Clinical Data Parsing & Transformation Workflows