diff --git a/changelog.d/719-work-experience-inputs.added.md b/changelog.d/719-work-experience-inputs.added.md new file mode 100644 index 00000000..b3223f9f --- /dev/null +++ b/changelog.d/719-work-experience-inputs.added.md @@ -0,0 +1 @@ +US `work_experience_inputs` source stage: `detailed_industry_recode` (ASEC WEIND, industry of longest job by detailed groups), `major_industry_recode` (WEMIND), and `worked_last_year` (WKSWORK > 0, the work-experience recode universe) on every person record, restored from the three SHA-pinned official ASEC archives via exact PERIDNUM joins; raw ASEC checkpoint schema version 4 (#719). diff --git a/packages/microcosm-build/src/microcosm/build/source_manifest.py b/packages/microcosm-build/src/microcosm/build/source_manifest.py index c4dafc79..add45833 100644 --- a/packages/microcosm-build/src/microcosm/build/source_manifest.py +++ b/packages/microcosm-build/src/microcosm/build/source_manifest.py @@ -87,6 +87,7 @@ "derive_workers_compensation", "derive_weeks_unemployed", "derive_wic_claim", + "derive_work_experience_inputs", "disaggregate_aggregate_records", "fit_labor_market_models", "fit_tip_income_model", diff --git a/packages/microcosm-build/src/microcosm/build/spec_engine/schema/sources.schema.json b/packages/microcosm-build/src/microcosm/build/spec_engine/schema/sources.schema.json index 1c7d9d1d..fb91ec50 100644 --- a/packages/microcosm-build/src/microcosm/build/spec_engine/schema/sources.schema.json +++ b/packages/microcosm-build/src/microcosm/build/spec_engine/schema/sources.schema.json @@ -1417,6 +1417,18 @@ "kind" ] }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "kind": { + "const": "derive_work_experience_inputs" + } + }, + "required": [ + "kind" + ] + }, { "type": "object", "additionalProperties": false, diff --git a/packages/microcosm-build/src/microcosm/build/us/release_input_coverage_manifest.json b/packages/microcosm-build/src/microcosm/build/us/release_input_coverage_manifest.json index 29fd5bf9..4445ad41 100644 --- a/packages/microcosm-build/src/microcosm/build/us/release_input_coverage_manifest.json +++ b/packages/microcosm-build/src/microcosm/build/us/release_input_coverage_manifest.json @@ -56,6 +56,9 @@ "cps_race": { "status": "required" }, + "detailed_industry_recode": { + "status": "required" + }, "detailed_occupation_recode": { "status": "required" }, @@ -268,6 +271,9 @@ "long_term_capital_gains_on_collectibles": { "status": "required" }, + "major_industry_recode": { + "status": "required" + }, "meets_ssi_disability_criteria": { "status": "required" }, @@ -523,6 +529,9 @@ "weeks_unemployed": { "status": "required" }, + "worked_last_year": { + "status": "required" + }, "workers_compensation": { "status": "required" }, @@ -534,9 +543,9 @@ } }, "counts": { - "required": 163, + "required": 166, "reviewed_exclusion": 7, - "total": 170 + "total": 173 }, "derivation": "Required surface = input columns in the pinned, sha-verified ecps_parity_reference.json populated layers, plus the documented post-reference fsla_overtime_premium, qualified_passenger_vehicle_loan_interest, five desired retirement-contribution inputs, meets_ssi_disability_criteria required by shipped validation probes, and the #282 Schedule-D capital-gain-distributions route leg schedule_d_capital_gain_distributions (PolicyEngine/microcosm#462). status='reviewed_exclusion' for ecps_parity_known_gaps.json entries (reason+issue from that register); EXCEPT every primary-source restoration pinned by RESTORED_REFERENCE_ECPS_REQUIRED_INPUTS (including the Section 199A QBI family), and the SSI countable-resource asset inputs (bank_account_assets, stock_assets, bond_assets), which are status='required' with NO exclusion per PolicyEngine/microcosm#368 so the gate fails on today's artifacts and asset restoration (Deliverable 2) turns it green. All other populated layers are 'required'. Regenerate with tools/build_us_release_input_coverage_manifest.py.", "description": "Declared full-coverage contract for a US release: every input column the reference eCPS exports must be persisted as a key with non-default signal, or carry a reviewed exclusion. Enforced as a hard release gate (microcosm.build.us_runtime.release_input_coverage) that generalizes assert_required_us_release_source_columns from 5 columns to the full eCPS input surface.", diff --git a/packages/microcosm-build/src/microcosm/build/us/source_stages.json b/packages/microcosm-build/src/microcosm/build/us/source_stages.json index b4f21562..bc7deda5 100644 --- a/packages/microcosm-build/src/microcosm/build/us/source_stages.json +++ b/packages/microcosm-build/src/microcosm/build/us/source_stages.json @@ -2907,6 +2907,77 @@ "health_insurance_premiums" ], "notes": "After ACA and Medicaid take-up inputs are materialized, the retired annual residual is max(PHIP_VAL - chip_premium - marketplace_net_premium - medicaid_premium, 0), with tax-unit modeled premiums assigned to the first person (archived commit 42ed5d45c56df80d754fbe24cce21cfeb8d05cbe, datasets/cps/cps.py lines 828-944). The archived CPS-only output list and exact eight-predictor, at-most-5,000-person joint QRF are extended_cps.py lines 135-194, 234-248, and 639-745, with only the PUF support half replaced at lines 1014-1076. Because the archive predicted the two PUF leaves jointly without clipping one to the other, PUF-only residual-order exceedances remain diagnostics; measured ASEC rows must preserve the exact residual identity. The archive did not explicitly pin tree count or weights; Microcosm pins 100 trees and typed person weights as reproducibility hardening. The final attribution operation (PolicyEngine/microcosm#451 item 2) is deterministic: it copies the finished reported non-Part-B premium onto the person health_insurance_premiums input exactly where combined Schedule C income (self_employment_income_before_lsr + sstb_self_employment_income_before_lsr) is strictly positive, the person is outside the Medicare proxy (age 65+ or any Social Security disability income), and the person is outside measured employer-sponsored coverage (has_esi - the conservative, incomplete proxy for the 26 USC 162(l)(2)(B) subsidized-employer-plan month exclusion; eligibility through a spouse's or dependent's employer is not measured and remains documented residual overbreadth), and marks is_self_employed for every strictly-positive Schedule C person. PolicyEngine-US 1.764.6 then computes self_employed_health_insurance_premiums (a formula-owned adds-aggregation gated by is_self_employed, so it cannot ship as a column) and the section 162(l) ALD as min(total_self_employment_income, premiums). The Medicare proxy keeps the statutory medical-expense premium concept numerically invariant because the direct input equals the decomposed reported premium whenever the modeled Part B add-on is zero. Beyond the federal ALD, PolicyEngine-US 1.764.6 reads the bare health_insurance_premiums input in five state/local formulas (MO qualified-premium subtraction, MI household resources, OH uninsured unreimbursed medical expenses, CA Orange County general relief countable income, MO child-care-subsidy adjusted income): attribution moves those programs from premiums-invisible toward the measured premium structure for the self-employed slice, outside the federal income_tax probe's scope. External anchor class: IRS SOI Pub 1304 Table 1.4 TY2023 self-employed health insurance deduction, 3,595,764 returns / $31.23B (ledger#105, buildn v9.2 feed)." + }, + { + "stage": "work_experience_inputs", + "survey": "Census CPS ASEC", + "source": "https://www.census.gov/programs-surveys/cps.html", + "grain": "person", + "artifacts": [ + { + "kind": "public_microdata", + "format": "zip_csv", + "vintage": "2023 ASEC / 2022 income reference year", + "locator": "https://www2.census.gov/programs-surveys/cps/datasets/2023/march/asecpub23csv.zip", + "sha256": "d2e000250782adfbdd7f29c82b66d866591a30f0d330496698ec19f9c784ce11", + "size_bytes": 150165063, + "member": "pppub23.csv", + "member_size_bytes": 281065733, + "member_crc32": "49c09e5f", + "member_sha256": "19b56537e50e7663f954361ef2bb5ce9cef8d9d45f156fe1a69a99b654198ffe" + }, + { + "kind": "public_microdata", + "format": "zip_csv", + "vintage": "2024 ASEC / 2023 income reference year", + "locator": "https://www2.census.gov/programs-surveys/cps/datasets/2024/march/asecpub24csv.zip", + "sha256": "cdb39cdac34bef99dd0940ab28e306f692404c2eea44d85dfd634214872a0a09", + "size_bytes": 148664101, + "member": "pppub24.csv", + "member_size_bytes": 277250415, + "member_crc32": "87950ece", + "member_sha256": "21a2b9e0e4b08534563578a45acad77868af4ae9a7d46f23776b707d4a559aa7" + }, + { + "kind": "public_microdata", + "format": "zip_csv", + "vintage": "2025 ASEC / 2024 income reference year", + "locator": "https://www2.census.gov/programs-surveys/cps/datasets/2025/march/asecpub25csv.zip", + "sha256": "318845a2b5e0034eb2973898de1738f4df0025727de38499e7669cb9c0deef0b", + "size_bytes": 147271429, + "member": "pppub25.csv", + "member_size_bytes": 277882549, + "member_crc32": "7dc2878f", + "member_sha256": "06921fe83fc66c907e6c7b86b82255dc70458ee7d76258fc48297cb34f0c06b5" + }, + { + "kind": "official_data_dictionary", + "format": "pdf", + "vintage": "2024", + "locator": "https://www2.census.gov/programs-surveys/cps/datasets/2024/march/asec2024_ddl_pub_full.pdf", + "lines": "person WEIND entry, position 326 length 2; WEMIND entry, position 329 length 2; WKSWORK/WORKYN work-experience universe statements" + } + ], + "operations": [ + { + "kind": "read_table", + "table": "person", + "weight": "person_weight" + }, + { + "kind": "derive_work_experience_inputs" + } + ], + "outputs": [ + "detailed_industry_recode", + "major_industry_recode", + "worked_last_year" + ], + "nonnegative_outputs": [ + "detailed_industry_recode", + "major_industry_recode" + ], + "notes": "Net-new factual-input coverage with no archived derivation to port: the retired eCPS build never surfaced industry or an explicit worked-last-year indicator (its cps.py carried only the POCCU2 occupation recode). WEIND ('IND. OF LONGEST JOB BY DETAILED GROUPS', position 326 length 2; 0 = NIU, 22 = Military, 23 = Never worked) and WEMIND ('IND. OF LONGEST JOB BY MAJOR IND. GROUPS', position 329 length 2), universe all persons 15+, restore from the three SHA-pinned official ASEC person archives via exact per-income-year PERIDNUM joins (the frozen census_cps inputs never carried them; identical zip/member identities to the education-assistance sidecar pins) and carry directly to both support clones through the shared source identity, exactly as the POCCU2 occupation recode carries. worked_last_year derives as WKSWORK > 0, the official universe condition of the work-experience recode block: measured on the pinned archives, WEIND holds a worker code 1-22 iff WKSWORK > 0 with zero violations (worked rows 73,186 / 73,471 / 72,460; A_FNLWGT-weighted worked shares 0.519968 / 0.523171 / 0.522794; nonzero-recode shares 0.821737 / 0.822002 / 0.824376 across income years 2022-2024), while WORKYN = 1 under-covers that universe by 642 / 572 / 594 allocation rows carrying positive weeks and a real industry (WORKYN = 1 never appears without positive WKSWORK), so the recode-universe definition is the indicator and WORKYN is load-time audit evidence only. The loader also enforces recode coherence ((WEIND = 0) iff (WEMIND = 0), zero violations measured) and the derive handler re-enforces both identities on the pool surface with zero tolerance. Downstream demand: PolicyEngine/microcosm#719 (Living Wage Institute MVP person schema: working indicator + industry + occupation on every person record). Industry-conditional modeling stays owned by PolicyEngine-US; this stage persists measured facts only." } ] } diff --git a/packages/microcosm-build/src/microcosm/build/us/spec/sources.yaml b/packages/microcosm-build/src/microcosm/build/us/spec/sources.yaml index fc74d6f8..e7f9511c 100644 --- a/packages/microcosm-build/src/microcosm/build/us/spec/sources.yaml +++ b/packages/microcosm-build/src/microcosm/build/us/spec/sources.yaml @@ -2881,3 +2881,72 @@ stages: toward the measured premium structure for the self-employed slice, outside the federal income_tax probe''s scope. External anchor class: IRS SOI Pub 1304 Table 1.4 TY2023 self-employed health insurance deduction, 3,595,764 returns / $31.23B (ledger#105, buildn v9.2 feed).' +- stage: work_experience_inputs + survey: Census CPS ASEC + source: https://www.census.gov/programs-surveys/cps.html + grain: person + artifacts: + - kind: public_microdata + format: zip_csv + vintage: 2023 ASEC / 2022 income reference year + locator: https://www2.census.gov/programs-surveys/cps/datasets/2023/march/asecpub23csv.zip + sha256: d2e000250782adfbdd7f29c82b66d866591a30f0d330496698ec19f9c784ce11 + size_bytes: 150165063 + member: pppub23.csv + member_size_bytes: 281065733 + member_crc32: 49c09e5f + member_sha256: 19b56537e50e7663f954361ef2bb5ce9cef8d9d45f156fe1a69a99b654198ffe + - kind: public_microdata + format: zip_csv + vintage: 2024 ASEC / 2023 income reference year + locator: https://www2.census.gov/programs-surveys/cps/datasets/2024/march/asecpub24csv.zip + sha256: cdb39cdac34bef99dd0940ab28e306f692404c2eea44d85dfd634214872a0a09 + size_bytes: 148664101 + member: pppub24.csv + member_size_bytes: 277250415 + member_crc32: 87950ece + member_sha256: 21a2b9e0e4b08534563578a45acad77868af4ae9a7d46f23776b707d4a559aa7 + - kind: public_microdata + format: zip_csv + vintage: 2025 ASEC / 2024 income reference year + locator: https://www2.census.gov/programs-surveys/cps/datasets/2025/march/asecpub25csv.zip + sha256: 318845a2b5e0034eb2973898de1738f4df0025727de38499e7669cb9c0deef0b + size_bytes: 147271429 + member: pppub25.csv + member_size_bytes: 277882549 + member_crc32: 7dc2878f + member_sha256: 06921fe83fc66c907e6c7b86b82255dc70458ee7d76258fc48297cb34f0c06b5 + - kind: official_data_dictionary + format: pdf + vintage: '2024' + locator: https://www2.census.gov/programs-surveys/cps/datasets/2024/march/asec2024_ddl_pub_full.pdf + lines: person WEIND entry, position 326 length 2; WEMIND entry, position 329 length 2; WKSWORK/WORKYN work-experience + universe statements + operations: + - kind: read_table + table: person + weight: person_weight + - kind: derive_work_experience_inputs + outputs: + - detailed_industry_recode + - major_industry_recode + - worked_last_year + nonnegative_outputs: + - detailed_industry_recode + - major_industry_recode + notes: 'Net-new factual-input coverage with no archived derivation to port: the retired eCPS build never surfaced industry + or an explicit worked-last-year indicator (its cps.py carried only the POCCU2 occupation recode). WEIND (''IND. OF LONGEST + JOB BY DETAILED GROUPS'', position 326 length 2; 0 = NIU, 22 = Military, 23 = Never worked) and WEMIND (''IND. OF LONGEST + JOB BY MAJOR IND. GROUPS'', position 329 length 2), universe all persons 15+, restore from the three SHA-pinned official + ASEC person archives via exact per-income-year PERIDNUM joins (the frozen census_cps inputs never carried them; identical + zip/member identities to the education-assistance sidecar pins) and carry directly to both support clones through the + shared source identity, exactly as the POCCU2 occupation recode carries. worked_last_year derives as WKSWORK > 0, the + official universe condition of the work-experience recode block: measured on the pinned archives, WEIND holds a worker + code 1-22 iff WKSWORK > 0 with zero violations (worked rows 73,186 / 73,471 / 72,460; A_FNLWGT-weighted worked shares + 0.519968 / 0.523171 / 0.522794; nonzero-recode shares 0.821737 / 0.822002 / 0.824376 across income years 2022-2024), while + WORKYN = 1 under-covers that universe by 642 / 572 / 594 allocation rows carrying positive weeks and a real industry (WORKYN + = 1 never appears without positive WKSWORK), so the recode-universe definition is the indicator and WORKYN is load-time + audit evidence only. The loader also enforces recode coherence ((WEIND = 0) iff (WEMIND = 0), zero violations measured) + and the derive handler re-enforces both identities on the pool surface with zero tolerance. Downstream demand: PolicyEngine/microcosm#719 + (Living Wage Institute MVP person schema: working indicator + industry + occupation on every person record). Industry-conditional + modeling stays owned by PolicyEngine-US; this stage persists measured facts only.' diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py b/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py index e55690cb..0d52631f 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py @@ -1061,6 +1061,26 @@ us_wic_claim_summary, with_us_wic_claim_input, ) +from microcosm.build.us_runtime.work_experience_inputs import ( + US_WORK_EXPERIENCE_NONCONSTANT_PERSON_COLUMNS, + US_WORK_EXPERIENCE_OUTPUT_COLUMNS, + US_WORK_EXPERIENCE_REQUIRED_SOURCE_COLUMNS, + US_WORK_EXPERIENCE_STAGE_NAME, + derive_us_work_experience_inputs_from_manifest, + us_work_experience_signal_gate, + us_work_experience_stage_spec, + us_work_experience_summary, + with_us_work_experience_inputs, +) +from microcosm.build.us_runtime.work_experience_source import ( + ASEC_WORK_EXPERIENCE_ARCHIVES, + ASEC_WORK_EXPERIENCE_INCOME_YEARS, + ASEC_WORK_EXPERIENCE_SOURCE_COLUMNS, + WORK_EXPERIENCE_OFFICIAL_DICTIONARY_URL, + fetch_asec_work_experience_source, + fill_asec_work_experience_source, + load_asec_work_experience_sources, +) from microcosm.build.us_runtime.workers_compensation import ( US_WORKERS_COMPENSATION_NONCONSTANT_PERSON_COLUMNS, US_WORKERS_COMPENSATION_OUTPUT_COLUMNS, @@ -1403,6 +1423,22 @@ "us_wic_claim_stage_spec", "us_wic_claim_summary", "with_us_wic_claim_input", + "ASEC_WORK_EXPERIENCE_ARCHIVES", + "ASEC_WORK_EXPERIENCE_INCOME_YEARS", + "ASEC_WORK_EXPERIENCE_SOURCE_COLUMNS", + "US_WORK_EXPERIENCE_NONCONSTANT_PERSON_COLUMNS", + "US_WORK_EXPERIENCE_OUTPUT_COLUMNS", + "US_WORK_EXPERIENCE_REQUIRED_SOURCE_COLUMNS", + "US_WORK_EXPERIENCE_STAGE_NAME", + "WORK_EXPERIENCE_OFFICIAL_DICTIONARY_URL", + "derive_us_work_experience_inputs_from_manifest", + "fetch_asec_work_experience_source", + "fill_asec_work_experience_source", + "load_asec_work_experience_sources", + "us_work_experience_signal_gate", + "us_work_experience_stage_spec", + "us_work_experience_summary", + "with_us_work_experience_inputs", "US_MISC_ITEMIZED_NONCONSTANT_PERSON_COLUMNS", "US_MISC_ITEMIZED_OUTPUT_COLUMNS", "US_MISC_ITEMIZED_STAGE_NAME", @@ -2355,6 +2391,17 @@ def to_manifest(self) -> dict[str, object]: "educational assistance carries directly from ASEC ED_VAL." ), ), + US_WORK_EXPERIENCE_STAGE_NAME: DonorSpec( + survey="Census CPS ASEC", + source="https://www.census.gov/programs-surveys/cps.html", + notes=( + "Industry of the longest job (WEIND detailed groups, WEMIND major " + "groups) restores from the three SHA-pinned official ASEC person " + "archives by exact PERIDNUM join and carries directly; " + "worked_last_year is WKSWORK > 0, the work-experience recode " + "universe. Nothing is imputed (microcosm#719)." + ), + ), US_RETIREMENT_CONTRIBUTION_STAGE_NAME: DonorSpec( survey="Census CPS ASEC + published retirement-contribution shares", source="https://www.census.gov/programs-surveys/cps.html", @@ -2437,6 +2484,7 @@ def to_manifest(self) -> dict[str, object]: US_WORKERS_COMPENSATION_STAGE_NAME, US_WEEKS_UNEMPLOYED_STAGE_NAME, US_EDUCATION_INPUTS_STAGE_NAME, + US_WORK_EXPERIENCE_STAGE_NAME, "capital_gain_distributions", "scf_wealth", US_SSI_DISABILITY_CRITERIA_STAGE_NAME, diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/asec_checkpoint.py b/packages/microcosm-build/src/microcosm/build/us_runtime/asec_checkpoint.py index 0c6ebe0f..b44fc86e 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/asec_checkpoint.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/asec_checkpoint.py @@ -60,7 +60,11 @@ # Version 3 added the PAW_TYP restoration that gates TANF enrollment; older # artifacts lack the gate column and must fail loudly rather than let # PAW_VAL-only conflation back in (microcosm#591). -ASEC_RAW_STAGE_SCHEMA_VERSION = 3 +# Version 4 added the WEIND/WEMIND work-experience industry restoration +# feeding detailed_industry_recode, major_industry_recode, and +# worked_last_year; older artifacts lack the industry columns and must fail +# loudly rather than ship an industry-blind person schema (microcosm#719). +ASEC_RAW_STAGE_SCHEMA_VERSION = 4 ASEC_RAW_STAGE_STAGE = "raw_source_mapping" _RAW_STAGE_BINDING_KEYS = frozenset( { @@ -75,9 +79,11 @@ "stage", } ) -_RAW_SOURCE_MAPPING_COLUMNS = frozenset({"ED_VAL", "LKWEEKS", "PAW_TYP"}) +_RAW_SOURCE_MAPPING_COLUMNS = frozenset( + {"ED_VAL", "LKWEEKS", "PAW_TYP", "WEIND", "WEMIND"} +) _RAW_STAGE_REQUIRED_PERSON_COLUMNS = frozenset( - {"ED_VAL", "LKWEEKS", "PAW_TYP", "PERIDNUM", "source_year"} + {"ED_VAL", "LKWEEKS", "PAW_TYP", "PERIDNUM", "WEIND", "WEMIND", "source_year"} ) _RAW_SOURCE_MAPPING_KEYS = frozenset( { @@ -458,6 +464,35 @@ def _validate_raw_stage_source_columns(frame: Frame, *, path: Path) -> None: "in {0, 1, 2, 3}." ) + detailed_industry = pd.to_numeric(person["WEIND"], errors="coerce").to_numpy( + dtype=np.float64 + ) + valid_detailed = np.isfinite(detailed_industry) + valid_detailed &= np.equal(detailed_industry, np.floor(detailed_industry)) + valid_detailed &= (detailed_industry >= 0.0) & (detailed_industry <= 23.0) + if not valid_detailed.all(): + raise ValueError( + f"ASEC raw-stage checkpoint {path} WEIND must be complete integers " + "in {0, ..., 23}." + ) + + major_industry = pd.to_numeric(person["WEMIND"], errors="coerce").to_numpy( + dtype=np.float64 + ) + valid_major = np.isfinite(major_industry) + valid_major &= np.equal(major_industry, np.floor(major_industry)) + valid_major &= (major_industry >= 0.0) & (major_industry <= 15.0) + if not valid_major.all(): + raise ValueError( + f"ASEC raw-stage checkpoint {path} WEMIND must be complete integers " + "in {0, ..., 15}." + ) + if int(np.count_nonzero((detailed_industry == 0.0) != (major_industry == 0.0))): + raise ValueError( + f"ASEC raw-stage checkpoint {path} WEIND and WEMIND must be zero " + "on exactly the same not-in-universe rows." + ) + def _validate_asec_frame( frame: Frame, diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/l0_refit_export.py b/packages/microcosm-build/src/microcosm/build/us_runtime/l0_refit_export.py index 80adf18a..3c560859 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/l0_refit_export.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/l0_refit_export.py @@ -137,6 +137,9 @@ from microcosm.build.us_runtime.wic_claim import ( US_WIC_CLAIM_NONCONSTANT_PERSON_COLUMNS, ) +from microcosm.build.us_runtime.work_experience_inputs import ( + US_WORK_EXPERIENCE_NONCONSTANT_PERSON_COLUMNS, +) from microcosm.build.us_runtime.workers_compensation import ( US_WORKERS_COMPENSATION_NONCONSTANT_PERSON_COLUMNS, ) @@ -175,6 +178,7 @@ *US_EDUCATOR_EXPENSE_NONCONSTANT_PERSON_COLUMNS, *US_MISC_ITEMIZED_NONCONSTANT_PERSON_COLUMNS, *US_EDUCATION_INPUTS_NONCONSTANT_PERSON_COLUMNS, + *US_WORK_EXPERIENCE_NONCONSTANT_PERSON_COLUMNS, *US_RETIREMENT_CONTRIBUTION_NONCONSTANT_PERSON_COLUMNS, *US_RETIREMENT_DISTRIBUTION_NONCONSTANT_PERSON_COLUMNS, *US_QBI_NONCONSTANT_PERSON_COLUMNS, diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/multispine_pool.py b/packages/microcosm-build/src/microcosm/build/us_runtime/multispine_pool.py index 3b67636b..6762af7a 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/multispine_pool.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/multispine_pool.py @@ -118,6 +118,9 @@ ) from microcosm.build.us_runtime.weeks_unemployed import with_us_weeks_unemployed from microcosm.build.us_runtime.wic_claim import with_us_wic_claim_input +from microcosm.build.us_runtime.work_experience_inputs import ( + with_us_work_experience_inputs, +) from microcosm.build.us_runtime.workers_compensation import ( with_us_workers_compensation, ) @@ -222,6 +225,7 @@ "with_us_retirement_distribution_inputs", "with_us_immigration_inputs", "with_us_education_inputs", + "with_us_work_experience_inputs", ) """Logical source-input ownership inventory in legacy relative order. @@ -594,6 +598,12 @@ class SourceOperatorContract: (_POST_CLONE_PHASE,), "deterministic rowwise derivation follows PUF tuition imputation", ), + "with_us_work_experience_inputs": SourceOperatorContract( + "work_experience_inputs", + (_POST_CLONE_PHASE,), + "exact sidecar identity join must cover both support clones of every " + "source person", + ), "_complete_schedule_d_input": SourceOperatorContract( "capital_gain_distributions", (_POST_CLONE_PHASE,), @@ -1952,6 +1962,14 @@ def _post_clone_source_operators() -> Mapping[str, SourceFrameOperator]: time_period=POOL_TIME_PERIOD, asec_education_source=None, ), + "with_us_work_experience_inputs": lambda current: ( + with_us_work_experience_inputs( + current, + seed=POOL_RANDOM_SEED, + time_period=POOL_TIME_PERIOD, + asec_work_experience_source=None, + ) + ), } return operators diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/operator_boundary.py b/packages/microcosm-build/src/microcosm/build/us_runtime/operator_boundary.py index 6023ef75..22dc3375 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/operator_boundary.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/operator_boundary.py @@ -91,6 +91,9 @@ US_WEEKS_UNEMPLOYED_OUTPUT_COLUMNS, ) from microcosm.build.us_runtime.wic_claim import US_WIC_CLAIM_OUTPUT_COLUMNS +from microcosm.build.us_runtime.work_experience_inputs import ( + US_WORK_EXPERIENCE_OUTPUT_COLUMNS, +) from microcosm.build.us_runtime.workers_compensation import ( US_WORKERS_COMPENSATION_OUTPUT_COLUMNS, ) @@ -334,6 +337,9 @@ "education_inputs": { "person": frozenset(US_EDUCATION_INPUTS_OWNED_OUTPUT_COLUMNS), }, + "work_experience_inputs": { + "person": frozenset(US_WORK_EXPERIENCE_OUTPUT_COLUMNS), + }, "scf_wealth": { "person": frozenset(US_SCF_FINANCIAL_ASSET_OUTPUT_COLUMNS), "household": frozenset(US_SCF_NET_WORTH_OUTPUT_COLUMNS), diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/release_input_coverage.py b/packages/microcosm-build/src/microcosm/build/us_runtime/release_input_coverage.py index 40d983c3..cdaad657 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/release_input_coverage.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/release_input_coverage.py @@ -148,6 +148,13 @@ "pre_subsidy_care_expenses", "is_incapable_of_self_care", "health_insurance_premiums", + # PolicyEngine/microcosm#719: the work-experience stage's measured + # person attributes (industry of longest job, worked-last-year). + # The reference artifact predates the stage; the committed person + # schema carries them on every release, so absence is a red gate. + "detailed_industry_recode", + "major_industry_recode", + "worked_last_year", "is_self_employed", } ) diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/source_runtime.py b/packages/microcosm-build/src/microcosm/build/us_runtime/source_runtime.py index fd0374c9..0ce1477c 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/source_runtime.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/source_runtime.py @@ -90,6 +90,9 @@ impute_us_weeks_unemployed_to_puf_support_from_manifest, ) from microcosm.build.us_runtime.wic_claim import derive_us_wic_claim_from_manifest +from microcosm.build.us_runtime.work_experience_inputs import ( + derive_us_work_experience_inputs_from_manifest, +) from microcosm.build.us_runtime.workers_compensation import ( derive_us_workers_compensation_from_manifest, impute_us_workers_compensation_to_puf_support_from_manifest, @@ -313,6 +316,9 @@ def us_source_operation_handlers() -> Mapping[str, SourceOperationHandler]: ), "derive_eligibility_inputs": derive_us_eligibility_inputs_from_manifest, "derive_education_inputs": derive_us_education_inputs_from_manifest, + "derive_work_experience_inputs": ( + derive_us_work_experience_inputs_from_manifest + ), "derive_hours_worked": derive_us_hours_worked_from_manifest, "derive_medicare_take_up": derive_us_medicare_take_up_from_manifest, "derive_pregnancy": derive_us_pregnancy_from_manifest, diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/stacked_spine.py b/packages/microcosm-build/src/microcosm/build/us_runtime/stacked_spine.py index 6b4ffc1f..b6647846 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/stacked_spine.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/stacked_spine.py @@ -2919,6 +2919,11 @@ def _terminal_surface_from_pool_registry() -> TargetFamilies: "source_operator_wic_claim", "would_claim_wic", ), + ( + "person", + "source_operator_work_experience_inputs", + "worked_last_year", + ), ("person", "take_up", "takes_up_basic_health_program_if_eligible"), ("person", "take_up", "takes_up_chip_if_eligible"), ("person", "take_up", "takes_up_early_head_start_if_eligible"), @@ -2951,6 +2956,16 @@ def _terminal_surface_from_pool_registry() -> TargetFamilies: "immigration_status_str", ), ("person", "source_operator_immigration", "ssn_card_type"), + ( + "person", + "source_operator_work_experience_inputs", + "detailed_industry_recode", + ), + ( + "person", + "source_operator_work_experience_inputs", + "major_industry_recode", + ), ( "tax_unit", "puf_tax_itemization", @@ -3759,7 +3774,7 @@ def validate_stacked_post_puf_transfer_receipt( if not isinstance(targets, Mapping) or set(targets) != expected_target_labels: raise ValueError( f"{boundary}: stacked post-PUF transfer target surface is not the " - "canonical 70-target surface; production manifest emission is " + "canonical 73-target surface; production manifest emission is " "forbidden." ) if any( @@ -3774,8 +3789,8 @@ def validate_stacked_post_puf_transfer_receipt( completion = receipt.get("completion") if completion != { "status": "complete", - "group_count": 19, - "target_count": 70, + "group_count": 20, + "target_count": 73, "residual_null_rows": 0, }: raise ValueError( @@ -5014,6 +5029,7 @@ def stacked_late_primary_checkpoint_input_binding( "with_us_retirement_distribution_inputs": "retirement_distributions", "with_us_immigration_inputs": "immigration_status", "with_us_education_inputs": "education_inputs", + "with_us_work_experience_inputs": "work_experience_inputs", } ) _SOURCE_STAGE_SPEC_RESOLVER_BY_OPERATOR: Mapping[str, tuple[str, str]] = ( @@ -5079,13 +5095,17 @@ def stacked_late_primary_checkpoint_input_binding( "microcosm.build.us_runtime.education_inputs", "us_education_inputs_stage_spec", ), + "with_us_work_experience_inputs": ( + "microcosm.build.us_runtime.work_experience_inputs", + "us_work_experience_stage_spec", + ), } ) ) _DIRECT_HOUSING_ASSISTANCE_SOURCE_OPERATOR = ( "impute_us_housing_assistance_to_puf_support" ) -if len(_SOURCE_MANIFEST_STAGE_BY_OPERATOR) != 15 or set( +if len(_SOURCE_MANIFEST_STAGE_BY_OPERATOR) != 16 or set( _SOURCE_MANIFEST_STAGE_BY_OPERATOR ) | {_DIRECT_HOUSING_ASSISTANCE_SOURCE_OPERATOR} != set( POOL_POST_CLONE_SOURCE_OPERATOR_ORDER @@ -5274,6 +5294,8 @@ def _late_source_execution_config_binding( if operator == "with_us_weeks_unemployed" else {"asec_education_source": {"mode": "not_supplied"}} if operator == "with_us_education_inputs" + else {"asec_work_experience_source": {"mode": "not_supplied"}} + if operator == "with_us_work_experience_inputs" else {} ), "source_stage_spec": _late_source_stage_spec_binding(operator), @@ -7493,12 +7515,12 @@ def validate_structural_absence_receipt( _surface_target_keys(_freeze_target_families(pool_transfer_target_families())) ) if ( - len(_canonical_surface_keys) != 131 - or len(set(_canonical_surface_keys)) != 131 + len(_canonical_surface_keys) != 134 + or len(set(_canonical_surface_keys)) != 134 or len(_canonical_early_transfer_keys) != 48 - or len(_canonical_late_transfer_keys) != 70 + or len(_canonical_late_transfer_keys) != 73 or len(_canonical_late_puf_producer_keys) != 43 - or len(_canonical_late_source_producer_keys) != 29 + or len(_canonical_late_source_producer_keys) != 32 or len(_canonical_late_puf_producer_keys & _canonical_late_source_producer_keys) != 2 or _canonical_late_puf_producer_keys | _canonical_late_source_producer_keys @@ -7506,7 +7528,7 @@ def validate_structural_absence_receipt( or _canonical_early_transfer_keys & _canonical_late_transfer_keys or _canonical_early_transfer_keys | _canonical_late_transfer_keys != _canonical_full_transfer_keys - or len(_canonical_full_transfer_keys) != 118 + or len(_canonical_full_transfer_keys) != 121 or set(_plan_target_keys(_CANONICAL_STACKED_GAP_FILL_PLAN_ANCHOR)) != _canonical_early_transfer_keys or not _canonical_full_transfer_keys.issubset(_canonical_surface_keys) @@ -7526,10 +7548,10 @@ def validate_structural_absence_receipt( ) ): raise RuntimeError( - "Canonical stacked authority must partition the exact 118-target " - "transfer surface into 48 early gap-fill and 70 post-PUF targets " - "inside an exact 131-target terminal surface and metric registry; " - "the late surface must be exactly covered by 43 PUF-clone and 29 " + "Canonical stacked authority must partition the exact 121-target " + "transfer surface into 48 early gap-fill and 73 post-PUF targets " + "inside an exact 134-target terminal surface and metric registry; " + "the late surface must be exactly covered by 43 PUF-clone and 32 " "ASEC-source producer targets with their declared two-target overlap." ) @@ -9063,7 +9085,7 @@ def _aggregate_late_transfer_result( ], execution_order: Sequence[str], ) -> StackedPostPufTransferResult: - """Bind all bounded group outcomes into the canonical 70-target receipt.""" + """Bind all bounded group outcomes into the canonical 73-target receipt.""" expected_groups = tuple(group.name for group in CANONICAL_US_LATE_TRANSFER_GROUPS) if set(group_results) != set(expected_groups): @@ -11314,7 +11336,7 @@ class OriginBatterySpec: """Test-seam grouping for per-column battery metrics. Production never accepts these specs from a caller: it consumes the - immutable 131-column canonical registry. The explicit test-authority seam + immutable 134-column canonical registry. The explicit test-authority seam groups its digested registry into specs so the comparison engine can reuse the same loop. ``clone_index`` scopes a fixture comparison to one clone role: 0 compares native rows and 1 compares a PUF arm. @@ -11368,7 +11390,7 @@ def by_origin_battery( *, tail_manifest: Mapping[str, object] | None = None, ) -> GateResult: - """Run the canonical 131-target plus joint by-origin battery.""" + """Run the canonical 134-target plus joint by-origin battery.""" return _by_origin_battery_evaluate( frame, diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/us_late_producer_registry.py b/packages/microcosm-build/src/microcosm/build/us_runtime/us_late_producer_registry.py index 4a6c9b78..28d489d7 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/us_late_producer_registry.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/us_late_producer_registry.py @@ -186,6 +186,7 @@ "is_pursuing_credential_for_american_opportunity_credit", "takes_up_medicare_if_eligible", "would_claim_wic", + "worked_last_year", } ) @@ -992,6 +993,22 @@ def _inventory( _single("person_id", "person", "person_id"), _single("resolved_person_weight", "person", "@resolved_weight"), ), + "with_us_work_experience_inputs": _inventory( + "with_us_work_experience_inputs", + _requirement( + "industry_source", + ( + _column("person", "WEIND", value_kind="finite_numeric"), + _column("person", "WEMIND", value_kind="finite_numeric"), + ), + ), + _requirement( + "weeks_source", + (_column("person", "WKSWORK", value_kind="finite_numeric"),), + ), + _single("person_id", "person", "person_id"), + _single("resolved_person_weight", "person", "@resolved_weight"), + ), } _source_input_inventories = { @@ -1395,25 +1412,25 @@ def _bounded_transfer_groups( max_targets_per_fit=_DEFAULT_MAX_TARGETS_PER_FIT, ) if ( - len(CANONICAL_US_LATE_TRANSFER_GROUPS) != 19 - or sum(len(group.targets) for group in CANONICAL_US_LATE_TRANSFER_GROUPS) != 70 + len(CANONICAL_US_LATE_TRANSFER_GROUPS) != 20 + or sum(len(group.targets) for group in CANONICAL_US_LATE_TRANSFER_GROUPS) != 73 ): raise RuntimeError( - "Canonical US late transfer must contain exactly 19 bounded groups " - "and 70 ordered targets." + "Canonical US late transfer must contain exactly 20 bounded groups " + "and 73 ordered targets." ) _canonical_late_targets = { target for group in CANONICAL_US_LATE_TRANSFER_GROUPS for target in group.targets } if ( - len(_BOOLEAN_LATE_TARGETS) != 17 + len(_BOOLEAN_LATE_TARGETS) != 18 or len(_STRING_LATE_TARGETS) != 2 or not (_BOOLEAN_LATE_TARGETS | _STRING_LATE_TARGETS) <= _canonical_late_targets - or len(_canonical_late_targets - _BOOLEAN_LATE_TARGETS - _STRING_LATE_TARGETS) != 51 + or len(_canonical_late_targets - _BOOLEAN_LATE_TARGETS - _STRING_LATE_TARGETS) != 53 ): raise RuntimeError( - "Canonical US late target kinds must partition 70 targets into " - "51 numeric, 17 boolean, and 2 string inputs." + "Canonical US late target kinds must partition 73 targets into " + "53 numeric, 18 boolean, and 2 string inputs." ) US_LATE_TRANSFER_INPUT_INVENTORIES: Mapping[str, SourceInputInventory] = ( MappingProxyType( diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/work_experience_inputs.py b/packages/microcosm-build/src/microcosm/build/us_runtime/work_experience_inputs.py new file mode 100644 index 00000000..bdebbdeb --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/work_experience_inputs.py @@ -0,0 +1,403 @@ +"""Work-experience industry recodes and the worked-last-year indicator. + +The retired eCPS build carried the occupation of the longest job +(``POCCU2`` -> ``detailed_occupation_recode``) but never surfaced its +industry siblings or an explicit worked-last-year indicator. This stage is +net-new factual-input coverage with no archived derivation to port: the +official ASEC work-experience recodes ``WEIND`` (industry of longest job by +detailed groups, 0--23) and ``WEMIND`` (industry of longest job by major +industry groups, 0--15) carry directly, and ``worked_last_year`` derives as +``WKSWORK > 0`` — the official universe condition of the work-experience +longest-job recode block, verified exact against every pinned archive +(``WEIND`` holds a worker code 1--22 iff ``WKSWORK > 0``; ``WORKYN = 1`` +under-covers that universe by ~600 allocation rows per year and is +deliberately not the indicator). + +The frozen census_cps inputs never carried ``WEIND``/``WEMIND``, so the +pinned work-experience sidecar (:mod:`.work_experience_source`) restores +them for every pooled income year via exact ``PERIDNUM`` joins before the +derivation runs; ``WKSWORK`` is a frozen census_cps person column. Both +support clones of one source person inherit identical values through the +shared source identity, matching how ``detailed_occupation_recode`` treats +the PUF tax-detail half. Industry-conditional modeling stays owned by +PolicyEngine-US; this stage persists measured facts only. +""" + +from __future__ import annotations + +from importlib.resources import files + +import numpy as np +import pandas as pd + +from microcosm.build.gates import GateResult +from microcosm.build.source_manifest import ( + SourceOperationSpec, + SourceStageSpec, + load_source_manifest, +) +from microcosm.build.source_runtime import ( + SourceRuntimeConfig, + SourceRuntimeContext, + SourceRuntimeError, + run_source_stage, +) +from microcosm.build.us_runtime.support_provenance import ( + BASE_ASEC_SUPPORT_CHANNEL, + PUF_TAX_DETAIL_SUPPORT_CHANNEL, + has_support_role_metadata, + support_role_series, +) +from microcosm.build.us_runtime.work_experience_source import ( + fill_asec_work_experience_source, +) +from microcosm.frame import Frame +from microcosm.frame.units import US_SCHEMA + +__all__ = [ + "US_WORK_EXPERIENCE_NONCONSTANT_PERSON_COLUMNS", + "US_WORK_EXPERIENCE_OUTPUT_COLUMNS", + "US_WORK_EXPERIENCE_REQUIRED_SOURCE_COLUMNS", + "US_WORK_EXPERIENCE_STAGE_NAME", + "derive_us_work_experience_inputs_from_manifest", + "us_work_experience_signal_gate", + "us_work_experience_stage_spec", + "us_work_experience_summary", + "with_us_work_experience_inputs", +] + +US_WORK_EXPERIENCE_STAGE_NAME = "work_experience_inputs" + +_DETAILED_OUTPUT = "detailed_industry_recode" +_MAJOR_OUTPUT = "major_industry_recode" +_WORKED_OUTPUT = "worked_last_year" + +US_WORK_EXPERIENCE_OUTPUT_COLUMNS: tuple[str, ...] = ( + _DETAILED_OUTPUT, + _MAJOR_OUTPUT, + _WORKED_OUTPUT, +) +US_WORK_EXPERIENCE_NONCONSTANT_PERSON_COLUMNS = US_WORK_EXPERIENCE_OUTPUT_COLUMNS + +_DETAILED_SOURCE = "WEIND" +_MAJOR_SOURCE = "WEMIND" +_WEEKS_SOURCE = "WKSWORK" + +US_WORK_EXPERIENCE_REQUIRED_SOURCE_COLUMNS: tuple[str, ...] = ( + _DETAILED_SOURCE, + _MAJOR_SOURCE, + _WEEKS_SOURCE, +) + +_PERSON_WEIGHT_COLUMN = "person_weight" +_DETAILED_MAX = 23 +_DETAILED_WORKER_MAX = 22 +_DETAILED_MILITARY_CODE = 22 +_DETAILED_NEVER_WORKED_CODE = 23 +_MAJOR_MAX = 15 + +# Deliberately broad pool-level plausibility bands. The pinned official +# archives measure A_FNLWGT-weighted worked (WKSWORK > 0) shares of +# 0.5200/0.5232/0.5228 and nonzero-recode shares of 0.8217/0.8220/0.8244 +# across income years 2022-2024; the pool reweights and clones that +# population without changing either concept, so these floors reject a +# defaulted or collapsed surface while allowing support selection and +# adjacent ASEC vintages ample room to move. +_WORKED_SHARE_BAND = (0.40, 0.62) +_RECODE_SHARE_BAND = (0.70, 0.92) +_DERIVE_WORK_EXPERIENCE_PARAMETER_KEYS = frozenset() + + +def us_work_experience_stage_spec() -> SourceStageSpec: + """Load the packaged ``work_experience_inputs`` source-stage declaration.""" + + manifest = load_source_manifest( + files("microcosm.build.us").joinpath("source_stages.json") + ) + stage_map = manifest.stage_map() + if US_WORK_EXPERIENCE_STAGE_NAME not in stage_map: + raise ValueError( + f"US source manifest declares no {US_WORK_EXPERIENCE_STAGE_NAME!r} stage." + ) + spec = stage_map[US_WORK_EXPERIENCE_STAGE_NAME] + missing = sorted(set(US_WORK_EXPERIENCE_OUTPUT_COLUMNS) - set(spec.outputs)) + if missing: + raise ValueError( + f"{US_WORK_EXPERIENCE_STAGE_NAME!r} manifest stage does not declare " + f"output(s) {missing}; the runtime and manifest have drifted." + ) + return spec + + +def _bounded_integer_source( + frame: pd.DataFrame, column: str, *, upper: int +) -> np.ndarray: + values = pd.to_numeric(frame[column], errors="coerce").to_numpy(dtype=np.float64) + valid = np.isfinite(values) & (values == np.floor(values)) + valid &= (values >= 0.0) & (values <= float(upper)) + if not valid.all(): + rows = np.flatnonzero(~valid)[:5].tolist() + raise SourceRuntimeError( + f"US work-experience source {column!r} must be an integer in " + f"[0, {upper}] at row(s): {rows}." + ) + return values.astype(np.int64) + + +def derive_us_work_experience_inputs_from_manifest( + frame: pd.DataFrame | None, + operation: SourceOperationSpec, + _context: SourceRuntimeContext | None, +) -> pd.DataFrame: + """Carry the industry recodes and derive the worked-last-year indicator.""" + + if operation.kind != "derive_work_experience_inputs": + raise SourceRuntimeError( + "US work-experience derivation received unexpected operation " + f"{operation.kind!r}." + ) + if frame is None: + raise SourceRuntimeError( + "US work-experience derivation requires the person table to be read first." + ) + unexpected = sorted( + set(operation.parameters) - _DERIVE_WORK_EXPERIENCE_PARAMETER_KEYS + ) + if unexpected: + raise SourceRuntimeError( + "US work-experience derivation received unsupported parameter(s): " + f"{unexpected}." + ) + missing = [ + column + for column in US_WORK_EXPERIENCE_REQUIRED_SOURCE_COLUMNS + if column not in frame.columns + ] + if missing: + raise SourceRuntimeError( + f"US work-experience derivation requires source column(s): {missing}." + ) + + detailed = _bounded_integer_source(frame, _DETAILED_SOURCE, upper=_DETAILED_MAX) + major = _bounded_integer_source(frame, _MAJOR_SOURCE, upper=_MAJOR_MAX) + weeks = _bounded_integer_source(frame, _WEEKS_SOURCE, upper=52) + worked = weeks > 0 + worker_code = (detailed >= 1) & (detailed <= _DETAILED_WORKER_MAX) + universe_breaks = int(np.count_nonzero(worker_code != worked)) + if universe_breaks: + raise SourceRuntimeError( + "US work-experience derivation breaks the recode universe " + f"identity (WEIND in 1..{_DETAILED_WORKER_MAX} iff WKSWORK > 0) " + f"on {universe_breaks} row(s)." + ) + recode_zero_breaks = int(np.count_nonzero((detailed == 0) != (major == 0))) + if recode_zero_breaks: + raise SourceRuntimeError( + "US work-experience derivation finds the detailed and major " + f"recodes disagreeing on not-in-universe rows: " + f"{recode_zero_breaks} row(s)." + ) + result = frame.copy(deep=True) + result[_DETAILED_OUTPUT] = detailed.astype(np.int16) + result[_MAJOR_OUTPUT] = major.astype(np.int16) + result[_WORKED_OUTPUT] = worked + return result + + +def with_us_work_experience_inputs( + frame: Frame, + *, + seed: int, + time_period: int, + asec_work_experience_source: pd.DataFrame | None = None, +) -> Frame: + """Materialize work-experience inputs on a US frame. + + The frozen census_cps inputs never carried raw ASEC ``WEIND`` or + ``WEMIND``, so when the person table lacks them the pinned + work-experience sidecar (:mod:`.work_experience_source`) must be + supplied; the fill is an exact per-income-year ``PERIDNUM`` join and + never predicts a value. ``WKSWORK`` is a frozen census_cps person + column and must already be present. + """ + + if frame.schema != US_SCHEMA: + raise ValueError("US work-experience inputs require the US schema.") + if _work_experience_surface_carries_signal(frame): + return frame + + person = frame.table("person") + stage_person = person.copy(deep=True) + if _WEEKS_SOURCE not in stage_person.columns: + raise SourceRuntimeError( + "US work-experience stage requires the frozen census_cps person " + f"column {_WEEKS_SOURCE!r}." + ) + if any( + column not in stage_person.columns + for column in (_DETAILED_SOURCE, _MAJOR_SOURCE) + ): + if asec_work_experience_source is None: + raise SourceRuntimeError( + "US work-experience stage requires the pinned ASEC " + "work-experience sidecar to restore WEIND/WEMIND (the frozen " + "census_cps inputs never carried them); pass " + "--asec-work-experience-source or allow the official fetch." + ) + stage_person = fill_asec_work_experience_source( + stage_person, asec_work_experience_source + ) + stage_person[_PERSON_WEIGHT_COLUMN] = frame.resolve_weights("person").values + output = run_source_stage( + us_work_experience_stage_spec(), + tables={"person": stage_person}, + operation_handlers={ + "derive_work_experience_inputs": ( + derive_us_work_experience_inputs_from_manifest + ), + }, + config=SourceRuntimeConfig(seed=int(seed), target_year=int(time_period)), + ) + aligned = output.set_index("person_id").reindex(person["person_id"]) + for column in US_WORK_EXPERIENCE_OUTPUT_COLUMNS: + if aligned[column].isna().any(): + raise ValueError( + "US work-experience stage output does not cover every person " + f"for {column!r}." + ) + + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + tables["person"][_DETAILED_OUTPUT] = aligned[_DETAILED_OUTPUT].to_numpy( + dtype=np.int16 + ) + tables["person"][_MAJOR_OUTPUT] = aligned[_MAJOR_OUTPUT].to_numpy(dtype=np.int16) + tables["person"][_WORKED_OUTPUT] = aligned[_WORKED_OUTPUT].to_numpy(dtype=bool) + return Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + frame.strata, + mass_log=frame.mass_log, + metadata=frame.metadata, + ) + + +def us_work_experience_summary(frame: Frame) -> dict[str, object]: + """Return weighted signal and coherence diagnostics for the stage.""" + + person = frame.table("person") + weights = np.asarray(frame.resolve_weights("person").values, dtype=np.float64) + total_weight = float(weights.sum()) + detailed = pd.to_numeric(person[_DETAILED_OUTPUT], errors="coerce").to_numpy( + dtype=np.float64 + ) + major = pd.to_numeric(person[_MAJOR_OUTPUT], errors="coerce").to_numpy( + dtype=np.float64 + ) + worked = person[_WORKED_OUTPUT].fillna(False).astype(bool).to_numpy() + worker_code = (detailed >= 1.0) & (detailed <= float(_DETAILED_WORKER_MAX)) + recode_positive = detailed != 0.0 + + def _share(mask: np.ndarray) -> float: + return float(weights[mask].sum()) / total_weight if total_weight > 0 else 0.0 + + channels: dict[str, dict[str, float | int]] = {} + if has_support_role_metadata(person, entity="person"): + channel = support_role_series(person, entity="person").to_numpy() + for name in ( + BASE_ASEC_SUPPORT_CHANNEL, + PUF_TAX_DETAIL_SUPPORT_CHANNEL, + ): + mask = channel == name + channel_weight = float(weights[mask].sum()) + channels[name] = { + "rows": int(np.count_nonzero(mask)), + "worked_share": ( + float(weights[mask & worked].sum()) / channel_weight + if channel_weight > 0.0 + else 0.0 + ), + "recode_positive_share": ( + float(weights[mask & recode_positive].sum()) / channel_weight + if channel_weight > 0.0 + else 0.0 + ), + } + return { + "worked_share": _share(worked), + "recode_positive_share": _share(recode_positive), + "military_share": _share(detailed == float(_DETAILED_MILITARY_CODE)), + "never_worked_share": _share(detailed == float(_DETAILED_NEVER_WORKED_CODE)), + "channels": channels, + "worked_share_band": list(_WORKED_SHARE_BAND), + "recode_share_band": list(_RECODE_SHARE_BAND), + "nonfinite_detailed": int(np.count_nonzero(~np.isfinite(detailed))), + "nonfinite_major": int(np.count_nonzero(~np.isfinite(major))), + "out_of_range_detailed": int( + np.count_nonzero((detailed < 0.0) | (detailed > float(_DETAILED_MAX))) + ), + "out_of_range_major": int( + np.count_nonzero((major < 0.0) | (major > float(_MAJOR_MAX))) + ), + "universe_identity_breaks": int(np.count_nonzero(worker_code != worked)), + "recode_zero_breaks": int( + np.count_nonzero((detailed == 0.0) != (major == 0.0)) + ), + } + + +def us_work_experience_signal_gate(frame: Frame) -> GateResult: + """Require nonzero, plausible, coherent work-experience signal.""" + + person = frame.table("person") + missing = [ + column + for column in US_WORK_EXPERIENCE_OUTPUT_COLUMNS + if column not in person.columns + ] + if missing: + return GateResult( + name="work_experience_signal", + passed=False, + failures=(f"person columns missing: {missing}.",), + details={"missing": missing}, + ) + + summary = us_work_experience_summary(frame) + failures: list[str] = [] + for count_key, label in ( + ("nonfinite_detailed", "detailed_industry_recode nonfinite values"), + ("nonfinite_major", "major_industry_recode nonfinite values"), + ("out_of_range_detailed", "detailed_industry_recode out-of-range values"), + ("out_of_range_major", "major_industry_recode out-of-range values"), + ("universe_identity_breaks", "worker-code rows disagreeing with worked"), + ("recode_zero_breaks", "detailed/major zero-row disagreements"), + ): + count = int(summary[count_key]) + if count: + failures.append(f"{label}: {count}.") + + for share_key, band_key, label in ( + ("worked_share", "worked_share_band", "worked-last-year share"), + ("recode_positive_share", "recode_share_band", "industry-recode share"), + ): + share = float(summary[share_key]) + low, high = summary[band_key] + if not (low <= share <= high): + failures.append( + f"{label} {share:.3f} outside plausibility band [{low}, {high}]." + ) + + return GateResult( + name="work_experience_signal", + passed=not failures, + failures=tuple(failures), + details=summary, + ) + + +def _work_experience_surface_carries_signal(frame: Frame) -> bool: + person = frame.table("person") + if not all(column in person for column in US_WORK_EXPERIENCE_OUTPUT_COLUMNS): + return False + return us_work_experience_signal_gate(frame).passed diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/work_experience_source.py b/packages/microcosm-build/src/microcosm/build/us_runtime/work_experience_source.py new file mode 100644 index 00000000..224e24a7 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/work_experience_source.py @@ -0,0 +1,679 @@ +"""Pinned ASEC work-experience industry recodes for the pooled person years. + +The frozen census_cps person inputs never carried the ASEC work-experience +industry recodes, so this sidecar restores ``WEIND`` (industry of longest job +by detailed groups, 0--23) and ``WEMIND`` (industry of longest job by major +industry groups, 0--15) for every pooled income year from the official, +immutable ASEC public-use archives. The restore is an exact per-income-year +``PERIDNUM`` join; it never predicts an ASEC source value. + +The loader also enforces the official universe identity of the +work-experience recode block against each archive's own ``WKSWORK`` and +``WORKYN`` columns: ``WEIND`` is a worker code (1--22) exactly where +``WKSWORK > 0``, ``WEIND`` and ``WEMIND`` are zero on exactly the same rows, +and ``WORKYN = 1`` never appears without positive ``WKSWORK``. Those audit +columns are consumed at load time only and are not part of the sidecar +payload. +""" + +from __future__ import annotations + +import hashlib +import os +import urllib.request +import zipfile +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import BinaryIO + +import numpy as np +import pandas as pd + +__all__ = [ + "ASEC_WORK_EXPERIENCE_ARCHIVES", + "ASEC_WORK_EXPERIENCE_INCOME_YEARS", + "ASEC_WORK_EXPERIENCE_SOURCE_COLUMNS", + "WORK_EXPERIENCE_OFFICIAL_DICTIONARY_URL", + "AsecWorkExperienceArchive", + "fetch_asec_work_experience_source", + "fill_asec_work_experience_source", + "load_asec_work_experience_sources", +] + +#: Official data dictionary naming the carried entries: person WEIND +#: ("IND. OF LONGEST JOB BY DETAILED GROUPS", position 326 length 2) and +#: WEMIND ("IND. OF LONGEST JOB BY MAJOR IND. GROUPS", position 329 length +#: 2), both with universe "All Persons aged 15+". +WORK_EXPERIENCE_OFFICIAL_DICTIONARY_URL = ( + "https://www2.census.gov/programs-surveys/cps/datasets/2024/march/" + "asec2024_ddl_pub_full.pdf" +) + +_DETAILED_SOURCE = "WEIND" +_MAJOR_SOURCE = "WEMIND" +ASEC_WORK_EXPERIENCE_SOURCE_COLUMNS: tuple[str, ...] = ( + "PH_SEQ", + "P_SEQ", + "A_LINENO", + "PERIDNUM", + _DETAILED_SOURCE, + _MAJOR_SOURCE, +) +_AUDIT_WEIGHT_COLUMN = "A_FNLWGT" +_AUDIT_WEEKS_COLUMN = "WKSWORK" +_AUDIT_WORKED_COLUMN = "WORKYN" +_AUDIT_COLUMNS: tuple[str, ...] = ( + _AUDIT_WEIGHT_COLUMN, + _AUDIT_WEEKS_COLUMN, + _AUDIT_WORKED_COLUMN, +) +_DETAILED_MAX = 23 +_DETAILED_WORKER_MAX = 22 +_MAJOR_MAX = 15 + + +@dataclass(frozen=True) +class AsecWorkExperienceArchive: + """Pinned identity of one official ASEC survey-year person archive.""" + + survey_year: int + income_year: int + zip_url: str + zip_size_bytes: int + zip_sha256: str + member: str + member_size_bytes: int + member_crc32: str + member_sha256: str + rows: int + worked_rows: int + weighted_worked_share: float + recode_rows: int + weighted_recode_share: float + + +#: One pinned archive per pooled income year. The survey-year file published +#: the March after each income year carries that income year's person +#: universe: row counts equal the pooled cohorts exactly and PERIDNUM +#: coverage is 100.0% per year (and only ~33% against any adjacent survey +#: year, the CPS rotation-group overlap — pinning the wrong vintage fails the +#: full-coverage join loudly). The zip and member identities equal the +#: education-assistance pins of the same archives; the audit statistics are +#: this sidecar's own work-experience measurements. +ASEC_WORK_EXPERIENCE_ARCHIVES: dict[int, AsecWorkExperienceArchive] = { + archive.income_year: archive + for archive in ( + AsecWorkExperienceArchive( + survey_year=2023, + income_year=2022, + zip_url=( + "https://www2.census.gov/programs-surveys/cps/datasets/2023/" + "march/asecpub23csv.zip" + ), + zip_size_bytes=150_165_063, + zip_sha256=( + "d2e000250782adfbdd7f29c82b66d866591a30f0d330496698ec19f9c784ce11" + ), + member="pppub23.csv", + member_size_bytes=281_065_733, + member_crc32="49c09e5f", + member_sha256=( + "19b56537e50e7663f954361ef2bb5ce9cef8d9d45f156fe1a69a99b654198ffe" + ), + rows=146_133, + worked_rows=73_186, + weighted_worked_share=0.519967970757664, + recode_rows=116_650, + weighted_recode_share=0.821736897883986, + ), + AsecWorkExperienceArchive( + survey_year=2024, + income_year=2023, + zip_url=( + "https://www2.census.gov/programs-surveys/cps/datasets/2024/" + "march/asecpub24csv.zip" + ), + zip_size_bytes=148_664_101, + zip_sha256=( + "cdb39cdac34bef99dd0940ab28e306f692404c2eea44d85dfd634214872a0a09" + ), + member="pppub24.csv", + member_size_bytes=277_250_415, + member_crc32="87950ece", + member_sha256=( + "21a2b9e0e4b08534563578a45acad77868af4ae9a7d46f23776b707d4a559aa7" + ), + rows=144_265, + worked_rows=73_471, + weighted_worked_share=0.523171279944173, + recode_rows=115_836, + weighted_recode_share=0.822002159972951, + ), + AsecWorkExperienceArchive( + survey_year=2025, + income_year=2024, + zip_url=( + "https://www2.census.gov/programs-surveys/cps/datasets/2025/" + "march/asecpub25csv.zip" + ), + zip_size_bytes=147_271_429, + zip_sha256=( + "318845a2b5e0034eb2973898de1738f4df0025727de38499e7669cb9c0deef0b" + ), + member="pppub25.csv", + member_size_bytes=277_882_549, + member_crc32="7dc2878f", + member_sha256=( + "06921fe83fc66c907e6c7b86b82255dc70458ee7d76258fc48297cb34f0c06b5" + ), + rows=142_125, + worked_rows=72_460, + weighted_worked_share=0.522793674686251, + recode_rows=114_446, + weighted_recode_share=0.824376426736239, + ), + ) +} + +ASEC_WORK_EXPERIENCE_INCOME_YEARS: tuple[int, ...] = tuple( + sorted(ASEC_WORK_EXPERIENCE_ARCHIVES) +) + + +def _sha256_stream(stream: BinaryIO, *, chunk_size: int) -> tuple[str, int]: + digest = hashlib.sha256() + size = 0 + for chunk in iter(lambda: stream.read(chunk_size), b""): + digest.update(chunk) + size += len(chunk) + return digest.hexdigest(), size + + +def _verify_file( + path: Path, + *, + label: str, + expected_size_bytes: int | None, + expected_sha256: str | None, + chunk_size: int, +) -> None: + if not path.is_file(): + raise FileNotFoundError(path) + size = path.stat().st_size + if expected_size_bytes is not None and size != expected_size_bytes: + raise ValueError( + f"{label} byte length mismatch: expected {expected_size_bytes}, got {size}." + ) + if expected_sha256 is not None: + with path.open("rb") as stream: + digest, _ = _sha256_stream(stream, chunk_size=chunk_size) + if digest != expected_sha256: + raise ValueError( + f"{label} SHA-256 mismatch: expected {expected_sha256}, got {digest}." + ) + + +def _verified_member_info( + archive_file: zipfile.ZipFile, + pins: AsecWorkExperienceArchive, +) -> zipfile.ZipInfo: + members = [info for info in archive_file.infolist() if info.filename == pins.member] + if len(members) != 1: + raise ValueError( + f"ASEC {pins.survey_year} archive must contain exactly one " + f"{pins.member!r} member; found {len(members)}." + ) + info = members[0] + if info.file_size != pins.member_size_bytes: + raise ValueError( + f"ASEC {pins.survey_year} person member byte length mismatch: " + f"expected {pins.member_size_bytes}, got {info.file_size}." + ) + actual_crc = f"{info.CRC:08x}" + if actual_crc != pins.member_crc32: + raise ValueError( + f"ASEC {pins.survey_year} person member CRC32 mismatch: expected " + f"{pins.member_crc32}, got {actual_crc}." + ) + return info + + +def fetch_asec_work_experience_source( + income_year: int, + cache_dir: str | Path | None = None, + *, + chunk_size: int = 8 * 1024 * 1024, +) -> Path: + """Download, verify, extract, and cache one income year's person member.""" + + if chunk_size < 1: + raise ValueError("chunk_size must be positive") + if income_year not in ASEC_WORK_EXPERIENCE_ARCHIVES: + raise ValueError( + f"No pinned ASEC work-experience archive covers income year " + f"{income_year}; pinned income years: " + f"{list(ASEC_WORK_EXPERIENCE_INCOME_YEARS)}." + ) + pins = ASEC_WORK_EXPERIENCE_ARCHIVES[income_year] + root = ( + Path(cache_dir).expanduser() + if cache_dir is not None + else Path.home() / ".cache" / "microcosm" / "cps" / "asec_work_experience" + ) + root.mkdir(parents=True, exist_ok=True) + archive_path = root / f"asecpub{str(pins.survey_year)[2:]}csv.zip" + member_path = root / pins.member + + if not archive_path.exists(): + temporary = archive_path.with_suffix(".zip.part") + request = urllib.request.Request( + pins.zip_url, + headers={"User-Agent": "microcosm-build/asec-work-experience"}, + ) + try: + with urllib.request.urlopen(request, timeout=180) as response: # noqa: S310 + with temporary.open("wb") as destination: + digest = hashlib.sha256() + size = 0 + for chunk in iter(lambda: response.read(chunk_size), b""): + destination.write(chunk) + digest.update(chunk) + size += len(chunk) + if size != pins.zip_size_bytes: + raise ValueError( + f"ASEC {pins.survey_year} archive download byte length " + f"mismatch: expected {pins.zip_size_bytes}, got {size}." + ) + actual_digest = digest.hexdigest() + if actual_digest != pins.zip_sha256: + raise ValueError( + f"ASEC {pins.survey_year} archive download SHA-256 mismatch: " + f"expected {pins.zip_sha256}, got {actual_digest}." + ) + os.replace(temporary, archive_path) + finally: + temporary.unlink(missing_ok=True) + + _verify_file( + archive_path, + label=f"ASEC {pins.survey_year} archive", + expected_size_bytes=pins.zip_size_bytes, + expected_sha256=pins.zip_sha256, + chunk_size=chunk_size, + ) + with zipfile.ZipFile(archive_path) as archive_file: + info = _verified_member_info(archive_file, pins) + if member_path.exists(): + _verify_file( + member_path, + label=f"ASEC {pins.survey_year} person member", + expected_size_bytes=pins.member_size_bytes, + expected_sha256=pins.member_sha256, + chunk_size=chunk_size, + ) + return member_path + temporary = member_path.with_suffix(".csv.part") + try: + with archive_file.open(info) as source, temporary.open("wb") as out: + digest = hashlib.sha256() + size = 0 + for chunk in iter(lambda: source.read(chunk_size), b""): + out.write(chunk) + digest.update(chunk) + size += len(chunk) + if size != pins.member_size_bytes: + raise ValueError( + f"ASEC {pins.survey_year} extracted person member byte " + f"length mismatch: expected {pins.member_size_bytes}, " + f"got {size}." + ) + actual_digest = digest.hexdigest() + if actual_digest != pins.member_sha256: + raise ValueError( + f"ASEC {pins.survey_year} extracted person member SHA-256 " + f"mismatch: expected {pins.member_sha256}, got {actual_digest}." + ) + os.replace(temporary, member_path) + finally: + temporary.unlink(missing_ok=True) + return member_path + + +def _fixed_width_peridnum(values: pd.Series, *, label: str) -> pd.Series: + if values.isna().any(): + rows = values.index[values.isna()].tolist()[:5] + raise ValueError(f"{label} PERIDNUM is missing at row(s): {rows}.") + decoded = values.map( + lambda value: ( + value.decode() + if isinstance(value, (bytes, bytearray, np.bytes_)) + else value + ) + ).astype(str) + valid = decoded.str.fullmatch(r"[0-9]{22}", na=False) + if not valid.all(): + rows = decoded.index[~valid].tolist()[:5] + raise ValueError( + f"{label} PERIDNUM must be an exact 22-digit string at row(s): {rows}." + ) + return decoded + + +def _integer_bounded( + frame: pd.DataFrame, + column: str, + *, + label: str, + upper: int, +) -> np.ndarray: + values = pd.to_numeric(frame[column], errors="coerce").to_numpy(dtype=np.float64) + valid = np.isfinite(values) & (values == np.floor(values)) + valid &= (values >= 0.0) & (values <= float(upper)) + if not valid.all(): + rows = np.flatnonzero(~valid)[:5].tolist() + raise ValueError( + f"{label} {column} must be an integer in [0, {upper}] at row(s): {rows}." + ) + return values.astype(np.int64) + + +def _load_one_source( + path: Path, pins: AsecWorkExperienceArchive, chunk_size: int +) -> pd.DataFrame: + usecols = [*ASEC_WORK_EXPERIENCE_SOURCE_COLUMNS, *_AUDIT_COLUMNS] + if zipfile.is_zipfile(path): + _verify_file( + path, + label=f"ASEC {pins.survey_year} archive", + expected_size_bytes=pins.zip_size_bytes, + expected_sha256=pins.zip_sha256, + chunk_size=chunk_size, + ) + with zipfile.ZipFile(path) as archive_file: + info = _verified_member_info(archive_file, pins) + with archive_file.open(info) as member: + digest, size = _sha256_stream(member, chunk_size=chunk_size) + if size != pins.member_size_bytes or digest != pins.member_sha256: + raise ValueError( + f"ASEC {pins.survey_year} person member identity mismatch." + ) + with archive_file.open(info) as member: + return pd.read_csv( + member, + usecols=usecols, + dtype={"PERIDNUM": "string"}, + low_memory=False, + ) + _verify_file( + path, + label=f"ASEC {pins.survey_year} person member", + expected_size_bytes=pins.member_size_bytes, + expected_sha256=pins.member_sha256, + chunk_size=chunk_size, + ) + return pd.read_csv( + path, usecols=usecols, dtype={"PERIDNUM": "string"}, low_memory=False + ) + + +def load_asec_work_experience_sources( + paths: Mapping[int, str | Path] | None = None, + *, + income_years: tuple[int, ...] = ASEC_WORK_EXPERIENCE_INCOME_YEARS, + chunk_size: int = 8 * 1024 * 1024, +) -> pd.DataFrame: + """Load and pin-verify the pooled work-experience sidecar. + + ``paths`` maps INCOME years (the ``--asec-h5`` vocabulary) to local copies + of the pinned survey archives (zip or extracted member); any income year + without a path is fetched from the official Census archive and verified + against the same pins. + """ + + unknown = sorted(set(paths or ()) - set(ASEC_WORK_EXPERIENCE_ARCHIVES)) + if unknown: + raise ValueError( + f"No pinned ASEC work-experience archive covers income year(s) " + f"{unknown}; pinned income years: " + f"{list(ASEC_WORK_EXPERIENCE_INCOME_YEARS)}." + ) + if not income_years: + empty = pd.DataFrame( + columns=["source_year", *ASEC_WORK_EXPERIENCE_SOURCE_COLUMNS] + ) + empty.attrs["source_audit"] = {} + return empty + parts: list[pd.DataFrame] = [] + audits: dict[int, dict[str, float | int]] = {} + for income_year in income_years: + pins = ASEC_WORK_EXPERIENCE_ARCHIVES.get(income_year) + if pins is None: + raise ValueError( + f"No pinned ASEC work-experience archive covers income year " + f"{income_year}." + ) + provided = None if paths is None else paths.get(income_year) + path = ( + Path(provided).expanduser() + if provided is not None + else fetch_asec_work_experience_source(income_year) + ) + raw = _load_one_source(path, pins, chunk_size) + missing = sorted( + {*ASEC_WORK_EXPERIENCE_SOURCE_COLUMNS, *_AUDIT_COLUMNS} - set(raw.columns) + ) + if missing: + raise ValueError( + f"ASEC {pins.survey_year} work-experience source missing " + f"column(s): {missing}." + ) + if len(raw) != pins.rows: + raise ValueError( + f"ASEC {pins.survey_year} work-experience source row count " + f"mismatch: expected {pins.rows}, got {len(raw)}." + ) + raw["PERIDNUM"] = _fixed_width_peridnum( + raw["PERIDNUM"], label=f"ASEC {pins.survey_year} source" + ) + if raw["PERIDNUM"].duplicated(keep=False).any(): + raise ValueError( + f"ASEC {pins.survey_year} work-experience source PERIDNUM " + "must be unique." + ) + label = f"ASEC {pins.survey_year} work-experience source" + detailed = _integer_bounded( + raw, _DETAILED_SOURCE, label=label, upper=_DETAILED_MAX + ) + major = _integer_bounded(raw, _MAJOR_SOURCE, label=label, upper=_MAJOR_MAX) + weeks = _integer_bounded(raw, _AUDIT_WEEKS_COLUMN, label=label, upper=52) + worked_yn = _integer_bounded(raw, _AUDIT_WORKED_COLUMN, label=label, upper=2) + worked = weeks > 0 + worker_code = (detailed >= 1) & (detailed <= _DETAILED_WORKER_MAX) + universe_breaks = int(np.count_nonzero(worker_code != worked)) + if universe_breaks: + raise ValueError( + f"{label} breaks the work-experience universe identity " + f"(WEIND in 1..{_DETAILED_WORKER_MAX} iff WKSWORK > 0) on " + f"{universe_breaks} row(s)." + ) + recode_zero_breaks = int(np.count_nonzero((detailed == 0) != (major == 0))) + if recode_zero_breaks: + raise ValueError( + f"{label} detailed and major recodes disagree on the " + f"not-in-universe rows: {recode_zero_breaks} row(s)." + ) + affirm_without_weeks = int(np.count_nonzero((worked_yn == 1) & ~worked)) + if affirm_without_weeks: + raise ValueError( + f"{label} reports WORKYN = 1 without positive WKSWORK on " + f"{affirm_without_weeks} row(s)." + ) + weights = pd.to_numeric(raw[_AUDIT_WEIGHT_COLUMN], errors="coerce").to_numpy( + dtype=np.float64 + ) + if ( + not np.isfinite(weights).all() + or (weights < 0.0).any() + or float(weights.sum()) <= 0.0 + ): + raise ValueError( + f"ASEC {pins.survey_year} A_FNLWGT must be finite and " + "nonnegative with positive total mass." + ) + scaled = weights / 100.0 + recode_positive = detailed != 0 + audit = { + "rows": int(len(raw)), + "worked_rows": int(np.count_nonzero(worked)), + "weighted_worked_share": float(scaled[worked].sum() / scaled.sum()), + "recode_rows": int(np.count_nonzero(recode_positive)), + "weighted_recode_share": float( + scaled[recode_positive].sum() / scaled.sum() + ), + } + pinned_input = provided is None or ( + len(raw) == pins.rows + and Path(path).stat().st_size + in (pins.zip_size_bytes, pins.member_size_bytes) + ) + if pinned_input: + if audit["worked_rows"] != pins.worked_rows: + raise ValueError( + f"ASEC {pins.survey_year} work-experience audit drifted: " + f"expected {pins.worked_rows} worked rows, got " + f"{audit['worked_rows']}." + ) + if audit["recode_rows"] != pins.recode_rows: + raise ValueError( + f"ASEC {pins.survey_year} work-experience audit drifted: " + f"expected {pins.recode_rows} recode rows, got " + f"{audit['recode_rows']}." + ) + for key, expected in ( + ("weighted_worked_share", pins.weighted_worked_share), + ("weighted_recode_share", pins.weighted_recode_share), + ): + if not np.isclose(audit[key], expected, rtol=0.0, atol=1e-12): + raise ValueError( + f"ASEC {pins.survey_year} work-experience audit " + f"drifted for {key}." + ) + audits[income_year] = audit + part = raw.loc[:, list(ASEC_WORK_EXPERIENCE_SOURCE_COLUMNS)].copy() + part.insert(0, "source_year", np.int64(income_year)) + parts.append(part) + result = pd.concat(parts, ignore_index=True) + result.attrs["source_audit"] = audits + return result + + +def fill_asec_work_experience_source( + person: pd.DataFrame, + source: pd.DataFrame, +) -> pd.DataFrame: + """Fill ``WEIND``/``WEMIND`` for every pooled person via Census identity.""" + + required_person = ("source_year", "PERIDNUM") + missing_person = [column for column in required_person if column not in person] + if missing_person: + raise ValueError( + f"ASEC work-experience repair requires person column(s): {missing_person}." + ) + required_source = ("source_year", *ASEC_WORK_EXPERIENCE_SOURCE_COLUMNS) + missing_source = [column for column in required_source if column not in source] + if missing_source: + raise ValueError( + f"ASEC work-experience sidecar missing column(s): {missing_source}." + ) + result = person.copy(deep=True) + person_years = pd.to_numeric(result["source_year"], errors="coerce") + if person_years.isna().any() or (person_years != np.floor(person_years)).any(): + raise ValueError("ASEC work-experience person source_year is invalid.") + needed_years = sorted(int(year) for year in person_years.unique()) + covered = set( + pd.to_numeric(source["source_year"], errors="coerce").astype(int).unique() + ) + uncovered = [year for year in needed_years if year not in covered] + if uncovered: + raise ValueError( + "ASEC work-experience sidecar does not cover pooled income " + f"year(s): {uncovered}." + ) + + donor = source.copy(deep=True) + donor["PERIDNUM"] = _fixed_width_peridnum( + donor["PERIDNUM"], label="ASEC work-experience sidecar" + ) + for column in (_DETAILED_SOURCE, _MAJOR_SOURCE): + if column in result.columns: + existing = pd.to_numeric(result[column], errors="coerce") + if existing.notna().any(): + raise ValueError( + f"ASEC work-experience fill found a preexisting {column} " + "column with values; refusing to overwrite measured data." + ) + result[column] = np.nan + + for year in needed_years: + year_mask = person_years.eq(year).to_numpy() + year_donor = donor.loc[ + pd.to_numeric(donor["source_year"], errors="coerce").eq(year) + ].set_index("PERIDNUM") + keys = _fixed_width_peridnum( + result.loc[year_mask, "PERIDNUM"], label="ASEC work-experience frame" + ) + missing_keys = keys[~keys.isin(year_donor.index)].drop_duplicates() + if not missing_keys.empty: + raise ValueError( + "ASEC work-experience sidecar does not cover frame PERIDNUM " + f"key(s) for income year {year}: {missing_keys.tolist()[:5]}." + ) + aligned = year_donor.reindex(keys.to_numpy()) + aligned.index = result.index[year_mask] + identity_pairs = [ + ( + "PH_SEQ", + "source_household_id" if "source_household_id" in result else "PH_SEQ", + ), + ("P_SEQ", "P_SEQ"), + ("A_LINENO", "A_LINENO"), + ] + for donor_column, frame_column in identity_pairs: + if frame_column not in result: + continue + observed = pd.to_numeric( + result.loc[year_mask, frame_column], errors="coerce" + ).to_numpy(dtype=np.float64) + expected = pd.to_numeric(aligned[donor_column], errors="coerce").to_numpy( + dtype=np.float64 + ) + mismatch = ( + ~np.isfinite(observed) | ~np.isfinite(expected) | (observed != expected) + ) + if mismatch.any(): + rows = result.index[year_mask].to_numpy()[mismatch][:5].tolist() + raise ValueError( + "ASEC work-experience redundant identity mismatch for " + f"{frame_column} against sidecar {donor_column} in income " + f"year {year} at row(s): {rows}." + ) + for column, upper in ( + (_DETAILED_SOURCE, _DETAILED_MAX), + (_MAJOR_SOURCE, _MAJOR_MAX), + ): + values = pd.to_numeric(aligned[column], errors="coerce").to_numpy( + dtype=np.float64 + ) + valid = np.isfinite(values) & (values == np.floor(values)) + valid &= (values >= 0.0) & (values <= float(upper)) + if not valid.all(): + raise ValueError( + f"ASEC work-experience sidecar {column} is invalid for income " + f"year {year}." + ) + result.loc[result.index[year_mask], column] = values + for column in (_DETAILED_SOURCE, _MAJOR_SOURCE): + result[column] = result[column].to_numpy(dtype=np.int64) + result.attrs["work_experience_source_audit"] = source.attrs.get("source_audit", {}) + return result diff --git a/packages/microcosm-build/tests/test_spec_engine_country_bundles.py b/packages/microcosm-build/tests/test_spec_engine_country_bundles.py index 617eff55..6d9e6c26 100644 --- a/packages/microcosm-build/tests/test_spec_engine_country_bundles.py +++ b/packages/microcosm-build/tests/test_spec_engine_country_bundles.py @@ -32,7 +32,7 @@ [ ( "be", - "262091db8c7b01b2a3b596aa2468d95855a63703ba9f8ebba2940cf5834c2c83", + "6d33bd183fdd2ba4b2a1f18461bf7e84775a9e2225f462a61b2c44d54c2b003b", { "household.household_id", "person.person_id", @@ -42,7 +42,7 @@ ), ( "uk", - "e12a2cb87c0e096af0173bd51fbede4b7df5e7c118ffe07ef616bbb30640e4ea", + "6aeb8727431952f713766da07b232e08e67a3ce2d61268371e7ddaed11a75db3", { "benunit.benunit_id", "household.household_id", diff --git a/packages/microcosm-build/tests/test_us_asec_checkpoint.py b/packages/microcosm-build/tests/test_us_asec_checkpoint.py index 4a604a3f..0ef58b7a 100644 --- a/packages/microcosm-build/tests/test_us_asec_checkpoint.py +++ b/packages/microcosm-build/tests/test_us_asec_checkpoint.py @@ -134,7 +134,7 @@ def _raw_binding(frame: Frame) -> dict[str, object]: "operation": "exact_source_join", "source_pins": [pin], } - for column in ("ED_VAL", "LKWEEKS", "PAW_TYP") + for column in ("ED_VAL", "LKWEEKS", "PAW_TYP", "WEIND", "WEMIND") }, "schema_version": ASEC_RAW_STAGE_SCHEMA_VERSION, "source_construction_identity": frame_identity(frame).to_payload(), @@ -166,6 +166,8 @@ def _raw_us_frame(*, id_offset: int = 0) -> Frame: tables["person"]["ED_VAL"] = [0.0, 500.0] tables["person"]["LKWEEKS"] = [-1, 12] tables["person"]["PAW_TYP"] = np.asarray([0, 1], dtype=np.int64) + tables["person"]["WEIND"] = np.asarray([0, 7], dtype=np.int64) + tables["person"]["WEMIND"] = np.asarray([0, 5], dtype=np.int64) return Frame( tables, source.schema, @@ -248,7 +250,7 @@ def test_loads_operator_untouched_raw_stage_checkpoint(tmp_path: Path) -> None: @pytest.mark.parametrize( "column", - ("ED_VAL", "LKWEEKS", "PAW_TYP", "PERIDNUM", "source_year"), + ("ED_VAL", "LKWEEKS", "PAW_TYP", "WEIND", "WEMIND", "PERIDNUM", "source_year"), ) def test_raw_loader_rejects_missing_input_complete_source_column( tmp_path: Path, @@ -278,6 +280,9 @@ def test_raw_loader_rejects_missing_input_complete_source_column( ("ED_VAL", [0.0, np.nan], "ED_VAL must be complete"), ("LKWEEKS", [-1, 53], "LKWEEKS must be complete"), ("PAW_TYP", [0, 4], "PAW_TYP must be complete integers"), + ("WEIND", [0, 24], "WEIND must be complete integers"), + ("WEMIND", [0, 16], "WEMIND must be complete integers"), + ("WEIND", [0, 0], "must be zero on exactly the same"), ), ) def test_raw_loader_rejects_invalid_input_complete_source_values( diff --git a/packages/microcosm-build/tests/test_us_l0_refit_export.py b/packages/microcosm-build/tests/test_us_l0_refit_export.py index d6a756d8..e507aeb8 100644 --- a/packages/microcosm-build/tests/test_us_l0_refit_export.py +++ b/packages/microcosm-build/tests/test_us_l0_refit_export.py @@ -128,6 +128,9 @@ def _us_frame(**person_extra: object) -> Frame: "cps_race": [1, 2, 4], "is_hispanic": [False, True, False], "detailed_occupation_recode": [1, 20, 53], + "detailed_industry_recode": np.asarray([7, 16, 23], dtype=np.int16), + "major_industry_recode": np.asarray([5, 10, 15], dtype=np.int16), + "worked_last_year": [True, True, False], "has_never_worked": [False, False, True], "is_military": [False, True, False], "is_computer_scientist": [False, True, False], diff --git a/packages/microcosm-build/tests/test_us_late_producer_dag.py b/packages/microcosm-build/tests/test_us_late_producer_dag.py index aa83a6f7..66cadb5d 100644 --- a/packages/microcosm-build/tests/test_us_late_producer_dag.py +++ b/packages/microcosm-build/tests/test_us_late_producer_dag.py @@ -343,9 +343,9 @@ def test_canonical_us_late_registry_has_exact_producer_surface() -> None: registry = CANONICAL_US_LATE_PRODUCER_REGISTRY groups = CANONICAL_US_LATE_TRANSFER_GROUPS - assert len(registry) == 38 - assert len(groups) == 19 - assert sum(len(group.targets) for group in groups) == 70 + assert len(registry) == 40 + assert len(groups) == 20 + assert sum(len(group.targets) for group in groups) == 73 assert {contract.kind for contract in registry.values()} == { "primary_puf", "acs_earnings_universe", @@ -435,9 +435,9 @@ def test_late_overlap_ownership_exhausts_every_permitted_dual_write() -> None: declared = set(US_LATE_OVERLAP_OWNERSHIP_TARGETS) assert len(primary) == 65 - assert len(source_writes) == 35 - assert len(source_writes & transfer) == 29 - assert len(transfer) == 70 + assert len(source_writes) == 38 + assert len(source_writes & transfer) == 32 + assert len(transfer) == 73 assert len(recipient_owned) == 60 assert ( callback_passthroughs @@ -626,7 +626,7 @@ def test_primary_puf_inventory_declares_exact_read_before_write_surface() -> Non def test_canonical_us_late_registry_declares_required_cross_producer_edges() -> None: edges = set(CANONICAL_US_LATE_PRODUCER_SCHEDULE.edges) - assert len(edges) == 71 + assert len(edges) == 75 assert ( US_LATE_ACS_EARNINGS_UNIVERSE_STAGE, US_LATE_PRIMARY_PUF_STAGE, @@ -805,10 +805,10 @@ def test_canonical_us_late_schedule_is_import_validated_and_byte_stable() -> Non } assert receipt["status"] == "derived_and_import_validated" assert receipt["schedule_sha256"] == reconstructed.sha256 - assert receipt["producer_count"] == 38 - assert receipt["source_producer_count"] == 16 - assert receipt["transfer_group_count"] == 19 - assert receipt["transfer_target_count"] == 70 + assert receipt["producer_count"] == 40 + assert receipt["source_producer_count"] == 17 + assert receipt["transfer_group_count"] == 20 + assert receipt["transfer_target_count"] == 73 assert receipt["order"][:2] == [ US_LATE_ACS_EARNINGS_UNIVERSE_STAGE, US_LATE_PRIMARY_PUF_STAGE, @@ -1016,9 +1016,9 @@ def test_every_origin_exclusive_raw_input_has_its_native_scope() -> None: ("household", "H_TENURE"): "asec_source", } expected_counts = { - ("household", "TYPEHUGQ"): 39, + ("household", "TYPEHUGQ"): 41, ("person", "MCARE"): 2, - ("person", "PERIDNUM"): 18, + ("person", "PERIDNUM"): 19, ("person", "SEMP"): 2, ("person", "WAGP"): 2, **{ @@ -1055,10 +1055,10 @@ def test_every_origin_exclusive_raw_input_has_its_native_scope() -> None: assert observed == Counter( {(*key, required_scope[key]): count for key, count in expected_counts.items()} ) - assert sum(observed.values()) == 101 - assert sum(count for key, count in observed.items() if key[2] == "acs_source") == 43 + assert sum(observed.values()) == 104 + assert sum(count for key, count in observed.items() if key[2] == "acs_source") == 45 assert ( - sum(count for key, count in observed.items() if key[2] == "asec_source") == 58 + sum(count for key, count in observed.items() if key[2] == "asec_source") == 59 ) assert receipts[("household", "TYPEHUGQ")] == set() @@ -1287,6 +1287,7 @@ def test_source_numeric_input_audit_is_fully_executable() -> None: "SPM_CAPHOUSESUB", }, "with_us_education_inputs": {"ED_VAL", "qualified_tuition_expenses"}, + "with_us_work_experience_inputs": {"WEIND", "WEMIND", "WKSWORK"}, } assert set(expected_finite) == set(US_LATE_SOURCE_INPUT_INVENTORIES) for operator, expected_columns in expected_finite.items(): @@ -1324,7 +1325,7 @@ def test_source_numeric_input_audit_is_fully_executable() -> None: } == {"finite_numeric"} -def test_late_target_dependency_kinds_partition_51_numeric_17_boolean_2_string() -> ( +def test_late_target_dependency_kinds_partition_53_numeric_18_boolean_2_string() -> ( None ): string_targets = {"ssn_card_type", "immigration_status_str"} @@ -1346,6 +1347,7 @@ def test_late_target_dependency_kinds_partition_51_numeric_17_boolean_2_string() "is_pursuing_credential_for_american_opportunity_credit", "takes_up_medicare_if_eligible", "would_claim_wic", + "worked_last_year", } observed: dict[str, set[str]] = {} for group in CANONICAL_US_LATE_TRANSFER_GROUPS: @@ -1361,8 +1363,8 @@ def test_late_target_dependency_kinds_partition_51_numeric_17_boolean_2_string() } numeric_targets = set(observed) - boolean_targets - string_targets assert (len(numeric_targets), len(boolean_targets), len(string_targets)) == ( - 51, - 17, + 53, + 18, 2, ) assert all(observed[target] == {"finite_numeric"} for target in numeric_targets) diff --git a/packages/microcosm-build/tests/test_us_multispine_pool.py b/packages/microcosm-build/tests/test_us_multispine_pool.py index 35e1b826..b2842dfc 100644 --- a/packages/microcosm-build/tests/test_us_multispine_pool.py +++ b/packages/microcosm-build/tests/test_us_multispine_pool.py @@ -126,6 +126,7 @@ "with_us_retirement_distribution_inputs", "with_us_immigration_inputs", "with_us_education_inputs", + "with_us_work_experience_inputs", ) _EXPECTED_PRE_CLONE_SOURCE_OPERATOR_ORDER = ( @@ -154,6 +155,7 @@ "with_us_retirement_distribution_inputs", "with_us_immigration_inputs", "with_us_education_inputs", + "with_us_work_experience_inputs", ) @@ -1167,10 +1169,10 @@ def test_pool_transfer_plan_extends_legacy_except_receipted_asset_deferrals() -> assert "has_marketplace_health_coverage" not in owners target_names = sorted(owners) - assert len(target_names) == 118 + assert len(target_names) == 121 assert ( hashlib.sha256(("\n".join(target_names) + "\n").encode()).hexdigest() - == "74fd985208c62ee51a96c161ee2766118e4d92020ce2897bf2942e2625db9484" + == "d9a204e09e31e9a79b070da1b7ad01d7dc6a49a99ac1b79b551f9a178252938d" ) @@ -1190,11 +1192,11 @@ def keys(target_families): source_producers = keys(pool_post_puf_source_producer_target_families()) assert len(early) == 48 - assert len(late) == 70 + assert len(late) == 73 assert early.isdisjoint(late) assert early | late == full assert len(puf_producers) == 43 - assert len(source_producers) == 29 + assert len(source_producers) == 32 assert len(puf_producers & source_producers) == 2 assert puf_producers | source_producers == late assert ("person", "source_operator_cps_carried", "strike_benefits") in early @@ -1225,13 +1227,13 @@ def test_pool_input_surface_normalizes_all_four_source_registries() -> None: surface = pool_input_surface() by_name = {entry.variable: entry for entry in surface} - assert len(surface) == len(by_name) == 139 + assert len(surface) == len(by_name) == 142 assert [entry.variable for entry in surface] == sorted(by_name) assert Counter( provenance for entry in surface for provenance in entry.provenance ) == Counter( { - "pool_transfer_target_families": 118, + "pool_transfer_target_families": 121, "POOL_DEFERRED_TRANSFER_INPUTS": 3, "PRIMARY_QRF_TARGET_ORDER": 65, "load_take_up_contract": 13, @@ -1859,6 +1861,9 @@ def _producer_dtype_source_frame() -> Frame: person["PEAFEVER"] = 2 person["ED_VAL"] = [0.0, 500.0, 0.0, 1_000.0] person["qualified_tuition_expenses"] = [0.0, 1_000.0, 0.0, 2_000.0] + # Work-experience recodes coherent with WKSWORK = [52, 0, 48, 0]. + person["WEIND"] = np.asarray([7, 0, 21, 0], dtype=np.int64) + person["WEMIND"] = np.asarray([5, 0, 13, 0], dtype=np.int64) return _replace_person(frame, person) @@ -2135,7 +2140,7 @@ def test_every_pool_transfer_target_is_an_installed_engine_input_leaf() -> None: for columns in families.values() for target in columns } - assert len(targets) == 118 + assert len(targets) == 121 acs_transfer_module.assert_acs_transfer_targets_are_input_leaves( targets, require_known=True, @@ -2160,7 +2165,7 @@ def test_every_pool_transfer_family_accepts_its_produced_physical_dtype( ) ) - assert len(targets) == 118 + assert len(targets) == 121 assert len(predictors) == 32 assert len(primary_predictor_sets) == 65 primary_targets = tuple( @@ -2177,13 +2182,13 @@ def test_every_pool_transfer_family_accepts_its_produced_physical_dtype( assert len(primary_predictor_sets[0][1]) == 8 assert len(primary_predictor_sets[-1][1]) == 72 assert len(POOL_DEFERRED_TRANSFER_INPUTS) == 3 - assert len(targets) + len(POOL_DEFERRED_TRANSFER_INPUTS) == 121 + assert len(targets) + len(POOL_DEFERRED_TRANSFER_INPUTS) == 124 assert set(POOL_SOURCE_OPERATOR_ORDER) <= set(calls) assert all(calls[name] > 0 for name in POOL_SOURCE_OPERATOR_ORDER) assert calls["with_us_prior_year_income_inputs"] == 2 assert calls["primary_puf_qrf.fit"] > 0 assert calls["primary_puf_qrf.predict"] > 0 - assert sum(calls[name] for name in POOL_SOURCE_OPERATOR_ORDER) == 22 + assert sum(calls[name] for name in POOL_SOURCE_OPERATOR_ORDER) == 23 def test_object_backed_is_female_becomes_nullable_before_production_transfer_fit( @@ -2803,7 +2808,7 @@ def materialize_once(frame: Frame) -> PoolStageOutput: assert deferred_calls == [finalized.frame] assert finalized.receipt["operator_order"] == list(execution_order) assert [item["order_index"] for item in finalized.receipt["suboperators"]] == list( - range(16) + range(17) ) assert finalized.receipt["deferred_transfer_inputs"] == { "inputs": {"fixture": {"status": "pending"}} @@ -2823,7 +2828,7 @@ def test_source_finalizer_rejects_incomplete_receipts_before_deferred_inputs( for operator in POOL_POST_CLONE_SOURCE_OPERATOR_ORDER[:-1] } - with pytest.raises(ValueError, match=r"exactly.*16.*missing=.*education"): + with pytest.raises(ValueError, match=r"exactly.*17.*missing=.*work_experience"): finalize_multispine_source_inputs( _source_frame(), operator_receipts=receipts, @@ -2983,8 +2988,8 @@ def observe_guarded_chain( for phase in contract.phases } assert observed_placements == registered_placements - assert len({name for name, _phase in observed_placements}) == 23 - assert len(observed_placements) == 24 + assert len({name for name, _phase in observed_placements}) == 24 + assert len(observed_placements) == 25 def test_derive_stage_rejects_preclone_pool_before_kernels( diff --git a/packages/microcosm-build/tests/test_us_multispine_pool_h5_io.py b/packages/microcosm-build/tests/test_us_multispine_pool_h5_io.py index 8f5dc55e..218a94aa 100644 --- a/packages/microcosm-build/tests/test_us_multispine_pool_h5_io.py +++ b/packages/microcosm-build/tests/test_us_multispine_pool_h5_io.py @@ -595,8 +595,8 @@ def _canonical_stacked_late_dag_receipt() -> dict[str, object]: "targets": aggregate_targets, "completion": { "status": "complete", - "group_count": 19, - "target_count": 70, + "group_count": 20, + "target_count": 73, "residual_null_rows": 0, }, } diff --git a/packages/microcosm-build/tests/test_us_plan.py b/packages/microcosm-build/tests/test_us_plan.py index 50c8b379..fb65e2a8 100644 --- a/packages/microcosm-build/tests/test_us_plan.py +++ b/packages/microcosm-build/tests/test_us_plan.py @@ -1249,8 +1249,10 @@ class TestBaseStageSourceClosure: #: Raw columns restored from pinned official sidecars because the frozen #: census_cps inputs never carried them (LKWEEKS only for income year - #: 2022; ED_VAL and PAW_TYP for every pooled year). - SIDECAR_RESTORED_COLUMNS = frozenset({"LKWEEKS", "ED_VAL", "PAW_TYP"}) + #: 2022; ED_VAL, PAW_TYP, WEIND, and WEMIND for every pooled year). + SIDECAR_RESTORED_COLUMNS = frozenset( + {"LKWEEKS", "ED_VAL", "PAW_TYP", "WEIND", "WEMIND"} + ) #: Release-time stage constants whose inputs are produced inside the #: fiscal-refresh release tool, not the base builder (org_wages consumes diff --git a/packages/microcosm-build/tests/test_us_pool_input_consumers.py b/packages/microcosm-build/tests/test_us_pool_input_consumers.py index 6d04b05a..296cb535 100644 --- a/packages/microcosm-build/tests/test_us_pool_input_consumers.py +++ b/packages/microcosm-build/tests/test_us_pool_input_consumers.py @@ -53,6 +53,49 @@ class _ReviewedNonEngineConsumer(TypedDict): "Consumed by QBI reconciliation and the qualified-BDC exposure invariant." ), }, + # Work-experience person attributes (PolicyEngine/microcosm#719): the + # committed county-file person schema carries a working indicator, + # industry, and occupation on every record for the Living Wage Institute's + # downstream OEWS x QCEW machinery. PolicyEngine-US carries them as input + # leaves without a formula consumer, exactly like detailed_occupation_recode + # before the FLSA flags existed; the stage's signal gate validates the + # recode-universe identity and plausibility bands. + "detailed_industry_recode": { + "consumer_class": NonEngineConsumerClass.CLIENT_SURFACE, + "consumer": ( + "microcosm.build.us_runtime.work_experience_inputs." + "us_work_experience_signal_gate" + ), + "justification": ( + "Longest-job industry (ASEC WEIND) shipped on every person record " + "for the Living Wage Institute county files (microcosm#719); gated " + "for the WKSWORK > 0 universe identity and nonzero-recode share." + ), + }, + "major_industry_recode": { + "consumer_class": NonEngineConsumerClass.CLIENT_SURFACE, + "consumer": ( + "microcosm.build.us_runtime.work_experience_inputs." + "us_work_experience_signal_gate" + ), + "justification": ( + "Longest-job major industry group (ASEC WEMIND) shipped with the " + "detailed recode (microcosm#719); gated for zero-row agreement with " + "detailed_industry_recode." + ), + }, + "worked_last_year": { + "consumer_class": NonEngineConsumerClass.CLIENT_SURFACE, + "consumer": ( + "microcosm.build.us_runtime.work_experience_inputs." + "us_work_experience_signal_gate" + ), + "justification": ( + "Measured worked-last-year indicator (ASEC WKSWORK > 0, the " + "work-experience recode universe) shipped on every person record " + "(microcosm#719); gated for the worker-code identity and share band." + ), + }, } _REQUIRED_ENGINE_CONSUMER_GRAINS = { @@ -193,6 +236,9 @@ def test_non_engine_consumer_allowlist_is_exact_and_current() -> None: assert set(NON_ENGINE_CONSUMER_ALLOWLIST) == { "previous_year_income_available", "qualified_bdc_income", + "detailed_industry_recode", + "major_industry_recode", + "worked_last_year", } _validate_allowlist( _POOL_INPUT_SURFACE, 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 7f2d44dd..eb423d03 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 @@ -253,6 +253,22 @@ def _public_assistance_type_source() -> pd.DataFrame: return source +def _work_experience_source() -> pd.DataFrame: + source = pd.DataFrame( + { + "source_year": [2022, 2022, 2022], + "PH_SEQ": [101, 101, 202], + "P_SEQ": [1, 2, 1], + "A_LINENO": [1, 2, 1], + "PERIDNUM": [f"{value:022d}" for value in (1, 2, 3)], + "WEIND": [7, 0, 23], + "WEMIND": [5, 0, 15], + } + ) + source.attrs["source_audit"] = {2022: {"rows": 3}} + return source + + def _pooled_source_receipt(tmp_path: Path) -> dict[str, object]: return { "kind": "pooled_asec", @@ -335,6 +351,11 @@ def _patch_raw_stage_sources( "load_asec_public_assistance_type_sources", lambda _paths, *, income_years: _public_assistance_type_source(), ) + monkeypatch.setattr( + builder, + "load_asec_work_experience_sources", + lambda _paths, *, income_years: _work_experience_source(), + ) monkeypatch.setattr( builder, "_builder_code_identity", @@ -787,6 +808,7 @@ def test_reconciled_outer_pipeline_order_is_locked() -> None: "retirement_contributions_post_clone", "retirement_distributions_post_clone", "education_inputs_post_clone", + "work_experience_inputs_post_clone", "congressional_district_assignment", "block_ladder_assignment", "final_export", @@ -820,6 +842,11 @@ def test_raw_stage_copy_adds_only_exact_source_mappings( "load_asec_public_assistance_type_sources", lambda _paths, *, income_years: _public_assistance_type_source(), ) + monkeypatch.setattr( + builder, + "load_asec_work_experience_sources", + lambda _paths, *, income_years: _work_experience_source(), + ) raw, mappings = builder._asec_raw_source_mapping_frame( args, @@ -834,7 +861,9 @@ def test_raw_stage_copy_adds_only_exact_source_mappings( assert raw.table("person")["LKWEEKS"].tolist() == [7.0, -1.0, 12.0] assert raw.table("person")["ED_VAL"].tolist() == [0.0, 500.0, 1_000.0] assert raw.table("person")["PAW_TYP"].tolist() == [0, 1, 2] - assert set(mappings) == {"ED_VAL", "LKWEEKS", "PAW_TYP"} + assert raw.table("person")["WEIND"].tolist() == [7, 0, 23] + assert raw.table("person")["WEMIND"].tolist() == [5, 0, 15] + assert set(mappings) == {"ED_VAL", "LKWEEKS", "PAW_TYP", "WEIND", "WEMIND"} assert all( mapping["operation"] == "exact_source_join" and mapping["join_keys"] == ["source_year", "PERIDNUM"] diff --git a/packages/microcosm-build/tests/test_us_spine_blindness.py b/packages/microcosm-build/tests/test_us_spine_blindness.py index 9267ed80..363bef54 100644 --- a/packages/microcosm-build/tests/test_us_spine_blindness.py +++ b/packages/microcosm-build/tests/test_us_spine_blindness.py @@ -170,6 +170,7 @@ "voluntary_filing.py", "weeks_unemployed.py", "wic_claim.py", + "work_experience_inputs.py", "workers_compensation.py", ) @@ -264,6 +265,8 @@ "us_late_producer_registry.py", "validation_input_coverage.py", "warm_start_selection.py", + # Pinned-archive sidecar restore (WEIND/WEMIND); no population treatment. + "work_experience_source.py", } ) _CLASSIFIED_US_RUNTIME_MODULES = frozenset(_SPINE_BLIND_OPERATOR_MODULES).union( @@ -3278,8 +3281,8 @@ def test_pool_build_tool_import_graph_is_source_spine_blind() -> None: for tool in _SPINE_BLIND_BUILD_TOOLS: runtime_graph, missing_modules = _us_runtime_import_graph(tool) - assert len(runtime_graph) == 65, ( - f"{tool.name} must reach the pinned 65-module runtime graph; " + assert len(runtime_graph) == 67, ( + f"{tool.name} must reach the pinned 67-module runtime graph; " f"reached {len(runtime_graph)}" ) assert not missing_modules, ( diff --git a/packages/microcosm-build/tests/test_us_stacked_spine.py b/packages/microcosm-build/tests/test_us_stacked_spine.py index 4e1b8f9a..fb41b01a 100644 --- a/packages/microcosm-build/tests/test_us_stacked_spine.py +++ b/packages/microcosm-build/tests/test_us_stacked_spine.py @@ -1722,7 +1722,7 @@ def test_canonical_authority_objects_are_deeply_immutable() -> None: profile.min_effective_support = 50 -def test_canonical_metric_registry_covers_the_declared_131_target_split() -> None: +def test_canonical_metric_registry_covers_the_declared_134_target_split() -> None: surface = stacked_spine_module.CANONICAL_STACKED_DECLARED_SURFACE registry = stacked_spine_module.CANONICAL_ORIGIN_BATTERY_METRIC_REGISTRY surface_targets = { @@ -1732,9 +1732,9 @@ def test_canonical_metric_registry_covers_the_declared_131_target_split() -> Non for target in targets } - assert len(surface_targets) == 131 + assert len(surface_targets) == 134 assert Counter(entity for entity, _family, _target, _clone in surface_targets) == { - "person": 114, + "person": 117, "tax_unit": 9, "spm_unit": 8, } @@ -1746,13 +1746,13 @@ def test_canonical_metric_registry_covers_the_declared_131_target_split() -> Non for family in families } ) - == 31 + == 32 ) assert set(registry) == surface_targets assert Counter(registry.values()) == { "monetary_sign_separated": 79, - "boolean_incidence": 48, - "categorical_tvd": 4, + "boolean_incidence": 49, + "categorical_tvd": 6, } assert ( registry[("person", "puf_tax_itemization", "taxable_interest_income", 0)] @@ -1791,13 +1791,13 @@ def test_canonical_metric_registry_covers_the_declared_131_target_split() -> Non for target in targets } assert len(gap_targets) == 48 - assert len(post_puf_targets) == 70 + assert len(post_puf_targets) == 73 assert len(puf_producer_targets) == 43 - assert len(source_producer_targets) == 29 + assert len(source_producer_targets) == 32 assert len(puf_producer_targets & source_producer_targets) == 2 assert puf_producer_targets | source_producer_targets == post_puf_targets assert gap_targets.isdisjoint(post_puf_targets) - assert len(gap_targets | post_puf_targets) == 118 + assert len(gap_targets | post_puf_targets) == 121 assert gap_targets | post_puf_targets < surface_targets assert not { "bank_account_assets", @@ -1897,8 +1897,8 @@ def test_canonical_metric_registry_drives_checkpoint_round_trip( nullable_booleans = _transferred_registry_boolean_targets() assert Counter(registry.values()) == { "monetary_sign_separated": 79, - "boolean_incidence": 48, - "categorical_tvd": 4, + "boolean_incidence": 49, + "categorical_tvd": 6, } for (entity, _family, column, _clone_index), metric in registry.items(): expected = frame.table(entity)[column] @@ -1975,15 +1975,16 @@ def test_checkpoint_boundary_extension_dtype_inventory_is_exact() -> None: ("person", "self_employment_income_would_be_qualified"), ("person", "sstb_self_employment_income_would_be_qualified"), ("person", "takes_up_medicare_if_eligible"), + ("person", "worked_last_year"), ("person", "would_claim_wic"), ("spm_unit", "is_tanf_enrolled"), ("spm_unit", "receives_housing_assistance"), ("spm_unit", "receives_snap"), ("spm_unit", "takes_up_housing_assistance_if_eligible"), } - assert len(transferred_registry) == 37 + assert len(transferred_registry) == 38 assert transferred == expected - assert len(transferred) == 39 + assert len(transferred) == 40 terminal_registry = { (entity, column) @@ -2047,11 +2048,11 @@ def test_checkpoint_boundary_extension_dtype_inventory_is_exact() -> None: for stage, families in boundary_inventory.items() } == { "assembled": {"boolean": 0, "string": 17}, - "transferred": {"boolean": 39, "string": 19}, - "simulated": {"boolean": 39, "string": 19}, + "transferred": {"boolean": 40, "string": 19}, + "simulated": {"boolean": 40, "string": 19}, } assert sum(map(len, boundary_inventory["assembled"].values())) == 17 - assert sum(map(len, boundary_inventory["transferred"].values())) == 58 + assert sum(map(len, boundary_inventory["transferred"].values())) == 59 assert boundary_inventory["transferred"] == boundary_inventory["simulated"] @@ -2061,7 +2062,7 @@ def test_registry_drives_every_late_callback_dtype_family_check() -> None: (entity, column): metric for (entity, _family, column, _clone_index), metric in registry.items() } - assert len(by_column) == len(registry) == 131 + assert len(by_column) == len(registry) == 134 representative = { "monetary_sign_separated": pd.Series([1.0, pd.NA], dtype="Float64"), @@ -2103,23 +2104,23 @@ def test_registry_drives_every_late_callback_dtype_family_check() -> None: for output in contract.outputs if (key := (output.entity, output.column)) in by_column ] - assert len(registered_occurrences) == 163 + assert len(registered_occurrences) == 169 assert Counter( metric for _producer, _entity, _column, metric in registered_occurrences ) == { "monetary_sign_separated": 120, - "boolean_incidence": 37, - "categorical_tvd": 6, + "boolean_incidence": 39, + "categorical_tvd": 10, } unique_late_targets = { (entity, column): metric for _producer, entity, column, metric in registered_occurrences } - assert len(unique_late_targets) == 90 + assert len(unique_late_targets) == 93 assert Counter(unique_late_targets.values()) == { "monetary_sign_separated": 67, - "boolean_incidence": 20, - "categorical_tvd": 3, + "boolean_incidence": 21, + "categorical_tvd": 5, } @@ -4224,6 +4225,10 @@ def test_late_source_resources_bind_all_callback_controls(tmp_path: Path) -> Non expected_sidecars = {"asec_2023_source": {"mode": "not_supplied"}} if operator == "with_us_education_inputs": expected_sidecars = {"asec_education_source": {"mode": "not_supplied"}} + if operator == "with_us_work_experience_inputs": + expected_sidecars = { + "asec_work_experience_source": {"mode": "not_supplied"} + } assert binding["external_sidecars"] == expected_sidecars assert binding["allow_existing_without_source"] is ( multispine_pool_module.POOL_SOURCE_ALLOW_EXISTING_WITHOUT_SOURCE @@ -4943,7 +4948,7 @@ def test_late_executor_authority_binds_every_transfer_bank_identity( transfer_rows = [ row for row in first.receipt["execution"] if row["kind"] == "late_transfer" ] - assert len(transfer_rows) == 19 + assert len(transfer_rows) == 20 for row in transfer_rows: available = row["available_input_receipts"] assert len(available) == 2 @@ -6822,7 +6827,7 @@ def test_completeness_receipts_bind_live_authority_per_target() -> None: ) canonical = stacked_completeness_gate(frame) assert canonical.passed, canonical.failures - assert canonical.details["declared_targets"] == 131 + assert canonical.details["declared_targets"] == 134 authority = canonical.details["authority"] assert authority["authority_form"] == "CANONICAL" assert authority["canonical"] is True @@ -7001,7 +7006,7 @@ def test_stacked_authority_binds_import_validated_late_producer_schedule() -> No component = receipt["components"]["late_producer_schedule"] assert receipt["version"] == 10 - assert component["producer_count"] == 38 + assert component["producer_count"] == 40 assert component["schedule_sha256"] == ( stacked_spine_module.CANONICAL_US_LATE_PRODUCER_SCHEDULE.sha256 ) @@ -7202,7 +7207,7 @@ def test_fresh_gate_result_cannot_graft_canonical_authority_onto_test_surface() with pytest.raises( ValueError, - match="must declare exactly 131 targets.*manifest emission is forbidden", + match="must declare exactly 134 targets.*manifest emission is forbidden", ): GateReport((grafted,)).to_manifest() @@ -7337,7 +7342,7 @@ def test_fresh_battery_result_cannot_forge_canonical_coverage_receipts() -> None with pytest.raises( ValueError, - match="coverage receipt must bind all 131 targets.*emission is forbidden", + match="coverage receipt must bind all 134 targets.*emission is forbidden", ): GateReport((forged,)).to_manifest() @@ -7870,7 +7875,7 @@ def test_battery_taxable_interest_metric_cannot_be_relabelled_rare_incidence() - assert not result.passed assert result.details["tested_comparisons"] == 0 metric_receipt = result.details["authority"]["components"]["metric_registry"] - assert metric_receipt["target_count"] == 131 + assert metric_receipt["target_count"] == 134 assert any( "person/puf_tax_itemization/taxable_interest_income[clone_0]" in failure and "authoritative metric 'monetary_sign_separated'" in failure diff --git a/packages/microcosm-build/tests/test_us_work_experience_inputs.py b/packages/microcosm-build/tests/test_us_work_experience_inputs.py new file mode 100644 index 00000000..adde7c4b --- /dev/null +++ b/packages/microcosm-build/tests/test_us_work_experience_inputs.py @@ -0,0 +1,356 @@ +"""US work-experience industry and worked-last-year input stage tests. + +The stage carries the ASEC work-experience industry recodes (``WEIND`` +detailed groups, ``WEMIND`` major groups) and derives ``worked_last_year`` as +``WKSWORK > 0`` — the official universe condition of the recode block. The +tests exercise the manifest declaration, the derive handler's fail-closed +identities, the operator's sidecar restore, idempotency, and the signal gate. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.source_runtime import SourceRuntimeError +from microcosm.build.us_runtime import ( + BASE_ASEC_SUPPORT_CHANNEL, + PUF_TAX_DETAIL_SUPPORT_CHANNEL, + US_WORK_EXPERIENCE_NONCONSTANT_PERSON_COLUMNS, + US_WORK_EXPERIENCE_OUTPUT_COLUMNS, + US_WORK_EXPERIENCE_REQUIRED_SOURCE_COLUMNS, + US_WORK_EXPERIENCE_STAGE_NAME, + clone_us_frame_for_puf_support, + derive_us_work_experience_inputs_from_manifest, + us_work_experience_signal_gate, + us_work_experience_stage_spec, + us_work_experience_summary, + with_us_work_experience_inputs, +) +from microcosm.build.us_runtime.source_runtime import us_source_operation_handlers +from microcosm.frame import US_SCHEMA, Frame, WeightKind, Weights + +TIME_PERIOD = 2024 +_OUTPUTS = ("detailed_industry_recode", "major_industry_recode", "worked_last_year") + + +def _person_table(rows: list[dict]) -> pd.DataFrame: + records: list[dict] = [] + for index, row in enumerate(rows): + record = {"WEIND": 0, "WEMIND": 0, "WKSWORK": 0} + record.update(row) + record.setdefault("person_id", index + 1) + record.setdefault("person_household_id", index + 1) + records.append(record) + return pd.DataFrame(records) + + +def _us_frame(person_rows: list[dict]) -> Frame: + person = _person_table(person_rows) + n = len(person) + household_ids = person["person_household_id"].to_numpy(dtype=np.int64) + unique_households = np.unique(household_ids) + person["person_tax_unit_id"] = household_ids + 1_000 + person["person_spm_unit_id"] = household_ids + 2_000 + person["person_family_id"] = household_ids + 3_000 + person["person_marital_unit_id"] = np.arange(n, dtype=np.int64) + 4_000 + tables = { + "person": person, + "household": pd.DataFrame({"household_id": unique_households}), + "tax_unit": pd.DataFrame({"tax_unit_id": unique_households + 1_000}), + "spm_unit": pd.DataFrame({"spm_unit_id": unique_households + 2_000}), + "family": pd.DataFrame({"family_id": unique_households + 3_000}), + "marital_unit": pd.DataFrame( + {"marital_unit_id": np.arange(n, dtype=np.int64) + 4_000} + ), + } + return Frame( + tables, + US_SCHEMA, + { + "household": Weights( + values=np.ones(len(unique_households), dtype=np.float64), + kind=WeightKind.DESIGN, + ) + }, + ) + + +def _plausible_rows() -> list[dict]: + """100 people: 52 workers across industries, 30 never-worked, 18 NIU.""" + + rows: list[dict] = [] + for index in range(52): + detailed = 1 + index % 21 + rows.append( + {"WEIND": detailed, "WEMIND": 1 + detailed % 14, "WKSWORK": 10 + index % 43} + ) + rows.extend({"WEIND": 23, "WEMIND": 15} for _ in range(30)) + rows.extend({} for _ in range(18)) + return rows + + +def _operation(): + spec = us_work_experience_stage_spec() + return next( + op for op in spec.operations if op.kind == "derive_work_experience_inputs" + ) + + +class TestStageSpec: + def test_manifest_declares_the_outputs(self) -> None: + spec = us_work_experience_stage_spec() + assert spec.stage == US_WORK_EXPERIENCE_STAGE_NAME == "work_experience_inputs" + assert tuple(spec.outputs) == US_WORK_EXPERIENCE_OUTPUT_COLUMNS == _OUTPUTS + assert tuple(spec.nonnegative_outputs) == ( + "detailed_industry_recode", + "major_industry_recode", + ) + assert US_WORK_EXPERIENCE_NONCONSTANT_PERSON_COLUMNS == _OUTPUTS + assert US_WORK_EXPERIENCE_REQUIRED_SOURCE_COLUMNS == ( + "WEIND", + "WEMIND", + "WKSWORK", + ) + + def test_stage_reads_person_then_derives(self) -> None: + kinds = [op.kind for op in us_work_experience_stage_spec().operations] + assert kinds == ["read_table", "derive_work_experience_inputs"] + + def test_manifest_pins_every_pooled_archive_and_the_dictionary(self) -> None: + artifacts = us_work_experience_stage_spec().artifacts + microdata = [item for item in artifacts if item["kind"] == "public_microdata"] + assert [item["member"] for item in microdata] == [ + "pppub23.csv", + "pppub24.csv", + "pppub25.csv", + ] + assert all(len(item["sha256"]) == 64 for item in microdata) + dictionary = [ + item for item in artifacts if item["kind"] == "official_data_dictionary" + ] + assert len(dictionary) == 1 and "WEIND" in dictionary[0]["lines"] + + def test_handler_is_registered(self) -> None: + handlers = us_source_operation_handlers() + assert ( + handlers["derive_work_experience_inputs"] + is derive_us_work_experience_inputs_from_manifest + ) + + +class TestDerivation: + def _derive(self, table: pd.DataFrame) -> pd.DataFrame: + return derive_us_work_experience_inputs_from_manifest(table, _operation(), None) + + def test_carries_recodes_and_derives_worked_from_weeks(self) -> None: + table = _person_table( + [ + {"WEIND": 7, "WEMIND": 5, "WKSWORK": 52}, + {"WEIND": 22, "WEMIND": 15, "WKSWORK": 40}, + {"WEIND": 23, "WEMIND": 15, "WKSWORK": 0}, + {}, + ] + ) + result = self._derive(table) + assert result["detailed_industry_recode"].tolist() == [7, 22, 23, 0] + assert result["major_industry_recode"].tolist() == [5, 15, 15, 0] + assert result["worked_last_year"].tolist() == [True, True, False, False] + assert result["detailed_industry_recode"].dtype == np.int16 + assert result["major_industry_recode"].dtype == np.int16 + assert result["worked_last_year"].dtype == bool + + def test_rejects_unexpected_operation_kind(self) -> None: + spec = us_work_experience_stage_spec() + read = next(op for op in spec.operations if op.kind == "read_table") + with pytest.raises(SourceRuntimeError, match="unexpected operation"): + derive_us_work_experience_inputs_from_manifest( + _person_table([{}]), read, None + ) + + def test_requires_the_person_table(self) -> None: + with pytest.raises(SourceRuntimeError, match="read first"): + derive_us_work_experience_inputs_from_manifest(None, _operation(), None) + + def test_rejects_missing_source_columns(self) -> None: + table = _person_table([{}]).drop(columns=["WEMIND"]) + with pytest.raises(SourceRuntimeError, match=r"source column\(s\).*WEMIND"): + self._derive(table) + + @pytest.mark.parametrize( + ("row", "message"), + ( + ({"WEIND": 24, "WEMIND": 1, "WKSWORK": 1}, r"WEIND.*\[0, 23\]"), + ({"WEIND": 1, "WEMIND": 16, "WKSWORK": 1}, r"WEMIND.*\[0, 15\]"), + ({"WEIND": 1, "WEMIND": 1, "WKSWORK": 53}, r"WKSWORK.*\[0, 52\]"), + ({"WEIND": 1, "WEMIND": 1, "WKSWORK": 0}, "universe identity"), + ({"WEIND": 0, "WEMIND": 0, "WKSWORK": 5}, "universe identity"), + ({"WEIND": 23, "WEMIND": 0, "WKSWORK": 0}, "disagreeing"), + ), + ) + def test_fails_closed_on_official_identities(self, row: dict, message: str) -> None: + with pytest.raises(SourceRuntimeError, match=message): + self._derive(_person_table([row])) + + +class TestOperator: + def test_materializes_outputs_from_present_source_columns(self) -> None: + frame = with_us_work_experience_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD + ) + person = frame.table("person") + assert person["detailed_industry_recode"].dtype == np.int16 + assert person["major_industry_recode"].dtype == np.int16 + assert person["worked_last_year"].dtype == bool + assert int(person["worked_last_year"].sum()) == 52 + assert int(person["detailed_industry_recode"].eq(23).sum()) == 30 + assert int(person["detailed_industry_recode"].eq(0).sum()) == 18 + worker = person["detailed_industry_recode"].between(1, 22) + assert worker.equals(person["worked_last_year"]) + + def test_requires_the_frozen_weeks_column(self) -> None: + source = _us_frame(_plausible_rows()) + tables = {entity: source.table(entity).copy() for entity in source.entities} + tables["person"] = tables["person"].drop(columns=["WKSWORK"]) + stripped = Frame( + tables, + source.schema, + {entity: source.weights_for(entity) for entity in source.weighted_entities}, + source.strata, + ) + with pytest.raises(SourceRuntimeError, match="WKSWORK"): + with_us_work_experience_inputs(stripped, seed=0, time_period=TIME_PERIOD) + + def test_requires_the_sidecar_when_recodes_are_absent(self) -> None: + source = _us_frame(_plausible_rows()) + tables = {entity: source.table(entity).copy() for entity in source.entities} + tables["person"] = tables["person"].drop(columns=["WEIND", "WEMIND"]) + stripped = Frame( + tables, + source.schema, + {entity: source.weights_for(entity) for entity in source.weighted_entities}, + source.strata, + ) + with pytest.raises(SourceRuntimeError, match="pinned ASEC work-experience"): + with_us_work_experience_inputs(stripped, seed=0, time_period=TIME_PERIOD) + + def test_restores_recodes_from_the_sidecar_by_identity(self) -> None: + rows = _plausible_rows() + source = _us_frame(rows) + tables = {entity: source.table(entity).copy() for entity in source.entities} + person = tables["person"] + n = len(person) + person["source_year"] = 2023 + person["PERIDNUM"] = [f"{index:022d}" for index in range(n)] + sidecar = pd.DataFrame( + { + "source_year": np.full(n, 2023, dtype=np.int64), + "PH_SEQ": np.arange(1, n + 1, dtype=np.int64), + "P_SEQ": np.ones(n, dtype=np.int64), + "A_LINENO": np.ones(n, dtype=np.int64), + "PERIDNUM": person["PERIDNUM"].to_numpy(), + "WEIND": person["WEIND"].to_numpy(), + "WEMIND": person["WEMIND"].to_numpy(), + } + ) + tables["person"] = person.drop(columns=["WEIND", "WEMIND"]) + stripped = Frame( + tables, + source.schema, + {entity: source.weights_for(entity) for entity in source.weighted_entities}, + source.strata, + ) + restored = with_us_work_experience_inputs( + stripped, + seed=0, + time_period=TIME_PERIOD, + asec_work_experience_source=sidecar, + ) + direct = with_us_work_experience_inputs(source, seed=0, time_period=TIME_PERIOD) + for column in _OUTPUTS: + assert restored.table("person")[column].tolist() == ( + direct.table("person")[column].tolist() + ) + + def test_passes_a_healthy_surface_through_unchanged(self) -> None: + first = with_us_work_experience_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD + ) + second = with_us_work_experience_inputs(first, seed=1, time_period=TIME_PERIOD) + assert second is first + + def test_both_support_clones_carry_identical_values(self) -> None: + expanded = clone_us_frame_for_puf_support(_us_frame(_plausible_rows())) + materialized = with_us_work_experience_inputs( + expanded, seed=0, time_period=TIME_PERIOD + ) + summary = us_work_experience_summary(materialized) + channels = summary["channels"] + assert set(channels) == { + BASE_ASEC_SUPPORT_CHANNEL, + PUF_TAX_DETAIL_SUPPORT_CHANNEL, + } + asec = channels[BASE_ASEC_SUPPORT_CHANNEL] + puf = channels[PUF_TAX_DETAIL_SUPPORT_CHANNEL] + assert asec["rows"] == puf["rows"] == 100 + assert asec["worked_share"] == pytest.approx(puf["worked_share"]) + assert asec["recode_positive_share"] == pytest.approx( + puf["recode_positive_share"] + ) + + +class TestSignalGate: + def test_plausible_surface_passes_with_zero_identity_breaks(self) -> None: + frame = with_us_work_experience_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD + ) + gate = us_work_experience_signal_gate(frame) + assert gate.passed, gate.failures + details = gate.details + assert details["worked_share"] == pytest.approx(0.52) + assert details["recode_positive_share"] == pytest.approx(0.82) + assert details["never_worked_share"] == pytest.approx(0.30) + assert details["universe_identity_breaks"] == 0 + assert details["recode_zero_breaks"] == 0 + + def test_missing_outputs_fail_the_gate(self) -> None: + gate = us_work_experience_signal_gate(_us_frame(_plausible_rows())) + assert not gate.passed + assert "person columns missing" in gate.failures[0] + + def test_collapsed_surface_fails_the_plausibility_bands(self) -> None: + frame = with_us_work_experience_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD + ) + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + tables["person"]["worked_last_year"] = False + tables["person"]["detailed_industry_recode"] = np.int16(0) + tables["person"]["major_industry_recode"] = np.int16(0) + collapsed = Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + frame.strata, + ) + gate = us_work_experience_signal_gate(collapsed) + assert not gate.passed + assert any("worked-last-year share" in failure for failure in gate.failures) + assert any("industry-recode share" in failure for failure in gate.failures) + + def test_identity_breaks_fail_the_gate(self) -> None: + frame = with_us_work_experience_inputs( + _us_frame(_plausible_rows()), seed=0, time_period=TIME_PERIOD + ) + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + person = tables["person"] + person.loc[person.index[0], "worked_last_year"] = False + broken = Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + frame.strata, + ) + gate = us_work_experience_signal_gate(broken) + assert not gate.passed + assert any("disagreeing with worked" in failure for failure in gate.failures) diff --git a/packages/microcosm-build/tests/test_us_work_experience_source.py b/packages/microcosm-build/tests/test_us_work_experience_source.py new file mode 100644 index 00000000..868fd3f4 --- /dev/null +++ b/packages/microcosm-build/tests/test_us_work_experience_source.py @@ -0,0 +1,250 @@ +"""The pinned ASEC work-experience sidecar: load, verify, and fill. + +The frozen census_cps inputs never carried the ASEC work-experience industry +recodes ``WEIND``/``WEMIND``, so the work-experience stage restores them from +the official survey archives via an exact per-income-year ``PERIDNUM`` join — +the education-assistance pattern applied to two columns. These tests run +the real loader and fill against synthetic archives with overridden pins, +exercising every fail-closed path including the official universe identity +(``WEIND`` in 1..22 iff ``WKSWORK > 0``) the loader enforces. +""" + +from __future__ import annotations + +import dataclasses +import hashlib +import zipfile +from pathlib import Path + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.us_runtime.work_experience_source import ( + ASEC_WORK_EXPERIENCE_ARCHIVES, + ASEC_WORK_EXPERIENCE_INCOME_YEARS, + ASEC_WORK_EXPERIENCE_SOURCE_COLUMNS, + AsecWorkExperienceArchive, + fill_asec_work_experience_source, + load_asec_work_experience_sources, +) + +_YEAR = 2023 # income year; survey archive 2024 + + +def _peridnum(index: int) -> str: + return f"{index:022d}" + + +def _source_frame(rows: int = 6) -> pd.DataFrame: + """Six people: three workers with industries, two nonworkers, one child.""" + + return pd.DataFrame( + { + "PH_SEQ": np.arange(1, rows + 1, dtype=np.int64), + "P_SEQ": np.ones(rows, dtype=np.int64), + "A_LINENO": np.ones(rows, dtype=np.int64), + "PERIDNUM": [_peridnum(index) for index in range(rows)], + "WEIND": [7, 0, 21, 23, 16, 0][:rows], + "WEMIND": [5, 0, 13, 15, 10, 0][:rows], + "WKSWORK": [52, 0, 48, 0, 10, 0][:rows], + "WORKYN": [1, 2, 1, 2, 2, 0][:rows], + "A_FNLWGT": np.full(rows, 100.0), + } + ) + + +def _write_archive(path: Path, frame: pd.DataFrame, member: str) -> None: + with zipfile.ZipFile(path, "w", compression=zipfile.ZIP_DEFLATED) as archive: + archive.writestr(member, frame.to_csv(index=False)) + + +def _pins_for( + path: Path, frame: pd.DataFrame, member: str +) -> AsecWorkExperienceArchive: + zip_bytes = path.read_bytes() + with zipfile.ZipFile(path) as archive: + info = archive.getinfo(member) + member_bytes = archive.read(member) + weights = frame["A_FNLWGT"].to_numpy(dtype=np.float64) / 100.0 + worked = frame["WKSWORK"].to_numpy() > 0 + recode = frame["WEIND"].to_numpy() != 0 + return AsecWorkExperienceArchive( + survey_year=_YEAR + 1, + income_year=_YEAR, + zip_url="https://example.invalid/asec.zip", + zip_size_bytes=len(zip_bytes), + zip_sha256=hashlib.sha256(zip_bytes).hexdigest(), + member=member, + member_size_bytes=info.file_size, + member_crc32=f"{info.CRC:08x}", + member_sha256=hashlib.sha256(member_bytes).hexdigest(), + rows=len(frame), + worked_rows=int(worked.sum()), + weighted_worked_share=float(weights[worked].sum() / weights.sum()), + recode_rows=int(recode.sum()), + weighted_recode_share=float(weights[recode].sum() / weights.sum()), + ) + + +def _pin(tmp_path: Path, monkeypatch, frame: pd.DataFrame) -> Path: + member = f"pppub{str(_YEAR + 1)[2:]}.csv" + path = tmp_path / f"asecpub{str(_YEAR + 1)[2:]}csv.zip" + _write_archive(path, frame, member) + monkeypatch.setitem( + ASEC_WORK_EXPERIENCE_ARCHIVES, _YEAR, _pins_for(path, frame, member) + ) + return path + + +@pytest.fixture +def pinned_archive(tmp_path, monkeypatch): + frame = _source_frame() + return _pin(tmp_path, monkeypatch, frame), frame + + +def test_income_years_map_to_next_survey_year() -> None: + """Every pinned archive is the survey published the year after income.""" + + for income_year, pins in ASEC_WORK_EXPERIENCE_ARCHIVES.items(): + assert pins.survey_year == income_year + 1 + assert str(pins.survey_year) in pins.zip_url + assert pins.member == f"pppub{str(pins.survey_year)[2:]}.csv" + assert 0.0 < pins.weighted_worked_share < pins.weighted_recode_share < 1.0 + assert 0 < pins.worked_rows < pins.recode_rows < pins.rows + assert ASEC_WORK_EXPERIENCE_INCOME_YEARS == (2022, 2023, 2024) + + +def test_source_columns_carry_only_identity_and_industry_recodes() -> None: + assert ASEC_WORK_EXPERIENCE_SOURCE_COLUMNS == ( + "PH_SEQ", + "P_SEQ", + "A_LINENO", + "PERIDNUM", + "WEIND", + "WEMIND", + ) + + +def test_loader_reads_pinned_zip_and_audits(pinned_archive) -> None: + path, frame = pinned_archive + source = load_asec_work_experience_sources({_YEAR: path}, income_years=(_YEAR,)) + assert list(source["source_year"].unique()) == [_YEAR] + assert len(source) == len(frame) + assert list(source.columns) == ["source_year", *ASEC_WORK_EXPERIENCE_SOURCE_COLUMNS] + audit = source.attrs["source_audit"][_YEAR] + assert audit["worked_rows"] == 3 + assert audit["recode_rows"] == 4 + assert audit["weighted_worked_share"] == pytest.approx(0.5) + assert audit["weighted_recode_share"] == pytest.approx(4 / 6) + + +def test_loader_rejects_wrong_zip_bytes(pinned_archive, tmp_path) -> None: + path, frame = pinned_archive + tampered = tmp_path / "tampered.zip" + member = f"pppub{str(_YEAR + 1)[2:]}.csv" + corrupted = frame.copy() + corrupted.loc[0, "WEIND"] = 8 + _write_archive(tampered, corrupted, member) + with pytest.raises(ValueError, match="mismatch"): + load_asec_work_experience_sources({_YEAR: tampered}, income_years=(_YEAR,)) + + +def test_loader_rejects_unpinned_income_year(pinned_archive) -> None: + path, _ = pinned_archive + with pytest.raises(ValueError, match="covers income year"): + load_asec_work_experience_sources({1999: path}, income_years=(1999,)) + + +@pytest.mark.parametrize( + ("mutation", "message"), + ( + ({"WEIND": 24}, r"WEIND must be an integer in \[0, 23\]"), + ({"WEMIND": 16}, r"WEMIND must be an integer in \[0, 15\]"), + ({"WEIND": 0, "WEMIND": 0}, "universe identity"), + ({"WEMIND": 0}, "disagree on the not-in-universe rows"), + ({"WORKYN": 1, "WKSWORK": 0, "WEIND": 23}, "WORKYN = 1 without positive"), + ), +) +def test_loader_enforces_official_universe_identities( + tmp_path, monkeypatch, mutation: dict, message: str +) -> None: + frame = _source_frame() + for column, value in mutation.items(): + frame.loc[0, column] = value + path = _pin(tmp_path, monkeypatch, frame) + with pytest.raises(ValueError, match=message): + load_asec_work_experience_sources({_YEAR: path}, income_years=(_YEAR,)) + + +def test_pins_reject_audit_drift(pinned_archive, monkeypatch) -> None: + path, _ = pinned_archive + pins = ASEC_WORK_EXPERIENCE_ARCHIVES[_YEAR] + monkeypatch.setitem( + ASEC_WORK_EXPERIENCE_ARCHIVES, + _YEAR, + dataclasses.replace(pins, worked_rows=pins.worked_rows + 1), + ) + with pytest.raises(ValueError, match="audit drifted"): + load_asec_work_experience_sources({_YEAR: path}, income_years=(_YEAR,)) + + +def _person_frame(rows: int = 4) -> pd.DataFrame: + return pd.DataFrame( + { + "source_year": np.full(rows, _YEAR, dtype=np.int64), + "PERIDNUM": [_peridnum(index) for index in range(rows)], + "PH_SEQ": np.arange(1, rows + 1, dtype=np.int64), + "P_SEQ": np.ones(rows, dtype=np.int64), + "A_LINENO": np.ones(rows, dtype=np.int64), + } + ) + + +def test_fill_joins_both_recodes_by_identity(pinned_archive) -> None: + path, frame = pinned_archive + source = load_asec_work_experience_sources({_YEAR: path}, income_years=(_YEAR,)) + filled = fill_asec_work_experience_source(_person_frame(), source) + expected = frame.set_index("PERIDNUM") + for _, row in filled.iterrows(): + assert row["WEIND"] == expected.loc[row["PERIDNUM"], "WEIND"] + assert row["WEMIND"] == expected.loc[row["PERIDNUM"], "WEMIND"] + assert filled["WEIND"].dtype == np.int64 + assert filled["WEMIND"].dtype == np.int64 + assert filled.attrs["work_experience_source_audit"][_YEAR]["worked_rows"] == 3 + + +def test_fill_fails_closed_on_uncovered_key(pinned_archive) -> None: + path, _ = pinned_archive + source = load_asec_work_experience_sources({_YEAR: path}, income_years=(_YEAR,)) + person = _person_frame() + person.loc[0, "PERIDNUM"] = _peridnum(999) + with pytest.raises(ValueError, match="does not cover frame PERIDNUM"): + fill_asec_work_experience_source(person, source) + + +def test_fill_fails_closed_on_uncovered_year(pinned_archive) -> None: + path, _ = pinned_archive + source = load_asec_work_experience_sources({_YEAR: path}, income_years=(_YEAR,)) + person = _person_frame() + person["source_year"] = _YEAR - 1 + with pytest.raises(ValueError, match="does not cover pooled income"): + fill_asec_work_experience_source(person, source) + + +def test_fill_fails_closed_on_identity_mismatch(pinned_archive) -> None: + path, _ = pinned_archive + source = load_asec_work_experience_sources({_YEAR: path}, income_years=(_YEAR,)) + person = _person_frame() + person.loc[1, "A_LINENO"] = 9 + with pytest.raises(ValueError, match="redundant identity mismatch"): + fill_asec_work_experience_source(person, source) + + +def test_fill_refuses_to_overwrite_measured_values(pinned_archive) -> None: + path, _ = pinned_archive + source = load_asec_work_experience_sources({_YEAR: path}, income_years=(_YEAR,)) + person = _person_frame() + person["WEMIND"] = 1 + with pytest.raises(ValueError, match="refusing to overwrite"): + fill_asec_work_experience_source(person, source) diff --git a/tools/build_us_puf_support_base.py b/tools/build_us_puf_support_base.py index 5bdd07fe..b08c31a1 100644 --- a/tools/build_us_puf_support_base.py +++ b/tools/build_us_puf_support_base.py @@ -51,6 +51,7 @@ ASEC_RAW_STAGE_OPERATOR_STATUS, ASEC_RAW_STAGE_SCHEMA_VERSION, ASEC_RAW_STAGE_STAGE, + ASEC_WORK_EXPERIENCE_ARCHIVES, BASE_ASEC_SUPPORT_CHANNEL, CONGRESSIONAL_DISTRICT_VINTAGE_CROSSWALK_SHA256_ATTR, CONGRESSIONAL_DISTRICT_VINTAGE_TARGET_ATTR, @@ -76,6 +77,7 @@ fill_asec_2022_weeks_unemployed_source, fill_asec_education_assistance_source, fill_asec_public_assistance_type_source, + fill_asec_work_experience_source, finalize_puf_e01000_reconciliation, impute_us_housing_assistance_to_puf_support, impute_us_puf_tax_detail_support, @@ -84,6 +86,7 @@ load_asec_education_assistance_sources, load_asec_public_assistance_type_sources, load_asec_raw_stage_checkpoint, + load_asec_work_experience_sources, load_congressional_district_vintage_crosswalk, load_us_block_ladder, puf_tax_unit_donor_from_arrays, @@ -122,6 +125,7 @@ us_source_operation_handlers, us_weeks_unemployed_signal_gate, us_wic_claim_signal_gate, + us_work_experience_signal_gate, us_workers_compensation_signal_gate, validate_puf_capital_gains_tail_manifest, with_household_congressional_districts, @@ -144,6 +148,7 @@ with_us_retirement_distribution_inputs, with_us_weeks_unemployed, with_us_wic_claim_input, + with_us_work_experience_inputs, with_us_workers_compensation, write_puf_capital_gains_tail_manifest, ) @@ -185,6 +190,7 @@ "retirement_contributions_post_clone", "retirement_distributions_post_clone", "education_inputs_post_clone", + "work_experience_inputs_post_clone", "congressional_district_assignment", "block_ladder_assignment", "final_export", @@ -260,6 +266,10 @@ ("with_us_retirement_distribution_inputs",), ), ("education_inputs_post_clone", ("with_us_education_inputs",)), + ( + "work_experience_inputs_post_clone", + ("with_us_work_experience_inputs",), + ), ( "congressional_district_assignment", ("with_household_congressional_districts",), @@ -322,10 +332,11 @@ 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 " - "official Census archive and verified against the same pins." + "pppub member) restoring that pooled income year's ED_VAL, " + "PAW_TYP, and WEIND/WEMIND (income year YYYY maps to the " + "survey-year YYYY+1 archive). Years without a mapping are " + "fetched from the official Census archive and verified against " + "the same pins." ), ) parser.add_argument( @@ -900,6 +911,10 @@ def _run_all( _asec_education_source_paths(args), income_years=_pooled_income_years(args), ) + work_experience_source = load_asec_work_experience_sources( + _asec_education_source_paths(args), + income_years=_pooled_income_years(args), + ) base = derive_us_cps_carried_inputs( raw_base, public_assistance_type_source=public_assistance_type_source, @@ -1326,6 +1341,21 @@ def _run_all( + "\n ".join(education_inputs_gate.failures) ) _observe_frame_boundary(boundary_observer, "education_inputs_post_clone", imputed) + imputed = with_us_work_experience_inputs( + imputed, + seed=args.seed, + time_period=args.target_year, + asec_work_experience_source=work_experience_source, + ) + work_experience_gate = us_work_experience_signal_gate(imputed) + if not work_experience_gate.passed: + raise SystemExit( + "Work-experience signal gate failed:\n " + + "\n ".join(work_experience_gate.failures) + ) + _observe_frame_boundary( + boundary_observer, "work_experience_inputs_post_clone", imputed + ) congressional_district_assignment = {"applied": False} if args.assign_congressional_districts: ledger_facts = load_ledger_consumer_artifact(args.ledger_facts).facts @@ -1909,6 +1939,10 @@ def _asec_raw_source_mapping_frame( _asec_education_source_paths(args), income_years=_pooled_income_years(args), ) + work_experience_source = load_asec_work_experience_sources( + _asec_education_source_paths(args), + income_years=_pooled_income_years(args), + ) tables = { entity: source_frame.table(entity).copy(deep=True) for entity in source_frame.entities @@ -1922,6 +1956,7 @@ def _asec_raw_source_mapping_frame( person, public_assistance_type_source, ) + person = fill_asec_work_experience_source(person, work_experience_source) tables["person"] = person raw_frame = Frame( tables, @@ -1953,6 +1988,18 @@ def _asec_raw_source_mapping_frame( } for income_year in _pooled_income_years(args) ] + work_experience_pins = [ + { + "income_year": income_year, + "locator": ASEC_WORK_EXPERIENCE_ARCHIVES[income_year].zip_url, + "member": ASEC_WORK_EXPERIENCE_ARCHIVES[income_year].member, + "member_sha256": ( + ASEC_WORK_EXPERIENCE_ARCHIVES[income_year].member_sha256 + ), + "sha256": ASEC_WORK_EXPERIENCE_ARCHIVES[income_year].zip_sha256, + } + for income_year in _pooled_income_years(args) + ] return raw_frame, { "ED_VAL": { "audit": dict(education_source.attrs.get("source_audit", {})), @@ -1988,6 +2035,25 @@ def _asec_raw_source_mapping_frame( "operation": "exact_source_join", "source_pins": education_pins, }, + # WEIND/WEMIND live in the same pinned survey-year person members; + # the work-experience sidecar pins the identical archives with its + # own audit statistics (microcosm#719). + "WEIND": { + "audit": dict(work_experience_source.attrs.get("source_audit", {})), + "column": "WEIND", + "entity": "person", + "join_keys": ["source_year", "PERIDNUM"], + "operation": "exact_source_join", + "source_pins": work_experience_pins, + }, + "WEMIND": { + "audit": dict(work_experience_source.attrs.get("source_audit", {})), + "column": "WEMIND", + "entity": "person", + "join_keys": ["source_year", "PERIDNUM"], + "operation": "exact_source_join", + "source_pins": work_experience_pins, + }, } @@ -2666,6 +2732,21 @@ def _post_qrf_frame_stage( us_education_inputs_signal_gate(frame), "Education-input signal gate failed", ) + elif stage == "work_experience_inputs_post_clone": + work_experience_source = load_asec_work_experience_sources( + _asec_education_source_paths(args), + income_years=_pooled_income_years(args), + ) + frame = with_us_work_experience_inputs( + frame, + seed=args.seed, + time_period=args.target_year, + asec_work_experience_source=work_experience_source, + ) + signals["work_experience_signal"] = _checked_gate_payload( + us_work_experience_signal_gate(frame), + "Work-experience signal gate failed", + ) elif stage == "congressional_district_assignment": assignment: dict[str, object] = {"applied": False} if args.assign_congressional_districts: diff --git a/tools/build_us_release_input_coverage_manifest.py b/tools/build_us_release_input_coverage_manifest.py index 2c4e18d4..e5472690 100644 --- a/tools/build_us_release_input_coverage_manifest.py +++ b/tools/build_us_release_input_coverage_manifest.py @@ -80,6 +80,9 @@ "is_self_employed", "pre_subsidy_care_expenses", "is_incapable_of_self_care", + "detailed_industry_recode", + "major_industry_recode", + "worked_last_year", ) # Per-column annotations for post-reference hard requirements whose absence