Agent trace

ls -la /task/ /task/data/ && cat /task/requirements.txt

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

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))
p=pq.ParquetFile('/task/data/train_listens.parquet'); print('train_listens rows', p.metadata.num_rows, p.schema_arrow)
d=pd.read_parquet('/task/data/val_listens.parquet'); print('val listens', d.shape, d.timestamp.min(), d.timestamp.max())
z=np.load('/task/data/validation_eligible_candidates.npz'); print({k:(v.shape,v.dtype) for k,v in z.items()})
"

free -g; nproc; mkdir -p /app/solution

cd /app/solution && python -c "
import numpy as np, pandas as pd, time
t=time.time()
z=np.load('/task/data/validation_eligible_candidates.npz')
uids=z['uids']; items=z['item_ids']; pk=z['packed_eligible']
E=np.unpackbits(pk,axis=1,count=len(items),bitorder='little').astype(bool)
print('elig per user', E.sum(1).mean(), E.sum(1).min(), E.sum(1).max(), time.time()-t)
tu=pd.read_parquet('/task/data/target_users.parquet')
print((uids==tu.uid.values).all() if len(uids)==len(tu) else 'diff')
# check whether eligible items overlap user's train history
import pyarrow.parquet as pq
"

cat > /app/solution/fallback.py <<'EOF'
import numpy as np, pandas as pd
z=np.load('/task/data/test_eligible_candidates.npz')
uids=z['uids'].astype(np.int64); items=z['item_ids'].astype(np.int64)
E=np.unpackbits(z['packed_eligible'],axis=1,count=len(items),bitorder='little').astype(bool)
cat=pd.read_parquet('/task/data/candidate_catalog.parquet')
pop=cat.set_index('item_id').loc[items,'organic_positive_listens'].to_numpy(float)
order=np.argsort(-pop)
rows=[]
for r,u in enumerate(uids):
    sel=[c for c in order if E[r,c]][:100]
    rows.append(pd.DataFrame({'uid':u,'item_id':items[sel],'rank':np.arange(1,101)}))
pd.concat(rows).to_parquet('/app/solution/submission.parquet',index=False)
print('ok')
EOF
timeout 600 python fallback.py

cat > /app/solution/build.py <<'PYEOF'
"""Discovery ranker: implicit-MF (randomized SVD) + repeat-affinity + popularity blend."""
import sys, time, json
import numpy as np, pandas as pd
import scipy.sparse as sp
from sklearn.utils.extmath import randomized_svd

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

DATA = "/task/data/"
CAT = pd.read_parquet(DATA + "candidate_catalog.parquet")
CAT_ITEMS = np.sort(CAT.item_id.to_numpy(np.int64))
NI = len(CAT_ITEMS)
POPV = CAT.set_index("item_id").loc[CAT_ITEMS, "organic_positive_listens"].to_numpy(np.float64)
TARGETS = pd.read_parquet(DATA + "target_users.parquet").uid.to_numpy(np.int64)

def cat_idx(x):
    x = np.asarray(x, np.int64)
    pos = np.searchsorted(CAT_ITEMS, x)
    pos[pos >= NI] = 0
    return np.where(CAT_ITEMS[pos] == x, pos, -1)

def load_events(name, stage):
    fs = [DATA + f"train_{name}.parquet"] + ([DATA + f"val_{name}.parquet"] if stage == "test" else [])
    return pd.concat([pd.read_parquet(f) for f in fs], ignore_index=True)

