EDI 852 Field Extraction with Python

An EDI 852 Product Activity file looks deceptively like a delimited text dump — asterisks between fields, tildes between segments — and that appearance is exactly what trips up a first-pass parser. The delimiters are not fixed by the standard; they are declared by the interchange itself, in the fixed-width ISA header, and a file that arrives from a different trading partner may use a different element separator, sub-element separator, or segment terminator. Hard-code split("*") and split("~") and the parser works on your test fixture and silently mangles the next distributor’s feed. Extracting fields from an 852 correctly means reading the delimiters out of the envelope first, then walking the segment stream to pair each LIN item with the QTY and DTM segments that describe its activity. This recipe does exactly that in pure Python, and maps the result to normalized records ready for reconciliation. It implements the field-extraction step of the parent reference, EDI 852 & 846 Parsing Pipelines.

The 852 answers “what moved?” — units received, dispensed, on-hand, and backordered over a reporting period — one product at a time. The extraction job is to turn each product’s segment group into a typed record whose NDC, activity qualifier, and quantity are unambiguous, so a later reconciliation pass can compare reported movement against physical counts without re-parsing raw X12.

Problem framing

The EDI 852 & 846 parsing pipeline has to accept 852 files from many trading partners, each of which composes a valid X12 interchange but may not use the same punctuation. The extraction layer therefore cannot assume delimiters; it must derive them from the ISA header, which is the one segment the standard fixes to a known width and layout. In a controlled-substance context the stakes are concrete: a mis-split segment can drop a digit from a quantity or misalign an NDC, and a wrong NDC or quantity in the perpetual record is a recordkeeping defect under 21 CFR § 1304.11. Correct extraction is the difference between a reconcilable movement record and a silently corrupted one.

The ISA segment is exactly 106 characters wide and self-describing. The element separator is the character immediately after the segment tag — position index 3, right after ISA — the sub-element (component) separator sits at index 104, and the segment terminator is the final character at index 105. Read those three bytes and the rest of the interchange can be split deterministically regardless of which partner sent it.

Prerequisites & environment

• Python 3.11+ — the parser uses dataclass(frozen=True), enum.Enum, and modern typing; no third-party library is required for extraction.
• Standard library only. hashlib for the audit hash, logging for the PHI-free audit line, datetime for timestamps. Add pytest for the test block.
• A working grasp of the X12 852 envelope. ISA/GS/ST open the interchange, functional group, and transaction set; LIN/QTY/DTM carry item, quantity, and date detail; SE/GE/IEA close them. The activity qualifier in QTY01 tells you what the quantity means.
• NDC normalization downstream. The LIN03 product identifier extracted here is an NDC that must be normalized to its canonical 11-digit form before it can serve as a ledger key; the deterministic, ReDoS-safe way to do that is covered in NDC parsing regex patterns for Python.

Implementation: delimiter-aware pure-Python extraction

The parser reads the three delimiters from the ISA header, splits the interchange into segments, then walks them with a small state loop that pairs each LIN with the QTY and DTM segments that follow it until the next LIN or the SE trailer. Each completed item becomes a frozen, typed record carrying a SHA-256 audit hash.

from __future__ import annotations

import hashlib
import json
import logging
from dataclasses import dataclass
from datetime import datetime, timezone

logger = logging.getLogger("pharmacy.edi.852")

# QTY01 activity qualifiers relevant to 852 product activity.
ACTIVITY = {"17": "received", "33": "on_hand", "38": "dispensed", "QB": "backorder"}


@dataclass(frozen=True)
class Delimiters:
    element: str
    component: str
    segment: str

    @classmethod
    def from_isa(cls, raw: str) -> "Delimiters":
        # ISA is a fixed 106-char segment. The element separator is the byte
        # right after the "ISA" tag; the component separator is ISA16 at
        # index 104; the segment terminator is the final byte at index 105.
        if len(raw) < 106 or not raw.startswith("ISA"):
            raise ValueError("Interchange does not begin with a 106-char ISA header")
        return cls(element=raw[3], component=raw[104], segment=raw[105])


