Recipe 3.10: Python Implementation Example

pip install boto3 scikit-learn pandas numpy statsmodels shapely

import io
import json
import logging
import math
import uuid
from collections import defaultdict, Counter
from datetime import datetime, timedelta, timezone, date
from decimal import Decimal
from typing import Optional

import boto3
import numpy as np
import pandas as pd
from botocore.config import Config
from boto3.dynamodb.conditions import Key, Attr
from shapely.geometry import Point, Polygon
from sklearn.linear_model import LogisticRegression
from sklearn.isotonic import IsotonicRegression

# Structured logging. Ship JSON records to CloudWatch Logs Insights.
# Encounter payloads, chief-complaint text, line lists, lab results,
# and Bedrock prompts contain PHI. Log structural metadata only. Never
# log full encounter payloads with patient identifiers, full chief-
# complaint text, full line lists, or Bedrock prompts in application
# logs. The line-list search index (OpenSearch) and the cluster-state
# store (DynamoDB) are the right home for full payloads, behind KMS
# and CloudTrail data events.
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Adaptive retry handles throttling across DynamoDB, Timestream, Kinesis,
# Comprehend Medical, SageMaker, and Bedrock with exponential backoff
# and jitter. Encounter ingest is bursty (EDs flush large batches at
# top-of-hour boundaries; lab results spike during morning rounds), and
# adaptive mode keeps burst windows from cascading into retry storms
# against Comprehend Medical and the syndrome classifier endpoint.
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)
kinesis = boto3.client("kinesis", region_name=REGION, config=BOTO3_RETRY_CONFIG)
timestream_write = boto3.client("timestream-write", region_name=REGION, config=BOTO3_RETRY_CONFIG)
timestream_query = boto3.client("timestream-query", 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)
eventbridge = boto3.client("events", region_name=REGION, config=BOTO3_RETRY_CONFIG)
comprehend_medical = boto3.client(
    "comprehendmedical", region_name=REGION, config=BOTO3_RETRY_CONFIG
)
sagemaker_runtime = boto3.client("sagemaker-runtime", region_name=REGION, config=BOTO3_RETRY_CONFIG)
bedrock_runtime = boto3.client("bedrock-runtime", region_name=REGION, config=BOTO3_RETRY_CONFIG)
location = boto3.client("location", region_name=REGION, config=BOTO3_RETRY_CONFIG)

# --- Resource Names ---
# Fill in with your actual resource names. These are placeholders.
CELL_STATE_TABLE             = "cell-state"
CLUSTER_STATE_TABLE          = "cluster-state"
GEOCODE_CACHE_TABLE          = "address-geocode-cache"
SUPPRESSION_RULES_TABLE      = "suppression-rules"

SURVEILLANCE_EVENTS_STREAM   = "surveillance-events"
TIMESTREAM_DATABASE          = "surveillance"
TIMESTREAM_CELL_TABLE        = "cell_time_series"

RAW_EVENTS_BUCKET            = "my-surveillance-raw-events"
BASELINE_STORE_BUCKET        = "my-surveillance-baseline-store"
TRAINING_LABELS_BUCKET       = "my-surveillance-training-labels"

CLUSTER_BUS                  = "surveillance-events"
SYNDROME_CLASSIFIER_ENDPOINT = "syndrome-classifier-v3"
BEDROCK_MODEL_ID             = "anthropic.claude-3-sonnet-20240229-v1:0"
LOCATION_PLACE_INDEX         = "surveillance-place-index"

# --- Syndromic Categories ---
# NSSP's standard syndromic categories form the baseline taxonomy.
# Production loads the full taxonomy from a versioned reference table
# the surveillance team owns. The teaching example uses a small subset.
SYNDROMIC_CATEGORIES = {
    "fever_respiratory",   # fever + cough or shortness of breath
    "ili",                 # influenza-like illness (fever + cough/sore throat)
    "gi",                  # gastrointestinal: vomiting, diarrhea, abdominal pain
    "rash",                # rash with or without fever
    "neuro",               # altered mental status, seizure, meningitis-like
    "sepsis",              # sepsis indicators
    "asthma_copd",         # asthma / COPD exacerbation
}

# ICD-10 to syndrome mapping (small subset). Production uses NSSP's
# official chief-complaint-and-discharge-diagnosis (CCDD) syndrome
# definitions plus organization-specific extensions. The CCDD library
# is regularly updated; mirror it in a versioned reference table.
ICD_TO_SYNDROME_RULES = [
    {"pattern": r"^J0[6-9]",           "syndrome": "fever_respiratory"},  # acute upper respiratory
    {"pattern": r"^J1[0-8]",           "syndrome": "ili"},                # influenza, pneumonia
    {"pattern": r"^J2[0-2]",           "syndrome": "fever_respiratory"},  # bronchitis, bronchiolitis
    {"pattern": r"^J45",               "syndrome": "asthma_copd"},        # asthma
    {"pattern": r"^J44",               "syndrome": "asthma_copd"},        # COPD
    {"pattern": r"^A0[0-9]",           "syndrome": "gi"},                  # cholera, salmonella
    {"pattern": r"^A09",               "syndrome": "gi"},                  # gastroenteritis
    {"pattern": r"^K5[0-2]",           "syndrome": "gi"},                  # IBD, gastritis
    {"pattern": r"^B0[0-9]",           "syndrome": "rash"},                # viral exanthems
    {"pattern": r"^L5[0-9]",           "syndrome": "rash"},                # urticaria, dermatitis
    {"pattern": r"^G0[0-3]",           "syndrome": "neuro"},               # meningitis
    {"pattern": r"^R56",               "syndrome": "neuro"},               # convulsions
    {"pattern": r"^A41",               "syndrome": "sepsis"},              # septicemia
    {"pattern": r"^R65",               "syndrome": "sepsis"},              # SIRS / sepsis
]

# Comprehend-Medical-detected entity to syndrome mapping. Production
# uses a richer ontology mapping (UMLS-codes-to-syndrome) plus a
# trained classifier; the teaching example uses keyword-matching on
# the entity text.
ENTITY_TO_SYNDROME_RULES = [
    {"keywords": ["cough", "shortness of breath", "wheezing"], "syndrome": "fever_respiratory"},
    {"keywords": ["influenza", "flu", "ili"],                    "syndrome": "ili"},
    {"keywords": ["fever"],                                       "syndrome": "fever_respiratory"},
    {"keywords": ["vomiting", "diarrhea", "abdominal pain"],     "syndrome": "gi"},
    {"keywords": ["rash"],                                        "syndrome": "rash"},
    {"keywords": ["seizure", "altered mental status", "confusion"], "syndrome": "neuro"},
    {"keywords": ["sepsis", "septic shock"],                      "syndrome": "sepsis"},
    {"keywords": ["asthma", "copd", "wheezing"],                 "syndrome": "asthma_copd"},
]

# Lab-positive pathogen to syndrome promotion. A positive flu PCR
# promotes the encounter to ILI. A positive Salmonella culture promotes
# to GI. Production uses LOINC plus organism-specific maps; the demo
# uses a small dict.
PATHOGEN_TO_SYNDROME_RULES = [
    {"pathogens": ["influenza_a", "influenza_b"],                  "syndrome": "ili"},
    {"pathogens": ["sars_cov_2"],                                   "syndrome": "fever_respiratory"},
    {"pathogens": ["rsv"],                                          "syndrome": "fever_respiratory"},
    {"pathogens": ["salmonella", "shigella", "campylobacter"],     "syndrome": "gi"},
    {"pathogens": ["norovirus"],                                    "syndrome": "gi"},
    {"pathogens": ["measles", "rubella", "varicella"],             "syndrome": "rash"},
    {"pathogens": ["neisseria_meningitidis"],                       "syndrome": "neuro"},
]

SYNDROME_CONFIDENCE_THRESHOLD = 0.65

# --- Detector Thresholds ---
# Tunable per cohort and per syndrome in production. Loaded from a
# versioned governance-approved table so the surveillance team can
# update without a code deploy.
CUSUM_REFERENCE_K        = 0.5     # reference value: half a SD shift
CUSUM_DECISION_H         = 4.0     # decision interval: ~4 SD cumulative
EWMA_LAMBDA              = 0.4     # smoothing factor
EWMA_CONTROL_LIMIT       = 3.0     # in standard deviations
SCAN_PVALUE_THRESHOLD    = 0.01    # spatiotemporal scan significance
CROSS_SYNDROME_THRESHOLD = 0.7

WW_ANOMALY_THRESHOLD       = 2.0    # standardized anomaly score
RX_SPIKE_THRESHOLD         = 2.5
ABSENTEEISM_THRESHOLD      = 2.0

# Multi-source fusion weights. Concordance-bonus rewards multiple
# sources moving together. Calibrated separately per syndrome class
# in production.
FUSION_WEIGHTS = {
    "clinical":    0.50,
    "wastewater":  0.20,
    "pharmacy":    0.15,
    "absenteeism": 0.15,
}
CONCORDANCE_SIGNAL_THRESHOLD = 0.5
CONCORDANCE_BONUS_PER_SOURCE = 0.05

# --- Tier Thresholds ---
# Tier-1 clusters get same-day surveillance team review. Tier-2 within
# 24 hours. Tier-3 next business day. Tunable per syndrome class in
# production.
TIER_THRESHOLDS = {
    "DEFAULT": {"tier_1": 0.85, "tier_2": 0.65, "tier_3": 0.45},
}

# --- Window Sizes ---
BASELINE_LOOKBACK_YEARS    = 5
MIN_HISTORY_DAYS           = 365
MIN_HISTORY_COUNT          = 100
CLUSTER_TEMPORAL_WINDOW    = 14   # days; group flagged cells within this
DISMISSAL_VALIDITY_PERIOD_DAYS = 21

# --- Aggregation Windows ---
# Multiple windows in parallel catch outbreaks at different temporal
# scales. Daily for fast-moving, 7-day for smoothing, 14-day and 28-day
# for slower-burn detection.
AGGREGATION_WINDOW_DAYS = [1, 7, 14, 28]

