pa-decision-automation

事先授权决策自动化流水线

概述

提供确定性的代码片段和流水线方案，用于自动化事先授权 （PA）工作流：解析入站请求、提取临床特征、应用基于规则的裁定、训练 ML 分类器，以及分析拒绝模式。

用法

• 构建或调试用于 PA 受理的 X12 278 或 FHIR PAS Bundle 解析器
• 基于历史 PA 决策训练 ML 分类器，或生成 SHAP 解释
• 分析拒绝模式，以识别系统性的文档或政策缺口

核心概念

自动化方案选择

条件	方案	理由
明确的政策规则（阶梯疗法、年龄、实验室阈值）	规则引擎	可审计、确定性强、符合监管要求
标准较灵活的模糊案例	ML 分类器 + SHAP	能处理细微差异；SHAP 可提供可解释性
可用的历史决策少于 1000 条	仅使用规则引擎	数据不足，无法可靠地训练 ML 模型
多付款方部署	每个付款方使用单独的模型	各方政策不同；跨付款方模型效果不佳
输入为 X12 278（EDI）	parse_278() → 遍历分段	以竖线分隔、基于分段
输入为 FHIR PAS Bundle	parse_pas_bundle() → 提取资源	JSON，条目按资源类型组织

响应格式

• 先给出用户需要的命令或代码，再作解释
• 结构为：确认输入 → 可运行代码 → 说明关键参数 → 注意事项
• 每项任务提供一个完整可运行的示例；不展示所有替代方案
• 代码注释应简洁且实用（说明做什么，而非解释存在原因）
• 目标：50-100 行代码，附简短说明

1. PA 请求解析

1.1 解析 X12 278 事务

X12 278 医疗服务审核事务承载 PA 请求和响应。

"""Parse X12 278 prior authorization request into structured dict."""
import re
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class PA278Request:
    member_id: str = ""
    provider_npi: str = ""
    diagnosis_codes: list[str] = field(default_factory=list)
    procedure_codes: list[str] = field(default_factory=list)
    service_date: str = ""
    quantity: int = 0
    place_of_service: str = ""


def parse_278(raw: str) -> PA278Request:
    """Parse X12 278 segments into a PA278Request."""
    req = PA278Request()
    segments = raw.replace("\n", "").split("~")
    for seg in segments:
        elements = seg.strip().split("*")
        seg_id = elements[0] if elements else ""
        if seg_id == "NM1" and len(elements) > 9:
            qualifier = elements[1]
            if qualifier == "IL":  # insured/member
                req.member_id = elements[9] if len(elements) > 9 else ""
            elif qualifier == "1P":  # provider
                req.provider_npi = elements[9] if len(elements) > 9 else ""
        elif seg_id == "HI":
            for el in elements[1:]:
                parts = el.split(":")
                if len(parts) >= 2:
                    code_qualifier, code = parts[0], parts[1]
                    if code_qualifier in ("ABK", "ABF"):  # ICD-10
                        req.diagnosis_codes.append(code)
        elif seg_id == "SV1" and len(elements) > 1:
            svc_parts = elements[1].split(":")
            if len(svc_parts) >= 2:
                req.procedure_codes.append(svc_parts[1])
        elif seg_id == "DTP" and len(elements) > 3:
            if elements[1] == "472":  # service date
                req.service_date = elements[3]
    return req

1.2 解析 FHIR Da Vinci PAS Bundle

"""Extract PA fields from a FHIR Da Vinci PAS Bundle."""
import json
from dataclasses import dataclass, field


@dataclass
class PASRequest:
    member_id: str = ""
    provider_npi: str = ""
    diagnosis_codes: list[str] = field(default_factory=list)
    service_codes: list[str] = field(default_factory=list)
    supporting_info_types: list[str] = field(default_factory=list)


