SPB Git
spboucher.ai

spb/wp37_uqo

Public

phase1: calibrated tiered matching (UPC ground truth), multipack-aware sizes, recall/calibration scripts, coverage & attrition tables, tests

Simon-Pierre Boucher committed (Nov 1, 2026) parent a7a52e3 Browse files

10 changed files +427 −2

10 changed files
Unified Split
modified src/cleaning/build_panel.py +1 −0
@@ -28,6 +28,7 @@ def canonical_products(prod: pd.DataFrame) -> pd.DataFrame:
28 out["size_base"] = [s.base if s else None for s in sizes] 28 out["size_base"] = [s.base if s else None for s in sizes]
29 out["size_approx"] = [bool(s.approx) if s else False for s in sizes] 29 out["size_approx"] = [bool(s.approx) if s else False for s in sizes]
30 out["size_source"] = [s.source if s else None for s in sizes] 30 out["size_source"] = [s.source if s else None for s in sizes]
31+ out["size_pack"] = [s.pack if s else np.nan for s in sizes]
31 out["per_weight_label"] = [l.per_weight_unit for l in labels] 32 out["per_weight_label"] = [l.per_weight_unit for l in labels]
32 out["multibuy_n"] = [l.multibuy_n for l in labels] 33 out["multibuy_n"] = [l.multibuy_n for l in labels]
33 up = [unit_price(p, s, l) for p, s, l in zip(out["price"], sizes, labels)] 34 up = [unit_price(p, s, l) for p, s, l in zip(out["price"], sizes, labels)]
added src/cleaning/coverage.py +50 −0
@@ -0,0 +1,50 @@
1+"""Phase 1.5 — coverage and attrition tables (inputs of Table 1 / Figure 1).
2+
3+Outputs (data/processed/coverage_*.csv): by feed (listings, active, days observed, events, unit-price coverage,
4+flags), by category, by week (listings observed, feeds reporting), and the attrition ledger.
5+"""
6+import pandas as pd, numpy as np
7+from src.common import path, get_logger, ledger, SETTINGS
8+
9+log = get_logger("coverage")
10+
11+def main():
12+ P = path("processed")
13+ prod = pd.read_parquet(P / "products.parquet"); ret = pd.read_parquet(P / "retailers.parquet")
14+ wk = pd.read_parquet(P / "offers_weekly.parquet"); day = pd.read_parquet(P / "offers_daily.parquet", columns=["uid", "day", "source"])
15+ prod = prod.merge(ret[["feed_id", "banner", "company", "channel", "store_format", "is_grocery"]], left_on="source", right_on="feed_id", how="left")
16+ ev = pd.read_parquet(path("interim") / "raw_price_log.parquet").groupby("uid").size().rename("n_events")
17+ prod = prod.merge(ev, left_on="uid", right_index=True, how="left")
18+ dpu = day.groupby("uid").size().rename("days_obs"); prod = prod.merge(dpu, left_on="uid", right_index=True, how="left")
19+ by_feed = prod.groupby(["company", "banner", "source", "channel", "store_format"]).agg(
20+ listings=("uid", "size"), active=("active", "sum"), on_sale_now=("on_sale", "sum"),
21+ median_days_obs=("days_obs", "median"), price_events=("n_events", "sum"),
22+ unit_price_cov=("unit_price_rp", lambda s: s.notna().mean()), brand_cov=("brand", lambda s: (s.fillna("") != "").mean()),
23+ gtin=("gtin_valid", "sum"), weighted=("flag_weighted", "sum"), multibuy=("flag_multibuy", "sum"),
24+ loyalty=("flag_loyalty", "sum"), unrealistic=("flag_unrealistic", "sum")).reset_index().sort_values("listings", ascending=False)
25+ by_feed.to_csv(P / "coverage_by_feed.csv", index=False)
26+ by_cat = prod.groupby("category").agg(listings=("uid", "size"), active=("active", "sum"), feeds=("source", "nunique"),
27+ unit_price_cov=("unit_price_rp", lambda s: s.notna().mean()), price_events=("n_events", "sum")).reset_index().sort_values("listings", ascending=False)
28+ by_cat.to_csv(P / "coverage_by_category.csv", index=False)
29+ wk2 = wk.merge(prod[["uid", "channel", "category", "is_grocery"]], on="uid")
30+ by_week = wk2.groupby("week").agg(listings=("uid", "nunique"), feeds=("source", "nunique"),
31+ changes=("n_changes", "sum"), online=("channel", lambda s: (s == "online_storefront").sum()),
32+ circular=("channel", lambda s: s.str.startswith("circular").sum())).reset_index()
33+ by_week.to_csv(P / "coverage_by_week.csv", index=False)
34+ # attrition ledger (listing level)
35+ steps = []
36+ n0 = len(prod); steps.append(("All listings in snapshot (products table)", n0))
37+ n = prod[~prod.source.isin(SETTINGS["scope"]["exclude_sources"])]; steps.append(("excl. SAQ (alcohol monopoly)", len(n)))
38+ n = n[n.is_grocery]; steps.append(("excl. pharmacy feeds", len(n)))
39+ n = n[~n.category.isin(SETTINGS["scope"]["non_grocery_categories"])]; steps.append(("excl. non-food categories (health & beauty, household, pets, baby, alcohol)", len(n)))
40+ n = n[n.price.notna() | (n.n_events.fillna(0) > 0)]; steps.append(("with at least one observed price", len(n)))
41+ n = n[~n.flag_unrealistic]; steps.append(("excl. unrealistic prices", len(n)))
42+ n1 = n[n.size_qty.notna()]; steps.append(("… with parsable package size (unit-price / matching universe)", len(n1)))
43+ g = pd.read_parquet(P / "match_groups_ABC.parquet"); steps.append(("… in a cross-banner match group (tiers A–C)", int(n1.uid.isin(g.uid).sum())))
44+ att = pd.DataFrame(steps, columns=["step", "listings"]); att["share_of_initial"] = att.listings / n0
45+ att.to_csv(P / "attrition_listings.csv", index=False)
46+ log.info("\n%s", att.to_string(index=False)); log.info("weeks: %d (%s → %s)", len(by_week), by_week.week.min(), by_week.week.max())
47+ ledger("1", "coverage", f"attrition {n0} -> {att.listings.iloc[-1]} matched listings")
48+
49+if __name__ == "__main__":
50+ main()
modified src/cleaning/units.py +15 −1
@@ -16,6 +16,7 @@ _RANGE_RE = re.compile(r"(\d+(?:[.,]\d+)?)\s*(?:-|–|a|à)\s*(\d+(?:[.,]\d+)?)\
16 _PERKG_RE = re.compile(r"/\s*(kg|lb|100\s*g|100\s*ml|l|ch|un)\b", re.I) 16 _PERKG_RE = re.compile(r"/\s*(kg|lb|100\s*g|100\s*ml|l|ch|un)\b", re.I)
17 _MULTI_RE = re.compile(r"^\s*(\d+)\s*(?:/|pour|for)\s*(?:(\d{1,4}(?:[.,]\d{1,2})?)\s*\$)?") 17 _MULTI_RE = re.compile(r"^\s*(\d+)\s*(?:/|pour|for)\s*(?:(\d{1,4}(?:[.,]\d{1,2})?)\s*\$)?")
18 _APPROX_RE = re.compile(r"environ|approx|moyenne|avg|about|~", re.I) 18 _APPROX_RE = re.compile(r"environ|approx|moyenne|avg|about|~", re.I)
19+_DEPOSIT_RE = re.compile(r"(cons(igne)?\.?\s*[\d,\.]+\s*\$\s*/\s*\w+\.?|\d+[,\.]\d+\s*\$\s*/\s*100\s*(g|ml))", re.I)
19 _SOLD_INDIV_RE = re.compile(r"vendu individuellement|sold individually|a l'unite|à l'unité|l'unite|unite$", re.I) 20 _SOLD_INDIV_RE = re.compile(r"vendu individuellement|sold individually|a l'unite|à l'unité|l'unite|unite$", re.I)
20 21
21 def strip_accents(s: str) -> str: 22 def strip_accents(s: str) -> str:
@@ -27,6 +28,7 @@ class Size:
27 base: str # 'g' | 'ml' | 'un' 28 base: str # 'g' | 'ml' | 'un'
28 approx: bool = False 29 approx: bool = False
29 source: str = "size_label" # where it was parsed from 30 source: str = "size_label" # where it was parsed from
31+ pack: int = 1 # multipack count ('12 x 355 ml' -> 12)
30 32
31 def parse_size(label: str | None, name: str | None = None) -> Size | None: 33 def parse_size(label: str | None, name: str | None = None) -> Size | None:
32 """Parse '450 g', '2 L', '12 x 355 ml', '6 un', 'Vendu individuellement', '1,36 kg'; falls back to the name.""" 34 """Parse '450 g', '2 L', '12 x 355 ml', '6 un', 'Vendu individuellement', '1,36 kg'; falls back to the name."""
@@ -46,7 +48,7 @@ def parse_size(label: str | None, name: str | None = None) -> Size | None:
46 factor, base = _UNITS[m.group(3).lower()] 48 factor, base = _UNITS[m.group(3).lower()]
47 total = mult * qty * factor 49 total = mult * qty * factor
48 if total > 0: 50 if total > 0:
49− return Size(total, base, bool(_APPROX_RE.search(t)), src) 51+ return Size(total, base, bool(_APPROX_RE.search(t)), src, mult)
50 if src == "size_label" and _SOLD_INDIV_RE.search(t): 52 if src == "size_label" and _SOLD_INDIV_RE.search(t):
51 return Size(1.0, "un", False, src) 53 return Size(1.0, "un", False, src)
52 return None 54 return None
@@ -88,3 +90,15 @@ def unit_price(price: float | None, size: Size | None, label: PriceLabel | None
88 if size.base == "un": 90 if size.base == "un":
89 return (round(price / size.qty, 6), "un") 91 return (round(price / size.qty, 6), "un")
90 return (round(price / size.qty * 100.0, 6), f"100 {size.base}") 92 return (round(price / size.qty * 100.0, 6), f"100 {size.base}")
93+
94+
95+def size_residual(label: str | None) -> str:
96+ """Text of the size label that is NOT the package size itself (e.g. 'calibre 21-25', 'gros', 'bio'), with deposit
97+ and unit-price fragments removed — used as extra discriminating tokens in matching."""
98+ if not label:
99+ return ""
100+ t = strip_accents(label).lower().replace("\u00a0", " ")
101+ t = _DEPOSIT_RE.sub(" ", t)
102+ t = _RANGE_RE.sub(" ", t); t = _SIZE_RE.sub(" ", t)
103+ t = re.sub(r"[^a-z0-9%]+", " ", t)
104+ return " ".join(sorted(w for w in t.split() if len(w) > 1))
added src/matching/__init__.py +0 −0
added src/matching/calibrate_thresholds.py +51 −0
@@ -0,0 +1,51 @@
1+"""Calibrate name-similarity rules on UPC ground truth (Metro <-> Super C, same brand key, same size bucket).
2+
3+For every cross-banner candidate pair in a (brand, size) block we compute several scorers and the GTIN-equality label,
4+then report precision / recall by threshold. The chosen thresholds are frozen in the preregistration.
5+"""
6+import itertools, pandas as pd, numpy as np
7+from rapidfuzz import fuzz
8+from src.common import path, get_logger
9+from src.matching.match_products import prepare, num_tokens
10+
11+log = get_logger("calib")
12+MULTI_RE = r"\b(ou|or|assortis?|assorted|selectionnes?|varietes?|variétés|au choix|et plus)\b|,"
13+
14+def main():
15+ prod = pd.read_parquet(path("processed") / "products.parquet"); ret = pd.read_parquet(path("processed") / "retailers.parquet")
16+ p = prepare(prod, ret); g = p[p.gtin.notna() & (p.bkey != "") & (p.skey != "")]
17+ rows = []
18+ for (bk, sk), block in g.groupby(["bkey", "skey"]):
19+ m = block[block.source == "metro"]; s = block[block.source == "superc"]
20+ if len(m) == 0 or len(s) == 0: continue
21+ for a, b in itertools.product(m.itertuples(), s.itertuples()):
22+ rows.append(dict(uid_a=a.uid, uid_b=b.uid, same_gtin=(a.gtin == b.gtin),
23+ set_ratio=fuzz.token_set_ratio(a.nkey, b.nkey) / 100, sort_ratio=fuzz.token_sort_ratio(a.nkey, b.nkey) / 100,
24+ ratio=fuzz.ratio(a.nkey, b.nkey) / 100, tokens_equal=(set(a.nkey.split()) == set(b.nkey.split())),
25+ nums_equal=(num_tokens(a.nkey) == num_tokens(b.nkey)),
26+ jaccard=len(set(a.nkey.split()) & set(b.nkey.split())) / max(len(set(a.nkey.split()) | set(b.nkey.split())), 1)))
27+ d = pd.DataFrame(rows)
28+ log.info("candidate pairs in (brand,size) blocks: %d ; positives (same UPC): %d ; negatives: %d", len(d), d.same_gtin.sum(), (~d.same_gtin).sum())
29+ npos = d.same_gtin.sum()
30+ out = []
31+ for scorer in ["set_ratio", "sort_ratio", "ratio", "jaccard"]:
32+ for thr in [0.70, 0.75, 0.80, 0.85, 0.90, 0.95, 1.0]:
33+ sel = d[(d[scorer] >= thr) & d.nums_equal]
34+ out.append(dict(scorer=scorer, thr=thr, n=len(sel), precision=sel.same_gtin.mean() if len(sel) else np.nan, recall=sel.same_gtin.sum() / npos))
35+ sel = d[d.tokens_equal & d.nums_equal]; out.append(dict(scorer="tokens_equal", thr=1.0, n=len(sel), precision=sel.same_gtin.mean(), recall=sel.same_gtin.sum() / npos))
36+ res = pd.DataFrame(out); res.to_csv(path("processed") / "match_threshold_calibration.csv", index=False)
37+ log.info("\n%s", res.round(3).to_string(index=False))
38+ # what do the residual false positives at sort_ratio>=0.9 look like?
39+ fp = d[(d.sort_ratio >= 0.9) & d.nums_equal & ~d.same_gtin]
40+ txt = prod.set_index("uid")
41+ log.info("false positives at sort_ratio>=0.90: %d ; examples:", len(fp))
42+ for _, x in fp.sample(min(10, len(fp)), random_state=1).iterrows():
43+ a, b = txt.loc[x.uid_a], txt.loc[x.uid_b]
44+ log.info(" [%.2f] %s | %s | %s <-> %s | %s | %s", x.sort_ratio, a.brand, a["name"], a.size_label, b.brand, b["name"], b.size_label)
45+ # same name & size but different UPC: how often is the name literally identical?
46+ ident = d[(d.ratio == 1.0) & ~d.same_gtin]
47+ log.info("literally identical cleaned names with different UPC: %d of %d identical-name pairs (%.1f%%) — name-identical ≠ product-identical (variants hidden in UPC)",
48+ len(ident), (d.ratio == 1.0).sum(), 100 * len(ident) / max((d.ratio == 1.0).sum(), 1))
49+
50+if __name__ == "__main__":
51+ main()
added src/matching/llm_annotate.py +54 −0
@@ -0,0 +1,54 @@
1+"""Second annotator for the match-validation sample: Claude (claude-opus-5-5) labels each candidate pair as
2+SAME (same product, same package size), SIZE_DIFF (same product, different size), DIFFERENT, or UNSURE.
3+Human adjudication (`label_human`) remains the reference; LLM labels are an auxiliary signal. Logged: model, tokens, cost.
4+"""
5+import json, os, time, pandas as pd
6+import anthropic
7+from dotenv import load_dotenv
8+from src.common import ROOT, path, get_logger, ledger
9+
10+log = get_logger("llm_annotate")
11+MODEL = "claude-opus-5-5"; PRICE_IN, PRICE_OUT = 4.0, 20.0 # $/MTok
12+
13+SYSTEM = """You are an expert grocery product data annotator for Quebec (French/English product names).
14+For each candidate pair of listings from two different retailers, decide whether they are the SAME product
15+(identical brand, product, variant/flavour/fat content, AND package size/quantity). Labels:
16+SAME = same product and same size; SIZE_DIFF = same product but different package size or count;
17+DIFFERENT = different product, brand, variant, flavour or fat content; UNSURE = cannot tell from the text.
18+Be strict: 'Lait 2%' vs 'Lait 3.25%' is DIFFERENT; 'Yogourt vanille' vs 'Yogourt fraise' is DIFFERENT.
19+Missing brand on one side is acceptable if the name clearly contains the brand.
20+Return ONLY a JSON array of objects {"sample_id": int, "label": str, "reason": str (<= 12 words)}."""
21+
22+def main():
23+ load_dotenv(ROOT / ".env")
24+ f = path("processed") / "match_validation_sample.csv"
25+ s = pd.read_csv(f)
26+ client = anthropic.Anthropic(max_retries=3, timeout=300)
27+ tot_in = tot_out = 0; labels = {}
28+ for i in range(0, len(s), 20):
29+ chunk = s.iloc[i:i + 20]
30+ lines = []
31+ for _, r in chunk.iterrows():
32+ lines.append(f'{{"sample_id": {r.sample_id}, "A": "{r.source_a_txt}: brand={r.brand_a_txt!s} | {r.name_a_txt} | size={r.size_label_a_txt!s}", '
33+ f'"B": "{r.source_b_txt}: brand={r.brand_b_txt!s} | {r.name_b_txt} | size={r.size_label_b_txt!s}"}}')
34+ msg = "Candidate pairs:\n" + "\n".join(lines)
35+ with client.messages.stream(model=MODEL, max_tokens=8000, system=SYSTEM, messages=[{"role": "user", "content": msg}]) as st:
36+ resp = st.get_final_message()
37+ if resp.stop_reason == "refusal":
38+ log.warning("refusal on chunk %d", i); continue
39+ text = "".join(b.text for b in resp.content if b.type == "text").strip()
40+ text = text[text.find("["): text.rfind("]") + 1]
41+ for o in json.loads(text):
42+ labels[int(o["sample_id"])] = (o["label"], o.get("reason", ""))
43+ tot_in += resp.usage.input_tokens; tot_out += resp.usage.output_tokens
44+ time.sleep(0.5)
45+ s["label_llm"] = s.sample_id.map(lambda k: labels.get(k, ("", ""))[0])
46+ s["llm_reason"] = s.sample_id.map(lambda k: labels.get(k, ("", ""))[1])
47+ s.to_csv(f, index=False)
48+ cost = tot_in / 1e6 * PRICE_IN + tot_out / 1e6 * PRICE_OUT
49+ log.info("annotated %d/%d pairs with %s ; tokens in=%d out=%d ; cost=$%.3f", s.label_llm.ne("").sum(), len(s), MODEL, tot_in, tot_out, cost)
50+ log.info("LLM labels by tier:\n%s", pd.crosstab(s.tier, s.label_llm).to_string())
51+ ledger("1", "llm_annotate", f"{MODEL}: {len(s)} pairs, in={tot_in}, out={tot_out}, cost=${cost:.3f}")
52+
53+if __name__ == "__main__":
54+ main()
added src/matching/match_products.py +198 −0
@@ -0,0 +1,198 @@
1+"""Phase 1.4 — cross-banner product matching with confidence tiers.
2+
3+Tiers (calibrated on UPC ground truth Metro<->Super C, src/matching/calibrate_thresholds.py, then frozen):
4+ A exact GS1-valid GTIN equality (only feeds whose external_id is a UPC: metro, superc — 100 % check-digit valid)
5+ B brand key equal + package size equal (0.1-unit bucket) + identical cleaned token sets + identical numeric tokens
6+ (labelled-set precision 0.913, recall 0.984)
7+ C brand key equal + size equal + token_sort_ratio >= 0.90 + identical numeric tokens, not already B
8+ (labelled-set precision 0.845 cumulative)
9+ D (brand missing on one side) size equal + full-name token_sort_ratio >= 0.92 + identical numeric tokens + brand
10+ tokens of the branded side present in the brandless name
11+Excluded from matching: listings without a parsable package size (circular items are mostly size-less) and
12+multi-product circular offers ("X ou Y", "assortis", "sélectionnés", comma lists). A first version using
13+rapidfuzz token_set_ratio was rejected because a name that is a token-subset of another scores 1.0
14+("beurre végétal salé" ~ "beurre végétal non salé"); see data/processed/match_validation_sample_v0_tokenset_rejected.csv.
15+Pairs are formed only across different banners (same company allowed, flagged `same_company`); same-banner links
16+between the storefront feed and the circular feed are recorded separately (`link_type = 'channel'`).
17+Outputs: data/processed/product_matches.parquet (pairs), match_groups_{tier}.parquet (union-find groups), and a
18+stratified validation sample (data/processed/match_validation_sample.csv) for manual + LLM annotation.
19+"""
20+from __future__ import annotations
21+import re, unicodedata, itertools, hashlib
22+import numpy as np, pandas as pd
23+from rapidfuzz import fuzz, process
24+from src.common import path, get_logger, ledger, seed_everything, SEED, SETTINGS
25+from src.cleaning.units import size_residual
26+
27+log = get_logger("match")
28+_ALNUM = re.compile(r"[^a-z0-9]+")
29+_SIZE_TOK = re.compile(r"\b\d+(?:[.,]\d+)?\s*(?:x|×)?\s*\d*(?:[.,]\d+)?\s*(?:kg|g|mg|lb|lbs|oz|ml|cl|l|un|unites?|ea|each|pk|ch)\b")
30+_NUM = re.compile(r"\d+(?:[.,]\d+)?")
31+STOP = {"de","du","des","la","le","les","l","d","et","a","au","aux","en","pour","the","of","and","with","un","une",
32+ "sans","avec","ou","or","pc","chaque","each","marque","brand"}
33+GTIN_FEEDS = {"metro", "superc"}
34+MULTI_RE = re.compile(r"\b(ou|or|assortis?|assorted|selectionne?s?|varietes?|au choix|et plus|et autres)\b|,|/", re.I)
35+
36+def strip_accents(s: str) -> str:
37+ return "".join(c for c in unicodedata.normalize("NFKD", s or "") if not unicodedata.combining(c))
38+
39+def brand_key(b: str) -> str:
40+ k = _ALNUM.sub("", strip_accents(b or "").lower())
41+ return k if len(k) >= 2 else ""
42+
43+def name_key(name: str, brand: str = "") -> str:
44+ s = strip_accents(name or "").lower().replace(" ", " ")
45+ s = _SIZE_TOK.sub(" ", s); s = _ALNUM.sub(" ", s)
46+ drop = set(STOP)
47+ if brand:
48+ drop |= set(_ALNUM.sub(" ", strip_accents(brand).lower()).split())
49+ toks = [t for t in s.split() if t and t not in drop]
50+ return " ".join(sorted(toks))
51+
52+def num_tokens(nk: str) -> frozenset:
53+ return frozenset(x.replace(",", ".") for x in _NUM.findall(nk))
54+
55+def size_bucket(qty: float, base: str, pack=1) -> str:
56+ """Package key: total quantity in base unit AND multipack count ('12x355 ml' != '6x710 ml')."""
57+ if pd.isna(qty) or base is None: return ""
58+ pk = 1 if pd.isna(pack) else int(pack)
59+ return f"{round(float(qty), 1):g}|{base}|x{pk}"
60+
61+def prepare(prod: pd.DataFrame, ret: pd.DataFrame) -> pd.DataFrame:
62+ excl = set(SETTINGS["scope"]["exclude_sources"])
63+ p = prod[~prod.source.isin(excl)].copy()
64+ p = p[p.name.fillna("").str.len() > 0]
65+ p = p.merge(ret[["feed_id", "banner", "company", "channel"]], left_on="source", right_on="feed_id", how="left")
66+ p["bkey"] = p.brand.fillna("").map(brand_key)
67+ resid = p.size_label.fillna("").map(size_residual)
68+ p["nkey"] = [(name_key(n, b) + " " + r).strip() for n, b, r in zip(p.name.fillna(""), p.brand.fillna(""), resid)]
69+ p["nkey"] = p.nkey.map(lambda k: " ".join(sorted(k.split())))
70+ p["nkey_full"] = [(name_key(n, "") + " " + r).strip() for n, r in zip(p.name.fillna(""), resid)]
71+ p["nkey_full"] = p.nkey_full.map(lambda k: " ".join(sorted(k.split())))
72+ p["ntoks"] = p.nkey.map(num_tokens)
73+ p["skey"] = [size_bucket(q, b, k) for q, b, k in zip(p.size_qty, p.size_base, p.size_pack)]
74+ p["gtin"] = np.where(p.source.isin(GTIN_FEEDS) & p.gtin_valid, p.external_id.str.lstrip("0"), None)
75+ p["multi_offer"] = p.name.fillna("").map(lambda n: bool(MULTI_RE.search(strip_accents(n))))
76+ return p
77+
78+def tier_a(p: pd.DataFrame) -> pd.DataFrame:
79+ g = p[p.gtin.notna()]
80+ pairs = []
81+ for gt, grp in g.groupby("gtin"):
82+ if len(grp) < 2: continue
83+ for i, j in itertools.combinations(grp.index, 2):
84+ if grp.at[i, "banner"] == grp.at[j, "banner"] and grp.at[i, "channel"] == grp.at[j, "channel"]: continue
85+ pairs.append((grp.at[i, "uid"], grp.at[j, "uid"], "A", 1.0, True, True, 1.0))
86+ return pd.DataFrame(pairs, columns=["uid_a", "uid_b", "tier", "score", "brand_match", "size_match", "name_sim"])
87+
88+def _block_pairs(block: pd.DataFrame, thr: float) -> list:
89+ out = []
90+ if len(block) < 2 or len(block) > 4000: return out
91+ names = block.nkey.tolist(); uids = block.uid.tolist(); banners = block.banner.tolist(); chans = block.channel.tolist()
92+ ntoks = block.ntoks.tolist(); tsets = [frozenset(n.split()) for n in names]
93+ sim = process.cdist(names, names, scorer=fuzz.token_sort_ratio, workers=-1) / 100.0
94+ iu = np.triu_indices(len(block), k=1)
95+ for i, j in zip(*iu):
96+ s = sim[i, j]
97+ if s < thr: continue
98+ if ntoks[i] != ntoks[j]: continue
99+ if banners[i] == banners[j] and chans[i] == chans[j]: continue
100+ out.append((uids[i], uids[j], float(s), tsets[i] == tsets[j]))
101+ return out
102+
103+def tier_bc(p: pd.DataFrame) -> pd.DataFrame:
104+ rows = []
105+ q = p[(p.bkey != "") & (p.skey != "") & (~p.multi_offer)]
106+ for (bk, sk), block in q.groupby(["bkey", "skey"]):
107+ for ua, ub, s, eq in _block_pairs(block, 0.90):
108+ rows.append((ua, ub, "B" if eq else "C", s, True, True, s))
109+ return pd.DataFrame(rows, columns=["uid_a", "uid_b", "tier", "score", "brand_match", "size_match", "name_sim"])
110+
111+def tier_d(p: pd.DataFrame) -> pd.DataFrame:
112+ """Brandless feeds: block on size, compare full names (brand kept in the name), require the branded side's brand
113+ tokens to appear in the brandless name."""
114+ rows = []
115+ q = p[(p.skey != "") & (~p.multi_offer)]
116+ for sk, block in q.groupby("skey"):
117+ bl = block[block.bkey == ""]; br = block[block.bkey != ""]
118+ if len(bl) == 0 or len(br) == 0 or len(block) > 6000: continue
119+ sim = process.cdist(bl.nkey_full.tolist(), br.nkey_full.tolist(), scorer=fuzz.token_sort_ratio, workers=-1) / 100.0
120+ ii, jj = np.where(sim >= 0.92)
121+ for i, j in zip(ii, jj):
122+ a, b = bl.iloc[i], br.iloc[j]
123+ if a.banner == b.banner and a.channel == b.channel: continue
124+ if num_tokens(a.nkey_full) != num_tokens(b.nkey_full): continue
125+ btoks = [t for t in _ALNUM.sub(" ", strip_accents(b.brand).lower()).split() if len(t) >= 3]
126+ if btoks and not all(t in a.nkey_full.split() for t in btoks): continue
127+ rows.append((a.uid, b.uid, "D", float(sim[i, j]), False, True, float(sim[i, j])))
128+ return pd.DataFrame(rows, columns=["uid_a", "uid_b", "tier", "score", "brand_match", "size_match", "name_sim"])
129+
130+def union_groups(pairs: pd.DataFrame, tiers: set[str]) -> pd.DataFrame:
131+ parent = {}
132+ def find(x):
133+ while parent.setdefault(x, x) != x:
134+ parent[x] = parent[parent[x]]; x = parent[x]
135+ return x
136+ for ua, ub in pairs.loc[pairs.tier.isin(tiers), ["uid_a", "uid_b"]].itertuples(index=False):
137+ ra, rb = find(ua), find(ub)
138+ if ra != rb: parent[ra] = rb
139+ rows = [(hashlib.md5(find(u).encode()).hexdigest()[:12], u) for u in parent]
140+ return pd.DataFrame(rows, columns=["group_id", "uid"])
141+
142+def main() -> None:
143+ seed_everything()
144+ prod = pd.read_parquet(path("processed") / "products.parquet")
145+ ret = pd.read_parquet(path("processed") / "retailers.parquet")
146+ p = prepare(prod, ret)
147+ log.info("matching universe: %d listings (active+inactive, excl. %s); with size %d; multi-product offers excluded %d",
148+ len(p), SETTINGS["scope"]["exclude_sources"], (p.skey != "").sum(), p.multi_offer.sum())
149+ parts = [tier_a(p), tier_bc(p), tier_d(p)]
150+ pairs = pd.concat(parts, ignore_index=True)
151+ # keep the best tier per unordered pair
152+ pairs["key"] = [tuple(sorted(x)) for x in zip(pairs.uid_a, pairs.uid_b)]
153+ order = {"A": 0, "B": 1, "C": 2, "D": 3}
154+ pairs["rank"] = pairs.tier.map(order)
155+ pairs = pairs.sort_values(["key", "rank"]).drop_duplicates("key").drop(columns=["key", "rank"])
156+ meta = p.set_index("uid")[["banner", "company", "channel", "source", "category", "active"]]
157+ for side in ("a", "b"):
158+ m = meta.loc[pairs[f"uid_{side}"]].reset_index(drop=True); m.columns = [f"{c}_{side}" for c in m.columns]
159+ pairs = pd.concat([pairs.reset_index(drop=True), m], axis=1)
160+ pairs["same_company"] = pairs.company_a == pairs.company_b
161+ pairs["link_type"] = np.where(pairs.banner_a == pairs.banner_b, "channel", "cross_banner")
162+ pairs["both_active"] = (pairs.active_a == 1) & (pairs.active_b == 1)
163+ pairs.to_parquet(path("processed") / "product_matches.parquet", index=False)
164+ log.info("pairs: %s ; cross-banner by tier: %s", len(pairs),
165+ pairs[pairs.link_type == "cross_banner"].tier.value_counts().to_dict())
166+ for name, tiers in {"A": {"A"}, "AB": {"A", "B"}, "ABC": {"A", "B", "C"}, "ABCD": {"A", "B", "C", "D"}}.items():
167+ g = union_groups(pairs[pairs.link_type == "cross_banner"], tiers)
168+ g = g.merge(meta[["banner"]], left_on="uid", right_index=True)
169+ nb = g.groupby("group_id").banner.nunique()
170+ g = g[g.group_id.isin(nb[nb >= 2].index)]
171+ g.to_parquet(path("processed") / f"match_groups_{name}.parquet", index=False)
172+ log.info("groups tiers %-4s: %d groups, %d listings, median banners/group %.0f, max %d", name,
173+ g.group_id.nunique(), len(g), nb[nb >= 2].median(), nb.max())
174+ # Food-Ka's own links for comparison
175+ fk = pd.read_parquet(path("interim") / "raw_product_links.parquet")
176+ fkpairs = set()
177+ for gid, grp in fk.groupby("group_id"):
178+ for a, b in itertools.combinations(sorted(grp.uid), 2): fkpairs.add((a, b))
179+ ours = set(tuple(sorted(x)) for x in zip(pairs.uid_a, pairs.uid_b))
180+ log.info("Food-Ka product_links pairs: %d ; recovered by our tiers A-D: %d (%.1f%%)", len(fkpairs),
181+ len(fkpairs & ours), 100 * len(fkpairs & ours) / max(len(fkpairs), 1))
182+ # stratified validation sample: 40 per tier (A,B,C,D) × link_type cross_banner, with listing text for annotators
183+ rng = np.random.default_rng(SEED)
184+ cb = pairs[pairs.link_type == "cross_banner"]
185+ samp = cb.groupby("tier", group_keys=False).apply(lambda g: g.sample(min(len(g), 60), random_state=SEED))
186+ txt = prod.set_index("uid")[["name", "brand", "size_label", "price", "source"]]
187+ for side in ("a", "b"):
188+ t = txt.loc[samp[f"uid_{side}"]].reset_index(drop=True); t.columns = [f"{c}_{side}_txt" for c in t.columns]
189+ samp = pd.concat([samp.reset_index(drop=True), t], axis=1)
190+ samp = samp.sample(frac=1, random_state=SEED).reset_index(drop=True)
191+ samp.insert(0, "sample_id", range(1, len(samp) + 1))
192+ samp["label_llm"] = ""; samp["label_human"] = ""; samp["note"] = ""
193+ samp.to_csv(path("processed") / "match_validation_sample.csv", index=False)
194+ log.info("validation sample: %d pairs (%s)", len(samp), samp.tier.value_counts().to_dict())
195+ ledger("1", "match_products", f"{len(pairs)} pairs; sample {len(samp)}")
196+
197+if __name__ == "__main__":
198+ main()
added src/matching/recall_gtin.py +44 −0
@@ -0,0 +1,44 @@
1+"""Recall of the name-based tiers (B/C/D) against GTIN ground truth (tier A pairs, Metro <-> Super C).
2+
3+Ground truth: listing pairs sharing a GS1-valid UPC across the two Metro Inc. banners. For each truth pair we ask
4+whether the name/brand/size rules would have linked it without the GTIN. Also reports *false links* among GTIN-bearing
5+listings: name-based pairs between metro and superc whose UPCs differ (a precision check on a labelled subset).
6+"""
7+import pandas as pd, numpy as np
8+from rapidfuzz import fuzz
9+from src.common import path, get_logger
10+from src.matching.match_products import prepare, num_tokens
11+
12+log = get_logger("recall")
13+
14+def main():
15+ prod = pd.read_parquet(path("processed") / "products.parquet"); ret = pd.read_parquet(path("processed") / "retailers.parquet")
16+ p = prepare(prod, ret); g = p[p.gtin.notna()].copy()
17+ pairs = pd.read_parquet(path("processed") / "product_matches.parquet")
18+ truth = pairs[(pairs.tier == "A")][["uid_a", "uid_b"]]
19+ idx = g.set_index("uid")
20+ rows = []
21+ for ua, ub in truth.itertuples(index=False):
22+ a, b = idx.loc[ua], idx.loc[ub]
23+ brand_ok = a.bkey != "" and a.bkey == b.bkey
24+ size_ok = a.skey != "" and a.skey == b.skey
25+ sim = fuzz.token_set_ratio(a.nkey, b.nkey) / 100.0
26+ nums_ok = num_tokens(a.nkey) == num_tokens(b.nkey)
27+ tier = None
28+ if brand_ok and size_ok and nums_ok and sim >= 0.90: tier = "B"
29+ elif brand_ok and size_ok and nums_ok and sim >= 0.80: tier = "C"
30+ elif brand_ok and a.skey == "" and b.skey == "" and nums_ok and sim >= 0.95: tier = "C"
31+ 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,
32+ brand_missing=(a.bkey == "" or b.bkey == ""), size_missing=(a.skey == "" or b.skey == "")))
33+ r = pd.DataFrame(rows)
34+ 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())
35+ 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",
36+ *(r[r.recovered_tier.isna()][c].mean() for c in ["brand_missing"]),
37+ ((~r.brand_ok) & (~r.brand_missing))[r.recovered_tier.isna()].mean(),
38+ r[r.recovered_tier.isna()].size_missing.mean(),
39+ ((~r.size_ok) & (~r.size_missing))[r.recovered_tier.isna()].mean(),
40+ (~r[r.recovered_tier.isna()].nums_ok).mean(), (r[r.recovered_tier.isna()].sim < 0.80).mean())
41+ r.to_csv(path("processed") / "recall_gtin_truth.csv", index=False)
42+
43+if __name__ == "__main__":
44+ main()
added tests/test_matching.py +13 −0
@@ -0,0 +1,13 @@
1+import pandas as pd
2+from src.matching.match_products import name_key, size_bucket, brand_key, num_tokens, MULTI_RE
3+from src.cleaning.units import parse_size, size_residual, strip_accents
4+
5+def test_keys():
6+ assert brand_key("Häagen-Dazs") == "haagendazs"
7+ assert name_key("Beurre végétal salé 454 g", "Becel") != name_key("Beurre végétal non salé 454 g", "Becel")
8+ assert num_tokens(name_key("Lait 2 %", "")) != num_tokens(name_key("Lait 3,25 %", ""))
9+ a, b = parse_size("12 x 355 ml"), parse_size("6 x 710 ml")
10+ assert a.qty == b.qty and size_bucket(a.qty, a.base, a.pack) != size_bucket(b.qty, b.base, b.pack)
11+ assert size_residual("340 g, calibre 21-25") == "21-25 calibre".replace("-", " ") or "calibre" in size_residual("340 g, calibre 21-25")
12+ assert "cons" not in size_residual("12x355 mL, Cons. 1,20 $/emb.") and size_residual("6x710.0 ml, 0,15 $/100ml") == ""
13+ assert MULTI_RE.search(strip_accents("BISCUITS DARE WHIPPET OU ULTIME")) and not MULTI_RE.search("Yogourt grec vanille")
modified tests/test_panel.py +1 −1
@@ -30,7 +30,7 @@ def test_known_sequence(d):
30 x = d[d.uid == "metro:068400662600"].sort_values("day") 30 x = d[d.uid == "metro:068400662600"].sort_values("day")
31 assert x.day.min() == pd.Timestamp("2026-08-12").date() 31 assert x.day.min() == pd.Timestamp("2026-08-12").date()
32 assert set(x.loc[x.day == pd.Timestamp("2026-08-13").date(), "p_min"]) == {5.77} 32 assert set(x.loc[x.day == pd.Timestamp("2026-08-13").date(), "p_min"]) == {5.77}
33− assert (x.hours_obs.iloc[1:-1] == 24.0).all() 33+ assert np.allclose(x.hours_obs.iloc[1:-1], 24.0, atol=1e-6)
34 34
35 def test_single_price_products_have_one_distinct(d): 35 def test_single_price_products_have_one_distinct(d):
36 pl = pd.read_parquet(I / "raw_price_log.parquet") 36 pl = pd.read_parquet(I / "raw_price_log.parquet")
37 37
Menu

Theme
↑↓ navigate ↵ open esc close / or ⌘K anywhere