Point-in-Time Inventory Reconstruction

Replay an append-only event stream to any timestamp to reconstruct the exact on-hand quantity per NDC, deterministically and idempotently for an inspection

An inspector rarely asks “how much do you have now.” They ask “how much did you have on the morning of March 14,” and a controlled-substance system has to answer that exactly, on demand, for any National Drug Code (NDC). This recipe reconstructs the on-hand quantity per NDC as of an arbitrary timestamp by replaying an append-only event stream up to that instant — deterministically, so two runs agree, and idempotently, so replaying the same events twice never double-counts. It is the computational core of Biennial Inventory Reconciliation, where the “system side” of every count is exactly this as-of balance, and it is the same fold that freezes the figure behind a Form 106 filing.

The problem is easy to underestimate. A running total in a column is trivial to read but impossible to defend: it reflects whatever writes have landed, in whatever order, including corrections applied after the date in question. Reconstructing the balance from immutable events instead makes the answer a pure function of history — and history does not move.

Folding an ordered event timeline up to an as-of timestamp A timeline of ledger events for one NDC runs left to right: receive plus 500, dispense minus 80, waste minus 5, then a dashed as-of cutoff line, after which a later dispense minus 20 lies beyond the cutoff. Events at or before the cutoff fold into a running balance of 415; the event after the cutoff is excluded. +500 receive −80 dispense −5 waste −20 after cutoff as-of T balance = 415 folded ≤ T

Prerequisites & environment

  • Python 3.11+ — uses dataclasses(frozen=True), enum.Enum, and modern type syntax.
  • Standard library only: hashlib, logging, datetime, collections. Add pytest for the test block.
  • Your events must carry three things for a defensible reconstruction: an effective timestamp (when the transaction took effect, not when it was ingested), a monotonic sequence number for stable tie-breaking, and a signed or typed quantity. Without the effective timestamp you cannot answer “as of” questions honestly.
  • Regulatory grounding: 21 CFR § 1304.11 (point-in-time reconstructability of the perpetual inventory) and 21 CFR § 1304.04 (the record must remain re-derivable across the retention horizon). Replaying an event stream during a disconnect follows the same deferred discipline as fallback routing for offline sync.

Implementation: a deterministic, idempotent fold

The reconstruction is a fold: start at zero per NDC, apply each event whose effective time is at or before the cutoff, in a canonical order. Two design choices make it defensible. First, ordering is by (effective_ts, sequence) — never by insertion order — so a late-ingested but early-dated event lands in its correct historical slot. Second, each event carries an idempotency key, and a seen set drops duplicates, so replaying an at-least-once stream twice yields the same balance.

python
from __future__ import annotations

import hashlib
import logging
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum

logger = logging.getLogger("pharmacy.reporting.reconstruct")


class EventType(str, Enum):
    RECEIVE = "RECEIVE"
    DISPENSE = "DISPENSE"
    WASTE = "WASTE"
    TRANSFER_IN = "TRANSFER_IN"
    TRANSFER_OUT = "TRANSFER_OUT"
    LOSS = "LOSS"


_SIGN = {
    EventType.RECEIVE: +1, EventType.TRANSFER_IN: +1,
    EventType.DISPENSE: -1, EventType.WASTE: -1,
    EventType.TRANSFER_OUT: -1, EventType.LOSS: -1,
}


@dataclass(frozen=True)
class Event:
    ndc_11: str
    event_type: EventType
    quantity: int            # positive; sign comes from event_type
    effective_ts: str        # ISO-8601 UTC — WHEN it took effect
    sequence: int            # monotonic tie-breaker

    @property
    def idempotency_key(self) -> str:
        raw = f"{self.ndc_11}|{self.event_type.value}|{self.quantity}|" \
              f"{self.effective_ts}|{self.sequence}"
        return hashlib.sha256(raw.encode("utf-8")).hexdigest()


def _as_utc(ts: str) -> datetime:
    """Parse to an aware UTC datetime so naive/localized inputs cannot skew ordering."""
    dt = datetime.fromisoformat(ts)
    if dt.tzinfo is None:
        raise ValueError(f"Event timestamp {ts!r} is naive; UTC offset required.")
    return dt.astimezone(timezone.utc)


def reconstruct_as_of(events: list[Event], as_of_ts: str) -> dict[str, int]:
    """Reconstruct on-hand quantity per NDC as of `as_of_ts`.

    Deterministic: ordered by (effective_ts, sequence).
    Idempotent: duplicate events (same idempotency key) are applied once.
    Satisfies point-in-time reconstruction under 21 CFR § 1304.11.
    """
    cutoff = _as_utc(as_of_ts)
    ordered = sorted(events, key=lambda e: (_as_utc(e.effective_ts), e.sequence))
    balances: dict[str, int] = defaultdict(int)
    seen: set[str] = set()
    applied = 0
    for ev in ordered:
        if _as_utc(ev.effective_ts) > cutoff:
            continue                       # beyond the as-of horizon
        key = ev.idempotency_key
        if key in seen:
            continue                       # duplicate delivery — apply once
        seen.add(key)
        balances[ev.ndc_11] += _SIGN[ev.event_type] * ev.quantity
        applied += 1
    result = {ndc: qty for ndc, qty in balances.items()}
    # PHI-free structured audit line: counts and horizon only.
    logger.info(
        "reconstruct_as_of ts=%s ndc_count=%d events_applied=%d",
        cutoff.isoformat(), len(result), applied,
    )
    return result


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    events = [
        Event("00093310501", EventType.RECEIVE, 500, "2026-06-01T09:00:00+00:00", 1),
        Event("00093310501", EventType.DISPENSE, 80, "2026-07-01T11:00:00+00:00", 2),
        Event("00093310501", EventType.WASTE, 5, "2026-07-10T08:00:00+00:00", 3),
        # After the as-of cutoff — must be excluded:
        Event("00093310501", EventType.DISPENSE, 20, "2026-07-16T17:00:00+00:00", 4),
    ]
    balance = reconstruct_as_of(events, as_of_ts="2026-07-16T07:00:00+00:00")
    print(balance)     # {"00093310501": 415}