def parse_pas_bundle(bundle: dict) -> PASRequest:
    """Extract PA data from a FHIR PAS transaction Bundle."""
    req = PASRequest()
    resources = {
        entry["resource"]["resourceType"]: entry["resource"]
        for entry in bundle.get("entry", [])
        if "resource" in entry
    }
    # Patient / member
    patient = resources.get("Patient", {})
    for ident in patient.get("identifier", []):
        if ident.get("type", {}).get("coding", [{}])[0].get("code") == "MB":
            req.member_id = ident.get("value", "")
            break
    # Practitioner / provider NPI
    practitioner = resources.get("Practitioner", {})
    for ident in practitioner.get("identifier", []):
        if ident.get("system", "").endswith("/npi"):
            req.provider_npi = ident.get("value", "")
            break
    # Claim resource — core of the PA request
    claim = resources.get("Claim", {})
    for dx in claim.get("diagnosis", []):
        coding = dx.get("diagnosisCodeableConcept", {}).get("coding", [])
        for c in coding:
            req.diagnosis_codes.append(c.get("code", ""))
    for item in claim.get("item", []):
        svc_coding = item.get("productOrService", {}).get("coding", [])
        for c in svc_coding:
            req.service_codes.append(c.get("code", ""))
    for info in claim.get("supportingInfo", []):
        cat_coding = info.get("category", {}).get("coding", [])
        for c in cat_coding:
            req.supporting_info_types.append(c.get("code", ""))
    return req

2. 临床特征提取

2.1 PA 裁定的特征集

特征	来源	类型	描述
dx_specificity	诊断代码	int	ICD-10 代码长度（3 = 类别，4-7 = 具体代码）
drug_class	服务代码（NDC/HCPCS）	categorical	治疗类别（例如，生物制剂、阿片类药物）
prior_treatment_count	理赔历史	int	同一类别中先前尝试过的药物数量
step_therapy_complete	理赔 + 处方目录	bool	是否记录了所有必需的前置步骤
days_since_last_treatment	理赔历史	int	距上次相关治疗的间隔
lab_value_in_range	实验室结果	bool	关键实验室指标（例如，HbA1c）是否达到阈值
provider_specialty	服务提供者数据	categorical	开具申请的服务提供者的专科
place_of_service	理赔	categorical	诊室、门诊、住院、居家
prior_pa_denials	PA 历史	int	同一服务先前被拒绝的次数
documentation_score	支持信息	float	根据所需文档计算的完整度评分（0-1）

2.2 特征提取代码

"""Extract clinical features from parsed PA request + claims history."""
import pandas as pd
from datetime import datetime


def extract_features(
    pa_request: dict,
    claims_history: pd.DataFrame,
    formulary: pd.DataFrame,
    lab_results: pd.DataFrame,
) -> dict:
    """Build feature vector for PA adjudication model.

    Args:
        pa_request: Parsed PA request (from parse_278 or parse_pas_bundle).
        claims_history: Member's prior claims with columns:
            [member_id, service_date, drug_class, ndc, diagnosis_code].
        formulary: Formulary table with columns:
            [ndc, tier, step_therapy_required, prior_drugs_required].
        lab_results: Lab results with columns:
            [member_id, test_code, result_value, result_date].
    """
    member_id = pa_request.get("member_id", "")
    dx_codes = pa_request.get("diagnosis_codes", [])
    svc_codes = pa_request.get("service_codes", [])

    # Diagnosis specificity: max ICD-10 code length
    dx_specificity = max((len(c.replace(".", "")) for c in dx_codes), default=3)

    # Prior treatment count in same drug class
    requested_drug = svc_codes[0] if svc_codes else ""
    drug_info = formulary[formulary["ndc"] == requested_drug]
    drug_class = drug_info["drug_class"].iloc[0] if len(drug_info) > 0 else "unknown"
    member_claims = claims_history[claims_history["member_id"] == member_id]
    prior_treatments = member_claims[member_claims["drug_class"] == drug_class]
    prior_treatment_count = prior_treatments["ndc"].nunique()

    # Step therapy completion
    required_steps = int(drug_info["prior_drugs_required"].iloc[0]) if len(drug_info) > 0 else 0
    step_therapy_complete = prior_treatment_count >= required_steps

    # Days since last treatment in class
    if len(prior_treatments) > 0:
        last_date = pd.to_datetime(prior_treatments["service_date"]).max()
        days_since = (datetime.now() - last_date).days
    else:
        days_since = -1  # no prior treatment

    # Lab value check (example: HbA1c for diabetes drugs)
    member_labs = lab_results[lab_results["member_id"] == member_id]
    recent_lab = member_labs.sort_values("result_date", ascending=False).head(1)
    lab_in_range = bool(recent_lab["result_value"].iloc[0] >= 7.0) if len(recent_lab) > 0 else False

    # Documentation completeness score
    supporting_info = pa_request.get("supporting_info_types", [])
    required_docs = {"clinical-note", "lab-result", "treatment-history", "diagnosis"}
    doc_score = len(set(supporting_info) & required_docs) / len(required_docs)

    return {
        "dx_specificity": dx_specificity,
        "drug_class": drug_class,
        "prior_treatment_count": prior_treatment_count,
        "step_therapy_complete": step_therapy_complete,
        "days_since_last_treatment": days_since,
        "lab_value_in_range": lab_in_range,
        "documentation_score": doc_score,
    }

3. 基于规则的裁定引擎

"""Rules-based PA adjudication engine."""
from dataclasses import dataclass
from enum import Enum


class Decision(Enum):
    APPROVE = "approve"
    DENY = "deny"
    PEND = "pend"


@dataclass
class AdjudicationResult:
    decision: Decision
    reason_code: str
    reason_text: str


def adjudicate(features: dict, policy: dict) -> AdjudicationResult:
    """Apply rules-based adjudication logic.

    Args:
        features: Feature dict from extract_features().
        policy: Policy config with keys:
            - require_step_therapy (bool)
            - min_dx_specificity (int)
            - min_documentation_score (float)
            - require_lab_in_range (bool)
    """
    # Rule 1: Documentation completeness
    if features["documentation_score"] < policy.get("min_documentation_score", 0.75):
        return AdjudicationResult(
            Decision.PEND, "PEND-001", "Insufficient documentation; request additional clinical records"
        )
    # Rule 2: Diagnosis specificity
    if features["dx_specificity"] < policy.get("min_dx_specificity", 4):
        return AdjudicationResult(
            Decision.DENY, "DENY-DX", "Diagnosis code lacks required specificity"
        )
    # Rule 3: Step therapy
    if policy.get("require_step_therapy", True) and not features["step_therapy_complete"]:
        return AdjudicationResult(
            Decision.DENY, "DENY-STEP", "Step therapy requirements not met"
        )
    # Rule 4: Lab value threshold
    if policy.get("require_lab_in_range", False) and not features["lab_value_in_range"]:
        return AdjudicationResult(
            Decision.DENY, "DENY-LAB", "Required lab value not within criteria range"
        )
    # All rules passed
    return AdjudicationResult(Decision.APPROVE, "APPROVE-001", "All clinical criteria met")

4. 用于 PA 决策的 ML 分类器

4.1 训练流水线

"""Train a gradient-boosted classifier on historical PA decisions."""
import pandas as pd
from sklearn.model_selection import StratifiedKFold
from sklearn.metrics import classification_report, roc_auc_score
import xgboost as xgb


def train_pa_classifier(
    data: pd.DataFrame, target_col: str = "decision",
    feature_cols: list[str] | None = None, n_folds: int = 5,
) -> tuple[xgb.XGBClassifier, pd.DataFrame]:
    """Train XGBoost on historical PA decisions (1=approved, 0=denied)."""
    if feature_cols is None:
        feature_cols = [c for c in data.columns if c != target_col]
    X, y = data[feature_cols].copy(), data[target_col].copy()
    cat_cols = X.select_dtypes(include=["object", "category"]).columns.tolist()
    X[cat_cols] = X[cat_cols].astype("category")
    model = xgb.XGBClassifier(
        n_estimators=300, max_depth=6, learning_rate=0.05,
        subsample=0.8, colsample_bytree=0.8, enable_categorical=True,
        eval_metric="logloss", random_state=42,
    )
    cv_results = []
    for fold, (ti, vi) in enumerate(StratifiedKFold(n_folds, shuffle=True, random_state=42).split(X, y)):
        model.fit(X.iloc[ti], y.iloc[ti], eval_set=[(X.iloc[vi], y.iloc[vi])], verbose=False)
        y_prob = model.predict_proba(X.iloc[vi])[:, 1]
        cv_results.append({"fold": fold, "auc": roc_auc_score(y.iloc[vi], y_prob)})
    model.fit(X, y, verbose=False)
    return model, pd.DataFrame(cv_results)

4.2 SHAP 可解释性

"""Generate SHAP explanations for PA decisions."""
import shap


def explain_pa_decision(model, X, instance_idx: int = 0) -> dict:
    """Generate SHAP values for a single PA decision."""
    explainer = shap.TreeExplainer(model)
    shap_values = explainer.shap_values(X)
    explanation = dict(sorted(
        zip(X.columns, shap_values[instance_idx]),
        key=lambda x: abs(x[1]), reverse=True,
    ))
    return {
        "base_value": float(explainer.expected_value),
        "prediction": float(model.predict_proba(X.iloc[[instance_idx]])[:, 1][0]),
        "feature_contributions": explanation,
    }

5. 拒绝原因分析

