Agent trace

ls -la /task/ /task/data/ && cat /task/requirements.txt && nproc && free -g

cat /task/ranking_evaluation.py; echo ===; cat /task/evaluate_public.py

mkdir -p solution && python -c "
import pandas as pd, numpy as np, pyarrow.parquet as pq
for f in ['candidate_catalog','target_users','validation_preference_state','test_preference_state']:
    d=pd.read_parquet('/task/data/%s.parquet'%f); print(f,d.shape, list(d.columns)); print(d.head(3))
print(pq.ParquetFile('/task/data/train_listens.parquet').metadata)
d=pd.read_parquet('/task/data/val_listens.parquet'); print(d.shape); print(d.head()); print(d.timestamp.min(),d.timestamp.max())
z=np.load('/task/data/validation_eligible_candidates.npz'); print({k:z[k].shape for k in z})
"

cd /app/solution && python -c "
import pandas as pd, numpy as np, time
t=time.time()
d=pd.read_parquet('/task/data/train_listens.parquet',columns=['uid','item_id','played_ratio_pct','is_organic'])
print('load',time.time()-t, d.dtypes.to_dict())
print('users',d.uid.nunique(),'items',d.item_id.nunique())
tg=set(pd.read_parquet('/task/data/target_users.parquet').uid)
print('target rows frac', d.uid.isin(tg).mean())
z=np.load('/task/data/validation_eligible_candidates.npz')
items=z['item_ids']; uids=z['uids']; pk=z['packed_eligible']
m=np.unpackbits(pk,axis=1,count=len(items),bitorder='little')
print('elig per user mean',m.sum(1).mean(), m.sum(1).min(), m.sum(1).max())
# check overlap with history
sub=d[(d.uid==uids[0])&(d.played_ratio_pct>=50)]
hist=set(sub.item_id)
elig=set(items[m[0]==1])
print('user0 hist',len(hist),'elig',len(elig),'overlap',len(hist&elig), 'hist in catalog', len(hist&set(items)))
" 2>&1 | tail -20

python -c "
import pandas as pd, numpy as np
z=np.load('/task/data/validation_eligible_candidates.npz')
items=z['item_ids']; uids=z['uids']
m=np.unpackbits(z['packed_eligible'],axis=1,count=len(items),bitorder='little').astype(bool)
ps=pd.read_parquet('/task/data/validation_preference_state.parquet')
icol={int(v):i for i,v in enumerate(items)}
urow={int(v):i for i,v in enumerate(uids)}
ps['c']=ps.item_id.map(icol); ps['r']=ps.uid.map(urow)
p=ps.dropna(subset=['c','r']); p=p.astype({'c':int,'r':int})
for st in ['liked','disliked']:
    q=p[p.state==st]; print(st, len(q), 'eligible frac', m[q.r.values,q.c.values].mean())
d=pd.read_parquet('/task/data/train_listens.parquet',columns=['uid','timestamp'])
print('train ts', d.timestamp.min(), d.timestamp.max())
print('A nnz est', pd.read_parquet('/task/data/train_listens.parquet',columns=['uid','item_id','played_ratio_pct']).query('played_ratio_pct>=50')[['uid','item_id']].drop_duplicates().shape)
"

/app/solution/rank.py

#!/usr/bin/env python3
"""Discovery ranker for the "Made for You" shelf (Yambda organic NDCG@10).

Signals are computed per stage (validation history = train_*, test history =
train_* + val_*) and blended with weights hill-climbed on the public
validation window.
"""
from __future__ import annotations

import json
import sys
import time
from pathlib import Path

import numpy as np
import pandas as pd
import scipy.sparse as sp
from sklearn.utils.extmath import randomized_svd

DATA = Path("/task/data")
OUT = Path("/app/solution")
T0 = time.time()
DAY = 86400


def log(*a):
    print(f"[{time.time()-T0:7.1f}s]", *a, flush=True)


# ---------------------------------------------------------------- data loading
def load_events(split: str):
    ls = pd.read_parquet(
        DATA / f"{split}_listens.parquet",
        columns=["uid", "item_id", "timestamp", "is_organic", "played_ratio_pct"],
    )
    lk = pd.read_parquet(
        DATA / f"{split}_likes.parquet", columns=["uid", "item_id", "timestamp", "is_organic"]
    )
    return ls, lk


LS_TR, LK_TR = load_events("train")
LS_VA, LK_VA = load_events("val")
CATALOG = pd.read_parquet(DATA / "candidate_catalog.parquet")
TARGETS = pd.read_parquet(DATA / "target_users.parquet")

# global user index over every user seen anywhere (CF neighbours need all users)
ALL_UIDS = np.union1d(LS_TR.uid.unique(), LS_VA.uid.unique()).astype(np.int64)
UROW = pd.Series(np.arange(len(ALL_UIDS)), index=ALL_UIDS)
# global item index: catalog items first, then the rest of the long tail
tail = np.setdiff1d(
    np.union1d(LS_TR.item_id.unique(), LS_VA.item_id.unique()).astype(np.int64),
    CATALOG.item_id.values.astype(np.int64),
)
ITEMS = np.concatenate([CATALOG.item_id.values.astype(np.int64), tail])
ICOL = pd.Series(np.arange(len(ITEMS)), index=ITEMS)
NCAT = len(CATALOG)
log(f"users={len(ALL_UIDS)} items={len(ITEMS)} catalog={NCAT}")


def build_history(stage: str):
    """Return (listens, likes) frames usable as observed history for a stage."""
    if stage == "val":
        return LS_TR, LK_TR
    return (
        pd.concat([LS_TR, LS_VA], ignore_index=True),
        pd.concat([LK_TR, LK_VA], ignore_index=True),
    )


def coo(rows, cols, vals, shape):
    m = sp.coo_matrix((vals, (rows, cols)), shape=shape, dtype=np.float32).tocsr()
    m.sum_duplicates()
    return m


