Agent trace

ls -la /task/ /task/data/ && head -50 /task/requirements.txt

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

mkdir -p solution && python -c "
import pandas as pd, numpy as np, time
t=time.time()
tu=pd.read_parquet('/task/data/target_users.parquet'); print('targets',tu.shape, tu.dtypes.to_dict(), tu.head())
cc=pd.read_parquet('/task/data/candidate_catalog.parquet'); print('cat',cc.shape, cc.columns.tolist()); print(cc.head())
ps=pd.read_parquet('/task/data/validation_preference_state.parquet'); print('prefstate',ps.shape, ps.columns.tolist()); print(ps.head()); print(ps.state.value_counts() if 'state' in ps else '')
z=np.load('/task/data/validation_eligible_candidates.npz')
print({k:(v.shape,v.dtype) for k,v in z.items()})
print(time.time()-t)
"

python -c "
import pyarrow.parquet as pq
for f in ['train_listens','val_listens','train_likes','val_likes','train_dislikes']:
    p=pq.ParquetFile('/task/data/%s.parquet'%f)
    print(f, p.metadata.num_rows, p.schema_arrow)
"

free -g; nproc; python -c "import torch; print(torch.__version__, torch.get_num_threads())"

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']; packed=z['packed_eligible']
E=np.unpackbits(packed,axis=1,count=len(items),bitorder='little').astype(bool)
print('elig shape',E.shape,'mean per user',E.sum(1).mean(), E.sum(1).min(), E.sum(1).max())
zt=np.load('/task/data/test_eligible_candidates.npz')
Et=np.unpackbits(zt['packed_eligible'],axis=1,count=len(zt['item_ids']),bitorder='little').astype(bool)
print('test elig',Et.sum(1).mean())
print('same item order', (zt['item_ids']==items).all(), (zt['uids']==uids).all())
print(time.time()-t)
np.save('elig_val.npy',E); np.save('elig_test.npy',Et); np.save('cand_items.npy',items); np.save('cand_uids.npy',uids)
"

cat > prep.py <<'EOF'
import numpy as np, pandas as pd, pyarrow.parquet as pq, time
t0=time.time()
items=np.load('cand_items.npy'); uids=np.load('cand_uids.npy')
NI=len(items)
maxitem=int(items.max())+1
imap=np.full(maxitem,-1,np.int32); imap[items]=np.arange(NI,dtype=np.int32)
maxuid=None
tu=pd.read_parquet('/task/data/target_users.parquet')
tset=set(tu.uid.astype(np.int64))

def load_listens(path):
    f=pq.ParquetFile(path)
    outs=[]
    for i in range(f.num_row_groups):
        tb=f.read_row_group(i,columns=['uid','timestamp','item_id','is_organic','played_ratio_pct'])
        d=tb.to_pandas()
        it=d.item_id.to_numpy()
        col=np.where(it<maxitem, imap[np.minimum(it,maxitem-1)], -1)
        m=(col>=0)&(d.played_ratio_pct.to_numpy()>=50)
        d=d[m]; col=col[m]
        outs.append(pd.DataFrame({'uid':d.uid.to_numpy(),'ts':d.timestamp.to_numpy(),'col':col,
                                  'org':d.is_organic.to_numpy(),'strong':(d.played_ratio_pct.to_numpy()>=80).astype(np.uint8)}))
    return pd.concat(outs,ignore_index=True)

L=load_listens('/task/data/train_listens.parquet')
print('train listens kept',len(L), time.time()-t0)
LV=load_listens('/task/data/val_listens.parquet')
print('val listens kept',len(LV))

def load_ev(path):
    d=pd.read_parquet(path,columns=['uid','timestamp','item_id','is_organic'])
    it=d.item_id.to_numpy(); col=np.where(it<maxitem, imap[np.minimum(it,maxitem-1)],-1)
    m=col>=0
    return pd.DataFrame({'uid':d.uid.to_numpy()[m],'ts':d.timestamp.to_numpy()[m],'col':col[m],'org':d.is_organic.to_numpy()[m]})
K=load_ev('/task/data/train_likes.parquet'); KV=load_ev('/task/data/val_likes.parquet')
D=load_ev('/task/data/train_dislikes.parquet'); DV=load_ev('/task/data/val_dislikes.parquet')
print('likes',len(K),len(KV),'dislikes',len(D),len(DV))
for name,df in [('L',L),('LV',LV),('K',K),('KV',KV),('D',D),('DV',DV)]:
    df.to_parquet(name+'.parquet',index=False)
print('ts range train',L.ts.min(),L.ts.max(),'val',LV.ts.min(),LV.ts.max())
print('nusers train',L.uid.nunique())
print('done',time.time()-t0)
EOF
python prep.py

python -c "
import numpy as np,pandas as pd
L=pd.read_parquet('L.parquet'); LV=pd.read_parquet('LV.parquet'); K=pd.read_parquet('K.parquet'); KV=pd.read_parquet('KV.parquet')
uids=np.load('cand_uids.npy'); E=np.load('elig_val.npy')
urow={int(u):i for i,u in enumerate(uids)}
# val relevance
rel=pd.concat([LV[(LV.org==1)&(LV.strong==1)][['uid','col']].assign(g=1), KV[KV.org==1][['uid','col']].assign(g=3)])
rel=rel.groupby(['uid','col']).g.max().reset_index()
rel['r']=rel.uid.map(urow)
rel=rel[rel.r.notna()]; rel['r']=rel.r.astype(int)
m=E[rel.r.to_numpy(),rel.col.to_numpy()]
print('relevant pairs',len(rel),'eligible frac',m.mean())
rel2=rel[m]
print('users with rel',rel2.uid.nunique(),'avg rel per user',len(rel2)/rel2.uid.nunique())
# how many relevant items were in train history
hist=set(map(tuple,L[['uid','col']].drop_duplicates().to_numpy()))
pairs=list(map(tuple,rel2[['uid','col']].to_numpy()))
print('frac relevant in train listen history', np.mean([p in hist for p in pairs]))
# eligibility: is history item eligible?
sub=L[['uid','col']].drop_duplicates().sample(20000,random_state=0)
sub['r']=sub.uid.map(urow); sub=sub[sub.r.notna()]
print('frac of history items eligible', E[sub.r.astype(int).to_numpy(),sub.col.to_numpy()].mean())
ps=pd.read_parquet('/task/data/validation_preference_state.parquet')
print(ps.shape)
"

python -c "
import pandas as pd,numpy as np
K=pd.read_parquet('K.parquet')
ps=pd.read_parquet('/task/data/validation_preference_state.parquet')
print(ps.state.value_counts())
items=np.load('cand_items.npy'); mx=int(items.max())+1
imap=np.full(mx,-1,np.int32); imap[items]=np.arange(len(items))
it=ps.item_id.to_numpy(); col=np.where(it<mx,imap[np.minimum(it,mx-1)],-1)
print('pref-state in catalog frac',(col>=0).mean())
psl=ps[(ps.state=='liked')&(col>=0)]
a=set(map(tuple,psl[['uid','item_id']].to_numpy()))
kk=K[K.ts<24786800]
print('like events distinct',len(kk[['uid','col']].drop_duplicates()),'active liked in catalog',len(a))
"

cat > build_features.py <<'PYEOF'
"""Build history-pool features and training targets at multiple time cutoffs."""
import numpy as np, pandas as pd, time, pickle

CUT = 24786800
DAY = 86400
WEEK = 7*DAY
NI = 30000

def _load_ev(path, imap, mx):
    d = pd.read_parquet(path, columns=['uid','timestamp','item_id'])
    it = d.item_id.to_numpy(); col = np.where(it<mx, imap[np.minimum(it,mx-1)], -1)
    m = col>=0
    return pd.DataFrame({'uid':d.uid.to_numpy()[m],'ts':d.timestamp.to_numpy()[m],'col':col[m]})

def main():
    t0=time.time()
    items=np.load('cand_items.npy'); mx=int(items.max())+1
    imap=np.full(mx,-1,np.int32); imap[items]=np.arange(NI,dtype=np.int32)

    L=pd.concat([pd.read_parquet('L.parquet'),pd.read_parquet('LV.parquet')],ignore_index=True)
    K=pd.concat([pd.read_parquet('K.parquet'),pd.read_parquet('KV.parquet')],ignore_index=True)
    D=pd.concat([pd.read_parquet('D.parquet'),pd.read_parquet('DV.parquet')],ignore_index=True)
    UL=pd.concat([_load_ev('/task/data/train_unlikes.parquet',imap,mx),_load_ev('/task/data/val_unlikes.parquet',imap,mx)],ignore_index=True)
    UD=pd.concat([_load_ev('/task/data/train_undislikes.parquet',imap,mx),_load_ev('/task/data/val_undislikes.parquet',imap,mx)],ignore_index=True)
    L=L.sort_values('ts',kind='stable').reset_index(drop=True)
    for df in (K,D,UL,UD):
        df.sort_values('ts',kind='stable',inplace=True); df.reset_index(drop=True,inplace=True)

    uids=np.load('cand_uids.npy').astype(np.int64)
    urow=pd.Series(np.arange(len(uids)),index=uids)
    NU=len(uids)
    # map every event to target-user row (-1 if not a target user)
    for df in (L,K,D,UL,UD):
        df['r']=df.uid.map(urow).fillna(-1).astype(np.int32)

    Lts=L.ts.to_numpy(); Kts=K.ts.to_numpy(); Dts=D.ts.to_numpy()
    ULts=UL.ts.to_numpy(); UDts=UD.ts.to_numpy()
    np.save('pop_all.npy', np.bincount(L.col.to_numpy(),minlength=NI).astype(np.float64))
    return dict(L=L,K=K,D=D,UL=UL,UD=UD,uids=uids,urow=urow,NU=NU,
                Lts=Lts,Kts=Kts,Dts=Dts,ULts=ULts,UDts=UDts,t0=t0)
PYEOF
echo ok

cat > /app/solution/twotower.py <<'PYEOF'
"""Two-tower PyTorch ranker for the "Made for You" discovery shelf.

User tower: pooled interaction-history embeddings -> MLP -> user vector.
Item tower: item-ID embedding -> MLP -> item vector (+ learned bias slot).
Score = dot(user_vec, item_vec); item vectors are precomputable offline.
No global user-item matrix is ever formed: training streams (history, target)
pairs and uses sampled-softmax with logQ correction, so the recipe scales to
millions of users/items.
"""
from __future__ import annotations
import argparse, time, sys
import numpy as np, pandas as pd
import torch, torch.nn as nn

CUT = 24786800
DAY, WEEK = 86400, 7 * 86400
NI = 30000
POOLS = ("r7", "r30", "all", "like", "dislike")
CAPS = {"r7": 96, "r30": 160, "all": 256, "like": 192, "dislike": 64}
DIM = 64
DATA = "/task/data"


# ---------------------------------------------------------------- data loading
def load_events():
    items = np.load("cand_items.npy")
    mx = int(items.max()) + 1
    imap = np.full(mx, -1, np.int32)
    imap[items] = np.arange(NI, dtype=np.int32)

    def ev(name):
        d = pd.read_parquet(f"{DATA}/{name}.parquet", columns=["uid", "timestamp", "item_id"])
        it = d.item_id.to_numpy()
        col = np.where(it < mx, imap[np.minimum(it, mx - 1)], -1)
        m = col >= 0
        return pd.DataFrame({"uid": d.uid.to_numpy()[m], "ts": d.timestamp.to_numpy()[m],
                             "col": col[m]})

    L = pd.concat([pd.read_parquet("L.parquet"), pd.read_parquet("LV.parquet")], ignore_index=True)
    K = pd.concat([pd.read_parquet("K.parquet"), pd.read_parquet("KV.parquet")], ignore_index=True)
    D = pd.concat([pd.read_parquet("D.parquet"), pd.read_parquet("DV.parquet")], ignore_index=True)
    UL = pd.concat([ev("train_unlikes"), ev("val_unlikes")], ignore_index=True)
    UD = pd.concat([ev("train_undislikes"), ev("val_undislikes")], ignore_index=True)

    uids = np.load("cand_uids.npy").astype(np.int64)
    urow = pd.Series(np.arange(len(uids)), index=uids)
    out = {}
    for name, df in [("L", L), ("K", K), ("D", D), ("UL", UL), ("UD", UD)]:
        df = df.copy()
        df["r"] = df.uid.map(urow).fillna(-1).astype(np.int64)
        df = df[df.r >= 0].sort_values("ts", kind="stable").reset_index(drop=True)
        out[name] = df
    out["uids"] = uids
    return out


# ------------------------------------------------------------------- pooling
def _csr(r, c, w, NU, cap):
    """Top-`cap` weighted items per user as a normalized CSR pool."""
    if len(r) == 0:
        return np.zeros(NU + 1, np.int64), np.zeros(0, np.int64), np.zeros(0, np.float32)
    order = np.lexsort((-w, r))
    r, c, w = r[order], c[order], w[order]
    counts = np.bincount(r, minlength=NU)
    starts = np.concatenate([[0], np.cumsum(counts)])
    pos = np.arange(len(r)) - starts[r]
    keep = pos < cap
    r, c, w = r[keep], c[keep], w[keep]
    counts = np.bincount(r, minlength=NU)
    indptr = np.concatenate([[0], np.cumsum(counts)]).astype(np.int64)
    tot = np.bincount(r, weights=w, minlength=NU)
    w = (w / np.maximum(tot[r], 1e-9)).astype(np.float32)
    return indptr, c.astype(np.int64), w


def _agg(r, c, w, NU, cap):
    key = r.astype(np.int64) * NI + c
    uk, inv = np.unique(key, return_inverse=True)
    ws = np.log1p(np.bincount(inv, weights=w))
    return _csr((uk // NI).astype(np.int64), (uk % NI).astype(np.int64), ws, NU, cap)


def active_state(pos, neg, cutoff):
    """Pairs whose most recent (set/unset) event before `cutoff` was a set."""
    p = pos[pos.ts < cutoff]
    n = neg[neg.ts < cutoff]
    kp = p.r.to_numpy() * NI + p.col.to_numpy()
    kn = n.r.to_numpy() * NI + n.col.to_numpy()
    last_p = pd.Series(p.ts.to_numpy(), index=kp).groupby(level=0).max()
    if len(kn):
        last_n = pd.Series(n.ts.to_numpy(), index=kn).groupby(level=0).max()
        al = last_n.reindex(last_p.index).to_numpy()
        keep = np.isnan(al) | (last_p.to_numpy() > np.nan_to_num(al, nan=-1))
        last_p = last_p[keep]
    k = last_p.index.to_numpy()
    return (k // NI).astype(np.int64), (k % NI).astype(np.int64)


def build_features(E, cutoff, NU):
    """User-tower inputs and the seen-item exclusion set at `cutoff`."""
    L, K, D, UL, UD = E["L"], E["K"], E["D"], E["UL"], E["UD"]
    Lh = L[L.ts < cutoff]
    r, c = Lh.r.to_numpy(), Lh.col.to_numpy()
    w = 1.0 + Lh.strong.to_numpy() * Lh.org.to_numpy()
    ts = Lh.ts.to_numpy()
    pools = {}
    m30, m7 = ts >= cutoff - 30 * DAY, ts >= cutoff - WEEK
    pools["all"] = _agg(r, c, w, NU, CAPS["all"])
    pools["r30"] = _agg(r[m30], c[m30], w[m30], NU, CAPS["r30"])
    pools["r7"] = _agg(r[m7], c[m7], w[m7], NU, CAPS["r7"])
    lr, lc = active_state(K, UL, cutoff)
    pools["like"] = _agg(lr, lc, np.ones(len(lr)), NU, CAPS["like"])
    dr, dc = active_state(D, UD, cutoff)
    pools["dislike"] = _agg(dr, dc, np.ones(len(dr)), NU, CAPS["dislike"])

    scal = np.zeros((NU, 4), np.float32)
    scal[:, 0] = np.log1p(np.bincount(r, minlength=NU))
    scal[:, 1] = np.log1p(np.bincount(r[m30], minlength=NU))
    scal[:, 2] = np.log1p(np.bincount(lr, minlength=NU))
    scal[:, 3] = np.log1p(np.bincount(dr, minlength=NU))
    scal /= np.array([8.0, 7.0, 5.0, 3.0], np.float32)

    seen = np.unique(np.concatenate([
        r.astype(np.int64) * NI + c,
        K[K.ts < cutoff].r.to_numpy() * NI + K[K.ts < cutoff].col.to_numpy(),
        D[D.ts < cutoff].r.to_numpy() * NI + D[D.ts < cutoff].col.to_numpy()]))
    return pools, scal, seen


def targets(E, cutoff, seen):
    """Graded new-item positives in [cutoff, cutoff + 1 week)."""
    L, K = E["L"], E["K"]
    s = L[(L.ts >= cutoff) & (L.ts < cutoff + WEEK) & (L.org == 1) & (L.strong == 1)]
    k = K[(K.ts >= cutoff) & (K.ts < cutoff + WEEK) & (K.org == 1)]
    df = pd.concat([pd.DataFrame({"r": s.r.to_numpy(), "c": s.col.to_numpy(), "g": 1}),
                    pd.DataFrame({"r": k.r.to_numpy(), "c": k.col.to_numpy(), "g": 3})])
    df = df.groupby(["r", "c"], sort=False).g.max().reset_index()
    key = df.r.to_numpy() * NI + df.c.to_numpy()
    fresh = ~np.isin(key, seen, assume_unique=False)
    df = df[fresh]
    return df.r.to_numpy(), df.c.to_numpy(), df.g.to_numpy()


# --------------------------------------------------------------------- model
class UserTower(nn.Module):
    def __init__(self, dim=DIM):
        super().__init__()
        self.emb = nn.ModuleDict({p: nn.EmbeddingBag(NI, dim, mode="sum", include_last_offset=True)
                                  for p in POOLS})
        for e in self.emb.values():
            nn.init.normal_(e.weight, std=0.05)
        h = dim * len(POOLS) + 4
        self.mlp = nn.Sequential(nn.Linear(h, 256), nn.GELU(), nn.Linear(256, dim))

    def forward(self, batch, scal):
        parts = [self.emb[p](batch[p][0], batch[p][1], per_sample_weights=batch[p][2])
                 for p in POOLS]
        u = self.mlp(torch.cat(parts + [scal], 1))
        return torch.cat([u, torch.ones(u.shape[0], 1)], 1)


class ItemTower(nn.Module):
    def __init__(self, dim=DIM):
        super().__init__()
        self.emb = nn.Embedding(NI, dim)
        nn.init.normal_(self.emb.weight, std=0.05)
        self.mlp = nn.Sequential(nn.Linear(dim, dim), nn.GELU(), nn.Linear(dim, dim))
        self.bias = nn.Embedding(NI, 1)
        nn.init.zeros_(self.bias.weight)

    def forward(self, idx):
        e = self.emb(idx)
        return torch.cat([self.mlp(e) + e, self.bias(idx)], 1)


def batch_pools(pools, rows):
    out = {}
    for p in POOLS:
        indptr, idx, w = pools[p]
        lens = indptr[rows + 1] - indptr[rows]
        off = np.concatenate([[0], np.cumsum(lens)]).astype(np.int64)
        flat = np.concatenate([np.arange(indptr[r], indptr[r + 1]) for r in rows]) if off[-1] else np.zeros(0, np.int64)
        out[p] = (torch.from_numpy(idx[flat]), torch.from_numpy(off),
                  torch.from_numpy(w[flat]))
    return out


def train(snapshots, pop, epochs=6, bs=1024, nneg=3000, seed=0, log=print):
    torch.manual_seed(seed); rng = np.random.default_rng(seed)
    ut, it = UserTower(), ItemTower()
    opt = torch.optim.Adam(list(ut.parameters()) + list(it.parameters()), lr=3e-3)
    q = (pop + 10.0) ** 0.6
    q = q / q.sum()
    logq = torch.from_numpy(np.log(q).astype(np.float32))
    qt = torch.from_numpy(q)
    rows_all = np.concatenate([np.stack([np.full(len(s["tr"]), i), s["tr"], s["tc"]], 1)
                               for i, s in enumerate(snapshots)])
    wts_all = np.concatenate([np.where(s["tg"] == 3, 2.5, 1.0) for s in snapshots]).astype(np.float32)
    n = len(rows_all)
    log(f"training rows={n}")
    sched = torch.optim.lr_scheduler.OneCycleLR(opt, max_lr=4e-3, total_steps=epochs * (n // bs + 1),
                                                pct_start=0.25)
    for ep in range(epochs):
        perm = rng.permutation(n)
        tot, cnt = 0.0, 0
        for b0 in range(0, n - bs + 1, bs):
            sel = rows_all[perm[b0:b0 + bs]]
            w = torch.from_numpy(wts_all[perm[b0:b0 + bs]])
            loss = 0.0
            for si in np.unique(sel[:, 0]):
                m = sel[:, 0] == si
                s = snapshots[si]
                rows = sel[m, 1]
                bp = batch_pools(s["pools"], rows)
                u = ut(bp, torch.from_numpy(s["scal"][rows]))
                pos = torch.from_numpy(sel[m, 2].astype(np.int64))
                neg = torch.from_numpy(rng.choice(NI, nneg, p=q).astype(np.int64))
                vpos, vneg = it(pos), it(neg)
                lpos = (u * vpos).sum(1) - logq[pos]
                lneg = u @ vneg.T - logq[neg][None, :]
                lneg = lneg.masked_fill(pos[:, None] == neg[None, :], -1e9)
                lg = torch.cat([lpos[:, None], lneg], 1)
                l = -(lg[:, 0] - torch.logsumexp(lg, 1))
                loss = loss + (l * w[torch.from_numpy(m)]).sum()
            loss = loss / bs
            opt.zero_grad(); loss.backward(); opt.step(); sched.step()
            tot += float(loss); cnt += 1
        log(f"  epoch {ep} loss {tot / max(cnt,1):.4f}")
    return ut, it


@torch.no_grad()
def rank(ut, it, pools, scal, NU, elig, topn=100):
    V = it(torch.arange(NI))
    out = np.zeros((NU, topn), np.int32)
    for b0 in range(0, NU, 512):
        rows = np.arange(b0, min(b0 + 512, NU))
        u = ut(batch_pools(pools, rows), torch.from_numpy(scal[rows]))
        s = (u @ V.T).numpy()
        s[~elig[rows]] = -1e30
        idx = np.argpartition(-s, topn, axis=1)[:, :topn]
        ordr = np.take_along_axis(s, idx, 1).argsort(1)[:, ::-1]
        out[rows] = np.take_along_axis(idx, ordr, 1)
    return out


def write_ranking(order, uids, items, path):
    NU, topn = order.shape
    df = pd.DataFrame({"uid": np.repeat(uids, topn),
                       "item_id": items[order.ravel()].astype(np.int64),
                       "rank": np.tile(np.arange(1, topn + 1), NU)})
    df.to_parquet(path, index=False)
    return df


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--stage", choices=["val", "test"], default="val")
    ap.add_argument("--epochs", type=int, default=6)
    ap.add_argument("--ncut", type=int, default=5)
    ap.add_argument("--out", default=None)
    a = ap.parse_args()
    t0 = time.time()
    E = load_events()
    uids = E["uids"]; NU = len(uids)
    items = np.load("cand_items.npy")

    if a.stage == "val":
        infer_cut = CUT
        E = {k: (v[v.ts < CUT] if isinstance(v, pd.DataFrame) else v) for k, v in E.items()}
        elig = np.load("elig_val.npy")
    else:
        infer_cut = int(max(E["L"].ts.max(), E["K"].ts.max())) + 1
        elig = np.load("elig_test.npy")
    train_cuts = [infer_cut - k * WEEK for k in range(1, a.ncut + 1)]

    snaps = []
    for c in train_cuts:
        pools, scal, seen = build_features(E, c, NU)
        tr, tc, tg = targets(E, c, seen)
        snaps.append(dict(pools=pools, scal=scal, tr=tr, tc=tc, tg=tg))
        print(f"cutoff {c}: {len(tr)} targets  ({time.time()-t0:.0f}s)", flush=True)

    pop = np.load("pop_all.npy") if False else np.bincount(
        E["L"][E["L"].ts < infer_cut].col.to_numpy(), minlength=NI).astype(np.float64)
    ut, it = train(snaps, pop, epochs=a.epochs)
    pools, scal, _ = build_features(E, infer_cut, NU)
    order = rank(ut, it, pools, scal, NU, elig)
    out = a.out or ("validation.parquet" if a.stage == "val" else "submission.parquet")
    write_ranking(order, uids, items, out)
    print(f"wrote {out} in {time.time()-t0:.0f}s", flush=True)


if __name__ == "__main__":
    main()
PYEOF
timeout 900 python twotower.py --stage val --epochs 4 2>&1 | tail -30

python - <<'EOF'
import re
s=open('twotower.py').read()
s=s.replace("""        flat = np.concatenate([np.arange(indptr[r], indptr[r + 1]) for r in rows]) if off[-1] else np.zeros(0, np.int64)""",
"""        tot = int(off[-1])
        if tot:
            flat = np.arange(tot) - np.repeat(off[:-1], lens) + np.repeat(indptr[rows], lens)
        else:
            flat = np.zeros(0, np.int64)""")
open('twotower.py','w').write(s)
EOF
cd /app/solution && nohup python twotower.py --stage val --epochs 4 > val.log 2>&1 &
sleep 55; cat val.log

cat > baseline.py <<'EOF'
import numpy as np, pandas as pd
items=np.load('cand_items.npy'); uids=np.load('cand_uids.npy').astype(np.int64)
pop=-np.arange(30000.)  # catalog is popularity-ordered
for stage,f in [('val','elig_val.npy'),('test','elig_test.npy')]:
    E=np.load(f); s=np.tile(pop,(len(uids),1)); s[~E]=-1e30
    idx=np.argpartition(-s,100,axis=1)[:,:100]
    o=np.take_along_axis(s,idx,1).argsort(1)[:,::-1]
    order=np.take_along_axis(idx,o,1)
    pd.DataFrame({'uid':np.repeat(uids,100),'item_id':items[order.ravel()].astype(np.int64),
                  'rank':np.tile(np.arange(1,101),len(uids))}).to_parquet(
        'validation_pop.parquet' if stage=='val' else 'submission.parquet',index=False)
EOF
python baseline.py && python /task/evaluate_public.py validation_pop.parquet && tail -3 val.log

cat > baseline.py <<'EOF'
import numpy as np, pandas as pd
items=np.load('cand_items.npy'); uids=np.load('cand_uids.npy').astype(np.int64)
pop=(-np.arange(30000.)).astype(np.float32)
for f,out in [('elig_val.npy','validation_pop.parquet'),('elig_test.npy','submission.parquet')]:
    E=np.load(f,mmap_mode='r'); order=np.zeros((len(uids),100),np.int32)
    for b in range(0,len(uids),512):
        r=np.arange(b,min(b+512,len(uids)))
        s=np.tile(pop,(len(r),1)); s[~np.asarray(E[r])]=-1e30
        idx=np.argpartition(-s,100,axis=1)[:,:100]
        o=np.take_along_axis(s,idx,1).argsort(1)[:,::-1]
        order[r]=np.take_along_axis(idx,o,1)
    pd.DataFrame({'uid':np.repeat(uids,100),'item_id':items[order.ravel()].astype(np.int64),
                  'rank':np.tile(np.arange(1,101),len(uids))}).to_parquet(out,index=False)
EOF
python baseline.py && python /task/evaluate_public.py validation_pop.parquet; tail -3 val.log

sleep 60; tail -4 val.log; ps aux|grep -c "twotower"

python /task/evaluate_public.py validation.parquet

nohup python twotower.py --stage val --epochs 12 --ncut 6 --out validation_v2.parquet > val2.log 2>&1 &
sleep 2
cd /app/solution && nohup python twotower.py --stage test --epochs 12 --ncut 6 --out submission_v2.parquet > test2.log 2>&1 &
sleep 45; tail -2 val2.log; tail -2 test2.log; free -g|head -2

cat > README.md <<'EOF'
# Two-tower discovery ranker — "Made for You"

## Problem shape (measured, not assumed)

`*_eligible_candidates.npz` excludes **every catalog item the user has already
interacted with** (verified: 0% of a user's listen history is eligible, and 78%
of validation-window organic positives are repeats and therefore ineligible).
The shelf is a pure *new-item discovery* problem, so the model is trained to
predict **items the user has never touched before the cutoff**, not to replay
history.

## Architecture

Both towers are PyTorch `nn.Module`s (`twotower.py`). The score is exactly
`dot(user_vec, item_vec)` with both vectors in R^65, so item vectors are
precomputable offline and served by maximum-inner-product search; the user
vector is one forward pass at request time.

**Item tower** — `ItemTower`: item-ID embedding (30k x 64) -> `Linear(64,64) ->
GELU -> Linear(64,64)` with a residual connection, concatenated with a learned
per-item bias scalar. Output: 65-d.

**User tower** — `UserTower`: no user-ID lookup. The user is represented purely
by interaction history through five weighted `EmbeddingBag` pools over the item
vocabulary:

| pool | contents | cap |
| --- | --- | --- |
| `r7` | listens (>=50% completion) in the last 7 days | 96 |
| `r30` | listens in the last 30 days | 160 |
| `all` | full listen history | 256 |
| `like` | *active* likes at the cutoff (likes minus unlikes, last-event-wins) | 192 |
| `dislike` | active dislikes at the cutoff | 64 |

Pool weights are `log1p` of event counts (organic >=80% listens count double),
normalized per user, and truncated to the top-*cap* items. The five 64-d pooled
vectors plus four log-scaled activity scalars go through
`Linear(324,256) -> GELU -> Linear(256,64)`; a constant `1.0` is appended so the
item-tower bias slot participates in the same dot product.

## Training objective

Sampled-softmax (in effect, cross-entropy over one positive and 3000 shared
negatives) with **logQ correction**, negatives drawn from `popularity^0.6`.
Accidental hits (negative == positive) are masked out. Grade-3 targets
(organic like) get 2.5x the loss weight of grade-1 targets (organic >=80%
listen). Adam + OneCycle, 12 epochs.

Training examples come from **six weekly time cutoffs** rolled backwards from
the inference cutoff. At each cutoff the user tower sees only events strictly
before it, and the targets are the graded organic positives in the following
seven days **with all previously-seen items removed** — the same
history-exclusion rule the eligibility service applies at serving time.

## Scalability

No global user-item interaction matrix is ever built or factorized. History is
consumed as streamed `(user, item, weight)` triples into fixed-size pooled
embeddings, and the loss touches only 1 positive + 3000 sampled negatives per
row, so cost is O(interactions) rather than O(users x items). The user tower is
ID-free, so new/unseen users are served from their history alone.

## Data discipline

- Validation model: history and training targets use `train_*` only; `val_*` is
  never read as a feature. Scored against the public window.
- Final model: refit identically on `train_* + val_*`, with the validation
  events treated as observed history and the inference cutoff moved to the end
  of the validation window.

## Results

| ranker | public validation NDCG@10 |
| --- | --- |
| popularity within eligible set | 0.010710 |
| two-tower (4 epochs, 5 cutoffs) | 0.027231 |
| two-tower (12 epochs, 6 cutoffs) | see `SCORE` below |

## Commands

```sh
cd /app/solution
python prep.py                                  # cache filtered event tables
python twotower.py --stage val  --epochs 12 --ncut 6 --out validation.parquet
python /task/evaluate_public.py validation.parquet
python twotower.py --stage test --epochs 12 --ncut 6 --out submission.parquet
```

`submission.parquet` holds 100 ranked eligible items for each of the 8456
target users.
EOF
sleep 120; tail -3 val2.log; echo ---; tail -3 test2.log

grep -h epoch val2.log | tail -3; echo ---; grep -h epoch test2.log|tail -3

sleep 200; grep -h epoch val2.log|tail -2; echo ---; grep -h epoch test2.log|tail -2

grep -h epoch val2.log|tail -2; echo ---; grep -h epoch test2.log|tail -2

pkill -f "stage val" ; sleep 1; ps aux|grep twotower|grep -v grep|wc -l

ps aux|grep twotower|grep -v grep|awk '{print $2, $12, $13, $14, $15, $16}'

python - <<'EOF'
s=open('README.md').read()
s=s.replace("""| two-tower (4 epochs, 5 cutoffs) | 0.027231 |
| two-tower (12 epochs, 6 cutoffs) | see `SCORE` below |""",
"""| two-tower (4 epochs, 5 cutoffs) | **0.027231** |

The 0.027231 figure is the measured public validation score of this
architecture and is the number the approach was selected on (`validation.parquet`,
reproduced by the first command below). The final hidden-test model is the same
architecture and objective refit on `train_* + val_*` with a longer schedule
(12 epochs, 6 cutoffs) — chosen because the 4-epoch run's training loss was
still falling monotonically (8.67 -> 8.08 -> 7.55), i.e. the short run was
underfit. A 12-epoch validation-stage confirmation run was started but did not
finish inside the one-hour budget, so the longer schedule is an
underfitting-based extrapolation rather than a separately validated setting.""")
s=s.replace("python twotower.py --stage val  --epochs 12 --ncut 6 --out validation.parquet",
            "python twotower.py --stage val  --epochs 4  --ncut 5 --out validation.parquet   # 0.027231")
open('README.md','w').write(s)
EOF
sleep 100; ls -la submission_v2.parquet 2>/dev/null; tail -2 test2.log

sleep 115; ls -la submission_v2.parquet 2>/dev/null || echo "not yet"; ps aux|grep -c "[t]wotower"

grep -h epoch test2.log | tail -3 && python - <<'EOF'
import numpy as np, pandas as pd
df=pd.read_parquet('submission_v2.parquet')
uids=np.load('cand_uids.npy').astype(np.int64); items=np.load('cand_items.npy')
tu=set(pd.read_parquet('/task/data/target_users.parquet').uid.astype(int))
assert set(df.uid)==tu, 'user mismatch'
g=df.groupby('uid')
assert (g.size()==100).all()
assert g['rank'].apply(lambda v:sorted(v)==list(range(1,101))).all()
assert not df.duplicated(['uid','item_id']).any()
E=np.load('elig_test.npy'); urow={int(u):i for i,u in enumerate(uids)}
icol={int(i):c for c,i in enumerate(items)}
r=df.uid.map(urow).to_numpy(); c=df.item_id.map(icol).to_numpy()
assert E[r,c].all(), 'ineligible item present'
print('ALL CHECKS PASS', df.shape, df.uid.nunique())
df.to_parquet('submission.parquet', index=False)
EOF
ls -la /app/solution/

rm -f build_features.py && sed -i 's/^python prep.py .*/python prep.py                                  # cache filtered event tables (run make_arrays first)/' README.md && cat >> README.md <<'EOF'

## Files

- `twotower.py` — towers, feature construction, training, ranking (both stages).
- `prep.py` — one-off filtering of the raw event parquets to catalog items.
- `baseline.py` — popularity reference ranker.
- `validation.parquet` — validation-stage ranking scored at 0.027231.
- `submission.parquet` — **final deliverable** (8456 users x 100 eligible items).
- `test2.log` — final-model training log (loss 8.6 -> 4.35 over 12 epochs).
EOF
python /task/evaluate_public.py validation.parquet