Signal/python-backend/core/doc_state_machine.py

"""
doc_state_machine.py
Signal — STTIL Solutions

Computes per-patient documentation status across 5 items.
Requirements are payer-dependent. Produces cascade consequence chain.

PHI Contract: no patient names, DOBs, or contact info in any parameter.
"""

from dataclasses import dataclass, field
from datetime import date
from typing import Optional


# Payer types that require PECOS enrollment
_PECOS_REQUIRED = {"medicare", "medicare_advantage"}

# Payer types that require Prior Authorization for CGM
_PA_REQUIRED = {"medicare_advantage", "medicaid", "commercial"}

# Medicare FFS: PA not required (CGM coverage expansion 2023 + CMS-1828-F exemption path)
_PA_NOT_REQUIRED = {"medicare"}


@dataclass
class DocState:
    swo: str = "Pending"          # "On File" | "Pending" | "Expired"
    visit: str = "Not Confirmed"  # "Confirmed YYYY-MM-DD" | "Estimated YYYY-MM-DD" | "Not Confirmed"
    pecos: str = "Not Verified"   # "Verified" | "Not Verified" | "N/A" | "Pending — Verify"
    pa: str = "Required — Not Started"  # "Not Required" | "Approved" | "Pending" | "Denied" | "Required — Not Started" | "Pending — Verify"
    diagnosis: str = "Missing"    # "On File" | "Missing" | "Pending — Verify"
    cascade: list = field(default_factory=list)


def _resolve_visit_display(
    csv_visit_date: Optional[date],
    confirmed_visit_date: Optional[date],
    is_transfer: bool,
) -> tuple[str, bool]:
    """
    Returns (display_string, is_confirmed).
    is_confirmed=True means the date is reliable (from CSV or staff-confirmed).
    """
    if is_transfer:
        return "Pending — Verify", False
    if csv_visit_date:
        return f"Confirmed {csv_visit_date.isoformat()}", True
    if confirmed_visit_date:
        return f"Confirmed {confirmed_visit_date.isoformat()}", True
    return "Not Confirmed — Estimated Only", False


def _build_cascade(
    swo: str,
    visit_confirmed: bool,
    pecos: str,
    pa: str,
    diagnosis: str,
    payer_type: str,
) -> list:
    """
    Build the full consequence chain for all documentation gaps.
    No early returns — every unfulfilled requirement is surfaced.
    Type A = doc gaps Signal can help close.
    Type B = route-to-action alerts for denials.
    """
    cascade = []

    # Type A: Documentation gaps
    if not visit_confirmed:
        cascade.append(
            "Qualifying visit not on file — confirm with prescriber before scheduling shipment"
        )

    if swo in ("Pending", "Expired"):
        label = "not on file" if swo == "Pending" else "expired"
        cascade.append(
            f"Written order {label} — obtain from prescriber before claim submission"
        )

    if diagnosis == "Missing":
        cascade.append(
            "Diagnosis code not on file — obtain from prescriber records before billing"
        )

    if payer_type in _PECOS_REQUIRED and pecos == "Not Verified":
        cascade.append(
            "Ordering provider PECOS enrollment not verified — confirm NPI is active before claim submission"
        )

    if payer_type not in _PA_NOT_REQUIRED and pa == "Required — Not Started":
        cascade.append(
            "Prior authorization not started — obtain before scheduling shipment"
        )

    # Type B: Route-to-action alert
    if pa == "Denied":
        cascade.append(
            "Prior authorization denied — file Level 1 appeal within 65 days. "
            "Approximately 80% of first-level Medicare Advantage appeals succeed."
        )

    # Consequence line — only appended when at least one gap exists
    if cascade:
        cascade.append("Device shipment at risk")

    return cascade


def compute_doc_state(
    payer_type: str,
    csv_swo_status: Optional[str],
    csv_visit_date: Optional[date],
    confirmed_visit_date: Optional[date],
    csv_pecos_verified: Optional[str],
    csv_pa_status: Optional[str],
    csv_diagnosis_on_file: Optional[str],
    is_transfer: bool = False,
) -> DocState:
    """
    Compute the full documentation state for one patient.

    payer_type: "medicare" | "medicare_advantage" | "medicaid" | "commercial"
    All csv_* fields are Optional — None means not present in the CSV.
    """

    # Transfer patients: everything is pending verification
    if is_transfer:
        return DocState(
            swo="Pending — Verify",
            visit="Pending — Verify",
            pecos="Pending — Verify",
            pa="Pending — Verify",
            diagnosis="Pending — Verify",
            cascade=[
                "Patient transferred from prior supplier",
                "Verify: SWO from current prescriber, qualifying visit within 6 months, PA status with current payer",
            ],
        )

    # SWO
    swo_map = {"on file": "On File", "pending": "Pending", "expired": "Expired"}
    swo = swo_map.get((csv_swo_status or "").strip().lower(), "Pending")

    # Visit
    visit_display, visit_confirmed = _resolve_visit_display(
        csv_visit_date, confirmed_visit_date, is_transfer
    )

    # PECOS — only required for Medicare
    if payer_type in _PECOS_REQUIRED:
        pecos_val = (csv_pecos_verified or "").strip().lower()
        pecos = "Verified" if pecos_val in ("yes", "verified", "true", "1") else "Not Verified"
    else:
        pecos = "N/A"

    # PA — not required for Medicare FFS; required for MA, Medicaid, Commercial
    if payer_type in _PA_NOT_REQUIRED:
        pa = "Not Required"
    elif csv_pa_status:
        pa_map = {
            "approved": "Approved",
            "pending": "Pending",
            "denied": "Denied",
            "not required": "Not Required",
            "not started": "Required — Not Started",
        }
        pa_key = csv_pa_status.strip().lower()
        pa = pa_map.get(pa_key, "Required — Not Started")
    else:
        pa = "Required — Not Started"

    # Diagnosis
    dx_val = (csv_diagnosis_on_file or "").strip().lower()
    diagnosis = "On File" if dx_val in ("yes", "true", "on file", "1") else "Missing"

    cascade = _build_cascade(swo, visit_confirmed, pecos, pa, diagnosis, payer_type)

    return DocState(
        swo=swo,
        visit=visit_display,
        pecos=pecos,
        pa=pa,
        diagnosis=diagnosis,
        cascade=cascade,
    )