Add maintenance loan targets

Author: MaxGhenis

changelog.d/maintenance-loan-targets.added.md

@@ -0,0 +1 @@
+Add Student Loans Company maintenance-loan recipient-count and spend targets for England full-time undergraduates.

policyengine_uk_data/targets/sources/slc.py

@@ -5,13 +5,19 @@
 Two target types are exposed:
 - `above_threshold`: borrowers liable to repay and earning above threshold
 - `liable`: all borrowers liable to repay, including below-threshold holders
+- `maintenance_loan`: full-time undergraduate England maintenance-loan
+  recipient counts and total amount paid
 
 Source: Explore Education Statistics — Student loan forecasts for England,
 Table 6a: Forecast number of student borrowers liable to repay and number
 earning above repayment threshold, by product. We use the "Higher education
 total" row which sums HE full-time, HE part-time, and Advanced Learner loans.
 Academic year 20XX-YY maps to calendar year 20XX+1 (e.g., 2024-25 → 2025).
 
+Maintenance-loan targets come from Student support for higher education in
+England 2025, Table 3A: Maintenance Loans paid to full-time undergraduate
+students. Academic year 20XX/YY maps to calendar year 20XX+1.
+
 Data permalink:
 https://explore-education-statistics.service.gov.uk/data-tables/permalink/6ff75517-7124-487c-cb4e-08de6eccf22d
 """
@@ -21,6 +27,7 @@
 import re
 from functools import lru_cache
 
+import pandas as pd
 import requests
 
 from policyengine_uk_data.targets.schema import Target, Unit
@@ -30,6 +37,10 @@
     f"https://explore-education-statistics.service.gov.uk"
     f"/data-tables/permalink/{_PERMALINK_ID}"
 )
+_MAINTENANCE_LOAN_URL = (
+    "https://assets.publishing.service.gov.uk/media/"
+    "691d9e662c6b98ecdbc5003f/slcsp052025.xlsx"
+)
 _TESTING_DATA = {
     "plan_2": {
         "above_threshold": {
@@ -68,6 +79,36 @@
         },
     },
 }
+_MAINTENANCE_LOAN_TESTING_DATA = {
+    "recipients": {
+        2014: 972_830,
+        2015: 963_084,
+        2016: 986_323,
+        2017: 1_013_354,
+        2018: 1_028_438,
+        2019: 1_044_973,
+        2020: 1_055_702,
+        2021: 1_117_591,
+        2022: 1_145_289,
+        2023: 1_151_607,
+        2024: 1_154_427,
+        2025: 1_159_761,
+    },
+    "amount_paid": {
+        2014: 3_783_626_551,
+        2015: 3_784_628_482,
+        2016: 3_996_708_360,
+        2017: 4_870_158_274,
+        2018: 5_746_431_691,
+        2019: 6_555_506_426,
+        2020: 7_113_141_652,
+        2021: 7_914_340_039,
+        2022: 8_332_837_845,
+        2023: 8_594_103_415,
+        2024: 8_881_701_387,
+        2025: 8_591_659_718,
+    },
+}
 
 
 def get_snapshot_data() -> dict:
@@ -80,6 +121,13 @@ def get_snapshot_data() -> dict:
     }
 
 
+def get_maintenance_loan_snapshot_data() -> dict:
+    """Return the checked-in maintenance-loan snapshot."""
+    return {
+        key: values.copy() for key, values in _MAINTENANCE_LOAN_TESTING_DATA.items()
+    }
+
+
 @lru_cache(maxsize=1)
 def _fetch_slc_data() -> dict:
     """Fetch and parse SLC Table 6a data from Explore Education Statistics.
@@ -166,9 +214,62 @@ def parse_values(row, start_index, years):
     }
 
 
