Tool call · Input
/app/solution/build_ranker.py
#!/usr/bin/env python3
"""Discovery ranker for the Yambda Made-for-You shelf.
Hybrid of recency-weighted implicit CF (user-kNN + PureSVD), short-term
taste, continuation, and organic popularity. Validation uses train history
only; the hidden-test ranker is refit on train+val.
"""
from __future__ import annotations
import json
import os
import sys
import time
from pathlib import Path
os.environ.setdefault("OMP_NUM_THREADS", "8")
os.environ.setdefault("OPENBLAS_NUM_THREADS", "8")
os.environ.setdefault("MKL_NUM_THREADS", "8")
os.environ.setdefault("NUMEXPR_NUM_THREADS", "8")
import numpy as np
import pandas as pd
from scipy.sparse import coo_matrix, csr_matrix
from sklearn.decomposition import TruncatedSVD
from sklearn.preprocessing import normalize
sys.path.insert(0, "/task")
from ranking_evaluation import ndcg_at_k, QUALITY_K, TOP_N
DATA = Path("/task/data")
OUT = Path("/app/solution")
OUT.mkdir(parents=True, exist_ok=True)
TRAIN_CUTOFF = 24_786_800
TEST_CUTOFF = 25_395_195
DAY = 86_400.0
HALF_LIFE_DAYS = 14.0
SHORT_DAYS = 7.0
SVD_COMPONENTS = 160
KNN_NEIGHBORS = 90
LIKE_WEIGHT = 10.0
PREF_LIKE_WEIGHT = 16.0
ORGANIC_MULT = 2.4
STRONG_MULT = 1.6
MIN_RATIO = 50
N_ITEMS_FIXED = 30_000
def log(msg: str) -> None:
print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)
def memmap_channel(name: str, n_users: int, n_items: int, mode: str = "w+") -> np.memmap:
path = OUT / f"channel_{name}.dat"
return np.memmap(path, dtype=np.float32, mode=mode, shape=(n_users, n_items))
def row_z_inplace(A: np.ndarray) -> None:
mean = A.mean(axis=1, keepdims=True)
std = np.maximum(A.std(axis=1, keepdims=True), 1e-6)
A -= mean
A /= std
def load_catalog_index():
el = np.load(DATA / "validation_eligible_candidates.npz")
item_ids = el["item_ids"].astype(np.int64)
target_uids = el["uids"].astype(np.int64)
catalog = pd.read_parquet(DATA / "candidate_catalog.parquet")
pop_map = dict(
zip(catalog.item_id.astype(int), catalog.organic_positive_listens.astype(np.float64))
)
pop = np.array([pop_map.get(int(i), 1.0) for i in item_ids], dtype=np.float32)
max_item = int(item_ids.max()) + 1
lut = np.full(max_item, -1, dtype=np.int32)
lut[item_ids] = np.arange(len(item_ids), dtype=np.int32)
return item_ids, target_uids, pop, lut
def unpack_eligible(path: Path, n_items: int) -> tuple[np.ndarray, np.ndarray]:
art = np.load(path)
uids = art["uids"].astype(np.int64)
packed = art["packed_eligible"]
elig = np.unpackbits(packed, axis=1, bitorder="little")[:, :n_items].astype(bool)
return uids, elig
def map_items(item_arr: np.ndarray, lut: np.ndarray) -> np.ndarray:
cols = np.full(len(item_arr), -1, dtype=np.int32)
valid = item_arr < len(lut)
cols[valid] = lut[item_arr[valid]]
return cols
def load_listens(paths: list[Path]) -> pd.DataFrame:
frames = [
pd.read_parquet(
p, columns=["uid", "timestamp", "item_id", "is_organic", "played_ratio_pct"]
)
for p in paths
]
return pd.concat(frames, ignore_index=True) if len(frames) > 1 else frames[0]
def listen_weights(df: pd.DataFrame, cutoff: int, lut: np.ndarray):
uid = df.uid.to_numpy(dtype=np.int64, copy=False)
ts = df.timestamp.to_numpy(dtype=np.int64, copy=False)
item = df.item_id.to_numpy(dtype=np.int64, copy=False)
organic = df.is_organic.to_numpy(dtype=np.int16, copy=False)
ratio = df.played_ratio_pct.to_numpy(dtype=np.int16, copy=False)
cols = map_items(item, lut)
keep = (cols >= 0) & (ts <= cutoff) & (ratio >= MIN_RATIO)
uid, ts, cols, organic, ratio = uid[keep], ts[keep], cols[keep], organic[keep], ratio[keep]
age_days = np.clip(cutoff - ts, 0, None) / DAY
decay = np.exp(-np.log(2.0) * age_days / HALF_LIFE_DAYS).astype(np.float32)
w = decay * np.clip(ratio.astype(np.float32) / 100.0, 0.0, 1.2)
w *= np.where(organic == 1, ORGANIC_MULT, 1.0).astype(np.float32)
w *= np.where(ratio >= 80, STRONG_MULT, 1.0).astype(np.float32)
w_short = w * (age_days <= SHORT_DAYS) * (organic == 1)
return uid, cols, w, w_short.astype(np.float32)
def add_pref_likes(uid, cols, w, pref_path: Path, lut: np.ndarray):
pref = pd.read_parquet(pref_path)
liked = pref[pref.state == "liked"]
p_uid = liked.uid.to_numpy(dtype=np.int64)
p_cols = map_items(liked.item_id.to_numpy(dtype=np.int64), lut)
keep = p_cols >= 0
p_uid, p_cols = p_uid[keep], p_cols[keep]
p_w = np.full(len(p_uid), PREF_LIKE_WEIGHT, dtype=np.float32)
return (
np.concatenate([uid, p_uid]),
np.concatenate([cols, p_cols]),
np.concatenate([w, p_w]),
)
def add_like_events(uid, cols, w, paths: list[Path], cutoff: int, lut: np.ndarray):
frames = [
pd.read_parquet(p, columns=["uid", "timestamp", "item_id", "is_organic"]) for p in paths
]
likes = pd.concat(frames, ignore_index=True) if len(frames) > 1 else frames[0]
l_uid = likes.uid.to_numpy(dtype=np.int64)
ts = likes.timestamp.to_numpy(dtype=np.int64)
l_cols = map_items(likes.item_id.to_numpy(dtype=np.int64), lut)
organic = likes.is_organic.to_numpy(dtype=np.int16)
keep = (l_cols >= 0) & (ts <= cutoff)
l_uid, ts, l_cols, organic = l_uid[keep], ts[keep], l_cols[keep], organic[keep]
age_days = np.clip(cutoff - ts, 0, None) / DAY
decay = np.exp(-np.log(2.0) * age_days / HALF_LIFE_DAYS).astype(np.float32)
lw = decay * LIKE_WEIGHT * np.where(organic == 1, 1.6, 1.0).astype(np.float32)
return (
np.concatenate([uid, l_uid]),
np.concatenate([cols, l_cols]),
np.concatenate([w, lw]),
)
def pref_like_matrix(pref_path: Path, lut, uid_lut, n_users: int, n_items: int) -> csr_matrix:
pref = pd.read_parquet(pref_path)
liked = pref[pref.state == "liked"]
p_uid = liked.uid.to_numpy(dtype=np.int64)
p_cols = map_items(liked.item_id.to_numpy(dtype=np.int64), lut)
keep = p_cols >= 0
p_uid, p_cols = p_uid[keep], p_cols[keep]
keep = p_uid < len(uid_lut)
p_uid, p_cols = p_uid[keep], p_cols[keep]
rows = uid_lut[p_uid]
keep = rows >= 0
return coo_matrix(
(np.ones(keep.sum(), dtype=np.float32), (rows[keep], p_cols[keep])),
shape=(n_users, n_items),
).tocsr()
def build_user_index(target_uids: np.ndarray, event_uids: np.ndarray):
extra = np.setdiff1d(np.unique(event_uids), target_uids, assume_unique=False)
all_uids = np.concatenate([target_uids, extra])
max_uid = int(all_uids.max()) + 1
lut = np.full(max_uid, -1, dtype=np.int32)
lut[all_uids] = np.arange(len(all_uids), dtype=np.int32)
return all_uids, lut, len(target_uids)
def to_csr(uids, cols, w, uid_lut, n_users, n_items) -> csr_matrix:
keep = uids < len(uid_lut)
uids, cols, w = uids[keep], cols[keep], w[keep]
rows = uid_lut[uids]
keep = rows >= 0
return coo_matrix(
(w[keep].astype(np.float32), (rows[keep], cols[keep])),
shape=(n_users, n_items),
).tocsr()
def apply_idf(X: csr_matrix) -> csr_matrix:
df = np.asarray((X > 0).sum(axis=0)).ravel().astype(np.float32)
idf = np.log((X.shape[0] + 1.0) / (df + 1.0)) + 1.0
return X.multiply(idf).tocsr()
def recent_organic_pop(df: pd.DataFrame, cutoff: int, lut: np.ndarray, n_items: int, window_days: float) -> np.ndarray:
ts = df.timestamp.to_numpy(dtype=np.int64, copy=False)
item = df.item_id.to_numpy(dtype=np.int64, copy=False)
organic = df.is_organic.to_numpy(dtype=np.int16, copy=False)
ratio = df.played_ratio_pct.to_numpy(dtype=np.int16, copy=False)
cols = map_items(item, lut)
keep = (
(cols >= 0)
& (ts <= cutoff)
& (ts >= cutoff - int(window_days * DAY))
& (organic == 1)
& (ratio >= 80)
)
return np.bincount(cols[keep], minlength=n_items).astype(np.float32)
def validation_truth(item_ids, target_set, elig, elig_uids) -> dict[int, dict[int, int]]:
uid_row = {int(u): i for i, u in enumerate(elig_uids)}
item_col = {int(it): i for i, it in enumerate(item_ids)}
listens = pd.read_parquet(DATA / "val_listens.parquet")
likes = pd.read_parquet(DATA / "val_likes.parquet")
strong = listens[
listens.uid.isin(target_set)
& (listens.is_organic == 1)
& (listens.played_ratio_pct >= 80)
][["uid", "item_id"]].drop_duplicates()
olikes = likes[likes.uid.isin(target_set) & (likes.is_organic == 1)][
["uid", "item_id"]
].drop_duplicates()
graded: dict[int, dict[int, int]] = {}
def eligible(uid: int, item: int) -> bool:
r = uid_row.get(uid)
c = item_col.get(item)
if r is None or c is None:
return False
return bool(elig[r, c])
for u, i in strong.itertuples(index=False, name=None):
u, i = int(u), int(i)
if eligible(u, i):
graded.setdefault(u, {})[i] = 1
for u, i in olikes.itertuples(index=False, name=None):
u, i = int(u), int(i)
if eligible(u, i):
graded.setdefault(u, {})[i] = 3
return graded
def ndcg_from_scores(scores, elig, uids, item_ids, graded, buf) -> float:
np.copyto(buf, scores)
buf[~elig] = np.float32(-1e4)
k = QUALITY_K
idx = np.argpartition(-buf, kth=k, axis=1)[:, :k]
part = np.take_along_axis(buf, idx, axis=1)
order = np.argsort(-part, axis=1)
idx = np.take_along_axis(idx, order, axis=1)
uid_row = {int(u): i for i, u in enumerate(uids)}
vals = [
ndcg_at_k(item_ids[idx[uid_row[uid]]].tolist(), gold, k) for uid, gold in graded.items()
]
return float(np.mean(vals))
def topn_ranking(scores, elig, uids, item_ids, buf, n: int = TOP_N) -> pd.DataFrame:
np.copyto(buf, scores)
buf[~elig] = np.float32(-1e4)
idx = np.argpartition(-buf, kth=n, axis=1)[:, :n]
part = np.take_along_axis(buf, idx, axis=1)
order = np.argsort(-part, axis=1)
idx = np.take_along_axis(idx, order, axis=1)
n_users = len(uids)
return pd.DataFrame(
{
"uid": np.repeat(uids, n),
"item_id": item_ids[idx].reshape(-1),
"rank": np.tile(np.arange(1, n + 1, dtype=np.int32), n_users),
}
)
def save_ranking(df: pd.DataFrame, path: Path) -> None:
df = df.astype({"uid": np.int64, "item_id": np.int64, "rank": np.int32})
df.to_parquet(path, index=False)
log(f"wrote {path} rows={len(df)}")
def user_knn_into(X_sim: csr_matrix, X_vote: csr_matrix, n_neighbors: int, n_targets: int, out: np.ndarray) -> None:
log("kNN cosine")
Xn = normalize(X_sim, norm="l2", axis=1)
S = Xn @ Xn.T
if not isinstance(S, np.ndarray):
S = S.toarray()
S = np.asarray(S, dtype=np.float32)
del Xn
np.fill_diagonal(S, 0.0)
np.maximum(S, 0.0, out=S)
k = min(n_neighbors, S.shape[1] - 1)
idx = np.argpartition(-S, kth=k, axis=1)[:, :k]
rows = np.arange(S.shape[0], dtype=np.int32)[:, None]
vals = S[rows, idx]
S = None
St = np.zeros((n_targets, X_vote.shape[0]), dtype=np.float32)
St[rows[:n_targets], idx[:n_targets]] = vals[:n_targets]
denom = St.sum(axis=1, keepdims=True)
denom[denom == 0] = 1.0
St /= denom
log("kNN vote")
# (items x users_all) @ (users_all x targets) -> items x targets
scored = X_vote.T @ St.T
out[:] = scored.T
del scored, St, idx, vals
def svd_into(X: csr_matrix, n_components: int, n_targets: int, out: np.ndarray):
k = min(n_components, min(X.shape) - 1)
svd = TruncatedSVD(n_components=k, n_iter=8, random_state=0)
user_f = svd.fit_transform(X).astype(np.float32)
item_f = svd.components_.T.astype(np.float32)
out[:] = user_f[:n_targets] @ item_f.T
return user_f, item_f
def short_into(X_short: csr_matrix, item_f: np.ndarray, n_targets: int, out: np.ndarray) -> None:
user = X_short[:n_targets] @ item_f
norms = np.linalg.norm(user, axis=1, keepdims=True)
norms[norms == 0] = 1.0
user = user / norms
item_n = item_f / np.maximum(np.linalg.norm(item_f, axis=1, keepdims=True), 1e-6)
out[:] = user @ item_n.T
def like_svd_into(X_like: csr_matrix, n_components: int, n_targets: int, out: np.ndarray) -> None:
k = min(n_components, max(2, min(X_like.shape) - 1), max(2, X_like.nnz // 20))
k = min(k, 96)
if X_like.nnz < 1000:
out[:] = 0
return
svd = TruncatedSVD(n_components=k, n_iter=7, random_state=1)
user_f = svd.fit_transform(X_like).astype(np.float32)
item_f = svd.components_.T.astype(np.float32)
out[:] = user_f[:n_targets] @ item_f.T
def build_channels(
listen_paths: list[Path],
like_paths: list[Path],
pref_path: Path,
cutoff: int,
item_ids: np.ndarray,
target_uids: np.ndarray,
lut: np.ndarray,
catalog_pop: np.ndarray,
tag: str,
):
n_items = len(item_ids)
n_targets = len(target_uids)
log(f"loading listens from {[p.name for p in listen_paths]}")
listens = load_listens(listen_paths)
log(f"listens rows={len(listens)}")
uid, cols, w, w_short = listen_weights(listens, cutoff, lut)
pop_recent = recent_organic_pop(listens, cutoff, lut, n_items, 10.0)
pop_month = recent_organic_pop(listens, cutoff, lut, n_items, 30.0)
del listens
uid_s, cols_s, w_s = uid.copy(), cols.copy(), w_short
uid, cols, w = add_like_events(uid, cols, w, like_paths, cutoff, lut)
uid, cols, w = add_pref_likes(uid, cols, w, pref_path, lut)
all_uids, uid_lut, n_targets = build_user_index(target_uids, uid)
n_users = len(all_uids)
log(f"users={n_users} targets={n_targets} events={len(w)}")
X = to_csr(uid, cols, w, uid_lut, n_users, n_items)
X_short = to_csr(uid_s, cols_s, w_s, uid_lut, n_users, n_items)
X_like = pref_like_matrix(pref_path, lut, uid_lut, n_users, n_items)
log(f"X nnz={X.nnz} short nnz={X_short.nnz} like nnz={X_like.nnz}")
pop_vec = np.log1p(0.55 * pop_recent + 0.25 * pop_month + 0.20 * catalog_pop).astype(np.float32)
pop_vec -= pop_vec.mean()
pop_vec /= max(float(pop_vec.std()), 1e-6)
names = ["knn", "svd", "short", "repeat", "like"]
ch = {name: memmap_channel(f"{tag}_{name}", n_targets, n_items) for name in names}
log("repeat channel")
ch["repeat"][:] = X[:n_targets].toarray()
np.log1p(ch["repeat"], out=ch["repeat"])
row_z_inplace(ch["repeat"])
ch["repeat"].flush()
X_idf = apply_idf(X)
log("user-kNN channel")
user_knn_into(X_idf, X, KNN_NEIGHBORS, n_targets, ch["knn"])
row_z_inplace(ch["knn"])
ch["knn"].flush()
log("svd channel")
_user_f, item_f = svd_into(X_idf, SVD_COMPONENTS, n_targets, ch["svd"])
row_z_inplace(ch["svd"])
ch["svd"].flush()
del _user_f, X_idf
log("short channel")
short_into(X_short, item_f, n_targets, ch["short"])
row_z_inplace(ch["short"])
ch["short"].flush()
del item_f, X_short
log("like-svd channel")
like_svd_into(apply_idf(X_like + X_short * 0), 96, n_targets, ch["like"]) if False else like_svd_into(
apply_idf(X_like), 96, n_targets, ch["like"]
)
row_z_inplace(ch["like"])
ch["like"].flush()
del X, X_like
return ch, pop_vec
def blend_into(out, channels, weights, pop_vec, w_pop):
first = True
for name, w in weights.items():
if w == 0:
continue
if first:
np.multiply(channels[name], np.float32(w), out=out)
first = False
else:
out += np.float32(w) * channels[name]
if first:
out[:] = 0
if w_pop:
out += np.float32(w_pop) * pop_vec[None, :]
def grid_search(channels, pop_vec, elig, uids, item_ids, graded, buf):
best_w = None
best_pop = 0.0
best = -1.0
trials = 0
knn_w = [0.7, 1.1, 1.5]
svd_w = [0.4, 0.8, 1.2]
like_w = [0.2, 0.6, 1.0]
pop_w = [0.15, 0.35, 0.55]
for a in knn_w:
for b in svd_w:
for lk in like_w:
for e in pop_w:
w = {"knn": a, "svd": b, "short": 0.55, "repeat": 0.12, "like": lk}
blend_into(buf, channels, w, pop_vec, e)
val = ndcg_from_scores(buf, elig, uids, item_ids, graded, buf)
# ndcg_from_scores copies buf onto itself first; we already blended into buf.
# Re-evaluate properly: blend into a dedicated scores view.
trials += 1
if val > best:
best, best_w, best_pop = val, w, e
log(f"best {best:.6f} {best_w} pop={best_pop}")
# The evaluate above is WRONG because ndcg_from_scores copies scores then masks,
# but we passed buf as both scores and buf, so copyto is no-op then mask destroys blend
# for subsequent? actually it masks in place so the NDCG is valid for that blend,
# but we cannot reuse. The NDCG itself is correct. Good.
base = dict(best_w)
for c in [0.15, 0.4, 0.7, 1.0]:
for d in [0.0, 0.1, 0.25, 0.45]:
w = {**base, "short": c, "repeat": d}
blend_into(buf, channels, w, pop_vec, best_pop)
val = ndcg_from_scores(buf, elig, uids, item_ids, graded, buf)
trials += 1
if val > best:
best, best_w = val, w
log(f"best {best:.6f} {best_w} pop={best_pop}")
base = dict(best_w)
for e in [0.05, 0.1, 0.2, 0.3, 0.45, 0.7, 1.0]:
blend_into(buf, channels, base, pop_vec, e)
val = ndcg_from_scores(buf, elig, uids, item_ids, graded, buf)
trials += 1
if val > best:
best, best_pop = val, e
log(f"best {best:.6f} {base} pop={best_pop}")
log(f"grid trials={trials} best={best:.6f} {best_w} pop={best_pop}")
return best_w, best_pop, best
def main() -> None:
t0 = time.time()
item_ids, target_uids, catalog_pop, lut = load_catalog_index()
n_items = len(item_ids)
n_targets = len(target_uids)
log(f"catalog items={n_items} targets={n_targets}")
val_uids, val_elig = unpack_eligible(DATA / "validation_eligible_candidates.npz", n_items)
assert np.array_equal(val_uids, target_uids)
buf = memmap_channel("buf", n_targets, n_items)
pop_fallback = np.log1p(catalog_pop).astype(np.float32)
buf[:] = pop_fallback[None, :]
save_ranking(
topn_ranking(buf, val_elig, target_uids, item_ids, buf),
OUT / "validation.parquet",
)
# restore note: topn_ranking masked buf in place; fallback submission for test too
buf[:] = pop_fallback[None, :]
test_uids, test_elig = unpack_eligible(DATA / "test_eligible_candidates.npz", n_items)
save_ranking(
topn_ranking(buf, test_elig, test_uids, item_ids, buf),
OUT / "submission.parquet",
)
target_set = set(map(int, target_uids))
log("building validation labels")
graded = validation_truth(item_ids, target_set, val_elig, val_uids)
log(f"labeled users={len(graded)} relevant={sum(len(v) for v in graded.values())}")
buf[:] = pop_fallback[None, :]
pop_ndcg = ndcg_from_scores(buf, val_elig, target_uids, item_ids, graded, buf)
log(f"organic-pop baseline ndcg@10={pop_ndcg:.6f}")
channels, pop_vec = build_channels(
listen_paths=[DATA / "train_listens.parquet"],
like_paths=[DATA / "train_likes.parquet"],
pref_path=DATA / "validation_preference_state.parquet",
cutoff=TRAIN_CUTOFF,
item_ids=item_ids,
target_uids=target_uids,
lut=lut,
catalog_pop=catalog_pop,
tag="val",
)
weights, w_pop, val_score = grid_search(
channels, pop_vec, val_elig, target_uids, item_ids, graded, buf
)
blend_into(buf, channels, weights, pop_vec, w_pop)
save_ranking(
topn_ranking(buf, val_elig, target_uids, item_ids, buf),
OUT / "validation.parquet",
)
with open(OUT / "val_metrics.json", "w") as f:
json.dump(
{"ndcg@10": val_score, "weights": weights, "w_pop": w_pop, "pop_baseline": pop_ndcg},
f,
indent=2,
)
log(f"validation ndcg@10={val_score:.6f}")
del channels
log("refitting on train+val for hidden test")
test_channels, test_pop = build_channels(
listen_paths=[DATA / "train_listens.parquet", DATA / "val_listens.parquet"],
like_paths=[DATA / "train_likes.parquet", DATA / "val_likes.parquet"],
pref_path=DATA / "test_preference_state.parquet",
cutoff=TEST_CUTOFF,
item_ids=item_ids,
target_uids=target_uids,
lut=lut,
catalog_pop=catalog_pop,
tag="test",
)
blend_into(buf, test_channels, weights, test_pop, w_pop)
save_ranking(
topn_ranking(buf, test_elig, target_uids, item_ids, buf),
OUT / "submission.parquet",
)
log(f"done in {time.time() - t0:.1f}s")
if __name__ == "__main__":
main()