claims-analytics

Claims Analytics — パイプラインスキル

概要

医療保険請求データの処理に使用できる、決定論的でコピー＆ペーストしてすぐに使える Python および SQL コードを提供します。対象は、X12 の解析、コードの操作、プロバイダープロファイリング、外れ値検出、NCCI 編集の検証、重複請求の特定です。

使用方法

• X12 837/835 請求ファイルを構造化された DataFrame に解析します
• プロバイダーの請求パターンをプロファイリングし、外れ値を検出します（E&M の過剰なレベル設定、実現不可能な診療日数）
• NCCI 編集に照らして請求を検証し、重複送信を検出します

主な概念

---

回答形式

• まずユーザーが必要としているコマンドまたはコードを示し、その後に説明します
• 次の構成にします：入力の確認 → 動作するコード → 主なパラメーターの説明 → 注意点
• タスクごとに完全に動作する例を1つ示します。代替案をすべて示す必要はありません
• コードコメントは最小限にし、機能的な内容（何をするか）にとどめます
• 目安：コードは50〜100行、前後の説明は簡潔にします

1. X12 837 Professional 請求の解析

import pandas as pd
from typing import Generator


def parse_x12_837(filepath: str) -> Generator[dict, None, None]:
    """Parse X12 837P file into claim records.

    Yields:
        dict with claim_id, charge_amount, place_of_service, dx_codes, proc_codes,
        provider_npi, member_id, service_date.
    """
    with open(filepath, "r") as f:
        content = f.read()

    seg_term = content[105] if len(content) > 105 else "~"
    elem_sep = content[3] if len(content) > 3 else "*"
    segments = content.split(seg_term)

    claim = {}
    for seg in segments:
        el = seg.strip().split(elem_sep)
        sid = el[0] if el else ""

        if sid == "CLM":
            if claim:
                yield claim
            claim = {
                "claim_id": el[1] if len(el) > 1 else None,
                "charge_amount": float(el[2]) if len(el) > 2 else 0.0,
                "place_of_service": el[5].split(":")[0] if len(el) > 5 else None,
                "dx_codes": [], "proc_codes": [],
            }
        elif sid == "HI" and claim:
            for e in el[1:]:
                parts = e.split(":")
                if len(parts) >= 2:
                    claim["dx_codes"].append(parts[1])
        elif sid == "SV1" and claim:
            pp = el[1].split(":") if len(el) > 1 else []
            claim["proc_codes"].append({
                "cpt": pp[1] if len(pp) > 1 else None,
                "modifiers": [m for m in pp[2:6] if m] if len(pp) > 2 else [],
                "charge": float(el[2]) if len(el) > 2 else 0.0,
                "units": int(el[4]) if len(el) > 4 else 1,
            })
        elif sid == "NM1" and claim:
            q = el[1] if len(el) > 1 else ""
            if q == "82":
                claim["provider_npi"] = el[9] if len(el) > 9 else None
            elif q == "IL":
                claim["member_id"] = el[9] if len(el) > 9 else None
        elif sid == "DTP" and claim:
            if len(el) > 1 and el[1] == "472":
                claim["service_date"] = el[3] if len(el) > 3 else None
    if claim:
        yield claim

# Usage
claims = pd.json_normalize(list(parse_x12_837("claims_837p.txt")))

---

2. E&M プロバイダープロファイリング

Python

import pandas as pd
import numpy as np

EM_CODES = ["99202","99203","99204","99205","99211","99212","99213","99214","99215"]


def profile_em_distribution(claims_df: pd.DataFrame) -> pd.DataFrame:
    """Compare each provider's E&M distribution to specialty peers via z-scores."""
    em = claims_df[claims_df["proc_code"].isin(EM_CODES)].copy()

    prov_dist = em.groupby(["provider_id", "specialty", "proc_code"]).size().unstack(fill_value=0)
    prov_pct = prov_dist.div(prov_dist.sum(axis=1), axis=0)

    spec_mean = prov_pct.groupby("specialty").mean()
    spec_std = prov_pct.groupby("specialty").std()

    results = []
    for (prov, spec), row in prov_pct.iterrows():
        if spec not in spec_mean.index:
            continue
        z = (row - spec_mean.loc[spec]) / spec_std.loc[spec].replace(0, np.nan)
        results.append({
            "provider_id": prov, "specialty": spec,
            "total_em": int(prov_dist.loc[(prov, spec)].sum()),
            "pct_99214_99215": row.get("99214", 0) + row.get("99215", 0),
            "z_99214": z.get("99214", np.nan), "z_99215": z.get("99215", np.nan),
            "flag_upcoding": z.get("99214", 0) > 2.0 or z.get("99215", 0) > 2.0,
        })
    return pd.DataFrame(results).sort_values("z_99215", ascending=False)

SQL

