claims-analytics

Claims Analytics — Pipeline Skill

Overview

Provide deterministic, copy-paste-ready Python and SQL code for healthcare claims data processing: X12 parsing, code manipulation, provider profiling, outlier detection, NCCI edit validation, and duplicate claim identification.

Usage

Parse X12 837/835 claim files into structured DataFrames
Profile provider billing patterns and detect outliers (E&M upcoding, impossible days)
Validate claims against NCCI edits and detect duplicate submissions

Core Concepts

Response Format

Lead with the command or code the user needs — explain after
Structure as: confirm inputs → working code → key parameters explained → gotchas
One complete working example per task; do not show every alternative
Keep code comments minimal and functional (what, not why-it-exists)
Target: 50-100 lines of code with brief surrounding explanation

1. X12 837 Professional Claim Parsing

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 Provider Profiling

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 Edit Validation

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. Duplicate Claim Detection

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. Benford's Law Analysis

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. Impossible Day Detection

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. Parameter Reference

Function	Key Parameter	Default	Guidance
`profile_em_distribution`	z-score threshold	2.0	Lower to 1.5 for sensitive screening
`validate_ncci`	NCCI edit file	CMS quarterly	Update quarterly from CMS
`detect_duplicates`	`fuzzy_window_days`	3	Increase to 7 for post-acute care
`benfords_test`	p-value threshold	0.01	Use 0.05 for initial screening
`detect_impossible_days`	`max_min`	1440	Reduce to 960 for stricter threshold

Common Mistakes

Wrong: Parsing X12 837 files using the newline character as segment terminator Right: Read the segment terminator from position 105 of the ISA segment (the character after the last ISA element) Why: X12 segment terminators are defined in the interchange header — they can be ~, \n, or any character; assuming ~ breaks on files using other terminators
Wrong: Comparing provider E&M distributions to a single national benchmark Right: Benchmark against same-specialty, same-region, same-payer-mix peers Why: Specialty and geography drive legitimate variation — a cardiologist's 99214 rate is naturally higher than a pediatrician's
Wrong: Flagging all exact duplicate claims as fraud Right: Distinguish true duplicates (same claim resubmitted) from legitimate corrections (different claim IDs, adjustment codes) Why: Payers routinely reprocess claims with new claim IDs after corrections — only same-claim-ID resubmissions without adjustment reason codes are suspect
Wrong: Running Benford's Law analysis on charge amounts below $10 Right: Filter to charges ≥ $10 before first-digit analysis; small charges cluster around fee-schedule amounts and naturally violate Benford's Why: Benford's Law applies to data spanning multiple orders of magnitude — low-value charges are constrained by fee schedules and produce false positives
Wrong: Using NCCI edit tables without checking effective and deletion dates Right: Filter NCCI edits to those active on the claim's date of service using effective_date <= service_date and (deletion_date IS NULL OR deletion_date > service_date) Why: NCCI edits are versioned quarterly — applying current edits to historical claims produces false violations for pairs that were valid at the time of service
Wrong: Detecting impossible days using only CPT time estimates without accounting for concurrent services Right: Distinguish sequential time-based services from concurrent ones (e.g., infusion supervision during E&M); only sum non-overlapping service minutes Why: Some services legitimately overlap (monitoring during infusion, teaching physician attestation) — naive summation inflates minutes and produces false flags

About this skill

Overview

Usage

Core Concepts

Response Format

1. X12 837 Professional Claim Parsing

2. E&M Provider Profiling

Python

SQL

3. NCCI Edit Validation

4. Duplicate Claim Detection

Python

SQL

5. Benford's Law Analysis

6. Impossible Day Detection

7. Parameter Reference

Common Mistakes