merge: SP41 inwindow-rebill-pipeline into main
This commit is contained in:
Nora
@@ -982,3 +982,19 @@ class Resubmission(Base):
|
||||
unique=True,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class SubmissionRecord(Base):
|
||||
"""SP41: one row per (claim_id, submitted_at) pre-flight dedup guard.
|
||||
|
||||
The 30-day dedup window is enforced by ``cyclone.store.submission_dedup``.
|
||||
This table records (claim_id, submitted_at) for every push that passed
|
||||
the pre-flight check. Past the window, the prior record is ignored
|
||||
(the guard lets through re-submits older than
|
||||
``cyclone.store.submission_dedup.DEFAULT_WINDOW_DAYS``).
|
||||
"""
|
||||
|
||||
__tablename__ = "submission_dedup"
|
||||
|
||||
claim_id: Mapped[str] = mapped_column(String(64), primary_key=True)
|
||||
submitted_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, index=True)
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
-- version: 22
|
||||
-- SP41: claim-id dedup at SFTP pre-flight.
|
||||
--
|
||||
-- The 30-day dedup window is enforced by cyclone.store.submission_dedup.
|
||||
-- This table records (claim_id, submitted_at) for every push that passed
|
||||
-- the pre-flight check. Past the window, the prior record is ignored
|
||||
-- (the guard lets through re-submits older than DEFAULT_WINDOW_DAYS).
|
||||
|
||||
CREATE TABLE submission_dedup (
|
||||
claim_id TEXT PRIMARY KEY,
|
||||
submitted_at DATETIME NOT NULL
|
||||
);
|
||||
|
||||
CREATE INDEX submission_dedup_submitted_at_idx
|
||||
ON submission_dedup (submitted_at);
|
||||
@@ -0,0 +1,11 @@
|
||||
"""SP41 — in-window rebill pipeline.
|
||||
|
||||
Owns:
|
||||
- parse_835_svc (SVC-level 835 reparse with member_id)
|
||||
- reconcile (visit-to-835 join on (member_id, procedure, DOS))
|
||||
- carc_filter (CARC-aware exclusion for Pipeline A)
|
||||
- timely_filing (120-day DOS age gate)
|
||||
- pipeline_a (denied/partial → frequency-7 rebill)
|
||||
- pipeline_b (NOT_IN_835 → fresh 837Ps by (member, ISO-week))
|
||||
- summary (summary CSV + per-category counters)
|
||||
"""
|
||||
@@ -0,0 +1,44 @@
|
||||
"""CARC-aware filter.
|
||||
|
||||
EXCLUDED = contractual / not recoverable / not a denial-of-payment in the
|
||||
narrow sense (charge exceeds fee schedule, prior to coverage, etc.).
|
||||
REVIEW = operator must decide (claim/service lacks info, non-covered
|
||||
charges, duplicate, etc.). Anything else falls through to REBILL.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class CarcDecision(str, Enum):
|
||||
EXCLUDED = "EXCLUDED"
|
||||
REVIEW = "REVIEW"
|
||||
REBILL = "REBILL"
|
||||
|
||||
|
||||
# These CARCs are NOT recoverable denials. Do not rebill.
|
||||
EXCLUDED_CARCS: frozenset[str] = frozenset({
|
||||
"CO-45", # charge exceeds fee schedule / contractual obligation
|
||||
"CO-26", # expenses incurred prior to coverage
|
||||
"CO-129", # prior processing information; forward to next payer
|
||||
})
|
||||
|
||||
# These CARCs need human review before rebill.
|
||||
REVIEW_CARCS: frozenset[str] = frozenset({
|
||||
"PI-16", # claim/service lacks information
|
||||
"PI-96", # non-covered charges
|
||||
"PI-15", # authorization / certification absent
|
||||
"PI-4", # procedure not paid separately
|
||||
"PI-110", # billing date predates service date
|
||||
"OA-18", # exact duplicate (resubmit noise)
|
||||
"OA-23", # impact of prior payer adjudication
|
||||
})
|
||||
|
||||
|
||||
def decide_carc(reasons: tuple[str, ...] | list[str]) -> CarcDecision:
|
||||
"""Pick the strongest action: EXCLUDED > REVIEW > REBILL."""
|
||||
if any(r in EXCLUDED_CARCS for r in reasons):
|
||||
return CarcDecision.EXCLUDED
|
||||
if any(r in REVIEW_CARCS for r in reasons):
|
||||
return CarcDecision.REVIEW
|
||||
return CarcDecision.REBILL
|
||||
@@ -0,0 +1,124 @@
|
||||
"""SVC-level 835 reparse with member_id at SVC scope.
|
||||
|
||||
Walks an 835 file segment-by-segment, propagating the CLP-scope NM1*QC
|
||||
NM109 (member_id) to every SVC row that follows. Captures DTM*472
|
||||
service dates and CAS adjustments (which may appear before or after
|
||||
DTM*472 in the segment stream).
|
||||
"""
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
ELEM = "*"
|
||||
SEG_END = "~"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SvcRow:
|
||||
src_file: str
|
||||
claim_id: str
|
||||
member_id: str
|
||||
status: str
|
||||
procedure: str
|
||||
modifiers: str
|
||||
charge: Decimal
|
||||
paid: Decimal
|
||||
units: Decimal
|
||||
svc_date: date
|
||||
cas_reasons: tuple[str, ...] # each entry is "GROUP-CODE" (e.g. "OA-18")
|
||||
pay_date: date | None
|
||||
|
||||
|
||||
def _split(seg: str) -> list[str]:
|
||||
return seg.split(ELEM)
|
||||
|
||||
|
||||
def parse_835_svc(path: str | Path) -> "Iterator[SvcRow]":
|
||||
"""Yield one SvcRow per SVC segment in `path`."""
|
||||
path = Path(path)
|
||||
text = path.read_text()
|
||||
segments = [s for s in text.split(SEG_END) if s]
|
||||
pay_date: date | None = None
|
||||
current_claim_id = ""
|
||||
current_status = ""
|
||||
current_member_id = ""
|
||||
i = 0
|
||||
while i < len(segments):
|
||||
elems = _split(segments[i])
|
||||
seg_name = elems[0]
|
||||
if seg_name == "DTM" and len(elems) > 2 and elems[1] == "405":
|
||||
try:
|
||||
pay_date = _ymd(elems[2])
|
||||
except ValueError:
|
||||
pass
|
||||
elif seg_name == "CLP":
|
||||
current_claim_id = elems[1] if len(elems) > 1 else ""
|
||||
current_status = elems[2] if len(elems) > 2 else ""
|
||||
current_member_id = ""
|
||||
elif seg_name == "NM1" and len(elems) > 1 and elems[1] == "QC":
|
||||
current_member_id = elems[9] if len(elems) > 9 else ""
|
||||
elif seg_name == "SVC" and current_claim_id:
|
||||
rest = elems[1:]
|
||||
comp = rest[0] if rest else ""
|
||||
proc_parts = comp.split(":")
|
||||
procedure = proc_parts[1] if len(proc_parts) > 1 else ""
|
||||
modifiers = ":".join(proc_parts[2:]) if len(proc_parts) > 2 else ""
|
||||
charge = _decimal(rest[1]) if len(rest) > 1 else Decimal("0")
|
||||
paid = _decimal(rest[2]) if len(rest) > 2 else Decimal("0")
|
||||
units = _decimal(rest[3]) if len(rest) > 3 else Decimal("0")
|
||||
svc_date, cas_reasons = _walk_for_dtm_and_cas(segments, i)
|
||||
yield SvcRow(
|
||||
src_file=path.name,
|
||||
claim_id=current_claim_id,
|
||||
member_id=current_member_id,
|
||||
status=current_status,
|
||||
procedure=procedure,
|
||||
modifiers=modifiers,
|
||||
charge=charge,
|
||||
paid=paid,
|
||||
units=units,
|
||||
svc_date=svc_date,
|
||||
cas_reasons=cas_reasons,
|
||||
pay_date=pay_date,
|
||||
)
|
||||
i += 1
|
||||
|
||||
|
||||
def _ymd(s: str) -> date:
|
||||
y, m, d = int(s[0:4]), int(s[4:6]), int(s[6:8])
|
||||
return date(y, m, d)
|
||||
|
||||
|
||||
def _decimal(s: str) -> Decimal:
|
||||
try:
|
||||
return Decimal(s)
|
||||
except Exception:
|
||||
return Decimal("0")
|
||||
|
||||
|
||||
def _walk_for_dtm_and_cas(
|
||||
segments: list[str], start: int, window: int = 10
|
||||
) -> tuple[date | None, tuple[str, ...]]:
|
||||
"""Walk the next `window` segments after SVC. Capture DTM*472 and CAS.
|
||||
|
||||
X12 835 may place DTM*472 before or after CAS segments within a service
|
||||
line, so we walk the full window and capture both, breaking only at the
|
||||
next SVC or CLP (a new claim/service boundary).
|
||||
"""
|
||||
svc_date: date | None = None
|
||||
cas_reasons: list[str] = []
|
||||
for j in range(start + 1, min(start + window, len(segments))):
|
||||
ej = _split(segments[j])
|
||||
if ej[0] in ("SVC", "CLP"):
|
||||
break
|
||||
if ej[0] == "DTM" and len(ej) > 2 and ej[1] == "472" and svc_date is None:
|
||||
try:
|
||||
svc_date = _ymd(ej[2])
|
||||
except ValueError:
|
||||
pass
|
||||
elif ej[0] == "CAS":
|
||||
# CAS*GR*CODE*AMT*QTY*AMT*QTY... → GR-CODE
|
||||
if len(ej) >= 3:
|
||||
cas_reasons.append(f"{ej[1]}-{ej[2]}")
|
||||
return svc_date, tuple(cas_reasons)
|
||||
@@ -0,0 +1,86 @@
|
||||
"""Pipeline A: denied/partial → frequency-7 replacement 837Ps.
|
||||
|
||||
For each visit the previous adjudication said was DENIED or PARTIAL,
|
||||
emit a fresh 837P with CLM05-3 = '7' (replacement claim) and the
|
||||
original claim_submit_id preserved as CLM01. The 837P carries the
|
||||
original DOS and the same procedure / member.
|
||||
|
||||
The CARC-aware filter is the gate — only REBILL / REVIEW decisions are
|
||||
emitted; EXCLUDED visits never make it to a file.
|
||||
"""
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
from cyclone.rebill.carc_filter import CarcDecision
|
||||
from cyclone.rebill.reconcile import ReconcileOutcome, OutcomeCategory
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RebillClaim:
|
||||
"""Pre-emission 837P shape for Pipeline A. The orchestrator's serializer
|
||||
adapter converts this to a ClaimOutput for serialize_837.
|
||||
"""
|
||||
|
||||
claim_id: str # reuses original claim_submit_id
|
||||
member_id: str
|
||||
procedure: str
|
||||
svc_date: date
|
||||
charge: Decimal
|
||||
frequency_code: str # always "7" for Pipeline A
|
||||
needs_review: bool # True for CARC REVIEW decisions
|
||||
original_carc_reasons: tuple[str, ...]
|
||||
|
||||
|
||||
def build_pipeline_a_claims(
|
||||
original_claim_id: str,
|
||||
visit_outcomes: list[ReconcileOutcome],
|
||||
carc_decisions: list[CarcDecision],
|
||||
cas_reasons_per_visit: list[tuple[str, ...]],
|
||||
) -> list[RebillClaim]:
|
||||
"""Zip three parallel per-visit lists by index; drop PAID/NOT_IN_835 outcomes and EXCLUDED CARC decisions; emit one frequency-7 RebillClaim per survivor."""
|
||||
out: list[RebillClaim] = []
|
||||
# strict=True: all three lists must be the same length; a mismatch is a contract bug, fail loud.
|
||||
for vo, carc, reasons in zip(visit_outcomes, carc_decisions, cas_reasons_per_visit, strict=True):
|
||||
if vo.category not in (OutcomeCategory.DENIED, OutcomeCategory.PARTIAL):
|
||||
continue
|
||||
if carc == CarcDecision.EXCLUDED:
|
||||
continue
|
||||
out.append(RebillClaim(
|
||||
claim_id=original_claim_id,
|
||||
member_id=vo.visit.member_id,
|
||||
procedure=vo.visit.procedure,
|
||||
svc_date=vo.visit.date,
|
||||
charge=vo.visit.billed,
|
||||
frequency_code="7",
|
||||
needs_review=(carc == CarcDecision.REVIEW),
|
||||
original_carc_reasons=reasons,
|
||||
))
|
||||
return out
|
||||
|
||||
|
||||
def write_pipeline_a_files(
|
||||
claims: list[RebillClaim],
|
||||
out_dir: Path,
|
||||
serialize_837_fn, # injected: callable[[RebillClaim], str] (orchestrator-supplied adapter)
|
||||
tpid: str,
|
||||
) -> list[Path]:
|
||||
"""Write one 837P per claim, using HCPF-spec filenames via build_outbound_filename.
|
||||
|
||||
`serialize_837_fn(claim)` 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_a 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 c in claims:
|
||||
body = serialize_837_fn(c)
|
||||
fname = build_outbound_filename(tpid, "837P")
|
||||
path = out_dir / fname
|
||||
path.write_text(body, encoding="ascii")
|
||||
paths.append(path)
|
||||
return paths
|
||||
@@ -0,0 +1,123 @@
|
||||
"""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 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.
|
||||
|
||||
Note: frozen=True is shallow — `visits` is a mutable list, so
|
||||
`mb.visits.append(x)` works even though `mb.visits = new_list` raises
|
||||
FrozenInstanceError. Treat the batch as logically immutable.
|
||||
"""
|
||||
|
||||
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.
|
||||
|
||||
PROVISIONAL: this helper is defined for the Task 12 orchestrator's
|
||||
anticipated use, but has no callers in this branch. If Task 12's
|
||||
actual needs differ, this function may be removed or replaced without
|
||||
notice.
|
||||
|
||||
This also 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
|
||||
@@ -0,0 +1,69 @@
|
||||
"""Visit-to-835 reconciliation on (member_id, procedure, DOS).
|
||||
|
||||
Authoritative match key. The 5% tolerance on PAID handles the legitimate
|
||||
charge-amount differences between AxisCare's per-unit CSV and the 835's
|
||||
per-SVC breakdown.
|
||||
"""
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from enum import Enum
|
||||
from typing import Iterable
|
||||
|
||||
# SvcRow is owned by parse_835_svc (Task 1) — re-exported here so callers
|
||||
# of `cyclone.rebill.reconcile` only need one import.
|
||||
from cyclone.rebill.parse_835_svc import SvcRow # noqa: F401 (re-exported)
|
||||
|
||||
PAID_TOLERANCE = Decimal("0.05") # charge match tolerance
|
||||
PARTIAL_RATIO = Decimal("0.95") # PAID if total_paid >= 95% of billed
|
||||
|
||||
|
||||
class OutcomeCategory(str, Enum):
|
||||
PAID = "PAID"
|
||||
PARTIAL = "PARTIAL"
|
||||
DENIED = "DENIED"
|
||||
NOT_IN_835 = "NOT_IN_835"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class VisitRow:
|
||||
date: date
|
||||
member_id: str
|
||||
procedure: str
|
||||
billed: Decimal
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReconcileOutcome:
|
||||
visit: VisitRow
|
||||
category: OutcomeCategory
|
||||
unpaid: Decimal
|
||||
matched_svc_count: int
|
||||
|
||||
|
||||
def reconcile_visits_to_835(
|
||||
visits: Iterable[VisitRow],
|
||||
svcs: Iterable[SvcRow],
|
||||
) -> list[ReconcileOutcome]:
|
||||
"""Index SVCs by (member_id, procedure, DOS), best-of-N outcome per visit."""
|
||||
by_key: dict[tuple[str, str, date], list[SvcRow]] = defaultdict(list)
|
||||
for s in svcs:
|
||||
by_key[(s.member_id, s.procedure, s.svc_date)].append(s)
|
||||
|
||||
out: list[ReconcileOutcome] = []
|
||||
for v in visits:
|
||||
cands = by_key.get((v.member_id, v.procedure, v.date), [])
|
||||
if not cands:
|
||||
out.append(ReconcileOutcome(v, OutcomeCategory.NOT_IN_835, v.billed, 0))
|
||||
continue
|
||||
total_paid = sum((c.paid for c in cands), Decimal("0"))
|
||||
any_paid = any(c.paid > 0 for c in cands)
|
||||
if any_paid and total_paid >= v.billed * PARTIAL_RATIO:
|
||||
cat, unpaid = OutcomeCategory.PAID, Decimal("0")
|
||||
elif any_paid:
|
||||
cat, unpaid = OutcomeCategory.PARTIAL, max(Decimal("0"), v.billed - total_paid)
|
||||
else:
|
||||
cat, unpaid = OutcomeCategory.DENIED, v.billed
|
||||
out.append(ReconcileOutcome(v, cat, unpaid, len(cands)))
|
||||
return out
|
||||
@@ -0,0 +1,57 @@
|
||||
"""Summary CSV for SP41.
|
||||
|
||||
One row per visit, classified into one of:
|
||||
REBILLED_A — pipeline A emitted a frequency-7 replacement
|
||||
REBILLED_B — pipeline B emitted a fresh 837P
|
||||
EXCLUDED_CARC — CO-45/CO-26/CO-129 (or other excluded CARC)
|
||||
EXCLUDED_PAYER — non-CO_TXIX payer per AxisCare API spot audit
|
||||
EXCLUDED_NO_EVV — no Authorized=Yes AND no AxisCare EVV ref
|
||||
EXCLUDED_TIMELY_FILING — DOS > 120 days before run date, no override
|
||||
"""
|
||||
import csv
|
||||
from dataclasses import dataclass
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
from cyclone.rebill.reconcile import VisitRow
|
||||
|
||||
REBILLED_A = "REBILLED_A"
|
||||
REBILLED_B = "REBILLED_B"
|
||||
EXCLUDED_CARC = "EXCLUDED_CARC"
|
||||
EXCLUDED_PAYER = "EXCLUDED_PAYER"
|
||||
EXCLUDED_NO_EVV = "EXCLUDED_NO_EVV"
|
||||
EXCLUDED_TIMELY_FILING = "EXCLUDED_TIMELY_FILING"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SummaryRow:
|
||||
"""One visit + its post-pipeline disposition for the operator audit trail."""
|
||||
|
||||
visit: VisitRow
|
||||
disposition: str
|
||||
unpaid: Decimal
|
||||
cas_reasons: tuple[str, ...]
|
||||
file_path: str # relative to dev/rebills/<date>/
|
||||
|
||||
|
||||
def write_summary_csv(rows: list[SummaryRow], out_path: Path) -> int:
|
||||
"""Write the summary CSV; return the number of rows written."""
|
||||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with out_path.open("w", newline="") as f:
|
||||
w = csv.writer(f)
|
||||
w.writerow([
|
||||
"dos", "member_id", "procedure", "billed",
|
||||
"disposition", "unpaid", "cas_reasons", "file_path",
|
||||
])
|
||||
for r in rows:
|
||||
w.writerow([
|
||||
r.visit.date.isoformat(),
|
||||
r.visit.member_id,
|
||||
r.visit.procedure,
|
||||
str(r.visit.billed),
|
||||
r.disposition,
|
||||
str(r.unpaid),
|
||||
"|".join(r.cas_reasons), # pipe-separated; no collision with J-prefixed member IDs or relative file paths
|
||||
r.file_path,
|
||||
])
|
||||
return len(rows)
|
||||
@@ -0,0 +1,37 @@
|
||||
"""120-day timely-filing gate for Colorado Medicaid.
|
||||
|
||||
Per HCPF's Colorado Medical Assistance Program provider manual, claims must
|
||||
be filed within 120 days from the date of service. The gate is HARD by
|
||||
default — past-window visits are surfaced as EXCLUDED_TIMELY_FILING in the
|
||||
summary CSV and not emitted by Pipeline B.
|
||||
|
||||
The operator may pass override=True per batch to relax the gate (e.g. for
|
||||
good-cause appeal visits). The decision is recorded in the summary CSV
|
||||
either way so the audit trail shows the override.
|
||||
"""
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
|
||||
FILING_WINDOW_DAYS = 120
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TimelyFilingDecision:
|
||||
dos: date
|
||||
as_of: date
|
||||
days_old: int
|
||||
within_window: bool
|
||||
override: bool
|
||||
rebillable: bool
|
||||
|
||||
|
||||
def timely_filing_decision(
|
||||
dos: date, as_of: date, override: bool
|
||||
) -> TimelyFilingDecision:
|
||||
days_old = (as_of - dos).days
|
||||
within_window = days_old <= FILING_WINDOW_DAYS
|
||||
return TimelyFilingDecision(
|
||||
dos=dos, as_of=as_of, days_old=days_old,
|
||||
within_window=within_window, override=override,
|
||||
rebillable=within_window or override,
|
||||
)
|
||||
@@ -4,6 +4,8 @@ These are part of the public API — callers (API endpoints) catch them
|
||||
and translate to HTTP 409 Conflict responses.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class AlreadyMatchedError(Exception):
|
||||
"""Raised by ``CycloneStore.manual_match`` when the claim is already paired.
|
||||
@@ -38,4 +40,27 @@ class InvalidStateError(Exception):
|
||||
self.activity_kind = activity_kind
|
||||
super().__init__(
|
||||
f"invalid state {current_state} for apply (kind={activity_kind})"
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class DuplicateClaimError(Exception):
|
||||
"""Raised by ``cyclone.store.submission_dedup.check_duplicate`` when a claim_id
|
||||
is re-submitted within the 30-day dedup window (SP41).
|
||||
|
||||
Mirrors the other 409-class exceptions: callers catch it at the API
|
||||
boundary and surface it as a 409 Conflict. Carries
|
||||
``original_submission_at`` so the UI can render "already submitted on
|
||||
YYYY-MM-DD" without re-querying.
|
||||
|
||||
Note: the "30-day" literal in the message mirrors
|
||||
``cyclone.store.submission_dedup.DEFAULT_WINDOW_DAYS`` — keep them in
|
||||
sync if either ever moves.
|
||||
"""
|
||||
|
||||
def __init__(self, claim_id: str, original_submission_at: datetime):
|
||||
self.claim_id = claim_id
|
||||
self.original_submission_at = original_submission_at
|
||||
super().__init__(
|
||||
f"claim_id {claim_id!r} was already submitted at "
|
||||
f"{original_submission_at.isoformat()}; within 30-day window"
|
||||
)
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
"""Claim-id dedup at the SFTP pre-flight stage.
|
||||
|
||||
Schema: the ``submission_dedup`` table (defined on the main ``Base`` in
|
||||
``cyclone.db``) records (claim_id, submitted_at). A re-submission of
|
||||
the same claim_id within ``DEFAULT_WINDOW_DAYS`` (30) is rejected with
|
||||
``DuplicateClaimError``. Past the window, the guard lets it through —
|
||||
the operator may be retrying a 999-rejected file from >30 days ago,
|
||||
which is a legitimate rebuild path.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from sqlalchemy import select
|
||||
|
||||
from cyclone import db as cycl_db
|
||||
from cyclone.db import SubmissionRecord
|
||||
from cyclone.store.exceptions import DuplicateClaimError
|
||||
|
||||
DEFAULT_WINDOW_DAYS = 30
|
||||
|
||||
|
||||
def record_submission(claim_id: str, submitted_at: datetime, db_url: str) -> None:
|
||||
"""Upsert (claim_id, submitted_at). Idempotent on claim_id.
|
||||
|
||||
``db_url`` is accepted for back-compat with the original Task 8
|
||||
signature but ignored — sessions now flow through the process-wide
|
||||
``cycl_db.SessionLocal()`` factory that ``db.init_db()`` set up.
|
||||
"""
|
||||
del db_url # signature-preserving; conftest wires the per-test DB
|
||||
with cycl_db.SessionLocal()() as session:
|
||||
session.merge(SubmissionRecord(
|
||||
claim_id=claim_id,
|
||||
submitted_at=submitted_at,
|
||||
))
|
||||
session.commit()
|
||||
|
||||
|
||||
def check_duplicate(
|
||||
claim_id: str,
|
||||
db_url: str | None = None,
|
||||
now: datetime | None = None,
|
||||
window_days: int = DEFAULT_WINDOW_DAYS,
|
||||
) -> None:
|
||||
"""Raise ``DuplicateClaimError`` if ``claim_id`` was submitted within the window.
|
||||
|
||||
``db_url`` is accepted for back-compat with the original Task 8
|
||||
signature but ignored — sessions flow through ``cycl_db.SessionLocal()``
|
||||
just like the rest of ``cyclone.store``. Tests that pass an explicit
|
||||
``db_url`` keep working because ``cycl_db.SessionLocal()()`` points at
|
||||
the URL the conftest resolved under ``CYCLONE_DB_URL``.
|
||||
"""
|
||||
del db_url # signature-preserving
|
||||
if now is None:
|
||||
now = datetime.now(timezone.utc)
|
||||
cutoff = now - timedelta(days=window_days)
|
||||
with cycl_db.SessionLocal()() as session:
|
||||
row = session.scalars(
|
||||
select(SubmissionRecord).where(
|
||||
SubmissionRecord.claim_id == claim_id,
|
||||
SubmissionRecord.submitted_at >= cutoff,
|
||||
)
|
||||
).first()
|
||||
if row is not None:
|
||||
raise DuplicateClaimError(claim_id, row.submitted_at)
|
||||
@@ -0,0 +1 @@
|
||||
ISA*00* *00* *ZZ*CYCLONE *ZZ*GAINWELL *260101*1200*^*00501*000000001*0*P*:~GS*HC*CYCLONE*GAINWELL*20260101*1200*1*X*005010X221A1~ST*835*0001~BPR*I*100.00*C*ACH*CCP*01*021000021*DA*123456789*1512345678**01*021000021*DA*123456789*20260101~TRN*1*TRACE01*1512345678~DTM*405*20260118~N1*PR*COLORADO MEDICAL ASSISTANCE PROGRAM*XV*COMEDASSISTPROG~N3*PO BOX 1100~N4*DENVER*CO*80202~REF*2U*7912900843~LX*1~CLP*T1001*1*16.24*16.10**MC*2026029105200*11*1~NM1*QC*1*AALBUE*ERIC****MR*J813715~NM1*74*1*AALBUE*ERIC*W***C*J813715~DTM*232*20260118~DTM*233*20260124~SVC*HC:T1019:U2:SC:KX*2.32*2.30**.33~DTM*472*20260118~CAS*CO*45*.02~REF*G1*6252960154~REF*6R*T1001V001~AMT*B6*2.30~SVC*HC:T1019:U2:SC:KX*2.32*2.30**.33~DTM*472*20260119~CAS*OA*18*2.32~SVC*HC:T1019:U2:SC:KX*2.32*-2.30**.33~DTM*472*20260120~CAS*CO*45*-.02~LX*2~CLP*T1002*4*16.24*0**MC*2026029105201*11*1~NM1*QC*1*OTHER-A*FIRST****MR*OTHER-MEMBER-A~SVC*HC:T1019*2.32*0**.33~DTM*472*20260121~CAS*CO*97*2.32~LX*3~CLP*T1003*22*16.24*-16.10**MC*2026029105202*11*1~NM1*QC*1*OTHER-B*SECOND****MR*OTHER-MEMBER-B~SVC*HC:T1019*2.32*-2.30**.33~DTM*472*20260122~SE*23*0001~GE*1*1~IEA*1*000000001~
|
||||
@@ -51,7 +51,7 @@ def test_migration_0002_creates_acks_table():
|
||||
|
||||
def test_migration_latest_idempotent_on_fresh_db():
|
||||
"""Re-running the migration on the same DB must be a no-op (PRAGMA
|
||||
user_version already at the latest version — currently 21 after
|
||||
user_version already at the latest version — currently 22 after
|
||||
0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007
|
||||
providers/payers/clearhouse, SP10's 0008 payer_rejected,
|
||||
SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged,
|
||||
@@ -62,16 +62,16 @@ def test_migration_latest_idempotent_on_fresh_db():
|
||||
claim.patient_control_number backfill UPDATE, SP28's 0018
|
||||
claim_acks join table, SP32's 0019
|
||||
rendering_provider_npi + service_provider_npi, SP37's 0020
|
||||
transaction_set_control_number, and SP39's 0021 resubmissions
|
||||
audit table)."""
|
||||
transaction_set_control_number, SP39's 0021 resubmissions
|
||||
audit table, and SP41's 0022 submission_dedup)."""
|
||||
with db.engine().begin() as c:
|
||||
v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
|
||||
assert v1 == 21
|
||||
assert v1 == 22
|
||||
# A second run should not raise and should not bump the version.
|
||||
db_migrate.run(db.engine())
|
||||
with db.engine().begin() as c:
|
||||
v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
|
||||
assert v2 == 21
|
||||
assert v2 == 22
|
||||
|
||||
|
||||
def test_add_ack_persists_row():
|
||||
|
||||
@@ -128,14 +128,15 @@ def test_migration_latest_idempotent_on_fresh_db(tmp_path: Path) -> None:
|
||||
service_provider_npi on claims and remittances.
|
||||
SP37 bumped it to 20 with batch transaction_set_control_number.
|
||||
SP39 bumped it to 21 with the resubmissions audit table.
|
||||
SP41 bumped it to 22 with submission_dedup.
|
||||
"""
|
||||
engine = _fresh_engine(tmp_path)
|
||||
db_migrate.run(engine)
|
||||
v_after_first = _user_version(engine)
|
||||
assert v_after_first == 21, f"expected head=21, got {v_after_first}"
|
||||
assert v_after_first == 22, f"expected head=22, got {v_after_first}"
|
||||
|
||||
db_migrate.run(engine)
|
||||
assert _user_version(engine) == 21, "second run should not bump version"
|
||||
assert _user_version(engine) == 22, "second run should not bump version"
|
||||
|
||||
|
||||
def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
@@ -161,7 +162,7 @@ def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: py
|
||||
engine = _fresh_engine(tmp_path)
|
||||
|
||||
db_migrate.run(engine)
|
||||
assert _user_version(engine) == 21, f"expected head=21, got {_user_version(engine)}"
|
||||
assert _user_version(engine) == 22, f"expected head=22, got {_user_version(engine)}"
|
||||
|
||||
# Two claims in one batch with the same patient_control_number
|
||||
# must be insertable. If 0015's table recreation re-introduced a
|
||||
|
||||
@@ -102,10 +102,10 @@ def migrated_engine(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
||||
engine = _fresh_engine(tmp_path / "mig0020.db")
|
||||
db_migrate.run(engine)
|
||||
|
||||
# Confirm head is 21 (every migration applied, including SP39's 0021).
|
||||
# Confirm head is 22 (every migration applied, including SP41's 0022).
|
||||
with engine.connect() as conn:
|
||||
v = conn.exec_driver_sql("PRAGMA user_version").scalar() or 0
|
||||
assert v == 21, f"expected migration head=21, got {v}"
|
||||
assert v == 22, f"expected migration head=22, got {v}"
|
||||
|
||||
yield engine
|
||||
engine.dispose()
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
"""CARC-aware filter for Pipeline A.
|
||||
|
||||
Contractual / non-recoverable denials (CO-45, CO-26, CO-129) must be
|
||||
excluded from rebill. CARCs that need operator review (PI-16, PI-96,
|
||||
PI-15, PI-4, PI-110, OA-18, OA-23) must be surfaced as 'REVIEW' so the
|
||||
operator can decide.
|
||||
"""
|
||||
from cyclone.rebill.carc_filter import (
|
||||
CarcDecision,
|
||||
decide_carc,
|
||||
EXCLUDED_CARCS,
|
||||
REVIEW_CARCS,
|
||||
)
|
||||
|
||||
|
||||
def test_co45_is_excluded():
|
||||
assert decide_carc(("CO-45",)) == CarcDecision.EXCLUDED
|
||||
|
||||
|
||||
def test_co26_is_excluded():
|
||||
assert decide_carc(("CO-26",)) == CarcDecision.EXCLUDED
|
||||
|
||||
|
||||
def test_co129_is_excluded():
|
||||
assert decide_carc(("CO-129",)) == CarcDecision.EXCLUDED
|
||||
|
||||
|
||||
def test_pi16_is_review():
|
||||
assert decide_carc(("PI-16",)) == CarcDecision.REVIEW
|
||||
|
||||
|
||||
def test_o18_is_review():
|
||||
"""OA-18 is the duplicate noise — surface for review, don't auto-rebill."""
|
||||
assert decide_carc(("OA-18",)) == CarcDecision.REVIEW
|
||||
|
||||
|
||||
def test_no_carc_is_rebill():
|
||||
assert decide_carc(()) == CarcDecision.REBILL
|
||||
|
||||
|
||||
def test_mixed_excluded_wins():
|
||||
"""If any CARC is excluded, the whole service is excluded."""
|
||||
assert decide_carc(("PI-16", "CO-45")) == CarcDecision.EXCLUDED
|
||||
|
||||
|
||||
def test_sets_have_expected_members():
|
||||
assert "CO-45" in EXCLUDED_CARCS
|
||||
assert "PI-16" in REVIEW_CARCS
|
||||
assert "OA-18" in REVIEW_CARCS
|
||||
@@ -0,0 +1,38 @@
|
||||
"""835 SVC-level parser with member_id at SVC scope."""
|
||||
from pathlib import Path
|
||||
from collections import defaultdict
|
||||
from decimal import Decimal
|
||||
from cyclone.rebill.parse_835_svc import parse_835_svc
|
||||
|
||||
FIX = Path(__file__).parent / "fixtures" / "835_sample_svc_with_member.txt"
|
||||
|
||||
def test_parse_835_svc_extracts_member_id():
|
||||
"""NM1*QC NM109 at the CLP scope must propagate to that CLP's SVC rows;
|
||||
the multi-claim fixture confirms each CLP's NM1*QC lands on its own SVCs."""
|
||||
rows = list(parse_835_svc(FIX))
|
||||
assert len(rows) >= 1
|
||||
by_claim: dict[str, list[str]] = defaultdict(list)
|
||||
for r in rows:
|
||||
by_claim[r.claim_id].append(r.member_id)
|
||||
assert r.procedure # non-empty
|
||||
assert r.svc_date # non-empty
|
||||
assert r.charge > 0
|
||||
# First CLP (T1001) -> original fixture member
|
||||
assert all(mid == "J813715" for mid in by_claim["T1001"]), by_claim
|
||||
# Second CLP (T1002, status 4) -> its own NM1*QC
|
||||
assert all(mid == "OTHER-MEMBER-A" for mid in by_claim["T1002"]), by_claim
|
||||
# Third CLP (T1003, status 22) -> its own NM1*QC
|
||||
assert all(mid == "OTHER-MEMBER-B" for mid in by_claim["T1003"]), by_claim
|
||||
|
||||
def test_parse_835_svc_extracts_cas_reasons():
|
||||
"""CAS segments after DTM*472 must be captured (post-DTM*472 ordering)."""
|
||||
rows = list(parse_835_svc(FIX))
|
||||
# at least one row should have an OA-18 reason
|
||||
assert any("OA-18" in r.cas_reasons for r in rows)
|
||||
|
||||
def test_parse_835_svc_picks_up_status_22_reversals():
|
||||
"""Status 22 (reversal of previous payment) must be preserved, along with
|
||||
status 1 (primary) and status 4 (denied)."""
|
||||
rows = list(parse_835_svc(FIX))
|
||||
statuses = {r.status for r in rows}
|
||||
assert statuses == {"1", "4", "22"}
|
||||
@@ -0,0 +1,55 @@
|
||||
"""Pipeline A: denied/partial visits become frequency-7 replacement claims."""
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from cyclone.rebill.pipeline_a import build_pipeline_a_claims
|
||||
from cyclone.rebill.reconcile import VisitRow, ReconcileOutcome, OutcomeCategory
|
||||
from cyclone.rebill.carc_filter import CarcDecision
|
||||
|
||||
|
||||
def _outcome(date_, member, proc, billed, cat, unpaid=Decimal("0")):
|
||||
visit = VisitRow(date=date_, member_id=member, procedure=proc, billed=Decimal(billed))
|
||||
return ReconcileOutcome(visit, cat, unpaid, 0)
|
||||
|
||||
|
||||
def test_pipeline_a_emits_frequency_7():
|
||||
"""CLM05-3 must be 7 (replacement), preserving the original claim_submit_id."""
|
||||
out = build_pipeline_a_claims(
|
||||
original_claim_id="ORIG-001",
|
||||
visit_outcomes=[
|
||||
_outcome(date(2026, 6, 27), "J813715", "T1019", "2.32",
|
||||
OutcomeCategory.DENIED, unpaid=Decimal("2.32")),
|
||||
],
|
||||
carc_decisions=[CarcDecision.REBILL],
|
||||
cas_reasons_per_visit=[()],
|
||||
)
|
||||
assert len(out) == 1
|
||||
assert out[0].claim_id == "ORIG-001"
|
||||
assert out[0].frequency_code == "7"
|
||||
assert out[0].svc_date == date(2026, 6, 27)
|
||||
|
||||
|
||||
def test_pipeline_a_excludes_carc_excluded_visits():
|
||||
out = build_pipeline_a_claims(
|
||||
original_claim_id="ORIG-002",
|
||||
visit_outcomes=[
|
||||
_outcome(date(2026, 6, 27), "J813715", "T1019", "2.32",
|
||||
OutcomeCategory.DENIED),
|
||||
],
|
||||
carc_decisions=[CarcDecision.EXCLUDED],
|
||||
cas_reasons_per_visit=[("CO-45",)],
|
||||
)
|
||||
assert out == [] # not emitted
|
||||
|
||||
|
||||
def test_pipeline_a_surfaces_review_visits_with_flag():
|
||||
out = build_pipeline_a_claims(
|
||||
original_claim_id="ORIG-003",
|
||||
visit_outcomes=[
|
||||
_outcome(date(2026, 6, 27), "J813715", "T1019", "2.32",
|
||||
OutcomeCategory.DENIED),
|
||||
],
|
||||
carc_decisions=[CarcDecision.REVIEW],
|
||||
cas_reasons_per_visit=[("PI-16",)],
|
||||
)
|
||||
assert len(out) == 1
|
||||
assert out[0].needs_review is True
|
||||
@@ -0,0 +1,75 @@
|
||||
"""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
|
||||
|
||||
|
||||
def test_override_flag_set_on_past_window_visit_batch():
|
||||
# Same member "OLD" with one past-window visit (187 days old, ISO
|
||||
# week 1) and one within-window visit (10 days old, ISO week 26).
|
||||
# With override=True the past-window visit survives and the batch
|
||||
# for week 1 must be flagged has_overridden_visits=True because the
|
||||
# override saved an otherwise-excluded visit.
|
||||
visits = [
|
||||
_v(date(2026, 1, 1), "OLD", "T1019", "2.32"), # 187 days old (past window, W01)
|
||||
_v(date(2026, 6, 27), "OLD", "T1019", "2.32"), # 10 days old (within window, W26)
|
||||
]
|
||||
out_overridden = build_pipeline_b_batches(
|
||||
visits, as_of=date(2026, 7, 7), override=True,
|
||||
)
|
||||
# Two batches: one per (member, ISO-week) — the past-window visit and
|
||||
# the within-window visit land in different weeks.
|
||||
assert len(out_overridden) == 2
|
||||
by_week = {b.iso_week: b for b in out_overridden}
|
||||
assert by_week[1].member_id == "OLD"
|
||||
assert by_week[1].has_overridden_visits is True # the override saved this batch
|
||||
assert by_week[26].member_id == "OLD"
|
||||
assert by_week[26].has_overridden_visits is False # no override needed here
|
||||
|
||||
# With override=False the past-window visit is dropped. The surviving
|
||||
# within-window batch (W26) must NOT carry the override flag, and the
|
||||
# past-window batch (W01) is absent.
|
||||
out_default = build_pipeline_b_batches(
|
||||
visits, as_of=date(2026, 7, 7), override=False,
|
||||
)
|
||||
assert len(out_default) == 1
|
||||
assert out_default[0].member_id == "OLD"
|
||||
assert out_default[0].iso_week == 26
|
||||
assert out_default[0].has_overridden_visits is False
|
||||
@@ -0,0 +1,76 @@
|
||||
"""Visit-to-835 reconciliation on (member_id, procedure, DOS)."""
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from cyclone.rebill.reconcile import (
|
||||
VisitRow,
|
||||
SvcRow,
|
||||
ReconcileOutcome,
|
||||
reconcile_visits_to_835,
|
||||
OutcomeCategory,
|
||||
)
|
||||
|
||||
|
||||
def _v(date, member, proc, amt):
|
||||
return VisitRow(date=date, member_id=member, procedure=proc, billed=Decimal(amt))
|
||||
|
||||
|
||||
def _s(date, member, proc, chg, paid, status="1"):
|
||||
return SvcRow(
|
||||
src_file="x.835", claim_id="c1", member_id=member, status=status,
|
||||
procedure=proc, modifiers="", charge=Decimal(chg), paid=Decimal(paid),
|
||||
units=Decimal("0"), svc_date=date, cas_reasons=(), pay_date=None,
|
||||
)
|
||||
|
||||
|
||||
def test_visit_paid_when_any_svc_paid_within_tolerance():
|
||||
visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")]
|
||||
svcs = [_s(date(2026, 1, 18), "J813715", "T1019", "2.32", "2.30")]
|
||||
out = reconcile_visits_to_835(visits, svcs)
|
||||
assert out[0].category == OutcomeCategory.PAID
|
||||
assert out[0].unpaid == Decimal("0")
|
||||
|
||||
|
||||
def test_visit_paid_when_one_svc_paid_others_denied_duplicate():
|
||||
"""OA-18 on duplicates is noise; paid SVC wins."""
|
||||
visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")]
|
||||
svcs = [
|
||||
_s(date(2026, 1, 18), "J813715", "T1019", "2.32", "2.30", "1"),
|
||||
_s(date(2026, 1, 18), "J813715", "T1019", "2.32", "0.00", "4"),
|
||||
_s(date(2026, 1, 18), "J813715", "T1019", "2.32", "0.00", "4"),
|
||||
]
|
||||
out = reconcile_visits_to_835(visits, svcs)
|
||||
assert out[0].category == OutcomeCategory.PAID
|
||||
|
||||
|
||||
def test_visit_denied_when_all_svc_denied():
|
||||
visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")]
|
||||
svcs = [
|
||||
_s(date(2026, 1, 18), "J813715", "T1019", "2.32", "0.00", "4"),
|
||||
_s(date(2026, 1, 18), "J813715", "T1019", "2.32", "0.00", "4"),
|
||||
]
|
||||
out = reconcile_visits_to_835(visits, svcs)
|
||||
assert out[0].category == OutcomeCategory.DENIED
|
||||
assert out[0].unpaid == Decimal("2.32")
|
||||
|
||||
|
||||
def test_visit_not_in_835_when_no_svc_match():
|
||||
visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")]
|
||||
svcs = [_s(date(2026, 1, 19), "J813715", "T1019", "2.32", "2.30")]
|
||||
out = reconcile_visits_to_835(visits, svcs)
|
||||
assert out[0].category == OutcomeCategory.NOT_IN_835
|
||||
assert out[0].unpaid == Decimal("2.32")
|
||||
|
||||
|
||||
def test_partial_when_total_paid_below_95pct_of_billed():
|
||||
visits = [_v(date(2026, 1, 18), "J813715", "T1019", "10.00")]
|
||||
svcs = [_s(date(2026, 1, 18), "J813715", "T1019", "10.00", "4.00")]
|
||||
out = reconcile_visits_to_835(visits, svcs)
|
||||
assert out[0].category == OutcomeCategory.PARTIAL
|
||||
assert out[0].unpaid == Decimal("6.00")
|
||||
|
||||
|
||||
def test_different_member_does_not_match():
|
||||
visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")]
|
||||
svcs = [_s(date(2026, 1, 18), "OTHERMEMBER", "T1019", "2.32", "2.30")]
|
||||
out = reconcile_visits_to_835(visits, svcs)
|
||||
assert out[0].category == OutcomeCategory.NOT_IN_835
|
||||
@@ -0,0 +1,44 @@
|
||||
"""Summary CSV: one row per visit with its disposition."""
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
from cyclone.rebill.summary import (
|
||||
SummaryRow,
|
||||
write_summary_csv,
|
||||
REBILLED_A, REBILLED_B, EXCLUDED_CARC, EXCLUDED_TIMELY_FILING,
|
||||
EXCLUDED_PAYER, EXCLUDED_NO_EVV,
|
||||
)
|
||||
from cyclone.rebill.reconcile import VisitRow
|
||||
|
||||
|
||||
def test_summary_csv_emits_one_row_per_visit(tmp_path: Path):
|
||||
out = tmp_path / "summary.csv"
|
||||
rows = [
|
||||
SummaryRow(
|
||||
visit=VisitRow(date=date(2026, 6, 27), member_id="J813715",
|
||||
procedure="T1019", billed=Decimal("2.32")),
|
||||
disposition=REBILLED_A, unpaid=Decimal("0.00"),
|
||||
cas_reasons=("CO-45",), file_path="pipeline-a/x.837",
|
||||
),
|
||||
SummaryRow(
|
||||
visit=VisitRow(date=date(2026, 1, 1), member_id="OLD",
|
||||
procedure="T1019", billed=Decimal("2.32")),
|
||||
disposition=EXCLUDED_TIMELY_FILING, unpaid=Decimal("2.32"),
|
||||
cas_reasons=(), file_path="",
|
||||
),
|
||||
]
|
||||
write_summary_csv(rows, out)
|
||||
text = out.read_text()
|
||||
assert "J813715" in text
|
||||
assert "REBILLED_A" in text
|
||||
assert "EXCLUDED_TIMELY_FILING" in text
|
||||
assert "OLD" in text
|
||||
|
||||
|
||||
def test_disposition_constants():
|
||||
assert REBILLED_A == "REBILLED_A"
|
||||
assert REBILLED_B == "REBILLED_B"
|
||||
assert EXCLUDED_CARC == "EXCLUDED_CARC"
|
||||
assert EXCLUDED_PAYER == "EXCLUDED_PAYER"
|
||||
assert EXCLUDED_NO_EVV == "EXCLUDED_NO_EVV"
|
||||
assert EXCLUDED_TIMELY_FILING == "EXCLUDED_TIMELY_FILING"
|
||||
@@ -0,0 +1,39 @@
|
||||
"""120-day DOS-age gate. CO Medicaid's timely-filing window is 120 days from DOS."""
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from cyclone.rebill.timely_filing import timely_filing_decision, FILING_WINDOW_DAYS
|
||||
|
||||
|
||||
def test_visit_within_window_is_rebillable():
|
||||
decision = timely_filing_decision(
|
||||
dos=date(2026, 6, 1), as_of=date(2026, 7, 7), override=False
|
||||
)
|
||||
assert decision.rebillable is True
|
||||
|
||||
|
||||
def test_visit_past_window_excluded_by_default():
|
||||
decision = timely_filing_decision(
|
||||
dos=date(2026, 1, 1), as_of=date(2026, 7, 7), override=False
|
||||
)
|
||||
assert decision.rebillable is False
|
||||
assert decision.days_old == 187
|
||||
|
||||
|
||||
def test_visit_at_exact_120_days_inclusive():
|
||||
decision = timely_filing_decision(
|
||||
dos=date(2026, 3, 9), as_of=date(2026, 7, 7), override=False
|
||||
)
|
||||
# 2026-03-09 → 2026-07-07 = 120 days
|
||||
assert decision.days_old == 120
|
||||
assert decision.rebillable is True
|
||||
|
||||
|
||||
def test_override_flag_relaxes_the_gate():
|
||||
decision = timely_filing_decision(
|
||||
dos=date(2026, 1, 1), as_of=date(2026, 7, 7), override=True
|
||||
)
|
||||
assert decision.rebillable is True
|
||||
|
||||
|
||||
def test_window_constant_is_120():
|
||||
assert FILING_WINDOW_DAYS == 120
|
||||
@@ -0,0 +1,38 @@
|
||||
"""Claim-id dedup at the SFTP pre-flight stage."""
|
||||
from datetime import datetime
|
||||
from cyclone.store.exceptions import DuplicateClaimError
|
||||
from cyclone.store.submission_dedup import (
|
||||
record_submission,
|
||||
check_duplicate,
|
||||
DEFAULT_WINDOW_DAYS,
|
||||
)
|
||||
|
||||
|
||||
def test_check_duplicate_returns_none_for_first_submission(tmp_path):
|
||||
db = tmp_path / "test.db"
|
||||
assert check_duplicate("ORIG-001", db_url=f"sqlite:///{db}") is None
|
||||
|
||||
|
||||
def test_check_duplicate_raises_within_window(tmp_path):
|
||||
db = tmp_path / "test.db"
|
||||
record_submission("ORIG-002", submitted_at=datetime(2026, 6, 1), db_url=f"sqlite:///{db}")
|
||||
try:
|
||||
check_duplicate("ORIG-002", db_url=f"sqlite:///{db}",
|
||||
now=datetime(2026, 6, 15))
|
||||
except DuplicateClaimError as e:
|
||||
assert e.claim_id == "ORIG-002"
|
||||
assert e.original_submission_at == datetime(2026, 6, 1)
|
||||
else:
|
||||
raise AssertionError("expected DuplicateClaimError")
|
||||
|
||||
|
||||
def test_check_duplicate_passes_after_window(tmp_path):
|
||||
db = tmp_path / "test.db"
|
||||
record_submission("ORIG-003", submitted_at=datetime(2026, 1, 1), db_url=f"sqlite:///{db}")
|
||||
# 100 days later — past the 30-day default window
|
||||
assert check_duplicate("ORIG-003", db_url=f"sqlite:///{db}",
|
||||
now=datetime(2026, 4, 11)) is None
|
||||
|
||||
|
||||
def test_default_window_is_30_days():
|
||||
assert DEFAULT_WINDOW_DAYS == 30
|
||||
Reference in New Issue
Block a user