Add ACA marketplace bronze-selection target ETL (#618)

* Rebase ACA marketplace ETL onto main

* Format CPS marketplace benchmark helper

* Fix bronze-stratum domain_variable ordering and drop dead ETL

Review fixes from the standing review of PR 618:

1. P0 bug: bronze stratum constraints were inserted in the order
   ``state_fips, used_aca_ptc, selected_marketplace_plan_benchmark_ratio``,
   which SQLite's ``GROUP_CONCAT(DISTINCT ...)`` preserves insertion
   order for. That produced ``domain_variable = "used_aca_ptc,...``, but
   ``target_config.yaml:68`` expects the alphabetical form
   ``selected_marketplace_plan_benchmark_ratio,used_aca_ptc``. The rule
   didn't match, so the bronze target silently dropped out of the loss.
   Reorder the inserts and add a comment explaining why order matters.

2. Delete the now-dead ``etl_aca_agi_state_targets.py`` — it still used
   ``source="CMS Marketplace"`` (rejected by ``create_field_valid_values``)
   and the Makefile no longer invokes it. Redirect
   ``tests/integration/test_database_build.py`` to the new
   ``etl_aca_marketplace.py``.

3. Add a ValueError guard for corrupt source data (bronze APTC consumers
   exceeding total APTC consumers for any state).

4. Add the CMS Marketplace PUF URL to the ETL extract docstring so the
   input CSV is actually refetchable.

5. Expand the unit test file: add a real-CSV regression test (expects
   27+ HC.gov states with bronze ≤ total and no SBM states leaking in)
   and a negative test for the new ValueError.

* Make domain_variable ordering deterministic and fix stale integration test

Codex review on 8fd8990 found two issues:

1. ``tests/integration/test_database_build.py::test_state_aca_and_agi_targets_loaded``
   still asserted legacy ``aca_ptc`` / ``person_count`` / ``adjusted_gross_income``
   state targets that the deleted ``etl_aca_agi_state_targets.py`` used to
   load, so it would fail against the rebuilt DB. Rename and rewrite it as
   ``test_state_marketplace_targets_loaded`` that asserts the new APTC and
   bronze-selection targets land with the canonical alphabetical
   ``domain_variable`` strings.

2. The previous constraint-insertion-order workaround relied on SQLite's
   ``GROUP_CONCAT(DISTINCT ...)`` preserving insertion order, which is
   undocumented. Add ``ORDER BY`` to the ``domain_variable`` aggregation in
   the ``stratum_domain`` view so the canonical form is enforced at the view
   level, regardless of how callers insert constraints. Drop the now-obsolete
   ordering comment in ``etl_aca_marketplace.py``.

* Use correlated subquery for domain_variable ordering (SQLite portability)

The prior ``GROUP_CONCAT(DISTINCT ... ORDER BY ...)`` form requires
SQLite >= 3.44 and failed on the Modal integration runner with
``sqlite3.OperationalError: near "ORDER": syntax error``.

Replace with a correlated subquery that selects distinct constraint
names ordered alphabetically and then concatenates them without an
inner ORDER BY. Works on all supported SQLite versions and still
produces the canonical form (e.g.
``selected_marketplace_plan_benchmark_ratio,used_aca_ptc``) regardless
of constraint insertion order.

Verified by running the real view against in-memory SQLite with
non-alphabetical insert order; result matches the expected canonical
string.

* Restore etl_aca_agi_state_targets.py alongside new marketplace ETL

The deletion in 8fd8990 was too aggressive. That file loaded three
distinct target families into the calibration DB:

1. state-level ``aca_ptc`` spending targets (sourced from
   ``aca_spending_and_enrollment_2024.csv``)
2. state-level ``person_count`` enrollment targets (same source)
3. state-level AGI bracket targets (sourced from ``agi_state.csv``)

This PR adds *new* marketplace APTC-count and bronze-count targets but
does not replace the ACA spending/enrollment or AGI targets. Without
them the calibrator has nothing to pin state-level ACA PTC spending,
and ``test_aca_calibration`` / ``test_sparse_aca_calibration`` fail
with >500% state deviations.

Restore the file verbatim from the pre-deletion state, keep the
``CMS Marketplace`` source string it uses (re-added to the
``create_field_valid_values`` allowlist alongside the newer
``CMS 2024 OEP state metal status PUF`` source the marketplace ETL
uses), re-add the Makefile invocation, and put its entry back in the
integration-build script list ahead of the marketplace ETL. Keep the
new ``test_state_marketplace_targets_loaded`` as a peer to the
restored ``test_state_aca_and_agi_targets_loaded``.

The long-term cleanup (migrating the spending/enrollment targets into
the marketplace ETL or deprecating them) is a follow-up.

---------

Co-authored-by: Max Ghenis <mghenis@gmail.com>

Author: daphnehanse11

.gitignore

@@ -27,6 +27,7 @@ node_modules
 !age_state.csv
 !agi_state.csv
 !soi_targets.csv
+!policyengine_us_data/storage/calibration_targets/aca_marketplace_state_metal_selection_2024.csv
 !policyengine_us_data/storage/social_security_aux.csv
 !policyengine_us_data/storage/SSPopJul_TR2024.csv
 !policyengine_us_data/storage/national_and_district_rents_2023.csv

Makefile

@@ -88,6 +88,7 @@ database:
 	python policyengine_us_data/db/etl_state_income_tax.py --year $(YEAR)
 	python policyengine_us_data/db/etl_irs_soi.py --year $(YEAR)
 	python policyengine_us_data/db/etl_aca_agi_state_targets.py --year $(YEAR)
+	python policyengine_us_data/db/etl_aca_marketplace.py --year $(YEAR)
 	python policyengine_us_data/db/etl_pregnancy.py --year $(YEAR)
 	python policyengine_us_data/db/validate_database.py
 

changelog.d/618.added.md

@@ -0,0 +1,2 @@
+Add an ACA marketplace ETL that loads state-level HC.gov bronze-plan
+selection targets for APTC recipients into the calibration database.

policyengine_us_data/calibration/target_config.yaml

@@ -59,6 +59,14 @@ include:
     geo_level: state
     domain_variable: tanf
 
+  # === STATE — ACA marketplace APTC and bronze-plan enrollment counts ===
+  - variable: tax_unit_count
+    geo_level: state
+    domain_variable: used_aca_ptc
+  - variable: tax_unit_count
+    geo_level: state
+    domain_variable: selected_marketplace_plan_benchmark_ratio,used_aca_ptc
+
   # === STATE — fine AGI bracket targets (stubs 9/10 from in55cmcsv) ===
   - variable: person_count
     geo_level: state

policyengine_us_data/db/create_database_tables.py

@@ -341,12 +341,24 @@ def validate_parent_child_constraints(mapper, connection, target: Stratum):
             THEN sc.value END),
         'US'
     ) AS geographic_id,