+def _row_contains_text(df: pd.DataFrame, row_index: int, text: str) -> bool:
+    row = df.iloc[row_index].dropna()
+    return any(str(value).strip() == text for value in row)
+
+
+def _find_row(df: pd.DataFrame, text: str, start: int = 0) -> int:
+    for row_index in range(start, len(df)):
+        if _row_contains_text(df, row_index, text):
+            return row_index
+    raise ValueError(f"Could not find row containing {text!r}")
+
+
+@lru_cache(maxsize=1)
+def _fetch_maintenance_loan_data() -> dict:
+    """Fetch full-time England maintenance-loan recipient counts and spend."""
+    if os.environ.get("TESTING", "0") == "1":
+        return get_maintenance_loan_snapshot_data()
+
+    df = pd.read_excel(_MAINTENANCE_LOAN_URL, sheet_name="Table 3A", header=None)
+
+    count_header_row = _find_row(df, "Number of students paid (000s) [27]")
+    count_year_row = count_header_row + 1
+    count_total_row = _find_row(df, "Grand total", start=count_year_row + 1)
+
+    amount_header_row = _find_row(df, "Amount paid (£m)")
+    amount_year_row = amount_header_row + 1
+    amount_total_row = _find_row(df, "Grand total", start=amount_year_row + 1)
+
+    year_columns = {}
+    for column, value in df.iloc[count_year_row].items():
+        if isinstance(value, str) and re.fullmatch(r"\d{4}/\d{2}", value):
+            year_columns[column] = int(value[:4]) + 1
+
+    if not year_columns:
+        raise ValueError("Could not find maintenance-loan year columns")
+
+    recipients = {}
+    amount_paid = {}
+    for column, year in year_columns.items():
+        count_value = df.iloc[count_total_row, column]
+        amount_value = df.iloc[amount_total_row, column]
+        if pd.notna(count_value):
+            recipients[year] = int(round(float(count_value) * 1_000))
+        if pd.notna(amount_value):
+            amount_paid[year] = int(round(float(amount_value) * 1_000_000))
+
+    return {
+        "recipients": recipients,
+        "amount_paid": amount_paid,
+    }
+
+
 def get_targets() -> list[Target]:
     """Generate SLC calibration targets by fetching live data."""
     slc_data = _fetch_slc_data()
+    maintenance_loan_data = _fetch_maintenance_loan_data()
 
     targets = []
 
@@ -189,4 +290,26 @@ def get_targets() -> list[Target]:
                 )
             )
 
+    targets.extend(
+        [
+            Target(
+                name="slc/maintenance_loan_recipients",
+                variable="maintenance_loan",
+                source="slc",
+                unit=Unit.COUNT,
+                is_count=True,
+                values=maintenance_loan_data["recipients"],
+                reference_url=_MAINTENANCE_LOAN_URL,
+            ),
+            Target(
+                name="slc/maintenance_loan_spend",
+                variable="maintenance_loan",
+                source="slc",
+                unit=Unit.GBP,
+                values=maintenance_loan_data["amount_paid"],
+                reference_url=_MAINTENANCE_LOAN_URL,
+            ),
+        ]
+    )
+
     return targets

policyengine_uk_data/tests/test_student_loan_targets.py

@@ -18,6 +18,16 @@ def test_slc_targets_registered():
     assert "slc/student_loan_repayment/england" in targets
     assert "slc/student_loan_repayment/scotland" in targets
     assert "slc/student_loan_repayment/england/plan_2" in targets
+    assert "slc/maintenance_loan_recipients" in targets
+    assert "slc/maintenance_loan_spend" in targets
+
+
+def test_policyengine_uk_release_exposes_maintenance_loan_variable():
+    """The lockfile should point at a policyengine-uk release with maintenance loans."""
+    from policyengine_uk import CountryTaxBenefitSystem
+
+    system = CountryTaxBenefitSystem()
+    assert "maintenance_loan" in system.variables
 
 
 def test_slc_snapshot_values_match_higher_education_total_rows():
@@ -97,6 +107,28 @@ def test_slc_england_plan_repayments_sum_to_england_total():
     assert england_plans == england_total
 
 
