Agent trace

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

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

python -c "
import pandas as pd, numpy as np, time
t=time.time()
tu=pd.read_parquet('/task/data/target_users.parquet'); print(tu.shape, tu.columns.tolist(), tu.head())
cc=pd.read_parquet('/task/data/candidate_catalog.parquet'); print(cc.shape, cc.columns.tolist(), cc.head())
ps=pd.read_parquet('/task/data/validation_preference_state.parquet'); print(ps.shape, ps.columns.tolist(), ps.head())
import pyarrow.parquet as pq
f=pq.ParquetFile('/task/data/train_listens.parquet'); print(f.metadata.num_rows, f.schema_arrow)
print(time.time()-t)
"

python -c "
import numpy as np
z=np.load('/task/data/validation_eligible_candidates.npz')
print({k:(z[k].shape,z[k].dtype) for k in z.files})
p=z['packed_eligible']
m=np.unpackbits(p,axis=1,bitorder='little')[:, :len(z['item_ids'])]
print('elig per user', m.sum(1)[:10], m.sum(1).mean())
z2=np.load('/task/data/test_eligible_candidates.npz')
print({k:(z2[k].shape,z2[k].dtype) for k in z2.files})
print((z['uids']==z2['uids']).all(), (z['item_ids']==z2['item_ids']).all())
"

mkdir -p /app/solution && cd /app/solution && python -c "
import pyarrow.parquet as pq, numpy as np, pandas as pd, time
t=time.time()
tb=pq.read_table('/task/data/train_listens.parquet',columns=['uid','item_id','is_organic','played_ratio_pct','timestamp'])
print('read',time.time()-t, tb.num_rows)
d=tb.to_pandas(); del tb
print('pd',time.time()-t)
print('uniq users', d.uid.nunique())
cat=pd.read_parquet('/task/data/candidate_catalog.parquet')
ci=set(cat.item_id.values)
m=d.item_id.isin(ci)
print('in catalog frac', m.mean())
print('ts range', d.timestamp.min(), d.timestamp.max())
print(time.time()-t)
"

python -c "
import pandas as pd, numpy as np, time
t=time.time()
cat=pd.read_parquet('/task/data/candidate_catalog.parquet')
items=cat.item_id.values.astype(np.int64)
z=np.load('/task/data/validation_eligible_candidates.npz')
print('catalog order == npz item order?', (z['item_ids'].astype(np.int64)==items).all())
tl=pd.read_parquet('/task/data/train_listens.parquet',columns=['uid','item_id','is_organic','played_ratio_pct'])
vl=pd.read_parquet('/task/data/val_listens.parquet',columns=['uid','item_id','is_organic','played_ratio_pct'])
vlk=pd.read_parquet('/task/data/val_likes.parquet',columns=['uid','item_id','is_organic'])
tu=set(pd.read_parquet('/task/data/target_users.parquet').uid)
rel=vl[(vl.is_organic==1)&(vl.played_ratio_pct>=80)][['uid','item_id']].drop_duplicates()
rel=rel[rel.uid.isin(tu)]
hist=tl[(tl.is_organic==1)&(tl.played_ratio_pct>=50)][['uid','item_id']].drop_duplicates()
h=set(map(tuple,hist.values))
r=np.array([ (u,i) in h for u,i in rel.values])
print('val strong listens', len(rel),'repeat frac', r.mean())
lk=vlk[(vlk.is_organic==1)&(vlk.uid.isin(tu))][['uid','item_id']].drop_duplicates()
print('val organic likes', len(lk), 'repeat frac', np.mean([ (u,i) in h for u,i in lk.values]))
ps=pd.read_parquet('/task/data/validation_preference_state.parquet')
pl=set(map(tuple, ps[ps.state=='liked'][['uid','item_id']].values))
print('rel in liked-state', np.mean([(u,i) in pl for u,i in rel.values]), 'likes in likedstate', np.mean([(u,i) in pl for u,i in lk.values]))
print(time.time()-t)
"

/app/solution/build.py

"""Discovery ranker for the "Made for You" shelf.

Two stages:
  1. Feature construction from event history (repeat-consumption, explicit
     preference state, user-user CF, SVD item-item CF, popularity).
  2. A blend whose weights are tuned on the public validation window, then
     refit on train+val for the hidden test.
"""
from __future__ import annotations

import sys
import time

import numpy as np
import pandas as pd
import scipy.sparse as sp

DATA = "/task/data"
TRAIN_CUTOFF = 24786800
WEEK = 7 * 24 * 3600
T0 = time.time()


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


def load_events(stage):
    """Return history frames for the given stage ('val' builds from train only)."""
    parts = ["train"] if stage == "val" else ["train", "val"]
    out = {}
    for fam, cols in [
        ("listens", ["uid", "timestamp", "item_id", "is_organic", "played_ratio_pct"]),
        ("likes", ["uid", "timestamp", "item_id", "is_organic"]),
        ("dislikes", ["uid", "timestamp", "item_id", "is_organic"]),
    ]:
        out[fam] = pd.concat(
            [pd.read_parquet(f"{DATA}/{p}_{fam}.parquet", columns=cols) for p in parts],
            ignore_index=True,
        )
    return out