def build_stage(stage):
    """stage in {'val','test'}: history = train (+val for test)."""
    cols = ["uid", "timestamp", "item_id", "is_organic", "played_ratio_pct"]
    parts = [pd.read_parquet(DATA + "train_listens.parquet", columns=cols)]
    if stage == "test":
        parts.append(pd.read_parquet(DATA + "val_listens.parquet", columns=cols))
    L = pd.concat(parts, ignore_index=True) if len(parts) > 1 else parts[0]
    log("listens loaded", L.shape)
    ii = cat_idx(L.item_id.to_numpy())
    keep = (ii >= 0) & (L.played_ratio_pct.to_numpy() >= 50)
    uid = L.uid.to_numpy(np.int64)[keep]
    ts = L.timestamp.to_numpy(np.int64)[keep]
    org = L.is_organic.to_numpy(np.uint8)[keep]
    pr = L.played_ratio_pct.to_numpy(np.int32)[keep]
    item = ii[keep]
    del L, ii, keep
    T_END = int(ts.max())
    log("positive listens in catalog", len(uid), "t_end", T_END)

    # ---- global item-user matrix (all users) for implicit MF
    urow, uuniq = pd.factorize(uid, sort=False)
    NU = len(uuniq)
    w = np.ones(len(uid), np.float32)
    R = sp.csr_matrix((w, (urow, item)), shape=(NU, NI), dtype=np.float32)
    R.sum_duplicates()
    R.data = np.log1p(R.data).astype(np.float32)          # count damping
    # bm25-ish: idf on items, l2 on users
    df = np.asarray((R > 0).sum(0)).ravel().astype(np.float64)
    idf = np.log1p(NU / np.maximum(df, 1)).astype(np.float32)
    R = R.multiply(idf[None, :]).tocsr().astype(np.float32)
    nr = np.sqrt(np.asarray(R.multiply(R).sum(1)).ravel()); nr[nr == 0] = 1
    R = sp.diags((1.0 / nr).astype(np.float32)) @ R
    R = R.astype(np.float32).tocsr()
    log("R built", R.shape, R.nnz)

    K = 160
    U_, S_, Vt = randomized_svd(R, n_components=K, n_iter=4, random_state=0)
    V = (Vt.T * S_).astype(np.float32)                     # item factors 30000 x K
    Vn = V / np.maximum(np.linalg.norm(V, axis=1, keepdims=True), 1e-8)
    log("svd done")
    del R, U_, Vt

    # ---- target-user histories
    tset = pd.Series(np.arange(len(TARGETS)), index=TARGETS)
    m = np.isin(uid, TARGETS)
    tu = tset.reindex(uid[m]).to_numpy()
    t_item, t_ts, t_org, t_pr = item[m], ts[m], org[m], pr[m]
    del uid, item, ts, org, pr

    HL = 14 * 86400.0
    age = (T_END - t_ts) / HL
    dec_long = np.exp2(-age).astype(np.float32)
    dec_short = np.exp2(-(T_END - t_ts) / (3 * 86400.0)).astype(np.float32)
    strong = ((t_pr >= 80) & (t_org == 1)).astype(np.float32)
    wgt = (0.4 + 0.6 * (t_pr >= 80)) * (0.5 + 0.5 * (t_org == 1))

    def spmat(vals):
        M = sp.csr_matrix((vals.astype(np.float32), (tu, t_item)),
                          shape=(len(TARGETS), NI), dtype=np.float32)
        M.sum_duplicates()
        return M

    P_long = spmat(dec_long * wgt)
    P_short = spmat(dec_short * wgt)
    CNT = spmat(np.ones(len(tu), np.float32))
    STR = spmat(strong * dec_long)
    log("profiles built")

    # explicit signals
    pref = pd.read_parquet(DATA + ("validation_preference_state.parquet" if stage == "val"
                                   else "test_preference_state.parquet"))
    pi = cat_idx(pref.item_id.to_numpy()); pu = tset.reindex(pref.uid.to_numpy()).to_numpy()
    okp = (pi >= 0) & ~np.isnan(pu)
    pu = pu[okp].astype(np.int64); pi = pi[okp]; pst = pref.state.to_numpy()[okp]
    LIKED = sp.csr_matrix((np.ones((pst == "liked").sum(), np.float32),
                           (pu[pst == "liked"], pi[pst == "liked"])), shape=(len(TARGETS), NI))
    DISL = sp.csr_matrix((np.ones((pst == "disliked").sum(), np.float32),
                          (pu[pst == "disliked"], pi[pst == "disliked"])), shape=(len(TARGETS), NI))
    LIKED.sum_duplicates(); DISL.sum_duplicates()

    likes = load_events("likes", stage)
    li = cat_idx(likes.item_id.to_numpy()); lu = tset.reindex(likes.uid.to_numpy()).to_numpy()
    okl = (li >= 0) & ~np.isnan(lu)
    LIKE_T = sp.csr_matrix((np.exp2(-(T_END - likes.timestamp.to_numpy(np.int64)[okl]) / HL).astype(np.float32),
                            (lu[okl].astype(np.int64), li[okl])), shape=(len(TARGETS), NI))
    LIKE_T.sum_duplicates()

    # profile embeddings (cosine-style user vectors)
    def emb(P):
        X = (P @ Vn).astype(np.float32)
        X /= np.maximum(np.linalg.norm(X, axis=1, keepdims=True), 1e-8)
        return X
    E_long, E_short = emb(P_long), emb(P_short)
    E_like = emb(LIKE_T + P_long * 0.0 + LIKE_T * 0.0) if LIKE_T.nnz else E_long
    log("embeddings built")

    # popularity: global + recent trend
    pop_all = np.log1p(POPV).astype(np.float32)
    pop_all = (pop_all - pop_all.mean()) / pop_all.std()
    return dict(Vn=Vn, E_long=E_long, E_short=E_short, E_like=E_like,
                CNT=CNT, STR=STR, P_long=P_long, LIKED=LIKED, DISL=DISL,
                LIKE_T=LIKE_T, pop=pop_all, T_END=T_END)

