Derive ACA takeup from marketplace coverage

src/microplex_us/pipelines/us.py

@@ -6915,7 +6915,7 @@ def _infer_role_flag_tax_unit_filing_status(
     def _aggregate_policyengine_tax_unit_input_columns(
         self,
         unit_persons: pd.DataFrame,
-    ) -> dict[str, float]:
+    ) -> dict[str, Any]:
         columns = (
             "domestic_production_ald",
             "health_savings_account_ald",
@@ -6925,7 +6925,7 @@ def _aggregate_policyengine_tax_unit_input_columns(
             "unrecaptured_section_1250_gain",
             "unreported_payroll_tax",
         )
-        aggregated: dict[str, float] = {}
+        aggregated: dict[str, Any] = {}
         for column in columns:
             if column not in unit_persons.columns:
                 continue
@@ -6935,8 +6935,46 @@ def _aggregate_policyengine_tax_unit_input_columns(
                 aggregated[column] = float(nonzero_values.iloc[0])
                 continue
             aggregated[column] = float(values.sum())
+        aca_takeup = self._infer_policyengine_aca_takeup_for_tax_unit(unit_persons)
+        if aca_takeup is not None:
+            aggregated["takes_up_aca_if_eligible"] = aca_takeup
         return aggregated
 
+    def _infer_policyengine_aca_takeup_for_tax_unit(
+        self,
+        unit_persons: pd.DataFrame,
+    ) -> bool | None:
+        if "takes_up_aca_if_eligible" in unit_persons.columns:
+            return bool(
+                pd.to_numeric(
+                    unit_persons["takes_up_aca_if_eligible"],
+                    errors="coerce",
+                )
+                .fillna(0.0)
+                .ne(0.0)
+                .any()
+            )
+        marketplace_columns = (
+            "has_marketplace_health_coverage",
+            "has_marketplace_health_coverage_at_interview",
+            "reported_has_marketplace_health_coverage_at_interview",
+            "reported_has_subsidized_marketplace_health_coverage_at_interview",
+            "reported_has_unsubsidized_marketplace_health_coverage_at_interview",
+        )
+        observed = [
+            column for column in marketplace_columns if column in unit_persons.columns
+        ]
+        if not observed:
+            return None
+        marketplace = pd.Series(False, index=unit_persons.index, dtype=bool)
+        for column in observed:
+            marketplace |= (
+                pd.to_numeric(unit_persons[column], errors="coerce")
+                .fillna(0.0)
+                .ne(0.0)
+            )
+        return bool(marketplace.any())
+
     def _split_preserved_tax_unit_members(
         self,
         unit_persons: pd.DataFrame,

tests/pipelines/test_us.py

@@ -1005,10 +1005,40 @@ def test_build_policyengine_entity_tables_adds_deterministic_aca_takeup(self):
                 "relationship_to_head": [0, 0, 0],
                 "state_fips": [6, 12, 48],
                 "tenure": [1, 1, 1],
+                "has_marketplace_health_coverage": [True, False, True],
             }
         )
 
-        pipeline.build_policyengine_entity_tables(population)
+        tables = pipeline.build_policyengine_entity_tables(population)
+
+        tax_units = tables.tax_units.sort_values("household_id").reset_index(drop=True)
+        assert tax_units["takes_up_aca_if_eligible"].tolist() == [True, False, True]
+
+    def test_build_policyengine_entity_tables_preserves_explicit_aca_takeup(self):
+        pipeline = USMicroplexPipeline(
+            USMicroplexBuildConfig(policyengine_dataset_year=2024)
+        )
+        population = pd.DataFrame(
+            {
+                "person_id": [1, 2],
+                "household_id": [10, 20],
+                "weight": [1.0, 1.0],
+                "age": [34, 42],
+                "sex": [2, 1],
+                "income": [40_000.0, 65_000.0],
+                "filing_status": ["SINGLE", "SINGLE"],
+                "relationship_to_head": [0, 0],
+                "state_fips": [6, 12],
+                "tenure": [1, 1],
+                "has_marketplace_health_coverage": [False, True],
+                "takes_up_aca_if_eligible": [True, False],
+            }
+        )
+
+        tables = pipeline.build_policyengine_entity_tables(population)
+
+        tax_units = tables.tax_units.sort_values("household_id").reset_index(drop=True)
+        assert tax_units["takes_up_aca_if_eligible"].tolist() == [True, False]
 
     def test_build_policyengine_entity_tables_fallback_employment_excludes_transfer_income(
         self,