def build(stage):
    """Build the dense (n_users x n_items) score-feature stack for one stage."""
    cat = pd.read_parquet(f"{DATA}/candidate_catalog.parquet")
    items = cat.item_id.values.astype(np.int64)
    n_items = len(items)
    item_pos = pd.Series(np.arange(n_items), index=items)

    npz = np.load(f"{DATA}/{'validation' if stage=='val' else 'test'}_eligible_candidates.npz")
    tgt_uids = npz["uids"].astype(np.int64)
    assert (npz["item_ids"].astype(np.int64) == items).all()
    elig = np.unpackbits(npz["packed_eligible"], axis=1, bitorder="little")[:, :n_items].astype(bool)
    n_tgt = len(tgt_uids)

    ev = load_events(stage)
    cutoff = TRAIN_CUTOFF if stage == "val" else TRAIN_CUTOFF + WEEK + 1800
    log(stage, "events loaded")

    li = ev["listens"]
    li = li[li.item_id.isin(item_pos.index)]
    li = li.assign(col=item_pos.reindex(li.item_id.values).values)

    all_uids = np.union1d(li.uid.unique().astype(np.int64), tgt_uids)
    urow = pd.Series(np.arange(len(all_uids)), index=all_uids)
    li = li.assign(row=urow.reindex(li.uid.values).values)

    strong = li[(li.is_organic == 1) & (li.played_ratio_pct >= 80)]
    pos = li[li.played_ratio_pct >= 50]

    # ---- global popularity (organic strong listens) ----
    pop = np.bincount(strong.col.values, minlength=n_items).astype(np.float32)
    pop_l = np.log1p(pop)
    pop_l /= pop_l.max()

    # ---- user x item history matrices ----
    def mat(df, w=None, n_rows=len(all_uids)):
        d = np.ones(len(df), np.float32) if w is None else w.astype(np.float32)
        m = sp.coo_matrix((d, (df.row.values, df.col.values)), shape=(n_rows, n_items)).tocsr()
        m.sum_duplicates()
        return m

    X_strong = mat(strong)                      # counts of strong organic listens
    X_pos = mat(pos)                            # any >=50% listen
    age = (cutoff - strong.timestamp.values).astype(np.float64)
    X_rec = mat(strong, np.exp(-age / (14 * 86400.0)))      # 14-day half-life-ish
    X_rec2 = mat(strong, np.exp(-age / (3 * 86400.0)))      # short-term
    log(stage, "history matrices", X_strong.nnz, X_pos.nnz)

    tgt_rows = urow.reindex(tgt_uids).values

    # ---- explicit preference state ----
    ps = pd.read_parquet(f"{DATA}/{'validation' if stage=='val' else 'test'}_preference_state.parquet")
    ps = ps[ps.item_id.isin(item_pos.index) & ps.uid.isin(set(tgt_uids))]
    trow = pd.Series(np.arange(n_tgt), index=tgt_uids)
    pr = trow.reindex(ps.uid.values).values
    pc = item_pos.reindex(ps.item_id.values).values
    liked = np.zeros((n_tgt, n_items), np.float32)
    disliked = np.zeros((n_tgt, n_items), np.float32)
    m = (ps.state == "liked").values
    liked[pr[m], pc[m]] = 1.0
    disliked[pr[~m], pc[~m]] = 1.0

    # ---- CF: user-user on cosine-normalised strong history ----
    Xn = X_strong.copy()
    Xn.data = np.log1p(Xn.data).astype(np.float32)
    norms = np.sqrt(Xn.multiply(Xn).sum(1)).A.ravel()
    norms[norms == 0] = 1.0
    Xn = sp.diags((1.0 / norms).astype(np.float32)) @ Xn
    Xn = Xn.astype(np.float32).tocsr()

    Xd = np.asarray(Xn.todense(), dtype=np.float32)
    sim = Xd[tgt_rows] @ Xd.T                      # (n_tgt x n_users)
    log(stage, "user-user sim", sim.shape)
    np.put_along_axis(sim, np.arange(len(tgt_rows))[:, None] * 0 + tgt_rows[:, None], 0.0, axis=1)
    K = 200
    idx = np.argpartition(-sim, K, axis=1)[:, :K]
    top = np.take_along_axis(sim, idx, axis=1)
    top = np.maximum(top, 0) ** 2
    rows = np.repeat(np.arange(n_tgt), K)
    Sk = sp.csr_matrix((top.ravel(), (rows, idx.ravel())), shape=sim.shape)
    del sim
    cf_uu = np.asarray((Sk @ Xn).todense(), dtype=np.float32)
    cf_uu /= cf_uu.max(1, keepdims=True).clip(1e-9)
    log(stage, "user-user CF done")

    # ---- CF: item-item over the same normalised matrix, low-rank ----
    from sklearn.utils.extmath import randomized_svd

    U, S, Vt = randomized_svd(Xn, n_components=192, n_iter=4, random_state=0)
    ii = (U[tgt_rows] * S) @ Vt
    ii /= np.abs(ii).max(1, keepdims=True).clip(1e-9)
    log(stage, "svd done")

    del Xd, Sk
    feats = {
        "pop": np.broadcast_to(pop_l, (n_tgt, n_items)).astype(np.float32),
        "rep": np.log1p(np.asarray(X_strong[tgt_rows].todense(), np.float32)),
        "rec": np.asarray(X_rec[tgt_rows].todense(), np.float32),
        "rec2": np.asarray(X_rec2[tgt_rows].todense(), np.float32),
        "posl": np.log1p(np.asarray(X_pos[tgt_rows].todense(), np.float32)),
        "liked": liked,
        "disliked": disliked,
        "cf_uu": cf_uu,
        "ii": ii.astype(np.float32),
    }
    for k in ("rep", "rec", "rec2", "posl"):
        feats[k] = feats[k] / feats[k].max(1, keepdims=True).clip(1e-9)
    log(stage, "features built")
    return dict(items=items, tgt_uids=tgt_uids, elig=elig, feats=feats)


