Recipe 3.6: Python Implementation Example

pip install boto3 scikit-learn pandas numpy networkx python-louvain

import hashlib
import hmac
import json
import logging
import uuid
from collections import Counter, defaultdict
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from typing import Optional

import boto3
import networkx as nx
import numpy as np
import pandas as pd
from botocore.config import Config
from boto3.dynamodb.conditions import Key
from sklearn.ensemble import IsolationForest

# Louvain community detection. Import guarded because python-louvain is an
# optional teaching dependency. In production on Neptune, community detection
# runs as a Neptune ML algorithm or a precomputed batch job, not in-process.
try:
    import community as community_louvain  # python-louvain
    HAS_LOUVAIN = True
except ImportError:
    HAS_LOUVAIN = False

# Visible when running this file directly; Lambda configures its own handler
# and this becomes a no-op there.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)

# Structured logging. Ship JSON records to CloudWatch Logs Insights. Claim,
# provider, and patient data is PHI-adjacent (an NPI plus a date range plus a
# patient population is re-identifying even without names). Log structural
# metadata only. Never log full claim bodies, patient identifiers, ownership
# chains, or raw Comprehend Medical output in application logs.
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Adaptive retry mode handles throttling across DynamoDB, OpenSearch,
# SageMaker Feature Store, Comprehend Medical, and Bedrock with exponential
# backoff and jitter. Monthly graph refresh plus daily scoring is naturally
# bursty, and adaptive mode keeps burst windows from cascading into retry
# storms. Setting a higher max on Bedrock specifically would be reasonable,
# since throttles there are model-capacity bound.
BOTO3_RETRY_CONFIG = Config(retries={"max_attempts": 5, "mode": "adaptive"})

REGION = "us-east-1"
dynamodb = boto3.resource("dynamodb", region_name=REGION, config=BOTO3_RETRY_CONFIG)
s3_client = boto3.client("s3", region_name=REGION, config=BOTO3_RETRY_CONFIG)
cloudwatch = boto3.client("cloudwatch", region_name=REGION, config=BOTO3_RETRY_CONFIG)
sns = boto3.client("sns", region_name=REGION, config=BOTO3_RETRY_CONFIG)
eventbridge = boto3.client("events", region_name=REGION, config=BOTO3_RETRY_CONFIG)
featurestore_runtime = boto3.client(
    "sagemaker-featurestore-runtime", region_name=REGION, config=BOTO3_RETRY_CONFIG
)
comprehend_medical = boto3.client("comprehendmedical", region_name=REGION, config=BOTO3_RETRY_CONFIG)
bedrock_runtime = boto3.client("bedrock-runtime", region_name=REGION, config=BOTO3_RETRY_CONFIG)

# --- Resource Names ---
# Fill in with your actual resource names. These are placeholders.
RESOLVED_ENTITIES_TABLE = "resolved-entities"
CASE_STATE_TABLE = "case-state"
# NOTE: The suppression-registry DynamoDB table (described in the main recipe)
# is not exercised in this teaching example. The suppression workflow requires
# the case management UI to capture suppression_requested and suppression_window
# from the investigator at outcome-capture time, plus a per-flag pre-emit check
# in the rules and statistical layers. Both are out of scope for the sketchpad
# version; see the main recipe's Alert Fatigue subsection for the design.

PROVIDER_FEATURES_FG = "provider-features"
PEER_GROUP_BASELINES_FG = "peer-group-baselines"

CLAIMS_LAKE_BUCKET = "my-claims-lake"
RESOLVED_ENTITIES_BUCKET = "my-resolved-entities"
MODEL_ARTIFACTS_BUCKET = "my-fwa-model-artifacts"
CASE_OUTCOMES_BUCKET = "my-fwa-case-outcomes"
SUBGRAPH_ARTIFACTS_BUCKET = "my-fwa-subgraph-exports"

# Customer-managed KMS key ARN for the case-outcomes bucket. Separate key per
# bucket so rotation and grants can be scoped independently. The labels-bucket
# key gets stricter access policy than the general-claims-lake key because
# labels carry adjudication outcomes subject to discovery in FCA proceedings.
CASE_OUTCOMES_CMK_ARN = "arn:aws:kms:us-east-1:123456789012:key/YOUR-KEY-ID-HERE"

# Stable salt for patient-ID hashing in graph nodes. In production, load this
# from Secrets Manager at module init and rotate on a documented schedule
# (annually or per SIU policy). A static placeholder here for teaching only.
PATIENT_HASH_SALT = b"REPLACE-WITH-SECRETS-MANAGER-VALUE"

FWA_FLAGS_BUS = "fwa-flags"
FWA_WORKFLOW_BUS = "fwa-workflow"
INVESTIGATOR_NOTIFICATION_TOPIC_ARN = (
    "arn:aws:sns:us-east-1:123456789012:fwa-investigator-notifications"
)
BEDROCK_MODEL_ID = "anthropic.claude-3-sonnet-20240229-v1:0"

# --- Thresholds ---
# These are teaching defaults. Real deployments tune them with SIU leadership
# against a labeled case stream and revisit them quarterly. A threshold too
# tight drops real cases; too loose buries investigators under noise.
PEER_ZSCORE_FLAG = 3.0                  # statistical layer flag threshold
PEER_ZSCORE_HIGH_SEVERITY = 5.0         # push to high-severity queue
CUSUM_H = 5.0                           # CUSUM decision threshold (self-history drift)
CUSUM_K = 0.5                           # CUSUM slack parameter, in standard deviations
ISOLATION_FOREST_CONTAMINATION = 0.02   # expected fraction of outlier providers
REFERRAL_CONCENTRATION_FLAG = 0.35      # > 35% of referrals to one downstream provider
HIGH_SEVERITY_EXPOSURE = Decimal("100000.00")  # dollar exposure cut for high severity
CRITICAL_EXPOSURE = Decimal("500000.00")       # dollar exposure cut for immediate escalation

# --- Code Families ---
# Real CPT/HCPCS code identifiers used illustratively. A production deployment
# loads these from the published CMS files and refreshes quarterly. Code set
# drift (ICD-10-CM annual updates in October, CPT updates in January, HCPCS
# quarterly) will silently invalidate rules that reference retired codes.
EM_OFFICE_VISIT_CODES = {"99202", "99203", "99204", "99205", "99212", "99213", "99214", "99215"}
HIGH_LEVEL_EM_CODES = {"99205", "99215"}          # level 5 E&M codes
DRUG_TEST_CODES = {"80305", "80306", "80307"}     # urine drug test code family
DME_OXYGEN_CODES = {"E1390", "E0431"}             # oxygen equipment HCPCS codes
MODIFIER_25 = "25"                                # "significant separate E&M service" modifier