@dataclass(frozen=True)
class ProductActivity:
    ndc: str                 # raw LIN03 identifier (normalize downstream)
    activity: str            # decoded QTY01 label
    qualifier: str           # raw QTY01 code
    quantity: float          # QTY02
    period_or_exp: str | None
    transaction_set: str     # ST01, expected "852"
    set_control: str         # ST02
    audit_hash: str

    @staticmethod
    def _hash(ndc: str, qualifier: str, quantity: float, st02: str, ts: str) -> str:
        material = f"{st02}|{ndc}|{qualifier}|{quantity}|{ts}"
        return hashlib.sha256(material.encode("utf-8")).hexdigest()


def extract_852(raw_x12: str) -> list[ProductActivity]:
    """Extract product-activity records from one 852 interchange.

    Delimiters are read from the ISA header, never assumed, so the parser is
    correct across trading partners that punctuate differently."""
    delims = Delimiters.from_isa(raw_x12)
    segments = [s for s in raw_x12.split(delims.segment) if s.strip()]

    records: list[ProductActivity] = []
    st01 = st02 = None
    pending_ndc: str | None = None
    pending_date: str | None = None

    for seg in segments:
        fields = seg.split(delims.element)
        tag = fields[0].strip()

        if tag == "ST":
            st01 = fields[1] if len(fields) > 1 else None
            st02 = fields[2] if len(fields) > 2 else None
        elif tag == "LIN":
            # LIN*<n>*N4*<ndc> — the N4 qualifier marks the NDC element.
            pending_ndc = (
                fields[3] if len(fields) > 3 and fields[2].strip() == "N4" else None
            )
            pending_date = None
        elif tag == "DTM" and len(fields) > 2:
            pending_date = fields[2]
        elif tag == "QTY" and len(fields) > 2 and pending_ndc:
            qualifier = fields[1].strip()
            # QTY02 may carry a component separator for unit-of-measure; take
            # the leading numeric component only.
            quantity = float(fields[2].split(delims.component)[0])
            ts = datetime.now(timezone.utc).isoformat()
            rec = ProductActivity(
                ndc=pending_ndc,
                activity=ACTIVITY.get(qualifier, "unknown"),
                qualifier=qualifier,
                quantity=quantity,
                period_or_exp=pending_date,
                transaction_set=st01 or "",
                set_control=st02 or "",
                audit_hash=ProductActivity._hash(
                    pending_ndc, qualifier, quantity, st02 or "", ts
                ),
            )
            records.append(rec)
            # PHI-free audit line: identifiers and hash only, no patient data.
            logger.info(json.dumps({
                "event": "edi_852_item",
                "ndc": rec.ndc, "activity": rec.activity,
                "qty": rec.quantity, "st02": rec.set_control,
                "audit_hash": rec.audit_hash,
            }, sort_keys=True))

    return records


if __name__ == "__main__":
    sample = (
        "ISA*00*          *00*          *ZZ*SENDER         *ZZ*RECEIVER       "
        "*260716*1403*U*00401*000000001*0*P*:~"
        "GS*PD*SENDER*RECEIVER*20260716*1403*1*X*004010~"
        "ST*852*0001~"
        "LIN*1*N4*00093721410~QTY*33*144~DTM*090*20261130~"
        "LIN*2*N4*00071015523~QTY*38*12~"
        "SE*7*0001~GE*1*1~IEA*1*000000001~"
    )
    for r in extract_852(sample):
        print(r.ndc, r.activity, r.quantity)

The LIN/QTY pairing is deliberately stateful: a LIN establishes the current product, any DTM refines its date, and each QTY under that product emits one record. Because delimiters come from the ISA header rather than a hard-coded constant, the same function correctly parses a partner who uses | for elements and \n for segments without a single code change.

Verification & testing

