Parsing EDI 852 files with Python pandas

Parse EDI 852 Product Activity Data into pandas DataFrames without losing the LIN/QTY/DTM loop hierarchy. Chunked, DEA-auditable ingestion with SHA-256 hashing.

The Problem: Flat Parsing Destroys the 852 Loop Hierarchy

The most common EDI 852 parsing defect is treating the file as flat text. An engineer reaches for pd.read_csv with a ~ line terminator or a naive str.split("*"), gets a rectangular table back, and assumes the job is done. It is not. The X12 852 transaction is a hierarchical document: every QTY and DTM segment belongs to the LIN item loop that immediately precedes it. Flatten the file and you sever that association — a quantity reported for one National Drug Code silently re-attaches to the next item in the stream.

For ordinary retail inventory that is a rounding annoyance. For a controlled-substance pharmacy it is a recordkeeping violation: 21 CFR § 1304.21 requires that every transaction be tied to the exact product identified, and a mis-attributed QTY for a Schedule II item is the kind of drift that surfaces as an unexplained shortage during a DEA audit. This page solves exactly that failure — preserving the LIN → QTY → DTM loop while streaming the file into pandas in bounded memory. It is a focused recipe within the EDI 852 & 846 Parsing Pipelines cluster, which in turn sits under the broader Data Ingestion & Inventory Sync Workflows architecture.

A second, quieter defect rides alongside the first: code that “chunks” by yielding one DataFrame per item. That produces thousands of single-row frames, defeats vectorization, and is slower than not chunking at all. The implementation below fixes both — it groups completed item records into fixed-size batches before yielding.

Loop-aware, chunked EDI 852 data flow: LIN opens a buffer, QTY/DTM fold in, completed items batch into a DataFrame A left-to-right pipeline. The raw 852 file is read line-by-line into a precompiled SEGMENT_RE matcher. Lines that fail the regex branch down to a quarantine queue. Matched lines are dispatched by segment id into a single open item buffer: a LIN segment finalizes the previous item and opens a new one keyed by NDC, while QTY and DTM segments fold their values into the currently open buffer; a QTY or DTM that arrives with no open LIN also branches to quarantine. When the next LIN arrives the finished item is appended to the rows accumulator, and once the row count reaches chunk_size the parser yields one pandas DataFrame and clears the list. Quarantined records are written to quarantine_852.json. stream match finalize ≥ N Raw 852 file read line-by-line ST*852*0001~ LIN**N4*…~ QTY*QA*144~ DTM*007*…~ LIN**N4*…~ @@@ malformed ~ Segment matcher compiled once · ReDoS-safe ^([A-Z]{2,3})\*(.*)~$ seg_id · elements[ ] dispatch by segment id open_item buffer one LIN loop open at a time LIN → opens · ndc finalizes the previous item first QTY → qty_qa · base_qa DTM → dtm_007 (ISO-8601) QTY/DTM fold into the open buffer rows[ ] completed items len ≥ chunk_size yield pd.DataFrame one frame · ≤ N rows then rows.clear() no regex match QTY/DTM · no open LIN quarantine[ ] → quarantine_852.json deferred review — never mis-attributed

Prerequisites & Environment

This recipe targets Python 3.11+ and pandas 2.x. It uses only the standard library plus pandas:

Component Purpose
pandas >= 2.0 Chunked DataFrame assembly and downstream vectorized validation
re Precompiled segment matcher (compile once, reuse per line)
hashlib SHA-256 audit_hash over the raw segment for tamper-evident retention
dataclasses Frozen AuditEvent record — no PHI fields, structured logging only
datetime DTM normalization to ISO-8601 UTC

You should already understand the X12 852 envelope (ISA/GS/ST … SE/GE/IEA) and the item loop (LIN item identification, QTY quantity, DTM date/time). You should also be comfortable with two pharmacy-specific rules that the parser depends on: the canonical-format conversion described in NDC-11 vs NDC-10 Parsing Standards, and the scheduling lookup performed downstream by DEA Schedule II-V Classification Mapping. This page does not re-derive those; it consumes them.

Implementation: Loop-Aware, Chunked 852 Parser

The parser is a generator. It reads the file one line at a time, keeps a single open item buffer, folds each QTY/DTM into the currently-open LIN, and only finalizes an item when the next LIN arrives (or end-of-file is reached). Finalized items accumulate in a rows list; when that list reaches chunk_size, the parser yields a real multi-row DataFrame and clears the list. Malformed segments never raise — they are routed to a quarantine list for deferred handling.

python
from __future__ import annotations

import hashlib
import json
import logging
import re
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterator, Optional

import pandas as pd

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)

# Compile the segment matcher ONCE, not per line. Anchored on both ends so a
# malformed line cannot trigger catastrophic backtracking (ReDoS-safe).
SEGMENT_RE = re.compile(r"^([A-Z]{2,3})\*(.*)~$")