def elig_mask(stage):
    z = np.load(DATA + ("validation_eligible_candidates.npz" if stage == "val"
                        else "test_eligible_candidates.npz"))
    uids = z["uids"].astype(np.int64); items = z["item_ids"].astype(np.int64)
    E = np.unpackbits(z["packed_eligible"], axis=1, count=len(items), bitorder="little").astype(bool)
    assert (uids == TARGETS).all()
    col = cat_idx(items)                       # npz item order -> catalog order
    Ec = np.zeros((len(TARGETS), NI), bool)
    Ec[:, col] = E
    return Ec

FEATS = ["cf_long", "cf_short", "cf_like", "cnt", "str", "pop", "liked", "disl"]

def feature_block(F, rows, cols):
    """features for given (row, col) arrays -> dict of 1d arrays"""
    out = {}
    out["cf_long"] = np.einsum("ij,ij->i", F["E_long"][rows], F["Vn"][cols])
    out["cf_short"] = np.einsum("ij,ij->i", F["E_short"][rows], F["Vn"][cols])
    out["cf_like"] = np.einsum("ij,ij->i", F["E_like"][rows], F["Vn"][cols])
    for name, M in (("cnt", F["CNT"]), ("str", F["STR"]), ("liked", F["LIKED"]), ("disl", F["DISL"])):
        v = np.asarray(M[rows, cols]).ravel()
        out[name] = np.log1p(v) if name in ("cnt", "str") else v
    out["pop"] = F["pop"][cols]
    return out

def full_score(F, w, E):
    """dense score matrix (nusers x NI)"""
    S = (F["E_long"] @ F["Vn"].T) * np.float32(w["cf_long"])
    S += (F["E_short"] @ F["Vn"].T) * np.float32(w["cf_short"])
    S += (F["E_like"] @ F["Vn"].T) * np.float32(w["cf_like"])
    S += F["pop"][None, :] * np.float32(w["pop"])
    for name, M in (("cnt", F["CNT"]), ("str", F["STR"])):
        C = M.tocoo()
        S[C.row, C.col] += np.log1p(C.data) * np.float32(w[name])
    for name, M in (("liked", F["LIKED"]), ("disl", F["DISL"])):
        C = M.tocoo()
        S[C.row, C.col] += C.data * np.float32(w[name])
    S[~E] = -1e9
    return S
PYEOF
echo written

cat >> /app/solution/build.py <<'PYEOF'

