File size: 4,545 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 | """Smoke tests on synthetic data (no dataset needed). Run: python -m hybrid_engine.tests.test_engine"""
import numpy as np
from hybrid_engine import HybridRecommender
rng = np.random.default_rng(0)
def vec():
v = rng.normal(size=1024); return v / np.linalg.norm(v)
def post(i, age):
return dict(post_id=f"p{i}", community=["r/MachineLearning", "r/robotics", "r/IOT"][i % 3], post_type="text", score=10 + i, comments=i, age_hours=age, title=f"post {i}",
body="body", tags={"machine-learning": 0.9} if i % 2 else {"robotics": 0.8}, embedding=vec(), author=f"a{i % 5}", impressions=i % 4)
def event(i, t, engaged):
return dict(post_id=f"h{i}", ts_ms=t, visible_ms=4000, focal_ms=4000 if engaged else 200, hover_ms=500, clicked=engaged, engaged=engaged, community="r/robotics", author="a1", embedding=vec())
def test_smoke():
eng = HybridRecommender(); now = 1_800_000_000_000.0
cands = [post(i, age) for i, age in enumerate([1, 5, 30, 100, 400, 2, 50, 800, 12, 200])]
new = eng.score({"interests": ["machine-learning", "robotics"], "history": []}, cands, now)
assert all(d["warm_weight"] == 0.0 and d["warm_probability"] is None for d in new), "a brand-new user must get pure v2b"
assert all(np.isfinite(d["score"]) for d in new)
weights = []
for n_eng in (1, 5, 20, 100):
hist = [event(i, now - 10_000_000 + i * 1000, i < n_eng) for i in range(max(n_eng, 10))]
res = eng.score({"interests": ["machine-learning"], "history": hist}, cands, now); weights.append(res[0]["warm_weight"])
assert res[0]["engaged_history"] == n_eng
assert weights == sorted(weights) and weights[-1] <= eng.cfg["w_max"] + 1e-9 and weights[0] > 0, f"warm weight must grow with history and stay <= w_max: {weights}"
future = [event(0, now + 5_000, True)] # an interaction after the request must not count
assert eng.score({"interests": [], "history": future}, cands[:3], now)[0]["engaged_history"] == 0
fm = {d["post_id"]: d["freshness_multiplier"] for d in new}; ages = {c["post_id"]: c["age_hours"] for c in cands}
s = sorted(fm, key=lambda p: ages[p]); assert all(fm[s[i]] >= fm[s[i + 1]] for i in range(len(s) - 1)), "older posts must never get a larger freshness multiplier"
assert all((d["push"] > 0) == (ages[d["post_id"]] <= eng.cfg["push_max_age_hours"]) for d in new), "push only for fresh posts"
many = [post(i, 5 + i) for i in range(30)] # 15 'machine-learning' posts (odd i), 15 'robotics' posts
feed = eng.rank({"interests": ["machine-learning"], "history": []}, many, now)
k = eng.cfg["topic_first_slots"]
assert all(d["topic_match"] for d in feed[:k]), "the first topic_first_slots positions must be topic matches when enough exist"
plain = HybridRecommender(topic_first_slots=0).rank({"interests": ["machine-learning"], "history": []}, many, now)
assert [d["score"] for d in plain] == sorted([d["score"] for d in plain], reverse=True), "with the rule off, the order is plain score order"
few = [post(i, 5) for i in (0, 1, 2, 4, 6)] # only one topic match: the rule must fall back gracefully
assert len(eng.rank({"interests": ["machine-learning"], "history": []}, few, now)) == 5
top = eng.recommend({"interests": ["robotics"], "history": []}, cands, now, k=3); assert len(top) == 3
print("all smoke tests passed | warm weight at 1/5/20/100 engaged posts:", [round(w, 3) for w in weights])
def test_no_community_or_flair_in_models():
"""Engine 2.0: community and flair are not model inputs, so renaming them changes no score for a new user
(with history, community only feeds the user's own per-community rates)."""
eng = HybridRecommender(); now = 1_800_000_000_000.0
for which in ("cold", "warm"):
assert not {"subreddit", "flair"} & set(eng.meta[which]["token_inputs"]), eng.meta[which]["token_inputs"]
assert "no community" in eng.meta["pca"].get("embedding_text", ""), "content vectors must come from title + body (+ caption) only"
cands = [post(i, 5 + i) for i in range(12)]
renamed = [{**c, "community": f"somewhere-{i}", "flair": "Anything"} for i, c in enumerate(cands)]
u = {"interests": ["machine-learning"], "history": []}
assert [d["score"] for d in eng.score(u, cands, now)] == [d["score"] for d in eng.score(u, renamed, now)]
if __name__ == "__main__":
test_smoke()
test_no_community_or_flair_in_models()
print("community / flair guard passed")
|