Dynamic Threshold Mapping for Multi-Product Pallets

In modern pharmaceutical logistics a single outbound pallet routinely consolidates temperature-sensitive biologics (2–8 °C), lyophilised vaccines (≤ −20 °C), and controlled room temperature (CRT) formulations (15–25 °C) onto one load. Applying a uniform temperature envelope across that mix creates operational bottlenecks, forces unnecessary product quarantines, and opens compliance gaps — a CRT reading of 22 °C is perfectly compliant for its own SKU yet would trip a single 2–8 °C limit applied to the whole pallet. Dynamic threshold mapping eliminates that constraint by binding each sensor to the SKU-specific temperature envelope, validated packaging configuration, and regulatory tolerance of the unit it actually measures. Implemented correctly, this is the detection layer that resolves limits for the broader Temperature Excursion Detection & Automated Rule Engines platform. Its single hard regulatory anchor is FDA 21 CFR Part 11 §11.10(e): the threshold matrix that produced any disposition must survive as a secure, time-stamped, tamper-evident record that an investigator cannot retroactively alter.

Problem Statement: Why One Pallet Needs Many Limits

A uniform envelope is the path of least engineering resistance, and it fails in three distinct ways that surface as audit findings rather than mere inconvenience:

• Proportionality is lost. ICH Q9 quality risk management expects corrective action to scale with quantified risk. Forcing the tightest SKU’s band onto the whole pallet means a benign CRT reading triggers the same heavyweight response as a genuine breach of a fragile biologic, generating CAPA noise that buries real signals.
• False quarantines erode supply. Holding an entire mixed load because one product’s limit was crossed condemns stable inventory that was never thermally stressed, inflating handling risk and waste — and each re-handling is itself a documented thermal event.
• Retrospective limit edits are indefensible. If thresholds live as editable rules rather than a sealed matrix, an investigator can never prove which limit was active at the moment of an excursion. §11.10(e) and EU GMP Annex 11 both treat that ambiguity as a data-integrity defect.

Dynamic mapping satisfies these mandates by decoupling physical sensors from hard-coded rules and replacing them with version-controlled, product-bound lookup matrices that are frozen at pallet seal. Downstream of the resolved band, deviations are weighted by the duration-based excursion scoring model, and spurious single-probe spikes are filtered by Multi-Sensor Correlation to Reduce False Positives before they ever reach disposition.

Concept & Specification: The Sealed Pallet Manifest

The detection workflow begins with master-data synchronisation. Validated temperature envelopes are maintained in a centralised, audit-controlled repository; each SKU profile carries a nominal range, a tolerance buffer, time-weighted excursion allowances, and a version hash with an effective date. The source stability data for those envelopes is established upstream by Establishing Temperature Excursion Thresholds by Product.

During palletisation, the Warehouse Execution System (WES) or Transport Management System (TMS) emits a structured pallet manifest. This manifest is the binding contract between the physical configuration and the digital rules — it maps every sensor to a SKU and pins the exact profile version that governs that sensor for the life of the shipment:

{
  "pallet_id": "PLT-8842-X",
  "seal_timestamp": "2024-06-14T08:32:11Z",
  "sensor_topology": [
    {"sensor_id": "SN-01A", "position": "top_center", "sku": "BIO-774"},
    {"sensor_id": "SN-01B", "position": "mid_left", "sku": "CRT-112"},
    {"sensor_id": "SN-01C", "position": "base_right", "sku": "VAC-990"}
  ],
  "profile_versions": {
    "BIO-774": "v3.1.0",
    "CRT-112": "v2.4.0",
    "VAC-990": "v4.0.0"
  }
}

The rule engine ingests this payload over a mutually authenticated channel, resolves each SKU to its threshold profile, validates the version hash against the master repository, and constructs an in-memory lookup table that is pushed to edge gateways or cloud nodes before the pallet leaves the staging area. The persisted manifest-and-threshold record carries the following fields; the Regulatory anchor column states why each must exist for the record to be inspection-ready.

Binding matrix_hash to every downstream evaluation is what makes the limits non-repudiable: a reviewer can prove which exact set of bands governed any reading, and any post-seal edit changes the hash and breaks the chain.

Architecture Diagram: Resolution and Real-Time Evaluation

For every inbound telemetry packet the rule engine performs five chained checks. The manifest seals the threshold matrix at pack time — no post-hoc edits — and the engine resolves each sensor_id to its SKU-bound profile before any threshold comparison runs.

