Automating POS Data Extraction for CPG

Delayed or malformed point-of-sale (POS) data is the single most common root cause of accrual disputes in CPG trade reconciliation: when a retailer’s sell-through feed arrives late, splits a segment across a line break, or reports volume in cases where the agreement prices in consumer units, the rebate engine silently posts the wrong accrual and the variance only surfaces at quarter-end. This page documents the exact procedure for automating retailer POS extraction into a deterministic, reconcilable, audit-ready dataset — idempotent polling with checksum validation, envelope-aware EDI 852/867 parsing, decimal-exact normalization, and tiered error routing — so fragmented SFTP drops and EDI gateways become a single trustworthy source for promotional spend and sell-through volume. It is the implementation-level companion to the POS & ERP sync patterns cluster, which frames where extracted feeds meet the general ledger inside the broader data ingestion normalization pipelines discipline.

Automated POS extraction data flow for CPG reconciliation A left-to-right pipeline. Retailer channels (SFTP drop, REST endpoint, EDI 852/867 gateway) feed an idempotent poller that computes a SHA-256 digest and claims it with SET NX EX; re-dropped files matching an existing digest branch down and are dropped silently into quarantine. Verified payloads fork into two parser lanes: an EDI 852/867 parser that validates the ISA/GS/ST envelope and splits on the tilde segment terminator, and a CSV dialect reader that sniffs the delimiter, decodes with utf-8-sig, and streams multi-GB files. Both lanes merge into a Normalize stage that casts money to decimal.Decimal with ROUND_HALF_UP and dates to ISO 8601, then into a Map stage that resolves SKUs and promo codes through an effective-dated lookup and converts every quantity to a single base unit of measure. Clean canonical records flow to the reconciliation engine. Rejected rows from the parser and normalize stages drop into a tiered error router that fans out by class: structural errors to quarantine.etl, referential errors to workbench.vendor, and business errors to hold.finance, each carrying lineage from source file to row hash to error code. EDI CSV Retailer channels SFTP drop REST endpoint EDI 852 / 867 gateway 1 Idempotent poller SHA-256 digest claim · SET NX EX duplicate digest Quarantine re-drops dropped silently 2 EDI 852 / 867 parser ISA / GS / ST envelope split on ~ terminator 3 CSV dialect reader sniff delimiter · utf-8-sig stream multi-GB files 4 Normalize Decimal · ROUND_HALF_UP dates → ISO 8601 5 Map to canonical effective-dated lookup convert to base UOM Reconciliation engine clean canonical records rejected rows 6 Tiered error router classify by error type file → row hash → code structural → quarantine.etl referential → workbench.vendor business → hold.finance

Prerequisites

Before automating extraction, confirm the following are in place:

  • A retailer data contract per source. Document the transport (SFTP, REST, or EDI gateway), file cadence, transaction set (EDI 852 Product Activity, EDI 867 Product Transfer/Resale, or a named CSV layout), encoding, and the reporting unit of measure for every retailer you onboard.
  • A version-pinned mapping table. Retailer SKUs, promotional identifiers, and store IDs must resolve against master data through field mapping strategies with effective dating; the downstream parsing in CSV & EDI parsing workflows consumes canonical codes, never raw retailer terminology.
  • A quarantine and dedup store. A low-latency store (Redis or a Postgres unique index) to hold the SHA-256 ingest index, plus an object store or table for quarantined raw payloads.
  • Python packages: paramiko>=3.4 or asyncssh>=2.14 for SFTP, pydantic>=2.6 for validation, polars>=0.20 or duckdb for streaming large files, and the standard-library decimal, csv, and hashlib modules. Every monetary field uses decimal.Decimal; never float.
  • Access role: read access to each retailer endpoint and write access to the quarantine, dedup store, and canonical staging tables (typically the reconciliation_etl service role).

Step-by-step implementation

Step 1 — Poll idempotently and checksum every payload

Retailers distribute POS files through fragmented channels and frequently re-drop the same file. Treat retrieval as idempotent: compute a SHA-256 over the raw bytes, and skip any payload whose digest has already been ingested. Use key-based SFTP auth with connection pooling so peak reporting windows do not exhaust sockets.

python
import hashlib

def ingest_digest(raw: bytes) -> str:
    return hashlib.sha256(raw).hexdigest()