def val_truth(items, tgt_uids, elig):
    item_pos = pd.Series(np.arange(len(items)), index=items)
    trow = pd.Series(np.arange(len(tgt_uids)), index=tgt_uids)
    grade = np.zeros((len(tgt_uids), len(items)), np.int8)
    vl = pd.read_parquet(f"{DATA}/val_listens.parquet")
    vk = pd.read_parquet(f"{DATA}/val_likes.parquet")
    for df, g in ((vl[(vl.is_organic == 1) & (vl.played_ratio_pct >= 80)], 1),
                  (vk[vk.is_organic == 1], 3)):
        df = df[df.uid.isin(set(tgt_uids)) & df.item_id.isin(item_pos.index)]
        grade[trow.reindex(df.uid.values).values, item_pos.reindex(df.item_id.values).values] = g
    grade[~elig] = 0
    return grade


def ndcg(score, grade, elig, k=10):
    s = np.where(elig, score, -np.inf)
    idx = np.argpartition(-s, k, axis=1)[:, :k]
    ordr = np.take_along_axis(s, idx, axis=1).argsort(axis=1)[:, ::-1]
    idx = np.take_along_axis(idx, ordr, axis=1)
    g = np.take_along_axis(grade, idx, axis=1).astype(np.float64)
    disc = 1.0 / np.log2(np.arange(2, k + 2))
    dcg = ((2.0 ** g - 1) * disc).sum(1)
    gs = -np.sort(-grade, axis=1)[:, :k].astype(np.float64)
    idcg = ((2.0 ** gs - 1) * disc).sum(1)
    m = idcg > 0
    return float(np.mean(dcg[m] / idcg[m])), int(m.sum())


def combine(feats, w):
    out = np.zeros_like(feats["pop"])
    for k, v in w.items():
        if v:
            out += np.float32(v) * feats[k]
    return out


def write_sub(res, score, path):
    elig, tgt_uids, items = res["elig"], res["tgt_uids"], res["items"]
    s = np.where(elig, score, -np.inf)
    idx = np.argpartition(-s, 100, axis=1)[:, :100]
    ordr = np.take_along_axis(s, idx, axis=1).argsort(axis=1)[:, ::-1]
    idx = np.take_along_axis(idx, ordr, axis=1)
    n = len(tgt_uids)
    df = pd.DataFrame({
        "uid": np.repeat(tgt_uids, 100),
        "item_id": items[idx.ravel()],
        "rank": np.tile(np.arange(1, 101), n),
    })
    assert np.take_along_axis(elig, idx, axis=1).all()
    df.to_parquet(path, index=False)
    log("wrote", path)


if __name__ == "__main__":
    np.save("/app/solution/_noop.npy", np.zeros(1))

python -c "
import build as B, numpy as np, itertools, pickle
r=B.build('val')
g=B.val_truth(r['items'],r['tgt_uids'],r['elig'])
B.log('truth users', (g.max(1)>0).sum())
f=r['feats']
for k in f: 
    print(k, B.ndcg(np.ascontiguousarray(f[k]),g,r['elig']))
np.save('/app/solution/_valgrade.npy', g)
" 2>&1 | tail -30

python -c "
import build as B, numpy as np
r=B.build('val')
g=B.val_truth(r['items'],r['tgt_uids'],r['elig'])
B.log('truth users', (g.max(1)>0).sum())
f=r['feats']
for k in f:
    print(k, B.ndcg(np.ascontiguousarray(f[k]),g,r['elig']), flush=True)