Once telemetry begins streaming (typically at 1–5 minute intervals) the detection layer performs continuous boundary resolution for each incoming reading:

1. Coordinate resolution. Match sensor_id to its assigned SKU and active threshold profile from the sealed matrix.
2. Tolerance evaluation. Compare the reading against the nominal range ± buffer. If in-band, log as compliant and advance.
3. Excursion classification. If out of band, classify as WARNING (still within the time allowance) or CRITICAL (allowance exhausted or an absolute limit breached).
4. Temporal aggregation. Accumulate excursion duration with sliding-window logic, feeding the duration-based excursion scoring engine so disposition is risk-weighted rather than binary.
5. Spatial validation. Cross-reference adjacent sensors to rule out localised thermal artefacts via Multi-Sensor Correlation to Reduce False Positives, preventing alert fatigue.

Accurate dt intervals depend on the telemetry already being clock-aligned; that normalisation is owned upstream by Time-Series Alignment for Multi-Zone Cold Storage, and a mutually authenticated source is enforced by Designing Secure IoT Gateways for Pharma Logistics before any reading is trusted.

Production Python Implementation

The module below is a complete, runnable resolution-and-evaluation engine. It loads a manifest, fingerprints and seals the threshold matrix, refuses to evaluate against an unverified profile version, classifies each reading against its SKU-bound band, and emits an append-only hash-chained audit record. Every block cites the clause it satisfies.

from __future__ import annotations

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

logger = logging.getLogger("excursion.threshold_mapping")


class Disposition(str, Enum):
    COMPLIANT = "COMPLIANT"
    WARNING = "WARNING"
    CRITICAL = "CRITICAL"


class ManifestError(ValueError):
    """Raised when a manifest cannot be sealed against the master registry."""


@dataclass(frozen=True)
class ThresholdProfile:
    # A validated, version-pinned storage envelope for one SKU. The object is
    # immutable so a sealed limit can never be edited in place — §11.10(e).
    sku: str
    version: str
    spec_low_c: Decimal
    spec_high_c: Decimal
    tolerance_c: Decimal
    allowance_minutes: Decimal

    def deviation(self, temp_c: Decimal) -> Decimal:
        # Signed distance OUTSIDE nominal ± tolerance; in-band == 0. This is the
        # deterministic sensor-to-limit binding required by §11.10(a) validation.
        upper = self.spec_high_c + self.tolerance_c
        lower = self.spec_low_c - self.tolerance_c
        if temp_c > upper:
            return temp_c - upper
        if temp_c < lower:
            return lower - temp_c
        return Decimal("0")


@dataclass
class SealedMatrix:
    pallet_id: str
    seal_timestamp: str
    sensor_to_sku: dict[str, str]
    profiles: dict[str, ThresholdProfile]
    matrix_hash: str = field(default="")

    def fingerprint(self) -> str:
        # Bind every limit, version and binding into one digest so a reviewer can
        # prove which matrix governed a reading — §11.10(k) change control.
        payload = {
            "pallet_id": self.pallet_id,
            "seal_timestamp": self.seal_timestamp,
            "sensor_to_sku": self.sensor_to_sku,
            "profiles": {
                sku: {
                    "version": p.version,
                    "spec_low_c": str(p.spec_low_c),
                    "spec_high_c": str(p.spec_high_c),
                    "tolerance_c": str(p.tolerance_c),
                    "allowance_minutes": str(p.allowance_minutes),
                }
                for sku, p in sorted(self.profiles.items())
            },
        }
        canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def seal_manifest(manifest: dict, registry: dict[str, ThresholdProfile]) -> SealedMatrix:
    """Resolve a manifest to a sealed, hash-fingerprinted threshold matrix.

    `registry` is keyed by "<sku>@<version>" so the pinned version in the
    manifest is the only profile that can be used — a newer cached profile is
    never silently substituted.
    """
    sensor_to_sku: dict[str, str] = {}
    profiles: dict[str, ThresholdProfile] = {}

    for entry in manifest["sensor_topology"]:
        sku = entry["sku"]
        sensor_to_sku[entry["sensor_id"]] = sku
        version = manifest["profile_versions"].get(sku)
        if version is None:
            # A sensor with no pinned version cannot be evaluated against a
            # defensible limit — fail closed rather than guess (§11.10(a)).
            raise ManifestError(f"no profile version pinned for SKU {sku}")
        key = f"{sku}@{version}"
        if key not in registry:
            # Validating the pinned version against the master repository is the
            # change-control checkpoint required by §11.10(k).
            raise ManifestError(f"profile {key} not found in validated registry")
        profiles[sku] = registry[key]

    matrix = SealedMatrix(
        pallet_id=manifest["pallet_id"],
        seal_timestamp=manifest["seal_timestamp"],
        sensor_to_sku=sensor_to_sku,
        profiles=profiles,
    )
    matrix.matrix_hash = matrix.fingerprint()
    logger.info("sealed matrix %s -> %s", matrix.pallet_id, matrix.matrix_hash[:12])
    return matrix


