tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(batch-diff): side-by-side claim diff between two batches

Browse Source
This commit is contained in:
Tyler
2026-06-20 17:17:09 -06:00
parent 4cd52c3084
commit 7b394fff1a
12 changed files with 2556 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/api.py
+52
View File
@@ -749,6 +749,58 @@ def get_reconciliation_unmatched() -> dict:
return store.list_unmatched(kind="both")
# --------------------------------------------------------------------------- #
# Side-by-side diff between two batches (SP3 P4 / T18)
# --------------------------------------------------------------------------- #
@app.get("/api/batch-diff")
def get_batch_diff(
a: str | None = Query(None),
b: str | None = Query(None),
) -> dict:
"""Return a side-by-side diff of two batches identified by id.
Query params: ``a=<batch_id>``, ``b=<batch_id>`` (both required).
Response body (snake_case keys, see :mod:`cyclone.batch_diff` for the
projector shapes):
- ``a`` / ``b`` — small metadata blocks (id, kind, parsedAt,
inputFilename, claimCount)
- ``added`` — claims present in B but not A
- ``removed`` — claims present in A but not B
- ``changed`` — claims present in both, with field deltas
- ``summary`` — precomputed counts
Errors:
- 400 — missing ``a`` or ``b``
- 404 — either batch id is unknown
Pure read endpoint — never mutates the store. Both 837P and 835
batches are accepted (mixed-kind diffs are valid: comparing the
submitted claims against the matching remittances).
"""
if not a or not b:
raise HTTPException(
status_code=400,
detail={"error": "Missing param", "detail": "Both ?a=<batch_id> and ?b=<batch_id> are required."},
)
try:
a_rec, b_rec = store.load_two_for_diff(a, b)
except LookupError as exc:
raise HTTPException(
status_code=404,
detail={"error": "Not found", "detail": str(exc)},
)
# Lazy import — keeps the module's import surface small until the
# endpoint is actually hit. Mirrors the same pattern used by other
# endpoint-local helpers (e.g. reconciler).
from cyclone.batch_diff import diff_batches_to_wire
return diff_batches_to_wire(a_rec, b_rec)
@app.post("/api/reconciliation/match")
def post_reconciliation_match(body: dict) -> dict:
"""Manually pair a Claim with a Remittance (operator override).
backend/src/cyclone/batch_diff.py
+401
View File
@@ -0,0 +1,401 @@
"""Side-by-side diff between two parsed batches.
Powers ``GET /api/batch-diff?a=<batch_id>&b=<batch_id>``. Given two
``BatchRecord`` objects (kind=837P or 835), project each contained
claim/remittance to a small ``BatchClaimSummary`` and split the union
into three buckets:
- ``added`` — in B, not in A (by claim_id / payer_claim_control_number)
- ``removed`` — in A, not in B
- ``changed`` — in both, with at least one differing field
The diff is purely value-based (no segment-level deep diff). The four
fields we surface as "changed" are:
- status (837P: validation-derived label, 835: CLP02 code)
- total_charge (Decimal → number; tolerance 1e-2 to ignore float jitter)
- service_date (ISO date; missing vs present counts as a change)
- cpt_codes (sorted list equality)
The module is a pure function on Pydantic models — no DB I/O — so the
API endpoint and tests can both call it without session management.
"""
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
from typing import Any, Iterable
from cyclone.parsers.models import ClaimOutput, ValidationReport
from cyclone.parsers.models_835 import ClaimPayment, ParseResult835
from cyclone.store import BatchRecord, BatchRecord835, BatchRecord837, _claim_status_from_validation
# ---------------------------------------------------------------------------
# Projected types
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class BatchClaimSummary:
"""Small per-claim shape returned in the diff payload.
Kept tiny on purpose — the diff view shouldn't carry the full EDI
payload. Raw segments and diagnoses live on the per-batch detail
endpoint; this view is what the operator uses to spot the deltas
between A and B at a glance.
``service_date`` and ``cpt_codes`` are optional because 835
remittance rows don't carry the patient name; we surface empty
strings / empty lists rather than synthesizing values so the UI can
distinguish "no data" from "blank string".
"""
claim_id: str
patient_name: str
total_charge: float
service_date: str | None
status: str
cpt_codes: list[str]
def to_dict(self) -> dict[str, Any]:
return {
"claim_id": self.claim_id,
"patient_name": self.patient_name,
"total_charge": self.total_charge,
"service_date": self.service_date,
"status": self.status,
"cpt_codes": list(self.cpt_codes),
}
@dataclass(frozen=True)
class FieldDiff:
"""One changed field between two ``BatchClaimSummary`` rows.
``a`` and ``b`` are the field values from each side. We don't
attempt to coerce types — the frontend renders them verbatim — but
the comparator picks the field-specific projection (e.g. charge as
a number) so equality is stable.
"""
field: str
a: Any
b: Any
def to_dict(self) -> dict[str, Any]:
return {"field": self.field, "a": self.a, "b": self.b}
@dataclass(frozen=True)
class ChangedRow:
"""One claim present in both batches with at least one field delta.
Always carries both summaries (so the renderer can fall back to
a single side if a field is missing) plus the diff list (possibly
multiple fields changed for the same claim).
"""
a: BatchClaimSummary
b: BatchClaimSummary
diff: list[FieldDiff]
def to_dict(self) -> dict[str, Any]:
return {
"a": self.a.to_dict(),
"b": self.b.to_dict(),
"diff": [d.to_dict() for d in self.diff],
}
@dataclass(frozen=True)
class BatchDiffResult:
"""Full diff between two batches.
Mirrors the API contract documented in the task spec. Counts are
precomputed so the frontend doesn't have to derive them.
"""
a_meta: dict[str, Any]
b_meta: dict[str, Any]
added: list[BatchClaimSummary]
removed: list[BatchClaimSummary]
changed: list[ChangedRow]
@property
def summary(self) -> dict[str, int]:
added_count = len(self.added)
removed_count = len(self.removed)
changed_count = len(self.changed)
# ``unchanged`` are the claims present in BOTH batches that have
# no field differences. Derivation from A's total:
# A's total = unchanged + changed + removed
# so unchanged = A_total - changed - removed. Clamp at zero so
# a malformed batch pair can never surface a negative count.
a_total = len(self.a_meta.get("claim_ids", []) or [])
unchanged_count = max(0, a_total - changed_count - removed_count)
return {
"addedCount": added_count,
"removedCount": removed_count,
"changedCount": changed_count,
"unchangedCount": unchanged_count,
}
def to_dict(self) -> dict[str, Any]:
return {
"a": self.a_meta,
"b": self.b_meta,
"added": [s.to_dict() for s in self.added],
"removed": [s.to_dict() for s in self.removed],
"changed": [c.to_dict() for c in self.changed],
"summary": self.summary,
}
# ---------------------------------------------------------------------------
# Projection helpers
# ---------------------------------------------------------------------------
def _summary_from_claim_837(claim: ClaimOutput) -> BatchClaimSummary:
"""Project a 837P ``ClaimOutput`` to the diff summary shape."""
# Patient name: same composition as to_ui_claim.
sub = claim.subscriber
patient_name = f"{sub.first_name} {sub.last_name}".strip()
# Total charge: coerce Decimal to float; 0.0 when None.
total_charge = float(claim.claim.total_charge or Decimal("0"))
# Earliest service date across the SV1 service lines (matches
# store._service_dates_from_claim convention).
service_dates: list = [
sl.service_date for sl in claim.service_lines if sl.service_date is not None
]
service_date: str | None = None
if service_dates:
earliest = min(service_dates)
# The pydantic model serializes dates to ISO strings via the
# @model_serializer; pull isoformat defensively for both
# date and datetime inputs (Pydantic coerces in either case).
service_date = earliest.isoformat() if hasattr(earliest, "isoformat") else str(earliest)
# CPT codes: collect the procedure.code across all SV1s, in line order.
cpt_codes = [
sl.procedure.code for sl in claim.service_lines if sl.procedure.code
]
return BatchClaimSummary(
claim_id=claim.claim_id,
patient_name=patient_name,
total_charge=total_charge,
service_date=service_date,
status=_claim_status_from_validation(claim),
cpt_codes=cpt_codes,
)
def _summary_from_remit_835(cp: ClaimPayment) -> BatchClaimSummary:
"""Project an 835 ``ClaimPayment`` to the diff summary shape.
Patient name isn't carried on an ERA — we leave it blank rather
than synthesize. Total charge / paid amount aren't on the
``BatchClaimSummary`` contract; we expose the 835's ``total_charge``
on the BPR total so it lines up with the 837P's CLM01 amount.
"""
total_charge = float(cp.total_charge or Decimal("0"))
service_date: str | None = None
cpt_codes: list[str] = []
if cp.service_payments:
sp = cp.service_payments[0]
if sp.service_date is not None:
service_date = (
sp.service_date.isoformat()
if hasattr(sp.service_date, "isoformat")
else str(sp.service_date)
)
# CPT codes from every SVC segment in the CLP loop.
for svc in cp.service_payments:
if svc.procedure_code:
cpt_codes.append(svc.procedure_code)
# CLP02 status code as the primary status label; fall back to the
# human-readable label when one is bundled in the parser.
status = cp.status_label or cp.status_code or ""
return BatchClaimSummary(
# ``payer_claim_control_number`` is the ERA's natural id and the
# match key against another batch's 835 (matches on CLP01).
claim_id=cp.payer_claim_control_number,
patient_name="",
total_charge=total_charge,
service_date=service_date,
status=status,
cpt_codes=cpt_codes,
)
def _summarize_record(rec: BatchRecord) -> list[BatchClaimSummary]:
"""Return the list of summaries for a batch — 837P or 835.
Dispatched by ``isinstance`` so callers don't need to branch on
``kind``. Returns an empty list when the record's ``result`` is
missing the expected claim array (defensive against partially-loaded
rows in tests).
"""
if isinstance(rec, BatchRecord837):
return [_summary_from_claim_837(c) for c in (rec.result.claims or [])]
if isinstance(rec, BatchRecord835):
return [_summary_from_remit_835(c) for c in (rec.result.claims or [])]
# Unknown record type — return empty so the diff still computes.
return []
# ---------------------------------------------------------------------------
# Field comparison
# ---------------------------------------------------------------------------
def _eq_charge(a: float, b: float) -> bool:
"""Two floats are equal if they differ by less than a cent.
Decimal → float coercion can introduce sub-cent noise when the
source had more than 2 decimal places; tolerate that here so we
don't surface spurious charge deltas.
"""
return abs(a - b) < 0.01
def _diff_pair(
a: BatchClaimSummary,
b: BatchClaimSummary,
*,
compare_status: bool,
) -> list[FieldDiff]:
"""Return the list of changed fields between two same-id summaries.
Compares the four fields documented in the module docstring; an
unchanged field is simply not in the returned list.
``compare_status`` controls whether ``status`` enters the diff:
- True when both batches are the same kind (837P vs 837P, or
835 vs 835). 837P's validation-derived status (``submitted`` /
``pending`` / ``denied`` / ``draft``) and 835's CLP02 code
(``1`` / ``4`` / ``21`` / …) live in different namespaces, so
comparing across kinds surfaces spurious deltas the operator
can't act on. We skip the comparison when kinds differ.
"""
out: list[FieldDiff] = []
if compare_status and a.status != b.status:
out.append(FieldDiff("status", a.status, b.status))
if not _eq_charge(a.total_charge, b.total_charge):
out.append(FieldDiff("total_charge", a.total_charge, b.total_charge))
if a.service_date != b.service_date:
out.append(FieldDiff("service_date", a.service_date, b.service_date))
# CPT codes are compared as sorted, deduplicated lists so reordering
# between SV1 segments doesn't register as a diff.
a_codes = sorted(set(a.cpt_codes))
b_codes = sorted(set(b.cpt_codes))
if a_codes != b_codes:
out.append(FieldDiff("cpt_codes", a_codes, b_codes))
return out
def _meta(rec: BatchRecord, *, claim_ids: Iterable[str]) -> dict[str, Any]:
"""Build the ``a``/``b`` metadata block for the diff envelope.
The ``claim_ids`` list is passed in by the caller (the same source
used for the summaries) so the meta block is consistent with what
the diff actually saw.
"""
parsed_at = rec.parsed_at
parsed_iso = parsed_at.isoformat().replace("+00:00", "Z") if parsed_at else ""
return {
"id": rec.id,
"kind": rec.kind,
"parsedAt": parsed_iso,
"inputFilename": rec.input_filename,
"claimCount": len(list(claim_ids)),
# Internal-only — stripped from the wire response below. The
# API layer recomputes counts via to_dict, so this key is
# used solely for the ``summary.unchangedCount`` derivation.
"claim_ids": list(claim_ids),
}
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def diff_batches(
a: BatchRecord,
b: BatchRecord,
) -> BatchDiffResult:
"""Compute the side-by-side diff between ``a`` and ``b``.
Both records can be 837P, both 835, or one of each — we project
each to ``BatchClaimSummary`` independently and match by the
summary's ``claim_id`` field (837P ``CLM01``, 835 ``CLP01``). When
the kinds differ the result is still well-defined; it just means
you're comparing e.g. an 837P claim set against the matching
remittances — useful when the operator wants to see which claims
were paid vs. submitted.
Order preservation:
- ``added`` follows B's input order (matches the upload narrative).
- ``removed`` follows A's input order.
- ``changed`` follows B's input order (the "current state" side).
This mirrors git-diff conventions where the right-hand side drives
the row order on the page.
"""
a_summaries = _summarize_record(a)
b_summaries = _summarize_record(b)
a_by_id = {s.claim_id: s for s in a_summaries}
b_by_id = {s.claim_id: s for s in b_summaries}
# Status only carries comparable meaning when both batches share a
# kind (837P/837P or 835/835). Across kinds the namespaces differ
# (validation-derived vs CLP02) so we skip that field.
compare_status = a.kind == b.kind
added: list[BatchClaimSummary] = []
removed: list[BatchClaimSummary] = []
changed: list[ChangedRow] = []
# Walk B's order for added/changed (B drives row order).
for s in b_summaries:
if s.claim_id not in a_by_id:
added.append(s)
continue
other = a_by_id[s.claim_id]
diff = _diff_pair(other, s, compare_status=compare_status)
if diff:
changed.append(ChangedRow(a=other, b=s, diff=diff))
# Walk A's order for removed (anything in A but not in B).
for s in a_summaries:
if s.claim_id not in b_by_id:
removed.append(s)
a_meta = _meta(a, claim_ids=[s.claim_id for s in a_summaries])
b_meta = _meta(b, claim_ids=[s.claim_id for s in b_summaries])
return BatchDiffResult(
a_meta=a_meta,
b_meta=b_meta,
added=added,
removed=removed,
changed=changed,
)
def diff_batches_to_wire(
a: BatchRecord,
b: BatchRecord,
) -> dict[str, Any]:
"""Convenience wrapper that returns the JSON-ready dict.
Strips the internal ``claim_ids`` key from the metadata blocks
before returning so the wire contract stays clean — the spec only
promises ``id, kind, parsedAt, inputFilename, claimCount`` per side.
"""
result = diff_batches(a, b)
wire = result.to_dict()
wire["a"].pop("claim_ids", None)
wire["b"].pop("claim_ids", None)
return wire
backend/src/cyclone/store.py
+25
View File
@@ -1144,6 +1144,31 @@ class CycloneStore:
rows = s.query(Batch).order_by(Batch.parsed_at.asc()).all()
return [self._row_to_record(r) for r in rows]
def load_two_for_diff(
self,
a_id: str,
b_id: str,
) -> tuple[BatchRecord, BatchRecord]:
"""Load two batches by id for the side-by-side diff view.
Returns ``(a, b)`` as ``BatchRecord`` objects. Raises
:class:`LookupError` when either id is missing — the API layer
catches it and maps it to ``404 Not Found`` (matching the
``GET /api/batches/{id}`` contract). The two loads happen in
independent sessions so a transient failure on one side can't
poison the other.
Used exclusively by :mod:`cyclone.batch_diff` via the
``/api/batch-diff`` endpoint.
"""
a = self.get(a_id)
if a is None:
raise LookupError(f"batch {a_id} not found")
b = self.get(b_id)
if b is None:
raise LookupError(f"batch {b_id} not found")
return a, b
def iter_claims(
self,
*,