def claim_payload(redis, digest: str, ttl: int = 7 * 86_400) -> bool:
    # True only the first time this exact file is seen
    return bool(redis.set(f"pos:ingest:{digest}", "1", nx=True, ex=ttl))

def retrieve(sftp, remote_path: str, redis) -> bytes | None:
    raw = sftp.getfo(remote_path)
    if not claim_payload(redis, ingest_digest(raw)):
        return None          # already ingested, drop silently
    return raw

Validation check: retrieve the same file twice and assert the second call returns None with no row written downstream. A re-dropped POS file must never produce a second accrual run.

Step 2 — Validate the transport envelope before parsing line items

For EDI, validate the ISA/GS/ST envelope before extracting any line item, and split on the segment terminator (~), not newlines — X12 segments end with ~, and a feed that packs multiple segments onto one line (or wraps one across several) will silently misparse if you split on \n.

python
def split_segments(edi_text: str, terminator: str = "~") -> list[str]:
    return [s.strip() for s in edi_text.split(terminator) if s.strip()]

REQUIRED_ENVELOPE = ("ISA", "GS", "ST")

def validate_envelope(segments: list[str]) -> None:
    seen = {seg.split("*")[0] for seg in segments}
    missing = [tag for tag in REQUIRED_ENVELOPE if tag not in seen]
    if missing:
        raise StructuralError(f"missing EDI envelope segments: {missing}")

Adhere to the ANSI ASC X12 transaction sets for delimiter positioning. Validation check: feed a fixture whose segments are joined with ~ on a single physical line and assert every line item is recovered; feed one with a missing GS and assert StructuralError.

Step 3 — Parse CSV variants with a dialect-aware streaming reader

For CSV feeds, auto-detect the delimiter, quote character, and encoding (UTF-8 vs CP1252 vs ISO-8859-1), and stream multi-GB daily files rather than loading them into memory. Python’s built-in csv module is a reliable baseline; polars or duckdb carry production volumes.

python
import csv, io

def sniff_dialect(sample: str) -> csv.Dialect:
    return csv.Sniffer().sniff(sample, delimiters=",;\t|")

def read_pos_csv(raw: bytes) -> list[dict]:
    text = raw.decode("utf-8-sig", errors="strict")   # strips BOM
    dialect = sniff_dialect(text[:8192])
    return list(csv.DictReader(io.StringIO(text), dialect=dialect))

Validation check: parse one comma-delimited and one pipe-delimited fixture with a leading BOM and assert both yield identical row counts and no  in the first header key.

Step 4 — Normalize money and dates deterministically

Reconciliation accuracy depends on deterministic casting: dates to ISO 8601, and monetary values to fixed-point decimals via Python’s decimal module. Any floating-point deviation here propagates as variance in rebate calculations and accrual postings. Model the canonical row in Pydantic v2 so a float amount or unparseable date is rejected at the boundary.

python
from datetime import date
from decimal import Decimal, ROUND_HALF_UP
from pydantic import BaseModel, field_validator

class PosLine(BaseModel):
    retailer_sku: str
    store_id: str
    promo_code: str | None
    txn_date: date            # normalized to ISO 8601
    units: Decimal            # base UOM, never cases/pallets here
    extended_amount: Decimal

    @field_validator("units", "extended_amount", mode="before")
    @classmethod
    def to_decimal(cls, v) -> Decimal:
        d = Decimal(str(v))                       # str() avoids float drift
        return d.quantize(Decimal("0.0001"), ROUND_HALF_UP)

Validation check: instantiate PosLine from a raw row where extended_amount is the float 19.99 and assert the stored value equals Decimal("19.9900") exactly, then assert a malformed txn_date raises ValidationError.

Step 5 — Map retailer codes to canonical identifiers in base UOM

Retailer schemas rarely align with internal hierarchies. Resolve each retailer_sku to an internal UPC/GTIN and each promo_code to a Trade Promotion Management campaign ID through the effective-dated mapping table, and convert every quantity to a single base UOM before any tiered rebate logic runs. Where the qualification rules themselves live is owned by the claim validation rule engine — this step only produces canonical, comparable records.

python
def to_base_uom(units: Decimal, uom: str, factors: dict[str, Decimal]) -> Decimal:
    # factors e.g. {"CASE": Decimal("12"), "PALLET": Decimal("960"), "EACH": Decimal("1")}
    return (units * factors[uom]).quantize(Decimal("0.0001"))