np.save('/app/solution/_valgrade.npy', g)
import pickle; pickle.dump(r, open('/app/solution/_val.pkl','wb'), protocol=4)
"; echo EXIT=$?

python - <<'EOF'
import re
s=open('build.py').read()
s=s.replace("""    X_strong = mat(strong)                      # counts of strong organic listens
    X_pos = mat(pos)                            # any >=50% listen
    age = (cutoff - strong.timestamp.values).astype(np.float64)
    X_rec = mat(strong, np.exp(-age / (14 * 86400.0)))      # 14-day half-life-ish
    X_rec2 = mat(strong, np.exp(-age / (3 * 86400.0)))      # short-term
    log(stage, "history matrices", X_strong.nnz, X_pos.nnz)
""","""    X_strong = mat(strong)                      # counts of strong organic listens
    X_pos = mat(pos)                            # any >=50% listen
    age = (cutoff - strong.timestamp.values).astype(np.float64)
    X_rec = mat(strong, np.exp(-age / (14 * 86400.0)))      # 14-day half-life-ish
    X_rec2 = mat(strong, np.exp(-age / (3 * 86400.0)))      # short-term
    log(stage, "history matrices", X_strong.nnz, X_pos.nnz)
    del ev, li, strong, pos
""")
s=s.replace("""    feats = {
        "pop": np.broadcast_to(pop_l, (n_tgt, n_items)).astype(np.float32),
        "rep": np.log1p(np.asarray(X_strong[tgt_rows].todense(), np.float32)),
        "rec": np.asarray(X_rec[tgt_rows].todense(), np.float32),
        "rec2": np.asarray(X_rec2[tgt_rows].todense(), np.float32),
        "posl": np.log1p(np.asarray(X_pos[tgt_rows].todense(), np.float32)),
        "liked": liked,
        "disliked": disliked,
        "cf_uu": cf_uu,
        "ii": ii.astype(np.float32),
    }
    for k in ("rep", "rec", "rec2", "posl"):
        feats[k] = feats[k] / feats[k].max(1, keepdims=True).clip(1e-9)
""","""    feats = {
        "pop": np.broadcast_to(pop_l, (n_tgt, n_items)),
        "rep": np.log1p(np.asarray(X_strong[tgt_rows].todense(), np.float32)),
        "rec": np.asarray(X_rec[tgt_rows].todense(), np.float32),
        "rec2": np.asarray(X_rec2[tgt_rows].todense(), np.float32),
        "posl": np.log1p(np.asarray(X_pos[tgt_rows].todense(), np.float32)),
        "liked": liked,
        "disliked": disliked,
        "cf_uu": cf_uu,
        "ii": ii.astype(np.float32),
    }
    del X_strong, X_pos, X_rec, X_rec2, Xn, U, S, Vt, ii
    for k in ("rep", "rec", "rec2", "posl"):
        feats[k] /= feats[k].max(1, keepdims=True).clip(1e-9)
""")
s=s.replace("""    del Xd, Sk
""","""    del Xd, Sk, U, S, Vt
""").replace("    del X_strong, X_pos, X_rec, X_rec2, Xn, U, S, Vt, ii","    del X_strong, X_pos, X_rec, X_rec2, Xn, ii")
open('build.py','w').write(s)
EOF
grep -n "del " build.py

python - <<'EOF'
s=open('build.py').read()
s=s.replace("""    liked = np.zeros((n_tgt, n_items), np.float32)
    disliked = np.zeros((n_tgt, n_items), np.float32)
    m = (ps.state == "liked").values
    liked[pr[m], pc[m]] = 1.0
    disliked[pr[~m], pc[~m]] = 1.0""",
"""    m = (ps.state == "liked").values
    liked = (pr[m], pc[m])
    disliked = (pr[~m], pc[~m])""")
s=s.replace("""        "rec2": np.asarray(X_rec2[tgt_rows].todense(), np.float32),
        "posl": np.log1p(np.asarray(X_pos[tgt_rows].todense(), np.float32)),
        "liked": liked,
        "disliked": disliked,
        "cf_uu": cf_uu,""","""        "rec2": np.asarray(X_rec2[tgt_rows].todense(), np.float32),
        "cf_uu": cf_uu,""")
s=s.replace('    for k in ("rep", "rec", "rec2", "posl"):','    for k in ("rep", "rec", "rec2"):')
s=s.replace('    return dict(items=items, tgt_uids=tgt_uids, elig=elig, feats=feats)',
            '    return dict(items=items, tgt_uids=tgt_uids, elig=elig, feats=feats,\n                liked=liked, disliked=disliked, pop=pop_l)')
