From 39b8e7b63afd4f52c4c1dce85ac89538ea2ea87a Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Wed, 23 Sep 2026 05:43:27 -0400 Subject: [PATCH 1/2] Restore the reviewed Census person columns the 2022/2023 ASEC inputs lack (#720) The pinned income-year 2022/2023 census_cps H5 inputs carry 2 of the 18 NOW_* at-interview recodes and lack A_EXPRRP, PTOTVAL, A_ENRLW and A_FTPT, which the 2024 input carries. Pooling leaves those cells NaN for two-thirds of the pool and every reader treats NaN as "no" or zero, so seven reported-coverage flags have zero reporters in the 2022/2023 vintages (the #744 gate's 14 failures), the older vintages use a derived relationship recode, and tax-unit construction sees no enrollment or total-income signal for them. The complete Census person files are already pinned build inputs (--asec-education-source; ASEC_SPM_ROLE_SOURCES). The new asec_census_person_columns module appends, per vintage and before any pooling, every reviewed column the H5 lacks, read from that year's pinned member through the SPM role stage's reader (generalised to take a column list, its default read unchanged) and joined on exact PERIDNUM. The join must be one-to-one and total; identity columns must agree after it; member columns must be integers inside their reviewed Census codes; a column the H5 already carries is verified equal and never overwritten (all eleven on 2024); every existing H5 column is checked unchanged. Each refusal is a named AsecCensusPersonColumnsError. Reviewed set (11, each with a cited build reader that a test resolves): NOW_MCAID, NOW_NONM, NOW_CHAMPVA, NOW_MIL, NOW_VACARE, NOW_OTHMT, NOW_IHSFLG, A_EXPRRP, PTOTVAL, A_ENRLW, A_FTPT. Of the 24 columns the 2026-08-23 offline fix appended, nine NOW_* have no reader, A_FAMTYP/A_FAMREL/PECOHAB are only self-checks of the same member in the SPM role stage, and LKWEEKS is already restored for 2022 from the same pinned archive. AsecSource gains census_person_source; build_us_puf_support_base.py and build_us_asec_pooled_source_base.py bind every pooled year to its pinned Census person file (the --asec-education-source mapping, else fetched and verified), so the source-construction frame, the raw-stage checkpoint and everything downstream carry the columns. Provenance (archive and member digests, columns added or verified, row counts, member code counts) is recorded per vintage in the pool metadata, and so in the source-construction receipt, the raw-stage checkpoint's source_receipt and the base summary. Co-Authored-By: Claude Opus 5.5 --- .../us_runtime/asec_census_person_columns.py | 506 +++++++++++++++ .../microcosm/build/us_runtime/asec_pool.py | 32 + .../build/us_runtime/spm_role_source.py | 64 +- .../test_us_asec_census_person_columns.py | 608 ++++++++++++++++++ .../tests/test_us_puf_support_base_builder.py | 99 ++- .../tests/test_us_spine_blindness.py | 5 + .../tests/test_us_spm_role_source.py | 46 ++ tools/build_us_asec_pooled_source_base.py | 45 +- tools/build_us_puf_support_base.py | 38 +- 9 files changed, 1420 insertions(+), 23 deletions(-) create mode 100644 packages/microcosm-build/src/microcosm/build/us_runtime/asec_census_person_columns.py create mode 100644 packages/microcosm-build/tests/test_us_asec_census_person_columns.py 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 556f92828a..36c1170bbe 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( From c9222e2da047ef1f6aaa21e1b08f1fd0cd0f14b0 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Wed, 23 Sep 2026 05:57:37 -0400 Subject: [PATCH 2/2] Document the #720 Census column review and commit its real-data receipts docs/us-asec-census-person-columns.md records the defect, the reviewed column table with each build reader, the refusals, the paths covered, and the 2026-09-23 source-construction evidence on route A's exact inputs: - #744 gate on derive_us_cps_carried_inputs of the raw-stage frame: route A's checkpoint (47976be6c) fails with 14 failures (Medicaid reporters 0 / 0 / 24,844); this branch passes (28,184 / 26,367 / 24,844, the 2026-08-23 receipt's numbers). - The eleven restored columns equal the 2026-08-23 corrected H5s in value, dtype and order; the whole raw-stage frame equals the 8/23 corrected raw-stage frame on every entity table and every shared column except the twelve columns deliberately not restored. - Structural effect: 2022/2023 use Census's A_EXPRRP (20,595 / 20,561 persons change), tax units 231,007 -> 226,102 (2024 unchanged). experiments/us-asec-720-census-person-columns holds the run script, the verification scripts and their JSON outputs, and the stage context. Co-Authored-By: Claude Opus 5.5 --- ...us-asec-720-census-person-columns.fixed.md | 1 + docs/us-asec-census-person-columns.md | 194 ++++ .../README.md | 16 + .../checkpoint_sha256.txt | 2 + .../compare_823_raw_stage.json | 158 ++++ .../compare_823_raw_stage.py | 113 +++ .../run_source_construction.sh | 31 + .../source_construction_stage_profile.json | 16 + ...source_construction_stage_run_context.json | 1 + .../verify_720.json | 886 ++++++++++++++++++ .../verify_720.py | 226 +++++ 11 files changed, 1644 insertions(+) create mode 100644 changelog.d/us-asec-720-census-person-columns.fixed.md create mode 100644 docs/us-asec-census-person-columns.md create mode 100644 experiments/us-asec-720-census-person-columns/README.md create mode 100644 experiments/us-asec-720-census-person-columns/checkpoint_sha256.txt create mode 100644 experiments/us-asec-720-census-person-columns/compare_823_raw_stage.json create mode 100644 experiments/us-asec-720-census-person-columns/compare_823_raw_stage.py create mode 100755 experiments/us-asec-720-census-person-columns/run_source_construction.sh create mode 100644 experiments/us-asec-720-census-person-columns/source_construction_stage_profile.json create mode 100644 experiments/us-asec-720-census-person-columns/source_construction_stage_run_context.json create mode 100644 experiments/us-asec-720-census-person-columns/verify_720.json create mode 100644 experiments/us-asec-720-census-person-columns/verify_720.py 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())