Entity resolution: blocking and the threshold

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Entity resolution: blocking and the threshold.

20,000 synthetic customers, 2,000 copied with typos. Comparing all pairs is
n(n-1)/2; blocking on the surname's first letter plus the postcode cuts that
and loses some true duplicates (recall). Then a threshold sweep: false merges
against missed duplicates.
"""
import argparse, json, random, string
from collections import defaultdict
from difflib import SequenceMatcher

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--customers", type=int, default=20_000, help="Distinct customers.")
p.add_argument("--duplicates", type=int, default=2_000, help="Copies with typos.")
p.add_argument("--postcodes", type=int, default=400, help="Distinct postcodes.")
p.add_argument("--postcode-typo", type=float, default=0.05, help="Share of copies whose postcode is also mistyped.")
p.add_argument("--seed", type=int, default=0, help="Random seed.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()
random.seed(args.seed)

FIRST = ["Anna", "Ben", "Chloe", "David", "Elena", "Farid", "Grace", "Hugo", "Ines", "Jonas", "Kira", "Liam", "Maya",
         "Noah", "Olga", "Pavel", "Quinn", "Rosa", "Sami", "Tara", "Umar", "Vera", "Wen", "Yara", "Zoe"]
def surname(): return "".join(random.choice(string.ascii_lowercase) for _ in range(random.randint(5, 9))).capitalize()
def typo(s):
    i = random.randrange(1, len(s) - 1)
    return s[:i] + s[i + 1] + s[i] + s[i + 2:] if random.random() < 0.5 else s[:i] + s[i + 1:]   # swap or drop a letter

people = [{"id": i, "first": random.choice(FIRST), "last": surname(), "pc": f"P{random.randrange(args.postcodes):04d}"}
          for i in range(args.customers)]
recs = [dict(r, rid=r["id"]) for r in people]
for r in random.sample(people, args.duplicates):
    pc = r["pc"] if random.random() > args.postcode_typo else f"P{random.randrange(args.postcodes):04d}"
    recs.append({"id": r["id"], "first": r["first"], "last": typo(r["last"]), "pc": pc, "rid": len(recs)})
n = len(recs)
all_pairs = n * (n - 1) // 2

blocks = defaultdict(list)
for r in recs: blocks[(r["last"][0], r["pc"])].append(r)
cand = [(a, b) for g in blocks.values() for i, a in enumerate(g) for b in g[i + 1:]]
true_pairs = args.duplicates
found_true = sum(1 for a, b in cand if a["id"] == b["id"])
print(f"{n:,} records: {all_pairs:,} pairs to compare without blocking; {len(cand):,} with blocking "
      f"({all_pairs / max(1, len(cand)):,.0f}x fewer)")
print(f"blocking keeps {found_true:,} of {true_pairs:,} true duplicate pairs: recall {found_true / true_pairs:.1%}")

def sim(a, b): return SequenceMatcher(None, f"{a['first']} {a['last']}", f"{b['first']} {b['last']}").ratio()
scored = [(sim(a, b), a["id"] == b["id"]) for a, b in cand]
rows = []
for t in [0.70, 0.75, 0.80, 0.85, 0.90, 0.95]:
    false_merges = sum(1 for s, same in scored if s >= t and not same)
    missed = true_pairs - sum(1 for s, same in scored if s >= t and same)
    rows.append({"threshold": t, "false_merges": false_merges, "missed": missed})
    print(f"threshold {t:.2f}: {false_merges:>5} false merges   {missed:>5} missed duplicates")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"thresholds": rows, "summary": [{"records": n, "all_pairs": all_pairs, "blocked_pairs": len(cand),
                   "reduction": all_pairs / max(1, len(cand)), "recall_pct": found_true / true_pairs * 100}]}}, f)