s=s.replace('''def combine(feats, w):
    out = np.zeros_like(feats["pop"])
    for k, v in w.items():
        if v:
            out += np.float32(v) * feats[k]
    return out''','''def combine(res, w, out=None):
    """Linear blend of the dense features plus sparse explicit-state terms."""
    f = res["feats"]
    if out is None:
        out = np.zeros((len(res["tgt_uids"]), len(res["items"])), np.float32)
    else:
        out[:] = 0
    out += np.float32(w.get("pop", 0.0)) * res["pop"]
    for k in ("rep", "rec", "rec2", "cf_uu", "ii"):
        v = w.get(k, 0.0)
        if v:
            out += np.float32(v) * f[k]
    if w.get("liked"):
        np.add.at(out, res["liked"], np.float32(w["liked"]))
    if w.get("disliked"):
        np.add.at(out, res["disliked"], np.float32(w["disliked"]))
    return out''')
open('build.py','w').write(s)
EOF
grep -n "feats = {" -A 12 build.py

/app/solution/tune.py

"""Tune blend weights on the public validation window, then refit for test."""
import gc
import json

import numpy as np

import build as B

res = B.build("val")
grade = B.val_truth(res["items"], res["tgt_uids"], res["elig"])
B.log("labelled users", int((grade.max(1) > 0).sum()))
gc.collect()

buf = np.zeros((len(res["tgt_uids"]), len(res["items"])), np.float32)
disc = 1.0 / np.log2(np.arange(2, 12))
gs = -np.sort(-grade, axis=1)[:, :10].astype(np.float64)
IDCG = ((2.0 ** gs - 1) * disc).sum(1)
MASK = IDCG > 0
del gs


def score_of(w):
    s = B.combine(res, w, out=buf)
    s[~res["elig"]] = -np.inf
    idx = np.argpartition(-s, 10, axis=1)[:, :10]
    ordr = np.take_along_axis(s, idx, axis=1).argsort(axis=1)[:, ::-1]
    idx = np.take_along_axis(idx, ordr, axis=1)
    g = np.take_along_axis(grade, idx, axis=1).astype(np.float64)
    dcg = ((2.0 ** g - 1) * disc).sum(1)
    return float(np.mean(dcg[MASK] / IDCG[MASK]))


for k in ("pop", "rep", "rec", "rec2", "cf_uu", "ii", "liked"):
    B.log("solo", k, round(score_of({k: 1.0}), 5))

w = {"pop": 0.05, "rep": 1.0, "rec": 0.5, "rec2": 0.3, "cf_uu": 0.3, "ii": 0.2,
     "liked": 0.3, "disliked": -0.5}
best = score_of(w)
B.log("init", best, w)
GRID = {
    "pop": [0.0, 0.02, 0.05, 0.1, 0.2, 0.4],
    "rep": [0.0, 0.25, 0.5, 1.0, 2.0],
    "rec": [0.0, 0.25, 0.5, 1.0, 2.0, 4.0],
    "rec2": [0.0, 0.15, 0.3, 0.6, 1.2, 2.5],
    "cf_uu": [0.0, 0.1, 0.2, 0.4, 0.8, 1.6],
    "ii": [0.0, 0.05, 0.1, 0.2, 0.4, 0.8],
    "liked": [0.0, 0.1, 0.2, 0.4, 0.8],
    "disliked": [0.0, -0.2, -0.5, -2.0],
}
for sweep in range(3):
    improved = False
    for k, vals in GRID.items():
        cur = w[k]
        for v in vals:
            if v == cur:
                continue
            w[k] = v
            s = score_of(w)
            if s > best + 1e-6:
                best, cur, improved = s, v, True
            w[k] = cur
        w[k] = cur
    B.log("sweep", sweep, round(best, 5), w)
    if not improved:
        break

json.dump({"weights": w, "val_ndcg@10": best}, open("/app/solution/weights.json", "w"), indent=2)
B.log("BEST", best, w)

s = B.combine(res, w, out=buf)
B.write_sub(res, s, "/app/solution/validation.parquet")
del res, grade, buf, s
gc.collect()

test = B.build("test")
s = B.combine(test, w)
B.write_sub(test, s, "/app/solution/submission.parquet")
B.log("done")

nohup python -u tune.py > tune.log 2>&1 &
echo started; sleep 60; tail -20 tune.log; free -g | head -2

cat /sys/fs/cgroup/memory.max 2>/dev/null || cat /sys/fs/cgroup/memory/memory.limit_in_bytes 2>/dev/null; echo; cat /sys/fs/cgroup/memory.peak 2>/dev/null

python - <<'EOF'
s=open('build.py').read()
s=s.replace('''    for k in ("rep", "rec", "rec2"):
        feats[k] /= feats[k].max(1, keepdims=True).clip(1e-9)''',
'''    for k in ("rep", "rec", "rec2"):
        feats[k] /= feats[k].max(1, keepdims=True).clip(1e-9)
    for k in ("rep", "rec", "rec2", "cf_uu", "ii"):
        feats[k] = feats[k].astype(np.float16)   # 8 GiB cgroup budget
        gc.collect()''')
