Small fixes

pipeline/transform/merge.py

@@ -1149,8 +1149,9 @@ def _canonical_epc_property_type_expr() -> pl.Expr:
 
 def _construction_year_expr(column: str = "construction_age_band") -> pl.Expr:
     # Use the shared band->midpoint-year mapping so the direct-EPC / listings
-    # path matches join_epc_pp (band midpoint, not lower bound; 'before 1900' and
-    # implausible years -> null). Already-numeric inputs pass through unchanged.
+    # path matches join_epc_pp (band midpoint, not lower bound; 'before 1900' ->
+    # 1890 representative year; implausible years -> null). Already-numeric inputs
+    # pass through unchanged.
     return epc_band_to_year(pl.col(column))
 
 
@@ -1728,7 +1729,7 @@ def _property_match_candidate_frame(wide: pl.LazyFrame) -> pl.DataFrame:
 
 
 def _index_candidates(
-    candidates: pl.DataFrame, postcode_key: str, uprn_key: str
+    candidates: pl.DataFrame, postcode_key: str, uprn_key: str, address_key: str
 ) -> tuple[dict[str, list[dict]], dict[str, dict]]:
     """Index candidate rows for matching, in a single pass over the frame.
 
@@ -1736,17 +1737,41 @@ def _index_candidates(
     fuzzy street-address match; the UPRN index drives the exact match and is
     postcode-independent, so it still resolves when a listing's postcode is
     slightly off.
+
+    The EPC register's UPRN is NOT unique: a single building/parent UPRN fans
+    across many distinct flats (up to 58 distinct (address, postcode) rows in
+    the 2026-06 data; ~9k UPRNs collide, touching ~20k epc_pp rows). Such a
+    UPRN cannot serve as a 1:1 exact-match key — it would mis-link a listing to
+    one arbitrary flat — so any UPRN that resolves to more than one distinct
+    ``(postcode_key, address_key)`` identity is dropped from ``uprn_index``;
+    those listings fall back to the fuzzy street-address matcher, which
+    disambiguates the specific flat. A UPRN repeated for the SAME identity
+    (one genuine property) is kept.
     """
     buckets: dict[str, list[dict]] = {}
-    uprn_index: dict[str, dict] = {}
+    # uprn -> first_row, plus the identity of that first row; a uprn drops out
+    # the moment a second distinct (postcode, address) identity appears.
+    uprn_first: dict[str, dict] = {}
+    uprn_identity: dict[str, tuple] = {}
+    uprn_dropped: set[str] = set()
     for row in candidates.iter_rows(named=True):
         postcode = row.get(postcode_key)
         if postcode:
             buckets.setdefault(postcode, []).append(row)
         uprn = _normalize_uprn(row.get(uprn_key))
-        if uprn and uprn not in uprn_index:
-            uprn_index[uprn] = row
-    return buckets, uprn_index
+        if not uprn or uprn in uprn_dropped:
+            continue
+        identity = (row.get(postcode_key), row.get(address_key))
+        if uprn not in uprn_first:
+            uprn_first[uprn] = row
+            uprn_identity[uprn] = identity
+        elif identity != uprn_identity[uprn]:
+            # Same UPRN, different (postcode, address): a non-unique parent/
+            # building UPRN. Remove it so it cannot act as a 1:1 key.
+            del uprn_first[uprn]
+            del uprn_identity[uprn]
+            uprn_dropped.add(uprn)
+    return buckets, uprn_first
 
 
 def _best_listing_property_candidate(
@@ -1783,7 +1808,10 @@ def _match_listing_properties(
         return _empty_listing_property_matches()
 
     buckets, uprn_index = _index_candidates(
-        property_candidates, "_property_match_postcode", "uprn"
+        property_candidates,
+        "_property_match_postcode",
+        "uprn",
+        "_property_match_address",
     )
     best_matches = []
     for listing in listing_matches.iter_rows(named=True):
@@ -1909,7 +1937,10 @@ def _match_direct_epc(
         return _empty_direct_epc_matches()
 
     buckets, uprn_index = _index_candidates(
-        epc_candidates, "_direct_epc_match_postcode", "_direct_epc_uprn"
+        epc_candidates,
+        "_direct_epc_match_postcode",
+        "_direct_epc_uprn",
+        "_direct_epc_match_address",
     )
     street_index, noise_tokens = _index_epc_streets(epc_candidates)
     street_score_cache: dict[tuple[str, str], list[tuple[int, str]]] = {}
@@ -2214,6 +2245,25 @@ class _BuildResult:
     listings: pl.DataFrame | None = None
 
 
+# Property-level Yes/No flags default to "No" once all EPC + listings overlays
+# have been coalesced. A property with no EPC match has no recorded social
+# tenure / listed status, which is "No", not "unknown": join_epc_pp fills
+# was_council_house with "No" only for EPC-matched rows (it runs before the
+# fuzzy join), so without this the ~32% of EPC-unmatched properties would
+# publish null instead of "No".
+_PROPERTY_LEVEL_NO_DEFAULT_COLUMNS = (LISTED_BUILDING_FEATURE, "was_council_house")
+
+
+def _fill_property_level_no_defaults(frame: pl.LazyFrame) -> pl.LazyFrame:
+    """Default the property-level Yes/No flag columns to "No" where null."""
+    return frame.with_columns(
+        *(
+            pl.col(column).fill_null("No")
+            for column in _PROPERTY_LEVEL_NO_DEFAULT_COLUMNS
+        )
+    )
+
+
 def _build(
     epc_pp_path: Path,
     arcgis_path: Path,
@@ -2303,7 +2353,12 @@ def _build(
         )
         wide = _filter_to_active_english_postcodes(wide, active_postcodes)
 
-    wide = wide.with_columns(pl.col(LISTED_BUILDING_FEATURE).fill_null("No"))
+    # Default property-level Yes/No flags to "No" here: after the listings
+    # overlay coalesce (so a directly-matched "Yes" survives) and before the
+    # rename to "Former council house". This is the single place the FINAL
+    # was_council_house column (null for ~32% EPC-unmatched rows) gets its "No"
+    # default, alongside Listed building.
+    wide = _fill_property_level_no_defaults(wide)
 
     # NSPL Feb 2026 renamed geographic code columns to {field}{year}cd.
     # `_active_english_postcode_area` aliases them back to the short canonical