# UOM -> base-unit multiplier. Controlled substances reconcile in base units
# (21 CFR 1304.04 inventory thresholds), so PK/BX/CS must be expanded.
UOM_TO_BASE: dict[str, float] = {"EA": 1.0, "PK": 10.0, "BX": 100.0, "CS": 500.0}


@dataclass(frozen=True)
class AuditEvent:
    """Tamper-evident, PHI-free audit record. Serialized to the log stream."""
    timestamp_utc: str
    segment_id: str
    action: str
    audit_hash: str
    detail: Optional[str] = None


def _emit_audit(segment_id: str, action: str, raw: str, detail: str | None = None) -> None:
    digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()
    event = AuditEvent(
        timestamp_utc=datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
        segment_id=segment_id,
        action=action,
        audit_hash=digest,
        detail=detail,
    )
    logging.info(json.dumps(asdict(event)))


def _normalize_ndc(raw_ndc: str) -> str:
    """Collapse hyphens and reduce to the FDA-canonical 10-digit form.
    Full directional rules live in the NDC-11 vs NDC-10 parsing standard."""
    cleaned = raw_ndc.replace("-", "").strip()
    if len(cleaned) == 11:
        return cleaned[1:] if cleaned.startswith("0") else cleaned[:10]
    if len(cleaned) == 10:
        return cleaned
    raise ValueError(f"non-compliant NDC length: {len(cleaned)}")


def _normalize_dtm(raw_ts: str) -> str:
    """X12 CCYYMMDD or CCYYMMDDHHMM -> ISO-8601 UTC with millisecond precision."""
    fmt = "%Y%m%d" if len(raw_ts) == 8 else "%Y%m%d%H%M"
    dt = datetime.strptime(raw_ts, fmt).replace(tzinfo=timezone.utc)
    return dt.isoformat(timespec="milliseconds")


def parse_edi_852(
    file_path: str,
    chunk_size: int = 5_000,
    quarantine_path: str = "quarantine_852.json",
) -> Iterator[pd.DataFrame]:
    """Yield loop-aware pandas DataFrames from an EDI 852 file.

    Each yielded frame contains up to `chunk_size` fully-assembled item rows;
    every QTY/DTM stays bound to the LIN that opened its loop.
    """
    rows: list[dict] = []
    quarantine: list[dict] = []
    open_item: Optional[dict] = None

    def _finalize() -> None:
        nonlocal open_item
        if open_item is not None:
            rows.append(open_item)
            open_item = None

    _emit_audit("INIT", "FILE_OPENED", file_path, file_path)

    with open(file_path, "r", encoding="utf-8-sig") as fh:
        for line_num, raw in enumerate(fh, start=1):
            line = raw.strip()
            if not line:
                continue

            match = SEGMENT_RE.match(line)
            if match is None:
                quarantine.append({"line": line_num, "raw": line, "error": "MALFORMED_SEGMENT"})
                continue

            seg_id, payload = match.groups()
            elements = payload.split("*")

            if seg_id == "LIN":
                _finalize()  # close the previous item before opening a new one
                try:
                    open_item = {"ndc": _normalize_ndc(elements[2]), "src_line": line_num}
                except (IndexError, ValueError) as exc:
                    quarantine.append({"line": line_num, "raw": line, "error": str(exc)})
                    open_item = None

            elif seg_id == "QTY" and open_item is not None:
                try:
                    qty_type = elements[1].lower()
                    raw_qty = float(elements[2])
                    uom = elements[3] if len(elements) > 3 else "EA"
                    open_item[f"qty_{qty_type}"] = raw_qty
                    open_item[f"base_{qty_type}"] = round(raw_qty * UOM_TO_BASE.get(uom, 1.0), 4)
                    open_item["uom"] = uom
                except (IndexError, ValueError) as exc:
                    quarantine.append({"line": line_num, "raw": line, "error": str(exc)})

            elif seg_id == "DTM" and open_item is not None:
                try:
                    open_item[f"dtm_{elements[1]}"] = _normalize_dtm(elements[2])
                except (IndexError, ValueError) as exc:
                    quarantine.append({"line": line_num, "raw": line, "error": str(exc)})

            if len(rows) >= chunk_size:
                yield pd.DataFrame(rows)
                rows.clear()

    _finalize()
    if rows:
        yield pd.DataFrame(rows)

    if quarantine:
        Path(quarantine_path).write_text(json.dumps(quarantine, indent=2), encoding="utf-8")
        _emit_audit("QUARANTINE", "MALFORMED_RECORDS_ROUTED", quarantine_path,
                    f"{len(quarantine)} records")

The two structural guarantees this code makes are worth stating plainly: an item is never yielded until the parser has seen the next LIN (so its loop is complete), and a QTY/DTM is dropped to quarantine rather than mis-attributed if no LIN is currently open. Those two rules are what keep the pandas output faithful to the source X12 hierarchy.