WITH provider_em AS (
    SELECT provider_id, specialty, proc_code, COUNT(*) AS cnt
    FROM claims
    WHERE proc_code IN ('99202','99203','99204','99205','99211','99212','99213','99214','99215')
    GROUP BY provider_id, specialty, proc_code
),
provider_total AS (
    SELECT provider_id, specialty, SUM(cnt) AS total FROM provider_em GROUP BY provider_id, specialty
),
provider_pct AS (
    SELECT e.provider_id, e.specialty, e.proc_code, t.total,
           ROUND(e.cnt * 100.0 / t.total, 2) AS pct
    FROM provider_em e JOIN provider_total t ON e.provider_id = t.provider_id
),
benchmark AS (
    SELECT specialty, proc_code, AVG(pct) AS avg_pct, STDDEV(pct) AS std_pct
    FROM provider_pct GROUP BY specialty, proc_code
)
SELECT p.provider_id, p.specialty, p.proc_code, p.pct, b.avg_pct,
       ROUND((p.pct - b.avg_pct) / NULLIF(b.std_pct, 0), 2) AS z_score
FROM provider_pct p
JOIN benchmark b ON p.specialty = b.specialty AND p.proc_code = b.proc_code
WHERE ABS((p.pct - b.avg_pct) / NULLIF(b.std_pct, 0)) > 2.0
ORDER BY z_score DESC;

---

3. NCCI 編集の検証

import pandas as pd


def validate_ncci(claims_df: pd.DataFrame, ncci_edits: pd.DataFrame) -> pd.DataFrame:
    """Validate claims against NCCI edits. Returns violations.

    Args:
        claims_df: [claim_id, provider_id, member_id, service_date, proc_code, modifier_1].
        ncci_edits: [column1_cpt, column2_cpt, modifier_indicator, effective_date, deletion_date].
    """
    violations = []
    for (prov, mem, dt), grp in claims_df.groupby(["provider_id", "member_id", "service_date"]):
        codes = grp["proc_code"].tolist()
        mods = grp["modifier_1"].fillna("").tolist()

        for i, c1 in enumerate(codes):
            for j, c2 in enumerate(codes):
                if i >= j:
                    continue
                for col1, col2, mi in [(c1, c2, j), (c2, c1, i)]:
                    match = ncci_edits[
                        (ncci_edits["column1_cpt"] == col1) & (ncci_edits["column2_cpt"] == col2) &
                        (ncci_edits["effective_date"] <= dt) &
                        (ncci_edits["deletion_date"].isna() | (ncci_edits["deletion_date"] > dt))
                    ]
                    if match.empty:
                        continue
                    ind = match.iloc[0]["modifier_indicator"]
                    has_mod = mods[mi] in ("59", "XE", "XS", "XP", "XU")
                    if ind == "0" or (ind == "1" and not has_mod):
                        violations.append({
                            "provider_id": prov, "member_id": mem, "service_date": dt,
                            "column1_cpt": col1, "column2_cpt": col2,
                            "modifier_indicator": ind,
                            "status": "DENIED" if ind == "0" else "DENIED — modifier required",
                        })
    return pd.DataFrame(violations)

---

4. 重複請求の検出

Python

import pandas as pd


def detect_duplicates(claims_df: pd.DataFrame, fuzzy_window_days: int = 3) -> pd.DataFrame:
    """Detect exact and near-duplicate claims."""
    key_cols = ["member_id", "provider_id", "proc_code", "service_date"]
    exact = claims_df[claims_df.duplicated(subset=key_cols, keep=False)].copy()
    exact["dup_type"] = "exact"

    s = claims_df.sort_values(key_cols)
    s["prev_date"] = s.groupby(key_cols[:3])["service_date"].shift(1)
    s["gap"] = (s["service_date"] - s["prev_date"]).dt.days
    near = s[(s["gap"] > 0) & (s["gap"] <= fuzzy_window_days)].copy()
    near["dup_type"] = "near"

    cols = ["claim_id", "member_id", "provider_id", "proc_code", "service_date", "dup_type"]
    return pd.concat([exact[cols], near[cols]])

SQL

-- Exact duplicates
SELECT a.claim_id AS id_1, b.claim_id AS id_2, a.member_id, a.proc_code, a.service_date
FROM claims a JOIN claims b
  ON a.member_id = b.member_id AND a.provider_id = b.provider_id
  AND a.proc_code = b.proc_code AND a.service_date = b.service_date
  AND a.claim_id < b.claim_id;

-- Near-duplicates (within 3 days)
SELECT a.claim_id AS id_1, b.claim_id AS id_2, a.member_id, a.proc_code,
       a.service_date AS date_1, b.service_date AS date_2
FROM claims a JOIN claims b
  ON a.member_id = b.member_id AND a.provider_id = b.provider_id
  AND a.proc_code = b.proc_code AND a.claim_id < b.claim_id
  AND ABS(DATEDIFF(a.service_date, b.service_date)) BETWEEN 1 AND 3;

---

5. ベンフォードの法則による分析

