feat(backend): tighten CLM05 validation for CO Medicaid (CLM05-1 POS, CLM05-2 qualifier)
This commit is contained in:
@@ -59,6 +59,7 @@ class ClaimHeader(_Base):
|
|||||||
claim_id: str
|
claim_id: str
|
||||||
total_charge: Decimal
|
total_charge: Decimal
|
||||||
place_of_service: str | None = None
|
place_of_service: str | None = None
|
||||||
|
facility_code_qualifier: str | None = None
|
||||||
frequency_code: str | None = None
|
frequency_code: str | None = None
|
||||||
provider_signature: str | None = None
|
provider_signature: str | None = None
|
||||||
assignment: str | None = None
|
assignment: str | None = None
|
||||||
|
|||||||
@@ -181,17 +181,21 @@ def _consume_claim(segments: list[list[str]], idx: int) -> tuple[ClaimOutput, in
|
|||||||
claim_id = clm[1] if len(clm) > 1 else ""
|
claim_id = clm[1] if len(clm) > 1 else ""
|
||||||
total_charge = Decimal(clm[2]) if len(clm) > 2 and clm[2] else Decimal("0")
|
total_charge = Decimal(clm[2]) if len(clm) > 2 and clm[2] else Decimal("0")
|
||||||
pos = ""
|
pos = ""
|
||||||
|
qualifier: str | None = None
|
||||||
freq = ""
|
freq = ""
|
||||||
if len(clm) > 5:
|
if len(clm) > 5:
|
||||||
clm05 = clm[5].split(":")
|
clm05 = clm[5].split(":")
|
||||||
if len(clm05) > 0:
|
if len(clm05) > 0:
|
||||||
pos = clm05[0]
|
pos = clm05[0]
|
||||||
|
if len(clm05) > 1:
|
||||||
|
qualifier = clm05[1] or None
|
||||||
if len(clm05) > 2:
|
if len(clm05) > 2:
|
||||||
freq = clm05[2]
|
freq = clm05[2]
|
||||||
claim_header = ClaimHeader(
|
claim_header = ClaimHeader(
|
||||||
claim_id=claim_id,
|
claim_id=claim_id,
|
||||||
total_charge=total_charge,
|
total_charge=total_charge,
|
||||||
place_of_service=pos or None,
|
place_of_service=pos or None,
|
||||||
|
facility_code_qualifier=qualifier,
|
||||||
frequency_code=freq or None,
|
frequency_code=freq or None,
|
||||||
provider_signature=clm[6] if len(clm) > 6 else None,
|
provider_signature=clm[6] if len(clm) > 6 else None,
|
||||||
assignment=clm[7] if len(clm) > 7 else None,
|
assignment=clm[7] if len(clm) > 7 else None,
|
||||||
|
|||||||
@@ -11,6 +11,21 @@ from __future__ import annotations
|
|||||||
from pydantic import BaseModel, ConfigDict, Field
|
from pydantic import BaseModel, ConfigDict, Field
|
||||||
|
|
||||||
|
|
||||||
|
# Full CMS Place of Service code list (X12 837P CLM05-1).
|
||||||
|
# Two-digit zero-padded codes covering the entire CMS POS set as of 2026.
|
||||||
|
CMS_PLACE_OF_SERVICE_CODES: set[str] = {
|
||||||
|
"01", "02", "03", "04", "05", "06", "07", "08", "09",
|
||||||
|
"10", "11", "12", "13", "14", "15", "16", "17", "18", "19",
|
||||||
|
"20", "21", "22", "23", "24", "25", "26", "27", "28", "29",
|
||||||
|
"30", "31", "32", "33", "34", "41", "42", "49", "50", "51",
|
||||||
|
"52", "53", "54", "55", "56", "57", "58", "59", "60", "61",
|
||||||
|
"62", "63", "64", "65", "66", "67", "68", "69", "70", "71",
|
||||||
|
"72", "73", "74", "75", "76", "77", "78", "79", "80", "81",
|
||||||
|
"82", "83", "84", "85", "86", "87", "88", "89", "90", "91",
|
||||||
|
"92", "93", "94", "95", "96", "97", "98", "99",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
class PayerConfig(BaseModel):
|
class PayerConfig(BaseModel):
|
||||||
model_config = ConfigDict(frozen=True)
|
model_config = ConfigDict(frozen=True)
|
||||||
|
|
||||||
@@ -23,6 +38,12 @@ class PayerConfig(BaseModel):
|
|||||||
payer_name: str = ""
|
payer_name: str = ""
|
||||||
no_patient_loop: bool = False
|
no_patient_loop: bool = False
|
||||||
encounter_claim_in_same_batch: bool = False
|
encounter_claim_in_same_batch: bool = False
|
||||||
|
# CLM05-2 Facility Code Qualifier. "B" = CMS Place of Service code.
|
||||||
|
allowed_facility_qualifiers: set[str] = Field(default_factory=lambda: {"B"})
|
||||||
|
# CLM05-1 Facility Type Code / Place of Service. Defaults to the full CMS POS list.
|
||||||
|
allowed_place_of_service_codes: set[str] = Field(
|
||||||
|
default_factory=lambda: set(CMS_PLACE_OF_SERVICE_CODES)
|
||||||
|
)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def co_medicaid(cls) -> "PayerConfig":
|
def co_medicaid(cls) -> "PayerConfig":
|
||||||
@@ -41,6 +62,8 @@ class PayerConfig(BaseModel):
|
|||||||
payer_name="COHCPF",
|
payer_name="COHCPF",
|
||||||
no_patient_loop=True,
|
no_patient_loop=True,
|
||||||
encounter_claim_in_same_batch=False,
|
encounter_claim_in_same_batch=False,
|
||||||
|
allowed_facility_qualifiers={"B"},
|
||||||
|
allowed_place_of_service_codes=set(CMS_PLACE_OF_SERVICE_CODES),
|
||||||
)
|
)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
@@ -56,4 +79,6 @@ class PayerConfig(BaseModel):
|
|||||||
payer_name="",
|
payer_name="",
|
||||||
no_patient_loop=False,
|
no_patient_loop=False,
|
||||||
encounter_claim_in_same_batch=True,
|
encounter_claim_in_same_batch=True,
|
||||||
|
allowed_facility_qualifiers={"B"},
|
||||||
|
allowed_place_of_service_codes=set(CMS_PLACE_OF_SERVICE_CODES),
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -57,6 +57,37 @@ def _r031_ref_g1_optional(claim: ClaimOutput, _: PayerConfig) -> Iterable[Valida
|
|||||||
return ()
|
return ()
|
||||||
|
|
||||||
|
|
||||||
|
def _r032_clm05_2_facility_qualifier(claim: ClaimOutput, cfg: PayerConfig) -> Iterable[ValidationIssue]:
|
||||||
|
"""CLM05-2 Facility Code Qualifier must be recognized for the payer.
|
||||||
|
|
||||||
|
When the field is absent (e.g. production data with only a 2-component
|
||||||
|
CLM05), the rule is skipped silently — the absence of a qualifier is
|
||||||
|
tolerated, but a present-and-unknown qualifier is a hard error.
|
||||||
|
"""
|
||||||
|
q = claim.claim.facility_code_qualifier
|
||||||
|
if not q:
|
||||||
|
return
|
||||||
|
if q not in cfg.allowed_facility_qualifiers:
|
||||||
|
yield ValidationIssue(
|
||||||
|
rule="R032_clm05_2_facility_qualifier",
|
||||||
|
severity="error",
|
||||||
|
message=f"CLM05-2 facility code qualifier {q!r} not in {sorted(cfg.allowed_facility_qualifiers)} for payer {cfg.name}",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _r033_clm05_1_place_of_service_code(claim: ClaimOutput, cfg: PayerConfig) -> Iterable[ValidationIssue]:
|
||||||
|
"""CLM05-1 Facility Type Code must be a valid CMS Place of Service code."""
|
||||||
|
pos = claim.claim.place_of_service
|
||||||
|
if not pos:
|
||||||
|
return
|
||||||
|
if pos not in cfg.allowed_place_of_service_codes:
|
||||||
|
yield ValidationIssue(
|
||||||
|
rule="R033_clm05_1_place_of_service_code",
|
||||||
|
severity="error",
|
||||||
|
message=f"CLM05-1 place of service {pos!r} not in CMS POS code list for payer {cfg.name}",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def _r050_diagnosis_present(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]:
|
def _r050_diagnosis_present(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]:
|
||||||
if not claim.diagnoses:
|
if not claim.diagnoses:
|
||||||
yield ValidationIssue(rule="R050_diagnosis_present", severity="error", message="HI segment missing — no diagnoses on claim")
|
yield ValidationIssue(rule="R050_diagnosis_present", severity="error", message="HI segment missing — no diagnoses on claim")
|
||||||
@@ -95,6 +126,8 @@ _RULES: list[Rule] = [
|
|||||||
_r020_npi_format,
|
_r020_npi_format,
|
||||||
_r030_frequency_allowed,
|
_r030_frequency_allowed,
|
||||||
_r031_ref_g1_optional,
|
_r031_ref_g1_optional,
|
||||||
|
_r032_clm05_2_facility_qualifier,
|
||||||
|
_r033_clm05_1_place_of_service_code,
|
||||||
_r050_diagnosis_present,
|
_r050_diagnosis_present,
|
||||||
_r060_service_dates_present,
|
_r060_service_dates_present,
|
||||||
_r070_charges_sum,
|
_r070_charges_sum,
|
||||||
|
|||||||
@@ -150,3 +150,46 @@ def test_r100_payer_id_warning_only():
|
|||||||
report = validate(claim, cfg)
|
report = validate(claim, cfg)
|
||||||
assert any(i.rule == "R100_payer_id_matches" and i.severity == "warning" for i in report.warnings)
|
assert any(i.rule == "R100_payer_id_matches" and i.severity == "warning" for i in report.warnings)
|
||||||
assert report.passed is True
|
assert report.passed is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_r032_passes_when_qualifier_b():
|
||||||
|
cfg = PayerConfig.co_medicaid()
|
||||||
|
claim = _build_claim()
|
||||||
|
claim.claim.facility_code_qualifier = "B"
|
||||||
|
report = validate(claim, cfg)
|
||||||
|
assert not any(i.rule == "R032_clm05_2_facility_qualifier" for i in report.errors + report.warnings)
|
||||||
|
|
||||||
|
|
||||||
|
def test_r032_errors_when_qualifier_unknown():
|
||||||
|
cfg = PayerConfig.co_medicaid()
|
||||||
|
claim = _build_claim()
|
||||||
|
claim.claim.facility_code_qualifier = "X"
|
||||||
|
report = validate(claim, cfg)
|
||||||
|
assert any(i.rule == "R032_clm05_2_facility_qualifier" and i.severity == "error" for i in report.errors)
|
||||||
|
assert report.passed is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_r032_skipped_when_qualifier_none():
|
||||||
|
cfg = PayerConfig.co_medicaid()
|
||||||
|
claim = _build_claim()
|
||||||
|
# Default _build_claim leaves facility_code_qualifier unset (None).
|
||||||
|
assert claim.claim.facility_code_qualifier is None
|
||||||
|
report = validate(claim, cfg)
|
||||||
|
assert not any(i.rule == "R032_clm05_2_facility_qualifier" for i in report.errors + report.warnings)
|
||||||
|
|
||||||
|
|
||||||
|
def test_r033_passes_with_valid_pos_code():
|
||||||
|
cfg = PayerConfig.co_medicaid()
|
||||||
|
claim = _build_claim()
|
||||||
|
claim.claim.place_of_service = "12" # Office — valid CMS POS
|
||||||
|
report = validate(claim, cfg)
|
||||||
|
assert not any(i.rule == "R033_clm05_1_place_of_service_code" for i in report.errors + report.warnings)
|
||||||
|
|
||||||
|
|
||||||
|
def test_r033_errors_with_invalid_pos_code():
|
||||||
|
cfg = PayerConfig.co_medicaid()
|
||||||
|
claim = _build_claim()
|
||||||
|
claim.claim.place_of_service = "999" # not a valid CMS POS code
|
||||||
|
report = validate(claim, cfg)
|
||||||
|
assert any(i.rule == "R033_clm05_1_place_of_service_code" and i.severity == "error" for i in report.errors)
|
||||||
|
assert report.passed is False
|
||||||
|
|||||||
Reference in New Issue
Block a user