# ---------------------------------------------------------------- signals
def build_signals(stage: str, target_uids: np.ndarray, elig_mask: np.ndarray):
    ls, lk = build_history(stage)
    cutoff = int(ls.timestamp.max())
    nU, nI = len(ALL_UIDS), len(ITEMS)

    pos = ls.played_ratio_pct.values >= 50
    strong = ls.played_ratio_pct.values >= 80
    lr = UROW.reindex(ls.uid.values).values
    lc = ICOL.reindex(ls.item_id.values).values
    kr = UROW.reindex(lk.uid.values).values
    kc = ICOL.reindex(lk.item_id.values).values
    lorg = ls.is_organic.values == 1

    # ---- user x item affinity matrices
    A_pos = coo(lr[pos], lc[pos], np.ones(pos.sum(), np.float32), (nU, nI))
    A_pos.data = np.log1p(A_pos.data).astype(np.float32)
    A_str = coo(lr[strong & lorg], lc[strong & lorg],
                np.ones(int((strong & lorg).sum()), np.float32), (nU, nI))
    A_str.data = np.log1p(A_str.data).astype(np.float32)
    A_like = coo(kr, kc, np.ones(len(kr), np.float32), (nU, nI))
    A_like.data = np.minimum(A_like.data, 1.0)
    A = (A_pos + 0.7 * A_str + 2.0 * A_like).tocsr()
    A.data = A.data.astype(np.float32)
    log(f"{stage}: A nnz={A.nnz}")

    # column (item) popularity normalisation -> cosine-ish item space
    cn = np.sqrt(np.asarray(A.multiply(A).sum(axis=0)).ravel()) + 1e-6
    C = A @ sp.diags((1.0 / cn).astype(np.float32))
    C = C.tocsr()
    rn = np.sqrt(np.asarray(C.multiply(C).sum(axis=1)).ravel()) + 1e-6
    Cn = sp.diags((1.0 / rn).astype(np.float32)) @ C  # row-normalised

    trow = UROW.reindex(target_uids).values
    nT = len(trow)
    Cn_t = Cn[trow]
    C_cat = C[:, :NCAT].tocsr()

    signals: dict[str, np.ndarray] = {}

    # ---- popularity signals (broadcast over users)
    def pop(mask, weight_rows, since=None):
        sel = mask
        if since is not None:
            sel = sel & (weight_rows >= since)
        cols = lc[sel]
        cols = cols[cols < NCAT]
        v = np.bincount(cols, minlength=NCAT).astype(np.float32)
        return np.log1p(v)

    ts = ls.timestamp.values
    signals["pop"] = np.tile(pop(pos & lorg, ts), (nT, 1))
    signals["pop7"] = np.tile(pop(strong & lorg, ts, cutoff - 7 * DAY), (nT, 1))
    signals["pop2"] = np.tile(pop(strong & lorg, ts, cutoff - 2 * DAY), (nT, 1))
    kts = lk.timestamp.values
    kcols = kc[(lk.is_organic.values == 1) & (kts >= cutoff - 14 * DAY)]
    kcols = kcols[kcols < NCAT]
    signals["poplike"] = np.tile(
        np.log1p(np.bincount(kcols, minlength=NCAT).astype(np.float32)), (nT, 1)
    )
    log(f"{stage}: popularity done")

    # ---- user-kNN CF
    K = 250
    G = np.empty((nT, nU), dtype=np.float32)
    step = 1500
    CnT = Cn.T.tocsr()
    for s in range(0, nT, step):
        G[s:s + step] = (Cn_t[s:s + step] @ CnT).toarray()
    G[np.arange(nT), trow] = 0.0
    thr = np.partition(G, nU - K, axis=1)[:, nU - K][:, None]
    G[G < thr] = 0.0
    Gs = sp.csr_matrix(G ** 3)
    del G
    knn = np.empty((nT, NCAT), dtype=np.float32)
    for s in range(0, nT, step):
        knn[s:s + step] = (Gs[s:s + step] @ C_cat).toarray()
    signals["knn"] = knn
    log(f"{stage}: knn done")

    # ---- latent factor signal (truncated SVD of the affinity matrix)
    U, S, Vt = randomized_svd(Cn, n_components=192, n_iter=4, random_state=0)
    svd = (U[trow] * S) @ Vt[:, :NCAT]
    signals["svd"] = svd.astype(np.float32)
    log(f"{stage}: svd done")

    # ---- item-item on catalog: pure co-occurrence from the user's own history
    #      (asymmetric, popularity-damped target side)
    Cpow = C.copy()
    Cpow.data = Cpow.data ** 0.5
    Ct = Cpow[trow]
    ii = np.empty((nT, NCAT), dtype=np.float32)
    CatT = Cpow[:, :NCAT].tocsr()
    for s in range(0, nT, step):
        Gm = (Ct[s:s + step] @ Cpow.T).toarray()
        np.power(Gm, 2.0, out=Gm)
        ii[s:s + step] = Gm @ CatT.toarray() if False else (sp.csr_matrix(Gm) @ CatT).toarray()
    signals["ii"] = ii
    log(f"{stage}: ii done")

    # per-user standardise each signal over eligible items, mask ineligible
    for k, v in signals.items():
        v = v.astype(np.float32)
        v[~elig_mask] = 0.0
        mu = v.sum(1, keepdims=True) / np.maximum(elig_mask.sum(1, keepdims=True), 1)
        sd = np.sqrt(((v - mu) ** 2 * elig_mask).sum(1, keepdims=True)
                     / np.maximum(elig_mask.sum(1, keepdims=True), 1)) + 1e-6
        v = (v - mu) / sd
        v[~elig_mask] = -1e9
        signals[k] = v
    return signals


# ---------------------------------------------------------------- eligibility
def load_elig(name: str, target_uids: np.ndarray):
    z = np.load(DATA / f"{name}_eligible_candidates.npz", allow_pickle=False)
    uids = z["uids"].astype(np.int64)
    items = z["item_ids"].astype(np.int64)
    packed = z["packed_eligible"]
    m = np.unpackbits(packed, axis=1, count=len(items), bitorder="little").astype(bool)
    # reorder columns to the global catalog-first ordering
    order = ICOL.reindex(items).values
    assert (order < NCAT).all()
    M = np.zeros((len(uids), NCAT), dtype=bool)
    M[:, order] = m
    row = pd.Series(np.arange(len(uids)), index=uids).reindex(target_uids).values
    return M[row]


