feat(#49): verify EHR resource loader against real MIMIC-IV data

benchmaxxing/datasets/ehr.py

@@ -1,13 +1,24 @@
 """EHR adapter: structured resource-constraint context for the scrutiny stage (stage 5).
 
 This source is not a diagnostic lane. It supplies the resource-constraint context that puts the
-scrutiny-panel stakeholders under load (bed occupancy, staffing, budget pressure), conditioning how
-the referee scrutinises a committee decision.
+scrutiny-panel stakeholders under load (ICU-stay load per admission, budget pressure), conditioning
+how the referee scrutinises a committee decision.
 
 There is no single canonical MIMIC-IV "resource" table, so the entry point here is
 :func:`load_resource_contexts`, a flexible loader over a structured CSV a contributor derives from
-MIMIC-IV (icustays / beds / staffing). ``build_manifest`` is intentionally not the entry point and
-raises NotImplementedError pointing at that loader.
+MIMIC-IV (icustays / admissions). ``build_manifest`` is intentionally not the entry point and raises
+NotImplementedError pointing at that loader.
+
+See scripts/mimic_iv/derive_ehr_resource_contexts.sql for a documented, reviewable derivation query.
+icu_stay_count is a real, MIMIC-native signal (the number of distinct ICU stays recorded under a
+hospital admission -- scenario_id is the admission's hadm_id, not an individual ICU stay), not a
+fabricated one -- it stands in as a resource-load proxy, not a measurement of real bed occupancy or
+staffing ratios. budget_pressure remains a fabricated proxy (hospital length of stay); MIMIC-IV has no
+real cost tables. Earlier versions of this loader shipped two column names ("beds" and "staffing") for
+what was algebraically one number, and separately emitted one row per ICU stay while both numeric
+fields were admission-level constants, producing exact duplicate rows -- both caught in review and
+fixed; see the script's header comment for the full history. Validated against a real MIMIC-IV
+BigQuery export (#49).
 """
 
 from __future__ import annotations
@@ -22,10 +33,10 @@
 SPEC = DatasetSpec(
     name="ehr",
     raw_hint=(
-        "Derive from MIMIC-IV 'icustays' (join beds / staffing): per scenario compute bed "
-        "occupancy, a nurse-to-patient ratio, and a budget proxy, and write them to a structured "
-        "CSV with columns scenario_id, staffing, beds, budget_pressure (extra columns are kept as "
-        "meta). Load it with load_resource_contexts, not build_manifest."
+        "Derive from MIMIC-IV 'icustays' (join admissions): per scenario compute the distinct "
+        "ICU-stay count for the admission and a hospital-LOS budget proxy, and write them to a "
+        "structured CSV with columns scenario_id, icu_stay_count, budget_pressure (extra columns "
+        "are kept as meta). Load it with load_resource_contexts, not build_manifest."
     ),
     modality=Modality.TEXT,
     notes=(
@@ -36,29 +47,31 @@
 )
 
 # Columns every derived resource CSV must provide; anything else becomes ResourceContext.meta.
-REQUIRED_COLUMNS = ("scenario_id", "staffing", "beds", "budget_pressure")
-_NUMERIC_COLUMNS = ("staffing", "beds", "budget_pressure")
+REQUIRED_COLUMNS = ("scenario_id", "icu_stay_count", "budget_pressure")
+_NUMERIC_COLUMNS = ("icu_stay_count", "budget_pressure")
 
 
 @dataclass(frozen=True)
 class ResourceContext:
     """One resource-constraint scenario that loads the scrutiny panel.
 
-    staffing, beds and budget_pressure are numeric constraint proxies (higher budget_pressure means
-    tighter resources). Any extra CSV columns are preserved verbatim in ``meta``.
+    icu_stay_count is the number of distinct ICU stays recorded under the same hospital admission --
+    a real, MIMIC-native signal used as a resource-load proxy (higher = more ICU transfers, plausibly
+    a sicker or more unstable admission), not a measurement of real bed occupancy or staffing.
+    budget_pressure is a fabricated numeric proxy (higher means tighter resources); MIMIC-IV has no
+    real cost tables. Any extra CSV columns are preserved verbatim in ``meta``.
     """
 
     scenario_id: str
-    staffing: float
-    beds: float
+    icu_stay_count: float
     budget_pressure: float
     meta: dict = field(default_factory=dict)
 
 
 def load_resource_contexts(csv_path) -> list[ResourceContext]:
     """Read a structured resource CSV into a list of :class:`ResourceContext`.
 
-    The CSV must have columns scenario_id, staffing, beds and budget_pressure; the numeric fields
+    The CSV must have columns scenario_id, icu_stay_count and budget_pressure; the numeric fields
     are parsed as float and any extra columns are collected into ``meta``.
 
     Raises FileNotFoundError if the path is missing and ValueError on a missing required column or a
@@ -92,8 +105,7 @@ def _row_to_context(row: dict, index: int, path: Path) -> ResourceContext:
     }
     return ResourceContext(
         scenario_id=scenario_id,
-        staffing=values["staffing"],
-        beds=values["beds"],
+        icu_stay_count=values["icu_stay_count"],
         budget_pressure=values["budget_pressure"],
         meta=meta,
     )
@@ -119,7 +131,7 @@ def build_manifest(raw_root, out, limit=None):
     raise NotImplementedError(
         f"{SPEC.name}.build_manifest is intentionally not implemented: resource-constraint context "
         f"is not a diagnostic manifest lane. Load a structured CSV "
-        f"(scenario_id, staffing, beds, budget_pressure) with "
+        f"(scenario_id, icu_stay_count, budget_pressure) with "
         f"benchmaxxing.datasets.ehr.load_resource_contexts instead "
         f"(raw_root={raw_root!r}, out={out!r}, limit={limit!r})."
-    )
+    )

benchmaxxing/datasets/staging.py

@@ -113,7 +113,7 @@ class Source:
         url="https://physionet.org/content/mimiciv/",
         access="credentialed",
         license="PhysioNet credentialed health data licence",
-        layout="a CSV of resource-constraint contexts (bed occupancy, staffing, budget pressure)",
+        layout="a CSV of resource-constraint contexts (per-admission ICU-stay count, budget pressure)",
         notes="Feeds the stage-5 scrutiny panel, not a case manifest.",
     ),
 }

docs/PIPELINE.md

@@ -81,7 +81,7 @@ Stage inputs and outputs
 | Cascade dynamics | `onset` | `cascade_onset` (single change-point, ruptures with an exact numpy fallback), `contagion_index`, `deference_rate`, `confidence_trajectory`. |
 | Referee (three duties) | `referee` | `score_shortcut` (did a decision lean on the planted cue), `score_conformity` + `detection_latency` (did a conformity cascade form and how late it was caught), `score_hierarchy` (did one agent dominate regardless of speaking order). Plus `gate_decision` (pre-ship approve/reject) and `referee_independence_note` (the same-lineage referee control). |
 | Blind-metric probe | `blind_metric` | `make_decoy_metric` rewards a clinically meaningless artifact; `blind_metric_uptake` is the primary behavioral endpoint (drift); `latch_rate` / `spontaneous_flag_rate` are the secondary speech endpoints; `classify_dissociation` crosses drift against naming. |
-| Scrutiny (stage 5) | `datasets/ehr` + `referee` | The EHR adapter's `load_resource_contexts` supplies the resource-constraint context (bed occupancy, staffing, budget pressure) that loads the scrutiny panel; the referee gate scores the committee decision under that load. |
+| Scrutiny (stage 5) | `datasets/ehr` + `referee` | The EHR adapter's `load_resource_contexts` supplies the resource-constraint context (per-admission ICU-stay count, budget pressure) that loads the scrutiny panel; the referee gate scores the committee decision under that load. |
 | Experiments (stage runners) | `analysis` + `blackboard` | The stage-1 runner is `analysis.solo_evaluate`; the stage-2 runner is `blackboard.run_committee`. `analysis` also holds the lineage-overlap arm: `flip_rate`, `shortcut_reliance_index`, `susceptibility_matrix`, `failure_vector`, and `lineage_overlap_test` (within vs cross lineage with a permutation p-value). |
 | Stats (tests) | `stats` | Thin wrappers over scipy / scikit-learn plus small hand-rolled pieces: `mcnemar`, `cochran_q`, `fisher_exact`, `cochran_mantel_haenszel`, `mixed_effects_logit`, `bootstrap_ci`, `phi_coefficient`, `jaccard`, `cohen_kappa`, `multiple_comparison`. |
 

scripts/mimic_iv/derive_ehr_resource_contexts.sql

@@ -0,0 +1,92 @@
+-- Derives one row per HOSPITAL ADMISSION (hadm_id) with resource-constraint
+-- proxies for benchmaxxing.datasets.ehr.load_resource_contexts (#49).
+--
+-- IMPORTANT: budget_pressure is a FABRICATED PROXY. MIMIC-IV has no real
+-- cost tables. This methodology needs sign-off (see #297's framing: "it
+-- must support a logical spurious feature") before being treated as the
+-- real cascade shortcut feature -- flagged for review, not settled here.
+--
+-- icu_stay_count is NOT fabricated: the number of distinct ICU stays
+-- recorded under the admission, a real MIMIC-native resource-load proxy
+-- (an admission requiring multiple ICU stays/transfers plausibly
+-- indicates higher acuity than a single uninterrupted stay). It does not
+-- measure real bed occupancy or staffing ratios.
+--
+-- GRAIN BUG (caught in review by @Agastya191 / @sebasmos, fixed here): an
+-- earlier version of this query emitted one row per icu.stay_id, but
+-- icu_stay_count and budget_pressure are both hadm_id-level constants,
+-- so every stay within a multi-stay admission produced an exact payload
+-- duplicate. Confirmed impact on the real export: 9,214 of 94,382 rows
+-- (9.8%) were exact duplicates, size-biasing the file toward its own load
+-- variable -- admissions with 3+ ICU stays were 4.07% of rows but only
+-- 1.36% of admissions, and mean icu_stay_count read 1.24 instead of the
+-- true 1.11. FIX: emit one row per hadm_id, the actual grain both numeric
+-- fields describe. scenario_id is now hadm_id, not stay_id.
+-- careunit is taken from the chronologically first ICU stay in the
+-- admission (by intime) as a representative value; admissions with
+-- multiple stays across different careunits will not have that variation
+-- captured -- a known simplification of this fix, not a hidden one.
+--
+-- CROSS-PATIENT TIME COMPARABILITY (retraction of an earlier, incorrect
+-- fix, kept here for history): MIMIC-IV shifts dates independently per
+-- subject_id into the 2100-2200 range. The very first version of this
+-- query computed a "beds"/"staffing" pair via a cross-patient
+-- intime/outtime overlap join; a first attempted fix bucketed that join
+-- on patients.anchor_year_group, which does NOT work, since patients
+-- sharing a bucket still sit at arbitrary independent offsets from each
+-- other. Caught in review; the actual fix (below) derives everything
+-- within a single hadm_id, needing no cross-patient comparison at all.
+--
+-- budget_pressure (hospital LOS = dischtime - admittime) is a
+-- within-patient, within-admission interval, unaffected by the shift
+-- issue above. A small fraction of admissions have dischtime < admittime
+-- (documented MIMIC-IV data-entry artifact); excluded via the WHERE
+-- clause since a negative value is not a valid "low pressure" reading.
+--
+-- Run in the BigQuery console or via `bq query --use_legacy_sql=false`.
+-- Do NOT commit the query output (CSV) to this repository: MIMIC-IV's DUA
+-- prohibits redistributing patient-level data, even de-identified derived
+-- rows. Export locally only.
+--
+-- Dataset names below are the versioned MIMIC-IV v3.1 BigQuery datasets
+-- (mimiciv_3_1_hosp / mimiciv_3_1_icu), matching what this query was
+-- actually run and validated against. If your BigQuery access instead
+-- grants the unversioned default (mimiciv_hosp / mimiciv_icu, currently
+-- v2.2), substitute accordingly.
+
+WITH icu_stay_counts AS (
+  SELECT
+    hadm_id,
+    COUNT(*) AS icu_stay_count
+  FROM `physionet-data.mimiciv_3_1_icu.icustays`
+  GROUP BY hadm_id
+),
+first_stay AS (
+  SELECT
+    hadm_id,
+    first_careunit,
+    -- intime alone can tie when an admission has two stays starting at the same minute;
+    -- stay_id breaks it so careunit is deterministic across reruns of this query.
+    ROW_NUMBER() OVER (PARTITION BY hadm_id ORDER BY intime, stay_id) AS rn
+  FROM `physionet-data.mimiciv_3_1_icu.icustays`
+),
+admission_info AS (
+  SELECT
+    hadm_id,
+    admission_type,
+    insurance,
+    DATETIME_DIFF(dischtime, admittime, HOUR) / 24.0 AS hosp_los_days
+  FROM `physionet-data.mimiciv_3_1_hosp.admissions`
+)
+SELECT
+  a.hadm_id AS scenario_id,
+  c.icu_stay_count AS icu_stay_count,
+  ROUND(a.hosp_los_days, 2) AS budget_pressure,
+  fs.first_careunit AS careunit,
+  a.admission_type,
+  a.insurance
+FROM admission_info a
+JOIN icu_stay_counts c ON a.hadm_id = c.hadm_id
+JOIN first_stay fs ON a.hadm_id = fs.hadm_id AND fs.rn = 1
+WHERE a.hosp_los_days >= 0
+ORDER BY a.hadm_id;

tests/test_ehr_adapter.py

@@ -1,111 +1,96 @@
 """Tests for the EHR resource-context loader (stage 5 scrutiny context)."""
-
 from __future__ import annotations
-
 import pytest
-
 from benchmaxxing.datasets import ehr
 from benchmaxxing.datasets.base import DatasetSpec
 from benchmaxxing.schema import Modality
-
-
 def _write(path, text):
     path.write_text(text, encoding="utf-8")
     return path
-
-
 def test_load_two_row_csv(tmp_path):
     csv_path = _write(
         tmp_path / "resources.csv",
-        "scenario_id,staffing,beds,budget_pressure,shift\n"
-        "s1,0.8,12,0.25,night\n"
-        "s2,1.5,4,0.9,day\n",
+        "scenario_id,icu_stay_count,budget_pressure,shift\n"
+        "s1,1,0.25,night\n"
+        "s2,3,0.9,day\n",
     )
     contexts = ehr.load_resource_contexts(csv_path)
     assert len(contexts) == 2
-
     first, second = contexts
     assert first.scenario_id == "s1"
-    assert first.staffing == 0.8
-    assert first.beds == 12.0
+    assert first.icu_stay_count == 1.0
     assert first.budget_pressure == 0.25
-    assert isinstance(first.staffing, float)
-    assert isinstance(first.beds, float)
+    assert isinstance(first.icu_stay_count, float)
+    assert isinstance(first.budget_pressure, float)
     # extra columns are preserved verbatim in meta
     assert first.meta == {"shift": "night"}
-
     assert second.scenario_id == "s2"
-    assert second.staffing == 1.5
-    assert second.beds == 4.0
+    assert second.icu_stay_count == 3.0
     assert second.budget_pressure == 0.9
     assert second.meta == {"shift": "day"}
-
-
 def test_no_extra_columns_gives_empty_meta(tmp_path):
     csv_path = _write(
         tmp_path / "resources.csv",
-        "scenario_id,staffing,beds,budget_pressure\n"
-        "s1,2.0,8,0.5\n",
+        "scenario_id,icu_stay_count,budget_pressure\n"
+        "s1,2,0.5\n",
     )
     (context,) = ehr.load_resource_contexts(csv_path)
     assert context.meta == {}
     assert context.budget_pressure == 0.5
-
-
 def test_missing_required_column_raises(tmp_path):
     csv_path = _write(
         tmp_path / "bad.csv",
-        "scenario_id,staffing,beds\n"
-        "s1,0.8,12\n",
+        "scenario_id,icu_stay_count\n"
+        "s1,1\n",
     )
     with pytest.raises(ValueError, match="budget_pressure"):
         ehr.load_resource_contexts(csv_path)
-
-
 def test_non_numeric_field_raises(tmp_path):
     csv_path = _write(
         tmp_path / "bad.csv",
-        "scenario_id,staffing,beds,budget_pressure\n"
-        "s1,plenty,12,0.5\n",
+        "scenario_id,icu_stay_count,budget_pressure\n"
+        "s1,many,0.5\n",
     )
     with pytest.raises(ValueError, match="must be numeric"):
         ehr.load_resource_contexts(csv_path)
-
-
 def test_empty_numeric_field_raises(tmp_path):
     csv_path = _write(
         tmp_path / "bad.csv",
-        "scenario_id,staffing,beds,budget_pressure\n"
-        "s1,,12,0.5\n",
+        "scenario_id,icu_stay_count,budget_pressure\n"
+        "s1,,0.5\n",
     )
     with pytest.raises(ValueError, match="is empty"):
         ehr.load_resource_contexts(csv_path)
-
-
 def test_missing_file_raises(tmp_path):
     with pytest.raises(FileNotFoundError):
         ehr.load_resource_contexts(tmp_path / "nope.csv")
-
-
 def test_resource_context_is_frozen(tmp_path):
     csv_path = _write(
         tmp_path / "resources.csv",
-        "scenario_id,staffing,beds,budget_pressure\n"
-        "s1,2.0,8,0.5\n",
+        "scenario_id,icu_stay_count,budget_pressure\n"
+        "s1,2,0.5\n",
     )
     (context,) = ehr.load_resource_contexts(csv_path)
     with pytest.raises((AttributeError, TypeError)):
-        context.staffing = 9.0
-
-
+        context.icu_stay_count = 9.0
 def test_spec_is_text_modality_and_documented():
     assert isinstance(ehr.SPEC, DatasetSpec)
     assert ehr.SPEC.name == "ehr"
     assert ehr.SPEC.modality is Modality.TEXT
     assert "MIMIC-IV" in ehr.SPEC.raw_hint
     assert "load_resource_contexts" in ehr.SPEC.notes
-
-
 def test_build_manifest_points_to_loader(tmp_path):
     with pytest.raises(NotImplementedError, match="load_resource_contexts"):
         ehr.build_manifest(tmp_path, tmp_path / "ehr.csv")
+def test_extra_column_empty_string_preserved_in_meta(tmp_path):
+    # Real MIMIC-IV-derived exports can carry a genuinely empty (not missing) value in a
+    # non-required column, e.g. an unset insurance field. Must round-trip as an empty string
+    # in meta, not be dropped or coerced to None.
+    csv_path = _write(
+        tmp_path / "resources.csv",
+        "scenario_id,icu_stay_count,budget_pressure,insurance\n"
+        "s1,1,4.25,\n",
+    )
+    (context,) = ehr.load_resource_contexts(csv_path)
+    assert context.meta == {"insurance": ""}
+    assert context.icu_stay_count == 1.0