Tool call · Input
/app/solution/build_ranker.py
#!/usr/bin/env python3
"""Discovery ranker for the Made-for-You shelf.
Fits collaborative models on cutoff-correct history, ranks eligible catalog
tracks, and writes validation + hidden-test submissions.
"""
from __future__ import annotations
import os
import sys
import time
from pathlib import Path
os.environ.setdefault("OMP_NUM_THREADS", "8")
os.environ.setdefault("MKL_NUM_THREADS", "8")
os.environ.setdefault("OPENBLAS_NUM_THREADS", "8")
os.environ.setdefault("NUMEXPR_NUM_THREADS", "8")
import numpy as np
import pandas as pd
from scipy import sparse
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")
TRAIN_CUTOFF = 24_786_800
TEST_CUTOFF = 25_395_200
DAY = 86_400.0
LISTEN_STRONG = 80
LISTEN_POS = 50
def log(msg: str) -> None:
print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)
def load_catalog() -> tuple[np.ndarray, dict[int, int]]:
cat = pd.read_parquet(DATA / "candidate_catalog.parquet")
item_ids = cat.item_id.to_numpy(dtype=np.int64)
item_to_col = {int(i): c for c, i in enumerate(item_ids)}
return item_ids, item_to_col
def load_targets() -> np.ndarray:
return pd.read_parquet(DATA / "target_users.parquet").uid.to_numpy(dtype=np.int64)
def load_eligibility(path: Path, n_items: int) -> tuple[np.ndarray, np.ndarray]:
with np.load(path, allow_pickle=False) as z:
uids = z["uids"].astype(np.int64, copy=True)
packed = z["packed_eligible"]
elig = np.unpackbits(packed, axis=1, bitorder="little")[:, :n_items].astype(bool)
return uids, elig
def item_index_lookup(item_ids: np.ndarray) -> np.ndarray:
vmax = int(item_ids.max()) + 1
table = np.full(vmax, -1, dtype=np.int32)
table[item_ids] = np.arange(len(item_ids), dtype=np.int32)
return table
def map_users(uids: np.ndarray) -> tuple[np.ndarray, dict[int, int], np.ndarray]:
uniq = pd.unique(uids)
uniq = uniq.astype(np.int64, copy=False)
table_max = int(uniq.max()) + 1
table = np.full(table_max, -1, dtype=np.int32)
table[uniq] = np.arange(len(uniq), dtype=np.int32)
return uniq, {int(u): i for i, u in enumerate(uniq)}, table
def decay_weights(ts: np.ndarray, now: int, half_life_days: float) -> np.ndarray:
age = np.maximum(0.0, (now - ts.astype(np.float64)) / DAY)
return np.exp2(-age / half_life_days).astype(np.float32)
def read_parquet(name: str, columns: list[str]) -> pd.DataFrame:
return pd.read_parquet(DATA / name, columns=columns)
def collect_positive_matrix(
listen_files: list[str],
like_files: list[str],
now: int,
item_table: np.ndarray,
n_items: int,
half_life: float,
query_half_life: float,
) -> tuple[sparse.csr_matrix, sparse.csr_matrix, np.ndarray, np.ndarray, np.ndarray]:
"""Build full-history and query user-item matrices over catalog items."""
parts_u = []
parts_i = []
parts_w = []
parts_q = []
parts_ts = []
for name in listen_files:
log(f" load {name}")
df = read_parquet(
name,
["uid", "timestamp", "item_id", "is_organic", "played_ratio_pct"],
)
items = df.item_id.to_numpy()
valid = items < len(item_table)
cols = np.full(len(df), -1, dtype=np.int32)
cols[valid] = item_table[items[valid]]
ratio = df.played_ratio_pct.to_numpy()
keep = (cols >= 0) & (ratio >= LISTEN_POS)
if not keep.any():
continue
cols = cols[keep]
uids = df.uid.to_numpy()[keep].astype(np.int64, copy=False)
ts = df.timestamp.to_numpy()[keep].astype(np.int64, copy=False)
organic = df.is_organic.to_numpy()[keep].astype(np.int8, copy=False)
ratio = ratio[keep].astype(np.float32, copy=False)
strong = ratio >= LISTEN_STRONG
# Stronger organic / completion signal, scaled by play-through.
base = np.where(strong, 1.6, 0.7).astype(np.float32)
base *= 0.55 + 0.45 * np.minimum(ratio, 100.0) / 100.0
base *= np.where(organic == 1, 1.7, 1.0).astype(np.float32)
parts_u.append(uids)
parts_i.append(cols)
parts_w.append(base * decay_weights(ts, now, half_life))
parts_q.append(base * decay_weights(ts, now, query_half_life))
parts_ts.append(ts)
del df
for name in like_files:
log(f" load {name}")
df = read_parquet(name, ["uid", "timestamp", "item_id", "is_organic"])
items = df.item_id.to_numpy()
valid = items < len(item_table)
cols = np.full(len(df), -1, dtype=np.int32)
cols[valid] = item_table[items[valid]]
keep = cols >= 0
if not keep.any():
continue
cols = cols[keep]
uids = df.uid.to_numpy()[keep].astype(np.int64, copy=False)
ts = df.timestamp.to_numpy()[keep].astype(np.int64, copy=False)
organic = df.is_organic.to_numpy()[keep].astype(np.int8, copy=False)
base = np.where(organic == 1, 8.0, 4.0).astype(np.float32)
parts_u.append(uids)
parts_i.append(cols)
parts_w.append(base * decay_weights(ts, now, half_life))
parts_q.append(base * decay_weights(ts, now, query_half_life))
parts_ts.append(ts)
del df
all_u = np.concatenate(parts_u)
all_i = np.concatenate(parts_i)
all_w = np.concatenate(parts_w)
all_q = np.concatenate(parts_q)
all_ts = np.concatenate(parts_ts)
user_ids, _, user_table = map_users(all_u)
rows = user_table[all_u]
n_users = len(user_ids)
log(f" aggregate {len(all_u):,} events, {n_users} users, {n_items} items")
X = sparse.csr_matrix((all_w, (rows, all_i)), shape=(n_users, n_items), dtype=np.float32)
Q = sparse.csr_matrix((all_q, (rows, all_i)), shape=(n_users, n_items), dtype=np.float32)
X.sum_duplicates()
Q.sum_duplicates()
X.data = np.maximum(X.data, 0.0)
Q.data = np.maximum(Q.data, 0.0)
return X, Q, user_ids, rows, all_ts
def apply_preference_state(
Q: sparse.csr_matrix,
user_ids: np.ndarray,
pref_path: Path,
item_table: np.ndarray,
like_boost: float = 6.0,
) -> sparse.csr_matrix:
pref = pd.read_parquet(pref_path)
liked = pref[pref.state == "liked"]
if liked.empty:
return Q
uid_max = int(user_ids.max()) + 1
urow = np.full(uid_max, -1, dtype=np.int32)
urow[user_ids] = np.arange(len(user_ids), dtype=np.int32)
u = liked.uid.to_numpy()
it = liked.item_id.to_numpy()
valid_u = u < len(urow)
rows = np.full(len(liked), -1, dtype=np.int32)
rows[valid_u] = urow[u[valid_u]]
valid_i = it < len(item_table)
cols = np.full(len(liked), -1, dtype=np.int32)
cols[valid_i] = item_table[it[valid_i]]
keep = (rows >= 0) & (cols >= 0)
boost = sparse.csr_matrix(
(np.full(int(keep.sum()), like_boost, dtype=np.float32), (rows[keep], cols[keep])),
shape=Q.shape,
dtype=np.float32,
)
return Q + boost
def bm25_weight(X: sparse.csr_matrix, k1: float = 1.2, b: float = 0.75) -> sparse.csr_matrix:
X = X.tocoo()
n_users = X.shape[0]
df = np.bincount(X.col, minlength=X.shape[1]).astype(np.float64)
idf = np.log((n_users - df + 0.5) / (df + 0.5) + 1.0).astype(np.float32)
row_sums = np.ravel(sparse.csr_matrix((X.data, (X.row, X.col)), shape=X.shape).sum(axis=1))
avg_len = float(row_sums.mean()) + 1e-6
length_norm = (1.0 - b) + b * row_sums[X.row] / avg_len
data = X.data.astype(np.float32, copy=True)
data = data * (k1 + 1.0) / (data + k1 * length_norm) * idf[X.col]
data = np.maximum(data, 0.0)
return sparse.csr_matrix((data, (X.row, X.col)), shape=X.shape, dtype=np.float32)
def item_topk_from_gram(
X: sparse.csr_matrix, k: int = 200, batch: int = 256
) -> sparse.csr_matrix:
"""Column-cosine item kNN. X is users x items."""
Xn = normalize(X, norm="l2", axis=0)
Xt = Xn.T.tocsr()
n_items = X.shape[1]
k = min(k, n_items - 1)
rows_out: list[np.ndarray] = []
cols_out: list[np.ndarray] = []
data_out: list[np.ndarray] = []
for start in range(0, n_items, batch):
end = min(start + batch, n_items)
block = Xt[start:end].toarray()
sim = block @ Xn # dense (batch, n_items)
bsz = end - start
for i in range(bsz):
sim[i, start + i] = -np.inf
idx = np.argpartition(sim, -k, axis=1)[:, -k:]
val = np.take_along_axis(sim, idx, axis=1)
order = np.argsort(-val, axis=1)
idx = np.take_along_axis(idx, order, axis=1)
val = np.take_along_axis(val, order, axis=1)
val = np.maximum(val, 0.0)
src = np.repeat(np.arange(start, end), k)
rows_out.append(src.astype(np.int32))
cols_out.append(idx.reshape(-1).astype(np.int32))
data_out.append(val.reshape(-1).astype(np.float32))
if start % 4096 == 0:
log(f" item-knn {end}/{n_items}")
S = sparse.csr_matrix(
(np.concatenate(data_out), (np.concatenate(rows_out), np.concatenate(cols_out))),
shape=(n_items, n_items),
dtype=np.float32,
)
S.eliminate_zeros()
return S
def rp3beta_topk(
Xbin: sparse.csr_matrix, k: int = 200, alpha: float = 0.5, beta: float = 0.4, batch: int = 256
) -> sparse.csr_matrix:
user_deg = np.ravel(Xbin.sum(axis=1)).astype(np.float64)
item_deg = np.ravel(Xbin.sum(axis=0)).astype(np.float64)
user_deg = np.power(np.maximum(user_deg, 1.0), alpha)
item_deg = np.power(np.maximum(item_deg, 1.0), beta)
Xu = Xbin.astype(np.float32).multiply(1.0 / user_deg[:, None])
Xi = Xbin.astype(np.float32).multiply(1.0 / item_deg[None, :])
Xt = Xu.T.tocsr()
n_items = Xbin.shape[1]
k = min(k, n_items - 1)
rows_out, cols_out, data_out = [], [], []
for start in range(0, n_items, batch):
end = min(start + batch, n_items)
block = Xt[start:end].toarray()
sim = block @ Xi
bsz = end - start
for i in range(bsz):
sim[i, start + i] = -np.inf
idx = np.argpartition(sim, -k, axis=1)[:, -k:]
val = np.take_along_axis(sim, idx, axis=1)
order = np.argsort(-val, axis=1)
idx = np.take_along_axis(idx, order, axis=1)
val = np.take_along_axis(val, order, axis=1)
val = np.maximum(val, 0.0)
src = np.repeat(np.arange(start, end), k)
rows_out.append(src.astype(np.int32))
cols_out.append(idx.reshape(-1).astype(np.int32))
data_out.append(val.reshape(-1).astype(np.float32))
if start % 4096 == 0:
log(f" rp3 {end}/{n_items}")
S = sparse.csr_matrix(
(np.concatenate(data_out), (np.concatenate(rows_out), np.concatenate(cols_out))),
shape=(n_items, n_items),
dtype=np.float32,
)
S.eliminate_zeros()
return S
def sequential_item_graph(
listen_files: list[str],
item_table: np.ndarray,
n_items: int,
now: int,
half_life: float,
gap: int = 1800,
topk: int = 80,
) -> sparse.csr_matrix:
src_all = []
dst_all = []
w_all = []
for name in listen_files:
log(f" sequential load {name}")
df = read_parquet(
name, ["uid", "timestamp", "item_id", "played_ratio_pct"]
)
items = df.item_id.to_numpy()
valid = items < len(item_table)
cols = np.full(len(df), -1, dtype=np.int32)
cols[valid] = item_table[items[valid]]
keep = (cols >= 0) & (df.played_ratio_pct.to_numpy() >= LISTEN_POS)
u = df.uid.to_numpy()[keep].astype(np.int64, copy=False)
ts = df.timestamp.to_numpy()[keep].astype(np.int64, copy=False)
it = cols[keep]
order = np.lexsort((ts, u))
u, ts, it = u[order], ts[order], it[order]
same = u[1:] == u[:-1]
close = (ts[1:] - ts[:-1]) <= gap
mask = same & close & (it[1:] != it[:-1])
src = it[:-1][mask]
dst = it[1:][mask]
w = decay_weights(ts[1:][mask], now, half_life)
src_all.append(src)
dst_all.append(dst)
w_all.append(w)
del df
src = np.concatenate(src_all)
dst = np.concatenate(dst_all)
w = np.concatenate(w_all)
# Symmetric transitions
G = sparse.csr_matrix(
(np.concatenate([w, w]), (np.concatenate([src, dst]), np.concatenate([dst, src]))),
shape=(n_items, n_items),
dtype=np.float32,
)
G.sum_duplicates()
G.data = np.log1p(G.data)
# Keep top-k outgoing per item
G = G.tocsr()
rows, cols, data = [], [], []
for i in range(n_items):
sl = slice(G.indptr[i], G.indptr[i + 1])
if sl.start == sl.stop:
continue
idx = G.indices[sl]
val = G.data[sl]
if len(val) > topk:
keep = np.argpartition(val, -topk)[-topk:]
idx = idx[keep]
val = val[keep]
rows.append(np.full(len(idx), i, dtype=np.int32))
cols.append(idx.astype(np.int32))
data.append(val.astype(np.float32))
if not rows:
return sparse.csr_matrix((n_items, n_items), dtype=np.float32)
S = sparse.csr_matrix(
(np.concatenate(data), (np.concatenate(rows), np.concatenate(cols))),
shape=(n_items, n_items),
dtype=np.float32,
)
S = normalize(S, norm="l2", axis=1)
return S
def popularity_vector(
listen_files: list[str],
like_files: list[str],
item_table: np.ndarray,
n_items: int,
now: int,
half_life: float,
) -> np.ndarray:
pop = np.zeros(n_items, dtype=np.float64)
for name in listen_files:
df = read_parquet(
name, ["timestamp", "item_id", "is_organic", "played_ratio_pct"]
)
items = df.item_id.to_numpy()
valid = items < len(item_table)
cols = np.full(len(df), -1, dtype=np.int32)
cols[valid] = item_table[items[valid]]
keep = (
(cols >= 0)
& (df.is_organic.to_numpy() == 1)
& (df.played_ratio_pct.to_numpy() >= LISTEN_STRONG)
)
if keep.any():
w = decay_weights(df.timestamp.to_numpy()[keep], now, half_life)
np.add.at(pop, cols[keep], w)
del df
for name in like_files:
df = read_parquet(name, ["timestamp", "item_id", "is_organic"])
items = df.item_id.to_numpy()
valid = items < len(item_table)
cols = np.full(len(df), -1, dtype=np.int32)
cols[valid] = item_table[items[valid]]
keep = (cols >= 0) & (df.is_organic.to_numpy() == 1)
if keep.any():
w = 3.0 * decay_weights(df.timestamp.to_numpy()[keep], now, half_life)
np.add.at(pop, cols[keep], w)
del df
return np.log1p(pop).astype(np.float32)
def svd_factors(X: sparse.csr_matrix, n_components: int = 192) -> tuple[np.ndarray, np.ndarray]:
log(f" TruncatedSVD k={n_components}")
svd = TruncatedSVD(n_components=n_components, n_iter=8, random_state=42)
user_f = svd.fit_transform(X).astype(np.float32) # U * S
item_f = svd.components_.T.astype(np.float32)
log(f" explained variance {svd.explained_variance_ratio_.sum():.4f}")
return user_f, item_f
def user_knn_scores(
X: sparse.csr_matrix,
target_rows: np.ndarray,
k: int = 80,
batch: int = 256,
) -> np.ndarray:
"""Cosine user-kNN scores for target rows."""
Xn = normalize(X, norm="l2", axis=1).tocsr()
n_tgt = len(target_rows)
n_items = X.shape[1]
scores = np.zeros((n_tgt, n_items), dtype=np.float32)
Xt = Xn.T.tocsr()
k = min(k, X.shape[0] - 1)
for start in range(0, n_tgt, batch):
end = min(start + batch, n_tgt)
rows = target_rows[start:end]
Q = Xn[rows].toarray()
sim = Q @ Xt # (batch, n_users)
for i, r in enumerate(rows):
sim[i, r] = -np.inf
idx = np.argpartition(sim, -k, axis=1)[:, -k:]
val = np.take_along_axis(sim, idx, axis=1)
val = np.maximum(val, 0.0)
# weighted sum of neighbor histories
for i in range(end - start):
neigh = idx[i]
w = val[i]
s = w @ Xn[neigh].toarray() if False else Xn[neigh].T.dot(w)
scores[start + i] = np.asarray(s).ravel().astype(np.float32)
if start % 1024 == 0:
log(f" user-knn {end}/{n_tgt}")
return scores
def zscore_rows(S: np.ndarray) -> np.ndarray:
mean = S.mean(axis=1, keepdims=True)
std = S.std(axis=1, keepdims=True)
std = np.maximum(std, 1e-6)
return ((S - mean) / std).astype(np.float32)
def topn_from_scores(
scores: np.ndarray, eligible: np.ndarray, item_ids: np.ndarray, n: int = TOP_N
) -> np.ndarray:
s = scores.copy()
s[~eligible] = -1e30
idx = np.argpartition(-s, n, axis=1)[:, :n]
val = np.take_along_axis(s, idx, axis=1)
order = np.argsort(-val, axis=1, kind="stable")
idx = np.take_along_axis(idx, order, axis=1)
return item_ids[idx]
def ranking_frame(uids: np.ndarray, top_items: np.ndarray) -> pd.DataFrame:
n_users, n = top_items.shape
return pd.DataFrame(
{
"uid": np.repeat(uids.astype(np.int64), n),
"item_id": top_items.reshape(-1).astype(np.int64),
"rank": np.tile(np.arange(1, n + 1, dtype=np.int64), n_users),
}
)
def validation_labels(item_ids: np.ndarray, uids: np.ndarray, elig: np.ndarray) -> dict[int, dict[int, int]]:
targets = set(map(int, uids))
item_col = {int(i): c for c, i in enumerate(item_ids)}
uid_row = {int(u): r for r, u in enumerate(uids)}
listens = pd.read_parquet(DATA / "val_listens.parquet")
likes = pd.read_parquet(DATA / "val_likes.parquet")
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]] = {}
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 uid, item in strong.itertuples(index=False, name=None):
uid, item = int(uid), int(item)
if eligible(uid, item):
graded.setdefault(uid, {})[item] = 1
for uid, item in olikes.itertuples(index=False, name=None):
uid, item = int(uid), int(item)
if eligible(uid, item):
graded.setdefault(uid, {})[item] = 3
return graded
def score_top_items(top_items: np.ndarray, uids: np.ndarray, graded: dict[int, dict[int, int]]) -> float:
uid_to_row = {int(u): i for i, u in enumerate(uids)}
vals = []
for uid, gold in graded.items():
row = uid_to_row[uid]
ranked = [int(x) for x in top_items[row, :QUALITY_K]]
vals.append(ndcg_at_k(ranked, gold, QUALITY_K))
return float(np.mean(vals)) if vals else 0.0
def target_row_index(user_ids: np.ndarray, target_uids: np.ndarray) -> np.ndarray:
vmax = int(user_ids.max()) + 1
table = np.full(vmax, -1, dtype=np.int32)
table[user_ids] = np.arange(len(user_ids), dtype=np.int32)
rows = np.full(len(target_uids), -1, dtype=np.int32)
valid = target_uids < vmax
rows[valid] = table[target_uids[valid]]
if (rows < 0).any():
log(f" warning: {(rows < 0).sum()} targets missing from matrix")
return rows
def slice_or_zeros(M: np.ndarray | sparse.spmatrix, rows: np.ndarray, n_items: int) -> np.ndarray | sparse.csr_matrix:
n_tgt = len(rows)
if sparse.issparse(M):
out = sparse.csr_matrix((n_tgt, n_items), dtype=np.float32)
good = rows >= 0
if good.any():
out[np.where(good)[0]] = M[rows[good]]
return out
out = np.zeros((n_tgt, M.shape[1] if M.ndim == 2 else n_items), dtype=np.float32)
good = rows >= 0
if good.any():
out[np.where(good)[0]] = M[rows[good]]
return out
def fit_and_score(
listen_files: list[str],
like_files: list[str],
pref_path: Path,
now: int,
item_ids: np.ndarray,
item_table: np.ndarray,
target_uids: np.ndarray,
elig: np.ndarray,
knn_k: int = 200,
) -> dict[str, np.ndarray]:
n_items = len(item_ids)
log("building interaction matrices")
X, Q, user_ids, _, _ = collect_positive_matrix(
listen_files,
like_files,
now,
item_table,
n_items,
half_life=28.0,
query_half_life=14.0,
)
Q = apply_preference_state(Q, user_ids, pref_path, item_table, like_boost=7.0)
# damp extreme repeaters
X = X.tocsr()
Q = Q.tocsr()
X.data = np.log1p(X.data)
Q.data = np.log1p(Q.data)
trows = target_row_index(user_ids, target_uids)
Q_tgt = slice_or_zeros(Q, trows, n_items).tocsr()
log("popularity")
pop = popularity_vector(listen_files, like_files, item_table, n_items, now, half_life=21.0)
pop_scores = np.broadcast_to(pop, (len(target_uids), n_items)).copy()
log("BM25 item-kNN")
X_bm25 = bm25_weight(X)
S_knn = item_topk_from_gram(X_bm25, k=knn_k, batch=192)
knn_scores = Q_tgt @ S_knn
if sparse.issparse(knn_scores):
knn_scores = knn_scores.toarray()
knn_scores = knn_scores.astype(np.float32, copy=False)
log("RP3beta")
Xbin = X.copy()
Xbin.data = np.ones_like(Xbin.data)
S_rp3 = rp3beta_topk(Xbin, k=knn_k, alpha=0.5, beta=0.35, batch=192)
rp3_scores = Q_tgt @ S_rp3
if sparse.issparse(rp3_scores):
rp3_scores = rp3_scores.toarray()
rp3_scores = rp3_scores.astype(np.float32, copy=False)
log("sequential co-visitation")
S_seq = sequential_item_graph(
listen_files, item_table, n_items, now, half_life=21.0, gap=1800, topk=60
)
seq_scores = Q_tgt @ S_seq
if sparse.issparse(seq_scores):
seq_scores = seq_scores.toarray()
seq_scores = seq_scores.astype(np.float32, copy=False)
log("SVD")
X_tfidf = bm25_weight(X, k1=1.5, b=0.4)
user_f, item_f = svd_factors(X_tfidf, n_components=160)
U_tgt = slice_or_zeros(user_f, trows, n_items)
svd_scores = U_tgt @ item_f.T
log("user-kNN")
uknn_scores = user_knn_scores(X_bm25, np.where(trows >= 0, trows, 0), k=70, batch=128)
# zero out users missing from matrix
missing = trows < 0
if missing.any():
uknn_scores[missing] = 0.0
models = {
"knn": knn_scores,
"rp3": rp3_scores,
"seq": seq_scores,
"svd": svd_scores,
"uknn": uknn_scores,
"pop": pop_scores,
}
return models
def blend_scores(models: dict[str, np.ndarray], weights: dict[str, float]) -> np.ndarray:
acc = None
for name, w in weights.items():
z = zscore_rows(models[name])
acc = w * z if acc is None else acc + w * z
return acc
def write_ranking(path: Path, uids: np.ndarray, top_items: np.ndarray) -> None:
frame = ranking_frame(uids, top_items)
frame.to_parquet(path, index=False)
log(f"wrote {path} rows={len(frame)}")
def main() -> None:
t0 = time.time()
OUT.mkdir(parents=True, exist_ok=True)
item_ids, _ = load_catalog()
item_table = item_index_lookup(item_ids)
target_uids = load_targets()
n_items = len(item_ids)
val_uids, val_elig = load_eligibility(DATA / "validation_eligible_candidates.npz", n_items)
assert np.array_equal(val_uids, target_uids)
log("=== validation fit (train history only) ===")
models = fit_and_score(
listen_files=["train_listens.parquet"],
like_files=["train_likes.parquet"],
pref_path=DATA / "validation_preference_state.parquet",
now=TRAIN_CUTOFF,
item_ids=item_ids,
item_table=item_table,
target_uids=target_uids,
elig=val_elig,
)
log("building validation labels")
graded = validation_labels(item_ids, target_uids, val_elig)
log(f"scored users with relevance: {len(graded)}")
weight_grid = [
{"knn": 1.2, "rp3": 1.0, "seq": 0.7, "svd": 0.8, "uknn": 0.5, "pop": 0.15},
{"knn": 1.4, "rp3": 1.1, "seq": 0.5, "svd": 0.6, "uknn": 0.4, "pop": 0.1},
{"knn": 1.0, "rp3": 1.2, "seq": 0.8, "svd": 1.0, "uknn": 0.6, "pop": 0.2},
{"knn": 1.3, "rp3": 0.8, "seq": 1.0, "svd": 0.7, "uknn": 0.3, "pop": 0.1},
{"knn": 1.5, "rp3": 1.3, "seq": 0.4, "svd": 0.5, "uknn": 0.2, "pop": 0.05},
{"knn": 0.9, "rp3": 0.9, "seq": 0.9, "svd": 0.9, "uknn": 0.9, "pop": 0.2},
]
# also score single models
best_w = None
best_ndcg = -1.0
for name, S in models.items():
top = topn_from_scores(S, val_elig, item_ids)
nd = score_top_items(top, target_uids, graded)
log(f" single {name}: ndcg@10={nd:.6f}")
if nd > best_ndcg:
best_ndcg = nd
best_w = {name: 1.0}
for w in weight_grid:
blended = blend_scores(models, w)
top = topn_from_scores(blended, val_elig, item_ids)
nd = score_top_items(top, target_uids, graded)
log(f" blend {w}: ndcg@10={nd:.6f}")
if nd > best_ndcg:
best_ndcg = nd
best_w = w
log(f"best validation ndcg@10={best_ndcg:.6f} weights={best_w}")
blended = blend_scores(models, best_w)
val_top = topn_from_scores(blended, val_elig, item_ids)
write_ranking(OUT / "validation.parquet", target_uids, val_top)
# free validation scores before refit
del models, blended, val_top
log("=== test fit (train+val history) ===")
test_uids, test_elig = load_eligibility(DATA / "test_eligible_candidates.npz", n_items)
assert np.array_equal(test_uids, target_uids)
models = fit_and_score(
listen_files=["train_listens.parquet", "val_listens.parquet"],
like_files=["train_likes.parquet", "val_likes.parquet"],
pref_path=DATA / "test_preference_state.parquet",
now=TEST_CUTOFF,
item_ids=item_ids,
item_table=item_table,
target_uids=target_uids,
elig=test_elig,
)
blended = blend_scores(models, best_w)
test_top = topn_from_scores(blended, test_elig, item_ids)
write_ranking(OUT / "submission.parquet", target_uids, test_top)
log(f"done in {(time.time() - t0) / 60:.1f} min")
if __name__ == "__main__":
main()