Hippo pharmacy-claims pipeline

This repository turns pharmacy, claim, and revert files into a durable PostgreSQL event ledger, deterministic canonical state, revisioned analytics, and reproducible CSV/JSON exports. Airflow orchestrates the same application operations exposed by the CLI; it contains no business SQL.

Quick start

Prerequisites are Docker Engine, Docker Compose v2, Make, Python 3, and a POSIX shell (Linux, macOS, or WSL2). No host PostgreSQL, Airflow, or project Python environment is required.

make up
make pipeline
make test
make metrics
make down

make setup creates ignored local credentials. make up builds pinned images, initializes separate application, test, and Airflow databases, and waits for the API and scheduler. make pipeline safely extracts sample-data.tar.gz, triggers the real DAG, waits for completion, and prints its run status. Repeating it is an idempotency check: identical content has no duplicate business effect.

The supplied sample must publish 41,549 in-scope claims, 461 out-of-scope claims, 2,792 canonical reverts, 2,739 reversed in-scope claims, and 38,810 non-reversed claims. Gross value is exactly 215,173,602.15; non-reversed value is exactly 200,781,043.93; reversal rate is about 6.5922164%.

Architecture and transaction boundaries

directories -> immutable retained bytes -> occurrence evidence
     -> conflict-aware canonical claims/reverts -> analytics views
     -> revisioned CSV/JSON exports

The DAG is freeze_manifest -> [stage_pharmacies, stage_claims, stage_reverts] -> reconcile_validate_publish -> export_metrics -> finalize_run. Source tasks retry twice with exponential backoff from 30 seconds to five minutes. The finalizer runs with all_done and fails a DAG whose persisted pipeline outcome is incomplete.

Each file stages atomically: malformed suffixes cannot leave partial evidence. Publication holds a PostgreSQL advisory lock and commits admissions, reference replacement, affected-ID reconciliation, assertions, canonical changes, and the revision together. Export uses a repeatable-read snapshot and an atomic directory rename; export failure never rolls back a valid database publication. Older revisions are reconstructed from admitted evidence and that revision's reference snapshot without mutating current state.

Inputs and event semantics

Source contracts are in contracts/sources.yaml. NPIs and NDCs remain strings, timestamps normalize to UTC, and prices/quantities use exact numeric(20,6)/Python Decimal; binary floats and numeric strings are rejected. Pharmacy files are complete current snapshots, not deltas, because no effective dates are supplied.

An event ID with equivalent normalized variants has one effect. Distinct variants for the same ID conflict, all variants are excluded, and a later conflict retracts any earlier canonical effect. Invalid evidence retaining a usable UUID participates in that trust decision. Reverts may arrive before claims; all late events reconcile. Multiple valid reverts are retained and reversal state is recomputed for every old and candidate target. Unknown-pharmacy claims remain canonical but out of scope for pharmacy analytics.

For custom host inputs, set HIPPO_INPUT_DIR before startup and pass directories relative to that root:

HIPPO_INPUT_DIR=/absolute/host/input make up
HIPPO_INPUT_DIR=/absolute/host/input docker compose run --rm pipeline run \
  --pharmacy-dir pharmacies --claim-dir claims --revert-dir reverts

Each directory flag is repeatable. Source symlinks and paths outside the mounted root are rejected. Retained content-addressed bytes live in a Docker volume.

Data products

Stable consumer views are analytics.claims, analytics.pharmacy_daily, analytics.ndc_pricing, analytics.revert_activity, and analytics.dataset_status. Quality views expose conflicts, record issues, revert relationships, and sanitized run summaries. Daily cohort metrics use claim date; revert activity uses revert date. Rates are claim-count weighted. Monetary values are exact decimals. Pricing compares a pharmacy's monthly same-NDC average unit price with the median of other eligible pharmacies; zero-price and insufficient-peer rows retain explicit statuses instead of producing misleading ratios.

