Calibrate fuel litre proxies to road clearances

Author: MaxGhenis

changelog.d/fuel-litre-calibration.md

@@ -0,0 +1 @@
+- Calibrate imputed petrol and diesel litre proxies to HMRC/OBR road-fuel clearances before publishing enhanced FRS datasets.

policyengine_uk_data/datasets/imputations/consumption.py

@@ -617,6 +617,58 @@ def fuel_spending_litre_proxy_uprating(
     )
 
 
+def _fuel_litre_proxy_mlitres(
+    household: pd.DataFrame,
+    fiscal_year: int,
+    parameters=None,
+) -> float:
+    """Return weighted petrol + diesel litres represented by spending proxies."""
+    from policyengine_uk.system import system
+
+    if parameters is None:
+        parameters = system.parameters
+
+    total_litres = 0.0
+    for variable, price_parameter_name in FUEL_PRICE_PARAMETER_NAME.items():
+        price = getattr(
+            parameters.household.consumption.fuel.prices,
+            price_parameter_name,
+        )(fiscal_year)
+        total_litres += household[variable] / price
+    return (total_litres * household["household_weight"]).sum() / 1_000_000
+
+
+def calibrate_fuel_litre_proxies_to_road_fuel(
+    household: pd.DataFrame,
+    fiscal_year: int,
+    parameters=None,
+) -> float:
+    """Scale imputed fuel proxies to HMRC/OBR road-fuel litre controls.
+
+    PolicyEngine UK derives petrol and diesel litres from spending divided by
+    pump prices. Applying one multiplicative factor to petrol and diesel
+    spending preserves the household distribution while making the resulting
+    litres reconcile to the external fuel-duty volume benchmark.
+    """
+    from policyengine_uk_data.sources.road_fuel_volume import (
+        road_fuel_clearances_mlitres,
+    )
+
+    actual_mlitres = _fuel_litre_proxy_mlitres(
+        household,
+        fiscal_year,
+        parameters=parameters,
+    )
+    if actual_mlitres <= 0:
+        return 1.0
+
+    target_mlitres = road_fuel_clearances_mlitres(end_year=fiscal_year)[fiscal_year]
+    scale = target_mlitres / actual_mlitres
+    for variable in FUEL_PRICE_PARAMETER_NAME:
+        household[variable] *= scale
+    return scale
+
+
 def save_imputation_models():
     from policyengine_uk_data.utils.qrf import QRF
 
@@ -739,5 +791,10 @@ def _wmean(arr, mask):
     dataset.household["petrol_spending"][no_fuel] = 0
     dataset.household["diesel_spending"][no_fuel] = 0
 
+    calibrate_fuel_litre_proxies_to_road_fuel(
+        dataset.household,
+        int(dataset.time_period),
+    )
+
     dataset.validate()
     return dataset

policyengine_uk_data/tests/test_road_fuel_volume_uprating.py

@@ -1,10 +1,13 @@
 """Tests for road-fuel volume uprating. See #402."""
 
 import pandas as pd
+import pytest
 
 from policyengine_uk_data.datasets.imputations.consumption import (
     CONSUMPTION_MODEL_FILENAME,
     IMPUTATIONS,
+    _fuel_litre_proxy_mlitres,
+    calibrate_fuel_litre_proxies_to_road_fuel,
     fuel_spending_litre_proxy_uprating,
     uprate_lcfs_table,
 )
@@ -148,6 +151,38 @@ def test__given_obr_2027_volume__then_rate_difference_matches_cost_benchmark():
     assert round(benchmark_cost_gbp_bn, 2) == 3.12
 
 
+def test__given_imputed_fuel_proxies__then_calibration_matches_road_fuel_litres():
+    # Given
+    from policyengine_uk.system import system
+
+    year = 2027
+    target_mlitres = road_fuel_clearances_mlitres()[year]
+    petrol_price = system.parameters.household.consumption.fuel.prices.petrol(year)
+    diesel_price = system.parameters.household.consumption.fuel.prices.diesel(year)
+    household = pd.DataFrame(
+        {
+            "household_weight": [1.0, 1.0],
+            "petrol_spending": [
+                target_mlitres * 1_000_000 * petrol_price,
+                0.0,
+            ],
+            "diesel_spending": [
+                0.0,
+                target_mlitres * 1_000_000 * diesel_price,
+            ],
+        }
+    )
+
+    # When
+    scale = calibrate_fuel_litre_proxies_to_road_fuel(household, year)
+
+    # Then
+    assert scale == pytest.approx(0.5)
+    assert _fuel_litre_proxy_mlitres(household, year) == pytest.approx(
+        target_mlitres
+    )
+
+
 def test__given_year_after_obr_horizon__then_last_forecast_is_carried_forward():
     # When
     series = road_fuel_clearances_mlitres(end_year=END_YEAR)