+def test_slc_maintenance_loan_targets_match_official_2025_values():
+    """Maintenance-loan targets should match Table 3A for 2024/25."""
+    from policyengine_uk_data.targets.registry import get_all_targets
+
+    targets = {t.name: t for t in get_all_targets()}
+
+    assert targets["slc/maintenance_loan_recipients"].values[2025] == 1_159_761
+    assert targets["slc/maintenance_loan_spend"].values[2025] == 8_591_659_718
+
+
+def test_slc_maintenance_loan_snapshot_matches_known_series_points():
+    """Snapshot should preserve the published maintenance-loan time series."""
+    from policyengine_uk_data.targets.sources import slc
+
+    data = slc.get_maintenance_loan_snapshot_data()
+
+    assert data["recipients"][2014] == 972_830
+    assert data["recipients"][2024] == 1_154_427
+    assert data["amount_paid"][2017] == 4_870_158_274
+    assert data["amount_paid"][2025] == 8_591_659_718
+
+
 def test_slc_testing_mode_uses_snapshot_without_network(monkeypatch):
     """Dataset-build CI should not depend on a live SLC endpoint."""
     from policyengine_uk_data.targets.sources import slc
@@ -113,6 +145,24 @@ def fail_network(*args, **kwargs):
     slc._fetch_slc_data.cache_clear()
 
 
+def test_slc_maintenance_loan_testing_mode_uses_snapshot_without_network(monkeypatch):
+    """Maintenance-loan targets should also avoid network in TESTING mode."""
+    from policyengine_uk_data.targets.sources import slc
+
+    slc._fetch_maintenance_loan_data.cache_clear()
+    monkeypatch.setenv("TESTING", "1")
+
+    def fail_excel(*args, **kwargs):
+        raise AssertionError("network should not be used in TESTING mode")
+
+    monkeypatch.setattr(slc.pd, "read_excel", fail_excel)
+
+    assert (
+        slc._fetch_maintenance_loan_data() == slc.get_maintenance_loan_snapshot_data()
+    )
+    slc._fetch_maintenance_loan_data.cache_clear()
+
+
 def test_slc_parser_uses_higher_education_total_rows(monkeypatch):
     """Parser should read HE-total rows, not the first matching above-threshold row."""
     from policyengine_uk_data.targets.sources import slc
@@ -218,6 +268,43 @@ def raise_for_status():
     slc._fetch_slc_data.cache_clear()
 
 
+def test_slc_maintenance_loan_parser_uses_grand_total_rows(monkeypatch):
+    """Maintenance-loan parser should read the grand-total rows from Table 3A."""
+    from policyengine_uk_data.targets.sources import slc
+
+    table = np.full((24, 16), np.nan, dtype=object)
+    table[6, 4] = "Number of students paid (000s) [27]"
+    table[7, 4] = "2013/14"
+    table[7, 5] = "2024/25"
+    table[12, 1] = "Grand total"
+    table[12, 4] = 972.830
+    table[12, 5] = 1159.761
+    table[15, 4] = "Amount paid (£m)"
+    table[16, 4] = "2013/14"
+    table[16, 5] = "2024/25"
+    table[21, 1] = "Grand total"
+    table[21, 4] = 3783.626551
+    table[21, 5] = 8591.659718
+
+    df = np.array(table, dtype=object)
+
+    slc._fetch_maintenance_loan_data.cache_clear()
+    monkeypatch.delenv("TESTING", raising=False)
+    monkeypatch.setattr(
+        slc.pd,
+        "read_excel",
+        lambda *args, **kwargs: __import__("pandas").DataFrame(df),
+    )
+
+    data = slc._fetch_maintenance_loan_data()
+    assert data["recipients"][2014] == 972_830
+    assert data["recipients"][2025] == 1_159_761
+    assert data["amount_paid"][2014] == 3_783_626_551
+    assert data["amount_paid"][2025] == 8_591_659_718
+
+    slc._fetch_maintenance_loan_data.cache_clear()
+
+
 def test_student_loan_target_compute_distinguishes_liable_from_repaying():
     """Above-threshold counts should require repayments, while liable counts should not."""
     from policyengine_uk_data.targets.compute.other import (