# ---------------- validation truth + fast NDCG ----------------
def val_gold(E):
    ln = pd.read_parquet(DATA + "val_listens.parquet")
    lk = pd.read_parquet(DATA + "val_likes.parquet")
    tset = pd.Series(np.arange(len(TARGETS)), index=TARGETS)
    G = np.zeros((len(TARGETS), NI), np.int8)
    def add(df, grade, cond):
        d = df[cond]
        r = tset.reindex(d.uid.to_numpy()).to_numpy()
        c = cat_idx(d.item_id.to_numpy())
        ok = (c >= 0) & ~np.isnan(r)
        r = r[ok].astype(np.int64); c = c[ok]
        ok2 = E[r, c]
        G[r[ok2], c[ok2]] = grade
    add(ln, 1, (ln.is_organic == 1) & (ln.played_ratio_pct >= 80))
    add(lk, 3, lk.is_organic == 1)
    return G

def ndcg_from_top(top, G, idcg, scored):
    g = G[np.arange(G.shape[0])[:, None], top].astype(np.float64)
    disc = 1.0 / np.log2(np.arange(2, g.shape[1] + 2))
    dcg = ((2.0 ** g - 1.0) * disc).sum(1)
    return float((dcg[scored] / idcg[scored]).mean())

def idcg_of(G):
    disc = 1.0 / np.log2(np.arange(2, 11))
    out = np.zeros(G.shape[0])
    for r in range(G.shape[0]):
        row = G[r]
        nz = row[row > 0]
        if nz.size:
            nz = np.sort(nz)[::-1][:10].astype(np.float64)
            out[r] = ((2.0 ** nz - 1.0) * disc[:len(nz)]).sum()
    return out
PYEOF
echo ok

cat > /app/solution/run_val.py <<'PYEOF'
import numpy as np, pandas as pd, itertools, json, pickle, time
import sys; sys.path.insert(0,'/app/solution')
from build import *

E = elig_mask("val")
F = build_stage("val")
log("stage built")
G = val_gold(E)
idcg = idcg_of(G); scored = idcg > 0
log("gold users", scored.sum())

nu = len(TARGETS)
# ---- candidate pool: union of tops from each component
def topk_cols(M, k):
    M = np.where(E, M, -1e9)
    return np.argpartition(-M, k, axis=1)[:, :k]
cf_l = F["E_long"] @ F["Vn"].T
pools = [topk_cols(cf_l, 250)]
del cf_l
cf_s = F["E_short"] @ F["Vn"].T; pools.append(topk_cols(cf_s, 120)); del cf_s
cf_k = F["E_like"] @ F["Vn"].T; pools.append(topk_cols(cf_k, 120)); del cf_k
cnt = np.zeros((nu, NI), np.float32); C = F["CNT"].tocoo(); cnt[C.row, C.col] = np.log1p(C.data)
cnt += F["pop"][None,:]*1e-3
pools.append(topk_cols(cnt, 120)); del cnt
pop_top = np.argsort(-F["pop"])[:400]
pools.append(np.where(E[:, pop_top], pop_top[None,:], pop_top[0])[:, :120])
POOL = np.concatenate(pools, axis=1)
log("pool", POOL.shape)
np.save('/app/solution/pool_val.npy', POOL)

rows = np.repeat(np.arange(nu), POOL.shape[1])
cols = POOL.ravel()
FB = feature_block(F, rows, cols)
FB = {k: v.reshape(POOL.shape).astype(np.float32) for k, v in FB.items()}
log("features extracted")

# duplicate mask within pool
srt = np.argsort(POOL, axis=1)
ps = np.take_along_axis(POOL, srt, 1)
dup = np.zeros(POOL.shape, bool)
d = ps[:, 1:] == ps[:, :-1]
np.put_along_axis(dup, srt[:, 1:], d, 1)
elig_pool = E[np.arange(nu)[:,None], POOL] & ~dup
log("dup masked")

def evalw(w):
    S = np.zeros(POOL.shape, np.float32)
    for k, v in w.items(): S += FB[k] * np.float32(v)
    S[~elig_pool] = -1e9
    top = np.argpartition(-S, 10, axis=1)[:, :10]
    tv = np.take_along_axis(S, top, 1)
    order = np.argsort(-tv, axis=1)
    top = np.take_along_axis(top, order, 1)
    return ndcg_from_top(np.take_along_axis(POOL, top, 1), G, idcg, scored), S

