The historic roof description conditions the cohort by form family 🟩

roof_construction codes group by FORM (empirical: 1=Flat 98%, 4/5/8=
Pitched 88-99%, 3=dwelling-above 100% over 7,974 certs; 7/9=premises-
above per #1452), so the filter matches families — an exact-code filter
would wrongly drop pitched neighbours lodged as 5/8. Historic prefixes
map to the same families; roof rooms and thatch stay unconditioned.
Harness ladder replay and telemetry mirror the new filter.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

domain/epc_prediction/comparable_properties.py

@@ -31,6 +31,26 @@ FLOOR_AREA_TOLERANCE = 0.20
 # RdSAP Table S1 band letters in chronological order, for band-distance checks.
 _AGE_BAND_ORDER = "ABCDEFGHIJKLM"
 
+# roof_construction codes by FORM family. Pinned empirically — a 7,974-cert
+# co-occurrence sweep of code x roofs[0].description over single-building-part
+# certs (scripts/roof_construction_code_sweep.py): 1=Flat 98%, 4=Pitched 99%,
+# 5/8=Pitched 88%, 3="(another dwelling above)" 100% — plus 7/9 = "(another
+# premises above)", established by the #1452 heat-loss suppression fix.
+ROOF_FORM_BY_CONSTRUCTION: dict[int, str] = {
+    1: "flat",
+    3: "dwelling_above",
+    4: "pitched",
+    5: "pitched",
+    7: "premises_above",
+    8: "pitched",
+    9: "premises_above",
+}
+
+
+def roof_form_of_construction(code: object) -> Optional[str]:
+    """The form family of a roof_construction code, or None when unknown."""
+    return ROOF_FORM_BY_CONSTRUCTION.get(code) if isinstance(code, int) else None
+
 
 def age_bands_within_one(candidate: object, target_band: object) -> bool:
     """Whether two Table S1 band letters are at most one band apart. Assessors
@@ -109,6 +129,12 @@ def select_comparables(
         active=target.wall_construction is not None,
         minimum_cohort=minimum_cohort,
     )
+    cohort = _maybe_filter(
+        cohort,
+        lambda c: _main_roof_form(c) == target.roof_form,
+        active=target.roof_form is not None,
+        minimum_cohort=minimum_cohort,
+    )
     cohort = _maybe_filter(
         cohort,
         lambda c: age_bands_within_one(
@@ -170,6 +196,12 @@ def _main_wall_construction(comparable: ComparableProperty) -> object:
     return parts[0].wall_construction if parts else None
 
 
+def _main_roof_form(comparable: ComparableProperty) -> Optional[str]:
+    """The main building part's roof-form family, or None when unresolvable."""
+    parts = comparable.epc.sap_building_parts
+    return roof_form_of_construction(parts[0].roof_construction) if parts else None
+
+
 def _main_construction_age_band(comparable: ComparableProperty) -> object:
     """The main building part's Table S1 band letter, or None when no part lodged."""
     parts = comparable.epc.sap_building_parts

domain/epc_prediction/historic_conditioning.py

@@ -82,6 +82,16 @@ _FUEL_CODES: dict[str, int] = {
     "biomass (community)": 31,
 }
 
+# Roof FORM families by the description's pre-comma half. Only forms whose
+# API code grouping is pinned (see comparable_properties.ROOF_FORM_BY_CONSTRUCTION
+# for the empirical evidence); roof rooms and thatch stay None — never guess.
+_ROOF_FORM_BY_PREFIX: dict[str, str] = {
+    "Pitched": "pitched",
+    "Flat": "flat",
+    "(another dwelling above)": "dwelling_above",
+    "(another premises above)": "premises_above",
+}
+
 _NOT_COMMUNITY_SUFFIX = " (not community)"
 
 # Pre-RdSAP-17 lodgements carry a deprecation rider after the fuel name; the
@@ -117,6 +127,11 @@ def _wall_construction(description: str) -> Optional[int]:
     return _WALL_MATERIAL_CONSTRUCTION.get(material)
 
 
+def _roof_form(description: str) -> Optional[str]:
+    form = description.split(",", 1)[0].strip()
+    return _ROOF_FORM_BY_PREFIX.get(form)
+
+
 def _main_fuel(description: str) -> Optional[int]:
     base = description.strip().removesuffix(_NOT_COMMUNITY_SUFFIX)
     base = base.removesuffix(_LEGACY_SUFFIX)
@@ -140,6 +155,7 @@ def conditioning_from_historic(record: HistoricEpc) -> HistoricConditioning:
         construction_age_band=_AGE_BAND_LETTERS.get(record.construction_age_band),
         main_fuel=_main_fuel(record.main_fuel),
         total_floor_area_m2=_floor_area(record.total_floor_area),
+        roof_form=_roof_form(record.roof_description),
     )
 
 
