Backfill the phantom Lodged Performance on predicted Properties to NULL 🟩

One-time script (dry-run default, --apply in a transaction, idempotent) that
NULLs the four lodged_* columns on every predicted-source baseline row — a
predicted EPC exists and no lodged one does — leaving Effective, the bill block,
and rebaseline_reason intact. Dry-run against the audited DB reports 12,236 rows.
Must run after the FE-owned Drizzle ALTER ... DROP NOT NULL lands.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

scripts/null_predicted_lodged_performance.py

@@ -0,0 +1,111 @@
+"""One-time backfill: NULL the phantom Lodged Performance on predicted Properties.
+
+#1361 Class B: ``PropertyBaselineOrchestrator`` used to read Lodged Performance
+off a predicted Property's neighbour-synthesised EPC, persisting a phantom set of
+``lodged_*`` figures on ``property_baseline_performance`` — a *different
+dwelling's* SAP / band / carbon / Primary Energy Intensity presented as this
+Property's government-register record. The orchestrator no longer does this
+(``lodged`` is ``None`` when ``source_path == "predicted"``, ADR-0004 amendment),
+but the ~12k rows written before the fix still carry the phantom.
+
+This NULLs the four ``lodged_*`` columns on every predicted-source baseline row,
+leaving the **Effective** half, the **bill** block, and ``rebaseline_reason``
+untouched (a predicted Property's Effective Performance is correct — it is a
+first-class modelled output).
+
+A predicted-source Property is identified exactly as the orchestrator's
+``source_path == "predicted"``: it has a **predicted** EPC and **no lodged** EPC.
+Site Notes keep their as-surveyed Lodged Performance and so are excluded — though
+no Site-Notes-sourced EPC exists in ``epc_property`` today, the predicate (a
+predicted EPC, no lodged EPC) holds regardless.
+
+DRY-RUN BY DEFAULT: prints the count it would change and writes nothing. Pass
+``--apply`` to execute inside a transaction. **Idempotent** — only rows whose
+``lodged_sap_score`` is still non-NULL are touched, so a second run is a no-op.
+
+Requires the FE-owned Drizzle ``ALTER ... DROP NOT NULL`` on the four ``lodged_*``
+columns to have landed first; without it the UPDATE to NULL violates the
+constraint.
+"""
+
+from __future__ import annotations
+
+import argparse
+
+from sqlalchemy import Connection, text
+
+from scripts.e2e_common import build_engine, load_env
+
+# A predicted-source baseline row: a predicted EPC exists for the property and no
+# lodged one does (``source_path == "predicted"``). ``lodged_sap_score IS NOT
+# NULL`` makes it idempotent — a row already nulled is skipped on a re-run.
+_PREDICTED_PHANTOM_PREDICATE = """
+    pbp.lodged_sap_score IS NOT NULL
+    AND EXISTS (
+        SELECT 1 FROM epc_property e
+         WHERE e.property_id = pbp.property_id AND e.source = 'predicted'
+    )
+    AND NOT EXISTS (
+        SELECT 1 FROM epc_property e
+         WHERE e.property_id = pbp.property_id AND e.source = 'lodged'
+    )
+"""
+
+_COUNT = text(
+    f"""
+    SELECT count(*) FROM property_baseline_performance pbp
+     WHERE {_PREDICTED_PHANTOM_PREDICATE}
+    """
+)
+
+_NULL_LODGED = text(
+    f"""
+    UPDATE property_baseline_performance AS pbp
+       SET lodged_sap_score = NULL,
+           lodged_epc_band = NULL,
+           lodged_co2_emissions_t_per_yr = NULL,
+           lodged_primary_energy_intensity_kwh_per_m2_yr = NULL
+     WHERE {_PREDICTED_PHANTOM_PREDICATE}
+    """
+)
+
+
+def backfill(conn: Connection, *, apply: bool) -> int:
+    """NULL the four ``lodged_*`` columns on predicted-source baseline rows.
+
+    Returns the number of phantom rows found (those that ``--apply`` would / did
+    null). Reads the count first so the dry-run reports it without writing.
+    """
+    found = conn.execute(_COUNT).scalar() or 0
+    if apply:
+        conn.execute(_NULL_LODGED)
+    return found
+
+
+def main() -> None:
+    load_env()
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="execute the update (default: dry-run, writes nothing)",
+    )
+    args = parser.parse_args()
+
+    engine = build_engine()
+    with engine.begin() as conn:
+        conn.execute(text("SET statement_timeout = 120000"))
+        found = backfill(conn, apply=args.apply)
+
+    verb = "NULLed" if args.apply else "would NULL"
+    print(
+        f"{verb} the Lodged Performance (lodged_* → NULL) on {found} "
+        "predicted-source baseline row(s); Effective / bill / rebaseline_reason "
+        "left intact."
+    )
+    if not args.apply:
+        print("\nDRY-RUN — nothing written. Re-run with --apply to execute.")
+
+
+if __name__ == "__main__":
+    main()

