backend/app.py

@@ -677,6 +677,7 @@ async def compare_schools(
         "deprivation": None,
     }
     supplementary_by_urn: dict = {}
+    census_benchmarks = None
     db = None
     try:
         db = database.SessionLocal()
@@ -688,6 +689,31 @@ async def compare_schools(
                 key: supp.get(key, default)
                 for key, default in _EMPTY_SUPPLEMENTARY.items()
             }
+        # Import-time census context benchmarks (fact_census_benchmarks);
+        # absent mart → None, and compute_benchmarks leaves those fields null.
+        try:
+            from .models import CensusBenchmark
+
+            rows = db.query(CensusBenchmark).all()
+            by_phase = {
+                r.phase: {
+                    "year": r.year,
+                    "fsm_pct": r.fsm_pct,
+                    "eal_pct": r.eal_pct,
+                    "median_pupils": r.median_pupils,
+                }
+                for r in rows
+                if getattr(r, "phase", None) in ("primary", "secondary")
+            }
+            if by_phase:
+                census_benchmarks = by_phase
+        except Exception:
+            # Missing mart (or a stubbed session in tests) must never break
+            # the compare payload — and not every session has rollback().
+            try:
+                db.rollback()
+            except Exception:
+                pass
     except Exception:
         supplementary_by_urn = {}
     finally:
@@ -728,7 +754,7 @@ async def compare_schools(
         # Official DfE anchors + computed state-school benchmarks so the
         # compare UI can label provenance correctly (spec §8.6).
         "national_averages": _national_averages_payload(df),
-        "benchmarks": compute_benchmarks(df),
+        "benchmarks": compute_benchmarks(df, census_benchmarks=census_benchmarks),
     }
 
 

backend/data_loader.py

@@ -525,13 +525,20 @@ def get_data_info(db: Session = None) -> dict:
 # SUPPLEMENTARY DATA — per-school detail page
 # =============================================================================
 
-def compute_benchmarks(df: pd.DataFrame) -> dict:
+def compute_benchmarks(df: pd.DataFrame, census_benchmarks: dict | None = None) -> 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.
+    small schools don't dominate.
+
+    Context measures (FSM/EAL/pupil counts) come from `census_benchmarks`
+    (the fact_census_benchmarks mart, pupil-weighted, keyed by phase): the
+    performance df has no fsm_pct at all, and its eal/disadvantaged columns
+    are KS2-only — medianing them for "secondary" produced junk anchors
+    from the handful of all-through schools. When the mart is unavailable
+    these are None; never fall back across measure definitions.
     """
     if df.empty or "year" not in df.columns:
         return {}
@@ -567,18 +574,14 @@ def compute_benchmarks(df: pd.DataFrame) -> dict:
         )
         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)
+    def _block(sub, phase, with_disadvantaged):
+        census = (census_benchmarks or {}).get(phase) or {}
         block = {
-            "eal_pct": _median(sub, "eal_pct"),
+            "eal_pct": census.get("eal_pct"),
             "sen_support_pct": _median(sub, "sen_support_pct"),
-            "disadvantaged_pct": _median(sub, "disadvantaged_pct"),
-            "fsm_pct": _median(sub, "fsm_pct"),
-            "median_pupils": median_pupils,
+            "disadvantaged_pct": _median(sub, "disadvantaged_pct") if with_disadvantaged else None,
+            "fsm_pct": census.get("fsm_pct"),
+            "median_pupils": census.get("median_pupils"),
         }
         if with_disadvantaged:
             block["disadvantaged_rwm_expected_pct"] = _weighted_disadvantaged(sub)
@@ -587,8 +590,8 @@ def compute_benchmarks(df: pd.DataFrame) -> dict:
     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),
+        "primary": _block(prim, "primary", with_disadvantaged=True),
+        "secondary": _block(sec, "secondary", with_disadvantaged=False),
     }
 
 

backend/models.py

@@ -234,6 +234,22 @@ class FactFinance(Base):
     premises_cost_pct = Column(Float)
 
 
+class CensusBenchmark(Base):
+    """State-school context benchmarks from the pupil census — one row per phase.
+
+    fsm_pct / eal_pct are pupil-weighted means. Computed at import time;
+    consumers label them "state-school average (computed from our dataset)".
+    """
+    __tablename__ = "fact_census_benchmarks"
+    __table_args__ = MARTS
+
+    phase = Column(String(20), primary_key=True)
+    year = Column(Integer)
+    fsm_pct = Column(Float)
+    eal_pct = Column(Float)
+    median_pupils = Column(Integer)
+
+
 class Ks4NationalAverage(Base):
     """Computed national KS4 averages (from our dataset) — one row per year."""
     __tablename__ = "fact_ks4_national_averages"

backend/tests/test_benchmarks.py

@@ -57,19 +57,40 @@ def test_weighted_disadvantaged_average():
 def test_medians_ignore_nan_and_older_years():
     b = compute_benchmarks(_df())
     assert b["year"] == LATEST
-    # eal medians over [10,20,30,40,50] = 30
-    assert b["primary"]["eal_pct"] == 30.0
-    # fsm medians over [15,17,19,21,23] = 19
-    assert b["primary"]["fsm_pct"] == 19.0
-    # median pupils over [200,280,300,350,400] = 300
-    assert b["primary"]["median_pupils"] == 300
+    # sen medians over [10,14,18,20,22] = 18 — the only context measure still
+    # sourced from the performance df (the rest come from the census mart).
+    assert b["primary"]["sen_support_pct"] == 18.0
+    # disadvantaged_pct medians over [20,24,30,40,44] = 30
+    assert b["primary"]["disadvantaged_pct"] == 30.0
+
+
+def test_benchmarks_use_census_mart_for_context():
+    census = {
+        "primary": {"year": LATEST, "fsm_pct": 25.3, "eal_pct": 21.8, "median_pupils": 240},
+        "secondary": {"year": LATEST, "fsm_pct": 24.1, "eal_pct": 18.9, "median_pupils": 980},
+    }
+    b = compute_benchmarks(_df(), census_benchmarks=census)
+    assert b["primary"]["fsm_pct"] == 25.3
+    assert b["primary"]["eal_pct"] == 21.8
+    assert b["secondary"]["eal_pct"] == 18.9
+    assert b["secondary"]["median_pupils"] == 980
+
+
+def test_benchmarks_context_none_when_mart_missing():
+    # The performance df has no fsm_pct and its eal/disadvantaged columns are
+    # KS2-only — never silently fall back to medianing them for context.
+    b = compute_benchmarks(_df(), census_benchmarks=None)
+    assert b["primary"]["fsm_pct"] is None
+    assert b["primary"]["eal_pct"] is None
+    assert b["primary"]["median_pupils"] is None
 
 
 def test_secondary_block_has_no_disadvantaged_rwm():
     b = compute_benchmarks(_df())
     assert "disadvantaged_rwm_expected_pct" not in b["secondary"]
-    assert b["secondary"]["fsm_pct"] == 13.0
-    assert b["secondary"]["median_pupils"] == 1100
+    # KS2-only columns must not produce a fake secondary disadvantaged anchor
+    # (the old median over all-through schools' KS2 rows produced 50%).
+    assert b["secondary"]["disadvantaged_pct"] is None
 
 
 def test_provenance_string():

nextjs-app/components/compare/CompareCommunity.tsx

@@ -33,7 +33,10 @@ export function CompareCommunity({
   const bench = isSecondary ? benchmarks?.secondary : benchmarks?.primary;
 
   const fsmChip = (value: number | null) => {
-    const anchor = bench?.fsm_pct ?? bench?.disadvantaged_pct ?? null;
+    // FSM is anchored only against a real FSM benchmark (census-sourced,
+    // pupil-weighted). disadvantaged_pct is a different measure (FSM6+CLA)
+    // — never fall back across definitions; no anchor means no chip.
+    const anchor = bench?.fsm_pct ?? null;
     if (value == null || anchor == null) return null;
     const v = verdict(value, anchor, 3);
     return (

pipeline/transform/models/marts/_marts_schema.yml

@@ -133,6 +133,16 @@ models:
       - name: year
         tests: [not_null]
 
+  - name: fact_census_benchmarks
+    description: >
+      State-school context benchmarks from the pupil census — one row per
+      phase (primary/secondary), latest census year. fsm_pct/eal_pct are
+      pupil-weighted means; consumers label them "state-school average
+      (computed from our dataset)", never "England average".
+    columns:
+      - name: phase
+        tests: [not_null, unique]
+
   - name: fact_admissions
     description: School admissions — one row per URN per year
     columns:

pipeline/transform/models/marts/fact_census_benchmarks.sql

@@ -0,0 +1,39 @@
+{{ config(materialized='table') }}
+
+-- Mart: state-school context benchmarks from the pupil census — one row per
+-- phase, latest census year. Computed at import time (never per request).
+-- fsm_pct / eal_pct are pupil-weighted means, i.e. "what % of pupils", not
+-- "the median school" — this matches how DfE quotes national FSM/EAL rates.
+-- Consumers must label these "state-school average (computed from our
+-- dataset)" (spec §8.6), never "England average".
+
+with latest as (
+    select max(year) as year from {{ ref('fact_pupil_characteristics') }}
+),
+
+classified as (
+    select
+        case
+            when p.phase_type_grouping ilike '%primary%'   then 'primary'
+            when p.phase_type_grouping ilike '%secondary%' then 'secondary'
+        end as phase,
+        p.total_pupils,
+        p.fsm_pct,
+        p.eal_pct,
+        l.year
+    from {{ ref('fact_pupil_characteristics') }} p
+    join latest l on p.year = l.year
+    where p.total_pupils is not null and p.total_pupils > 0
+)
+
+select
+    phase,
+    max(year) as year,
+    round((sum(fsm_pct * total_pupils) filter (where fsm_pct is not null)
+        / nullif(sum(total_pupils) filter (where fsm_pct is not null), 0))::numeric, 1) as fsm_pct,
+    round((sum(eal_pct * total_pupils) filter (where eal_pct is not null)
+        / nullif(sum(total_pupils) filter (where eal_pct is not null), 0))::numeric, 1) as eal_pct,
+    round(percentile_cont(0.5) within group (order by total_pupils))::integer as median_pupils
+from classified
+where phase is not null
+group by phase