# CCI (Correct Coding Initiative) edit pairs. Illustrative only. Production
# loads the full CMS NCCI PTP edit file (tens of thousands of pairs) and
# refreshes quarterly.
CCI_EDIT_PAIRS = {
    # (primary_code, secondary_code): (edit_type, default_allowed)
    ("99213", "99214"): ("mutually_exclusive", False),
    ("36415", "36416"): ("mutually_exclusive", False),
}

# MUE (Medically Unlikely Edit) thresholds per code per date of service.
# Illustrative; production loads from the CMS MUE files.
MUE_THRESHOLDS = {
    "E1390": 1,   # one oxygen concentrator per day is already generous
    "80307": 1,   # one drug screen per day
    "99213": 1,   # one level-3 E&M per patient per day
}

def _to_decimal(value, precision="0.01"):
    """Convert numeric input to Decimal for DynamoDB storage.

    DynamoDB rejects Python float for numeric attributes because float
    arithmetic introduces rounding drift that makes threshold comparisons
    unreliable over time. Always pass dollar amounts, rates, and z-scores
    through Decimal on the way in and back out.
    """
    if value is None:
        return None
    return Decimal(str(value)).quantize(Decimal(precision))

def _floats_to_decimal(obj):
    """Recursively convert Python floats in a dict/list tree to Decimal.

    DynamoDB rejects Python float for numeric attributes; the resource-API
    serializer raises TypeError at put time. Apply this to any structured
    payload at the put-item boundary so flag dicts produced by sklearn or
    by Python ratio math can be persisted without per-call-site coercion.
    """
    if isinstance(obj, float):
        return Decimal(str(obj))
    if isinstance(obj, dict):
        return {k: _floats_to_decimal(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_floats_to_decimal(v) for v in obj]
    return obj

def _hash_patient_id(patient_id, salt=None):
    """Stable HMAC-SHA256 hash for patient identifiers used as graph node IDs.

    The FWA graph crosses provider, payment, and ownership data with patient
    connections. An investigator (or a downstream system, or a subpoena
    response) traversing the graph should see the structural relationship,
    not a re-identifiable patient pointer. The hash is one-way for the
    pipeline's purposes; a downstream system that needs to look up the
    underlying patient (the case management UI surfacing the patient's full
    record to an investigator with appropriate access) does the reverse
    lookup through the patient master, not by reversing the hash.

    The salt must be stable across runs so the same patient produces the same
    node ID across batches; rotate on a long cadence (annually or per SIU
    policy), not per-run.
    """
    if salt is None:
        salt = PATIENT_HASH_SALT
    return hmac.new(salt, patient_id.encode("utf-8"),
                    hashlib.sha256).hexdigest()[:16]

def _redact_for_logs(canonical_claim):
    """Produce a log-safe structural summary of a claim.

    Keeps claim_id, provider_id, service_date, and CPT set. Drops patient
    identifiers, billed amounts, and diagnosis codes to minimize PHI in
    CloudWatch Logs. Full claim bodies stay in S3 under the claims lake.
    """
    return {
        "claim_id": canonical_claim.get("claim_id"),
        "rendering_provider": canonical_claim.get("rendering_provider_npi"),
        "service_date": canonical_claim.get("service_date"),
        "num_lines": len(canonical_claim.get("lines", [])),
    }

def normalize_claim(raw_claim):
    """Convert a raw payer/clearinghouse claim into the canonical shape used
    by every downstream step.

    In production, the raw_claim argument is the output of an X12 837 parser
    (for payer claims) or a native EHR/adjudication-system export. This
    example accepts a dict in the approximate shape that parser output takes
    so you can see the normalization logic without also reading a 600-page
    X12 specification.
    """
    # Normalize the service date. Timezones are the first silent bug class in
    # claims pipelines: claims span a whole day, so service_date should always
    # be a date (no time, no tz). Billing jurisdictions with business-day
    # nuances (nursing homes, inpatient) need separate handling outside scope.
    raw_date = raw_claim.get("service_date")
    if isinstance(raw_date, str):
        service_date = datetime.strptime(raw_date, "%Y-%m-%d").date().isoformat()
    elif hasattr(raw_date, "isoformat"):
        service_date = raw_date.isoformat()
    else:
        raise ValueError(f"Unparseable service_date on claim {raw_claim.get('claim_id')}")

    # Normalize line items. Each line is one procedure code plus modifiers
    # plus billed amount. The main source of error here is splitting versus
    # collapsing lines that share a code but differ in modifier, site of
    # service, or units.
    lines = []
    for raw_line in raw_claim.get("lines", []):
        code = (raw_line.get("cpt") or raw_line.get("hcpcs") or "").strip().upper()
        modifiers = raw_line.get("modifiers") or []
        if isinstance(modifiers, str):
            modifiers = [m.strip() for m in modifiers.split(",") if m.strip()]
        lines.append({
            "line_number": raw_line.get("line_number"),
            "code": code,
            "modifiers": [m.strip().upper() for m in modifiers],
            "units": int(raw_line.get("units", 1)),
            "billed_amount": _to_decimal(raw_line.get("billed_amount", 0)),
            "diagnosis_pointers": raw_line.get("diagnosis_pointers", []),
            "place_of_service": raw_line.get("place_of_service"),
        })

    canonical = {
        "claim_id": raw_claim["claim_id"],
        "claim_type": raw_claim.get("claim_type", "professional"),
        "service_date": service_date,
        "submission_date": raw_claim.get("submission_date"),
        "patient_id": raw_claim.get("patient_id"),             # internal patient ID
        "rendering_provider_npi": raw_claim.get("rendering_provider_npi"),
        "billing_provider_npi": raw_claim.get("billing_provider_npi"),
        "referring_provider_npi": raw_claim.get("referring_provider_npi"),
        "facility_npi": raw_claim.get("facility_npi"),
        "diagnosis_codes": [d.strip().upper() for d in raw_claim.get("diagnosis_codes", [])],
        "lines": lines,
        "billed_amount_total": _to_decimal(
            raw_claim.get("billed_amount_total")
            or sum((line["billed_amount"] or Decimal("0")) for line in lines)
        ),
        "paid_amount": _to_decimal(raw_claim.get("paid_amount", 0)),
        "status": raw_claim.get("status", "adjudicated"),      # adjudicated | denied | pending
        "source_system": raw_claim.get("source_system"),       # provenance for audit
        "ingestion_timestamp": datetime.now(timezone.utc).isoformat(),
    }

    logger.info(
        "claim normalized",
        extra={"event": "normalize_claim", **_redact_for_logs(canonical)},
    )
    return canonical

def _load_external_reference_data():
    """Load NPPES, LEIE, SAM, and Sunshine Act reference data.

    In production, these are downloaded monthly (NPPES and LEIE are monthly
    files; SAM is weekly; Open Payments/Sunshine Act is annual with quarterly
    refinements). A Glue job parses each file into Parquet and writes to the
    reference-data prefix of the entity-resolution pipeline's S3 bucket.

    This example returns a tiny in-memory dict so the resolve step runs.
    """
    return {
        "nppes": {
            # canonical public provider registry, keyed by NPI
            "1234567890": {
                "npi": "1234567890",
                "entity_type": "individual",
                "full_name": "Smith, John",
                "credentials": "MD",
                "primary_specialty": "Internal Medicine",
                "practice_address": "123 Main St, Austin, TX 78701",
            },
        },
        "leie": set(),  # NPIs on the HHS-OIG exclusion list
        "sam": set(),   # entity IDs debarred from federal contracting
        "sunshine_act": defaultdict(list),  # NPI -> list of industry payments
    }

def resolve_providers(claims_batch, external_reference):
    """Produce a canonical provider ID for each NPI referenced in the batch.

    This example uses a simple NPI-based lookup. A production pipeline runs a
    probabilistic match across name, address, DOB, degree, and specialty from
    NPPES plus internal provider-master records (see Recipe 5.x for the full
    matching pattern). The output is written both to a DynamoDB lookup table
    (so the online path can cache canonical IDs) and to Parquet in S3 (so
    downstream Spark jobs can join by canonical ID).
    """
    table = dynamodb.Table(RESOLVED_ENTITIES_TABLE)
    canonical_ids = {}
    exclusion_flags = []

    # Collect the unique NPIs from this batch. Rendering, billing, referring,
    # and facility NPIs all need resolution; the graph wires them with
    # different edge types.
    npis_in_batch = set()
    for claim in claims_batch:
        for role in ["rendering_provider_npi", "billing_provider_npi",
                     "referring_provider_npi", "facility_npi"]:
            if claim.get(role):
                npis_in_batch.add(claim[role])

    for npi in npis_in_batch:
        # Check the canonical cache first. In production this is either a
        # DynamoDB read (for the online path) or a broadcast join in Spark
        # (for the batch path).
        response = table.get_item(Key={"canonical_id": f"NPI:{npi}"})
        if "Item" in response:
            canonical_ids[npi] = response["Item"]["canonical_id"]
        else:
            # Not in cache. Attempt resolution against NPPES.
            nppes_record = external_reference["nppes"].get(npi)
            if nppes_record:
                canonical_id = f"NPI:{npi}"   # NPIs are globally unique, use directly
                table.put_item(Item={
                    "canonical_id": canonical_id,
                    "npi": npi,
                    "source_records": [{"source": "nppes", "record": nppes_record}],
                    "resolved_at": datetime.now(timezone.utc).isoformat(),
                })
                canonical_ids[npi] = canonical_id
            else:
                # NPI not found in NPPES. Two possibilities: the NPI is
                # invalid (typo, test data) or the NPPES refresh is stale.
                # Emit a metric and route to the entity-resolution DLQ.
                logger.warning("npi not found in nppes", extra={"npi_prefix": npi[:4]})
                canonical_ids[npi] = f"UNRESOLVED:{npi}"

        # Exclusion checks fire unconditionally. The LEIE match is a trump
        # card: any claim touching an excluded provider is invalid, period.
        if npi in external_reference["leie"]:
            exclusion_flags.append({
                "canonical_id": canonical_ids[npi],
                "exclusion_source": "LEIE",
                "severity": "critical",
            })
        if npi in external_reference["sam"]:
            exclusion_flags.append({
                "canonical_id": canonical_ids[npi],
                "exclusion_source": "SAM",
                "severity": "high",
            })

    return canonical_ids, exclusion_flags

def refresh_graph(resolved_entities, claims_batch, ownership_edges=None):
    """Build or refresh the provider/patient/claim relationship graph.

    Nodes: providers, organizations, patients, claims, payments.
    Edges: rendered_by, billed_by, referred_by, rendered_at, patient_of,
           payment_for, owns (for organizational ownership), co_located_at.

    In production on Neptune, each add_node and add_edge call below maps to a
    Gremlin upsert (`.g.V().has('id', ...).fold().coalesce(...)` style) or a
    Cypher MERGE, and traversals run as long-lived gremlin or openCypher
    queries. The shape of the data and the query patterns is identical here;
    only the runtime and the scale change.
    """
    graph = nx.MultiDiGraph()

    # Provider nodes. Type annotations on nodes are essential for later
    # filtering in community detection and referral-concentration queries.
    for original_npi, canonical_id in resolved_entities.items():
        graph.add_node(canonical_id, node_type="provider", npi=original_npi)

    # Claim nodes plus the edges that link them to providers and patients.
    # Patients are a separate node type because the ownership graph does not
    # include them (patient-level privacy separation is important for legal
    # review) but the care graph does.
    for claim in claims_batch:
        claim_node = f"CLAIM:{claim['claim_id']}"
        graph.add_node(
            claim_node, node_type="claim",
            service_date=claim["service_date"],
            billed=float(claim["billed_amount_total"] or 0),
        )

        if claim.get("patient_id"):
            patient_node = f"PATIENT:{_hash_patient_id(claim['patient_id'])}"
            graph.add_node(patient_node, node_type="patient")
            graph.add_edge(patient_node, claim_node, edge_type="patient_of",
                           service_date=claim["service_date"])

        rendering = resolved_entities.get(claim.get("rendering_provider_npi"))
        if rendering:
            graph.add_edge(rendering, claim_node, edge_type="rendered_by")

        billing = resolved_entities.get(claim.get("billing_provider_npi"))
        if billing:
            graph.add_edge(billing, claim_node, edge_type="billed_by")

        referring = resolved_entities.get(claim.get("referring_provider_npi"))
        if referring and rendering and referring != rendering:
            # Referral edges are the substrate for referral-concentration
            # analytics in Step 6. Self-referrals are excluded here because
            # they look like data-entry errors more than scheme signal.
            graph.add_edge(referring, rendering, edge_type="referred_to",
                           service_date=claim["service_date"])

    # Ownership edges. In production these come from state filings, Sunshine
    # Act data, and corporate registry feeds. They are the edges that expose
    # common-ownership cascades (one LLC owning ten clinics that all refer
    # to the same DME supplier, for example).
    for edge in (ownership_edges or []):
        parent_id = edge["parent"]
        child_id = edge["child"]
        # Create endpoint nodes with node_type="organization" so the
        # OWNERSHIP_CASCADE detector in Step 6 can find them. In production,
        # entity resolution (Step 2) writes organization nodes alongside
        # providers; this branch is the teaching-example shortcut.
        if parent_id not in graph:
            graph.add_node(parent_id, node_type="organization")
        if child_id not in graph:
            graph.add_node(child_id, node_type="organization")
        graph.add_edge(parent_id, child_id, edge_type="owns",
                       percentage=edge.get("percentage"))

    logger.info("graph refreshed", extra={
        "event": "refresh_graph",
        "nodes": graph.number_of_nodes(),
        "edges": graph.number_of_edges(),
    })
    return graph

def run_rules_on_claim(canonical_claim, resolved_entities, patient_vital_status=None):
    """Evaluate the rules layer against a single canonical claim.

    Returns a list of flag dicts. Each flag is explainable: rule_id, the
    values that triggered it, and a severity tier. The rule catalog grows
    with the SIU's experience; this function is a teaching subset (CCI edit,
    MUE, exclusion check, post-mortem billing, and a medical-necessity gate
    for DME). A production deployment has hundreds of rules partitioned by
    claim type and jurisdiction.
    """
    flags = []
    claim_id = canonical_claim["claim_id"]
    service_date = canonical_claim["service_date"]

    # Gather codes on this claim for CCI and MUE evaluation. A claim with
    # multiple lines of the same code is collapsed for CCI (which looks at
    # pairs) but preserved for MUE (which looks at total units).
    codes_on_claim = [line["code"] for line in canonical_claim["lines"]]
    code_counter = Counter()
    for line in canonical_claim["lines"]:
        code_counter[line["code"]] += line["units"]

    # Rule 1: CCI edits. If two mutually-exclusive codes appear on the same
    # claim without a modifier that documents the override, flag it.
    for (code_a, code_b), (edit_type, default_allowed) in CCI_EDIT_PAIRS.items():
        if code_a in codes_on_claim and code_b in codes_on_claim and not default_allowed:
            flags.append({
                "rule_id": "CCI_EDIT_MX",
                "claim_id": claim_id,
                "severity": "high",
                "details": {
                    "pair": [code_a, code_b],
                    "edit_type": edit_type,
                },
                "explain": (
                    f"Codes {code_a} and {code_b} are flagged as {edit_type} "
                    "by CMS CCI PTP edits; no overriding modifier present."
                ),
            })

    # Rule 2: Medically Unlikely Edit. If units for a code exceed the CMS MUE
    # threshold for that code, flag it. The MUE file is a hard ceiling; units
    # above the MUE are almost always data-entry errors or upcoding.
    for code, total_units in code_counter.items():
        mue = MUE_THRESHOLDS.get(code)
        if mue is not None and total_units > mue:
            flags.append({
                "rule_id": "MUE_EXCEEDED",
                "claim_id": claim_id,
                "severity": "high",
                "details": {"code": code, "units": total_units, "threshold": mue},
                "explain": (
                    f"Code {code} billed with {total_units} units on a single date of "
                    f"service (MUE threshold: {mue})."
                ),
            })

    # Rule 3: Unresolved provider role. The actual LEIE/SAM exclusion check
    # ran in Step 2 (resolve_providers returns exclusion_flags); the pipeline
    # driver concatenates those into rule_flags. Here we surface the
    # data-quality flag for any provider role that fell through entity
    # resolution, which is worth investigating separately (missing from NPPES
    # can indicate credential issues, practice closure, or data-entry errors).
    for role in ["rendering_provider_npi", "billing_provider_npi",
                 "referring_provider_npi", "facility_npi"]:
        npi = canonical_claim.get(role)
        if npi and resolved_entities.get(npi, "").startswith("UNRESOLVED:"):
            flags.append({
                "rule_id": "UNRESOLVED_PROVIDER",
                "claim_id": claim_id,
                "severity": "medium",
                "details": {"role": role, "npi_prefix": npi[:4]},
                "explain": f"Provider role {role} could not be resolved to NPPES.",
            })

    # Rule 4: Post-mortem billing. Services billed after a patient's date of
    # death are never legitimate. The date-of-death lookup comes from a
    # vital-status feed (state death-record integrations, CMS 20% sample for
    # Medicare, internal mortality data). Not every payer has this, and
    # without it this rule simply does not fire.
    if patient_vital_status and canonical_claim.get("patient_id"):
        dod = patient_vital_status.get(canonical_claim["patient_id"])
        if dod and service_date > dod:
            flags.append({
                "rule_id": "POST_MORTEM_BILLING",
                "claim_id": claim_id,
                "severity": "critical",
                "details": {"service_date": service_date, "date_of_death": dod},
                "explain": (
                    f"Service date {service_date} is after recorded date of death "
                    f"{dod}. This claim should not have been paid."
                ),
            })

    # Rule 5: Medical-necessity gate for DME oxygen (E1390/E0431). Oxygen is
    # only covered with a documented hypoxemia diagnosis. If the claim has no
    # supporting diagnosis code, flag for medical-necessity review.
    dme_oxygen_lines = [line for line in canonical_claim["lines"]
                        if line["code"] in DME_OXYGEN_CODES]
    if dme_oxygen_lines:
        hypoxemia_dx_prefix = ("J96", "J44", "J95", "R09")   # illustrative subset
        has_supporting_dx = any(
            any(dx.startswith(p) for p in hypoxemia_dx_prefix)
            for dx in canonical_claim.get("diagnosis_codes", [])
        )
        if not has_supporting_dx:
            flags.append({
                "rule_id": "MEDNEC_DME_OXYGEN",
                "claim_id": claim_id,
                "severity": "high",
                "details": {"codes": [line["code"] for line in dme_oxygen_lines]},
                "explain": (
                    "Oxygen DME billed without a documented hypoxemia diagnosis. "
                    "Requires medical-necessity review."
                ),
            })

    return flags

def _safe_zscore(value, peer_mean, peer_std, floor_std=0.01):
    """Compute a z-score with a floor on the standard deviation.

    Protects against divide-by-zero when a peer group is homogeneous, which
    otherwise produces spurious infinite z-scores for any nonzero deviation.
    """
    denom = max(peer_std or 0, floor_std)
    return (value - peer_mean) / denom

def _cusum(series, k, h):
    """Classic one-sided upper CUSUM.

    Returns the first index where the cumulative positive deviation from the
    reference mean exceeds h standard deviations. None if no detection.
    Tracks when a provider's own rate started drifting upward (level-5 E&M
    rate creeping up, drug-test units per patient climbing, and so on).

    NOTE: This teaching implementation uses the full series mean as the
    reference value. Production CUSUM uses an explicit baseline window (e.g.,
    the first half of the series) or a target mean from the SIU's process
    spec. The full-series-mean approach is structurally less sensitive to
    shifts because the shift itself pulls the reference up. See Recipe 3.3
    for the billing-code-drift detection pattern with an explicit baseline.
    """
    mean = np.mean(series)
    std = np.std(series) or 1.0
    c = 0.0
    for i, x in enumerate(series):
        c = max(0, c + (x - mean - k * std))
        if c > h * std:
            return i
    return None

def score_provider_statistics(provider_id, provider_features, peer_baselines):
    """Score a provider's monthly feature vector against peer baselines and
    self-history.

    Arguments:
      provider_features: dict with keys like 'level5_em_rate', 'modifier_25_rate',
                         'drug_test_units_per_patient', 'unique_patients',
                         plus 12 months of each as 'history_level5_em_rate' list.
      peer_baselines:    dict keyed by peer_group_feature_name, with 'mean' and
                         'std' entries. Loaded from Feature Store or DynamoDB.
    """
    flags = []
    peer_key = provider_features["peer_group_key"]

    # Peer z-scores across the features that matter for fraud detection.
    # These feature names are illustrative; a real production pipeline has
    # dozens of features per provider per period.
    for feature in ["level5_em_rate", "modifier_25_rate", "drug_test_units_per_patient"]:
        baseline = peer_baselines.get(f"{peer_key}:{feature}")
        if not baseline:
            continue
        z = _safe_zscore(
            provider_features[feature],
            float(baseline["mean"]),
            float(baseline["std"]),
        )
        if abs(z) >= PEER_ZSCORE_FLAG:
            severity = "high" if abs(z) >= PEER_ZSCORE_HIGH_SEVERITY else "medium"
            flags.append({
                "rule_id": f"PEER_ZSCORE_{feature.upper()}",
                "provider_id": provider_id,
                "severity": severity,
                "details": {
                    "feature": feature,
                    "value": provider_features[feature],
                    "peer_mean": float(baseline["mean"]),
                    "peer_std": float(baseline["std"]),
                    "z_score": round(z, 2),
                    "peer_group": peer_key,
                },
                "explain": (
                    f"Provider's {feature} is {round(z, 2)} standard deviations "
                    f"from peer mean in {peer_key}."
                ),
            })

    # Self-history CUSUM. Catches upward drift from the provider's own
    # baseline, which is different from being an outlier against peers
    # (a provider whose peers all upcode at the same rate as they do).
    for history_feature in ["history_level5_em_rate", "history_modifier_25_rate"]:
        series = provider_features.get(history_feature, [])
        if len(series) >= 6:
            drift_index = _cusum(series, k=CUSUM_K, h=CUSUM_H)
            if drift_index is not None:
                flags.append({
                    "rule_id": f"CUSUM_{history_feature.upper()}",
                    "provider_id": provider_id,
                    "severity": "medium",
                    "details": {
                        "feature": history_feature,
                        "drift_start_period": drift_index,
                        "series_length": len(series),
                    },
                    "explain": (
                        f"Provider's own {history_feature} began drifting upward "
                        f"at period index {drift_index}."
                    ),
                })

    return flags

def train_isolation_forest(feature_matrix):
    """Train a multivariate Isolation Forest on provider-period feature vectors.

    The training matrix contains features per provider per month: level-5 E&M
    rate, modifier-25 rate, units per patient by code family, diagnosis-code
    entropy, billed-amount per patient, and so on. In production, this is a
    SageMaker Training Job; joblib.dump the model to S3, tag the model version
    in the Model Registry, and run quarterly.
    """
    model = IsolationForest(
        n_estimators=200,
        contamination=ISOLATION_FOREST_CONTAMINATION,
        random_state=42,
    )
    model.fit(feature_matrix)
    return model

def score_isolation_forest(model, feature_matrix, provider_ids):
    """Score a batch of providers against the trained Isolation Forest.

    Returns a list of flags for providers the model identifies as outliers.
    """
    scores = model.decision_function(feature_matrix)   # higher = more normal
    predictions = model.predict(feature_matrix)        # -1 = anomaly

    flags = []
    for provider_id, score, prediction in zip(provider_ids, scores, predictions):
        if prediction == -1:
            flags.append({
                "rule_id": "ISOLATION_FOREST_OUTLIER",
                "provider_id": provider_id,
                "severity": "medium" if score > -0.1 else "high",
                "details": {"isolation_score": round(float(score), 4)},
                "explain": (
                    "Provider's multivariate feature vector is flagged as an "
                    "outlier by the Isolation Forest model. No single feature "
                    "is extreme; the combination is unusual relative to peers."
                ),
            })
    return flags

def run_graph_analytics(graph):
    """Run the graph-analytics detectors against the refreshed graph.

    On Neptune, each of these is a Gremlin query or a Neptune ML graph
    algorithm invocation. Here we use NetworkX's in-memory equivalents so the
    teaching example runs in a notebook. Output shape is identical.
    """
    flags = []

    # Detector 1: Louvain community detection. Communities are dense
    # subgraphs of providers who bill, refer, and render together more than
    # the network average would predict. High-risk communities share patients
    # heavily, cross-refer in tight loops, and have elevated per-patient
    # exposure. A community alone is not fraud; a community with elevated
    # exposure against peer benchmarks is worth investigating.
    if HAS_LOUVAIN:
        provider_subgraph = graph.subgraph([
            n for n, d in graph.nodes(data=True) if d.get("node_type") == "provider"
        ]).to_undirected()
        if provider_subgraph.number_of_nodes() >= 3:
            partition = community_louvain.best_partition(provider_subgraph)
            community_sizes = Counter(partition.values())
            for community_id, size in community_sizes.items():
                if size >= 5:
                    members = [n for n, c in partition.items() if c == community_id]
                    flags.append({
                        "rule_id": "COMMUNITY_CLUSTER",
                        "severity": "medium",
                        "details": {
                            "community_id": community_id,
                            "size": size,
                            "members_sample": members[:10],
                        },
                        "explain": (
                            f"Provider community of size {size} detected by Louvain "
                            "community detection; warrants referral-pattern review."
                        ),
                    })

    # Detector 2: Referral concentration. If more than 35% of a provider's
    # outbound referrals go to a single downstream provider, flag it. Some
    # specialties legitimately have concentrated referral networks (a small-
    # town PCP with one local specialist), so this is a flag not a rule;
    # investigators validate with the local-market context.
    for provider in [n for n, d in graph.nodes(data=True) if d.get("node_type") == "provider"]:
        outbound_referrals = Counter()
        for _, target, edge_data in graph.out_edges(provider, data=True):
            if edge_data.get("edge_type") == "referred_to":
                outbound_referrals[target] += 1
        total = sum(outbound_referrals.values())
        if total >= 20:
            top_target, top_count = outbound_referrals.most_common(1)[0]
            concentration = top_count / total
            if concentration >= REFERRAL_CONCENTRATION_FLAG:
                flags.append({
                    "rule_id": "REFERRAL_CONCENTRATION",
                    "provider_id": provider,
                    "severity": "medium",
                    "details": {
                        "top_target": top_target,
                        "concentration": round(concentration, 3),
                        "total_referrals": total,
                    },
                    "explain": (
                        f"Provider sends {round(concentration*100, 1)}% of referrals "
                        f"to a single downstream provider; review for common ownership."
                    ),
                })

    # Detector 3: Ownership cascade. If a single owner controls organizations
    # that together exceed a concentration threshold in a tight referral
    # network, escalate. Computed by walking `owns` edges and aggregating
    # downstream claim exposure.
    owners = [n for n, d in graph.nodes(data=True) if d.get("node_type") == "organization"]
    for owner in owners:
        owned = [t for _, t, d in graph.out_edges(owner, data=True)
                 if d.get("edge_type") == "owns"]
        if len(owned) >= 5:
            flags.append({
                "rule_id": "OWNERSHIP_CASCADE",
                "organization_id": owner,
                "severity": "high",
                "details": {"owned_count": len(owned), "owned_sample": owned[:5]},
                "explain": (
                    f"Organization owns {len(owned)} downstream entities; "
                    "aggregate exposure review recommended."
                ),
            })

    return flags

def _overall_severity(flag_severities):
    """Fold individual flag severities into a case-level severity.

    Critical wins. Otherwise, three or more high flags promote to critical.
    One high or any combination below that is high. Otherwise medium.
    """
    if "critical" in flag_severities:
        return "critical"
    if flag_severities.count("high") >= 3:
        return "critical"
    if "high" in flag_severities:
        return "high"
    return "medium"

def _determine_routing(severity, exposure):
    """Routing policy by case severity and dollar exposure.

    Critical cases go to the lead investigator and legal immediately.
    High-severity plus high-exposure cases go to the priority queue.
    Everything else lands in the standard queue.
    """
    if severity == "critical" or exposure >= CRITICAL_EXPOSURE:
        return {"queue": "priority", "notify": ["lead-investigator", "legal"]}
    if severity == "high" and exposure >= HIGH_SEVERITY_EXPOSURE:
        return {"queue": "priority", "notify": ["lead-investigator"]}
    return {"queue": "standard", "notify": []}

def _derive_case_id(entity_id, rule_ids, window_key):
    """Deterministic case_id so retries collapse to the same record.

    window_key is a coarse bucket like 'year=2026/week=20'. It collapses
    retries within a detection window to a single case while letting a
    week-over-week recurrence open a new case.
    """
    key_material = "|".join([
        entity_id,
        ",".join(sorted(set(rule_ids))),
        window_key,
    ])
    digest = hashlib.sha256(key_material.encode("utf-8")).hexdigest()
    return f"CASE:{digest[:24]}"

def aggregate_flags_to_cases(all_flags, claim_exposure_by_entity):
    """Group flags by target entity, compute case severity and exposure,
    and write case records to DynamoDB.

    all_flags: list of flag dicts from rules, stats, and graph layers.
               Each flag has an implicit or explicit target entity
               (provider_id, organization_id, or claim_id which maps back
               to a rendering provider).
    claim_exposure_by_entity: dict {entity_id -> Decimal dollar exposure}.
    """
    table = dynamodb.Table(CASE_STATE_TABLE)
    flags_by_entity = defaultdict(list)

    # Group flags by the target entity the investigator will open. Most flags
    # hang off a provider_id; some hang off an organization_id (ownership
    # cascade) or a claim_id (which we roll up to the rendering provider).
    for flag in all_flags:
        if flag.get("provider_id"):
            entity_id = flag["provider_id"]
        elif flag.get("organization_id"):
            entity_id = flag["organization_id"]
        elif flag.get("claim_id"):
            # Map the claim back to its rendering provider for case grouping.
            # In production, the flag is enriched with the rendering provider
            # canonical ID at rule-firing time so this lookup is not needed.
            entity_id = flag.get("rendering_provider_canonical_id", flag["claim_id"])
        else:
            continue
        flags_by_entity[entity_id].append(flag)

    cases_written = []
    for entity_id, flags in flags_by_entity.items():
        severities = [f["severity"] for f in flags]
        overall = _overall_severity(severities)
        exposure = claim_exposure_by_entity.get(entity_id, Decimal("0"))
        routing = _determine_routing(overall, exposure)

        # Deterministic case_id derived from entity, rules, and time window.
        # On retry (Step Functions retries a transient DynamoDB throttle), the
        # function produces the same case_id, and the ConditionExpression
        # below catches the duplicate write. This is real retry idempotency,
        # not the uuid-per-call pattern that silently creates duplicates.
        window_key = datetime.now(timezone.utc).strftime("year=%Y/week=%V")
        case_id = _derive_case_id(
            entity_id=entity_id,
            rule_ids=[f["rule_id"] for f in flags],
            window_key=window_key,
        )

        # DynamoDB rejects Python float for numeric attributes; the
        # resource-API serializer raises TypeError at put time. Coerce
        # numeric fields to Decimal at the type boundary for nested payloads
        # like evidence_summary (which contains z-scores, isolation scores,
        # and referral concentrations from sklearn and Python ratio math).
        case_record = {
            "target_entity_id": entity_id,
            "case_id": case_id,
            "status": "open",
            "severity": overall,
            "exposure_amount": _to_decimal(exposure),
            "num_flags": len(flags),
            "flag_types": sorted(set(f["rule_id"] for f in flags)),
            "evidence_summary": _floats_to_decimal(flags[:20]),
            "routing_queue": routing["queue"],
            "created_at": datetime.now(timezone.utc).isoformat(),
            "case_bundle_s3_uri": (
                f"s3://{SUBGRAPH_ARTIFACTS_BUCKET}/cases/{case_id}/bundle.json"
            ),
        }

        # ConditionExpression with a deterministic case_id means retries
        # hit the condition and fail closed. Catch the exception and treat
        # as success: the case is already there from the first attempt.
        try:
            table.put_item(
                Item=_floats_to_decimal(case_record),
                ConditionExpression="attribute_not_exists(case_id)",
            )
        except table.meta.client.exceptions.ClientError as e:
            if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
                logger.info("case already exists; idempotent retry",
                            extra={"case_id": case_id, "entity_id": entity_id})
                continue
            raise
        cases_written.append(case_record)

        # Publish to the flags EventBridge bus so the evidence-aggregator,
        # notifier, and audit consumers can fan out without this function
        # knowing about them.
        eventbridge.put_events(Entries=[{
            "Source": "fwa.detection",
            "DetailType": "CaseCreated",
            "Detail": json.dumps({
                "case_id": case_id,
                "entity_id": entity_id,
                "severity": overall,
                "exposure_amount": str(exposure),
                "routing_queue": routing["queue"],
            }),
            "EventBusName": FWA_FLAGS_BUS,
        }])

        # High-priority cases also page the lead investigator through SNS.
        if routing["queue"] == "priority":
            sns.publish(
                TopicArn=INVESTIGATOR_NOTIFICATION_TOPIC_ARN,
                Subject=f"[{overall.upper()}] FWA Case Opened: {entity_id}",
                Message=json.dumps({
                    "case_id": case_id,
                    "entity_id": entity_id,
                    "exposure": str(exposure),
                    "flag_types": case_record["flag_types"],
                }),
            )

    _emit_metric("CasesCreated", len(cases_written))
    return cases_written

def _emit_metric(metric_name, value, unit="Count"):
    """Emit a CloudWatch metric."""
    cloudwatch.put_metric_data(
        Namespace="FWA/Detection",
        MetricData=[{
            "MetricName": metric_name,
            "Value": float(value),
            "Unit": unit,
            "Timestamp": datetime.now(timezone.utc),
        }],
    )

def assist_documentation_review(case_id, clinical_note_text, billed_codes):
    """Produce a structured documentation-review finding for a case.

    Uses Comprehend Medical to extract entities (pre-processing, privacy-
    bounded) and Bedrock with a constrained prompt to produce the reviewer
    draft. The output is never the final case summary. It is an input to a
    human reviewer.
    """
    # Step 8a: Extract clinical entities with Comprehend Medical. This step
    # is partly privacy hygiene (we want to log what entities the LLM saw
    # without passing raw text to logs) and partly feature extraction (the
    # LLM works better when pointed at specific findings).
    entities_response = comprehend_medical.detect_entities_v2(Text=clinical_note_text)
    entities = entities_response.get("Entities", [])
    key_findings = [
        {"category": e["Category"], "type": e["Type"], "text": e["Text"]}
        for e in entities
        if e["Score"] >= 0.8
    ]

    # Step 8b: Invoke Bedrock with a constrained prompt. The system prompt
    # anchors the model to a structured output and prohibits clinical
    # recommendations. Output is a finding object, not free-text opinion.
    prompt = (
        "You are reviewing whether clinical documentation supports a set of "
        "billed procedure codes. You are not making a clinical judgment. You "
        "are identifying whether the documentation, as written, contains the "
        "elements that each code requires. Produce a JSON object with keys: "
        "\"supported_codes\" (list), \"unsupported_codes\" (list), "
        "\"ambiguous_codes\" (list), \"missing_elements\" (dict of code -> "
        "list of missing documentation elements).\n\n"
        f"Billed codes: {billed_codes}\n\n"
        f"Clinical documentation:\n{clinical_note_text}"
    )

    bedrock_response = bedrock_runtime.invoke_model(
        modelId=BEDROCK_MODEL_ID,
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 2048,
            "temperature": 0.1,   # low temperature, we want structured output
            "messages": [{"role": "user", "content": prompt}],
        }),
    )

    bedrock_body = json.loads(bedrock_response["body"].read())
    model_output = bedrock_body["content"][0]["text"]

    # Attempt to parse the JSON out of the model output. Bedrock may wrap
    # the JSON in prose or code fences; defend against both.
    try:
        # Extract content between the first and last brace if prose is present.
        first = model_output.find("{")
        last = model_output.rfind("}")
        parsed = json.loads(model_output[first:last + 1]) if first >= 0 and last >= 0 else {}
    except json.JSONDecodeError:
        parsed = {"parse_error": True, "raw_output_length": len(model_output)}

    finding = {
        "case_id": case_id,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "key_findings_count": len(key_findings),
        "billed_codes": billed_codes,
        "reviewer_draft": parsed,
        "needs_human_review": True,   # always. never auto-close on LLM output.
    }

    logger.info("documentation review draft generated", extra={
        "event": "assist_documentation_review",
        "case_id": case_id,
        "supported_count": len(parsed.get("supported_codes", [])),
        "unsupported_count": len(parsed.get("unsupported_codes", [])),
    })
    return finding

