1"""Model-based category labels (Amendment A-002): the platform's keyword rules reach only ~46 % precision against annotator2consensus, so we train a supervised classifier on LLM-annotated listings and validate it on the independent 500-listing audit3sample (never used for training).45Steps (python -m src.features.category_model <step>):6 sample 6 000 listings stratified by rule category (proportional to size, floor 150; audit-sample uids excluded) -> data/processed/category_train_sample.csv7 annotate claude-sonnet-5-5 (adaptive thinking, low effort) labels each listing with one of the 20 keys; usage logged; never overwritten8 embed OpenAI text-embedding-3-small on "title | product_type | merchant_categories | vendor | tags" for ALL assortment listings9 -> data/interim/listing_embeddings.npy (+ uid index); cached10 train multinomial logistic regression on [embedding, one-hot rule label]; 5-fold CV; evaluation on the audit consensus sample11 -> data/processed/category_model.parquet (uid, category_model, p_max), experiments/category_model.json12"""13from __future__ import annotations14import json, sys, time, concurrent.futures as cf15import numpy as np, pandas as pd16from dotenv import load_dotenv17from src.common import ROOT, P, SEED, get_logger, ledger, register_output, write_json, BASE_PROC, BASE_EXP18from src.validation.category_audit import TAXO, SYSTEM, fiche, parse, SAMPLE as AUDIT1920log = get_logger("features.category_model"); load_dotenv(ROOT / ".env")21PROC = BASE_PROC; TRAIN = PROC / "category_train_sample.csv"; EMB = P["interim"] / "listing_embeddings.npy"; EMB_IDX = P["interim"] / "listing_embeddings_uids.csv"22N_TRAIN, FLOOR = 6000, 150232425def sample():26 if TRAIN.exists(): log.error("%s exists — refusing to overwrite", TRAIN); return27 li = pd.read_parquet(P["interim"] / "listings_clean.parquet"); a = li[li.f_analysis_assort]28 audit = set(pd.read_csv(AUDIT).uid); a = a[~a.uid.isin(audit)]29 share = a.category.value_counts(normalize=True); n = (share * N_TRAIN).round().clip(lower=FLOOR).astype(int)30 s = pd.concat([g.sample(min(int(n[c]), len(g)), random_state=SEED) for c, g in a.groupby("category")])31 cols = ["uid", "store_id", "category", "title", "product_type", "merchant_categories", "tags_list", "vendor", "price", "currency", "description_head"]32 s[cols].to_csv(TRAIN, index=False); log.info("train sample %s", len(s))333435def _label(text):36 import anthropic37 m = anthropic.Anthropic().messages.create(model="claude-sonnet-5-5", max_tokens=600, thinking={"type": "adaptive"}, output_config={"effort": "low"}, system=SYSTEM, messages=[{"role": "user", "content": text}])38 return next((b.text for b in m.content if getattr(b, "type", "") == "text"), ""), {"in": m.usage.input_tokens, "out": m.usage.output_tokens}394041def annotate():42 df = pd.read_csv(TRAIN)43 if "llm" not in df.columns: df["llm"] = None44 df.loc[df.llm.isin(["PARSE_ERROR", "INVALID"]), "llm"] = None45 todo = df.index[df.llm.isna()].tolist(); usage = {"in": 0, "out": 0, "calls": 0, "errors": 0}; t0 = time.time()46 def work(i):47 r = df.loc[i]48 for k in range(6):49 try: return i, *_label(fiche(r))50 except Exception as e: time.sleep(min(60, 3 * 2 ** k)); err = str(e)[:100]51 return i, f"ERROR {err}", {"in": 0, "out": 0}52 with cf.ThreadPoolExecutor(8) as ex:53 for j, (i, out, u) in enumerate(ex.map(work, todo)):54 df.loc[i, "llm"] = parse(out)[0]; usage["in"] += u["in"]; usage["out"] += u["out"]; usage["calls"] += 1; usage["errors"] += out.startswith("ERROR")55 if j % 500 == 499: df.to_csv(TRAIN, index=False); log.info("%s/%s %.0fs", j + 1, len(todo), time.time() - t0)56 df.to_csv(TRAIN, index=False); ledger("category_model.annotate", model="claude-sonnet-5-5", **usage); log.info("annotate done %s", usage)575859def text_of(r):60 return " | ".join(str(x) for x in (r.title, r.product_type or "", r.merchant_categories or "", r.vendor or "", r.tags_list if r.tags_list != "[]" else "") if x)[:600]616263def embed():64 from openai import OpenAI65 li = pd.read_parquet(P["interim"] / "listings_clean.parquet"); a = li[li.f_analysis_assort | li.uid.isin(pd.read_csv(AUDIT).uid)]66 texts = [text_of(r) for r in a.itertuples()]; uids = a.uid.tolist()67 done = {}68 if EMB.exists():69 old = np.load(EMB); oid = pd.read_csv(EMB_IDX).uid.tolist(); done = dict(zip(oid, old))70 todo = [i for i, u in enumerate(uids) if u not in done]; log.info("embedding %s of %s listings", len(todo), len(uids))71 client = OpenAI(); usage = 0; t0 = time.time()72 def batch(idx):73 for k in range(5):74 try:75 r = client.embeddings.create(model="text-embedding-3-small", input=[texts[i] for i in idx], dimensions=512); return idx, [e.embedding for e in r.data], r.usage.total_tokens76 except Exception as e: time.sleep(min(60, 3 * 2 ** k)); err = str(e)[:100]77 raise RuntimeError(err)78 batches = [todo[i:i + 256] for i in range(0, len(todo), 256)]79 with cf.ThreadPoolExecutor(8) as ex:80 for j, (idx, vecs, tok) in enumerate(ex.map(batch, batches)):81 for i, v in zip(idx, vecs): done[uids[i]] = np.asarray(v, np.float32)82 usage += tok83 if j % 200 == 199: log.info("%s/%s batches %.0fs", j + 1, len(batches), time.time() - t0)84 M = np.stack([done[u] for u in uids]); np.save(EMB, M); pd.DataFrame({"uid": uids}).to_csv(EMB_IDX, index=False)85 ledger("category_model.embed", model="text-embedding-3-small", listings=len(uids), tokens=usage); log.info("embeddings saved %s tokens", usage)868788def train():89 from sklearn.linear_model import LogisticRegression90 from sklearn.model_selection import cross_val_predict91 M = np.load(EMB); uids = pd.read_csv(EMB_IDX).uid.tolist(); pos = {u: i for i, u in enumerate(uids)}92 li = pd.read_parquet(P["interim"] / "listings_clean.parquet").set_index("uid")93 cats = list(TAXO); onehot = lambda s: np.eye(len(cats))[[cats.index(c) if c in cats else cats.index("autre") for c in s]]94 tr = pd.read_csv(TRAIN); tr = tr[tr.llm.isin(cats)]95 Xtr = np.hstack([M[[pos[u] for u in tr.uid]], onehot(tr.category)]); ytr = tr.llm.values96 clf = LogisticRegression(max_iter=2000, C=2.0, n_jobs=-1)97 cvp = cross_val_predict(clf, Xtr, ytr, cv=5); cv_acc = float((cvp == ytr).mean())98 clf.fit(Xtr, ytr)99 au = pd.read_csv(AUDIT); cons = au[au.llm_a == au.llm_b]; Xau = np.hstack([M[[pos[u] for u in cons.uid]], onehot(cons.category)]); pred_au = clf.predict(Xau)100 acc_au = float((pred_au == cons.llm_a).mean()); rule_au = float((cons.category == cons.llm_a).mean())101 # listing-weighted accuracy (audit sample is stratified 25/category -> reweight by assortment shares of the RULE label)102 a = li[li.f_analysis_assort]; w = a.category.value_counts(normalize=True); ww = cons.category.map(w).values; ww = ww / ww.sum()103 acc_au_w = float(((pred_au == cons.llm_a) * ww).sum()); rule_au_w = float(((cons.category == cons.llm_a) * ww).sum())104 per = {c: {"n": int((cons.llm_a == c).sum()), "recall_model": round(float((pred_au[cons.llm_a == c] == c).mean()), 3) if (cons.llm_a == c).any() else None, "precision_model": round(float((cons.llm_a[pred_au == c] == c).mean()), 3) if (pred_au == c).any() else None} for c in cats}105 conf = pd.crosstab(cons.llm_a.rename("annotated"), pd.Series(pred_au, index=cons.index, name="model"))106 # predict all assortment listings107 allu = a.index.tolist(); Xall = np.hstack([M[[pos[u] for u in allu]], onehot(a.category)]); prob = clf.predict_proba(Xall); pred = clf.classes_[prob.argmax(1)]108 out = pd.DataFrame({"uid": allu, "category_rule": a.category.values, "category_model": pred, "p_max": prob.max(1)})109 out.to_parquet(PROC / "category_model.parquet", index=False); register_output(PROC / "category_model.parquet", "category_model")110 res = {"train_n": int(len(tr)), "cv5_accuracy_vs_llm": round(cv_acc, 4), "audit_consensus_n": int(len(cons)), "accuracy_model_on_audit": round(acc_au, 4), "accuracy_rule_on_audit": round(rule_au, 4),111 "listing_weighted_accuracy_model": round(acc_au_w, 4), "listing_weighted_accuracy_rule": round(rule_au_w, 4), "share_relabelled": round(float((out.category_rule != out.category_model).mean()), 4),112 "category_shares_rule": a.category.value_counts(normalize=True).round(4).to_dict(), "category_shares_model": out.category_model.value_counts(normalize=True).round(4).to_dict(),113 "per_category_on_audit": per, "confusion_audit_annotated_vs_model": {r: {c: int(v) for c, v in row.items() if v} for r, row in conf.to_dict("index").items()}, "features": "text-embedding-3-small (512 d) + one-hot rule label; multinomial logistic regression C=2"}114 write_json(res, BASE_EXP / "category_model.json", "category_model"); ledger("category_model.train", **{k: v for k, v in res.items() if "per_" not in k and "confusion" not in k and "shares" not in k})115 log.info(json.dumps({k: v for k, v in res.items() if "per_" not in k and "confusion" not in k and "shares" not in k}))116117118if __name__ == "__main__":119 {"sample": sample, "annotate": annotate, "embed": embed, "train": train}[sys.argv[1]]()120