"""HybridRecommender: v2b for new users, gradually blended toward a warm DeepFM as the user's history grows, plus a post-age decay and a cold-start push for fresh posts. score = (1 - w) * rank(v2b) + w * rank(DeepFM) w = min(w_max, engaged / (engaged + n0)) (w = 0 for a brand-new user) * (floor + (1 - floor) * 0.5 ** (age_hours / half_life)) (older posts slowly sink) + push_beta * [age_hours <= push_max_age] / sqrt(1 + impressions) (fresh, rarely shown posts get a lift) v2b = calibrated cold-model probability + tag_bonus * rank(tag closeness) """ import json, math, os import numpy as np import onnxruntime as ort from .features import FeatureBuilder, content_vector, rank01 ART = os.path.join(os.path.dirname(os.path.abspath(__file__)), "artifacts") class HybridRecommender: def __init__(self, artifacts_dir=ART, **overrides): self.meta = json.load(open(os.path.join(artifacts_dir, "meta.json"), encoding="utf8")) self.cfg = {**self.meta["defaults"], **overrides} self.fb = FeatureBuilder(self.meta) so = ort.SessionOptions(); so.intra_op_num_threads = 1 self.cold = ort.InferenceSession(os.path.join(artifacts_dir, self.meta["cold"]["onnx"]), so, providers=["CPUExecutionProvider"]) self.warm = ort.InferenceSession(os.path.join(artifacts_dir, self.meta["warm"]["onnx"]), so, providers=["CPUExecutionProvider"]) # ---------- helpers ---------- def _tokens(self, which, toks): spec = self.meta[which] return [spec["vocab"][f].get(toks[f], 0) for f in spec["token_inputs"]] def _floats(self, which, raw): return [raw[f] for f in self.meta[which]["float_inputs"]] @staticmethod def _prep_history(history, fb): hs = [] for h in sorted(history or [], key=lambda x: x["ts_ms"]): e = content_vector(h.get("embedding"), fb.emb_dim) hs.append({**h, "embedding": e, "pca16": None if e is None else fb.pca16(e)}) return hs def warm_weight(self, engaged_count): n0, wmax = self.cfg["n0_engaged_posts"], self.cfg["w_max"] return min(wmax, engaged_count / (engaged_count + n0)) if engaged_count > 0 else 0.0 # ---------- main API ---------- def score(self, user, candidates, now_ms): """user: {"interests": [tag slugs], "history": [interaction dicts]}; candidates: [post dicts]; now_ms: request time (or a list, one per candidate, for replay). Returns one dict per candidate with the final score and every component (for logging / explanations).""" fb, cfg = self.fb, self.cfg if not candidates: return [] nows = list(now_ms) if isinstance(now_ms, (list, tuple, np.ndarray)) else [now_ms] * len(candidates) q, r, ms = fb.interest_profile(user.get("interests")) hist = self._prep_history(user.get("history"), fb) Tc, Fc, Tw, Fw, s1, eng, ages, imps, topic, content = [], [], [], [], [], [], [], [], [], [] for post, now in zip(candidates, nows): X = content_vector(post.get("embedding"), fb.emb_dim); post = {**post, "embedding": X}; content.append(X is not None) base = fb.post_floats(post); toks = fb.post_tokens(post) tm = fb.tag_match(fb.tag_vector(post.get("tags")), q, r, ms); s1.append(fb.tag_closeness(tm)) topic.append(bool(tm[0] > 0 or tm[1] > 0)) # shares a picked tag, or that tag's broad domain zb = {k: fb.z(k, v) for k, v in base.items()} zt = dict(zip(self.meta["stage1"]["features"], (tm - fb.s1_mu) / fb.s1_sd)) Tc.append(self._tokens("cold", toks)); Fc.append(self._floats("cold", {**zb, **zt})) hf, n_eng = fb.history_floats(hist, post, now, X); eng.append(n_eng) Tw.append(self._tokens("warm", toks)); Fw.append(self._floats("warm", {**zb, **{k: fb.z(k, v) for k, v in hf.items()}})) ages.append(fb.age_median if post.get("age_hours") is None else float(post["age_hours"])); imps.append(float(post.get("impressions") or 0)) p_cold = self.cold.run(None, {"token_ids": np.array(Tc, np.int64), "float_features": np.array(Fc, np.float32)})[0] a, b = self.meta["cold"]["platt"]["coef"], self.meta["cold"]["platt"]["intercept"] lgt = np.log(np.clip(p_cold, 1e-6, 1 - 1e-6) / (1 - np.clip(p_cold, 1e-6, 1 - 1e-6))) p_cal = 1 / (1 + np.exp(-(a * lgt + b))) v2b = p_cal + cfg["tag_bonus"] * rank01(s1) w = np.array([self.warm_weight(e) for e in eng]) if (w > 0).any(): p_warm = self.warm.run(None, {"token_ids": np.array(Tw, np.int64), "float_features": np.array(Fw, np.float32)})[0] else: p_warm = np.full(len(candidates), np.nan) blend = (1 - w) * rank01(v2b) + w * (rank01(np.nan_to_num(p_warm)) if (w > 0).any() else 0) ages = np.array(ages, dtype=float); imps = np.array(imps) fresh = cfg["decay_floor"] + (1 - cfg["decay_floor"]) * 0.5 ** (np.maximum(ages, 0) / cfg["decay_half_life_hours"]) push = cfg["push_beta"] * (ages <= cfg["push_max_age_hours"]) / np.sqrt(1 + imps) final = blend * fresh + push return [dict(post_id=c.get("post_id"), score=float(final[i]), topic_match=topic[i], cold_probability=float(p_cal[i]), tag_closeness=float(s1[i]), v2b=float(v2b[i]), warm_probability=None if np.isnan(p_warm[i]) else float(p_warm[i]), warm_weight=float(w[i]), engaged_history=int(eng[i]), freshness_multiplier=float(fresh[i]), push=float(push[i]), has_content_embedding=content[i]) for i, c in enumerate(candidates)] def order(self, scored): """Topic-first ordering: in every block of `topic_block` positions, the first `topic_first_slots` go to topic-matching posts (best score first); the remaining positions take the best remaining posts of any kind. If too few topic matches exist, the slots fall back to the best remaining posts.""" slots, block = int(self.cfg.get("topic_first_slots", 0)), int(self.cfg.get("topic_block", 10)) rest = sorted(scored, key=lambda d: -d["score"]) if slots <= 0: return rest out = [] while rest: topical = [d for d in rest if d["topic_match"]][:slots] for d in topical: rest.remove(d) out += topical fill = block - len(topical) out += rest[:fill]; rest = rest[fill:] for pos, d in enumerate(out): d["position"] = pos + 1 return out def rank(self, user, candidates, now_ms): """All candidates in final feed order (scores + topic-first rule).""" return self.order(self.score(user, candidates, now_ms)) def recommend(self, user, candidates, now_ms, k=10): """Top-k candidates in final feed order.""" return self.rank(user, candidates, now_ms)[:k]