-    GROUP_CONCAT(DISTINCT CASE
-        WHEN sc.constraint_variable NOT IN (
-            'state_fips', 'congressional_district_geoid',
-            'tax_unit_is_filer', 'ucgid_str'
-        ) THEN sc.constraint_variable
-    END) AS domain_variable
+    -- Compute domain_variable via a correlated subquery so we can sort
+    -- the distinct constraint names alphabetically before concatenation.
+    -- We can't use `GROUP_CONCAT(DISTINCT ... ORDER BY ...)` because the
+    -- `ORDER BY` form inside aggregates requires SQLite >= 3.44, and the
+    -- Modal runner ships an older libsqlite.
+    (
+        SELECT GROUP_CONCAT(cv, ',')
+        FROM (
+            SELECT DISTINCT sc2.constraint_variable AS cv
+            FROM stratum_constraints sc2
+            WHERE sc2.stratum_id = t.stratum_id
+              AND sc2.constraint_variable NOT IN (
+                  'state_fips', 'congressional_district_geoid',
+                  'tax_unit_is_filer', 'ucgid_str'
+              )
+            ORDER BY sc2.constraint_variable
+        )
+    ) AS domain_variable
 FROM targets t
 LEFT JOIN stratum_constraints sc ON t.stratum_id = sc.stratum_id
 GROUP BY t.target_id, t.stratum_id, t.variable,

policyengine_us_data/db/create_field_valid_values.py

@@ -70,6 +70,7 @@ def populate_field_valid_values(session: Session) -> None:
         ("source", "Census ACS S0101", "survey"),
         ("source", "IRS SOI", "administrative"),
         ("source", "CMS Marketplace", "administrative"),
+        ("source", "CMS 2024 OEP state metal status PUF", "administrative"),
         ("source", "CMS Medicaid", "administrative"),
         ("source", "Census ACS S2704", "survey"),
         ("source", "USDA FNS SNAP", "administrative"),

policyengine_us_data/db/etl_aca_marketplace.py