The fold is a pure function of its inputs: same events, same cutoff, same answer, every time and on every machine. That property is what lets a reconstructed balance stand as evidence — it is not a stored number someone could have edited, it is a re-derivation of immutable history.

Verification & testing

The tests must prove three independent properties: the cutoff is respected, duplicate delivery is absorbed, and out-of-order ingestion still lands events in historical order.

python
import pytest

BASE = [
    Event("00093310501", EventType.RECEIVE, 500, "2026-06-01T09:00:00+00:00", 1),
    Event("00093310501", EventType.DISPENSE, 80, "2026-07-01T11:00:00+00:00", 2),
    Event("00093310501", EventType.WASTE, 5, "2026-07-10T08:00:00+00:00", 3),
    Event("00093310501", EventType.DISPENSE, 20, "2026-07-16T17:00:00+00:00", 4),
]


def test_cutoff_excludes_later_events():
    # 500 - 80 - 5 = 415; the 17:00 dispense is after the 07:00 cutoff.
    assert reconstruct_as_of(BASE, "2026-07-16T07:00:00+00:00") == {"00093310501": 415}


def test_replay_is_idempotent():
    doubled = BASE + BASE                       # at-least-once redelivery
    assert reconstruct_as_of(doubled, "2026-07-16T07:00:00+00:00") == {"00093310501": 415}


def test_out_of_order_ingestion_is_deterministic():
    shuffled = [BASE[2], BASE[0], BASE[3], BASE[1]]
    assert reconstruct_as_of(shuffled, "2026-07-16T07:00:00+00:00") == {"00093310501": 415}


def test_naive_timestamp_is_rejected():
    bad = [Event("00093310501", EventType.RECEIVE, 10, "2026-06-01T09:00:00", 9)]
    with pytest.raises(ValueError):
        reconstruct_as_of(bad, "2026-07-16T07:00:00+00:00")

A successful reconstruction emits one structured, PHI-free line that records the horizon and the work done, with no patient data:

text
INFO pharmacy.reporting.reconstruct reconstruct_as_of \
     ts=2026-07-16T07:00:00+00:00 ndc_count=1 events_applied=3

The events_applied=3 (not 4) is the visible proof that the post-cutoff dispense was excluded, and re-running the doubled stream still shows events_applied=3 because the duplicates were absorbed.

Gotchas & compliance pitfalls

  • Order by effective time, never by insertion order. An event ingested late but timestamped early must fold into its historical position. Sorting by (effective_ts, sequence) guarantees it; sorting by arrival silently corrupts every as-of query before the late event.
  • Reject naive timestamps. A timestamp without a UTC offset can be compared inconsistently against an aware cutoff, shifting an event across the boundary by the local offset. Parse to aware UTC and refuse naive input rather than guessing the zone.
  • The cutoff is inclusive at the boundary. Decide explicitly whether an event exactly at T belongs in the balance. For an opening-of-business inventory, events at or before T are included; document the choice because a one-second ambiguity can move a dosage unit.
  • Idempotency needs a stable key. Deriving the key from the full event content (NDC, type, quantity, effective time, sequence) means a genuine duplicate collapses while two legitimately distinct events never do. Keying on a mutable field would drop real events.
  • Monotonic sequence breaks ties, it does not order time. Two events with the same effective timestamp must resolve in a stable order; the sequence does that. Do not use the sequence instead of the timestamp — a high sequence number does not mean a later effective time.
  • A reconstruction is read-only. The fold must never write back a “materialized” balance that a later query then trusts. The whole value is that the answer is re-derived from immutable events; caching it reintroduces the mutable-total problem the technique exists to solve.

Frequently Asked Questions

Why replay events instead of storing a running balance?

A running balance is a single mutable number that reflects whatever writes have landed, including corrections applied after the date you are asking about. Replaying immutable events makes the as-of balance a pure function of history, so it is reproducible and cannot have been silently edited — which is precisely what 21 CFR § 1304.11 point-in-time reconstruction requires.

What makes the reconstruction idempotent?

Each event has a SHA-256 idempotency key derived from its content, and a seen set applies each key once. If an at-least-once event stream delivers the same event twice, the second application is skipped, so the folded balance is identical whether the stream was clean or contained duplicates.

How are out-of-order events handled?

Events are sorted by (effective_ts, sequence) before folding, not by arrival order. An event that was ingested late but occurred early is placed in its correct historical position, so a backfilled or delayed transaction reconstructs the balance as though it had always been there.

Does the timezone of the timestamp matter?

Yes, decisively. All timestamps are parsed to aware UTC before comparison; naive timestamps are rejected. A naive value compared against an aware cutoff could shift an event across the as-of boundary by the local offset, moving dosage units in or out of the reported balance.

Can this run while the site is offline?

Yes. The fold only needs a locally-durable copy of the events up to the horizon, so a disconnected terminal can reconstruct an as-of balance from its local replay and reconcile with the central ledger when connectivity returns, following the same deferred-replay discipline as the offline-sync fallback path.