关于 claims-analytics
claims-analytics 是面向医疗理赔数据处理的 Pipeline Skill，为分析人员提供 Python 和 SQL 示例，涵盖 X12 解析、提供者画像、异常检测及理赔校验。
主要功能
- 01解析 X12 837 理赔文件
- 02按专科同业分析 E&M 分布
- 03验证 NCCI 编辑规则
- 04检测精确和近似重复理赔
- 05Benford 首位数分析
- 06识别不可能服务时长
适用场景
- 01将 X12 837 文件解析为结构化记录
- 02筛查提供者 E&M 编码异常
- 03核验理赔是否违反 NCCI 编辑
- 04查找重复提交及异常服务时长
安装
复制下面的命令，在终端中运行。
$
npx skills add https:/安装后在 Claude Code 中输入/claims-analytics确认已加载。
详细介绍
理赔分析 — 流水线技能
概述
提供确定性、可直接复制粘贴的 Python 和 SQL 代码，用于医疗理赔数据处理：X12 解析、代码操作、提供者画像分析、异常值检测、NCCI 编辑验证和重复理赔识别。
用法
- 将 X12 837/835 理赔文件解析为结构化 DataFrame
- 分析提供者的账单模式并检测异常值（E&M 编码升级、每日服务量不可能）
- 根据 NCCI 编辑验证理赔并检测重复提交
核心概念
响应格式
- 优先给出用户需要的命令或代码，然后再解释
- 按以下结构组织：确认输入 → 可运行代码 → 解释关键参数 → 注意事项
- 每项任务提供一个完整可运行的示例；不要列出所有替代方案
- 代码注释保持精简且实用（说明做什么，而非解释存在原因）
- 目标：50–100 行代码，并附简短说明
1. X12 837 专业理赔解析
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 季度版 | 每季度从 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 的收费金额进行 Benford 定律分析 正确： 在分析首位数字前筛选出 ≥ $10 的收费；小额收费会聚集在费用表金额附近，因此自然不符合 Benford 定律 原因： Benford 定律适用于跨越多个数量级的数据；低价值收费受费用表限制，会产生误报
-
错误： 使用 NCCI 编辑表，却不检查生效日期和删除日期 正确： 根据索赔的服务日期筛选当时有效的 NCCI 编辑项，使用
effective_date <= service_date和(deletion_date IS NULL OR deletion_date > service_date)原因： NCCI 编辑项按季度进行版本管理；将当前编辑项应用于历史索赔，会错误地判定当时有效的组合存在违规 -
错误： 仅使用 CPT 时间估算检测不可能的服务时长，却未考虑并行服务 正确： 区分按顺序进行的计时服务与并行服务（例如，E&M 期间的输液监督）；只合计互不重叠的服务分钟数 原因： 有些服务会合理地重叠（输液期间的监测、教学医师签注）；简单相加会夸大分钟数并产生误报
评论
支持 Markdown