diff --git a/changelog.d/1155.fixed.md b/changelog.d/1155.fixed.md new file mode 100644 index 0000000000..5e890d0aba --- /dev/null +++ b/changelog.d/1155.fixed.md @@ -0,0 +1 @@ +Generate complete WIC participation decisions for ACS local-area records after demographic completion and before donor assembly, preserving existing donor decisions and rejecting missing participation at release boundaries. diff --git a/docs/agent-guide.md b/docs/agent-guide.md index ca981de83d..a927db55a9 100644 --- a/docs/agent-guide.md +++ b/docs/agent-guide.md @@ -251,6 +251,14 @@ register entry with its reason, bound to its producer in name. A new US lane that writes a release H5 must run the check where it records `build.built_with_model_package`. +The supported ACS local-area path additionally requires complete boolean +`takes_up_wic_if_eligible`: ACS decisions are generated after demographic +completion and before donor assembly; donor decisions are not regenerated. +Calculation/export cannot default-fill missing WIC decisions, and packaging +checks actual person values as well as stored names. The code repair does not +replace the published dataset; see [the WIC replacement instructions](us-acs-local-wic-replacement.md) +and #1154 for the separate rebuild, publication, certification, and `.py` removal. + A US release or release-gate preflight that receives a multispine pool through `--base-h5` must authenticate its sibling terminal manifest. A current stacked pool whose terminal battery is red remains fail-closed unless the operator diff --git a/docs/evidence/spec-engine/us-f0-coverage.json b/docs/evidence/spec-engine/us-f0-coverage.json index 0fabd95a86..565f342ff0 100644 --- a/docs/evidence/spec-engine/us-f0-coverage.json +++ b/docs/evidence/spec-engine/us-f0-coverage.json @@ -1656,13 +1656,13 @@ "compiler_ir.node_slices" ], "expected": { - "map_sha256": "b4d6ec9c11ce323d83319a3d0289e0cf7216a7936d6c6f08cef0a463ad369641", - "protocol_sha256": "114217a2918329b0601d94deb8ed83f73919584b85ca8aabe45a40665afb6560" + "map_sha256": "02ec8173113ac6d7256181439831e1c51ad8fca5bbcda657f1630bfc6b5a1c28", + "protocol_sha256": "e3035d45cd8e5ce29fc3291894a28596897dfd58955fcf0139eaafb7409a0bc6" }, "failures": [], "observed": { - "map_sha256": "b4d6ec9c11ce323d83319a3d0289e0cf7216a7936d6c6f08cef0a463ad369641", - "protocol_sha256": "114217a2918329b0601d94deb8ed83f73919584b85ca8aabe45a40665afb6560" + "map_sha256": "02ec8173113ac6d7256181439831e1c51ad8fca5bbcda657f1630bfc6b5a1c28", + "protocol_sha256": "e3035d45cd8e5ce29fc3291894a28596897dfd58955fcf0139eaafb7409a0bc6" }, "status": "covered" }, @@ -1677,7 +1677,7 @@ "compiler_ir.seed_stream_map" ], "expected": { - "implementation_sha256": "114217a2918329b0601d94deb8ed83f73919584b85ca8aabe45a40665afb6560", + "implementation_sha256": "e3035d45cd8e5ce29fc3291894a28596897dfd58955fcf0139eaafb7409a0bc6", "protocol": "legacy-v1", "streams": [ "build_model", @@ -1698,7 +1698,7 @@ }, "failures": [], "observed": { - "implementation_sha256": "114217a2918329b0601d94deb8ed83f73919584b85ca8aabe45a40665afb6560", + "implementation_sha256": "e3035d45cd8e5ce29fc3291894a28596897dfd58955fcf0139eaafb7409a0bc6", "protocol": "legacy-v1", "streams": [ "build_model", @@ -2599,7 +2599,7 @@ "country": "us", "schema_id": "country_spec", "schema_version": 1, - "spec_sha256": "f76f921d4f74392b106b8c0729ce250494e0a55455eaba7b910b19cdf7825178" + "spec_sha256": "5048cabc59e5d7d9789e751cc9a5c5b4d97ab6fb3f15429cabce3b568615073b" } }, "report_schema_version": 3, @@ -2609,7 +2609,7 @@ "country": "us", "schema_id": "country_spec", "schema_version": 1, - "spec_sha256": "f76f921d4f74392b106b8c0729ce250494e0a55455eaba7b910b19cdf7825178" + "spec_sha256": "5048cabc59e5d7d9789e751cc9a5c5b4d97ab6fb3f15429cabce3b568615073b" }, "status": "pass" } diff --git a/docs/us-acs-local-wic-replacement.md b/docs/us-acs-local-wic-replacement.md new file mode 100644 index 0000000000..2738dcf183 --- /dev/null +++ b/docs/us-acs-local-wic-replacement.md @@ -0,0 +1,108 @@ +# Replace incomplete ACS local-area WIC participation + +## Scope and ownership + +[Issue #1154](https://github.com/PolicyEngine/microcosm/issues/1154) remains open +until a dataset owner rebuilds and publishes the replacement, `.py` certifies +it, and the temporary assumption is removed. [Issue #1155](https://github.com/PolicyEngine/microcosm/issues/1155) +covers this code repair only. Merging the code does not change existing H5 +files or certify a population. + +The temporary `.py` change is [PR #562](https://github.com/PolicyEngine/policyengine.py/pull/562). +Its separate removal draft, [PR #563](https://github.com/PolicyEngine/policyengine.py/pull/563), +must remain blocked until the actual replacement is published and certified. +It contains a deliberately failing check against the known incomplete pin. + +This work does not publish a dataset, modify the national default/latest +pointer, deploy services, or add environment variables or database changes. + +## Generation order + +The supported ACS local-area build calls `build_optional_acs_multispine`: + +1. Require complete boolean `takes_up_wic_if_eligible` on the donor population. + This is the participation decision conditional on eligibility, not the + separate `receives_wic` reported-receipt input. Do not recreate donor draws. +2. Load ACS source records and map their native age, sex, income, and family + membership. Keep `source_year`, `source_household_id`, and `source_person_id`. +3. Complete the demographic inputs through the existing transfer plan, + including pregnancy and own children. Missing required inputs remain errors. +4. On ACS records only, run `with_acs_wic_claim_input`, which requires complete + source identity and delegates to `with_us_wic_claim_input`. Use the existing + build seed and ACS source year. The existing generator owns category order, + sourced FNS CY2022 rates, and deterministic source-person hash draws. It + assigns pregnant, postpartum, infant, child, or no-category probabilities; + it does not invent a breastfeeding observation or change eligibility rules. +5. Combine ACS records with donors, preserving donor decisions exactly. Require + complete boolean participation over the combined population. Record the + generator seed, source year, output column, row count, and rate-source URL + in the existing orchestration provenance. + +The country engine still determines income and demographic eligibility and +benefit amounts. The generator does not assume every eligible ACS person +claims WIC. Reordering records cannot change a person's seeded decision. + +## Release validation + +The local release tool rejects missing or malformed participation before +sampling/materialization, before consumer export, and before packaging. +`fill_reviewed_nulls` checks WIC before loading the reviewed-null register: +even an explicit register entry cannot fill participation from the engine's +`True` default. Other reviewed-null handling remains unchanged. + +Packaging retains the existing stored-input-name checks, including rejection +of obsolete `would_claim_wic`. It additionally checks the actual participation +column in every person row and records the checked row count in the existing +stored-input report, bound to the packaged H5 hash. H5 reads request batches +of 65,536 rows. Table-format files project only the participation column; +fixed-format files require all columns and pandas may unpickle an entire +object block despite the requested row bounds. This is not a claim of a strict +memory bound for historical fixed-format files. + +## Dataset-owner steps (not performed by this PR) + +1. Choose a qualified donor release with complete current-name participation + and every other required donor input. Preserve its immutable identity. + Existing donor qualification, receipt, hours, geography, and model checks + still apply; an incompatible donor must fail, not receive an implicit fill. +2. Use a new output directory and rebuild ACS staging with the supported + `tools/build_us_acs_multispine_base.py` entry point. It is deprecated for + new general multispine builds but still supports this local-area release + path. Use pinned ACS archives, the PUMA mapping, donor inputs, and explicit + build configuration. A small smoke build may validate the setup but cannot + be published as the full replacement. +3. Check `orchestration.provenance.wic_claim` in the new staging summary and + complete boolean participation in the H5. Compare donor decisions to the + selected parent; inspect ACS rates by demographic category. Never edit the + existing published artifact in place. +4. Rebuild materialization, calibration, consumer export, QA, finalization, and + packaging using `tools/build_us_acs_local_release.py --stage all` and the + recorded full-scale configuration. Do not reuse old target matrices, + calibrated weights, consumer evidence, prepared years, or QA reports: + changing participation can change WIC and dependent outputs. Satisfy every + existing release requirement, not just the new participation check. +5. Review and publish the new immutable **non-default local-area** release with + the existing `microcosm-publish-release --no-latest` route. + Do not change the national latest pointer or default dataset. +6. Supply the actual revision, artifact hash, and regional manifest to `.py` + PR #563. Use its existing certification process and model-compatibility + requirements; do not invent replacement metadata or bypass validation. +7. After the compatibility parent merges, rebase and retarget the removal PR + to `main`, certify the replacement, run its checks, and only then merge it. + Keep ordinary legacy-name mapping for other datasets. Close #1154 only + after publication, certification, and removal are all complete. + +## Focused tests + +Existing CI discovers the new tests through the test-directory registry: + +- `engine_free/us/test_us_acs_wic.py`: deterministic generator delegation, + reordered identities, required source inputs, strict participation validation. +- `engine_free/us/test_us_acs_multispine.py`: stage order, donor preservation, + early donor validation, and unchanged disabled behavior. +- `engine_free/us/test_us_acs_local_release_tool.py`: no WIC default fills, + export refusal, packaging refusal, and hash-bound row-count evidence. +- `engine_workflow/us/test_us_acs_wic.py`: H5 round-trip, live country-engine + consumption, fixed/table validation, and reads over multiple batches. + +These small synthetic tests do not rebuild, publish, or certify the real dataset. diff --git a/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py b/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py index 673ab08d35..1f00e8958c 100644 --- a/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py +++ b/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py @@ -359,8 +359,8 @@ "late_schedule": "e59c019d3d454eac99ac0ac209b6c5b6faaf9bdfcaeee18c36a25be19bf7da2f", "ownership": "5f64f0aac49e2313177564f71876bffc8c81b3ded4df701e70930e60e9c98356", "primary_tuples": "fdf23da429f8c501e198a2f5b719763d523a41565c2153fdf17b0344f21bcd7c", - "seed_map": "b4d6ec9c11ce323d83319a3d0289e0cf7216a7936d6c6f08cef0a463ad369641", - "seed_protocol": "114217a2918329b0601d94deb8ed83f73919584b85ca8aabe45a40665afb6560", + "seed_map": "02ec8173113ac6d7256181439831e1c51ad8fca5bbcda657f1630bfc6b5a1c28", + "seed_protocol": "e3035d45cd8e5ce29fc3291894a28596897dfd58955fcf0139eaafb7409a0bc6", "source_manifest": "f4f8bfeb79268a4043a843b47cec1d1c0e58e9f5dd6bc23417ada95446dd7b8e", "take_up": "9522ce40f7dea569312dd7a7beb474e5a5afadd2f1e10cb5876534c3ef623d35", "tail": "ac92829c88a1a4fb6460d61190918d5d99c6c377fc8dd8f62f02b332d09bf59c", diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/acs_multispine.py b/packages/microcosm-build/src/microcosm/build/us_runtime/acs_multispine.py index c6fe766d3f..a942f99cf9 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/acs_multispine.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/acs_multispine.py @@ -52,6 +52,12 @@ us_puma_ladder_assignment_summary, ) from microcosm.build.us_runtime.support_provenance import BASE_ASEC_SUPPORT_CHANNEL +from microcosm.build.us_runtime.wic_claim import ( + US_WIC_CLAIM_OUTPUT_COLUMNS, + WIC_CLAIM_FNS_SOURCE_URL, + require_complete_us_wic_claim_input, + with_acs_wic_claim_input, +) from microcosm.frame import Frame __all__ = ["AcsMultispineResult", "build_optional_acs_multispine"] @@ -115,6 +121,9 @@ def build_optional_acs_multispine( provenance={"enabled": False}, ) + # Donors must already carry their qualified decisions. Never regenerate + # them with ACS categories or silently use the country model's True default. + _require_complete_pre_acs_donor_wic_input(base) raw_acs, loader_metadata = build_acs_pums_unit_frame( source, chunksize=chunksize, @@ -206,6 +215,19 @@ def build_optional_acs_multispine( transferred_frame = transferred.frame del transferred + # Pregnancy/parent inputs are now complete. Generate decisions on ACS only, + # while its stable Census identities are intact, before donor assembly. + transferred_frame = with_acs_wic_claim_input( + transferred_frame, seed=seed, time_period=source.vintage + ) + wic_provenance: dict[str, object] = { + "seed": seed, + "source_year": source.vintage, + "person_rows": len(transferred_frame.table("person")), + "output_columns": list(US_WIC_CLAIM_OUTPUT_COLUMNS), + "category_rates_source": WIC_CLAIM_FNS_SOURCE_URL, + } + pool_options: dict[str, Any] = {"acs_share": acs_share} if puma_ladder is not None: pool_options.update( @@ -219,12 +241,14 @@ def build_optional_acs_multispine( ) pooled = with_optional_acs_spine(base, transferred_frame, **pool_options) del transferred_frame + require_complete_us_wic_claim_input(pooled.table("person")) provenance = { "enabled": True, "acs_share": _json_ready(acs_share), "loader": loader_provenance, "native_inputs": native_provenance, + "wic_claim": wic_provenance, "imputed_inputs": imputed_provenance, "deferred_inputs": deferred_provenance, "adult_care_recipient_gate": _json_ready(adult_care_gate) @@ -283,6 +307,11 @@ def build_optional_acs_multispine( ) +def _require_complete_pre_acs_donor_wic_input(base: Frame) -> None: + """Validate donor participation before loading or appending ACS records.""" + require_complete_us_wic_claim_input(base.table("person")) + + def _require_recipient_adult_care_structure( frame: Frame, ) -> dict[str, Any] | None: diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/wic_claim.py b/packages/microcosm-build/src/microcosm/build/us_runtime/wic_claim.py index 33858a4458..834fd7c51e 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/wic_claim.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/wic_claim.py @@ -29,6 +29,7 @@ import hashlib from collections.abc import Mapping from importlib.resources import files +from pathlib import Path import numpy as np import pandas as pd @@ -63,6 +64,9 @@ "WIC_CLAIM_ARCHIVED_RANDOMNESS_URL", "WIC_CLAIM_FNS_SOURCE_URL", "derive_us_wic_claim_from_manifest", + "require_complete_us_wic_claim_input", + "require_complete_us_wic_claim_h5", + "with_acs_wic_claim_input", "us_wic_claim_signal_gate", "us_wic_claim_stage_spec", "us_wic_claim_summary", @@ -88,9 +92,7 @@ ) US_WIC_CLAIM_STAGE_NAME = "wic_claim_input" -US_WIC_CLAIM_OUTPUT_COLUMNS: tuple[str, ...] = ( - "takes_up_wic_if_eligible", -) +US_WIC_CLAIM_OUTPUT_COLUMNS: tuple[str, ...] = ("takes_up_wic_if_eligible",) US_WIC_CLAIM_NONCONSTANT_PERSON_COLUMNS = US_WIC_CLAIM_OUTPUT_COLUMNS # These are persisted PolicyEngine-facing columns already carried or derived @@ -527,6 +529,63 @@ def with_us_wic_claim_input( ) +def require_complete_us_wic_claim_input(person: pd.DataFrame) -> None: + """Reject absent, nullable, or non-boolean participation, without filling.""" + if _OUTPUT not in person: + raise ValueError(f"WIC participation requires person column {_OUTPUT!r}.") + if person[_OUTPUT].isna().any(): + raise ValueError(f"WIC participation {_OUTPUT!r} contains missing values.") + if not pd.api.types.is_bool_dtype(person[_OUTPUT].dtype): + raise ValueError(f"WIC participation {_OUTPUT!r} must have boolean dtype.") + + +def with_acs_wic_claim_input(frame: Frame, *, seed: int, time_period: int) -> Frame: + """Generate ACS decisions after demographic completion, before assembly. + + Require the actual source identity instead of accepting the generic + generator's person-id fallback. No donor records may be passed here. + Categories, rates, and hash draws remain owned by the existing generator. + """ + person = frame.table("person") + missing = [column for column in _SOURCE_IDENTITY_COLUMNS if column not in person] + if missing: + raise ValueError( + f"ACS WIC generation requires source identity columns: {missing}." + ) + identity = person.loc[:, list(_SOURCE_IDENTITY_COLUMNS)] + if identity.isna().any(axis=None): + raise ValueError("ACS WIC generation requires complete source identity.") + result = with_us_wic_claim_input(frame, seed=seed, time_period=time_period) + require_complete_us_wic_claim_input(result.table("person")) + return result + + +def require_complete_us_wic_claim_h5(path: Path) -> int: + """Check persisted participation in bounded row batches; return row count. + + Table-format H5 reads only the participation column. Fixed-format H5 + requires all columns for each row batch; pandas cannot project columns + from that format. Numeric/boolean blocks are sliced, but pandas may + unpickle an entire fixed-format object block despite the row bounds. + """ + with pd.HDFStore(path, mode="r") as store: + if "/person" not in store.keys(): + raise ValueError("WIC participation requires a person H5 table.") + storer = store.get_storer("person") + rows = int(storer.nrows if storer.is_table else storer.shape[0]) + if rows == 0: + raise ValueError("WIC participation cannot certify an empty person table.") + for start in range(0, rows, 65_536): + if storer.is_table: + person = store.select( + "person", start=start, stop=start + 65_536, columns=[_OUTPUT] + ) + else: + person = store.select("person", start=start, stop=start + 65_536) + require_complete_us_wic_claim_input(person) + return rows + + def _person_weights(frame: Frame) -> np.ndarray: weights = np.asarray(frame.resolve_weights("person").values, dtype=np.float64) valid = np.isfinite(weights) & (weights >= 0.0) diff --git a/packages/microcosm-build/tests/engine_free/shared/test_spec_engine_loader.py b/packages/microcosm-build/tests/engine_free/shared/test_spec_engine_loader.py index 17cd0b715a..e6954bd5d9 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_spec_engine_loader.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_spec_engine_loader.py @@ -12,7 +12,7 @@ def test_semantic_hash_has_golden_vector_and_surface_separation(tmp_path) -> Non # Pin the domain separator, normalization rules, schema-set receipt, and # exact normative projection as one reviewable golden vector. assert first.spec_sha256 == ( - "69de1ffc232c3a4da09925658c9d105af55bdaae90852824ac6283306932aab9" + "e66fe425edb9b814a9133801fadf2ebf5dade584bd250e2124f6a51a57704bf8" ) second_root = _rich_minimal(tmp_path / "xy", note="second", store="local:b") diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_hours.py b/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_hours.py index ded9e53d4b..dd5815db3f 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_hours.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_hours.py @@ -43,8 +43,16 @@ def _frame( person["hours_worked_last_week"] = person["A_HRS1"] person["age"] = np.resize([40, 30, 35, 32], n) person["is_female"] = np.resize([False, True], n) + person["is_pregnant"] = False + person["own_children_in_household"] = 0 + person["source_year"] = 2024 + person["source_household_id"] = ids + person["source_person_id"] = ids tables["household"]["state_fips"] = 6 if role is not None: + # Already-built donors have participation decisions. Raw ACS fixtures + # deliberately omit them so the real post-transfer generator is used. + person["takes_up_wic_if_eligible"] = np.resize([False, True], n) for entity, table in tables.items(): table[support_channel_column(entity)] = role table[support_source_id_column(entity)] = ids @@ -417,6 +425,15 @@ def fit(self, frame, predictors, targets, *, weights): acs = result.frame.person.loc[ result.frame.person["person_spine"].eq("acs_2024_1yr") ] + assert acs["takes_up_wic_if_eligible"].dtype == bool + assert not acs["takes_up_wic_if_eligible"].isna().any() + donors = result.frame.person.loc[ + ~result.frame.person["person_spine"].eq("acs_2024_1yr") + ].set_index("person_id") + pd.testing.assert_series_equal( + donors["takes_up_wic_if_eligible"].sort_index(), + original.person.set_index("person_id")["takes_up_wic_if_eligible"].sort_index(), + ) assert acs["weekly_hours_worked_before_lsr"].tolist() == ( [40, 0, 19.5 if unresolved else 38, 0] * 2 ) diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_release_tool.py b/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_release_tool.py index e4c841adb2..dfc06f3bd3 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_release_tool.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_release_tool.py @@ -14,6 +14,7 @@ import pandas as pd import pytest +from microcosm.build.us_runtime import wic_claim from microcosm.data import stored_inputs from test_support.paths import paths_for @@ -33,6 +34,7 @@ #: The stored-input contract's real entry points, kept before the autouse #: fixture below replaces them, for the tests that pin the contract itself. _REAL_REQUIRE_H5_STORED_INPUTS = stored_inputs.require_h5_stored_inputs +_REAL_REQUIRE_WIC_H5 = wic_claim.require_complete_us_wic_claim_h5 #: A policyengine-us stand-in for the package stage's stored-input contract. _PACKAGE_ENGINE = stored_inputs.CertifiedEngine( label="policyengine-us 2.2.1", @@ -67,6 +69,9 @@ def _stored_input_contract_passes(monkeypatch): against :data:`_PACKAGE_ENGINE`.""" monkeypatch.setattr(stored_inputs, "installed_us_engine", lambda: _PACKAGE_ENGINE) + # Placeholder-byte tests do not exercise H5 participation. Dedicated + # packaging tests below restore the real validator against actual H5. + monkeypatch.setattr(wic_claim, "require_complete_us_wic_claim_h5", lambda path: 0) monkeypatch.setattr( stored_inputs, "require_h5_stored_inputs", @@ -124,13 +129,17 @@ def test_household_chunks_refuse_population_aggregates(monkeypatch, tmp_path) -> monkeypatch.setattr(module, "project_input_only", lambda frame, **kw: (frame, {})) monkeypatch.setattr(module, "fill_reviewed_nulls", lambda *args, **kw: None) specs = (fixtures._variable("aggregate_total", base_variable="aggregate_probe"),) + from test_support.microcosm_build.us_wic_claim import _replace_person + + frame = fixtures._nested_frame() + frame = _replace_person(frame, frame.person.assign(takes_up_wic_if_eligible=False)) with pytest.raises( ValueError, match=rf"household batch 1/1 computed .*{aggregate}@2024", ): module.materialize_chunked( - fixtures._nested_frame(), + frame, specs, hh_chunk=2, batch=2, @@ -984,6 +993,7 @@ def _staging_frame_with_hours(weekly: list[float], last_week: list[float]): "person_marital_unit_id": ids, "weekly_hours_worked_before_lsr": np.asarray(weekly, dtype=float), "hours_worked_last_week": np.asarray(last_week, dtype=float), + "takes_up_wic_if_eligible": np.zeros(n, dtype=bool), } ) tables = { @@ -1436,6 +1446,9 @@ def test_package_records_the_stored_input_gate_bound_to_the_packaged_bytes( monkeypatch.setattr( stored_inputs, "require_h5_stored_inputs", _REAL_REQUIRE_H5_STORED_INPUTS ) + monkeypatch.setattr( + wic_claim, "require_complete_us_wic_claim_h5", _REAL_REQUIRE_WIC_H5 + ) module = _load_tool_module() args = _package_args_with_hours(module, tmp_path, monkeypatch, gate_state="passed") result = module.do_package(args) @@ -1451,6 +1464,7 @@ def test_package_records_the_stored_input_gate_bound_to_the_packaged_bytes( "registered_non_variables": [], "artifact_sha256": result["root_artifact"]["sha256"], "checked_at_stage": "package", + "wic_participation_person_rows": 8, } @@ -1537,6 +1551,81 @@ def test_package_refuses_a_stale_stored_input_before_any_release_dir( assert Path(result["release_dir"]).is_dir() +@requires_pytables +@pytest.mark.parametrize("values", [None, [False, None] * 4, ["False", "True"] * 4]) +def test_package_refuses_incomplete_wic_before_creating_release_directory( + tmp_path, monkeypatch, values +): + from test_support.microcosm_build.us_wic_claim import _replace_person + + monkeypatch.setattr( + wic_claim, "require_complete_us_wic_claim_h5", _REAL_REQUIRE_WIC_H5 + ) + module = _load_tool_module() + args = _package_args_with_hours(module, tmp_path, monkeypatch, gate_state="passed") + frame = _plausible_hours_frame() + person = frame.person.drop(columns="takes_up_wic_if_eligible") + if values is not None: + person["takes_up_wic_if_eligible"] = values + _write_frame_h5(args.out_h5, _replace_person(frame, person)) + with pytest.raises(SystemExit, match="WIC participation"): + module.do_package(args) + assert not (args.out / "releases").exists() + assert not (args.out / "package_result.json").exists() + + +@pytest.mark.parametrize("values", [None, [False, None], ["False", "True"]]) +def test_reviewed_null_register_cannot_default_fill_wic(tmp_path, values): + from test_support.microcosm_build.us_wic_claim import _frame, _replace_person + + module = _load_tool_module() + frame = _frame([{}, {}]) + person = frame.person.copy() + if values is not None: + person["takes_up_wic_if_eligible"] = values + summary = tmp_path / "summary.json" + summary.write_text( + json.dumps( + { + "reviewed_engine_input_nulls": [ + { + "entity": "person", + "column": "takes_up_wic_if_eligible", + "default": True, + } + ] + } + ) + ) + original = person.copy(deep=True) + manifest = tmp_path / "fills.json" + with pytest.raises(ValueError, match="WIC participation"): + module.fill_reviewed_nulls( + _replace_person(frame, person), summary, manifest_path=manifest + ) + pd.testing.assert_frame_equal(person, original) + assert not manifest.exists() + + +def test_consumer_export_refuses_missing_wic_before_writing(tmp_path, monkeypatch): + from test_support.microcosm_build.us_wic_claim import _frame + + module = _load_tool_module() + frame = _frame([{}]) + monkeypatch.setattr(module, "_load_staging_frame", lambda path: frame) + monkeypatch.setattr( + module, + "_require_local_hours", + lambda *args: pytest.fail("must refuse WIC first"), + ) + args = SimpleNamespace( + out_h5=tmp_path / "output.h5", staging_h5=tmp_path / "input.h5" + ) + with pytest.raises(ValueError, match="WIC participation"): + module._write_calibrated_artifact(args, np.ones(1), {}, {}) + assert not args.out_h5.exists() + + def _refresh_package_evidence_bindings(module, args) -> None: """Rebind the package evidence to rewritten artifact bytes.""" diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_sparse_cd.py b/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_sparse_cd.py index c2ecceecda..aaf3fd7d1d 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_sparse_cd.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_sparse_cd.py @@ -168,8 +168,12 @@ def _materialize(module, fixtures, release, monkeypatch, tmp_path, specs, hh_chu fixtures._install_fake_engine(release, monkeypatch, reform_specs=fixtures._REFORMS) monkeypatch.setattr(module, "project_input_only", lambda frame, **kw: (frame, {})) monkeypatch.setattr(module, "fill_reviewed_nulls", lambda *args, **kw: None) + from test_support.microcosm_build.us_wic_claim import _replace_person + + frame = fixtures._nested_frame() + frame = _replace_person(frame, frame.person.assign(takes_up_wic_if_eligible=False)) return module.materialize_chunked( - fixtures._nested_frame(), + frame, specs, hh_chunk=hh_chunk, batch=2, diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_acs_multispine.py b/packages/microcosm-build/tests/engine_free/us/test_us_acs_multispine.py index afc38b72d4..4f886646d8 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_acs_multispine.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_acs_multispine.py @@ -19,6 +19,7 @@ ) from microcosm.build.us_runtime.base_pool import spine_column from microcosm.build.us_runtime.puma_ladder import UsPumaLadder +from microcosm.build.us_runtime.wic_claim import WIC_CLAIM_FNS_SOURCE_URL from microcosm.frame import US_SCHEMA, Frame, WeightKind, Weights @@ -38,6 +39,10 @@ def test__given_no_source__then_base_frame_is_an_untouched_identity( monkeypatch.setattr(acs_multispine, "map_acs_native_inputs", _must_not_run) monkeypatch.setattr(acs_multispine, "transfer_acs_inputs", _must_not_run) monkeypatch.setattr(acs_multispine, "with_optional_acs_spine", _must_not_run) + monkeypatch.setattr(acs_multispine, "with_acs_wic_claim_input", _must_not_run) + monkeypatch.setattr( + acs_multispine, "require_complete_us_wic_claim_input", _must_not_run + ) monkeypatch.setattr( acs_multispine, "us_puma_ladder_assignment_summary", @@ -91,13 +96,13 @@ def test__given_source__then_stages_run_in_order_and_provenance_is_json_ready( tmp_path, ) -> None: events: list[tuple[Any, ...]] = [] - base = object() + base = SimpleNamespace(table=lambda entity: pd.DataFrame()) raw_acs = object() mapped_acs = object() # The post-transfer adult-care gate probes the transferred frame's person # table; an empty table means "columns absent", so the gate scopes out. transferred_acs = SimpleNamespace(table=lambda entity: pd.DataFrame()) - pooled = object() + pooled = SimpleNamespace(table=lambda entity: pd.DataFrame()) source = AcsPumsSource( tmp_path / "csv_hus.zip", tmp_path / "csv_pus.zip", @@ -149,10 +154,18 @@ def fake_pool(actual_base, actual_acs, *, acs_share): events.append(("pool", actual_base, actual_acs, acs_share)) return pooled + def fake_wic(actual_acs, *, seed, time_period): + events.append(("wic", actual_acs, seed, time_period)) + return actual_acs + monkeypatch.setattr(acs_multispine, "build_acs_pums_unit_frame", fake_load) monkeypatch.setattr(acs_multispine, "map_acs_native_inputs", fake_map) monkeypatch.setattr(acs_multispine, "transfer_acs_inputs", fake_transfer) monkeypatch.setattr(acs_multispine, "with_optional_acs_spine", fake_pool) + monkeypatch.setattr(acs_multispine, "with_acs_wic_claim_input", fake_wic) + monkeypatch.setattr( + acs_multispine, "require_complete_us_wic_claim_input", lambda person: None + ) result = acs_multispine.build_optional_acs_multispine( cast(Frame, base), @@ -182,6 +195,7 @@ def fake_pool(actual_base, actual_acs, *, acs_share): "max_targets_per_fit": 8, }, ), + ("wic", transferred_acs, 19, source.vintage), ("pool", base, transferred_acs, 0.4), ] assert result.frame is pooled @@ -218,6 +232,13 @@ def fake_pool(actual_base, actual_acs, *, acs_share): "structural_receipt": None, } ], + "wic_claim": { + "seed": 19, + "source_year": source.vintage, + "person_rows": 0, + "output_columns": ["takes_up_wic_if_eligible"], + "category_rates_source": WIC_CLAIM_FNS_SOURCE_URL, + }, "deferred_inputs": ["congressional_district_geoid"], "adult_care_recipient_gate": None, "fit_records": [ @@ -278,6 +299,15 @@ def test__given_tiny_frames__then_mapping_transfer_and_pool_interoperate( [55_000.0, 0.0] ) assert acs_people["qualified_dividend_income"].notna().all() + assert person.takes_up_wic_if_eligible.dtype == bool + assert person.takes_up_wic_if_eligible.notna().all() + pd.testing.assert_series_equal( + person.loc[ + person[spine_column("person")].ne("acs_2024_1yr"), + "takes_up_wic_if_eligible", + ].reset_index(drop=True), + base.person.takes_up_wic_if_eligible, + ) assert ( result.frame.weights_for("household").total == base.weights_for("household").total @@ -370,6 +400,29 @@ def _test_puma_ladder() -> UsPumaLadder: ) +def test_pre_acs_donor_wic_check_preserves_existing_decisions(): + base = _base_donor_frame() + original_person = base.person.copy(deep=True) + + acs_multispine._require_complete_pre_acs_donor_wic_input(base) + + pd.testing.assert_frame_equal(base.person, original_person) + + +def test_missing_pre_acs_donor_decisions_fail_before_acs_loading(monkeypatch, tmp_path): + from test_support.microcosm_build.us_wic_claim import _replace_person + + base = _base_donor_frame() + missing = _replace_person( + base, base.person.drop(columns="takes_up_wic_if_eligible") + ) + monkeypatch.setattr(acs_multispine, "build_acs_pums_unit_frame", _must_not_run) + with pytest.raises(ValueError, match="WIC participation"): + acs_multispine.build_optional_acs_multispine( + missing, AcsPumsSource(tmp_path / "hus.zip", tmp_path / "pus.zip") + ) + + def _base_donor_frame() -> Frame: person = pd.DataFrame( { @@ -381,6 +434,9 @@ def _base_donor_frame() -> Frame: "person_marital_unit_id": [1, 1, 2, 2], "age": [42.0, 38.0, 65.0, 12.0], "is_female": [False, True, True, False], + "is_pregnant": [False, False, False, False], + "own_children_in_household": [0, 0, 0, 0], + "takes_up_wic_if_eligible": [True, False, True, False], "qualified_dividend_income": [100.0, 20.0, 2_000.0, 0.0], } ) @@ -409,6 +465,11 @@ def _raw_acs_frame() -> Frame: "person_family_id": [10, 10], "person_marital_unit_id": [10, 10], "AGEP": [40, 15], + "is_pregnant": [False, False], + "own_children_in_household": [0, 0], + "source_year": [2024, 2024], + "source_household_id": ["ACS10", "ACS10"], + "source_person_id": [1, 2], "SEX": [1, 2], "RELSHIPP": [20, 25], "ADJINC": [1_100_000, 1_100_000], diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_acs_wic.py b/packages/microcosm-build/tests/engine_free/us/test_us_acs_wic.py new file mode 100644 index 0000000000..93bd5fc7ca --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/us/test_us_acs_wic.py @@ -0,0 +1,85 @@ +"""ACS participation uses the existing category generator, never default fill.""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.source_runtime import SourceRuntimeError +from microcosm.build.us_runtime.wic_claim import ( + US_WIC_CLAIM_REQUIRED_SOURCE_COLUMNS, + require_complete_us_wic_claim_input, + with_acs_wic_claim_input, + with_us_wic_claim_input, +) +from test_support.microcosm_build.us_wic_claim import _frame, _replace_person + +OUTPUT = "takes_up_wic_if_eligible" + + +def test_acs_generation_matches_existing_category_rates_and_is_deterministic(): + frame = _frame( + [{"age": 0}, {"age": 3}, {"is_female": True, "is_pregnant": True}, {}] * 32 + ) + before = frame.person.copy(deep=True) + result = with_acs_wic_claim_input(frame, seed=19, time_period=2024) + expected = with_us_wic_claim_input(frame, seed=19, time_period=2024) + pd.testing.assert_series_equal(result.person[OUTPUT], expected.person[OUTPUT]) + assert result.person[OUTPUT].dtype == bool + assert result.person[OUTPUT].nunique() == 2 + assert not result.person.loc[ + result.person.age.eq(30) & ~result.person.is_pregnant, OUTPUT + ].any() + pd.testing.assert_frame_equal(frame.person, before) + assert with_acs_wic_claim_input(result, seed=19, time_period=2024) is result + reordered = _replace_person(frame, frame.person.iloc[::-1].reset_index(drop=True)) + actual = with_acs_wic_claim_input(reordered, seed=19, time_period=2024) + pd.testing.assert_series_equal( + actual.person.set_index("person_id")[OUTPUT].sort_index(), + result.person.set_index("person_id")[OUTPUT].sort_index(), + ) + + +@pytest.mark.parametrize("column", US_WIC_CLAIM_REQUIRED_SOURCE_COLUMNS[:-1]) +def test_missing_demographic_inputs_fail_before_acs_wic_generation(column): + frame = _frame([{}]) + missing = _replace_person(frame, frame.person.drop(columns=column)) + with pytest.raises(SourceRuntimeError, match="source column"): + with_acs_wic_claim_input(missing, seed=0, time_period=2024) + + +@pytest.mark.parametrize( + "column", ["source_year", "source_household_id", "source_person_id"] +) +def test_missing_acs_identity_is_not_replaced_with_generated_person_ids(column): + frame = _frame([{}]) + with pytest.raises(ValueError, match="source identity"): + with_acs_wic_claim_input( + _replace_person(frame, frame.person.drop(columns=column)), + seed=0, + time_period=2024, + ) + with pytest.raises(ValueError, match="complete source identity"): + with_acs_wic_claim_input( + _replace_person(frame, frame.person.assign(**{column: [None]})), + seed=0, + time_period=2024, + ) + + +@pytest.mark.parametrize( + "values", [[None], [np.nan], [pd.NA], ["False"], [1], [2], [np.inf]] +) +def test_incomplete_or_non_boolean_participation_is_refused(values): + with pytest.raises(ValueError, match="WIC participation"): + require_complete_us_wic_claim_input(pd.DataFrame({OUTPUT: values})) + + +def test_absent_participation_is_refused_and_complete_decisions_are_unchanged(): + with pytest.raises(ValueError, match="requires person column"): + require_complete_us_wic_claim_input(pd.DataFrame({"person_id": [1]})) + person = pd.DataFrame({OUTPUT: [False, True]}) + before = person.copy(deep=True) + require_complete_us_wic_claim_input(person) + pd.testing.assert_frame_equal(person, before) diff --git a/packages/microcosm-build/tests/engine_workflow/us/test_us_acs_wic.py b/packages/microcosm-build/tests/engine_workflow/us/test_us_acs_wic.py new file mode 100644 index 0000000000..8f39c08d36 --- /dev/null +++ b/packages/microcosm-build/tests/engine_workflow/us/test_us_acs_wic.py @@ -0,0 +1,78 @@ +"""Persist ACS decisions and confirm the country engine consumes them.""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.us_runtime.h5_io import ( + LEGACY_NULLABLE_STAGING_ARTIFACT_KIND, + load_legacy_calibrated_us_h5, + write_nullable_us_h5, +) +from microcosm.build.us_runtime.wic_claim import ( + require_complete_us_wic_claim_h5, + with_acs_wic_claim_input, +) +from microcosm.frame.adapters.policyengine_us import PolicyEngineUSEngine +from test_support.microcosm_build.us_wic_claim import _frame + +OUTPUT = "takes_up_wic_if_eligible" + + +def test_generated_acs_participation_round_trips_and_controls_wic_benefits(tmp_path): + frame = _frame([{"age": 3}, {"age": 40}] * 16) + frame = with_acs_wic_claim_input(frame, seed=19, time_period=2024) + path = tmp_path / "complete-acs.h5" + write_nullable_us_h5( + frame, path, period=2024, artifact_kind=LEGACY_NULLABLE_STAGING_ARTIFACT_KIND + ) + assert require_complete_us_wic_claim_h5(path) == len(frame.person) + loaded = load_legacy_calibrated_us_h5(path) + pd.testing.assert_series_equal(frame.person[OUTPUT], loaded.person[OUTPUT]) + assert "would_claim_wic" not in loaded.person + result = PolicyEngineUSEngine().materialize( + loaded, variables=["wic", "is_wic_eligible"], period=2024 + ) + claimed = loaded.person[OUTPUT].to_numpy() + adult = loaded.person.age.eq(40).to_numpy() + child = ~adult + assert claimed[child].any() and (~claimed[child]).any() + assert not result["is_wic_eligible"][adult].any() + assert not result["wic"][~claimed].any() + assert (result["wic"][child & claimed] > 0).all() + + +@pytest.mark.parametrize("hdf_format", ["fixed", "table"]) +def test_h5_validation_reads_every_row_in_bounded_batches( + tmp_path, monkeypatch, hdf_format +): + path = tmp_path / "multi-batch.h5" + pd.DataFrame({OUTPUT: np.zeros(65_537, dtype=bool)}).to_hdf( + path, key="person", format=hdf_format + ) + reads = [] + original = pd.HDFStore.select + + def record_read(store, key, *args, **kwargs): + reads.append((kwargs["start"], kwargs["stop"])) + return original(store, key, *args, **kwargs) + + monkeypatch.setattr(pd.HDFStore, "select", record_read) + assert require_complete_us_wic_claim_h5(path) == 65_537 + assert reads == [(0, 65_536), (65_536, 131_072)] + + +@pytest.mark.parametrize("hdf_format", ["fixed", "table"]) +@pytest.mark.parametrize("values", [None, [1.0, np.nan], ["False", "True"], [1, 0]]) +def test_h5_validation_refuses_missing_or_non_boolean_decisions( + tmp_path, values, hdf_format +): + path = tmp_path / "incomplete.h5" + person = pd.DataFrame({"person_id": [1, 2]}) + if values is not None: + person[OUTPUT] = values + person.to_hdf(path, key="person", format=hdf_format) + with pytest.raises(ValueError, match="WIC participation"): + require_complete_us_wic_claim_h5(path) diff --git a/packages/microcosm-build/tests/engine_workflow/us/test_us_multispine_pool_tool.py b/packages/microcosm-build/tests/engine_workflow/us/test_us_multispine_pool_tool.py index 9606d75b4c..c2e6383773 100644 --- a/packages/microcosm-build/tests/engine_workflow/us/test_us_multispine_pool_tool.py +++ b/packages/microcosm-build/tests/engine_workflow/us/test_us_multispine_pool_tool.py @@ -338,7 +338,7 @@ def capture_equality(expected: object, actual: object) -> None: "country": "us", "schema_id": "country_spec", "schema_version": 1, - "spec_sha256": "f76f921d4f74392b106b8c0729ce250494e0a55455eaba7b910b19cdf7825178", + "spec_sha256": "5048cabc59e5d7d9789e751cc9a5c5b4d97ab6fb3f15429cabce3b568615073b", }, } diff --git a/tools/build_us_acs_local_release.py b/tools/build_us_acs_local_release.py index 464a658fd9..4f0bf195b4 100644 --- a/tools/build_us_acs_local_release.py +++ b/tools/build_us_acs_local_release.py @@ -79,6 +79,7 @@ import pandas as pd from microcosm.build.us_runtime import target_loss_weights as loss_weighting +from microcosm.build.us_runtime import wic_claim from microcosm.build.us_runtime.acs_local_hours import acs_local_hours_signal_gate _TOOLS_DIR = Path(__file__).resolve().parent @@ -505,6 +506,10 @@ def fill_reviewed_nulls( defect, surfaced not filled. """ + # Participation is a required generated input, not a reviewed engine + # default. A null register must never turn missing decisions into True. + wic_claim.require_complete_us_wic_claim_input(frame.table("person")) + from policyengine_us import CountryTaxBenefitSystem from microcosm.build.us_runtime.base_pool import spine_column @@ -749,6 +754,7 @@ def materialize_chunked( import build_us_fiscal_refresh_release as release_tool + wic_claim.require_complete_us_wic_claim_input(base_frame.table("person")) projected, dropped = project_input_only(base_frame, period=period) total_held = sum(len(columns) for columns in dropped.values()) log(f"input-schema projection held back {total_held} formula-owned column(s)") @@ -1344,6 +1350,7 @@ def do_materialize(args) -> None: staging_sha = _sha256(args.staging_h5) ladder_sha = _sha256(args.ladder) frame = _load_staging_frame(args.staging_h5) + wic_claim.require_complete_us_wic_claim_input(frame.table("person")) _require_local_hours(frame, _load_json(summary_path)) frame, sampling = cd_surface.sample_staging_frame( frame, fraction=args.sample_fraction, seed=args.sample_seed @@ -2181,6 +2188,7 @@ def _write_calibrated_artifact( from microcosm.frame import Frame, WeightKind, Weights frame = _load_staging_frame(args.staging_h5) + wic_claim.require_complete_us_wic_claim_input(frame.table("person")) _require_local_hours(frame, _load_json(_staging_summary_path(args))) frame = _resample_like_materialize(frame, identity) staging_ids = frame.table("household")["household_id"].to_numpy() @@ -3046,7 +3054,8 @@ def _require_stored_inputs(calibrated_h5: Path) -> dict[str, object]: :mod:`microcosm.data.stored_inputs` owns the rule and the reviewed register of deliberately non-variable columns. The engine is the installed policyengine-us, the one :func:`do_package` records as - ``build.built_with_model_package``. Only HDF metadata is read. + ``build.built_with_model_package``. Stored-name validation reads metadata; + WIC completeness also validates actual participation in bounded row batches. Returns the gate entry the release's gate summary records. """ @@ -3068,7 +3077,17 @@ def _require_stored_inputs(calibrated_h5: Path) -> dict[str, object]: stored_inputs.StoredTableLayoutError, ) as error: raise SystemExit(f"Refusing to package: {error}") from error - return {"passed": True, "failures": [], "engine": engine.label, **summary} + try: + wic_rows = wic_claim.require_complete_us_wic_claim_h5(calibrated_h5) + except (ValueError, KeyError) as error: + raise SystemExit(f"Refusing to package: {error}") from error + return { + "passed": True, + "failures": [], + "engine": engine.label, + **summary, + "wic_participation_person_rows": wic_rows, + } def do_package(args) -> dict: