Tool call · Input
/app/solution/build_ranker.py
#!/usr/bin/env python3
"""Made-for-You discovery ranker.
Unsupervised collaborative-filtering scores are fit on history at a cutoff,
blended on public validation labels, then refit on train+val for the hidden test.
"""
from __future__ import annotations
import gc
import sys
import time
from pathlib import Path
import numpy as np
import pandas as pd
from scipy import sparse
from sklearn.decomposition import TruncatedSVD
from sklearn.linear_model import Ridge
from sklearn.preprocessing import normalize
sys.path.append("/task")
from ranking_evaluation import QUALITY_K, ndcg_at_k # noqa: E402
DATA = Path("/task/data")
OUT = Path("/app/solution")
OUT.mkdir(parents=True, exist_ok=True)
TRAIN_CUTOFF = 24_786_800
VAL_END = 25_393_400
LISTEN_POS = 50
LISTEN_STRONG = 80
TOP_N = 100
DAY = 86_400.0
CATALOG = pd.read_parquet(DATA / "candidate_catalog.parquet")
ITEM_IDS = CATALOG.item_id.to_numpy(dtype=np.int64)
N_ITEMS = len(ITEM_IDS)
POP_RANK = CATALOG.popularity_rank.to_numpy(dtype=np.int32)
POP_SCORE = np.log1p(CATALOG.organic_positive_listens.to_numpy(dtype=np.float32))
TARGETS = pd.read_parquet(DATA / "target_users.parquet")
TARGET_UIDS = TARGETS.uid.to_numpy(dtype=np.int64)
N_TGT = len(TARGET_UIDS)
def log(msg: str) -> None:
print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)
def item_mapper(item_ids: np.ndarray) -> np.ndarray:
inv = np.full(int(item_ids.max()) + 1, -1, dtype=np.int32)
inv[item_ids] = np.arange(len(item_ids), dtype=np.int32)
return inv
ITEM_INV = item_mapper(ITEM_IDS)
def map_items(raw: np.ndarray) -> np.ndarray:
out = np.full(len(raw), -1, dtype=np.int32)
ok = (raw >= 0) & (raw < len(ITEM_INV))
out[ok] = ITEM_INV[raw[ok]]
return out
def load_eligibility(path: Path, target_uids: np.ndarray) -> np.ndarray:
z = np.load(path)
uids = z["uids"].astype(np.int64)
packed = z["packed_eligible"]
items = z["item_ids"].astype(np.int64)
if not np.array_equal(items, ITEM_IDS):
raise RuntimeError(f"eligibility item order mismatch in {path}")
row_of = {int(u): i for i, u in enumerate(uids)}
order = np.array([row_of[int(u)] for u in target_uids], dtype=np.int32)
bits = np.unpackbits(packed[order], axis=1, bitorder="little")[:, :N_ITEMS]
return bits.astype(bool)
def load_pref_sets(path: Path) -> tuple[dict[int, set[int]], dict[int, set[int]]]:
df = pd.read_parquet(path)
liked: dict[int, set[int]] = {}
disliked: dict[int, set[int]] = {}
for uid, item, state in df.itertuples(index=False, name=None):
uid = int(uid)
item = int(item)
bucket = liked if state == "liked" else disliked
bucket.setdefault(uid, set()).add(item)
return liked, disliked
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(names: list[str]) -> dict[str, pd.DataFrame]:
parts = [read_events(n) for n in names]
return {k: pd.concat([p[k] for p in parts], ignore_index=True) for k in parts[0]}
def decay_from_age(age_sec: np.ndarray, half_life_days: float) -> np.ndarray:
return np.exp(-np.log(2.0) * np.clip(age_sec, 0, None) / (half_life_days * DAY)).astype(
np.float32
)
def collect_listen_triples(
listens: pd.DataFrame, cutoff: int, min_ratio: int
) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
uid = listens.uid.to_numpy(dtype=np.int64)
ts = listens.timestamp.to_numpy(dtype=np.int64)
item = listens.item_id.to_numpy(dtype=np.int64)
organic = listens.is_organic.to_numpy(dtype=np.int8)
ratio = listens.played_ratio_pct.to_numpy(dtype=np.int16)
cols = map_items(item)
keep = (ts <= cutoff) & (cols >= 0) & (ratio >= min_ratio)
return uid[keep], ts[keep], cols[keep], organic[keep], ratio[keep]
def collect_flag_triples(
frame: pd.DataFrame, cutoff: int
) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
uid = frame.uid.to_numpy(dtype=np.int64)
ts = frame.timestamp.to_numpy(dtype=np.int64)
cols = map_items(frame.item_id.to_numpy(dtype=np.int64))
organic = frame.is_organic.to_numpy(dtype=np.int8)
keep = (ts <= cutoff) & (cols >= 0)
return uid[keep], ts[keep], cols[keep], organic[keep]
def unique_sorted_uids(*arrays: np.ndarray) -> np.ndarray:
return np.unique(np.concatenate(arrays))
def uid_mapper(uids: np.ndarray) -> dict[int, int]:
return {int(u): i for i, u in enumerate(uids.tolist())}
def aggregate_pairs(
rows: np.ndarray,
cols: np.ndarray,
weights: np.ndarray,
n_rows: int,
n_cols: int,
) -> sparse.csr_matrix:
if len(rows) == 0:
return sparse.csr_matrix((n_rows, n_cols), dtype=np.float32)
coo = sparse.coo_matrix((weights.astype(np.float32), (rows, cols)), shape=(n_rows, n_cols))
return coo.tocsr().astype(np.float32)
def build_matrices(
events: dict[str, pd.DataFrame],
cutoff: int,
all_uids: np.ndarray,
) -> dict[str, sparse.csr_matrix | np.ndarray]:
umap = uid_mapper(all_uids)
n_users = len(all_uids)
lu, lts, lc, lo, lr = collect_listen_triples(events["listens"], cutoff, min_ratio=1)
pu, pts, pc, po, pr = collect_listen_triples(events["listens"], cutoff, min_ratio=LISTEN_POS)
su, sts, sc, so, sr = collect_listen_triples(events["listens"], cutoff, min_ratio=LISTEN_STRONG)
ku, kts, kc, ko = collect_flag_triples(events["likes"], cutoff)
du, dts, dc, do = collect_flag_triples(events["dislikes"], cutoff)
def rows_of(uids: np.ndarray) -> np.ndarray:
return np.fromiter((umap[int(u)] for u in uids), dtype=np.int32, count=len(uids))
def w_listen(ts, organic, ratio, half, org_boost: float, min_ratio: int) -> np.ndarray:
dec = decay_from_age(cutoff - ts, half)
org = np.where(organic == 1, org_boost, 1.0).astype(np.float32)
strength = np.clip(ratio.astype(np.float32) / 100.0, 0.05, 1.0)
strength = np.where(ratio >= min_ratio, strength, 0.1 * strength)
return (dec * org * strength).astype(np.float32)
lu_rows = rows_of(lu)
pu_rows = rows_of(pu)
su_rows = rows_of(su)
ku_rows = rows_of(ku)
du_rows = rows_of(du)
X_bin = aggregate_pairs(pu_rows, pc, np.ones(len(pc), dtype=np.float32), n_users, N_ITEMS)
X_bin.data = np.minimum(X_bin.data, 1.0)
X_org_bin = aggregate_pairs(
pu_rows[po == 1],
pc[po == 1],
np.ones((po == 1).sum(), dtype=np.float32),
n_users,
N_ITEMS,
)
X_org_bin.data = np.minimum(X_org_bin.data, 1.0)
X_strong_org = aggregate_pairs(
su_rows[so == 1],
sc[so == 1],
np.ones((so == 1).sum(), dtype=np.float32),
n_users,
N_ITEMS,
)
X_strong_org.data = np.minimum(X_strong_org.data, 1.0)
X_w14 = aggregate_pairs(
pu_rows, pc, w_listen(pts, po, pr, 14.0, 2.5, LISTEN_POS), n_users, N_ITEMS
)
X_w30 = aggregate_pairs(
pu_rows, pc, w_listen(pts, po, pr, 30.0, 2.0, LISTEN_POS), n_users, N_ITEMS
)
X_w7 = aggregate_pairs(
pu_rows, pc, w_listen(pts, po, pr, 7.0, 3.0, LISTEN_POS), n_users, N_ITEMS
)
X_org_w = aggregate_pairs(
pu_rows[po == 1],
pc[po == 1],
w_listen(pts[po == 1], po[po == 1], pr[po == 1], 21.0, 1.0, LISTEN_POS),
n_users,
N_ITEMS,
)
like_w = (
decay_from_age(cutoff - kts, 45.0)
* np.where(ko == 1, 6.0, 3.0).astype(np.float32)
)
X_like = aggregate_pairs(ku_rows, kc, like_w, n_users, N_ITEMS)
X_like_bin = aggregate_pairs(ku_rows, kc, np.ones(len(kc), dtype=np.float32), n_users, N_ITEMS)
X_like_bin.data = np.minimum(X_like_bin.data, 1.0)
X_org_like = aggregate_pairs(
ku_rows[ko == 1],
kc[ko == 1],
np.ones((ko == 1).sum(), dtype=np.float32),
n_users,
N_ITEMS,
)
X_org_like.data = np.minimum(X_org_like.data, 1.0)
# Boost likes into weighted profiles.
X_w14 = X_w14 + X_like
X_w30 = X_w30 + X_like
X_w7 = X_w7 + sparse.csr_matrix(
(
decay_from_age(cutoff - kts, 14.0) * np.where(ko == 1, 8.0, 3.5).astype(np.float32),
(ku_rows, kc),
),
shape=(n_users, N_ITEMS),
dtype=np.float32,
)
X_org_w = X_org_w + X_org_like.multiply(4.0)
skip_mask = lr <= 20
X_skip = aggregate_pairs(
lu_rows[skip_mask],
lc[skip_mask],
decay_from_age(cutoff - lts[skip_mask], 30.0),
n_users,
N_ITEMS,
)
X_dis = aggregate_pairs(
du_rows, dc, decay_from_age(cutoff - dts, 60.0), n_users, N_ITEMS
)
# Session / sequential co-visitation on positive catalog listens.
session = _session_matrix(pu_rows, pts, pc, po, n_users)
pop7, pop30, pop_like, pop_org_strong = _popularity_vectors(
pts, pc, po, pr, kts, kc, ko, cutoff
)
return {
"X_bin": X_bin,
"X_org_bin": X_org_bin,
"X_strong_org": X_strong_org,
"X_w7": X_w7.tocsr(),
"X_w14": X_w14.tocsr(),
"X_w30": X_w30.tocsr(),
"X_org_w": X_org_w.tocsr(),
"X_like_bin": X_like_bin,
"X_org_like": X_org_like,
"X_skip": X_skip,
"X_dis": X_dis,
"session": session,
"pop7": pop7,
"pop30": pop30,
"pop_like": pop_like,
"pop_org_strong": pop_org_strong,
}
def _popularity_vectors(pts, pc, po, pr, kts, kc, ko, cutoff):
pop7 = np.zeros(N_ITEMS, dtype=np.float32)
pop30 = np.zeros(N_ITEMS, dtype=np.float32)
pop_org_strong = np.zeros(N_ITEMS, dtype=np.float32)
pop_like = np.zeros(N_ITEMS, dtype=np.float32)
rec7 = pts >= cutoff - 7 * DAY
rec30 = pts >= cutoff - 30 * DAY
strong_org = (po == 1) & (pr >= LISTEN_STRONG)
np.add.at(pop7, pc[rec7 & strong_org], 1.0)
np.add.at(pop30, pc[rec30 & strong_org], 1.0)
np.add.at(pop_org_strong, pc[strong_org], 1.0)
rec_like = (kts >= cutoff - 30 * DAY) & (ko == 1)
np.add.at(pop_like, kc[rec_like], 1.0)
return (
np.log1p(pop7),
np.log1p(pop30),
np.log1p(pop_like),
np.log1p(pop_org_strong),
)
def _session_matrix(rows, ts, cols, organic, n_users) -> sparse.csr_matrix:
order = np.lexsort((ts, rows))
rows, ts, cols, organic = rows[order], ts[order], cols[order], organic[order]
same = rows[1:] == rows[:-1]
dt = ts[1:] - ts[:-1]
close = (dt >= 0) & (dt <= 1800)
mask = same & close
src, dst = cols[:-1][mask], cols[1:][mask]
org_pair = ((organic[:-1] == 1) | (organic[1:] == 1))[mask]
w = np.where(org_pair, 1.5, 0.75).astype(np.float32)
w *= (1.0 / (1.0 + dt[mask] / 300.0)).astype(np.float32)
keep = src != dst
src, dst, w = src[keep], dst[keep], w[keep]
fwd = sparse.coo_matrix((w, (src, dst)), shape=(N_ITEMS, N_ITEMS))
# Symmetric short-window co-listen (album/artist clusters).
close2 = same & (dt >= 0) & (dt <= 240)
src2, dst2 = cols[:-1][close2], cols[1:][close2]
keep2 = src2 != dst2
w2 = np.ones(keep2.sum(), dtype=np.float32)
sym = sparse.coo_matrix((w2, (src2[keep2], dst2[keep2])), shape=(N_ITEMS, N_ITEMS))
S = (fwd + fwd.T.multiply(0.35) + (sym + sym.T)).tocsr()
S = _topk_sparse_rows(S, 150)
S = normalize(S, norm="l1", axis=1, copy=False)
return S.astype(np.float32)
def _topk_sparse_rows(mat: sparse.csr_matrix, k: int) -> sparse.csr_matrix:
mat = mat.tocsr()
indptr = [0]
indices = []
data = []
for r in range(mat.shape[0]):
sl = slice(mat.indptr[r], mat.indptr[r + 1])
idx = mat.indices[sl]
val = mat.data[sl]
if idx.size == 0:
indptr.append(len(indices))
continue
if idx.size > k:
take = np.argpartition(val, -k)[-k:]
idx, val = idx[take], val[take]
indices.append(idx)
data.append(val)
indptr.append(len(indices) if not indices else indptr[-1] + idx.size)
# rebuild more carefully
indptr = [0]
indices_cat = []
data_cat = []
for r in range(mat.shape[0]):
sl = slice(mat.indptr[r], mat.indptr[r + 1])
idx = mat.indices[sl]
val = mat.data[sl]
if idx.size > k:
take = np.argpartition(val, -k)[-k:]
idx, val = idx[take], val[take]
indices_cat.append(idx)
data_cat.append(val)
indptr.append(indptr[-1] + idx.size)
if not indices_cat:
return sparse.csr_matrix(mat.shape, dtype=np.float32)
return sparse.csr_matrix(
(np.concatenate(data_cat), np.concatenate(indices_cat), np.asarray(indptr)),
shape=mat.shape,
dtype=np.float32,
)
def idf_weights(X: sparse.csr_matrix) -> np.ndarray:
df = np.asarray((X > 0).sum(axis=0)).ravel().astype(np.float32)
n = X.shape[0]
return np.log((n + 1.0) / (df + 1.0)).astype(np.float32) + 1.0
def bm25_weight(X: sparse.csr_matrix, k1: float = 1.2, b: float = 0.75) -> sparse.csr_matrix:
"""BM25-weight a user-item CSR matrix (users as documents)."""
X = X.tocsr().astype(np.float32)
n_users, n_items = X.shape
dl = np.asarray(X.sum(axis=1)).ravel()
avgdl = float(dl.mean()) + 1e-6
df = np.bincount(X.indices, minlength=n_items).astype(np.float32)
idf = np.log((n_users - df + 0.5) / (df + 0.5) + 1.0).astype(np.float32)
X = X.copy()
for u in range(n_users):
sl = slice(X.indptr[u], X.indptr[u + 1])
tf = X.data[sl]
denom = tf + k1 * (1.0 - b + b * dl[u] / avgdl)
X.data[sl] = ((k1 + 1.0) * tf / denom) * idf[X.indices[sl]]
return X
def l2_normalize_columns(X: sparse.csr_matrix) -> np.ndarray:
Xc = X.tocsc().astype(np.float32)
norms = np.sqrt(np.asarray(Xc.power(2).sum(axis=0))).ravel()
norms = np.maximum(norms, 1e-8)
Xc = Xc.multiply(1.0 / norms)
return np.ascontiguousarray(Xc.toarray().T.astype(np.float32).T) # users x items C-order
def column_normalized_dense(X: sparse.csr_matrix) -> np.ndarray:
dense = np.ascontiguousarray(X.astype(np.float32).toarray())
norms = np.sqrt((dense * dense).sum(axis=0, keepdims=True))
norms = np.maximum(norms, 1e-8)
dense /= norms
return dense
def score_item_cosine(profile: sparse.csr_matrix, item_user: np.ndarray) -> np.ndarray:
"""profile: n_tgt x n_items, item_user: n_users x n_items with L2 columns."""
mid = profile @ item_user.T
return np.ascontiguousarray(mid @ item_user, dtype=np.float32)
def score_rp3beta(
profile: sparse.csr_matrix,
X: sparse.csr_matrix,
beta: float,
) -> np.ndarray:
"""P3 * pop^{-beta} via user space: (p @ X_item_norm.T) @ X_user_norm."""
X = X.tocsr().astype(np.float32)
item_pop = np.asarray(X.sum(axis=0)).ravel().astype(np.float32)
user_deg = np.asarray(X.sum(axis=1)).ravel().astype(np.float32)
Xu = normalize(X, norm="l1", axis=1) # user -> item
Xi = X.multiply(1.0 / np.maximum(item_pop, 1e-8)) # item -> user (col L1)
Xi = Xi.tocsr()
# p @ Xi.T : n_tgt x n_users (profile lives on items)
# Xi is users x items, columns scale 1/pop, so Xi.T is items x users
mid = profile @ Xi.T # tgt x users
scores = mid @ Xu # tgt x items
scores = np.ascontiguousarray(scores, dtype=np.float32)
scores *= np.power(np.maximum(item_pop, 1.0), -beta).astype(np.float32)
return scores
def score_user_knn(
profile: sparse.csr_matrix,
X: sparse.csr_matrix,
target_rows: np.ndarray,
k: int = 80,
) -> np.ndarray:
Xn = normalize(X.astype(np.float32), norm="l2", axis=1)
Pn = normalize(profile.astype(np.float32), norm="l2", axis=1)
sim = Pn @ Xn.T # tgt x users
if sparse.issparse(sim):
sim = sim.toarray()
sim = np.ascontiguousarray(sim, dtype=np.float32)
sim[np.arange(len(target_rows)), target_rows] = -1.0
k = min(k, sim.shape[1] - 1)
idx = np.argpartition(-sim, k, axis=1)[:, :k]
rows = np.repeat(np.arange(sim.shape[0]), k)
cols = idx.ravel()
vals = sim[np.arange(sim.shape[0])[:, None], idx].ravel()
vals = np.clip(vals, 0.0, None)
Sk = sparse.csr_matrix((vals, (rows, cols)), shape=sim.shape, dtype=np.float32)
scores = Sk @ X.astype(np.float32)
if sparse.issparse(scores):
scores = scores.toarray()
return np.ascontiguousarray(scores, dtype=np.float32)
def score_svd(X: sparse.csr_matrix, target_rows: np.ndarray, n_comp: int = 128) -> np.ndarray:
n_comp = min(n_comp, min(X.shape) - 1)
svd = TruncatedSVD(n_components=n_comp, random_state=42, n_iter=7)
user_f = svd.fit_transform(X.astype(np.float32)) # n_users x f (U * S)
item_f = svd.components_.astype(np.float32) # f x n_items
scores = user_f[target_rows] @ item_f
return np.ascontiguousarray(scores, dtype=np.float32)
def implicit_als(
R: sparse.csr_matrix,
target_rows: np.ndarray,
factors: int = 64,
iterations: int = 8,
alpha: float = 20.0,
reg: float = 0.15,
seed: int = 0,
) -> np.ndarray:
R = R.tocsr().astype(np.float32)
Rt = R.T.tocsr()
n_users, n_items = R.shape
rng = np.random.default_rng(seed)
X = rng.normal(0.0, 0.01, size=(n_users, factors)).astype(np.float64)
Y = rng.normal(0.0, 0.01, size=(n_items, factors)).astype(np.float64)
def _update(factors_out, other, mat, n_entity):
YtY = other.T @ other
eye = np.eye(factors, dtype=np.float64) * reg
for i in range(n_entity):
sl = slice(mat.indptr[i], mat.indptr[i + 1])
idx = mat.indices[sl]
if idx.size == 0:
factors_out[i] = 0.0
continue
data = mat.data[sl].astype(np.float64)
c = 1.0 + alpha * data
Y_i = other[idx]
A = YtY + Y_i.T @ ((c - 1.0)[:, None] * Y_i) + eye
b = Y_i.T @ c
try:
factors_out[i] = np.linalg.solve(A, b)
except np.linalg.LinAlgError:
factors_out[i] = np.linalg.lstsq(A, b, rcond=None)[0]
t0 = time.time()
for it in range(iterations):
_update(X, Y, R, n_users)
_update(Y, X, Rt, n_items)
log(f" ALS iter {it + 1}/{iterations} ({time.time() - t0:.1f}s)")
scores = X[target_rows] @ Y.T
return np.ascontiguousarray(scores, dtype=np.float32)
def zscore_masked(scores: np.ndarray, eligible: np.ndarray) -> np.ndarray:
scores = scores.astype(np.float32, copy=False)
el = eligible
cnt = el.sum(axis=1, keepdims=True).astype(np.float32)
cnt = np.maximum(cnt, 1.0)
mean = np.where(el, scores, 0.0).sum(axis=1, keepdims=True) / cnt
diff = np.where(el, scores - mean, 0.0)
std = np.sqrt((diff * diff).sum(axis=1, keepdims=True) / cnt)
z = np.where(el, (scores - mean) / (std + 1e-6), 0.0)
return z.astype(np.float32)
def apply_ineligible(scores: np.ndarray, eligible: np.ndarray) -> np.ndarray:
out = scores.astype(np.float32, copy=True)
out[~eligible] = -1e9
# tiny popularity tie-break
out += (1e-7 * (N_ITEMS - POP_RANK)).astype(np.float32)
return out
def topk_indices(scores: np.ndarray, k: int = TOP_N) -> np.ndarray:
idx = np.argpartition(-scores, kth=k - 1, axis=1)[:, :k]
row = np.arange(scores.shape[0])[:, None]
part = scores[row, idx]
order = np.argsort(-part, axis=1)
return idx[row, order]
def ranking_frame(uids: np.ndarray, item_idx: np.ndarray) -> pd.DataFrame:
n_users, k = item_idx.shape
return pd.DataFrame(
{
"uid": np.repeat(uids, k).astype(np.int64),
"item_id": ITEM_IDS[item_idx.reshape(-1)].astype(np.int64),
"rank": np.tile(np.arange(1, k + 1, dtype=np.int64), n_users),
}
)
def save_ranking(df: pd.DataFrame, path: Path) -> None:
df.to_parquet(path, index=False)
log(f"wrote {path} rows={len(df)}")
def validation_graded(eligible: np.ndarray) -> dict[int, dict[int, int]]:
uid_row = {int(u): i for i, u in enumerate(TARGET_UIDS)}
listens = pd.read_parquet(DATA / "val_listens.parquet")
likes = pd.read_parquet(DATA / "val_likes.parquet")
targets = set(map(int, TARGET_UIDS))
strong = listens[
listens.uid.isin(targets)
& (listens.is_organic == 1)
& (listens.played_ratio_pct >= LISTEN_STRONG)
][["uid", "item_id"]].drop_duplicates()
olikes = likes[likes.uid.isin(targets) & (likes.is_organic == 1)][
["uid", "item_id"]
].drop_duplicates()
graded: dict[int, dict[int, int]] = {}
for uid, item in strong.itertuples(index=False, name=None):
uid, item = int(uid), int(item)
r = uid_row.get(uid)
c = ITEM_INV[item] if 0 <= item < len(ITEM_INV) else -1
if r is None or c < 0 or not eligible[r, c]:
continue
graded.setdefault(uid, {})[item] = 1
for uid, item in olikes.itertuples(index=False, name=None):
uid, item = int(uid), int(item)
r = uid_row.get(uid)
c = ITEM_INV[item] if 0 <= item < len(ITEM_INV) else -1
if r is None or c < 0 or not eligible[r, c]:
continue
graded.setdefault(uid, {})[item] = 3
return graded
def ndcg_from_indices(item_idx: np.ndarray, graded: dict[int, dict[int, int]]) -> float:
uid_row = {int(u): i for i, u in enumerate(TARGET_UIDS)}
scores = []
for uid, gold in graded.items():
ranked = ITEM_IDS[item_idx[uid_row[uid]]].tolist()
scores.append(ndcg_at_k(ranked, gold, QUALITY_K))
return float(np.mean(scores)) if scores else 0.0
def write_pop_ranking(eligible: np.ndarray, path: Path, uids: np.ndarray) -> np.ndarray:
scores = np.broadcast_to(POP_SCORE, (len(uids), N_ITEMS)).copy()
scores = apply_ineligible(scores, eligible)
idx = topk_indices(scores, TOP_N)
save_ranking(ranking_frame(uids, idx), path)
return idx
def target_slice(mat: sparse.csr_matrix, target_rows: np.ndarray) -> sparse.csr_matrix:
return mat[target_rows]
def scale_profile_idf(P: sparse.csr_matrix, idf: np.ndarray) -> sparse.csr_matrix:
return P.multiply(idf).tocsr()
def build_score_bank(
mats: dict,
target_rows: np.ndarray,
eligible: np.ndarray,
als: bool = True,
) -> dict[str, np.ndarray]:
X_bin = mats["X_bin"]
X_org = mats["X_org_bin"]
X_strong = mats["X_strong_org"]
X_w7 = mats["X_w7"]
X_w14 = mats["X_w14"]
X_w30 = mats["X_w30"]
X_org_w = mats["X_org_w"]
X_like = mats["X_like_bin"]
session = mats["session"]
P14 = target_slice(X_w14, target_rows)
P7 = target_slice(X_w7, target_rows)
P30 = target_slice(X_w30, target_rows)
Porg = target_slice(X_org_w, target_rows)
Plike = target_slice(X_like, target_rows)
bank: dict[str, np.ndarray] = {}
log("column-normalize binary / organic matrices")
Xn_bin = column_normalized_dense(X_bin)
log("item-cosine full profile")
idf_bin = idf_weights(X_bin)
scores = score_item_cosine(scale_profile_idf(P14, idf_bin), Xn_bin)
bank["knn14"] = zscore_masked(scores, eligible)
del scores
log("item-cosine recent profile")
scores = score_item_cosine(scale_profile_idf(P7, idf_bin), Xn_bin)
bank["knn7"] = zscore_masked(scores, eligible)
del scores
log("item-cosine 30d profile")
scores = score_item_cosine(scale_profile_idf(P30, idf_bin), Xn_bin)
bank["knn30"] = zscore_masked(scores, eligible)
del scores, Xn_bin
gc.collect()
log("organic item-cosine")
Xn_org = column_normalized_dense(X_org + X_like)
idf_org = idf_weights(X_org + X_like)
scores = score_item_cosine(scale_profile_idf(Porg, idf_org), Xn_org)
bank["knn_org"] = zscore_masked(scores, eligible)
del scores, Xn_org
gc.collect()
log("BM25 item-cosine")
X_bm = bm25_weight(X_bin + X_like.multiply(2.0))
Xn_bm = column_normalized_dense(X_bm)
scores = score_item_cosine(P14, Xn_bm)
bank["bm25"] = zscore_masked(scores, eligible)
del scores, Xn_bm, X_bm
gc.collect()
log("RP3beta")
scores = score_rp3beta(P14, X_bin + X_like, beta=0.4)
bank["rp3"] = zscore_masked(scores, eligible)
del scores
scores = score_rp3beta(Porg, X_org + X_like, beta=0.55)
bank["rp3_org"] = zscore_masked(scores, eligible)
del scores
gc.collect()
log("user-kNN")
scores = score_user_knn(P14, X_bin + X_like.multiply(1.5), target_rows, k=90)
bank["uknn"] = zscore_masked(scores, eligible)
del scores
gc.collect()
log("SVD")
X_svd = bm25_weight(X_w30 + X_like.multiply(2.0))
scores = score_svd(X_svd, target_rows, n_comp=128)
bank["svd"] = zscore_masked(scores, eligible)
del scores
scores = score_svd(bm25_weight(X_org_w + X_like.multiply(3.0)), target_rows, n_comp=96)
bank["svd_org"] = zscore_masked(scores, eligible)
del scores, X_svd
gc.collect()
log("session expansion")
sess_scores = P7 @ session
if sparse.issparse(sess_scores):
sess_scores = sess_scores.toarray()
bank["session"] = zscore_masked(np.ascontiguousarray(sess_scores, dtype=np.float32), eligible)
del sess_scores
sess_scores = P14 @ session
if sparse.issparse(sess_scores):
sess_scores = sess_scores.toarray()
bank["session14"] = zscore_masked(np.ascontiguousarray(sess_scores, dtype=np.float32), eligible)
del sess_scores
gc.collect()
log("like-knn")
if X_like.nnz > 0:
Xn_like = column_normalized_dense(X_like + X_strong.multiply(0.25))
scores = score_item_cosine(scale_profile_idf(Plike + Porg.multiply(0.15), idf_weights(X_like + 1e-6)), Xn_like)
bank["like_knn"] = zscore_masked(scores, eligible)
del scores, Xn_like
gc.collect()
log("popularity features")
bank["pop"] = zscore_masked(np.broadcast_to(POP_SCORE, (P14.shape[0], N_ITEMS)), eligible)
bank["pop7"] = zscore_masked(np.broadcast_to(mats["pop7"], (P14.shape[0], N_ITEMS)), eligible)
bank["pop30"] = zscore_masked(np.broadcast_to(mats["pop30"], (P14.shape[0], N_ITEMS)), eligible)
bank["pop_like"] = zscore_masked(
np.broadcast_to(mats["pop_like"], (P14.shape[0], N_ITEMS)), eligible
)
bank["pop_org"] = zscore_masked(
np.broadcast_to(mats["pop_org_strong"], (P14.shape[0], N_ITEMS)), eligible
)
skip_item = np.asarray(mats["X_skip"].sum(axis=0)).ravel().astype(np.float32)
listen_item = np.asarray(X_bin.sum(axis=0)).ravel().astype(np.float32)
skip_rate = skip_item / np.maximum(skip_item + listen_item, 1.0)
bank["skip_pen"] = zscore_masked(
np.broadcast_to(-skip_rate, (P14.shape[0], N_ITEMS)), eligible
)
if als:
log("implicit ALS")
R = (X_w14 + X_like.multiply(2.0) + X_org_w.multiply(0.5)).tocsr()
R.data = np.maximum(R.data, 0.0)
try:
scores = implicit_als(R, target_rows, factors=48, iterations=7, alpha=25.0, reg=0.2)
bank["als"] = zscore_masked(scores, eligible)
del scores
except Exception as exc:
log(f"ALS failed: {exc}")
del R
gc.collect()
return bank
def blend_equal(bank: dict[str, np.ndarray], keys: list[str], weights: list[float] | None = None) -> np.ndarray:
acc = None
if weights is None:
weights = [1.0] * len(keys)
for k, w in zip(keys, weights):
if k not in bank or w == 0:
continue
acc = bank[k] * w if acc is None else acc + bank[k] * w
return acc
def sample_ridge_dataset(
bank: dict[str, np.ndarray],
keys: list[str],
graded: dict[int, dict[int, int]],
eligible: np.ndarray,
rng: np.random.Generator,
n_neg: int = 60,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
uid_row = {int(u): i for i, u in enumerate(TARGET_UIDS)}
xs = []
ys = []
ws = []
for uid, gold in graded.items():
r = uid_row[uid]
pos_cols = []
pos_grades = []
for item, grade in gold.items():
c = int(ITEM_INV[item]) if item < len(ITEM_INV) else -1
if c >= 0 and eligible[r, c]:
pos_cols.append(c)
pos_grades.append(grade)
if not pos_cols:
continue
el_idx = np.flatnonzero(eligible[r])
pos_set = set(pos_cols)
neg_pool = el_idx[np.fromiter((c not in pos_set for c in el_idx), dtype=bool, count=len(el_idx))]
if len(neg_pool) == 0:
continue
take = min(n_neg, len(neg_pool))
# mix random and hard (high knn) negatives
n_rand = take // 2
n_hard = take - n_rand
rand_cols = rng.choice(neg_pool, size=n_rand, replace=False)
knn_key = "knn14" if "knn14" in bank else keys[0]
hard_scores = bank[knn_key][r, neg_pool]
hard_idx = np.argpartition(-hard_scores, min(n_hard, len(neg_pool) - 1))[:n_hard]
hard_cols = neg_pool[hard_idx]
cols = np.concatenate([np.asarray(pos_cols, dtype=np.int32), rand_cols.astype(np.int32), hard_cols.astype(np.int32)])
grades = np.concatenate(
[np.asarray(pos_grades, dtype=np.float32), np.zeros(len(rand_cols) + len(hard_cols), dtype=np.float32)]
)
feat = np.stack([bank[k][r, cols] for k in keys], axis=1)
xs.append(feat)
ys.append(grades)
wrow = np.where(grades >= 3, 4.0, np.where(grades >= 1, 1.5, 1.0)).astype(np.float32)
ws.append(wrow)
X = np.concatenate(xs, axis=0)
y = np.concatenate(ys, axis=0)
w = np.concatenate(ws, axis=0)
return X, y, w
def ridge_weights(bank, keys, graded, eligible) -> np.ndarray:
rng = np.random.default_rng(42)
X, y, w = sample_ridge_dataset(bank, keys, graded, eligible, rng)
y_bin = (y > 0).astype(np.float32)
y_gain = np.where(y >= 3, 7.0, y).astype(np.float32)
model = Ridge(alpha=15.0, random_state=42)
model.fit(X, y_gain, sample_weight=w)
coef = model.coef_.astype(np.float64)
log(f"ridge coefs: {list(zip(keys, np.round(coef, 4)))} intercept={model.intercept_:.4f}")
return coef
def scores_to_ranking(scores: np.ndarray, eligible: np.ndarray) -> np.ndarray:
return topk_indices(apply_ineligible(scores, eligible), TOP_N)
def rrf_from_bank(bank: dict[str, np.ndarray], keys: list[str], eligible: np.ndarray, k: int = 60, topm: int = 400) -> np.ndarray:
n = next(iter(bank.values())).shape[0]
acc = np.zeros((n, N_ITEMS), dtype=np.float32)
for key in keys:
s = apply_ineligible(bank[key], eligible)
idx = np.argpartition(-s, kth=topm - 1, axis=1)[:, :topm]
row = np.arange(n)[:, None]
order = np.argsort(-s[row, idx], axis=1)
ranked = idx[row, order]
ranks = np.empty_like(ranked)
ranks[row, ranked.argsort(axis=1)] = np.arange(topm, dtype=np.int32)
# only the topm get rrf credit; reconstruct via scatter
contrib = 1.0 / (k + 1.0 + np.arange(topm, dtype=np.float32))
for i in range(n):
acc[i, ranked[i]] += contrib
return acc
def main() -> None:
t_all = time.time()
log("loading eligibility + preference")
val_el = load_eligibility(DATA / "validation_eligible_candidates.npz", TARGET_UIDS)
test_el = load_eligibility(DATA / "test_eligible_candidates.npz", TARGET_UIDS)
log("writing popularity fallback submissions")
write_pop_ranking(test_el, OUT / "submission.parquet", TARGET_UIDS)
pop_val_idx = write_pop_ranking(val_el, OUT / "validation.parquet", TARGET_UIDS)
log("loading train events")
train_events = read_events("train")
log("building validation-cutoff matrices")
all_uids = unique_sorted_uids(
TARGET_UIDS,
train_events["listens"].uid.to_numpy(dtype=np.int64),
train_events["likes"].uid.to_numpy(dtype=np.int64),
)
uid_index = uid_mapper(all_uids)
target_rows = np.array([uid_index[int(u)] for u in TARGET_UIDS], dtype=np.int32)
mats = build_matrices(train_events, TRAIN_CUTOFF, all_uids)
log("score bank (validation cutoff)")
bank = build_score_bank(mats, target_rows, val_el, als=True)
graded = validation_graded(val_el)
log(f"validation users with relevance: {len(graded)}")
pop_ndcg = ndcg_from_indices(pop_val_idx, graded)
log(f"popularity NDCG@10={pop_ndcg:.6f}")
method_ndcg = {}
for key in bank:
idx = scores_to_ranking(bank[key], val_el)
method_ndcg[key] = ndcg_from_indices(idx, graded)
log(f" {key:12s} NDCG@10={method_ndcg[key]:.6f}")
cf_keys = [
k
for k in [
"knn14",
"knn7",
"knn30",
"knn_org",
"bm25",
"rp3",
"rp3_org",
"uknn",
"svd",
"svd_org",
"session",
"session14",
"like_knn",
"als",
]
if k in bank
]
pop_keys = [k for k in ["pop7", "pop30", "pop_like", "pop_org", "skip_pen", "pop"] if k in bank]
keys = cf_keys + pop_keys
# NDCG-weighted z blend (unsupervised, honest)
ndcgs = np.array([method_ndcg[k] for k in keys], dtype=np.float64)
w_ndcg = np.clip(ndcgs - pop_ndcg, 0, None)
w_ndcg = w_ndcg / (w_ndcg.sum() + 1e-9)
# boost methods clearly above median
med = np.median(ndcgs)
w_ndcg = w_ndcg * (1.0 + 1.5 * (ndcgs > med))
w_ndcg = w_ndcg / (w_ndcg.sum() + 1e-9)
log(f"ndcg weights: {list(zip(keys, np.round(w_ndcg, 4)))}")
blend_ndcg_scores = blend_equal(bank, keys, w_ndcg.tolist())
idx_ndcg = scores_to_ranking(blend_ndcg_scores, val_el)
ndcg_w = ndcg_from_indices(idx_ndcg, graded)
log(f"ndcg-weighted blend NDCG@10={ndcg_w:.6f}")
# equal top methods
top_methods = [k for k, _ in sorted(method_ndcg.items(), key=lambda kv: -kv[1])[:8]]
idx_eq = scores_to_ranking(blend_equal(bank, top_methods), val_el)
ndcg_eq = ndcg_from_indices(idx_eq, graded)
log(f"equal top8 {top_methods} NDCG@10={ndcg_eq:.6f}")
# ridge on val labels
coef = ridge_weights(bank, keys, graded, val_el)
blend_ridge = blend_equal(bank, keys, coef.tolist())
idx_ridge = scores_to_ranking(blend_ridge, val_el)
ndcg_ridge = ndcg_from_indices(idx_ridge, graded)
log(f"ridge blend NDCG@10={ndcg_ridge:.6f}")
# RRF of strongest methods
rrf_keys = [k for k, _ in sorted(method_ndcg.items(), key=lambda kv: -kv[1])[:10]]
rrf_scores = rrf_from_bank(bank, rrf_keys, val_el)
idx_rrf = scores_to_ranking(rrf_scores, val_el)
ndcg_rrf = ndcg_from_indices(idx_rrf, graded)
log(f"RRF top10 NDCG@10={ndcg_rrf:.6f}")
# hybrid: ridge + small ndcg-weighted
hybrid = blend_ridge + 0.35 * blend_ndcg_scores
idx_hyb = scores_to_ranking(hybrid, val_el)
ndcg_hyb = ndcg_from_indices(idx_hyb, graded)
log(f"hybrid ridge+ndcg NDCG@10={ndcg_hyb:.6f}")
candidates = {
"ndcg_w": (ndcg_w, idx_ndcg, "ndcg", w_ndcg, keys),
"equal": (ndcg_eq, idx_eq, "equal", np.ones(len(top_methods)), top_methods),
"ridge": (ndcg_ridge, idx_ridge, "ridge", coef, keys),
"rrf": (ndcg_rrf, idx_rrf, "rrf", None, rrf_keys),
"hybrid": (ndcg_hyb, idx_hyb, "hybrid", coef, keys),
}
best_name, (best_ndcg, best_idx, best_kind, best_coef, best_keys) = max(
candidates.items(), key=lambda kv: kv[1][0]
)
log(f"selected validation blend={best_name} NDCG@10={best_ndcg:.6f}")
save_ranking(ranking_frame(TARGET_UIDS, best_idx), OUT / "validation.parquet")
# Free validation bank before refit
del bank, mats, blend_ndcg_scores, blend_ridge, hybrid, rrf_scores
gc.collect()
log("loading val events for refit")
all_events = concat_splits(["train", "val"])
all_uids2 = unique_sorted_uids(
TARGET_UIDS,
all_events["listens"].uid.to_numpy(dtype=np.int64),
all_events["likes"].uid.to_numpy(dtype=np.int64),
)
uid_index2 = uid_mapper(all_uids2)
target_rows2 = np.array([uid_index2[int(u)] for u in TARGET_UIDS], dtype=np.int32)
log("building test-cutoff matrices (train+val)")
mats2 = build_matrices(all_events, VAL_END, all_uids2)
log("score bank (test cutoff)")
bank2 = build_score_bank(mats2, target_rows2, test_el, als=True)
if best_kind == "rrf":
test_scores = rrf_from_bank(bank2, best_keys, test_el)
elif best_kind == "hybrid":
test_scores = blend_equal(bank2, keys, coef.tolist()) + 0.35 * blend_equal(
bank2, keys, w_ndcg.tolist()
)
elif best_kind == "equal":
test_scores = blend_equal(bank2, best_keys)
elif best_kind == "ndcg":
test_scores = blend_equal(bank2, best_keys, np.asarray(best_coef).tolist())
else:
test_scores = blend_equal(bank2, best_keys, np.asarray(best_coef).tolist())
# extra insurance: if a method is missing after refit, skip
test_idx = scores_to_ranking(test_scores, test_el)
save_ranking(ranking_frame(TARGET_UIDS, test_idx), OUT / "submission.parquet")
log(f"done in {time.time() - t_all:.1f}s val_best={best_name} {best_ndcg:.6f}")
if __name__ == "__main__":
main()