@@ -0,0 +1,236 @@
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+import pandas as pd
+from sqlmodel import Session, create_engine
+
+from policyengine_us_data.calibration.calibration_utils import STATE_CODES
+from policyengine_us_data.db.create_database_tables import (
+    Stratum,
+    StratumConstraint,
+    Target,
+)
+from policyengine_us_data.storage import CALIBRATION_FOLDER, STORAGE_FOLDER
+from policyengine_us_data.utils.db import etl_argparser, get_geographic_strata
+
+logger = logging.getLogger(__name__)
+
+# `selected_marketplace_plan_benchmark_ratio == 1.0` represents benchmark
+# silver coverage, so bronze plan selections are the subset below this ratio.
+BENCHMARK_SILVER_RATIO = 1.0
+
+STATE_METAL_SELECTION_PATH = (
+    CALIBRATION_FOLDER / "aca_marketplace_state_metal_selection_2024.csv"
+)
+
+STATE_ABBR_TO_FIPS = {abbr: fips for fips, abbr in STATE_CODES.items()}
+
+
+def _extra_args(parser) -> None:
+    parser.add_argument(
+        "--state-metal-csv",
+        type=Path,
+        default=STATE_METAL_SELECTION_PATH,
+        help=("State-metal CMS OEP proxy CSV. Default: %(default)s"),
+    )
+
+
+def extract_aca_marketplace_state_metal_data(
+    state_metal_csv_path: Path,
+) -> pd.DataFrame:
+    """Extract CMS marketplace state metal-status inputs from the checked-in CSV.
+
+    This ETL keeps an explicit extract step even though the source file already
+    lives in the repository. The original CMS 2024 OEP state metal status PUF
+    is not currently pulled from a stable direct-download endpoint in CI, so we
+    store the normalized input CSV at
+    `policyengine_us_data/storage/calibration_targets/aca_marketplace_state_metal_selection_2024.csv`.
+
+    Source (CMS Marketplace Open Enrollment Period Public Use Files):
+    https://www.cms.gov/marketplace/resources/data/public-use-files
+
+    To reproduce or update that file:
+    1. Download the CMS 2024 OEP State, Metal Level, and Enrollment Status PUF
+       from the URL above.
+    2. Preserve one row per state/platform/metal/enrollment-status combination.
+    3. Keep the `state_code`, `platform`, `metal_level`,
+       `enrollment_status`, `consumers`, and `aptc_consumers` columns.
+    4. Save the normalized output back to `state_metal_csv_path`.
+    """
+    return pd.read_csv(state_metal_csv_path)
+
+
+def build_state_marketplace_bronze_aptc_targets(
+    state_metal_df: pd.DataFrame,
+) -> pd.DataFrame:
+    """
+    Build HC.gov state bronze-selection targets among APTC consumers.
+
+    The 2024 CMS state-metal-status PUF exposes:
+    - metal rows (`B`, `G`, `S`) with enrollment_status=`All`
+    - aggregate rows (`All`) broken out by enrollment status (`01-atv`, etc.)
+
+    We use:
+    - total APTC consumers = sum of `aptc_consumers` for `metal_level == All`
+      across enrollment statuses
+    - bronze APTC consumers = `aptc_consumers` on the bronze row
+    """
+    df = state_metal_df.copy()
+    df = df[df["platform"] == "HC.gov"].copy()
+
+    total_rows = df[
+        (df["metal_level"] == "All") & (df["aptc_consumers"].notna())
+    ].copy()
+    bronze_rows = df[
+        (df["metal_level"] == "B")
+        & (df["enrollment_status"] == "All")
+        & (df["aptc_consumers"].notna())
+    ].copy()
+
+    total_aptc = total_rows.groupby("state_code", as_index=False).agg(
+        marketplace_aptc_consumers=("aptc_consumers", "sum"),
+        marketplace_consumers=("consumers", "sum"),
+    )
+    bronze_aptc = bronze_rows[["state_code", "aptc_consumers", "consumers"]].rename(
+        columns={
+            "aptc_consumers": "bronze_aptc_consumers",
+            "consumers": "bronze_consumers",
+        }
+    )
+
+    result = total_aptc.merge(bronze_aptc, on="state_code", how="inner")
+    result["state_fips"] = result["state_code"].map(STATE_ABBR_TO_FIPS)
+    result = result[result["state_fips"].notna()].copy()
+    result["state_fips"] = result["state_fips"].astype(int)
+    invalid_bronze = (
+        result["bronze_aptc_consumers"] > result["marketplace_aptc_consumers"]
+    )
+    if invalid_bronze.any():
+        bad_states = result.loc[invalid_bronze, "state_code"].tolist()
+        raise ValueError(
+            "Bronze APTC consumers exceed total APTC consumers for states: "
+            f"{bad_states}. Source CSV likely corrupted."
+        )
+    result["bronze_aptc_share"] = (
+        result["bronze_aptc_consumers"] / result["marketplace_aptc_consumers"]
+    )
+    result.insert(0, "year", 2024)
+    result.insert(1, "source", "cms_2024_oep_state_metal_status_puf")
+    return result.sort_values("state_code").reset_index(drop=True)
+
+
+def load_state_marketplace_bronze_aptc_targets(
+    targets_df: pd.DataFrame,
+    year: int,
+) -> None:
+    db_url = f"sqlite:///{STORAGE_FOLDER / 'calibration' / 'policy_data.db'}"
+    engine = create_engine(db_url)
+
+    with Session(engine) as session:
+        geo_strata = get_geographic_strata(session)
+
+        for row in targets_df.itertuples(index=False):
+            state_fips = int(row.state_fips)
+            parent_id = geo_strata["state"].get(state_fips)
+            if parent_id is None:
+                logger.warning(
+                    "No state geographic stratum for FIPS %s, skipping", state_fips
+                )
+                continue
+
+            # We intentionally do not subset to `tax_unit_is_filer == 1`.
+            # These CMS targets describe marketplace coverage groups rather
+            # than the IRS filer universe, so the closest calibration entity is
+            # a tax unit with positive modeled APTC use.
+            aptc_stratum = Stratum(
+                parent_stratum_id=parent_id,
+                notes=f"State FIPS {state_fips} Marketplace APTC recipients",
+            )
+            aptc_stratum.constraints_rel = [
+                StratumConstraint(
+                    constraint_variable="state_fips",
+                    operation="==",
+                    value=str(state_fips),
+                ),
+                StratumConstraint(
+                    constraint_variable="used_aca_ptc",
+                    operation=">",
+                    value="0",
+                ),
+            ]
+            aptc_stratum.targets_rel.append(
+                Target(
+                    # We use `tax_unit_count` rather than household/person
+                    # counts because insurance groups map most closely to
+                    # PolicyEngine tax units in the current calibration schema.
+                    variable="tax_unit_count",
+                    period=year,
+                    value=float(row.marketplace_aptc_consumers),
+                    active=True,
+                    source="CMS 2024 OEP state metal status PUF",
+                    notes="HC.gov APTC consumers across all enrollment statuses",
+                )
+            )
+            session.add(aptc_stratum)
+            session.flush()
+
+            bronze_stratum = Stratum(
+                parent_stratum_id=aptc_stratum.stratum_id,
+                notes=f"State FIPS {state_fips} Marketplace bronze APTC recipients",
+            )
+            bronze_stratum.constraints_rel = [
+                StratumConstraint(
+                    constraint_variable="state_fips",
+                    operation="==",
+                    value=str(state_fips),
+                ),
+                StratumConstraint(
+                    constraint_variable="selected_marketplace_plan_benchmark_ratio",
+                    operation="<",
+                    value=str(BENCHMARK_SILVER_RATIO),
+                ),
+                StratumConstraint(
+                    constraint_variable="used_aca_ptc",
+                    operation=">",
+                    value="0",
+                ),
+            ]
+            bronze_stratum.targets_rel.append(
+                Target(
+                    variable="tax_unit_count",
+                    period=year,
+                    value=float(row.bronze_aptc_consumers),
+                    active=True,
+                    source="CMS 2024 OEP state metal status PUF",
+                    notes="HC.gov bronze plan selections among APTC consumers",
+                )
+            )
+            session.add(bronze_stratum)
+            session.flush()
+
+        session.commit()
+
+
+def main() -> None:
+    args, year = etl_argparser(
+        "ETL for ACA marketplace bronze-selection calibration targets",
+        extra_args_fn=_extra_args,
+    )
+
+    state_metal = extract_aca_marketplace_state_metal_data(args.state_metal_csv)
+    targets_df = build_state_marketplace_bronze_aptc_targets(state_metal)
+    if targets_df.empty:
+        raise RuntimeError("No HC.gov marketplace bronze/APTC targets were generated.")
+
+    print(
+        "Loading ACA marketplace bronze/APTC state targets for "
+        f"{len(targets_df)} states from {args.state_metal_csv}"
+    )
+    load_state_marketplace_bronze_aptc_targets(targets_df, year)
+    print("ACA marketplace bronze/APTC targets loaded.")
+
+
+if __name__ == "__main__":
+    main()