Guard property_type to its leading dwelling-type token

The landlord property-type description is a "<dwelling type>: <built form>: <floor>"
split whose leading token IS the dwelling type; the built-form tail is not. The
LLM occasionally over-read the tail and flipped the type — a handful of
"Bungalow: EndTerrace" / "Bungalow: MidTerrace" dwellings were stored as House.

Adds property_type_guard (claims the recognised leading token: House / Bungalow /
Flat / Maisonette / Park home; defers unrecognised phrasings to the LLM) and wires
property_type through a GuardedColumnClassifier, so the built-form tail can never
flip the type and the live path is deterministic.

Applied the scoped backfill to portfolio 796 (Hyde): 3 rows corrected from House
back to Bungalow. No enum migration needed — the targets are original members.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jun-te Kim 2026-07-01 17:24:37 +00:00
parent de9159f9b8
commit 88bae2166f
4 changed files with 216 additions and 2 deletions

View file

@ -15,6 +15,7 @@ from domain.epc.property_overrides.main_fuel_type import MainFuelType
from domain.epc.property_overrides.main_heating_system_type import MainHeatingSystemType
from domain.epc.property_overrides.main_heating_guard import main_heating_guard
from domain.epc.property_overrides.property_type import PropertyType
from domain.epc.property_overrides.property_type_guard import property_type_guard
from domain.epc.property_overrides.roof_type import RoofType
from domain.epc.property_overrides.roof_party_ceiling_guard import (
roof_party_ceiling_guard,
@ -91,8 +92,15 @@ def _build_columns(
"property_type": lambda src: ClassifiableColumn(
name="property_type",
source_column=src,
classifier=ChatGptColumnClassifier(
chat_gpt, PropertyType, PropertyType.UNKNOWN
# The dwelling type is the leading token of the "<type>: <built form>"
# split; the deterministic guard claims it so the built-form tail can't
# flip the type (the LLM read "Bungalow: EndTerrace" as House), and the
# LLM handles varied phrasings (#1376).
classifier=GuardedColumnClassifier(
guard=property_type_guard,
fallback=ChatGptColumnClassifier(
chat_gpt, PropertyType, PropertyType.UNKNOWN
),
),
repo=LandlordOverridesRepository[PropertyType](
session, LandlordPropertyTypeOverrideRow

View file

@ -0,0 +1,32 @@
from __future__ import annotations
from typing import Optional
from domain.epc.property_overrides.property_type import PropertyType
# The landlord property-type description is a "<dwelling type>: <built form>: …"
# split (e.g. "Bungalow: EndTerrace"). The dwelling type is the leading token and
# is independent of the built-form tail — which the LLM sometimes over-reads,
# mislabelling a "Bungalow: EndTerrace" as House (#1376).
_LEADING_TYPE: dict[str, PropertyType] = {
"house": PropertyType.HOUSE,
"bungalow": PropertyType.BUNGALOW,
"flat": PropertyType.FLAT,
"maisonette": PropertyType.MAISONETTE,
"park home": PropertyType.PARK_HOME,
}
def property_type_guard(description: str) -> Optional[PropertyType]:
"""Deterministically resolve the dwelling type from the leading token of a
landlord property-type description.
The description is a structured ``"<dwelling type>: <built form>: <floor>"``
split whose first token is the dwelling type; the tail (EndTerrace, Mid Floor,
) is built form, not type. This guard claims the recognised leading tokens so
the built-form tail can never flip the type (the LLM read "Bungalow: EndTerrace"
as House). Returns ``None`` for an unrecognised leading token, so varied
phrasings still reach the LLM classifier via ``GuardedColumnClassifier``.
"""
leading = description.split(":", 1)[0].strip().lower()
return _LEADING_TYPE.get(leading)

View file

@ -0,0 +1,125 @@
"""Backfill property_type overrides the LLM mislabelled off their leading dwelling
type (e.g. "Bungalow: EndTerrace" stored as House).
The property-type description is a "<dwelling type>: <built form>: <floor>" split
whose leading token IS the dwelling type; the built-form tail is not. The LLM
occasionally over-read the tail and flipped the type. ``property_type_guard`` now
resolves the leading token deterministically, so this fixes the rows written
before that guard.
Uses the SAME guard as the live path, so the backfill and the classifier cannot
drift. SCOPED TO ONE PORTFOLIO (``--portfolio``, default 796 = Hyde) and only
touches ``property_overrides.override_value`` (TEXT). No enum migration is needed
the target values (House / Bungalow / Flat / Maisonette) are original members
but the description-keyed classifier cache is left alone here (a property-level
data fix, not a cache rewrite).
DRY-RUN BY DEFAULT: prints what it would change and writes nothing. Pass
``--apply`` to execute inside a transaction; it also writes an audit CSV of every
row changed (property_id, uprn, old value, new value) so the change is reversible.
Idempotent only rows whose stored value differs from the guard's target member
are touched.
python -m scripts.lisasrequest.reclassify_property_type # dry run
python -m scripts.lisasrequest.reclassify_property_type --apply # write
"""
from __future__ import annotations
import argparse
import csv
import sys
from collections import Counter
from pathlib import Path
from sqlalchemy import text
_REPO_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(_REPO_ROOT))
from domain.epc.property_overrides.property_type_guard import ( # noqa: E402
property_type_guard,
)
from scripts.e2e_common import ENV_PATH, build_engine, load_env # noqa: E402
_SELECT = text(
"""
SELECT po.property_id, pr.uprn,
po.original_spreadsheet_description AS description,
po.override_value AS value
FROM property_overrides po
JOIN property pr ON pr.id = po.property_id
WHERE po.portfolio_id = :portfolio
AND po.override_component = 'property_type'
"""
)
_UPDATE = text(
"""
UPDATE property_overrides
SET override_value = :new_value
WHERE portfolio_id = :portfolio
AND override_component = 'property_type'
AND property_id = :property_id
AND override_value <> :new_value
"""
)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--portfolio", type=int, default=796)
parser.add_argument(
"--apply",
action="store_true",
help="execute the updates (default: dry-run, writes nothing)",
)
args = parser.parse_args()
load_env(ENV_PATH)
engine = build_engine()
audit: list[tuple[int, object, str, str, str]] = [] # pid, uprn, descr, old, new
tally: Counter[str] = Counter()
with engine.begin() as conn:
conn.execute(text("SET statement_timeout = 120000"))
for property_id, uprn, description, value in conn.execute(
_SELECT, {"portfolio": args.portfolio}
):
member = property_type_guard(description or "")
if member is None or value == member.value:
continue
tally[f"{description!r}: {value!r} -> {member.value!r}"] += 1
audit.append((property_id, uprn, description, value, member.value))
if args.apply:
conn.execute(
_UPDATE,
{
"portfolio": args.portfolio,
"property_id": property_id,
"new_value": member.value,
},
)
if not args.apply:
conn.rollback()
verb = "re-classified" if args.apply else "would re-classify"
print(f"portfolio {args.portfolio}: {verb} {len(audit)} property_type override row(s)")
for change, n in tally.most_common():
print(f" {n:5d} {change}")
if args.apply and audit:
out = _REPO_ROOT / "scripts" / "lisasrequest" / (
f"reclassify_property_type_{args.portfolio}_audit.csv"
)
with out.open("w", newline="") as fh:
w = csv.writer(fh)
w.writerow(["property_id", "uprn", "description", "old_value", "new_value"])
w.writerows(audit)
print(f"\naudit trail: {out}")
if not args.apply:
print("\nDRY-RUN — nothing written. Re-run with --apply to execute.")
return 0
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -0,0 +1,49 @@
from __future__ import annotations
import pytest
from domain.epc.property_overrides.property_type import PropertyType
from domain.epc.property_overrides.property_type_guard import property_type_guard
@pytest.mark.parametrize(
("description", "expected"),
[
# The bug: the LLM read the "EndTerrace/MidTerrace" tail as house-like and
# mislabelled a handful of bungalows as House. The dwelling type is the
# leading token, independent of the built-form tail.
("Bungalow: EndTerrace", PropertyType.BUNGALOW),
("Bungalow: MidTerrace", PropertyType.BUNGALOW),
("House: SemiDetached", PropertyType.HOUSE),
("Flat: Mid Terrace: Mid Floor", PropertyType.FLAT),
("Maisonette: Mid Terrace: Top Floor", PropertyType.MAISONETTE),
# Bare type with no built-form tail.
("Flat", PropertyType.FLAT),
("Maisonette", PropertyType.MAISONETTE),
],
)
def test_guard_resolves_the_leading_dwelling_type(
description: str, expected: PropertyType
) -> None:
# Act
result = property_type_guard(description)
# Assert
assert result is expected
@pytest.mark.parametrize(
"description",
[
# No recognised leading dwelling type — left to the LLM classifier.
"Studio apartment",
"Converted barn",
"",
],
)
def test_guard_defers_unrecognised_descriptions_to_the_llm(description: str) -> None:
# Act
result = property_type_guard(description)
# Assert
assert result is None