feat(api): computed state-school benchmarks

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0146VHeLAWjDVE2B5uU67jCB
2026-07-13 18:45:30 +01:00
parent dbaa15c099
commit cec7941b44
2 changed files with 144 additions and 0 deletions
@@ -525,6 +525,72 @@ def get_data_info(db: Session = None) -> dict:
 # SUPPLEMENTARY DATA — per-school detail page
 # =============================================================================

+def compute_benchmarks(df: pd.DataFrame) -> dict:
+    """State-school benchmarks computed from our dataset (spec §5/§8.6).
+
+    NOT official DfE figures — consumers must label them
+    "state-school average (computed from our dataset)". The disadvantaged
+    attainment average is weighted by cohort size (eligible_pupils) so
+    small schools don't dominate; context measures are medians.
+    """
+    if df.empty or "year" not in df.columns:
+        return {}
+    latest_year = df["year"].max()
+    if pd.isna(latest_year):
+        return {}
+    d = df[df["year"] == latest_year]
+    if d.empty:
+        return {}
+    is_secondary = (
+        d["attainment_8_score"].notna()
+        if "attainment_8_score" in d.columns
+        else pd.Series(False, index=d.index)
+    )
+    prim, sec = d[~is_secondary], d[is_secondary]
+
+    def _median(sub, col):
+        if col not in sub.columns:
+            return None
+        v = sub[col].median()
+        return round(float(v), 1) if pd.notna(v) else None
+
+    def _weighted_disadvantaged(sub):
+        needed = {"rwm_expected_disadvantaged_pct", "eligible_pupils"}
+        if not needed <= set(sub.columns):
+            return None
+        s = sub.dropna(subset=list(needed))
+        if s.empty or s["eligible_pupils"].sum() == 0:
+            return None
+        w = (
+            (s["rwm_expected_disadvantaged_pct"] * s["eligible_pupils"]).sum()
+            / s["eligible_pupils"].sum()
+        )
+        return round(float(w), 1)
+
+    def _block(sub, with_disadvantaged):
+        median_pupils = None
+        if "total_pupils" in sub.columns:
+            mp = sub["total_pupils"].median()
+            if pd.notna(mp):
+                median_pupils = int(mp)
+        block = {
+            "eal_pct": _median(sub, "eal_pct"),
+            "sen_support_pct": _median(sub, "sen_support_pct"),
+            "disadvantaged_pct": _median(sub, "disadvantaged_pct"),
+            "median_pupils": median_pupils,
+        }
+        if with_disadvantaged:
+            block["disadvantaged_rwm_expected_pct"] = _weighted_disadvantaged(sub)
+        return block
+
+    return {
+        "source": "state-school average (computed from our dataset)",
+        "year": int(latest_year),
+        "primary": _block(prim, with_disadvantaged=True),
+        "secondary": _block(sec, with_disadvantaged=False),
+    }
+
+
 def _ofsted_block(o, urn: int) -> dict:
    """Serialize the latest Ofsted inspection row for API responses.