new demo mode & tenure

pipeline/transform/join_epc_pp.py

@@ -105,6 +105,35 @@ def epc_band_to_year(band: pl.Expr) -> pl.Expr:
     )
 
 
+# Coarse occupancy statuses derived from the raw EPC TENURE field, in the order
+# the timeline reads them. EPC lodges tenure as one of "Owner-occupied",
+# "Rented (private)", "Rented (social)" (plus blanks / "unknown" /
+# "Not defined - ..." for new dwellings). Anything unrecognised maps to null
+# (unknown) so it neither appears on the timeline nor breaks the change chain.
+TENURE_OWNER_OCCUPIED = "Owner-occupied"
+TENURE_RENTED_PRIVATE = "Rented (private)"
+TENURE_RENTED_SOCIAL = "Rented (social)"
+
+
+def tenure_status(tenure: pl.Expr) -> pl.Expr:
+    """Normalise the raw EPC TENURE field to a coarse occupancy status.
+
+    Matching is case-insensitive and order-sensitive: "social" is tested before
+    the generic "rent" so "Rented (social)" lands on the social bucket rather
+    than the private one. Null/blank/unrecognised tenures yield null.
+    """
+    lowered = tenure.str.to_lowercase()
+    return (
+        pl.when(lowered.str.contains("owner"))
+        .then(pl.lit(TENURE_OWNER_OCCUPIED))
+        .when(lowered.str.contains("social"))
+        .then(pl.lit(TENURE_RENTED_SOCIAL))
+        .when(lowered.str.contains("rent"))
+        .then(pl.lit(TENURE_RENTED_PRIVATE))
+        .otherwise(pl.lit(None, dtype=pl.String))
+    )
+
+
 EPC_SOURCE_COLUMNS = [
     "address",
     # The individual lines behind `address` (= address1+2+3): address2/3
@@ -484,6 +513,53 @@ def _run(epc_path: Path, price_paid_path: Path, output_path: Path, temp_dir: Pat
     print(f"Renovation events: {events.height} properties with events")
     print(event_counts)
 
+    # Tenure-history fork: a chronological timeline of owner-occupied <-> rented
+    # transitions, derived from the per-certificate EPC TENURE field. Each
+    # certificate is one tenure observation; we keep only the change points so
+    # the property history can show *when* a home was let out vs lived in.
+    #
+    # Emission rule (walking certificates oldest-first, ignoring unknown-tenure
+    # ones so they neither appear nor break the chain):
+    #   - the first known status is emitted only when it is a rental — an
+    #     owner-occupied baseline is the unremarkable default for a property
+    #     that has changed hands and would only clutter the timeline;
+    #   - every later certificate is emitted when its status differs from the
+    #     previous known one (this is what surfaces the return to owner-occupied
+    #     that closes a rental period).
+    # Eager like the events/social forks: shift().over() does not run under the
+    # streaming sink used by fuzzy_join_on_postcode.
+    tenure_events = (
+        epc_base.with_columns(tenure_status(pl.col("tenure")).alias("_tenure_status"))
+        .filter(
+            pl.col("inspection_date").is_not_null()
+            & pl.col("_tenure_status").is_not_null()
+        )
+        .sort("inspection_date")
+        .with_columns(
+            pl.col("_tenure_status")
+            .shift(1)
+            .over("_epc_match_address", "_epc_match_postcode")
+            .alias("_prev_tenure_status"),
+        )
+        .filter(
+            pl.when(pl.col("_prev_tenure_status").is_null())
+            .then(pl.col("_tenure_status") != pl.lit(TENURE_OWNER_OCCUPIED))
+            .otherwise(pl.col("_tenure_status") != pl.col("_prev_tenure_status"))
+        )
+        .with_columns(
+            pl.col("inspection_date").dt.year().cast(pl.Int32).alias("_event_year"),
+        )
+        .group_by("_epc_match_address", "_epc_match_postcode")
+        .agg(
+            pl.struct(
+                pl.col("_event_year").alias("year"),
+                pl.col("_tenure_status").alias("status"),
+            ).alias("tenure_history"),
+        )
+        .collect()
+    )
+    print(f"Tenure timelines: {tenure_events.height} properties with tenure changes")
+
     # Social tenure fork: flag properties that were ever social housing
     social_tenure = (
         epc_base.filter(pl.col("tenure").str.to_lowercase().str.contains("social"))
@@ -494,13 +570,18 @@ def _run(epc_path: Path, price_paid_path: Path, output_path: Path, temp_dir: Pat
     )
     print(f"Former council houses (EPC social tenure): {social_tenure.height}")
 
-    # Left-join events and social tenure back onto dedup EPC
+    # Left-join events, tenure history and social tenure back onto dedup EPC
     epc = (
         epc.join(
             events.lazy(),
             on=["_epc_match_address", "_epc_match_postcode"],
             how="left",
         )
+        .join(
+            tenure_events.lazy(),
+            on=["_epc_match_address", "_epc_match_postcode"],
+            how="left",
+        )
         .join(
             social_tenure.lazy(),
             on=["_epc_match_address", "_epc_match_postcode"],
@@ -600,9 +681,7 @@ def _run(epc_path: Path, price_paid_path: Path, output_path: Path, temp_dir: Pat
     # flags are computed here and joined back into the lazy stream.
     outliers = flag_price_outliers(
         price_paid_base.filter(value_ok)
-        .select(
-            "_pp_group_address", "_pp_group_postcode", "date_of_transfer", "price"
-        )
+        .select("_pp_group_address", "_pp_group_postcode", "date_of_transfer", "price")
         .collect(engine="streaming")
     )
     print(f"Implausible consecutive-sale price jumps flagged: {outliers.height}")
@@ -665,7 +744,15 @@ def _run(epc_path: Path, price_paid_path: Path, output_path: Path, temp_dir: Pat
             right_postcode_col="epc_postcode",
             left_variant_cols=["pp_address_loc"],
             right_variant_cols=["epc_address_a1", "epc_address_a12"],
+            # Include EPC-only dwellings (an energy certificate but no Land
+            # Registry sale) so the property universe is "any dwelling we hold a
+            # record for", not just ones that have sold since 1995.
+            keep_unmatched_right=True,
         )
+        # EPC-only rows have no price-paid postcode; fall back to the EPC postcode
+        # so every row carries one (the active-English-postcode filter and the
+        # postcode->coordinates join downstream both key on it).
+        .with_columns(pl.coalesce("postcode", "epc_postcode").alias("postcode"))
         .drop("epc_postcode")
         # Audit trail: keep the fuzzy-match confidence (100 = exact address
         # match) in the published output; null means no EPC match.
@@ -676,10 +763,17 @@ def _run(epc_path: Path, price_paid_path: Path, output_path: Path, temp_dir: Pat
     matched = joined.filter(
         pl.col("epc_address").is_not_null() & pl.col("pp_address").is_not_null()
     )
+    pp_only = joined.filter(
+        pl.col("pp_address").is_not_null() & pl.col("epc_address").is_null()
+    )
+    epc_only = joined.filter(pl.col("pp_address").is_null())
     total = joined.height
     print(f"Unique properties: {total}")
-    print(f"Matched: {matched.height} ({100 * matched.height / total:.1f}%)")
-    print(f"Unmatched: {total - matched.height}")
+    print(
+        f"Matched (sale + EPC): {matched.height} ({100 * matched.height / total:.1f}%)"
+    )
+    print(f"Sale only (no EPC): {pp_only.height}")
+    print(f"EPC only (never sold): {epc_only.height}")
 
     # For new-builds (old_new == "Y"), use the first transaction date year as
     # the exact construction date; otherwise fall back to the EPC age band.
@@ -689,10 +783,23 @@ def _run(epc_path: Path, price_paid_path: Path, output_path: Path, temp_dir: Pat
     )
     is_new_build = pl.col("old_new") == "Y"
 
+    # A dwelling cannot have been built after it was first sold, yet the EPC age
+    # band (a coarse range midpoint) sometimes lands later than the earliest
+    # Land Registry transfer. Cap the band-derived year at the first transfer
+    # year so the published build year is never after the first known sale. The
+    # cap only fires when both years exist: the transfer year alone is an upper
+    # bound, not an estimate, so we never fabricate a build year from it for a
+    # non-new-build that lacks an EPC band.
+    capped_band_year = (
+        pl.when(transfer_year.is_not_null() & (transfer_year < epc_band_year))
+        .then(transfer_year)
+        .otherwise(epc_band_year)
+    )
+
     joined = joined.with_columns(
         pl.when(is_new_build & transfer_year.is_not_null())
         .then(transfer_year)
-        .otherwise(epc_band_year)
+        .otherwise(capped_band_year)
         .alias("construction_age_band"),
         pl.when(is_new_build & transfer_year.is_not_null())
         .then(pl.lit(0, dtype=pl.UInt8))