feat(backend): structural + CO Medicaid validation rules

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)