# ---------------------------------------------------------------- evaluation
def val_gold(target_uids, elig_mask):
    tg = set(target_uids.tolist())
    st = LS_VA[(LS_VA.is_organic == 1) & (LS_VA.played_ratio_pct >= 80)
               & LS_VA.uid.isin(tg)][["uid", "item_id"]].drop_duplicates()
    lk = LK_VA[(LK_VA.is_organic == 1) & LK_VA.uid.isin(tg)][
        ["uid", "item_id"]].drop_duplicates()
    urow = pd.Series(np.arange(len(target_uids)), index=target_uids)
    G = np.zeros((len(target_uids), NCAT), dtype=np.int8)
    for frame, grade in ((st, 1), (lk, 3)):
        r = urow.reindex(frame.uid.values).values
        c = ICOL.reindex(frame.item_id.values).values
        ok = (c < NCAT) & ~pd.isna(r)
        r, c = r[ok].astype(int), c[ok].astype(int)
        keep = elig_mask[r, c]
        G[r[keep], c[keep]] = grade
    return G


DISC = 1.0 / np.log2(np.arange(2, 12))


def ndcg10(scores: np.ndarray, gold: np.ndarray) -> float:
    top = np.argpartition(-scores, 10, axis=1)[:, :10]
    ord_ = np.take_along_axis(scores, top, 1).argsort(axis=1)[:, ::-1]
    top = np.take_along_axis(top, ord_, 1)
    g = np.take_along_axis(gold, top, 1).astype(np.float64)
    dcg = ((2.0 ** g - 1.0) * DISC).sum(1)
    idcg = np.empty(len(gold))
    for i in range(len(gold)):
        gv = gold[i][gold[i] > 0]
        gv = -np.sort(-gv)[:10].astype(np.float64)
        idcg[i] = ((2.0 ** gv - 1.0) * DISC[:len(gv)]).sum()
    sel = idcg > 0
    return float((dcg[sel] / idcg[sel]).mean()), int(sel.sum())


# ---------------------------------------------------------------- write out
def write_ranking(scores, target_uids, path):
    top = np.argpartition(-scores, 100, axis=1)[:, :100]
    ord_ = np.take_along_axis(scores, top, 1).argsort(axis=1)[:, ::-1]
    top = np.take_along_axis(top, ord_, 1)
    items = ITEMS[top]
    n = len(target_uids)
    df = pd.DataFrame({
        "uid": np.repeat(target_uids, 100).astype(np.int64),
        "item_id": items.ravel().astype(np.int64),
        "rank": np.tile(np.arange(1, 101), n).astype(np.int64),
    })
    df.to_parquet(path, index=False)
    log(f"wrote {path}")


def main():
    target_uids = TARGETS.uid.values.astype(np.int64)
    em_val = load_elig("validation", target_uids)
    sig_val = build_signals("val", target_uids, em_val)
    gold = val_gold(target_uids, em_val)
    log(f"gold users={int((gold>0).any(1).sum())}")

    keys = list(sig_val.keys())
    base = {"pop": 0.15, "pop7": 0.25, "pop2": 0.1, "poplike": 0.15,
            "knn": 1.0, "svd": 0.6, "ii": 0.6}
    w = {k: base.get(k, 0.0) for k in keys}

    def blend(sig, w):
        out = np.zeros_like(sig[keys[0]])
        for k in keys:
            if w[k]:
                out += w[k] * sig[k]
        return out

    for k in keys:
        s, n = ndcg10(sig_val[k], gold)
        log(f"single {k:8s} ndcg@10={s:.5f} users={n}")

    best, nu = ndcg10(blend(sig_val, w), gold)
    log(f"init blend {best:.5f} users={nu}")
    grid = [0.0, 0.05, 0.1, 0.2, 0.35, 0.5, 0.75, 1.0, 1.5, 2.0]
    for rnd in range(3):
        for k in keys:
            cur = w[k]
            for cand in grid:
                if cand == cur:
                    continue
                w[k] = cand
                s, _ = ndcg10(blend(sig_val, w), gold)
                if s > best + 1e-6:
                    best, cur = s, cand
                w[k] = cur
        log(f"round {rnd} best={best:.5f} w={w}")
        if time.time() - T0 > 900:
            break
    log(f"FINAL val ndcg@10={best:.5f} weights={w}")
    (OUT / "weights.json").write_text(json.dumps({"weights": w, "val_ndcg": best}, indent=2))

    write_ranking(blend(sig_val, w), target_uids, OUT / "validation.parquet")
    del sig_val

    em_test = load_elig("test", target_uids)
    sig_test = build_signals("test", target_uids, em_test)
    write_ranking(blend(sig_test, w), target_uids, OUT / "submission.parquet")


if __name__ == "__main__":
    main()

python -c "
import numpy as np,pandas as pd,scipy.sparse as sp,time
ls=pd.read_parquet('/task/data/train_listens.parquet',columns=['uid','item_id','played_ratio_pct'])
ls=ls[ls.played_ratio_pct>=50]
u,ui=np.unique(ls.uid.values,return_inverse=True); it,ii=np.unique(ls.item_id.values,return_inverse=True)
A=sp.coo_matrix((np.ones(len(ui),np.float32),(ui,ii)),shape=(len(u),len(it))).tocsr(); A.sum_duplicates()
A.data=np.log1p(A.data).astype(np.float32)
cnt=np.asarray((A>0).sum(0)).ravel()
print('items',len(it),'nnz',A.nnz, 'itemcount>3000:',(cnt>3000).sum(),'sumsq',(np.minimum(cnt,10**9).astype(float)**2).sum())
cn=np.sqrt(np.asarray(A.multiply(A).sum(0)).ravel())+1e-6
C=([email protected]((1/cn).astype(np.float32))).tocsr()
rn=np.sqrt(np.asarray(C.multiply(C).sum(1)).ravel())+1e-6
Cn=(sp.diags((1/rn).astype(np.float32))@C).tocsr()
t=time.time(); G=(Cn[:500]@Cn.T).toarray(); print('chunk500 exact',time.time()-t, G.shape)
keep=cnt<=3000; Cp=Cn[:,keep].tocsr(); print('pruned nnz',Cp.nnz)
t=time.time(); G2=(Cp[:500]@Cp.T.tocsr()).toarray(); print('chunk500 pruned',time.time()-t)
" 2>&1|tail

/app/solution/rank.py

#!/usr/bin/env python3
"""Discovery ranker for the "Made for You" shelf (Yambda organic NDCG@10).

Pipeline
--------
1. Build a user x item affinity matrix from observed history (positive listens,
   strong organic listens, likes; explicit dislikes subtracted).
2. Derive several ranking signals over the 30k retrieval catalog:
   popularity / trending / like-velocity (global), two user-kNN collaborative
   variants (exact cosine neighbours, top-K truncated, different weightings and
   similarity sharpening) and a truncated-SVD latent-factor score.
3. Per-user standardise each signal over that user's eligible candidates and
   blend them with weights hill-climbed on the public validation window.
4. Refit every signal on train + val history and emit the hidden-test ranking
   with the selected weights.
"""
from __future__ import annotations

