Tool call · Input
/app/solution/build_ranker.py
#!/usr/bin/env python3
"""Made-for-You discovery ranker: collaborative models + z-score blend."""
from __future__ import annotations
import gc
import json
import os
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.preprocessing import normalize
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")
DATA = Path("/task/data")
OUT = Path("/app/solution")
CACHE = OUT / "cache"
OUT.mkdir(parents=True, exist_ok=True)
CACHE.mkdir(parents=True, exist_ok=True)
TRAIN_CUTOFF = 24_786_800
TEST_CUTOFF = 25_393_400
DAY = 86_400
TOP_N = 100
QUALITY_K = 10
LISTEN_STRONG = 80
def log(msg: str) -> None:
print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)
def load_catalog():
cat = pd.read_parquet(DATA / "candidate_catalog.parquet")
item_ids = cat.item_id.to_numpy(np.int64)
pop = cat.positive_listens.to_numpy(np.float64)
opop = cat.organic_positive_listens.to_numpy(np.float64)
return item_ids, pop, opop
def load_targets():
t = pd.read_parquet(DATA / "target_users.parquet")
return t.uid.to_numpy(np.int64)
def unpack_eligibility(path: Path, n_items: int) -> tuple[np.ndarray, np.ndarray]:
z = np.load(path)
uids = z["uids"].astype(np.int64)
packed = z["packed_eligible"]
elig = np.unpackbits(packed, axis=1, bitorder="little")[:, :n_items].astype(bool)
return uids, elig
def item_lookup(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
def decay_weights(ts: np.ndarray, cutoff: int, half_life_days: float) -> np.ndarray:
age = np.clip(cutoff - ts.astype(np.float64), 0, None) / DAY
return np.exp(-np.log(2.0) * age / half_life_days).astype(np.float32)
def listen_base_weight(played: np.ndarray, organic: np.ndarray) -> np.ndarray:
played = played.astype(np.float32)
organic = organic.astype(np.float32)
w = np.zeros(len(played), dtype=np.float32)
strong = played >= 80
mid = (played >= 50) & ~strong
weak = (played >= 20) & ~mid & ~strong
w[strong] = np.where(organic[strong] >= 0.5, 1.6, 0.45)
w[mid] = np.where(organic[mid] >= 0.5, 0.7, 0.18)
w[weak] = np.where(organic[weak] >= 0.5, 0.12, 0.04)
return w
def like_base_weight(organic: np.ndarray) -> np.ndarray:
return np.where(organic.astype(np.float32) >= 0.5, 8.0, 3.0).astype(np.float32)
def load_split_events(split: str) -> dict[str, pd.DataFrame]:
cols_l = ["uid", "timestamp", "item_id", "is_organic", "played_ratio_pct"]
cols_e = ["uid", "timestamp", "item_id", "is_organic"]
return {
"listens": pd.read_parquet(DATA / f"{split}_listens.parquet", columns=cols_l),
"likes": pd.read_parquet(DATA / f"{split}_likes.parquet", columns=cols_e),
}
def filter_catalog(df: pd.DataFrame, inv: np.ndarray) -> pd.DataFrame:
iid = df.item_id.to_numpy(np.int64)
ok = (iid >= 0) & (iid < len(inv))
keep = np.zeros(len(df), dtype=bool)
keep[ok] = inv[iid[ok]] >= 0
if keep.all():
return df
return df.loc[keep]
def events_to_triples(listens, likes, inv, cutoff, half_life):
parts_u, parts_i, parts_w, parts_t = [], [], [], []
if len(listens):
listens = filter_catalog(listens, inv)
u = listens.uid.to_numpy(np.int64)
col = inv[listens.item_id.to_numpy(np.int64)].astype(np.int32)
ts = listens.timestamp.to_numpy(np.int64)
w = listen_base_weight(
listens.played_ratio_pct.to_numpy(np.float32),
listens.is_organic.to_numpy(np.float32),
)
w *= decay_weights(ts, cutoff, half_life)
parts_u.append(u)
parts_i.append(col)
parts_w.append(w)
parts_t.append(ts)
if len(likes):
likes = filter_catalog(likes, inv)
u = likes.uid.to_numpy(np.int64)
col = inv[likes.item_id.to_numpy(np.int64)].astype(np.int32)
ts = likes.timestamp.to_numpy(np.int64)
w = like_base_weight(likes.is_organic.to_numpy(np.float32))
w *= decay_weights(ts, cutoff, half_life)
parts_u.append(u)
parts_i.append(col)
parts_w.append(w)
parts_t.append(ts)
return (
np.concatenate(parts_u),
np.concatenate(parts_i),
np.concatenate(parts_w),
np.concatenate(parts_t),
)
def uid_lookup(all_uids: np.ndarray) -> np.ndarray:
inv = np.full(int(all_uids.max()) + 1, -1, dtype=np.int32)
inv[all_uids] = np.arange(len(all_uids), dtype=np.int32)
return inv
def aggregate_ui(uid, col, wgt, uinv, n_users, n_items) -> sparse.csr_matrix:
rows = uinv[uid]
mat = sparse.csr_matrix(
(wgt.astype(np.float32), (rows, col)),
shape=(n_users, n_items),
dtype=np.float32,
)
mat.sum_duplicates()
mat.data = np.log1p(np.maximum(mat.data, 0.0)).astype(np.float32)
mat.eliminate_zeros()
return mat
def binary_from_weighted(X: sparse.csr_matrix) -> sparse.csr_matrix:
B = X.copy()
B.data = np.ones_like(B.data, dtype=np.float32)
return B
def bm25_weight(X: sparse.csr_matrix, k1: float = 1.2, b: float = 0.75) -> sparse.csr_matrix:
X = X.tocsr().astype(np.float32)
n_users = X.shape[0]
df = np.asarray(X.getnnz(axis=0), dtype=np.float64)
idf = np.log((n_users - df + 0.5) / (df + 0.5) + 1.0).astype(np.float32)
dl = np.asarray(X.sum(axis=1)).ravel().astype(np.float32)
avgdl = float(dl.mean()) if n_users else 1.0
X = X.tocoo()
tf = X.data
row_len = dl[X.row]
denom = tf + k1 * (1.0 - b + b * row_len / max(avgdl, 1e-6))
X.data = idf[X.col] * (tf * (k1 + 1.0)) / np.maximum(denom, 1e-6)
return sparse.csr_matrix((X.data.astype(np.float32), (X.row, X.col)), shape=X.shape)
def topk_item_sim(item_factors: sparse.csr_matrix, k: int, block: int = 400) -> sparse.csr_matrix:
"""item_factors is items x dim; similarity = left @ left.T, keep top-k per row."""
n_items = item_factors.shape[0]
left = item_factors.tocsr()
right_t = left # items x dim, we want left[block] @ left.T
rows, cols, data = [], [], []
kk = min(k, n_items - 1)
for start in range(0, n_items, block):
end = min(start + block, n_items)
chunk = left[start:end].dot(left.T)
if sparse.issparse(chunk):
chunk = chunk.toarray()
chunk = np.asarray(chunk, dtype=np.float32)
for i, gi in enumerate(range(start, end)):
chunk[i, gi] = 0.0
idx = np.argpartition(chunk, -kk, axis=1)[:, -kk:]
vals = np.take_along_axis(chunk, idx, axis=1)
mask = vals > 0
ridx = np.repeat(np.arange(start, end), kk).reshape(end - start, kk)
rows.append(ridx[mask])
cols.append(idx[mask])
data.append(vals[mask])
return sparse.csr_matrix(
(np.concatenate(data), (np.concatenate(rows), np.concatenate(cols))),
shape=(n_items, n_items),
dtype=np.float32,
)
def topk_product(left: sparse.csr_matrix, right: sparse.csr_matrix, k: int, block: int = 300) -> sparse.csr_matrix:
"""left: items x dim, right: dim x items, sim = left @ right, top-k."""
n_items = left.shape[0]
left = left.tocsr()
right = right.tocsr()
rows, cols, data = [], [], []
kk = min(k, n_items - 1)
for start in range(0, n_items, block):
end = min(start + block, n_items)
chunk = left[start:end].dot(right)
if sparse.issparse(chunk):
chunk = chunk.toarray()
chunk = np.asarray(chunk, dtype=np.float32)
for i, gi in enumerate(range(start, end)):
chunk[i, gi] = 0.0
idx = np.argpartition(chunk, -kk, axis=1)[:, -kk:]
vals = np.take_along_axis(chunk, idx, axis=1)
mask = vals > 0
ridx = np.repeat(np.arange(start, end), kk).reshape(end - start, kk)
rows.append(ridx[mask])
cols.append(idx[mask])
data.append(vals[mask])
return sparse.csr_matrix(
(np.concatenate(data), (np.concatenate(rows), np.concatenate(cols))),
shape=(n_items, n_items),
dtype=np.float32,
)
def score_itemknn(profile: sparse.csr_matrix, sim: sparse.csr_matrix) -> np.ndarray:
out = profile.dot(sim)
if sparse.issparse(out):
out = out.toarray()
return np.asarray(out, dtype=np.float32)
def userknn_scores(X_all: sparse.csr_matrix, target_rows: np.ndarray, k: int = 80) -> np.ndarray:
Xn = normalize(X_all, norm="l2", axis=1)
Xt = Xn[target_rows]
sim = Xt.dot(Xn.T)
if sparse.issparse(sim):
sim = sim.toarray()
sim = np.asarray(sim, dtype=np.float32)
n_t, n_u = sim.shape
sim[np.arange(n_t), target_rows] = 0.0
kk = min(k, n_u - 1)
idx = np.argpartition(sim, -kk, axis=1)[:, -kk:]
vals = np.clip(np.take_along_axis(sim, idx, axis=1), 0, None)
S = sparse.csr_matrix(
(vals.ravel(), (np.repeat(np.arange(n_t), kk), idx.ravel())),
shape=(n_t, n_u),
dtype=np.float32,
)
S.data[S.data <= 0] = 0
S.eliminate_zeros()
scores = S.dot(X_all)
if sparse.issparse(scores):
scores = scores.toarray()
return np.asarray(scores, dtype=np.float32)
def ials(R: sparse.csr_matrix, factors=48, n_iters=5, reg=5.0, alpha=28.0, seed=7):
R = R.tocsr()
n_users, n_items = R.shape
rng = np.random.default_rng(seed)
X = rng.normal(0, 0.1, size=(n_users, factors)).astype(np.float64)
Y = rng.normal(0, 0.1, size=(n_items, factors)).astype(np.float64)
Rt = R.T.tocsr()
eye = np.eye(factors, dtype=np.float64)
def update(factors_out, other, mat, n_rows):
YtY = other.T @ other
indptr, indices, data = mat.indptr, mat.indices, mat.data
for u in range(n_rows):
s, e = indptr[u], indptr[u + 1]
if s == e:
factors_out[u] = 0
continue
idx = indices[s:e]
r = data[s:e].astype(np.float64, copy=False)
conf = 1.0 + alpha * r
Yu = other[idx]
A = YtY + Yu.T @ ((conf - 1.0)[:, None] * Yu) + reg * eye
b = (conf[:, None] * Yu).sum(axis=0)
factors_out[u] = np.linalg.solve(A, b)
for it in range(n_iters):
t1 = time.time()
update(X, Y, R, n_users)
update(Y, X, Rt, n_items)
log(f" iALS iter {it + 1}/{n_iters} {time.time() - t1:.1f}s")
return X.astype(np.float32), Y.astype(np.float32)
def session_item_sim(uid, col, ts, wgt, n_items, window=2400, k=120) -> sparse.csr_matrix:
order = np.lexsort((ts, uid))
uid, col, ts, wgt = uid[order], col[order], ts[order], wgt[order]
C = np.zeros((n_items, n_items), dtype=np.float32)
n = len(uid)
start = 0
while start < n:
end = start + 1
while end < n and uid[end] == uid[start]:
end += 1
items = col[start:end]
times = ts[start:end]
weights = wgt[start:end]
j = 0
m = end - start
for i in range(m):
while times[i] - times[j] > window:
j += 1
if i > j:
seg = items[j:i]
ww = np.sqrt(np.maximum(weights[i], 1e-6) * np.maximum(weights[j:i], 1e-6))
C[items[i], seg] += ww
C[seg, items[i]] += ww
start = end
np.fill_diagonal(C, 0.0)
norms = np.sqrt(np.maximum((C * C).sum(axis=1), 1e-12))
rows, cols, data = [], [], []
kk = min(k, n_items - 1)
for i in range(0, n_items, 250):
end = min(i + 250, n_items)
block = C[i:end]
if block.max() <= 0:
continue
idx = np.argpartition(block, -kk, axis=1)[:, -kk:]
vals = np.take_along_axis(block, idx, axis=1)
nn = norms[i:end][:, None] * norms[idx] + 1e-12
vals = vals / nn
mask = vals > 0
ridx = np.repeat(np.arange(i, end), kk).reshape(end - i, kk)
rows.append(ridx[mask])
cols.append(idx[mask])
data.append(vals[mask].astype(np.float32))
del C
gc.collect()
if not rows:
return sparse.csr_matrix((n_items, n_items), dtype=np.float32)
return sparse.csr_matrix(
(np.concatenate(data), (np.concatenate(rows), np.concatenate(cols))),
shape=(n_items, n_items),
dtype=np.float32,
)
def pop_window_scores(col, wgt, ts, cutoff, n_items, window_days):
keep = ts >= (cutoff - window_days * DAY)
counts = np.zeros(n_items, dtype=np.float64)
if keep.any():
np.add.at(counts, col[keep], wgt[keep])
return np.log1p(counts).astype(np.float32)
def zscore_eligible(scores: np.ndarray, elig: np.ndarray, chunk: int = 256) -> np.ndarray:
n = scores.shape[0]
out = np.zeros_like(scores, dtype=np.float32)
for i in range(0, n, chunk):
sl = slice(i, min(i + chunk, n))
block = scores[sl].astype(np.float64, copy=True)
m = elig[sl]
block[~m] = np.nan
mean = np.nanmean(block, axis=1, keepdims=True)
std = np.nanstd(block, axis=1, keepdims=True)
std = np.where(std < 1e-8, 1.0, std)
z = (block - mean) / std
z[~m] = 0.0
np.nan_to_num(z, copy=False, nan=0.0, posinf=0.0, neginf=0.0)
out[sl] = z.astype(np.float32)
return out
def zscore_global_item(vec: np.ndarray, elig: np.ndarray, chunk: int = 256) -> np.ndarray:
n, m = elig.shape
out = np.zeros((n, m), dtype=np.float32)
v = vec.astype(np.float64)
for i in range(0, n, chunk):
sl = slice(i, min(i + chunk, n))
block = np.broadcast_to(v, (sl.stop - sl.start, m)).copy()
mask = elig[sl]
block[~mask] = np.nan
mean = np.nanmean(block, axis=1, keepdims=True)
std = np.nanstd(block, axis=1, keepdims=True)
std = np.where(std < 1e-8, 1.0, std)
z = (block - mean) / std
z[~mask] = 0.0
np.nan_to_num(z, copy=False, nan=0.0)
out[sl] = z.astype(np.float32)
return out
def rank_topn(scores: np.ndarray, elig: np.ndarray, n: int = TOP_N) -> np.ndarray:
s = np.where(elig, scores, np.float32(-1e18))
idx = np.argpartition(-s, n - 1, axis=1)[:, :n]
row = np.arange(s.shape[0])[:, None]
order = np.argsort(-s[row, idx], axis=1)
return np.take_along_axis(idx, order, axis=1)
def ranking_frame(uids, item_ids, top_idx) -> pd.DataFrame:
n_users, n = top_idx.shape
return pd.DataFrame(
{
"uid": np.repeat(uids, n).astype(np.int64),
"item_id": item_ids[top_idx.reshape(-1)].astype(np.int64),
"rank": np.tile(np.arange(1, n + 1, dtype=np.int64), n_users),
}
)
def validation_labels(item_ids, target_uids, elig):
uid_row = {int(u): i for i, u in enumerate(target_uids)}
inv = item_lookup(item_ids)
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"])
targets = set(map(int, target_uids))
graded = [dict() for _ in range(len(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()
def add(df, grade):
for u, it in df.itertuples(index=False, name=None):
r = uid_row.get(int(u))
if r is None:
continue
it = int(it)
if it >= len(inv):
continue
c = int(inv[it])
if c < 0 or not elig[r, c]:
continue
if grade >= graded[r].get(c, 0):
graded[r][c] = grade
add(strong, 1)
add(olikes, 3)
return graded
def ndcg_from_top(top_idx, graded_by_row, k=QUALITY_K) -> float:
discounts = 1.0 / np.log2(np.arange(2, k + 2))
vals = []
for row, gold in enumerate(graded_by_row):
if not gold:
continue
ranked = top_idx[row, :k]
gains = np.array([gold.get(int(c), 0) for c in ranked], dtype=np.float64)
dcg = float(np.sum((np.power(2.0, gains) - 1.0) * discounts[: len(gains)]))
ideal = np.array(sorted(gold.values(), reverse=True)[:k], dtype=np.float64)
idcg = float(np.sum((np.power(2.0, ideal) - 1.0) * discounts[: len(ideal)]))
vals.append(dcg / idcg if idcg else 0.0)
return float(np.mean(vals)) if vals else 0.0
def ndcg_from_scores(scores, elig, graded_by_row, k=QUALITY_K) -> float:
s = np.where(elig, scores, np.float32(-1e18))
idx = np.argpartition(-s, k - 1, axis=1)[:, :k]
row = np.arange(s.shape[0])[:, None]
order = np.argsort(-s[row, idx], axis=1)
top = np.take_along_axis(idx, order, axis=1)
return ndcg_from_top(top, graded_by_row, k)
def dislike_profile(pref_path, uinv, inv, n_users, n_items, target_rows):
pref = pd.read_parquet(pref_path)
d = pref[pref.state == "disliked"]
if d.empty:
return sparse.csr_matrix((len(target_rows), n_items), dtype=np.float32)
uid = d.uid.to_numpy(np.int64)
iid = d.item_id.to_numpy(np.int64)
ok = (uid >= 0) & (uid < len(uinv)) & (iid >= 0) & (iid < len(inv))
uid, iid = uid[ok], iid[ok]
rows = uinv[uid]
cols = inv[iid]
keep = (rows >= 0) & (cols >= 0)
M = sparse.csr_matrix(
(np.ones(keep.sum(), dtype=np.float32), (rows[keep], cols[keep])),
shape=(n_users, n_items),
dtype=np.float32,
)
return M[target_rows]
def mmap_save(name: str, arr: np.ndarray) -> np.ndarray:
path = CACHE / f"{name}.npy"
np.save(path, np.ascontiguousarray(arr, dtype=np.float32))
return np.load(path, mmap_mode="r")
def collect_events(use_val: bool):
tr = load_split_events("train")
if not use_val:
return tr["listens"], tr["likes"]
va = load_split_events("val")
listens = pd.concat([tr["listens"], va["listens"]], ignore_index=True)
likes = pd.concat([tr["likes"], va["likes"]], ignore_index=True)
return listens, likes
def store_channel(name, scores, elig, zchan):
z = zscore_eligible(scores, elig)
del scores
zchan[name] = mmap_save(name, z)
del z
gc.collect()
log(f" stored {name}")
def build_zchannels(
X_all,
P_mix,
P_rec,
target_rows,
n_items,
pop_log,
pop_30,
pop_7,
dislike,
uid_e,
col_e,
wgt_e,
ts_e,
elig,
):
zchan = {}
B = binary_from_weighted(X_all)
log("popularity channels")
zchan["pop_log"] = mmap_save("pop_log", zscore_global_item(pop_log, elig))
zchan["pop_30"] = mmap_save("pop_30", zscore_global_item(pop_30, elig))
zchan["pop_7"] = mmap_save("pop_7", zscore_global_item(pop_7, elig))
gc.collect()
log("BM25 item-KNN")
Xb = bm25_weight(B, k1=1.2, b=0.75)
sim_bm25 = topk_item_sim(Xb.T.tocsr(), k=220, block=350)
del Xb
gc.collect()
store_channel("knn_bm25", score_itemknn(P_mix, sim_bm25), elig, zchan)
store_channel("knn_bm25_recent", score_itemknn(P_rec, sim_bm25), elig, zchan)
if dislike is not None and dislike.nnz:
store_channel("pen_bm25", score_itemknn(dislike, sim_bm25), elig, zchan)
del sim_bm25
gc.collect()
log("cosine item-KNN")
Xn = normalize(X_all.astype(np.float32), norm="l2", axis=0)
sim_cos = topk_item_sim(Xn.T.tocsr(), k=200, block=350)
del Xn
gc.collect()
store_channel("knn_cos", score_itemknn(P_mix, sim_cos), elig, zchan)
del sim_cos
gc.collect()
log("RP3beta")
Piu = normalize(B.T.tocsr().astype(np.float32), norm="l1", axis=1) # items x users
Pui = normalize(B.astype(np.float32), norm="l1", axis=1) # users x items
sim_rp = topk_product(Piu, Pui, k=250, block=300)
di = np.asarray(B.sum(axis=0)).ravel().astype(np.float32)
di[di <= 0] = 1.0
del Piu, Pui
gc.collect()
for beta, name in ((0.45, "rp3_b045"), (0.7, "rp3_b07")):
scale = np.power(di, -beta).astype(np.float32)
sim_b = sim_rp.copy()
sim_b.data *= scale[sim_b.indices]
store_channel(name, score_itemknn(P_mix, sim_b), elig, zchan)
del sim_b
gc.collect()
del sim_rp
gc.collect()
log("PureSVD")
dfreq = np.asarray(B.getnnz(axis=0), dtype=np.float32)
idf = np.log((B.shape[0] + 1.0) / (dfreq + 1.0)).astype(np.float32)
Xt = normalize(B.multiply(idf[np.newaxis, :]).tocsr(), norm="l2", axis=1)
svd = TruncatedSVD(n_components=128, n_iter=8, random_state=7)
U = svd.fit_transform(Xt)
V = svd.components_.astype(np.float32)
scores = U[target_rows].astype(np.float32) @ V
del Xt, U, V, svd
store_channel("svd", scores, elig, zchan)
gc.collect()
log("User-KNN")
store_channel("uknn", userknn_scores(X_all, target_rows, k=90), elig, zchan)
gc.collect()
log("iALS")
try:
Xu, Yi = ials(X_all, factors=48, n_iters=5, reg=5.0, alpha=28.0)
scores = Xu[target_rows] @ Yi.T
del Xu, Yi
store_channel("ials", scores, elig, zchan)
except Exception as exc:
log(f"iALS failed: {exc}")
gc.collect()
log("session co-occurrence")
try:
sim_ses = session_item_sim(uid_e, col_e, ts_e, wgt_e, n_items, window=2400, k=120)
store_channel("session", score_itemknn(P_rec, sim_ses), elig, zchan)
del sim_ses
except Exception as exc:
log(f"session failed: {exc}")
gc.collect()
return zchan
def blend_search(zchan, elig, graded):
names = list(zchan.keys())
prior = {
"knn_bm25": 1.2,
"knn_cos": 0.7,
"knn_bm25_recent": 0.85,
"rp3_b045": 0.9,
"rp3_b07": 0.65,
"svd": 0.85,
"uknn": 0.45,
"ials": 1.0,
"session": 0.55,
"pop_log": 0.12,
"pop_30": 0.22,
"pop_7": 0.18,
"pen_bm25": -0.22,
}
w = {n: float(prior.get(n, 0.4 if not n.startswith("pen") else -0.2)) for n in names}
def assemble(weights):
acc = np.zeros(elig.shape, dtype=np.float32)
for n, wt in weights.items():
if abs(wt) < 1e-8:
continue
acc += np.float32(wt) * np.asarray(zchan[n])
return acc
acc = assemble(w)
best = ndcg_from_scores(acc, elig, graded)
log(f"blend start ndcg={best:.6f}")
scales = [0.0, 0.2, 0.4, 0.7, 1.0, 1.4, 1.9]
for rnd in range(3):
improved = False
for n in names:
base = w[n]
sign = -1.0 if prior.get(n, 0) < 0 or n.startswith("pen") else 1.0
local_best, local_val = best, base
for s in scales:
new = sign * s
if abs(new - base) < 1e-12:
continue
acc += np.float32(new - base) * np.asarray(zchan[n])
sc = ndcg_from_scores(acc, elig, graded)
acc += np.float32(base - new) * np.asarray(zchan[n])
if sc > local_best + 1e-6:
local_best, local_val = sc, new
if local_val != base:
acc += np.float32(local_val - base) * np.asarray(zchan[n])
w[n] = local_val
best = local_best
improved = True
log(f" r{rnd} {n} -> {local_val:.2f} ndcg={best:.6f}")
if not improved:
break
for n in names:
base = w[n]
for delta in (-0.25, -0.12, 0.12, 0.25, 0.4):
new = base + delta
acc += np.float32(delta) * np.asarray(zchan[n])
sc = ndcg_from_scores(acc, elig, graded)
if sc > best + 1e-6:
w[n] = new
best = sc
base = new
log(f" fine {n} -> {new:.2f} ndcg={best:.6f}")
else:
acc += np.float32(-delta) * np.asarray(zchan[n])
log(f"blend weights {w} ndcg={best:.6f}")
return w, best, acc
def apply_blend(zchan, weights, elig):
acc = np.zeros(elig.shape, dtype=np.float32)
for n, wt in weights.items():
if n not in zchan or abs(wt) < 1e-8:
continue
acc += np.float32(wt) * np.asarray(zchan[n])
return rank_topn(acc, elig, TOP_N)
def run_stage(stage, item_ids, pop, opop, target_uids, blend_weights):
log(f"===== STAGE {stage} =====")
use_val = stage == "test"
cutoff = TEST_CUTOFF if use_val else TRAIN_CUTOFF
elig_path = DATA / ("test_eligible_candidates.npz" if use_val else "validation_eligible_candidates.npz")
pref_path = DATA / ("test_preference_state.parquet" if use_val else "validation_preference_state.parquet")
uids, elig = unpack_eligibility(elig_path, len(item_ids))
assert np.array_equal(uids, target_uids)
listens, likes = collect_events(use_val)
inv = item_lookup(item_ids)
log(f"events listens={len(listens):,} likes={len(likes):,}")
uid_e, col_e, wgt_e, ts_e = events_to_triples(listens, likes, inv, cutoff, 50.0)
del listens, likes
gc.collect()
log(f"catalog events {len(uid_e):,}")
all_uids = np.unique(np.concatenate([uid_e, target_uids]))
uinv = uid_lookup(all_uids)
n_users = len(all_uids)
n_items = len(item_ids)
target_rows = uinv[target_uids]
X_all = aggregate_ui(uid_e, col_e, wgt_e, uinv, n_users, n_items)
log(f"X_all shape={X_all.shape} nnz={X_all.nnz}")
recent_cut = cutoff - 21 * DAY
m_rec = ts_e >= recent_cut
if m_rec.any():
ratio = decay_weights(ts_e[m_rec], cutoff, 10.0) / np.maximum(
decay_weights(ts_e[m_rec], cutoff, 50.0), 1e-6
)
X_recent = aggregate_ui(
uid_e[m_rec], col_e[m_rec], (wgt_e[m_rec] * ratio).astype(np.float32), uinv, n_users, n_items
)
else:
X_recent = X_all.copy()
P_mix = X_all[target_rows] + X_recent[target_rows] * 1.35
P_rec = X_recent[target_rows]
pop_log = np.log1p(opop).astype(np.float32)
pop_30 = pop_window_scores(col_e, wgt_e, ts_e, cutoff, n_items, 30.0)
pop_7 = pop_window_scores(col_e, wgt_e, ts_e, cutoff, n_items, 7.0)
dislike = dislike_profile(pref_path, uinv, inv, n_users, n_items, target_rows)
log(f"dislike nnz={dislike.nnz}")
zchan = build_zchannels(
X_all, P_mix, P_rec, target_rows, n_items, pop_log, pop_30, pop_7, dislike, uid_e, col_e, wgt_e, ts_e, elig
)
log(f"channels: {list(zchan)}")
ndcg = None
if stage == "val":
graded = validation_labels(item_ids, target_uids, elig)
for n, s in zchan.items():
sc = ndcg_from_scores(np.asarray(s), elig, graded)
log(f"channel {n:16s} ndcg@10={sc:.6f}")
blend_weights, ndcg, acc = blend_search(zchan, elig, graded)
(OUT / "blend_weights.json").write_text(json.dumps({"ndcg": ndcg, "weights": blend_weights}, indent=2))
top = rank_topn(acc, elig, TOP_N)
ndcg = ndcg_from_top(top, graded)
log(f"VAL NDCG@10={ndcg:.6f}")
ranking_frame(target_uids, item_ids, top).to_parquet(OUT / "validation.parquet", index=False)
del acc
else:
if blend_weights is None:
blend_weights = json.loads((OUT / "blend_weights.json").read_text())["weights"]
top = apply_blend(zchan, blend_weights, elig)
ranking_frame(target_uids, item_ids, top).to_parquet(OUT / "submission.parquet", index=False)
log("wrote submission")
del zchan
gc.collect()
return blend_weights, ndcg
def main():
t0 = time.time()
item_ids, pop, opop = load_catalog()
target_uids = load_targets()
log(f"users={len(target_uids)} items={len(item_ids)}")
weights, ndcg = run_stage("val", item_ids, pop, opop, target_uids, None)
run_stage("test", item_ids, pop, opop, target_uids, weights)
log(f"done in {(time.time() - t0) / 60:.1f} min val_ndcg={ndcg}")
if __name__ == "__main__":
main()