def _to_decimal(value, precision="0.0001"):
    """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 baseline expected counts, deviation
    z-scores, p-values, and composite scores through Decimal on the way
    in and back out.
    """
    if value is None:
        return None
    return Decimal(str(value)).quantize(Decimal(precision))

def _decimalize(obj):
    """Recursively convert floats to Decimals for DynamoDB write."""
    if isinstance(obj, float):
        return _to_decimal(obj)
    if isinstance(obj, dict):
        return {k: _decimalize(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_decimalize(v) for v in obj]
    return obj

def _undecimalize(obj):
    """Inverse of _decimalize for read-side conversion to Python-native types."""
    if isinstance(obj, Decimal):
        return float(obj)
    if isinstance(obj, dict):
        return {k: _undecimalize(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_undecimalize(v) for v in obj]
    return obj

def generate_event_id():
    """Generate a surveillance event identifier."""
    return f"EV-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}-{uuid.uuid4().hex[:10]}"

def pseudonymize(patient_identifier):
    """Replace a clinical patient identifier with a surveillance pseudonym.

    Production maintains a separate, access-controlled mapping service
    that holds the clinical-to-surveillance mapping. The analytic
    pipeline operates on the pseudonym only; investigators access the
    case detail (with the clinical identifier) under specific authority
    through a separately controlled service.

    The teaching example uses a deterministic hash so the same patient
    always maps to the same pseudonym across encounters; production uses
    a keyed HMAC with the key stored in KMS / Secrets Manager.
    """
    import hashlib
    return "SP-" + hashlib.sha256(patient_identifier.encode("utf-8")).hexdigest()[:12]

def bucket_age(age_years):
    """Bucket continuous age into surveillance age groups."""
    if age_years is None:
        return "unknown"
    if age_years < 5:
        return "under_5"
    if age_years < 18:
        return "5_17"
    if age_years < 50:
        return "18_49"
    if age_years < 65:
        return "50_64"
    return "65_plus"

def parse_by_source(raw_message, source_id):
    """Parse a source-specific message into a normalized intermediate dict.

    Production has per-source parsers: HL7 v2 ADT (hl7apy), HL7 v2 ORU,
    FHIR Encounter and Observation (fhir.resources), eCR documents, NWSS
    wastewater records, NHSN HAI reports. Each has its own schema and
    quirks. The teaching example treats raw_message as a pre-shaped dict
    so the demo runs without a real parser stack.
    """
    return raw_message

def ingest_encounter(raw_message, source_id):
    """Receive an encounter event, normalize, and put on the event stream.

    Production wires this as a Lambda triggered by the encounter-feed
    connector for each source class. The Lambda parses the source-
    specific payload, performs identifier resolution, builds the
    canonical event, and writes to Kinesis. The downstream geocoding
    Lambda picks it up.
    """
    parsed = parse_by_source(raw_message, source_id)

    # Build the canonical surveillance event. Note the deliberate
    # separation: the analytic event uses surveillance_pid (pseudonym),
    # not the clinical patient identifier. The clinical identifier
    # stays in the case-detail store, accessed only when an
    # investigator opens a cluster under appropriate authority.
    canonical = {
        "event_id":           generate_event_id(),
        "source_id":          source_id,
        "source_event_id":    parsed.get("encounter_id"),
        "observed_at":        parsed["arrival_at"],
        "ingested_at":        datetime.now(timezone.utc).isoformat(),
        "encounter_type":     parsed.get("encounter_type"),
        "facility_id":        parsed.get("facility_id"),
        "surveillance_pid":   pseudonymize(parsed["patient_identifier"]),
        "chief_complaint":    parsed.get("chief_complaint_text"),
        "triage_note":        parsed.get("triage_note_text"),
        "diagnoses_admit":    parsed.get("diagnosis_codes_admit", []),
        "diagnoses_final":    parsed.get("diagnosis_codes_final", []),
        "age_years":          parsed.get("age_years"),
        "age_group":          bucket_age(parsed.get("age_years")),
        "sex":                parsed.get("sex"),
        "race_ethnicity":     parsed.get("race_ethnicity"),
        "residence_address":  parsed.get("patient_address", {}),
        "residence_zip":      parsed.get("patient_address", {}).get("zip"),
    }

    # Append to the canonical surveillance-event stream. Partition by
    # surveillance_pid so repeat encounters from the same patient
    # arrive in order to the same shard.
    kinesis.put_record(
        StreamName=SURVEILLANCE_EVENTS_STREAM,
        Data=json.dumps(canonical, default=str).encode("utf-8"),
        PartitionKey=canonical["surveillance_pid"],
    )

    # Persist the raw event in the lake for replay and audit. The raw
    # event still contains the clinical patient identifier; it lives
    # behind KMS in S3 and is accessed only by authorized roles for
    # retrospective analysis and outbreak investigation.
    obs_at = canonical["observed_at"]
    s3_client.put_object(
        Bucket=RAW_EVENTS_BUCKET,
        Key=(
            f"source={source_id}/year={obs_at[:4]}/"
            f"month={obs_at[5:7]}/day={obs_at[8:10]}/"
            f"{canonical['event_id']}.json"
        ),
        Body=json.dumps(parsed, default=str).encode("utf-8"),
        ServerSideEncryption="aws:kms",
    )

    return {"statusCode": 200, "event_id": canonical["event_id"], "canonical": canonical}

# Synthetic geography reference for the demo. Production stores the
# full geography hierarchy in Aurora PostgreSQL with PostGIS, with
# annual refreshes from TIGER/Line and operational geography updates.
_DEMO_GEOGRAPHY = {
    "census_tracts": [
        {"id": "36055-001100", "polygon": Polygon([(-77.65, 43.10), (-77.60, 43.10),
                                                    (-77.60, 43.15), (-77.65, 43.15)]),
         "zcta": "14620", "county": "36055", "school_district": "ROC-CITY",
         "sewershed": "ROC-CENTRAL", "hospital_service_area": "HSA-ROC-1"},
        {"id": "36055-001200", "polygon": Polygon([(-77.60, 43.10), (-77.55, 43.10),
                                                    (-77.55, 43.15), (-77.60, 43.15)]),
         "zcta": "14620", "county": "36055", "school_district": "ROC-CITY",
         "sewershed": "ROC-CENTRAL", "hospital_service_area": "HSA-ROC-1"},
        {"id": "36055-001300", "polygon": Polygon([(-77.65, 43.05), (-77.60, 43.05),
                                                    (-77.60, 43.10), (-77.65, 43.10)]),
         "zcta": "14620", "county": "36055", "school_district": "ROC-CITY",
         "sewershed": "ROC-CENTRAL", "hospital_service_area": "HSA-ROC-1"},
        {"id": "36055-001400", "polygon": Polygon([(-77.60, 43.05), (-77.55, 43.05),
                                                    (-77.55, 43.10), (-77.60, 43.10)]),
         "zcta": "14620", "county": "36055", "school_district": "ROC-CITY",
         "sewershed": "ROC-CENTRAL", "hospital_service_area": "HSA-ROC-1"},
        {"id": "36055-002100", "polygon": Polygon([(-77.50, 43.10), (-77.45, 43.10),
                                                    (-77.45, 43.15), (-77.50, 43.15)]),
         "zcta": "14622", "county": "36055", "school_district": "ROC-EAST",
         "sewershed": "ROC-EAST", "hospital_service_area": "HSA-ROC-2"},
    ],
}

def format_address(address):
    """Format an address dict as a single string for the geocoder."""
    parts = [
        address.get("street", ""),
        address.get("city", ""),
        address.get("state", ""),
        address.get("zip", ""),
    ]
    return ", ".join(p for p in parts if p)

def geocode_with_amazon_location(address):
    """Geocode an address using Amazon Location Service.

    Production calls Amazon Location's place-index search. The teaching
    example returns a stubbed coordinate based on the ZIP code so the
    demo runs without provisioning a place index.
    """
    # Production:
    # response = location.search_place_index_for_text(
    #     IndexName=LOCATION_PLACE_INDEX,
    #     Text=format_address(address),
    #     MaxResults=1,
    # )
    # if response["Results"]:
    #     coords = response["Results"][0]["Place"]["Geometry"]["Point"]
    #     return {"lon": coords[0], "lat": coords[1]}
    # return None
    zip_to_coords = {
        "14620": {"lon": -77.625, "lat": 43.125},
        "14622": {"lon": -77.475, "lat": 43.125},
        "10001": {"lon": -73.997, "lat": 40.751},
        "94110": {"lon": -122.418, "lat": 37.748},
    }
    return zip_to_coords.get(address.get("zip"), {"lon": -77.625, "lat": 43.125})

def query_postgis_for_admin_geographies(coords):
    """Spatial join: find every administrative geography that contains the point.

    Production runs SQL like:
        SELECT
          t.id   AS census_tract,
          t.zcta AS zcta,
          t.county AS county,
          t.school_district AS school_district,
          t.sewershed AS sewershed,
          t.hospital_service_area AS hospital_service_area
        FROM   census_tracts t
        WHERE  ST_Contains(t.geometry, ST_SetSRID(ST_MakePoint(:lon, :lat), 4326))

    Against an Aurora PostgreSQL with PostGIS database. The teaching
    example walks the in-process Shapely polygons.
    """
    point = Point(coords["lon"], coords["lat"])
    for tract in _DEMO_GEOGRAPHY["census_tracts"]:
        if tract["polygon"].contains(point):
            return {
                "census_tract":           tract["id"],
                "zcta":                   tract["zcta"],
                "county":                 tract["county"],
                "school_district":        tract["school_district"],
                "sewershed":              tract["sewershed"],
                "hospital_service_area":  tract["hospital_service_area"],
            }
    return {
        "census_tract":           None,
        "zcta":                   None,
        "county":                 None,
        "school_district":        None,
        "sewershed":              None,
        "hospital_service_area":  None,
    }

def hash_address(address):
    """Compute a stable hash key for the geocode cache."""
    import hashlib
    canonical = format_address(address).lower().strip()
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:24]

def geocode_and_stratify(canonical_event):
    """Attach geographic stratification to a canonical event.

    Production caches the geocode-and-spatial-join result keyed on a
    stable address hash so repeat encounters from the same patient
    don't re-geocode unnecessarily. The cache lives in DynamoDB; entries
    expire after a configurable TTL to handle address changes and
    boundary refreshes.
    """
    if not canonical_event.get("residence_address"):
        # No address. Some encounters arrive without one (homeless
        # patients, data-entry omissions, walk-in encounters with
        # incomplete registration). Production routes these to a
        # facility-only geographic assignment as a fallback; the
        # teaching example sets all geographies to None.
        canonical_event.update({
            "coords":                None,
            "census_tract":          None,
            "zcta":                  canonical_event.get("residence_zip"),
            "county":                None,
            "school_district":       None,
            "sewershed":             None,
            "hospital_service_area": None,
        })
        return canonical_event

    cache_table = dynamodb.Table(GEOCODE_CACHE_TABLE)
    address_hash = hash_address(canonical_event["residence_address"])

    cached_response = cache_table.get_item(Key={"address_hash": address_hash})
    cached = _undecimalize(cached_response.get("Item"))

    if cached:
        coords = cached["coords"]
        admin_geographies = cached["admin_geographies"]
    else:
        coords = geocode_with_amazon_location(canonical_event["residence_address"])
        if coords is None:
            # Geocode failure (ungeocodable address, service error). Fall
            # back to ZCTA-level assignment based on the residence ZIP.
            admin_geographies = {
                "census_tract":          None,
                "zcta":                  canonical_event.get("residence_zip"),
                "county":                None,
                "school_district":       None,
                "sewershed":             None,
                "hospital_service_area": None,
            }
        else:
            admin_geographies = query_postgis_for_admin_geographies(coords)

        cache_table.put_item(Item=_decimalize({
            "address_hash":      address_hash,
            "coords":            coords,
            "admin_geographies": admin_geographies,
            "geocoded_at":       datetime.now(timezone.utc).isoformat(),
        }))

    canonical_event["coords"]                = coords
    canonical_event.update(admin_geographies)
    return canonical_event

import re

def apply_icd_rules(diagnosis_codes, syndromes_set):
    """Apply ICD-10-pattern-to-syndrome rules to a list of diagnosis codes."""
    for code in diagnosis_codes or []:
        for rule in ICD_TO_SYNDROME_RULES:
            if re.match(rule["pattern"], code or ""):
                syndromes_set.add(rule["syndrome"])
    return syndromes_set

def call_comprehend_medical(text):
    """Call Amazon Comprehend Medical for entity extraction.

    Comprehend Medical extracts conditions, anatomy, medications, and
    signs/symptoms from unstructured clinical text. The teaching
    example wraps the call in a try/except so the demo can run without
    a live API call (returns an empty entity list on failure).
    """
    if not text or not text.strip():
        return []
    try:
        response = comprehend_medical.detect_entities_v2(Text=text)
        return response.get("Entities", [])
    except Exception as e:
        logger.warning("comprehend medical call failed", extra={"error": str(e)})
        _emit_metric("ComprehendMedicalFailed", 1)
        return []

def apply_entity_rules(entities, syndromes_set):
    """Map Comprehend-Medical-detected entities to syndromic categories."""
    for entity in entities or []:
        category = entity.get("Category", "").upper()
        if category not in {"MEDICAL_CONDITION", "SIGN_SYMPTOM", "DX_NAME"}:
            continue
        text = entity.get("Text", "").lower()
        for rule in ENTITY_TO_SYNDROME_RULES:
            for keyword in rule["keywords"]:
                if keyword in text:
                    syndromes_set.add(rule["syndrome"])
                    break
    return syndromes_set

def score_via_syndrome_classifier(text, age_group):
    """Invoke the deployed SageMaker syndrome-classifier endpoint.

    Production hosts a fine-tuned transformer (or a similar text
    classifier) on a SageMaker endpoint. The teaching example returns
    a small synthetic prediction list so the path runs without a
    deployed endpoint.
    """
    if not text or not text.strip():
        return []
    try:
        payload = json.dumps({"text": text, "age_group": age_group})
        response = sagemaker_runtime.invoke_endpoint(
            EndpointName=SYNDROME_CLASSIFIER_ENDPOINT,
            ContentType="application/json",
            Body=payload.encode("utf-8"),
        )
        body = json.loads(response["Body"].read().decode("utf-8"))
        return body.get("predictions", [])
    except Exception as e:
        # In the demo, fall back to a tiny heuristic so the path runs.
        logger.debug("syndrome classifier endpoint unavailable; using fallback",
                     extra={"error": str(e)})
        text_lower = text.lower()
        predictions = []
        if "fever" in text_lower and ("cough" in text_lower or "respiratory" in text_lower):
            predictions.append({"syndrome": "fever_respiratory", "confidence": 0.85})
        if "vomit" in text_lower or "diarrhea" in text_lower:
            predictions.append({"syndrome": "gi", "confidence": 0.82})
        if "rash" in text_lower:
            predictions.append({"syndrome": "rash", "confidence": 0.78})
        if "seizure" in text_lower or "altered mental" in text_lower:
            predictions.append({"syndrome": "neuro", "confidence": 0.80})
        return predictions

def lookup_recent_lab_results(surveillance_pid, window_days=14):
    """Look up recent lab results for the patient.

    Production queries the lab-event store (a separate stream and
    OpenSearch index for lab data). The teaching example returns an
    empty list; the lab integration path is documented but not
    exercised in the demo.
    """
    return []

def apply_lab_rules(lab_results, syndromes_set):
    """Promote encounters to pathogen-specific syndromic categories."""
    for lab in lab_results or []:
        if not lab.get("result_positive"):
            continue
        for rule in PATHOGEN_TO_SYNDROME_RULES:
            if lab.get("pathogen") in rule["pathogens"]:
                syndromes_set.add(rule["syndrome"])
    return syndromes_set

def classify_syndrome(canonical_event):
    """Classify an encounter into one or more syndromic categories.

    Returns the event with a `syndromes` list attached. Multi-label
    classification is intentional: a single encounter often hits
    multiple syndromic categories, and the surveillance system runs
    detectors against each category independently.
    """
    syndromes = set()

    # Structured-data rules first. Cheap, deterministic, high-precision
    # for the patterns they cover.
    apply_icd_rules(canonical_event.get("diagnoses_admit", []), syndromes)
    apply_icd_rules(canonical_event.get("diagnoses_final", []), syndromes)

    # Free-text NLP. The bulk of real-time signal lives here because
    # final diagnoses arrive hours or days after the encounter.
    free_text_parts = [
        canonical_event.get("chief_complaint") or "",
        canonical_event.get("triage_note") or "",
    ]
    free_text = " ".join(p for p in free_text_parts if p).strip()

    if free_text:
        entities = call_comprehend_medical(free_text)
        apply_entity_rules(entities, syndromes)

        # Custom syndrome classifier handles patterns the rule library
        # misses. Fine-tuned periodically on adjudicated cases.
        predictions = score_via_syndrome_classifier(
            free_text, canonical_event.get("age_group")
        )
        for prediction in predictions:
            if prediction.get("confidence", 0.0) > SYNDROME_CONFIDENCE_THRESHOLD:
                syndromes.add(prediction["syndrome"])

    # Lab-confirmed pathogens promote the encounter to specific
    # syndromic categories. Lab events arrive on a separate stream;
    # this is the encounter-side hook that joins recent lab results
    # to the encounter.
    lab_results = lookup_recent_lab_results(
        canonical_event["surveillance_pid"], window_days=14
    )
    apply_lab_rules(lab_results, syndromes)

    canonical_event["syndromes"] = sorted(syndromes)
    return canonical_event

def stratifications_for(canonical_event):
    """Return the demographic stratifications the event participates in.

    Multiple stratifications in parallel let the detector catch
    subgroup signals (age-skewed pediatric outbreaks, sex-skewed
    foodborne clusters) that population-wide aggregation misses.
    Multiple-comparison handling is the cost; pay it explicitly.
    """
    strats = ["all_ages"]
    if canonical_event.get("age_group") and canonical_event["age_group"] != "unknown":
        strats.append(f"age_{canonical_event['age_group']}")
    if canonical_event.get("sex") in ("male", "female"):
        strats.append(f"sex_{canonical_event['sex']}")
    return strats

def cell_keys_for(canonical_event):
    """Build the cell key set this event participates in.

    A single event might update dozens of cells. The cardinality is:
    geographies (5-6) x stratifications (3-4) x syndromes (1-3) x
    temporal windows (4). Production keeps the cell-counter table
    sized for the resulting volume; daily writes can run into hundreds
    of millions per state-level program.
    """
    geographies = []
    for geo_type, geo_id in [
        ("census_tract",          canonical_event.get("census_tract")),
        ("zcta",                  canonical_event.get("zcta")),
        ("county",                canonical_event.get("county")),
        ("school_district",       canonical_event.get("school_district")),
        ("sewershed",             canonical_event.get("sewershed")),
        ("hospital_service_area", canonical_event.get("hospital_service_area")),
    ]:
        if geo_id:
            geographies.append((geo_type, geo_id))

    cell_keys = []
    for geo_type, geo_id in geographies:
        for strat in stratifications_for(canonical_event):
            for syndrome in canonical_event.get("syndromes", []):
                for window_days in AGGREGATION_WINDOW_DAYS:
                    cell_keys.append({
                        "geo_type":  geo_type,
                        "geo_id":    geo_id,
                        "strat":     strat,
                        "syndrome":  syndrome,
                        "window":    f"{window_days}d",
                    })
    return cell_keys

def cell_partition_key(cell_key):
    """Render a cell key as a single DynamoDB partition key string."""
    return (
        f"{cell_key['geo_type']}|{cell_key['geo_id']}|"
        f"{cell_key['strat']}|{cell_key['syndrome']}|{cell_key['window']}"
    )

def increment_cell(cell_key, observed_at):
    """Atomically increment a cell's count and update last-event timestamp."""
    cell_table = dynamodb.Table(CELL_STATE_TABLE)
    cell_table.update_item(
        Key={"cell_id": cell_partition_key(cell_key)},
        UpdateExpression=(
            "ADD #c :one "
            "SET last_event_at = :now, "
            "    geo_type = if_not_exists(geo_type, :gt), "
            "    geo_id   = if_not_exists(geo_id, :gi), "
            "    strat    = if_not_exists(strat, :s), "
            "    syndrome = if_not_exists(syndrome, :sy), "
            "    #w       = if_not_exists(#w, :win)"
        ),
        ExpressionAttributeNames={"#c": "count", "#w": "window"},
        ExpressionAttributeValues={
            ":one": 1,
            ":now": observed_at,
            ":gt":  cell_key["geo_type"],
            ":gi":  cell_key["geo_id"],
            ":s":   cell_key["strat"],
            ":sy":  cell_key["syndrome"],
            ":win": cell_key["window"],
        },
    )

def write_to_timestream(cell_key, observed_at):
    """Append a per-cell increment to the Timestream cell-time-series table.

    Production writes one record per increment to Timestream with
    cell-key dimensions and a count_increment measure. Baseline
    computation reads multi-year history from Timestream. The teaching
    example wraps the call in try/except so the path runs without a
    Timestream database; an in-memory store handles the demo flow
    (see _IN_MEMORY_TIMESTREAM below).
    """
    record = {
        "Dimensions": [
            {"Name": "geo_type", "Value": cell_key["geo_type"]},
            {"Name": "geo_id",   "Value": cell_key["geo_id"]},
            {"Name": "strat",    "Value": cell_key["strat"]},
            {"Name": "syndrome", "Value": cell_key["syndrome"]},
            {"Name": "window",   "Value": cell_key["window"]},
        ],
        "MeasureName":      "count_increment",
        "MeasureValue":     "1",
        "MeasureValueType": "BIGINT",
        "Time":             str(int(datetime.fromisoformat(
                                 observed_at.replace("Z", "+00:00")
                             ).timestamp() * 1000)),
        "TimeUnit":         "MILLISECONDS",
    }
    try:
        timestream_write.write_records(
            DatabaseName=TIMESTREAM_DATABASE,
            TableName=TIMESTREAM_CELL_TABLE,
            Records=[record],
        )
    except Exception as e:
        logger.debug("timestream write fallback to in-memory store",
                     extra={"error": str(e)})
        _in_memory_timestream_append(cell_key, observed_at)

# In-memory Timestream stand-in for the demo. Production reads from
# real Timestream; the demo populates and reads this dict.
_IN_MEMORY_TIMESTREAM = defaultdict(list)

def _in_memory_timestream_append(cell_key, observed_at):
    pk = cell_partition_key(cell_key)
    _IN_MEMORY_TIMESTREAM[pk].append(observed_at)

def update_cell_counters(canonical_event):
    """Increment every cell the event participates in."""
    cell_keys = cell_keys_for(canonical_event)
    for cell_key in cell_keys:
        increment_cell(cell_key, canonical_event["observed_at"])
        write_to_timestream(cell_key, canonical_event["observed_at"])
    return cell_keys

def query_timestream_for_history(cell_key, lookback_years):
    """Query Timestream for multi-year per-day count history.

    Production runs SQL like:
        SELECT bin(time, 1d) AS day, sum(measure_value) AS count
        FROM   surveillance.cell_time_series
        WHERE  geo_type = '<geo_type>' AND geo_id = '<geo_id>'
          AND  strat = '<strat>' AND syndrome = '<syndrome>'
          AND  window = '1d'
          AND  time BETWEEN ago(5y) AND now()
        GROUP BY bin(time, 1d)

    The teaching example reads from the in-memory store and synthesizes
    a per-day count series. Production reads from real Timestream.
    """
    pk = cell_partition_key(cell_key)
    timestamps = _IN_MEMORY_TIMESTREAM.get(pk, [])
    if not timestamps:
        return pd.DataFrame(columns=["day", "count"])

    days = [datetime.fromisoformat(t.replace("Z", "+00:00")).date()
            for t in timestamps]
    series = pd.Series(days).value_counts().sort_index()
    df = pd.DataFrame({"day": series.index, "count": series.values})
    df["day"] = pd.to_datetime(df["day"])
    return df

def known_outbreak_dates(cell_key):
    """Return historical outbreak windows to exclude from baseline fitting.

    Baseline contamination is a real problem: if last winter's flu
    season is in the training window, the baseline absorbs it and the
    detector under-flags the next flu season. Production maintains a
    known-outbreak registry (curated by the surveillance team) and
    excludes those date windows during baseline fitting. The teaching
    example returns an empty list.
    """
    return []

def fit_negative_binomial_glm(history_df, exclude_dates=None):
    """Fit a seasonal-and-trend GLM to per-day count history.

    Negative binomial regression handles the overdispersion that
    surveillance count data typically exhibits. Annual harmonics
    capture seasonal patterns; weekly harmonics capture day-of-week
    effects; a linear trend term captures secular drift.

    Production uses statsmodels' GLM with NegativeBinomial family or
    the equivalent in R's surveillance package. The teaching example
    fits a small Poisson GLM as a stand-in (negative binomial
    convergence is unreliable on tiny synthetic datasets).
    """
    if len(history_df) < 30:
        # Insufficient data; return a constant-mean baseline.
        mean_count = max(history_df["count"].mean(), 0.5)
        return {
            "model_type":      "constant_mean_fallback",
            "mean_count":      float(mean_count),
            "std_count":       float(max(history_df["count"].std() or 1.0, 1.0)),
        }

    df = history_df.copy()
    if exclude_dates:
        df = df[~df["day"].isin(pd.to_datetime(exclude_dates))]
    df = df.sort_values("day").reset_index(drop=True)

    # Build features: linear trend, annual harmonics, weekly harmonics.
    df["t"] = (df["day"] - df["day"].min()).dt.days.astype(float)
    df["annual_sin"] = np.sin(2 * np.pi * df["t"] / 365.25)
    df["annual_cos"] = np.cos(2 * np.pi * df["t"] / 365.25)
    df["weekly_sin"] = np.sin(2 * np.pi * df["t"] / 7.0)
    df["weekly_cos"] = np.cos(2 * np.pi * df["t"] / 7.0)
    df["dow"] = df["day"].dt.dayofweek

    # statsmodels Poisson GLM as a robust stand-in. Production uses
    # NegativeBinomial.
    try:
        import statsmodels.api as sm
        feature_cols = ["t", "annual_sin", "annual_cos", "weekly_sin", "weekly_cos"]
        X = sm.add_constant(df[feature_cols])
        y = df["count"]
        glm = sm.GLM(y, X, family=sm.families.Poisson()).fit()
        predicted = glm.predict(X)
        residuals = y - predicted
        std_resid = float(np.std(residuals)) or 1.0
        return {
            "model_type":         "poisson_glm",
            "feature_cols":       feature_cols,
            "params":             glm.params.to_dict(),
            "first_day":          df["day"].min().isoformat(),
            "predicted_per_day":  predicted.tolist(),
            "actual_per_day":     y.tolist(),
            "residual_std":       std_resid,
        }
    except Exception as e:
        logger.warning("GLM fit failed; using constant-mean fallback",
                       extra={"error": str(e)})
        return {
            "model_type":  "constant_mean_fallback",
            "mean_count":  float(history_df["count"].mean()),
            "std_count":   float(max(history_df["count"].std() or 1.0, 1.0)),
        }

def predict_baseline_for_date(model_summary, target_date):
    """Predict expected count and prediction intervals for a target date."""
    if model_summary["model_type"] == "constant_mean_fallback":
        mean = model_summary["mean_count"]
        std  = model_summary["std_count"]
        return {
            "expected": mean,
            "upper_95": mean + 1.96 * std,
            "upper_99": mean + 2.58 * std,
            "std":      std,
        }

    first_day = datetime.fromisoformat(model_summary["first_day"]).date()
    t = (target_date - first_day).days

    annual_sin = math.sin(2 * math.pi * t / 365.25)
    annual_cos = math.cos(2 * math.pi * t / 365.25)
    weekly_sin = math.sin(2 * math.pi * t / 7.0)
    weekly_cos = math.cos(2 * math.pi * t / 7.0)

    p = model_summary["params"]
    log_mu = (
        p.get("const", 0.0)
      + p.get("t", 0.0)            * t
      + p.get("annual_sin", 0.0)   * annual_sin
      + p.get("annual_cos", 0.0)   * annual_cos
      + p.get("weekly_sin", 0.0)   * weekly_sin
      + p.get("weekly_cos", 0.0)   * weekly_cos
    )
    expected = math.exp(log_mu)
    std      = max(model_summary["residual_std"], 1.0)

    return {
        "expected": expected,
        "upper_95": expected + 1.96 * std,
        "upper_99": expected + 2.58 * std,
        "std":      std,
    }

def downscale_parent(parent_model, cell):
    """Use the parent geography's model with a cell-specific offset.

    Hierarchical pooling: when a cell has insufficient history, fall back
    to the parent geography's baseline scaled by the cell's population
    fraction. Production runs this through a hierarchical Bayesian model
    (INLA, brms, or a custom Stan model). The teaching example just
    halves the parent's expected count as a placeholder.
    """
    parent_summary = parent_model
    if parent_summary["model_type"] == "constant_mean_fallback":
        return {
            "model_type": "constant_mean_fallback",
            "mean_count": parent_summary["mean_count"] * 0.5,
            "std_count":  max(parent_summary["std_count"] * 0.7, 1.0),
        }
    return parent_summary

def get_parent_baseline_model(cell_key):
    """Return the baseline model for the cell's parent geography.

    Census tract -> ZCTA -> county -> state. The teaching example just
    returns a constant-mean fallback so the path runs.
    """
    return {
        "model_type":  "constant_mean_fallback",
        "mean_count":  3.0,
        "std_count":   1.5,
    }

def enumerate_active_cells():
    """Return all currently-active cells in the system.

    Production iterates the cell-state table (or a derived index of
    active cells) so every cell with recent activity gets a baseline
    refresh. The teaching example walks the in-memory Timestream
    store.
    """
    cells = []
    for pk in _IN_MEMORY_TIMESTREAM.keys():
        parts = pk.split("|")
        if len(parts) == 5:
            cells.append({
                "geo_type":  parts[0],
                "geo_id":    parts[1],
                "strat":     parts[2],
                "syndrome":  parts[3],
                "window":    parts[4],
            })
    return cells

def compute_baselines(reference_date):
    """Recompute baselines for every active cell.

    Production runs this as a daily Step Functions workflow that fans
    out across active cells using AWS Batch or SageMaker Processing.
    Each cell's baseline is persisted to S3 (versioned per
    reference_date) and the per-day prediction is updated in DynamoDB
    cell-state for fast scoring.
    """
    cells = enumerate_active_cells()
    if not cells:
        return []

    baselines = []
    for cell in cells:
        # Only compute for the daily-window cells; the 7d/14d/28d
        # baselines are derived from the 1d series.
        if cell["window"] != "1d":
            continue

        history = query_timestream_for_history(cell, BASELINE_LOOKBACK_YEARS)
        excluded = known_outbreak_dates(cell)

        if (len(history) < MIN_HISTORY_DAYS
                or history["count"].sum() < MIN_HISTORY_COUNT):
            parent_model = get_parent_baseline_model(cell)
            cell_baseline = downscale_parent(parent_model, cell)
            baseline_source = "parent_pooled"
        else:
            cell_baseline = fit_negative_binomial_glm(
                history, exclude_dates=excluded
            )
            baseline_source = "cell_specific"

        prediction = predict_baseline_for_date(cell_baseline, reference_date)

        # Persist to S3 (versioned baseline store).
        baseline_record = {
            "cell":             cell,
            "expected":         prediction["expected"],
            "upper_95":         prediction["upper_95"],
            "upper_99":         prediction["upper_99"],
            "std":              prediction["std"],
            "model_summary":    cell_baseline,
            "source":           baseline_source,
            "reference_date":   reference_date.isoformat(),
            "computed_at":      datetime.now(timezone.utc).isoformat(),
        }

        try:
            s3_client.put_object(
                Bucket=BASELINE_STORE_BUCKET,
                Key=(
                    f"baseline/{cell['geo_type']}/{cell['geo_id']}/"
                    f"{cell['strat']}/{cell['syndrome']}/"
                    f"{reference_date.isoformat()}.json"
                ),
                Body=json.dumps(baseline_record, default=str).encode("utf-8"),
                ServerSideEncryption="aws:kms",
            )
        except Exception as e:
            logger.debug("baseline S3 write skipped",
                         extra={"error": str(e)})

        # Update cell-state with the latest expected and intervals.
        cell_table = dynamodb.Table(CELL_STATE_TABLE)
        cell_table.update_item(
            Key={"cell_id": cell_partition_key(cell)},
            UpdateExpression=(
                "SET expected = :e, upper_95 = :u95, upper_99 = :u99, "
                "    baseline_source = :src, baseline_at = :t"
            ),
            ExpressionAttributeValues={
                ":e":   _to_decimal(prediction["expected"]),
                ":u95": _to_decimal(prediction["upper_95"]),
                ":u99": _to_decimal(prediction["upper_99"]),
                ":src": baseline_source,
                ":t":   datetime.now(timezone.utc).isoformat(),
            },
        )

        baselines.append(baseline_record)

    return baselines

def get_recent_counts(cell_key, days=60):
    """Return the recent daily count series for a cell."""
    history = query_timestream_for_history(cell_key, lookback_years=1)
    if history.empty:
        return pd.DataFrame(columns=["day", "count"])
    cutoff = pd.Timestamp(datetime.now(timezone.utc).date() - timedelta(days=days))
    return history[history["day"] >= cutoff].copy()

def load_baseline_for_cell(cell_key, reference_date):
    """Load the baseline record persisted for the cell on this reference date."""
    cell_table = dynamodb.Table(CELL_STATE_TABLE)
    response = cell_table.get_item(Key={"cell_id": cell_partition_key(cell_key)})
    item = _undecimalize(response.get("Item")) or {}
    return {
        "expected":  item.get("expected", 1.0),
        "upper_95":  item.get("upper_95", 5.0),
        "upper_99":  item.get("upper_99", 7.0),
        "std":       max(item.get("upper_95", 5.0) - item.get("expected", 1.0), 1.0) / 1.96,
    }

def compute_cusum(observed, expected, sigma, k, h):
    """Compute a CUSUM control chart on the observed series.

    CUSUM (Cumulative Sum) accumulates standardized deviations above
    the expected level and signals when the running sum exceeds the
    decision threshold h. The reference value k is the size of shift
    (in standard deviations) the chart is tuned to detect quickly.

    Returns the most recent CUSUM value and a boolean flag.
    """
    if len(observed) == 0:
        return {"cusum_value": 0.0, "signal": False}
    z = (observed - expected) / max(sigma, 1e-6)
    cumsum = 0.0
    history = []
    for zi in z:
        cumsum = max(0.0, cumsum + zi - k)
        history.append(cumsum)
    return {
        "cusum_value":  history[-1],
        "signal":       history[-1] > h,
        "history":      history,
    }

def compute_ewma(observed, expected, sigma, lambda_val, limit):
    """Compute an EWMA control chart on the observed series.

    EWMA (Exponentially Weighted Moving Average) smooths the standardized
    deviations and flags when the smoothed value exceeds the control
    limit (in standard deviations of the smoothed series).
    """
    if len(observed) == 0:
        return {"ewma_value": 0.0, "signal": False}
    z = (observed - expected) / max(sigma, 1e-6)
    ewma_value = 0.0
    history = []
    for zi in z:
        ewma_value = lambda_val * zi + (1 - lambda_val) * ewma_value
        history.append(ewma_value)
    # The EWMA's standard deviation is sigma * sqrt(lambda / (2 - lambda)).
    ewma_sigma = math.sqrt(lambda_val / (2 - lambda_val))
    return {
        "ewma_value":   history[-1],
        "signal":       abs(history[-1]) > limit * ewma_sigma,
        "history":      history,
    }

def run_control_chart_detectors(reference_date):
    """Run CUSUM and EWMA per active cell."""
    results = []
    for cell in enumerate_active_cells():
        if cell["window"] != "1d":
            continue
        recent = get_recent_counts(cell, days=60)
        if recent.empty:
            continue
        baseline = load_baseline_for_cell(cell, reference_date)
        cusum_result = compute_cusum(
            recent["count"].values,
            baseline["expected"],
            baseline["std"],
            CUSUM_REFERENCE_K,
            CUSUM_DECISION_H,
        )
        ewma_result = compute_ewma(
            recent["count"].values,
            baseline["expected"],
            baseline["std"],
            EWMA_LAMBDA,
            EWMA_CONTROL_LIMIT,
        )
        flagged = bool(cusum_result["signal"] or ewma_result["signal"])
        results.append({
            "detector":        "control_chart",
            "cell":            cell,
            "cusum_value":     float(cusum_result["cusum_value"]),
            "ewma_value":      float(ewma_result["ewma_value"]),
            "flagged":         flagged,
            "reference_date":  reference_date.isoformat(),
        })
    return results

def run_farrington_flexible_detector(reference_date):
    """Run the Farrington Flexible regression-based detector per cell.

    Production runs this as a SageMaker Processing job per cell (or
    batched across cells), using the R `surveillance` package or a
    Python port. The teaching example computes a tail-probability
    score against the Poisson GLM baseline.
    """
    results = []
    for cell in enumerate_active_cells():
        if cell["window"] != "1d":
            continue
        baseline = load_baseline_for_cell(cell, reference_date)
        recent = get_recent_counts(cell, days=7)
        if recent.empty:
            continue
        observed_today = recent["count"].iloc[-1] if len(recent) else 0
        expected = baseline["expected"]
        upper_99 = baseline["upper_99"]

        # Standardized exceedance score: (observed - expected) / std.
        std = baseline["std"]
        exceedance = (observed_today - expected) / max(std, 1e-6)
        flagged = observed_today > upper_99

        results.append({
            "detector":        "farrington_flexible",
            "cell":            cell,
            "observed":        float(observed_today),
            "expected":        float(expected),
            "upper_99":        float(upper_99),
            "exceedance":      float(exceedance),
            "flagged":         bool(flagged),
            "reference_date":  reference_date.isoformat(),
        })
    return results

def submit_satscan_batch_job(scan_geography, reference_date, method,
                              max_window_pct=25, max_temporal_window_days=14):
    """Submit a SaTScan run as an AWS Batch job.

    Production stages a case file, a population file, and a parameter
    file into S3, then submits an AWS Batch job that runs the SaTScan
    binary in a container against the staged inputs. The job parses the
    output and writes significant clusters back to a results store.

    The teaching example replaces this with a tiny stub that flags
    cells where today's count substantially exceeds the upper-99
    interval and groups geographically adjacent flagged cells into
    a synthetic cluster. Use real SaTScan in production; this stub
    exists only so the path is exercised.
    """
    # Stub: scan all daily-window cells and group adjacent flagged cells.
    flagged_cells = []
    for cell in enumerate_active_cells():
        if cell["window"] != "1d" or cell["geo_type"] != "census_tract":
            continue
        baseline = load_baseline_for_cell(cell, reference_date)
        recent = get_recent_counts(cell, days=1)
        if recent.empty:
            continue
        observed_today = recent["count"].iloc[-1]
        if observed_today > baseline["upper_99"]:
            flagged_cells.append({
                "cell":       cell,
                "observed":   float(observed_today),
                "expected":   float(baseline["expected"]),
                "upper_99":   float(baseline["upper_99"]),
            })

    if len(flagged_cells) < 2:
        return {"significant_clusters": []}

    # Group flagged cells into a single cluster (the demo doesn't
    # actually do spatial proximity testing; real SaTScan does).
    cluster = {
        "geographies":     [c["cell"]["geo_id"] for c in flagged_cells],
        "syndromes":       list({c["cell"]["syndrome"] for c in flagged_cells}),
        "observed":        sum(c["observed"] for c in flagged_cells),
        "expected":        sum(c["expected"] for c in flagged_cells),
        "p_value":         0.005,    # synthetic; real SaTScan computes via permutation
        "relative_risk":   sum(c["observed"] for c in flagged_cells)
                            / max(sum(c["expected"] for c in flagged_cells), 1.0),
        "method":          method,
        "reference_date":  reference_date.isoformat(),
    }
    return {"significant_clusters": [cluster]}

def run_satscan_detectors(reference_date):
    """Run spatial and spatiotemporal scan statistics."""
    results = []
    spatial = submit_satscan_batch_job(
        "county_zcta", reference_date, "poisson", max_window_pct=25
    )
    for cluster in spatial["significant_clusters"]:
        if cluster["p_value"] < SCAN_PVALUE_THRESHOLD:
            results.append({
                "detector":        "satscan_spatial",
                "cluster":         cluster,
                "p_value":         cluster["p_value"],
                "relative_risk":   cluster["relative_risk"],
                "flagged":         True,
                "reference_date":  reference_date.isoformat(),
            })

    spacetime = submit_satscan_batch_job(
        "county_zcta", reference_date, "space_time_permutation",
        max_temporal_window_days=14,
    )
    for cluster in spacetime["significant_clusters"]:
        if cluster["p_value"] < SCAN_PVALUE_THRESHOLD:
            results.append({
                "detector":        "satscan_spacetime",
                "cluster":         cluster,
                "p_value":         cluster["p_value"],
                "relative_risk":   cluster["relative_risk"],
                "flagged":         True,
                "reference_date":  reference_date.isoformat(),
            })
    return results

def run_detector_bank(reference_date):
    """Run all detectors and return the combined result list."""
    all_results = []
    all_results.extend(run_control_chart_detectors(reference_date))
    all_results.extend(run_farrington_flexible_detector(reference_date))
    all_results.extend(run_satscan_detectors(reference_date))

    # Persist to OpenSearch and EventBridge for the cluster builder
    # and the audit archive.
    for result in all_results:
        try:
            eventbridge.put_events(Entries=[{
                "Source":      "surveillance.detector-bank",
                "DetailType":  "DetectorResult",
                "Detail":      json.dumps(result, default=str),
                "EventBusName": CLUSTER_BUS,
            }])
        except Exception as e:
            logger.debug("detector result event publish failed",
                         extra={"error": str(e)})
    return all_results

# Synthetic auxiliary-data store for the demo. Production reads from
# real source feeds (NWSS for wastewater, retail-pharmacy partnerships
# for pharmacy, state-DOE feeds for school absenteeism).
_DEMO_WASTEWATER = {
    "ROC-CENTRAL": {
        "sars_cov_2": [
            {"date": "2026-10-13", "concentration": 1.2e6},
            {"date": "2026-10-15", "concentration": 1.4e6},
            {"date": "2026-10-17", "concentration": 2.1e6},
            {"date": "2026-10-19", "concentration": 2.6e6},
            {"date": "2026-10-21", "concentration": 3.0e6},
            {"date": "2026-10-23", "concentration": 3.4e6},
        ],
    },
}

_DEMO_ABSENTEEISM = {
    "ROC-CITY": [
        {"date": "2026-10-15", "rate": 0.075},
        {"date": "2026-10-16", "rate": 0.080},
        {"date": "2026-10-19", "rate": 0.110},
        {"date": "2026-10-20", "rate": 0.140},
        {"date": "2026-10-21", "rate": 0.165},
        {"date": "2026-10-22", "rate": 0.180},
    ],
}

def load_ww_concentration_history(sewershed, pathogen):
    """Load wastewater concentration history for a sewershed and pathogen."""
    return _DEMO_WASTEWATER.get(sewershed, {}).get(pathogen, [])

def fit_ww_baseline(history):
    """Fit a simple baseline mean and std on log-transformed concentrations.

    Production normalizes against PMMoV (a viral indicator that controls
    for sample dilution) or against sewershed flow rate, fits a seasonal
    baseline, and applies pathogen-specific quality-control rules. The
    teaching example is a stand-in.
    """
    if not history:
        return {"mean": 1.0, "std": 1.0}
    values = np.array([math.log10(max(h["concentration"], 1.0)) for h in history[:-3]])
    if len(values) == 0:
        return {"mean": 1.0, "std": 1.0}
    return {"mean": float(values.mean()),
            "std":  float(max(values.std(), 0.1))}

def current_ww_concentration(sewershed, pathogen, reference_date):
    """Return the most recent wastewater concentration for the sewershed."""
    history = _DEMO_WASTEWATER.get(sewershed, {}).get(pathogen, [])
    if not history:
        return None
    return history[-1]["concentration"]

def run_wastewater_detector(reference_date):
    """Detect wastewater pathogen-concentration anomalies."""
    results = []
    for sewershed, pathogens in _DEMO_WASTEWATER.items():
        for pathogen, _ in pathogens.items():
            history = load_ww_concentration_history(sewershed, pathogen)
            if len(history) < 3:
                continue
            baseline = fit_ww_baseline(history)
            current_raw = current_ww_concentration(sewershed, pathogen, reference_date)
            if current_raw is None:
                continue
            current_log = math.log10(max(current_raw, 1.0))
            ww_score = (current_log - baseline["mean"]) / max(baseline["std"], 1e-6)
            if ww_score > WW_ANOMALY_THRESHOLD:
                results.append({
                    "detector":        "wastewater",
                    "sewershed":       sewershed,
                    "pathogen":        pathogen,
                    "anomaly_score":   float(ww_score),
                    "current_value":   float(current_raw),
                    "flagged":         True,
                    "reference_date":  reference_date.isoformat(),
                })
    return results

def run_absenteeism_detector(reference_date):
    """Detect school absenteeism spikes."""
    results = []
    for district, history in _DEMO_ABSENTEEISM.items():
        if len(history) < 5:
            continue
        baseline_values = np.array([h["rate"] for h in history[:-3]])
        baseline_mean = baseline_values.mean()
        baseline_std  = max(baseline_values.std(), 0.001)
        current_rate = history[-1]["rate"]
        score = (current_rate - baseline_mean) / baseline_std
        if score > ABSENTEEISM_THRESHOLD:
            results.append({
                "detector":        "school_absenteeism",
                "district":        district,
                "current_rate":    float(current_rate),
                "baseline_rate":   float(baseline_mean),
                "score":           float(score),
                "flagged":         True,
                "reference_date":  reference_date.isoformat(),
            })
    return results

def run_pharmacy_detector(reference_date):
    """Detect pharmacy-volume spikes (placeholder; demo returns empty).

    Production wires retail-pharmacy partnerships for OTC product sales
    and prescription-fill data. The teaching example skips this path.
    """
    return []

def candidate_geo_windows(clinical_results, auxiliary_results):
    """Build the set of geography-window candidates from detector outputs.

    Each detector flags either cells (clinical) or higher-level
    geographies (sewersheds for wastewater, school districts for
    absenteeism). The fusion layer aligns these to a common geography
    (typically the ZCTA or census-tract level) and combines signals
    that overlap.
    """
    candidates = {}

    # Clinical signals: extract per-syndrome geographies.
    for result in clinical_results:
        if not result.get("flagged"):
            continue
        cell = result.get("cell")
        if cell:
            key = (cell["geo_type"], cell["geo_id"], cell["syndrome"])
            candidates.setdefault(key, {
                "geo_type":     cell["geo_type"],
                "geo_id":       cell["geo_id"],
                "syndrome":     cell["syndrome"],
                "syndrome_class": classify_syndrome_class(cell["syndrome"]),
                "geo_class":    classify_geo_class(cell["geo_type"]),
            })

    # Wastewater signals: associate with the sewershed and the
    # respiratory-or-GI syndrome class implied by the pathogen.
    pathogen_to_syndrome = {
        "sars_cov_2":   "fever_respiratory",
        "influenza_a":  "ili",
        "influenza_b":  "ili",
        "rsv":          "fever_respiratory",
        "norovirus":    "gi",
    }
    for result in auxiliary_results:
        if result.get("detector") == "wastewater":
            syndrome = pathogen_to_syndrome.get(result.get("pathogen"), "fever_respiratory")
            key = ("sewershed", result["sewershed"], syndrome)
            candidates.setdefault(key, {
                "geo_type":     "sewershed",
                "geo_id":       result["sewershed"],
                "syndrome":     syndrome,
                "syndrome_class": classify_syndrome_class(syndrome),
                "geo_class":    "sewershed",
            })
        elif result.get("detector") == "school_absenteeism":
            key = ("school_district", result["district"], "ili")
            candidates.setdefault(key, {
                "geo_type":     "school_district",
                "geo_id":       result["district"],
                "syndrome":     "ili",
                "syndrome_class": "respiratory",
                "geo_class":    "school_district",
            })

    return list(candidates.values())

def classify_syndrome_class(syndrome):
    """Group syndromes into broader classes for fusion calibration."""
    if syndrome in {"fever_respiratory", "ili", "asthma_copd"}:
        return "respiratory"
    if syndrome == "gi":
        return "gi"
    if syndrome == "rash":
        return "rash"
    if syndrome == "neuro":
        return "neuro"
    if syndrome == "sepsis":
        return "sepsis"
    return "other"

def classify_geo_class(geo_type):
    """Group geography types into broader classes for fusion calibration."""
    if geo_type in {"census_tract", "zcta"}:
        return "neighborhood"
    if geo_type == "county":
        return "county"
    return geo_type

def max_score_for(results, candidate, detector=None):
    """Return the maximum score for the candidate from results."""
    max_score = 0.0
    for result in results:
        if detector and result.get("detector") != detector:
            continue
        if not result.get("flagged"):
            continue
        # Match by geography overlap.
        cell = result.get("cell")
        if cell and cell.get("geo_id") == candidate["geo_id"]:
            score = abs(result.get("exceedance",
                                    result.get("ewma_value",
                                              result.get("cusum_value", 0.0))))
            max_score = max(max_score, min(score / 5.0, 1.0))
        elif result.get("sewershed") == candidate["geo_id"]:
            max_score = max(max_score,
                            min(result.get("anomaly_score", 0.0) / 5.0, 1.0))
        elif result.get("district") == candidate["geo_id"]:
            max_score = max(max_score,
                            min(result.get("score", 0.0) / 5.0, 1.0))
    return max_score

def count_concordant_sources(*signals, threshold):
    """Count how many signal scores exceed the concordance threshold."""
    return sum(1 for s in signals if s and s > threshold)

def apply_fusion_calibration(score, cohort, calibration):
    """Apply a per-cohort calibration mapping to the fused score.

    Production maintains per-cohort calibration curves fit on labeled
    historical adjudications. The teaching example clips to [0, 1].
    """
    return min(max(score, 0.0), 1.0)

def run_auxiliary_and_fuse(reference_date, clinical_results):
    """Run auxiliary detectors and fuse with clinical signals."""
    auxiliary_results = []
    auxiliary_results.extend(run_wastewater_detector(reference_date))
    auxiliary_results.extend(run_absenteeism_detector(reference_date))
    auxiliary_results.extend(run_pharmacy_detector(reference_date))

    fused_signals = []
    for candidate in candidate_geo_windows(clinical_results, auxiliary_results):
        clinical_signal     = max_score_for(clinical_results, candidate)
        wastewater_signal   = max_score_for(auxiliary_results, candidate, "wastewater")
        pharmacy_signal     = max_score_for(auxiliary_results, candidate, "pharmacy_spike")
        absenteeism_signal  = max_score_for(auxiliary_results, candidate, "school_absenteeism")

        composite = (
            FUSION_WEIGHTS["clinical"]    * clinical_signal
          + FUSION_WEIGHTS["wastewater"]  * wastewater_signal
          + FUSION_WEIGHTS["pharmacy"]    * pharmacy_signal
          + FUSION_WEIGHTS["absenteeism"] * absenteeism_signal
        )
        concordance_bonus = (
            count_concordant_sources(
                clinical_signal, wastewater_signal,
                pharmacy_signal, absenteeism_signal,
                threshold=CONCORDANCE_SIGNAL_THRESHOLD,
            )
            * CONCORDANCE_BONUS_PER_SOURCE
        )
        composite_with_concordance = composite + concordance_bonus
        calibrated = apply_fusion_calibration(
            composite_with_concordance,
            cohort=(candidate["geo_class"], candidate["syndrome_class"]),
            calibration=None,
        )

        fused_signals.append({
            "candidate":             candidate,
            "clinical_signal":       float(clinical_signal),
            "wastewater_signal":     float(wastewater_signal),
            "pharmacy_signal":       float(pharmacy_signal),
            "absenteeism_signal":    float(absenteeism_signal),
            "composite_raw":         float(composite),
            "concordance_bonus":     float(concordance_bonus),
            "composite_calibrated":  float(calibrated),
            "concordant_source_count": count_concordant_sources(
                clinical_signal, wastewater_signal,
                pharmacy_signal, absenteeism_signal,
                threshold=CONCORDANCE_SIGNAL_THRESHOLD,
            ),
            "reference_date":        reference_date.isoformat(),
        })

    return fused_signals

def aggregate_adjacent_flags(signals, temporal_window_days):
    """Group geographically and temporally adjacent flagged signals.

    Production runs spatial proximity testing in PostGIS to identify
    cells whose geographies share boundaries or are within a distance
    threshold. The teaching example groups by syndrome and county; a
    real spatial proximity check is what scales.
    """
    groups = defaultdict(list)
    for signal in signals:
        candidate = signal["candidate"]
        # Group key: (county-equivalent, syndrome). For the demo, we
        # use the geography ID prefix as a proxy for county.
        county_key = (candidate["geo_id"][:5]
                      if candidate["geo_type"] == "census_tract"
                      else candidate["geo_id"])
        group_key = (county_key, candidate["syndrome"])
        groups[group_key].append(signal)

    proto_clusters = []
    for (county, syndrome), group_signals in groups.items():
        if not group_signals:
            continue
        # Proto-cluster aggregates the flagged geographies, the
        # window, and the strongest signal across sources.
        geographies = list({(s["candidate"]["geo_type"], s["candidate"]["geo_id"])
                            for s in group_signals})
        composite = max(s["composite_calibrated"] for s in group_signals)
        signal_summary = {
            "clinical_signal":         max(s["clinical_signal"] for s in group_signals),
            "wastewater_signal":       max(s["wastewater_signal"] for s in group_signals),
            "pharmacy_signal":         max(s["pharmacy_signal"] for s in group_signals),
            "absenteeism_signal":      max(s["absenteeism_signal"] for s in group_signals),
            "concordant_source_count": max(s["concordant_source_count"]
                                            for s in group_signals),
        }

        ref_date = datetime.fromisoformat(group_signals[0]["reference_date"]).date()
        proto_clusters.append({
            "geographies":           [{"type": gt, "id": gi} for gt, gi in geographies],
            "syndromes":             [syndrome],
            "window_start":          (ref_date - timedelta(days=temporal_window_days)).isoformat(),
            "window_end":            ref_date.isoformat(),
            "composite_calibrated":  composite,
            "signal_summary":        signal_summary,
        })
    return proto_clusters

def find_existing_open_cluster(proto):
    """Search for an open cluster covering the same geography and syndrome."""
    cluster_table = dynamodb.Table(CLUSTER_STATE_TABLE)
    geo_ids = sorted([g["id"] for g in proto["geographies"]])
    syndromes = sorted(proto["syndromes"])

    response = cluster_table.scan(
        FilterExpression=Attr("status").eq("open_for_review"),
    )
    for item in response.get("Items", []):
        existing = _undecimalize(item)
        existing_geo_ids = sorted([g["id"] for g in existing.get("geographies", [])])
        existing_syndromes = sorted(existing.get("syndromes", []))
        if (existing_geo_ids == geo_ids
                and existing_syndromes == syndromes):
            return existing
    return None

def update_existing_cluster(existing_cluster, proto):
    """Append the new signal to an existing open cluster."""
    cluster_table = dynamodb.Table(CLUSTER_STATE_TABLE)
    composite = max(existing_cluster.get("composite_score", 0.0),
                    proto["composite_calibrated"])
    cluster_table.update_item(
        Key={"cluster_id": existing_cluster["cluster_id"]},
        UpdateExpression="SET composite_score = :c, updated_at = :t",
        ExpressionAttributeValues={
            ":c": _to_decimal(composite),
            ":t": datetime.now(timezone.utc).isoformat(),
        },
    )

def check_recent_dismissal(proto, reason_class):
    """Was this same pattern recently dismissed for the relevant geographies?"""
    suppression_table = dynamodb.Table(SUPPRESSION_RULES_TABLE)
    geo_ids = {g["id"] for g in proto["geographies"]}
    syndromes = set(proto["syndromes"])
    now_iso = datetime.now(timezone.utc).isoformat()

    response = suppression_table.scan()
    for item in response.get("Items", []):
        rule = _undecimalize(item)
        if rule.get("valid_until", "") < now_iso:
            continue
        if rule.get("reason_class") != reason_class:
            continue
        rule_geo_ids = set(rule.get("geographies", []))
        rule_syndromes = set(rule.get("syndromes", []))
        if rule_geo_ids & geo_ids and rule_syndromes & syndromes:
            return True
    return False

def log_suppression(proto, reason):
    """Log that a proto-cluster was suppressed."""
    logger.info("cluster suppressed",
                extra={"reason": reason,
                       "geographies": [g["id"] for g in proto["geographies"]],
                       "syndromes": proto["syndromes"]})
    _emit_metric(f"Suppressed_{reason}", 1)

def line_list_build(geographies, syndromes, window_start, window_end):
    """Build the line list for a cluster.

    Production queries the case-detail store (a separate, access-
    controlled service) for encounters matching the cluster's
    geography x syndrome x time window. Investigators access the line
    list under specific authority. The teaching example returns a
    synthetic line list summary.
    """
    geo_ids = {g["id"] for g in geographies}
    line_list = []
    # In the demo, we return a small synthetic line list keyed off
    # the geographies and syndromes; production runs a real query.
    for i, geo_id in enumerate(geo_ids):
        for syndrome in syndromes:
            for j in range(3):
                line_list.append({
                    "case_index":   i * 10 + j,
                    "encounter_id": f"DEMO-{geo_id}-{syndrome}-{j}",
                    "syndrome":     syndrome,
                    "geo_id":       geo_id,
                    "age_group":    "5_17" if j < 2 else "18_49",
                    "sex":          "male" if j % 2 else "female",
                    "observed_at":  window_end,
                })
    return line_list

def line_list_summary(line_list):
    """Return a de-identified summary of the line list for case display."""
    if not line_list:
        return {}
    return {
        "case_count":     len(line_list),
        "syndromes":      sorted(list({c["syndrome"] for c in line_list})),
        "geographies":    sorted(list({c["geo_id"] for c in line_list})),
        "age_distribution": dict(Counter(c["age_group"] for c in line_list)),
    }

def persist_line_list(line_list):
    """Persist the line list (with case-detail authority required to read).

    Production writes to OpenSearch with field-level access controls
    so only authorized investigators can retrieve the full line list.
    The teaching example just returns a placeholder pointer.
    """
    line_list_id = f"LL-{uuid.uuid4().hex[:12]}"
    return line_list_id

def sum_of_expected_for(geographies, syndromes, window_start, window_end):
    """Sum the baseline expected counts across the cluster's cells and window."""
    expected = 0.0
    days = max((datetime.fromisoformat(window_end)
                - datetime.fromisoformat(window_start)).days, 1)
    for geo in geographies:
        for syndrome in syndromes:
            cell_key = {
                "geo_type":  geo["type"],
                "geo_id":    geo["id"],
                "strat":     "all_ages",
                "syndrome":  syndrome,
                "window":    "1d",
            }
            baseline = load_baseline_for_cell(cell_key,
                                               datetime.fromisoformat(window_end).date())
            expected += baseline["expected"] * days
    return expected

def summarize_demographics(line_list):
    """Compute a demographic distribution summary for the line list."""
    if not line_list:
        return {}
    return {
        "age_groups":  dict(Counter(c.get("age_group") for c in line_list)),
        "sex":         dict(Counter(c.get("sex") for c in line_list)),
    }

def build_geo_payload(line_list_coords, heatmap_resolution="h3_level_8"):
    """Build a geography visualization payload for the surveillance UI.

    Production returns H3-cell counts at the configured resolution; the
    teaching example returns a placeholder.
    """
    return {"resolution": heatmap_resolution, "cell_count": len(line_list_coords)}

def lookup_lab_results_for_cases(line_list, days):
    """Look up lab results associated with cases in the line list."""
    return {
        "respiratory_panels_run":      0,
        "respiratory_panels_positive": 0,
        "panels_pending":              0,
        "novel_pathogen_signals":      0,
    }

def lookup_genomic_clusters(lab_context):
    """Look up genomic-cluster context for the cases."""
    return {
        "sequencing_in_progress":  False,
        "preliminary_clusters":    [],
        "alerting_signal":         "no_genomic_data",
    }

def build_cluster_narrative_prompt(structured_evidence):
    """Build the Bedrock prompt for cluster-narrative generation.

    Constrained: cite the structured evidence, describe the cluster,
    never assert intent or recommend an outcome. The LLM produces
    investigator-readable narrative; investigator judgment makes the
    decisions.
    """
    geos = ", ".join(f"{g['type']}={g['id']}" for g in structured_evidence["geographies"])
    syndromes = ", ".join(structured_evidence["syndromes"])
    window = (f"{structured_evidence['window'][0]} through "
              f"{structured_evidence['window'][1]}")
    obs_exp = (
        f"observed={structured_evidence['observed']:.0f}, "
        f"expected={structured_evidence['expected']:.1f}, "
        f"relative_risk={structured_evidence['relative_risk']:.2f}"
    )
    demo = json.dumps(structured_evidence["demographic_breakdown"])
    concordance = structured_evidence["multi_source_concordance"]
    concordance_text = (
        f"clinical={concordance.get('clinical_signal', 0):.2f}, "
        f"wastewater={concordance.get('wastewater_signal', 0):.2f}, "
        f"absenteeism={concordance.get('absenteeism_signal', 0):.2f}, "
        f"concordant_sources={concordance.get('concordant_source_count', 0)}"
    )

    return (
        "You are summarizing a public health outbreak-detection cluster "
        "for a surveillance epidemiologist. You are not making an "
        "outbreak determination and you are not asserting a specific "
        "pathogen. You are translating the structured evidence into a "
        "3-5 sentence investigator-readable narrative that cites the "
        "evidence and describes the spatiotemporal pattern. End with "
        "the phrase 'This is decision support; investigator judgment "
        "governs the public health response.'\n\n"
        f"Geographies: {geos}\n"
        f"Syndromes: {syndromes}\n"
        f"Window: {window}\n"
        f"Counts: {obs_exp}\n"
        f"Demographics: {demo}\n"
        f"Multi-source concordance: {concordance_text}\n\n"
        "Produce 3-5 sentences of plain narrative. No bullet points. "
        "No assertions of specific pathogens beyond what the lab and "
        "genomic context indicates. No recommended public health "
        "actions. End with the required closing phrase."
    )

def parse_bedrock_response(response):
    """Parse a Bedrock InvokeModel response into the narrative text."""
    body = json.loads(response["body"].read())
    return body["content"][0]["text"].strip()

def invoke_bedrock_narrative(structured_evidence):
    """Generate the surveillance-team-facing cluster narrative via Bedrock."""
    prompt = build_cluster_narrative_prompt(structured_evidence)
    try:
        response = bedrock_runtime.invoke_model(
            modelId=BEDROCK_MODEL_ID,
            body=json.dumps({
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens":        800,
                "temperature":       0.0,
                "messages":          [{"role": "user", "content": prompt}],
            }),
        )
        return parse_bedrock_response(response)
    except Exception as e:
        logger.warning("bedrock invocation failed", extra={"error": str(e)})
        _emit_metric("BedrockNarrativeFailed", 1)
        # Fall back to a structured-only summary so cluster-builder
        # doesn't fail when Bedrock is unavailable.
        return (
            f"Cluster of {structured_evidence['observed']:.0f} cases of "
            f"{', '.join(structured_evidence['syndromes'])} across "
            f"{len(structured_evidence['geographies'])} geographies "
            f"between {structured_evidence['window'][0]} and "
            f"{structured_evidence['window'][1]}. "
            f"Observed exceeds expected by a factor of "
            f"{structured_evidence['relative_risk']:.2f}. "
            "This is decision support; investigator judgment governs "
            "the public health response."
        )

def _emit_metric(metric_name, value, unit="Count"):
    """Emit a CloudWatch metric for operational monitoring."""
    try:
        cloudwatch.put_metric_data(
            Namespace="Surveillance/ClusterBuilder",
            MetricData=[{
                "MetricName": metric_name,
                "Value":      float(value),
                "Unit":       unit,
                "Timestamp":  datetime.now(timezone.utc),
            }],
        )
    except Exception as e:
        logger.debug("metric emit failed",
                     extra={"metric": metric_name, "error": str(e)})

def generate_cluster_id():
    """Generate a cluster identifier."""
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    return f"CL-{today}-{uuid.uuid4().hex[:8]}"

def tier_from_composite(composite_score, cohort="DEFAULT"):
    """Map a composite score to a surveillance-team tier."""
    thresholds = TIER_THRESHOLDS.get(cohort, TIER_THRESHOLDS["DEFAULT"])
    if composite_score >= thresholds["tier_1"]:
        return "tier_1"
    if composite_score >= thresholds["tier_2"]:
        return "tier_2"
    if composite_score >= thresholds["tier_3"]:
        return "tier_3"
    return "below_threshold"

def build_clusters(fused_signals, reference_date):
    """Build cluster candidates, attach evidence, and persist to cluster-state."""
    proto_clusters = aggregate_adjacent_flags(
        fused_signals, temporal_window_days=CLUSTER_TEMPORAL_WINDOW
    )

    cluster_candidates = []
    for proto in proto_clusters:
        existing = find_existing_open_cluster(proto)
        if existing:
            update_existing_cluster(existing, proto)
            continue

        if check_recent_dismissal(proto, reason_class="false_alarm"):
            log_suppression(proto, reason="matches_recent_dismissal")
            continue

        line_list = line_list_build(
            proto["geographies"], proto["syndromes"],
            proto["window_start"], proto["window_end"],
        )

        observed = len(line_list)
        expected = sum_of_expected_for(
            proto["geographies"], proto["syndromes"],
            proto["window_start"], proto["window_end"],
        )
        relative_risk = observed / max(expected, 1.0)
        excess = max(0, observed - expected)

        demographic_breakdown = summarize_demographics(line_list)
        geo_visualization = build_geo_payload(
            [], heatmap_resolution="h3_level_8",
        )
        lab_context = lookup_lab_results_for_cases(line_list, days=14)
        genomic_context = lookup_genomic_clusters(lab_context)

        structured_evidence = {
            "geographies":              proto["geographies"],
            "syndromes":                proto["syndromes"],
            "window":                    (proto["window_start"], proto["window_end"]),
            "observed":                 observed,
            "expected":                 expected,
            "relative_risk":            relative_risk,
            "demographic_breakdown":    demographic_breakdown,
            "lab_context":              lab_context,
            "genomic_context":          genomic_context,
            "multi_source_concordance":  proto["signal_summary"],
        }
        narrative = invoke_bedrock_narrative(structured_evidence)
        composite_tier = tier_from_composite(proto["composite_calibrated"])

        if composite_tier == "below_threshold":
            continue

        cluster = {
            "cluster_id":             generate_cluster_id(),
            "opened_at":              datetime.now(timezone.utc).isoformat(),
            "reference_date":         reference_date.isoformat(),
            "geographies":            proto["geographies"],
            "syndromes":              proto["syndromes"],
            "window_start":           proto["window_start"],
            "window_end":             proto["window_end"],
            "observed":               observed,
            "expected":               expected,
            "relative_risk":          relative_risk,
            "excess":                 excess,
            "composite_score":        float(proto["composite_calibrated"]),
            "multi_source_concordance": proto["signal_summary"],
            "line_list_summary":      line_list_summary(line_list),
            "line_list_pointer":      persist_line_list(line_list),
            "demographic_breakdown":  demographic_breakdown,
            "geo_visualization":      geo_visualization,
            "lab_context":            lab_context,
            "genomic_context":        genomic_context,
            "narrative":              narrative,
            "tier":                   composite_tier,
            "status":                  "open_for_review",
            "assigned_to":             None,
            "outcome":                  None,
        }

        cluster_table = dynamodb.Table(CLUSTER_STATE_TABLE)
        cluster_table.put_item(Item=_decimalize(cluster))

        try:
            eventbridge.put_events(Entries=[{
                "Source":      "surveillance.cluster-builder",
                "DetailType":  "ClusterOpened",
                "Detail":      json.dumps({
                    "cluster_id":      cluster["cluster_id"],
                    "tier":            cluster["tier"],
                    "composite_score": cluster["composite_score"],
                    "geographies":     cluster["geographies"],
                    "syndromes":       cluster["syndromes"],
                }, default=str),
                "EventBusName": CLUSTER_BUS,
            }])
        except Exception as e:
            logger.debug("cluster event publish failed",
                         extra={"error": str(e)})

        _emit_metric(f"ClustersOpened_{cluster['tier']}", 1)
        cluster_candidates.append(cluster)

    return cluster_candidates