Agent trace

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

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

mkdir -p solution && python -c "
import pandas as pd, numpy as np, time
t=time.time()
for f in ['candidate_catalog','target_users','validation_preference_state','test_preference_state','val_likes','val_listens','train_likes']:
    d=pd.read_parquet('/task/data/%s.parquet'%f)
    print(f, d.shape, list(d.columns))
    print(d.head(3))
print(time.time()-t)
"

python -c "
import numpy as np,time
t=time.time()
z=np.load('/task/data/validation_eligible_candidates.npz')
print(list(z.keys()))
for k in z: print(k, z[k].shape, z[k].dtype)
u=z['uids']; it=z['item_ids']; p=z['packed_eligible']
m=np.unpackbits(p,axis=1,count=len(it),bitorder='little')
print('elig per user mean',m.sum(1).mean(), m.sum(1).min(), m.sum(1).max())
print(time.time()-t)
"

cd /app/solution && python -c "
import pyarrow.parquet as pq, time
f=pq.ParquetFile('/task/data/train_listens.parquet')
print(f.metadata)
print(f.schema)
"

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

/app/solution/data_utils.py

"""Shared data loading helpers for the two-tower discovery ranker."""
from __future__ import annotations

from pathlib import Path

import numpy as np
import pandas as pd
import pyarrow.parquet as pq

DATA = Path("/task/data")
TRAIN_CUTOFF = 24786800
DAY = 86400


def load_listens(split: str, min_ratio: int = 0, since: int | None = None,
                 organic_only: bool = False) -> pd.DataFrame:
    """Stream a listens parquet row-group at a time with early filtering."""
    cols = ["uid", "timestamp", "item_id", "is_organic", "played_ratio_pct"]
    reader = pq.ParquetFile(DATA / f"{split}_listens.parquet")
    parts = []
    for rg in range(reader.metadata.num_row_groups):
        tbl = reader.read_row_group(rg, columns=cols)
        df = tbl.to_pandas()
        if min_ratio:
            df = df[df.played_ratio_pct.values >= min_ratio]
        if since is not None:
            df = df[df.timestamp.values >= since]
        if organic_only:
            df = df[df.is_organic.values == 1]
        parts.append(df)
    out = pd.concat(parts, ignore_index=True)
    return out


def load_events(family: str, split: str) -> pd.DataFrame:
    return pd.read_parquet(DATA / f"{split}_{family}.parquet")


def candidate_items() -> np.ndarray:
    cat = pd.read_parquet(DATA / "candidate_catalog.parquet")
    return cat


def target_users() -> np.ndarray:
    return pd.read_parquet(DATA / "target_users.parquet").uid.astype(np.int64).values


def eligibility_matrix(stage: str):
    """Return (uids, item_ids, bool mask [n_users, n_items])."""
    with np.load(DATA / f"{stage}_eligible_candidates.npz", allow_pickle=False) as z:
        uids = z["uids"].astype(np.int64)
        item_ids = z["item_ids"].astype(np.int64)
        packed = z["packed_eligible"]
        mask = np.unpackbits(packed, axis=1, count=len(item_ids),
                             bitorder="little").astype(bool)
    return uids, item_ids, mask

time python -c "
import numpy as np, pandas as pd, data_utils as du
tg = set(du.target_users().tolist())
uids, items, mask = du.eligibility_matrix('validation')
itempos = {int(i):c for c,i in enumerate(items)}
# val relevance
vl = du.load_events('listens','val')
vlk = du.load_events('likes','val')
strong = vl[(vl.is_organic==1)&(vl.played_ratio_pct>=80)&vl.uid.isin(tg)][['uid','item_id']].drop_duplicates()
lk = vlk[(vlk.is_organic==1)&vlk.uid.isin(tg)][['uid','item_id']].drop_duplicates()
rel = pd.concat([strong,lk]).drop_duplicates()
print('rel pairs', len(rel), 'users', rel.uid.nunique())
incat = rel.item_id.isin(itempos.keys()).mean()
print('rel in catalog frac', incat)
# eligible frac
urow={int(u):r for r,u in enumerate(uids)}
rr = rel[rel.item_id.isin(itempos.keys())]
r_idx = rr.uid.map(urow).values; c_idx=rr.item_id.map(itempos).values
el = mask[r_idx,c_idx]
print('eligible frac of in-cat rel', el.mean())
np.save('/tmp/nothing.npy',np.array([0]))
" 2>&1 | tail -20

time python -c "
import numpy as np, pandas as pd, data_utils as du
uids, items, mask = du.eligibility_matrix('validation')
uids2, items2, mask2 = du.eligibility_matrix('test')
print('same uids', np.array_equal(uids,uids2), 'same items', np.array_equal(items,items2))
print('val elig',mask.sum(1).mean(),'test elig',mask2.sum(1).mean())
print('test subset of val?', (mask2 & ~mask).sum())
# check excluded = history
tg=du.target_users()
print('uids==targets sorted', np.array_equal(np.sort(uids), np.sort(tg)))
"

time python -c "
import numpy as np, pandas as pd, data_utils as du, time
t=time.time()
df = du.load_listens('train', min_ratio=50, since=du.TRAIN_CUTOFF-180*86400)
print('rows',len(df), time.time()-t)
print('users', df.uid.nunique(), 'items', df.item_id.nunique())
vc = df.item_id.value_counts()
print(vc.head())
print('items with >=20 plays', (vc>=20).sum(), '>=50', (vc>=50).sum(), '>=100',(vc>=100).sum())
cat=du.candidate_items()
print('catalog coverage of rows', df.item_id.isin(set(cat.item_id)).mean())
" 

/app/solution/features.py

"""Build cutoff-correct user history channels and training targets.

Every representation consumed by the user tower is a bag of item ids drawn
strictly from events before the relevant cutoff, so the same code path serves
training slices, the validation ranking, and the final hidden-test ranking.
"""
from __future__ import annotations

import numpy as np
import pandas as pd

import data_utils as du

DAY = 86400
# Channel capacities: recent positive listens, habitual top items, likes,
# dislikes.
CH_SIZES = (192, 96, 96, 32)
N_CH = 4


def _sorted_events(df: pd.DataFrame) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Return uid-major, time-ordered (uid, ts, item) arrays."""
    order = np.lexsort((df.timestamp.values, df.uid.values))
    return (df.uid.values[order].astype(np.int64),
            df.timestamp.values[order].astype(np.int64),
            df.item_id.values[order].astype(np.int64))


class EventStore:
    """Per-user, time-sorted event streams with cutoff slicing."""

    def __init__(self, frames: dict[str, pd.DataFrame], users: np.ndarray):
        self.users = users
        self.upos = {int(u): i for i, u in enumerate(users)}
        self.streams = {}
        for name, df in frames.items():
            df = df[df.uid.isin(self.upos)]
            uid, ts, item = _sorted_events(df)
            row = np.array([self.upos[int(u)] for u in uid], dtype=np.int64) \
                if len(uid) else np.zeros(0, np.int64)
            # user boundaries
            starts = np.searchsorted(row, np.arange(len(users)), "left")
            ends = np.searchsorted(row, np.arange(len(users)), "right")
            self.streams[name] = (ts, item, starts, ends)

    def slice(self, name: str, u: int, cutoff: int):
        ts, item, starts, ends = self.streams[name]
        s, e = starts[u], ends[u]
        if s == e:
            return ts[0:0], item[0:0]
        p = s + np.searchsorted(ts[s:e], cutoff, "left")
        return ts[s:p], item[s:p]


def build_profiles(store: EventStore, vocab_map: np.ndarray, cutoff: int,
                   half_life_days: float = 21.0):
    """Return padded id/weight tensors for all users at ``cutoff``.

    Args:
        store: Event store holding listens/likes/dislikes streams.
        vocab_map: Dense lookup from raw item id to vocab index (-1 = unknown).
        cutoff: Exclusive upper bound on event timestamps.
        half_life_days: Recency half-life for the recent-listen channel.

    Returns:
        ``(ids, weights, seen)`` where ``ids``/``weights`` are lists of
        ``[n_users, cap]`` arrays (one per channel) and ``seen`` is a list of
        per-user sets of already-consumed vocab indices.
    """
    n = len(store.users)
    ids = [np.zeros((n, c), np.int64) for c in CH_SIZES]
    wts = [np.zeros((n, c), np.float32) for c in CH_SIZES]
    seen: list[np.ndarray] = []
    hl = half_life_days * DAY

    for u in range(n):
        lts, litem = store.slice("listen", u, cutoff)
        lidx = vocab_map[litem]
        ok = lidx >= 0
        lidx_ok, lts_ok = lidx[ok], lts[ok]

        # Channel 0: most recent positive listens, recency weighted.
        c = CH_SIZES[0]
        take = lidx_ok[-c:]
        tts = lts_ok[-c:]
        if take.size:
            w = np.exp2(-(cutoff - tts) / hl).astype(np.float32)
            ids[0][u, :take.size] = take
            wts[0][u, :take.size] = w / max(w.sum(), 1e-6)

        # Channel 1: habitual favourites -- top items by lifetime play count.
        c = CH_SIZES[1]
        if lidx_ok.size:
            uniq, cnt = np.unique(lidx_ok, return_counts=True)
            if uniq.size > c:
                sel = np.argpartition(-cnt, c - 1)[:c]
                uniq, cnt = uniq[sel], cnt[sel]
            w = np.log1p(cnt).astype(np.float32)
            ids[1][u, :uniq.size] = uniq
            wts[1][u, :uniq.size] = w / max(w.sum(), 1e-6)
            seen.append(np.unique(lidx_ok))
        else:
            seen.append(np.zeros(0, np.int64))

        # Channels 2/3: explicit feedback.
        for ci, name in ((2, "like"), (3, "dislike")):
            c = CH_SIZES[ci]
            _, eitem = store.slice(name, u, cutoff)
            eidx = vocab_map[eitem]
            eidx = eidx[eidx >= 0][-c:]
            if eidx.size:
                ids[ci][u, :eidx.size] = eidx
                wts[ci][u, :eidx.size] = 1.0 / eidx.size
    return ids, wts, seen


def build_targets(store: EventStore, vocab_map: np.ndarray, n_catalog: int,
                  cutoff: int, horizon: int, seen: list[np.ndarray]):
    """Collect new-item organic positives in ``(cutoff, cutoff + horizon]``.

    Mirrors the evaluation objective: grade-3 organic likes and grade-1
    organic 80%-completion listens on items the user has not consumed before
    the cutoff.
    """
    rows_u, rows_i, rows_w = [], [], []
    for u in range(len(store.users)):
        hi = cutoff + horizon
        got: dict[int, float] = {}
        for name, weight in (("strong", 1.0), ("like_org", 2.0)):
            ts, item, starts, ends = store.streams[name]
            s, e = starts[u], ends[u]
            if s == e:
                continue
            a = s + np.searchsorted(ts[s:e], cutoff, "right")
            b = s + np.searchsorted(ts[s:e], hi, "right")
            if a >= b:
                continue
            idx = vocab_map[item[a:b]]
            idx = idx[(idx >= 0) & (idx < n_catalog)]
            for it in np.unique(idx):
                got[int(it)] = max(got.get(int(it), 0.0), weight)
        if not got:
            continue
        prior = seen[u]
        cand = np.fromiter(got.keys(), np.int64, len(got))
        keep = ~np.isin(cand, prior)
        for it in cand[keep]:
            rows_u.append(u)
            rows_i.append(int(it))
            rows_w.append(got[int(it)])
    return (np.asarray(rows_u, np.int64), np.asarray(rows_i, np.int64),
            np.asarray(rows_w, np.float32))


def load_frames(include_val: bool):
    """Load listen/like/dislike frames, optionally including the val window."""
    splits = ["train"] + (["val"] if include_val else [])
    listens, likes, dislikes = [], [], []
    for sp in splits:
        since = du.TRAIN_CUTOFF - 200 * DAY if sp == "train" else None
        listens.append(du.load_listens(sp, min_ratio=50, since=since))
        likes.append(du.load_events("likes", sp))
        dislikes.append(du.load_events("dislikes", sp))
    lis = pd.concat(listens, ignore_index=True)
    lik = pd.concat(likes, ignore_index=True)
    dis = pd.concat(dislikes, ignore_index=True)
    strong = lis[(lis.is_organic.values == 1) & (lis.played_ratio_pct.values >= 80)]
    frames = {
        "listen": lis[["uid", "timestamp", "item_id"]],
        "like": lik[["uid", "timestamp", "item_id"]],
        "dislike": dis[["uid", "timestamp", "item_id"]],
        "strong": strong[["uid", "timestamp", "item_id"]],
        "like_org": lik[lik.is_organic.values == 1][["uid", "timestamp", "item_id"]],
    }
    return frames, lis


def build_vocab(listens: pd.DataFrame, catalog: pd.DataFrame, extra: int = 50000):
    """Catalog items first, then the most played non-catalog history items."""
    cat_items = catalog.item_id.astype(np.int64).values
    n_catalog = len(cat_items)
    vc = listens.item_id.value_counts()
    non_cat = vc[~vc.index.isin(set(cat_items.tolist()))]
    extra_items = non_cat.index.values[:extra].astype(np.int64)
    vocab = np.concatenate([cat_items, extra_items])
    vmax = int(vocab.max()) + 1
    vocab_map = np.full(vmax, -1, np.int64)
    vocab_map[vocab] = np.arange(len(vocab))
    return vocab, vocab_map, n_catalog

python - <<'EOF'
import re
p='features.py'
s=open(p).read()
s = s.replace("""def _sorted_events""", """def _map(vocab_map: np.ndarray, items: np.ndarray) -> np.ndarray:
    \"\"\"Vocab lookup that tolerates item ids beyond the lookup table.\"\"\"
    if items.size == 0:
        return items.astype(np.int64)
    it = items.astype(np.int64)
    out = np.full(it.shape, -1, np.int64)
    ok = it < vocab_map.shape[0]
    out[ok] = vocab_map[it[ok]]
    return out


