392 lines
15 KiB
Python
392 lines
15 KiB
Python
"""
|
|
coverage_calculator.py
|
|
Signal CGM — STTIL Solutions
|
|
|
|
Calculates CGM coverage status per patient based on shipment history
|
|
and payer-specific wear-day rules.
|
|
|
|
PHI CONTRACT:
|
|
This module receives only: patient_id, device_type, shipment_date,
|
|
quantity, payer, component, and optional doc status fields.
|
|
No names, SSNs, DOBs, or contact fields may be added to any function
|
|
signature or data structure in this file.
|
|
|
|
Flag types emitted:
|
|
SUPPLY_LAPSED — supply cycle ended, no new shipment
|
|
VISIT_REQUIRED — visit date past due, no confirmed new visit
|
|
TRANSFER_PENDING — transferred patient, all docs need verification
|
|
RENEWAL_CRITICAL — <= 45 days to next visit due
|
|
RENEWAL_ELEVATED — <= 60 days to next visit due
|
|
RENEWAL_SOON — <= 90 days to next visit due
|
|
RESUPPLY_READY — within refill window, visit not yet urgent
|
|
ACTIVE — all clear
|
|
NO_RECENT_SHIPMENT — last shipment 12+ months ago, no confirmed active care;
|
|
verify patient status with prescriber office
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from datetime import date, timedelta
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
RULE_VERSION = "v0.2"
|
|
|
|
PAYER_RULES_PATH = Path(__file__).parent.parent / "config" / "payer_rules.json"
|
|
|
|
|
|
class CoverageFlag(str, Enum):
|
|
SUPPLY_LAPSED = "SUPPLY_LAPSED" # supply cycle ended, no new shipment
|
|
VISIT_REQUIRED = "VISIT_REQUIRED" # visit date past due, no confirmed new visit
|
|
TRANSFER_PENDING = "TRANSFER_PENDING" # transferred patient, all docs need verification
|
|
RENEWAL_CRITICAL = "RENEWAL_CRITICAL" # <= 45 days to next visit due
|
|
RENEWAL_ELEVATED = "RENEWAL_ELEVATED" # <= 60 days to next visit due
|
|
RENEWAL_SOON = "RENEWAL_SOON" # <= 90 days to next visit due
|
|
RESUPPLY_READY = "RESUPPLY_READY" # within refill window, visit not yet urgent
|
|
ACTIVE = "ACTIVE" # all clear
|
|
NO_RECENT_SHIPMENT = "NO_RECENT_SHIPMENT" # last shipment 12+ months ago, no confirmed active care
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ShipmentRecord:
|
|
"""
|
|
Minimal shipment record. Only non-PHI fields allowed.
|
|
|
|
patient_id: Supplier's internal MRN or account number.
|
|
This is the sole crosswalk key — no real identity data here.
|
|
csv_* fields: Optional doc status columns parsed from the CSV.
|
|
csv_transfer_from: Non-empty string means this is a transferred patient.
|
|
"""
|
|
patient_id: str
|
|
device_type: str
|
|
shipment_date: date
|
|
quantity: int
|
|
payer: str
|
|
component: str = "sensor"
|
|
# Optional doc fields — populated when CSV contains these columns
|
|
csv_visit_date: Optional[date] = None
|
|
csv_swo_status: Optional[str] = None # "On File" | "Pending" | "Expired"
|
|
csv_pecos_verified: Optional[str] = None # "Yes" | "No"
|
|
csv_pa_status: Optional[str] = None # "Approved" | "Pending" | "Denied" | "Not Required"
|
|
csv_diagnosis_on_file: Optional[str] = None # "Yes" | "No"
|
|
csv_transfer_from: Optional[str] = None # prior supplier name if transfer
|
|
order_number: Optional[str] = None # pass-through — billing system order/claim number
|
|
hcpcs: Optional[str] = None # pass-through — HCPCS code from supplier CSV
|
|
|
|
|
|
@dataclass
|
|
class CoverageResult:
|
|
patient_id: str
|
|
device_type: str
|
|
payer: str
|
|
component: str
|
|
last_shipment_date: date
|
|
coverage_end_date: date
|
|
next_visit_due_date: Optional[date]
|
|
flag: CoverageFlag
|
|
days_until_coverage_end: int
|
|
days_until_visit_due: Optional[int]
|
|
priority_score: int
|
|
visit_date_confidence: str = "estimated" # "confirmed" | "estimated"
|
|
is_transfer: bool = False
|
|
rule_version: str = RULE_VERSION
|
|
|
|
|
|
def _load_payer_rules() -> dict:
|
|
with open(PAYER_RULES_PATH, "r") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def _get_wear_days(rules: dict, device_type: str, component: str) -> int:
|
|
"""
|
|
Return wear days for a given device type and component.
|
|
Raises ValueError for unknown device types.
|
|
"""
|
|
devices = rules.get("devices", {})
|
|
device = devices.get(device_type)
|
|
if device is None:
|
|
raise ValueError(f"Unknown device_type: '{device_type}'. "
|
|
f"Valid types: {list(devices.keys())}")
|
|
|
|
component_key = f"{component}_wear_days"
|
|
wear_days = device.get(component_key)
|
|
if wear_days is None:
|
|
raise ValueError(
|
|
f"Device '{device_type}' has no wear-day rule for component "
|
|
f"'{component}'. Check payer_rules.json."
|
|
)
|
|
return wear_days
|
|
|
|
|
|
_COMMERCIAL_KEYWORDS = {
|
|
"aetna", "united", "unitedhealthcare", "cigna", "humana",
|
|
"bcbs", "blue cross", "blue shield", "anthem", "molina",
|
|
"centene", "wellcare", "oscar", "ambetter", "carefirst",
|
|
"highmark", "geisinger", "kaiser", "commercial",
|
|
}
|
|
|
|
|
|
def _normalize_payer(payer: str) -> str:
|
|
"""
|
|
Map raw DME CSV payer strings to canonical payer_rules.json keys.
|
|
|
|
"Medicare Part B" → "medicare"
|
|
"Medicare Advantage" → "medicare_advantage"
|
|
"Medicaid - GA" → "medicaid"
|
|
"Aetna PPO" → "commercial"
|
|
|
|
Preserves original value in CoverageResult.payer for display.
|
|
"""
|
|
s = payer.strip().lower()
|
|
if "medicare advantage" in s or "medicare_advantage" in s:
|
|
return "medicare_advantage"
|
|
if "medicare" in s:
|
|
return "medicare"
|
|
if "tricare for life" in s or s == "tfl":
|
|
return "medicare"
|
|
if "medicaid" in s:
|
|
return "medicaid"
|
|
if "tricare" in s:
|
|
return "commercial"
|
|
for keyword in _COMMERCIAL_KEYWORDS:
|
|
if keyword in s:
|
|
return "commercial"
|
|
return s
|
|
|
|
|
|
def _get_payer_config(rules: dict, payer: str) -> dict:
|
|
payer_rules = rules.get("payer_rules", {})
|
|
normalized = _normalize_payer(payer)
|
|
return payer_rules.get(normalized, payer_rules.get("default", {}))
|
|
|
|
|
|
def _resolve_visit_date(
|
|
record: ShipmentRecord,
|
|
confirmed_visit_date: Optional[date] = None,
|
|
) -> tuple[Optional[date], str]:
|
|
"""
|
|
Visit date priority chain:
|
|
1. CSV visit date column (if present + valid)
|
|
2. Supabase confirmed visit date (stored by staff)
|
|
3. Estimated proxy: shipment_date - 30 days
|
|
Returns (resolved_date, confidence) where confidence is 'confirmed' or 'estimated'.
|
|
"""
|
|
if record.csv_visit_date:
|
|
return record.csv_visit_date, "confirmed"
|
|
if confirmed_visit_date:
|
|
return confirmed_visit_date, "confirmed"
|
|
estimated = record.shipment_date - timedelta(days=30)
|
|
return estimated, "estimated"
|
|
|
|
|
|
def _compute_visit_flag(
|
|
next_visit_due: date,
|
|
today: date,
|
|
) -> CoverageFlag:
|
|
"""Map days-to-next-visit to the appropriate tier flag."""
|
|
days = (next_visit_due - today).days
|
|
if days < 0:
|
|
return CoverageFlag.VISIT_REQUIRED
|
|
if days <= 45:
|
|
return CoverageFlag.RENEWAL_CRITICAL
|
|
if days <= 60:
|
|
return CoverageFlag.RENEWAL_ELEVATED
|
|
if days <= 90:
|
|
return CoverageFlag.RENEWAL_SOON
|
|
return CoverageFlag.ACTIVE
|
|
|
|
|
|
def _compute_priority(flag: CoverageFlag, days_until_visit: Optional[int]) -> int:
|
|
"""
|
|
Priority score for worklist sort. Higher = more urgent.
|
|
Supply Lapsed and Transfer Pending are always highest.
|
|
Within visit tiers, urgency increases as days decrease.
|
|
"""
|
|
urgency_days = abs(days_until_visit) if days_until_visit is not None else 0
|
|
if flag == CoverageFlag.SUPPLY_LAPSED:
|
|
return 2000 + urgency_days
|
|
if flag == CoverageFlag.TRANSFER_PENDING:
|
|
return 1800
|
|
if flag == CoverageFlag.VISIT_REQUIRED:
|
|
return 1500 + urgency_days
|
|
if flag == CoverageFlag.RENEWAL_CRITICAL:
|
|
return 1000 + (45 - max(0, urgency_days))
|
|
if flag == CoverageFlag.RENEWAL_ELEVATED:
|
|
return 700 + (60 - max(0, urgency_days))
|
|
if flag == CoverageFlag.RENEWAL_SOON:
|
|
return 400 + (90 - max(0, urgency_days))
|
|
if flag == CoverageFlag.RESUPPLY_READY:
|
|
return 200
|
|
return 0 # ACTIVE
|
|
|
|
|
|
def calculate_coverage(
|
|
record: ShipmentRecord,
|
|
as_of: Optional[date] = None,
|
|
confirmed_visit_date: Optional[date] = None,
|
|
) -> CoverageResult:
|
|
"""
|
|
Calculate coverage status for a single shipment record.
|
|
|
|
Args:
|
|
record: ShipmentRecord with non-PHI fields only.
|
|
as_of: Date to evaluate against. Defaults to today.
|
|
confirmed_visit_date: Confirmed visit date from Supabase (staff-entered).
|
|
"""
|
|
rules = _load_payer_rules()
|
|
today = as_of or date.today()
|
|
|
|
# Transferred patients: all docs need verification before anything else
|
|
if record.csv_transfer_from:
|
|
return CoverageResult(
|
|
patient_id=record.patient_id,
|
|
device_type=record.device_type,
|
|
payer=record.payer,
|
|
component=record.component,
|
|
last_shipment_date=record.shipment_date,
|
|
coverage_end_date=record.shipment_date,
|
|
next_visit_due_date=None,
|
|
flag=CoverageFlag.TRANSFER_PENDING,
|
|
days_until_coverage_end=0,
|
|
days_until_visit_due=None,
|
|
priority_score=1800,
|
|
visit_date_confidence="estimated",
|
|
is_transfer=True,
|
|
rule_version=RULE_VERSION,
|
|
)
|
|
|
|
wear_days = _get_wear_days(rules, record.device_type, record.component)
|
|
payer_config = _get_payer_config(rules, record.payer)
|
|
|
|
total_wear_days = wear_days * record.quantity
|
|
coverage_end = record.shipment_date + timedelta(days=total_wear_days)
|
|
days_until_end = (coverage_end - today).days
|
|
|
|
# Catch-all FIRST: patient shows no active-care signals and last shipment was
|
|
# 12+ months ago. Must precede the supply-lapsed check — a 12-month-stale record
|
|
# always reads as lapsed, so checking lapsed first made this flag unreachable.
|
|
# A confirmed or CSV visit date within the last 12 months overrides this check,
|
|
# treating the patient as still active even if the shipment record is old.
|
|
NO_RECENT_SHIPMENT_DAYS = 365
|
|
days_since_shipment = (today - record.shipment_date).days
|
|
confirmed_visit_within_12m = (
|
|
confirmed_visit_date is not None
|
|
and (today - confirmed_visit_date).days < NO_RECENT_SHIPMENT_DAYS
|
|
)
|
|
csv_visit_within_12m = (
|
|
record.csv_visit_date is not None
|
|
and (today - record.csv_visit_date).days < NO_RECENT_SHIPMENT_DAYS
|
|
)
|
|
if (
|
|
days_since_shipment >= NO_RECENT_SHIPMENT_DAYS
|
|
and not confirmed_visit_within_12m
|
|
and not csv_visit_within_12m
|
|
):
|
|
_, confidence = _resolve_visit_date(record, confirmed_visit_date)
|
|
return CoverageResult(
|
|
patient_id=record.patient_id,
|
|
device_type=record.device_type,
|
|
payer=record.payer,
|
|
component=record.component,
|
|
last_shipment_date=record.shipment_date,
|
|
coverage_end_date=coverage_end,
|
|
next_visit_due_date=None,
|
|
flag=CoverageFlag.NO_RECENT_SHIPMENT,
|
|
days_until_coverage_end=days_until_end,
|
|
days_until_visit_due=None,
|
|
priority_score=5, # lowest — drops to bottom of worklist sort
|
|
visit_date_confidence=confidence,
|
|
rule_version=RULE_VERSION,
|
|
)
|
|
|
|
# Supply lapsed — coverage cycle ended
|
|
if days_until_end < 0:
|
|
visit_date, confidence = _resolve_visit_date(record, confirmed_visit_date)
|
|
next_visit_due = visit_date + timedelta(days=180) if visit_date else None
|
|
days_visit = (next_visit_due - today).days if next_visit_due else None
|
|
return CoverageResult(
|
|
patient_id=record.patient_id,
|
|
device_type=record.device_type,
|
|
payer=record.payer,
|
|
component=record.component,
|
|
last_shipment_date=record.shipment_date,
|
|
coverage_end_date=coverage_end,
|
|
next_visit_due_date=next_visit_due,
|
|
flag=CoverageFlag.SUPPLY_LAPSED,
|
|
days_until_coverage_end=days_until_end,
|
|
days_until_visit_due=days_visit,
|
|
priority_score=_compute_priority(CoverageFlag.SUPPLY_LAPSED, days_visit),
|
|
visit_date_confidence=confidence,
|
|
rule_version=RULE_VERSION,
|
|
)
|
|
|
|
visit_renewal_days = payer_config.get("visit_renewal_days")
|
|
refill_window_days = payer_config.get("refill_window_days", 30)
|
|
|
|
# Resolve visit date using priority chain
|
|
visit_date, confidence = _resolve_visit_date(record, confirmed_visit_date)
|
|
next_visit_due: Optional[date] = None
|
|
days_until_visit: Optional[int] = None
|
|
flag = CoverageFlag.ACTIVE
|
|
|
|
if visit_renewal_days and visit_date:
|
|
next_visit_due = visit_date + timedelta(days=visit_renewal_days)
|
|
days_until_visit = (next_visit_due - today).days
|
|
flag = _compute_visit_flag(next_visit_due, today)
|
|
|
|
# If no visit urgency but within refill window, mark resupply ready
|
|
if flag == CoverageFlag.ACTIVE and days_until_end <= refill_window_days:
|
|
flag = CoverageFlag.RESUPPLY_READY
|
|
|
|
priority = _compute_priority(flag, days_until_visit)
|
|
|
|
return CoverageResult(
|
|
patient_id=record.patient_id,
|
|
device_type=record.device_type,
|
|
payer=record.payer,
|
|
component=record.component,
|
|
last_shipment_date=record.shipment_date,
|
|
coverage_end_date=coverage_end,
|
|
next_visit_due_date=next_visit_due,
|
|
flag=flag,
|
|
days_until_coverage_end=days_until_end,
|
|
days_until_visit_due=days_until_visit,
|
|
priority_score=priority,
|
|
visit_date_confidence=confidence,
|
|
rule_version=RULE_VERSION,
|
|
)
|
|
|
|
|
|
def calculate_batch(
|
|
records: list[ShipmentRecord],
|
|
as_of: Optional[date] = None,
|
|
confirmed_visits: Optional[dict[str, date]] = None,
|
|
) -> list[CoverageResult]:
|
|
"""
|
|
Calculate coverage for a list of shipment records and return a
|
|
worklist sorted by priority (highest first).
|
|
|
|
confirmed_visits: dict mapping patient_id_hash -> confirmed visit date.
|
|
Loaded from Supabase by persistence layer before calling this function.
|
|
|
|
Skips records that raise ValueError (unknown device/component) and
|
|
logs a warning so the batch continues.
|
|
"""
|
|
import hashlib
|
|
confirmed_visits = confirmed_visits or {}
|
|
results = []
|
|
for record in records:
|
|
patient_hash = hashlib.sha256(record.patient_id.encode()).hexdigest()
|
|
confirmed_date = confirmed_visits.get(patient_hash)
|
|
try:
|
|
result = calculate_coverage(record, as_of=as_of, confirmed_visit_date=confirmed_date)
|
|
results.append(result)
|
|
except ValueError as exc:
|
|
logger.warning("Skipping record for patient_id hash: %s", exc)
|
|
|
|
results.sort(key=lambda r: r.priority_score, reverse=True)
|
|
return results
|