import json
import time
from pathlib import Path

import numpy as np
import pandas as pd
import scipy.sparse as sp
from sklearn.utils.extmath import randomized_svd

DATA = Path("/task/data")
OUT = Path("/app/solution")
T0 = time.time()
DAY = 86400
NEG_INF = -1e9


def log(*a):
    print(f"[{time.time()-T0:7.1f}s]", *a, flush=True)


def load_events(split: str):
    ls = pd.read_parquet(
        DATA / f"{split}_listens.parquet",
        columns=["uid", "item_id", "timestamp", "is_organic", "played_ratio_pct"],
    )
    lk = pd.read_parquet(DATA / f"{split}_likes.parquet",
                         columns=["uid", "item_id", "timestamp", "is_organic"])
    dk = pd.read_parquet(DATA / f"{split}_dislikes.parquet",
                         columns=["uid", "item_id", "timestamp"])
    return ls, lk, dk


LS_TR, LK_TR, DK_TR = load_events("train")
LS_VA, LK_VA, DK_VA = load_events("val")
CATALOG = pd.read_parquet(DATA / "candidate_catalog.parquet")
TARGETS = pd.read_parquet(DATA / "target_users.parquet")

ALL_UIDS = np.union1d(LS_TR.uid.unique(), LS_VA.uid.unique()).astype(np.int64)
UROW = pd.Series(np.arange(len(ALL_UIDS)), index=ALL_UIDS)
tail = np.setdiff1d(np.union1d(LS_TR.item_id.unique(), LS_VA.item_id.unique()).astype(np.int64),
                    CATALOG.item_id.values.astype(np.int64))
ITEMS = np.concatenate([CATALOG.item_id.values.astype(np.int64), tail])
ICOL = pd.Series(np.arange(len(ITEMS)), index=ITEMS)
NCAT = len(CATALOG)
NU, NI = len(ALL_UIDS), len(ITEMS)
log(f"users={NU} items={NI} catalog={NCAT}")


def history(stage: str):
    if stage == "val":
        return LS_TR, LK_TR, DK_TR
    return (pd.concat([LS_TR, LS_VA], ignore_index=True),
            pd.concat([LK_TR, LK_VA], ignore_index=True),
            pd.concat([DK_TR, DK_VA], ignore_index=True))


def mat(rows, cols, vals, log1p=False, clip=None):
    m = sp.coo_matrix((vals.astype(np.float32), (rows, cols)), shape=(NU, NI)).tocsr()
    m.sum_duplicates()
    if log1p:
        m.data = np.log1p(m.data).astype(np.float32)
    if clip is not None:
        m.data = np.minimum(m.data, clip).astype(np.float32)
    return m


def normalise(A, col_pow=1.0, row_norm=True):
    cn = np.sqrt(np.asarray(A.multiply(A).sum(0)).ravel()) + 1e-6
    X = (A @ sp.diags((cn ** -col_pow).astype(np.float32))).tocsr()
    if row_norm:
        rn = np.sqrt(np.asarray(X.multiply(X).sum(1)).ravel()) + 1e-6
        X = (sp.diags((1.0 / rn).astype(np.float32)) @ X).tocsr()
    return X


def neighbour_score(X, trow, Ycat, K, power, chunk=2000):
    """Top-K cosine-neighbour aggregation of Ycat over the catalog."""
    nT = len(trow)
    out = np.empty((nT, NCAT), dtype=np.float32)
    XT = X.T.tocsr()
    for s in range(0, nT, chunk):
        idx = trow[s:s + chunk]
        G = (X[idx] @ XT).toarray()
        G[np.arange(len(idx)), idx] = 0.0
        k = min(K, G.shape[1] - 1)
        thr = np.partition(G, G.shape[1] - k, axis=1)[:, G.shape[1] - k][:, None]
        G[G < thr] = 0.0
        if power != 1.0:
            np.power(G, power, out=G)
        out[s:s + chunk] = (sp.csr_matrix(G) @ Ycat).toarray()
        del G
    return out


def build_signals(stage: str, target_uids: np.ndarray, elig: np.ndarray):
    ls, lk, dk = history(stage)
    cutoff = int(ls.timestamp.max())
    pr = ls.played_ratio_pct.values
    org = ls.is_organic.values == 1
    pos, strong = pr >= 50, pr >= 80
    lr = UROW.reindex(ls.uid.values).values
    lc = ICOL.reindex(ls.item_id.values).values
    ts = ls.timestamp.values

    A_pos = mat(lr[pos], lc[pos], np.ones(pos.sum()), log1p=True)
    sel = strong & org
    A_str = mat(lr[sel], lc[sel], np.ones(sel.sum()), log1p=True)
    A_like = mat(UROW.reindex(lk.uid.values).values, ICOL.reindex(lk.item_id.values).values,
                 np.ones(len(lk)), clip=1.0)
    A_dis = mat(UROW.reindex(dk.uid.values).values, ICOL.reindex(dk.item_id.values).values,
                np.ones(len(dk)), clip=1.0)
    A = (A_pos + 0.7 * A_str + 2.0 * A_like - 1.0 * A_dis).tocsr()
    A.data = np.maximum(A.data, 0.0).astype(np.float32)
    A.eliminate_zeros()
    log(f"{stage}: A nnz={A.nnz}")

    trow = UROW.reindex(target_uids).values
    nT = len(trow)
    sig: dict[str, np.ndarray] = {}

    def popvec(mask, tmin=None, frame_ts=None, cols=None):
        m = mask if tmin is None else (mask & (frame_ts >= tmin))
        c = cols[m]
        c = c[c < NCAT]
        return np.log1p(np.bincount(c, minlength=NCAT).astype(np.float32))

    sig["pop"] = popvec(pos & org, cols=lc)
    sig["pop7"] = popvec(strong & org, cutoff - 7 * DAY, ts, lc)
    sig["pop2"] = popvec(strong & org, cutoff - 2 * DAY, ts, lc)
    kc = ICOL.reindex(lk.item_id.values).values
    korg = lk.is_organic.values == 1
    sig["poplike"] = popvec(korg, cutoff - 14 * DAY, lk.timestamp.values, kc)
    log(f"{stage}: popularity done")

    Cn = normalise(A, col_pow=1.0, row_norm=True)
    C_cat = normalise(A, col_pow=1.0, row_norm=False)[:, :NCAT].tocsr()
    sig["knn"] = neighbour_score(Cn, trow, C_cat, K=250, power=3.0)
    log(f"{stage}: knn done")

    Ch = normalise(A, col_pow=0.4, row_norm=True)
    Ch_cat = Ch[:, :NCAT].tocsr()
    sig["knn2"] = neighbour_score(Ch, trow, Ch_cat, K=800, power=1.0)
    log(f"{stage}: knn2 done")

    U, S, Vt = randomized_svd(Cn, n_components=192, n_iter=4, random_state=0)
    sig["svd"] = ((U[trow] * S) @ Vt[:, :NCAT]).astype(np.float32)
    log(f"{stage}: svd done")

    ncand = np.maximum(elig.sum(1, keepdims=True), 1).astype(np.float32)
    for k, v in list(sig.items()):
        if v.ndim == 1:
            v = np.tile(v, (1, 1))
            mu, sd = v.mean(), v.std() + 1e-6
            sig[k] = ((v - mu) / sd).astype(np.float32)  # shape (1, NCAT)
        else:
            v = np.where(elig, v, 0.0).astype(np.float32)
            mu = v.sum(1, keepdims=True) / ncand
            sd = np.sqrt(np.maximum((v * v).sum(1, keepdims=True) / ncand - mu * mu, 0)) + 1e-6
            sig[k] = ((v - mu) / sd).astype(np.float32)
    return sig


