diff --git a/changelog.d/597.fixed.md b/changelog.d/597.fixed.md new file mode 100644 index 00000000..157aa839 --- /dev/null +++ b/changelog.d/597.fixed.md @@ -0,0 +1 @@ +Measure SPM poverty on legacy certified US sources that predate SPM measurement-scope declarations, such as the ACS-local Build O release. Following Census's ACS household-population method, ACS group-quarters units are outside the SPM measurement: their SPM results are missing instead of stopping the run with `SPM_COMPOSITION_REQUIRED`, and poverty rates count measured people only. Ordinary tax and benefit results are unchanged. diff --git a/changelog.d/598.fixed.md b/changelog.d/598.fixed.md new file mode 100644 index 00000000..61065982 --- /dev/null +++ b/changelog.d/598.fixed.md @@ -0,0 +1 @@ +Supply the certified SPM independence role from the national release for the ASEC and PUF records of legacy certified US sources, so independent-teen SPM units are measured instead of stopping SPM calculations. diff --git a/changelog.d/poverty-rate-unmeasured.changed.md b/changelog.d/poverty-rate-unmeasured.changed.md new file mode 100644 index 00000000..72261a07 --- /dev/null +++ b/changelog.d/poverty-rate-unmeasured.changed.md @@ -0,0 +1 @@ +`Poverty.rate` is now `None`, not `0.0`, when nobody in the group is measured (its weighted population with a poverty status is zero), for every source. `headcount` and `total_population` are unchanged, and the poverty DataFrame helpers report the rate as missing. Callers that format or subtract rates must handle `None`. diff --git a/docs/engineering/runbooks/us-legacy-spm-sources.md b/docs/engineering/runbooks/us-legacy-spm-sources.md new file mode 100644 index 00000000..20f90d84 --- /dev/null +++ b/docs/engineering/runbooks/us-legacy-spm-sources.md @@ -0,0 +1,122 @@ +# Legacy US SPM sources (backward compatibility) + +> **Backward compatibility, not a data contract.** This layer exists so that +> certified US sources published *before* SPM declarations existed can still +> be measured. New certified sources must carry the declarations themselves; +> do not register them here. + +policyengine-us measures SPM poverty only when a source declares two things: + +- **Measurement scope:** which SPM units are inside the SPM measurement + population (`spm_unit_spm_universe_status`: `INCLUDED`/`OUTSIDE`, the + contract of PolicyEngine/microcosm#977). +- **Independence role:** which 15–17-year-olds independently head their own + SPM unit (`is_spm_independent_minor_role`). + +A legacy source carries neither. Every SPM calculation on it stops with +`SPM_COMPOSITION_REQUIRED` ("SPM unit has no classified adult"). + +The package +[`policyengine.tax_benefit_models.us.legacy_spm`](../../../src/policyengine/tax_benefit_models/us/legacy_spm/__init__.py) +derives the missing declarations from evidence a legacy source does carry. +Nobody is dropped: memberships, weights and every ordinary tax, benefit and +income calculation are unchanged. + +## Registered sources + +Sources are keyed by certified SHA-256 in `legacy_spm/registry.py`. + +| Source | Scope rule | Independence list | +| --- | --- | --- | +| `populace_us_2024_acs_local` Build O (`769756c3…a7ec`) | `acs_household_population` | national release `source_spm_person_independence.csv` (`22b5968d…3ce5`) | + +### Scope rule `acs_household_population` (#597) + +Census's ACS SPM methodology measures the household population only. + +- ACS records with `TYPEHUGQ` 2 or 3 and the matching `RELSHIPP` (37 or 38) are + group quarters, so their SPM units are `OUTSIDE`. +- ACS housing-unit records are `INCLUDED`. +- Records from the `asec` and `puf_tax_detail` channels are `INCLUDED`. + +Age is never used. Unknown channels, missing or inconsistent evidence, and SPM +units that mix group quarters with household people all raise. + +On the full Build O source, 183,206 SPM units are outside the measurement. + +### Independence list (#598) + +The source's ASEC/PUF records are the same records the certified national +release carries with a source-backed independence role. The list is joined by +`person_id` and `person_spm_unit_id`, and the join must be exact: + +- every ASEC/PUF person appears once, in the same SPM unit; +- the list names nobody else. + +ACS people keep the model's own fallback (household head or spouse). + +On Build O, exactly 28 units change: Census-built units around an independent +teen. + +## How runs get the declarations + +| Path | What happens | +| --- | --- | +| `create_datasets` / `ensure_datasets` | Year files store the declarations as ordinary input columns, with a `policyengine_legacy_spm_declarations` record. | +| `Simulation.run()` over a raw legacy source file | The declarations are derived in memory. The file is recognised by its evidence columns and then its SHA-256, which is hashed once per process. | +| `managed_microsimulation` | The declarations are derived from the loaded tables and set on the simulation and its branches. | + +`legacy_spm.model_adapter` then makes the model honour the scope: + +- **native:** policyengine-us defines `spm_unit_spm_universe_status` + (PolicyEngine/policyengine-us#9462). The declaration is just an input, and + nothing is overridden. +- **override:** policyengine-us 2.2.1 cannot read it. A variable-only reform: + - adds the input; + - wraps the installed SPM chain so that outside units never reach the + canonical provider and every SPM outcome for them is NaN ("not measured"). + + Poverty indicators become floats (1.0, 0.0 or NaN). The override wraps the + installed formulas, with one exception: `spm_unit_income_decile` ranks + units, so it is recalculated over measured units with the installed method. + A contract test pins that method. The override refuses to run if the SPM + chain no longer has the verified shape. + +Poverty rates count measured people only. A group with nobody measured has a +rate of `None`, not 0. + +## Staleness + +The record holds a `LegacySPMProfile`: the registry entry plus +`ADAPTER_REVISION`. Checks are total and exact, like the year-file stamp: + +- `ensure_datasets` cuts a year file again, and `load_datasets` refuses it, when + its record is absent, unreadable, malformed, or from another profile. +- `Simulation.load()` refuses an output saved under a different profile. + +Raise `ADAPTER_REVISION` whenever a change to derivation or the override means +existing files must be made again. + +## When the model or data changes + +- **policyengine-us gains native scope (#9462).** + `test_the_installed_model_does_not_read_scope_natively_yet` fails. Check that + the native members include `INCLUDED` and `OUTSIDE`, and confirm the native + runs in the integration tests. Then delete the override half of + `model_adapter` (keep the derivation) and raise `ADAPTER_REVISION`. +- **The SPM chain changes.** `test_the_override_covers_every_variable_that_depends_on_the_provider`, + `test_the_installed_chain_has_the_verified_shape` or + `test_the_recalculated_decile_matches_the_installed_method` fails. Re-verify + the override against the new chain before bumping the pin. +- **A rebuilt source declares scope and role itself.** Point the bundle at it. + Keep the Build O registry entry only while anything still reads Build O, + then delete it. + +## Related + +- Microcosm #977: scope classification at source. +- Microcosm #959: independence role in US builds. +- policyengine-us #9462: native scope. +- policyengine.py #519: nullable SPM summaries. +- The year-file and output stamps (#590, and #593/#594 in flight). Fold this + record into them once those land. diff --git a/docs/microsim.md b/docs/microsim.md index 1e63432a..98dd0244 100644 --- a/docs/microsim.md +++ b/docs/microsim.md @@ -87,6 +87,24 @@ before passing them to the country model. Generic adult/child counts remain separate. The native source enrichment is not recalibration, and inherited schema-5 calibration diagnostics cannot be relabeled as schema 6. +**Legacy SPM sources (backward compatibility).** Some certified US sources +predate SPM declarations, for example the ACS-local Build O release. For those +sources, .py derives the declarations from evidence they carry, following +Census's ACS SPM methodology, which measures the household population only: + +- ACS group-quarters SPM units are outside the SPM measurement. Their SPM + thresholds, resources and poverty indicators are missing (NaN). +- Poverty indicators on those sources are floats: 1.0, 0.0, or NaN for "not + measured". +- Poverty rates count measured people only. A group with nobody measured has + a rate of `None`, not 0. +- The sources' ASEC and PUF records take the certified SPM independence role + from the national release. + +Ordinary tax, benefit and income results are unchanged. Outputs record the +declarations they used under `legacy_spm_declarations`. See +[the runbook](engineering/runbooks/us-legacy-spm-sources.md). + The wrapper's native pandas HDF loader accepts files that store only calibrated `household_weight`. It maps missing person and group weight columns through native membership IDs in memory, retaining supplied weight columns and the diff --git a/docs/outputs.md b/docs/outputs.md index ff1708e1..94a538dc 100644 --- a/docs/outputs.md +++ b/docs/outputs.md @@ -167,6 +167,12 @@ rate.run() rate.headcount, rate.total_population, rate.rate ``` +`rate.rate` is `None` when nobody in the group is measured: the group's +weighted population with a poverty status is zero, for example a filter that +matches no one, or a group whose members are all outside the SPM measurement +on a legacy source. That is not a measured rate of zero, so handle `None` +before formatting or subtracting rates. + For all canonical poverty measures over one simulation: ```python diff --git a/examples/us_budgetary_impact.py b/examples/us_budgetary_impact.py index f52b835f..1dc32445 100644 --- a/examples/us_budgetary_impact.py +++ b/examples/us_budgetary_impact.py @@ -32,6 +32,11 @@ ) +def _format_rate(rate: float | None) -> str: + """Format a poverty rate; ``None`` means nobody in the group is measured.""" + return "n/a" if rate is None else f"{rate:.4f}" + + def main(): year = 2026 @@ -147,11 +152,15 @@ def main(): analysis.reform_poverty.outputs, strict=True, ): + # A rate is None when nobody in the group is measured. + change = ( + "n/a" if bp.rate is None or rp.rate is None else f"{rp.rate - bp.rate:+.4f}" + ) print( f" {bp.metric:30s} " - f"baseline={bp.rate:.4f} " - f"reform={rp.rate:.4f} " - f"change={rp.rate - bp.rate:+.4f}" + f"baseline={_format_rate(bp.rate)} " + f"reform={_format_rate(rp.rate)} " + f"change={change}" ) print("\n=== Inequality ===") diff --git a/src/policyengine/core/run_record.py b/src/policyengine/core/run_record.py index f9f552aa..126e49f5 100644 --- a/src/policyengine/core/run_record.py +++ b/src/policyengine/core/run_record.py @@ -193,6 +193,15 @@ def build_simulation_run_record_payloads( if renames is not None: results_payload["legacy_input_renames"] = dict(renames) + # BACKWARD COMPATIBILITY (#597, #598): the legacy SPM declarations a + # US run applied, which change its SPM results. See + # ``policyengine.tax_benefit_models.us.legacy_spm``. + legacy_spm_declarations = ( + getattr(simulation.output_dataset, "metadata", None) or {} + ).get("legacy_spm_declarations") + if legacy_spm_declarations is not None: + results_payload["legacy_spm_declarations"] = dict(legacy_spm_declarations) + return {"reform": reform, "input": input_payload, "results": results_payload} diff --git a/src/policyengine/core/simulation.py b/src/policyengine/core/simulation.py index 3dd537b2..93762b88 100644 --- a/src/policyengine/core/simulation.py +++ b/src/policyengine/core/simulation.py @@ -258,4 +258,12 @@ def release_bundle(self) -> dict[str, Any]: ) if renames is not None: result["legacy_input_renames"] = dict(renames) + # BACKWARD COMPATIBILITY (#597, #598): the legacy SPM declarations a + # US run applied, which change its SPM results. See + # ``policyengine.tax_benefit_models.us.legacy_spm``. + legacy_spm_declarations = ( + getattr(self.output_dataset, "metadata", None) or {} + ).get("legacy_spm_declarations") + if legacy_spm_declarations is not None: + result["legacy_spm_declarations"] = dict(legacy_spm_declarations) return result diff --git a/src/policyengine/outputs/poverty.py b/src/policyengine/outputs/poverty.py index 85a761a5..5ba56ed0 100644 --- a/src/policyengine/outputs/poverty.py +++ b/src/policyengine/outputs/poverty.py @@ -81,6 +81,21 @@ def run(self): # Map poverty variable to target entity if needed if poverty_var_obj.entity != target_entity: + source = getattr( + self.simulation.output_dataset.data, poverty_var_obj.entity + ) + if ( + target_entity != "person" + and pd.isna(source[self.poverty_variable]).any() + ): + # Mapping onto a group sums members and fills empty groups + # with zero, which would count a person outside the SPM + # measurement as measured and not poor. + raise ValueError( + f"{self.poverty_variable} is missing for units outside " + f"the poverty measurement; summarise it for people, not " + f"for {target_entity} units." + ) mapped = self.simulation.output_dataset.data.map_to_entity( poverty_var_obj.entity, target_entity, @@ -121,8 +136,13 @@ def run(self): # Calculate results using weighted counts self.headcount = float((poverty_series == True).sum()) # noqa: E712 self.total_population = float(poverty_series.count()) + # ``None`` means nobody in the group is measured (for example, every + # member is outside the SPM measurement population), which is not + # a measured rate of zero. self.rate = ( - self.headcount / self.total_population if self.total_population > 0 else 0.0 + self.headcount / self.total_population + if self.total_population > 0 + else None ) diff --git a/src/policyengine/tax_benefit_models/common/model_version.py b/src/policyengine/tax_benefit_models/common/model_version.py index e10d3330..4d3cce7a 100644 --- a/src/policyengine/tax_benefit_models/common/model_version.py +++ b/src/policyengine/tax_benefit_models/common/model_version.py @@ -417,6 +417,14 @@ def load(self, simulation: Simulation) -> None: RENAMES_RECORD_KEY, read_renames_record, ) + from policyengine.tax_benefit_models.us.legacy_spm import ( + RECORD_KEY as LEGACY_SPM_RECORD_KEY, + ) + from policyengine.tax_benefit_models.us.legacy_spm import ( + expected_profile_for_dataset, + read_record_value, + record_problem, + ) with h5py.File(filepath, "r") as stream: raw = ( @@ -442,6 +450,17 @@ def load(self, simulation: Simulation) -> None: "Saved US simulation predates the mapping of renamed " "stored inputs (it has no record of them); run it again" ) + # BACKWARD COMPATIBILITY (#597, #598): an output is reused only + # under exactly the legacy SPM declarations its input needs. + recorded_legacy_spm = read_record_value(filepath) + legacy_problem = record_problem( + recorded_legacy_spm, expected_profile_for_dataset(simulation.dataset) + ) + if legacy_problem is not None: + raise ValueError( + f"Saved US simulation output is out of date: {legacy_problem}; " + "run it again" + ) if self.country_code == "uk": from policyengine.tax_benefit_models.uk.datasets import ( @@ -474,6 +493,10 @@ def load(self, simulation: Simulation) -> None: simulation.output_dataset.metadata.update( {"spm_config": recorded["config"], RENAMES_RECORD_KEY: recorded_renames} ) + if isinstance(recorded_legacy_spm, dict): + simulation.output_dataset.metadata[LEGACY_SPM_RECORD_KEY] = ( + recorded_legacy_spm + ) if os.path.exists(filepath): simulation.created_at = datetime.datetime.fromtimestamp( diff --git a/src/policyengine/tax_benefit_models/us/datasets.py b/src/policyengine/tax_benefit_models/us/datasets.py index a5fbfb9b..320a538a 100644 --- a/src/policyengine/tax_benefit_models/us/datasets.py +++ b/src/policyengine/tax_benefit_models/us/datasets.py @@ -25,6 +25,9 @@ from policyengine.tax_benefit_models.common.model_version import ( build_runtime_dataset_provenance, ) +from policyengine.tax_benefit_models.us.legacy_inputs import ( + _ABSENT as _ABSENT_RECORD, +) from policyengine.tax_benefit_models.us.legacy_inputs import ( RENAMES_RECORD_KEY, apply_legacy_input_renames_to_microsimulation, @@ -36,6 +39,21 @@ year_file_stamp, year_file_stamp_problem, ) +from policyengine.tax_benefit_models.us.legacy_spm import ( + RECORD_KEY as LEGACY_SPM_RECORD_KEY, +) +from policyengine.tax_benefit_models.us.legacy_spm import ( + LegacySPMDeclarations, + LegacySPMProfile, + add_to_year_file_frames, + derive_from_stored_tables, + expected_profile, + legacy_spm_source, + parse_record, + read_record_value, + write_record, + year_file_problem, +) from policyengine.utils.hashing import sha256_file logger = logging.getLogger(__name__) @@ -117,6 +135,12 @@ def save(self) -> None: renames = self.metadata.get(RENAMES_RECORD_KEY) if renames is not None: write_renames_record(filepath, renames) + # BACKWARD COMPATIBILITY (#597, #598): the legacy SPM declarations a + # year file was cut with, or an output calculated with (see + # ``legacy_spm.record``). + legacy_spm_record = parse_record(self.metadata.get(LEGACY_SPM_RECORD_KEY)) + if legacy_spm_record is not None: + write_record(filepath, legacy_spm_record) def load(self) -> None: """Load dataset from HDF5 file into this instance.""" @@ -125,6 +149,15 @@ def load(self) -> None: renames = read_renames_record(path) if renames is not None: self.metadata[RENAMES_RECORD_KEY] = renames + # BACKWARD COMPATIBILITY (#597, #598): keep the stored record as it + # is, readable or not; runs and loaders judge it (``legacy_spm.record``). + legacy_spm_record = read_record_value(path) + if isinstance(legacy_spm_record, dict): + self.metadata[LEGACY_SPM_RECORD_KEY] = legacy_spm_record + elif legacy_spm_record is not _ABSENT_RECORD: + self.metadata[LEGACY_SPM_RECORD_KEY] = None + else: + self.metadata.pop(LEGACY_SPM_RECORD_KEY, None) if _is_policyengine_core_h5(path): self.data = _load_policyengine_core_h5(path, self.year) return @@ -411,6 +444,12 @@ def create_datasets( # Map stored inputs the engine has since renamed before extracting, # so each year's file stores the live input rather than dropping it. legacy_input_renames = apply_legacy_input_renames_to_microsimulation(sim) + # BACKWARD COMPATIBILITY (#597, #598): a legacy certified source + # carries no SPM declarations, so derive them while its stored + # evidence is still available; year files keep model inputs only. + legacy_spm = _legacy_spm_declarations( + sim, source.bundle_dataset, Path(data_folder) + ) for year in years: # Get all input variables from the simulation @@ -548,15 +587,24 @@ def create_datasets( elif entity_name == "tax_unit": tax_unit_df = entity_df + metadata: dict[str, Any] = { + RENAMES_RECORD_KEY: dict(sorted(legacy_input_renames.items())) + } + if legacy_spm is not None: + # BACKWARD COMPATIBILITY (#597, #598): store the derived + # declarations as the ordinary inputs newer sources carry. + add_to_year_file_frames(person_df, spm_unit_df, legacy_spm) + metadata[LEGACY_SPM_RECORD_KEY] = legacy_spm.record.model_dump( + mode="json" + ) + us_dataset = PolicyEngineUSDataset( id=f"{dataset_stem}_year_{year}", name=f"{dataset_stem}-year-{year}", description=f"US Dataset for year {year} based on {dataset_stem}", filepath=f"{data_folder}/{dataset_stem}_year_{year}.h5", year=int(year), - metadata={ - RENAMES_RECORD_KEY: dict(sorted(legacy_input_renames.items())) - }, + metadata=metadata, data=USYearData( person=MicroDataFrame(person_df, weights="person_weight"), household=MicroDataFrame(household_df, weights="household_weight"), @@ -579,6 +627,57 @@ def create_datasets( return result +def _legacy_spm_declarations( + sim: Any, + bundle_dataset: Optional[MaterializedDataset], + data_folder: Path, +) -> Optional[LegacySPMDeclarations]: + """Derive the SPM declarations a legacy certified source lacks. + + BACKWARD COMPATIBILITY (#597, #598). Only a bundle dataset whose verified + SHA-256 is registered in ``legacy_spm.registry`` gets any; see + ``legacy_spm``. + """ + source = legacy_spm_source(bundle_dataset.sha256 if bundle_dataset else None) + if source is None: + return None + declarations, _ = derive_from_stored_tables( + source, getattr(sim, "dataset", None), data_dir=data_folder + ) + return declarations + + +def _managed_source_sha256(dataset: Optional[str]) -> Optional[str]: + """Return the certified SHA-256 of a managed bundle dataset reference. + + A managed dataset may be named by its manifest name or by the URI that + name resolves to (the form the region registry's dataset paths take). + Any other reference is unmanaged and has no certified hash. + """ + manifest = get_release_manifest("us") + if dataset is None or dataset == manifest.default_dataset_uri: + name: Optional[str] = manifest.default_dataset + elif dataset in manifest.datasets: + name = dataset + else: + name = next( + ( + candidate + for candidate in manifest.datasets + if resolve_dataset_reference("us", candidate) == dataset + ), + None, + ) + reference = manifest.datasets.get(name) if name is not None else None + return reference.sha256 if reference is not None else None + + +def _expected_legacy_spm_profile(dataset: Optional[str]) -> Optional[LegacySPMProfile]: + """Return the legacy SPM profile a dataset reference's year files need.""" + source = legacy_spm_source(_managed_source_sha256(dataset)) + return expected_profile(source) if source is not None else None + + def load_datasets( datasets: Optional[list[str]] = None, years: list[int] = [2024, 2025, 2026, 2027, 2028], @@ -617,6 +716,17 @@ def load_datasets( "Regenerate it with ensure_datasets() or " "create_datasets()." ) + # BACKWARD COMPATIBILITY (#597, #598): a legacy source's year + # file must carry exactly the declarations this release derives. + legacy_problem = year_file_problem( + filepath, _expected_legacy_spm_profile(dataset) + ) + if legacy_problem is not None: + raise ValueError( + f"US year file {filepath} is out of date: " + f"{legacy_problem}. Regenerate it with ensure_datasets() " + "or create_datasets()." + ) us_dataset = PolicyEngineUSDataset( name=f"{dataset_stem}-year-{year}", description=f"US Dataset for year {year} based on {dataset_stem}", @@ -1301,6 +1411,19 @@ def ensure_datasets( filepath, problem, ) + continue + # BACKWARD COMPATIBILITY (#597, #598): see ``legacy_spm.record``. + legacy_problem = year_file_problem( + filepath, _expected_legacy_spm_profile(dataset) + ) + if legacy_problem is not None: + all_current = False + logger.warning( + "US year file %s is out of date: %s. Creating the year " + "files again.", + filepath, + legacy_problem, + ) if all_current: return load_datasets(datasets=datasets, years=years, data_folder=data_folder) diff --git a/src/policyengine/tax_benefit_models/us/legacy_spm/__init__.py b/src/policyengine/tax_benefit_models/us/legacy_spm/__init__.py new file mode 100644 index 00000000..26905ff3 --- /dev/null +++ b/src/policyengine/tax_benefit_models/us/legacy_spm/__init__.py @@ -0,0 +1,120 @@ +"""Backward compatibility for certified US sources without SPM declarations. + +############################################################################# +# BACKWARD COMPATIBILITY — LEGACY CERTIFIED SOURCES ONLY. # +# # +# policyengine-us measures SPM poverty only when a source declares two # +# things: which SPM units are inside the SPM measurement population, and # +# which 15–17-year-olds independently head their own SPM unit. Certified # +# sources published before those declarations existed do not carry them, # +# and every SPM calculation on them stops with SPM_COMPOSITION_REQUIRED. # +# # +# This package DERIVES the declarations from evidence those sources do # +# carry, for the sources registered in ``registry.LEGACY_SPM_SOURCES`` and # +# for nothing else. New certified sources must carry the declarations # +# themselves: do NOT register them here. # +# # +# ``model_adapter`` is the part that must eventually go: it overrides SPM # +# variables only while the installed policyengine-us cannot read the scope # +# declaration, and switches itself off once it can # +# (PolicyEngine/policyengine-us#9462). # +# # +# Issues: PolicyEngine/policyengine.py#597, #598. Runbook: # +# docs/engineering/runbooks/us-legacy-spm-sources.md # +############################################################################# + +The pieces: + +- :mod:`.registry` names each legacy source by its certified SHA-256 and the + declarations it needs. +- :mod:`.universe` derives the SPM measurement scope (Census's ACS SPM + methodology measures the household population, so ACS group quarters are + outside it). +- :mod:`.independence` takes the certified SPM independence role for + ASEC/PUF records from the national release's list. +- :mod:`.declarations` derives both for a source and writes them as the + ordinary dataset inputs ``spm_unit_spm_universe_status`` and + ``is_spm_independent_minor_role``, the contract that newer sources carry. +- :mod:`.record` stamps year files and outputs with the profile they were + derived under, so that stale files are cut again and stale outputs + recalculated. +- :mod:`.runs` decides which declarations a run needs, from a year file or + from a raw legacy source file. +- :mod:`.model_adapter` lets a policyengine-us release that cannot read the + scope declaration honour it. +""" + +from .declarations import ( + LegacySPMDeclarations, + add_to_year_file_frames, + check_year_file_frames, + derive_declarations, + derive_from_stored_tables, + set_declarations, +) +from .model_adapter import ( + INDEPENDENCE_ROLE, + UNIVERSE_VARIABLE, + ModelMode, + prepare_microsimulation, +) +from .record import ( + ADAPTER_REVISION, + RECORD_H5_DATASET, + RECORD_KEY, + LegacySPMProfile, + LegacySPMRecord, + expected_profile, + parse_record, + read_record_value, + record_problem, + write_record, + year_file_problem, +) +from .registry import ( + LEGACY_SPM_SOURCES, + IndependenceListPin, + LegacySPMSource, + legacy_spm_source, +) +from .runs import ( + LegacySPMRun, + expected_profile_for_dataset, + legacy_spm_for_run, + raw_legacy_source, +) +from .universe import SPMUniverseStatus, classify_acs_household_population + +__all__ = [ + "ADAPTER_REVISION", + "INDEPENDENCE_ROLE", + "LEGACY_SPM_SOURCES", + "RECORD_H5_DATASET", + "RECORD_KEY", + "UNIVERSE_VARIABLE", + "IndependenceListPin", + "LegacySPMDeclarations", + "LegacySPMProfile", + "LegacySPMRecord", + "LegacySPMRun", + "LegacySPMSource", + "ModelMode", + "SPMUniverseStatus", + "add_to_year_file_frames", + "check_year_file_frames", + "classify_acs_household_population", + "derive_declarations", + "derive_from_stored_tables", + "expected_profile", + "expected_profile_for_dataset", + "legacy_spm_for_run", + "legacy_spm_source", + "parse_record", + "prepare_microsimulation", + "raw_legacy_source", + "read_record_value", + "record_problem", + "set_declarations", + "write_record", + "year_file_problem", +] diff --git a/src/policyengine/tax_benefit_models/us/legacy_spm/declarations.py b/src/policyengine/tax_benefit_models/us/legacy_spm/declarations.py new file mode 100644 index 00000000..5fc4a66d --- /dev/null +++ b/src/policyengine/tax_benefit_models/us/legacy_spm/declarations.py @@ -0,0 +1,218 @@ +"""Derive a legacy source's SPM declarations and hand them to the model. + +BACKWARD COMPATIBILITY (#597, #598). The declarations are written as the +ordinary dataset inputs newer sources carry: + +- ``spm_unit_spm_universe_status`` (SPM unit, per year): ``INCLUDED`` or + ``OUTSIDE``; +- ``is_spm_independent_minor_role`` (person): the SPM independence role. + +``create_datasets`` stores them as year-file columns +(:func:`add_to_year_file_frames`). A run straight over a legacy source file, +and ``managed_microsimulation``, set them on the simulation instead +(:func:`set_declarations`). +""" + +from __future__ import annotations + +from collections.abc import Iterable, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Optional + +import numpy as np +import pandas as pd + +from policyengine.provenance.dataset_materialization import materialize_dataset +from policyengine.tax_benefit_models.us.legacy_inputs import ( + _record_input_variables, + _simulation_and_branches, + stored_entity_tables, +) + +from .independence import INDEPENDENCE_ROLE, independence_roles +from .model_adapter import UNIVERSE_VARIABLE +from .record import LegacySPMRecord, expected_profile +from .registry import IndependenceListPin, LegacySPMSource +from .universe import ( + NON_ACS_CHANNELS, + SPMUniverseStatus, + classify_acs_household_population, + decode_text, +) + +#: Declared statuses a stored scope column may hold. +_DECLARED_STATUSES = frozenset( + {SPMUniverseStatus.INCLUDED.value, SPMUniverseStatus.OUTSIDE.value} +) + + +@dataclass(frozen=True) +class LegacySPMDeclarations: + """The declarations derived for one legacy source.""" + + record: LegacySPMRecord + #: Measurement status by SPM unit ID, if the source needs a scope. + universe: Optional[pd.Series] + #: Independence role by person ID, if the source needs one. + independence_role: Optional[pd.Series] + + +def derive_declarations( + source: LegacySPMSource, + person: pd.DataFrame, + household: pd.DataFrame, + *, + data_dir: Path, +) -> LegacySPMDeclarations: + """Derive a legacy source's declarations from its stored tables. + + ``data_dir`` is where a certified independence list is downloaded to. + """ + universe = None + if source.spm_universe_rule == "acs_household_population": + universe = classify_acs_household_population(person, household) + roles = None + list_people = None + if source.independence_list is not None: + list_path = _materialize_independence_list(source.independence_list, data_dir) + roles = independence_roles(person, list_path) + channel = person["person_support_channel"].map(decode_text) + list_people = int(channel.isin(NON_ACS_CHANNELS).sum()) + record = LegacySPMRecord( + profile=expected_profile(source), + outside_spm_units=_count(universe, SPMUniverseStatus.OUTSIDE), + included_spm_units=_count(universe, SPMUniverseStatus.INCLUDED), + independence_list_people=list_people, + ) + return LegacySPMDeclarations( + record=record, universe=universe, independence_role=roles + ) + + +def derive_from_stored_tables( + source: LegacySPMSource, dataset: Any, *, data_dir: Path +) -> tuple[LegacySPMDeclarations, list[int]]: + """Derive declarations from the tables a country Microsimulation loaded. + + Returns the declarations and the dataset years they apply to. + """ + stored = stored_entity_tables(dataset, {"person": {}, "household": {}}) + if not stored: + raise ValueError( + "Legacy SPM declarations need the stored person and household " + "tables of the source." + ) + tables = stored[min(stored)] + declarations = derive_declarations( + source, + pd.DataFrame(tables["person"]), + pd.DataFrame(tables["household"]), + data_dir=data_dir, + ) + return declarations, sorted(stored) + + +def add_to_year_file_frames( + person: pd.DataFrame, spm_unit: pd.DataFrame, declarations: LegacySPMDeclarations +) -> None: + """Store the declarations as columns of a year file's tables.""" + if declarations.universe is not None: + spm_unit[UNIVERSE_VARIABLE] = _align( + spm_unit["spm_unit_id"], declarations.universe, "SPM units" + ).astype(object) + if declarations.independence_role is not None: + person[INDEPENDENCE_ROLE] = _align( + person["person_id"], declarations.independence_role, "people" + ).astype(bool) + + +def check_year_file_frames( + person: pd.DataFrame, spm_unit: pd.DataFrame, record: LegacySPMRecord +) -> None: + """Refuse a year file whose tables lack the declarations it records.""" + source = record.profile.source + if source.spm_universe_rule is not None: + if UNIVERSE_VARIABLE not in spm_unit.columns: + raise ValueError( + f"This dataset records legacy SPM declarations but stores no " + f"{UNIVERSE_VARIABLE!r} column. Regenerate it with " + "ensure_datasets() or create_datasets()." + ) + statuses = spm_unit[UNIVERSE_VARIABLE].map(decode_text) + if not statuses.isin(_DECLARED_STATUSES).all(): + raise ValueError( + f"The stored {UNIVERSE_VARIABLE!r} column must declare every SPM " + f"unit as one of {sorted(_DECLARED_STATUSES)}." + ) + if source.independence_list is not None: + if INDEPENDENCE_ROLE not in person.columns: + raise ValueError( + f"This dataset records legacy SPM declarations but stores no " + f"{INDEPENDENCE_ROLE!r} column. Regenerate it with " + "ensure_datasets() or create_datasets()." + ) + roles = person[INDEPENDENCE_ROLE] + if roles.isna().any() or not pd.api.types.is_bool_dtype(roles.dtype): + raise ValueError( + f"The stored {INDEPENDENCE_ROLE!r} column must be complete and boolean." + ) + + +def set_declarations( + simulation: Any, declarations: LegacySPMDeclarations, years: Sequence[int] +) -> None: + """Set the declarations on a built simulation and every branch it has. + + The scope is set for each of ``years``; the role is per person. Values + are aligned to each simulation's own entity IDs, so a region-scoped + simulation takes only its own units. + """ + for country_simulation in _simulation_and_branches(simulation): + names = [] + if declarations.universe is not None: + statuses = _align( + country_simulation.populations["spm_unit"].ids, + declarations.universe, + "SPM units", + ).astype(object) + for year in years: + country_simulation.set_input(UNIVERSE_VARIABLE, int(year), statuses) + names.append(UNIVERSE_VARIABLE) + if declarations.independence_role is not None: + roles = _align( + country_simulation.populations["person"].ids, + declarations.independence_role, + "people", + ).astype(bool) + country_simulation.set_input(INDEPENDENCE_ROLE, int(years[0]), roles) + names.append(INDEPENDENCE_ROLE) + _record_input_variables(country_simulation, names) + + +def _materialize_independence_list(pin: IndependenceListPin, data_dir: Path) -> Path: + verified = materialize_dataset( + "us", pin.bundle_dataset, data_dir=data_dir + ).bundle_dataset + if verified is None or verified.sha256 != pin.sha256: + raise ValueError( + f"The bundle's {pin.bundle_dataset!r} is not the certified " + f"{pin.sha256} file the legacy SPM registry pins." + ) + return Path(verified.path) + + +def _align(ids: Iterable[Any], values: pd.Series, what: str) -> np.ndarray: + aligned = pd.Series(np.asarray(ids)).map(values) + if aligned.isna().any(): + raise ValueError( + f"Legacy SPM declarations have no value for {int(aligned.isna().sum())} " + f"{what}." + ) + return np.asarray(aligned.to_numpy()) + + +def _count(universe: Optional[pd.Series], status: SPMUniverseStatus) -> Optional[int]: + if universe is None: + return None + return int(universe.eq(status.value).sum()) diff --git a/src/policyengine/tax_benefit_models/us/legacy_spm/independence.py b/src/policyengine/tax_benefit_models/us/legacy_spm/independence.py new file mode 100644 index 00000000..06cc52ac --- /dev/null +++ b/src/policyengine/tax_benefit_models/us/legacy_spm/independence.py @@ -0,0 +1,121 @@ +"""Take a legacy source's SPM independence roles from a certified list. + +BACKWARD COMPATIBILITY (#598). policyengine-us counts a 15–17-year-old as an +SPM adult only when ``is_spm_independent_minor_role`` is true; without the +input it falls back to "household head or spouse". A legacy source whose +ASEC/PUF records lack the role leaves Census-built units around an +independent teen without a classified adult. + +The certified national release carries a source-backed list of the role for +the same records. :func:`independence_roles` joins it on exactly: + +- ASEC/PUF people take the list's role. Every one of them must appear once, in + the same SPM unit, and the list must name nobody else. +- ACS people take ``is_household_head`` (or a stored + ``is_household_spouse``), the value the model's own fallback gives them. +""" + +from __future__ import annotations + +from pathlib import Path + +import numpy as np +import pandas as pd + +from .universe import ACS_CHANNEL, NON_ACS_CHANNELS, decode_text, person_link + +#: Model input the role is supplied as. +INDEPENDENCE_ROLE = "is_spm_independent_minor_role" + +_LIST_COLUMNS = ("person_id", "person_spm_unit_id", "is_spm_independence_role") + + +def independence_roles(person: pd.DataFrame, list_path: str | Path) -> pd.Series: + """Return each source person's SPM independence role, by person ID. + + ``person`` is the legacy source's stored person table and ``list_path`` + the hash-verified certified list. + + Raises: + ValueError: if the source already stores the role, if a person's + source channel is unknown, or if the list does not match the + source's ASEC/PUF people exactly. + """ + if INDEPENDENCE_ROLE in person.columns: + raise ValueError( + f"This source stores {INDEPENDENCE_ROLE!r} itself, so it is not a " + "legacy source for the independence role; remove its " + "independence_list from the legacy SPM registry." + ) + unit_link = person_link(person, "spm_unit") + for column in ("person_id", "person_support_channel"): + if column not in person.columns: + raise ValueError(f"The SPM independence role needs person {column!r}.") + if person["person_id"].duplicated().any(): + raise ValueError("The SPM independence role needs unique person IDs.") + + channel = person["person_support_channel"].map(decode_text) + is_acs = channel.eq(ACS_CHANNEL).to_numpy() + is_non_acs = channel.isin(NON_ACS_CHANNELS).to_numpy() + if not (is_acs | is_non_acs).all(): + raise ValueError("The SPM independence role found unknown source channels.") + + listed = read_independence_list(list_path) + source_units = person.loc[is_non_acs, ["person_id", unit_link]].set_index( + "person_id" + )[unit_link] + missing = source_units.index.difference(listed.index) + extra = listed.index.difference(source_units.index) + if len(missing) or len(extra): + raise ValueError( + "The certified SPM independence list does not cover this source's " + f"ASEC/PUF people exactly: {len(missing)} missing, {len(extra)} not " + "in the source." + ) + listed = listed.loc[source_units.index] + moved = listed["person_spm_unit_id"].to_numpy() != source_units.to_numpy() + if moved.any(): + raise ValueError( + f"The certified SPM independence list places {int(moved.sum())} " + "ASEC/PUF people in a different SPM unit from this source." + ) + + roles = pd.Series(False, index=person["person_id"].to_numpy(), dtype=bool) + roles.loc[source_units.index] = listed["is_spm_independence_role"].to_numpy( + dtype=bool + ) + roles.loc[person.loc[is_acs, "person_id"].to_numpy()] = model_fallback_roles( + person.loc[is_acs] + ) + roles.index.name = "person_id" + return roles + + +def read_independence_list(path: str | Path) -> pd.DataFrame: + """Read a certified independence list, refusing any malformed content.""" + listed = pd.read_csv(path, dtype={"is_spm_independence_role": str}) + if tuple(listed.columns) != _LIST_COLUMNS: + raise ValueError( + f"Unexpected SPM independence list columns {list(listed.columns)}." + ) + if listed["person_id"].duplicated().any(): + raise ValueError("The SPM independence list repeats person IDs.") + roles = listed["is_spm_independence_role"] + if not roles.isin(["True", "False"]).all(): + raise ValueError("The SPM independence list has non-boolean roles.") + listed["is_spm_independence_role"] = roles.eq("True") + return listed.set_index("person_id") + + +def model_fallback_roles(person: pd.DataFrame) -> np.ndarray: + """The model's own fallback role: household head or spouse.""" + roles = np.zeros(len(person), dtype=bool) + for column in ("is_household_head", "is_household_spouse"): + if column in person.columns: + values = person[column] + if values.isna().any(): + raise ValueError( + f"The SPM independence role needs complete {column!r}." + ) + roles |= values.to_numpy(dtype=bool) + return roles diff --git a/src/policyengine/tax_benefit_models/us/legacy_spm/model_adapter.py b/src/policyengine/tax_benefit_models/us/legacy_spm/model_adapter.py new file mode 100644 index 00000000..b1cd4e2c --- /dev/null +++ b/src/policyengine/tax_benefit_models/us/legacy_spm/model_adapter.py @@ -0,0 +1,337 @@ +"""Let a policyengine-us release that cannot read SPM scope honour it. + +############################################################################# +# BACKWARD COMPATIBILITY — MODEL OVERRIDE, REMOVE WHEN THE MODEL READS SCOPE. # +# # +# policyengine-us 2.2.1 has no measurement-scope input, so it asks the # +# canonical SPM provider for every SPM unit. ``prepare_microsimulation`` # +# decides per simulation: # +# # +# - "native": the model defines ``spm_unit_spm_universe_status`` itself # +# (PolicyEngine/policyengine-us#9462). The derived declaration is just # +# an input, and nothing here changes the model. # +# - "override": it does not. A variable-only reform adds the declaration # +# input and wraps the SPM chain so that units OUTSIDE the measurement # +# never reach the provider and every SPM outcome for them is NaN ("not # +# measured"). Measured units keep the canonical values. # +# # +# The override wraps the installed formulas, with ONE exception: # +# ``spm_unit_income_decile`` ranks units, so it is recalculated over # +# measured units with the installed method (same weights, same ranking). # +# ``test_us_legacy_spm_model_contract`` pins that installed method, so a # +# model change cannot silently diverge from it. Before wrapping, the # +# override checks that the SPM chain still has the shape it was verified # +# against, and refuses to run otherwise. # +############################################################################# + +The override must be applied after the simulation is constructed (so policy +and structural reforms are in place) and before populations are built (so the +stored declaration loads as an input). Applying it again changes nothing. +""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping +from typing import Any, Literal, Protocol + +import numpy as np + +from policyengine.tax_benefit_models.us.legacy_inputs import ( + _simulation_and_branches, +) + +from .independence import INDEPENDENCE_ROLE +from .record import ADAPTER_REVISION +from .universe import SPMUniverseStatus + +#: The measurement-scope input newer sources carry +#: (PolicyEngine/microcosm#977, PolicyEngine/policyengine-us#9462). +UNIVERSE_VARIABLE = "spm_unit_spm_universe_status" + +#: Override mode only: whether an SPM unit is outside the measurement. +OUTSIDE_VARIABLE = "policyengine_legacy_spm_outside" + +#: ``Variable.metadata`` key marking the variables the override adds. +ADAPTER_MARKER = "policyengine_legacy_spm_adapter" + +ModelMode = Literal["native", "override"] + +#: Canonical amount variables built by spm-calculator, and the +#: ``masked_policyengine_amount`` field each one returns. +THRESHOLD_FIELDS = { + "spm_unit_reference_spm_threshold": "reference_threshold", + "spm_unit_unadjusted_spm_threshold": "unadjusted_threshold", + "spm_unit_geographic_adjustment": "geographic_factor", + "spm_unit_spm_threshold": "threshold", + "spm_unit_spm_threshold_housing_portion": "housing_portion", +} + +#: Formula variables in the SPM measurement whose results become NaN for +#: outside units. The housing allocation comes first: with it NaN, the +#: installed capped-subsidy formula treats an outside unit as unassisted, so +#: it never asks the provider about one. +MASKED_FORMULA_VARIABLES = ( + "spm_unit_allocated_housing_subsidy", + "spm_unit_allocated_tenant_payment", + "spm_unit_capped_housing_subsidy", + "spm_unit_benefits", + "poverty_gap", + "deep_poverty_line", + "deep_poverty_gap", + "spm_unit_is_in_spm_poverty", + "spm_unit_is_in_deep_spm_poverty", + "in_poverty", + "in_deep_poverty", + "person_in_poverty", + "spm_unit_oecd_equiv_net_income", +) + +#: Indicators that become floats (1.0, 0.0, or NaN when not measured). +FLOAT_INDICATORS = frozenset( + { + "spm_unit_is_in_spm_poverty", + "spm_unit_is_in_deep_spm_poverty", + "in_poverty", + "in_deep_poverty", + "person_in_poverty", + } +) + +#: ``adds``/``subtracts`` variables that carry NaN from a masked component. +NAN_PROPAGATING_VARIABLES = ("spm_unit_net_income", "poverty_line") + +#: Ranks units, so it is recalculated over measured units only. +DECILE_VARIABLE = "spm_unit_income_decile" + +#: Every SPM measurement variable that is NaN for an outside unit. +SPM_MEASUREMENT_VARIABLES = ( + *THRESHOLD_FIELDS, + *MASKED_FORMULA_VARIABLES, + *NAN_PROPAGATING_VARIABLES, + DECILE_VARIABLE, +) + + +class CountrySimulation(Protocol): + """The parts of a policyengine-us simulation the adapter uses.""" + + tax_benefit_system: Any + branches: Mapping[str, Any] + + def apply_reform(self, reform: Any) -> None: ... + + +class Population(Protocol): + """A policyengine-core population, as a formula receives it.""" + + entity: Any + + def __call__(self, variable_name: str, period: Any) -> Any: ... + + +def prepare_microsimulation(microsimulation: CountrySimulation) -> ModelMode: + """Make a simulation and its branches read the legacy declarations. + + Returns the mode every simulation runs in. + + Raises: + ValueError: if the model defines the scope input with members the + declaration cannot use, if its SPM chain no longer has the shape + the override was verified against, or if branches disagree. + """ + modes: set[ModelMode] = set() + for simulation in _simulation_and_branches(microsimulation): + variable = simulation.tax_benefit_system.variables.get(UNIVERSE_VARIABLE) + if variable is None: + simulation.apply_reform(_override_reform()) + modes.add("override") + elif ADAPTER_MARKER in (variable.metadata or {}): + modes.add("override") + else: + _check_native_universe(variable) + modes.add("native") + if len(modes) != 1: + raise ValueError( + "A simulation and its branches disagree on whether policyengine-us " + "reads SPM measurement scope natively." + ) + return modes.pop() + + +def _check_native_universe(variable: Any) -> None: + members = {item.name for item in getattr(variable, "possible_values", ()) or ()} + needed = {SPMUniverseStatus.INCLUDED.value, SPMUniverseStatus.OUTSIDE.value} + if not needed <= members: + raise ValueError( + f"policyengine-us defines {UNIVERSE_VARIABLE!r} with members " + f"{sorted(members)}; the legacy SPM declarations need {sorted(needed)}." + ) + + +def _check_spm_chain(variables: Mapping[str, Any]) -> None: + """Refuse a model whose SPM chain differs from the one verified.""" + problems = [] + for name in (*THRESHOLD_FIELDS, *MASKED_FORMULA_VARIABLES, DECILE_VARIABLE): + if name not in variables: + problems.append(f"{name} is not defined") + elif not variables[name].formulas: + problems.append(f"{name} has no formula") + for name in NAN_PROPAGATING_VARIABLES: + if name not in variables: + problems.append(f"{name} is not defined") + elif variables[name].formulas or not variables[name].adds: + problems.append(f"{name} no longer adds up its components") + if INDEPENDENCE_ROLE not in variables: + problems.append(f"{INDEPENDENCE_ROLE} is not defined") + if problems: + raise ValueError( + "The legacy SPM model override was verified against a different " + "policyengine-us SPM chain: " + "; ".join(problems) + ". Re-verify it " + "(docs/engineering/runbooks/us-legacy-spm-sources.md) before using " + "legacy sources with this release." + ) + + +def _override_reform() -> type: + from policyengine_core.enums import Enum as CoreEnum + from policyengine_core.periods import YEAR + from policyengine_core.reforms import Reform + from policyengine_core.variables import Variable + from policyengine_us.entities import SPMUnit + from spm_calculator.errors import SPMInputError + + marker = {ADAPTER_MARKER: ADAPTER_REVISION} + + class SPMUniverseStatusInput(CoreEnum): + INCLUDED = "Inside the SPM measurement population" + OUTSIDE = "Outside the SPM measurement population" + UNRESOLVED = "No SPM measurement-scope declaration" + + class spm_unit_spm_universe_status(Variable): + value_type = CoreEnum + possible_values = SPMUniverseStatusInput + default_value = SPMUniverseStatusInput.UNRESOLVED + entity = SPMUnit + definition_period = YEAR + label = "SPM measurement scope" + documentation = ( + "BACKWARD COMPATIBILITY: added by policyengine.py for a " + "policyengine-us release without a native SPM measurement-scope " + "input, to carry the declaration derived for a legacy source." + ) + metadata = marker + + class policyengine_legacy_spm_outside(Variable): + value_type = bool + entity = SPMUnit + definition_period = YEAR + label = "Outside the SPM measurement population" + metadata = marker + + def formula(spm_unit, period, parameters): + status = spm_unit(UNIVERSE_VARIABLE, period) + unresolved = status == SPMUniverseStatusInput.UNRESOLVED + if np.any(unresolved): + raise SPMInputError( + "SPM_UNIVERSE_REQUIRED", + f"{int(np.sum(unresolved))} SPM units have no SPM " + f"measurement-scope declaration for {period}", + ) + return status == SPMUniverseStatusInput.OUTSIDE + + class legacy_spm_declarations(Reform): + def apply(self): + _check_spm_chain(self.variables) + self.add_variable(spm_unit_spm_universe_status) + self.add_variable(policyengine_legacy_spm_outside) + for name, field in THRESHOLD_FIELDS.items(): + self.update_variable(_threshold_variable(name, field)) + for name in MASKED_FORMULA_VARIABLES: + self.update_variable(_masked_variable(self.variables[name])) + self.update_variable(_income_decile_variable()) + + return legacy_spm_declarations + + +def _outside(population: Population, period: Any) -> np.ndarray: + if population.entity.key == "spm_unit": + values = population(OUTSIDE_VARIABLE, period) + else: + values = population.spm_unit(OUTSIDE_VARIABLE, period) # type: ignore[attr-defined] + return np.asarray(values, dtype=bool) + + +def _threshold_variable(name: str, field: str) -> type: + from policyengine_core.variables import Variable + from policyengine_us.spm import masked_policyengine_amount + + def formula(spm_unit: Population, period: Any, parameters: Any) -> np.ndarray: + outside = _outside(spm_unit, period) + amounts = masked_policyengine_amount(spm_unit, period, field, ~outside) + return np.where(outside, np.nan, amounts) + + return type(name, (Variable,), {"formula": formula, "__module__": __name__}) + + +def _masked_variable(original: Any) -> type: + """Wrap an installed formula variable so that outside units are NaN.""" + from policyengine_core.variables import Variable + + def formula(population: Population, period: Any, parameters: Any) -> np.ndarray: + original_formula: Callable[..., Any] = original.get_formula(period) + if original_formula.__code__.co_argcount == 2: + values = original_formula(population, period) + else: + values = original_formula(population, period, parameters) + return np.where( + _outside(population, period), + np.nan, + np.asarray(values, dtype=np.float64), + ) + + attributes: dict[str, Any] = {"formula": formula, "__module__": __name__} + if original.name in FLOAT_INDICATORS: + attributes["value_type"] = float + attributes["default_value"] = 0.0 + return type(original.name, (Variable,), attributes) + + +def _income_decile_variable() -> type: + """Rank measured SPM units only; outside units have no decile. + + The installed formula ranks every unit by OECD-equivalised SPM income, + weighted by people. With outside units' income NaN, it would place them + in the top decile and count their weight, so the ranking is recalculated + over measured units, with the same weights and the same ranking method. + + This is the one installed formula the override re-implements rather than + wraps. ``test_the_recalculated_decile_matches_the_installed_method`` pins + the installed formula it mirrors. + """ + from microdf import MicroSeries + from policyengine_core.variables import Variable + + def formula(spm_unit: Population, period: Any, parameters: Any) -> np.ndarray: + outside = _outside(spm_unit, period) + income = np.asarray(spm_unit("spm_unit_oecd_equiv_net_income", period)) + weights = np.asarray(spm_unit("spm_unit_weight", period)) * np.asarray( + spm_unit.nb_persons() # type: ignore[attr-defined] + ) + deciles = np.full(len(income), np.nan) + measured = ~outside + if measured.any(): + deciles[measured] = np.asarray( + MicroSeries(income[measured], weights=weights[measured]).decile_rank() + ) + return deciles + + return type( + DECILE_VARIABLE, + (Variable,), + { + "value_type": float, + "default_value": 0.0, + "formula": formula, + "__module__": __name__, + }, + ) diff --git a/src/policyengine/tax_benefit_models/us/legacy_spm/record.py b/src/policyengine/tax_benefit_models/us/legacy_spm/record.py new file mode 100644 index 00000000..cfbc2513 --- /dev/null +++ b/src/policyengine/tax_benefit_models/us/legacy_spm/record.py @@ -0,0 +1,167 @@ +"""Records of the legacy SPM declarations a year file or output was made with. + +BACKWARD COMPATIBILITY (#597, #598). A year file cut from a legacy source, and +every output calculated from one, records the :class:`LegacySPMProfile` its +declarations were derived under. The profile is everything that determines +them: the registry entry (source hash, scope rule, independence list hash) +and :data:`ADAPTER_REVISION`. Checks follow the year-file stamp +(``legacy_inputs.year_file_stamp_problem``): + +- **total**: any stored value, readable or not, gives either ``None`` or a + reason, and never raises; +- **exact**: a file is current only when its profile equals the one this + release derives for its source. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from pathlib import Path +from typing import Any, Final, Literal, Optional + +from pydantic import BaseModel, ConfigDict, Field, ValidationError + +from policyengine.tax_benefit_models.us.legacy_inputs import ( + _ABSENT, + _UNREADABLE, + _read_json_dataset, + _write_json_dataset, +) + +from .registry import LegacySPMSource + +#: Raise when a change to how declarations are derived or applied means files +#: and outputs made before it must be made again. +ADAPTER_REVISION: Final = 1 + +#: Key of the record in a US dataset's ``metadata``. +RECORD_KEY: Final = "legacy_spm_declarations" + +#: H5 dataset holding the record as UTF-8 JSON in a US file written by +#: ``PolicyEngineUSDataset.save()``. +RECORD_H5_DATASET: Final = "policyengine_legacy_spm_declarations" + +#: How a run applied the declarations to the model (see +#: :mod:`.model_adapter`). +ModelModeName = Literal["native", "override"] + + +class LegacySPMProfile(BaseModel): + """Everything that determines a legacy source's derived declarations.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + adapter_revision: int + source: LegacySPMSource + + +class LegacySPMRecord(BaseModel): + """What a year file or output records about its legacy SPM declarations.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + schema_version: Literal[1] = 1 + profile: LegacySPMProfile + #: SPM units the declaration puts outside the SPM measurement. + outside_spm_units: Optional[int] = Field(default=None, ge=0) + #: SPM units it puts inside. + included_spm_units: Optional[int] = Field(default=None, ge=0) + #: People whose independence role came from the certified list. + independence_list_people: Optional[int] = Field(default=None, ge=0) + #: Outputs only: how the run's model applied the declarations. + model_mode: Optional[ModelModeName] = None + + +def expected_profile(source: LegacySPMSource) -> LegacySPMProfile: + """Return the profile this release derives declarations under.""" + return LegacySPMProfile(adapter_revision=ADAPTER_REVISION, source=source) + + +def parse_record(value: Any) -> Optional[LegacySPMRecord]: + """Return a stored record as a model, or ``None`` if it is not one.""" + if value is None or value is _ABSENT or value is _UNREADABLE: + return None + try: + return LegacySPMRecord.model_validate(value) + except ValidationError: + return None + + +def record_problem(stored: Any, expected: Optional[LegacySPMProfile]) -> Optional[str]: + """Return why a stored record does not match ``expected``, or ``None``. + + ``stored`` is a record's JSON value, a record model, or ``_ABSENT``. + ``expected`` is ``None`` for data whose source needs no legacy + declarations. The reason is a clause starting "it" or "its". + """ + if isinstance(stored, LegacySPMRecord): + stored = stored.model_dump(mode="json") + if expected is None: + if stored is _ABSENT: + return None + return "it records legacy SPM declarations that its source does not need" + if stored is _ABSENT: + return ( + "it has no record of the legacy SPM declarations its source needs, " + "so it was made by an older policyengine.py or not written completely" + ) + record = parse_record(stored) + if record is None: + return ( + "its record of legacy SPM declarations is unreadable, malformed or " + "from another version of policyengine.py" + ) + if record.profile != expected: + return ( + "its legacy SPM declarations were derived under profile " + f"{_describe(record.profile)}, and this release derives them under " + f"{_describe(expected)}" + ) + return None + + +def read_record_value(path: str | Path) -> Any: + """Return the JSON value of a file's record, ``_ABSENT`` or ``_UNREADABLE``. + + Never raises for the file's contents. Errors that say nothing about them + (no permission, a directory, a lock) are raised, as for the year-file + stamp. + """ + import h5py + + try: + with h5py.File(path, "r") as file: + return _read_json_dataset(file, RECORD_H5_DATASET) + except (PermissionError, IsADirectoryError, BlockingIOError): + raise + except OSError: + return _UNREADABLE + + +def year_file_problem( + path: str | Path, expected: Optional[LegacySPMProfile] +) -> Optional[str]: + """Return why a year file's legacy SPM declarations are out of date.""" + return record_problem(read_record_value(path), expected) + + +def write_record(path: str | Path, record: LegacySPMRecord) -> None: + """Store a record in an H5 file, replacing any it has.""" + _write_json_dataset(path, RECORD_H5_DATASET, record.model_dump(mode="json")) + + +def metadata_value(metadata: Mapping[str, Any]) -> Any: + """Return the record stored in a dataset's ``metadata``, or ``_ABSENT``.""" + return metadata.get(RECORD_KEY, _ABSENT) + + +def _describe(profile: LegacySPMProfile) -> str: + source = profile.source + independence = ( + source.independence_list.sha256[:12] if source.independence_list else None + ) + return ( + f"(revision {profile.adapter_revision}, source " + f"{source.source_sha256[:12]}, scope rule {source.spm_universe_rule}, " + f"independence list {independence})" + ) diff --git a/src/policyengine/tax_benefit_models/us/legacy_spm/registry.py b/src/policyengine/tax_benefit_models/us/legacy_spm/registry.py new file mode 100644 index 00000000..2d2b3592 --- /dev/null +++ b/src/policyengine/tax_benefit_models/us/legacy_spm/registry.py @@ -0,0 +1,76 @@ +"""Certified US sources that predate SPM declarations. + +BACKWARD COMPATIBILITY (#597, #598): register a source here only when it was +certified before SPM measurement-scope and independence-role declarations +existed. New certified sources must carry the declarations themselves. +""" + +from __future__ import annotations + +from typing import Literal, Optional + +from pydantic import BaseModel, ConfigDict + +#: How a legacy source's SPM measurement scope is derived. +#: +#: ``acs_household_population``: Census's ACS SPM methodology measures the +#: household population only, so ACS group-quarters units are outside the +#: measurement (see :mod:`.universe`). +SPMUniverseRule = Literal["acs_household_population"] + + +class IndependenceListPin(BaseModel): + """A certified list of SPM independence roles, pinned by bundle entry.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + #: Name of the list's entry in the bundle manifest's datasets. + bundle_dataset: str + #: Its certified SHA-256. The bundle manifest must pin the same value. + sha256: str + + +class LegacySPMSource(BaseModel): + """A certified source and the SPM declarations it needs derived.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + #: Human-readable identity: logical dataset name and certified revision. + name: str + #: Certified SHA-256 of the source file; the registry key. + source_sha256: str + #: How to derive its SPM measurement scope, if it needs one. + spm_universe_rule: Optional[SPMUniverseRule] = None + #: Where its ASEC/PUF records' SPM independence role comes from, if it + #: needs one. + independence_list: Optional[IndependenceListPin] = None + + +#: ``populace_us_2024_acs_local`` Build O. It keeps ACS group-quarters +#: residents without declaring SPM scope, and it was built a few hours before +#: PolicyEngine/microcosm#959, so its ASEC/PUF records lack the independence +#: role that the national release carries for the same records. +POPULACE_US_2024_ACS_LOCAL_BUILD_O = LegacySPMSource( + name=( + "populace_us_2024_acs_local@" + "populace-us-2024-buildo-acs-local-767312d60-20260923T074941Z" + ), + source_sha256="769756c31f3ca646d12c272511744dec04c0e68870c6946dd945fbba65b6a7ec", + spm_universe_rule="acs_household_population", + independence_list=IndependenceListPin( + bundle_dataset="source_spm_person_independence", + sha256="22b5968d90fecfeef7614583e493fe10cc16bda8b5be82e6f49a5bc2102d3ce5", + ), +) + +LEGACY_SPM_SOURCES: tuple[LegacySPMSource, ...] = (POPULACE_US_2024_ACS_LOCAL_BUILD_O,) + + +def legacy_spm_source(source_sha256: Optional[str]) -> Optional[LegacySPMSource]: + """Return the registered legacy source with this certified SHA-256.""" + if source_sha256 is None: + return None + for source in LEGACY_SPM_SOURCES: + if source.source_sha256 == source_sha256: + return source + return None diff --git a/src/policyengine/tax_benefit_models/us/legacy_spm/runs.py b/src/policyengine/tax_benefit_models/us/legacy_spm/runs.py new file mode 100644 index 00000000..21a29f64 --- /dev/null +++ b/src/policyengine/tax_benefit_models/us/legacy_spm/runs.py @@ -0,0 +1,140 @@ +"""Decide which legacy SPM declarations a US run needs. + +BACKWARD COMPATIBILITY (#597, #598). A run gets its declarations in one of +two ways: + +- from a ``create_datasets`` year file, which stores them as columns and + records the profile they were derived under; or +- straight from a legacy source file (a run over the raw certified file), + whose stored evidence they are derived from in memory. Such a file is + recognised by its evidence columns and then its SHA-256, so a file whose + bytes differ from a certified legacy source gets nothing. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Optional + +import pandas as pd + +from policyengine.tax_benefit_models.us.legacy_inputs import _ABSENT +from policyengine.utils.hashing import sha256_file + +from .declarations import ( + LegacySPMDeclarations, + check_year_file_frames, + derive_declarations, +) +from .record import ( + LegacySPMProfile, + LegacySPMRecord, + expected_profile, + metadata_value, + parse_record, + record_problem, +) +from .registry import LegacySPMSource, legacy_spm_source + +#: Columns only a legacy source file (not a year file) stores. +_SOURCE_EVIDENCE = {"person": "person_support_channel", "household": "TYPEHUGQ"} + +#: ``{(path, size, mtime_ns): sha256}``, so a file is hashed once per process. +_SHA256_CACHE: dict[tuple[str, int, int], str] = {} + + +@dataclass(frozen=True) +class LegacySPMRun: + """The declarations a run applies, and where they come from.""" + + record: LegacySPMRecord + #: Set when they were derived in memory rather than stored as columns. + in_memory: Optional[LegacySPMDeclarations] = None + + +def legacy_spm_for_run(dataset: Any) -> Optional[LegacySPMRun]: + """Return the declarations a run over ``dataset`` must apply. + + Raises: + ValueError: if the dataset records declarations that are out of + date, unreadable, or missing from its tables. + """ + stored = metadata_value(dataset.metadata) + if stored is not _ABSENT: + record = parse_record(stored) + expected = _expected_for_record(record) + problem = record_problem(stored, expected) + if problem is not None: + raise ValueError( + f"This US dataset is out of date: {problem}. Regenerate it with " + "ensure_datasets() or create_datasets()." + ) + assert record is not None + check_year_file_frames( + pd.DataFrame(dataset.data.person), + pd.DataFrame(dataset.data.spm_unit), + record, + ) + return LegacySPMRun(record=record) + + source = raw_legacy_source(dataset) + if source is None: + return None + declarations = derive_declarations( + source, + pd.DataFrame(dataset.data.person), + pd.DataFrame(dataset.data.household), + data_dir=Path(dataset.filepath).parent, + ) + return LegacySPMRun(record=declarations.record, in_memory=declarations) + + +def expected_profile_for_dataset(dataset: Any) -> Optional[LegacySPMProfile]: + """Return the profile a run over ``dataset`` applies, without deriving.""" + stored = metadata_value(getattr(dataset, "metadata", None) or {}) + if stored is not _ABSENT: + return _expected_for_record(parse_record(stored)) + source = raw_legacy_source(dataset) + return expected_profile(source) if source is not None else None + + +def raw_legacy_source(dataset: Any) -> Optional[LegacySPMSource]: + """Return the legacy source a dataset's file is, if it is one. + + Only a dataset loaded from a file that stores legacy source evidence is + hashed, and only the first time in a process. + """ + data = getattr(dataset, "data", None) + filepath = getattr(dataset, "filepath", None) + if data is None or not filepath or getattr(dataset, "is_output_dataset", False): + return None + for entity, column in _SOURCE_EVIDENCE.items(): + if column not in getattr(data, entity).columns: + return None + path = Path(filepath) + if not path.is_file(): + return None + status = path.stat() + key = (os.fspath(path.resolve()), status.st_size, status.st_mtime_ns) + if key not in _SHA256_CACHE: + _SHA256_CACHE[key] = sha256_file(path) + return legacy_spm_source(_SHA256_CACHE[key]) + + +def _expected_for_record( + record: Optional[LegacySPMRecord], +) -> Optional[LegacySPMProfile]: + if record is None: + # Judge it against a profile it cannot match, so it reads as stale. + return _UNMATCHABLE + source = legacy_spm_source(record.profile.source.source_sha256) + return expected_profile(source) if source is not None else None + + +#: A profile no stored record carries. +_UNMATCHABLE = LegacySPMProfile( + adapter_revision=-1, + source=LegacySPMSource(name="", source_sha256=""), +) diff --git a/src/policyengine/tax_benefit_models/us/legacy_spm/universe.py b/src/policyengine/tax_benefit_models/us/legacy_spm/universe.py new file mode 100644 index 00000000..9e41bce0 --- /dev/null +++ b/src/policyengine/tax_benefit_models/us/legacy_spm/universe.py @@ -0,0 +1,137 @@ +"""Derive a legacy source's SPM measurement scope from its record types. + +BACKWARD COMPATIBILITY (#597). Census's ACS SPM methodology measures the +household population only, so an ACS group-quarters SPM unit is outside the +SPM measurement population. The declaration uses the member names of the +measurement-scope contract newer sources carry (PolicyEngine/microcosm#977): +``INCLUDED`` and ``OUTSIDE``. Nothing here reads age or adult counts. +""" + +from __future__ import annotations + +from enum import Enum + +import numpy as np +import pandas as pd + + +class SPMUniverseStatus(str, Enum): + """Whether an SPM unit is inside the SPM measurement population.""" + + INCLUDED = "INCLUDED" + OUTSIDE = "OUTSIDE" + #: No declaration. Calculations refuse it rather than guess. + UNRESOLVED = "UNRESOLVED" + + +ACS_CHANNEL = "acs_2024_1yr" +NON_ACS_CHANNELS = frozenset({"asec", "puf_tax_detail"}) + +#: ACS household record type (``TYPEHUGQ``) of a housing unit. +ACS_HOUSING_UNIT = 1 +#: ACS group-quarters record types (``TYPEHUGQ``) and the relationship code +#: (``RELSHIPP``) each requires: 2 institutional (37), 3 noninstitutional (38). +ACS_GROUP_QUARTERS = {2: 37, 3: 38} + + +def classify_acs_household_population( + person: pd.DataFrame, household: pd.DataFrame +) -> pd.Series: + """Return each SPM unit's measurement status under the ACS household rule. + + ``person`` and ``household`` are the source's stored entity tables. The + result is indexed by SPM unit ID and holds ``"INCLUDED"`` or + ``"OUTSIDE"``: + + - ACS records with ``TYPEHUGQ`` 2 or 3 and the matching ``RELSHIPP`` are + group quarters, so ``OUTSIDE``; + - ACS housing-unit records (``TYPEHUGQ`` 1, ``RELSHIPP`` not 37 or 38) and + records from the ``asec`` and ``puf_tax_detail`` channels (no ACS + record-type evidence at all) are ``INCLUDED``. + + Raises: + ValueError: if any evidence is missing, unrecognised or inconsistent, + or if an SPM unit mixes group-quarters and household people. + """ + unit_link = person_link(person, "spm_unit") + household_link = person_link(person, "household") + for column in ("person_support_channel", "RELSHIPP"): + if column not in person.columns: + raise ValueError(f"SPM measurement scope needs person {column!r}.") + for column in ("household_id", "TYPEHUGQ"): + if column not in household.columns: + raise ValueError(f"SPM measurement scope needs household {column!r}.") + if household["household_id"].duplicated().any(): + raise ValueError("SPM measurement scope needs unique household_id values.") + if not person[household_link].isin(household["household_id"]).all(): + raise ValueError("SPM measurement scope found people with unknown households.") + + channel = person["person_support_channel"].map(decode_text) + relationship = pd.to_numeric(person["RELSHIPP"], errors="raise") + record_type = pd.to_numeric( + person[household_link].map(household.set_index("household_id")["TYPEHUGQ"]), + errors="raise", + ) + + is_acs = channel.eq(ACS_CHANNEL) + is_non_acs = channel.isin(NON_ACS_CHANNELS) + unknown_channel = ~(is_acs | is_non_acs) + if unknown_channel.any(): + channels = sorted(map(str, channel[unknown_channel].unique())) + raise ValueError( + f"SPM measurement scope has no rule for source channels {channels}." + ) + + in_housing_unit = ( + is_acs + & record_type.eq(ACS_HOUSING_UNIT) + & relationship.notna() + & ~relationship.isin(list(ACS_GROUP_QUARTERS.values())) + ) + in_group_quarters = pd.Series(False, index=person.index) + for quarters_type, quarters_relationship in ACS_GROUP_QUARTERS.items(): + in_group_quarters |= ( + is_acs + & record_type.eq(quarters_type) + & relationship.eq(quarters_relationship) + ) + recognised_non_acs = is_non_acs & record_type.isna() & relationship.isna() + unrecognised = ~(in_housing_unit | in_group_quarters | recognised_non_acs) + if unrecognised.any(): + raise ValueError( + f"SPM measurement scope cannot classify {int(unrecognised.sum())} " + "people: their TYPEHUGQ/RELSHIPP evidence is missing or " + "inconsistent with their source channel." + ) + + group_quarters_by_unit = in_group_quarters.groupby(person[unit_link]) + mixed_units = group_quarters_by_unit.nunique() > 1 + if mixed_units.any(): + raise ValueError( + f"SPM measurement scope found {int(mixed_units.sum())} SPM units " + "mixing group-quarters and household people." + ) + outside = group_quarters_by_unit.first().astype(bool) + status = pd.Series( + np.where( + outside, + SPMUniverseStatus.OUTSIDE.value, + SPMUniverseStatus.INCLUDED.value, + ), + index=outside.index, + dtype=object, + ) + status.index.name = "spm_unit_id" + return status + + +def person_link(person: pd.DataFrame, entity: str) -> str: + """Return the person table's link column to a group entity.""" + for column in (f"person_{entity}_id", f"{entity}_id"): + if column in person.columns: + return column + raise ValueError(f"Legacy SPM declarations need a person {entity} link.") + + +def decode_text(value: object) -> object: + return value.decode("utf-8") if isinstance(value, bytes) else value diff --git a/src/policyengine/tax_benefit_models/us/model.py b/src/policyengine/tax_benefit_models/us/model.py index f07f1611..b008dc97 100644 --- a/src/policyengine/tax_benefit_models/us/model.py +++ b/src/policyengine/tax_benefit_models/us/model.py @@ -1,4 +1,5 @@ import datetime +from pathlib import Path from typing import TYPE_CHECKING, Optional import pandas as pd @@ -22,6 +23,16 @@ apply_legacy_input_renames, apply_legacy_input_renames_to_microsimulation, ) +from .legacy_spm import ( + RECORD_KEY as LEGACY_SPM_RECORD_KEY, +) +from .legacy_spm import ( + derive_from_stored_tables, + legacy_spm_for_run, + legacy_spm_source, + prepare_microsimulation, + set_declarations, +) from .spm import ( SPMProvenance, SPMSelection, @@ -174,6 +185,11 @@ def run(self, simulation: "Simulation") -> "Simulation": validate_policyengine_inputs(dataset.data.entity_data) + # BACKWARD COMPATIBILITY (#597, #598): a legacy certified source + # carries no SPM declarations. A year file cut from one stores them; + # a run straight over one derives them now (see ``legacy_spm``). + legacy_spm = legacy_spm_for_run(dataset) + # Apply regional scoping if specified if simulation.scoping_strategy: scoped_data = simulation.scoping_strategy.apply( @@ -218,6 +234,13 @@ class InputMicrosimulation(Microsimulation): spm=simulation.spm_config, default_input_period=dataset.year, ) + legacy_spm_mode = None + if legacy_spm is not None: + # BACKWARD COMPATIBILITY (#597, #598): after construction, so every + # policy and structural reform is in place, and before populations + # are built, so stored declarations load as inputs in the reform + # and its baseline. + legacy_spm_mode = prepare_microsimulation(microsim) # Use ``microsim.tax_benefit_system``, not the module-level # ``system``: ``Microsimulation.__init__`` applies structural # reforms (e.g. ``gov.contrib.ctc.*``) to its per-sim system but @@ -244,6 +267,10 @@ class InputMicrosimulation(Microsimulation): microsim, dataset, microsim.tax_benefit_system ) ) + if legacy_spm is not None and legacy_spm.in_memory is not None: + # BACKWARD COMPATIBILITY (#597, #598): declarations derived from a + # raw legacy source are set on the simulation and its baseline. + set_declarations(microsim, legacy_spm.in_memory, [dataset.year]) data = { "person": pd.DataFrame(), @@ -325,6 +352,14 @@ class InputMicrosimulation(Microsimulation): data["household"], weights="household_weight" ) + output_metadata = { + "spm_config": dict(microsim.spm_config), + RENAMES_RECORD_KEY: dict(sorted(legacy_input_renames.items())), + } + if legacy_spm is not None: + output_metadata[LEGACY_SPM_RECORD_KEY] = legacy_spm.record.model_copy( + update={"model_mode": legacy_spm_mode} + ).model_dump(mode="json") simulation.output_dataset = PolicyEngineUSDataset( id=simulation.id, name=dataset.name, @@ -332,10 +367,7 @@ class InputMicrosimulation(Microsimulation): filepath=str(_output_dataset_filepath(simulation)), year=simulation.dataset.year, is_output_dataset=True, - metadata={ - "spm_config": dict(microsim.spm_config), - RENAMES_RECORD_KEY: dict(sorted(legacy_input_renames.items())), - }, + metadata=output_metadata, data=USYearData( person=data["person"], marital_unit=data["marital_unit"], @@ -534,6 +566,22 @@ def managed_microsimulation( ) microsim = Microsimulation(dataset=source.path, spm=selection, **kwargs) legacy_input_renames = apply_legacy_input_renames_to_microsimulation(microsim) + # BACKWARD COMPATIBILITY (#597, #598): a legacy certified source carries + # no SPM declarations, so derive them from its stored evidence and set + # them on the simulation and its branches (see ``legacy_spm``). + legacy_spm_record = None + legacy_source = legacy_spm_source( + source.bundle_dataset.sha256 if source.bundle_dataset else None + ) + if legacy_source is not None: + declarations, years = derive_from_stored_tables( + legacy_source, microsim.dataset, data_dir=Path(source.path).parent + ) + mode = prepare_microsimulation(microsim) + set_declarations(microsim, declarations, years) + legacy_spm_record = declarations.record.model_copy( + update={"model_mode": mode} + ).model_dump(mode="json") microsim.policyengine_bundle = dict(us_latest.release_bundle) microsim.policyengine_bundle.update( build_runtime_dataset_provenance( @@ -546,6 +594,8 @@ def managed_microsimulation( microsim.policyengine_bundle[RENAMES_RECORD_KEY] = dict( sorted(legacy_input_renames.items()) ) + if legacy_spm_record is not None: + microsim.policyengine_bundle[LEGACY_SPM_RECORD_KEY] = legacy_spm_record return microsim diff --git a/tests/test_poverty_run.py b/tests/test_poverty_run.py index 4122b242..1367d2e7 100644 --- a/tests/test_poverty_run.py +++ b/tests/test_poverty_run.py @@ -242,10 +242,10 @@ def test__given_range_filter__then_only_counts_persons_in_range(self): class TestPovertyRunZeroPopulation: """Tests for Poverty.run() with zero matching population.""" - def test__given_no_matching_persons__then_rate_is_zero(self): + def test__given_no_matching_persons__then_rate_is_none(self): """Given: Filter matches no persons When: Running poverty - Then: headcount=0, total=0, rate=0.0 + Then: headcount=0, total=0, rate=None (nobody measured) """ sim = _make_sim( poverty_values=[True, True], @@ -262,4 +262,5 @@ def test__given_no_matching_persons__then_rate_is_zero(self): assert pov.headcount == 0.0 assert pov.total_population == 0.0 - assert pov.rate == 0.0 + # Nobody in the group is measured, which is not a measured rate of 0. + assert pov.rate is None diff --git a/tests/test_us_legacy_spm.py b/tests/test_us_legacy_spm.py new file mode 100644 index 00000000..26329c02 --- /dev/null +++ b/tests/test_us_legacy_spm.py @@ -0,0 +1,565 @@ +"""Unit checks for backward compatibility with legacy US SPM sources. + +Covers PolicyEngine/policyengine.py#597 and #598 with synthetic tables only: +no country model and no data download. The country-model behaviour is in +``test_us_legacy_spm_integration`` and the model contract in +``test_us_legacy_spm_model_contract``. +""" + +from __future__ import annotations + +import json + +import h5py +import numpy as np +import pandas as pd +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +from policyengine.provenance.manifest import get_release_manifest +from policyengine.tax_benefit_models.us import datasets as us_datasets +from policyengine.tax_benefit_models.us.legacy_inputs import _ABSENT +from policyengine.tax_benefit_models.us.legacy_spm import ( + ADAPTER_REVISION, + LEGACY_SPM_SOURCES, + RECORD_H5_DATASET, + LegacySPMRecord, + LegacySPMSource, + expected_profile, + legacy_spm_source, + parse_record, + read_record_value, + record_problem, + write_record, + year_file_problem, +) +from policyengine.tax_benefit_models.us.legacy_spm.independence import ( + INDEPENDENCE_ROLE, + independence_roles, +) +from policyengine.tax_benefit_models.us.legacy_spm.registry import ( + POPULACE_US_2024_ACS_LOCAL_BUILD_O, +) +from policyengine.tax_benefit_models.us.legacy_spm.universe import ( + classify_acs_household_population, +) + +ACS = "acs_2024_1yr" +BUILD_O = POPULACE_US_2024_ACS_LOCAL_BUILD_O + + +def _tables(): + """One ACS household, two group quarters, and two non-ACS households. + + SPM units: 10 an ACS family whose reference person is 17, 20 an + institutional group-quarters adult, 30 a noninstitutional group-quarters + child, 40 an ASEC family, 50 a PUF teen living alone. + """ + person = pd.DataFrame( + { + "person_id": [1, 2, 3, 4, 5, 6, 7], + "person_household_id": [1, 1, 2, 3, 4, 4, 5], + "person_spm_unit_id": [10, 10, 20, 30, 40, 40, 50], + "age": [17, 9, 45, 16, 38, 6, 16], + "person_support_channel": [ + ACS, + ACS, + ACS, + ACS, + "asec", + "asec", + "puf_tax_detail", + ], + "RELSHIPP": [20.0, 25.0, 37.0, 38.0, np.nan, np.nan, np.nan], + "is_household_head": [True, False, False, False, True, False, False], + } + ) + household = pd.DataFrame( + {"household_id": [1, 2, 3, 4, 5], "TYPEHUGQ": [1.0, 2.0, 3.0, np.nan, np.nan]} + ) + return person, household + + +# --- Registry ------------------------------------------------------------- + + +def test_registry_keys_are_unique_and_match_the_bundle(): + hashes = [source.source_sha256 for source in LEGACY_SPM_SOURCES] + assert len(hashes) == len(set(hashes)) + manifest = get_release_manifest("us") + assert manifest.datasets["populace_us_2024_acs_local"].sha256 == ( + BUILD_O.source_sha256 + ) + pin = BUILD_O.independence_list + assert pin is not None + assert manifest.datasets[pin.bundle_dataset].sha256 == pin.sha256 + + +def test_only_registered_sources_are_legacy(): + assert legacy_spm_source(BUILD_O.source_sha256) == BUILD_O + assert legacy_spm_source(None) is None + assert legacy_spm_source("0" * 64) is None + default = get_release_manifest("us").datasets["populace_us_2024"] + assert legacy_spm_source(default.sha256) is None + + +def test_registry_entries_are_validated(): + with pytest.raises(ValueError): + LegacySPMSource(name="x", source_sha256="y", spm_universe_rule="other") + with pytest.raises(ValueError): + LegacySPMSource(name="x", source_sha256="y", unknown=True) + + +# --- Measurement scope (#597) ---------------------------------------------- + + +def test_acs_group_quarters_are_outside_and_everyone_else_included(): + person, household = _tables() + assert classify_acs_household_population(person, household).to_dict() == { + 10: "INCLUDED", + 20: "OUTSIDE", + 30: "OUTSIDE", + 40: "INCLUDED", + 50: "INCLUDED", + } + + +def test_classification_never_reads_age(): + person, household = _tables() + person["age"] = person["age"][::-1].to_numpy() + reversed_ages = classify_acs_household_population(person, household) + assert reversed_ages.to_dict() == ( + classify_acs_household_population(*_tables()).to_dict() + ) + + +def _set_person(row, column, value): + def change(person, household): + person.loc[row, column] = value + + return change + + +def _set_household(row, column, value): + def change(person, household): + household.loc[row, column] = value + + return change + + +def _drop(table_name, column): + def change(person, household): + table = person if table_name == "person" else household + table.drop(columns=[column], inplace=True) + + return change + + +@pytest.mark.parametrize( + "change, message", + [ + (_set_person(0, "person_support_channel", "acs_2023"), "no rule for source"), + (_set_person(0, "person_support_channel", None), "no rule for source"), + (_set_person(2, "RELSHIPP", np.nan), "cannot classify 1 people"), + (_set_person(2, "RELSHIPP", 38.0), "cannot classify 1 people"), + (_set_person(0, "RELSHIPP", 37.0), "cannot classify 1 people"), + (_set_household(0, "TYPEHUGQ", np.nan), "cannot classify 2 people"), + (_set_household(3, "TYPEHUGQ", 1.0), "cannot classify 2 people"), + (_set_person(4, "RELSHIPP", 20.0), "cannot classify 1 people"), + (_set_person(2, "person_spm_unit_id", 10), "mixing group-quarters"), + (_set_person(0, "person_household_id", 99), "unknown households"), + (_drop("person", "RELSHIPP"), "needs person 'RELSHIPP'"), + (_drop("household", "TYPEHUGQ"), "needs household 'TYPEHUGQ'"), + ], +) +def test_unrecognised_or_inconsistent_evidence_raises(change, message): + person, household = _tables() + change(person, household) + with pytest.raises(ValueError, match=message): + classify_acs_household_population(person, household) + + +#: Per-person evidence: (channel, household TYPEHUGQ, RELSHIPP). +_EVIDENCE = st.sampled_from( + [ + (ACS, 1.0, 20.0), + (ACS, 1.0, 25.0), + (ACS, 2.0, 37.0), + (ACS, 3.0, 38.0), + ("asec", np.nan, np.nan), + ("puf_tax_detail", np.nan, np.nan), + (ACS, 2.0, 38.0), + (ACS, 1.0, 37.0), + (ACS, np.nan, 20.0), + (ACS, 1.0, np.nan), + ("asec", 1.0, np.nan), + ("other", np.nan, np.nan), + ] +) + + +def _oracle_status(channel, record_type, relationship) -> str | None: + """The rule for one person, stated independently of the classifier.""" + if channel == ACS: + if record_type == 1 and relationship == relationship: + return None if relationship in (37, 38) else "INCLUDED" + if (record_type, relationship) in {(2, 37), (3, 38)}: + return "OUTSIDE" + return None + if channel in ("asec", "puf_tax_detail"): + if record_type != record_type and relationship != relationship: + return "INCLUDED" + return None + + +@settings(max_examples=300, deadline=None) +@given( + units=st.lists(st.lists(_EVIDENCE, min_size=1, max_size=3), min_size=1, max_size=6), + ages=st.lists(st.integers(0, 90), min_size=18, max_size=18), +) +def test_classification_agrees_with_the_rule_for_any_population(units, ages): + """Each SPM unit is its own household; ages are arbitrary and unused.""" + rows, households = [], [] + expected: dict[int, str] = {} + valid = True + for unit_index, members in enumerate(units): + household_id = unit_index + 1 + record_type = members[0][1] + households.append({"household_id": household_id, "TYPEHUGQ": record_type}) + statuses = set() + for channel, _, relationship in members: + rows.append( + { + "person_id": len(rows) + 1, + "person_household_id": household_id, + "person_spm_unit_id": household_id * 10, + "age": ages[len(rows) % len(ages)], + "person_support_channel": channel, + "RELSHIPP": relationship, + } + ) + statuses.add(_oracle_status(channel, record_type, relationship)) + if None in statuses or len(statuses) > 1: + valid = False + else: + expected[household_id * 10] = statuses.pop() + person, household = pd.DataFrame(rows), pd.DataFrame(households) + if valid: + assert classify_acs_household_population(person, household).to_dict() == ( + expected + ) + else: + with pytest.raises(ValueError): + classify_acs_household_population(person, household) + + +# --- Independence role (#598) ------------------------------------------------ + + +def _write_list(path, rows): + pd.DataFrame( + rows, columns=["person_id", "person_spm_unit_id", "is_spm_independence_role"] + ).to_csv(path, index=False) + return path + + +def _list_rows(): + return [(5, 40, True), (6, 40, False), (7, 50, True)] + + +def test_non_acs_people_take_the_list_and_acs_people_the_model_fallback(tmp_path): + person, _ = _tables() + roles = independence_roles(person, _write_list(tmp_path / "l.csv", _list_rows())) + assert roles.to_dict() == { + 1: True, + 2: False, + 3: False, + 4: False, + 5: True, + 6: False, + 7: True, + } + + +def test_a_stored_spouse_counts_for_acs_people(tmp_path): + person, _ = _tables() + person["is_household_spouse"] = [False, False, True, False, False, False, False] + roles = independence_roles(person, _write_list(tmp_path / "l.csv", _list_rows())) + assert roles[3] + + +@pytest.mark.parametrize( + "rows, message", + [ + (_list_rows()[:2], "1 missing, 0 not in the source"), + ([*_list_rows(), (8, 50, False)], "0 missing, 1 not in the source"), + ([*_list_rows(), (1, 10, False)], "0 missing, 1 not in the source"), + ([(5, 40, True), (6, 40, False), (7, 40, True)], "different SPM unit"), + ([*_list_rows(), (7, 50, True)], "repeats person IDs"), + ([(5, 40, True), (6, 40, "yes"), (7, 50, True)], "non-boolean"), + ([(5, 40, True), (6, 40, None), (7, 50, True)], "non-boolean"), + ], +) +def test_the_list_must_match_the_non_acs_people_exactly(tmp_path, rows, message): + person, _ = _tables() + with pytest.raises(ValueError, match=message): + independence_roles(person, _write_list(tmp_path / "l.csv", rows)) + + +def test_a_source_that_stores_the_role_is_not_legacy_for_it(tmp_path): + person, _ = _tables() + person[INDEPENDENCE_ROLE] = False + with pytest.raises(ValueError, match="not a legacy source"): + independence_roles(person, _write_list(tmp_path / "l.csv", _list_rows())) + + +def test_unexpected_list_columns_are_refused(tmp_path): + person, _ = _tables() + path = tmp_path / "l.csv" + pd.DataFrame({"person_id": [5], "role": [True]}).to_csv(path, index=False) + with pytest.raises(ValueError, match="Unexpected SPM independence list"): + independence_roles(person, path) + + +# --- Records: total and exact --------------------------------------------- + + +def _record(**source_changes) -> LegacySPMRecord: + source = BUILD_O.model_copy(update=source_changes) + return LegacySPMRecord( + profile=expected_profile(source), + outside_spm_units=2, + included_spm_units=3, + independence_list_people=3, + ) + + +def _empty_h5(path): + pd.DataFrame({"x": [1]}).to_hdf(path, key="person") + return path + + +def test_a_record_round_trips_through_h5(tmp_path): + path = _empty_h5(tmp_path / "file.h5") + assert read_record_value(path) is _ABSENT + write_record(path, _record()) + write_record(path, _record()) + assert parse_record(read_record_value(path)) == _record() + assert year_file_problem(path, expected_profile(BUILD_O)) is None + + +def test_the_profile_must_match_exactly(): + expected = expected_profile(BUILD_O) + assert record_problem(_record(), expected) is None + other_list = BUILD_O.independence_list.model_copy(update={"sha256": "0" * 64}) + for changed in ( + _record(independence_list=other_list), + _record(independence_list=None), + _record(spm_universe_rule=None), + _record(source_sha256="0" * 64), + ): + assert "derived under profile" in record_problem(changed, expected) + newer = _record().model_copy( + update={ + "profile": expected.model_copy( + update={"adapter_revision": ADAPTER_REVISION + 1} + ) + } + ) + assert "derived under profile" in record_problem(newer, expected) + + +def test_a_records_model_mode_is_provenance_not_profile(): + with_mode = _record().model_copy(update={"model_mode": "override"}) + assert record_problem(with_mode, expected_profile(BUILD_O)) is None + + +def test_records_are_required_exactly_when_the_source_needs_them(): + expected = expected_profile(BUILD_O) + assert record_problem(_ABSENT, None) is None + assert "no record" in record_problem(_ABSENT, expected) + assert "does not need" in record_problem(_record(), None) + + +#: Arbitrary JSON values a record node might hold. +_JSON = st.recursive( + st.none() | st.booleans() | st.integers() | st.floats(allow_nan=False) | st.text(), + lambda children: ( + st.lists(children, max_size=4) + | st.dictionaries(st.text(max_size=12), children, max_size=4) + ), + max_leaves=12, +) + + +@settings(max_examples=200, deadline=None) +@given( + stored=st.one_of( + _JSON, + st.builds( + lambda key, value: {**_record().model_dump(mode="json"), key: value}, + st.sampled_from(["schema_version", "profile", "extra", "model_mode"]), + _JSON, + ), + st.just(_record().model_dump(mode="json")), + ) +) +def test_the_check_is_total_and_accepts_only_the_expected_profile( + stored, tmp_path_factory +): + path = _empty_h5(tmp_path_factory.mktemp("record") / "file.h5") + with h5py.File(path, "a") as file: + file.create_dataset( + RECORD_H5_DATASET, + data=json.dumps(stored), + dtype=h5py.string_dtype("utf-8"), + ) + expected = expected_profile(BUILD_O) + problem = year_file_problem(path, expected) + record = parse_record(stored) + assert (problem is None) == (record is not None and record.profile == expected) + + +@pytest.mark.parametrize( + "stored", + [ + np.void(b"\xff\xfe not utf-8"), + "{not json", + "[" * 100_000 + "]" * 100_000, + np.arange(3), + ], + # Explicit IDs: pytest would otherwise use the 200,000-character string as + # its test ID, which exceeds Windows' environment-variable limit for + # PYTEST_CURRENT_TEST and floods verbose CI logs. + ids=["bytes_not_utf8", "text_not_json", "json_nested_too_deeply", "number_array"], +) +def test_unreadable_records_are_out_of_date_not_errors(tmp_path, stored): + path = _empty_h5(tmp_path / "file.h5") + with h5py.File(path, "a") as file: + if isinstance(stored, str): + file.create_dataset( + RECORD_H5_DATASET, data=stored, dtype=h5py.string_dtype("utf-8") + ) + else: + file.create_dataset(RECORD_H5_DATASET, data=stored) + assert "unreadable" in year_file_problem(path, expected_profile(BUILD_O)) + + +def test_a_file_that_is_not_hdf5_is_out_of_date(tmp_path): + path = tmp_path / "file.h5" + path.write_bytes(b"not hdf5") + assert year_file_problem(path, expected_profile(BUILD_O)) is not None + + +# --- Year-file decisions --------------------------------------------------- + + +def test_only_legacy_year_files_must_record_declarations(tmp_path): + path = _empty_h5(tmp_path / "year.h5") + acs_local = us_datasets._expected_legacy_spm_profile("populace_us_2024_acs_local") + assert acs_local == expected_profile(BUILD_O) + assert us_datasets._expected_legacy_spm_profile("populace_us_2024") is None + assert us_datasets._expected_legacy_spm_profile("hf://other/repo/x.h5") is None + assert year_file_problem(path, acs_local) is not None + write_record(path, _record()) + assert year_file_problem(path, acs_local) is None + assert year_file_problem(path, None) is not None + + +def test_a_managed_dataset_is_recognised_by_its_uri_too(): + from policyengine.provenance.manifest import resolve_dataset_reference + + uri = resolve_dataset_reference("us", "populace_us_2024_acs_local") + assert uri.startswith("hf://") + assert us_datasets._expected_legacy_spm_profile(uri) == expected_profile(BUILD_O) + default_uri = resolve_dataset_reference("us", "populace_us_2024") + assert us_datasets._expected_legacy_spm_profile(default_uri) is None + + +# --- Year-file frame guards ------------------------------------------------- + + +def _declared_frames(): + from policyengine.tax_benefit_models.us.legacy_spm import UNIVERSE_VARIABLE + + person = pd.DataFrame({"person_id": [1, 2], INDEPENDENCE_ROLE: [True, False]}) + spm_unit = pd.DataFrame( + {"spm_unit_id": [1, 2], UNIVERSE_VARIABLE: ["INCLUDED", "OUTSIDE"]} + ) + return person, spm_unit + + +def test_complete_declarations_pass_the_frame_check(): + from policyengine.tax_benefit_models.us.legacy_spm import check_year_file_frames + + person, spm_unit = _declared_frames() + check_year_file_frames(person, spm_unit, _record()) + spm_unit["spm_unit_spm_universe_status"] = [b"INCLUDED", b"OUTSIDE"] + check_year_file_frames(person, spm_unit, _record()) + + +def _drop_column(table, column): + def change(person, spm_unit): + (person if table == "person" else spm_unit).drop(columns=[column], inplace=True) + + return change + + +def _set_column(table, column, values): + def change(person, spm_unit): + (person if table == "person" else spm_unit)[column] = values + + return change + + +@pytest.mark.parametrize( + "change, message", + [ + ( + _drop_column("spm_unit", "spm_unit_spm_universe_status"), + "stores no 'spm_unit_spm_universe_status'", + ), + ( + _set_column( + "spm_unit", "spm_unit_spm_universe_status", ["INCLUDED", "UNRESOLVED"] + ), + "must declare every SPM unit", + ), + ( + _set_column("spm_unit", "spm_unit_spm_universe_status", ["INCLUDED", None]), + "must declare every SPM unit", + ), + ( + _drop_column("person", INDEPENDENCE_ROLE), + "stores no 'is_spm_independent_minor_role'", + ), + ( + _set_column("person", INDEPENDENCE_ROLE, [True, None]), + "must be complete and boolean", + ), + ( + _set_column("person", INDEPENDENCE_ROLE, [1, 0]), + "must be complete and boolean", + ), + ], +) +def test_incomplete_or_invalid_declarations_fail_the_frame_check(change, message): + from policyengine.tax_benefit_models.us.legacy_spm import check_year_file_frames + + person, spm_unit = _declared_frames() + change(person, spm_unit) + with pytest.raises(ValueError, match=message): + check_year_file_frames(person, spm_unit, _record()) + + +def test_only_the_declarations_a_source_needs_are_required(): + from policyengine.tax_benefit_models.us.legacy_spm import check_year_file_frames + + person, spm_unit = _declared_frames() + person.drop(columns=[INDEPENDENCE_ROLE], inplace=True) + check_year_file_frames(person, spm_unit, _record(independence_list=None)) + spm_unit.drop(columns=["spm_unit_spm_universe_status"], inplace=True) + check_year_file_frames( + person, spm_unit, _record(independence_list=None, spm_universe_rule=None) + ) diff --git a/tests/test_us_legacy_spm_integration.py b/tests/test_us_legacy_spm_integration.py new file mode 100644 index 00000000..45bdc894 --- /dev/null +++ b/tests/test_us_legacy_spm_integration.py @@ -0,0 +1,772 @@ +"""Backward compatibility with legacy US SPM sources, in real model runs. + +#597 keeps ACS group-quarters units out of the SPM measurement; #598 supplies +the certified SPM independence role for ASEC/PUF records. The tests use tiny +synthetic households in California: + +- SPM unit 1: an ACS family (adult head, child), receiving housing assistance. +- SPM unit 2: an ACS institutional group-quarters child aged 16. +- SPM unit 3: an ACS noninstitutional group-quarters adult aged 70, receiving + housing assistance. +- SPM unit 4: an ASEC adult head. +- SPM unit 5: an ASEC teen aged 16, alone in an SPM unit inside household 4, + whom the certified independence list marks independent. + +Without the declarations, units 2 and 5 have no classified adult and stop +every SPM calculation. Synthetic files are served under the registered legacy +source's identity by patching the materialiser, so nothing is downloaded. +""" + +from __future__ import annotations + +import importlib +from pathlib import Path + +import h5py +import numpy as np +import pandas as pd +import pytest +from microdf import MicroDataFrame + +pytest.importorskip("policyengine_us") +pytest.importorskip("spm_calculator.policyengine_adapter") + +import policyengine as pe # noqa: E402 +from policyengine.core.run_record import ( # noqa: E402 + build_simulation_run_record_payloads, +) +from policyengine.core.simulation import Simulation, _cache # noqa: E402 +from policyengine.outputs.poverty import ( # noqa: E402 + Poverty, + calculate_us_poverty_by_age, + calculate_us_poverty_rates, +) +from policyengine.provenance.dataset_materialization import ( # noqa: E402 + DatasetSource, + MaterializedDataset, +) +from policyengine.tax_benefit_models.us import datasets as us_datasets # noqa: E402 +from policyengine.tax_benefit_models.us.datasets import ( # noqa: E402 + PolicyEngineUSDataset, + USYearData, + create_datasets, + ensure_datasets, + load_datasets, +) +from policyengine.tax_benefit_models.us.legacy_spm import ( # noqa: E402 + INDEPENDENCE_ROLE, + RECORD_H5_DATASET, + RECORD_KEY, + UNIVERSE_VARIABLE, + LegacySPMRecord, + declarations, + expected_profile, + record, + runs, +) +from policyengine.tax_benefit_models.us.legacy_spm.model_adapter import ( # noqa: E402 + FLOAT_INDICATORS, + SPM_MEASUREMENT_VARIABLES, +) +from policyengine.tax_benefit_models.us.legacy_spm.registry import ( # noqa: E402 + POPULACE_US_2024_ACS_LOCAL_BUILD_O, +) +from policyengine.tax_benefit_models.us.model import PolicyEngineUSLatest # noqa: E402 + +# The ``us`` package's ``model`` attribute is the model version, not the module. +us_model = importlib.import_module("policyengine.tax_benefit_models.us.model") + +BUILD_O = POPULACE_US_2024_ACS_LOCAL_BUILD_O +LIST_PIN = BUILD_O.independence_list +YEAR = 2026 +ACS = "acs_2024_1yr" +OUTSIDE_UNITS = [2, 3] +MEASURED_UNITS = [1, 4, 5] +GROUP_QUARTERS_COUNTIES = {"06075", "06001"} +REFORM = {"gov.irs.credits.ctc.amount.base[0].amount": 3_000} +SPM_UNIT_OUTPUTS = [ + name for name in SPM_MEASUREMENT_VARIABLES if name != "person_in_poverty" +] +SPM_EXTRAS = {"spm_unit": SPM_UNIT_OUTPUTS, "person": ["person_in_poverty"]} +ORDINARY = { + "household": [ + "household_net_income", + "household_benefits", + "household_tax", + "household_market_income", + ], + "tax_unit": [ + "income_tax", + "employee_payroll_tax", + "state_income_tax", + "eitc", + "ctc", + ], + "spm_unit": ["snap", "tanf"], +} + + +def _groups(count: int, key: str) -> pd.DataFrame: + return pd.DataFrame( + {f"{key}_id": range(1, count + 1), f"{key}_weight": [1.0] * count} + ) + + +def _source_frames() -> dict[str, pd.DataFrame]: + """Tables in the legacy source's layout, with its record-type evidence.""" + person = pd.DataFrame( + { + "person_id": [1, 2, 3, 4, 5, 6], + "person_household_id": [1, 1, 2, 3, 4, 4], + "person_spm_unit_id": [1, 1, 2, 3, 4, 5], + "person_tax_unit_id": [1, 1, 2, 3, 4, 5], + "person_family_id": [1, 1, 2, 3, 4, 5], + "person_marital_unit_id": [1, 2, 3, 4, 5, 6], + "age": [40, 8, 16, 70, 30, 16], + "employment_income": [30_000.0, 0, 0, 0, 50_000.0, 4_000.0], + "is_household_head": [True, False, False, False, True, False], + "person_support_channel": [ACS, ACS, ACS, ACS, "asec", "asec"], + "RELSHIPP": [20.0, 25.0, 37.0, 38.0, np.nan, np.nan], + } + ) + household = _groups(4, "household") + household["state_code"] = ["CA"] * 4 + household["county_fips"] = ["06037", "06075", "06001", "06037"] + household["TYPEHUGQ"] = [1.0, 2.0, 3.0, np.nan] + spm_unit = _groups(5, "spm_unit") + spm_unit["housing_assistance"] = [6_000.0, 0.0, 5_000.0, 0.0, 0.0] + return { + "person": person, + "household": household, + "spm_unit": spm_unit, + "tax_unit": _groups(5, "tax_unit"), + "family": _groups(5, "family"), + "marital_unit": _groups(6, "marital_unit"), + } + + +def _record() -> LegacySPMRecord: + return LegacySPMRecord( + profile=expected_profile(BUILD_O), + outside_spm_units=2, + included_spm_units=3, + independence_list_people=2, + ) + + +def _dataset(directory, *, declared=True, drop_units=(), teen_independent=True): + """A year-file-like dataset; ``declared`` adds the stored declarations.""" + frames = {name: frame.copy() for name, frame in _source_frames().items()} + person = frames["person"] + person["person_weight"] = 1.0 + for entity in ("household", "spm_unit", "tax_unit", "family", "marital_unit"): + person[f"{entity}_id"] = person.pop(f"person_{entity}_id") + if declared: + frames["spm_unit"][UNIVERSE_VARIABLE] = [ + "INCLUDED", + "OUTSIDE", + "OUTSIDE", + "INCLUDED", + "INCLUDED", + ] + person[INDEPENDENCE_ROLE] = [True, False, False, False, True, teen_independent] + if drop_units: + frames["person"] = person = person[ + ~person["spm_unit_id"].isin(drop_units) + ].reset_index(drop=True) + for entity in ("household", "spm_unit", "tax_unit", "family", "marital_unit"): + kept = frames[entity][f"{entity}_id"].isin(set(person[f"{entity}_id"])) + frames[entity] = frames[entity][kept].reset_index(drop=True) + person.drop(columns=["RELSHIPP", "person_support_channel"], inplace=True) + frames["household"].drop(columns="TYPEHUGQ", inplace=True) + return PolicyEngineUSDataset( + name="legacy-spm-source", + description="Uncertified synthetic legacy-source households", + filepath=str(directory / "input.h5"), + year=YEAR, + metadata={RECORD_KEY: _record().model_dump(mode="json")} if declared else {}, + data=USYearData( + **{ + entity: MicroDataFrame(frame, weights=f"{entity}_weight") + for entity, frame in frames.items() + } + ), + ) + + +def _frame(simulation, entity) -> pd.DataFrame: + return pd.DataFrame(getattr(simulation.output_dataset.data, entity)).set_index( + f"{entity}_id" + ) + + +@pytest.fixture(scope="module") +def provider_calls(): + return [] + + +@pytest.fixture(scope="module") +def declared_run(tmp_path_factory, provider_calls): + """A reform run over the declared households, recording provider calls.""" + from policyengine_us.spm import CountyRequiringSPMProvider + + original = CountyRequiringSPMProvider._amounts + + def spy(self, year, adults, children, tenure, county): + provider_calls.append((adults, children, county)) + return original(self, year, adults, children, tenure, county) + + with pytest.MonkeyPatch.context() as patch: + patch.setattr(CountyRequiringSPMProvider, "_amounts", spy) + simulation = Simulation( + dataset=_dataset(tmp_path_factory.mktemp("declared")), + tax_benefit_model_version=pe.us.model, + policy=REFORM, + extra_variables=SPM_EXTRAS, + ) + simulation.run() + return simulation + + +@pytest.fixture(scope="module") +def measured_reference(tmp_path_factory): + """The unmodified model on the households it can measure. + + Group-quarters unit 3 has an adult, so the unmodified model measures it; + group-quarters child unit 2 cannot be measured and is dropped. A unit's SPM + threshold depends only on its own members, tenure and county. + """ + dataset = _dataset( + tmp_path_factory.mktemp("reference"), declared=False, drop_units=(2,) + ) + # The same certified role, so the independent teen's unit (5) is measured. + dataset.data.person[INDEPENDENCE_ROLE] = [True, False, False, True, True] + simulation = Simulation( + dataset=dataset, + tax_benefit_model_version=pe.us.model, + policy=REFORM, + extra_variables=SPM_EXTRAS, + ) + simulation.run() + return simulation + + +@pytest.fixture(scope="module") +def ordinary_reference(tmp_path_factory): + """The unmodified model's ordinary outputs on the complete population.""" + original = PolicyEngineUSLatest.resolve_entity_variables + + def ordinary_only(self, simulation): + return { + entity: [name for name in names if name not in SPM_MEASUREMENT_VARIABLES] + for entity, names in original(self, simulation).items() + } + + with pytest.MonkeyPatch.context() as patch: + patch.setattr(PolicyEngineUSLatest, "resolve_entity_variables", ordinary_only) + simulation = Simulation( + dataset=_dataset(tmp_path_factory.mktemp("ordinary"), declared=False), + tax_benefit_model_version=pe.us.model, + policy=REFORM, + ) + simulation.run() + return simulation + + +def test_the_unmodified_model_cannot_measure_the_population(tmp_path): + from spm_calculator.errors import SPMInputError + + simulation = Simulation( + dataset=_dataset(tmp_path, declared=False), + tax_benefit_model_version=pe.us.model, + ) + with pytest.raises(SPMInputError) as error: + simulation.run() + assert error.value.code == "SPM_COMPOSITION_REQUIRED" + + +def test_outside_units_have_no_spm_values(declared_run): + spm_unit = _frame(declared_run, "spm_unit") + for name in SPM_UNIT_OUTPUTS: + assert spm_unit.loc[OUTSIDE_UNITS, name].isna().all(), name + assert spm_unit.loc[MEASURED_UNITS, name].notna().all(), name + person = _frame(declared_run, "person") + assert person["person_in_poverty"].isna().tolist() == [ + False, + False, + True, + True, + False, + False, + ] + for name in FLOAT_INDICATORS - {"person_in_poverty"}: + assert spm_unit[name].dtype.kind == "f" + recorded = LegacySPMRecord.model_validate( + declared_run.output_dataset.metadata[RECORD_KEY] + ) + assert recorded.model_mode == "override" + assert recorded.profile == expected_profile(BUILD_O) + + +def test_outside_units_never_reach_the_spm_provider(declared_run, provider_calls): + assert provider_calls + assert not {county for _, _, county in provider_calls} & GROUP_QUARTERS_COUNTIES + assert all(adults >= 1 for adults, _, _ in provider_calls) + + +def test_measured_units_keep_the_canonical_spm_values(declared_run, measured_reference): + declared = _frame(declared_run, "spm_unit") + reference = _frame(measured_reference, "spm_unit") + for name in SPM_UNIT_OUTPUTS: + if name == "spm_unit_income_decile": + continue # ranks depend on which units are ranked + np.testing.assert_array_equal( + declared.loc[MEASURED_UNITS, name].to_numpy(dtype=float), + reference.loc[MEASURED_UNITS, name].to_numpy(dtype=float), + err_msg=name, + ) + # The measured family's subsidy is capped through the installed formula; + # the assisted group-quarters adult is not measured. + assert declared.loc[1, "spm_unit_capped_housing_subsidy"] > 0 + assert reference.loc[3, "spm_unit_capped_housing_subsidy"] > 0 + assert np.isnan(declared.loc[3, "spm_unit_capped_housing_subsidy"]) + + +def test_ordinary_calculations_are_unchanged_for_everyone( + declared_run, ordinary_reference +): + for entity, names in ORDINARY.items(): + declared = _frame(declared_run, entity) + reference = _frame(ordinary_reference, entity) + assert declared.index.equals(reference.index) + for name in names: + np.testing.assert_array_equal( + declared[name].to_numpy(), reference[name].to_numpy(), err_msg=name + ) + assert np.isfinite(declared[name].to_numpy(dtype=float)).all(), name + + +def test_poverty_rates_cover_measured_people_only(declared_run): + rates = calculate_us_poverty_rates(declared_run).dataframe + # People 3 and 4 are outside; everyone has weight 1. + assert rates["total_population"].tolist() == [4.0, 4.0] + by_age = calculate_us_poverty_by_age(declared_run).dataframe + child = by_age[by_age["filter_group"] == "child"].iloc[0] + # Child 2 and teen 6 are measured; group-quarters teen 3 is not. + assert child["total_population"] == 2.0 + + +def test_a_group_with_nobody_measured_has_no_rate(declared_run): + seniors = Poverty( + simulation=declared_run, + poverty_variable="spm_unit_is_in_spm_poverty", + entity="person", + filter_variable="age", + filter_variable_geq=65, + ) + seniors.run() + # The only senior lives in group quarters, outside the measurement. + assert seniors.total_population == 0.0 + assert seniors.rate is None + + +def test_poverty_flags_are_not_summed_onto_groups(declared_run): + poverty = Poverty( + simulation=declared_run, + poverty_variable="spm_unit_is_in_spm_poverty", + entity="household", + ) + with pytest.raises(ValueError, match="summarise it for people"): + poverty.run() + + +def test_the_declarations_reach_the_baseline_and_stay_private(tmp_path, monkeypatch): + from policyengine_us.system import system + + systems = [] + original = PolicyEngineUSLatest._build_simulation_from_dataset + + def capture(self, country_simulation, dataset, tax_benefit_system): + systems.append(tax_benefit_system) + return original(self, country_simulation, dataset, tax_benefit_system) + + monkeypatch.setattr(PolicyEngineUSLatest, "_build_simulation_from_dataset", capture) + Simulation( + dataset=_dataset(tmp_path), tax_benefit_model_version=pe.us.model, policy=REFORM + ).run() + assert len(systems) == 2 + assert all(UNIVERSE_VARIABLE in policy.variables for policy in systems) + + systems.clear() + undeclared = _dataset(tmp_path, declared=False, drop_units=(2,)) + undeclared.data.person[INDEPENDENCE_ROLE] = [True, False, False, True, True] + unaffected = Simulation(dataset=undeclared, tax_benefit_model_version=pe.us.model) + unaffected.run() + assert UNIVERSE_VARIABLE not in systems[0].variables + assert UNIVERSE_VARIABLE not in system.variables + assert system.variables["spm_unit_is_in_spm_poverty"].value_type is bool + assert _frame(unaffected, "spm_unit")["spm_unit_is_in_spm_poverty"].dtype == bool + assert RECORD_KEY not in unaffected.output_dataset.metadata + + +@pytest.mark.parametrize("abolish_hud", [False, True]) +def test_structural_resource_reforms_cannot_bypass_the_declarations( + tmp_path, abolish_hud +): + simulation = Simulation( + dataset=_dataset(tmp_path), + tax_benefit_model_version=pe.us.model, + policy={ + "gov.contrib.congress.tlaib.end_child_poverty_act.in_effect": True, + "gov.hud.abolition": abolish_hud, + }, + extra_variables={"spm_unit": ["spm_unit_benefits"]}, + ) + simulation.run() + spm_unit = _frame(simulation, "spm_unit") + for name in ( + "spm_unit_benefits", + "spm_unit_net_income", + "spm_unit_is_in_spm_poverty", + ): + assert spm_unit.loc[OUTSIDE_UNITS, name].isna().all(), name + assert spm_unit.loc[MEASURED_UNITS, name].notna().all(), name + + +def test_a_measured_unit_without_an_adult_still_raises(tmp_path): + from spm_calculator.errors import SPMInputError + + simulation = Simulation( + dataset=_dataset(tmp_path, teen_independent=False), + tax_benefit_model_version=pe.us.model, + ) + with pytest.raises(SPMInputError) as error: + simulation.run() + assert error.value.code == "SPM_COMPOSITION_REQUIRED" + + +def test_regional_filtering_keeps_the_declarations(tmp_path): + from policyengine.core.scoping_strategy import RowFilterStrategy + + simulation = Simulation( + dataset=_dataset(tmp_path), + tax_benefit_model_version=pe.us.model, + scoping_strategy=RowFilterStrategy( + variable_name="county_fips", variable_value=["06037", "06075"] + ), + ) + simulation.run() + spm_unit = _frame(simulation, "spm_unit") + assert spm_unit.index.tolist() == [1, 2, 4, 5] + assert spm_unit["spm_unit_net_income"].isna().tolist() == [ + False, + True, + False, + False, + ] + + +def test_a_declared_dataset_must_be_current_and_complete(tmp_path): + stale = _dataset(tmp_path) + stale.metadata[RECORD_KEY]["profile"]["adapter_revision"] = 0 + with pytest.raises(ValueError, match="out of date"): + Simulation(dataset=stale, tax_benefit_model_version=pe.us.model).run() + + unreadable = _dataset(tmp_path) + unreadable.metadata[RECORD_KEY] = None + with pytest.raises(ValueError, match="unreadable"): + Simulation(dataset=unreadable, tax_benefit_model_version=pe.us.model).run() + + missing_column = _dataset(tmp_path) + missing_column.data.spm_unit = MicroDataFrame( + pd.DataFrame(missing_column.data.spm_unit).drop(columns=[UNIVERSE_VARIABLE]), + weights="spm_unit_weight", + ) + with pytest.raises(ValueError, match="stores no"): + Simulation(dataset=missing_column, tax_benefit_model_version=pe.us.model).run() + + +def test_saved_outputs_are_reused_only_under_the_same_declarations(tmp_path): + _cache.clear() + dataset = _dataset(tmp_path) + # A run record binds the input file's bytes, so write it first. + dataset.save() + simulation = Simulation( + id="legacy-spm", dataset=dataset, tax_benefit_model_version=pe.us.model + ) + simulation.ensure() + with h5py.File(simulation.output_dataset.filepath, "r") as file: + assert RECORD_H5_DATASET in file + recorded = simulation.release_bundle[RECORD_KEY] + assert recorded["profile"] == _record().model_dump(mode="json")["profile"] + payloads = build_simulation_run_record_payloads(simulation) + assert payloads["results"][RECORD_KEY] == recorded + + _cache.clear() + restored = Simulation( + id=simulation.id, dataset=dataset, tax_benefit_model_version=pe.us.model + ) + restored.load() + assert restored.output_dataset.metadata[RECORD_KEY] == recorded + + undeclared = Simulation( + id=simulation.id, + dataset=_dataset(tmp_path, declared=False), + tax_benefit_model_version=pe.us.model, + ) + with pytest.raises(ValueError, match="does not need"): + undeclared.load() + + +# --- Legacy source files: year files, raw runs, managed simulations ------- + + +def _write_source(path) -> str: + with pd.HDFStore(path, mode="w") as store: + for entity, frame in _source_frames().items(): + store.put(entity, frame, format="table", data_columns=True) + store.put("_time_period", pd.Series([2024]), format="table") + return str(path) + + +def _write_list(path) -> Path: + pd.DataFrame( + { + "person_id": [5, 6], + "person_spm_unit_id": [4, 5], + "is_spm_independence_role": [True, True], + } + ).to_csv(path, index=False) + return path + + +def _bundle_source(path, sha256) -> DatasetSource: + uri = f"hf://policyengine/populace-us/{Path(path).name}" + return DatasetSource( + source_uri=uri, + path=str(path), + bundle_dataset=MaterializedDataset( + data_package_name="microcosm-data", + repo_type="dataset", + revision="synthetic", + source_uri=uri, + sha256=sha256, + path=path, + ), + ) + + +@pytest.fixture +def legacy_files(tmp_path, monkeypatch): + """Serve the synthetic source and list as the registered certified files.""" + source = _write_source(tmp_path / "populace_us_2024_acs_local.h5") + independence_list = _write_list(tmp_path / "source_spm_person_independence.csv") + requests = [] + + def materialize(country, dataset=None, **kwargs): + requests.append(dataset) + if dataset == LIST_PIN.bundle_dataset: + return _bundle_source(independence_list, LIST_PIN.sha256) + return _bundle_source(source, BUILD_O.source_sha256) + + for module in (us_datasets, us_model, declarations): + monkeypatch.setattr(module, "materialize_dataset", materialize) + return SimpleSources(source=source, requests=requests) + + +class SimpleSources: + def __init__(self, source, requests): + self.source = source + self.requests = requests + + +def test_year_files_carry_the_declarations_for_every_year(tmp_path, legacy_files): + data_folder = str(tmp_path / "data") + years = [2025, 2026, 2027] + created = create_datasets( + datasets=["populace_us_2024_acs_local"], years=years, data_folder=data_folder + ) + assert legacy_files.requests == [ + "populace_us_2024_acs_local", + LIST_PIN.bundle_dataset, + ] + for year in years: + dataset = created[f"populace_us_2024_acs_local_{year}"] + spm_unit = pd.DataFrame(dataset.data.spm_unit).set_index("spm_unit_id") + person = pd.DataFrame(dataset.data.person).set_index("person_id") + assert spm_unit[UNIVERSE_VARIABLE].tolist() == [ + "INCLUDED", + "OUTSIDE", + "OUTSIDE", + "INCLUDED", + "INCLUDED", + ] + # ACS: household head, as the model computes; ASEC: the list. + assert person[INDEPENDENCE_ROLE].tolist() == [ + True, + False, + False, + False, + True, + True, + ] + assert LegacySPMRecord.model_validate(dataset.metadata[RECORD_KEY]) == _record() + + (loaded,) = load_datasets( + datasets=["populace_us_2024_acs_local"], years=[YEAR], data_folder=data_folder + ).values() + simulation = Simulation(dataset=loaded, tax_benefit_model_version=pe.us.model) + simulation.run() + spm_unit = _frame(simulation, "spm_unit") + assert spm_unit.loc[OUTSIDE_UNITS, "spm_unit_net_income"].isna().all() + assert spm_unit.loc[MEASURED_UNITS, "spm_unit_net_income"].notna().all() + + +@pytest.mark.parametrize("corruption", ["delete", "garbage", "revision"]) +def test_out_of_date_year_files_are_cut_again( + tmp_path, legacy_files, monkeypatch, corruption +): + data_folder = tmp_path / "data" + create_datasets( + datasets=["populace_us_2024_acs_local"], + years=[YEAR], + data_folder=str(data_folder), + ) + path = data_folder / f"populace_us_2024_acs_local_year_{YEAR}.h5" + if corruption == "revision": + monkeypatch.setattr(record, "ADAPTER_REVISION", record.ADAPTER_REVISION + 1) + else: + with h5py.File(path, "a") as file: + del file[RECORD_H5_DATASET] + if corruption == "garbage": + file.create_dataset( + RECORD_H5_DATASET, + data="{not json", + dtype=h5py.string_dtype("utf-8"), + ) + with pytest.raises(ValueError, match="out of date"): + load_datasets( + datasets=["populace_us_2024_acs_local"], + years=[YEAR], + data_folder=str(data_folder), + ) + legacy_files.requests.clear() + (regenerated,) = ensure_datasets( + datasets=["populace_us_2024_acs_local"], + years=[YEAR], + data_folder=str(data_folder), + ).values() + assert legacy_files.requests == [ + "populace_us_2024_acs_local", + LIST_PIN.bundle_dataset, + ] + regenerated_record = LegacySPMRecord.model_validate( + regenerated.metadata[RECORD_KEY] + ) + assert regenerated_record.profile.adapter_revision == record.ADAPTER_REVISION + + +def test_a_run_straight_over_a_legacy_source_derives_the_declarations( + tmp_path, legacy_files, monkeypatch +): + source_hash = runs.sha256_file(Path(legacy_files.source)) + monkeypatch.setattr( + runs, + "legacy_spm_source", + lambda sha256: BUILD_O if sha256 == source_hash else None, + ) + raw = PolicyEngineUSDataset( + name="raw-legacy-source", + description="The synthetic legacy source file itself", + filepath=legacy_files.source, + year=2024, + ) + _cache.clear() + simulation = Simulation( + id="raw-legacy", + dataset=raw, + tax_benefit_model_version=pe.us.model, + policy=REFORM, + ) + simulation.ensure() + spm_unit = _frame(simulation, "spm_unit") + assert spm_unit.loc[OUTSIDE_UNITS, "spm_unit_net_income"].isna().all() + assert spm_unit.loc[MEASURED_UNITS, "spm_unit_net_income"].notna().all() + assert ( + LegacySPMRecord.model_validate( + simulation.output_dataset.metadata[RECORD_KEY] + ).independence_list_people + == 2 + ) + + _cache.clear() + restored = Simulation( + id=simulation.id, + dataset=raw, + tax_benefit_model_version=pe.us.model, + policy=REFORM, + ) + restored.load() + assert restored.output_dataset.metadata[RECORD_KEY]["model_mode"] == "override" + + +def test_a_managed_simulation_of_a_legacy_source_applies_the_declarations( + legacy_files, +): + simulation = pe.us.managed_microsimulation(dataset="populace_us_2024_acs_local") + thresholds = simulation.calculate("spm_unit_spm_threshold", YEAR).values + assert np.isnan(thresholds[[1, 2]]).all() + assert np.isfinite(thresholds[[0, 3, 4]]).all() + in_poverty = simulation.calculate("spm_unit_is_in_spm_poverty", YEAR).values + assert np.isnan(in_poverty[[1, 2]]).all() + recorded = LegacySPMRecord.model_validate( + simulation.policyengine_bundle[RECORD_KEY] + ) + assert recorded.model_mode == "override" + assert recorded.outside_spm_units == 2 + + +def test_year_files_load_by_the_sources_uri_as_well_as_its_name(tmp_path, legacy_files): + """The region registry names the source by URI; that form must work too.""" + from policyengine.provenance.manifest import resolve_dataset_reference + + data_folder = str(tmp_path / "data") + create_datasets( + datasets=["populace_us_2024_acs_local"], years=[YEAR], data_folder=data_folder + ) + uri = resolve_dataset_reference("us", "populace_us_2024_acs_local") + (loaded,) = load_datasets( + datasets=[uri], years=[YEAR], data_folder=data_folder + ).values() + assert LegacySPMRecord.model_validate(loaded.metadata[RECORD_KEY]) == _record() + + legacy_files.requests.clear() + (ensured,) = ensure_datasets( + datasets=[uri], years=[YEAR], data_folder=data_folder + ).values() + assert legacy_files.requests == [] # current, so nothing is cut again + assert ensured.metadata[RECORD_KEY] == loaded.metadata[RECORD_KEY] + + +def test_a_year_without_a_declaration_is_refused_not_guessed(legacy_files): + """The source stores 2024 onwards; an earlier year has no declaration.""" + from spm_calculator.errors import SPMInputError + + simulation = pe.us.managed_microsimulation(dataset="populace_us_2024_acs_local") + with pytest.raises(SPMInputError) as error: + simulation.calculate("spm_unit_spm_threshold", 2023) + assert error.value.code == "SPM_UNIVERSE_REQUIRED" + + +def test_other_sources_get_no_declarations(tmp_path): + source = _write_source(tmp_path / "other.h5") + created = create_datasets( + datasets=[source], + years=[YEAR], + data_folder=str(tmp_path / "data"), + allow_unmanaged=True, + ) + (dataset,) = created.values() + assert RECORD_KEY not in dataset.metadata + assert UNIVERSE_VARIABLE not in pd.DataFrame(dataset.data.spm_unit).columns + assert INDEPENDENCE_ROLE not in pd.DataFrame(dataset.data.person).columns diff --git a/tests/test_us_legacy_spm_model_contract.py b/tests/test_us_legacy_spm_model_contract.py new file mode 100644 index 00000000..7db52051 --- /dev/null +++ b/tests/test_us_legacy_spm_model_contract.py @@ -0,0 +1,243 @@ +"""Keep the legacy SPM model override aligned with the installed model. + +The override in ``legacy_spm.model_adapter`` must cover every policyengine-us +variable that depends on the canonical SPM provider; otherwise an upgrade +could add a consumer that sends units outside the measurement to the +provider. These checks rebuild that dependency set from the installed model +and from its formula sources, so a pin bump that changes the SPM chain fails +here first. +""" + +from __future__ import annotations + +import ast +import inspect +import textwrap +from types import SimpleNamespace + +import pytest + +pytest.importorskip("policyengine_us") +pytest.importorskip("spm_calculator.policyengine_adapter") + +from policyengine.tax_benefit_models.us.legacy_spm import ( # noqa: E402 + model_adapter, + prepare_microsimulation, +) +from policyengine.tax_benefit_models.us.legacy_spm.model_adapter import ( # noqa: E402 + ADAPTER_MARKER, + MASKED_FORMULA_VARIABLES, + NAN_PROPAGATING_VARIABLES, + SPM_MEASUREMENT_VARIABLES, + THRESHOLD_FIELDS, + UNIVERSE_VARIABLE, + _check_spm_chain, +) + +#: Names whose call in a formula reaches the canonical SPM provider. +_PROVIDER_CALLS = {"policyengine_amount", "masked_policyengine_amount"} + + +def _formula_references(variable) -> tuple[set[str], set[str]]: + """Return the string literals and called names in a variable's formulas.""" + strings: set[str] = set() + calls: set[str] = set() + for formula in variable.formulas.values(): + try: + source = textwrap.dedent(inspect.getsource(formula)) + except (OSError, TypeError): + continue + for node in ast.walk(ast.parse(source)): + if isinstance(node, ast.Constant) and isinstance(node.value, str): + strings.add(node.value) + elif isinstance(node, ast.Call): + function = node.func + name = getattr(function, "id", None) or getattr(function, "attr", None) + if name: + calls.add(name) + return strings, calls + + +def _component_names(variable, system) -> set[str]: + names: set[str] = set() + for components in (variable.adds, variable.subtracts): + if isinstance(components, str): + parameter = system.parameters.get_child(components) + components = parameter("2026-01-01") + names.update(components or []) + return names + + +def _provider_dependents(system) -> set[str]: + consumers: dict[str, set[str]] = {} + roots: set[str] = set() + for name, variable in system.variables.items(): + strings, calls = _formula_references(variable) + if calls & _PROVIDER_CALLS: + roots.add(name) + for reference in (strings | _component_names(variable, system)) & set( + system.variables + ): + consumers.setdefault(reference, set()).add(name) + dependents, frontier = set(roots), list(roots) + while frontier: + for consumer in consumers.get(frontier.pop(), ()): + if consumer not in dependents: + dependents.add(consumer) + frontier.append(consumer) + return dependents + + +def test_the_override_covers_every_variable_that_depends_on_the_provider(): + from policyengine_us.system import system + + dependents = _provider_dependents(system) + assert set(THRESHOLD_FIELDS) <= dependents + assert dependents <= set(SPM_MEASUREMENT_VARIABLES), sorted( + dependents - set(SPM_MEASUREMENT_VARIABLES) + ) + + +def test_the_installed_chain_has_the_verified_shape(): + from policyengine_us.system import system + + _check_spm_chain(system.variables) + + +def test_the_installed_model_does_not_read_scope_natively_yet(): + """When this fails, policyengine-us reads scope (#9462): see the runbook.""" + from policyengine_us.system import system + + assert UNIVERSE_VARIABLE not in system.variables + + +#: The installed ``spm_unit_income_decile`` formula the override recalculates +#: over measured units (policyengine-us 2.2.1). The override re-implements +#: this one formula rather than wrapping it, so a different installed method +#: must fail here before legacy-source deciles can drift from the model's. +_VERIFIED_DECILE_FORMULA = """ +def formula(spm_unit, period, parameters): + income = spm_unit("spm_unit_oecd_equiv_net_income", period) + weights = spm_unit("spm_unit_weight", period) * spm_unit.nb_persons() + return MicroSeries(income, weights=weights).decile_rank() +""" + + +def _normalised_ast(source: str) -> str: + return ast.dump(ast.parse(textwrap.dedent(source)), annotate_fields=False) + + +def test_the_recalculated_decile_matches_the_installed_method(): + from policyengine_us.system import system + + formulas = list(system.variables["spm_unit_income_decile"].formulas.values()) + assert len(formulas) == 1 + installed = _normalised_ast(inspect.getsource(formulas[0])) + assert installed == _normalised_ast(_VERIFIED_DECILE_FORMULA), ( + "policyengine-us changed how spm_unit_income_decile ranks units; update " + "legacy_spm.model_adapter._income_decile_variable to the new method " + "before updating _VERIFIED_DECILE_FORMULA." + ) + + +# --- Chain-shape guard ------------------------------------------------------- + + +def _variable(*, formulas=True, adds=None): + return SimpleNamespace( + formulas={"0001-01-01": object()} if formulas else {}, adds=adds + ) + + +def _verified_chain() -> dict: + chain = { + name: _variable() + for name in ( + *THRESHOLD_FIELDS, + *MASKED_FORMULA_VARIABLES, + model_adapter.DECILE_VARIABLE, + ) + } + for name in NAN_PROPAGATING_VARIABLES: + chain[name] = _variable(formulas=False, adds=["component"]) + chain[model_adapter.INDEPENDENCE_ROLE] = _variable() + return chain + + +def test_a_verified_chain_passes(): + _check_spm_chain(_verified_chain()) + + +@pytest.mark.parametrize( + "change, message", + [ + (lambda chain: chain.pop("spm_unit_spm_threshold"), "is not defined"), + ( + lambda chain: chain.__setitem__("poverty_gap", _variable(formulas=False)), + "poverty_gap has no formula", + ), + ( + lambda chain: chain.__setitem__("spm_unit_net_income", _variable()), + "spm_unit_net_income no longer adds up", + ), + ( + lambda chain: chain.pop(model_adapter.INDEPENDENCE_ROLE), + "is_spm_independent_minor_role is not defined", + ), + ], +) +def test_a_changed_chain_is_refused(change, message): + chain = _verified_chain() + change(chain) + with pytest.raises(ValueError, match=message): + _check_spm_chain(chain) + + +# --- Mode selection ---------------------------------------------------------- + + +class _FakeSimulation: + def __init__(self, variables, branches=None): + self.tax_benefit_system = SimpleNamespace(variables=variables) + self.branches = branches or {} + self.reforms = [] + + def apply_reform(self, reform): + self.reforms.append(reform) + self.tax_benefit_system.variables[UNIVERSE_VARIABLE] = SimpleNamespace( + metadata={ADAPTER_MARKER: 1} + ) + + +def _native_universe(members=("INCLUDED", "OUTSIDE", "UNRESOLVED")): + return SimpleNamespace( + metadata={}, + possible_values=[SimpleNamespace(name=member) for member in members], + ) + + +def test_a_model_without_native_scope_is_overridden_once_per_simulation(): + baseline = _FakeSimulation({}) + simulation = _FakeSimulation({}, {"baseline": baseline}) + assert prepare_microsimulation(simulation) == "override" + assert prepare_microsimulation(simulation) == "override" + assert len(simulation.reforms) == len(baseline.reforms) == 1 + + +def test_a_model_with_native_scope_is_left_alone(): + simulation = _FakeSimulation({UNIVERSE_VARIABLE: _native_universe()}) + assert prepare_microsimulation(simulation) == "native" + assert simulation.reforms == [] + + +def test_a_native_scope_without_the_declared_members_is_refused(): + simulation = _FakeSimulation({UNIVERSE_VARIABLE: _native_universe(("IN", "OUT"))}) + with pytest.raises(ValueError, match="declarations need"): + prepare_microsimulation(simulation) + + +def test_branches_must_agree_on_the_mode(): + native = _FakeSimulation({UNIVERSE_VARIABLE: _native_universe()}) + simulation = _FakeSimulation({}, {"baseline": native}) + with pytest.raises(ValueError, match="disagree"): + prepare_microsimulation(simulation) diff --git a/tests/test_us_year_file_stamp.py b/tests/test_us_year_file_stamp.py index 7d11eef0..38d4acba 100644 --- a/tests/test_us_year_file_stamp.py +++ b/tests/test_us_year_file_stamp.py @@ -65,6 +65,9 @@ year_file_stamp, year_file_stamp_problem, ) +from policyengine.tax_benefit_models.us.legacy_spm import ( + RECORD_H5_DATASET as LEGACY_SPM_RECORD_H5_DATASET, +) LEGACY = "would_claim_wic" LIVE = "takes_up_wic_if_eligible" @@ -975,7 +978,13 @@ def __exit__(self, *exc_info): self._file.close() def get(self, name, default=None): - assert name in (YEAR_FILE_STAMP_H5_DATASET, RENAMES_H5_DATASET), name + # The legacy SPM declarations record is a small JSON node like the + # stamp, read to tell a legacy source's year file is current. + assert name in ( + YEAR_FILE_STAMP_H5_DATASET, + RENAMES_H5_DATASET, + LEGACY_SPM_RECORD_H5_DATASET, + ), name return self._file.get(name, default) with (