def capture_case_outcome(case_id, entity_id, outcome, recovery_amount=None, notes=None):
    """Update a case with its investigation outcome and write the label to S3
    for the retraining pipeline.

    outcome must be one of: CONFIRMED, CLEARED, PENDING_PAYER_RECOVERY,
    REFERRED_TO_REGULATOR, DUPLICATE_OF_EARLIER_CASE.
    """
    valid_outcomes = {
        "CONFIRMED", "CLEARED", "PENDING_PAYER_RECOVERY",
        "REFERRED_TO_REGULATOR", "DUPLICATE_OF_EARLIER_CASE",
    }
    if outcome not in valid_outcomes:
        raise ValueError(f"invalid outcome: {outcome}")

    # Input validation: catch obvious bad inputs before any side effects.
    if recovery_amount is not None and recovery_amount < 0:
        raise ValueError(f"recovery_amount must be non-negative; got {recovery_amount}")
    if not case_id or not case_id.startswith("CASE:"):
        raise ValueError(f"case_id must be a CASE: prefixed identifier; got {case_id!r}")
    if not entity_id:
        raise ValueError("entity_id is required")

    table = dynamodb.Table(CASE_STATE_TABLE)

    # Atomic write counter for audit. The version attribute increments on
    # every outcome write so retroactive review can tell whether the outcome
    # was overwritten and how many times. ConditionExpression ensures we never
    # write to a case that has been deleted. EventBridge at-least-once
    # delivery is fine here because outcome writes are terminal and
    # idempotent; if a real concurrent-writer scenario emerges, switch to
    # read-modify-write with a version check on the ConditionExpression.
    table.update_item(
        Key={"target_entity_id": entity_id, "case_id": case_id},
        UpdateExpression=(
            "SET #status = :status, outcome = :outcome, recovery_amount = :recovery, "
            "investigator_notes = :notes, outcome_captured_at = :ts "
            "ADD version :one"
        ),
        ExpressionAttributeNames={"#status": "status"},
        ExpressionAttributeValues={
            ":status": "closed",
            ":outcome": outcome,
            ":recovery": _to_decimal(recovery_amount or 0),
            ":notes": notes or "",
            ":ts": datetime.now(timezone.utc).isoformat(),
            ":one": 1,
        },
        ConditionExpression="attribute_exists(case_id)",
    )

    # Derive and store the label for the retraining pipeline. One label row
    # per case, with its feature-snapshot key so the training job can join
    # the label back to the exact feature vector that was scored at case
    # creation. The label derivation is organization-specific; revisit the
    # mapping with SIU leadership quarterly.
    # Date-partitioned key so Athena and Glue can prune at the partition level
    # when reading labels for retraining; case_id uniqueness inside the
    # partition is preserved by the deterministic case_id.
    decision_dt = datetime.now(timezone.utc)
    label_key = (
        f"labels/year={decision_dt.year:04d}/"
        f"month={decision_dt.month:02d}/"
        f"day={decision_dt.day:02d}/"
        f"{case_id.replace(':', '-')}.json"
    )
    label_record = {
        "case_id": case_id,
        "entity_id": entity_id,
        "label": 1 if outcome in {"CONFIRMED", "PENDING_PAYER_RECOVERY",
                                  "REFERRED_TO_REGULATOR"} else 0,
        "outcome_raw": outcome,
        "recovery_amount": str(_to_decimal(recovery_amount or 0)),
        "labeled_at": decision_dt.isoformat(),
    }
    s3_client.put_object(
        Bucket=CASE_OUTCOMES_BUCKET,
        Key=label_key,
        Body=json.dumps(label_record).encode("utf-8"),
        ServerSideEncryption="aws:kms",
        SSEKMSKeyId=CASE_OUTCOMES_CMK_ARN,
    )

    # Publish the outcome to the workflow bus. Consumers include the
    # retraining trigger, the SIU dashboard updater, and the legal-packaging
    # path for REFERRED_TO_REGULATOR.
    eventbridge.put_events(Entries=[{
        "Source": "fwa.workflow",
        "DetailType": "CaseClosed",
        "Detail": json.dumps(label_record),
        "EventBusName": FWA_WORKFLOW_BUS,
    }])

    _emit_metric("CasesClosed", 1)
    _emit_metric(f"Outcome_{outcome}", 1)
    return label_record