W0 = dict(cf_long=1.0, cf_short=0.3, cf_like=0.3, cnt=0.3, str=0.3, pop=0.05, liked=0.0, disl=-1.0)
base, _ = evalw(W0); log("base ndcg", base)
pickle.dump(dict(FB=FB, elig_pool=elig_pool, POOL=POOL, G=G, idcg=idcg, scored=scored),
            open('/app/solution/valtune.pkl','wb'), protocol=4)
log("saved tune data")
PYEOF
timeout 900 python run_val.py 2>&1 | tail -30

python - <<'EOF'
import re
s=open('build.py').read()
s=s.replace("    disc = 1.0 / np.log2(np.arange(2, 11))\n    out","    disc = 1.0 / np.log2(np.arange(2, 12))\n    out")
open('build.py','w').write(s)
EOF
timeout 900 python run_val.py 2>&1 | tail -20

cat > /app/solution/run_val.py <<'PYEOF'
import numpy as np, pandas as pd, pickle, sys
sys.path.insert(0,'/app/solution')
from build import *

E = elig_mask("val")
F = build_stage("val")
G = val_gold(E); idcg = idcg_of(G); scored = idcg > 0
log("gold users", scored.sum())
nu = len(TARGETS)
CNTd = F["CNT"].tocsr(); STRd = F["STR"].tocsr()
pop_top = np.argsort(-F["pop"])[:200]
POOLS = []
CH = 1000
for s in range(0, nu, CH):
    e = min(s + CH, nu)
    sl = slice(s, e)
    Eb = E[sl]
    def tk(M, k):
        M = np.where(Eb, M, -1e9)
        return np.argpartition(-M, k, axis=1)[:, :k]
    parts = []
    for emb_, k in (("E_long", 250), ("E_short", 120), ("E_like", 120)):
        parts.append(tk(F[emb_][sl] @ F["Vn"].T, k))
    c = np.zeros((e - s, NI), np.float32)
    C = CNTd[sl].tocoo(); c[C.row, C.col] = np.log1p(C.data)
    C = STRd[sl].tocoo(); c[C.row, C.col] += np.log1p(C.data)
    c += F["pop"][None, :] * 1e-3
    parts.append(tk(c, 120)); del c
    parts.append(np.broadcast_to(pop_top[None, :120], (e - s, 120)))
    POOLS.append(np.concatenate(parts, 1))
POOL = np.concatenate(POOLS, 0).astype(np.int32)
log("pool", POOL.shape)

rows = np.repeat(np.arange(nu), POOL.shape[1]); cols = POOL.ravel()
FB = {k: v.reshape(POOL.shape).astype(np.float32) for k, v in feature_block(F, rows, cols).items()}
log("features")
srt = np.argsort(POOL, 1); ps = np.take_along_axis(POOL, srt, 1)
dup = np.zeros(POOL.shape, bool)
np.put_along_axis(dup, srt[:, 1:], ps[:, 1:] == ps[:, :-1], 1)
elig_pool = E[np.arange(nu)[:, None], POOL] & ~dup
pickle.dump(dict(FB=FB, elig_pool=elig_pool, POOL=POOL, G=G, idcg=idcg, scored=scored),
            open('/app/solution/valtune.pkl','wb'), protocol=4)
log("saved")

def evalw(w):
    S = np.zeros(POOL.shape, np.float32)
    for k, v in w.items(): S += FB[k] * np.float32(v)
    S[~elig_pool] = -1e9
    top = np.argpartition(-S, 10, 1)[:, :10]
    tv = np.take_along_axis(S, top, 1)
    top = np.take_along_axis(top, np.argsort(-tv, 1), 1)
    return ndcg_from_top(np.take_along_axis(POOL, top, 1), G, idcg, scored)