s=s.replace('''    Xd = np.asarray(Xn.todense(), dtype=np.float32)
    sim = Xd[tgt_rows] @ Xd.T''','''    Xd = np.asarray(Xn.todense(), dtype=np.float32)
    sim = np.ascontiguousarray(Xd[tgt_rows]) @ Xd.T''')
s=s.replace("import sys\nimport time","import gc\nimport time")
s=s.replace('''    U, S, Vt = randomized_svd(Xn, n_components=192, n_iter=4, random_state=0)''',
'''    del Xd
    gc.collect()
    U, S, Vt = randomized_svd(Xn, n_components=192, n_iter=4, random_state=0)''')
s=s.replace("    del Xd, Sk, U, S, Vt","    del Sk, U, S, Vt\n    gc.collect()")
open('build.py','w').write(s)
EOF
sed -i 's/^import json$/import json\nimport gc/' tune.py; python -c "import ast;ast.parse(open('/app/solution/build.py').read())" && echo OK
nohup python -u tune.py > tune.log 2>&1 &
sleep 75; tail -25 tune.log

sleep 50; tail -8 tune.log

pkill -f tune.py; cat > /app/solution/tune2.py <<'EOF'
"""Lean weight search (discovery-only features) + final artifacts."""
import gc, json
import numpy as np
import build as B

def prep(res, grade):
    disc = 1.0 / np.log2(np.arange(2, 12))
    gs = -np.sort(-grade, axis=1)[:, :10].astype(np.float64)
    idcg = ((2.0 ** gs - 1) * disc).sum(1); m = idcg > 0
    del gs; gc.collect()
    return disc, idcg, m

res = B.build("val")
grade = B.val_truth(res["items"], res["tgt_uids"], res["elig"])
disc, IDCG, MASK = prep(res, grade)
buf = np.zeros(grade.shape, np.float32)
NEG = res["elig"] == False

def score_of(w):
    s = B.combine(res, w, out=buf)
    s[NEG] = -np.inf
    idx = np.argpartition(-s, 10, axis=1)[:, :10]
    o = np.take_along_axis(s, idx, axis=1).argsort(axis=1)[:, ::-1]
    g = np.take_along_axis(grade, np.take_along_axis(idx, o, axis=1), axis=1).astype(np.float64)
    return float(np.mean(((2.0 ** g - 1) * disc).sum(1)[MASK] / IDCG[MASK]))

w = {"pop": 0.05, "cf_uu": 0.5, "ii": 0.5, "disliked": -1.0}
best = score_of(w); B.log("init", round(best, 5))
GRID = {"cf_uu": [0.2, 0.35, 0.5, 0.8, 1.2], "ii": [0.2, 0.35, 0.5, 0.8, 1.2],
        "pop": [0.0, 0.02, 0.05, 0.12, 0.25], "disliked": [0.0, -1.0]}
for sweep in range(2):
    for k, vals in GRID.items():
        cur = w[k]
        for v in vals:
            if v == cur: continue
            w[k] = v; s = score_of(w)
            if s > best + 1e-6: best, cur = s, v
            w[k] = cur
        w[k] = cur
    B.log("sweep", sweep, round(best, 5), w)
json.dump({"weights": w, "val_ndcg@10": best}, open("/app/solution/weights.json", "w"), indent=2)
B.log("BEST", best, w)
B.write_sub(res, B.combine(res, w, out=buf), "/app/solution/validation.parquet")
del res, grade, buf, NEG; gc.collect()
test = B.build("test")
B.write_sub(test, B.combine(test, w), "/app/solution/submission.parquet")
B.log("done")
EOF
nohup python -u tune2.py > tune2.log 2>&1 &
sleep 100; tail -12 tune2.log

sleep 45; tail -12 /app/solution/tune2.log

pgrep -af python | head; ls /app/solution/

cat > tune2.py <<'PYEOF'
"""Lean weight search (discovery-only features) + final artifacts."""
import gc, json
import numpy as np
import build as B

res = B.build("val")
grade = B.val_truth(res["items"], res["tgt_uids"], res["elig"])
disc = 1.0 / np.log2(np.arange(2, 12))
gs = -np.sort(-grade, axis=1)[:, :10].astype(np.float64)
IDCG = ((2.0 ** gs - 1) * disc).sum(1); MASK = IDCG > 0
del gs; gc.collect()
buf = np.zeros(grade.shape, np.float32)
NEG = ~res["elig"]

def score_of(w):
    s = B.combine(res, w, out=buf)
    s[NEG] = -np.inf
    idx = np.argpartition(-s, 10, axis=1)[:, :10]
    o = np.take_along_axis(s, idx, axis=1).argsort(axis=1)[:, ::-1]
    g = np.take_along_axis(grade, np.take_along_axis(idx, o, axis=1), axis=1).astype(np.float64)
    return float(np.mean(((2.0 ** g - 1) * disc).sum(1)[MASK] / IDCG[MASK]))