def _sorted_events""")
s = s.replace("lidx = vocab_map[litem]", "lidx = _map(vocab_map, litem)")
s = s.replace("eidx = vocab_map[eitem]", "eidx = _map(vocab_map, eitem)")
s = s.replace("idx = vocab_map[item[a:b]]", "idx = _map(vocab_map, item[a:b])")
open(p,'w').write(s)
EOF
grep -n "_map" features.py

/app/solution/model.py

"""Two-tower ranker: a history-based user tower and an item tower.

The final relevance score is exactly ``dot(user_vector, item_vector)`` with
both vectors of dimension ``out_dim``, so item vectors can be precomputed
offline into a maximum-inner-product index and request-time work is one user
tower forward pass. The user tower never sees a user id: it reads only pooled
interaction history, so new users are served without retraining.
"""
from __future__ import annotations

import torch
import torch.nn as nn
import torch.nn.functional as F


class ItemTower(nn.Module):
    """Map an item id plus its popularity prior to an item vector.

    The last output coordinate carries a learned item bias; the user tower
    pins its matching coordinate to 1.0, keeping the score a plain dot product
    of equal-length vectors.
    """

    def __init__(self, emb: nn.Embedding, n_catalog: int, log_pop: torch.Tensor,
                 hidden: int, out_dim: int, dropout: float = 0.1):
        super().__init__()
        self.emb = emb
        self.n_catalog = n_catalog
        self.register_buffer("log_pop", log_pop)
        d = emb.embedding_dim
        self.net = nn.Sequential(
            nn.Linear(d + 1, hidden), nn.GELU(), nn.Dropout(dropout),
            nn.Linear(hidden, out_dim - 1),
        )
        self.bias = nn.Embedding(n_catalog, 1)
        nn.init.zeros_(self.bias.weight)

    def forward(self, items: torch.Tensor) -> torch.Tensor:
        e = self.emb(items)
        x = torch.cat([e, self.log_pop[items].unsqueeze(-1)], dim=-1)
        return torch.cat([self.net(x), self.bias(items)], dim=-1)


class UserTower(nn.Module):
    """Pool per-channel interaction bags into one user vector."""

    def __init__(self, emb: nn.Embedding, n_channels: int, hidden: int,
                 out_dim: int, dropout: float = 0.1):
        super().__init__()
        self.emb = emb
        self.n_channels = n_channels
        d = emb.embedding_dim
        self.net = nn.Sequential(
            nn.Linear(d * n_channels, hidden), nn.GELU(), nn.Dropout(dropout),
            nn.Linear(hidden, hidden), nn.GELU(),
            nn.Linear(hidden, out_dim - 1),
        )

    def forward(self, ids: list[torch.Tensor], wts: list[torch.Tensor]) -> torch.Tensor:
        pooled = []
        for c in range(self.n_channels):
            pooled.append(F.embedding_bag(
                ids[c], self.emb.weight, mode="sum",
                per_sample_weights=wts[c]))
        x = torch.cat(pooled, dim=-1)
        h = self.net(x)
        ones = torch.ones(h.shape[0], 1, dtype=h.dtype, device=h.device)
        return torch.cat([h, ones], dim=-1)


class TwoTower(nn.Module):
    """Container holding the shared embedding table and both towers."""

    def __init__(self, vocab: int, n_catalog: int, log_pop: torch.Tensor,
                 dim: int = 64, hidden: int = 256, out_dim: int = 64,
                 n_channels: int = 4, dropout: float = 0.1):
        super().__init__()
        self.emb = nn.Embedding(vocab, dim)
        nn.init.normal_(self.emb.weight, std=0.05)
        self.user = UserTower(self.emb, n_channels, hidden, out_dim, dropout)
        self.item = ItemTower(self.emb, n_catalog, log_pop, hidden, out_dim, dropout)

    def score(self, u: torch.Tensor, items: torch.Tensor) -> torch.Tensor:
        return u @ self.item(items).t()

/app/solution/run.py

"""Train the two-tower ranker and emit a ranking for one evaluation stage.

Usage::

    python run.py validation solution/validation.parquet
    python run.py test solution/submission.parquet
"""
from __future__ import annotations

import sys
import time
from pathlib import Path

import numpy as np
import pandas as pd
import torch
import torch.nn.functional as F

import data_utils as du
import features as ft
from model import TwoTower

torch.set_num_threads(8)
DAY = 86400
HORIZON = 7 * DAY
SEED = 17

# Training slice cutoffs, expressed as days before the profile cutoff. Each
# slice contributes (history-before-cutoff, new organic positives after) pairs.
SLICE_DAYS = (7, 14, 21, 28, 35, 42, 49, 56)
EPOCHS = 3
BATCH = 1024
N_NEG = 2000
LR = 3e-3
DIM = 64
OUT_DIM = 64
HIDDEN = 256


def log(msg: str, t0: float) -> None:
    print(f"[{time.time() - t0:7.1f}s] {msg}", flush=True)


def to_tensors(ids, wts, rows=None):
    if rows is None:
        return ([torch.from_numpy(a) for a in ids],
                [torch.from_numpy(a) for a in wts])
    return ([torch.from_numpy(a[rows]) for a in ids],
            [torch.from_numpy(a[rows]) for a in wts])


def main() -> None:
    stage = sys.argv[1]
    out_path = Path(sys.argv[2])
    assert stage in ("validation", "test")
    include_val = stage == "test"
    t0 = time.time()
    rng = np.random.default_rng(SEED)
    torch.manual_seed(SEED)

    frames, listens = ft.load_frames(include_val)
    log(f"loaded {len(listens)} positive listens", t0)

    catalog = du.candidate_items()
    vocab, vocab_map, n_catalog = ft.build_vocab(listens, catalog)
    users = np.sort(np.unique(np.concatenate(
        [listens.uid.unique().astype(np.int64), du.target_users()])))
    store = ft.EventStore(frames, users)
    log(f"vocab={len(vocab)} catalog={n_catalog} users={len(users)}", t0)

    # Popularity prior over the catalog, from the training-history counts.
    counts = np.zeros(n_catalog, np.float64)
    lidx = ft._map(vocab_map, listens.item_id.values)
    keep = (lidx >= 0) & (lidx < n_catalog)
    np.add.at(counts, lidx[keep], 1.0)
    log_pop = torch.from_numpy(
        (np.log1p(counts) - np.log1p(counts).mean()).astype(np.float32))
    # Sampling distribution for shared negatives.
    neg_p = torch.from_numpy((counts + 1.0) ** 0.75)
    neg_p = neg_p / neg_p.sum()

    # ---- Training slices -------------------------------------------------
    final_cutoff = du.TRAIN_CUTOFF if stage == "validation" else int(
        max(frames["listen"].timestamp.max(), frames["like"].timestamp.max())) + 1
    slice_ids, slice_wts, tu, ti, tw, tslice = [], [], [], [], [], []
    for k, days in enumerate(SLICE_DAYS):
        cutoff = final_cutoff - days * DAY
        ids, wts, seen = ft.build_profiles(store, vocab_map, cutoff)
        u, i, w = ft.build_targets(store, vocab_map, n_catalog, cutoff,
                                   HORIZON, seen)
        slice_ids.append(ids)
        slice_wts.append(wts)
        tu.append(u)
        ti.append(i)
        tw.append(w)
        tslice.append(np.full(len(u), k, np.int64))
        log(f"slice -{days}d: {len(u)} pairs", t0)
    tu = np.concatenate(tu)
    ti = np.concatenate(ti)
    tw = np.concatenate(tw)
    tslice = np.concatenate(tslice)
    log(f"total training pairs: {len(tu)}", t0)

    slice_t = [to_tensors(slice_ids[k], slice_wts[k])
               for k in range(len(SLICE_DAYS))]

    # ---- Model -----------------------------------------------------------
    model = TwoTower(len(vocab), n_catalog, log_pop, dim=DIM, hidden=HIDDEN,
                     out_dim=OUT_DIM, n_channels=ft.N_CH)
    opt = torch.optim.AdamW(model.parameters(), lr=LR, weight_decay=1e-5)
    n = len(tu)
    steps_per_epoch = max(1, n // BATCH)
    sched = torch.optim.lr_scheduler.OneCycleLR(
        opt, max_lr=LR, total_steps=EPOCHS * steps_per_epoch, pct_start=0.15)
    log_q = torch.log(neg_p + 1e-12)

    ti_t = torch.from_numpy(ti)
    tw_t = torch.from_numpy(tw)
    model.train()
    for epoch in range(EPOCHS):
        perm = rng.permutation(n)
        running, seen_n = 0.0, 0
        for step in range(steps_per_epoch):
            b = perm[step * BATCH:(step + 1) * BATCH]
            pos = ti_t[b]
            negs = torch.multinomial(neg_p, N_NEG, replacement=True)
            cand = torch.cat([pos, negs])

            # Group the batch by slice so each user bag uses its own cutoff.
            uvecs = torch.empty(len(b), OUT_DIM)
            sl = tslice[b]
            for k in np.unique(sl):
                m = np.nonzero(sl == k)[0]
                rows = torch.from_numpy(tu[b][m])
                ids_k, wts_k = slice_t[k]
                uvecs[m] = model.user([x[rows] for x in ids_k],
                                      [x[rows] for x in wts_k])

            ivecs = model.item(cand)
            logits = uvecs @ ivecs.t()
            logits[:, len(b):] -= log_q[negs].unsqueeze(0)

            # Suppress duplicates of the true positive among the candidates.
            dup = cand.unsqueeze(0) == pos.unsqueeze(1)
            dup[torch.arange(len(b)), torch.arange(len(b))] = False
            logits = logits.masked_fill(dup, -1e9)

            target = torch.arange(len(b))
            loss = (F.cross_entropy(logits, target, reduction="none")
                    * tw_t[b]).mean()
            opt.zero_grad()
            loss.backward()
            torch.nn.utils.clip_grad_norm_(model.parameters(), 5.0)
            opt.step()
            sched.step()
            running += float(loss) * len(b)
            seen_n += len(b)
            if step % 200 == 0:
                log(f"epoch {epoch} step {step}/{steps_per_epoch} "
                    f"loss {running / max(seen_n, 1):.4f}", t0)
        log(f"epoch {epoch} done, loss {running / max(seen_n, 1):.4f}", t0)

    # ---- Inference -------------------------------------------------------
    model.eval()
    ids, wts, _ = ft.build_profiles(store, vocab_map, final_cutoff)
    targets = du.target_users()
    rows = np.searchsorted(users, targets)
    eu, eitems, emask = du.eligibility_matrix(stage)
    # Column order of the eligibility matrix, expressed in vocab indices.
    col_vocab = ft._map(vocab_map, eitems)
    assert (col_vocab >= 0).all() and (col_vocab < n_catalog).all()
    col_vocab_t = torch.from_numpy(col_vocab)
    urow = {int(u): r for r, u in enumerate(eu)}

    with torch.no_grad():
        item_vecs = torch.cat([model.item(col_vocab_t[s:s + 4096])
                               for s in range(0, len(col_vocab), 4096)])
        out_u, out_i, out_r = [], [], []
        for s in range(0, len(targets), 512):
            blk = rows[s:s + 512]
            bt = torch.from_numpy(blk)
            uv = model.user([torch.from_numpy(a)[bt] for a in ids],
                            [torch.from_numpy(a)[bt] for a in wts])
            scores = (uv @ item_vecs.t()).numpy()
            for j, uid in enumerate(targets[s:s + 512]):
                m = emask[urow[int(uid)]]
                sc = np.where(m, scores[j], -np.inf)
                top = np.argpartition(-sc, 100)[:100]
                top = top[np.argsort(-sc[top])]
                out_u.append(np.full(100, uid, np.int64))
                out_i.append(eitems[top])
                out_r.append(np.arange(1, 101, dtype=np.int64))

    frame = pd.DataFrame({
        "uid": np.concatenate(out_u),
        "item_id": np.concatenate(out_i),
        "rank": np.concatenate(out_r),
    })
    out_path.parent.mkdir(parents=True, exist_ok=True)
    frame.to_parquet(out_path, index=False)
    log(f"wrote {out_path} ({len(frame)} rows)", t0)


if __name__ == "__main__":
    main()

/app/solution/run.py · Replace one occurrence

Before

            # Group the batch by slice so each user bag uses its own cutoff.
            uvecs = torch.empty(len(b), OUT_DIM)
            sl = tslice[b]
            for k in np.unique(sl):
                m = np.nonzero(sl == k)[0]
                rows = torch.from_numpy(tu[b][m])
                ids_k, wts_k = slice_t[k]
                uvecs[m] = model.user([x[rows] for x in ids_k],
                                      [x[rows] for x in wts_k])

            ivecs = model.item(cand)
            logits = uvecs @ ivecs.t()
            logits[:, len(b):] -= log_q[negs].unsqueeze(0)

After

            # Group the batch by slice so each user bag uses its own cutoff.
            sl = tslice[b]
            parts, order = [], []
            for k in np.unique(sl):
                m = np.nonzero(sl == k)[0]
                rows = torch.from_numpy(tu[b][m])
                ids_k, wts_k = slice_t[k]
                parts.append(model.user([x[rows] for x in ids_k],
                                        [x[rows] for x in wts_k]))
                order.append(m)
            inv = np.empty(len(b), np.int64)
            inv[np.concatenate(order)] = np.arange(len(b))
            uvecs = torch.cat(parts)[torch.from_numpy(inv)]

            ivecs = model.item(cand)
            corr = torch.cat([torch.zeros(len(b)), log_q[negs]])
            logits = uvecs @ ivecs.t() - corr.unsqueeze(0)

python -c "
import numpy as np, pandas as pd, data_utils as du
for stage,out in [('validation','validation_pop.parquet'),('test','submission.parquet')]:
    eu, eitems, emask = du.eligibility_matrix(stage)
    cat = du.candidate_items().sort_values('popularity_rank')
    rank_of = {int(i):r for r,i in enumerate(cat.item_id.values)}
    score = np.array([-rank_of[int(i)] for i in eitems], dtype=np.float64)
    tg = du.target_users()
    urow={int(u):r for r,u in enumerate(eu)}
    U,I,R=[],[],[]
    for uid in tg:
        m=emask[urow[int(uid)]]
        sc=np.where(m,score,-np.inf)
        top=np.argpartition(-sc,100)[:100]; top=top[np.argsort(-sc[top])]
        U.append(np.full(100,uid)); I.append(eitems[top]); R.append(np.arange(1,101))
    pd.DataFrame({'uid':np.concatenate(U),'item_id':np.concatenate(I),'rank':np.concatenate(R)}).to_parquet(out,index=False)
    print('wrote',out)
" && python /task/evaluate_public.py /app/solution/validation_pop.parquet

python - <<'EOF'
p='run.py'; s=open(p).read()
s=s.replace("""SLICE_DAYS = (7, 14, 21, 28, 35, 42, 49, 56)
EPOCHS = 3""","""SLICE_DAYS = tuple(int(x) for x in
                   os.environ.get("SLICES", "7,14,21,28,35,42,49,56").split(","))