W = dict(cf_long=1.0, cf_short=0.3, cf_like=0.3, cnt=0.3, str=0.3, pop=0.05, liked=0.0, disl=-1.0)
best = evalw(W); log("base", round(best, 5), W)
grid = {"cf_long":[0,.5,1,2], "cf_short":[0,.1,.3,.6,1.2], "cf_like":[0,.1,.3,.6,1.2],
        "cnt":[0,.05,.15,.3,.6,1.2], "str":[0,.05,.15,.3,.6,1.2], "pop":[0,.02,.05,.1,.2],
        "liked":[-1,0,.5], "disl":[-2,0]}
for it in range(3):
    for k, vals in grid.items():
        cur = W[k]; bv = cur
        for v in vals:
            if v == cur: continue
            W[k] = v; s = evalw(W)
            if s > best: best, bv = s, v
        W[k] = bv
    log("iter", it, round(best, 5), W)
json.dump({"w": W, "ndcg": best}, open('/app/solution/best_weights.json','w'))
log("DONE", best, W)
PYEOF
timeout 900 python run_val.py 2>&1 | tail -20; echo "EXIT=$?"

cat > /app/solution/pipeline.py <<'PYEOF'
import numpy as np, pandas as pd, pickle, sys, json
sys.path.insert(0,'/app/solution')
from build import *

def feats_for_pool(F, POOL, E):
    nu, P = POOL.shape
    out = {k: np.zeros((nu, P), np.float32) for k in FEATS}
    CH = 400
    for s in range(0, nu, CH):
        e = min(s + CH, nu); pl = POOL[s:e]
        for name, key in (("cf_long","E_long"), ("cf_short","E_short"), ("cf_like","E_like")):
            out[name][s:e] = np.einsum("ij,ikj->ik", F[key][s:e], F["Vn"][pl])
        for name, M in (("cnt", F["CNT"]), ("str", F["STR"]), ("liked", F["LIKED"]), ("disl", F["DISL"])):
            d = np.asarray(M[s:e].todense())
            v = np.take_along_axis(d, pl, 1)
            out[name][s:e] = np.log1p(v) if name in ("cnt", "str") else v
        out["pop"][s:e] = F["pop"][pl]
    return out

def make_pool(F, E, sizes=(250,120,120,120,120)):
    nu = len(TARGETS); pop_top = np.argsort(-F["pop"])[:200]
    CNTd, STRd = F["CNT"].tocsr(), F["STR"].tocsr()
    POOLS = []
    for s in range(0, nu, 1000):
        e = min(s + 1000, nu); sl = slice(s, e); Eb = E[sl]
        def tk(M, k):
            M = np.where(Eb, M, -1e9)
            return np.argpartition(-M, k, 1)[:, :k]
        parts = [tk(F[k_][sl] @ F["Vn"].T, n) for k_, n in
                 (("E_long", sizes[0]), ("E_short", sizes[1]), ("E_like", sizes[2]))]
        c = np.zeros((e - s, NI), np.float32)
        C = CNTd[sl].tocoo(); c[C.row, C.col] = np.log1p(C.data)
        C = STRd[sl].tocoo(); c[C.row, C.col] += np.log1p(C.data)
        c += F["pop"][None, :] * 1e-3
        parts.append(tk(c, sizes[3])); del c
        parts.append(np.broadcast_to(pop_top[None, :sizes[4]], (e - s, sizes[4])))
        POOLS.append(np.concatenate(parts, 1))
    return np.concatenate(POOLS, 0).astype(np.int32)

def dedup_mask(POOL, E):
    nu = POOL.shape[0]
    srt = np.argsort(POOL, 1); ps = np.take_along_axis(POOL, srt, 1)
    dup = np.zeros(POOL.shape, bool)
    np.put_along_axis(dup, srt[:, 1:], ps[:, 1:] == ps[:, :-1], 1)
    return E[np.arange(nu)[:, None], POOL] & ~dup

# ---------- validation: tune ----------
Ev = elig_mask("val"); Fv = build_stage("val")
G = val_gold(Ev); idcg = idcg_of(G); scored = idcg > 0
POOL = make_pool(Fv, Ev); log("pool", POOL.shape)
FB = feats_for_pool(Fv, POOL, Ev); log("feats")
ep = dedup_mask(POOL, Ev)

