1"""Phase 2.5 — Cross-merchant product matching (candidates for H5 price dispersion).23There is no GTIN in the source, so identity across merchants is inferred, tiered, and later validated on a labelled sample4(src/validation/match_validation.py). Only listings in the price sample are matched.56Tiers (pairs of listings in DIFFERENT stores):7 A same merchant SKU (>= 4 chars, not a short pure number) and token_set_ratio(title) >= 708 B same normalised brand (vendor), identical normalised title9 C same normalised brand, same category, token_sort_ratio(title) >= 9010 D identical normalised title (>= 3 tokens, >= 15 chars), same category, brand missing or different [sensitivity only]11Comparability guard (kept as a flag, used by the primary H5 specification):12 comparable = both single-variant, or equal variant counts and (grams equal or both missing)1314Outputs: data/interim/match_pairs.parquet, data/interim/match_groups.parquet (connected components over tiers A–C),15 experiments/matching.json16"""17from __future__ import annotations18import itertools, json, re, unicodedata19import numpy as np, pandas as pd, networkx as nx20from rapidfuzz import fuzz21from src.common import ROOT, P, get_logger, ledger, register_output, write_json22from src.clean.listings import norm_title2324log = get_logger("features.match")25GENERIC_VENDORS = {"", "general", "generale", "default", "default title", "unknown", "inconnu", "divers", "vendor", "brand", "marque", "none", "n a", "na", "autre", "autres"}262728def norm_vendor(v: str) -> str:29 v = norm_title(v)30 v = re.sub(r"\b(inc|ltee|ltd|llc|enr|senc|co|cie|boutique|la|le|les|the|de|du|des|d)\b", " ", v)31 return re.sub(r"\s+", " ", v).strip()323334def sku_ok(s) -> bool:35 if not isinstance(s, str): return False36 s = s.strip()37 if len(s) < 4: return False38 if s.isdigit() and len(s) < 6: return False39 return not re.fullmatch(r"(default|none|null|n/a|sku|test)[-_ ]*\d*", s.lower())404142def main():43 li = pd.read_parquet(P["interim"] / "listings_clean.parquet")44 u = li[li.f_analysis_price].copy()45 u["vendor_norm"] = u.vendor.fillna("").map(norm_vendor)46 u.loc[u.vendor_norm.isin(GENERIC_VENDORS), "vendor_norm"] = ""47 u["sku_norm"] = u.sku.where(u.sku.map(sku_ok)).fillna("").str.strip().str.upper()48 u["n_tok"] = u.title_norm.str.split().str.len()49 u["nv"] = u.n_variants.fillna(1).astype(int)50 idx = u.set_index("uid")51 pairs = []5253 def add(a, b, tier, score):54 if a == b: return55 pairs.append((min(a, b), max(a, b), tier, score))5657 # Tier A — shared SKU across stores58 sk = u[u.sku_norm != ""].groupby("sku_norm")59 for s, g in sk:60 if g.store_id.nunique() < 2 or len(g) > 60: continue61 for (ua, ta, sa), (ub, tb, sb) in itertools.combinations(zip(g.uid, g.title_norm, g.store_id), 2):62 if sa == sb: continue63 sc = fuzz.token_set_ratio(ta, tb)64 if sc >= 70: add(ua, ub, "A", sc)65 nA = len(pairs); log.info("tier A pairs %s", nA)6667 # Tier B/C — same brand68 vb = u[u.vendor_norm != ""].groupby("vendor_norm")69 for v, g in vb:70 if g.store_id.nunique() < 2 or len(g) > 4000: continue71 # B: identical title72 for t, gg in g.groupby("title_norm"):73 if gg.store_id.nunique() < 2 or len(t) < 8: continue74 for (ua, sa), (ub, sb) in itertools.combinations(zip(gg.uid, gg.store_id), 2):75 if sa != sb: add(ua, ub, "B", 100)76 # C: fuzzy within category77 for c, gc in g.groupby("category"):78 if gc.store_id.nunique() < 2 or len(gc) > 1500: continue79 stores = gc.store_id.values; uids = gc.uid.values; titles = gc.title_norm.values80 for i in range(len(gc)):81 for j in range(i + 1, len(gc)):82 if stores[i] == stores[j] or titles[i] == titles[j]: continue83 sc = fuzz.token_sort_ratio(titles[i], titles[j])84 if sc >= 90: add(uids[i], uids[j], "C", sc)85 nBC = len(pairs) - nA; log.info("tier B+C pairs %s", nBC)8687 # Tier D — identical title, brand missing/different88 gt = u[(u.n_tok >= 3) & (u.title_norm.str.len() >= 15)].groupby(["category", "title_norm"])89 for (c, t), g in gt:90 if g.store_id.nunique() < 2 or len(g) > 40: continue91 for (ua, sa, va), (ub, sb, vb_) in itertools.combinations(zip(g.uid, g.store_id, g.vendor_norm), 2):92 if sa == sb: continue93 if va and vb_ and va == vb_: continue # already tier B94 add(ua, ub, "D", 100)95 nD = len(pairs) - nA - nBC; log.info("tier D pairs %s", nD)9697 pr = pd.DataFrame(pairs, columns=["uid_a", "uid_b", "tier", "score"])98 pr = pr.sort_values(["uid_a", "uid_b", "tier"]).drop_duplicates(["uid_a", "uid_b"], keep="first") # best tier (A<B<C<D)99 for side in ("a", "b"):100 pr[f"store_{side}"] = pr[f"uid_{side}"].map(idx.store_id); pr[f"price_{side}"] = pr[f"uid_{side}"].map(idx.price)101 pr[f"nv_{side}"] = pr[f"uid_{side}"].map(idx.nv); pr[f"grams_{side}"] = pr[f"uid_{side}"].map(idx.grams)102 pr[f"cat_{side}"] = pr[f"uid_{side}"].map(idx.category)103 pr["comparable"] = ((pr.nv_a == 1) & (pr.nv_b == 1)) | ((pr.nv_a == pr.nv_b) & ((pr.grams_a == pr.grams_b) | (pr.grams_a.isna() & pr.grams_b.isna())))104 pr["log_gap"] = (np.log(pr.price_a) - np.log(pr.price_b)).abs()105 pr.to_parquet(P["interim"] / "match_pairs.parquet", index=False); register_output(P["interim"] / "match_pairs.parquet", "match")106107 # groups over A-C108 G = nx.Graph(); G.add_edges_from(pr.loc[pr.tier.isin(list("ABC")), ["uid_a", "uid_b"]].itertuples(index=False, name=None))109 rows = []110 for gid, comp in enumerate(nx.connected_components(G)):111 for uid in comp: rows.append((gid, uid))112 gr = pd.DataFrame(rows, columns=["group_id", "uid"])113 gr["store_id"] = gr.uid.map(idx.store_id); gr["price"] = gr.uid.map(idx.price); gr["category"] = gr.uid.map(idx.category)114 gr["nv"] = gr.uid.map(idx.nv); gr["grams"] = gr.uid.map(idx.grams); gr["vendor_norm"] = gr.uid.map(idx.vendor_norm)115 gs = gr.groupby("group_id").agg(n=("uid", "size"), n_stores=("store_id", "nunique"), n_cat=("category", "nunique"))116 gr = gr.merge(gs, left_on="group_id", right_index=True)117 gr.to_parquet(P["interim"] / "match_groups.parquet", index=False); register_output(P["interim"] / "match_groups.parquet", "match")118119 summ = {"universe_listings": len(u), "universe_stores": int(u.store_id.nunique()), "with_sku": int((u.sku_norm != "").sum()), "with_brand": int((u.vendor_norm != "").sum()),120 "pairs_by_tier": pr.tier.value_counts().to_dict(), "pairs_comparable_by_tier": pr[pr.comparable].tier.value_counts().to_dict(),121 "median_abs_log_gap_by_tier": pr.groupby("tier").log_gap.median().round(4).to_dict(),122 "share_zero_gap_by_tier": pr.groupby("tier").log_gap.apply(lambda s: (s < 1e-9).mean()).round(4).to_dict(),123 "groups_AC": int(gs.shape[0]), "groups_AC_multi_store": int((gs.n_stores >= 2).sum()), "listings_in_groups": len(gr),124 "stores_in_groups": int(gr.store_id.nunique()), "groups_size_gt_10": int((gs.n > 10).sum()),125 "regions_pairs": None}126 write_json(summ, ROOT / "experiments" / "matching.json", "match"); ledger("features.match_products", **summ)127 log.info(json.dumps(summ, ensure_ascii=False))128129130if __name__ == "__main__":131 main()132