def load_elig(name: str, target_uids: np.ndarray):
    with np.load(DATA / f"{name}_eligible_candidates.npz", allow_pickle=False) as z:
        uids = z["uids"].astype(np.int64)
        items = z["item_ids"].astype(np.int64)
        m = np.unpackbits(z["packed_eligible"], axis=1, count=len(items),
                          bitorder="little").astype(bool)
    order = ICOL.reindex(items).values
    assert (order < NCAT).all()
    M = np.zeros((len(uids), NCAT), dtype=bool)
    M[:, order] = m
    row = pd.Series(np.arange(len(uids)), index=uids).reindex(target_uids).values
    return M[row]


def val_gold(target_uids, elig):
    tg = set(target_uids.tolist())
    st = LS_VA[(LS_VA.is_organic == 1) & (LS_VA.played_ratio_pct >= 80)
               & LS_VA.uid.isin(tg)][["uid", "item_id"]].drop_duplicates()
    lk = LK_VA[(LK_VA.is_organic == 1) & LK_VA.uid.isin(tg)][
        ["uid", "item_id"]].drop_duplicates()
    urow = pd.Series(np.arange(len(target_uids)), index=target_uids)
    G = np.zeros((len(target_uids), NCAT), dtype=np.int8)
    for frame, grade in ((st, 1), (lk, 3)):
        r = urow.reindex(frame.uid.values).values.astype(np.float64)
        c = ICOL.reindex(frame.item_id.values).values.astype(np.float64)
        ok = np.isfinite(r) & np.isfinite(c) & (np.nan_to_num(c, nan=1e18) < NCAT)
        r, c = r[ok].astype(int), c[ok].astype(int)
        keep = elig[r, c]
        G[r[keep], c[keep]] = grade
    return G


DISC = 1.0 / np.log2(np.arange(2, 12))


def ideal(gold):
    idcg = np.zeros(len(gold))
    r, c = np.nonzero(gold)
    df = pd.DataFrame({"r": r, "g": gold[r, c].astype(np.float64)})
    for rr, grp in df.groupby("r").g:
        gv = -np.sort(-grp.to_numpy())[:10]
        idcg[rr] = ((2.0 ** gv - 1.0) * DISC[:len(gv)]).sum()
    return idcg


def blend_full(sig, w, keys):
    out = None
    for k in keys:
        if not w[k]:
            continue
        v = sig[k] * w[k]
        out = v if out is None else out + v
    return out


def main():
    target_uids = TARGETS.uid.values.astype(np.int64)
    em = load_elig("validation", target_uids)
    sig = build_signals("val", target_uids, em)
    gold = val_gold(target_uids, em)
    idcg = ideal(gold)
    scored = idcg > 0
    log(f"gold users={int(scored.sum())}")

    keys = ["pop", "pop7", "pop2", "poplike", "knn", "knn2", "svd"]
    w = {"pop": 0.1, "pop7": 0.3, "pop2": 0.1, "poplike": 0.2,
         "knn": 1.0, "knn2": 0.5, "svd": 0.5}

    # Reduced tuning arena: union of each signal's per-user top items.
    nT = len(target_uids)
    tops = []
    for k in keys:
        v = sig[k]
        if v.shape[0] == 1:
            t = np.argpartition(-v[0], 60)[:60]
            tops.append(np.tile(t, (nT, 1)))
        else:
            vm = np.where(em, v, NEG_INF)
            tops.append(np.argpartition(-vm, 120, axis=1)[:, :120])
    CAND = np.concatenate(tops, axis=1)
    CANDW = CAND.shape[1]
    small = {}
    for k in keys:
        v = sig[k]
        small[k] = (np.take_along_axis(np.broadcast_to(v, (nT, NCAT)), CAND, 1)
                    .astype(np.float32).copy())
    elig_s = np.take_along_axis(em, CAND, 1)
    gold_s = np.take_along_axis(gold, CAND, 1).astype(np.float64)
    # duplicate columns across signals: keep first occurrence only
    order = np.argsort(CAND, axis=1, kind="stable")
    sorted_c = np.take_along_axis(CAND, order, 1)
    dup = np.zeros_like(elig_s)
    dup[:, 1:] = sorted_c[:, 1:] == sorted_c[:, :-1]
    dupmask = np.zeros_like(elig_s)
    np.put_along_axis(dupmask, order, dup, 1)
    valid = elig_s & ~dupmask
    log(f"arena width={CANDW}, gold recall="
        f"{float((gold_s * valid).sum() / max(gold.sum(), 1)):.3f}")

    def score_small(w):
        out = np.zeros((nT, CANDW), dtype=np.float32)
        for k in keys:
            if w[k]:
                out += w[k] * small[k]
        out[~valid] = NEG_INF
        top = np.argpartition(-out, 10, axis=1)[:, :10]
        o = np.take_along_axis(out, top, 1).argsort(axis=1)[:, ::-1]
        top = np.take_along_axis(top, o, 1)
        g = np.take_along_axis(gold_s, top, 1)
        dcg = ((2.0 ** g - 1.0) * DISC).sum(1)
        return float((dcg[scored] / idcg[scored]).mean())

    best = score_small(w)
    log(f"init blend ndcg@10={best:.5f}")
    for k in keys:
        w0 = dict.fromkeys(keys, 0.0)
        w0[k] = 1.0
        log(f"single {k:8s} {score_small(w0):.5f}")

    grid = [0.0, 0.05, 0.1, 0.2, 0.3, 0.45, 0.6, 0.8, 1.0, 1.3, 1.7, 2.2]
    for rnd in range(4):
        improved = False
        for k in keys:
            cur = w[k]
            for c in grid:
                if c == cur:
                    continue
                w[k] = c
                s = score_small(w)
                if s > best + 1e-6:
                    best, cur, improved = s, c, True
                w[k] = cur
        log(f"round {rnd}: best={best:.5f} w={w}")
        if not improved:
            break
    log(f"FINAL val ndcg@10={best:.5f} weights={w}")
    (OUT / "weights.json").write_text(json.dumps({"weights": w, "val_ndcg_arena": best},
                                                 indent=2))

    def write(sig, w, path):
        s = blend_full(sig, w, keys)
        s = np.where(em_cur, s, NEG_INF)
        top = np.argpartition(-s, 100, axis=1)[:, :100]
        o = np.take_along_axis(s, top, 1).argsort(axis=1)[:, ::-1]
        top = np.take_along_axis(top, o, 1)
        pd.DataFrame({
            "uid": np.repeat(target_uids, 100),
            "item_id": ITEMS[top].ravel().astype(np.int64),
            "rank": np.tile(np.arange(1, 101), nT).astype(np.int64),
        }).to_parquet(path, index=False)
        log(f"wrote {path}")

    em_cur = em
    write(sig, w, OUT / "validation.parquet")
    del sig, small
    em_cur = load_elig("test", target_uids)
    sig_t = build_signals("test", target_uids, em_cur)
    write(sig_t, w, OUT / "submission.parquet")


