diff --git a/changelog.d/us-asec-720-census-person-columns.fixed.md b/changelog.d/us-asec-720-census-person-columns.fixed.md new file mode 100644 index 0000000000..e2c0e67958 --- /dev/null +++ b/changelog.d/us-asec-720-census-person-columns.fixed.md @@ -0,0 +1 @@ +The US base's source construction restores eleven reviewed Census person columns that the pinned income-year 2022/2023 ASEC inputs lack: seven `NOW_*` at-interview coverage recodes, `A_EXPRRP`, `PTOTVAL`, `A_ENRLW` and `A_FTPT`. They are read from each year's pinned Census person archive (`--asec-education-source`) by exact one-to-one `PERIDNUM` identity, and a column the input already carries is verified, never overwritten (microcosm #720). On route A's inputs the 2022/2023 vintages gain Medicaid reporters (28,184 / 26,367), the #744 reported-coverage vintage gate passes, and the older vintages switch to Census's relationship recode (231,007 → 226,102 tax units). diff --git a/docs/us-asec-census-person-columns.md b/docs/us-asec-census-person-columns.md new file mode 100644 index 0000000000..73592bac7a --- /dev/null +++ b/docs/us-asec-census-person-columns.md @@ -0,0 +1,194 @@ +# Census person columns the 2022/2023 ASEC inputs lack (#720, layer 1) + +Status as of 2026-09-23, branch `us-asec-720-coverage-recodes` (stacked on +`us-spm-role-stage`, #959). Code: `microcosm.build.us_runtime.asec_census_person_columns`. + +## The defect + +The base pools three processed CPS ASEC inputs, pinned in +`us_runtime/asec_sources.py`: + +| income year | file | SHA-256 | person rows | `NOW_*` recodes | +|---|---|---|---:|---:| +| 2022 | `census_cps_2022.h5` | `7ccca976…` | 146,133 | 2 of 18 | +| 2023 | `census_cps_2023.h5` | `cb578173…` | 144,265 | 2 of 18 | +| 2024 | `census_cps_2024.h5` | `ec36604c…` | 142,125 | 18 of 18 | + +The 2022 and 2023 files were extracted with an older column list. Besides 16 +`NOW_*` at-interview coverage recodes they lack `A_EXPRRP`, `PTOTVAL`, +`A_ENRLW`, `A_FTPT`, `A_FAMREL`, `A_FAMTYP` and `PECOHAB`; the 2022 file also +lacks `LKWEEKS`. `asec_pool.pool_asec_sources` concatenates the per-year +person tables, so a column only the 2024 file carries is NaN on every +2022/2023 row, and each reader treats NaN as "no" or zero: + +- `cps_carried._fill_health_coverage_inputs` maps seven reported-coverage + flags from `NOW_*` codes. They were False for every 2022/2023 person, and + the #744 gate (`us_reported_coverage_vintage_signal_gate`) refused route A's + release with 14 failures (seven flags times two vintages). +- `asec_pool._with_relationship_recode` found no `A_EXPRRP` for those years + and derived one from line, spouse and parent pointers; it never produces the + Census partner/roommate code (13) and differs from Census's own recode for + about 14% of 2022/2023 persons. +- microunit's tax-unit construction (fed by `microcosm.frame.units`) read + `PTOTVAL` as 0 and saw no enrollment (`A_ENRLW`, `A_FTPT`) for those years; + `eligibility_inputs` set `is_full_time_college_student` False for all of + them. + +## The fix + +The official Census archives the inputs were extracted from are already pinned +build inputs. The base stage passes them as `--asec-education-source`, and +`spm_role_source.ASEC_SPM_ROLE_SOURCES` pins each archive's SHA-256 and its +person member's name, size and SHA-256. `restore_asec_census_person_columns` +runs in `asec_pool._prepare_year_input` for each vintage. It runs straight +after the H5 is read, before the relationship recode and before any +smoke-household limit. It appends every reviewed column the H5 lacks, reads +those columns from that year's pinned member, and joins them on exact +`PERIDNUM`. It uses `spm_role_source.read_pinned_asec_person_columns`, the SPM +role stage's own reader. That reader now takes a column list, and its default +read is unchanged. + +`build_us_puf_support_base.py` and `build_us_asec_pooled_source_base.py` bind +every pooled year to its Census person file. They use the +`--asec-education-source` mapping or, for an unmapped year, a fetch from the +official archive that is verified against the same pins. They resolve the +file the same way the SPM role stage does. Every base built from `--asec-h5` +inputs therefore carries the columns. The source-construction frame, the +`asec_raw_stage` checkpoint, pre-clone enrichment and the export all see them. + +### Refusals (`AsecCensusPersonColumnsError`, a `ValueError`) + +The restoration refuses any of the following: + +- The H5 lacks `PERIDNUM`, `PH_SEQ`, `P_SEQ`, `A_LINENO` or `A_AGE`. It also + refuses repeated column names or a repeated row index. +- The archive or member fails its pins, or the member's person count differs + from the pin. +- The member lacks a reviewed column. A reviewed column parses as anything but + integers (a blank, a fraction or text), or a value falls outside the column's + reviewed Census codes. +- A `PERIDNUM` is not an exact 22-digit string, or either side repeats one. +- An H5 person has no member row, or a member person is absent from the H5. + The join must be one-to-one and total over the complete member. +- `PH_SEQ`, `P_SEQ`, `A_LINENO` or `A_AGE` disagree after the join. +- A reviewed column the H5 already carries has missing or non-integer values, + or differs from the member on any row. Such a column is never overwritten. + On 2024, where the H5 carries all eleven, this check validates the join on + real data. +- Any existing H5 column changes position, dtype or value. This condition is + checked explicitly, not asserted. + +No value is predicted, imputed or defaulted. + +## Column review + +Start: the 24 columns the 2026-08-23 offline fix appended +(`_buildo-runtime/inputs/asec-720/receipt_720.json`). A column is restored +only if build code reads it. Readers come from a grep of `packages/*/src` and +`tools/` on 2026-09-23. `test_every_restored_column_cites_readers_that_really_read_it` +resolves every citation and requires the cited function or constant to name +the column. + +| column | restored | reader, or why not | +|---|---|---| +| `NOW_MCAID` | yes | `cps_carried._fill_health_coverage_inputs` → `has_medicaid_health_coverage_at_interview` | +| `NOW_NONM` | yes | same → `has_non_marketplace_direct_purchase_health_coverage_at_interview` | +| `NOW_CHAMPVA` | yes | same → `has_champva_health_coverage_at_interview` | +| `NOW_MIL` | yes | same → `has_tricare_health_coverage_at_interview` | +| `NOW_VACARE` | yes | same → `has_va_health_coverage_at_interview` | +| `NOW_OTHMT` | yes | same → `has_other_means_tested_health_coverage_at_interview` | +| `NOW_IHSFLG` | yes | same → `has_indian_health_service_coverage_at_interview` | +| `A_EXPRRP` | yes | `asec_pool._with_relationship_recode` (the Census recode replaces the derived fallback); `frame.units.MICROUNIT_REQUIRED_COLUMNS` → microunit `_prepare_household_people` (tax-unit head choice); `acs_transfer._person_head_feature`; `us_late_producer_registry._transfer_input_inventory` | +| `PTOTVAL` | yes | `frame.units._is_microunit_optional_column` → microunit `_precompute_tax_unit_inputs` | +| `A_ENRLW` | yes | `frame.units._is_microunit_optional_column` → microunit `_is_full_time_student` | +| `A_FTPT` | yes | same; `eligibility_inputs.derive_us_eligibility_inputs_from_manifest` (`is_full_time_college_student`) | +| `NOW_CAID`, `NOW_COV`, `NOW_DIR`, `NOW_MCARE`, `NOW_MRKS`, `NOW_MRKUN`, `NOW_PCHIP`, `NOW_PRIV`, `NOW_PUB` | no | No build reader. Restoring them would only widen the exported person table. A test fails if code starts reading one. | +| `A_FAMTYP`, `A_FAMREL`, `PECOHAB` | no | Read only as optional cross-checks by the SPM role derivation (`spm_role_source._OPTIONAL_RAW_CHECKS`). That derivation reads this same member itself and derives the role from it, so restoring these columns would make the check compare the member with itself. #38's partner derivation should restore them together with its reader. | +| `LKWEEKS` | no | Already restored for 2022, the only vintage without it, by `weeks_unemployed.fill_asec_2022_weeks_unemployed_source` from the same pinned `pppub23.csv`. The 2023 and 2024 inputs carry it. On the real run it equals the 8/23 file on all 146,133 rows. | + +`CENSUS_TAX_ID` was never appended: the extractor derives it, and the Census +files do not carry it. + +## Real-data evidence (2026-09-23, commit `39b8e7b63`) + +The real-data run used route A's exact base inputs and flags +(`route-a/run-47976be6ce75/base-config.json`) with +`--stage source_construction` only. Its checkpoints are under +`_recovered/scratch-backup/893/overnight-20260923/asec-720-fix/`. The run took +69 s at 11.2 GB peak RSS. The inputs matched their pins: the H5 files +`7ccca976…`, `cb578173…` and `ec36604c…`, and the archives `d2e00025…`, +`cdb39cda…` and `318845a2…`. + +### The #744 gate + +The gate ran on `derive_us_cps_carried_inputs` of the ASEC raw-stage frame. +"Before" is route A's raw-stage checkpoint (commit `47976be6c`); "after" is +this branch. + +| | 2022 | 2023 | 2024 | gate | +|---|---:|---:|---:|---| +| Medicaid reporters, before | 0 | 0 | 24,844 | **FAIL, 14 failures** | +| Medicaid reporters, after | 28,184 | 26,367 | 24,844 | **PASS, 0 failures** | +| 2026-08-23 receipt | 28,184 | 26,367 | 24,844 | pass | + +The other restored flags after the fix, as 2022 / 2023 / 2024 reporters: + +- non-marketplace direct purchase: 8,666 / 8,435 / 8,486 +- TRICARE: 3,882 / 4,318 / 4,413 +- VA: 1,271 / 1,178 / 1,270 +- IHS: 687 / 651 / 685 +- CHAMPVA: 260 / 335 / 368 +- other means-tested: 275 / 331 / 312 + +There are no nulls. `has_esi` and marketplace are unchanged. Every 2024 +reported-coverage count is unchanged. + +### Equality with the 2026-08-23 corrected inputs + +- Restoring each pinned H5 as the pool does yields the corrected files' + columns exactly, in value, dtype and row order, for all 11 restored columns + in both 2022 and 2023. Every pre-existing column is identical too. The + corrected files differ only in the 13 columns listed above as not restored. +- The whole raw-stage frame equals the 8/23 corrected raw-stage frame on + every entity table: family, household, marital unit, person, SPM unit and + tax unit, including the 226,102 tax units. It also matches on all 176 + shared person columns except exactly the 12 unrestored columns the 8/23 + frame also carried. `LKWEEKS` is identical. +- On 2024 the H5 carries all eleven columns, and all eleven were verified + equal to `pppub25.csv` on 142,125 persons. + +### Structural effect (intended; flag for downstream) + +| | before | after | +|---|---:|---:| +| tax units, all vintages | 231,007 | 226,102 | +| tax units 2022 / 2023 / 2024 | 78,221 / 78,089 / 74,697 | 75,797 / 75,608 / 74,697 | +| tax units per person 2022 / 2023 / 2024 | 0.535 / 0.541 / 0.526 | 0.519 / 0.524 / 0.526 | +| 2022 / 2023 persons whose `A_EXPRRP` changes | — | 20,595 / 20,561 | +| `A_EXPRRP == 13` persons 2022 / 2023 | 0 / 0 | 4,312 / 4,387 | + +Household, family, SPM-unit, marital-unit and person counts are unchanged, as +is every 2024 value. The 8/23 run produced the same 226,102 tax units. The +older vintages' tax-unit rate moves toward the 2024 rate, which always used +the Census recode. + +## Coverage + +- Covered: every `tools/build_us_puf_support_base.py` run that pools + `--asec-h5` inputs, in both staged and monolithic runs and with a support + spine spec, and `tools/build_us_asec_pooled_source_base.py`. +- Covered only after regeneration: `tools/build_us_multispine_pool.py` reads + the SHA-pinned `asec_raw_stage.checkpoint.h5` that the base's source + construction writes. A checkpoint built before this change keeps the hole, + and the #744 gate refuses a release built from it. +- Not affected: `--base-h5` builds, which start from an existing frame, and the + enrichment and qualification lanes, which start from published parents + (`build_us_spm_role_enrichment.py`, + `build_us_acs_donor_receipt_qualification.py`). ACS-spine rows come from ACS + PUMS, not these H5 files. +- Unchanged: the #744 gate, poverty and SPM outcomes (neither a target nor a + gate), and the reviewed `is_unmarried_partner_of_household_head` exclusion. + That exclusion describes the inputs and remains true of them. With the + Census recode restored, `A_EXPRRP == 13` now appears in every vintage, but + it still conflates partners and roommates, so the exclusion's conclusion + stands. diff --git a/experiments/us-asec-720-census-person-columns/README.md b/experiments/us-asec-720-census-person-columns/README.md new file mode 100644 index 0000000000..4d811cd22b --- /dev/null +++ b/experiments/us-asec-720-census-person-columns/README.md @@ -0,0 +1,16 @@ +# #720 layer 1: real-data receipts (2026-09-23) + +Evidence for `docs/us-asec-census-person-columns.md`. The scripts ran from +`_recovered/scratch-backup/893/overnight-20260923/asec-720-fix/` (they resolve +`checkpoints/` next to themselves) on commit `39b8e7b63`: + +- `run_source_construction.sh`: route A's base inputs and flags, `--stage source_construction` only. + `source_construction_stage_run_context.json` / `source_construction_stage_profile.json` + are its stage context and profile; `checkpoint_sha256.txt` the checkpoints it wrote. +- `verify_720.py` → `verify_720.json`: the #744 gate before (route A's raw-stage + checkpoint, commit `47976be6c`) and after; per-vintage reporter counts; column-by-column + equality with the 2026-08-23 corrected H5s; structural deltas. +- `compare_823_raw_stage.py` → `compare_823_raw_stage.json`: this branch's raw-stage frame + against the 2026-08-23 corrected raw-stage frame, entity by entity. + +Checkpoints are local build evidence, not release artifacts. diff --git a/experiments/us-asec-720-census-person-columns/checkpoint_sha256.txt b/experiments/us-asec-720-census-person-columns/checkpoint_sha256.txt new file mode 100644 index 0000000000..d78acde2c2 --- /dev/null +++ b/experiments/us-asec-720-census-person-columns/checkpoint_sha256.txt @@ -0,0 +1,2 @@ +be8c268930ff90a46af5225cc95244cc3db9f66ff37b26211d7515b75d2338cf /Users/maxghenis/PolicyEngine/_recovered/scratch-backup/893/overnight-20260923/asec-720-fix/checkpoints/000_source_construction.frame.h5 +0ae0b7558094ee1175fc03494e07ef79554e40b253a9373ed3f62c1224435395 /Users/maxghenis/PolicyEngine/_recovered/scratch-backup/893/overnight-20260923/asec-720-fix/checkpoints/asec_raw_stage.checkpoint.h5 diff --git a/experiments/us-asec-720-census-person-columns/compare_823_raw_stage.json b/experiments/us-asec-720-census-person-columns/compare_823_raw_stage.json new file mode 100644 index 0000000000..5537487a46 --- /dev/null +++ b/experiments/us-asec-720-census-person-columns/compare_823_raw_stage.json @@ -0,0 +1,158 @@ +{ + "family": { + "differing_shared_columns": [], + "only_ours": [], + "only_theirs": [], + "rows": [ + 189401, + 189401 + ], + "shared_columns": 1 + }, + "household": { + "differing_shared_columns": [], + "only_ours": [], + "only_theirs": [], + "rows": [ + 168852, + 168852 + ], + "shared_columns": 3 + }, + "marital_unit": { + "differing_shared_columns": [], + "only_ours": [], + "only_theirs": [], + "rows": [ + 345873, + 345873 + ], + "shared_columns": 1 + }, + "ours": "/Users/maxghenis/PolicyEngine/_recovered/scratch-backup/893/overnight-20260923/asec-720-fix/checkpoints/asec_raw_stage.checkpoint.h5", + "peak_rss_gb": 4.952326144, + "person": { + "differing_shared_columns": [ + "PECOHAB", + "A_FAMREL", + "A_FAMTYP", + "NOW_CAID", + "NOW_COV", + "NOW_DIR", + "NOW_MCARE", + "NOW_MRKS", + "NOW_MRKUN", + "NOW_PCHIP", + "NOW_PRIV", + "NOW_PUB" + ], + "only_ours": [], + "only_theirs": [], + "rows": [ + 432523, + 432523 + ], + "shared_columns": 176 + }, + "restored_columns_identical_across_all_rows": { + "A_ENRLW": true, + "A_EXPRRP": true, + "A_FTPT": true, + "NOW_CHAMPVA": true, + "NOW_IHSFLG": true, + "NOW_MCAID": true, + "NOW_MIL": true, + "NOW_NONM": true, + "NOW_OTHMT": true, + "NOW_VACARE": true, + "PTOTVAL": true + }, + "spm_unit": { + "differing_shared_columns": [], + "only_ours": [], + "only_theirs": [], + "rows": [ + 176039, + 176039 + ], + "shared_columns": 1 + }, + "tax_unit": { + "differing_shared_columns": [], + "only_ours": [], + "only_theirs": [], + "rows": [ + 226102, + 226102 + ], + "shared_columns": 2 + }, + "theirs": "/Users/maxghenis/PolicyEngine/_buildo-runtime/out/asec-720-source/checkpoints/asec_raw_stage.checkpoint.h5", + "unrestored_columns": { + "A_FAMREL": { + "2024_rows_identical": true, + "ours_null_rows_2022_2023": 290398, + "theirs_null_rows_2022_2023": 0 + }, + "A_FAMTYP": { + "2024_rows_identical": true, + "ours_null_rows_2022_2023": 290398, + "theirs_null_rows_2022_2023": 0 + }, + "LKWEEKS": { + "2024_rows_identical": true, + "ours_null_rows_2022_2023": 0, + "theirs_null_rows_2022_2023": 0 + }, + "NOW_CAID": { + "2024_rows_identical": true, + "ours_null_rows_2022_2023": 290398, + "theirs_null_rows_2022_2023": 0 + }, + "NOW_COV": { + "2024_rows_identical": true, + "ours_null_rows_2022_2023": 290398, + "theirs_null_rows_2022_2023": 0 + }, + "NOW_DIR": { + "2024_rows_identical": true, + "ours_null_rows_2022_2023": 290398, + "theirs_null_rows_2022_2023": 0 + }, + "NOW_MCARE": { + "2024_rows_identical": true, + "ours_null_rows_2022_2023": 290398, + "theirs_null_rows_2022_2023": 0 + }, + "NOW_MRKS": { + "2024_rows_identical": true, + "ours_null_rows_2022_2023": 290398, + "theirs_null_rows_2022_2023": 0 + }, + "NOW_MRKUN": { + "2024_rows_identical": true, + "ours_null_rows_2022_2023": 290398, + "theirs_null_rows_2022_2023": 0 + }, + "NOW_PCHIP": { + "2024_rows_identical": true, + "ours_null_rows_2022_2023": 290398, + "theirs_null_rows_2022_2023": 0 + }, + "NOW_PRIV": { + "2024_rows_identical": true, + "ours_null_rows_2022_2023": 290398, + "theirs_null_rows_2022_2023": 0 + }, + "NOW_PUB": { + "2024_rows_identical": true, + "ours_null_rows_2022_2023": 290398, + "theirs_null_rows_2022_2023": 0 + }, + "PECOHAB": { + "2024_rows_identical": true, + "ours_null_rows_2022_2023": 290398, + "theirs_null_rows_2022_2023": 0 + } + } +} \ No newline at end of file diff --git a/experiments/us-asec-720-census-person-columns/compare_823_raw_stage.py b/experiments/us-asec-720-census-person-columns/compare_823_raw_stage.py new file mode 100644 index 0000000000..01321d4017 --- /dev/null +++ b/experiments/us-asec-720-census-person-columns/compare_823_raw_stage.py @@ -0,0 +1,113 @@ +"""Compare this branch's ASEC raw-stage frame with the 2026-08-23 corrected one. + +The 8/23 frame was built from H5s with 24 appended columns (receipt_720.json); +this branch restores 11 of them in the pool. Every entity table's rows, every +shared column, and the unit assignments should agree; the only expected +differences are the 13 unrestored columns (all-populated there, NaN on the +2022/2023 rows here) and columns other code changes since 8/23 added. +""" + +from __future__ import annotations + +import gc +import json +import resource +from pathlib import Path + +import numpy as np +import pandas as pd + +from microcosm.build.frame_checkpoint import load_frame_checkpoint +from microcosm.build.us_runtime.asec_census_person_columns import ( + ASEC_CENSUS_PERSON_COLUMN_NAMES, + ASEC_CENSUS_PERSON_COLUMNS_NOT_RESTORED, +) + +HERE = Path(__file__).resolve().parent +OURS = HERE / "checkpoints" / "asec_raw_stage.checkpoint.h5" +THEIRS = Path( + "/Users/maxghenis/PolicyEngine/_buildo-runtime/out/asec-720-source/" + "checkpoints/asec_raw_stage.checkpoint.h5" +) + + +def tables(path: Path) -> dict[str, pd.DataFrame]: + frame = load_frame_checkpoint(path).frame + return {entity: frame.table(entity) for entity in frame.entities} + + +def same(left: pd.Series, right: pd.Series) -> bool: + if left.equals(right): + return True + try: + a = pd.to_numeric(left, errors="raise").to_numpy(dtype=np.float64) + b = pd.to_numeric(right, errors="raise").to_numpy(dtype=np.float64) + except (TypeError, ValueError): + return bool( + ( + left.astype("string").fillna("") + == right.astype("string").fillna("") + ).all() + ) + return bool(np.array_equal(a, b, equal_nan=True)) + + +def main() -> None: + ours = tables(OURS) + theirs = tables(THEIRS) + report: dict[str, object] = {"ours": str(OURS), "theirs": str(THEIRS)} + for entity in sorted(set(ours) | set(theirs)): + left, right = ours.get(entity), theirs.get(entity) + if left is None or right is None: + report[entity] = {"present": [left is not None, right is not None]} + continue + shared = [c for c in left.columns if c in right.columns] + differing = [ + c + for c in shared + if len(left) != len(right) + or not same(left[c].reset_index(drop=True), right[c].reset_index(drop=True)) + ] + report[entity] = { + "rows": [len(left), len(right)], + "shared_columns": len(shared), + "differing_shared_columns": differing, + "only_ours": sorted(set(left.columns) - set(right.columns)), + "only_theirs": sorted(set(right.columns) - set(left.columns)), + } + person_ours, person_theirs = ours["person"], theirs["person"] + year = person_ours["source_year"].to_numpy() + restored = {} + for column in ASEC_CENSUS_PERSON_COLUMN_NAMES: + restored[column] = same(person_ours[column], person_theirs[column]) + report["restored_columns_identical_across_all_rows"] = restored + unrestored = {} + for column in ASEC_CENSUS_PERSON_COLUMNS_NOT_RESTORED: + if column not in person_ours or column not in person_theirs: + unrestored[column] = "absent on one side" + continue + is_2024 = year == 2024 + unrestored[column] = { + "2024_rows_identical": same( + person_ours.loc[is_2024, column].reset_index(drop=True), + person_theirs.loc[is_2024, column].reset_index(drop=True), + ), + "ours_null_rows_2022_2023": int( + person_ours.loc[~is_2024, column].isna().sum() + ), + "theirs_null_rows_2022_2023": int( + person_theirs.loc[~is_2024, column].isna().sum() + ), + } + report["unrestored_columns"] = unrestored + report["peak_rss_gb"] = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1e9 + (HERE / "compare_823_raw_stage.json").write_text( + json.dumps(report, indent=1, sort_keys=True) + ) + print(json.dumps(report, indent=1, sort_keys=True)) + del ours, theirs + gc.collect() + + +if __name__ == "__main__": + main() diff --git a/experiments/us-asec-720-census-person-columns/run_source_construction.sh b/experiments/us-asec-720-census-person-columns/run_source_construction.sh new file mode 100755 index 0000000000..1f0e0ecaf2 --- /dev/null +++ b/experiments/us-asec-720-census-person-columns/run_source_construction.sh @@ -0,0 +1,31 @@ +#!/bin/zsh +# Source-construction stage only (issue #720 fix verification), route A's inputs. +set -euo pipefail +WT=/Users/maxghenis/PolicyEngine/_worktrees/us-asec-720-coverage-recodes +OUT=/Users/maxghenis/PolicyEngine/_recovered/scratch-backup/893/overnight-20260923/asec-720-fix +STO=/Users/maxghenis/PolicyEngine/policyengine-us-data/policyengine_us_data/storage +ARC=/Users/maxghenis/PolicyEngine/_buildm-runtime/inputs/asec_education +cd "$WT" +export PYTHONHASHSEED=0 PYTHONUNBUFFERED=1 +/usr/bin/time -l "$WT/.venv/bin/python" -B tools/build_us_puf_support_base.py \ + --stage source_construction \ + --checkpoint-dir "$OUT/checkpoints" \ + --asec-h5 2024=$STO/census_cps_2024.h5 \ + --asec-h5 2023=$STO/census_cps_2023.h5 \ + --asec-h5 2022=$STO/census_cps_2022.h5 \ + --asec-h5-sha256 2024=ec36604cb735a660b51b0b2f90be27d803b5878f3464fb30d0eacead59c1260d \ + --asec-h5-sha256 2023=cb57817327799f42b741caed5f9be94d04021c2e6809c1ad7bd0686da5428d88 \ + --asec-h5-sha256 2022=7ccca976284bb47815d84460cc4f75a0a65d26d7754ab0a0f417de351b3d474e \ + --puf-h5 $STO/puf_2024.h5 \ + --puf-source-year-csv $STO/puf_2015.csv \ + --acs-h5 $STO/acs_2022.h5 \ + --asec-education-source 2022=$ARC/asecpub23csv.zip \ + --asec-education-source 2023=$ARC/asecpub24csv.zip \ + --asec-education-source 2024=$ARC/asecpub25csv.zip \ + --target-year 2024 --seed 0 --n-estimators 32 \ + --ledger-facts /Users/maxghenis/PolicyEngine/_buildh-runtime/inputs/consumer_facts_us_c5e5bf8.jsonl \ + --assign-congressional-districts \ + --congressional-district-vintage-crosswalk "$WT/packages/microcosm-build/src/microcosm/build/us_runtime/data/congressional_district_vintage_crosswalk.csv" \ + --congressional-district-seed 0 \ + --block-ladder-artifact /Users/maxghenis/PolicyEngine/_buildf-runtime/inputs/us_block_ladder_2020.npz \ + --out "$OUT/base-out" diff --git a/experiments/us-asec-720-census-person-columns/source_construction_stage_profile.json b/experiments/us-asec-720-census-person-columns/source_construction_stage_profile.json new file mode 100644 index 0000000000..3041d0475a --- /dev/null +++ b/experiments/us-asec-720-census-person-columns/source_construction_stage_profile.json @@ -0,0 +1,16 @@ +{ + "schema_version": 1, + "stages": { + "source_construction": { + "entry_rss_bytes": 2057764864, + "error": null, + "exit_rss_bytes": 10696212480, + "peak_rss_bytes": 11240964096, + "rss_sample_count": 138, + "sampling_error": null, + "stage": "source_construction", + "status": "succeeded", + "wall_seconds": 43.59271766699385 + } + } +} diff --git a/experiments/us-asec-720-census-person-columns/source_construction_stage_run_context.json b/experiments/us-asec-720-census-person-columns/source_construction_stage_run_context.json new file mode 100644 index 0000000000..745217b652 --- /dev/null +++ b/experiments/us-asec-720-census-person-columns/source_construction_stage_run_context.json @@ -0,0 +1 @@ +{"artifact_kind":"populace_outer_stage_run_context","completed":["source_construction"],"pipeline":{"stages":[{"description":"_load_base_frame_from_args","name":"source_construction"},{"description":"derive_us_cps_carried_inputs -> with_us_prior_year_income_inputs -> with_us_relationship_inputs -> with_us_spm_independence_role -> with_us_medicare_take_up_input -> with_us_housing_inputs[includes_acs_rent_in_current_order] -> with_us_eligibility_inputs -> with_us_pregnancy_inputs -> with_us_wic_claim_input -> with_us_child_support_inputs -> with_us_disability_benefits -> with_us_workers_compensation -> with_us_weeks_unemployed -> with_us_childcare_inputs -> with_us_energy_subsidy_input -> with_us_retirement_contribution_inputs -> with_us_retirement_distribution_inputs -> with_us_immigration_inputs","name":"pre_clone_enrichment"},{"description":"clone_us_frame_for_puf_support -> puf_tax_unit_donor_from_arrays -> initialize_primary_puf_qrf_chain","name":"clone_feature_extraction"},{"description":"run_primary_puf_qrf_chain[target_subprocesses]","name":"primary_qrf_chain"},{"description":"finalize_primary_puf_qrf_chain","name":"qrf_finalization"},{"description":"transfer_puf_capital_gains_tail","name":"capital_gains_tail_transfer"},{"description":"run_source_stage[capital_gain_distributions]","name":"capital_gain_distributions"},{"description":"with_us_qbi_input_reconciliation","name":"qbi_reconciliation"},{"description":"with_us_wic_claim_input","name":"wic_post_clone"},{"description":"impute_us_housing_assistance_to_puf_support","name":"housing_assistance"},{"description":"with_us_prior_year_income_inputs","name":"prior_year_income_post_clone"},{"description":"with_us_child_support_inputs","name":"child_support_post_clone"},{"description":"with_us_disability_benefits","name":"disability_benefits_post_clone"},{"description":"with_us_workers_compensation","name":"workers_compensation_post_clone"},{"description":"with_us_weeks_unemployed","name":"weeks_unemployed_post_clone"},{"description":"with_us_childcare_inputs","name":"childcare_post_clone"},{"description":"with_us_adult_care_inputs","name":"adult_care_post_clone"},{"description":"with_us_energy_subsidy_input","name":"energy_subsidy_post_clone"},{"description":"with_us_retirement_contribution_inputs","name":"retirement_contributions_post_clone"},{"description":"with_us_retirement_distribution_inputs","name":"retirement_distributions_post_clone"},{"description":"with_us_education_inputs","name":"education_inputs_post_clone"},{"description":"with_household_congressional_districts","name":"congressional_district_assignment"},{"description":"with_household_us_geography_ladder","name":"block_ladder_assignment"},{"description":"PolicyEngineUSEngine.write_dataset","name":"final_export"}]},"pipeline_sha256":"906c9c022de6d154771d8250897a77430b0f457371b3d76a6bd046bceaccd266","run_config":{"acs_h5":"/Users/maxghenis/PolicyEngine/policyengine-us-data/policyengine_us_data/storage/acs_2022.h5","allow_geography_ladder_gate_failures":false,"asec_2023_weeks_unemployed_source":null,"asec_education_source":{"2022":"/Users/maxghenis/PolicyEngine/_buildm-runtime/inputs/asec_education/asecpub23csv.zip","2023":"/Users/maxghenis/PolicyEngine/_buildm-runtime/inputs/asec_education/asecpub24csv.zip","2024":"/Users/maxghenis/PolicyEngine/_buildm-runtime/inputs/asec_education/asecpub25csv.zip"},"asec_h5":["2024=/Users/maxghenis/PolicyEngine/policyengine-us-data/policyengine_us_data/storage/census_cps_2024.h5","2023=/Users/maxghenis/PolicyEngine/policyengine-us-data/policyengine_us_data/storage/census_cps_2023.h5","2022=/Users/maxghenis/PolicyEngine/policyengine-us-data/policyengine_us_data/storage/census_cps_2022.h5"],"asec_h5_sha256":{"2022":"7ccca976284bb47815d84460cc4f75a0a65d26d7754ab0a0f417de351b3d474e","2023":"cb57817327799f42b741caed5f9be94d04021c2e6809c1ad7bd0686da5428d88","2024":"ec36604cb735a660b51b0b2f90be27d803b5878f3464fb30d0eacead59c1260d"},"asec_max_households":null,"assign_congressional_districts":true,"base_h5":null,"block_ladder_artifact":"/Users/maxghenis/PolicyEngine/_buildf-runtime/inputs/us_block_ladder_2020.npz","builder_code_identity":{"dependency_versions":{"h5py":"3.16.0","numpy":"2.4.6","pandas":"3.0.3","policyengine-us":"2.2.1","quantile-forest":"1.4.2","scikit-learn":"1.8.0"},"python":"3.14.4 (main, Apr 14 2026, 14:46:33) [Clang 22.1.3 ]","source_sha256":"2ca84e747c2d83f220c1c39a8202c13102c8e3989fe5b267a4e378612874b80e"},"congressional_district_seed":0,"congressional_district_vintage_crosswalk":"/Users/maxghenis/PolicyEngine/_worktrees/us-asec-720-coverage-recodes/packages/microcosm-build/src/microcosm/build/us_runtime/data/congressional_district_vintage_crosswalk.csv","equivalence_deterministic_h5_metadata":false,"geography_ladder_seed":0,"ledger_facts":"/Users/maxghenis/PolicyEngine/_buildh-runtime/inputs/consumer_facts_us_c5e5bf8.jsonl","n_estimators":32,"out":"/Users/maxghenis/PolicyEngine/_recovered/scratch-backup/893/overnight-20260923/asec-720-fix/base-out","puf_h5":"/Users/maxghenis/PolicyEngine/policyengine-us-data/policyengine_us_data/storage/puf_2024.h5","puf_h5_sha256":"7669f5b5281f20080e77204f9bd4aabfad0aa101fa283e22caf9ba8d61d4d6df","puf_source_year_csv":"/Users/maxghenis/PolicyEngine/policyengine-us-data/policyengine_us_data/storage/puf_2015.csv","puf_source_year_csv_sha256":"0a7fd643edb1acc55c507db795914b41d232922be78c149b58d111f4672499df","seed":0,"support_spine_spec":null,"target_year":2024,"thread_environment":{"BLIS_NUM_THREADS":null,"MKL_NUM_THREADS":null,"NUMEXPR_NUM_THREADS":null,"OMP_NUM_THREADS":null,"OPENBLAS_NUM_THREADS":null,"POPULACE_FIT_N_JOBS":null,"POPULACE_FIT_PREDICT_WORKERS":null,"PYTHONHASHSEED":"0","VECLIB_MAXIMUM_THREADS":null},"without_block_ladder":false},"schema_version":2,"stage_records":[{"checkpoint_filename":"000_source_construction.frame.h5","checkpoint_sha256":"be8c268930ff90a46af5225cc95244cc3db9f66ff37b26211d7515b75d2338cf","checkpoint_stage":"source_construction","identity":{"columns":[{"columns":["person_id","person_household_id","person_tax_unit_id","person_spm_unit_id","person_family_id","person_marital_unit_id","source_year","source_household_id","source_person_id","source_row_id"],"table":"person"},{"columns":["household_id"],"table":"household"},{"columns":["tax_unit_id"],"table":"tax_unit"},{"columns":["spm_unit_id"],"table":"spm_unit"},{"columns":["family_id"],"table":"family"},{"columns":["marital_unit_id"],"table":"marital_unit"}],"row_counts":[{"rows":432523,"table":"person"},{"rows":168852,"table":"household"},{"rows":226102,"table":"tax_unit"},{"rows":176039,"table":"spm_unit"},{"rows":189401,"table":"family"},{"rows":345873,"table":"marital_unit"}],"sha256":"e043124aaecceae3484d6323d5bcc1341b7ce726d6c376fa412323478eddd068"},"metadata":{"base_household_weight_total":134690323.34024483,"base_rows":{"family":189401,"household":168852,"marital_unit":345873,"person":432523,"spm_unit":176039,"tax_unit":226102},"base_source":{"kind":"pooled_asec","metadata":{"anchor_year":2024,"sources":[{"census_person_columns":{"archive_sha256":"318845a2b5e0034eb2973898de1738f4df0025727de38499e7669cb9c0deef0b","columns_added":[],"columns_verified_equal":["NOW_MCAID","NOW_NONM","NOW_CHAMPVA","NOW_MIL","NOW_VACARE","NOW_OTHMT","NOW_IHSFLG","A_EXPRRP","PTOTVAL","A_ENRLW","A_FTPT"],"h5_person_rows":142125,"identity_columns_verified_equal":["PH_SEQ","P_SEQ","A_LINENO","A_AGE"],"income_year":2024,"join_key":"PERIDNUM (exact 22-digit string), one-to-one and total","joined_person_rows":142125,"member":"pppub25.csv","member_person_rows":142125,"member_sha256":"06921fe83fc66c907e6c7b86b82255dc70458ee7d76258fc48297cb34f0c06b5","member_size_bytes":277882549,"member_values":{"A_ENRLW":{"0":72079,"1":11445,"2":58601},"A_EXPRRP":{"1":37382,"10":2659,"11":134,"12":1734,"13":4257,"14":844,"2":18380,"3":11334,"4":15918,"5":42053,"7":2570,"8":2876,"9":1984},"A_FTPT":{"0":130680,"1":10014,"2":1431},"NOW_CHAMPVA":{"1":368,"2":141757},"NOW_IHSFLG":{"1":685,"2":141440},"NOW_MCAID":{"1":24844,"2":117281},"NOW_MIL":{"1":4413,"2":137712},"NOW_NONM":{"1":8486,"2":133639},"NOW_OTHMT":{"1":312,"2":141813},"NOW_VACARE":{"1":1270,"2":140855},"PTOTVAL":{"negative_rows":75,"nonzero_rows":101163}},"official_archive_url":"https://www2.census.gov/programs-surveys/cps/datasets/2025/march/asecpub25csv.zip","operation":"exact_source_join","reviewed_columns":["NOW_MCAID","NOW_NONM","NOW_CHAMPVA","NOW_MIL","NOW_VACARE","NOW_OTHMT","NOW_IHSFLG","A_EXPRRP","PTOTVAL","A_ENRLW","A_FTPT"],"source_form":"archive","source_path":"/Users/maxghenis/PolicyEngine/_buildm-runtime/inputs/asec_education/asecpub25csv.zip","survey_year":2025},"raw_household_rows":55762,"raw_person_population":325952369.06,"raw_person_rows":142125,"relationship_recode_source":"source:A_EXPRRP","scale":0.3333333333333333,"share":0.3333333333333333,"source_file":"census_cps_2024.h5","weighted_person_population":108650789.68666667,"year":2024},{"census_person_columns":{"archive_sha256":"cdb39cdac34bef99dd0940ab28e306f692404c2eea44d85dfd634214872a0a09","columns_added":["NOW_MCAID","NOW_NONM","NOW_CHAMPVA","NOW_MIL","NOW_VACARE","NOW_OTHMT","NOW_IHSFLG","A_EXPRRP","PTOTVAL","A_ENRLW","A_FTPT"],"columns_verified_equal":[],"h5_person_rows":144265,"identity_columns_verified_equal":["PH_SEQ","P_SEQ","A_LINENO","A_AGE"],"income_year":2023,"join_key":"PERIDNUM (exact 22-digit string), one-to-one and total","joined_person_rows":144265,"member":"pppub24.csv","member_person_rows":144265,"member_sha256":"21a2b9e0e4b08534563578a45acad77868af4ae9a7d46f23776b707d4a559aa7","member_size_bytes":277250415,"member_values":{"A_ENRLW":{"0":72991,"1":11620,"2":59654},"A_EXPRRP":{"1":37875,"10":2763,"11":147,"12":1752,"13":4387,"14":968,"2":18376,"3":11652,"4":16147,"5":42756,"7":2661,"8":2854,"9":1927},"A_FTPT":{"0":132645,"1":10174,"2":1446},"NOW_CHAMPVA":{"1":335,"2":143930},"NOW_IHSFLG":{"1":651,"2":143614},"NOW_MCAID":{"1":26367,"2":117898},"NOW_MIL":{"1":4318,"2":139947},"NOW_NONM":{"1":8435,"2":135830},"NOW_OTHMT":{"1":331,"2":143934},"NOW_VACARE":{"1":1178,"2":143087},"PTOTVAL":{"negative_rows":59,"nonzero_rows":102389}},"official_archive_url":"https://www2.census.gov/programs-surveys/cps/datasets/2024/march/asecpub24csv.zip","operation":"exact_source_join","reviewed_columns":["NOW_MCAID","NOW_NONM","NOW_CHAMPVA","NOW_MIL","NOW_VACARE","NOW_OTHMT","NOW_IHSFLG","A_EXPRRP","PTOTVAL","A_ENRLW","A_FTPT"],"source_form":"archive","source_path":"/Users/maxghenis/PolicyEngine/_buildm-runtime/inputs/asec_education/asecpub24csv.zip","survey_year":2024},"raw_household_rows":56251,"raw_person_population":320890854.43000007,"raw_person_rows":144265,"relationship_recode_source":"source:A_EXPRRP","scale":0.3385911071839787,"share":0.3333333333333333,"source_file":"census_cps_2023.h5","weighted_person_population":108650789.68666664,"year":2023},{"census_person_columns":{"archive_sha256":"d2e000250782adfbdd7f29c82b66d866591a30f0d330496698ec19f9c784ce11","columns_added":["NOW_MCAID","NOW_NONM","NOW_CHAMPVA","NOW_MIL","NOW_VACARE","NOW_OTHMT","NOW_IHSFLG","A_EXPRRP","PTOTVAL","A_ENRLW","A_FTPT"],"columns_verified_equal":[],"h5_person_rows":146133,"identity_columns_verified_equal":["PH_SEQ","P_SEQ","A_LINENO","A_AGE"],"income_year":2022,"join_key":"PERIDNUM (exact 22-digit string), one-to-one and total","joined_person_rows":146133,"member":"pppub23.csv","member_person_rows":146133,"member_sha256":"19b56537e50e7663f954361ef2bb5ce9cef8d9d45f156fe1a69a99b654198ffe","member_size_bytes":281065733,"member_values":{"A_ENRLW":{"0":74477,"1":11613,"2":60043},"A_EXPRRP":{"1":38440,"10":2799,"11":132,"12":1735,"13":4312,"14":919,"2":18399,"3":11887,"4":16284,"5":43805,"7":2792,"8":2765,"9":1864},"A_FTPT":{"0":134520,"1":10164,"2":1449},"NOW_CHAMPVA":{"1":260,"2":145873},"NOW_IHSFLG":{"1":687,"2":145446},"NOW_MCAID":{"1":28184,"2":117949},"NOW_MIL":{"1":3882,"2":142251},"NOW_NONM":{"1":8666,"2":137467},"NOW_OTHMT":{"1":275,"2":145858},"NOW_VACARE":{"1":1271,"2":144862},"PTOTVAL":{"negative_rows":66,"nonzero_rows":102532}},"official_archive_url":"https://www2.census.gov/programs-surveys/cps/datasets/2023/march/asecpub23csv.zip","operation":"exact_source_join","reviewed_columns":["NOW_MCAID","NOW_NONM","NOW_CHAMPVA","NOW_MIL","NOW_VACARE","NOW_OTHMT","NOW_IHSFLG","A_EXPRRP","PTOTVAL","A_ENRLW","A_FTPT"],"source_form":"archive","source_path":"/Users/maxghenis/PolicyEngine/_buildm-runtime/inputs/asec_education/asecpub23csv.zip","survey_year":2023},"raw_household_rows":56839,"raw_person_population":318771270.71000004,"raw_person_rows":146133,"relationship_recode_source":"source:A_EXPRRP","scale":0.34084247756916264,"share":0.3333333333333333,"source_file":"census_cps_2022.h5","weighted_person_population":108650789.68666667,"year":2022}],"target_person_population":325952369.06,"weighted_person_population":325952369.05999994},"sources":[{"max_households":null,"path":"/Users/maxghenis/PolicyEngine/policyengine-us-data/policyengine_us_data/storage/census_cps_2024.h5","sha256":"ec36604cb735a660b51b0b2f90be27d803b5878f3464fb30d0eacead59c1260d","share":null,"year":2024},{"max_households":null,"path":"/Users/maxghenis/PolicyEngine/policyengine-us-data/policyengine_us_data/storage/census_cps_2023.h5","sha256":"cb57817327799f42b741caed5f9be94d04021c2e6809c1ad7bd0686da5428d88","share":null,"year":2023},{"max_households":null,"path":"/Users/maxghenis/PolicyEngine/policyengine-us-data/policyengine_us_data/storage/census_cps_2022.h5","sha256":"7ccca976284bb47815d84460cc4f75a0a65d26d7754ab0a0f417de351b3d474e","share":null,"year":2022}],"support_spine_spec":null,"target_year":2024},"weeks_unemployed_source_path":"/Users/maxghenis/.cache/microcosm/cps/asec_2023/pppub23.csv"},"stage":"source_construction","wrote_frame":true}]} diff --git a/experiments/us-asec-720-census-person-columns/verify_720.json b/experiments/us-asec-720-census-person-columns/verify_720.json new file mode 100644 index 0000000000..f4287463ab --- /dev/null +++ b/experiments/us-asec-720-census-person-columns/verify_720.json @@ -0,0 +1,886 @@ +{ + "after": { + "census_person_columns": { + "2022": { + "archive_sha256": "d2e000250782adfbdd7f29c82b66d866591a30f0d330496698ec19f9c784ce11", + "columns_added": [ + "NOW_MCAID", + "NOW_NONM", + "NOW_CHAMPVA", + "NOW_MIL", + "NOW_VACARE", + "NOW_OTHMT", + "NOW_IHSFLG", + "A_EXPRRP", + "PTOTVAL", + "A_ENRLW", + "A_FTPT" + ], + "columns_verified_equal": [], + "h5_person_rows": 146133, + "identity_columns_verified_equal": [ + "PH_SEQ", + "P_SEQ", + "A_LINENO", + "A_AGE" + ], + "income_year": 2022, + "join_key": "PERIDNUM (exact 22-digit string), one-to-one and total", + "joined_person_rows": 146133, + "member": "pppub23.csv", + "member_person_rows": 146133, + "member_sha256": "19b56537e50e7663f954361ef2bb5ce9cef8d9d45f156fe1a69a99b654198ffe", + "member_size_bytes": 281065733, + "member_values": { + "A_ENRLW": { + "0": 74477, + "1": 11613, + "2": 60043 + }, + "A_EXPRRP": { + "1": 38440, + "10": 2799, + "11": 132, + "12": 1735, + "13": 4312, + "14": 919, + "2": 18399, + "3": 11887, + "4": 16284, + "5": 43805, + "7": 2792, + "8": 2765, + "9": 1864 + }, + "A_FTPT": { + "0": 134520, + "1": 10164, + "2": 1449 + }, + "NOW_CHAMPVA": { + "1": 260, + "2": 145873 + }, + "NOW_IHSFLG": { + "1": 687, + "2": 145446 + }, + "NOW_MCAID": { + "1": 28184, + "2": 117949 + }, + "NOW_MIL": { + "1": 3882, + "2": 142251 + }, + "NOW_NONM": { + "1": 8666, + "2": 137467 + }, + "NOW_OTHMT": { + "1": 275, + "2": 145858 + }, + "NOW_VACARE": { + "1": 1271, + "2": 144862 + }, + "PTOTVAL": { + "negative_rows": 66, + "nonzero_rows": 102532 + } + }, + "official_archive_url": "https://www2.census.gov/programs-surveys/cps/datasets/2023/march/asecpub23csv.zip", + "operation": "exact_source_join", + "reviewed_columns": [ + "NOW_MCAID", + "NOW_NONM", + "NOW_CHAMPVA", + "NOW_MIL", + "NOW_VACARE", + "NOW_OTHMT", + "NOW_IHSFLG", + "A_EXPRRP", + "PTOTVAL", + "A_ENRLW", + "A_FTPT" + ], + "source_form": "archive", + "source_path": "/Users/maxghenis/PolicyEngine/_buildm-runtime/inputs/asec_education/asecpub23csv.zip", + "survey_year": 2023 + }, + "2023": { + "archive_sha256": "cdb39cdac34bef99dd0940ab28e306f692404c2eea44d85dfd634214872a0a09", + "columns_added": [ + "NOW_MCAID", + "NOW_NONM", + "NOW_CHAMPVA", + "NOW_MIL", + "NOW_VACARE", + "NOW_OTHMT", + "NOW_IHSFLG", + "A_EXPRRP", + "PTOTVAL", + "A_ENRLW", + "A_FTPT" + ], + "columns_verified_equal": [], + "h5_person_rows": 144265, + "identity_columns_verified_equal": [ + "PH_SEQ", + "P_SEQ", + "A_LINENO", + "A_AGE" + ], + "income_year": 2023, + "join_key": "PERIDNUM (exact 22-digit string), one-to-one and total", + "joined_person_rows": 144265, + "member": "pppub24.csv", + "member_person_rows": 144265, + "member_sha256": "21a2b9e0e4b08534563578a45acad77868af4ae9a7d46f23776b707d4a559aa7", + "member_size_bytes": 277250415, + "member_values": { + "A_ENRLW": { + "0": 72991, + "1": 11620, + "2": 59654 + }, + "A_EXPRRP": { + "1": 37875, + "10": 2763, + "11": 147, + "12": 1752, + "13": 4387, + "14": 968, + "2": 18376, + "3": 11652, + "4": 16147, + "5": 42756, + "7": 2661, + "8": 2854, + "9": 1927 + }, + "A_FTPT": { + "0": 132645, + "1": 10174, + "2": 1446 + }, + "NOW_CHAMPVA": { + "1": 335, + "2": 143930 + }, + "NOW_IHSFLG": { + "1": 651, + "2": 143614 + }, + "NOW_MCAID": { + "1": 26367, + "2": 117898 + }, + "NOW_MIL": { + "1": 4318, + "2": 139947 + }, + "NOW_NONM": { + "1": 8435, + "2": 135830 + }, + "NOW_OTHMT": { + "1": 331, + "2": 143934 + }, + "NOW_VACARE": { + "1": 1178, + "2": 143087 + }, + "PTOTVAL": { + "negative_rows": 59, + "nonzero_rows": 102389 + } + }, + "official_archive_url": "https://www2.census.gov/programs-surveys/cps/datasets/2024/march/asecpub24csv.zip", + "operation": "exact_source_join", + "reviewed_columns": [ + "NOW_MCAID", + "NOW_NONM", + "NOW_CHAMPVA", + "NOW_MIL", + "NOW_VACARE", + "NOW_OTHMT", + "NOW_IHSFLG", + "A_EXPRRP", + "PTOTVAL", + "A_ENRLW", + "A_FTPT" + ], + "source_form": "archive", + "source_path": "/Users/maxghenis/PolicyEngine/_buildm-runtime/inputs/asec_education/asecpub24csv.zip", + "survey_year": 2024 + }, + "2024": { + "archive_sha256": "318845a2b5e0034eb2973898de1738f4df0025727de38499e7669cb9c0deef0b", + "columns_added": [], + "columns_verified_equal": [ + "NOW_MCAID", + "NOW_NONM", + "NOW_CHAMPVA", + "NOW_MIL", + "NOW_VACARE", + "NOW_OTHMT", + "NOW_IHSFLG", + "A_EXPRRP", + "PTOTVAL", + "A_ENRLW", + "A_FTPT" + ], + "h5_person_rows": 142125, + "identity_columns_verified_equal": [ + "PH_SEQ", + "P_SEQ", + "A_LINENO", + "A_AGE" + ], + "income_year": 2024, + "join_key": "PERIDNUM (exact 22-digit string), one-to-one and total", + "joined_person_rows": 142125, + "member": "pppub25.csv", + "member_person_rows": 142125, + "member_sha256": "06921fe83fc66c907e6c7b86b82255dc70458ee7d76258fc48297cb34f0c06b5", + "member_size_bytes": 277882549, + "member_values": { + "A_ENRLW": { + "0": 72079, + "1": 11445, + "2": 58601 + }, + "A_EXPRRP": { + "1": 37382, + "10": 2659, + "11": 134, + "12": 1734, + "13": 4257, + "14": 844, + "2": 18380, + "3": 11334, + "4": 15918, + "5": 42053, + "7": 2570, + "8": 2876, + "9": 1984 + }, + "A_FTPT": { + "0": 130680, + "1": 10014, + "2": 1431 + }, + "NOW_CHAMPVA": { + "1": 368, + "2": 141757 + }, + "NOW_IHSFLG": { + "1": 685, + "2": 141440 + }, + "NOW_MCAID": { + "1": 24844, + "2": 117281 + }, + "NOW_MIL": { + "1": 4413, + "2": 137712 + }, + "NOW_NONM": { + "1": 8486, + "2": 133639 + }, + "NOW_OTHMT": { + "1": 312, + "2": 141813 + }, + "NOW_VACARE": { + "1": 1270, + "2": 140855 + }, + "PTOTVAL": { + "negative_rows": 75, + "nonzero_rows": 101163 + } + }, + "official_archive_url": "https://www2.census.gov/programs-surveys/cps/datasets/2025/march/asecpub25csv.zip", + "operation": "exact_source_join", + "reviewed_columns": [ + "NOW_MCAID", + "NOW_NONM", + "NOW_CHAMPVA", + "NOW_MIL", + "NOW_VACARE", + "NOW_OTHMT", + "NOW_IHSFLG", + "A_EXPRRP", + "PTOTVAL", + "A_ENRLW", + "A_FTPT" + ], + "source_form": "archive", + "source_path": "/Users/maxghenis/PolicyEngine/_buildm-runtime/inputs/asec_education/asecpub25csv.zip", + "survey_year": 2025 + } + }, + "checkpoint": "/Users/maxghenis/PolicyEngine/_recovered/scratch-backup/893/overnight-20260923/asec-720-fix/checkpoints/asec_raw_stage.checkpoint.h5", + "checkpoint_sha256": "0ae0b7558094ee1175fc03494e07ef79554e40b253a9373ed3f62c1224435395", + "gate": { + "failure_count": 0, + "failures": [], + "passed": true, + "vintages": { + "2022": { + "enforced": true, + "null_counts": { + "has_champva_health_coverage_at_interview": 0, + "has_esi": 0, + "has_indian_health_service_coverage_at_interview": 0, + "has_marketplace_health_coverage_at_interview": 0, + "has_medicaid_health_coverage_at_interview": 0, + "has_non_marketplace_direct_purchase_health_coverage_at_interview": 0, + "has_other_means_tested_health_coverage_at_interview": 0, + "has_tricare_health_coverage_at_interview": 0, + "has_va_health_coverage_at_interview": 0 + }, + "reporter_counts": { + "has_champva_health_coverage_at_interview": 260, + "has_esi": 78773, + "has_indian_health_service_coverage_at_interview": 687, + "has_marketplace_health_coverage_at_interview": 4666, + "has_medicaid_health_coverage_at_interview": 28184, + "has_non_marketplace_direct_purchase_health_coverage_at_interview": 8666, + "has_other_means_tested_health_coverage_at_interview": 275, + "has_tricare_health_coverage_at_interview": 3882, + "has_va_health_coverage_at_interview": 1271 + }, + "rows": 146133 + }, + "2023": { + "enforced": true, + "null_counts": { + "has_champva_health_coverage_at_interview": 0, + "has_esi": 0, + "has_indian_health_service_coverage_at_interview": 0, + "has_marketplace_health_coverage_at_interview": 0, + "has_medicaid_health_coverage_at_interview": 0, + "has_non_marketplace_direct_purchase_health_coverage_at_interview": 0, + "has_other_means_tested_health_coverage_at_interview": 0, + "has_tricare_health_coverage_at_interview": 0, + "has_va_health_coverage_at_interview": 0 + }, + "reporter_counts": { + "has_champva_health_coverage_at_interview": 335, + "has_esi": 76710, + "has_indian_health_service_coverage_at_interview": 651, + "has_marketplace_health_coverage_at_interview": 5530, + "has_medicaid_health_coverage_at_interview": 26367, + "has_non_marketplace_direct_purchase_health_coverage_at_interview": 8435, + "has_other_means_tested_health_coverage_at_interview": 331, + "has_tricare_health_coverage_at_interview": 4318, + "has_va_health_coverage_at_interview": 1178 + }, + "rows": 144265 + }, + "2024": { + "enforced": true, + "null_counts": { + "has_champva_health_coverage_at_interview": 0, + "has_esi": 0, + "has_indian_health_service_coverage_at_interview": 0, + "has_marketplace_health_coverage_at_interview": 0, + "has_medicaid_health_coverage_at_interview": 0, + "has_non_marketplace_direct_purchase_health_coverage_at_interview": 0, + "has_other_means_tested_health_coverage_at_interview": 0, + "has_tricare_health_coverage_at_interview": 0, + "has_va_health_coverage_at_interview": 0 + }, + "reporter_counts": { + "has_champva_health_coverage_at_interview": 368, + "has_esi": 75388, + "has_indian_health_service_coverage_at_interview": 685, + "has_marketplace_health_coverage_at_interview": 5828, + "has_medicaid_health_coverage_at_interview": 24844, + "has_non_marketplace_direct_purchase_health_coverage_at_interview": 8486, + "has_other_means_tested_health_coverage_at_interview": 312, + "has_tricare_health_coverage_at_interview": 4413, + "has_va_health_coverage_at_interview": 1270 + }, + "rows": 142125 + } + } + }, + "raw_source_mappings": [ + "ED_VAL", + "LKWEEKS", + "PAW_TYP" + ], + "relationship_recode_source": { + "2022": "source:A_EXPRRP", + "2023": "source:A_EXPRRP", + "2024": "source:A_EXPRRP" + }, + "rows": { + "family": 189401, + "household": 168852, + "marital_unit": 345873, + "person": 432523, + "spm_unit": 176039, + "tax_unit": 226102 + } + }, + "before": { + "census_person_columns": { + "2022": null, + "2023": null, + "2024": null + }, + "checkpoint": "/Users/maxghenis/PolicyEngine/_recovered/scratch-backup/893/overnight-20260923/route-a/run-47976be6ce75/base-checkpoints/asec_raw_stage.checkpoint.h5", + "checkpoint_sha256": "0e828cca816940b7e227ca806a7db3749f656fe7616f8a809d4eb107b78bedb8", + "gate": { + "failure_count": 14, + "failures": [ + "has_champva_health_coverage_at_interview: vintage 2022 has 0 reporters over 146133 person rows (consistent with a source input lacking the at-interview recode, microcosm #720).", + "has_indian_health_service_coverage_at_interview: vintage 2022 has 0 reporters over 146133 person rows (consistent with a source input lacking the at-interview recode, microcosm #720).", + "has_medicaid_health_coverage_at_interview: vintage 2022 has 0 reporters over 146133 person rows (consistent with a source input lacking the at-interview recode, microcosm #720).", + "has_non_marketplace_direct_purchase_health_coverage_at_interview: vintage 2022 has 0 reporters over 146133 person rows (consistent with a source input lacking the at-interview recode, microcosm #720).", + "has_other_means_tested_health_coverage_at_interview: vintage 2022 has 0 reporters over 146133 person rows (consistent with a source input lacking the at-interview recode, microcosm #720).", + "has_tricare_health_coverage_at_interview: vintage 2022 has 0 reporters over 146133 person rows (consistent with a source input lacking the at-interview recode, microcosm #720).", + "has_va_health_coverage_at_interview: vintage 2022 has 0 reporters over 146133 person rows (consistent with a source input lacking the at-interview recode, microcosm #720).", + "has_champva_health_coverage_at_interview: vintage 2023 has 0 reporters over 144265 person rows (consistent with a source input lacking the at-interview recode, microcosm #720).", + "has_indian_health_service_coverage_at_interview: vintage 2023 has 0 reporters over 144265 person rows (consistent with a source input lacking the at-interview recode, microcosm #720).", + "has_medicaid_health_coverage_at_interview: vintage 2023 has 0 reporters over 144265 person rows (consistent with a source input lacking the at-interview recode, microcosm #720).", + "has_non_marketplace_direct_purchase_health_coverage_at_interview: vintage 2023 has 0 reporters over 144265 person rows (consistent with a source input lacking the at-interview recode, microcosm #720).", + "has_other_means_tested_health_coverage_at_interview: vintage 2023 has 0 reporters over 144265 person rows (consistent with a source input lacking the at-interview recode, microcosm #720).", + "has_tricare_health_coverage_at_interview: vintage 2023 has 0 reporters over 144265 person rows (consistent with a source input lacking the at-interview recode, microcosm #720).", + "has_va_health_coverage_at_interview: vintage 2023 has 0 reporters over 144265 person rows (consistent with a source input lacking the at-interview recode, microcosm #720)." + ], + "passed": false, + "vintages": { + "2022": { + "enforced": true, + "null_counts": { + "has_champva_health_coverage_at_interview": 0, + "has_esi": 0, + "has_indian_health_service_coverage_at_interview": 0, + "has_marketplace_health_coverage_at_interview": 0, + "has_medicaid_health_coverage_at_interview": 0, + "has_non_marketplace_direct_purchase_health_coverage_at_interview": 0, + "has_other_means_tested_health_coverage_at_interview": 0, + "has_tricare_health_coverage_at_interview": 0, + "has_va_health_coverage_at_interview": 0 + }, + "reporter_counts": { + "has_champva_health_coverage_at_interview": 0, + "has_esi": 78773, + "has_indian_health_service_coverage_at_interview": 0, + "has_marketplace_health_coverage_at_interview": 4666, + "has_medicaid_health_coverage_at_interview": 0, + "has_non_marketplace_direct_purchase_health_coverage_at_interview": 0, + "has_other_means_tested_health_coverage_at_interview": 0, + "has_tricare_health_coverage_at_interview": 0, + "has_va_health_coverage_at_interview": 0 + }, + "rows": 146133 + }, + "2023": { + "enforced": true, + "null_counts": { + "has_champva_health_coverage_at_interview": 0, + "has_esi": 0, + "has_indian_health_service_coverage_at_interview": 0, + "has_marketplace_health_coverage_at_interview": 0, + "has_medicaid_health_coverage_at_interview": 0, + "has_non_marketplace_direct_purchase_health_coverage_at_interview": 0, + "has_other_means_tested_health_coverage_at_interview": 0, + "has_tricare_health_coverage_at_interview": 0, + "has_va_health_coverage_at_interview": 0 + }, + "reporter_counts": { + "has_champva_health_coverage_at_interview": 0, + "has_esi": 76710, + "has_indian_health_service_coverage_at_interview": 0, + "has_marketplace_health_coverage_at_interview": 5530, + "has_medicaid_health_coverage_at_interview": 0, + "has_non_marketplace_direct_purchase_health_coverage_at_interview": 0, + "has_other_means_tested_health_coverage_at_interview": 0, + "has_tricare_health_coverage_at_interview": 0, + "has_va_health_coverage_at_interview": 0 + }, + "rows": 144265 + }, + "2024": { + "enforced": true, + "null_counts": { + "has_champva_health_coverage_at_interview": 0, + "has_esi": 0, + "has_indian_health_service_coverage_at_interview": 0, + "has_marketplace_health_coverage_at_interview": 0, + "has_medicaid_health_coverage_at_interview": 0, + "has_non_marketplace_direct_purchase_health_coverage_at_interview": 0, + "has_other_means_tested_health_coverage_at_interview": 0, + "has_tricare_health_coverage_at_interview": 0, + "has_va_health_coverage_at_interview": 0 + }, + "reporter_counts": { + "has_champva_health_coverage_at_interview": 368, + "has_esi": 75388, + "has_indian_health_service_coverage_at_interview": 685, + "has_marketplace_health_coverage_at_interview": 5828, + "has_medicaid_health_coverage_at_interview": 24844, + "has_non_marketplace_direct_purchase_health_coverage_at_interview": 8486, + "has_other_means_tested_health_coverage_at_interview": 312, + "has_tricare_health_coverage_at_interview": 4413, + "has_va_health_coverage_at_interview": 1270 + }, + "rows": 142125 + } + } + }, + "raw_source_mappings": [ + "ED_VAL", + "LKWEEKS", + "PAW_TYP" + ], + "relationship_recode_source": { + "2022": "derived:line_spouse_parent", + "2023": "derived:line_spouse_parent", + "2024": "source:A_EXPRRP" + }, + "rows": { + "family": 189401, + "household": 168852, + "marital_unit": 345873, + "person": 432523, + "spm_unit": 176039, + "tax_unit": 231007 + } + }, + "corrected_h5_comparison": { + "2022": { + "NOW_MCAID_yes_rows": 28184, + "added_columns_vs_corrected": { + "A_ENRLW": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "A_EXPRRP": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "A_FTPT": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "NOW_CHAMPVA": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "NOW_IHSFLG": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "NOW_MCAID": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "NOW_MIL": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "NOW_NONM": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "NOW_OTHMT": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "NOW_VACARE": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "PTOTVAL": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + } + }, + "all_added_columns_identical": true, + "columns_added": [ + "NOW_MCAID", + "NOW_NONM", + "NOW_CHAMPVA", + "NOW_MIL", + "NOW_VACARE", + "NOW_OTHMT", + "NOW_IHSFLG", + "A_EXPRRP", + "PTOTVAL", + "A_ENRLW", + "A_FTPT" + ], + "columns_verified_equal": [], + "corrected_h5": "/Users/maxghenis/PolicyEngine/_buildo-runtime/inputs/asec-720/census_cps_2022.h5", + "corrected_h5_sha256": "797fb7e0ea3e94a52dc6c9c7f5a01d70d28a6c5620f3c112bcbd7c6cf6cffe9f", + "corrected_only_columns": [ + "A_FAMREL", + "A_FAMTYP", + "LKWEEKS", + "NOW_CAID", + "NOW_COV", + "NOW_DIR", + "NOW_MCARE", + "NOW_MRKS", + "NOW_MRKUN", + "NOW_PCHIP", + "NOW_PRIV", + "NOW_PUB", + "PECOHAB" + ], + "corrected_only_columns_all_in_not_restored": true, + "existing_columns_identical_to_corrected": true, + "h5_sha256": "7ccca976284bb47815d84460cc4f75a0a65d26d7754ab0a0f417de351b3d474e", + "joined_person_rows": 146133, + "source_form": "archive" + }, + "2023": { + "NOW_MCAID_yes_rows": 26367, + "added_columns_vs_corrected": { + "A_ENRLW": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "A_EXPRRP": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "A_FTPT": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "NOW_CHAMPVA": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "NOW_IHSFLG": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "NOW_MCAID": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "NOW_MIL": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "NOW_NONM": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "NOW_OTHMT": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "NOW_VACARE": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + }, + "PTOTVAL": { + "dtype": [ + "int64", + "int64" + ], + "identical": true + } + }, + "all_added_columns_identical": true, + "columns_added": [ + "NOW_MCAID", + "NOW_NONM", + "NOW_CHAMPVA", + "NOW_MIL", + "NOW_VACARE", + "NOW_OTHMT", + "NOW_IHSFLG", + "A_EXPRRP", + "PTOTVAL", + "A_ENRLW", + "A_FTPT" + ], + "columns_verified_equal": [], + "corrected_h5": "/Users/maxghenis/PolicyEngine/_buildo-runtime/inputs/asec-720/census_cps_2023.h5", + "corrected_h5_sha256": "5fc2feca08aa125ac2ed127711457038e0d493a5ad1d60aff763fb8e42a9d60a", + "corrected_only_columns": [ + "A_FAMREL", + "A_FAMTYP", + "NOW_CAID", + "NOW_COV", + "NOW_DIR", + "NOW_MCARE", + "NOW_MRKS", + "NOW_MRKUN", + "NOW_PCHIP", + "NOW_PRIV", + "NOW_PUB", + "PECOHAB" + ], + "corrected_only_columns_all_in_not_restored": true, + "existing_columns_identical_to_corrected": true, + "h5_sha256": "cb57817327799f42b741caed5f9be94d04021c2e6809c1ad7bd0686da5428d88", + "joined_person_rows": 144265, + "source_form": "archive" + }, + "2024": { + "NOW_MCAID_yes_rows": 24844, + "columns_added": [], + "columns_verified_equal": [ + "NOW_MCAID", + "NOW_NONM", + "NOW_CHAMPVA", + "NOW_MIL", + "NOW_VACARE", + "NOW_OTHMT", + "NOW_IHSFLG", + "A_EXPRRP", + "PTOTVAL", + "A_ENRLW", + "A_FTPT" + ], + "h5_sha256": "ec36604cb735a660b51b0b2f90be27d803b5878f3464fb30d0eacead59c1260d", + "joined_person_rows": 142125, + "source_form": "archive" + } + }, + "lkweeks_2022_vs_corrected": { + "equal_rows": 146133, + "rows_joined": 146133 + }, + "peak_rss_gb": 10.541154304, + "structural_deltas": { + "A_ENRLW_null_by_year_after": { + "2022": 0, + "2023": 0, + "2024": 0 + }, + "A_ENRLW_null_by_year_before": { + "2022": 146133, + "2023": 144265, + "2024": 0 + }, + "A_EXPRRP_changed_2022": 20595, + "A_EXPRRP_changed_2023": 20561, + "A_EXPRRP_changed_2024": 0, + "A_EXPRRP_partner_roommate_13_2022": { + "after": 4312, + "before": 0 + }, + "A_EXPRRP_partner_roommate_13_2023": { + "after": 4387, + "before": 0 + }, + "A_EXPRRP_partner_roommate_13_2024": { + "after": 4257, + "before": 4257 + }, + "A_FTPT_null_by_year_after": { + "2022": 0, + "2023": 0, + "2024": 0 + }, + "A_FTPT_null_by_year_before": { + "2022": 146133, + "2023": 144265, + "2024": 0 + }, + "PTOTVAL_null_by_year_after": { + "2022": 0, + "2023": 0, + "2024": 0 + }, + "PTOTVAL_null_by_year_before": { + "2022": 146133, + "2023": 144265, + "2024": 0 + }, + "persons_joined": 432523, + "tax_units_after": 226102, + "tax_units_before": 231007, + "tax_units_by_year_after": { + "2022": 75797, + "2023": 75608, + "2024": 74697 + }, + "tax_units_by_year_before": { + "2022": 78221, + "2023": 78089, + "2024": 74697 + } + } +} \ No newline at end of file diff --git a/experiments/us-asec-720-census-person-columns/verify_720.py b/experiments/us-asec-720-census-person-columns/verify_720.py new file mode 100644 index 0000000000..73156bcc2c --- /dev/null +++ b/experiments/us-asec-720-census-person-columns/verify_720.py @@ -0,0 +1,226 @@ +"""Verify the #720 restoration on real inputs (source construction only). + +1. Gate before/after: us_reported_coverage_vintage_signal_gate on + derive_us_cps_carried_inputs of the ASEC raw-stage frame, for route A's + raw-stage checkpoint (commit 47976be6c, before) and this branch's (after). +2. Column-by-column equality of the restored per-vintage person tables with + the 2026-08-23 corrected H5s (_buildo-runtime/inputs/asec-720). +3. Structural deltas the restored A_EXPRRP/PTOTVAL/A_ENRLW/A_FTPT cause. + +Writes verify_720.json next to this script. +""" + +from __future__ import annotations + +import gc +import hashlib +import json +import resource +import sys +from pathlib import Path + +import pandas as pd + +from microcosm.build.us_runtime import ( + derive_us_cps_carried_inputs, + load_asec_raw_stage_checkpoint, + us_reported_coverage_vintage_signal_gate, +) +from microcosm.build.us_runtime.asec_census_person_columns import ( + ASEC_CENSUS_PERSON_COLUMN_NAMES, + ASEC_CENSUS_PERSON_COLUMNS_NOT_RESTORED, + restore_asec_census_person_columns, +) +from microcosm.build.us_runtime.asec_pool import load_asec_h5_tables + +HERE = Path(__file__).resolve().parent +BEFORE = Path( + "/Users/maxghenis/PolicyEngine/_recovered/scratch-backup/893/overnight-20260923/" + "route-a/run-47976be6ce75/base-checkpoints/asec_raw_stage.checkpoint.h5" +) +AFTER = HERE / "checkpoints" / "asec_raw_stage.checkpoint.h5" +STO = Path( + "/Users/maxghenis/PolicyEngine/policyengine-us-data/policyengine_us_data/storage" +) +ARC = Path("/Users/maxghenis/PolicyEngine/_buildm-runtime/inputs/asec_education") +CORRECTED = Path("/Users/maxghenis/PolicyEngine/_buildo-runtime/inputs/asec-720") +ARCHIVES = { + 2022: "asecpub23csv.zip", + 2023: "asecpub24csv.zip", + 2024: "asecpub25csv.zip", +} + + +def sha256(path: Path) -> str: + with path.open("rb") as stream: + return hashlib.file_digest(stream, "sha256").hexdigest() + + +def peak_rss_gb() -> float: + return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1e9 # bytes on macOS + + +def gate_on(path: Path) -> tuple[dict, pd.DataFrame]: + frame, metadata = load_asec_raw_stage_checkpoint(path) + carried = derive_us_cps_carried_inputs(frame) + gate = us_reported_coverage_vintage_signal_gate(carried) + keep = [ + "source_year", + "PERIDNUM", + "A_EXPRRP", + "PTOTVAL", + "A_ENRLW", + "A_FTPT", + "LKWEEKS", + "person_tax_unit_id", + *[c for c in ASEC_CENSUS_PERSON_COLUMN_NAMES if c.startswith("NOW_")], + ] + person = frame.table("person") + slim = person[[c for c in keep if c in person.columns]].copy() + rows = {entity: int(frame.n(entity)) for entity in frame.entities} + sources = metadata["source_receipt"]["metadata"]["sources"] + result = { + "checkpoint": str(path), + "checkpoint_sha256": sha256(path), + "rows": rows, + "relationship_recode_source": { + str(item["year"]): item["relationship_recode_source"] for item in sources + }, + "census_person_columns": { + str(item["year"]): item.get("census_person_columns") for item in sources + }, + "raw_source_mappings": sorted(metadata["raw_source_mappings"]), + "gate": { + "passed": bool(gate.passed), + "failure_count": len(gate.failures), + "failures": list(gate.failures), + "vintages": gate.details.get("vintages"), + }, + } + del frame, carried, person + gc.collect() + return result, slim + + +def compare_with_corrected() -> dict: + """Restore each pinned H5 exactly as the pool does; compare to 8/23 files.""" + + out = {} + for year in (2022, 2023, 2024): + h5 = STO / f"census_cps_{year}.h5" + person = load_asec_h5_tables(h5)["person"].reset_index(drop=True) + restored, record = restore_asec_census_person_columns( + person, income_year=year, source_path=ARC / ARCHIVES[year] + ) + entry = { + "h5_sha256": sha256(h5), + "columns_added": record["columns_added"], + "columns_verified_equal": record["columns_verified_equal"], + "joined_person_rows": record["joined_person_rows"], + "NOW_MCAID_yes_rows": record["member_values"]["NOW_MCAID"]["1"], + "source_form": record["source_form"], + } + if year in (2022, 2023): + corrected_path = CORRECTED / f"census_cps_{year}.h5" + with pd.HDFStore(corrected_path, mode="r") as store: + corrected = store["person"].reset_index(drop=True) + entry["corrected_h5"] = str(corrected_path) + entry["corrected_h5_sha256"] = sha256(corrected_path) + comparison = {} + for column in ASEC_CENSUS_PERSON_COLUMN_NAMES: + ours, theirs = restored[column], corrected[column] + comparison[column] = { + "identical": bool( + ours.dtype == theirs.dtype + and len(ours) == len(theirs) + and ours.equals(theirs) + ), + "dtype": [str(ours.dtype), str(theirs.dtype)], + } + entry["added_columns_vs_corrected"] = comparison + entry["all_added_columns_identical"] = all( + item["identical"] for item in comparison.values() + ) + # Every pre-existing column: our table leaves them exactly as the + # pinned H5 has them, as the corrected file did. + entry["existing_columns_identical_to_corrected"] = all( + restored[column].equals(corrected[column]) + and restored[column].dtype == corrected[column].dtype + for column in person.columns + ) + entry["corrected_only_columns"] = sorted( + set(corrected.columns) - set(restored.columns) + ) + entry["corrected_only_columns_all_in_not_restored"] = set( + entry["corrected_only_columns"] + ) <= set(ASEC_CENSUS_PERSON_COLUMNS_NOT_RESTORED) + del corrected + out[str(year)] = entry + del person, restored + gc.collect() + return out + + +def lkweeks_2022_vs_corrected(after: pd.DataFrame) -> dict: + """The dropped LKWEEKS is still restored for 2022 by the existing sidecar.""" + + with pd.HDFStore(CORRECTED / "census_cps_2022.h5", mode="r") as store: + corrected = store["person"][["PERIDNUM", "LKWEEKS"]] + ours = after.loc[after["source_year"] == 2022, ["PERIDNUM", "LKWEEKS"]] + merged = ours.merge(corrected, on="PERIDNUM", suffixes=("_ours", "_corrected")) + return { + "rows_joined": int(len(merged)), + "equal_rows": int( + (merged["LKWEEKS_ours"].astype(float) == merged["LKWEEKS_corrected"]).sum() + ), + } + + +def structural_deltas(before: pd.DataFrame, after: pd.DataFrame) -> dict: + key = ["source_year", "PERIDNUM"] + merged = before[[*key, "A_EXPRRP"]].merge( + after[[*key, "A_EXPRRP"]], on=key, suffixes=("_before", "_after") + ) + out = {"persons_joined": int(len(merged))} + for year in (2022, 2023, 2024): + rows = merged[merged["source_year"] == year] + differ = rows["A_EXPRRP_before"] != rows["A_EXPRRP_after"] + out[f"A_EXPRRP_changed_{year}"] = int(differ.sum()) + out[f"A_EXPRRP_partner_roommate_13_{year}"] = { + "before": int((rows["A_EXPRRP_before"] == 13).sum()), + "after": int((rows["A_EXPRRP_after"] == 13).sum()), + } + for name, frame in (("before", before), ("after", after)): + out[f"tax_units_{name}"] = int(frame["person_tax_unit_id"].nunique()) + out[f"tax_units_by_year_{name}"] = { + str(year): int(group["person_tax_unit_id"].nunique()) + for year, group in frame.groupby("source_year") + } + for column in ("PTOTVAL", "A_ENRLW", "A_FTPT"): + if column in frame: + out[f"{column}_null_by_year_{name}"] = { + str(year): int(group[column].isna().sum()) + for year, group in frame.groupby("source_year") + } + return out + + +def main() -> int: + report: dict[str, object] = {} + report["before"], before = gate_on(BEFORE) + print("before gate passed:", report["before"]["gate"]["passed"], flush=True) + report["after"], after = gate_on(AFTER) + print("after gate passed:", report["after"]["gate"]["passed"], flush=True) + report["structural_deltas"] = structural_deltas(before, after) + report["lkweeks_2022_vs_corrected"] = lkweeks_2022_vs_corrected(after) + del before, after + gc.collect() + report["corrected_h5_comparison"] = compare_with_corrected() + report["peak_rss_gb"] = peak_rss_gb() + (HERE / "verify_720.json").write_text(json.dumps(report, indent=1, sort_keys=True)) + print(json.dumps(report, indent=1, sort_keys=True)[:200]) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/asec_census_person_columns.py b/packages/microcosm-build/src/microcosm/build/us_runtime/asec_census_person_columns.py new file mode 100644 index 0000000000..b0c5650405 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/asec_census_person_columns.py @@ -0,0 +1,506 @@ +"""Restore the reviewed Census person columns the pinned ASEC H5 inputs lack. + +Microcosm #720. The base pools three processed ASEC HDF5 inputs (pins in +:mod:`.asec_sources`: ``census_cps_2022.h5`` 7ccca976…, ``census_cps_2023.h5`` +cb578173…, ``census_cps_2024.h5`` ec36604c…). The income-year 2022 and 2023 +files were extracted with an older column list: they carry 2 of the 18 +``NOW_*`` at-interview coverage recodes (``NOW_GRP``, ``NOW_MRK``) and lack +``A_EXPRRP``, ``PTOTVAL``, ``A_ENRLW`` and ``A_FTPT``, which the 2024 file +carries. :func:`.asec_pool.pool_asec_sources` concatenates per-year person +tables, so every 2022/2023 person gets NaN in a column only the 2024 file +carries, and each reader treats NaN as "no" or zero: seven reported-coverage +flags are False for two-thirds of the pool (the 14 failures of the #744 +``us_reported_coverage_vintage_signal_gate``), the 2022/2023 relationship +recode falls back to a line/spouse/parent derivation, and tax-unit +construction sees no enrollment or total-income signal for those vintages. + +The complete Census person files those inputs were extracted from are already +pinned build inputs: the base stage's ``--asec-education-source`` names the +official archives (``asecpub23csv.zip`` d2e00025…, ``asecpub24csv.zip``, +``asecpub25csv.zip``), pinned in +:data:`.spm_role_source.ASEC_SPM_ROLE_SOURCES` (archive SHA-256, member name, +member size and member SHA-256) and read through +:func:`.spm_role_source.read_pinned_asec_person_columns`, the reader the SPM +role stage uses. :func:`restore_asec_census_person_columns` appends, per +vintage and before any pooling, every reviewed column the H5 lacks, read from +that vintage's pinned person member by exact ``PERIDNUM`` identity. It never +predicts, imputes or defaults a value, and it never overwrites a column the H5 +carries: such a column is instead checked equal to the archive, which on the +2024 vintage (where the H5 carries all of them) validates the join on real +data. See ``docs/us-asec-census-person-columns.md`` for the review. + +Reviewed columns and their build readers (grep of ``packages/*/src`` and +``tools/`` on 2026-09-23; a column is restored only if a build reader exists): + +====================== ====================================================== +Column Build reader +====================== ====================================================== +``NOW_MCAID`` ``cps_carried._fill_health_coverage_inputs`` -> + ``has_medicaid_health_coverage_at_interview`` +``NOW_NONM`` same -> ``has_non_marketplace_direct_purchase_…`` +``NOW_CHAMPVA`` same -> ``has_champva_health_coverage_at_interview`` +``NOW_MIL`` same -> ``has_tricare_health_coverage_at_interview`` +``NOW_VACARE`` same -> ``has_va_health_coverage_at_interview`` +``NOW_OTHMT`` same -> ``has_other_means_tested_…_at_interview`` +``NOW_IHSFLG`` same -> ``has_indian_health_service_coverage_…`` +``A_EXPRRP`` ``asec_pool._with_relationship_recode`` (the Census + recode instead of the derived fallback); + ``microcosm.frame.units.MICROUNIT_REQUIRED_COLUMNS`` + -> microunit tax-unit head choice; + ``acs_transfer._person_head_feature``; the + ``optional_household_head`` requirement in + ``us_late_producer_registry._transfer_input_inventory`` +``PTOTVAL`` ``microcosm.frame.units._is_microunit_optional_column`` + -> microunit ``_precompute_tax_unit_inputs`` (total + money income; a pooled NaN reads as 0) +``A_ENRLW`` same -> microunit ``_is_full_time_student`` +``A_FTPT`` same; ``eligibility_inputs`` + ``derive_us_eligibility_inputs_from_manifest`` + (``is_full_time_college_student``) +====================== ====================================================== + +The other thirteen columns the 2026-08-23 offline fix appended are not +restored; :data:`ASEC_CENSUS_PERSON_COLUMNS_NOT_RESTORED` records why. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from types import MappingProxyType + +import numpy as np +import pandas as pd + +from .spm_role_source import ( + ASEC_SPM_ROLE_SOURCES, + AsecSpmRoleSource, + _exact_person_keys, + read_pinned_asec_person_columns, +) + +__all__ = [ + "ASEC_CENSUS_PERSON_COLUMNS", + "ASEC_CENSUS_PERSON_COLUMNS_NOT_RESTORED", + "ASEC_CENSUS_PERSON_COLUMN_NAMES", + "ASEC_CENSUS_PERSON_IDENTITY_COLUMNS", + "AsecCensusPersonColumn", + "AsecCensusPersonColumnsError", + "restore_asec_census_person_columns", +] + + +class AsecCensusPersonColumnsError(ValueError): + """The #720 Census person-column restoration refused its inputs.""" + + +@dataclass(frozen=True) +class AsecCensusPersonColumn: + """One reviewed Census person column and the build code that reads it.""" + + name: str + #: Census codes the column may take; ``None`` admits any integer + #: (``PTOTVAL`` is a signed dollar amount). + domain: frozenset[int] | None + readers: tuple[str, ...] + + +_NOW_YES_NO = frozenset({1, 2}) +_ENROLLMENT = frozenset({0, 1, 2}) +_HEALTH_COVERAGE_READER = ( + "microcosm.build.us_runtime.cps_carried._fill_health_coverage_inputs" +) +_MICROUNIT_OPTIONAL_READER = ( + "microcosm.frame.units._is_microunit_optional_column (passes it to microunit)" +) + +#: The reviewed restoration set, in the order columns are appended. Domains are +#: the Census codes observed in all three pinned person members (pppub23/24/25, +#: 2026-09-23); ``A_EXPRRP`` has no code 6 in the Census codebook or the data. +ASEC_CENSUS_PERSON_COLUMNS: tuple[AsecCensusPersonColumn, ...] = ( + AsecCensusPersonColumn( + "NOW_MCAID", + _NOW_YES_NO, + (f"{_HEALTH_COVERAGE_READER} -> has_medicaid_health_coverage_at_interview",), + ), + AsecCensusPersonColumn( + "NOW_NONM", + _NOW_YES_NO, + ( + f"{_HEALTH_COVERAGE_READER} -> " + "has_non_marketplace_direct_purchase_health_coverage_at_interview", + ), + ), + AsecCensusPersonColumn( + "NOW_CHAMPVA", + _NOW_YES_NO, + (f"{_HEALTH_COVERAGE_READER} -> has_champva_health_coverage_at_interview",), + ), + AsecCensusPersonColumn( + "NOW_MIL", + _NOW_YES_NO, + (f"{_HEALTH_COVERAGE_READER} -> has_tricare_health_coverage_at_interview",), + ), + AsecCensusPersonColumn( + "NOW_VACARE", + _NOW_YES_NO, + (f"{_HEALTH_COVERAGE_READER} -> has_va_health_coverage_at_interview",), + ), + AsecCensusPersonColumn( + "NOW_OTHMT", + _NOW_YES_NO, + ( + f"{_HEALTH_COVERAGE_READER} -> " + "has_other_means_tested_health_coverage_at_interview", + ), + ), + AsecCensusPersonColumn( + "NOW_IHSFLG", + _NOW_YES_NO, + ( + f"{_HEALTH_COVERAGE_READER} -> " + "has_indian_health_service_coverage_at_interview", + ), + ), + AsecCensusPersonColumn( + "A_EXPRRP", + frozenset({1, 2, 3, 4, 5, 7, 8, 9, 10, 11, 12, 13, 14}), + ( + "microcosm.build.us_runtime.asec_pool._with_relationship_recode " + "(source recode instead of the line/spouse/parent fallback)", + "microcosm.frame.units.MICROUNIT_REQUIRED_COLUMNS -> " + "microunit.tax_unit_construction._prepare_household_people " + "(relationship code for the tax-unit head choice)", + "microcosm.build.us_runtime.acs_transfer._person_head_feature", + "microcosm.build.us_runtime.us_late_producer_registry." + "_transfer_input_inventory (optional_household_head)", + ), + ), + AsecCensusPersonColumn( + "PTOTVAL", + None, + ( + _MICROUNIT_OPTIONAL_READER, + "microunit.tax_unit_construction._precompute_tax_unit_inputs " + "(total money income; a pooled NaN reads as 0)", + ), + ), + AsecCensusPersonColumn( + "A_ENRLW", + _ENROLLMENT, + ( + _MICROUNIT_OPTIONAL_READER, + "microunit.tax_unit_construction._is_full_time_student", + ), + ), + AsecCensusPersonColumn( + "A_FTPT", + _ENROLLMENT, + ( + _MICROUNIT_OPTIONAL_READER, + "microunit.tax_unit_construction._is_full_time_student", + "microcosm.build.us_runtime.eligibility_inputs." + "derive_us_eligibility_inputs_from_manifest " + "(is_full_time_college_student <- A_HSCOL == 2 & A_FTPT == 1)", + ), + ), +) +ASEC_CENSUS_PERSON_COLUMN_NAMES: tuple[str, ...] = tuple( + column.name for column in ASEC_CENSUS_PERSON_COLUMNS +) + +_UNREAD = ( + "No build reader: no code under packages/*/src or tools/ reads it " + "(2026-09-23), so restoring it would only widen the exported person table." +) +_SELF_CHECK = ( + "Read only as an optional cross-check by the SPM role derivation " + "(spm_role_source._OPTIONAL_RAW_CHECKS), which reads this same pinned member " + "itself and derives the role from it, not from the frame; restoring it from " + "that member would make the check compare the member with itself. #38's " + "unmarried-partner derivation, if built, should restore it with its reader." +) + +#: Columns the 2026-08-23 offline fix (``receipt_720.json``) appended that this +#: restoration deliberately leaves out, and why. ``CENSUS_TAX_ID`` was never +#: appended: it is extractor-derived and absent from the Census files. +ASEC_CENSUS_PERSON_COLUMNS_NOT_RESTORED: MappingProxyType[str, str] = MappingProxyType( + { + "NOW_CAID": _UNREAD, + "NOW_COV": _UNREAD, + "NOW_DIR": _UNREAD, + "NOW_MCARE": _UNREAD, + "NOW_MRKS": _UNREAD, + "NOW_MRKUN": _UNREAD, + "NOW_PCHIP": _UNREAD, + "NOW_PRIV": _UNREAD, + "NOW_PUB": _UNREAD, + "A_FAMTYP": _SELF_CHECK, + "A_FAMREL": _SELF_CHECK, + "PECOHAB": _SELF_CHECK, + "LKWEEKS": ( + "Already restored for income year 2022, the only vintage lacking it, " + "by weeks_unemployed.fill_asec_2022_weeks_unemployed_source from the " + "same pinned pppub23.csv (asecpub23csv.zip d2e00025…) before its only " + "reader; census_cps_2023.h5 and census_cps_2024.h5 carry it." + ), + } +) + +#: Census identity carried by both the H5 and the person member. After the +#: PERIDNUM join each must agree row for row, so a join that pairs the wrong +#: people cannot pass on a coincidence of keys. +ASEC_CENSUS_PERSON_IDENTITY_COLUMNS: tuple[str, ...] = ( + "PH_SEQ", + "P_SEQ", + "A_LINENO", + "A_AGE", +) + +_JOIN_KEY = "PERIDNUM" +_EXAMPLES = 5 + + +def _refuse(message: str) -> None: + raise AsecCensusPersonColumnsError(message) + + +def _is_integer(values: pd.Series) -> bool: + return pd.api.types.is_integer_dtype(values) and not pd.api.types.is_bool_dtype( + values + ) + + +def _examples(values: pd.Series | pd.Index) -> list[str]: + return [str(value) for value in list(values)[:_EXAMPLES]] + + +def _resolve_pin(income_year: int, pin: AsecSpmRoleSource | None) -> AsecSpmRoleSource: + if pin is None: + pin = ASEC_SPM_ROLE_SOURCES.get(income_year) + if pin is None: + _refuse( + "No pinned Census ASEC person source covers income year " + f"{income_year}; pinned income years: " + f"{sorted(ASEC_SPM_ROLE_SOURCES)}." + ) + if pin.income_year != income_year or pin.survey_year != income_year + 1: + _refuse( + f"Census ASEC person pin is for income year {pin.income_year} (survey " + f"{pin.survey_year}); the H5 input is income year {income_year}." + ) + return pin + + +def _read_member( + path: Path, pin: AsecSpmRoleSource, label: str +) -> tuple[pd.DataFrame, str]: + columns = (_JOIN_KEY, *ASEC_CENSUS_PERSON_IDENTITY_COLUMNS) + columns += ASEC_CENSUS_PERSON_COLUMN_NAMES + try: + source, form = read_pinned_asec_person_columns(path, pin, label, columns) + except ValueError as error: + # Pin mismatches (already labelled) and a member header missing a + # reviewed column (pandas' usecols message, unlabelled). + message = str(error) + raise AsecCensusPersonColumnsError( + message if message.startswith(label) else f"{label}: {message}" + ) from error + if len(source) != pin.persons: + _refuse(f"{label} holds {len(source)} persons; the pin records {pin.persons}.") + for column in columns[1:]: + if not _is_integer(source[column]): + _refuse( + f"{label} column {column} parsed as {source[column].dtype}, not " + "integer: the member has blank, fractional or text values there." + ) + for spec in ASEC_CENSUS_PERSON_COLUMNS: + if spec.domain is None: + continue + outside = ~source[spec.name].isin(spec.domain) + if outside.any(): + _refuse( + f"{label} column {spec.name} has {int(outside.sum())} values " + f"outside its reviewed Census codes {sorted(spec.domain)}, e.g. " + f"{_examples(source.loc[outside, spec.name].drop_duplicates())}." + ) + return source, form + + +def _one_to_one_positions( + h5_keys: pd.Series, source_keys: pd.Series, *, h5_label: str, label: str +) -> np.ndarray: + """Positions of each H5 person's archive row; refuse anything not a bijection.""" + + for side, keys in ((h5_label, h5_keys), (label, source_keys)): + duplicated = keys.duplicated(keep=False) + if duplicated.any(): + _refuse( + f"{side} repeats {int(duplicated.sum())} PERIDNUM rows, e.g. " + f"{_examples(keys[duplicated].drop_duplicates())}; the join must " + "be one-to-one." + ) + source_index = pd.Index(source_keys) + positions = source_index.get_indexer(h5_keys) + missing = positions < 0 + if missing.any(): + _refuse( + f"{int(missing.sum())} {h5_label} persons have no row in {label}, e.g. " + f"PERIDNUM {_examples(h5_keys[missing])}; the join must be total." + ) + extra = ~source_index.isin(h5_keys) + if extra.any(): + _refuse( + f"{int(extra.sum())} {label} persons are absent from {h5_label}, e.g. " + f"PERIDNUM {_examples(source_keys[extra])}; the H5 must be the complete " + "person universe of its Census member." + ) + return positions + + +def _value_summary(spec: AsecCensusPersonColumn, values: pd.Series) -> dict[str, int]: + if spec.domain is None: + return { + "nonzero_rows": int(values.ne(0).sum()), + "negative_rows": int(values.lt(0).sum()), + } + counts = values.value_counts() + return {str(code): int(counts.get(code, 0)) for code in sorted(spec.domain)} + + +def restore_asec_census_person_columns( + person: pd.DataFrame, + *, + income_year: int, + source_path: str | Path, + pin: AsecSpmRoleSource | None = None, +) -> tuple[pd.DataFrame, dict[str, object]]: + """Append the reviewed Census person columns one ASEC H5 input lacks. + + ``person`` is one vintage's complete raw person table as read from its + pinned H5 (before pooling or any smoke-household limit). ``source_path`` + is that income year's pinned Census person file: the official archive or + its extracted CSV, verified against ``pin`` (default: + :data:`.spm_role_source.ASEC_SPM_ROLE_SOURCES` for ``income_year``; an + explicit pin is for synthetic tests). + + Refuses, with :class:`AsecCensusPersonColumnsError`, when the H5 lacks the + join key or an identity column; when the archive or member fails its pins + or its person count; when a member column is not integer or leaves its + reviewed Census codes; when either side repeats a ``PERIDNUM``, an H5 + person has no member row, or a member person is absent from the H5; when + ``PH_SEQ``/``P_SEQ``/``A_LINENO``/``A_AGE`` disagree after the join; and + when a reviewed column the H5 already carries is not integer or differs + from the member on any row. A column the H5 carries is never overwritten. + New columns are appended as ``int64`` after the H5's own, in reviewed + order; every H5 column keeps its position, dtype and values (checked). + + Returns the widened table and a JSON-ready provenance record. + """ + + pin = _resolve_pin(int(income_year), pin) + label = f"ASEC {pin.survey_year} Census person member {pin.member}" + h5_label = f"income-year {income_year} ASEC H5 person table" + required = (_JOIN_KEY, *ASEC_CENSUS_PERSON_IDENTITY_COLUMNS) + missing_columns = [column for column in required if column not in person.columns] + if missing_columns: + _refuse(f"{h5_label} lacks join/identity column(s) {missing_columns}.") + if person.columns.duplicated().any(): + _refuse(f"{h5_label} repeats column names; cannot append by name.") + if not person.index.is_unique: + _refuse(f"{h5_label} has a repeated row index; cannot append by row.") + + path = Path(source_path).expanduser() + source, form = _read_member(path, pin, label) + try: + h5_keys = _exact_person_keys(person[_JOIN_KEY], h5_label).reset_index(drop=True) + source_keys = _exact_person_keys(source[_JOIN_KEY], label).reset_index( + drop=True + ) + except ValueError as error: + raise AsecCensusPersonColumnsError(str(error)) from error + positions = _one_to_one_positions( + h5_keys, source_keys, h5_label=h5_label, label=label + ) + aligned = source.iloc[positions].reset_index(drop=True) + if not np.array_equal(source_keys.to_numpy()[positions], h5_keys.to_numpy()): + _refuse(f"{label} rows did not align to {h5_label} by PERIDNUM.") + + for column in ASEC_CENSUS_PERSON_IDENTITY_COLUMNS: + if not _is_integer(person[column]) or person[column].isna().any(): + _refuse( + f"{h5_label} identity column {column} is {person[column].dtype}, " + "not integer." + ) + disagree = person[column].to_numpy() != aligned[column].to_numpy() + if disagree.any(): + _refuse( + f"{h5_label} {column} disagrees with {label} on " + f"{int(disagree.sum())} PERIDNUM-joined rows, e.g. PERIDNUM " + f"{_examples(h5_keys[disagree])}." + ) + + added: dict[str, np.ndarray] = {} + verified: list[str] = [] + values: dict[str, dict[str, int]] = {} + for spec in ASEC_CENSUS_PERSON_COLUMNS: + member_values = aligned[spec.name].to_numpy(dtype=np.int64) + if spec.name in person.columns: + carried = person[spec.name] + if not _is_integer(carried) or carried.isna().any(): + _refuse( + f"{h5_label} already carries {spec.name} as {carried.dtype} " + "(missing or non-integer values); it cannot be verified " + "against the Census member and is never overwritten." + ) + disagree = carried.to_numpy() != member_values + if disagree.any(): + _refuse( + f"{h5_label} {spec.name} disagrees with {label} on " + f"{int(disagree.sum())} PERIDNUM-joined rows, e.g. PERIDNUM " + f"{_examples(h5_keys[disagree])}." + ) + verified.append(spec.name) + else: + added[spec.name] = member_values + values[spec.name] = _value_summary(spec, pd.Series(member_values)) + + result = pd.concat( + [person, pd.DataFrame(added, index=person.index)], + axis=1, + ) + width = len(person.columns) + unchanged = ( + result.index.equals(person.index) + and list(result.columns[:width]) == list(person.columns) + and list(result.columns[width:]) == list(added) + and all( + result[column].dtype == person[column].dtype + and result[column].equals(person[column]) + for column in person.columns + ) + ) + if not unchanged: + _refuse(f"Restoring Census columns changed an existing {h5_label} column.") + + return result, { + "operation": "exact_source_join", + "join_key": "PERIDNUM (exact 22-digit string), one-to-one and total", + "income_year": pin.income_year, + "survey_year": pin.survey_year, + "source_path": str(path.resolve()), + "source_form": form, + "official_archive_url": pin.official_archive_url, + "archive_sha256": pin.archive_sha256, + "member": pin.member, + "member_sha256": pin.csv_sha256, + "member_size_bytes": pin.csv_size_bytes, + "h5_person_rows": len(person), + "member_person_rows": len(source), + "joined_person_rows": len(aligned), + "identity_columns_verified_equal": list(ASEC_CENSUS_PERSON_IDENTITY_COLUMNS), + "reviewed_columns": list(ASEC_CENSUS_PERSON_COLUMN_NAMES), + "columns_added": list(added), + "columns_verified_equal": verified, + "member_values": values, + } diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/asec_pool.py b/packages/microcosm-build/src/microcosm/build/us_runtime/asec_pool.py index 039ccef547..685296dc49 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/asec_pool.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/asec_pool.py @@ -6,6 +6,13 @@ load raw Census CPS ASEC H5 files, make household ids globally unique across years, scale each year to an explicit person-population share, and construct the Microcosm US unit frame from the pooled person table. + +When a source names its pinned complete Census person file +(``AsecSource.census_person_source``), the reviewed Census person columns its +H5 lacks are restored from that file by exact ``PERIDNUM`` identity before the +year is prepared (:mod:`.asec_census_person_columns`, microcosm #720), so the +relationship recode, unit construction and every later reader see the same +columns for every vintage. """ from __future__ import annotations @@ -18,6 +25,10 @@ import numpy as np import pandas as pd +from microcosm.build.us_runtime.asec_census_person_columns import ( + restore_asec_census_person_columns, +) +from microcosm.build.us_runtime.spm_role_source import AsecSpmRoleSource from microcosm.frame import ( MICROUNIT_REQUIRED_COLUMNS, US_SCHEMA, @@ -47,6 +58,14 @@ class AsecSource: path: Path share: float | None = None max_households: int | None = None + #: This income year's pinned complete Census ASEC person file (the + #: official archive or its extracted CSV). When set, the reviewed Census + #: person columns the H5 lacks are restored from it before anything else + #: reads the year (microcosm #720); production build tools always set it. + census_person_source: Path | None = None + #: Synthetic-test pin override for ``census_person_source``; production + #: uses ``spm_role_source.ASEC_SPM_ROLE_SOURCES[year]``. + census_person_pin: AsecSpmRoleSource | None = None @dataclass(frozen=True) @@ -133,6 +152,7 @@ def pool_asec_sources( "raw_household_rows": int(len(household)), "raw_person_population": raw_person_population, "relationship_recode_source": prepared["relationship_recode_source"], + "census_person_columns": prepared["census_person_columns"], "scale": scale, "weighted_person_population": weighted_population, } @@ -238,6 +258,17 @@ def _prepare_year_input(source: AsecSource, *, weight_scale: float) -> dict[str, tables = load_asec_h5_tables(source.path) person = tables["person"].reset_index(drop=True) household = tables["household"].reset_index(drop=True) + census_person_columns: dict[str, Any] | None = None + if source.census_person_source is not None: + # Before the relationship recode (which prefers a source A_EXPRRP) and + # before the smoke-household limit, so the PERIDNUM join is checked + # against the complete person universe of the Census member. + person, census_person_columns = restore_asec_census_person_columns( + person, + income_year=source.year, + source_path=source.census_person_source, + pin=source.census_person_pin, + ) person, relationship_source = _with_relationship_recode(person) _require_columns(person, ("PH_SEQ", *MICROUNIT_REQUIRED_COLUMNS), label="person") _require_columns(household, ("H_SEQ", "HSUP_WGT"), label="household") @@ -259,6 +290,7 @@ def _prepare_year_input(source: AsecSource, *, weight_scale: float) -> dict[str, "household": household.reset_index(drop=True), "raw_person_population": float(person_weights.sum()), "relationship_recode_source": relationship_source, + "census_person_columns": census_person_columns, } diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/spm_role_source.py b/packages/microcosm-build/src/microcosm/build/us_runtime/spm_role_source.py index 03cc5c2777..af384ff964 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/spm_role_source.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/spm_role_source.py @@ -153,17 +153,27 @@ def _reconcile_units(persons: pd.DataFrame, key: str, label: str) -> pd.DataFram return units -def _read_source_columns(source: Path | Any) -> pd.DataFrame: +#: Census identifiers read as exact text, never as numbers (a numeric parse +#: would silently drop PERIDNUM's leading zeros). +_TEXT_SOURCE_COLUMNS = ("PERIDNUM", "SPM_ID") + + +def _read_source_columns( + source: Path | Any, columns: tuple[str, ...] = _SOURCE_COLUMNS +) -> pd.DataFrame: return pd.read_csv( source, - usecols=list(_SOURCE_COLUMNS), - dtype={"PERIDNUM": str, "SPM_ID": str}, + usecols=list(columns), + dtype={column: str for column in _TEXT_SOURCE_COLUMNS if column in columns}, low_memory=False, ) def _read_archive_member( - path: Path, pin: AsecSpmRoleSource, label: str + path: Path, + pin: AsecSpmRoleSource, + label: str, + columns: tuple[str, ...] = _SOURCE_COLUMNS, ) -> pd.DataFrame: """Read the pinned person CSV from inside the official Census archive. @@ -171,7 +181,10 @@ def _read_archive_member( person CSV (``education_assistance_source._load_one_source`` accepts both), so the role stage accepts both too. The archive must be the pinned one and hold exactly one pinned member whose size and SHA-256 equal the CSV pins; - the archive is re-hashed after reading. + the archive is re-hashed after reading. ``columns`` defaults to the role + stage's own; the #720 Census person-column restoration + (:mod:`.asec_census_person_columns`) reads its reviewed set through the + same verified path. """ _require(_sha256(path) == pin.archive_sha256, f"{label} archive SHA-256 mismatch.") @@ -191,13 +204,41 @@ def _read_archive_member( digest.update(chunk) _require(digest.hexdigest() == pin.csv_sha256, f"{label} CSV SHA-256 mismatch.") with archive.open(info) as member: - source = _read_source_columns(member) + source = _read_source_columns(member, columns) _require( _sha256(path) == pin.archive_sha256, f"{label} archive changed while reading." ) return source +def read_pinned_asec_person_columns( + path: Path, + pin: AsecSpmRoleSource, + label: str, + columns: tuple[str, ...] = _SOURCE_COLUMNS, +) -> tuple[pd.DataFrame, str]: + """Read ``columns`` of one pinned complete Census ASEC person CSV. + + ``path`` is the official archive (verified by :func:`_read_archive_member`) + or its extracted person CSV, whose byte length and SHA-256 must equal the + pins before reading and whose SHA-256 is re-checked after. Returns the + columns and the path form read (``"archive"`` or ``"csv"``). The row count + is the caller's check. This is the one reader the SPM role derivation and + the #720 Census person-column restoration share. + """ + + if zipfile.is_zipfile(path): + return _read_archive_member(path, pin, label, columns), "archive" + _require( + path.stat().st_size == pin.csv_size_bytes, + f"{label} CSV byte length mismatch.", + ) + _require(_sha256(path) == pin.csv_sha256, f"{label} CSV SHA-256 mismatch.") + source = _read_source_columns(path, columns) + _require(_sha256(path) == pin.csv_sha256, f"{label} CSV changed while reading.") + return source, "csv" + + def _load_source( path: Path, pin: AsecSpmRoleSource ) -> tuple[pd.DataFrame, dict[str, Any]]: @@ -205,16 +246,7 @@ def _load_source( _require( pin.survey_year == pin.income_year + 1, f"{label} income/survey year mismatch." ) - if zipfile.is_zipfile(path): - source = _read_archive_member(path, pin, label) - else: - _require( - path.stat().st_size == pin.csv_size_bytes, - f"{label} CSV byte length mismatch.", - ) - _require(_sha256(path) == pin.csv_sha256, f"{label} CSV SHA-256 mismatch.") - source = _read_source_columns(path) - _require(_sha256(path) == pin.csv_sha256, f"{label} CSV changed while reading.") + source, _form = read_pinned_asec_person_columns(path, pin, label) _require(len(source) == pin.persons, f"{label} CSV person count mismatch.") source["PERIDNUM"] = _exact_person_keys(source.PERIDNUM, label) _require(bool(source.PERIDNUM.is_unique), f"{label} has duplicate PERIDNUM keys.") diff --git a/packages/microcosm-build/tests/test_us_asec_census_person_columns.py b/packages/microcosm-build/tests/test_us_asec_census_person_columns.py new file mode 100644 index 0000000000..1087649aa5 --- /dev/null +++ b/packages/microcosm-build/tests/test_us_asec_census_person_columns.py @@ -0,0 +1,608 @@ +"""The #720 Census person-column restoration: pinned, one-to-one, never overwriting.""" + +from __future__ import annotations + +import hashlib +import importlib +import inspect +import re +import zipfile +from dataclasses import replace +from pathlib import Path + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.us_runtime.asec_census_person_columns import ( + ASEC_CENSUS_PERSON_COLUMN_NAMES, + ASEC_CENSUS_PERSON_COLUMNS, + ASEC_CENSUS_PERSON_COLUMNS_NOT_RESTORED, + ASEC_CENSUS_PERSON_IDENTITY_COLUMNS, + AsecCensusPersonColumnsError, + restore_asec_census_person_columns, +) +from microcosm.build.us_runtime.spm_role_source import ( + _OPTIONAL_RAW_CHECKS, + ASEC_SPM_ROLE_SOURCES, + AsecSpmRoleSource, +) + +_REPOSITORY_ROOT = Path(__file__).resolve().parents[3] +_MODULE = ( + _REPOSITORY_ROOT + / "packages/microcosm-build/src/microcosm/build/us_runtime" + / "asec_census_person_columns.py" +) + +#: The 24 person columns the 2026-08-23 offline fix appended to the 2022 input +#: (``_buildo-runtime/inputs/asec-720/receipt_720.json``; the 2023 input +#: already carried LKWEEKS). The review must place every one of them. +_RECEIPT_720_ADDED = frozenset( + { + "PECOHAB", + "A_ENRLW", + "A_FTPT", + "A_FAMREL", + "A_FAMTYP", + "A_EXPRRP", + "LKWEEKS", + "NOW_CAID", + "NOW_CHAMPVA", + "NOW_COV", + "NOW_DIR", + "NOW_IHSFLG", + "NOW_MCAID", + "NOW_MCARE", + "NOW_MIL", + "NOW_MRKS", + "NOW_MRKUN", + "NOW_NONM", + "NOW_OTHMT", + "NOW_PCHIP", + "NOW_PRIV", + "NOW_PUB", + "NOW_VACARE", + "PTOTVAL", + } +) + +#: H5 row order differs from the member's, so a positional copy would fail. +_H5_ORDER = [3, 0, 7, 5, 1, 6, 2, 4] + + +def _digest(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _member() -> pd.DataFrame: + """A complete synthetic Census person member for eight people.""" + + member = pd.DataFrame( + { + # Leading zeros: a numeric parse would change the key. + "PERIDNUM": [f"{number:022}" for number in range(101, 109)], + "PH_SEQ": [1, 1, 2, 2, 2, 3, 4, 4], + "P_SEQ": [1, 2, 1, 2, 3, 1, 1, 2], + "A_LINENO": [1, 2, 1, 2, 3, 1, 1, 2], + "A_AGE": [44, 41, 70, 16, 19, 30, 52, 23], + "A_SEX": [1, 2, 2, 1, 2, 1, 1, 2], + } + ) + for spec in ASEC_CENSUS_PERSON_COLUMNS: + if spec.domain is None: + member[spec.name] = [-1_200, 0, 18_500, 0, 3_000, 61_000, 0, 9_999] + else: + codes = sorted(spec.domain) + member[spec.name] = [codes[(row * 3) % len(codes)] for row in range(8)] + return member + + +def _write(tmp_path: Path, member: pd.DataFrame, *, name: str = "pppub23.csv"): + """Write ``member`` as a CSV and as a Census-style archive; return their pin.""" + + csv = tmp_path / name + member.to_csv(csv, index=False) + archive = tmp_path / "asecpub23csv.zip" + with zipfile.ZipFile(archive, "w", compression=zipfile.ZIP_DEFLATED) as out: + out.write(csv, arcname=name) + out.writestr("hhpub23.csv", "H_SEQ\n1\n") + pin = AsecSpmRoleSource( + income_year=2022, + survey_year=2023, + csv_sha256=_digest(csv), + csv_size_bytes=csv.stat().st_size, + persons=len(member), + units=0, + official_archive_url="https://example.invalid/asecpub23csv.zip", + archive_sha256=_digest(archive), + member=name, + ) + return csv, archive, pin + + +def _h5_person(member: pd.DataFrame, *, carry: tuple[str, ...] = ()) -> pd.DataFrame: + """The H5 side: the same people in another order, lacking reviewed columns.""" + + rows = member.iloc[_H5_ORDER].reset_index(drop=True) + person = rows[["PERIDNUM", *ASEC_CENSUS_PERSON_IDENTITY_COLUMNS]].copy() + person["PERIDNUM"] = person["PERIDNUM"].astype("string") + person["NOW_GRP"] = np.array([1, 2, 2, 1, 1, 2, 1, 2], dtype=np.int64) + # An existing float column with a missing value must survive untouched. + person["WSAL_VAL"] = [0.0, np.nan, 5.5, 0.0, 1.0, 2.0, 3.0, 4.0] + for column in carry: + person[column] = rows[column].to_numpy() + return person + + +@pytest.fixture +def sources(tmp_path): + member = _member() + csv, archive, pin = _write(tmp_path, member) + return member, csv, archive, pin + + +def _restore(person, path, pin): + return restore_asec_census_person_columns( + person, income_year=2022, source_path=path, pin=pin + ) + + +@pytest.mark.parametrize("form", ["archive", "csv"]) +def test_appends_every_reviewed_column_the_h5_lacks_by_peridnum(sources, form): + member, csv, archive, pin = sources + person = _h5_person(member) + before = person.copy(deep=True) + + result, record = _restore(person, archive if form == "archive" else csv, pin) + + pd.testing.assert_frame_equal(person, before) # input not mutated + width = len(before.columns) + pd.testing.assert_frame_equal(result.iloc[:, :width], before) + assert list(result.columns[width:]) == list(ASEC_CENSUS_PERSON_COLUMN_NAMES) + expected = member.iloc[_H5_ORDER].reset_index(drop=True) + for column in ASEC_CENSUS_PERSON_COLUMN_NAMES: + assert result[column].dtype == np.int64 + np.testing.assert_array_equal( + result[column].to_numpy(), expected[column].to_numpy() + ) + assert "A_SEX" not in result # an unreviewed member column is never read in + assert record["source_form"] == form + assert record["columns_added"] == list(ASEC_CENSUS_PERSON_COLUMN_NAMES) + assert record["columns_verified_equal"] == [] + assert record["h5_person_rows"] == record["member_person_rows"] == 8 + assert record["joined_person_rows"] == 8 + assert record["archive_sha256"] == pin.archive_sha256 + assert record["member_sha256"] == pin.csv_sha256 + assert record["member"] == "pppub23.csv" + assert record["identity_columns_verified_equal"] == list( + ASEC_CENSUS_PERSON_IDENTITY_COLUMNS + ) + assert record["member_values"]["NOW_MCAID"] == { + "1": int((member["NOW_MCAID"] == 1).sum()), + "2": int((member["NOW_MCAID"] == 2).sum()), + } + assert record["member_values"]["PTOTVAL"] == { + "nonzero_rows": 5, + "negative_rows": 1, + } + + +def test_carried_columns_are_verified_against_the_member_never_overwritten(sources): + """The 2024 case: the H5 carries every reviewed column, so all are checked.""" + + member, _, archive, pin = sources + person = _h5_person(member, carry=ASEC_CENSUS_PERSON_COLUMN_NAMES) + + result, record = _restore(person, archive, pin) + + pd.testing.assert_frame_equal(result, person) + assert record["columns_added"] == [] + assert record["columns_verified_equal"] == list(ASEC_CENSUS_PERSON_COLUMN_NAMES) + + +def test_partially_carried_h5_gets_only_the_missing_columns(sources): + member, _, archive, pin = sources + person = _h5_person(member, carry=("NOW_MCAID", "A_EXPRRP")) + + result, record = _restore(person, archive, pin) + + assert list(result.columns[: len(person.columns)]) == list(person.columns) + assert record["columns_verified_equal"] == ["NOW_MCAID", "A_EXPRRP"] + assert record["columns_added"] == [ + name + for name in ASEC_CENSUS_PERSON_COLUMN_NAMES + if name not in {"NOW_MCAID", "A_EXPRRP"} + ] + + +def test_refuses_a_carried_column_that_disagrees_with_the_member(sources): + member, _, archive, pin = sources + person = _h5_person(member, carry=ASEC_CENSUS_PERSON_COLUMN_NAMES) + person.loc[2, "NOW_MCAID"] = 3 - person.loc[2, "NOW_MCAID"] + + with pytest.raises( + AsecCensusPersonColumnsError, match=r"NOW_MCAID disagrees .* on 1 " + ): + _restore(person, archive, pin) + + +def test_refuses_a_carried_column_with_missing_values(sources): + member, _, archive, pin = sources + person = _h5_person(member, carry=("NOW_MCAID",)) + person["NOW_MCAID"] = person["NOW_MCAID"].astype(float) + person.loc[0, "NOW_MCAID"] = np.nan + + with pytest.raises(AsecCensusPersonColumnsError, match="never overwritten"): + _restore(person, archive, pin) + + +def test_refuses_duplicate_h5_keys(sources): + member, _, archive, pin = sources + person = _h5_person(member) + person.loc[1, "PERIDNUM"] = person.loc[0, "PERIDNUM"] + + with pytest.raises(AsecCensusPersonColumnsError, match="repeats 2 PERIDNUM"): + _restore(person, archive, pin) + + +def test_refuses_duplicate_member_keys(tmp_path): + member = _member() + member.loc[7, "PERIDNUM"] = member.loc[6, "PERIDNUM"] + _, archive, pin = _write(tmp_path, member) + person = _h5_person(_member()) + + with pytest.raises(AsecCensusPersonColumnsError, match="repeats 2 PERIDNUM"): + _restore(person, archive, pin) + + +def test_refuses_an_h5_person_missing_from_the_member(sources): + member, _, archive, pin = sources + person = _h5_person(member) + person.loc[4, "PERIDNUM"] = "9" * 22 + + with pytest.raises(AsecCensusPersonColumnsError, match="join must be total"): + _restore(person, archive, pin) + + +def test_refuses_a_member_person_missing_from_the_h5(sources): + member, _, archive, pin = sources + person = _h5_person(member).iloc[:-1] + + with pytest.raises(AsecCensusPersonColumnsError, match="absent from"): + _restore(person, archive, pin) + + +def test_refuses_a_non_census_key(sources): + member, _, archive, pin = sources + person = _h5_person(member) + person.loc[0, "PERIDNUM"] = person.loc[0, "PERIDNUM"][1:] + + with pytest.raises(AsecCensusPersonColumnsError, match="22-digit"): + _restore(person, archive, pin) + + +@pytest.mark.parametrize("column", ASEC_CENSUS_PERSON_IDENTITY_COLUMNS) +def test_refuses_identity_disagreement_after_the_join(sources, column): + member, _, archive, pin = sources + person = _h5_person(member) + person.loc[5, column] = person.loc[5, column] + 1 + + with pytest.raises(AsecCensusPersonColumnsError, match=f"{column} disagrees"): + _restore(person, archive, pin) + + +def test_refuses_an_h5_without_the_join_or_identity_columns(sources): + member, _, archive, pin = sources + with pytest.raises(AsecCensusPersonColumnsError, match="P_SEQ"): + _restore(_h5_person(member).drop(columns=["P_SEQ"]), archive, pin) + + +@pytest.mark.parametrize( + ("value", "match"), + [ + ("", "parsed as float64"), + ("yes", "parsed as"), + (3, "outside its reviewed Census codes"), + ], +) +def test_refuses_member_dtype_and_code_surprises(tmp_path, value, match): + member = _member().astype({"NOW_MCAID": object}) + member.loc[2, "NOW_MCAID"] = value + _, archive, pin = _write(tmp_path, member) + + with pytest.raises(AsecCensusPersonColumnsError, match=match): + _restore(_h5_person(_member()), archive, pin) + + +def test_refuses_a_member_without_a_reviewed_column(tmp_path): + _, archive, pin = _write(tmp_path, _member().drop(columns=["NOW_IHSFLG"])) + + with pytest.raises(AsecCensusPersonColumnsError, match="NOW_IHSFLG"): + _restore(_h5_person(_member()), archive, pin) + + +def test_refuses_sources_that_fail_their_pins(sources): + member, csv, archive, pin = sources + person = _h5_person(member) + with pytest.raises(AsecCensusPersonColumnsError, match="archive SHA-256"): + _restore(person, archive, replace(pin, archive_sha256="0" * 64)) + with pytest.raises(AsecCensusPersonColumnsError, match="CSV SHA-256"): + _restore(person, archive, replace(pin, csv_sha256="0" * 64)) + with pytest.raises(AsecCensusPersonColumnsError, match="CSV SHA-256"): + _restore(person, csv, replace(pin, csv_sha256="0" * 64)) + with pytest.raises(AsecCensusPersonColumnsError, match="exactly one"): + _restore(person, archive, replace(pin, member="pppub99.csv")) + with pytest.raises(AsecCensusPersonColumnsError, match="the pin records 9"): + _restore(person, archive, replace(pin, persons=9)) + + +def test_refuses_an_unpinned_or_mismatched_income_year(sources): + member, _, archive, pin = sources + person = _h5_person(member) + with pytest.raises(AsecCensusPersonColumnsError, match="No pinned"): + restore_asec_census_person_columns( + person, income_year=2019, source_path=archive + ) + with pytest.raises(AsecCensusPersonColumnsError, match="income year 2023"): + restore_asec_census_person_columns( + person, income_year=2023, source_path=archive, pin=pin + ) + + +def test_default_pins_are_the_shared_census_person_pins(sources): + """Without an explicit pin the production table applies (and refuses here).""" + + member, _, archive, _ = sources + assert set(ASEC_SPM_ROLE_SOURCES) == {2022, 2023, 2024} + with pytest.raises(AsecCensusPersonColumnsError, match="archive SHA-256"): + restore_asec_census_person_columns( + _h5_person(member), income_year=2022, source_path=archive + ) + + +def test_the_review_places_every_offline_fix_column_exactly_once(): + restored = set(ASEC_CENSUS_PERSON_COLUMN_NAMES) + not_restored = set(ASEC_CENSUS_PERSON_COLUMNS_NOT_RESTORED) + assert len(restored) == len(ASEC_CENSUS_PERSON_COLUMN_NAMES) + assert restored.isdisjoint(not_restored) + assert restored | not_restored == _RECEIPT_720_ADDED + assert not restored & set(ASEC_CENSUS_PERSON_IDENTITY_COLUMNS) + + +def _cited_objects(reader: str): + for dotted in re.findall(r"\b(?:microcosm|microunit)(?:\.\w+)+", reader): + parts = dotted.split(".") + for split in range(len(parts), 0, -1): + try: + module = importlib.import_module(".".join(parts[:split])) + except ImportError: + continue + target = module + for attribute in parts[split:]: + target = getattr(target, attribute) + yield dotted, module, target + break + else: + raise AssertionError(f"cited reader {dotted!r} does not import") + + +def test_every_restored_column_cites_readers_that_really_read_it(): + """Each cited function or constant must resolve and itself name the column.""" + + pytest.importorskip("microunit") + for spec in ASEC_CENSUS_PERSON_COLUMNS: + assert spec.readers, spec.name + for reader in spec.readers: + cited = list(_cited_objects(reader)) + assert cited, reader + for dotted, _module, target in cited: + if inspect.isfunction(target): + names_it = bool( + re.search(rf"\b{spec.name}\b", inspect.getsource(target)) + ) + else: + names_it = spec.name in target + assert names_it, ( + f"{dotted} is cited as a reader of {spec.name} but does not " + "name it." + ) + + +def test_columns_left_unrestored_for_want_of_a_reader_still_have_none(): + """If code starts reading one, it must join the reviewed restoration set.""" + + unread = sorted( + column + for column, reason in ASEC_CENSUS_PERSON_COLUMNS_NOT_RESTORED.items() + if reason.startswith("No build reader") + ) + assert unread == [ + "NOW_CAID", + "NOW_COV", + "NOW_DIR", + "NOW_MCARE", + "NOW_MRKS", + "NOW_MRKUN", + "NOW_PCHIP", + "NOW_PRIV", + "NOW_PUB", + ] + pattern = re.compile(rf"\b({'|'.join(unread)})\b") + roots = [*(_REPOSITORY_ROOT / "packages").glob("*/src"), _REPOSITORY_ROOT / "tools"] + readers = { + str(path.relative_to(_REPOSITORY_ROOT)): sorted(set(pattern.findall(text))) + for root in roots + for path in root.rglob("*.py") + if path.resolve() != _MODULE.resolve() + and pattern.search(text := path.read_text(encoding="utf-8")) + } + assert readers == {} + + +def test_other_unrestored_columns_keep_their_recorded_reasons(): + from microcosm.build.us_runtime.weeks_unemployed import ( + ASEC_2023_WEEKS_UNEMPLOYED_MEMBER, + ASEC_2023_WEEKS_UNEMPLOYED_MEMBER_SHA256, + ASEC_2023_WEEKS_UNEMPLOYED_SOURCE_YEAR, + ASEC_2023_WEEKS_UNEMPLOYED_ZIP_SHA256, + ) + + for column in ("A_FAMTYP", "A_FAMREL", "PECOHAB"): + assert column in _OPTIONAL_RAW_CHECKS + # LKWEEKS: the existing 2022 repair reads the very member pinned here. + pin = ASEC_SPM_ROLE_SOURCES[ASEC_2023_WEEKS_UNEMPLOYED_SOURCE_YEAR] + assert ASEC_2023_WEEKS_UNEMPLOYED_ZIP_SHA256 == pin.archive_sha256 + assert ASEC_2023_WEEKS_UNEMPLOYED_MEMBER == pin.member + assert ASEC_2023_WEEKS_UNEMPLOYED_MEMBER_SHA256 == pin.csv_sha256 + + +# --- the pool: every vintage is restored before anything reads it ----------- + + +def _write_asec_h5(path: Path, member: pd.DataFrame, *, carry: bool) -> None: + person = member[["PERIDNUM", *ASEC_CENSUS_PERSON_IDENTITY_COLUMNS, "A_SEX"]].copy() + person["A_MARITL"] = [1, 1, 7, 7, 7, 7, 7, 7] + person["A_SPOUSE"] = [2, 1, 0, 0, 0, 0, 0, 0] + person["PEPAR1"] = [0, 0, 0, 1, 1, 0, 0, 0] + person["PEPAR2"] = 0 + person["PF_SEQ"] = 1 + person["SPM_ID"] = person["PH_SEQ"] + person["NOW_GRP"] = 2 + person["NOW_MRK"] = 2 + if carry: + for column in ASEC_CENSUS_PERSON_COLUMN_NAMES: + person[column] = member[column].to_numpy() + household = pd.DataFrame( + { + "H_SEQ": [1, 2, 3, 4], + "HSUP_WGT": [10_000, 20_000, 15_000, 5_000], + "GESTFIPS": [6, 36, 48, 12], + "H_TENURE": [1, 2, 1, 2], + } + ) + person.to_hdf(path, key="person", mode="w") + household.to_hdf(path, key="household", mode="a") + + +@pytest.fixture +def pool_inputs(tmp_path): + pytest.importorskip("tables") + pytest.importorskip("microunit") + from microcosm.build.us_runtime.asec_pool import AsecSource + + old = tmp_path / "old" + new = tmp_path / "new" + old.mkdir() + new.mkdir() + member = _member() + # Person 4 is a partner the derived fallback cannot identify (it codes 12). + member.loc[4, "A_EXPRRP"] = 13 + _, old_archive, old_pin = _write(old, member) + _, new_archive, new_pin = _write(new, member, name="pppub25.csv") + new_pin = replace(new_pin, income_year=2024, survey_year=2025) + _write_asec_h5(old / "census_cps_2022.h5", member, carry=False) + _write_asec_h5(new / "census_cps_2024.h5", member, carry=True) + return member, [ + AsecSource( + year=2024, + path=new / "census_cps_2024.h5", + census_person_source=new_archive, + census_person_pin=new_pin, + ), + AsecSource( + year=2022, + path=old / "census_cps_2022.h5", + census_person_source=old_archive, + census_person_pin=old_pin, + ), + ] + + +def test_pool_restores_every_vintage_before_the_relationship_recode(pool_inputs): + from microcosm.build.us_runtime.asec_pool import pool_asec_sources + + member, sources = pool_inputs + pooled = pool_asec_sources(sources) + person = pooled.person + + for column in ASEC_CENSUS_PERSON_COLUMN_NAMES: + assert person[column].notna().all(), column + assert person[column].dtype == np.int64, column + for year in (2022, 2024): + rows = person.loc[person["source_year"] == year] + expected = member.set_index("PERIDNUM").loc[rows["PERIDNUM"], column] + np.testing.assert_array_equal(rows[column].to_numpy(), expected) + by_year = {item["year"]: item for item in pooled.metadata["sources"]} + assert { + year: item["relationship_recode_source"] for year, item in by_year.items() + } == { + 2022: "source:A_EXPRRP", + 2024: "source:A_EXPRRP", + } + old, new = ( + by_year[2022]["census_person_columns"], + by_year[2024]["census_person_columns"], + ) + assert old["columns_added"] == list(ASEC_CENSUS_PERSON_COLUMN_NAMES) + assert new["columns_added"] == [] + assert new["columns_verified_equal"] == list(ASEC_CENSUS_PERSON_COLUMN_NAMES) + partner = person["PERIDNUM"].eq(member.loc[4, "PERIDNUM"]) + assert set(person.loc[partner, "A_EXPRRP"]) == {13} + + +def test_pool_without_a_census_person_source_keeps_the_720_hole(pool_inputs): + """The library default documents the defect the build tools now close.""" + + from microcosm.build.us_runtime.asec_pool import pool_asec_sources + + _, sources = pool_inputs + bare = [ + replace(source, census_person_source=None, census_person_pin=None) + for source in sources + ] + pooled = pool_asec_sources(bare) + old = pooled.person["source_year"] == 2022 + assert pooled.person.loc[old, "NOW_MCAID"].isna().all() + assert { + item["year"]: item["relationship_recode_source"] + for item in pooled.metadata["sources"] + } == {2022: "derived:line_spouse_parent", 2024: "source:A_EXPRRP"} + assert all( + item["census_person_columns"] is None for item in pooled.metadata["sources"] + ) + + +def test_pool_smoke_limit_still_checks_the_complete_join(pool_inputs): + from microcosm.build.us_runtime.asec_pool import pool_asec_sources + + _, sources = pool_inputs + smoke = [replace(source, max_households=1) for source in sources] + pooled = pool_asec_sources(smoke) + assert set(pooled.person["source_household_id"]) == {1} + for item in pooled.metadata["sources"]: + assert item["census_person_columns"]["joined_person_rows"] == 8 + + +def test_pool_refuses_a_source_that_fails_the_restoration(pool_inputs, tmp_path): + from microcosm.build.us_runtime.asec_pool import pool_asec_sources + + _, sources = pool_inputs + broken = list(sources) + broken[1] = replace( + broken[1], + census_person_pin=replace(broken[1].census_person_pin, archive_sha256="0" * 64), + ) + with pytest.raises(AsecCensusPersonColumnsError, match="archive SHA-256"): + pool_asec_sources(broken) + + +def test_pooled_unit_frame_builds_on_the_restored_columns(pool_inputs): + from microcosm.build.us_runtime.asec_pool import build_pooled_asec_unit_frame + + _, sources = pool_inputs + frame, metadata = build_pooled_asec_unit_frame(sources, target_year=2024) + person = frame.table("person") + assert person["NOW_MCAID"].notna().all() + assert frame.n("tax_unit") > 0 + assert all(item["census_person_columns"] for item in metadata["sources"]) diff --git a/packages/microcosm-build/tests/test_us_puf_support_base_builder.py b/packages/microcosm-build/tests/test_us_puf_support_base_builder.py index e6c335b8a2..196b803639 100644 --- a/packages/microcosm-build/tests/test_us_puf_support_base_builder.py +++ b/packages/microcosm-build/tests/test_us_puf_support_base_builder.py @@ -1831,6 +1831,10 @@ def fake_build_pooled_asec_unit_frame(sources, *, target_year): f"2023={asec_2023}", "--asec-h5", f"2024={asec_2024}", + "--asec-education-source", + f"2023={tmp_path / 'asecpub24csv.zip'}", + "--asec-education-source", + f"2024={tmp_path / 'asecpub25csv.zip'}", "--target-year", "2024", "--asec-max-households", @@ -1848,11 +1852,17 @@ def fake_build_pooled_asec_unit_frame(sources, *, target_year): assert frame is sentinel_frame assert captured["target_year"] == 2024 assert [ - (source.year, source.path.name, source.max_households) + ( + source.year, + source.path.name, + source.max_households, + source.census_person_source, + source.census_person_pin, + ) for source in captured["sources"] ] == [ - (2023, "asec_2023.h5", 50), - (2024, "asec_2024.h5", 50), + (2023, "asec_2023.h5", 50, tmp_path / "asecpub24csv.zip", None), + (2024, "asec_2024.h5", 50, tmp_path / "asecpub25csv.zip", None), ] assert metadata == { "kind": "pooled_asec", @@ -1933,6 +1943,13 @@ def fake_build_pooled_asec_unit_frame(sources, *, target_year): fake_build_pooled_asec_unit_frame, ) monkeypatch.setattr(builder, "_sha256", lambda path: f"sha:{Path(path).name}") + resolved = {} + + def fake_resolve(paths, *, income_years): + resolved["call"] = (paths, income_years) + return {year: tmp_path / f"census_{year}.zip" for year in income_years} + + monkeypatch.setattr(builder, "resolve_asec_spm_role_source_paths", fake_resolve) args = builder._parse_args( [ @@ -1962,11 +1979,87 @@ def fake_build_pooled_asec_unit_frame(sources, *, target_year): (2024, "asec_2024.h5", 0.25), (2025, "asec_2025.h5", 0.75), ] + # Spine-spec sources are bound to their Census person files like any other. + assert resolved["call"] == (None, (2024, 2025)) + assert [source.census_person_source for source in captured["sources"]] == [ + tmp_path / "census_2024.zip", + tmp_path / "census_2025.zip", + ] assert metadata["support_spine_spec"]["path"] == str(spec_path.resolve()) assert metadata["support_spine_spec"]["sources"][0]["resolved_year"] == 2024 assert metadata["support_spine_spec"]["sources"][1]["resolved_year"] == 2025 +def test_pooled_asec_sources_fetch_unmapped_census_person_files( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """A pooled year without --asec-education-source is fetched and pinned (#720). + + Source construction resolves the Census person files exactly as the SPM + role stage does, so no base built from --asec-h5 inputs can skip the + restoration of the columns its H5 lacks. + """ + + from microcosm.build.us_runtime import spm_independence_role + + builder = _load_support_builder_module() + fetched = [] + + def fake_fetch(year): + fetched.append(year) + return tmp_path / f"fetched_{year}.csv" + + monkeypatch.setattr( + spm_independence_role, "fetch_asec_education_assistance_source", fake_fetch + ) + args = builder._parse_args( + [ + "--asec-h5", + f"2024={tmp_path / 'asec_2024.h5'}", + "--asec-h5", + f"2022={tmp_path / 'asec_2022.h5'}", + "--asec-education-source", + f"2024={tmp_path / 'asecpub25csv.zip'}", + "--puf-h5", + "puf.h5", + "--out", + "out", + "--without-block-ladder", + ] + ) + + sources = builder._asec_sources_from_args(args, support_spine_spec=None) + + assert fetched == [2022] + assert [(source.year, source.census_person_source) for source in sources] == [ + (2024, tmp_path / "asecpub25csv.zip"), + (2022, tmp_path / "fetched_2022.csv"), + ] + + +def test_pooled_asec_sources_refuse_an_income_year_without_census_pins( + tmp_path: Path, +) -> None: + builder = _load_support_builder_module() + args = builder._parse_args( + [ + "--asec-h5", + f"2021={tmp_path / 'asec_2021.h5'}", + "--asec-education-source", + f"2021={tmp_path / 'asecpub22csv.zip'}", + "--puf-h5", + "puf.h5", + "--out", + "out", + "--without-block-ladder", + ] + ) + + with pytest.raises(ValueError, match=r"income year\(s\) \[2021\]"): + builder._asec_sources_from_args(args, support_spine_spec=None) + + def test_support_spine_spec_requires_mapped_asec_year(tmp_path: Path) -> None: builder = _load_support_builder_module() diff --git a/packages/microcosm-build/tests/test_us_spine_blindness.py b/packages/microcosm-build/tests/test_us_spine_blindness.py index 8347a606c1..a18f026f5c 100644 --- a/packages/microcosm-build/tests/test_us_spine_blindness.py +++ b/packages/microcosm-build/tests/test_us_spine_blindness.py @@ -234,6 +234,11 @@ "acs_release_predictors.py", # Pinned release join; provenance owner. "acs_sources.py", "acs_transfer_bank.py", # Bounded checkpoint I/O; no population treatment. + # Reviewed Census person columns restored into each raw ASEC vintage by + # exact pinned PERIDNUM identity before pooling (#720), through + # spm_role_source's pinned reader; no population treatment. Remains + # subject to the all-runtime source-identity scan. + "asec_census_person_columns.py", "asec_checkpoint.py", # Bounded checkpoint I/O; no population treatment. "asec_pool.py", # Pinned ASEC source coordinates and verified fetch; no population diff --git a/packages/microcosm-build/tests/test_us_spm_role_source.py b/packages/microcosm-build/tests/test_us_spm_role_source.py index 6989cfa9bc..d1979449ce 100644 --- a/packages/microcosm-build/tests/test_us_spm_role_source.py +++ b/packages/microcosm-build/tests/test_us_spm_role_source.py @@ -9,11 +9,13 @@ import pytest from microcosm.build.us_runtime.spm_role_source import ( + _SOURCE_COLUMNS, ASEC_SPM_ROLE_SOURCES, EVIDENCE_SPM_ROLE, AsecSpmRoleSource, derive_spm_role_source, independent_minor_role, + read_pinned_asec_person_columns, ) pytest.importorskip("tables") @@ -255,6 +257,50 @@ def test_rejects_unpinned_csv(population): _derive((parent, source, replace(pin, csv_sha256="0" * 64))) +def test_shared_reader_default_is_the_role_read_from_either_form(population): + """Generalising the reader (#720) left the role stage's read unchanged.""" + + _, source, pin = population + _, archive, archive_pin = _archive(population) + expected = pd.read_csv( + source, + usecols=list(_SOURCE_COLUMNS), + dtype={"PERIDNUM": str, "SPM_ID": str}, + low_memory=False, + ) + from_csv, csv_form = read_pinned_asec_person_columns(source, pin, "ASEC 2025") + from_archive, archive_form = read_pinned_asec_person_columns( + archive, archive_pin, "ASEC 2025" + ) + assert (csv_form, archive_form) == ("csv", "archive") + pd.testing.assert_frame_equal(from_csv, expected) + pd.testing.assert_frame_equal(from_archive, expected) + + +def test_shared_reader_reads_other_columns_through_the_same_pins(population): + _, source, pin = population + _, archive, archive_pin = _archive(population) + columns = ("PERIDNUM", "A_AGE", "PECOHAB") + for path, path_pin in ((source, pin), (archive, archive_pin)): + frame, _ = read_pinned_asec_person_columns(path, path_pin, "ASEC 2025", columns) + assert set(frame.columns) == set(columns) + # Exact text keys (leading zeros kept); other columns parse as integers. + assert frame["PERIDNUM"].tolist() == [f"{n:022}" for n in range(1, 7)] + assert frame["A_AGE"].dtype == np.int64 + with pytest.raises(ValueError, match="archive SHA-256"): + read_pinned_asec_person_columns( + archive, replace(archive_pin, archive_sha256="0" * 64), "x", columns + ) + with pytest.raises(ValueError, match="CSV SHA-256"): + read_pinned_asec_person_columns( + source, replace(pin, csv_sha256="0" * 64), "x", columns + ) + with pytest.raises(ValueError, match="CSV byte length"): + read_pinned_asec_person_columns( + source, replace(pin, csv_size_bytes=pin.csv_size_bytes + 1), "x", columns + ) + + def test_rejects_missing_source_year(population): parent, _, pin = population with pytest.raises(ValueError, match="must match exactly"): diff --git a/tools/build_us_asec_pooled_source_base.py b/tools/build_us_asec_pooled_source_base.py index 51cd3dcab0..5564ce5c15 100644 --- a/tools/build_us_asec_pooled_source_base.py +++ b/tools/build_us_asec_pooled_source_base.py @@ -3,18 +3,26 @@ This is a pre-calibration support construction step. It pools raw ASEC years before unit assignment and writes diagnostics proving source-year shares. It does not run donor imputations, fiscal calibration, or release certification. +Like the base builder's source construction, it restores the reviewed Census +person columns each ASEC H5 lacks from that year's pinned Census person file +(microcosm #720). """ from __future__ import annotations import argparse import json +from dataclasses import replace from pathlib import Path import numpy as np import pandas as pd -from microcosm.build.us_runtime import AsecSource, build_pooled_asec_unit_frame +from microcosm.build.us_runtime import ( + AsecSource, + build_pooled_asec_unit_frame, + resolve_asec_spm_role_source_paths, +) def _parse_args() -> argparse.Namespace: @@ -31,6 +39,18 @@ def _parse_args() -> argparse.Namespace: type=int, help="Optional smoke limit applied to every ASEC source.", ) + parser.add_argument( + "--asec-education-source", + action="append", + metavar="YEAR=PATH", + help=( + "Optional INCOME_YEAR=PATH mapping to the SHA-pinned official ASEC " + "survey archive (zip or extracted pppub member) the reviewed Census " + "person columns are restored from (microcosm #720). Years without " + "a mapping are fetched from the official Census archive and " + "verified against the same pins." + ), + ) parser.add_argument("--summary", type=Path, required=True) parser.add_argument( "--out-h5", @@ -49,6 +69,14 @@ def main() -> int: _parse_source(value, max_households=args.max_households) for value in args.asec_h5 ) + census_person_sources = resolve_asec_spm_role_source_paths( + _census_person_source_paths(args.asec_education_source), + income_years=tuple(sorted({source.year for source in sources})), + ) + sources = tuple( + replace(source, census_person_source=census_person_sources[source.year]) + for source in sources + ) frame, metadata = build_pooled_asec_unit_frame( sources, target_year=args.target_year, @@ -76,6 +104,21 @@ def _parse_source(value: str, *, max_households: int | None) -> AsecSource: return AsecSource(year=year, path=path, max_households=max_households) +def _census_person_source_paths(values: list[str] | None) -> dict[int, Path]: + paths: dict[int, Path] = {} + for value in values or (): + raw_year, _, raw_path = value.partition("=") + if not raw_path: + raise ValueError( + f"--asec-education-source must be INCOME_YEAR=PATH, got {value!r}." + ) + year = int(raw_year) + if year in paths: + raise ValueError(f"--asec-education-source repeats income year {year}.") + paths[year] = Path(raw_path) + return paths + + def _summary(frame, metadata: dict) -> dict[str, object]: person = frame.table("person") household = frame.table("household") diff --git a/tools/build_us_puf_support_base.py b/tools/build_us_puf_support_base.py index 029f24891a..b5f8663ea1 100644 --- a/tools/build_us_puf_support_base.py +++ b/tools/build_us_puf_support_base.py @@ -17,6 +17,7 @@ import sys from collections.abc import Callable, Iterator, Mapping, Sequence from contextlib import contextmanager +from dataclasses import replace from pathlib import Path import numpy as np @@ -87,6 +88,7 @@ load_congressional_district_vintage_crosswalk, load_us_block_ladder, puf_tax_unit_donor_from_arrays, + resolve_asec_spm_role_source_paths, source_year_puf_adjusted_gross_income, support_channel_column, transfer_puf_capital_gains_tail, @@ -340,9 +342,12 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: help=( "Optional INCOME_YEAR=PATH mapping to a local copy of the " "SHA-pinned official ASEC survey archive (zip or extracted " - "pppub member) restoring that pooled income year's ED_VAL and " - "PAW_TYP (income year YYYY maps to the survey-year YYYY+1 " - "archive). Years without a mapping are fetched from the " + "pppub member) for that pooled income year (income year YYYY " + "maps to the survey-year YYYY+1 archive). Source construction " + "restores from it the reviewed Census person columns the " + "--asec-h5 input lacks (microcosm #720); the raw-stage mapping " + "restores ED_VAL and PAW_TYP; the SPM independence role stage " + "derives the role. Years without a mapping are fetched from the " "official Census archive and verified against the same pins." ), ) @@ -3253,6 +3258,33 @@ def _asec_sources_from_args( args: argparse.Namespace, *, support_spine_spec: SupportSpineSpec | None, +) -> tuple[AsecSource, ...]: + """Pooled ASEC sources, each bound to its pinned Census person file. + + Every source carries ``census_person_source`` so the pool restores the + reviewed Census person columns its H5 lacks (microcosm #720). The files + are the ``--asec-education-source`` mappings, with any pooled income year + left unmapped fetched from the official Census archive and verified, + exactly as the SPM independence role stage resolves them. + """ + + sources = _pooled_asec_sources_from_args( + args, support_spine_spec=support_spine_spec + ) + census_person_sources = resolve_asec_spm_role_source_paths( + _asec_education_source_paths(args), + income_years=tuple(sorted({source.year for source in sources})), + ) + return tuple( + replace(source, census_person_source=census_person_sources[source.year]) + for source in sources + ) + + +def _pooled_asec_sources_from_args( + args: argparse.Namespace, + *, + support_spine_spec: SupportSpineSpec | None, ) -> tuple[AsecSource, ...]: if support_spine_spec is None: return tuple(