def evalw(w, K=10):
    S = np.zeros(POOL.shape, np.float32)
    for k, v in w.items(): S += FB[k] * np.float32(v)
    S[~ep] = -1e9
    top = np.argpartition(-S, K, 1)[:, :K]
    top = np.take_along_axis(top, np.argsort(-np.take_along_axis(S, top, 1), 1), 1)
    return ndcg_from_top(np.take_along_axis(POOL, top, 1), G, idcg, scored)

W = dict(cf_long=1.0, cf_short=0.3, cf_like=0.3, cnt=0.3, str=0.3, pop=0.05, liked=0.0, disl=-1.0)
best = evalw(W); log("base", round(best, 5))
grid = {"cf_long":[0,.5,1,2,4], "cf_short":[0,.1,.3,.6,1.2], "cf_like":[0,.1,.3,.6,1.2],
        "cnt":[0,.05,.15,.3,.6,1.2,2.5], "str":[0,.05,.15,.3,.6,1.2,2.5],
        "pop":[0,.02,.05,.1,.25,.5], "liked":[-1,0,.5,1.5], "disl":[-2,0,.5]}
for it in range(3):
    改 = False
    for k, vals in grid.items():
        cur = W[k]; bv = cur
        for v in vals:
            if v == cur: continue
            W[k] = v; s = evalw(W)
            if s > best + 1e-6: best, bv, 改 = s, v, True
        W[k] = bv
    log("iter", it, round(best, 5), W)
    if not 改: break
json.dump({"w": W, "val_ndcg": best}, open('/app/solution/best_weights.json','w'), indent=1)
log("TUNED", best, W)

# validation artifact for the public evaluator
def write_ranking(F, E, W, path):
    nu = len(TARGETS); rows = []
    CNTc, STRc, LKc, DSc = F["CNT"].tocsr(), F["STR"].tocsr(), F["LIKED"].tocsr(), F["DISL"].tocsr()
    for s in range(0, nu, 1000):
        e = min(s + 1000, nu); sl = slice(s, e)
        S = (F["E_long"][sl] @ F["Vn"].T) * np.float32(W["cf_long"])
        S += (F["E_short"][sl] @ F["Vn"].T) * np.float32(W["cf_short"])
        S += (F["E_like"][sl] @ F["Vn"].T) * np.float32(W["cf_like"])
        S += F["pop"][None, :] * np.float32(W["pop"])
        for name, M in (("cnt", CNTc), ("str", STRc)):
            C = M[sl].tocoo(); S[C.row, C.col] += np.log1p(C.data) * np.float32(W[name])
        for name, M in (("liked", LKc), ("disl", DSc)):
            C = M[sl].tocoo(); S[C.row, C.col] += C.data.astype(np.float32) * np.float32(W[name])
        S[~E[sl]] = -1e9
        top = np.argpartition(-S, 100, 1)[:, :100]
        top = np.take_along_axis(top, np.argsort(-np.take_along_axis(S, top, 1), 1), 1)
        rows.append(pd.DataFrame({
            "uid": np.repeat(TARGETS[sl], 100),
            "item_id": CAT_ITEMS[top.ravel()],
            "rank": np.tile(np.arange(1, 101), e - s)}))
    pd.concat(rows, ignore_index=True).to_parquet(path, index=False)
    log("wrote", path)

write_ranking(Fv, Ev, W, '/app/solution/validation.parquet')
del Fv, FB, POOL, ep, G, Ev

# ---------- test: refit on train+val ----------
Et = elig_mask("test"); Ft = build_stage("test")
write_ranking(Ft, Et, W, '/app/solution/submission_new.parquet')
PYEOF
nohup timeout 1500 python /app/solution/pipeline.py > /app/solution/pipeline.log 2>&1 &
sleep 60; tail -5 /app/solution/pipeline.log

cat /app/solution/pipeline.log; ls -la /app/solution/

timeout 400 python /task/evaluate_public.py validation.parquet; tail -3 pipeline.log