import pandas as pd
import numpy as np
from scipy.stats import chisquare

BENFORD = {1: 0.301, 2: 0.176, 3: 0.125, 4: 0.097,
           5: 0.079, 6: 0.067, 7: 0.058, 8: 0.051, 9: 0.046}


def benfords_test(charges: pd.Series, provider_id: str = "") -> dict:
    """Test charge first-digit distribution against Benford's Law."""
    first = charges[charges > 0].apply(lambda x: int(str(x).lstrip("0").lstrip(".")[0]))
    first = first[first.between(1, 9)]
    counts = first.value_counts().sort_index()
    n = counts.sum()
    obs = np.array([counts.get(d, 0) for d in range(1, 10)])
    exp = np.array([BENFORD[d] * n for d in range(1, 10)])
    chi2, p = chisquare(obs, exp)
    return {"provider_id": provider_id, "chi2": round(chi2, 2),
            "p_value": round(p, 4), "flag": p < 0.01}

---

6. 実現不可能な診療日数の検出

import pandas as pd

CPT_MINUTES = {
    "99213": 15, "99214": 25, "99215": 40, "99203": 30, "99204": 45, "99205": 60,
    "90837": 53, "90834": 38, "97110": 15, "97140": 15, "99291": 74, "99292": 30,
}


def detect_impossible_days(claims_df: pd.DataFrame, max_min: int = 1440) -> pd.DataFrame:
    """Flag providers billing >24 hours of services in a single day."""
    df = claims_df.copy()
    df["est_min"] = df["proc_code"].map(CPT_MINUTES).fillna(0) * df["units"].fillna(1)
    daily = df.groupby(["provider_id", "service_date"]).agg(
        total_min=("est_min", "sum"), claims=("claim_id", "nunique")
    ).reset_index()
    return daily[daily["total_min"] > max_min].sort_values("total_min", ascending=False)

---

7. パラメーターリファレンス

関数	主なパラメーター	デフォルト	ガイダンス
profile_em_distribution	z スコアのしきい値	2.0	感度を高めてスクリーニングする場合は1.5に下げます
validate_ncci	NCCI 編集ファイル	CMS quarterly	CMS から四半期ごとに更新します
detect_duplicates	fuzzy_window_days	3	急性期後ケアでは7に増やします
benfords_test	p 値のしきい値	0.01	初回のスクリーニングでは0.05を使用します
detect_impossible_days	max_min	1440	より厳しいしきい値にするには960に下げます

---

よくある間違い

• 誤り： 改行文字をセグメント終端として、X12 837 ファイルを解析する 正しい方法： ISA セグメントの位置 105（最後の ISA 要素の後の文字）からセグメント終端を読み取ります 理由： X12 のセグメント終端はインターチェンジヘッダーで定義されており、~、\n、その他任意の文字を使用できます。~ と決めつけると、別の終端文字を使用するファイルで問題が発生します

• 誤り： プロバイダーの E&M 分布を、単一の全国ベンチマークと比較する 正しい方法： 同じ専門分野、同じ地域、同じ支払者構成の同業者をベンチマークにします 理由： 専門分野と地域によって正当なばらつきが生じます。心臓専門医の 99214 の割合が小児科医より自然に高くなるのはそのためです

• 誤り： 完全に重複する請求をすべて不正としてフラグ付けする 正しい方法： 真の重複（同じ請求の再提出）と正当な訂正（異なる請求 ID、調整コード）を区別します 理由： 支払者は訂正後に新しい請求 ID で請求を再処理することがよくあります。疑わしいのは、調整理由コードのない同一請求 ID の再提出だけです

• 誤り： 10 ドル未満の請求額にベンフォードの法則の分析を適用する 正しい方法： 先頭桁の分析を行う前に、請求額を 10 ドル以上に絞り込みます。少額の請求は料金表の金額付近に集中し、ベンフォードの法則に自然に反します 理由： ベンフォードの法則は、複数の桁にわたるデータに適用されます。少額の請求額は料金表によって制約され、誤検出を生みます

• 誤り： 有効日と削除日を確認せずに NCCI 編集テーブルを使用する 正しい方法： effective_date <= service_date および (deletion_date IS NULL OR deletion_date > service_date) を使って、請求のサービス日に有効な NCCI 編集に絞り込みます 理由： NCCI 編集は四半期ごとにバージョン管理されています。現在の編集を過去の請求に適用すると、サービス当時は有効だったペアが誤って違反と判定されます

• 誤り： 同時に行われるサービスを考慮せず、CPT の時間推定値だけで実現不可能な日程を検出する 正しい方法： 時間ベースのサービスを、順次行われるものと同時に行われるもの（例：E&M 中の注入療法の監督）に区別し、重複しないサービス時間だけを合計します 理由： 一部のサービスは正当に重複します（注入中のモニタリング、指導医の確認など）。単純に合計すると時間が過大になり、誤検出につながります