junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-12 13:29:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1487 from Hestia-Homes/feature/uprn-confirmation-before-finalise

Browse source 
Confirm UPRNs before finalise: withhold ambiguous address2uprn matches (ADR-0057)
This commit is contained in:
Jun-te Kim 2026-07-07 17:12:14 +01:00 • committed by GitHub
commit c9b5fc095b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 418 additions and 45 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

123
backend/address2UPRN/main.py
View file

@ -14,8 +14,13 @@ from utils.s3 import (
)
from datetime import datetime
from datatypes.address_match import UprnMatch
from backend.utils.addressMatch import AddressMatch
from backend.address2UPRN.scoring import all_uprns_match, rank_address_similarity
from backend.address2UPRN.scoring import (
all_uprns_match,
rank_address_similarity,
resolve_group_ambiguity,
)
from infrastructure.epc_client.epc_client_service import EpcClientService
from repositories.historic_epc.historic_epc_resolver import HistoricEpcResolver
from repositories.historic_epc.historic_epc_s3_repository import (
@ -34,32 +39,39 @@ def get_epc_data_with_postcode(postcode: str) -> pd.DataFrame:
service = EpcClientService(auth_token=token)
results = service.search_by_postcode(postcode)
return pd.DataFrame(
[{"address": r.address_line_1, "uprn": r.uprn} for r in results]
[
{
"address": r.address_line_1,
"uprn": r.uprn,
"certificate_number": r.certificate_number,
}
for r in results
]
)
def get_uprn_from_historic_epc(
user_inputed_address: str,
postcode: str,
) -> Optional[tuple[str, str, float]]:
) -> Optional[UprnMatch]:
"""Resolve a UPRN via historic EPC S3 data.
Returns (uprn, address, lexiscore) when the historic dataset agrees on a
single rank-1 UPRN, None otherwise (no stored data, zero score, or
ambiguous top rank). The score gate is `unambiguous_uprn`'s own (score > 0);
the 0.7 heuristic used for the new-EPC source isn't applied here because
historic addresses use a more verbose format that systematically depresses
lexiscores.
Returns (uprn, address, lexiscore, certificate_number) when the historic
dataset agrees on a single rank-1 UPRN, None otherwise (no stored data,
zero score, or ambiguous top rank). The score gate is `unambiguous_uprn`'s
own (score > 0); the 0.7 heuristic used for the new-EPC source isn't
applied here because historic addresses use a more verbose format that
systematically depresses lexiscores.
"""
repo = HistoricEpcS3Repository.with_default_s3_client()
return HistoricEpcResolver(repo).resolve_uprn(user_inputed_address, postcode)
def get_uprn_with_epc_df(
def get_uprn_from_epc_df(
user_inputed_address: str,
epc_df: pd.DataFrame,
verbose: bool = False,
) -> Optional[str | tuple[str, str, float]]:
) -> Optional[str | UprnMatch]:
"""
Return uprn (str) using a pre-fetched EPC dataframe.
This avoids calling the API multiple times for the same postcode.
@ -88,6 +100,7 @@ def get_uprn_with_epc_df(
address = top_rank_df["address"].values[0]
score = float(top_rank_df["lexiscore"].values[0])
certificate_number = top_rank_df["certificate_number"].values[0]
logger.info(f"Address found to be: {address}, with lexiscore {score}")
# Safe to return the agreed UPRN
@ -98,7 +111,7 @@ def get_uprn_with_epc_df(
return None
if verbose:
return (found_uprn, address, score)
return UprnMatch(found_uprn, address, score, certificate_number)
else:
return found_uprn
@ -116,11 +129,11 @@ def get_uprn(
back to the historic EPC dataset on S3.
For processing multiple addresses in the same postcode, use
get_uprn_with_epc_df instead.
get_uprn_from_epc_df instead.
"""
df = get_epc_data_with_postcode(postcode=postcode)
result: Optional[tuple[str, str, float]] = get_uprn_with_epc_df(
result: Optional[str | UprnMatch] = get_uprn_from_epc_df(
user_inputed_address=user_inputed_address,
epc_df=df,
verbose=True,
@ -393,6 +406,8 @@ def handler(event, context, local=False):
"address2uprn_uprn": "invalid postcode",
"address2uprn_address": "invalid postcode",
"address2uprn_lexiscore": "invalid postcode",
"address2uprn_certificate_number": "invalid postcode",
"address2uprn_status": "invalid_postcode",
}
)
continue
@ -409,7 +424,13 @@ def handler(event, context, local=False):
)
continue
# Process each address in this postcode with the same EPC data
# Match each address in this postcode against the same EPC data,
# collecting per-row results first. Cross-row ambiguity is resolved
# for the whole group afterwards (ADR-0057) — a UPRN that is the
# best match for two *different* addresses is withheld, so distinct
# addresses are never coerced onto one UPRN (which the identity
# insert would then silently merge).
group_records: list[dict] = []
for row in postcode_rows:
try:
# Concatenate Address columns directly
@ -428,7 +449,7 @@ def handler(event, context, local=False):
continue
# Get UPRN using the pre-fetched EPC data with all return options
result: Optional[tuple[str, str, float]] = get_uprn_with_epc_df(
result: Optional[UprnMatch] = get_uprn_from_epc_df(
user_inputed_address=address2uprn_user_input,
epc_df=epc_df,
verbose=True,
@ -451,31 +472,40 @@ def handler(event, context, local=False):
f"Historic EPC matched {address2uprn_user_input} in {postcode}"
)
norm_address = AddressMatch.normalise_address(
address2uprn_user_input
)
# Parse result tuple if successful
if result:
uprn, found_address, score = result
uprn, found_address, score, certificate_number = result
logger.info(
f"Found UPRN for {address2uprn_user_input} in {postcode}: {uprn} (score: {score})"
)
results_data.append(
group_records.append(
{
**row, # Include all original data
"address2uprn_uprn": uprn,
"address2uprn_address": found_address,
"address2uprn_lexiscore": score,
"row": row,
"uprn": uprn,
"address": found_address,
"lexiscore": score,
"certificate_number": certificate_number,
"norm_address": norm_address,
"error": None,
}
)
else:
logger.warning(
f"No UPRN found for {address2uprn_user_input} in {postcode}"
)
results_data.append(
group_records.append(
{
**row, # Include all original data
"address2uprn_uprn": None,
"address2uprn_address": None,
"address2uprn_lexiscore": None,
"row": row,
"uprn": None,
"address": None,
"lexiscore": None,
"certificate_number": None,
"norm_address": norm_address,
"error": None,
}
)
@ -483,18 +513,43 @@ def handler(event, context, local=False):
logger.error(
f"Error processing address {row.get('address2uprn_user_input', 'unknown')}: {e}"
)
# Still add the row with error markers
results_data.append(
# Still record the row with error markers
group_records.append(
{
**row,
"address2uprn_uprn": None,
"address2uprn_address": None,
"address2uprn_lexiscore": None,
"row": row,
"uprn": None,
"address": None,
"lexiscore": None,
"certificate_number": None,
"norm_address": "",
"error": str(e),
}
)
continue
# Resolve cross-row ambiguity for the whole postcode group, then
# emit. A withheld (ambiguous) UPRN drops to None but keeps its
# lexiscore so the confirmation page can triage it (ADR-0057).
decisions = resolve_group_ambiguity(
[(rec["uprn"], rec["norm_address"]) for rec in group_records]
)
for rec, (final_uprn, status) in zip(group_records, decisions):
emitted = {
**rec["row"], # Include all original data
"address2uprn_uprn": final_uprn,
"address2uprn_address": rec["address"] if final_uprn else None,
"address2uprn_lexiscore": rec["lexiscore"],
"address2uprn_certificate_number": (
rec["certificate_number"] if final_uprn else None
),
"address2uprn_status": (
"error" if rec["error"] is not None else status
),
}
if rec["error"] is not None:
emitted["error"] = rec["error"]
results_data.append(emitted)
# Create results DataFrame
result_df = pd.DataFrame(results_data)

37
backend/address2UPRN/scoring.py
View file

@ -1,8 +1,45 @@
from collections import defaultdict
from typing import NamedTuple, Optional
import pandas as pd
from backend.utils.addressMatch import AddressMatch
class GroupDecision(NamedTuple):
"""One row's outcome after cross-row ambiguity resolution (ADR-0057)."""
uprn: Optional[str]
status: str # "matched" | "ambiguous_duplicate" | "unmatched"
def resolve_group_ambiguity(
matches: list[tuple[Optional[str], str]],
) -> list[GroupDecision]:
"""Resolve cross-row UPRN ambiguity within one postcode group (ADR-0057).
``matches`` is ``(uprn, normalised_address)`` per row. A UPRN that is the
best match for two rows with *different* normalised addresses is withheld
on both (a coarse EPC record absorbing several real addresses, e.g. flats in
a block); a UPRN shared only by identical addresses is a genuine re-listing
and kept. Returns a ``GroupDecision`` per row, in input order.
"""
distinct_addresses: dict[str, set[str]] = defaultdict(set)
for uprn, norm_address in matches:
if uprn:
distinct_addresses[uprn].add(norm_address)
resolved: list[GroupDecision] = []
for uprn, _norm_address in matches:
if not uprn:
resolved.append(GroupDecision(None, "unmatched"))
elif len(distinct_addresses[uprn]) > 1:
resolved.append(GroupDecision(None, "ambiguous_duplicate"))
else:
resolved.append(GroupDecision(uprn, "matched"))
return resolved
def all_uprns_match(
df: pd.DataFrame,
target_uprn: str,

13
datatypes/address_match.py Normal file
View file

@ -0,0 +1,13 @@
from __future__ import annotations
from typing import NamedTuple, Optional
class UprnMatch(NamedTuple):
"""A confident address→UPRN match from either EPC source (new API or
historic). Tuple-compatible, so existing unpacking keeps working."""
uprn: str
address: str
lexiscore: float
certificate_number: Optional[str]

80
docs/adr/0057-uprn-confirmation-precedes-finalise.md Normal file
View file

@ -0,0 +1,80 @@
# UPRN confirmation precedes finalise; the bulk-upload review is one two-tab page and the portfolio "unmatched" tab is retired
## Status
proposed
## Context
The bulk-upload pipeline is a chain of SQS-driven Lambda stages:
`address2uprn → landlord-overrides → combiner → (review) → finaliser`.
`address2uprn` matches each input row independently against per-postcode EPC
candidates, keeping the best candidate above a `lexiscore >= 0.7` floor. There
is **no cross-row uniqueness guard**: nothing stops one UPRN being the best
match for two *different* addresses. The textbook case is a coarse EPC record
`42 Moreton Road` (no flat token) that `Flat 1, 42 Moreton Road` and
`Flat 2, 42 Moreton Road` both clear — the building-number guard passes and the
flat guard never fires, so both distinct flats collapse onto one UPRN.
Downstream this is corrosive, not just noisy:
- The `property` identity insert is `on_conflict_do_nothing` on the partial
unique index `(portfolio_id, uprn) WHERE uprn IS NOT NULL`. Two distinct
addresses carrying the same UPRN therefore **merge into a single property** —
one address silently loses its identity.
- `_build_overrides` (the finaliser) keys `property_overrides` on
`(property_id, override_component, building_part)`. Both merged rows emit the
same tuple, so the single `INSERT ... ON CONFLICT DO UPDATE` raises
`CardinalityViolation` ("cannot affect row a second time").
The classifier and finaliser were built assuming a UPRN is always intact and
unique; it is not. Separately, a *post-onboarding* portfolio-level "Unmatched
properties" tab let users fix no-UPRN properties after the fact — too late (the
identity merge has already happened) and in a different place from the rest of
review.
## Decision
1. **`address2uprn` emits a UPRN only when confident *and* unambiguous.** Within
a postcode group, a UPRN that is the best match for two or more rows whose
*normalised* input addresses differ is **withheld** (dropped to null) on
every such row; a UPRN shared only by rows with the *same* normalised address
is a genuine re-listing and is kept. `address2uprn` emits an
`address2uprn_status` column (`matched | ambiguous_duplicate | unmatched |
invalid_postcode | error`); withheld rows keep their `lexiscore` for triage.
A `lexiscore` review band `[0.7, X)` that also withholds shaky matches is a
follow-on once `X` is calibrated (see Consequences).
2. **One pre-finalise review page, two tabs**, styled like the retired unmatched
tab:
- **Addresses** — the flagged rows (`unmatched` / `ambiguous_duplicate` /
low-score); the user resolves each via OS Places (reusing the existing
`MatchAddress` / `searchAddresses` / `useAssignUprn` flow + `postcode_search`
cache), producing a valid residential UPRN, or explicitly marks *no-UPRN*.
- **Classification** — the existing description→value verify gate, semantics
unchanged (every `Unknown` must be mapped; the finaliser still fails loud on
an unresolved value).
Finalise is enabled only when **both** tabs are clear.
3. The finaliser runs only after confirmation, so every row carries a confirmed
UPRN **or** an explicit no-UPRN (a first-class terminal state: `property`
identity with `uprn NULL`, no `property_overrides`, building-passport shows
the "not matched yet" card).
4. **Retire the portfolio-level "Unmatched properties" tab** — properties now
arrive pre-matched, so post-onboarding UPRN cleanup is unnecessary.
5. `PropertyOverridePostgresRepository.upsert_all` keeps a **defensive
last-write-wins dedup** as a backstop for the single-statement invariant, but
it is no longer the primary fix.
## Consequences
- Distinct addresses can no longer silently merge into one property; the
`CardinalityViolation` can no longer arise from real data.
- Onboarding reaching "complete" now implies confirmed identities, so portfolio
pages need no unmatched tab.
- The ambiguous/low-score subset adds bounded user work at review time; confident
rows pass untouched.
- The `lexiscore` review-band threshold `X` needs calibrating against a real
portfolio's score distribution before it is switched on; until then only the
ambiguous-duplicate rule (which needs no threshold) is active.
- Migration is safe stage-by-stage: withheld UPRNs are caught by the *existing*
unmatched tab until the two-tab page ships, and the tab is removed only after
it does — so an ambiguous property always has somewhere to be resolved.

12
repositories/historic_epc/historic_epc_resolver.py
View file

@ -2,6 +2,7 @@ from __future__ import annotations
from typing import Optional
from datatypes.address_match import UprnMatch
from datatypes.epc.domain.historic_epc import HistoricEpc
from datatypes.epc.domain.historic_epc_matching import (
HistoricEpcMatches,
@ -47,11 +48,10 @@ class HistoricEpcResolver:
return None
return max(certs, key=lambda r: r.lodgement_date)
def resolve_uprn(
self, user_address: str, postcode: str
) -> Optional[tuple[str, str, float]]:
"""``(uprn, matched_address, lexiscore)`` for an unambiguous rank-1
match, else None (no data / ambiguous tie / zero score)."""
def resolve_uprn(self, user_address: str, postcode: str) -> Optional[UprnMatch]:
"""A ``UprnMatch`` for an unambiguous rank-1 match, else None (no data /
ambiguous tie / zero score). ``certificate_number`` is the historic
dataset's ``lmk_key``."""
matches: HistoricEpcMatches = self.match(user_address, postcode)
uprn: Optional[str] = matches.unambiguous_uprn()
if not uprn or uprn == "nan":
@ -59,4 +59,4 @@ class HistoricEpcResolver:
top: Optional[ScoredHistoricEpc] = matches.top()
if top is None:
return None
return uprn, top.record.address, top.lexiscore
return UprnMatch(uprn, top.record.address, top.lexiscore, top.record.lmk_key)

13
repositories/property/property_override_postgres_repository.py
View file

@ -31,6 +31,17 @@ class PropertyOverridePostgresRepository(PropertyOverrideRepository):
return 0
now = datetime.now(timezone.utc)
# Defensive backstop (ADR-0057): a single INSERT ... ON CONFLICT DO
# UPDATE may not touch the same target row twice — two batch rows sharing
# the conflict key (property_id, override_component, building_part) make
# Postgres raise CardinalityViolation ("cannot affect row a second
# time"). ADR-0057 stops distinct addresses being coerced onto one UPRN
# upstream, so this should no longer arise from real data; collapse to
# the last occurrence (last-write-wins, the same outcome DO UPDATE gives
# across separate statements) so the statement can never be invalid.
deduped: dict[tuple[int, str, int], PropertyOverrideInsert] = {
(r.property_id, r.override_component, r.building_part): r for r in rows
}
values = [
{
"property_id": r.property_id,
@ -42,7 +53,7 @@ class PropertyOverridePostgresRepository(PropertyOverrideRepository):
"created_at": now,
"updated_at": now,
}
for r in rows
for r in deduped.values()
]
stmt = pg_insert(self._table).values(values)

6
repositories/property/property_override_repository.py
View file

@ -34,5 +34,9 @@ class PropertyOverrideRepository(ABC):
"""Upsert each row on ``(property_id, override_component, building_part)``,
refreshing ``override_value`` + ``original_spreadsheet_description`` +
``updated_at`` on conflict (recalculate-on-rerun). Returns the number of
rows affected; an empty list is a no-op returning 0."""
rows affected; an empty list is a no-op returning 0.
Rows sharing a conflict key within one call are collapsed to the last
occurrence (last-write-wins) before writing, so a single batch never
upserts the same target row twice (ADR-0057 backstop)."""
...

4
scripts/resolve_uprns_for_finaliser.py
View file

@ -47,7 +47,7 @@ sys.path.insert(0, str(_REPO_ROOT)) # worktree root first — avoid the import
from backend.address2UPRN.main import ( # noqa: E402
get_epc_data_with_postcode,
get_uprn_from_historic_epc,
get_uprn_with_epc_df,
get_uprn_from_epc_df,
)
from backend.ordnanceSurvey.helpers import ( # noqa: E402
lookup_os_places,
@ -108,7 +108,7 @@ def resolve_epc(
epc_df = get_epc_data_with_postcode(postcode=postcode_clean)
epc_cache[postcode_clean] = epc_df
result = get_uprn_with_epc_df(
result = get_uprn_from_epc_df(
user_inputed_address=address, epc_df=epc_df, verbose=True
)
if isinstance(result, tuple):

0
tests/backend/address2UPRN/__init__.py Normal file
View file

70
tests/backend/address2UPRN/test_resolve_group_ambiguity.py Normal file
View file

@ -0,0 +1,70 @@
"""Unit tests for ``resolve_group_ambiguity`` (ADR-0057).
Each address is matched independently, so one UPRN can be the best match for
two *different* addresses (a coarse EPC record absorbing several real
addresses). Withholding that UPRN is a pure function of a postcode group's
``(uprn, normalised_address)`` pairs — no S3, EPC API, or scoring needed — so
it is unit-tested directly here.
"""
from __future__ import annotations
from backend.address2UPRN.scoring import resolve_group_ambiguity
def test_distinct_addresses_sharing_one_uprn_are_withheld() -> None:
# Flat 1 and Flat 2 both matched the same coarse EPC UPRN — distinct
# addresses, so both are withheld rather than silently merged.
result = resolve_group_ambiguity(
[
("100", "flat 1 42 moreton road"),
("100", "flat 2 42 moreton road"),
]
)
assert result == [(None, "ambiguous_duplicate"), (None, "ambiguous_duplicate")]
def test_same_address_listed_twice_keeps_the_uprn() -> None:
# A genuine re-listing of one property (identical normalised address) is a
# real duplicate, not an ambiguous match — the UPRN is kept on both.
result = resolve_group_ambiguity(
[
("100", "42 moreton road"),
("100", "42 moreton road"),
]
)
assert result == [("100", "matched"), ("100", "matched")]
def test_distinct_uprns_are_each_matched() -> None:
result = resolve_group_ambiguity(
[
("100", "42 moreton road"),
("101", "44 moreton road"),
]
)
assert result == [("100", "matched"), ("101", "matched")]
def test_none_uprn_is_unmatched() -> None:
result = resolve_group_ambiguity([(None, "99 nowhere lane")])
assert result == [(None, "unmatched")]
def test_order_is_preserved_across_mixed_group() -> None:
# A withheld pair, an unmatched row, and a clean match — all in one group;
# the output aligns positionally with the input.
result = resolve_group_ambiguity(
[
("100", "flat 1 42 moreton road"), # ambiguous (with row 3)
(None, "no epc candidate"), # unmatched
("100", "flat 2 42 moreton road"), # ambiguous (with row 0)
("200", "sole match road"), # matched
]
)
assert result == [
(None, "ambiguous_duplicate"),
(None, "unmatched"),
(None, "ambiguous_duplicate"),
("200", "matched"),
]

3
tests/repositories/historic_epc/test_historic_epc_resolver.py
View file

@ -73,10 +73,11 @@ def test_resolve_uprn_returns_unambiguous_match():
# Assert
assert result is not None
uprn, address, score = result
uprn, address, score, certificate_number = result
assert uprn == "100"
assert address == "47 GORDON ROAD"
assert score > 0
assert certificate_number == ""
def test_resolve_uprn_is_none_when_postcode_has_no_data():

102
tests/repositories/property/test_property_override_postgres_repository.py Normal file
View file

@ -0,0 +1,102 @@
"""Integration tests for ``PropertyOverridePostgresRepository.upsert_all``.
The conflict handling lives entirely in SQL (``INSERT ... ON CONFLICT
(property_id, override_component, building_part) DO UPDATE``), so it can only be
verified against a real Postgres -- the ``db_engine`` fixture in
``tests/conftest.py`` spins one up per test. The batch-dedup below is exactly
the case an in-memory fake cannot catch: only a real ``ON CONFLICT`` statement
raises ``CardinalityViolation`` when a batch would touch one target row twice
(the ADR-0057 backstop).
"""
from __future__ import annotations
from collections.abc import Iterator
import pytest
from sqlalchemy import Engine
from sqlmodel import Session, select
from infrastructure.postgres.property_override_table import PropertyOverrideRow
from repositories.property.property_override_postgres_repository import (
PropertyOverridePostgresRepository,
)
from repositories.property.property_override_repository import PropertyOverrideInsert
@pytest.fixture
def session(db_engine: Engine) -> Iterator[Session]:
with Session(db_engine) as s:
yield s
def _insert(
property_id: int,
override_value: str,
description: str,
*,
building_part: int = 0,
component: str = "property_type",
) -> PropertyOverrideInsert:
return PropertyOverrideInsert(
property_id=property_id,
portfolio_id=820,
building_part=building_part,
override_component=component,
override_value=override_value,
original_spreadsheet_description=description,
)
def _all_rows(session: Session) -> list[PropertyOverrideRow]:
return list(session.exec(select(PropertyOverrideRow)).all())
def test_duplicate_conflict_key_in_one_batch_collapses_to_last(
session: Session,
) -> None:
# arrange: two batch rows sharing the conflict key
# (property_id, override_component, building_part) — before the backstop this
# raised CardinalityViolation ("cannot affect row a second time").
repo = PropertyOverridePostgresRepository(session)
# act
repo.upsert_all(
[
_insert(742961, "House", "House: Semi-Detached"),
_insert(742961, "House", "House: Mid-Terrace"),
]
)
session.commit()
# assert: one row, last occurrence wins (matches ON CONFLICT DO UPDATE).
rows = _all_rows(session)
assert len(rows) == 1
assert rows[0].override_value == "House"
assert rows[0].original_spreadsheet_description == "House: Mid-Terrace"
def test_distinct_conflict_keys_are_all_written(session: Session) -> None:
# arrange: dedup must only collapse exact conflict-key duplicates — a
# different property_id, building_part, or component is a distinct row.
repo = PropertyOverridePostgresRepository(session)
# act
repo.upsert_all(
[
_insert(742961, "House", "House: Semi-Detached"),
_insert(742962, "House", "House: Mid-Terrace"), # different property
_insert(742961, "House", "ext", building_part=1), # different part
_insert(742961, "SolidBrick", "solid", component="wall_type"), # component
]
)
session.commit()
# assert
assert len(_all_rows(session)) == 4
def test_empty_batch_is_a_noop(session: Session) -> None:
repo = PropertyOverridePostgresRepository(session)
assert repo.upsert_all([]) == 0
assert _all_rows(session) == []