def _q(value: Decimal, places: int = 4) -> Decimal:
    # Deterministic half-up rounding so a replayed evaluation is bit-identical,
    # avoiding IEEE-754 drift in an audit-critical value (§11.10(b)).
    return value.quantize(Decimal(1).scaleb(-places), rounding=ROUND_HALF_UP)


def evaluate_reading(
    matrix: SealedMatrix,
    sensor_id: str,
    value_c: Decimal,
    ts: str,
    cumulative_minutes: Decimal,
    prev_hash: str = "0" * 64,
) -> dict:
    """Classify one reading against its SKU-bound band and emit an audit record.

    `cumulative_minutes` is the in-excursion time accrued so far for this sensor
    (supplied by the caller's sliding window). `ts` must be ISO 8601 UTC.
    """
    sku = matrix.sensor_to_sku.get(sensor_id)
    if sku is None:
        # An unmapped sensor falls back to the MOST RESTRICTIVE band on the
        # pallet and forces manual review — never silently dropped (ICH Q9).
        sku = min(matrix.profiles, key=lambda s: matrix.profiles[s].spec_high_c)
        logger.warning("unmapped sensor %s -> restrictive fallback %s", sensor_id, sku)

    profile = matrix.profiles[sku]
    deviation = profile.deviation(value_c)

    if deviation == 0:
        disposition = Disposition.COMPLIANT
    elif cumulative_minutes <= profile.allowance_minutes:
        disposition = Disposition.WARNING
    else:
        disposition = Disposition.CRITICAL

    record = {
        "pallet_id": matrix.pallet_id,
        "matrix_hash": matrix.matrix_hash,        # §11.10(e) which limits applied
        "sensor_id": sensor_id,
        "sku": sku,
        "profile_version": profile.version,
        "value_c": str(_q(value_c)),
        "deviation_c": str(_q(deviation)),
        "cumulative_minutes": str(_q(cumulative_minutes)),
        "disposition": disposition.value,
        "reading_ts": ts,
        # Contemporaneous, timezone-aware UTC stamp for the evaluation — §11.10(e).
        "evaluated_at": datetime.now(timezone.utc).isoformat(),
        "prev_hash": prev_hash,
    }
    # Append-only hash chain: each record commits to its predecessor, so any
    # retrospective edit breaks the chain — tamper-evidence for §11.10(e).
    canonical = json.dumps(record, sort_keys=True, separators=(",", ":"))
    record["record_hash"] = hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    logger.info("%s/%s %s -> %s", matrix.pallet_id, sensor_id, record["value_c"],
                disposition.value)
    return record

For sub-second lookups when thousands of pallets are in flight, the sealed matrix is kept hot in memory rather than read from a database per reading — the warm-up and invalidation strategy is covered by Cache-Warming Strategies for Real-Time Rule Engines. When the evaluation pipeline runs as an async stream processor, use asyncio with aiokafka; avoid the synchronous kafka-python client inside an event loop, because every consumer/producer call blocks the loop and creates unbounded latency under load.

Configuration & Deployment Parameters

Behaviour is driven entirely by configuration so the decision logic stays version-controlled and re-validatable. Treat the parameter set as a controlled document under your ICH Q10 pharmaceutical quality system: any change to limits, tolerances, or fallback policy is a change-management event that may require re-validation.

Pre-compile each sealed matrix into a hash map keyed by sensor_id rather than querying a relational store per telemetry event, and cache profiles on gateways with TTL-based invalidation so transit continues during a network partition without compromising compliance. The manifest’s profile_versions object is the source of truth for which version was active during each reading — never let a stale-but-newer cached profile be substituted for a shipment that was sealed against an older one. Rotate the broker’s mTLS client certificates on a fixed schedule and fail closed if a certificate is expired; an unauthenticated telemetry source must never be able to write into the regulated record stream.

Verification & Testing

A mapping engine that gates disposition is GxP-relevant software, so its tests are validation evidence, not a developer convenience. Build the Operational Qualification suite around the edge cases an inspector will probe:

• Resolution tests. Assert that each sensor_id resolves to the exact SKU and pinned profile_version from the manifest, and that a SKU missing from profile_versions raises ManifestError rather than evaluating against a guessed band.
• Version-pinning tests. Place a newer profile in the registry and assert the engine still uses the manifest’s pinned version, proving the change-control checkpoint of §11.10(k).
• Boundary tests. Assert the exact COMPLIANT → WARNING transition at the tolerance edge and WARNING → CRITICAL at the allowance edge, including the equality case, so routing under ICH Q9 is deterministic.
• Fallback tests. Feed an unmapped sensor_id and assert it maps to the most restrictive band and flags for manual review, never a silent drop.
• Audit-chain tests. Mutate one stored field and assert the recomputed record_hash no longer matches the successor’s prev_hash, demonstrating tamper-evidence for §11.10(e).
• CSV protocol hook. Expose a fixture loader that reads an OQ test-vector CSV (sensor_id, value_c, cumulative_minutes, expected_disposition) so qualification can be executed and signed against documented expected outputs.

from decimal import Decimal


def _registry():
    return {
        "BIO-774@v3.1.0": ThresholdProfile(
            "BIO-774", "v3.1.0", Decimal("2.0"), Decimal("8.0"),
            Decimal("0.5"), Decimal("120"),
        ),
        "CRT-112@v2.4.0": ThresholdProfile(
            "CRT-112", "v2.4.0", Decimal("15.0"), Decimal("25.0"),
            Decimal("1.0"), Decimal("240"),
        ),
    }


_MANIFEST = {
    "pallet_id": "PLT-TEST-1",
    "seal_timestamp": "2026-06-28T08:00:00Z",
    "sensor_topology": [
        {"sensor_id": "SN-01A", "position": "top_center", "sku": "BIO-774"},
        {"sensor_id": "SN-01B", "position": "mid_left", "sku": "CRT-112"},
    ],
    "profile_versions": {"BIO-774": "v3.1.0", "CRT-112": "v2.4.0"},
}


def test_pinned_version_is_not_overridden():
    reg = _registry()
    # A newer profile must NOT be substituted for the sealed version (§11.10(k)).
    reg["BIO-774@v4.0.0"] = ThresholdProfile(
        "BIO-774", "v4.0.0", Decimal("0.0"), Decimal("10.0"),
        Decimal("0.0"), Decimal("0"),
    )
    matrix = seal_manifest(_MANIFEST, reg)
    assert matrix.profiles["BIO-774"].version == "v3.1.0"


def test_critical_when_allowance_exhausted():
    matrix = seal_manifest(_MANIFEST, _registry())
    rec = evaluate_reading(
        matrix, "SN-01A", Decimal("11.0"), "2026-06-28T10:00:00Z",
        cumulative_minutes=Decimal("121"),  # 1 minute past the 120 allowance
    )
    # Allowance exhausted -> CRITICAL, the proportional ICH Q9 escalation.
    assert rec["disposition"] == "CRITICAL"


def test_unmapped_sensor_uses_restrictive_band():
    matrix = seal_manifest(_MANIFEST, _registry())
    rec = evaluate_reading(
        matrix, "SN-99Z", Decimal("12.0"), "2026-06-28T10:00:00Z",
        cumulative_minutes=Decimal("0"),
    )
    # Falls back to the tightest band (BIO-774, ≤8 °C) -> not COMPLIANT.
    assert rec["sku"] == "BIO-774"
    assert rec["disposition"] != "COMPLIANT"

Known Failure Modes & Mitigations

When a sensor’s SKU binding cannot be resolved, applying the most restrictive band with a manual-review flag is defensible because it never under-protects product; suspending monitoring is not, because a coverage gap is itself an audit finding.

Compliance FAQ

Related

• Duration-Based Scoring for Temperature Excursions — weights the deviations this engine resolves into a proportional risk score.
• Multi-Sensor Correlation to Reduce False Positives — filters spurious single-probe spikes before disposition.
• Cache-Warming Strategies for Real-Time Rule Engines — keeps the sealed matrix hot for sub-second lookups.
• Establishing Temperature Excursion Thresholds by Product — the stability-data source for each SKU envelope.
• Time-Series Alignment for Multi-Zone Cold Storage — the UTC alignment that makes duration aggregation trustworthy.

For architectural context, this page sits under Temperature Excursion Detection & Automated Rule Engines.