Exports appear under the persistent /data/output/<revision>/ volume as pharmacy_daily.csv, ndc_pricing.csv, quality.json, and manifest.json. Rows have declared deterministic sort keys; the manifest records versions, counts, checksums, and encoding rules. Empty CSV fields are null and JSON financial values are decimal strings.

contracts/analytics.yaml is the machine-readable semantic catalog, validated by contracts/catalog.schema.json and checked against the live views. Parameterized examples are in queries. They intentionally require explicit historical periods rather than pretending the sample represents the current week.

Operations and access

Open Airflow at http://localhost:8080; make airflow-login prints the URL, the local admin username, and the generated password. A manual Trigger DAG with no JSON configuration uses the mounted sample directories (pharmacies, claims, and reverts). On a fresh checkout, run make pipeline once (or python3 scripts/extract_sample.py after make up) before triggering manually. For custom mounted inputs, provide those directory lists in the trigger configuration, for example {"pharmacy_dirs":["pharmacies"],"claim_dirs":["claims"],"revert_dirs":["reverts"]}. Deterministic source/configuration failures are non-retryable; recognized transient infrastructure failures retain task retries. Correct inputs and trigger a new run to recover. make pipeline-cli runs the same services sequentially without Airflow.

make metrics prints the latest publication and key metrics and verifies or regenerates its export without ingesting data. make psql opens PostgreSQL as hippo_reader; useful first queries are:

SELECT * FROM analytics.dataset_status ORDER BY revision DESC LIMIT 1;
SELECT * FROM analytics.claims LIMIT 20;

That role is transaction-read-only with a statement timeout and can select analytics plus sanitized quality summary only; evidence and writes are denied. Migration ownership, runtime writing, and reading use separate roles.

Useful lifecycle commands:

make up             # build, initialize, and wait
make pipeline       # real Airflow sample run
make pipeline-cli   # sequential application path
make test           # unit, PostgreSQL, golden, and real Airflow tests
make lint           # Ruff lint and format checks
make benchmark      # supplied sample, 100k, and 1M measurements
make down           # stop, retain volumes
make reset          # destructive: remove local volumes/artifacts

CLI exit codes are 0 success, 2 configuration/schema, 3 deterministic source/publication, 4 recognized transient infrastructure, and 1 unexpected. Ordinary logs contain operational identifiers and counts, never raw payloads or credentials.

Correctness, performance, and design

Tests cover contracts, bounded parsers, identity normalization, rollback, concurrency, replay, input ordering/batching, revert-first and late events, late conflict retraction, reference changes, exact analytics, export recovery and historical parity, catalog permissions, the complete golden sample, DAG structure, and real DAG execution. A deliberately independent Python oracle checks randomized canonical state rather than reusing production reconciliation logic.

See docs/performance.md for measured sample/100k/1M results and limitations, and docs/decisions.md for assumptions and tradeoffs. Discovery still hashes old bytes, analytical views and exports may scan history, and historical export is intentionally a full recompute. No distributed system or physical aggregate was added without a measured need.

For AWS, source versions/manifests and retained bytes move to encrypted S3, orchestration to managed Airflow, secrets to Secrets Manager, revisioned exports to governed S3, and analytics toward Redshift. Streaming arrival may use Kinesis only after durable landing and checkpoint design. Redshift constraints are not an integrity substitute, so event conflict/reconciliation checks must remain explicit. Spark is appropriate only when measured distributed joins or execution windows justify it. Contracts, identities, metric formulas, lineage concepts, and tests remain stable while physical execution is redesigned. Production also requires IAM, audit, backup, retention/deletion, environment isolation, encryption, and a formal PHI/security review; this submission claims no compliance certification.

The project intentionally defers dashboards, chatbot/LLM integration, anomaly models, CDC, and a 5M benchmark. Codex assisted implementation; the checked-in contracts, tests, decisions, and measured evidence are the reviewable basis for implementation ownership.