5.1 拒绝模式分析

"""Analyze PA denial patterns to identify systemic issues."""
import pandas as pd


def analyze_denials(pa_decisions: pd.DataFrame, group_cols: list[str] | None = None) -> dict[str, pd.DataFrame]:
    """Analyze denial patterns across dimensions.

    Args:
        pa_decisions: DataFrame [pa_id, member_id, provider_npi, drug_class, denial_reason, decision, decision_date].
        group_cols: Columns to group by. Defaults to [denial_reason, drug_class, provider_npi].
    """
    denied = pa_decisions[pa_decisions["decision"] == "denied"].copy()
    if group_cols is None:
        group_cols = ["denial_reason", "drug_class", "provider_npi"]
    analyses = {}
    for col in group_cols:
        if col not in denied.columns:
            continue
        g = denied.groupby(col).agg(denial_count=("pa_id", "count"), unique_members=("member_id", "nunique")).sort_values("denial_count", ascending=False).reset_index()
        g["pct_of_denials"] = (g["denial_count"] / len(denied) * 100).round(1)
        analyses[col] = g
    if "denial_reason" in denied.columns:
        doc_gaps = denied[denied["denial_reason"].str.contains("documentation|insufficient", case=False, na=False)]
        analyses["documentation_gaps"] = doc_gaps.groupby("drug_class").agg(gap_count=("pa_id", "count")).sort_values("gap_count", ascending=False).reset_index()
    return analyses

5.2 拒绝原因代码参考

原因代码	描述	补救措施
DENY-STEP	未完成阶梯疗法	记录先前治疗及其日期
DENY-DX	诊断不够具体	使用具体程度最高的 ICD-10 代码
DENY-LAB	未满足实验室检查标准	重新提交当前实验室结果
DENY-MN	未能证明医疗必要性	提交医疗必要性说明函
DENY-EXP	实验性/研究性治疗	引用同行评审证据
PEND-001	文档不完整	提交临床记录、实验室结果和病史

6. 参数参考

6.1 PA 分类的 XGBoost 超参数

参数	默认值	范围
n_estimators	300	100–1000
max_depth	6	3–10
learning_rate	0.05	0.01–0.3
subsample	0.8	0.5–1.0
colsample_bytree	0.8	0.5–1.0
scale_pos_weight	1.0	对于类别不平衡的数据，设为负类/正类比例

6.2 文档完整性评分

文档类型	权重	适用范围
临床记录	0.30	所有 PA 申请
化验结果	0.25	有化验标准的药物 PA
治疗史	0.25	阶梯治疗药物
诊断确认	0.10	所有 PA 申请
专科转诊	0.10	专科药物

7. 常见错误

• 错误： 在类别不平衡的数据上训练 PA 分类器，却不进行校正 正确做法： 将 scale_pos_weight 设为负类/正类比例，或应用 SMOTE 平衡训练集 原因： PA 数据集通常有 70–80% 的申请获批；不进行校正时，模型会学会批准所有申请

• 错误： 纳入基于 PA 申请时尚不可用的信息派生的特征 正确做法： 确保所有特征仅使用提交前或提交时可用的数据（理赔历史，而非未来结果） 原因： 泄漏未来数据会抬高模型在训练中的表现，但在生产环境中会彻底失效

• 错误： 用一个付款方的决策训练模型，再将其用于裁定另一个付款方的申请 正确做法： 为每个付款方分别训练和验证模型，或将付款方身份作为特征，并确保每个付款方都有足够的训练数据 原因： 每个付款方都有独立的临床政策；根据付款方 A 标准训练的模型会对付款方 B 作出错误决定

• 错误： 未经临床验证就全盘接受 SHAP 解释 正确做法： 验证排名靠前的 SHAP 特征是否与付款方已发布政策中的已知临床标准一致 原因： 训练数据中的虚假相关性可能产生看似合理、但在临床上毫无意义的解释

• 错误： 通过自由文本匹配将拒绝原因字符串硬编码 正确做法： 使用标准化原因代码（CARC/RARC），并将其映射到结构化枚举 原因： 自由文本匹配很脆弱——措辞稍有变化就会破坏逻辑并导致静默故障

• 错误： 用 ML 分类器完全取代规则引擎 正确做法： 使用 ML 增强确定性政策规则；让明确的案例走规则，让有歧义的案例走 ML 原因： 可审计、可解释的决策需要确定性规则来满足监管合规要求；仅靠 ML 的决策无法通过审计辩护