diff --git a/backend/src/cyclone/parsers/validator.py b/backend/src/cyclone/parsers/validator.py new file mode 100644 index 0000000..f64cf79 --- /dev/null +++ b/backend/src/cyclone/parsers/validator.py @@ -0,0 +1,115 @@ +"""Structural and payer-specific validation rules for 837P claims. + +Each rule is a pure function that takes a :class:`ClaimOutput` and a +:class:`PayerConfig` and yields zero or more :class:`ValidationIssue`. The +top-level :func:`validate` function runs every rule and aggregates results. +""" + +from __future__ import annotations + +import re +from collections.abc import Iterable +from decimal import Decimal +from typing import Callable + +from cyclone.parsers.models import ClaimOutput, ValidationIssue, ValidationReport +from cyclone.parsers.payer import PayerConfig + +NPI_RE = re.compile(r"^\d{10}$") + +Rule = Callable[[ClaimOutput, PayerConfig], Iterable[ValidationIssue]] + + +def _r010_clm01_present(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if not claim.claim.claim_id or not claim.claim.claim_id.strip(): + yield ValidationIssue(rule="R010_clm01_present", severity="error", message="CLM01 (claim id) is empty") + + +def _r011_total_charge_positive(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if claim.claim.total_charge <= 0: + yield ValidationIssue(rule="R011_total_charge_positive", severity="error", message=f"CLM02 must be > 0, got {claim.claim.total_charge}") + + +def _r020_npi_format(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if claim.billing_provider.npi and not NPI_RE.match(claim.billing_provider.npi): + yield ValidationIssue(rule="R020_npi_format", severity="error", message=f"Billing provider NPI must be 10 digits, got {claim.billing_provider.npi!r}") + + +def _r030_frequency_allowed(claim: ClaimOutput, cfg: PayerConfig) -> Iterable[ValidationIssue]: + if not claim.claim.frequency_code: + return + try: + freq = int(claim.claim.frequency_code) + except ValueError: + yield ValidationIssue(rule="R030_frequency_allowed", severity="error", message=f"CLM05-3 not numeric: {claim.claim.frequency_code!r}") + return + if freq not in cfg.allowed_claim_frequencies: + yield ValidationIssue( + rule="R030_frequency_allowed", + severity="error", + message=f"CLM05-3 {freq} not in {sorted(cfg.allowed_claim_frequencies)} for payer {cfg.name}", + ) + + +def _r031_ref_g1_optional(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + """REF*G1 is informational in v1 (see spec ยง9 R031).""" + # We yield no errors; presence is recorded on the model. + return () + + +def _r050_diagnosis_present(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if not claim.diagnoses: + yield ValidationIssue(rule="R050_diagnosis_present", severity="error", message="HI segment missing โ€” no diagnoses on claim") + + +def _r060_service_dates_present(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + for sl in claim.service_lines: + if sl.service_date is None: + yield ValidationIssue(rule="R060_service_dates_present", severity="error", message=f"Service line {sl.line_number} missing service date (DTP*472)") + + +def _r070_charges_sum(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if not claim.service_lines: + return + line_sum = sum((sl.charge for sl in claim.service_lines), start=Decimal("0")) + if abs(line_sum - claim.claim.total_charge) > Decimal("0.01"): + yield ValidationIssue( + rule="R070_charges_sum", + severity="warning", + message=f"Sum of service lines ({line_sum}) does not match CLM02 ({claim.claim.total_charge})", + ) + + +def _r100_payer_id_matches(claim: ClaimOutput, cfg: PayerConfig) -> Iterable[ValidationIssue]: + if cfg.payer_id and claim.payer.id and claim.payer.id != cfg.payer_id: + yield ValidationIssue( + rule="R100_payer_id_matches", + severity="warning", + message=f"Payer id {claim.payer.id!r} != configured {cfg.payer_id!r}", + ) + + +_RULES: list[Rule] = [ + _r010_clm01_present, + _r011_total_charge_positive, + _r020_npi_format, + _r030_frequency_allowed, + _r031_ref_g1_optional, + _r050_diagnosis_present, + _r060_service_dates_present, + _r070_charges_sum, + _r100_payer_id_matches, +] + + +def validate(claim: ClaimOutput, config: PayerConfig) -> ValidationReport: + """Run every rule against a claim and return a report. + + Never raises โ€” all issues are accumulated into the report. + """ + errors: list[ValidationIssue] = [] + warnings: list[ValidationIssue] = [] + for rule in _RULES: + for issue in rule(claim, config): + (errors if issue.severity == "error" else warnings).append(issue) + return ValidationReport(passed=not errors, errors=errors, warnings=warnings) diff --git a/backend/tests/test_validator.py b/backend/tests/test_validator.py new file mode 100644 index 0000000..0bd016f --- /dev/null +++ b/backend/tests/test_validator.py @@ -0,0 +1,152 @@ +from datetime import date +from decimal import Decimal + +import pytest + +from cyclone.parsers.models import ( + Address, + BillingProvider, + ClaimHeader, + ClaimOutput, + Diagnosis, + Envelope, + Payer, + Procedure, + ServiceLine, + Subscriber, + ValidationReport, +) +from cyclone.parsers.payer import PayerConfig +from cyclone.parsers.validator import validate + + +def _build_claim(**overrides) -> ClaimOutput: + """Build a passing claim for tests; override fields to break specific rules.""" + base = dict( + claim_id="C1", + control_number="991102977", + transaction_date=date(2026, 6, 11), + billing_provider=BillingProvider( + name="Test Provider", + npi="1234567890", + tax_id="123456789", + address=Address(line1="1 Main", city="X", state="CO", zip="80000"), + ), + subscriber=Subscriber( + first_name="John", + last_name="Doe", + member_id="M1", + dob=date(1980, 1, 1), + gender="M", + address=Address(line1="1 Main", city="X", state="CO", zip="80000"), + ), + payer=Payer(name="COHCPF", id="SKCO0"), + claim=ClaimHeader( + claim_id="C1", + total_charge=Decimal("100.00"), + place_of_service="12", + frequency_code="1", + provider_signature="Y", + assignment="Y", + release_of_info="Y", + ), + diagnoses=[Diagnosis(code="Z00", qualifier="ABK")], + service_lines=[ + ServiceLine( + line_number=1, + procedure=Procedure(qualifier="HC", code="99213", modifiers=[]), + charge=Decimal("100.00"), + unit_type="UN", + units=Decimal("1.0"), + service_date=date(2026, 6, 11), + ) + ], + validation=ValidationReport(passed=True, errors=[], warnings=[]), + raw_segments=[], + ) + base.update(overrides) + return ClaimOutput(**base) + + +def test_validate_passing_claim(): + cfg = PayerConfig.co_medicaid() + report = validate(_build_claim(), cfg) + assert report.passed is True + assert report.errors == [] + + +def test_r010_clm01_required(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.claim.claim_id = "" + report = validate(claim, cfg) + assert not report.passed + assert any(i.rule == "R010_clm01_present" for i in report.errors) + + +def test_r011_total_charge_positive(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.claim.total_charge = Decimal("0.00") + report = validate(claim, cfg) + assert any(i.rule == "R011_total_charge_positive" for i in report.errors) + + +def test_r020_npi_must_be_ten_digits(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.billing_provider.npi = "12345" + report = validate(claim, cfg) + assert any(i.rule == "R020_npi_format" for i in report.errors) + + +def test_r030_frequency_allowed(): + cfg = PayerConfig.co_medicaid() # only 1, 7, 8 + claim = _build_claim() + claim.claim.frequency_code = "5" + report = validate(claim, cfg) + assert any(i.rule == "R030_frequency_allowed" for i in report.errors) + + +def test_r031_ref_g1_optional_no_error(): + """R031 is informational in v1 โ€” no REF*G1 should not error.""" + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + report = validate(claim, cfg) + assert not any(i.rule == "R031_ref_g1_optional" and i.severity == "error" for i in report.errors) + + +def test_r050_diagnosis_required(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim(diagnoses=[]) + report = validate(claim, cfg) + assert any(i.rule == "R050_diagnosis_present" for i in report.errors) + + +def test_r060_service_dates_required(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.service_lines[0].service_date = None + report = validate(claim, cfg) + assert any(i.rule == "R060_service_dates_present" for i in report.errors) + # Now add the date: + claim.service_lines[0].service_date = date(2026, 6, 11) + report = validate(claim, cfg) + assert not any(i.rule == "R060_service_dates_present" for i in report.errors + report.warnings) + + +def test_r070_charges_sum_warning(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.claim.total_charge = Decimal("999.00") # mismatch + report = validate(claim, cfg) + assert any(i.rule == "R070_charges_sum" and i.severity == "warning" for i in report.warnings) + + +def test_r100_payer_id_warning_only(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.payer.id = "WRONG" + report = validate(claim, cfg) + assert any(i.rule == "R100_payer_id_matches" and i.severity == "warning" for i in report.warnings) + assert report.passed is True