| """Verify and evaluate engine 3.0 through its REAL code path (raw post/user data -> features -> ONNX -> score). |
| 1. feature parity : engine features == the training files, every validation and test row |
| 2. score parity : engine model outputs == the offline ensemble predictions |
| 3. replay (NDCG@10): validation / test periods (whole feeds and 25-post pages) + the two later P09 sessions, with |
| trained vectors (captions included, as in training) | serving vectors (title + body only, what the app stores) | no vectors |
| vs v2b alone, the warm model alone and random |
| 4. brand-new users : same feeds with no history (interests only) and with no interests |
| Label everywhere: engaged OR active time >= 10 s (the app-style label the models were trained on).""" |
| import os, sys, json, time, collections |
| HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))); sys.path.insert(0, HERE); sys.path.insert(0, os.path.join(HERE, "engine_v2")) |
| import numpy as np, pandas as pd |
| from sklearn.metrics import roc_auc_score |
| from hybrid_engine import HybridRecommender |
| from engine_lib import CHILD, load_post_tags |
| ART = os.path.join(HERE, "hybrid_engine", "artifacts_v3"); OUT = os.path.join(HERE, "out"); DATA = os.path.join(HERE, "recbole_data") |
| eng = HybridRecommender(artifacts_dir=ART); fb = eng.fb; meta = eng.meta; K10 = 10 |
| assert meta["version"] == "hybrid-engine-3.0-bge384" and fb.emb_dim == 384 |
| rd = lambda name, sp: pd.read_csv(os.path.join(DATA, name, f"{name}.{sp}.inter"), sep="\t") |
|
|
| |
| sl = pd.read_csv(os.path.join(OUT, "split_labels_bge.csv")); sl = pd.concat([sl[sl.split == s] for s in ("train", "valid", "test")], ignore_index=True) |
| it = pd.read_csv(os.path.join(OUT, "interactions_v2.csv"), low_memory=False).drop_duplicates(["user_id", "post_id", "ts"]) |
| mm = sl[["user_id", "post_id", "ts", "split", "label"]].merge(it, on=["user_id", "post_id", "ts"], how="left"); assert len(mm) == len(sl) and mm.title.notna().sum() > 0 |
| rr = np.load(os.path.join(OUT, "engine_v3", "real_rows_v3.npz"), allow_pickle=True); assert (rr["post"] == mm.post_id.values).all() |
| Qrows, split, label, user, ts = rr["Q"], rr["split"], rr["label"].astype(int), rr["user"], mm.ts.values |
| z = np.load(os.path.join(OUT, "post_emb_bge384.npz"), allow_pickle=True); EMB = {"fused": dict(zip(z["post_ids"], z["fused"])), "text": dict(zip(z["post_ids"], z["text"])), "none": None} |
| TAGS = {k: {CHILD[j]: 1.0 for j in np.where(v > 0)[0]} for k, v in load_post_tags(os.path.join(OUT, "post_tags.jsonl")).items()} |
| s_ = lambda x: x if isinstance(x, str) else "" |
| def post_dict(i, vm="text"): |
| r = mm.iloc[i] |
| return dict(post_id=r.post_id, score=r.score, comments=r.comment_count, age_hours=None if pd.isna(r.age_hours) else float(r.age_hours), title=s_(r.title), body=s_(r.body), |
| tags=TAGS.get(r.post_id, {}), embedding=None if vm == "none" else EMB[vm][r.post_id], author=r.author_hash if isinstance(r.author_hash, str) else None, impressions=0) |
| def hist_item(i, vm="text"): |
| r = mm.iloc[i] |
| return dict(post_id=r.post_id, ts_ms=float(r.ts), visible_ms=float(r.visible_ms), focal_ms=float(r.active_ms), hover_ms=0.0, clicked=bool(r.clicked), engaged=bool(label[i]), |
| community=None, author=r.author_hash if isinstance(r.author_hash, str) else None, embedding=None if vm == "none" else EMB[vm][r.post_id]) |
| HIST = {vm: {u: [hist_item(i, vm) for i in np.where(user == u)[0]] for u in np.unique(user)} for vm in ("fused", "text", "none")} |
| picks = lambda i: [CHILD[j] for j in np.where(Qrows[i] > 0)[0]] |
|
|
| |
| HP = {u: HybridRecommender._prep_history(h, fb) for u, h in HIST["fused"].items()} |
| worst = {"cold": 0.0, "warm": 0.0}; perf = {}; n_rows = 0 |
| for sp in ("valid", "test"): |
| ce, we = rd("reco_engine_bge", sp), rd("reco_ctr_bge_noid_strict", sp); ix = np.where(split == sp)[0] |
| assert np.allclose(ts[ix] / 1000.0, ce["timestamp:float"].values, atol=1e-3) and np.allclose(ts[ix] / 1000.0, we["timestamp:float"].values, atol=1e-3), "row order" |
| for k, i in enumerate(ix): |
| post = post_dict(i, "fused"); X = post["embedding"] / np.linalg.norm(post["embedding"]); post = {**post, "embedding": X} |
| base = fb.post_floats(post); q, r_, ms = fb.interest_profile(picks(i)); tm = fb.tag_match(fb.tag_vector(post["tags"]), q, r_, ms) |
| zb = {a: fb.z(a, v) for a, v in base.items()}; zt = dict(zip(meta["stage1"]["features"], (tm - fb.s1_mu) / fb.s1_sd)) |
| hf, _ = fb.history_floats(HP[user[i]], post, float(ts[i]), X); zh = {a: fb.z(a, v) for a, v in hf.items()} |
| for f in meta["cold"]["float_inputs"]: d = abs({**zb, **zt}[f] - ce.iloc[k][f"{f}:float"]); worst["cold"] = max(worst["cold"], d); perf["cold:" + f] = max(perf.get("cold:" + f, 0), d) |
| for f in meta["warm"]["float_inputs"]: d = abs({**zb, **zh}[f] - we.iloc[k][f"{f}:float"]); worst["warm"] = max(worst["warm"], d); perf["warm:" + f] = max(perf.get("warm:" + f, 0), d) |
| n_rows += 1 |
| print("1) feature parity over", n_rows, "rows: max |engine - training file| =", {k: round(v, 5) for k, v in worst.items()}, "| worst:", sorted(((k, round(v, 4)) for k, v in perf.items()), key=lambda x: -x[1])[:4], flush=True) |
|
|
| |
| off = np.load(os.path.join(OUT, "engine_v3", "ship3_offline_scores.npz")); SC = {}; t0 = time.time() |
| for sp in ("valid", "test"): |
| for u in np.unique(user[split == sp]): |
| ix = np.where((user == u) & (split == sp))[0]; ix = ix[np.argsort(ts[ix], kind="stable")] |
| res = eng.score({"interests": picks(ix[0]), "history": HIST["fused"][u]}, [post_dict(i, "fused") for i in ix], [float(ts[i]) for i in ix]) |
| for i, d in zip(ix, res): SC[i] = d |
| lat = (time.time() - t0) / len(SC) * 1000 |
| ti = np.where(split == "test")[0]; a_, b_ = meta["cold"]["platt"]["coef"], meta["cold"]["platt"]["intercept"] |
| lgt = np.log(np.clip(off["cold_test"], 1e-6, 1 - 1e-6) / (1 - np.clip(off["cold_test"], 1e-6, 1 - 1e-6))); cold_off = 1 / (1 + np.exp(-(a_ * lgt + b_))) |
| cd = float(np.abs(np.array([SC[i]["cold_probability"] for i in ti]) - cold_off).max()); wd = float(np.abs(np.array([SC[i]["warm_probability"] for i in ti]) - off["warm_test"]).max()) |
| print(f"2) score parity (test rows): cold max diff {cd:.1e} | warm max diff {wd:.1e} | {lat:.2f} ms per candidate incl. history features", flush=True) |
|
|
| |
| def ndcg(order_ids, lab): |
| y = np.array([lab[p] for p in order_ids]); r = np.zeros(K10); r[:min(K10, len(y))] = y[:K10]; disc = 1 / np.log2(np.arange(2, K10 + 2)) |
| return (r * disc).sum() / disc[:min(int(y.sum()), K10)].sum() if y.sum() else np.nan |
| def run(requests, vm_rank=None): |
| """requests: (user, candidates, now, labels). Returns mean NDCG@10 of: shipped engine, v2b alone, warm alone (None if no history), and random.""" |
| out = {"engine": [], "v2b only": [], "warm only": [], "random": []}; rs = np.random.default_rng(0) |
| for u, c, now, lab in requests: |
| if sum(lab.values()) == 0 or len(c) < K10 + 2: continue |
| sc = eng.score(u, c, now); ids = [d["post_id"] for d in sorted(eng.order([dict(d) for d in sc]), key=lambda d: d["position"])] |
| out["engine"].append(ndcg(ids, lab)); out["v2b only"].append(ndcg([d["post_id"] for d in sorted(sc, key=lambda d: -d["v2b"])], lab)) |
| if sc[0]["warm_probability"] is not None: out["warm only"].append(ndcg([d["post_id"] for d in sorted(sc, key=lambda d: -d["warm_probability"])], lab)) |
| allids = [p["post_id"] for p in c]; out["random"].append(np.mean([ndcg(list(rs.permutation(allids)), lab) for _ in range(50)])) |
| return {k: (round(float(np.nanmean(v)), 3) if v else None) for k, v in out.items()} | {"feeds": len(out["engine"])} |
| def period_requests(vm, history=True, interests=True): |
| res = {} |
| for sp in ("valid", "test"): |
| for size in (None, 25): |
| reqs = [] |
| for u in np.unique(user[split == sp]): |
| ix = np.where((user == u) & (split == sp))[0]; ix = ix[np.argsort(ts[ix], kind="stable")] |
| for blk in ([ix] if size is None else [ix[k:k + size] for k in range(0, len(ix) - size // 2, size)]): |
| reqs.append(({"interests": picks(blk[0]) if interests else [], "history": HIST[vm][u] if history else []}, [post_dict(i, vm) for i in blk], float(ts[blk[0]]) - 1, |
| {post_dict(i, vm)["post_id"]: int(label[i]) for i in blk})) |
| res[f"{sp} period, {'whole feeds' if size is None else '25-post pages'}"] = reqs |
| return res |
| |
| import importlib |
| lib = {"__name__": "lib", "__file__": os.path.join(HERE, "new_session2", "validate_session.py")}; exec(open(lib["__file__"], encoding="utf8").read(), lib) |
| from fastembed import TextEmbedding |
| _bge = TextEmbedding(model_name="BAAI/bge-small-en-v1.5", cache_dir=os.path.join(HERE, "bge_cache"), threads=4) |
| def load_ses(d): |
| S, _, T = lib["load_session"](os.path.join(HERE, d)); best = collections.defaultdict(float) |
| for l in open(os.path.join(HERE, d, "events.jsonl"), encoding="utf8"): |
| e = json.loads(l) |
| if e["type"] == "attention_update" and e["d"].get("target_type") == "post": best[e["d"]["post_id"]] = max(best[e["d"]["post_id"]], e["d"].get("active_ms") or 0) |
| S["active_ms"] = S.post_id.map(best).fillna(0.0); S["lab"] = ((S.engaged.astype(bool)) | (S.active_ms >= 10000)).astype(int) |
| cache = os.path.join(HERE, d, "new_post_emb_bge.npz") |
| if not os.path.exists(cache): |
| v = np.asarray(list(_bge.embed(["\n".join([s_(r.title), s_(r.body), "", ""])[:12000] for r in S.itertuples()], batch_size=32)), dtype=np.float64); v = v / np.linalg.norm(v, axis=1, keepdims=True) |
| np.savez_compressed(cache, post_ids=S.post_id.values, emb=v.astype("float32")) |
| c = np.load(cache, allow_pickle=True); return S, dict(zip(c["post_ids"], c["emb"])), {k: {t: 1.0 for t, w in v.items()} for k, v in T.items()} |
| def ses_hist(S, E, vm): |
| return [dict(post_id=r.post_id, ts_ms=float(r.ts), visible_ms=float(r.visible_ms), focal_ms=float(r.active_ms), hover_ms=0.0, clicked=bool(r.clicked), engaged=bool(r.lab), community=None, |
| author=r.author, embedding=None if vm == "none" else E[r.post_id]) for r in S.itertuples()] |
| def ses_cand(S, E, T, vm): |
| return [dict(post_id=r.post_id, score=r.score, comments=r.comments, age_hours=r.age_hours, title=s_(r.title), body=s_(r.body), tags=T.get(r.post_id, {}), |
| embedding=None if vm == "none" else E[r.post_id], author=r.author, impressions=0) for r in S.itertuples()] |
| S1, E1, T1 = load_ses("new_session"); S2, E2, T2 = load_ses("new_session2") |
| p09 = [i for i in range(len(user)) if user[i] == "P09" and split[i] == "test"][0]; P09_PICKS = picks(p09) |
| print(f" later sessions: session 1 {len(S1)} posts, {int(S1.lab.sum())} positives | session 2 {len(S2)} posts, {int(S2.lab.sum())} positives | P09 interests {P09_PICKS}", flush=True) |
| def session_requests(vm, history=True, interests=True): |
| base = HIST[vm]["P09"] if history else []; res = {} |
| pg = lambda n: [list(range(k, min(k + 25, n))) for k in range(0, n - 12, 25)] |
| for name, S, E, T, prior in (("later session 1", S1, E1, T1, []), ("later session 2", S2, E2, T2, ses_hist(S1, E1, vm) if history else [])): |
| C = ses_cand(S, E, T, vm); lab = dict(zip(S.post_id, S.lab)); hs = base + prior; pick = P09_PICKS if interests else [] |
| res[f"{name}, one feed"] = [({"interests": pick, "history": hs}, C, float(S.ts.min()) - 1, lab)] |
| res[f"{name}, 25-post pages"] = [({"interests": pick, "history": hs + (ses_hist(S, E, vm) if history else [])}, [C[i] for i in p], float(S.ts.iloc[p[0]]) - 1, {C[i]["post_id"]: int(S.lab.iloc[i]) for i in p}) for p in pg(len(S))] |
| return res |
| results = {} |
| for vm, nm in (("fused", "trained vectors (captions in)"), ("text", "serving vectors (title + body only)"), ("none", "no content vectors")): |
| reqs = {**period_requests(vm), **session_requests(vm)}; results[nm] = {k: run(v) for k, v in reqs.items()}; print(f"3) done: {nm}", flush=True) |
| |
| results["new user: interests only, no history"] = {k: run(v) for k, v in {**period_requests("text", history=False), **session_requests("text", history=False)}.items()} |
| results["new user: no interests, no history"] = {k: run(v) for k, v in {**period_requests("text", history=False, interests=False), **session_requests("text", history=False, interests=False)}.items()} |
| feeds = list(results["serving vectors (title + body only)"].keys()) |
| def mean(nm, key="engine"): return round(float(np.mean([results[nm][f][key] for f in feeds if results[nm][f][key] is not None])), 3) |
| summary = {nm: mean(nm) for nm in results}; summary["random (expected)"] = mean("serving vectors (title + body only)", "random") |
| summary["v2b only (serving vectors)"] = mean("serving vectors (title + body only)", "v2b only"); summary["warm only (serving vectors)"] = mean("serving vectors (title + body only)", "warm only") |
| print("\nNDCG@10 mean over", len(feeds), "feed settings:"); [print(f" {k:45s} {v}") for k, v in summary.items()] |
| print("\nper setting (engine):"); [print(f" {f:34s} " + " | ".join(f"{nm[:18]:18s} {results[nm][f]['engine']:.3f}" for nm in list(results)[:5]) + f" | random {results['serving vectors (title + body only)'][f]['random']:.3f}") for f in feeds] |
| json.dump(dict(engine=meta["version"], label="engaged OR active_ms >= 10 s", parity=dict(features_max_abs=worst, rows=n_rows, scores=dict(cold=cd, warm=wd), ms_per_candidate=lat), summary=summary, per_setting=results), |
| open(os.path.join(ART, "evaluation_report.json"), "w"), indent=1, default=float) |
| json.dump(dict(feature_parity_max_abs=worst, rows_checked=n_rows, score_parity=dict(cold=cd, warm=wd), ms_per_candidate=lat), open(os.path.join(ART, "verification_report.json"), "w"), indent=1) |
| print("VERIFY V3 DONE") |
|
|