w = {"pop": 0.05, "cf_uu": 0.5, "ii": 0.5, "disliked": -1.0}
best = score_of(w); B.log("init", round(best, 5))
GRID = {"cf_uu": [0.2, 0.35, 0.8, 1.2], "ii": [0.2, 0.35, 0.8, 1.2],
        "pop": [0.0, 0.02, 0.12, 0.25], "disliked": [0.0]}
for sweep in range(2):
    for k, vals in GRID.items():
        cur = w[k]
        for v in vals:
            if v == cur: continue
            w[k] = v; s = score_of(w)
            if s > best + 1e-6: best, cur = s, v
            w[k] = cur
        w[k] = cur
    B.log("sweep", sweep, round(best, 5), w)
json.dump({"weights": w, "val_ndcg@10": best}, open("/app/solution/weights.json", "w"), indent=2)
B.log("BEST", best, w)
B.write_sub(res, B.combine(res, w, out=buf), "/app/solution/validation.parquet")
del res, grade, buf, NEG; gc.collect()
test = B.build("test")
B.write_sub(test, B.combine(test, w), "/app/solution/submission.parquet")
B.log("done")
PYEOF
nohup python -u tune2.py > tune2.log 2>&1 &
sleep 90; tail -12 tune2.log

python - <<'EOF'
s=open('build.py').read()
s=s.replace('''    X_pos = mat(pos)                            # any >=50% listen
    age = (cutoff - strong.timestamp.values).astype(np.float64)
    X_rec = mat(strong, np.exp(-age / (14 * 86400.0)))      # 14-day half-life-ish
    X_rec2 = mat(strong, np.exp(-age / (3 * 86400.0)))      # short-term
    log(stage, "history matrices", X_strong.nnz, X_pos.nnz)
    del ev, li, strong, pos''','''    log(stage, "history matrix", X_strong.nnz)
    del ev, li, strong, pos
    gc.collect()''')
s=s.replace('''        "rep": np.log1p(np.asarray(X_strong[tgt_rows].todense(), np.float32)),
        "rec": np.asarray(X_rec[tgt_rows].todense(), np.float32),
        "rec2": np.asarray(X_rec2[tgt_rows].todense(), np.float32),
        "cf_uu"''','''        "cf_uu"''')
s=s.replace('''    del X_strong, X_pos, X_rec, X_rec2, Xn, ii
    for k in ("rep", "rec", "rec2"):
        feats[k] /= feats[k].max(1, keepdims=True).clip(1e-9)
    for k in ("rep", "rec", "rec2", "cf_uu", "ii"):''','''    del X_strong, Xn, ii
    for k in ("cf_uu", "ii"):''')
s=s.replace('    for k in ("rep", "rec", "rec2", "cf_uu", "ii"):\n        v = w.get','    for k in ("cf_uu", "ii"):\n        v = w.get')
open('build.py','w').write(s)
EOF
grep -n "X_pos\|X_rec\|rep\|rec2" build.py; python -c "import ast;ast.parse(open('build.py').read())" && echo SYNTAX_OK

cat > final.py <<'PYEOF'
"""Fit the discovery ranker and write the hidden-test submission (test first)."""
import gc, json, sys
import numpy as np
import build as B

W = {"pop": 0.05, "cf_uu": 0.5, "ii": 0.5, "disliked": -1.0}
test = B.build("test")
B.write_sub(test, B.combine(test, W), "/app/solution/submission.parquet")
del test; gc.collect()
B.log("submission written")

res = B.build("val")
grade = B.val_truth(res["items"], res["tgt_uids"], res["elig"])
disc = 1.0 / np.log2(np.arange(2, 12))
IDCG = ((2.0 ** (-np.sort(-grade, axis=1)[:, :10].astype(np.float64)) - 1) * disc).sum(1)
MASK = IDCG > 0
gc.collect()
buf = np.zeros(grade.shape, np.float32)
NEG = ~res["elig"]

def score_of(w):
    s = B.combine(res, w, out=buf)
    s[NEG] = -np.inf
    idx = np.argpartition(-s, 10, axis=1)[:, :10]
    o = np.take_along_axis(s, idx, axis=1).argsort(axis=1)[:, ::-1]
    g = np.take_along_axis(grade, np.take_along_axis(idx, o, axis=1), axis=1).astype(np.float64)
    return float(np.mean(((2.0 ** g - 1) * disc).sum(1)[MASK] / IDCG[MASK]))

w = dict(W); best = score_of(w); B.log("init", round(best, 5))
for k, vals in {"cf_uu": [0.2, 0.35, 0.8, 1.5], "ii": [0.2, 0.35, 0.8, 1.5],
                "pop": [0.0, 0.02, 0.12, 0.3]}.items():
    cur = w[k]
    for v in vals:
        w[k] = v; s = score_of(w)
        B.log(" ", k, v, round(s, 5))
        if s > best + 1e-6: best, cur = s, v
        w[k] = cur
    w[k] = cur
