File size: 8,749 Bytes
1f9a1bf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | """Feature builders for the hybrid engine. They reproduce, from raw post/user data, exactly the inputs the models were trained on
(verified against the training files by verify_hybrid_engine.py)."""
import math, re
import numpy as np
def clean(s):
return " ".join(str(s).split()) if isinstance(s, str) else ""
def token(v):
"""Category value -> the token string used in training ('none' for missing/empty, whitespace -> '_')."""
if v is None or (isinstance(v, float) and math.isnan(v)):
return "none"
s = re.sub(r"\s+", "_", str(v))
return s if s != "" else "none"
def unit(v):
return v / (np.linalg.norm(v) + 1e-9)
EMB_DIM = 1024 # Qwen/Qwen3-Embedding-0.6B
def content_vector(emb):
"""A usable Qwen3 vector as a unit numpy array, or None when the embedding is missing, has the wrong size
(for example the backend's 512-d CLIP vectors, which live in a different space) or is all zeros / not finite."""
if emb is None:
return None
v = np.asarray(emb, dtype=float).ravel()
if v.shape[0] != EMB_DIM or not np.isfinite(v).all() or not v.any():
return None
return unit(v)
def rank01(x):
"""Rank-normalise scores inside one request: lowest -> 1/n, highest -> 1."""
x = np.asarray(x, dtype=float)
return (np.argsort(np.argsort(x)) + 1) / len(x) if len(x) else x
class FeatureBuilder:
def __init__(self, meta):
self.m = meta
self.stats = meta["feature_stats"]
c = meta["constants"]
self.gl, self.age_median, self.W, self.top_domains, self.click_prior = c["global_positive_rate"], c["age_hours_median"], c["history_window"], set(c["top_domains"]), c["click_prior"]
p = meta["pca"]
self.pca_mean, self.pca_comp, self.pca_std = np.array(p["mean"]), np.array(p["components"]), np.array(p["component_std"])
t = meta["taxonomy"]
self.child = t["child"]; self.cidx = {s: i for i, s in enumerate(self.child)}; self.parent_of = np.array(t["parent_of"]); self.n_broad = len(t["broad"])
self.S = np.array(t["tag_similarity"], dtype=float)
s1 = meta["stage1"]
self.s1_mu, self.s1_sd, self.s1_coef, self.s1_b, self.idf = np.array(s1["mean"]), np.array(s1["std"]), np.array(s1["coef"]), float(s1["intercept"]), np.array(s1["idf"])
# ---------- post content ----------
def pca16(self, emb):
"""1024-d Qwen3-Embedding-0.6B vector (unit length) -> the 16 content components the models use."""
return ((np.asarray(emb, dtype=float) - self.pca_mean) @ self.pca_comp.T) / self.pca_std
def z(self, name, raw):
mu, sd = self.stats[name]
return (raw - mu) / sd
def post_floats(self, post):
"""Raw post statistics + content components (before standardisation)."""
score = max(float(post.get("score") or 0), 0.0)
age = post.get("age_hours")
age = self.age_median if age is None or (isinstance(age, float) and math.isnan(age)) else max(float(age), 0.0)
ur = post.get("upvote_ratio")
missing = ur is None or (isinstance(ur, float) and math.isnan(ur))
raw = {
"log_score": math.log1p(score), "log_comments": math.log1p(float(post.get("comments") or 0)), "log_age_h": math.log1p(age),
"title_len": math.log1p(len(clean(post.get("title")))), "body_len": math.log1p(len(clean(post.get("body")))),
"has_caption": float(bool(clean(post.get("image_caption")))), "is_ad": float(bool(post.get("is_ad"))),
"upvote_ratio": 1.0 if missing else float(ur), "upvote_missing": float(missing),
"log_awards": math.log1p(float(post.get("awards") or 0)), "is_spoiler": float(bool(post.get("is_spoiler"))),
}
emb = post.get("embedding") # None = content unknown -> 0 on every component (the average post)
for i, v in enumerate(self.pca16(emb) if emb is not None else np.zeros(16)):
raw[f"pca{i}"] = float(v)
return raw
def post_tokens(self, post):
dom = post.get("domain")
dom = dom if isinstance(dom, str) and dom in self.top_domains else "other"
mem = post.get("is_member")
return {"user_id": "unk", "item_id": "unk", "subreddit": token(post.get("community")), "post_type": token(post.get("post_type")), "flair": token(post.get("flair")),
"domain": token(dom), "post_language": token(post.get("language")), "member": {True: "yes", False: "no"}.get(mem, "unk")}
# ---------- tags ----------
def tag_vector(self, tags):
"""{slug: confidence} or [slug, ...] -> confidence vector over the taxonomy's child tags (unknown slugs ignored)."""
v = np.zeros(len(self.child))
items = tags.items() if isinstance(tags, dict) else ((t, 1.0) for t in (tags or []))
for s, c in items:
if s in self.cidx:
v[self.cidx[s]] = max(v[self.cidx[s]], float(c))
return v
def interest_profile(self, picks):
q = np.zeros(len(self.child))
for s in picks or []:
if s in self.cidx:
q[self.cidx[s]] = 1.0
r = np.array([q[self.parent_of == b].max() if (self.parent_of == b).any() else 0.0 for b in range(self.n_broad)])
ms = (self.S * q[None, :]).max(1) if q.any() else np.zeros(len(self.child))
return q, r, ms
def tag_match(self, P, q, r, ms):
"""The 8 tag-match features (exact, parent, rarity-weighted exact, soft, top-tag hit, #matches, #tags, mean confidence)."""
sumc = P.sum(); safe = sumc if sumc > 0 else 1.0; nt = int((P > 0).sum())
exact = (P * q).sum() / safe
parent = (P * (1 - q) * r[self.parent_of]).sum() / safe
den = (P * self.idf).sum(); idf_e = (P * q * self.idf).sum() / (den if den > 0 else 1.0)
soft = (P * ms).sum() / safe
top = float(q[int(P.argmax())]) if nt > 0 else 0.0
return np.array([exact, parent, idf_e, soft, top, ((P > 0) * q).sum(), nt, sumc / max(nt, 1) if nt > 0 else 0.0])
def tag_closeness(self, tm):
"""Stage-1 tag closeness (fitted logistic score) from the raw tag-match features."""
return float(((tm - self.s1_mu) / self.s1_sd) @ self.s1_coef + self.s1_b)
# ---------- user history ----------
def history_floats(self, history, post, now_ms, post_X, strict_prefix=False):
"""Causal history features for one candidate: only past posts whose viewing had ended before `now_ms` count.
strict_prefix=True means `history` is already exactly the earlier interactions (used by the parity test).
history: list of dicts sorted by ts with keys ts_ms, visible_ms, focal_ms, hover_ms, clicked, engaged, community, author, embedding (unit 1024-d), pca16."""
J = [h for h in history if (h["ts_ms"] < now_ms or strict_prefix) and h["ts_ms"] + (h.get("visible_ms") or 0) <= now_ms] # past interactions whose viewing had ended
R = J[-self.W:]; n = len(R); gl = self.gl
lab = lambda xs: sum(1 for h in xs if h["engaged"])
out = {"h_n": math.log1p(len(J)), "h_pos_rate": (lab(R) + 3 * gl) / (n + 3),
"h_mean_focal": sum(math.log1p((h.get("focal_ms") or 0) / 1000) for h in R) / (n + 3),
"h_hover": sum(math.log1p(h.get("hover_ms") or 0) for h in R) / (n + 3),
"h_click": (sum(1 for h in R if h.get("clicked")) + 3 * self.click_prior) / (n + 3)}
ur = (lab(J) + 3 * gl) / (len(J) + 3)
com = post.get("community")
sm = [h for h in J if isinstance(com, str) and h.get("community") == com]
out["sub_seen"], out["sub_rate"] = math.log1p(len(sm)), (lab(sm) + 3 * ur) / (len(sm) + 3)
au = post.get("author")
am = [h for h in J if isinstance(au, str) and h.get("author") == au]
out["auth_seen"], out["auth_rate"] = math.log1p(len(am)), (lab(am) + 3 * ur) / (len(am) + 3)
# content features use only past posts that have an embedding; without one (post or history) they stay 0, as for a user with no history
Rp = [h for h in R if h["engaged"] and h.get("embedding") is not None]
Rn = [h for h in R if not h["engaged"] and h.get("embedding") is not None]
out["aff_recent"] = float(post_X @ unit(np.mean([h["embedding"] for h in Rp], 0))) if Rp and post_X is not None else 0.0
out["aff_neg"] = float(post_X @ unit(np.mean([h["embedding"] for h in Rn], 0))) if Rn and post_X is not None else 0.0
ue = np.mean([h["pca16"] for h in Rp], 0) if Rp else np.zeros(16)
for i in range(16):
out[f"uemb{i}"] = float(ue[i])
return out, lab(J)
|