tests/scripts/test_null_predicted_lodged_performance.py

@@ -0,0 +1,120 @@
+"""The one-time backfill nulls a predicted Property's phantom Lodged Performance,
+leaves a lodged Property's intact, and is idempotent (#1361 Class B)."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import Engine
+from sqlmodel import Session
+
+from datatypes.epc.domain.epc import Epc
+from datatypes.epc.domain.epc_property_data import EpcPropertyData
+from datatypes.epc.domain.mapper import EpcPropertyDataMapper
+from domain.property_baseline.performance import Performance
+from domain.property_baseline.property_baseline_performance import (
+    PropertyBaselinePerformance,
+)
+from repositories.epc.epc_postgres_repository import EpcPostgresRepository
+from repositories.property_baseline.property_baseline_postgres_repository import (
+    PropertyBaselinePostgresRepository,
+)
+from scripts.null_predicted_lodged_performance import backfill
+
+_JSON_SAMPLES = Path(__file__).resolve().parents[2] / "backend/epc_api/json_samples"
+
+_PHANTOM_LODGED = Performance(
+    sap_score=14, epc_band=Epc.G, co2_emissions=5.0, primary_energy_intensity=400
+)
+_EFFECTIVE = Performance(
+    sap_score=57, epc_band=Epc.D, co2_emissions=1.5, primary_energy_intensity=160
+)
+
+
+def _epc() -> EpcPropertyData:
+    raw: dict[str, Any] = json.loads(
+        (_JSON_SAMPLES / "RdSAP-Schema-21.0.0" / "epc.json").read_text()
+    )
+    return EpcPropertyDataMapper.from_api_response(raw)
+
+
+def _baseline(lodged: Performance) -> PropertyBaselinePerformance:
+    return PropertyBaselinePerformance(
+        lodged=lodged,
+        effective=_EFFECTIVE,
+        rebaseline_reason="physical_state_changed",
+        space_heating_kwh=4200.0,
+        water_heating_kwh=1600.0,
+    )
+
+
+def _seed(db_engine: Engine) -> None:
+    """A predicted Property (id 1, predicted EPC only, phantom lodged) and a
+    lodged Property (id 2, lodged EPC, real lodged) — both with a baseline row
+    carrying a populated lodged half (the pre-fix state)."""
+    epc = _epc()
+    with Session(db_engine) as session:
+        epc_repo = EpcPostgresRepository(session)
+        epc_repo.save(epc, property_id=1, source="predicted")
+        epc_repo.save(epc, property_id=2, source="lodged")
+        baseline_repo = PropertyBaselinePostgresRepository(session)
+        baseline_repo.save(_baseline(_PHANTOM_LODGED), property_id=1)
+        baseline_repo.save(_baseline(_PHANTOM_LODGED), property_id=2)
+        session.commit()
+
+
+def test_apply_nulls_only_the_predicted_propertys_lodged_half(
+    db_engine: Engine,
+) -> None:
+    # Arrange
+    _seed(db_engine)
+
+    # Act
+    with db_engine.begin() as conn:
+        found = backfill(conn, apply=True)
+
+    # Assert — one phantom found and nulled; the predicted Property loses its
+    # lodged half but keeps its Effective half, while the lodged Property is
+    # untouched.
+    assert found == 1
+    with Session(db_engine) as session:
+        repo = PropertyBaselinePostgresRepository(session)
+        predicted = repo.get_for_property(1)
+        lodged = repo.get_for_property(2)
+    assert predicted is not None
+    assert predicted.lodged is None
+    assert predicted.effective == _EFFECTIVE
+    assert lodged is not None
+    assert lodged.lodged == _PHANTOM_LODGED
+
+
+def test_dry_run_reports_the_phantom_but_writes_nothing(db_engine: Engine) -> None:
+    # Arrange
+    _seed(db_engine)
+
+    # Act
+    with db_engine.begin() as conn:
+        found = backfill(conn, apply=False)
+
+    # Assert — the phantom is counted, but the row is left untouched.
+    assert found == 1
+    with Session(db_engine) as session:
+        predicted = PropertyBaselinePostgresRepository(session).get_for_property(1)
+    assert predicted is not None
+    assert predicted.lodged == _PHANTOM_LODGED
+
+
+def test_re_running_apply_is_a_no_op(db_engine: Engine) -> None:
+    # Arrange — the first apply nulls the phantom.
+    _seed(db_engine)
+    with db_engine.begin() as conn:
+        backfill(conn, apply=True)
+
+    # Act — a second apply finds nothing left to null.
+    with db_engine.begin() as conn:
+        found = backfill(conn, apply=True)
+
+    # Assert
+    assert found == 0