def resolve_sku(retailer_sku: str, txn_date: date, lookup) -> str:
    match = lookup.effective(retailer_sku, on=txn_date)
    if match is None or match.confidence < Decimal("0.85"):
        raise ReferentialError(f"unmapped SKU {retailer_sku} @ {txn_date}")
    return match.gtin

Validation check: map a row reported in CASE with a factor of 12 and assert the canonical units is 12× the input; assert an unmapped SKU below the 0.85 confidence threshold raises ReferentialError and is flagged for vendor-manager review rather than silently dropped.

Step 6 — Classify failures into actionable tiers and route each

Automated extraction inevitably hits malformed payloads, missing reference data, and business-rule violations. Classify failures into tiers rather than dumping unstructured logs, and persist every failure with traceable lineage (source file → row hash → error code).

python
class StructuralError(Exception): ...    # missing column, bad encoding, broken envelope
class ReferentialError(Exception): ...   # unmapped SKU, invalid promo, unknown store
class BusinessError(Exception): ...      # negative units, out-of-range discount, dup hash

ROUTES = {
    StructuralError:  "quarantine.etl",        # alert ETL devs immediately
    ReferentialError: "workbench.vendor",      # vendor-manager reconciliation queue
    BusinessError:    "hold.finance",          # trade-finance validation before posting
}

def route_failure(audit, src_file: str, row_hash: str, exc: Exception) -> str:
    queue = ROUTES.get(type(exc), "quarantine.etl")
    audit.write(source=src_file, row_hash=row_hash,
                error_code=type(exc).__name__, queue=queue)
    return queue

Validation check: raise one of each error type and assert each lands on its mapped queue with a complete audit row (source, row_hash, error_code). A schema break must page ETL; an unmapped SKU must reach a vendor manager, not a developer.

Common failure modes and fixes

  1. Splitting EDI on newlines. X12 segments terminate with ~, and many gateways pack several segments onto one physical line. Splitting on \n silently drops or merges line items. Split on the declared segment terminator (Step 2) and validate the ISA/GS/ST envelope before trusting any extracted detail.
  2. Floating-point drift in extended amounts. Casting money through float lets rounding error compound across millions of POS lines and breaks ledger ties. Keep every monetary value in decimal.Decimal, construct it from a str (never a float), and quantize explicitly with ROUND_HALF_UP.
  3. Re-dropped files double-posting. Retailers routinely re-upload the same file after a perceived failure. Without the Step 1 ingest digest, the duplicate triggers a second accrual run. Claim the SHA-256 digest with SET NX EX before parsing and drop redeliveries silently.
  4. Unit-of-measure mismatch. A retailer reporting in cases while the agreement prices per consumer unit inflates accruals by the case factor. Convert every quantity to a single base UOM (Step 5) before any tier logic, and store the original UOM in the audit row for traceability.
  5. BOM and encoding corruption. A leading byte-order mark welds  onto the first header key, so retailer_sku lookups miss and every row is flagged unmapped. Decode with utf-8-sig to strip the BOM and sniff the dialect (Step 3) instead of assuming UTF-8 comma-delimited.

Operational checklist

Frequently asked questions

Why split EDI on ~ instead of line breaks? X12 defines ~ as the segment terminator; line breaks are cosmetic. Gateways may emit one segment per line, many segments per line, or wrap a single segment across lines. Splitting on \n silently misparses any of those layouts, so always split on the declared terminator and validate the ISA/GS/ST envelope first.

How do I stop a re-dropped file from double-counting sell-through? Compute a SHA-256 over the raw bytes and claim that digest atomically before parsing. A redelivery hashes to the same digest, fails the claim, and is dropped — the extraction stays idempotent at the file level regardless of how many times a retailer re-uploads.

What happens when a retailer reports volume in cases but the agreement prices per unit? Convert every quantity to a single base UOM during normalization, before any tier math runs. The conversion factor and original UOM are stored in the audit row so the figure remains reconstructable. The tier arithmetic itself lives in payout structure modeling.

Where do unmapped SKUs and invalid promo codes go? They raise a referential error and route to a vendor-manager reconciliation queue, not the developer quarantine. Persistent gaps are resolved against master data and the originating record is reprocessed; defaulting policy for records that cannot be matched is governed by fallback routing logic.

Back to