diff --git a/backend/src/cyclone/rebill/pipeline_b.py b/backend/src/cyclone/rebill/pipeline_b.py new file mode 100644 index 0000000..71f1c5d --- /dev/null +++ b/backend/src/cyclone/rebill/pipeline_b.py @@ -0,0 +1,116 @@ +"""Pipeline B: NOT_IN_835 visits → fresh 837Ps, batched by (member_id, ISO-week). + +One 837P per (member, ISO-week) pair. Per-visit files would be 4,509 SFTP +round-trips; per-batch (no member split) would lose the operator's +ability to track per-member. Member-week is the standard clearinghouse +pattern and matches AxisCare's billing cycle. + +The timely-filing gate is applied per visit before batching. The +override flag is the same for the whole batch (a per-batch override). +""" +from collections import defaultdict +from dataclasses import dataclass +from datetime import date +from decimal import Decimal +from pathlib import Path + +from cyclone.rebill.reconcile import VisitRow +from cyclone.rebill.timely_filing import timely_filing_decision + + +@dataclass(frozen=True) +class MemberWeekBatch: + """Pre-emission 837P shape for Pipeline B. The orchestrator's serializer + adapter converts this to a ClaimOutput for serialize_837 (one 837P + carrying every visit in `visits` as a CLM segment). + """ + + member_id: str + iso_year: int + iso_week: int + visits: list[VisitRow] + has_overridden_visits: bool + + +def build_pipeline_b_batches( + visits: list[VisitRow], + as_of: date, + override: bool, +) -> list[MemberWeekBatch]: + """Group rebillable visits by (member_id, ISO-week). + + Two filters are applied per visit before batching: + 1. timely_filing_decision(...).rebillable — drops past-window + visits unless `override=True` (per-batch override flag). + 2. implicit: only visits that survive (1) are grouped; the rest + are surfaced as EXCLUDED_TIMELY_FILING in the summary CSV. + + Returns batches sorted by (member_id, iso_year, iso_week) for + deterministic filenames downstream. + """ + by_key: dict[tuple[str, int, int], list[VisitRow]] = defaultdict(list) + overridden_keys: set[tuple[str, int, int]] = set() + for v in visits: + decision = timely_filing_decision(v.date, as_of, override) + if not decision.rebillable: + continue + iso = v.date.isocalendar() + key = (v.member_id, iso.year, iso.week) + by_key[key].append(v) + if decision.override and not decision.within_window: + overridden_keys.add(key) + return [ + MemberWeekBatch( + member_id=member, iso_year=yr, iso_week=wk, + visits=vs, has_overridden_visits=(member, yr, wk) in overridden_keys, + ) + for (member, yr, wk), vs in sorted(by_key.items()) + ] + + +def batch_visits_by_member_week( + visits: list[VisitRow], + as_of: date, + override: bool, +) -> dict[str, list[VisitRow]]: + """Thin wrapper over build_pipeline_b_batches. + + Returns ``{member_id: [VisitRow, ...]}`` — collapsing the (member, + ISO-week) batches back down to a per-member visit list. This is the + shape the orchestrator's summary aggregation expects. + + NOTE: This collapses multiple weeks for the same member into one + list. If the orchestrator needs week-aware per-member iteration, + use build_pipeline_b_batches directly. + """ + batches = build_pipeline_b_batches(visits, as_of=as_of, override=override) + out: dict[str, list[VisitRow]] = defaultdict(list) + for b in batches: + out[b.member_id].extend(b.visits) + return dict(out) + + +def write_pipeline_b_files( + batches: list[MemberWeekBatch], + out_dir: Path, + serialize_837_fn, + tpid: str, +) -> list[Path]: + """Write one 837P per batch, using HCPF-spec filenames. + + `serialize_837_fn(batch)` returns the X12 string. The 837P envelope + is pure ASCII so we write it as text. The filename is + `cyclone.edi.filenames.build_outbound_filename(tpid, '837P')`. + """ + # Lazy: keep cyclone.rebill.pipeline_b importable without zoneinfo from filenames. + from cyclone.edi.filenames import build_outbound_filename + + out_dir.mkdir(parents=True, exist_ok=True) + paths: list[Path] = [] + for b in batches: + body = serialize_837_fn(b) + fname = build_outbound_filename(tpid, "837P") + path = out_dir / fname + path.write_text(body, encoding="ascii") + paths.append(path) + return paths \ No newline at end of file diff --git a/backend/tests/test_rebill_pipeline_b.py b/backend/tests/test_rebill_pipeline_b.py new file mode 100644 index 0000000..86811d3 --- /dev/null +++ b/backend/tests/test_rebill_pipeline_b.py @@ -0,0 +1,41 @@ +"""Pipeline B: NOT_IN_835 visits → fresh 837Ps, batched by (member, ISO-week).""" +from datetime import date +from decimal import Decimal +from cyclone.rebill.pipeline_b import build_pipeline_b_batches +from cyclone.rebill.reconcile import VisitRow + + +def _v(date_, member, proc, amt): + return VisitRow(date=date_, member_id=member, procedure=proc, billed=Decimal(amt)) + + +def test_batches_split_by_member_and_iso_week(): + visits = [ + _v(date(2026, 6, 23), "J813715", "T1019", "2.32"), + _v(date(2026, 6, 25), "J813715", "T1019", "2.32"), # same week + _v(date(2026, 6, 23), "OTHER", "T1019", "2.32"), # different member + _v(date(2026, 6, 30), "J813715", "T1019", "2.32"), # different week + ] + out = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False) + assert len(out) == 3 + sizes = sorted(len(b.visits) for b in out) + assert sizes == [1, 1, 2] + + +def test_timely_filing_excludes_old_visits(): + visits = [ + _v(date(2026, 1, 1), "OLD", "T1019", "2.32"), # 187 days old + _v(date(2026, 6, 27), "NEW", "T1019", "2.32"), # 10 days old + ] + out = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False) + members = {b.member_id for b in out} + assert "OLD" not in members + assert "NEW" in members + + +def test_override_relaxes_timely_filing(): + visits = [_v(date(2026, 1, 1), "OLD", "T1019", "2.32")] + out_default = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False) + out_overridden = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=True) + assert out_default == [] + assert len(out_overridden) == 1 \ No newline at end of file