fix(api): serve DfE's LA averages for "vs LA avg" (H2)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
TudorandClaude Opus 5.5 committed 2026-10-06 12:35:11 +01:00
1 parent c26b65246f
commit e659867590
3 files changed
+185 -8

No files matched your search

+54 -8
View File
@@ -1273,17 +1273,63 @@ async def get_filter_options(request: Request):
}
def _la_averages_payload(df: pd.DataFrame) -> dict:
"""Per-LA Attainment 8 for the "vs LA avg" comparison: DfE's own LA
averages (fact_ks4_la_averages, all state-funded schools), never a mean of
the dataframe. That mean counted independent and special schools and put
most LAs about 7 points low (audit H2).
The year is the latest with any school Attainment 8, so the average and
the scores set against it are the same year. Figures are keyed by our LA
name through the LA code. An LA without a DfE figure for that year (City of
London) is absent; no figures for the year, or no mart, give an empty map,
so rows show no comparison, never another year's figure.
"""
empty = {"year": 0, "secondary": {"attainment_8_by_la": {}}}
if df.empty or "attainment_8_score" not in df.columns:
return empty
scored = df[df["attainment_8_score"].notna()]
if scored.empty:
return empty
year = int(scored["year"].max())
la = (df[["local_authority_code", "local_authority"]]
.dropna()
.drop_duplicates("local_authority_code"))
name_by_code = {int(code): name for code, name in
zip(la["local_authority_code"], la["local_authority"])}
from . import database
from .models import Ks4LaAverage
rows: list = []
db = None
try:
db = database.SessionLocal()
rows = db.query(Ks4LaAverage).filter(Ks4LaAverage.year == year).all()
except Exception:
import logging
logging.getLogger(__name__).warning(
"DfE LA averages unavailable for %s", year, exc_info=True)
if db is not None:
db.rollback()
finally:
if db is not None:
db.close()
by_la = {
name_by_code[row.la_code]: row.attainment_8_score
for row in rows
if row.attainment_8_score is not None and row.la_code in name_by_code
}
return {"year": year, "secondary": {"attainment_8_by_la": by_la}}
@app.get("/api/la-averages")
@limiter.limit(f"{settings.rate_limit_per_minute}/minute")
async def get_la_averages(request: Request):
"""Get per-LA average Attainment 8 score for secondary schools in the latest year."""
df = load_school_data()
if df.empty:
return {"year": 0, "secondary": {"attainment_8_by_la": {}}}
latest_year = int(df["year"].max())
sec_df = df[(df["year"] == latest_year) & df["attainment_8_score"].notna()]
la_avg = sec_df.groupby("local_authority")["attainment_8_score"].mean().round(1).to_dict()
return {"year": latest_year, "secondary": {"attainment_8_by_la": la_avg}}
"""DfE's per-LA Attainment 8 averages for the latest year with results."""
return _la_averages_payload(load_school_data())
_KS2_NATIONAL_METRICS = [