Project labelled energy inputs from their declared price year

changelog.d/2526.fixed.md

@@ -0,0 +1 @@
+Added explicit energy price-period declarations for dataset projection using the existing annual CPI assumption, while preserving flat energy amounts in unlabelled inputs.

docs/engineering/skills/dataset-sources.md

@@ -52,6 +52,63 @@ configured environment variable.
 Keep default dataset behavior explicit so calculator-style simulations,
 microsimulations, and API workers do not accidentally diverge.
 
+## Energy price declarations
+
+Unlabelled dataset energy spending stays flat through the projection horizon.
+Do not infer its price year from the dataset year, filename, URL or file format.
+Legacy builds can hold FY2026-27 energy prices in a dataset labelled 2024.
+
+An explicit declaration opts the three stored energy amounts into their
+existing annual OBR consumer-price-index assumption. For an input whose
+producer receipt declares calendar-2024 prices, a caller can label a local
+UK single-year H5 before constructing a simulation:
+
+```python
+from policyengine_uk import Microsimulation
+from policyengine_uk.data import UKSingleYearDataset
+
+dataset = UKSingleYearDataset(
+    file_path="verified-input.h5",
+    energy_price_basis={"period_type": "calendar_year", "period_value": 2024},
+)
+simulation = Microsimulation(dataset=dataset)
+```
+
+Use `fiscal_year` with the starting year when the receipt describes a UK
+financial year. Both keys are required. The year must be an integer from 1
+through 9999; malformed or conflicting caller/file declarations raise a clear
+error. A caller's `None` leaves an existing file declaration intact.
+
+The model retains the period kind as provenance but uses its existing annual
+index-year convention: CY2024 and FY2024-25 both use `I[2024]`. Projection
+multiplies the supplied amount by `I[target year] / I[price year]`. This is
+the existing generic CPI assumption, with no new energy-specific forecast or
+precise fiscal/calendar price adjustment. A 2026 basis therefore stays unchanged
+at target 2026 and can grow at target 2027. The original observation remains
+unchanged; a projected child's declaration records its effective fiscal-year
+price basis so later extension does not apply the original ratio twice.
+
+Single-year H5 files store a JSON declaration in the root attribute
+`energy_price_basis`. Multi-year files store it on each `/time_period/<year>`
+group. Copy, save/load and filtering preserve the declaration. A Core Dataset
+can declare the same named field or H5 attribute before UK conversion. HF/GCS
+materializers retain the existing local-file routing. External multi-year
+amounts and DataFrame inputs remain authoritative when loaded.
+
+Disabling economic assumptions resets energy amounts and their declaration as
+one group. A base year that supplies every stored later energy column replaces
+those amounts and their declaration. With no overlapping base energy, later
+amounts and metadata stay intact. A partial overlap resets shared columns only
+when both declarations match exactly; otherwise the entire later energy group
+and its declaration remain authoritative. Missing columns are never created.
+
+Existing metadata-free private Microcosm releases stay flat under this
+compatibility rule. They require an explicit caller migration based on a
+verified receipt, or a producer release that exports the canonical declaration
+matching that receipt. The producer must carry both period kind and year into
+the H5 artifact; this consumer change does not retrofit that metadata or verify
+the released private file.
+
 ## Tests
 
 For dataset source routing, prefer stubbed tests that prove dispatch behavior

policyengine_uk/data/dataset_schema.py

@@ -8,6 +8,13 @@
 
 from pathlib import Path
 import h5py
+from policyengine_uk.data.energy_price_basis import (
+    EnergyPriceBasis,
+    merge_energy_price_basis,
+    parse_energy_price_basis,
+    read_energy_price_basis,
+    write_energy_price_basis,
+)
 
 
 class UKSingleYearDataset:
@@ -57,7 +64,9 @@ def __init__(
         benunit: pd.DataFrame = None,
         household: pd.DataFrame = None,
         fiscal_year: int = None,
+        energy_price_basis: dict[str, str | int] | EnergyPriceBasis | None = None,
     ):
+        basis = parse_energy_price_basis(energy_price_basis)
         file_path = str(file_path) if file_path else None
         if file_path is not None:
             if fiscal_year is not None:
@@ -76,6 +85,7 @@ def __init__(
                 self._benunit = f["benunit"]
                 self._household = f["household"]
                 self.time_period = str(f["time_period"].iloc[0])
+            basis = merge_energy_price_basis(basis, read_energy_price_basis(file_path))
         else:
             if person is None or benunit is None or household is None:
                 raise ValueError(
@@ -97,6 +107,7 @@ def __init__(
             self._household = household
             self.time_period = str(fiscal_year)
 
+        self._energy_price_basis = basis
         self.data_format = "arrays"
         # tables tuple points at the raw (int16) DataFrames used by the
         # simulation engine internals - do not decode these.
@@ -106,6 +117,15 @@ def __init__(
         # dataset is loaded and enum columns have been converted to int16.
         self._enum_columns: dict = {}
 
+    @property
+    def energy_price_basis(self) -> dict[str, str | int] | None:
+        """A defensive copy of the stored amounts' declared price period."""
+        return self._energy_price_basis.as_dict() if self._energy_price_basis else None
+
+    @energy_price_basis.setter
+    def energy_price_basis(self, value: object) -> None:
+        self._energy_price_basis = parse_energy_price_basis(value)
+
     def _decode_enum_df(self, df: pd.DataFrame) -> pd.DataFrame:
         """Return a copy of *df* with int16 enum columns decoded to strings."""
         if not self._enum_columns:
@@ -156,6 +176,7 @@ def save(self, file_path: str):
             f.put("benunit", self.benunit, format="table", data_columns=True)
             f.put("household", self.household, format="table", data_columns=True)
             f.put("time_period", pd.Series([self.time_period]), format="table")
+        write_energy_price_basis(file_path, self._energy_price_basis)
 
     def load(self):
         data = {}
@@ -171,6 +192,7 @@ def copy(self):
             benunit=self.benunit.copy(),
             household=self.household.copy(),
             fiscal_year=self.time_period,
+            energy_price_basis=self._energy_price_basis,
         )
 
     def validate(self):
@@ -181,7 +203,11 @@ def validate(self):
                     raise ValueError(f"Column '{col}' contains NaN values.")
 
     @staticmethod
-    def from_simulation(simulation: "Microsimulation", fiscal_year: int = 2025):
+    def from_simulation(
+        simulation: "Microsimulation",
+        fiscal_year: int = 2025,
+        energy_price_basis: dict[str, str | int] | EnergyPriceBasis | None = None,
+    ):
         entity_dfs = {}
 
         for entity in ["person", "benunit", "household"]:
@@ -203,6 +229,7 @@ def from_simulation(simulation: "Microsimulation", fiscal_year: int = 2025):
             benunit=entity_dfs["benunit"],
             household=entity_dfs["household"],
             fiscal_year=fiscal_year,
+            energy_price_basis=energy_price_basis,
         )
 
 
@@ -212,6 +239,7 @@ def __init__(
         file_path: str = None,
         datasets: list[UKSingleYearDataset] | None = None,
     ):
+        file_path = str(file_path) if file_path else None
         if datasets is not None:
             self.datasets = {}
             for dataset in datasets:
@@ -238,6 +266,10 @@ def __init__(
                             household=household_df,
                             fiscal_year=fiscal_year,
                         )
+            for year, dataset in self.datasets.items():
+                dataset.energy_price_basis = read_energy_price_basis(
+                    file_path, f"/time_period/{year}"
+                )
 
         self.data_format = "time_period_arrays"
         self.time_period = list(sorted(self.datasets.keys()))[0]
@@ -283,6 +315,10 @@ def save(self, file_path: str):
                     format="table",
                     data_columns=True,
                 )
+        for year, dataset in self.datasets.items():
+            write_energy_price_basis(
+                file_path, dataset.energy_price_basis, f"/time_period/{year}"
+            )
 
     def copy(self):
         new_datasets = {year: dataset.copy() for year, dataset in self.datasets.items()}

policyengine_uk/data/economic_assumptions.py

@@ -102,8 +102,8 @@
 # Before this horizon was extended, Core uprated unmapped inputs from 2030.
 # Preserve that behavior in the newly materialized years without changing
 # projections for earlier years. Correcting historical mapping omissions is
-# a separate change. Energy values already have FY26/27 prices (#1868), as
-# documented in uprating_indices.yaml, and must remain flat in the dataset.
+# a separate change. Energy has its own declaration-aware projection below,
+# so it must not also grow through the generic variable-index fallback.
 LEGACY_PROJECTION_END_YEAR = 2030
 DATASET_UPRATING_EXCLUSIONS = frozenset(
     {
@@ -137,6 +137,7 @@ def _unmapped_uprating_indices(uprating):
 SOCIAL_RENT_GROWTH = "gov.economic_assumptions.yoy_growth.obr.social_rent"
 PRIVATE_RENT_GROWTH = "gov.economic_assumptions.yoy_growth.ons.private_rental_prices"
 RESIDENTIAL_PROPERTY_INDEX = "gov.economic_assumptions.indices.obr.house_prices"
+ENERGY_PRICE_INDEX = "gov.economic_assumptions.indices.obr.consumer_price_index"
 RESIDENTIAL_PROPERTY_INPUTS = (
     "main_residence_value",
     "other_residential_property_value",
@@ -154,7 +155,7 @@ def uprating_input_paths() -> list:
     ``apply_single_year_uprating`` and the functions it calls read these, and
     no other parameters: the growth rates in ``uprating_indices.yaml``, the
     council tax and rent series, the cumulative residential property index,
-    the indices of the input variables that
+    the conditional energy CPI index, the indices of the input variables that
     file leaves out (after ``LEGACY_PROJECTION_END_YEAR``) and the capital
     gains projection's metadata (``CGT_INDEX_PATH``). A path may name a node
     (the regional private rent series), which covers its parameters.
@@ -171,6 +172,7 @@ def uprating_input_paths() -> list:
             PRIVATE_RENT_GROWTH,
             CGT_INDEX_PATH,
             RESIDENTIAL_PROPERTY_INDEX,
+            ENERGY_PRICE_INDEX,
         }
         | set(_unmapped_uprating_indices(uprating).values())
     )
@@ -270,6 +272,9 @@ def apply_uprating(
         uprate_residential_property(
             current_year, dataset.datasets[first_year], tax_benefit_system_parameters
         )
+        uprate_energy(
+            current_year, dataset.datasets[first_year], tax_benefit_system_parameters
+        )
 
     return dataset
 
@@ -320,6 +325,8 @@ def apply_single_year_uprating(
 
     # Next, apply custom uprating.
 
+    current_year = uprate_energy(current_year, previous_year, parameters)
+
     # apply_uprating replaces these chained stocks from the original observation.
     current_year = uprate_residential_property(current_year, previous_year, parameters)
 
@@ -370,6 +377,42 @@ def apply_single_year_uprating(
     return current_year
 
 
+def uprate_energy(
+    current_year: UKSingleYearDataset,
+    observed_year: UKSingleYearDataset,
+    parameters: ParameterNode,
+) -> UKSingleYearDataset:
+    """Project declared energy prices with the existing annual CPI assumption."""
+    basis = observed_year.energy_price_basis
+    variables = [
+        name
+        for name in DATASET_UPRATING_EXCLUSIONS
+        if name in observed_year.household.columns
+        and name in current_year.household.columns
+    ]
+    ratio = 1
+    projected = False
+    if basis is not None and variables:
+        index = parameters.get_child(ENERGY_PRICE_INDEX)
+        source_index = _uprating_index_value(
+            index, f"{basis['period_value']:04d}-01-01"
+        )
+        target_index = _uprating_index_value(index, f"{current_year.time_period}-01-01")
+        projected = source_index not in (None, 0) and target_index is not None
+        if projected:
+            ratio = target_index / source_index
+    for variable in variables:
+        current_year.household[variable] = (
+            observed_year.household[variable].to_numpy() * ratio
+        )
+    current_year.energy_price_basis = (
+        {"period_type": "fiscal_year", "period_value": int(current_year.time_period)}
+        if projected
+        else basis
+    )
+    return current_year
+
+
 def uprate_residential_property(
     current_year: UKSingleYearDataset,
     previous_year: UKSingleYearDataset,
@@ -666,19 +709,44 @@ def reset_growthfactor_uprating(
     growthfactor_uprated_variables.update({"council_tax", "rent"})
     growthfactor_uprated_variables.update(COUNTRY_UPRATED_INPUTS)
     growthfactor_uprated_variables.update(_unmapped_uprating_indices(uprating))
+    growthfactor_uprated_variables.difference_update(DATASET_UPRATING_EXCLUSIONS)
 
     first_year = min(dataset.datasets.keys())
     base_year_dataset = dataset.datasets[first_year]
+    base_energy = DATASET_UPRATING_EXCLUSIONS.intersection(
+        base_year_dataset.household.columns
+    )
 
     for year in dataset.datasets:
         if year == first_year:
             continue
         current_year_dataset = dataset.datasets[year]
+        current_energy = DATASET_UPRATING_EXCLUSIONS.intersection(
+            current_year_dataset.household.columns
+        )
+        shared_energy = current_energy.intersection(base_energy)
+        replace_energy_group = bool(current_energy) and current_energy <= base_energy
+        reset_energy = replace_energy_group or (
+            current_year_dataset.energy_price_basis
+            == base_year_dataset.energy_price_basis
+        )
+
+        # One declaration describes every stored energy amount. Keep the
+        # later group intact if a partial reset would mix different bases.
         for table_name in current_year_dataset.table_names:
             base_table = getattr(base_year_dataset, table_name)
             current_table = getattr(current_year_dataset, table_name)
             for variable in growthfactor_uprated_variables:
                 if variable in current_table.columns and variable in base_table.columns:
                     current_table[variable] = base_table[variable].values
+        if reset_energy:
+            for variable in shared_energy:
+                current_year_dataset.household[variable] = base_year_dataset.household[
+                    variable
+                ].values
+        if replace_energy_group:
+            current_year_dataset.energy_price_basis = (
+                base_year_dataset.energy_price_basis
+            )
 
     return dataset