def run_fwa_pipeline(
    raw_claims_batch,
    ownership_edges=None,
    patient_vital_status=None,
    peer_baselines=None,
    isolation_forest_model=None,
):
    """End-to-end FWA detection pipeline against a batch of raw claims.

    Returns (cases_written, all_flags). Prints per-step progress so readers
    can trace the data flow.
    """
    print(f"[1/9] normalizing {len(raw_claims_batch)} claims")
    canonical_claims = [normalize_claim(raw) for raw in raw_claims_batch]

    print("[2/9] resolving providers")
    external_ref = _load_external_reference_data()
    resolved, exclusion_flags = resolve_providers(canonical_claims, external_ref)

    print("[3/9] refreshing graph")
    graph = refresh_graph(resolved, canonical_claims, ownership_edges)

    print("[4/9] running rules layer")
    rule_flags = []
    for claim in canonical_claims:
        rule_flags.extend(run_rules_on_claim(claim, resolved, patient_vital_status))
    rule_flags.extend(exclusion_flags)

    print("[5/9] running statistical layer")
    stat_flags = []
    if peer_baselines:
        # Build a per-provider monthly feature frame from the claims batch.
        # In production this is a separate Glue job; we inline a tiny version
        # for the teaching example.
        per_provider = _build_provider_features(canonical_claims)
        provider_ids = sorted(per_provider.keys())
        for pid in provider_ids:
            stat_flags.extend(score_provider_statistics(pid, per_provider[pid], peer_baselines))

        # Isolation Forest scoring if a model is loaded. In production you
        # would load this from S3 via joblib at job start.
        if isolation_forest_model is not None:
            feature_columns = ["level5_em_rate", "modifier_25_rate",
                               "drug_test_units_per_patient"]
            matrix = np.array([
                [per_provider[pid][c] for c in feature_columns]
                for pid in provider_ids
            ])
            stat_flags.extend(score_isolation_forest(isolation_forest_model, matrix, provider_ids))

    print("[6/9] running graph analytics")
    graph_flags = run_graph_analytics(graph)

    print("[7/9] aggregating flags into cases")
    all_flags = rule_flags + stat_flags + graph_flags
    exposure_by_entity = _compute_exposure(canonical_claims, resolved)
    cases = aggregate_flags_to_cases(all_flags, exposure_by_entity)

    # Steps 8 and 9 are event-triggered in production (LLM review fires from
    # the CaseCreated event; outcome capture fires from the investigator UI).
    # They are shown here as explicit calls for clarity.
    print(f"[8/9] LLM-assisted documentation review is event-triggered; "
          f"{len(cases)} cases will fan out through EventBridge")
    print("[9/9] outcome capture fires from the investigator UI; "
          "call capture_case_outcome when a case is closed")

    return cases, all_flags