Verification & Testing

Correctness here is binary: a QTY either stays with its own LIN or it does not. The cheapest way to prove it is a two-item fixture where the second item carries a deliberately distinct quantity, then assert on the assembled DataFrame.

python
import io
import pandas as pd

FIXTURE = (
    "ST*852*0001~\n"
    "LIN**N4*00093721410~\n"   # NDC-11 -> canonical 10-digit
    "QTY*QA*144*EA~\n"          # on-hand 144 each
    "DTM*007*20260628~\n"
    "LIN**N4*00054327199~\n"
    "QTY*QA*12*BX~\n"           # 12 boxes -> 1200 base units
    "DTM*007*20260628~\n"
    "SE*7*0001~\n"
)


def test_loop_isolation(tmp_path):
    p = tmp_path / "sample.edi"
    p.write_text(FIXTURE, encoding="utf-8")

    frame = pd.concat(parse_edi_852(str(p), chunk_size=10), ignore_index=True)

    assert len(frame) == 2
    # Item 1 keeps its 144 EA; item 2 keeps its 12 BX expanded to base units.
    item1 = frame.loc[frame["ndc"] == "0093721410"].iloc[0]
    item2 = frame.loc[frame["ndc"] == "0054327199"].iloc[0]
    assert item1["qty_qa"] == 144.0 and item1["base_qa"] == 144.0
    assert item2["qty_qa"] == 12.0 and item2["base_qa"] == 1200.0

For the compliance dimension, inspect the structured log. Each _emit_audit call writes a single JSON line whose audit_hash is the SHA-256 of the exact bytes it acted on — an inspector can recompute that digest from the retained raw segment and confirm the record was never altered, satisfying the tamper-evident retention expectation behind 21 CFR § 1304.22. A healthy run emits a FILE_OPENED line at the start and, only if anything was rejected, a MALFORMED_RECORDS_ROUTED line at the end:

text
2026-06-28T14:02:11.004+00:00 [INFO] {"timestamp_utc": "2026-06-28T14:02:11.004+00:00", "segment_id": "INIT", "action": "FILE_OPENED", "audit_hash": "9f2c...", "detail": "sample.edi"}

Gotchas & Compliance Pitfalls

  • Ambiguous NDC segment padding. _normalize_ndc strips a single leading zero from an 11-digit code, but the reverse conversion is not self-describing — an 11-digit 5-4-2 string may originate from 4-4-2, 5-3-2, or 5-4-1. Treat the helper here as a fast path and resolve anything that fails a directory cross-reference through the full rules in NDC-11 vs NDC-10 Parsing Standards before it touches the controlled-substance ledger.
  • EDI field-width truncation. Some distributors right-pad QTY values or transmit them with implied decimals. A float() cast hides this; reconcile base units against the Barcode Scan Log Routing Logic dispensing counts and flag deltas above your tolerance rather than trusting the feed.
  • Custom element separators. The * separator and ~ terminator are declared in the ISA segment and are not guaranteed. A robust deployment reads them from ISA rather than hard-coding SEGMENT_RE; assuming */~ is the single most common cause of a file that parses to zero rows.
  • Repeated QTY qualifiers in one loop. If a LIN carries two QTY*QA segments, the later one overwrites the earlier in the dict. Decide deliberately whether to sum, keep-last, or quarantine — silent overwrite is a defect for Schedule II reconciliation.
  • Quarantine is not optional. Routing malformed segments aside only matters if something consumes the queue. Wire quarantine_852.json into the deferred-review path described in Error Handling & Retry Mechanisms; an unread quarantine file is an unreported gap.

Frequently Asked Questions

Why not load the whole 852 with pd.read_csv and a ~ separator?

Because read_csv produces a flat table and the 852 is hierarchical. A QTY row carries no NDC of its own — its product identity comes from the LIN above it. Flattening discards that vertical relationship, so quantities re-attach to the wrong item. The generator above keeps a single open item buffer precisely to preserve the loop.

How do I keep memory bounded on multi-gigabyte distributor files?

The parser never holds the whole file. It reads line-by-line, accumulates at most chunk_size completed item rows, and yields a DataFrame the moment that threshold is reached. Lower chunk_size to cap peak memory; raise it to favor vectorized throughput downstream.

Where should validated chunks go next?

Pass each yielded DataFrame through your compliance gate, then dispatch it concurrently to the pharmacy management system via the Async Batch Processing for Inventory Updates path, which handles back-pressure and idempotent posting so a retried chunk cannot double-post.

Is the audit_hash enough for a DEA inspection?

The hash makes a record tamper-evident, not complete. You must also retain the raw segment, the normalized output, and the timestamp so the digest can be recomputed. Stored together, those fields let an inspector reproduce the SHA-256 and confirm integrity for the two-year window 21 CFR § 1304.22 requires.