"""Recall of the name-based tiers (B/C/D) against GTIN ground truth (tier A pairs, Metro <-> Super C). Ground truth: listing pairs sharing a GS1-valid UPC across the two Metro Inc. banners. For each truth pair we ask whether the name/brand/size rules would have linked it without the GTIN. Also reports *false links* among GTIN-bearing listings: name-based pairs between metro and superc whose UPCs differ (a precision check on a labelled subset). """ import pandas as pd, numpy as np from rapidfuzz import fuzz from src.common import path, get_logger from src.matching.match_products import prepare, num_tokens log = get_logger("recall") def main(): prod = pd.read_parquet(path("processed") / "products.parquet"); ret = pd.read_parquet(path("processed") / "retailers.parquet") p = prepare(prod, ret); g = p[p.gtin.notna()].copy() pairs = pd.read_parquet(path("processed") / "product_matches.parquet") truth = pairs[(pairs.tier == "A")][["uid_a", "uid_b"]] idx = g.set_index("uid") rows = [] for ua, ub in truth.itertuples(index=False): a, b = idx.loc[ua], idx.loc[ub] brand_ok = a.bkey != "" and a.bkey == b.bkey size_ok = a.skey != "" and a.skey == b.skey sim = fuzz.token_set_ratio(a.nkey, b.nkey) / 100.0 nums_ok = num_tokens(a.nkey) == num_tokens(b.nkey) tier = None if brand_ok and size_ok and nums_ok and sim >= 0.90: tier = "B" elif brand_ok and size_ok and nums_ok and sim >= 0.80: tier = "C" elif brand_ok and a.skey == "" and b.skey == "" and nums_ok and sim >= 0.95: tier = "C" rows.append(dict(uid_a=ua, uid_b=ub, brand_ok=brand_ok, size_ok=size_ok, nums_ok=nums_ok, sim=sim, recovered_tier=tier, brand_missing=(a.bkey == "" or b.bkey == ""), size_missing=(a.skey == "" or b.skey == ""))) r = pd.DataFrame(rows) log.info("GTIN truth pairs: %d ; recovered by B: %.3f ; by B+C: %.3f", len(r), (r.recovered_tier == "B").mean(), r.recovered_tier.notna().mean()) log.info("failure reasons among non-recovered: brand missing %.3f, brand differs %.3f, size missing %.3f, size differs %.3f, numeric tokens differ %.3f, sim<0.80 %.3f", *(r[r.recovered_tier.isna()][c].mean() for c in ["brand_missing"]), ((~r.brand_ok) & (~r.brand_missing))[r.recovered_tier.isna()].mean(), r[r.recovered_tier.isna()].size_missing.mean(), ((~r.size_ok) & (~r.size_missing))[r.recovered_tier.isna()].mean(), (~r[r.recovered_tier.isna()].nums_ok).mean(), (r[r.recovered_tier.isna()].sim < 0.80).mean()) r.to_csv(path("processed") / "recall_gtin_truth.csv", index=False) if __name__ == "__main__": main()