pa-decision-automation

事前承認判定自動化パイプライン

概要

事前承認（PA）ワークフローを自動化するための、決定論的なコードスニペットとパイプラインのレシピを提供します。受信リクエストの解析、臨床的特徴の抽出、ルールベースの判定、ML 分類器のトレーニング、否認パターンの分析を行います。

使用方法

• PA 受付用の X12 278 または FHIR PAS バンドルパーサーの構築やデバッグ
• 過去の PA 判定に基づく ML 分類器のトレーニング、または SHAP 説明の生成
• 否認パターンを分析し、体系的な文書化の不足やポリシー上の問題を特定

主な概念

自動化アプローチの選択

条件	アプローチ	理由
明確なポリシールール（段階的治療、年齢、検査値のしきい値）	ルールエンジン	監査可能で決定論的、規制要件にも適合
基準が柔軟な曖昧なケース	ML 分類器 + SHAP	微妙な違いに対応し、SHAP が説明可能性を提供
利用できる過去の判定が 1000 件未満	ルールエンジンのみ	信頼性の高い ML トレーニングにはデータが不十分
複数の支払者への導入	支払者ごとに個別のモデル	ポリシーが異なるため、支払者横断モデルでは適切に機能しない
入力が X12 278（EDI）	parse_278() → セグメントの反復処理	パイプ区切り、セグメントベース
入力が FHIR PAS Bundle	parse_pas_bundle() → リソースの抽出	JSON、リソース型のエントリー

応答形式

• 必要なコマンドまたはコードを最初に提示し、その後に説明する
• 次の構成にする：入力の確認 → 動作するコード → 主なパラメーターの説明 → 注意点
• タスクごとに完全に動作する例を 1 つ提示し、すべての代替案は示さない
• コードコメントは最小限かつ実用的にする（存在理由ではなく、機能を説明する）
• 目安：コード 50〜100 行と、簡潔な説明

1. PA リクエストの解析

1.1 X12 278 トランザクションの解析

X12 278 Health Care Services Review トランザクションは、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 バンドルの解析

"""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 だけでは監査に耐えられません