From 7c797c4df914c651ad822a3e0fe1f639832435c0 Mon Sep 17 00:00:00 2001 From: Daphne Hansell <128793799+daphnehanse11@users.noreply.github.com> Date: Mon, 6 Jul 2026 09:23:32 -0400 Subject: [PATCH] Derive hours-worked inputs from ASEC reported hours Add a hours_worked source stage (manifest entry, runtime handler, frame transform, release gate, builder wiring) following the immigration template. Direct-maps measured ASEC variables (HRSWK, A_HRS1, WKSWORK), heals the published constant-40 landmine, and hands weekly-hours ownership off from org_wages. Closes #242; unblocks #248. --- .../src/populace/build/source_manifest.py | 1 + .../src/populace/build/us/source_stages.json | 39 ++- .../src/populace/build/us_runtime/__init__.py | 31 ++ .../populace/build/us_runtime/hours_worked.py | 328 ++++++++++++++++++ .../build/us_runtime/l0_refit_export.py | 8 +- .../build/us_runtime/source_runtime.py | 4 + .../tests/test_us_fiscal_refresh_builder.py | 14 + .../tests/test_us_hours_worked.py | 298 ++++++++++++++++ .../tests/test_us_l0_refit_export.py | 3 + tools/build_us_fiscal_refresh_release.py | 75 +++- 10 files changed, 793 insertions(+), 8 deletions(-) create mode 100644 packages/populace-build/src/populace/build/us_runtime/hours_worked.py create mode 100644 packages/populace-build/tests/test_us_hours_worked.py diff --git a/packages/populace-build/src/populace/build/source_manifest.py b/packages/populace-build/src/populace/build/source_manifest.py index e6304963..cc2f183a 100644 --- a/packages/populace-build/src/populace/build/source_manifest.py +++ b/packages/populace-build/src/populace/build/source_manifest.py @@ -50,6 +50,7 @@ "compute_ratio", "declare_income_reference_offset", "derive", + "derive_hours_worked", "derive_immigration_status", "derive_puf_policyengine_variables", "derive_mortgage_balance_hints", diff --git a/packages/populace-build/src/populace/build/us/source_stages.json b/packages/populace-build/src/populace/build/us/source_stages.json index f3a65a5e..c923569e 100644 --- a/packages/populace-build/src/populace/build/us/source_stages.json +++ b/packages/populace-build/src/populace/build/us/source_stages.json @@ -346,14 +346,13 @@ "hourly_wage", "is_paid_hourly", "is_union_member_or_covered", - "weekly_hours_worked_before_lsr", "fsla_overtime_premium" ], "nonnegative_outputs": [ "hourly_wage", - "weekly_hours_worked_before_lsr", "fsla_overtime_premium" - ] + ], + "notes": "Usual weekly hours are owned by the hours_worked stage (ASEC HRSWK), matching the retired enhanced-CPS pipeline mapping; this stage consumes them as predictors." }, { "stage": "meps_esi_premiums", @@ -486,6 +485,40 @@ ], "notes": "Citizenship is measured (PRCITSHP), not imputed. Non-citizens with any ASEC-UA legal-status indicator (Van Hook et al., SSRN 4662801) become OTHER_NON_CITIZEN. The CPS carries no work-authorization variable, so among residual non-citizens, workers and students spill to NON_CITIZEN_VALID_EAD in deterministic seeded order until the remaining undocumented worker count matches Pew's 8.3M and the undocumented student count matches the Higher Ed Immigration Portal's ~408k \u2014 the only forced margins. The total undocumented population is emergent (\u224813M on the 2024 ASEC, within the range of published 2023-24 estimates) and is gated against Pew's 11.0M 2022 anchor with a coarse plausibility band rather than forced; reconciling the level is the calibration lane's job. Immigration-status tags carry only statutory tests the data supports: DACA (arrival cohort among EAD holders) and CUBAN_HAITIAN_ENTRANT (nativity plus post-1980 arrival); other documented non-citizens stay LEGAL_PERMANENT_RESIDENT, the modal true status \u2014 blanket REFUGEE/TPS labels are deliberately not fabricated because they would mislabel millions and over-grant refugee-class benefit exemptions." }, + { + "stage": "hours_worked", + "survey": "Census CPS ASEC", + "source": "https://www.census.gov/programs-surveys/cps.html", + "grain": "person", + "artifacts": [ + { + "kind": "public_microdata", + "format": "census_asec", + "vintage": "build_year", + "locator": "Census CPS ASEC person file usual weekly hours (HRSWK), reference-week hours (A_HRS1), and weeks worked (WKSWORK)" + } + ], + "operations": [ + { + "kind": "read_table", + "table": "person" + }, + { + "kind": "derive_hours_worked" + } + ], + "outputs": [ + "weekly_hours_worked_before_lsr", + "hours_worked_last_week", + "weeks_worked" + ], + "nonnegative_outputs": [ + "weekly_hours_worked_before_lsr", + "hours_worked_last_week", + "weeks_worked" + ], + "notes": "Direct mappings of measured ASEC hours variables, identical to the retired enhanced-CPS pipeline: weekly_hours_worked_before_lsr <- HRSWK (usual hours per week last year, 0 for non-workers), hours_worked_last_week <- A_HRS1, weeks_worked <- WKSWORK clipped to [0, 52]. Nothing is imputed. Without this stage PolicyEngine-US defaults every person to 40 weekly hours, so the SNAP 30-hour general and 20-hour ABAWD work requirements pass universally (populace #242/#248). HRSWK is an annual usual-hours measure while the ABAWD standard is a monthly 80-hour test; week-by-week volatility is not observable in the CPS and is deliberately out of scope." + }, { "stage": "mortgage_conversion", "survey": "IRS PUF 2015 (uprated) + Fed SCF 2022", diff --git a/packages/populace-build/src/populace/build/us_runtime/__init__.py b/packages/populace-build/src/populace/build/us_runtime/__init__.py index 6fb0e12b..39be1c1e 100644 --- a/packages/populace-build/src/populace/build/us_runtime/__init__.py +++ b/packages/populace-build/src/populace/build/us_runtime/__init__.py @@ -108,6 +108,17 @@ us_geography_ladder_gate, with_household_us_geography_ladder, ) +from populace.build.us_runtime.hours_worked import ( + US_HOURS_WORKED_NONCONSTANT_PERSON_COLUMNS, + US_HOURS_WORKED_OUTPUT_COLUMNS, + US_HOURS_WORKED_REQUIRED_SOURCE_COLUMNS, + US_HOURS_WORKED_STAGE_NAME, + derive_us_hours_worked_from_manifest, + us_hours_worked_signal_gate, + us_hours_worked_stage_spec, + us_hours_worked_summary, + with_us_hours_worked_inputs, +) from populace.build.us_runtime.immigration import ( IMMIGRATION_STATUS_VALUES, SSN_CARD_TYPE_VALUES, @@ -259,6 +270,15 @@ "US_IMMIGRATION_REQUIRED_SOURCE_COLUMNS", "US_IMMIGRATION_STAGE_NAME", "UndocumentedControls", + "US_HOURS_WORKED_NONCONSTANT_PERSON_COLUMNS", + "US_HOURS_WORKED_OUTPUT_COLUMNS", + "US_HOURS_WORKED_REQUIRED_SOURCE_COLUMNS", + "US_HOURS_WORKED_STAGE_NAME", + "derive_us_hours_worked_from_manifest", + "us_hours_worked_signal_gate", + "us_hours_worked_stage_spec", + "us_hours_worked_summary", + "with_us_hours_worked_inputs", "derive_us_immigration_status_from_manifest", "us_immigration_composition_gate", "us_immigration_composition_summary", @@ -452,6 +472,16 @@ def to_manifest(self) -> dict[str, object]: "undocumented population/worker/student control totals." ), ), + US_HOURS_WORKED_STAGE_NAME: DonorSpec( + survey="Census CPS ASEC", + source="https://www.census.gov/programs-surveys/cps.html", + notes=( + "Hours-worked inputs mapped directly from measured ASEC person " + "variables (HRSWK, A_HRS1, WKSWORK) — nothing imputed. Without " + "them the engine defaults every person to 40 weekly hours and " + "hours-conditioned rules (SNAP work requirements) become no-ops." + ), + ), "puf_tax_detail": DonorSpec( survey="IRS PUF 2015 (uprated)", source="https://www.irs.gov/statistics/soi-tax-stats-individual-public-use-microdata-files", @@ -498,6 +528,7 @@ def to_manifest(self) -> dict[str, object]: "unit_assignment", "derive_cps_carried", US_IMMIGRATION_STAGE_NAME, + US_HOURS_WORKED_STAGE_NAME, US_PUF_SUPPORT_STAGE_NAME, "puf_tax_detail", "capital_gain_distributions", diff --git a/packages/populace-build/src/populace/build/us_runtime/hours_worked.py b/packages/populace-build/src/populace/build/us_runtime/hours_worked.py new file mode 100644 index 00000000..cbd1f92c --- /dev/null +++ b/packages/populace-build/src/populace/build/us_runtime/hours_worked.py @@ -0,0 +1,328 @@ +"""Hours-worked inputs from CPS ASEC reported hours and weeks. + +Without this stage the published dataset stores no +``weekly_hours_worked_before_lsr`` (or stores it constant at the engine's +40-hour default), so PolicyEngine-US treats every person as working 40 +hours per week and every hours-conditioned rule becomes a no-op — the +failure mode of populace issues #242/#248 (SNAP's 30-hour general and +20-hour ABAWD work requirements pass for everyone, erasing work-requirement +reform exposure entirely). + +The stage writes three PolicyEngine-facing person input columns, each a +direct mapping of a measured CPS ASEC person variable — the same mappings +the retired enhanced-CPS pipeline used, so parity with the retired +enhanced-CPS hours surface is by construction rather than by imputation: + +- ``weekly_hours_worked_before_lsr`` ← ``HRSWK`` (usual hours worked per + week last year; zero for persons who did not work). +- ``hours_worked_last_week`` ← ``A_HRS1`` (hours worked at all jobs in the + reference week). +- ``weeks_worked`` ← ``WKSWORK`` clipped to [0, 52]. + +Semantics note (documented, not hidden): SNAP's ABAWD standard is a monthly +80-hour test. ``HRSWK`` is an annual usual-hours measure, so a person who +worked half the year at 40 hours/week carries 40 here — the engine's +work-requirement variables read usual weekly hours, matching how the +retired enhanced-CPS pipeline fed them. Week-by-week volatility is not in +the CPS and is out of scope for this stage. + +Healing behavior: a frame that already carries all three columns *with +signal* passes through untouched (idempotent). A frame carrying a constant +``weekly_hours_worked_before_lsr`` — the published constant-40 landmine — +is recomputed from the raw columns rather than trusted, because a constant +hours column is indistinguishable from the engine default and masks the +missing derivation this stage exists to fix. +""" + +from __future__ import annotations + +from importlib.resources import files + +import numpy as np +import pandas as pd + +from populace.build.gates import GateResult +from populace.build.source_manifest import ( + SourceOperationSpec, + SourceStageSpec, + load_source_manifest, +) +from populace.build.source_runtime import ( + SourceRuntimeConfig, + SourceRuntimeContext, + SourceRuntimeError, + run_source_stage, +) +from populace.frame import Frame +from populace.frame.units import US_SCHEMA + +__all__ = [ + "US_HOURS_WORKED_NONCONSTANT_PERSON_COLUMNS", + "US_HOURS_WORKED_OUTPUT_COLUMNS", + "US_HOURS_WORKED_REQUIRED_SOURCE_COLUMNS", + "US_HOURS_WORKED_STAGE_NAME", + "derive_us_hours_worked_from_manifest", + "us_hours_worked_signal_gate", + "us_hours_worked_summary", + "us_hours_worked_stage_spec", + "with_us_hours_worked_inputs", +] + +US_HOURS_WORKED_STAGE_NAME = "hours_worked" + +#: The PolicyEngine-facing person input columns this stage owns. +US_HOURS_WORKED_OUTPUT_COLUMNS: tuple[str, ...] = ( + "weekly_hours_worked_before_lsr", + "hours_worked_last_week", + "weeks_worked", +) + +#: Release gates require these person columns to carry signal (≥2 values). +US_HOURS_WORKED_NONCONSTANT_PERSON_COLUMNS: tuple[str, ...] = ( + US_HOURS_WORKED_OUTPUT_COLUMNS +) + +#: Raw CPS ASEC person columns the derivation reads. +US_HOURS_WORKED_REQUIRED_SOURCE_COLUMNS: tuple[str, ...] = ( + "HRSWK", + "A_HRS1", + "WKSWORK", +) + +_WEEKS_PER_YEAR = 52 + +#: Weighted share of persons with positive usual weekly hours must land in +#: this band. CPS ASEC 2024: ~50% of all persons (children included) worked +#: last year; the retired enhanced-CPS surface measured 50.4%. A share +#: outside the band means the derivation read the wrong column or the raw +#: hours collapsed. +_WORKED_SHARE_BAND = (0.35, 0.62) + +#: Weighted mean usual weekly hours among persons with positive hours must +#: land in this band (BLS/CPS average weekly hours for workers ≈ 38.5). +_MEAN_WEEKLY_HOURS_BAND = (30.0, 45.0) + +_PERSON_WEIGHT_COLUMN = "person_weight" + +_DERIVE_HOURS_WORKED_PARAMETER_KEYS = frozenset() + + +def us_hours_worked_stage_spec() -> SourceStageSpec: + """Load the packaged ``hours_worked`` source-stage manifest entry.""" + + manifest = load_source_manifest( + files("populace.build.us").joinpath("source_stages.json") + ) + stage_map = manifest.stage_map() + if US_HOURS_WORKED_STAGE_NAME not in stage_map: + raise ValueError( + f"US source manifest declares no {US_HOURS_WORKED_STAGE_NAME!r} stage." + ) + return stage_map[US_HOURS_WORKED_STAGE_NAME] + + +def derive_us_hours_worked_from_manifest( + frame: pd.DataFrame | None, + operation: SourceOperationSpec, + _context: SourceRuntimeContext, +) -> pd.DataFrame: + """Map raw ASEC hours/weeks columns onto the PolicyEngine input names. + + The current frame must be the raw-column person table (from the stage's + ``read_table`` operation). + """ + + if operation.kind != "derive_hours_worked": + raise SourceRuntimeError( + f"US hours derivation received unexpected operation {operation.kind!r}." + ) + if frame is None: + raise SourceRuntimeError( + "US hours derivation requires the person table to be read first." + ) + unexpected = sorted(set(operation.parameters) - _DERIVE_HOURS_WORKED_PARAMETER_KEYS) + if unexpected: + raise SourceRuntimeError( + f"US hours derivation received unsupported parameter(s): {unexpected}." + ) + missing = [ + column + for column in US_HOURS_WORKED_REQUIRED_SOURCE_COLUMNS + if column not in frame.columns + ] + if missing: + raise SourceRuntimeError( + f"US hours derivation requires raw ASEC column(s): {missing}." + ) + + def _nonnegative(column: str) -> np.ndarray: + values = pd.to_numeric(frame[column], errors="coerce").fillna(0.0) + return np.maximum(values.to_numpy(dtype=np.float64), 0.0) + + result = frame.copy(deep=True) + result["weekly_hours_worked_before_lsr"] = _nonnegative("HRSWK") + result["hours_worked_last_week"] = _nonnegative("A_HRS1") + result["weeks_worked"] = np.minimum(_nonnegative("WKSWORK"), _WEEKS_PER_YEAR) + return result + + +def _weekly_hours_carry_signal(person: pd.DataFrame) -> bool: + """Whether the persisted weekly-hours column is trustworthy as-is. + + A constant column (one observed value) is indistinguishable from the + engine's broadcast default and must be recomputed, not passed through. + """ + + values = pd.to_numeric( + person["weekly_hours_worked_before_lsr"], errors="coerce" + ).dropna() + return values.nunique() > 1 + + +def with_us_hours_worked_inputs(frame: Frame, *, seed: int, time_period: int) -> Frame: + """Run the ``hours_worked`` manifest stage over a US frame. + + A frame already carrying all three output columns with a non-constant + weekly-hours distribution passes through untouched (idempotent). Any + other surface — columns missing, or weekly hours constant at the + engine default — is recomputed from the raw ASEC columns. + + Args: + frame: A US-schema frame whose person table still carries the raw + CPS ASEC source columns (unless the outputs already carry + signal). + seed: Build-wide imputation seed (recorded in the stage run; the + derivation itself is deterministic). + time_period: The dataset's time period. + + Returns: + A new frame whose person table carries all three hours columns. + + Raises: + ValueError: If the frame is not US-schema or the stage output does + not cover every person. + SourceRuntimeError: If required raw ASEC columns are missing. + """ + + if frame.schema != US_SCHEMA: + raise ValueError("US hours-worked inputs require the US schema.") + person = frame.table("person") + have_all = all( + column in person.columns for column in US_HOURS_WORKED_OUTPUT_COLUMNS + ) + if have_all and _weekly_hours_carry_signal(person): + return frame + + stage_person = person.copy(deep=True) + stage_person[_PERSON_WEIGHT_COLUMN] = frame.resolve_weights("person").values + output = run_source_stage( + us_hours_worked_stage_spec(), + tables={"person": stage_person}, + operation_handlers={ + "derive_hours_worked": derive_us_hours_worked_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_HOURS_WORKED_OUTPUT_COLUMNS: + if aligned[column].isna().any(): + raise ValueError( + f"US hours-worked stage output does not cover every person for " + f"{column!r}." + ) + + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + for column in US_HOURS_WORKED_OUTPUT_COLUMNS: + tables["person"][column] = aligned[column].to_numpy(dtype=np.float64) + return Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + frame.strata, + mass_log=frame.mass_log, + ) + + +def us_hours_worked_summary(frame: Frame) -> dict[str, object]: + """Weighted hours-distribution summary for gates and release manifests.""" + + person = frame.table("person") + weights = np.asarray(frame.resolve_weights("person").values, dtype=np.float64) + weekly = pd.to_numeric( + person["weekly_hours_worked_before_lsr"], errors="coerce" + ).fillna(0.0) + weekly_values = weekly.to_numpy(dtype=np.float64) + total_weight = float(weights.sum()) + positive = weekly_values > 0 + positive_weight = float(weights[positive].sum()) + worked_share = positive_weight / total_weight if total_weight > 0 else 0.0 + mean_weekly_hours_workers = ( + float(np.average(weekly_values[positive], weights=weights[positive])) + if positive_weight > 0 + else 0.0 + ) + unique_counts = { + column: int(pd.to_numeric(person[column], errors="coerce").dropna().nunique()) + for column in US_HOURS_WORKED_OUTPUT_COLUMNS + if column in person.columns + } + return { + "worked_share": worked_share, + "mean_weekly_hours_workers": mean_weekly_hours_workers, + "worked_share_band": list(_WORKED_SHARE_BAND), + "mean_weekly_hours_band": list(_MEAN_WEEKLY_HOURS_BAND), + "unique_counts": unique_counts, + } + + +def us_hours_worked_signal_gate(frame: Frame) -> GateResult: + """Require the hours surface to carry a plausible worked distribution. + + Fails when a column is missing or constant, when the weighted share of + persons with positive usual weekly hours leaves the plausibility band, + or when mean weekly hours among workers does — each of which reproduces + (or inverts) the everyone-works-40-hours failure of populace #242. + """ + + person = frame.table("person") + failures: list[str] = [] + missing = [ + column + for column in US_HOURS_WORKED_OUTPUT_COLUMNS + if column not in person.columns + ] + if missing: + return GateResult( + name="hours_worked_signal", + passed=False, + failures=(f"person columns missing: {missing}.",), + details={"missing": missing}, + ) + + summary = us_hours_worked_summary(frame) + for column, count in summary["unique_counts"].items(): + if count < 2: + failures.append( + f"{column}: constant column (one observed value) — the hours " + "surface carries no signal." + ) + worked_share = float(summary["worked_share"]) + low, high = _WORKED_SHARE_BAND + if not (low <= worked_share <= high): + failures.append( + f"worked share {worked_share:.3f} outside plausibility band " + f"[{low}, {high}]." + ) + mean_hours = float(summary["mean_weekly_hours_workers"]) + low, high = _MEAN_WEEKLY_HOURS_BAND + if not (low <= mean_hours <= high): + failures.append( + f"mean weekly hours among workers {mean_hours:.1f} outside " + f"plausibility band [{low}, {high}]." + ) + return GateResult( + name="hours_worked_signal", + passed=not failures, + failures=tuple(failures), + details=summary, + ) diff --git a/packages/populace-build/src/populace/build/us_runtime/l0_refit_export.py b/packages/populace-build/src/populace/build/us_runtime/l0_refit_export.py index dc3607d7..b4cbdaf4 100644 --- a/packages/populace-build/src/populace/build/us_runtime/l0_refit_export.py +++ b/packages/populace-build/src/populace/build/us_runtime/l0_refit_export.py @@ -20,6 +20,9 @@ US_GEOGRAPHY_LADDER_COLUMNS, us_geography_ladder_gate, ) +from populace.build.us_runtime.hours_worked import ( + US_HOURS_WORKED_NONCONSTANT_PERSON_COLUMNS, +) from populace.build.us_runtime.immigration import ( US_IMMIGRATION_NONCONSTANT_PERSON_COLUMNS, ) @@ -32,7 +35,10 @@ "selected_marketplace_plan_benchmark_ratio", ) -US_RELEASE_REQUIRED_PERSON_SOURCE_COLUMNS = US_IMMIGRATION_NONCONSTANT_PERSON_COLUMNS +US_RELEASE_REQUIRED_PERSON_SOURCE_COLUMNS = ( + *US_IMMIGRATION_NONCONSTANT_PERSON_COLUMNS, + *US_HOURS_WORKED_NONCONSTANT_PERSON_COLUMNS, +) #: The geography spine a US release carries by default: state and district, #: plus the block-anchored ladder (populace #275). A release missing or diff --git a/packages/populace-build/src/populace/build/us_runtime/source_runtime.py b/packages/populace-build/src/populace/build/us_runtime/source_runtime.py index 94bb161f..d04ac35f 100644 --- a/packages/populace-build/src/populace/build/us_runtime/source_runtime.py +++ b/packages/populace-build/src/populace/build/us_runtime/source_runtime.py @@ -17,6 +17,9 @@ from populace.build.us_runtime.capital_gain_distributions import ( split_us_component_by_share_from_manifest, ) +from populace.build.us_runtime.hours_worked import ( + derive_us_hours_worked_from_manifest, +) from populace.build.us_runtime.immigration import ( derive_us_immigration_status_from_manifest, ) @@ -179,6 +182,7 @@ def us_source_operation_handlers() -> Mapping[str, SourceOperationHandler]: calibrate_us_binary_assignment_joint_targets_from_manifest ), "compute_ratio": compute_us_ratio_from_manifest, + "derive_hours_worked": derive_us_hours_worked_from_manifest, "derive_immigration_status": derive_us_immigration_status_from_manifest, "derive_puf_policyengine_variables": ( derive_us_puf_policyengine_variables_from_manifest diff --git a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py index 55cdb569..571f4590 100644 --- a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py +++ b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py @@ -1825,6 +1825,11 @@ def n(self, entity): "with_us_take_up_inputs", lambda frame, *, seed, time_period: frame, ) + monkeypatch.setattr( + builder, + "with_us_hours_worked_inputs", + lambda frame, *, seed, time_period: frame, + ) monkeypatch.setattr( builder, "us_take_up_signal_gate", @@ -1844,6 +1849,15 @@ def n(self, entity): "gate": {"passed": True}, }, ) + monkeypatch.setattr( + builder, + "us_hours_worked_signal_gate", + lambda frame: builder.GateResult( + name="hours_worked_signal", + passed=True, + details={"checked": True}, + ), + ) monkeypatch.setattr( builder, "_ecps_parity_gate", diff --git a/packages/populace-build/tests/test_us_hours_worked.py b/packages/populace-build/tests/test_us_hours_worked.py new file mode 100644 index 00000000..82f46767 --- /dev/null +++ b/packages/populace-build/tests/test_us_hours_worked.py @@ -0,0 +1,298 @@ +"""US hours-worked stage tests (populace #242/#248).""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from populace.build.source_manifest import SourceStageSpec +from populace.build.source_runtime import SourceRuntimeError +from populace.build.us_runtime import ( + US_HOURS_WORKED_OUTPUT_COLUMNS, + US_HOURS_WORKED_REQUIRED_SOURCE_COLUMNS, + US_HOURS_WORKED_STAGE_NAME, + US_SOURCE_MANIFEST, + derive_us_hours_worked_from_manifest, + us_hours_worked_signal_gate, + us_hours_worked_stage_spec, + us_hours_worked_summary, + with_us_hours_worked_inputs, +) +from populace.build.us_runtime.source_runtime import us_source_operation_handlers +from populace.frame import US_SCHEMA, Frame, WeightKind, Weights + +TIME_PERIOD = 2024 + + +def _person_table(rows: list[dict]) -> pd.DataFrame: + """A raw-ASEC person table: baseline is a non-working adult.""" + + baseline = {"HRSWK": 0, "A_HRS1": 0, "WKSWORK": 0} + records = [] + for index, row in enumerate(rows): + record = dict(baseline) + 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], + *, + household_weights: list[float] | None = None, +) -> Frame: + person = _person_table(person_rows) + n = len(person) + household_ids = person["person_household_id"].to_numpy() + unique_households = np.unique(household_ids) + person["person_tax_unit_id"] = person["person_household_id"] + 1_000 + person["person_spm_unit_id"] = person["person_household_id"] + 2_000 + person["person_family_id"] = person["person_household_id"] + 3_000 + person["person_marital_unit_id"] = np.arange(n, dtype="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="int64") + 4_000} + ), + } + weights = household_weights or [1.0] * len(unique_households) + return Frame( + tables, + US_SCHEMA, + { + "household": Weights( + values=np.asarray(weights, dtype=np.float64), + kind=WeightKind.DESIGN, + ) + }, + ) + + +def _worker(hrswk: float, a_hrs1: float, wkswork: float, **extra) -> dict: + return {"HRSWK": hrswk, "A_HRS1": a_hrs1, "WKSWORK": wkswork, **extra} + + +#: Half workers at typical full-time hours, half non-workers: lands inside +#: both plausibility bands. +_PLAUSIBLE_ROWS = [ + _worker(40, 40, 52), + _worker(38, 35, 50), + _worker(20, 22, 26), + _worker(45, 40, 52), + _worker(0, 0, 0), + _worker(0, 0, 0), + _worker(0, 0, 0), + _worker(0, 5, 0), +] + + +class TestManifestDeclaration: + def test_stage_is_declared_with_the_three_output_columns(self) -> None: + spec = us_hours_worked_stage_spec() + assert spec.stage == US_HOURS_WORKED_STAGE_NAME + assert tuple(spec.outputs) == US_HOURS_WORKED_OUTPUT_COLUMNS + assert set(spec.nonnegative_outputs) == set(US_HOURS_WORKED_OUTPUT_COLUMNS) + + def test_org_wages_no_longer_claims_weekly_hours(self) -> None: + org_wages = US_SOURCE_MANIFEST.stage_map()["org_wages"] + assert "weekly_hours_worked_before_lsr" not in org_wages.outputs + + def test_handler_is_registered(self) -> None: + handlers = us_source_operation_handlers() + assert handlers["derive_hours_worked"] is derive_us_hours_worked_from_manifest + + +class TestDerivation: + def _derive(self, table: pd.DataFrame) -> pd.DataFrame: + spec = us_hours_worked_stage_spec() + operation = next( + op for op in spec.operations if op.kind == "derive_hours_worked" + ) + return derive_us_hours_worked_from_manifest(table, operation, None) + + def test_maps_the_asec_columns_directly(self) -> None: + table = _person_table([_worker(38, 40, 50), _worker(0, 0, 0)]) + result = self._derive(table) + assert result["weekly_hours_worked_before_lsr"].tolist() == [38.0, 0.0] + assert result["hours_worked_last_week"].tolist() == [40.0, 0.0] + assert result["weeks_worked"].tolist() == [50.0, 0.0] + + def test_negative_sentinels_floor_at_zero_and_weeks_clip_at_52(self) -> None: + table = _person_table([_worker(-1, -4, 99)]) + result = self._derive(table) + assert result["weekly_hours_worked_before_lsr"].tolist() == [0.0] + assert result["hours_worked_last_week"].tolist() == [0.0] + assert result["weeks_worked"].tolist() == [52.0] + + def test_nan_raw_values_become_zero(self) -> None: + table = _person_table([_worker(np.nan, np.nan, np.nan)]) + result = self._derive(table) + for column in US_HOURS_WORKED_OUTPUT_COLUMNS: + assert result[column].tolist() == [0.0] + + def test_missing_raw_column_is_named(self) -> None: + table = _person_table([_worker(40, 40, 52)]).drop(columns=["HRSWK"]) + with pytest.raises(SourceRuntimeError, match="HRSWK"): + self._derive(table) + + def test_unexpected_parameters_are_refused(self) -> None: + operation_spec = SourceStageSpec.from_mapping( + { + "stage": US_HOURS_WORKED_STAGE_NAME, + "survey": "test ASEC", + "source": "https://example.com", + "grain": "person", + "operations": [ + {"kind": "read_table", "table": "person"}, + {"kind": "derive_hours_worked", "surprise": True}, + ], + "outputs": list(US_HOURS_WORKED_OUTPUT_COLUMNS), + } + ) + operation = operation_spec.operations[1] + with pytest.raises(SourceRuntimeError, match="surprise"): + derive_us_hours_worked_from_manifest( + _person_table([_worker(40, 40, 52)]), operation, None + ) + + +class TestFrameIntegration: + def test_with_us_hours_worked_inputs_writes_all_three_columns(self) -> None: + frame = with_us_hours_worked_inputs( + _us_frame(_PLAUSIBLE_ROWS), seed=0, time_period=TIME_PERIOD + ) + person = frame.table("person") + assert person["weekly_hours_worked_before_lsr"].tolist()[:4] == [ + 40.0, + 38.0, + 20.0, + 45.0, + ] + assert person["hours_worked_last_week"].iloc[7] == 5.0 + assert person["weeks_worked"].max() == 52.0 + + def test_frame_with_signal_passes_through_untouched(self) -> None: + derived = with_us_hours_worked_inputs( + _us_frame(_PLAUSIBLE_ROWS), seed=0, time_period=TIME_PERIOD + ) + again = with_us_hours_worked_inputs(derived, seed=1, time_period=TIME_PERIOD) + assert again is derived + + def test_constant_forty_landmine_is_recomputed_from_raw(self) -> None: + # The published failure mode: all three columns present, but weekly + # hours broadcast at the engine's 40-hour default. + rows = [ + _worker( + 38, + 40, + 50, + weekly_hours_worked_before_lsr=40.0, + hours_worked_last_week=40.0, + weeks_worked=52.0, + ), + _worker( + 0, + 0, + 0, + weekly_hours_worked_before_lsr=40.0, + hours_worked_last_week=40.0, + weeks_worked=52.0, + ), + ] + frame = with_us_hours_worked_inputs( + _us_frame(rows), seed=0, time_period=TIME_PERIOD + ) + person = frame.table("person") + assert person["weekly_hours_worked_before_lsr"].tolist() == [38.0, 0.0] + assert person["hours_worked_last_week"].tolist() == [40.0, 0.0] + assert person["weeks_worked"].tolist() == [50.0, 0.0] + + def test_partial_surface_is_recomputed(self) -> None: + rows = [ + _worker(38, 40, 50, hours_worked_last_week=40.0, weeks_worked=52.0), + _worker(0, 0, 0, hours_worked_last_week=0.0, weeks_worked=0.0), + ] + frame = with_us_hours_worked_inputs( + _us_frame(rows), seed=0, time_period=TIME_PERIOD + ) + person = frame.table("person") + assert person["weekly_hours_worked_before_lsr"].tolist() == [38.0, 0.0] + + def test_missing_raw_columns_without_signal_raise(self) -> None: + frame = _us_frame([{"person_id": 1}]) + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + tables["person"] = tables["person"].drop( + columns=list(US_HOURS_WORKED_REQUIRED_SOURCE_COLUMNS) + ) + stripped = Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + ) + with pytest.raises(SourceRuntimeError, match="HRSWK"): + with_us_hours_worked_inputs(stripped, seed=0, time_period=TIME_PERIOD) + + +class TestGate: + def test_plausible_distribution_passes(self) -> None: + frame = with_us_hours_worked_inputs( + _us_frame(_PLAUSIBLE_ROWS), seed=0, time_period=TIME_PERIOD + ) + gate = us_hours_worked_signal_gate(frame) + assert gate.passed + summary = us_hours_worked_summary(frame) + assert 0.35 <= summary["worked_share"] <= 0.62 + assert 30.0 <= summary["mean_weekly_hours_workers"] <= 45.0 + + def test_missing_columns_fail(self) -> None: + gate = us_hours_worked_signal_gate(_us_frame([_worker(40, 40, 52)])) + assert not gate.passed + assert "missing" in gate.failures[0] + + def test_constant_forty_fails(self) -> None: + rows = [ + _worker( + 0, + 0, + 0, + weekly_hours_worked_before_lsr=40.0, + hours_worked_last_week=40.0, + weeks_worked=52.0, + ) + for _ in range(4) + ] + frame = _us_frame(rows) + gate = us_hours_worked_signal_gate(frame) + assert not gate.passed + assert any("constant" in failure for failure in gate.failures) + + def test_everyone_working_fails_the_share_band(self) -> None: + frame = with_us_hours_worked_inputs( + _us_frame([_worker(40, 40, 52), _worker(38, 38, 52)]), + seed=0, + time_period=TIME_PERIOD, + ) + gate = us_hours_worked_signal_gate(frame) + assert not gate.passed + assert any("worked share" in failure for failure in gate.failures) + + def test_implausible_mean_hours_fail(self) -> None: + rows = [ + _worker(3, 3, 52), + _worker(2, 2, 52), + _worker(0, 0, 0), + _worker(0, 0, 0), + ] + frame = with_us_hours_worked_inputs( + _us_frame(rows), seed=0, time_period=TIME_PERIOD + ) + gate = us_hours_worked_signal_gate(frame) + assert not gate.passed + assert any("mean weekly hours" in failure for failure in gate.failures) diff --git a/packages/populace-build/tests/test_us_l0_refit_export.py b/packages/populace-build/tests/test_us_l0_refit_export.py index af43c5b7..6cf12d40 100644 --- a/packages/populace-build/tests/test_us_l0_refit_export.py +++ b/packages/populace-build/tests/test_us_l0_refit_export.py @@ -30,6 +30,9 @@ def _us_frame(**person_extra: object) -> Frame: ), "ssn_card_type": ["CITIZEN", "CITIZEN", "NONE"], "immigration_status_str": ["CITIZEN", "CITIZEN", "UNDOCUMENTED"], + "weekly_hours_worked_before_lsr": [40.0, 20.0, 0.0], + "hours_worked_last_week": [40.0, 18.0, 0.0], + "weeks_worked": [52.0, 26.0, 0.0], **person_extra, } ) diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index 6b668078..6fc83a4b 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -55,12 +55,14 @@ compile_us_fiscal_target_registry, hard_target_package_aliases, load_congressional_district_vintage_crosswalk, + us_hours_worked_signal_gate, us_immigration_composition_gate, us_source_coverage_diagnostics, us_source_operation_handlers, us_take_up_participation_diagnostics, us_take_up_signal_gate, us_validation_input_coverage_gate, + with_us_hours_worked_inputs, with_us_immigration_inputs, with_us_take_up_inputs, write_us_source_coverage_diagnostics, @@ -387,10 +389,6 @@ def _automatic_gc_suspended(): # degenerate column, or one of these becoming non-degenerate, fails the # default-valued-columns gate so this list cannot rot. US_DEGENERATE_INPUT_REVIEWED_EXCLUSIONS = { - "weekly_hours_worked_before_lsr": ( - "Hours-worked inputs not yet carried through (PolicyEngine/populace" - "#242, #248); constant at the 40-hour engine default." - ), "takes_up_snap_if_eligible": ( "SNAP take-up inputs not yet carried through (PolicyEngine/populace" "#243); constant True forces 100% take-up." @@ -3857,6 +3855,7 @@ def _release_gate_failures( input_mass_reference_gate: GateResult | None = None, degenerate_input_gate: GateResult | None = None, ecps_parity_gate: GateResult | None = None, + hours_worked_gate: GateResult | None = None, ) -> list[str]: failures: list[str] = [] if target_profile_gate is not None and not target_profile_gate.passed: @@ -3879,6 +3878,11 @@ def _release_gate_failures( f"Immigration composition failed: {failure}" for failure in immigration_gate.failures ) + if hours_worked_gate is not None and not hours_worked_gate.passed: + failures.extend( + f"Hours-worked signal failed: {failure}" + for failure in hours_worked_gate.failures + ) if input_mass_reference_gate is not None and not input_mass_reference_gate.passed: failures.extend( f"Input mass parity failed: {failure}" @@ -4155,6 +4159,7 @@ def _write_release_calibration_diagnostics( audit_export_targets: bool, immigration_gate: GateResult | None = None, input_mass_reference_gate: GateResult | None = None, + hours_worked_gate: GateResult | None = None, gate_failures: Iterable[str], timing: Mapping[str, object] | None = None, warm_start_calibration: Mapping[str, object] | None = None, @@ -4227,6 +4232,15 @@ def _write_release_calibration_diagnostics( if immigration_gate is not None else None ), + "hours_worked_signal": ( + { + "passed": hours_worked_gate.passed, + "failures": list(hours_worked_gate.failures), + "details": dict(hours_worked_gate.details), + } + if hours_worked_gate is not None + else None + ), "input_mass_reference": ( { "passed": input_mass_reference_gate.passed, @@ -4465,6 +4479,7 @@ def _build_manifests( input_mass_reference_gate: GateResult | None = None, degenerate_input_gate: GateResult | None = None, ecps_parity_gate: GateResult | None = None, + hours_worked_gate: GateResult | None = None, timing: Mapping[str, object] | None = None, warm_start_calibration: Mapping[str, object] | None = None, default_dataset: Mapping[str, object] | None = None, @@ -4491,6 +4506,7 @@ def _build_manifests( input_mass_reference_gate, degenerate_input_gate, ecps_parity_gate=ecps_parity_gate, + hours_worked_gate=hours_worked_gate, ) commit = _git_output("rev-parse", "HEAD") @@ -4607,6 +4623,17 @@ def _build_manifests( if ecps_parity_gate is not None else {} ), + **( + { + "hours_worked_signal": { + "passed": hours_worked_gate.passed, + "failures": list(hours_worked_gate.failures), + "details": dict(hours_worked_gate.details), + } + } + if hours_worked_gate is not None + else {} + ), }, } (release_dir / "build_manifest.json").write_text( @@ -4665,6 +4692,16 @@ def _build_manifests( if ecps_parity_gate is not None else {} ), + **( + { + "hours_worked_signal": { + "passed": hours_worked_gate.passed, + "details": dict(hours_worked_gate.details), + } + } + if hours_worked_gate is not None + else {} + ), }, "compatible_core_packages": [ { @@ -5043,6 +5080,33 @@ def main() -> None: f"Take-up signal failed: {failure}" for failure in take_up_gate.failures ) ) + if telemetry is not None: + telemetry.stage( + "hours_worked_inputs", + message="Deriving hours-worked inputs from ASEC reported hours.", + ) + base_frame = with_us_hours_worked_inputs( + base_frame, + seed=args.seed, + time_period=PERIOD, + ) + hours_worked_gate = us_hours_worked_signal_gate(base_frame) + if not hours_worked_gate.passed: + if telemetry is not None: + telemetry.stage( + "hours_worked_gate", + status="failed", + message="Hours-worked signal gate failed.", + failures=list(hours_worked_gate.failures), + force_upload=True, + ) + raise RuntimeError( + "Release gates failed: " + + "; ".join( + f"Hours-worked signal failed: {failure}" + for failure in hours_worked_gate.failures + ) + ) if telemetry is not None: telemetry.stage( "source_inputs", @@ -5355,6 +5419,7 @@ def main() -> None: enforced_input_mass_reference_gate, degenerate_input_gate, ecps_parity_gate=enforced_ecps_parity_gate, + hours_worked_gate=hours_worked_gate, ) _write_release_calibration_diagnostics( result=result, @@ -5367,6 +5432,7 @@ def main() -> None: base_population_gate=base_population_gate, immigration_gate=immigration_gate, input_mass_reference_gate=input_mass_reference_gate, + hours_worked_gate=hours_worked_gate, support_value_repairs={ "social_security_components": social_security_component_repair }, @@ -5570,6 +5636,7 @@ def main() -> None: input_mass_reference_gate=enforced_input_mass_reference_gate, degenerate_input_gate=degenerate_input_gate, ecps_parity_gate=enforced_ecps_parity_gate, + hours_worked_gate=hours_worked_gate, timing=timing, warm_start_calibration=warm_start_calibration, ledger_artifact=ledger_artifact.provenance(),