if __name__ == "__main__":
    main()

nohup python -u rank.py > run.log 2>&1 &
echo started; sleep 60; tail -30 run.log

python - <<'EOF'
import re
s=open('rank.py').read()
s=s.replace('''ALL_UIDS = np.union1d(LS_TR.uid.unique(), LS_VA.uid.unique()).astype(np.int64)''',
'''ALL_UIDS = np.unique(np.concatenate([
    f.uid.unique().astype(np.int64)
    for f in (LS_TR, LS_VA, LK_TR, LK_VA, DK_TR, DK_VA)]))''')
s=s.replace('''tail = np.setdiff1d(np.union1d(LS_TR.item_id.unique(), LS_VA.item_id.unique()).astype(np.int64),
                    CATALOG.item_id.values.astype(np.int64))''',
'''_all_items = np.unique(np.concatenate([
    f.item_id.unique().astype(np.int64)
    for f in (LS_TR, LS_VA, LK_TR, LK_VA, DK_TR, DK_VA)]))
tail = np.setdiff1d(_all_items, CATALOG.item_id.values.astype(np.int64))''')
s=s.replace('''def mat(rows, cols, vals, log1p=False, clip=None):
    m = sp.coo_matrix''','''def mat(rows, cols, vals, log1p=False, clip=None):
    rows = np.asarray(rows, dtype=np.float64)
    cols = np.asarray(cols, dtype=np.float64)
    ok = np.isfinite(rows) & np.isfinite(cols)
    rows, cols, vals = rows[ok].astype(np.int64), cols[ok].astype(np.int64), np.asarray(vals)[ok]
    m = sp.coo_matrix''')
open('rank.py','w').write(s)
EOF
nohup python -u rank.py > run.log 2>&1 & echo started; sleep 75; tail -25 run.log

sleep 75; tail -25 run.log

sleep 70; tail -25 run.log

sleep 45; tail -5 run.log; ps -o pid,pcpu,rss,etime,cmd -C python | head

tail -30 run.log; echo "---"; ps aux | grep -c rank.py; dmesg 2>/dev/null | tail -3

python - <<'EOF'
s=open('rank.py').read()
s=s.replace('''    U, S, Vt = randomized_svd(Cn, n_components=192, n_iter=4, random_state=0)
    sig["svd"] = ((U[trow] * S) @ Vt[:, :NCAT]).astype(np.float32)
    log(f"{stage}: svd done")
''','''    U, S, Vt = randomized_svd(Cn, n_components=160, n_iter=4, random_state=0)
    sig["svd"] = np.ascontiguousarray(((U[trow] * S) @ Vt[:, :NCAT]), dtype=np.float32)
    del U, S, Vt, Cn, Ch, C_cat, Ch_cat, A, A_pos, A_str, A_like, A_dis
    log(f"{stage}: svd done")
''')
s=s.replace('''        else:
            v = np.where(elig, v, 0.0).astype(np.float32)
            mu = v.sum(1, keepdims=True) / ncand
            sd = np.sqrt(np.maximum((v * v).sum(1, keepdims=True) / ncand - mu * mu, 0)) + 1e-6
            sig[k] = ((v - mu) / sd).astype(np.float32)''','''        else:
            v *= elig
            mu = v.sum(1, dtype=np.float64, keepdims=True) / ncand
            sq = np.einsum("ij,ij->i", v, v).reshape(-1, 1).astype(np.float64) / ncand
            sd = (np.sqrt(np.maximum(sq - mu * mu, 0)) + 1e-6).astype(np.float32)
            v -= mu.astype(np.float32)
            v /= sd
            sig[k] = v''')
s=s.replace('''def blend_full(sig, w, keys):
    out = None
    for k in keys:
        if not w[k]:
            continue
        v = sig[k] * w[k]
        out = v if out is None else out + v
    return out''','''def blend_full(sig, w, keys, nT):
    out = np.zeros((nT, NCAT), dtype=np.float32)
    for k in keys:
        if not w[k]:
            continue
        v = sig[k]
        if v.shape[0] == 1:
            out += w[k] * v[0]
        else:
            out += w[k] * v
    return out''')