def _build_provider_features(canonical_claims):
    """Compute a tiny subset of per-provider features from the claims batch.

    Real implementation uses a Glue job reading the full claims warehouse
    plus the patient master; this inline version is enough to show the
    shape.
    """
    by_provider = defaultdict(lambda: {
        "level5_claims": 0, "em_claims": 0,
        "mod25_claims": 0, "drug_test_units": 0,
        "unique_patients": set(), "peer_group_key": "internal_medicine:southwest",
        "history_level5_em_rate": [], "history_modifier_25_rate": [],
    })
    for claim in canonical_claims:
        pid = claim.get("rendering_provider_npi")
        if not pid:
            continue
        agg = by_provider[pid]
        if claim.get("patient_id"):
            agg["unique_patients"].add(claim["patient_id"])
        for line in claim["lines"]:
            if line["code"] in EM_OFFICE_VISIT_CODES:
                agg["em_claims"] += 1
                if line["code"] in HIGH_LEVEL_EM_CODES:
                    agg["level5_claims"] += 1
                if MODIFIER_25 in line["modifiers"]:
                    agg["mod25_claims"] += 1
            if line["code"] in DRUG_TEST_CODES:
                agg["drug_test_units"] += line["units"]

    for pid, agg in by_provider.items():
        em_total = max(agg["em_claims"], 1)
        agg["level5_em_rate"] = agg["level5_claims"] / em_total
        agg["modifier_25_rate"] = agg["mod25_claims"] / em_total
        agg["drug_test_units_per_patient"] = (
            agg["drug_test_units"] / max(len(agg["unique_patients"]), 1)
        )
        agg["unique_patients"] = len(agg["unique_patients"])
    return dict(by_provider)

def _compute_exposure(canonical_claims, resolved_entities):
    """Sum billed amounts by rendering-provider canonical ID for exposure."""
    exposure = defaultdict(lambda: Decimal("0"))
    for claim in canonical_claims:
        provider = resolved_entities.get(claim.get("rendering_provider_npi"))
        if provider:
            exposure[provider] += (claim.get("billed_amount_total") or Decimal("0"))
    return dict(exposure)