Add PIP points from disability categories

changelog.d/pip-points.md

@@ -0,0 +1 @@
+Add category-consistent PIP points imputation for future PIP point-threshold modeling.

policyengine_uk_data/datasets/disability_benefits.py

@@ -33,8 +33,15 @@
     "pip_dl_category",
 )
 
+PIP_POINT_COLUMNS = (
+    "pip_m_points",
+    "pip_dl_points",
+)
+
 SAFETY_MARGIN = 0.1
 SURVEY_REPORTED_AMOUNT_WEEKS_IN_YEAR = 365.25 / 7
+PIP_STANDARD_POINTS = 8
+PIP_ENHANCED_POINTS = 12
 
 
 @lru_cache(maxsize=None)
@@ -50,6 +57,12 @@ def _dwp_flag_parameters(year: int):
     return CountryTaxBenefitSystem().parameters(year).gov.dwp
 
 
+@lru_cache(maxsize=None)
+def _model_supports_pip_point_inputs() -> bool:
+    variables = CountryTaxBenefitSystem().variables
+    return all(column in variables for column in PIP_POINT_COLUMNS)
+
+
 def _reported_amount(person: pd.DataFrame, column: str) -> pd.Series:
     if column not in person.columns:
         return pd.Series(0.0, index=person.index)
@@ -70,6 +83,44 @@ def _category_from_reported_amount(
     return category
 
 
+def _minimum_pip_points_from_category(category: pd.Series) -> np.ndarray:
+    category = category.fillna("NONE").astype(str)
+    return np.select(
+        [
+            category == "ENHANCED",
+            category == "STANDARD",
+        ],
+        [
+            PIP_ENHANCED_POINTS,
+            PIP_STANDARD_POINTS,
+        ],
+        default=0,
+    ).astype(int)
+
+
+def add_pip_points_from_categories(
+    person: pd.DataFrame,
+    *,
+    inplace: bool = False,
+) -> pd.DataFrame:
+    """Assign the minimum PIP points consistent with observed categories."""
+
+    if not inplace:
+        person = person.copy()
+
+    mappings = (
+        ("pip_m_category", "pip_m_points"),
+        ("pip_dl_category", "pip_dl_points"),
+    )
+    for category_column, points_column in mappings:
+        if category_column in person.columns:
+            person[points_column] = _minimum_pip_points_from_category(
+                person[category_column]
+            )
+
+    return person
+
+
 def add_disability_benefit_categories_from_reported_amounts(
     person: pd.DataFrame,
     year: int,
@@ -133,6 +184,9 @@ def add_disability_benefit_categories_from_reported_amounts(
                 thresholds,
             )
 
+    if _model_supports_pip_point_inputs():
+        add_pip_points_from_categories(person, inplace=True)
+
     return person
 
 

policyengine_uk_data/tests/test_disability_benefits.py

@@ -8,6 +8,7 @@
 from policyengine_uk_data.datasets.disability_benefits import (
     add_disability_benefit_categories_from_reported_amounts,
     add_disability_benefit_flags_from_reported_amounts,
+    add_pip_points_from_categories,
     drop_internal_disability_reported_amounts,
     strip_internal_disability_reported_amounts,
 )
@@ -55,6 +56,22 @@ def test_reported_amounts_map_to_disability_categories():
     assert result["pip_dl_category"].tolist() == ["NONE", "STANDARD", "ENHANCED"]
 
 
+def test_pip_categories_map_to_minimum_qualifying_points():
+    person = pd.DataFrame(
+        {
+            "pip_m_category": ["NONE", "STANDARD", "ENHANCED", None],
+            "pip_dl_category": ["ENHANCED", "STANDARD", "NONE", None],
+        }
+    )
+
+    result = add_pip_points_from_categories(person)
+
+    assert result["pip_m_points"].tolist() == [0, 8, 12, 0]
+    assert result["pip_dl_points"].tolist() == [12, 8, 0, 0]
+    assert "pip_m_points" not in person.columns
+    assert "pip_dl_points" not in person.columns
+
+
 def test_reported_amounts_recompute_disability_flags():
     year = 2025
     dwp = CountryTaxBenefitSystem().parameters(year).gov.dwp