timeout 300 python -c "
import numpy as np, pandas as pd, sys
sys.path.insert(0,'/task')
from ranking_evaluation import validate_ranking
d=pd.read_parquet('submission_new.parquet')
t=set(pd.read_parquet('/task/data/target_users.parquet').uid.astype(int))
n=validate_ranking(d,t); print('schema ok',n.shape)
z=np.load('/task/data/test_eligible_candidates.npz')
uids=z['uids'].astype(np.int64); items=z['item_ids'].astype(np.int64)
E=np.unpackbits(z['packed_eligible'],axis=1,count=len(items),bitorder='little').astype(bool)
ur={int(u):i for i,u in enumerate(uids)}; ic={int(i):c for c,i in enumerate(items)}
r=n.uid.map(ur).to_numpy(); c=n.item_id.map(ic).to_numpy()
assert not pd.isna(c).any(); print('all eligible:', E[r,c.astype(int)].all())
" && cp submission_new.parquet submission.parquet && echo SWAPPED

cat > /app/solution/README.md <<'EOF'
# Made-for-You discovery ranker

Final artifact: `submission.parquet` (8,456 users x 100 ranked eligible items).
Validation-stage ranking: `validation.parquet` — public evaluator score
**quality_ndcg@10 = 0.023055** over 4,263 scored users.

## Approach

A linear blend of collaborative-filtering and behavioural features, with blend
weights tuned directly on the public validation window.

1. **History construction** (`build.py:build_stage`). Positive listens
   (`played_ratio_pct >= 50`) restricted to the 30k candidate catalog. For the
   validation stage history = `train_*` only; for the hidden-test stage the
   whole pipeline is refit with history = `train_* + val_*` (the same code path,
   `stage="test"`), so validation events are used as observed history only.
2. **Implicit MF item factors.** User x item matrix with `log1p` count damping,
   item IDF weighting and L2 user normalisation, factorised by randomized SVD
   (k=160). Item factors are L2-normalised so CF scores are cosine-like and do
   not simply re-rank by popularity.
3. **User profiles / features** scored per (user, candidate):
   - `cf_long`, `cf_short`: fold-in profile embeddings over the user's positive
     listens with 14-day and 3-day half-life decay, weighted by completion
     (>=80%) and organic flag, dotted with item factors.
   - `cf_like`: same fold-in over the user's decayed explicit likes.
   - `cnt`, `str`: log repeat-affinity — the user's own past listen count and
     recency-decayed strong-organic-listen count for that exact track.
   - `pop`: standardised `log1p(organic_positive_listens)` catalog prior.
   - `liked`, `disl`: active explicit preference state at the correct cutoff
     (`validation_preference_state` / `test_preference_state`).
4. **Weight tuning.** A generous per-user candidate pool (730 = union of top-N
   by each individual signal) is materialised; coordinate-ascent over a weight
   grid maximises graded NDCG@10 directly on the validation window. The internal
   scorer reproduces `evaluate_public.py` exactly (0.023055 both ways).
   Selected weights are in `best_weights.json`; the short-horizon CF term plus
   repeat-affinity dominate, dislikes are strongly penalised.
5. **Final ranking.** Scores are recomputed densely in user chunks, ineligible
   candidates masked to `-inf`, top-100 emitted per user. Every submitted item
   was verified present in `test_eligible_candidates.npz`.

## Files
- `build.py` — data loading, feature/profile construction, eligibility, NDCG.
- `pipeline.py` — end-to-end: tune on validation, refit on train+val, write both
  rankings. Run: `python pipeline.py`.
- `fallback.py` — popularity-only safety submission (superseded).

## Notes / limits
Only the checked-in environment (numpy/scipy/sklearn/pandas) was available, so
no GBDT ranker; with more time the natural next steps are a full item-item
cosine KNN term (feasible here — only ~9.1k distinct users) and a learned
pointwise ranker trained on an in-train temporal split instead of a hand-tuned
linear blend.
EOF
rm -f /app/solution/run_val.py /app/solution/pool_val.npy; ls /app/solution