`services/matching/entity_match.py`

150 lines
"""Entity resolution — probabilistic linkage, DISTINCT from public search ranking.

Search ranking favors recall (help a person find a record). This favors precision (decide
whether two records refer to the same real owner) and keeps a manual-review band. A score
here NEVER auto-approves a claim.

The weights below are an ILLUSTRATIVE logistic model. Production replaces them with a
Fellegi-Sunter model whose agreement/disagreement weights are frequency-adjusted (rare-name
agreement counts more than a common surname) and calibrated against state-adjudicated pairs,
with per-jurisdiction error analysis and protected-class fairness testing.

Stdlib-only: uses difflib.SequenceMatcher for string similarity in place of rapidfuzz.
"""
from __future__ import annotations

import math
from dataclasses import dataclass, field
from difflib import SequenceMatcher

from services.common.normalize import (
    normalize_business, normalize_postal, normalize_text, phonetic_key,
)


@dataclass(frozen=True)
class MatchInput:
    name: str
    address: str | None = None
    city: str | None = None
    region: str | None = None
    postal_code: str | None = None
    is_business: bool = False


@dataclass(frozen=True)
class MatchResult:
    probability: float
    disposition: str            # likely_duplicate_candidate | manual_review | distinct
    features: dict = field(default_factory=dict)


def _ratio(a: str, b: str) -> float:
    if not a or not b:
        return 0.0
    return SequenceMatcher(None, a, b).ratio()


def _token_set_ratio(a: str, b: str) -> float:
    """Order-independent token overlap (Jaccard-ish), a cheap token_set_ratio stand-in."""
    ta, tb = set(a.split()), set(b.split())
    if not ta or not tb:
        return 0.0
    return len(ta & tb) / len(ta | tb)


def similarity(left: str | None, right: str | None) -> float:
    return _ratio(normalize_text(left), normalize_text(right))


def blocking_keys(inp: MatchInput) -> list[str]:
    """Return index/lookup keys for candidate reduction (blocking step).

    A candidate pair is only scored by entity_match if the two records share ≥1
    blocking key — this reduces the O(n²) all-pairs comparison to O(n × k) where k
    is the average blocking-bucket size (typically O(1) to O(100)).

    Index pattern: build a {key → list[record_id]} map at ingest time; at query time
    generate the query record's keys, union the candidate sets across all matching
    keys, then call entity_match on each candidate. Sharded by jurisdiction for
    scalability.

    Fairness note: phonetic keys must cover international name distributions.
    The current phonetic_key uses a Soundex-like scheme — adequate for English
    but degrades on Polish/Vietnamese/Arabic names.  Production should add a
    Double Metaphone or Metaphone 3 path and test recall across name corpora
    (see tests/test_cycle4_fairness.py).
    """
    norm = normalize_business(inp.name) if inp.is_business else normalize_text(inp.name)
    keys: list[str] = []
    # Name-derived keys only when the name yields latin-script tokens. A name that
    # normalizes to empty (pure CJK/Arabic script under the Soundex-era normalizer) must
    # STILL fall through to the geo keys below — otherwise those owners are un-indexable
    # even with a valid address, a coverage/fairness gap (see tests/test_cycle4_fairness.py).
    if norm:
        pk = phonetic_key(norm)
        if pk:
            keys.append(f"ph:{pk}")
        toks = norm.split()
        if toks:
            keys.append(f"tok0:{toks[0]}")
            if len(toks) > 1:
                keys.append(f"tok01:{toks[0]}_{toks[1]}")
    if inp.postal_code:
        z5 = (normalize_postal(inp.postal_code) or "")[:5]
        if z5:
            keys.append(f"zip:{z5}")
    if inp.region:
        nr = normalize_text(inp.region)
        if nr:
            keys.append(f"region:{nr}")
    return keys


def entity_match(left: MatchInput, right: MatchInput) -> MatchResult:
    # A person and a business are never the same entity.
    if left.is_business != right.is_business:
        return MatchResult(0.0, "distinct", {"type_conflict": 1.0})

    ln = normalize_business(left.name) if left.is_business else normalize_text(left.name)
    rn = normalize_business(right.name) if right.is_business else normalize_text(right.name)

    features = {
        "name_edit": _ratio(ln, rn),
        "name_token": _token_set_ratio(ln, rn),
        "phonetic": _ratio(phonetic_key(ln), phonetic_key(rn)),
        "address": similarity(left.address, right.address),
        "city": similarity(left.city, right.city),
        "region_exact": float(
            bool(left.region and right.region)
            and normalize_text(left.region) == normalize_text(right.region)
        ),
        "postal_exact": float(
            bool(left.postal_code and right.postal_code)
            and (normalize_postal(left.postal_code) or "")[:5]
            == (normalize_postal(right.postal_code) or "")[:5]
        ),
    }

    # ILLUSTRATIVE weights — train + calibrate before production use.
    log_odds = (
        -8.0
        + 3.6 * features["name_edit"]
        + 2.8 * features["name_token"]
        + 2.2 * features["phonetic"]
        + 2.3 * features["address"]
        + 1.1 * features["city"]
        + 1.0 * features["region_exact"]
        + 1.5 * features["postal_exact"]
    )
    probability = 1.0 / (1.0 + math.exp(-log_odds))

    if probability >= 0.995:
        disposition = "likely_duplicate_candidate"
    elif probability >= 0.90:
        disposition = "manual_review"
    else:
        disposition = "distinct"

    return MatchResult(round(probability, 6), disposition, features)