1"""Phase 2.6 — Validation of the cross-merchant matching tiers on a labelled pair sample.23sample : 60 random pairs per tier (A, B, C, D), seed 20261101 -> data/processed/match_validation_sample.csv (never overwritten)4annotate : two independent LLM annotators decide whether the two listings are the SAME product (same brand, same item,5 same format) sold by two different merchants: {"same": "yes"|"no"|"unsure", "reason": "..."}; usage logged6report : precision per tier (share 'yes' among consensus labels), disagreement list for operator adjudication7Usage: python -m src.validation.match_validation sample | annotate | report8"""9from __future__ import annotations10import json, sys, time, concurrent.futures as cf11import pandas as pd12from dotenv import load_dotenv13from src.common import ROOT, P, SEED, get_logger, ledger, write_json, BASE_PROC, BASE_EXP1415log = get_logger("validation.match")16load_dotenv(ROOT / ".env")17SAMPLE = BASE_PROC / "match_validation_sample.csv"18SYSTEM = ("Tu es un expert en appariement de catalogues. On te montre deux fiches produit provenant de deux boutiques en ligne québécoises "19 "différentes. Décide si elles décrivent le MÊME produit (même marque/fabricant, même article, même format/contenance ; des variantes de "20 "couleur ou de taille du même article comptent comme le même produit si le prix affiché est le prix de base). Deux produits artisanaux "21 "similaires faits par des fabricants différents ne sont PAS le même produit. Réponds UNIQUEMENT en JSON: "22 "{\"same\": \"yes\"|\"no\"|\"unsure\", \"reason\": \"<10 mots max>\"}")232425def card(r, s):26 return (f"Boutique: {r[f'store_{s}']}\nTitre: {r[f'title_{s}']}\nMarque/vendor: {r[f'vendor_{s}'] or ''}\nSKU: {r[f'sku_{s}'] or ''}\n"27 f"Catégorie marchand: {r[f'mcat_{s}'] or ''}\nPrix: {r[f'price_{s}']} CAD\nVariantes: {r[f'nv_{s}']}\nDescription: {str(r[f'desc_{s}'] or '')[:250]}")282930def sample():31 if SAMPLE.exists():32 log.error("%s exists — refusing to overwrite", SAMPLE); return33 pr = pd.read_parquet(P["interim"] / "match_pairs.parquet")34 li = pd.read_parquet(P["interim"] / "listings_clean.parquet").set_index("uid")35 s = pr.groupby("tier", group_keys=False).apply(lambda g: g.sample(min(60, len(g)), random_state=SEED))36 for side in ("a", "b"):37 u = s[f"uid_{side}"]38 s[f"title_{side}"] = u.map(li.title); s[f"vendor_{side}"] = u.map(li.vendor); s[f"sku_{side}"] = u.map(li.sku)39 s[f"mcat_{side}"] = u.map(li.merchant_categories); s[f"desc_{side}"] = u.map(li.description_head)40 cols = ["uid_a", "uid_b", "tier", "score", "comparable", "store_a", "store_b", "price_a", "price_b", "nv_a", "nv_b"] + \41 [f"{k}_{s}" for s in ("a", "b") for k in ("title", "vendor", "sku", "mcat", "desc")]42 s[cols].to_csv(SAMPLE, index=False); log.info("sample %s pairs", len(s))434445def _anthropic(text):46 import anthropic47 m = anthropic.Anthropic().messages.create(model="claude-opus-5-5", max_tokens=600, thinking={"type": "adaptive"}, output_config={"effort": "low"}, system=SYSTEM, messages=[{"role": "user", "content": text}])48 return next((b.text for b in m.content if getattr(b, "type", "") == "text"), ""), {"in": m.usage.input_tokens, "out": m.usage.output_tokens}495051def _openai(text):52 from openai import OpenAI53 m = OpenAI().chat.completions.create(model="gpt-5.5", messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": text}],54 max_completion_tokens=400, response_format={"type": "json_object"})55 return m.choices[0].message.content or "", {"in": m.usage.prompt_tokens, "out": m.usage.completion_tokens}565758def parse(out):59 try:60 j = json.loads(out[out.find("{"): out.rfind("}") + 1]); v = str(j.get("same", "")).lower()61 return v if v in ("yes", "no", "unsure") else "INVALID"62 except Exception:63 return "PARSE_ERROR"646566def annotate():67 df = pd.read_csv(SAMPLE)68 for col, fn in (("llm_a", _anthropic), ("llm_b", _openai)):69 if col in df.columns and df[col].notna().all() and not df[col].isin(["PARSE_ERROR", "INVALID"]).any():70 continue71 if col not in df.columns: df[col] = None72 df.loc[df[col].isin(["PARSE_ERROR", "INVALID"]), col] = None73 todo = df.index[df[col].isna()].tolist(); usage = {"in": 0, "out": 0, "calls": 0, "errors": 0}74 def work(i):75 r = df.loc[i]; text = "FICHE 1\n" + card(r, "a") + "\n\nFICHE 2\n" + card(r, "b")76 for k in range(6):77 try: return i, *fn(text)78 except Exception as e: time.sleep(min(60, 3 * 2 ** k)); err = str(e)[:100]79 return i, f"ERROR {err}", {"in": 0, "out": 0}80 with cf.ThreadPoolExecutor(6) as ex:81 for i, out, u in ex.map(work, todo):82 df.loc[i, col] = parse(out); usage["in"] += u["in"]; usage["out"] += u["out"]; usage["calls"] += 1; usage["errors"] += out.startswith("ERROR")83 df.to_csv(SAMPLE, index=False); ledger("match_validation.annotate", annotator=col, **usage); log.info("%s %s", col, usage)848586def report():87 df = pd.read_csv(SAMPLE)88 cons = df.llm_a.where(df.llm_a == df.llm_b)89 out = {"n": len(df), "agree_llm": round(float((df.llm_a == df.llm_b).mean()), 4), "consensus_share": round(float(cons.notna().mean()), 4)}90 out["precision_by_tier_consensus"] = {t: {"n": int((cons.notna() & (df.tier == t)).sum()), "yes": round(float((cons[df.tier == t] == "yes").mean()), 3),91 "no": round(float((cons[df.tier == t] == "no").mean()), 3)} for t in sorted(df.tier.unique())}92 out["precision_by_tier_comparable_only"] = {t: round(float((cons[(df.tier == t) & df.comparable] == "yes").mean()), 3) for t in sorted(df.tier.unique())}93 if "operator" in df.columns:94 out["precision_by_tier_operator"] = {t: round(float((df.loc[df.tier == t, "operator"] == "yes").mean()), 3) for t in sorted(df.tier.unique())}95 dis = df[(df.llm_a != df.llm_b) | (df.llm_a == "unsure")]96 dis.to_csv(BASE_PROC / "match_validation_disagreements.csv", index=False); out["n_disagreements"] = len(dis)97 write_json(out, BASE_EXP / "match_validation.json", "match_validation.report"); log.info(json.dumps(out))9899100if __name__ == "__main__":101 {"sample": sample, "annotate": annotate, "report": report}[sys.argv[1]]()102