EPOCHS = int(os.environ.get("EPOCHS", "3"))""")
s=s.replace("import sys\nimport time","import os\nimport sys\nimport time")
s=s.replace('STEPCAP_PLACEHOLDER','')
s=s.replace("        steps_per_epoch = ","        steps_per_epoch = ")
open(p,'w').write(s)
EOF
grep -n "SLICE_DAYS\|EPOCHS =" run.py

SLICES=7 EPOCHS=1 timeout 600 python run.py validation /app/solution/smoke.parquet 2>&1 | tail -30

python /task/evaluate_public.py /app/solution/smoke.parquet

EPOCHS=6 timeout 1200 python run.py validation /app/solution/validation.parquet 2>&1 | grep -v UserWarning | grep -v "Consider using" | grep -v "running +=" | tail -30 && python /task/evaluate_public.py /app/solution/validation.parquet

sleep 100; tail -5 /tmp/claude-1000/-app/53bed668-7041-4440-8306-d01153e497d0/tasks/bah2z5a7c.output

grep -v Warning /tmp/claude-1000/-app/53bed668-7041-4440-8306-d01153e497d0/tasks/bah2z5a7c.output | tail -12

ls -la /tmp/claude-1000/-app/53bed668-7041-4440-8306-d01153e497d0/tasks/ 2>/dev/null; cat /tmp/claude-1000/-app/53bed668-7041-4440-8306-d01153e497d0/tasks/bah2z5a7c.output 2>/dev/null | tail -20

cat >> features.py <<'EOF'


class PopStats:
    """Time-sorted catalog event streams for cutoff-correct item priors."""

    def __init__(self, frames: dict[str, pd.DataFrame], vocab_map: np.ndarray,
                 n_catalog: int):
        self.n_catalog = n_catalog
        self.streams = {}
        for name in ("listen", "strong", "like_org"):
            df = frames[name]
            idx = _map(vocab_map, df.item_id.values)
            ts = df.timestamp.values.astype(np.int64)
            keep = (idx >= 0) & (idx < n_catalog)
            idx, ts = idx[keep], ts[keep]
            o = np.argsort(ts, kind="stable")
            self.streams[name] = (ts[o], idx[o])

    def _count(self, name: str, lo: int | None, hi: int) -> np.ndarray:
        ts, idx = self.streams[name]
        b = np.searchsorted(ts, hi, "left")
        a = 0 if lo is None else np.searchsorted(ts, lo, "left")
        return np.bincount(idx[a:b], minlength=self.n_catalog).astype(np.float32)

    def features(self, cutoff: int) -> np.ndarray:
        """Return standardized log-popularity features as of ``cutoff``."""
        d = DAY
        p7 = self._count("listen", cutoff - 7 * d, cutoff)
        p28 = self._count("listen", cutoff - 28 * d, cutoff)
        pall = self._count("listen", None, cutoff)
        s28 = self._count("strong", cutoff - 28 * d, cutoff)
        l28 = self._count("like_org", cutoff - 28 * d, cutoff)
        cols = [np.log1p(p7), np.log1p(p28), np.log1p(pall),
                np.log1p(s28), np.log1p(l28),
                np.log1p(p7) - np.log1p(p28 / 4.0)]
        f = np.stack(cols, axis=1).astype(np.float32)
        f -= f.mean(0, keepdims=True)
        f /= (f.std(0, keepdims=True) + 1e-6)
        return f


N_POP_FEATS = 6
EOF
python -c "import features; print(features.N_POP_FEATS)"

python - <<'EOF'
p='model.py'; s=open(p).read()
s=s.replace('''    def __init__(self, emb: nn.Embedding, n_catalog: int, log_pop: torch.Tensor,
                 hidden: int, out_dim: int, dropout: float = 0.1):
        super().__init__()
        self.emb = emb
        self.n_catalog = n_catalog
        self.register_buffer("log_pop", log_pop)
        d = emb.embedding_dim
        self.net = nn.Sequential(
            nn.Linear(d + 1, hidden), nn.GELU(), nn.Dropout(dropout),
            nn.Linear(hidden, out_dim - 1),
        )
        self.bias = nn.Embedding(n_catalog, 1)
        nn.init.zeros_(self.bias.weight)

    def forward(self, items: torch.Tensor) -> torch.Tensor:
        e = self.emb(items)
        x = torch.cat([e, self.log_pop[items].unsqueeze(-1)], dim=-1)
        return torch.cat([self.net(x), self.bias(items)], dim=-1)''','''    def __init__(self, emb: nn.Embedding, n_catalog: int, n_feats: int,
                 hidden: int, out_dim: int, dropout: float = 0.1):
        super().__init__()
        self.emb = emb
        self.n_catalog = n_catalog
        d = emb.embedding_dim
        self.net = nn.Sequential(
            nn.Linear(d + n_feats, hidden), nn.GELU(), nn.Dropout(dropout),
            nn.Linear(hidden, out_dim - 1),
        )
        self.bias = nn.Sequential(
            nn.Linear(d + n_feats, 64), nn.GELU(), nn.Linear(64, 1))

    def forward(self, items: torch.Tensor, feats: torch.Tensor) -> torch.Tensor:
        x = torch.cat([self.emb(items), feats], dim=-1)
        return torch.cat([self.net(x), self.bias(x)], dim=-1)''')
s=s.replace('''    def __init__(self, vocab: int, n_catalog: int, log_pop: torch.Tensor,
                 dim: int = 64, hidden: int = 256, out_dim: int = 64,
                 n_channels: int = 4, dropout: float = 0.1):''','''    def __init__(self, vocab: int, n_catalog: int, n_feats: int,
                 dim: int = 64, hidden: int = 256, out_dim: int = 64,
                 n_channels: int = 4, dropout: float = 0.1):''')
s=s.replace('self.item = ItemTower(self.emb, n_catalog, log_pop, hidden, out_dim, dropout)',
            'self.item = ItemTower(self.emb, n_catalog, n_feats, hidden, out_dim, dropout)')
s=s.replace('''    def score(self, u: torch.Tensor, items: torch.Tensor) -> torch.Tensor:
        return u @ self.item(items).t()''','''    def score(self, u: torch.Tensor, items: torch.Tensor,
              feats: torch.Tensor) -> torch.Tensor:
        return u @ self.item(items, feats).t()''')
open(p,'w').write(s)
EOF
python -c "import model; print('ok')"

python - <<'EOF'
p='run.py'; s=open(p).read()

s=s.replace('''    # Popularity prior over the catalog, from the training-history counts.
    counts = np.zeros(n_catalog, np.float64)
    lidx = ft._map(vocab_map, listens.item_id.values)
    keep = (lidx >= 0) & (lidx < n_catalog)
    np.add.at(counts, lidx[keep], 1.0)
    log_pop = torch.from_numpy(
        (np.log1p(counts) - np.log1p(counts).mean()).astype(np.float32))
    # Sampling distribution for shared negatives.
    neg_p = torch.from_numpy((counts + 1.0) ** 0.75)
    neg_p = neg_p / neg_p.sum()
''','''    pop = ft.PopStats(frames, vocab_map, n_catalog)
    counts = np.zeros(n_catalog, np.float64)
    lidx = ft._map(vocab_map, listens.item_id.values)
    keep = (lidx >= 0) & (lidx < n_catalog)
    np.add.at(counts, lidx[keep], 1.0)
    # Sampling distribution for shared negatives.
    neg_p = torch.from_numpy((counts + 1.0) ** 0.75)
    neg_p = neg_p / neg_p.sum()
''')

s=s.replace('''    slice_ids, slice_wts, tu, ti, tw, tslice = [], [], [], [], [], []
    for k, days in enumerate(SLICE_DAYS):
        cutoff = final_cutoff - days * DAY
        ids, wts, seen = ft.build_profiles(store, vocab_map, cutoff)
        u, i, w = ft.build_targets(store, vocab_map, n_catalog, cutoff,
                                   HORIZON, seen)
        slice_ids.append(ids)''','''    slice_ids, slice_wts, tu, ti, tw, tslice = [], [], [], [], [], []
    slice_feats = []
    for k, days in enumerate(SLICE_DAYS):
        cutoff = final_cutoff - days * DAY
        ids, wts, seen = ft.build_profiles(store, vocab_map, cutoff)
        u, i, w = ft.build_targets(store, vocab_map, n_catalog, cutoff,
                                   HORIZON, seen)
        slice_feats.append(torch.from_numpy(pop.features(cutoff)))
        slice_ids.append(ids)''')

old_loop = s[s.index('    ti_t = torch.from_numpy(ti)'):s.index('    # ---- Inference')]
new_loop = '''    ti_t = torch.from_numpy(ti)
    tw_t = torch.from_numpy(tw)
    slice_rows = [np.nonzero(tslice == k)[0] for k in range(len(SLICE_DAYS))]
    model.train()
    for epoch in range(EPOCHS):
        # Batches are drawn within a slice so every candidate item shares one
        # cutoff-correct popularity feature vector.
        batches = []
        for k, rows_k in enumerate(slice_rows):
            perm = rng.permutation(rows_k)
            for s0 in range(0, len(perm) - BATCH + 1, BATCH):
                batches.append((k, perm[s0:s0 + BATCH]))
        rng.shuffle(batches)
        running, seen_n = 0.0, 0
        for step, (k, b) in enumerate(batches):
            pos = ti_t[b]
            negs = torch.multinomial(neg_p, N_NEG, replacement=True)
            cand = torch.cat([pos, negs])
            feats = slice_feats[k][cand]

            rows = torch.from_numpy(tu[b])
            ids_k, wts_k = slice_t[k]
            uvecs = model.user([x[rows] for x in ids_k],
                               [x[rows] for x in wts_k])

            ivecs = model.item(cand, feats)
            corr = torch.cat([torch.zeros(len(b)), log_q[negs]])
            logits = uvecs @ ivecs.t() - corr.unsqueeze(0)

            # Suppress duplicates of the true positive among the candidates.
            dup = cand.unsqueeze(0) == pos.unsqueeze(1)
            dup[torch.arange(len(b)), torch.arange(len(b))] = False
            logits = logits.masked_fill(dup, -1e9)

            target = torch.arange(len(b))
            loss = (F.cross_entropy(logits, target, reduction="none")
                    * tw_t[b]).mean()
            opt.zero_grad()
            loss.backward()
            torch.nn.utils.clip_grad_norm_(model.parameters(), 5.0)
            opt.step()
            sched.step()
            running += float(loss.detach()) * len(b)
            seen_n += len(b)
            if step % 300 == 0:
                log(f"epoch {epoch} step {step}/{len(batches)} "
                    f"loss {running / max(seen_n, 1):.4f}", t0)
        log(f"epoch {epoch} done, loss {running / max(seen_n, 1):.4f}", t0)

'''
s = s.replace(old_loop, new_loop)

s=s.replace('''    model = TwoTower(len(vocab), n_catalog, log_pop, dim=DIM, hidden=HIDDEN,''',
            '''    model = TwoTower(len(vocab), n_catalog, ft.N_POP_FEATS, dim=DIM, hidden=HIDDEN,''')
s=s.replace('''    n = len(tu)
    steps_per_epoch = max(1, n // BATCH)''','''    steps_per_epoch = max(1, sum(int((tslice == k).sum()) // BATCH
                                 for k in range(len(SLICE_DAYS))))''')

s=s.replace('''        item_vecs = torch.cat([model.item(col_vocab_t[s:s + 4096])
                               for s in range(0, len(col_vocab), 4096)])''',
'''        final_feats = torch.from_numpy(pop.features(final_cutoff))
        item_vecs = torch.cat([
            model.item(col_vocab_t[s:s + 4096], final_feats[col_vocab_t[s:s + 4096]])
            for s in range(0, len(col_vocab), 4096)])''')
open(p,'w').write(s)
EOF
python -c "import ast;ast.parse(open('run.py').read());print('ok')" && grep -n "log_pop\|steps_per_epoch\|slice_feats" run.py

cat /tmp/claude-1000/-app/53bed668-7041-4440-8306-d01153e497d0/tasks/bah2z5a7c.output; ls -la /app/solution/

ps aux | grep -c "[r]un.py"; sleep 60; ls -la /app/solution/*.parquet

sleep 90; ls -la /app/solution/validation.parquet 2>/dev/null && python /task/evaluate_public.py /app/solution/validation.parquet || echo "still running"

python - <<'EOF'
p='run.py'; s=open(p).read()

# factor inference into a function
old = s[s.index('    # ---- Inference'):]
new = '''    # ---- Inference -------------------------------------------------------
    _, path = rank_and_write(model, store, vocab_map, pop, final_cutoff, stage,
                             users, out_path)
    log(f"wrote {path}", t0)


def rank_and_write(model, store, vocab_map, pop, final_cutoff, stage, users,
                   out_path):
    """Score every eligible candidate and write the top-100 ranking."""
    model.eval()
    ids, wts, _ = ft.build_profiles(store, vocab_map, final_cutoff)
    targets = du.target_users()
    rows = np.searchsorted(users, targets)
    eu, eitems, emask = du.eligibility_matrix(stage)
    col_vocab = ft._map(vocab_map, eitems)
    col_vocab_t = torch.from_numpy(col_vocab)
    urow = {int(u): r for r, u in enumerate(eu)}

    with torch.no_grad():
        feats = torch.from_numpy(pop.features(final_cutoff))
        item_vecs = torch.cat([
            model.item(col_vocab_t[s:s + 4096], feats[col_vocab_t[s:s + 4096]])
            for s in range(0, len(col_vocab), 4096)])
        out_u, out_i, out_r = [], [], []
        for s in range(0, len(targets), 512):
            bt = torch.from_numpy(rows[s:s + 512])
            uv = model.user([torch.from_numpy(a)[bt] for a in ids],
                            [torch.from_numpy(a)[bt] for a in wts])
            scores = (uv @ item_vecs.t()).numpy()
            for j, uid in enumerate(targets[s:s + 512]):
                sc = np.where(emask[urow[int(uid)]], scores[j], -np.inf)
                top = np.argpartition(-sc, 100)[:100]
                top = top[np.argsort(-sc[top])]
                out_u.append(np.full(100, uid, np.int64))
                out_i.append(eitems[top].astype(np.int64))
                out_r.append(np.arange(1, 101, dtype=np.int64))
    frame = pd.DataFrame({
        "uid": np.concatenate(out_u),
        "item_id": np.concatenate(out_i),
        "rank": np.concatenate(out_r),
    })
    out_path.parent.mkdir(parents=True, exist_ok=True)
    frame.to_parquet(out_path, index=False)
    model.train()
    return frame, out_path


if __name__ == "__main__":
    main()
'''
s = s[:s.index('    # ---- Inference')] + new
open(p,'w').write(s)
EOF
python -c "import ast;ast.parse(open('run.py').read());print('ok')"; tail -5 run.py

python - <<'EOF'
p='run.py'; s=open(p).read()
s=s.replace('''        log(f"epoch {epoch} done, loss {running / max(seen_n, 1):.4f}", t0)
''','''        log(f"epoch {epoch} done, loss {running / max(seen_n, 1):.4f}", t0)
        if stage == "validation":
            ep_path = out_path.with_name(f"{out_path.stem}_ep{epoch}.parquet")
            rank_and_write(model, store, vocab_map, pop, final_cutoff, stage,
                           users, ep_path)
            log(f"epoch {epoch} val ndcg@10 = {score_public(ep_path):.5f}", t0)
''')
s=s.replace('''def log(msg: str, t0: float) -> None:''','''def score_public(path: Path) -> float:
    """Score a validation ranking with the public evaluator."""
    sys.path.insert(0, "/task")
    import evaluate_public as ev
    targets = set(pd.read_parquet("/task/data/target_users.parquet").uid.astype(int))
    if not hasattr(score_public, "_truth"):
        elig = ev.CandidateSet(Path("/task/data/validation_eligible_candidates.npz"))
        score_public._truth = (elig, ev.validation_truth(elig, targets))
    elig, truth = score_public._truth
    ranked = ev.load_submission(path, elig, targets)
    return float(ev.score(ranked, truth)["quality_ndcg@10"])


def log(msg: str, t0: float) -> None:''')
open(p,'w').write(s)
EOF
python -c "import ast;ast.parse(open('run.py').read());print('ok')"

(SLICES=7,14,21,28 EPOCHS=4 python run.py validation /app/solution/v2.parquet > /app/solution/v2.log 2>&1; echo DONE >> /app/solution/v2.log) &
sleep 115; grep -E "slice|epoch .* val|loaded|vocab" /app/solution/v2.log | tail -20

python - <<'EOF'
p='run.py'; s=open(p).read()
s=s.replace('LR = 3e-3\nDIM = 64\nOUT_DIM = 64\nHIDDEN = 256',
 'LR = float(os.environ.get("LR", "3e-3"))\nDIM = int(os.environ.get("DIM", "64"))\nOUT_DIM = int(os.environ.get("OUT_DIM", "64"))\nHIDDEN = int(os.environ.get("HIDDEN", "256"))\nDROPOUT = float(os.environ.get("DROPOUT", "0.1"))\nWD = float(os.environ.get("WD", "1e-5"))\nN_NEG = int(os.environ.get("NNEG", "2000"))')
s=s.replace('N_NEG = 2000\n','')
s=s.replace('out_dim=OUT_DIM, n_channels=ft.N_CH)','out_dim=OUT_DIM, n_channels=ft.N_CH, dropout=DROPOUT)')
s=s.replace('weight_decay=1e-5)','weight_decay=WD)')
open(p,'w').write(s)
EOF
grep -n "^LR\|^DIM\|^DROPOUT\|^WD\|^N_NEG\|dropout=DROPOUT\|weight_decay" run.py

(SLICES=7,14,21,28,35,42,49,56 EPOCHS=3 LR=1.5e-3 DIM=48 OUT_DIM=48 DROPOUT=0.3 WD=1e-4 NNEG=3000 python run.py validation /app/solution/v3.parquet > /app/solution/v3.log 2>&1; echo DONE >> /app/solution/v3.log) &
sleep 110; grep -E "epoch .* val|DONE|Error|Trace" /app/solution/v3.log | tail

sleep 105; grep -E "epoch .* val|DONE" /app/solution/v3.log | tail

(SLICES=7,14,21,28 EPOCHS=1 NNEG=3000 python run.py validation /app/solution/v4.parquet > v4.log 2>&1; echo DONE >> v4.log) & sleep 105; grep -E "epoch .* val|DONE|wrote" /app/solution/v4.log

cp /app/solution/v4.parquet /app/solution/validation.parquet
(SLICES=7,14,21,28 EPOCHS=1 NNEG=3000 python run.py test /app/solution/submission_nn.parquet > test.log 2>&1; echo DONE >> test.log) &
sleep 5; echo launched

/app/solution/README.md

# Two-tower discovery ranker — "Made for You"

## Architecture

A PyTorch two-tower network (`model.py`). The final relevance score is exactly
`dot(user_vector, item_vector)` with both vectors of dimension 64, so item
vectors are precomputable offline into a MIPS index and request-time work is a
single user-tower forward pass.

**Shared embedding table** — 80,000 rows: the 30,000 retrieval-catalog tracks
followed by the 50,000 most-played non-catalog tracks (history-only vocabulary),
`dim = 64`.

**User tower** (`UserTower`) — no user-ID lookup. It reads four weighted bags of
history item ids, pooled with `F.embedding_bag`:

| Channel | Content | Cap | Weighting |
| --- | --- | --- | --- |
| 0 | most recent positive listens (`played_ratio_pct >= 50`) | 192 | 21-day recency half-life |
| 1 | habitual favourites: top items by lifetime play count | 96 | `log1p(count)` |
| 2 | most recent likes | 96 | uniform |
| 3 | most recent dislikes | 32 | uniform |

The four pooled vectors are concatenated (256-d) and passed through
`Linear(256→256) → GELU → Dropout → Linear(256→256) → GELU → Linear(256→63)`.
A constant `1.0` is appended as the 64th coordinate.

**Item tower** (`ItemTower`) — input is the item embedding concatenated with six
cutoff-correct popularity features (log plays in the trailing 7d / 28d /
all-time window, organic 80%-completion listens and organic likes in 28d, and a
7d-vs-28d trend ratio), all standardized. Two heads:
`Linear→GELU→Dropout→Linear` producing 63 dims, plus a small MLP producing the
64th coordinate, which acts as a learned item bias against the user tower's
pinned `1.0`. The item vector depends only on the item and the serving cutoff,
never on the user, so the offline/online split holds.

## Training objective

Training slices reproduce the evaluation protocol rather than a global
interaction matrix — nothing is materialized or factorized over users × items,
so the method scales to millions of users.

For each cutoff `T ∈ {C−7d, C−14d, C−21d, C−28d}` (`C` = the stage's cutoff):

* the user tower sees only events strictly before `T`;
* positives are the user's **new** items (never listened to before `T`) with an
  organic like (weight 2.0) or an organic ≥80%-completion listen (weight 1.0) in
  `(T, T+7d]`, restricted to the retrieval catalog.

This yields ~198k (history, positive) pairs. The loss is a **sampled softmax**:
per batch of 1,024 pairs, 3,000 shared negatives are drawn from `popularity^0.75`
and combined with the in-batch positives; logits carry a `−log q` sampling
correction, duplicates of the true positive are masked out, and the
cross-entropy is weighted by the grade weight. Batches are drawn within a single
slice so every candidate shares one cutoff-correct popularity feature vector.
AdamW, OneCycle LR (max 3e-3), 1 epoch — held-out validation NDCG degrades
monotonically past one epoch, so training stops there.

Ranking masks each user's ineligible items to `−inf` and emits the top 100.

## Results (public validation window)

| Model | NDCG@10 |
| --- | --- |
| Popularity baseline | 0.01071 |
| **Two-tower (this model)** | **0.03061** |

Scored over the 4,263 users with at least one eligible relevant item.

## Commands

```sh
# validation-stage ranker: history = train_* only, scored against val_*
python run.py validation solution/validation.parquet
python /task/evaluate_public.py /app/solution/validation.parquet

# final hidden-test ranker: refit on train_* + val_*, val events are history
python run.py test /app/solution/submission.parquet
```

## Files

* `data_utils.py` — parquet/eligibility loading.
* `features.py` — cutoff-correct user history channels, training targets,
  vocabulary, and `PopStats` item popularity features.
* `model.py` — `UserTower`, `ItemTower`, `TwoTower`.
* `run.py` — end-to-end train + rank for either stage.
* `validation.parquet` — validation-stage ranking (score above).
* `submission.parquet` — final hidden-test deliverable.

sleep 75; cat /app/solution/test.log | grep -E "slice|epoch|wrote|DONE|Error|Trace" | tail

python -c "
import sys; sys.path.insert(0,'/task')
from pathlib import Path
import pandas as pd, numpy as np
from ranking_evaluation import CandidateSet, validate_ranking, read_ranking
tg = set(pd.read_parquet('/task/data/target_users.parquet').uid.astype(int))
f = validate_ranking(read_ranking(Path('submission_nn.parquet')), tg)
el = CandidateSet(Path('/task/data/test_eligible_candidates.npz'))
bad = sum(1 for u,i in f[['uid','item_id']].itertuples(index=False,name=None) if not el.is_eligible(u,i))
print('rows',len(f),'users',f.uid.nunique(),'ineligible',bad)
assert bad==0
f.to_parquet('submission.parquet',index=False)
print('installed submission.parquet')
" && rm -f smoke.parquet v2.parquet v3.parquet validation_pop.parquet v2_ep*.parquet v3_ep*.parquet v4_ep*.parquet validation_ep*.parquet && ls -la

rm -f v4.parquet submission_nn.parquet v2.log v3.log v4.log && python /task/evaluate_public.py /app/solution/validation.parquet