s=s.replace('''        s = blend_full(sig, w, keys)
        s = np.where(em_cur, s, NEG_INF)''','''        s = blend_full(sig, w, keys, nT)
        s[~em_cur] = NEG_INF''')
s=s.replace('''    small = {}
    for k in keys:
        v = sig[k]
        small[k] = (np.take_along_axis(np.broadcast_to(v, (nT, NCAT)), CAND, 1)
                    .astype(np.float32).copy())''','''    small = {}
    for k in keys:
        v = sig[k]
        if v.shape[0] == 1:
            small[k] = np.ascontiguousarray(v[0][CAND], dtype=np.float32)
        else:
            small[k] = np.ascontiguousarray(np.take_along_axis(v, CAND, 1), dtype=np.float32)''')
s=s.replace('''            vm = np.where(em, v, NEG_INF)
            tops.append(np.argpartition(-vm, 120, axis=1)[:, :120])''','''            tops.append(np.argpartition(-v, 120, axis=1)[:, :120])''')
s=s.replace('for rnd in range(4):','for rnd in range(3):')
s=s.replace('''    em_cur = em
    write(sig, w, OUT / "validation.parquet")
    del sig, small''','''    em_cur = em
    write(sig, w, OUT / "validation.parquet")
    sig.clear()
    del small, gold, gold_s, CAND''')
open('rank.py','w').write(s)
EOF
nohup python -u rank.py > run.log 2>&1 & echo started; sleep 100; tail -20 run.log

python - <<'EOF'
s=open('rank.py').read()
s=s.replace('''def build_signals(stage''','''def topk_idx(v, k, chunk=1000):
    """Chunked per-row top-k indices (descending) as int32."""
    n = v.shape[0]
    out = np.empty((n, k), dtype=np.int32)
    for st in range(0, n, chunk):
        b = v[st:st + chunk]
        p = np.argpartition(b, b.shape[1] - k, axis=1)[:, b.shape[1] - k:]
        o = np.take_along_axis(b, p, 1).argsort(axis=1)[:, ::-1]
        out[st:st + chunk] = np.take_along_axis(p, o, 1)
    return out


def build_signals(stage''')
s=s.replace('''            t = np.argpartition(-v[0], 60)[:60]
            tops.append(np.tile(t, (nT, 1)))
        else:
            tops.append(np.argpartition(-v, 120, axis=1)[:, :120])''','''            t = np.argpartition(v[0], NCAT - 60)[NCAT - 60:]
            tops.append(np.tile(t.astype(np.int32), (nT, 1)))
        else:
            tops.append(topk_idx(v, 120))''')
s=s.replace('''        out[~valid] = NEG_INF
        top = np.argpartition(-out, 10, axis=1)[:, :10]
        o = np.take_along_axis(out, top, 1).argsort(axis=1)[:, ::-1]
        top = np.take_along_axis(top, o, 1)
        g''','''        out[~valid] = NEG_INF
        top = np.argpartition(out, CANDW - 10, axis=1)[:, CANDW - 10:]
        o = np.take_along_axis(out, top, 1).argsort(axis=1)[:, ::-1]
        top = np.take_along_axis(top, o, 1)
        g''')
s=s.replace('''        top = np.argpartition(-s, 100, axis=1)[:, :100]
        o = np.take_along_axis(s, top, 1).argsort(axis=1)[:, ::-1]
        top = np.take_along_axis(top, o, 1)
        pd''','''        top = topk_idx(s, 100)
        del s
        pd''')
s=s.replace('CAND = np.concatenate(tops, axis=1)','CAND = np.concatenate(tops, axis=1).astype(np.int64)\n    del tops')
open('rank.py','w').write(s)
EOF
nohup python -u rank.py > run.log 2>&1 & echo started; sleep 105; tail -30 run.log

sleep 95; tail -6 run.log; ls -la /app/solution/

sleep 40; tail -3 run.log; ls -la /app/solution/submission.parquet 2>&1

sleep 45; tail -4 run.log; ls -la /app/solution/submission.parquet 2>&1 | tail -1