json.dump({"weights": w, "val_ndcg@10": best, "baseline_weights": W},
          open("/app/solution/weights.json", "w"), indent=2)
B.log("BEST", best, w)
B.write_sub(res, B.combine(res, w, out=buf), "/app/solution/validation.parquet")
if w != W:
    del res, grade, buf, NEG; gc.collect()
    test = B.build("test")
    B.write_sub(test, B.combine(test, w), "/app/solution/submission.parquet")
B.log("done")
PYEOF
nohup python -u final.py > final.log 2>&1 &
sleep 80; tail -15 final.log; ls -la /app/solution/

/app/solution/README.md

# "Made for You" discovery ranker

`python final.py` builds `submission.parquet` (hidden test) and
`validation.parquet` (public window); `build.py` holds the model.

## Key observation

The supplied candidate eligibility masks out each user's *own* interaction
history (~900 items/user). Measured on the public window, repeat-consumption
features (past strong-listen counts, recency-decayed counts, `liked`
preference state) score NDCG@10 ≈ 7e-5 on their own — they rank items that are
never eligible. This is a pure **discovery** problem, so the model is entirely
collaborative-filtering driven and those features were dropped.

## Model

History = organic listens completed to ≥80 % (the objective's own positive
definition), as a user × item count matrix `X` over the 30 000-track catalog,
`log1p`-damped and L2-row-normalised (`Xn`).

Three scores are blended linearly:

| Feature | Definition |
| --- | --- |
| `cf_uu` | User-user CF: cosine neighbours over `Xn`, self-similarity zeroed, top-200 neighbours, weights squared to sharpen, then propagated back through `Xn`. Row-max normalised. |
| `ii` | Low-rank item-item CF: rank-192 randomised SVD of `Xn`, reconstructed scores. Row-max normalised. |
| `pop` | Global organic strong-listen popularity, `log1p`, max-normalised — a prior for thin users. |

Plus a `disliked` penalty applied to active explicit-dislike pairs from the
stage's preference-state file.

Weights are chosen by coordinate ascent directly on validation NDCG@10
(`weights.json` records the result). Solo validation NDCG@10: `ii` 0.0225,
`cf_uu` 0.0219, `pop` 0.0106 — the blend beats all three.

## Stage handling

`build("val")` uses only `train_*` as history and the validation preference
state / eligibility. `build("test")` refits the identical pipeline on
`train_* + val_*` with the test preference state and test eligibility, so the
validation week is observed history rather than a label source. Recency decay
is anchored to each stage's own cutoff.

## Caveats

Fit under a 30-minute wall clock and an 8 GiB memory cap: dense
(8456 × 30 000) feature planes are stored as float16, which bounds the feature
count. Given more time the obvious next steps are a full item-item cosine with
popularity damping / EASE-style closed form, and a GBDT reranker over a top-300
prefilter trained on a pseudo-validation split carved from the tail of train.

sleep 60; tail -20 final.log

sleep 20; tail -3 final.log && python /task/evaluate_public.py /app/solution/validation.parquet

rm -f tune.py tune2.py tune.log tune2.log _noop.npy && python -c "
import pandas as pd,numpy as np
from ranking_evaluation import CandidateSet, validate_ranking, read_ranking
import sys; sys.path.insert(0,'/task')
d=pd.read_parquet('submission.parquet'); print(d.shape, d.dtypes.to_dict())
t=set(pd.read_parquet('/task/data/target_users.parquet').uid.astype(int))
f=validate_ranking(d,t)
cs=CandidateSet('/task/data/test_eligible_candidates.npz')
el=np.unpackbits(cs.packed,axis=1,bitorder='little')[:,:30000].astype(bool)
row={int(u):r for r,u in enumerate(cs.uids)}; col={int(i):c for c,i in enumerate(cs.item_ids)}
r=np.array([row[u] for u in f.uid]); c=np.array([col[i] for i in f.item_id])
print('all eligible:', el[r,c].all())
print(d.head(3))
" 2>&1 | tail -20; ls -la /app/solution

PYTHONPATH=/task python -c "
import pandas as pd,numpy as np
from ranking_evaluation import CandidateSet, validate_ranking
d=pd.read_parquet('/app/solution/submission.parquet'); print(d.shape, d.dtypes.to_dict())
t=set(pd.read_parquet('/task/data/target_users.parquet').uid.astype(int))
f=validate_ranking(d,t); print('schema OK')
cs=CandidateSet('/task/data/test_eligible_candidates.npz')
el=np.unpackbits(cs.packed,axis=1,bitorder='little')[:,:30000].astype(bool)
row={int(u):r for r,u in enumerate(cs.uids)}; col={int(i):c for c,i in enumerate(cs.item_ids)}
r=np.array([row[u] for u in f.uid]); c=np.array([col[i] for i in f.item_id])
print('all eligible:', bool(el[r,c].all()))
"