The tests prove that delimiters are honored from the header, that the LIN/QTY pairing produces one record per activity line, and that a quantity carrying a unit-of-measure component is not corrupted.

import pytest


def test_delimiters_read_from_header_not_assumed():
    pipe = (
        "ISA|00|          |00|          |ZZ|SENDER         |ZZ|RECEIVER       "
        "|260716|1403|U|00401|000000001|0|P|:^"
        "ST|852|0007^LIN|1|N4|00093721410^QTY|33|144^SE|3|0007^"
    )
    recs = extract_852(pipe)
    assert len(recs) == 1
    assert recs[0].ndc == "00093721410"
    assert recs[0].activity == "on_hand"
    assert recs[0].quantity == 144.0


def test_one_record_per_qty_line():
    raw = (
        "ISA*00*          *00*          *ZZ*S              *ZZ*R              "
        "*260716*1403*U*00401*000000001*0*P*:~"
        "ST*852*0001~LIN*1*N4*00093721410~QTY*17*50~QTY*38*8~SE*4*0001~"
    )
    recs = extract_852(raw)
    assert [r.activity for r in recs] == ["received", "dispensed"]
    assert [r.quantity for r in recs] == [50.0, 8.0]


def test_component_separator_does_not_corrupt_quantity():
    raw = (
        "ISA*00*          *00*          *ZZ*S              *ZZ*R              "
        "*260716*1403*U*00401*000000001*0*P*:~"
        "ST*852*0001~LIN*1*N4*00093721410~QTY*33*144:EA~SE*3*0001~"
    )
    recs = extract_852(raw)
    assert recs[0].quantity == 144.0     # "EA" unit-of-measure component dropped

A captured audit line shows the extraction trail — NDC, decoded activity, quantity, and hash — with no patient identifier:

INFO pharmacy.edi.852 {"activity": "on_hand", "audit_hash": "b71d...9c02", \
     "event": "edi_852_item", "ndc": "00093721410", "qty": 144.0, "st02": "0001"}

Re-hashing a stored record and comparing it to its persisted audit_hash is the tamper check: a mismatch proves the extracted movement was altered after parsing, the signal an audit under 21 CFR § 1304.11 is meant to surface.

Gotchas & compliance pitfalls

• Never assume the delimiters. The element separator, component separator, and segment terminator are declared in the ISA header at fixed positions and vary by trading partner. Hard-coding * and ~ parses your fixture and corrupts the next partner’s feed. Read ISA[3], ISA[104], and ISA[105] and split with those.
• Segment-terminator variants. Some partners append a carriage-return or newline after the terminator for readability. Stripping empty segments after the split (as above) tolerates the extra whitespace; failing to strip yields phantom empty segments that break tag dispatch.
• Repeated LIN loops share DTM context. A DTM applies to the current LIN until the next LIN resets it. Forgetting to clear pending_date on a new LIN bleeds one product’s expiration onto the next, misattributing lot dates. Reset per-item state on every LIN.
• Field-width and unit-of-measure suffixes. QTY02 can carry a unit-of-measure component after the component separator (144:EA). Parsing the whole element as a float raises or truncates; take the leading numeric component only. Likewise, never coerce quantities through an integer type that could drop a leading zero or a fractional pack size.
• LIN03 is an NDC, not a ledger key yet. The extracted identifier still has to be normalized to canonical NDC-11 before it can key the perpetual record, or two segment layouts of the same product will fork into two ledger identities. Route it through the NDC parsing recipe before commitment.

Frequently Asked Questions

Related

• Up: EDI 852 & 846 Parsing Pipelines — the parent reference where this extraction step feeds the audit ledger.
• EDI 852 vs 846: Reconciliation Differences — how the extracted 852 activity reconciles against 846 availability.
• NDC parsing regex patterns for Python — normalizing the extracted LIN03 NDC to a canonical ledger key.
• Data Ingestion & Inventory Sync Workflows — the ingestion architecture this parser serves.