@@ -173,4 +189,5 @@ def target_with_conditioning(
         construction_age_band=conditioning.construction_age_band,
         main_fuel=conditioning.main_fuel,
         total_floor_area_m2=conditioning.total_floor_area_m2,
+        roof_form=conditioning.roof_form,
     )

scripts/expired_prediction_pairs_harness.py

@@ -46,6 +46,7 @@ from domain.epc_prediction.comparable_properties import (  # noqa: E402
     FLOOR_AREA_TOLERANCE,
     ComparableProperty,
     age_bands_within_one,
+    roof_form_of_construction,
     select_comparables,
 )
 from domain.epc_prediction.epc_prediction import EpcPrediction  # noqa: E402
@@ -210,6 +211,11 @@ def _part0(epc: EpcPropertyData) -> Optional[object]:
     return parts[0] if parts else None
 
 
+def _actual_roof_form(epc: EpcPropertyData) -> Optional[str]:
+    part = _part0(epc)
+    return roof_form_of_construction(getattr(part, "roof_construction", None))
+
+
 def _actual_age_band(epc: EpcPropertyData) -> Optional[object]:
     part = _part0(epc)
     return getattr(part, "construction_age_band", None)
@@ -247,6 +253,7 @@ def simulate_conditioning_ladder(
     age_band: Optional[str],
     main_fuel: Optional[int],
     total_floor_area_m2: Optional[float],
+    roof_form: Optional[str] = None,
     minimum_cohort: int = 5,
 ) -> dict[str, Optional[LadderStep]]:
     """Replay select_comparables' age->fuel->TFA conditioning sequence over the
@@ -266,6 +273,11 @@ def simulate_conditioning_ladder(
         if engaged:
             cohort = matches
 
+    apply(
+        "roof_form",
+        roof_form is not None,
+        [c for c in cohort if _actual_roof_form(c.epc) == roof_form],
+    )
     apply(
         "construction_age_band",
         age_band is not None,
@@ -296,7 +308,7 @@ def format_diagnosis(rows: list[dict[str, object]]) -> str:
     with the newly lodged one (the staleness measurement)."""
     if not rows:
         return ""
-    filters = ("construction_age_band", "main_fuel", "total_floor_area")
+    filters = ("roof_form", "construction_age_band", "main_fuel", "total_floor_area")
     lines = [
         "",
         "## Diagnosis — filter engagement (conditioned arm)",
@@ -321,6 +333,7 @@ def format_diagnosis(rows: list[dict[str, object]]) -> str:
         "property_type",
         "built_form",
         "wall_construction",
+        "roof_form",
         "construction_age_band",
         "main_fuel",
         "tfa_within_band",
@@ -478,6 +491,7 @@ def run(  # pragma: no cover - live IO composition
                     age_band=conditioning.construction_age_band,
                     main_fuel=conditioning.main_fuel,
                     total_floor_area_m2=conditioning.total_floor_area_m2,
+                    roof_form=conditioning.roof_form,
                 )
                 emit_telemetry(
                     {
@@ -504,6 +518,11 @@ def run(  # pragma: no cover - live IO composition
                             if conditioning.built_form is None
                             else conditioning.built_form == actual.built_form
                         ),
+                        "agrees_roof_form": (
+                            None
+                            if conditioning.roof_form is None
+                            else conditioning.roof_form == _actual_roof_form(actual)
+                        ),
                         "agrees_wall_construction": (
                             None
                             if conditioning.wall_construction is None

tests/scripts/test_expired_prediction_pairs_harness.py

@@ -142,6 +142,7 @@ def test_ladder_simulation_skips_unresolved_attributes():
     )
 
     assert steps == {
+        "roof_form": None,
         "construction_age_band": None,
         "main_fuel": None,
         "total_floor_area": None,