1"""Phase 2.4 — Audit of the rule-based category labels on a stratified random sample.23Step 1 (sample) : 25 listings per category (20 categories) from the assortment sample, seed 202611014 -> data/processed/category_audit_sample.csv (NEVER overwritten once annotations exist)5Step 2 (annotate) : two independent LLM annotators (Anthropic claude-opus-5-5, OpenAI gpt-5.5) label each listing with one of6 the 20 canonical keys using the platform's own taxonomy definitions; every call is logged with usage.7Step 3 (report) : agreement matrices, per-category precision of the rule label against the LLM consensus,8 list of disagreements for operator adjudication -> experiments/category_audit.json910Usage: python -m src.validation.category_audit sample | annotate | report11"""12from __future__ import annotations13import json, os, sys, time, concurrent.futures as cf14import pandas as pd15from dotenv import load_dotenv16from src.common import ROOT, P, SEED, get_logger, ledger, register_output, write_json, BASE_PROC, BASE_EXP1718log = get_logger("validation.category_audit")19load_dotenv(ROOT / ".env")20SAMPLE = BASE_PROC / "category_audit_sample.csv"21TAXO = {22 "erable": "Érable & miel: sirop, beurre, flocons d'érable; miel, pollen, hydromel non alcoolisé",23 "epicerie": "Épicerie fine: confitures, sauces, condiments, épices, vinaigres, huiles, farines, granolas, collations, marinades, produits d'épicerie en général (hors catégories plus spécifiques)",24 "cafe_the": "Café & thé: café torréfié, thés, tisanes, matcha, infusions",25 "chocolat": "Chocolat & confiseries: chocolat, caramels, guimauves, bonbons",26 "boulangerie": "Boulangerie & pâtisserie: pains, biscuits, pâtisseries",27 "viande_poisson": "Viandes & poissons: boucherie, charcuterie, jerky, gibier, fumoir, poissons, fruits de mer",28 "fromage_laitier": "Fromages & produits laitiers: fromages, yogourts, crème glacée",29 "boissons": "Boissons non alcoolisées: kombucha, limonade, jus, sodas, eau",30 "alcool": "Bières, vins, cidres, spiritueux, hydromel alcoolisé",31 "sante_beaute": "Beauté & soins: savons, cosmétiques, soins du corps/barbe/cheveux, parfums, suppléments et produits de santé",32 "maison": "Maison & déco: chandelles, céramique, ébénisterie, textiles de maison, literie, vaisselle, décoration, meubles",33 "mode": "Mode & accessoires: vêtements, tuques, lainages, maroquinerie, chaussures, sacs, accessoires de mode, lunettes",34 "bijoux": "Bijoux: joaillerie, bijoux faits main, montres",35 "art": "Art & artisanat: estampes, affiches, illustrations, photographie, papeterie, verre soufflé, sculpture, macramé, livres et objets culturels, fournitures d'artisanat",36 "enfants": "Enfants & bébés: jouets, doudous, vêtements enfants, puériculture",37 "animaux": "Animaux: gâteries, colliers, laisses, lits, jouets pour animaux, aliments pour animaux",38 "plein_air": "Plein air & sport: camping, raquettes, chasse/pêche, vélo, équipement sportif",39 "jardin": "Jardin & agriculture: semences, plantes, serres, compost, engrais, outils de jardin",40 "industriel": "Manufacturier & spécialisé: outillage, équipements, quincaillerie, composants, pièces, véhicules et remorques, matériaux de construction, fournitures professionnelles",41 "autre": "Autre: ne correspond clairement à aucune catégorie ci-dessus",42}43SYSTEM = ("Tu es un annotateur de catalogue. On te donne une fiche produit d'une boutique en ligne québécoise. Choisis la catégorie canonique la plus "44 "appropriée parmi la taxonomie fournie (une seule clé). Réponds UNIQUEMENT en JSON: {\"category\": \"<clé>\", \"confidence\": <0-1>}.\n\nTaxonomie:\n" +45 "\n".join(f"- {k}: {v}" for k, v in TAXO.items()))464748def fiche(r) -> str:49 return (f"Titre: {r.title}\nType (marchand): {r.product_type or ''}\nCatégories marchand: {r.merchant_categories or ''}\n"50 f"Tags: {r.tags_list}\nMarque/vendor: {r.vendor or ''}\nPrix: {r.price} {r.currency}\nDescription (début): {str(r.description_head or '')[:300]}")515253def sample():54 if SAMPLE.exists():55 log.error("%s exists — refusing to overwrite an annotated sample", SAMPLE); return56 li = pd.read_parquet(P["interim"] / "listings_clean.parquet")57 a = li[li.f_analysis_assort]58 s = a.groupby("category", group_keys=False).apply(lambda g: g.sample(min(25, len(g)), random_state=SEED))59 cols = ["uid", "store_id", "category", "title", "product_type", "merchant_categories", "tags_list", "vendor", "price", "currency", "description_head"]60 s[cols].to_csv(SAMPLE, index=False)61 log.info("sample %s rows -> %s", len(s), SAMPLE)626364def _anthropic(text, model="claude-opus-5-5"):65 import anthropic66 c = anthropic.Anthropic()67 t0 = time.time()68 m = c.messages.create(model=model, max_tokens=600, thinking={"type": "adaptive"}, output_config={"effort": "low"}, system=SYSTEM, messages=[{"role": "user", "content": text}])69 out = next((b.text for b in m.content if getattr(b, "type", "") == "text"), "").strip()70 return out, {"in": m.usage.input_tokens, "out": m.usage.output_tokens, "s": round(time.time() - t0, 2)}717273def _openai(text, model="gpt-5.5"):74 from openai import OpenAI75 c = OpenAI()76 t0 = time.time()77 m = c.chat.completions.create(model=model, messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": text}],78 max_completion_tokens=400, response_format={"type": "json_object"})79 out = (m.choices[0].message.content or "").strip()80 return out, {"in": m.usage.prompt_tokens, "out": m.usage.completion_tokens, "s": round(time.time() - t0, 2)}818283def parse(out):84 try:85 j = json.loads(out[out.find("{"): out.rfind("}") + 1])86 k = str(j.get("category", "")).strip().lower()87 return (k if k in TAXO else "INVALID"), float(j.get("confidence", float("nan")))88 except Exception:89 return "PARSE_ERROR", float("nan")909192def annotate():93 df = pd.read_csv(SAMPLE)94 for col, fn in (("llm_a", _anthropic), ("llm_b", _openai)):95 if col in df.columns and df[col].notna().all() and not df[col].isin(["PARSE_ERROR", "INVALID"]).any():96 log.info("%s already annotated", col); continue97 if col not in df.columns:98 df[col] = None; df[col + "_conf"] = None99 df.loc[df[col].isin(["PARSE_ERROR", "INVALID"]), col] = None100 todo = df.index[df[col].isna()].tolist()101 usage = {"in": 0, "out": 0, "calls": 0, "errors": 0}102 def work(i):103 r = df.loc[i]104 for attempt in range(6):105 try:106 out, u = fn(fiche(r)); return i, out, u107 except Exception as e: # noqa: BLE001108 time.sleep(min(60, 3 * 2 ** attempt)); err = str(e)[:120]109 return i, f"ERROR {err}", {"in": 0, "out": 0, "s": 0}110 with cf.ThreadPoolExecutor(6) as ex:111 for i, out, u in ex.map(work, todo):112 k, conf = parse(out); df.loc[i, col] = k; df.loc[i, col + "_conf"] = conf113 usage["in"] += u["in"]; usage["out"] += u["out"]; usage["calls"] += 1; usage["errors"] += out.startswith("ERROR")114 df.to_csv(SAMPLE, index=False)115 ledger("category_audit.annotate", annotator=col, model=("claude-opus-5-5" if col == "llm_a" else "gpt-5.5"), **usage)116 log.info("%s done: %s", col, usage)117118119def report():120 df = pd.read_csv(SAMPLE)121 out = {"n": len(df)}122 for a, b in (("category", "llm_a"), ("category", "llm_b"), ("llm_a", "llm_b")):123 out[f"agree_{a}_{b}"] = round(float((df[a] == df[b]).mean()), 4)124 cons = df.llm_a.where(df.llm_a == df.llm_b)125 out["consensus_share"] = round(float(cons.notna().mean()), 4)126 sub = df[cons.notna()]127 out["rule_precision_vs_consensus"] = round(float((sub.category == sub.llm_a).mean()), 4)128 out["per_category_precision_vs_consensus"] = {k: {"n_consensus": int((sub.category == k).sum()), "precision": round(float((sub[sub.category == k].llm_a == k).mean()), 3) if (sub.category == k).any() else None} for k in TAXO}129 conf = pd.crosstab(sub.category, sub.llm_a)130 out["confusion_rule_vs_consensus"] = {r: {c: int(v) for c, v in row.items() if v} for r, row in conf.to_dict("index").items()}131 dis = df[(df.category != df.llm_a) | (df.category != df.llm_b)]132 dis[["uid", "store_id", "category", "llm_a", "llm_b", "title", "product_type"]].to_csv(BASE_PROC / "category_audit_disagreements.csv", index=False)133 out["n_disagreements"] = len(dis)134 write_json(out, BASE_EXP / "category_audit.json", "category_audit.report")135 log.info(json.dumps({k: v for k, v in out.items() if "per_" not in k and "confusion" not in k}))136137138if __name__ == "__main__":139 {"sample": sample, "annotate": annotate, "report": report}[sys.argv[1]]()140