Tool call · Input
/app/solution/rank_discovery.py
#!/usr/bin/env python3
"""Personalized discovery ranker for the Made-for-You shelf.
Candidate eligibility already removes catalog tracks a user has heard or
explicitly rated. The model ranks remaining catalog tracks by how likely the
user is to organically like them or complete them in the next seven days.
Signals (collaborative only; no audio/artist metadata is available):
- RP3beta-style 3-step walks on the user-item graph
- Neighbors' recent organic completions
- TF-IDF SVD and item-cosine in user space
- Session co-occurrence
- Time-decayed organic popularity
Blend weights are chosen on the public validation window, then the same
recipe is refit on train+val for the hidden test ranking.
"""
from __future__ import annotations
import json
import time
from pathlib import Path
import numpy as np
import pandas as pd
from scipy.sparse import coo_matrix, csr_matrix, diags
from sklearn.preprocessing import normalize
from sklearn.utils.extmath import randomized_svd
from ranking_evaluation import ndcg_at_k
DATA = Path("/task/data")
OUT = Path("/app/solution")
CACHE = OUT / "cache"
TRAIN_CUTOFF = 24_786_800
VAL_END = 25_393_395
TEST_CUTOFF = VAL_END + 1_800
DAY = 86_400
LISTEN_STRONG = 80
TOP_N = 100
POOL_K = 400
QUALITY_K = 10
RRF_K = 40
CATALOG_N = 30_000
def log(msg: str) -> None:
print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)
def load_catalog_and_targets():
cat = pd.read_parquet(DATA / "candidate_catalog.parquet")
targets = pd.read_parquet(DATA / "target_users.parquet")
return cat, targets
def load_eligibility(path: Path):
z = np.load(path, allow_pickle=False)
uids = z["uids"].astype(np.int64)
item_ids = z["item_ids"].astype(np.int64)
packed = z["packed_eligible"]
elig = np.unpackbits(packed, axis=1, bitorder="little")[:, : len(item_ids)].astype(bool)
return uids, item_ids, elig
def read_events(split: str) -> dict[str, pd.DataFrame]:
frames = {}
for fam in ("listens", "likes", "dislikes", "unlikes", "undislikes"):
frames[fam] = pd.read_parquet(DATA / f"{split}_{fam}.parquet")
return frames
def concat_splits(train: dict, val: dict | None) -> dict[str, pd.DataFrame]:
if val is None:
return train
return {k: pd.concat([train[k], val[k]], ignore_index=True) for k in train}
def build_user_index(target_uids: np.ndarray, frames: dict) -> tuple[np.ndarray, dict[int, int]]:
extra = set()
for df in frames.values():
extra.update(int(u) for u in df.uid.unique())
extra.difference_update(int(u) for u in target_uids)
extra_uids = np.fromiter(sorted(extra), dtype=np.int64)
all_uids = np.concatenate([target_uids, extra_uids]) if len(extra_uids) else target_uids.copy()
row = {int(u): i for i, u in enumerate(all_uids)}
return all_uids, row
def map_ids(series: pd.Series, mapping: dict[int, int]) -> np.ndarray:
return np.fromiter((mapping.get(int(x), -1) for x in series.to_numpy()), dtype=np.int32, count=len(series))
def decay_weights(timestamps: np.ndarray, cutoff: int, half_life_days: float) -> np.ndarray:
age = np.maximum(0.0, (cutoff - timestamps.astype(np.float64)) / DAY)
return np.exp(-np.log(2.0) * age / half_life_days).astype(np.float32)
def assemble_matrix(
frames: dict,
user_row: dict[int, int],
item_col: dict[int, int],
n_users: int,
n_items: int,
cutoff: int,
*,
half_life: float,
listen_mode: str,
min_played: int,
organic_only: bool,
like_weight: float,
dislike_weight: float,
max_age_days: float | None = None,
) -> csr_matrix:
"""Build a user x item CSR of time-decayed implicit feedback on mapped items."""
listens = frames["listens"]
ts = listens.timestamp.to_numpy()
keep = np.ones(len(listens), dtype=bool)
if max_age_days is not None:
keep &= ts >= (cutoff - max_age_days * DAY)
if organic_only:
keep &= listens.is_organic.to_numpy() == 1
if min_played > 0:
keep &= listens.played_ratio_pct.to_numpy() >= min_played
u = np.fromiter((user_row.get(int(x), -1) for x in listens.uid.to_numpy()), dtype=np.int32, count=len(listens))
it = np.fromiter((item_col.get(int(x), -1) for x in listens.item_id.to_numpy()), dtype=np.int32, count=len(listens))
keep &= (u >= 0) & (it >= 0)
u, it, ts = u[keep], it[keep], ts[keep]
played = listens.played_ratio_pct.to_numpy()[keep].astype(np.float32)
organic = listens.is_organic.to_numpy()[keep].astype(np.int8)
w = decay_weights(ts, cutoff, half_life)
if listen_mode == "strong":
w *= np.where(played >= LISTEN_STRONG, 1.0, np.where(played >= 50, 0.45, 0.08)).astype(np.float32)
elif listen_mode == "binary_strong":
w = np.where(played >= LISTEN_STRONG, w, 0.0).astype(np.float32)
else:
w *= np.clip(played / 100.0, 0.05, 1.0)
w *= np.where(organic == 1, 1.6, 0.55).astype(np.float32)
rows = [u]
cols = [it]
data = [w]
likes = frames["likes"]
if len(likes):
lu = np.fromiter((user_row.get(int(x), -1) for x in likes.uid.to_numpy()), dtype=np.int32, count=len(likes))
li = np.fromiter((item_col.get(int(x), -1) for x in likes.item_id.to_numpy()), dtype=np.int32, count=len(likes))
lts = likes.timestamp.to_numpy()
lk = (lu >= 0) & (li >= 0)
if max_age_days is not None:
lk &= lts >= (cutoff - max_age_days * DAY)
if organic_only:
lk &= likes.is_organic.to_numpy() == 1
lw = like_weight * decay_weights(lts[lk], cutoff, half_life)
org = likes.is_organic.to_numpy()[lk]
lw *= np.where(org == 1, 1.3, 1.0).astype(np.float32)
rows.append(lu[lk])
cols.append(li[lk])
data.append(lw)
if dislike_weight != 0.0 and len(frames["dislikes"]):
d = frames["dislikes"]
du = np.fromiter((user_row.get(int(x), -1) for x in d.uid.to_numpy()), dtype=np.int32, count=len(d))
di = np.fromiter((item_col.get(int(x), -1) for x in d.item_id.to_numpy()), dtype=np.int32, count=len(d))
dts = d.timestamp.to_numpy()
dk = (du >= 0) & (di >= 0)
if max_age_days is not None:
dk &= dts >= (cutoff - max_age_days * DAY)
dw = dislike_weight * decay_weights(dts[dk], cutoff, half_life)
rows.append(du[dk])
cols.append(di[dk])
data.append(dw)
row = np.concatenate(rows)
col = np.concatenate(cols)
val = np.concatenate(data)
ok = val != 0
mat = coo_matrix((val[ok], (row[ok], col[ok])), shape=(n_users, n_items), dtype=np.float32)
return mat.tocsr()
def add_preference_boost(
X: csr_matrix,
pref: pd.DataFrame,
user_row: dict[int, int],
item_col: dict[int, int],
like_boost: float = 8.0,
dislike_boost: float = -2.0,
) -> csr_matrix:
if pref is None or len(pref) == 0:
return X
u = np.fromiter((user_row.get(int(x), -1) for x in pref.uid.to_numpy()), dtype=np.int32, count=len(pref))
it = np.fromiter((item_col.get(int(x), -1) for x in pref.item_id.to_numpy()), dtype=np.int32, count=len(pref))
state = pref.state.to_numpy()
w = np.where(state == "liked", like_boost, np.where(state == "disliked", dislike_boost, 0.0)).astype(np.float32)
ok = (u >= 0) & (it >= 0) & (w != 0)
extra = coo_matrix((w[ok], (u[ok], it[ok])), shape=X.shape, dtype=np.float32).tocsr()
return X + extra
def tfidf(X: csr_matrix) -> csr_matrix:
n_users = X.shape[0]
df = np.asarray((X > 0).sum(axis=0)).ravel().astype(np.float64)
idf = np.log((n_users + 1.0) / (df + 1.0)) + 1.0
return X.multiply(idf.astype(np.float32)).tocsr()
def l1_rows(X: csr_matrix) -> csr_matrix:
s = np.asarray(X.sum(axis=1)).ravel().astype(np.float32)
s = np.maximum(s, 1e-8)
return diags(1.0 / s) @ X
def dense_mm_sparse(UU: np.ndarray, X: csr_matrix) -> np.ndarray:
"""Compute UU @ X for dense UU (U,U) and sparse X (U,I), return float32 (U,I)."""
uu = np.ascontiguousarray(UU, dtype=np.float32)
xt = X.T.tocsr().astype(np.float32)
out = xt @ uu.T
if hasattr(out, "toarray"):
out = out.toarray()
return np.ascontiguousarray(out.T, dtype=np.float32)
def rp3_scores(
query: csr_matrix,
x_graph: csr_matrix,
x_target: csr_matrix,
beta: float,
zero_self: bool = True,
) -> np.ndarray:
"""3-step scores: query -> users-via-items -> target items, with RP3beta penalty."""
p_iu = l1_rows(x_graph.T.tocsr()) # items x users
q = l1_rows(query)
mid = q @ p_iu # users x users
if hasattr(mid, "toarray"):
mid = mid.toarray()
mid = np.asarray(mid, dtype=np.float32)
if zero_self:
np.fill_diagonal(mid, 0.0)
p_target = l1_rows(x_target)
scores = dense_mm_sparse(mid, p_target)
deg = np.asarray(x_target.sum(axis=0)).ravel().astype(np.float32)
scores /= np.power(np.maximum(deg, 1.0), beta)
return scores
def item_cosine_scores(query: csr_matrix, item_space: csr_matrix) -> np.ndarray:
"""Exact item-item cosine via the user-dimensional inner product."""
V = normalize(item_space.T.tocsr(), norm="l2", axis=1) # items x users
q = query.astype(np.float32)
mid = q @ V # users x users
if hasattr(mid, "toarray"):
mid = mid.toarray()
mid = np.ascontiguousarray(mid, dtype=np.float32)
scores = mid @ V.T
if hasattr(scores, "toarray"):
scores = scores.toarray()
return np.ascontiguousarray(scores, dtype=np.float32)
def svd_scores(x_fit: csr_matrix, query: csr_matrix, n_comp: int = 128) -> np.ndarray:
xw = tfidf(x_fit)
n_comp = min(n_comp, min(xw.shape) - 1)
u, s, vt = randomized_svd(xw.astype(np.float32), n_components=n_comp, n_iter=7, random_state=42)
item_f = np.ascontiguousarray(vt.T * s, dtype=np.float32) # items x d, scaled
# recency-biased query projected into the same space
q = tfidf(query)
user_q = q @ item_f # users x d (item_f is items x d, q is users x items)
if hasattr(user_q, "toarray"):
user_q = user_q.toarray()
user_q = np.ascontiguousarray(user_q, dtype=np.float32)
return user_q @ item_f.T
def user_knn_recent(x_sim: csr_matrix, x_recent: csr_matrix, k_neighbors: int = 80) -> np.ndarray:
U = normalize(tfidf(x_sim), norm="l2", axis=1)
uu = U @ U.T
if hasattr(uu, "toarray"):
uu = uu.toarray()
uu = np.asarray(uu, dtype=np.float32)
np.fill_diagonal(uu, 0.0)
# keep top-k neighbors per user
k = min(k_neighbors, uu.shape[1] - 1)
idx = np.argpartition(-uu, kth=k, axis=1)[:, :k]
rows = np.arange(uu.shape[0])[:, None]
mask = np.zeros_like(uu, dtype=bool)
mask[rows, idx] = True
uu = np.where(mask, uu, 0.0)
# cosine already in uu; clip tiny negatives
uu = np.clip(uu, 0.0, None)
return dense_mm_sparse(uu, x_recent.astype(np.float32))
def session_item_scores(
frames: dict,
user_row: dict[int, int],
item_col: dict[int, int],
n_users: int,
n_items: int,
cutoff: int,
query: csr_matrix,
window_sec: int = 7200,
) -> np.ndarray:
"""Item-item from same-session co-occurrence, scored with the user query."""
listens = frames["listens"]
ts = listens.timestamp.to_numpy()
played = listens.played_ratio_pct.to_numpy()
keep = (played >= 50) & (ts <= cutoff)
u_raw = listens.uid.to_numpy()
i_raw = listens.item_id.to_numpy()
u = np.fromiter((user_row.get(int(x), -1) for x in u_raw), dtype=np.int32, count=len(listens))
it = np.fromiter((item_col.get(int(x), -1) for x in i_raw), dtype=np.int32, count=len(listens))
keep &= (u >= 0) & (it >= 0)
u, it, ts = u[keep], it[keep], ts[keep]
sess = u.astype(np.int64) * 100_000 + (ts // window_sec).astype(np.int64)
# factorize sessions
_, sess_idx = np.unique(sess, return_inverse=True)
ones = np.ones(len(it), dtype=np.float32)
S = coo_matrix((ones, (sess_idx, it)), shape=(int(sess_idx.max()) + 1, n_items), dtype=np.float32)
S = (S.tocsr() > 0).astype(np.float32)
# item-item co-counts; drop diagonal
co = S.T @ S
co = co.tocsr()
co.setdiag(0)
co.eliminate_zeros()
# PMI-ish: divide by item popularity^0.5
deg = np.asarray(S.sum(axis=0)).ravel().astype(np.float32)
inv = 1.0 / np.sqrt(np.maximum(deg, 1.0))
co = diags(inv) @ co @ diags(inv)
# keep top 150 neighbors per item
co = topk_sparse_rows(co, 150)
scores = query.astype(np.float32) @ co.T.T # query (U,I) @ (I,I)
if hasattr(scores, "toarray"):
scores = scores.toarray()
return np.ascontiguousarray(scores, dtype=np.float32)
def topk_sparse_rows(mat: csr_matrix, k: int) -> csr_matrix:
mat = mat.tocsr()
data = []
indices = []
indptr = [0]
for r in range(mat.shape[0]):
start, end = mat.indptr[r], mat.indptr[r + 1]
if end <= start:
indptr.append(len(indices))
continue
vals = mat.data[start:end]
cols = mat.indices[start:end]
if end - start > k:
keep = np.argpartition(vals, -k)[-k:]
vals = vals[keep]
cols = cols[keep]
data.append(vals)
indices.append(cols)
indptr.append(indptr[-1] + len(cols))
if data:
data_a = np.concatenate(data)
idx_a = np.concatenate(indices)
else:
data_a = np.zeros(0, dtype=np.float32)
idx_a = np.zeros(0, dtype=np.int32)
return csr_matrix((data_a, idx_a, np.array(indptr, dtype=np.int32)), shape=mat.shape, dtype=np.float32)
def popularity_vector(
frames: dict,
item_col: dict[int, int],
n_items: int,
cutoff: int,
half_life: float,
organic_only: bool = True,
min_played: int = LISTEN_STRONG,
max_age_days: float | None = 45.0,
) -> np.ndarray:
listens = frames["listens"]
ts = listens.timestamp.to_numpy()
keep = np.ones(len(listens), dtype=bool)
if organic_only:
keep &= listens.is_organic.to_numpy() == 1
if min_played:
keep &= listens.played_ratio_pct.to_numpy() >= min_played
if max_age_days is not None:
keep &= ts >= (cutoff - max_age_days * DAY)
it = np.fromiter((item_col.get(int(x), -1) for x in listens.item_id.to_numpy()), dtype=np.int32, count=len(listens))
keep &= it >= 0
w = decay_weights(ts[keep], cutoff, half_life)
pop = np.zeros(n_items, dtype=np.float32)
np.add.at(pop, it[keep], w)
likes = frames["likes"]
if len(likes):
li = np.fromiter((item_col.get(int(x), -1) for x in likes.item_id.to_numpy()), dtype=np.int32, count=len(likes))
lts = likes.timestamp.to_numpy()
lk = li >= 0
if organic_only:
lk &= likes.is_organic.to_numpy() == 1
if max_age_days is not None:
lk &= lts >= (cutoff - max_age_days * DAY)
np.add.at(pop, li[lk], 3.0 * decay_weights(lts[lk], cutoff, half_life))
return np.log1p(pop)
def topk_indices(scores: np.ndarray, elig: np.ndarray, k: int) -> np.ndarray:
s = scores.astype(np.float32, copy=True)
s[~elig] = -np.inf
finite = np.isfinite(s)
# popularity-order tie-break: lower column index is more popular
s -= np.arange(s.shape[1], dtype=np.float32) * 1e-7
s[~finite] = -np.inf
k = min(k, s.shape[1])
idx = np.argpartition(s, -k, axis=1)[:, -k:]
part = np.take_along_axis(s, idx, axis=1)
order = np.argsort(-part, axis=1)
return np.take_along_axis(idx, order, axis=1)
def lists_from_idx(idx: np.ndarray, item_ids: np.ndarray) -> np.ndarray:
return item_ids[idx]
def validation_graded(target_set: set[int], item_set: set[int], elig_map) -> dict[int, dict[int, int]]:
listens = pd.read_parquet(DATA / "val_listens.parquet", columns=["uid", "item_id", "is_organic", "played_ratio_pct"])
likes = pd.read_parquet(DATA / "val_likes.parquet", columns=["uid", "item_id", "is_organic"])
uids, item_ids, elig = elig_map
uid_row = {int(u): i for i, u in enumerate(uids)}
item_c = {int(it): i for i, it in enumerate(item_ids)}
def eligible(u, i) -> bool:
r = uid_row.get(int(u))
c = item_c.get(int(i))
if r is None or c is None:
return False
return bool(elig[r, c])
graded: dict[int, dict[int, int]] = {}
strong = listens[
listens.uid.isin(target_set) & (listens.is_organic == 1) & (listens.played_ratio_pct >= LISTEN_STRONG)
][["uid", "item_id"]].drop_duplicates()
olikes = likes[likes.uid.isin(target_set) & (likes.is_organic == 1)][["uid", "item_id"]].drop_duplicates()
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_of_idx(idx: np.ndarray, item_ids: np.ndarray, uids: np.ndarray, graded: dict[int, dict[int, int]]) -> float:
ranked_items = item_ids[idx[:, :QUALITY_K]]
vals = []
for row, uid in enumerate(uids):
gold = graded.get(int(uid))
if not gold:
continue
vals.append(ndcg_at_k(ranked_items[row].tolist(), gold, QUALITY_K))
return float(np.mean(vals)) if vals else 0.0
def rrf_fuse(rank_lists: list[np.ndarray], weights: list[float], k_rrf: int = RRF_K, top: int = POOL_K) -> np.ndarray:
"""rank_lists: each (n_users, pool) item-column indices, best first."""
n_users = rank_lists[0].shape[0]
# accumulate in dicts per user would be slow; use a dense pool via unique per user
# vectorized: for each method add weight/(k+rank) into a score table over union
# Use a hash-on-the-fly dense of shape (n_users, n_items) is 1GB; we don't know n_items here.
raise NotImplementedError
def rrf_from_indices(
methods: dict[str, np.ndarray],
weights: dict[str, float],
n_items: int,
elig: np.ndarray,
k_rrf: int,
) -> np.ndarray:
n_users = elig.shape[0]
acc = np.zeros((n_users, n_items), dtype=np.float32)
rows = np.arange(n_users)[:, None]
for name, idx in methods.items():
w = weights.get(name, 0.0)
if w == 0.0:
continue
ranks = np.arange(1, idx.shape[1] + 1, dtype=np.float32)
contrib = (w / (k_rrf + ranks)).astype(np.float32)
np.add.at(acc, (rows, idx), contrib)
return topk_indices(acc, elig, TOP_N)
def minmax_fuse(
score_map: dict[str, np.ndarray],
weights: dict[str, float],
elig: np.ndarray,
) -> np.ndarray:
acc = None
for name, s in score_map.items():
w = weights.get(name, 0.0)
if w == 0.0:
continue
v = s.astype(np.float32, copy=True)
v[~elig] = np.nan
mn = np.nanmin(v, axis=1, keepdims=True)
mx = np.nanmax(v, axis=1, keepdims=True)
v = (v - mn) / (mx - mn + 1e-6)
np.nan_to_num(v, copy=False, nan=0.0)
acc = w * v if acc is None else acc + w * v
if acc is None:
acc = np.zeros(elig.shape, dtype=np.float32)
return topk_indices(acc, elig, TOP_N)
def zscore_fuse(
score_map: dict[str, np.ndarray],
weights: dict[str, float],
elig: np.ndarray,
) -> np.ndarray:
acc = None
for name, s in score_map.items():
w = weights.get(name, 0.0)
if w == 0.0:
continue
v = s.astype(np.float64, copy=True)
v[~elig] = np.nan
mu = np.nanmean(v, axis=1, keepdims=True)
sd = np.nanstd(v, axis=1, keepdims=True)
v = (v - mu) / (sd + 1e-6)
np.nan_to_num(v, copy=False, nan=0.0, posinf=0.0, neginf=0.0)
v = v.astype(np.float32)
acc = w * v if acc is None else acc + w * v
if acc is None:
acc = np.zeros(elig.shape, dtype=np.float32)
return topk_indices(acc, elig, TOP_N)
def write_ranking(path: Path, uids: np.ndarray, item_ids: np.ndarray, idx: np.ndarray) -> None:
n_users, k = idx.shape
out_uid = np.repeat(uids.astype(np.int64), k)
out_item = item_ids[idx.reshape(-1)].astype(np.int64)
out_rank = np.tile(np.arange(1, k + 1, dtype=np.int64), n_users)
df = pd.DataFrame({"uid": out_uid, "item_id": out_item, "rank": out_rank})
path.parent.mkdir(parents=True, exist_ok=True)
df.to_parquet(path, index=False)
log(f"wrote {path} rows={len(df)}")
def build_scores(
frames: dict,
pref: pd.DataFrame,
target_uids: np.ndarray,
item_ids: np.ndarray,
cutoff: int,
) -> dict[str, np.ndarray]:
n_targets = len(target_uids)
n_items = len(item_ids)
item_col = {int(it): i for i, it in enumerate(item_ids)}
all_uids, user_row = build_user_index(target_uids, frames)
n_users = len(all_uids)
log(f"users={n_users} targets={n_targets} items={n_items} cutoff={cutoff}")
log("building catalog matrices")
X = assemble_matrix(
frames, user_row, item_col, n_users, n_items, cutoff,
half_life=60.0, listen_mode="strong", min_played=0,
organic_only=False, like_weight=5.0, dislike_weight=-0.8,
)
X = add_preference_boost(X, pref, user_row, item_col, like_boost=10.0, dislike_boost=-3.0)
X.data = np.maximum(X.data, 0.0) # clip residual negatives for graph walks
X.eliminate_zeros()
Q = assemble_matrix(
frames, user_row, item_col, n_users, n_items, cutoff,
half_life=18.0, listen_mode="strong", min_played=0,
organic_only=False, like_weight=6.0, dislike_weight=0.0,
max_age_days=120.0,
)
Q = add_preference_boost(Q, pref, user_row, item_col, like_boost=12.0, dislike_boost=0.0)
X_org = assemble_matrix(
frames, user_row, item_col, n_users, n_items, cutoff,
half_life=40.0, listen_mode="binary_strong", min_played=LISTEN_STRONG,
organic_only=True, like_weight=4.0, dislike_weight=0.0,
)
X_recent = assemble_matrix(
frames, user_row, item_col, n_users, n_items, cutoff,
half_life=10.0, listen_mode="binary_strong", min_played=LISTEN_STRONG,
organic_only=True, like_weight=3.0, dislike_weight=0.0,
max_age_days=21.0,
)
Q_org = assemble_matrix(
frames, user_row, item_col, n_users, n_items, cutoff,
half_life=14.0, listen_mode="strong", min_played=50,
organic_only=True, like_weight=5.0, dislike_weight=0.0,
max_age_days=45.0,
)
Q_org = add_preference_boost(Q_org, pref, user_row, item_col, like_boost=8.0, dislike_boost=0.0)
log(f"X nnz={X.nnz} Q nnz={Q.nnz} X_recent nnz={X_recent.nnz}")
scores: dict[str, np.ndarray] = {}
log("RP3beta full")
s = rp3_scores(Q, X, X, beta=0.45, zero_self=True)
scores["rp3"] = s[:n_targets]
del s
log("RP3 recent-organic target")
s = rp3_scores(Q_org, X, X_recent, beta=0.35, zero_self=True)
scores["rp3_recent"] = s[:n_targets]
del s
log("RP3 organic graph")
s = rp3_scores(Q_org, X_org, X_org, beta=0.5, zero_self=True)
scores["rp3_org"] = s[:n_targets]
del s
log("item cosine")
s = item_cosine_scores(tfidf(Q), tfidf(X))
scores["item_cos"] = s[:n_targets]
del s
log("SVD")
s = svd_scores(X, Q, n_comp=128)
scores["svd"] = s[:n_targets]
del s
log("user kNN recent organic")
s = user_knn_recent(tfidf(X), X_recent, k_neighbors=90)
scores["uknn"] = s[:n_targets]
del s
log("popularity")
pop = popularity_vector(frames, item_col, n_items, cutoff, half_life=14.0, max_age_days=30.0)
scores["pop"] = np.broadcast_to(pop, (n_targets, n_items)).copy()
pop_long = popularity_vector(frames, item_col, n_items, cutoff, half_life=45.0, max_age_days=180.0, organic_only=True)
scores["pop_long"] = np.broadcast_to(pop_long, (n_targets, n_items)).copy()
log("session co-occurrence")
try:
s = session_item_scores(frames, user_row, item_col, n_users, n_items, cutoff, Q)
scores["session"] = s[:n_targets]
del s
except Exception as exc:
log(f"session failed: {exc}")
# drop full user matrices
del X, Q, X_org, X_recent, Q_org
return scores
def method_topk(scores: dict[str, np.ndarray], elig: np.ndarray, k: int) -> dict[str, np.ndarray]:
out = {}
for name, s in scores.items():
log(f"topk {name}")
out[name] = topk_indices(s, elig, k)
return out
def evaluate_methods(top: dict[str, np.ndarray], item_ids, uids, graded) -> dict[str, float]:
report = {}
for name, idx in top.items():
report[name] = round(ndcg_of_idx(idx, item_ids, uids, graded), 6)
log(f"NDCG {name}: {report[name]}")
return report
def search_blend(
scores: dict[str, np.ndarray],
top: dict[str, np.ndarray],
elig: np.ndarray,
item_ids: np.ndarray,
uids: np.ndarray,
graded: dict,
) -> tuple[str, dict[str, float], np.ndarray, float]:
rng = np.random.default_rng(42)
names = list(scores.keys())
best = ("none", {}, None, -1.0)
# single methods as candidates
for name, idx in top.items():
sc = ndcg_of_idx(idx, item_ids, uids, graded)
if sc > best[3]:
best = ("single", {name: 1.0}, idx[:, :TOP_N], sc)
def consider(kind: str, weights: dict[str, float], idx: np.ndarray) -> None:
nonlocal best
sc = ndcg_of_idx(idx, item_ids, uids, graded)
if sc > best[3]:
best = (kind, dict(weights), idx, sc)
log(f"new best {kind} {sc:.6f} {weights}")
# RRF grid
base_w = {n: 1.0 for n in names}
# drop weak pops a bit
rrf_templates = [
{n: 1.0 for n in names},
{n: (0.3 if n.startswith("pop") else 1.0) for n in names},
{n: (1.4 if n.startswith("rp3") else 0.8 if n in ("item_cos", "uknn", "svd") else 0.25) for n in names},
{n: (1.5 if n in ("rp3_recent", "uknn") else 1.0 if n.startswith("rp3") else 0.6 if n != "pop" else 0.2) for n in names},
{n: (1.2 if n in ("rp3", "rp3_org", "item_cos", "session") else 0.7) for n in names},
]
for k_rrf in (20, 40, 60):
for tmpl in rrf_templates:
idx = rrf_from_indices(top, tmpl, elig.shape[1], elig, k_rrf)
consider(f"rrf{k_rrf}", tmpl, idx)
# random RRF weights
for k_rrf in (30, 50):
for _ in range(18):
w = {n: float(rng.uniform(0.0, 1.5)) for n in names}
idx = rrf_from_indices(top, w, elig.shape[1], elig, k_rrf)
consider(f"rrf{k_rrf}r", w, idx)
# z-score linear combos
z_templates = [
{n: (1.3 if n.startswith("rp3") else 0.8 if n in ("item_cos", "uknn", "svd", "session") else 0.2) for n in names},
{n: (1.6 if n == "rp3_recent" else 1.1 if n.startswith("rp3") else 0.7 if n in ("uknn", "item_cos") else 0.4 if n == "svd" else 0.15) for n in names},
{n: (1.0 if n in ("rp3", "rp3_recent", "uknn") else 0.5) for n in names},
]
for tmpl in z_templates:
idx = zscore_fuse(scores, tmpl, elig)
consider("z", tmpl, idx)
for _ in range(12):
w = {n: float(rng.uniform(0.0, 1.6)) for n in names}
idx = zscore_fuse(scores, w, elig)
consider("zr", w, idx)
mm_templates = z_templates
for tmpl in mm_templates:
idx = minmax_fuse(scores, tmpl, elig)
consider("mm", tmpl, idx)
log(f"best blend {best[0]} ndcg={best[3]:.6f} w={best[1]}")
return best
def ranking_for_stage(
frames: dict,
pref: pd.DataFrame,
elig_path: Path,
cutoff: int,
graded: dict | None,
blend: tuple | None,
) -> tuple[np.ndarray, np.ndarray, np.ndarray, tuple]:
uids, item_ids, elig = load_eligibility(elig_path)
scores = build_scores(frames, pref, uids, item_ids, cutoff)
top = method_topk(scores, elig, POOL_K)
if graded is not None:
evaluate_methods(top, item_ids, uids, graded)
kind, weights, idx, sc = search_blend(scores, top, elig, item_ids, uids, graded)
log(f"selected {kind} val_ndcg={sc:.6f}")
return uids, item_ids, idx, (kind, weights, sc)
assert blend is not None
kind, weights, _sc = blend[0], blend[1], blend[2] if len(blend) > 2 else None
log(f"applying blend {kind} {weights}")
if kind.startswith("rrf"):
k_rrf = 40
digits = "".join(ch for ch in kind if ch.isdigit())
if digits:
k_rrf = int(digits)
idx = rrf_from_indices(top, weights, elig.shape[1], elig, k_rrf)
elif kind.startswith("z"):
idx = zscore_fuse(scores, weights, elig)
elif kind.startswith("mm"):
idx = minmax_fuse(scores, weights, elig)
elif kind == "single":
name = next(iter(weights))
idx = top[name][:, :TOP_N]
else:
idx = zscore_fuse(scores, weights, elig)
return uids, item_ids, idx, blend
def main() -> None:
CACHE.mkdir(parents=True, exist_ok=True)
OUT.mkdir(parents=True, exist_ok=True)
log("loading events")
train = read_events("train")
val = read_events("val")
pref_val = pd.read_parquet(DATA / "validation_preference_state.parquet")
pref_test = pd.read_parquet(DATA / "test_preference_state.parquet")
uids_v, item_ids_v, elig_v = load_eligibility(DATA / "validation_eligible_candidates.npz")
target_set = set(int(u) for u in uids_v)
log("building validation labels")
graded = validation_graded(target_set, set(int(i) for i in item_ids_v), (uids_v, item_ids_v, elig_v))
log(f"scored users={len(graded)} rel={sum(len(v) for v in graded.values())}")
del uids_v, item_ids_v, elig_v
log("==== VALIDATION FIT (train history only) ====")
uids, item_ids, idx, blend = ranking_for_stage(
train, pref_val, DATA / "validation_eligible_candidates.npz", TRAIN_CUTOFF, graded, None
)
write_ranking(OUT / "validation.parquet", uids, item_ids, idx)
with open(CACHE / "blend.json", "w") as f:
json.dump({"kind": blend[0], "weights": blend[1], "val_ndcg": blend[2]}, f, indent=2)
log("==== TEST FIT (train+val history) ====")
both = concat_splits(train, val)
uids, item_ids, idx, _ = ranking_for_stage(
both, pref_test, DATA / "test_eligible_candidates.npz", TEST_CUTOFF, None, blend
)
write_ranking(OUT / "submission.parquet", uids, item_ids, idx)
log("done")
if __name__ == "__main__":
main()