ps aux|grep -c "[r]ank.py"; python - <<'EOF'
s=open('/app/solution/rank.py').read()
# accumulate signals one at a time instead of holding all dense matrices
s=s.replace('''def main():''','''def build_blend(stage, target_uids, elig, w, keys):
    """Memory-lean scoring: standardise and accumulate each signal in turn."""
    nT = len(target_uids)
    acc = np.zeros((nT, NCAT), dtype=np.float32)
    ncand = np.maximum(elig.sum(1, keepdims=True), 1).astype(np.float32)

    def add(v, weight):
        if weight == 0.0:
            return
        if v.ndim == 1:
            acc.__iadd__(weight * ((v - v.mean()) / (v.std() + 1e-6)).astype(np.float32))
            return
        v *= elig
        mu = v.sum(1, dtype=np.float64, keepdims=True) / ncand
        sq = np.einsum("ij,ij->i", v, v).reshape(-1, 1).astype(np.float64) / ncand
        sd = (np.sqrt(np.maximum(sq - mu * mu, 0)) + 1e-6).astype(np.float32)
        v -= mu.astype(np.float32)
        v /= sd
        v *= np.float32(weight)
        acc.__iadd__(v)

    ls, lk, dk = history(stage)
    cutoff = int(ls.timestamp.max())
    pr, org = ls.played_ratio_pct.values, ls.is_organic.values == 1
    pos, strong = pr >= 50, pr >= 80
    lr = UROW.reindex(ls.uid.values).values
    lc = ICOL.reindex(ls.item_id.values).values
    ts = ls.timestamp.values
    A_pos = mat(lr[pos], lc[pos], np.ones(pos.sum()), log1p=True)
    sel = strong & org
    A = (A_pos + 0.7 * mat(lr[sel], lc[sel], np.ones(sel.sum()), log1p=True)
         + 2.0 * mat(UROW.reindex(lk.uid.values).values,
                     ICOL.reindex(lk.item_id.values).values, np.ones(len(lk)), clip=1.0)
         - 1.0 * mat(UROW.reindex(dk.uid.values).values,
                     ICOL.reindex(dk.item_id.values).values, np.ones(len(dk)), clip=1.0)).tocsr()
    A.data = np.maximum(A.data, 0.0).astype(np.float32)
    A.eliminate_zeros()
    del A_pos
    trow = UROW.reindex(target_uids).values

    def popvec(mask, tmin, tvec, cols):
        m = mask if tmin is None else (mask & (tvec >= tmin))
        c = cols[m]
        c = c[c < NCAT]
        return np.log1p(np.bincount(c, minlength=NCAT).astype(np.float32))

    add(popvec(pos & org, None, ts, lc), w["pop"])
    add(popvec(strong & org, cutoff - 7 * DAY, ts, lc), w["pop7"])
    add(popvec(strong & org, cutoff - 2 * DAY, ts, lc), w["pop2"])
    add(popvec(lk.is_organic.values == 1, cutoff - 14 * DAY, lk.timestamp.values,
               ICOL.reindex(lk.item_id.values).values), w["poplike"])
    log(f"{stage}: pop accumulated")
    Cn = normalise(A, col_pow=1.0, row_norm=True)
    if w["knn"]:
        C_cat = normalise(A, col_pow=1.0, row_norm=False)[:, :NCAT].tocsr()
        add(neighbour_score(Cn, trow, C_cat, K=250, power=3.0), w["knn"])
        del C_cat
        log(f"{stage}: knn accumulated")
    if w["knn2"]:
        Ch = normalise(A, col_pow=0.4, row_norm=True)
        add(neighbour_score(Ch, trow, Ch[:, :NCAT].tocsr(), K=800, power=1.0), w["knn2"])
        del Ch
        log(f"{stage}: knn2 accumulated")
    if w["svd"]:
        U, S, Vt = randomized_svd(Cn, n_components=160, n_iter=4, random_state=0)
        Z = np.ascontiguousarray((U[trow] * S) @ Vt[:, :NCAT], dtype=np.float32)
        del U, S, Vt, Cn, A
        add(Z, w["svd"])
        del Z
        log(f"{stage}: svd accumulated")
    acc[~elig] = NEG_INF
    return acc


def write_ranking(scores, target_uids, path):
    top = topk_idx(scores, 100)
    pd.DataFrame({
        "uid": np.repeat(target_uids, 100).astype(np.int64),
        "item_id": ITEMS[top].ravel().astype(np.int64),
        "rank": np.tile(np.arange(1, 101), len(target_uids)).astype(np.int64),
    }).to_parquet(path, index=False)
    log(f"wrote {path}")


def main_test(weights):
    target_uids = TARGETS.uid.values.astype(np.int64)
    em = load_elig("test", target_uids)
    keys = ["pop", "pop7", "pop2", "poplike", "knn", "knn2", "svd"]
    acc = build_blend("test", target_uids, em, weights, keys)
    write_ranking(acc, target_uids, OUT / "submission.parquet")


def main():''')
s=s.replace('''if __name__ == "__main__":
    main()''','''if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1 and sys.argv[1] == "test":
        main_test(json.loads((OUT / "weights.json").read_text())["weights"])
    else:
        main()''')
open('/app/solution/rank.py','w').write(s)
EOF
nohup python -u rank.py test > run_test.log 2>&1 & echo started; sleep 90; tail -12 run_test.log

python -c "
import pandas as pd,numpy as np,sys
sys.path.insert(0,'/task')
from ranking_evaluation import CandidateSet, validate_ranking, read_ranking
from pathlib import Path
t=set(pd.read_parquet('/task/data/target_users.parquet').uid.astype(int))
f=validate_ranking(read_ranking(Path('submission.parquet')),t)
cs=CandidateSet(Path('/task/data/test_eligible_candidates.npz'))
row=np.array([cs.uid_row[u] for u in f.uid]); col=np.array([cs.item_column[i] for i in f.item_id])
bits=(cs.packed[row,col//8]>>(col%8))&1
print('rows',len(f),'all eligible',bool(bits.all()))
" && python /task/evaluate_public.py validation.parquet

/app/solution/README.md

# Made-for-You discovery ranker

`rank.py` builds both artifacts. Public validation score of the selected
configuration: **graded NDCG@10 = 0.038596** over 4263 scored users
(`python /task/evaluate_public.py solution/validation.parquet`).

## Run

```sh
python rank.py        # fits on train_*, tunes blend weights, writes validation.parquet + weights.json
python rank.py test   # refits every signal on train_* + val_*, writes submission.parquet
```

The two entry points exist because the full-signal tuning path holds several
dense `8456 x 30000` score matrices; `rank.py test` uses `build_blend`, which
standardises and accumulates one signal at a time so the refit fits in memory.

## Model

One affinity matrix per stage over all 10,000 users and 871,816 items (the
similarity space is the full item space, only the scoring target is the 30k
retrieval catalog):

```
A = log1p(positive listens) + 0.7*log1p(strong organic listens) + 2*like - dislike   (clipped at 0)
```

History is `train_*` for the validation ranker and `train_* + val_*` for the
hidden-test ranker; validation events are never used as features when scoring
the validation window.

Signals, each per-user standardised over that user's eligible candidates only:

| signal | definition |
| --- | --- |
| `pop` | log organic positive-listen count over all history |
| `pop7` | log strong-organic-listen count in the last 7 days before the cutoff |
| `pop2` | same over the last 2 days (trending) |
| `poplike` | log organic-like count over the last 14 days |
| `knn` | user-kNN: exact cosine neighbours on the item-normalised matrix, top-250, similarity cubed, aggregating un-row-normalised catalog affinity |
| `knn2` | user-kNN variant with weaker item-popularity damping (`col_pow=0.4`), top-800, linear similarity |
| `svd` | rank-160 randomized SVD reconstruction of the normalised affinity matrix |

Weights are hill-climbed directly on validation NDCG@10 (3 coordinate rounds
over a 12-point grid) inside a reduced arena: the union of each signal's
per-user top items, with duplicate and ineligible columns masked. Tuning on the
arena reproduced the evaluator's score to 4 decimals (0.03856 vs 0.038596).
Selected weights are in `weights.json`.

## Notes

- Explicit preference state is informative but not needed for filtering: every
  `liked`/`disliked` pair in both state files is already ineligible, so
  eligibility masks are used verbatim and dislikes only enter `A` as a negative
  affinity term.
- Trending popularity carries most of the signal on this objective
  (`pop2`/`poplike` get the largest weights); collaborative signals add on top
  of it, while all-time popularity (`pop`) is tuned to zero once the recent
  windows are present.
- Not attempted for lack of time: a learned reranker (sklearn
  `HistGradientBoostingClassifier`) over the top-200 blended candidates with
  the signals above as features, and item-item kNN with an explicitly truncated
  catalog similarity matrix.