Add household_options.time_period to the country metadata

changelog.d/household-time-period.added.md

@@ -0,0 +1 @@
+Add `household_options.time_period` to the `/{country}/metadata` response: the years offered for household calculations, in the shape of `economy_options.time_period`. The US years are the years the certified SPM forecast covers, up to the last year of the CPI-U parameter; the UK years follow the rule for the UK society-wide years. `economy_options` is unchanged.

policyengine_api/country.py

@@ -38,10 +38,12 @@
 )
 from policyengine_api.data.uk_regions import uk_areas
 from policyengine_api.economy_years import uk_economy_years
+from policyengine_api.household_years import us_household_years
 from policyengine_api.services.household_calculation_service import CalculationResult
 from policyengine_api.spm import (
     SPM_INPUT_ERROR_CODES,
     calculation_spm_receipt,
+    certified_spm_forecast_years,
     normalize_spm_selection,
     spm_error_detail,
     spm_metadata,
@@ -169,13 +171,15 @@ def parameter_names(self) -> frozenset[str]:
 
     def build_metadata(self) -> None:
         """Build the JSON-safe metadata document served by the metadata endpoints."""
+        economy_options = self.build_microsimulation_options()
         self.metadata = self._json_safe(
             {
                 "variables": self.build_variables(),
                 "parameters": self.build_parameters(),
                 "entities": self.build_entities(),
                 "variableModules": self.tax_benefit_system.variable_module_metadata,
-                "economy_options": self.build_microsimulation_options(),
+                "economy_options": economy_options,
+                "household_options": self.build_household_options(economy_options),
                 "current_law_id": get_current_law_policy_id(self.country_id),
                 "basicInputs": self.tax_benefit_system.basic_inputs,
                 "modelled_policies": self.tax_benefit_system.modelled_policies,
@@ -315,6 +319,37 @@ def build_microsimulation_options(self) -> dict:
             options["datasets"] = datasets
         return options
 
+    def build_household_options(self, economy_options: dict) -> dict:
+        """Build the years offered for household calculations.
+
+        Each year has the shape of an ``economy_options`` year. The US years
+        are the certified SPM forecast's years up to the last CPI-U year
+        (``us_household_years``), or the society-wide years when the bundle
+        has no certified SPM forecast. The UK years follow the rule for the
+        UK society-wide years (``uk_economy_years``). Any other country
+        offers its society-wide years.
+        """
+        parameters = self.tax_benefit_system.parameters
+        years = None
+        if self.country_id == "uk":
+            years = uk_economy_years(parameters)
+        elif self.country_id == "us":
+            forecast_years = certified_spm_forecast_years()
+            if forecast_years is None:
+                logger.warning(
+                    "The bundle has no certified SPM forecast; the US household "
+                    "years are the US society-wide years"
+                )
+            else:
+                years = us_household_years(parameters, forecast_years)
+        if years is None:
+            return {
+                "time_period": [
+                    dict(period) for period in economy_options.get("time_period", [])
+                ]
+            }
+        return {"time_period": [{"name": year, "label": str(year)} for year in years]}
+
     def build_variables(self) -> dict:
         """Describe every variable by name; enum variables list their values."""
         variables = self.tax_benefit_system.variables

policyengine_api/household_years.py

@@ -0,0 +1,39 @@
+"""The US years offered for household calculations.
+
+A US household calculation of an SPM measure, such as the SPM poverty
+threshold, needs a year that the bundle's certified SPM forecast covers: the
+US model answers any other year with ``SPM_YEAR_UNAVAILABLE``. The US household
+years are therefore the forecast's years, ending no later than the last year
+written in the CPI-U parameter's source file (``gov.bls.cpi.cpi_u``), which is
+also the last year policyengine-us projects a dataset to by default. Both
+come from the installed model and bundle, so building the list loads no
+dataset and runs no simulation.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from policyengine_us.data.economic_assumptions import (
+    CPI_U_PARAM_PATH,
+    get_parameter_last_year,
+)
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
+    from policyengine_core.parameters import ParameterNode
+
+
+def us_household_years(
+    parameters: ParameterNode, forecast_years: Iterable[int]
+) -> list[int]:
+    """Return the forecast years up to the last CPI-U year, newest first.
+
+    Newest first is the order of the US society-wide years
+    (``economy_options.time_period``).
+    """
+    last_cpi_u_year = get_parameter_last_year(parameters.get_child(CPI_U_PARAM_PATH))
+    return sorted(
+        {year for year in forecast_years if year <= last_cpi_u_year}, reverse=True
+    )

policyengine_api/spm.py

@@ -739,6 +739,25 @@ def spm_metadata(country_id: str) -> dict:
     }
 
 
+def certified_spm_forecast_years() -> tuple[int, ...] | None:
+    """Return the years the bundle's certified US SPM forecast covers.
+
+    The US model refuses an SPM calculation in any other year
+    (``SPM_YEAR_UNAVAILABLE``). Return None in the cases in which
+    ``spm_metadata`` reports SPM as unavailable: the bundle certifies no
+    forecast, or the certified one cannot be loaded. The forecast file is read
+    once per process and shared with ``spm_metadata``.
+    """
+    try:
+        defaults = normalize_spm_selection("us", None)
+    except SPMValidationError:
+        return None
+    if defaults is None:
+        return None
+    forecast = _selected_forecast(defaults["forecast_content_sha256"])
+    return tuple(int(year) for year in forecast.years)
+
+
 def calculation_spm_receipt(simulation: Simulation) -> dict:
     """Reduce a country's existing receipt; never request calculations here."""
     if not hasattr(simulation, "spm_config"):

tests/unit/test_household_options.py

@@ -0,0 +1,147 @@
+"""The household years in each country's metadata (``household_options``)."""
+
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+import yaml
+from policyengine_us.data.economic_assumptions import CPI_U_PARAM_PATH
+from spm_calculator.rolling_forecast import load_forecast
+
+from policyengine_api import country as country_module
+from policyengine_api.country import COUNTRIES, PolicyEngineCountry
+from policyengine_api.economy_years import uk_economy_years
+from policyengine_api.household_years import us_household_years
+from policyengine_api.services.metadata_service import MetadataService
+from policyengine_api.spm import normalize_spm_selection
+
+# The US society-wide years as master lists them, newest first. This pull
+# request leaves them unchanged.
+US_SOCIETY_WIDE_YEARS = list(range(2035, 2021, -1))
+
+
+def _metadata(country_id):
+    return MetadataService().get_metadata(country_id)
+
+
+def _year_names(time_period):
+    return [period["name"] for period in time_period]
+
+
+class _ParameterTreeStandIn:
+    """Stands in for a parameter tree holding only the CPI-U parameter."""
+
+    def __init__(self, cpi_u_file):
+        self.cpi_u_file = cpi_u_file
+
+    def get_child(self, path):
+        assert path == CPI_U_PARAM_PATH
+        return SimpleNamespace(file_path=str(self.cpi_u_file))
+
+
+def _country_stand_in(country_id):
+    country = PolicyEngineCountry.__new__(PolicyEngineCountry)
+    country.country_id = country_id
+    country.tax_benefit_system = SimpleNamespace(parameters=object())
+    return country
+
+
+def test_us_household_years_stop_at_the_last_cpi_u_year_newest_first(tmp_path):
+    cpi_u_file = tmp_path / "cpi_u.yaml"
+    cpi_u_file.write_text(
+        yaml.safe_dump({"values": {"2013-01-01": 233.0, "2033-01-01": 340.0}})
+    )
+
+    years = us_household_years(
+        _ParameterTreeStandIn(cpi_u_file), [2023, 2022, 2033, 2034, 2035, 2023]
+    )
+
+    assert years == [2033, 2023, 2022]
+
+
+def test_us_household_years_are_the_society_wide_years_without_a_forecast(
+    monkeypatch,
+):
+    monkeypatch.setattr(country_module, "certified_spm_forecast_years", lambda: None)
+    economy_time_period = [
+        {"name": year, "label": str(year)} for year in US_SOCIETY_WIDE_YEARS
+    ]
+
+    options = _country_stand_in("us").build_household_options(
+        {"time_period": economy_time_period}
+    )
+
+    assert options == {"time_period": economy_time_period}
+
+
+def test_a_country_without_household_rules_offers_its_society_wide_years():
+    economy_time_period = [{"name": 2026, "label": "2026"}]
+
+    options = _country_stand_in("ca").build_household_options(
+        {"time_period": economy_time_period}
+    )
+
+    assert options == {"time_period": economy_time_period}
+
+
+def test_us_household_years_are_the_certified_forecast_years_to_cpi_u():
+    """The US years come from the installed bundle's forecast and CPI-U file."""
+    forecast_sha256 = normalize_spm_selection("us", None)["forecast_content_sha256"]
+    forecast_years = load_forecast(expected_sha256=forecast_sha256).years
+    cpi_u_file = Path(
+        COUNTRIES["us"]
+        .tax_benefit_system.parameters.get_child(CPI_U_PARAM_PATH)
+        .file_path
+    )
+    cpi_u_values = yaml.safe_load(cpi_u_file.read_text())["values"]
+    last_cpi_u_year = max(int(str(instant)[:4]) for instant in cpi_u_values)
+
+    time_period = _metadata("us")["household_options"]["time_period"]
+
+    assert _year_names(time_period) == sorted(
+        (year for year in forecast_years if year <= last_cpi_u_year), reverse=True
+    )
+
+
+def test_us_household_years_include_2022_and_2023():
+    """Household calculations stay available for 2022 and 2023."""
+    time_period = _metadata("us")["household_options"]["time_period"]
+
+    assert {2022, 2023} <= set(_year_names(time_period))
+
+
+def test_uk_household_years_follow_the_uk_society_wide_rule():
+    parameters = COUNTRIES["uk"].tax_benefit_system.parameters
+
+    time_period = _metadata("uk")["household_options"]["time_period"]
+
+    assert _year_names(time_period) == uk_economy_years(parameters)
+
+
+@pytest.mark.parametrize("country_id", COUNTRIES)
+def test_household_years_have_the_shape_of_society_wide_years(country_id):
+    metadata = _metadata(country_id)
+    household_time_period = metadata["household_options"]["time_period"]
+    economy_keys = {
+        frozenset(period) for period in metadata["economy_options"]["time_period"]
+    }
+
+    assert household_time_period
+    for period in household_time_period:
+        assert {frozenset(period)} == economy_keys == {frozenset({"name", "label"})}
+        assert type(period["name"]) is int
+        assert period["label"] == str(period["name"])
+
+
+def test_us_society_wide_years_are_unchanged():
+    assert _metadata("us")["economy_options"]["time_period"] == [
+        {"name": year, "label": str(year)} for year in US_SOCIETY_WIDE_YEARS
+    ]
+
+
+def test_uk_society_wide_years_are_unchanged():
+    parameters = COUNTRIES["uk"].tax_benefit_system.parameters
+
+    assert _metadata("uk")["economy_options"]["time_period"] == [
+        {"name": year, "label": str(year)} for year in uk_economy_years(parameters)
+    ]

tests/unit/test_spm_settings.py

@@ -237,6 +237,39 @@ def test_metadata_exposes_typed_defaults_only_when_available():
     assert result["settings_schema"] == spm.SPMSelection.model_json_schema()
 
 
+@pytest.mark.usefixtures("certified_bundle")
+def test_certified_forecast_years_are_the_artifacts_years():
+    assert spm.certified_spm_forecast_years() == (2022, 2035)
+
+
+@pytest.mark.parametrize("model_supports_spm", [False, True])
+def test_certified_forecast_years_are_none_without_a_certified_forecast(
+    monkeypatch, model_supports_spm
+):
+    """No certified forecast means no years, whether or not the model has SPM."""
+    monkeypatch.setattr(spm, "_current_bundle", lambda: deepcopy(LEGACY_BUNDLE))
+    monkeypatch.setattr(spm, "simulation_supports_spm", lambda _: model_supports_spm)
+
+    def must_not_load(_expected):
+        msg = "No certified forecast should be loaded"
+        raise AssertionError(msg)
+
+    monkeypatch.setattr(spm, "_selected_forecast", must_not_load)
+    assert spm.certified_spm_forecast_years() is None
+
+
+@pytest.mark.usefixtures("certified_bundle")
+def test_certified_forecast_years_are_none_when_the_forecast_cannot_load(
+    monkeypatch,
+):
+    def mismatched_artifact(_expected):
+        msg = "Forecast content SHA256 mismatch"
+        raise ValueError(msg)
+
+    monkeypatch.setattr(spm, "_selected_forecast", mismatched_artifact)
+    assert spm.certified_spm_forecast_years() is None
+
+
 @pytest.mark.usefixtures("certified_bundle")
 def test_selected_area_must_exist_in_artifact():
     with pytest.raises(spm.SPMValidationError) as caught: