Agent trace

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

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

mkdir -p solution && python -c "
import pandas as pd, numpy as np, time
t=time.time()
tu=pd.read_parquet('/task/data/target_users.parquet'); print(tu.shape, tu.columns.tolist(), tu.head())
cc=pd.read_parquet('/task/data/candidate_catalog.parquet'); print(cc.shape, cc.columns.tolist()); print(cc.head())
ps=pd.read_parquet('/task/data/validation_preference_state.parquet'); print(ps.shape, ps.columns.tolist()); print(ps.head())
z=np.load('/task/data/validation_eligible_candidates.npz'); print({k:(v.shape,v.dtype) for k,v in z.items()})
import pyarrow.parquet as pq
print(pq.ParquetFile('/task/data/train_listens.parquet').metadata)
print(time.time()-t)
"

cd /app/solution && python -c "
import pandas as pd, numpy as np, time
t=time.time()
for f in ['train_likes','train_dislikes','train_unlikes','val_listens','val_likes']:
    d=pd.read_parquet(f'/task/data/{f}.parquet'); print(f, d.shape, d.dtypes.to_dict())
print(time.time()-t)
"

python -c "
import numpy as np, pandas as pd, time
t=time.time()
z=np.load('/task/data/validation_eligible_candidates.npz')
p=z['packed_eligible']; items=z['item_ids']; uids=z['uids']
m=np.unpackbits(p,axis=1,bitorder='little')[:,:len(items)]
print('elig per user', m.sum(1).mean(), m.sum(1).min(), m.sum(1).max())
zt=np.load('/task/data/test_eligible_candidates.npz')
print('test uids equal', np.array_equal(zt['uids'],uids), np.array_equal(zt['item_ids'],items))
mt=np.unpackbits(zt['packed_eligible'],axis=1,bitorder='little')[:,:len(items)]
print('test elig', mt.sum(1).mean())
print('overlap', (m&mt).sum()/m.sum())
print(time.time()-t)
"

python -c "
import pandas as pd, numpy as np, time
t=time.time()
tu=pd.read_parquet('/task/data/target_users.parquet').uid.astype(np.uint32).values
tus=set(tu.tolist())
vl=pd.read_parquet('/task/data/val_listens.parquet')
print('val users', vl.uid.nunique(), 'target in val', vl.uid.isin(tus).mean())
tl=pd.read_parquet('/task/data/train_listens.parquet', columns=['uid','timestamp','item_id','is_organic','played_ratio_pct'])
print('train shape', tl.shape, 'users', tl.uid.nunique(), time.time()-t)
print('ts range', tl.timestamp.min(), tl.timestamp.max())
print('target frac', tl.uid.isin(tus).mean())
tl.to_pickle('/tmp/none') if False else None
print(time.time()-t)
" 2>&1 | tail -20

cat > prep.py <<'EOF'
import pandas as pd, numpy as np, time, pickle
t=time.time()
D='/task/data/'
cat=pd.read_parquet(D+'candidate_catalog.parquet')
items=np.load(D+'validation_eligible_candidates.npz')['item_ids'].astype(np.int64)
uids=np.load(D+'validation_eligible_candidates.npz')['uids'].astype(np.int64)
imap=pd.Series(np.arange(len(items)),index=items)
umap=pd.Series(np.arange(len(uids)),index=uids)
def load(name, cols):
    d=pd.read_parquet(D+name+'.parquet', columns=cols)
    d['u']=umap.reindex(d.uid.values).values
    d['i']=imap.reindex(d.item_id.values).values
    d=d[d.u.notna()&d.i.notna()]
    d['u']=d.u.astype(np.int32); d['i']=d.i.astype(np.int32)
    return d.drop(columns=['uid','item_id'])
out={}
for split in ['train','val']:
    out[split+'_listens']=load(split+'_listens',['uid','timestamp','item_id','is_organic','played_ratio_pct'])
    for f in ['likes','dislikes','unlikes','undislikes']:
        out[split+'_'+f]=load(split+'_'+f,['uid','timestamp','item_id','is_organic'])
    print(split, time.time()-t)
with open('/app/solution/cache.pkl','wb') as f: pickle.dump({'items':items,'uids':uids,'cat':cat,**out},f,protocol=4)
print('done',time.time()-t)
EOF
python prep.py

python -c "
import pickle,numpy as np,pandas as pd,time
t=time.time()
c=pickle.load(open('cache.pkl','rb'))
print({k:(v.shape if hasattr(v,'shape') else len(v)) for k,v in c.items()})
tl=c['train_listens']; vl=c['val_listens']
nU=len(c['uids']); nI=len(c['items'])
# relevance
vs=vl[(vl.is_organic==1)&(vl.played_ratio_pct>=80)][['u','i']].drop_duplicates()
vlk=c['val_likes']; vlk=vlk[vlk.is_organic==1][['u','i']].drop_duplicates()
print('rel strong',len(vs),'likes',len(vlk))
# history pairs
hist=tl[tl.played_ratio_pct>=50][['u','i']].drop_duplicates()
hs=set(map(tuple,hist.values.tolist()))
print('repeat frac of strong rel', np.mean([tuple(x) in hs for x in vs.values.tolist()[:50000]]))
print('repeat frac of like rel', np.mean([tuple(x) in hs for x in vlk.values.tolist()[:50000]]))
print(time.time()-t)
"

cat > feats.py <<'PYEOF'
import numpy as np, pandas as pd, pickle, time
from scipy import sparse

NI=30000
DAY=86400.0

def cat_events(c, splits, kind):
    return pd.concat([c[s+'_'+kind] for s in splits], ignore_index=True)

def active_state(likes, unlikes, dislikes, undislikes, cutoff):
    """last-wins active like/dislike state per (u,i)"""
    def last(df, v):
        d=df[df.timestamp<cutoff]
        if len(d)==0: return pd.DataFrame({'k':[],'ts':[],'v':[]})
        k=d.u.values.astype(np.int64)*NI+d.i.values
        return pd.DataFrame({'k':k,'ts':d.timestamp.values.astype(np.int64),'v':v})
    lk=pd.concat([last(likes,1),last(unlikes,0)],ignore_index=True)
    dk=pd.concat([last(dislikes,1),last(undislikes,0)],ignore_index=True)
    res={}
    for name,df in [('like',lk),('dis',dk)]:
        if len(df)==0: res[name]=(np.array([],dtype=np.int64),np.array([],dtype=np.int64)); continue
        df=df.sort_values('ts').groupby('k').tail(1)
        df=df[df.v==1]
        res[name]=(df.k.values, df.ts.values)
    return res

def build_features(c, hist_splits, cutoff, n_extra=200):
    t0=time.time()
    L=cat_events(c,hist_splits,'listens'); L=L[L.timestamp<cutoff]
    likes=cat_events(c,hist_splits,'likes'); unlikes=cat_events(c,hist_splits,'unlikes')
    dis=cat_events(c,hist_splits,'dislikes'); undis=cat_events(c,hist_splits,'undislikes')
    nU=len(c['uids'])
    u=L.u.values.astype(np.int64); i=L.i.values.astype(np.int64)
    ts=L.timestamp.values.astype(np.float64); pr=L.played_ratio_pct.values.astype(np.float32)
    org=L.is_organic.values.astype(np.float32)
    age=(cutoff-ts)/DAY
    key=u*NI+i
    upairs, code = np.unique(key, return_inverse=True)
    P=len(upairs)
    pu=(upairs//NI).astype(np.int32); pi=(upairs%NI).astype(np.int32)
    F={}
    def bc(w): return np.bincount(code,weights=w,minlength=P)
    pos=(pr>=50); strong=(pr>=80)
    F['n_all']=bc(None if False else np.ones(len(u)))
    F['n_pos']=bc(pos.astype(np.float64))
    F['n_strong']=bc(strong.astype(np.float64))
    F['n_org_strong']=bc((strong&(org>0)).astype(np.float64))
    F['n_org']=bc(org.astype(np.float64))
    F['sum_pr']=bc(pr.astype(np.float64))
    F['mean_pr']=F['sum_pr']/np.maximum(F['n_all'],1)
    for hl,nm in [(1.0,'d1'),(3.0,'d3'),(7.0,'d7'),(30.0,'d30')]:
        w=np.exp(-age/hl)
        F['dec_'+nm]=bc(w)
        F['decs_'+nm]=bc(w*strong)
    for wd in [1,3,7,14,30,90]:
        m=(age<wd)
        F[f'c{wd}']=bc(m.astype(np.float64))
        F[f'cs{wd}']=bc((m&strong).astype(np.float64))
    # recency
    neg=-age
    rec=np.full(P,-1e4); np.maximum.at(rec,code,neg); F['last_age']=-rec
    recs=np.full(P,-1e4); np.maximum.at(recs,code,np.where(strong,neg,-1e4)); F['last_strong_age']=-recs
    reco=np.full(P,-1e4); np.maximum.at(reco,code,np.where(org>0,neg,-1e4)); F['last_org_age']=-reco
    first=np.full(P,1e4); np.minimum.at(first,code,age); F['first_age']=first
    # user aggregates
    ucnt=np.bincount(u,minlength=nU).astype(np.float64)
    ucnt7=np.bincount(u,weights=(age<7).astype(np.float64),minlength=nU)
    ucnt_s=np.bincount(u,weights=strong.astype(np.float64),minlength=nU)
    udist=np.bincount(pu,minlength=nU).astype(np.float64)
    uorg=np.bincount(u,weights=org.astype(np.float64),minlength=nU)/np.maximum(ucnt,1)
    # item aggregates
    icnt=np.bincount(i,minlength=NI).astype(np.float64)
    icnt7=np.bincount(i,weights=(age<7).astype(np.float64),minlength=NI)
    icnt_s=np.bincount(i,weights=strong.astype(np.float64),minlength=NI)
    icnt_os=np.bincount(i,weights=(strong&(org>0)).astype(np.float64),minlength=NI)
    iusers=np.bincount(pi,minlength=NI).astype(np.float64)
    irepeat=icnt/np.maximum(iusers,1)
    istrength=icnt_s/np.maximum(icnt,1)
    iorg=np.bincount(i,weights=org.astype(np.float64),minlength=NI)/np.maximum(icnt,1)
    # state
    st=active_state(likes,unlikes,dis,undis,cutoff)
    lk_k,lk_ts=st['like']; ds_k,ds_ts=st['dis']
    # CF: item-item cosine on strong-listen matrix, capped per user
    ord_=np.lexsort((-F['decs_d30'],pu))
    cap=np.zeros(P,dtype=bool)
    rk=np.arange(P)-np.searchsorted(pu[ord_],pu[ord_])
    cap[ord_[rk<300]]=True
    sel=cap&(F['n_strong']>0)
    val=np.sqrt(F['n_strong'][sel])
    X=sparse.csr_matrix((val,(pu[sel],pi[sel])),shape=(nU,NI),dtype=np.float32)
    norm=np.sqrt(np.asarray(X.multiply(X).sum(0)).ravel())+1e-6
    Xn=X.multiply(sparse.csr_matrix(1.0/norm)).tocsr().astype(np.float32)
    G=(Xn.T@Xn).tocsr()
    G.setdiag(0); G.eliminate_zeros()
    # recent-profile user vector
    valr=F['dec_d30'][sel].astype(np.float32)
    Xr=sparse.csr_matrix((valr,(pu[sel],pi[sel])),shape=(nU,NI),dtype=np.float32)
    Xr=sparse.diags(1.0/(np.asarray(Xr.sum(1)).ravel()+1e-6)).astype(np.float32)@Xr
    S=(Xr@G)  # nU x NI dense-ish
    S=np.asarray(S.todense()) if sparse.issparse(S) else S
    print('cf done',time.time()-t0, S.shape)
    # likes-based CF
    lkh=likes[likes.timestamp<cutoff]
    Xl=sparse.csr_matrix((np.ones(len(lkh),dtype=np.float32),(lkh.u.values,lkh.i.values)),shape=(nU,NI))
    Xl.data[:]=1.0
    nl=np.sqrt(np.asarray(Xl.multiply(Xl).sum(0)).ravel())+1e-6
    Xln=Xl.multiply(sparse.csr_matrix(1.0/nl)).tocsr().astype(np.float32)
    Gl=(Xln.T@Xln).tocsr(); Gl.setdiag(0); Gl.eliminate_zeros()
    Xlu=sparse.diags(1.0/(np.asarray(Xl.sum(1)).ravel()+1e-6)).astype(np.float32)@Xl
    Sl=np.asarray((Xlu@Gl).todense())
    print('cf2 done',time.time()-t0)
    return dict(pu=pu,pi=pi,F=F,ucnt=ucnt,ucnt7=ucnt7,ucnt_s=ucnt_s,udist=udist,uorg=uorg,
                icnt=icnt,icnt7=icnt7,icnt_s=icnt_s,icnt_os=icnt_os,iusers=iusers,irepeat=irepeat,
                istrength=istrength,iorg=iorg,lk_k=lk_k,lk_ts=lk_ts,ds_k=ds_k,ds_ts=ds_ts,
                S=S,Sl=Sl,cutoff=cutoff,nU=nU,n_extra=n_extra)
PYEOF
echo ok

cat >> feats.py <<'PYEOF'

FEATNAMES=None

def assemble(B, eligible_mask):
    global FEATNAMES
    t0=time.time()
    nU=B['nU']; F=B['F']; pu=B['pu']; pi=B['pi']; S=B['S']; Sl=B['Sl']
    P=len(pu)
    # discovery candidates: top n_extra by blended CF among non-history eligible
    blend=S+0.5*Sl
    blend=np.where(eligible_mask,blend,-1e9)
    bh=blend.copy()
    bh[pu,pi]=-1e9
    ne=B['n_extra']
    idx=np.argpartition(-bh,ne,axis=1)[:,:ne]
    eu=np.repeat(np.arange(nU,dtype=np.int32),ne); ei=idx.ravel().astype(np.int32)
    keep=bh[eu,ei]>-1e8
    eu=eu[keep]; ei=ei[keep]
    cu=np.concatenate([pu,eu]); ci=np.concatenate([pi,ei])
    ish=np.concatenate([np.ones(P,dtype=np.float32),np.zeros(len(eu),dtype=np.float32)])
    E=len(cu)
    def ext(a):
        return np.concatenate([a,np.zeros(E-P,dtype=a.dtype)])
    cols={}; 
    for k,v in F.items(): cols[k]=ext(v.astype(np.float32))
    for k in ['last_age','last_strong_age','last_org_age']:
        cols[k]=np.where(ish>0,cols[k],999.0)
        cols[k]=np.minimum(cols[k],999.0)
    cols['first_age']=np.where(ish>0,np.minimum(cols['first_age'],999.0),999.0)
    cols['is_hist']=ish
    # user feats
    for nm,arr in [('u_cnt',B['ucnt']),('u_cnt7',B['ucnt7']),('u_cnt_s',B['ucnt_s']),('u_dist',B['udist']),('u_org',B['uorg'])]:
        cols[nm]=arr[cu].astype(np.float32)
    for nm,arr in [('i_cnt',B['icnt']),('i_cnt7',B['icnt7']),('i_cnt_s',B['icnt_s']),('i_cnt_os',B['icnt_os']),
                   ('i_users',B['iusers']),('i_rep',B['irepeat']),('i_str',B['istrength']),('i_org',B['iorg'])]:
        cols[nm]=arr[ci].astype(np.float32)
    cols['cf']=S[cu,ci]; cols['cf_l']=Sl[cu,ci]
    # normalized shares
    cols['share']=cols['n_all']/np.maximum(cols['u_cnt'],1)
    cols['share_s']=cols['n_strong']/np.maximum(cols['u_cnt_s'],1)
    cols['share7']=cols['c7']/np.maximum(cols['u_cnt7'],1)
    cols['i_pop_rank']=np.log1p(ci.astype(np.float32))
    # state
    ck=cu.astype(np.int64)*NI+ci
    order=np.argsort(ck); cks=ck[order]
    def statefeat(k_arr,ts_arr):
        f=np.zeros(E,dtype=np.float32); fa=np.full(E,999.0,dtype=np.float32)
        if len(k_arr)==0: return f,fa
        p=np.searchsorted(cks,k_arr)
        ok=(p<E)&(cks[np.minimum(p,E-1)]==k_arr)
        pos=order[p[ok]]
        f[pos]=1.0; fa[pos]=(B['cutoff']-ts_arr[ok])/DAY
        return f,fa
    cols['liked'],cols['liked_age']=statefeat(B['lk_k'],B['lk_ts'])
    cols['disliked'],cols['disliked_age']=statefeat(B['ds_k'],B['ds_ts'])
    names=sorted(cols.keys())
    FEATNAMES=names
    X=np.column_stack([cols[n] for n in names]).astype(np.float32)
    print('assemble',time.time()-t0,X.shape)
    return cu,ci,X,names
PYEOF
echo ok

cat > run.py <<'PYEOF'
import numpy as np, pandas as pd, pickle, time, sys
from feats import build_features, assemble, NI, DAY
t0=time.time()
c=pickle.load(open('/app/solution/cache.pkl','rb'))
nU=len(c['uids'])
ez=np.load('/task/data/validation_eligible_candidates.npz')
Mval=np.unpackbits(ez['packed_eligible'],axis=1,bitorder='little')[:,:NI].astype(bool)
ezt=np.load('/task/data/test_eligible_candidates.npz')
Mtest=np.unpackbits(ezt['packed_eligible'],axis=1,bitorder='little')[:,:NI].astype(bool)
TRAIN_CUT=24786800; WIN=604800; GAP=1800

def labels(splits, lo, hi):
    L=pd.concat([c[s+'_listens'] for s in splits],ignore_index=True)
    L=L[(L.timestamp>=lo)&(L.timestamp<hi)&(L.is_organic==1)&(L.played_ratio_pct>=80)]
    K=pd.concat([c[s+'_likes'] for s in splits],ignore_index=True)
    K=K[(K.timestamp>=lo)&(K.timestamp<hi)&(K.is_organic==1)]
    g={}
    k1=(L.u.values.astype(np.int64)*NI+L.i.values).astype(np.int64)
    k3=(K.u.values.astype(np.int64)*NI+K.i.values).astype(np.int64)
    return np.unique(k1), np.unique(k3)

def grade_of(cu,ci,k1,k3):
    ck=cu.astype(np.int64)*NI+ci
    g=np.zeros(len(ck),dtype=np.int8)
    g[np.isin(ck,k1)]=1
    g[np.isin(ck,k3)]=3
    return g

# --- pseudo-train fold ---
T1=TRAIN_CUT-WIN-GAP
print('building pseudo-train',flush=True)
B1=build_features(c,['train'],T1)
cu1,ci1,X1,names=assemble(B1,Mval)
k1,k3=labels(['train'],T1+GAP,T1+GAP+WIN)
y1=grade_of(cu1,ci1,k1,k3)
print('fold1',X1.shape,(y1>0).mean(),(y1==3).sum(),time.time()-t0,flush=True)
del B1
np.save('/tmp/x1.npy',X1)
pickle.dump((cu1,ci1,y1,names),open('/tmp/f1.pkl','wb'))
PYEOF
python run.py 2>&1 | tail -20

python run.py; echo "EXIT=$?"

python - <<'EOF'
import re
s=open('/app/solution/feats.py').read()
s=s.replace("""    G=(Xn.T@Xn).tocsr()
    G.setdiag(0); G.eliminate_zeros()
    # recent-profile user vector
    valr=F['dec_d30'][sel].astype(np.float32)
    Xr=sparse.csr_matrix((valr,(pu[sel],pi[sel])),shape=(nU,NI),dtype=np.float32)
    Xr=sparse.diags(1.0/(np.asarray(Xr.sum(1)).ravel()+1e-6)).astype(np.float32)@Xr
    S=(Xr@G)  # nU x NI dense-ish
    S=np.asarray(S.todense()) if sparse.issparse(S) else S
    print('cf done',time.time()-t0, S.shape)""",
"""    valr=F['dec_d30'][sel].astype(np.float32)
    Xr=sparse.csr_matrix((valr,(pu[sel],pi[sel])),shape=(nU,NI),dtype=np.float32)
    Xr=sparse.diags(1.0/(np.asarray(Xr.sum(1)).ravel()+1e-6)).astype(np.float32)@Xr
    UU=np.asarray(([email protected]).todense(),dtype=np.float32)
    np.fill_diagonal(UU,0.0)
    [email protected](Xn.todense(),dtype=np.float32)
    del UU
    print('cf done',time.time()-t0, S.shape)""")
s=s.replace("""    Gl=(Xln.T@Xln).tocsr(); Gl.setdiag(0); Gl.eliminate_zeros()
    Xlu=sparse.diags(1.0/(np.asarray(Xl.sum(1)).ravel()+1e-6)).astype(np.float32)@Xl
    Sl=np.asarray((Xlu@Gl).todense())""",
"""    Xlu=sparse.diags(1.0/(np.asarray(Xl.sum(1)).ravel()+1e-6)).astype(np.float32)@Xl
    UUl=np.asarray(([email protected]).todense(),dtype=np.float32)
    np.fill_diagonal(UUl,0.0)
    [email protected](Xln.todense(),dtype=np.float32)
    del UUl""")
# add capping of pairs at end of build_features
s=s.replace("""    print('cf2 done',time.time()-t0)
    return dict(""",
"""    print('cf2 done',time.time()-t0)
    prio=F['dec_d30']+0.3*F['dec_d7']+0.05*F['n_strong']+0.002*F['n_all']
    o=np.lexsort((-prio,pu))
    rank=np.arange(len(pu))-np.searchsorted(pu[o],pu[o])
    keep=np.zeros(len(pu),dtype=bool); keep[o[rank<cap_pairs]]=True
    pu=pu[keep]; pi=pi[keep]
    F={k:v[keep] for k,v in F.items()}
    print('pairs kept',keep.sum(),'of',len(keep))
    return dict(""")
s=s.replace("def build_features(c, hist_splits, cutoff, n_extra=200):","def build_features(c, hist_splits, cutoff, n_extra=250, cap_pairs=500):")
open('/app/solution/feats.py','w').write(s)
print('patched')
EOF
python run.py; echo "EXIT=$?"

python - <<'EOF'
s=open('feats.py').read()
s=s.replace("import numpy as np, pandas as pd, pickle, time\nfrom scipy import sparse",
"import numpy as np, pandas as pd, pickle, time, os, gc\nfrom scipy import sparse\ndef rss(tag):\n    import resource; print('RSS',tag,round(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/1e6,2),'GB',flush=True)")
s=s.replace("    upairs, code = np.unique(key, return_inverse=True)","    rss('pre-unique')\n    upairs, code = np.unique(key, return_inverse=True)\n    del key; gc.collect(); rss('post-unique')")
s=s.replace("    # user aggregates","    rss('post-F')\n    # user aggregates")
s=s.replace("    print('cf done',time.time()-t0, S.shape)","    rss('post-cf'); print('cf done',time.time()-t0, S.shape)")
open('feats.py','w').write(s)
EOF
cd /app/solution && timeout 600 python run.py; echo "EXIT=$?"

python - <<'EOF'
s=open('feats.py').read()
old_start=s.index("    key=u*NI+i")
old_end=s.index("    # user aggregates")
new="""    key=u*NI+i
    rss('pre-unique')
    upairs, code = np.unique(key, return_inverse=True)
    del key; gc.collect()
    P=len(upairs); print('P0',P,flush=True)
    pu=(upairs//NI).astype(np.int32); pi=(upairs%NI).astype(np.int32)
    pos=(pr>=50); strong=(pr>=80)
    prio=np.bincount(code,weights=np.exp(-age/30.0),minlength=P)+0.3*np.bincount(code,weights=np.exp(-age/7.0),minlength=P)+0.05*np.bincount(code,weights=strong.astype(np.float64),minlength=P)
    o=np.lexsort((-prio,pu))
    rank=np.arange(P)-np.searchsorted(pu[o],pu[o])
    keep=np.zeros(P,dtype=bool); keep[o[rank<cap_pairs]]=True
    del prio,o,rank; gc.collect()
    newidx=np.full(P,-1,dtype=np.int32); newidx[keep]=np.arange(keep.sum(),dtype=np.int32)
    # global aggregates from ALL events (before filtering)
    ucnt=np.bincount(u,minlength=nU).astype(np.float32)
    ucnt7=np.bincount(u,weights=(age<7).astype(np.float64),minlength=nU).astype(np.float32)
    ucnt_s=np.bincount(u,weights=strong.astype(np.float64),minlength=nU).astype(np.float32)
    udist=np.bincount(pu,minlength=nU).astype(np.float32)
    uorg=(np.bincount(u,weights=org.astype(np.float64),minlength=nU)/np.maximum(ucnt,1)).astype(np.float32)
    icnt=np.bincount(i,minlength=NI).astype(np.float32)
    icnt7=np.bincount(i,weights=(age<7).astype(np.float64),minlength=NI).astype(np.float32)
    icnt_s=np.bincount(i,weights=strong.astype(np.float64),minlength=NI).astype(np.float32)
    icnt_os=np.bincount(i,weights=(strong&(org>0)).astype(np.float64),minlength=NI).astype(np.float32)
    iusers=np.bincount(pi,minlength=NI).astype(np.float32)
    irepeat=icnt/np.maximum(iusers,1); istrength=icnt_s/np.maximum(icnt,1)
    iorg=(np.bincount(i,weights=org.astype(np.float64),minlength=NI)/np.maximum(icnt,1)).astype(np.float32)
    # CF matrices built from kept pairs (need F-lite): use strong counts
    nstrong_all=np.bincount(code,weights=strong.astype(np.float64),minlength=P)
    dec30_all=np.bincount(code,weights=np.exp(-age/30.0),minlength=P)
    del u,i,ts; gc.collect()
    # filter events
    em=keep[code]
    code=newidx[code[em]]; age=age[em]; pr=pr[em]; org=org[em]
    pos=pos[em]; strong=strong[em]
    pu_k=pu[keep]; pi_k=pi[keep]
    nstrong_k=nstrong_all[keep]; dec30_k=dec30_all[keep]
    del em; gc.collect()
    P=keep.sum(); print('Pkept',P,flush=True)
    F={}
    def bc(w): return np.bincount(code,weights=w,minlength=P).astype(np.float32)
    F['n_all']=bc(np.ones(len(age)))
    F['n_pos']=bc(pos.astype(np.float64))
    F['n_strong']=bc(strong.astype(np.float64))
    F['n_org_strong']=bc((strong&(org>0)).astype(np.float64))
    F['n_org']=bc(org.astype(np.float64))
    F['mean_pr']=bc(pr.astype(np.float64))/np.maximum(F['n_all'],1)
    for hl,nm in [(1.0,'d1'),(3.0,'d3'),(7.0,'d7'),(30.0,'d30')]:
        w=np.exp(-age/hl); F['dec_'+nm]=bc(w); F['decs_'+nm]=bc(w*strong)
    for wd in [1,3,7,14,30,90]:
        m=(age<wd); F[f'c{wd}']=bc(m.astype(np.float64)); F[f'cs{wd}']=bc((m&strong).astype(np.float64))
    neg=-age
    rec=np.full(P,-1e4); np.maximum.at(rec,code,neg); F['last_age']=(-rec).astype(np.float32)
    recs=np.full(P,-1e4); np.maximum.at(recs,code,np.where(strong,neg,-1e4)); F['last_strong_age']=(-recs).astype(np.float32)
    reco=np.full(P,-1e4); np.maximum.at(reco,code,np.where(org>0,neg,-1e4)); F['last_org_age']=(-reco).astype(np.float32)
    first=np.full(P,1e4); np.minimum.at(first,code,age); F['first_age']=first.astype(np.float32)
    del rec,recs,reco,first,age,pr,org,pos,strong,code; gc.collect()
    pu=pu_k; pi=pi_k
    rss('post-F')
"""
s=s[:old_start]+new+s[old_end:]
# remove the old aggregate block (now duplicated) -> old block between '# user aggregates' and '# state'
a=s.index("    # user aggregates"); b=s.index("    # state\n")
s=s[:a]+s[b:]
# fix CF section references
s=s.replace("""    ord_=np.lexsort((-F['decs_d30'],pu))
    cap=np.zeros(P,dtype=bool)
    rk=np.arange(P)-np.searchsorted(pu[ord_],pu[ord_])
    cap[ord_[rk<300]]=True
    sel=cap&(F['n_strong']>0)
    val=np.sqrt(F['n_strong'][sel])""","""    sel=(nstrong_k>0)
    val=np.sqrt(nstrong_k[sel]).astype(np.float32)""")
s=s.replace("    valr=F['dec_d30'][sel].astype(np.float32)","    valr=dec30_k[sel].astype(np.float32)")
s=s.replace("""    prio=F['dec_d30']+0.3*F['dec_d7']+0.05*F['n_strong']+0.002*F['n_all']
    o=np.lexsort((-prio,pu))
    rank=np.arange(len(pu))-np.searchsorted(pu[o],pu[o])
    keep=np.zeros(len(pu),dtype=bool); keep[o[rank<cap_pairs]]=True
    pu=pu[keep]; pi=pi[keep]
    F={k:v[keep] for k,v in F.items()}
    print('pairs kept',keep.sum(),'of',len(keep))
""","")
open('feats.py','w').write(s)
print(s[s.index('def build_features'):s.index('def assemble')])
EOF

timeout 900 python run.py; echo "EXIT=$?"

cat > safety.py <<'PYEOF'
import numpy as np, pandas as pd, pickle, time
NI=30000; DAY=86400.0
c=pickle.load(open('/app/solution/cache.pkl','rb'))
uids=c['uids']; items=c['items']; nU=len(uids)
CUT=24786800+1800+604800  # test history includes train+val
L=pd.concat([c['train_listens'],c['val_listens']],ignore_index=True)
u=L.u.values.astype(np.int64); i=L.i.values.astype(np.int64)
age=(CUT-L.timestamp.values.astype(np.float64))/DAY
strong=(L.played_ratio_pct.values>=80)
key=u*NI+i
up,code=np.unique(key,return_inverse=True)
P=len(up)
sc=np.bincount(code,weights=np.exp(-age/30.0)*(1+strong),minlength=P)+0.5*np.bincount(code,weights=np.exp(-age/7.0)*(1+strong),minlength=P)
pu=(up//NI).astype(np.int32); pi=(up%NI).astype(np.int32)
M=np.unpackbits(np.load('/task/data/test_eligible_candidates.npz')['packed_eligible'],axis=1,bitorder='little')[:,:NI].astype(bool)
pop=c['cat'].sort_values('popularity_rank').item_id.values
imap=pd.Series(np.arange(NI),index=items); popcol=imap.reindex(pop).values
ok=M[pu,pi]
pu,pi,sc=pu[ok],pi[ok],sc[ok]
o=np.lexsort((-sc,pu)); pu,pi,sc=pu[o],pi[o],sc[o]
starts=np.searchsorted(pu,np.arange(nU)); ends=np.searchsorted(pu,np.arange(nU),side='right')
rows=[]
for r in range(nU):
    sel=pi[starts[r]:ends[r]][:100]
    if len(sel)<100:
        extra=[x for x in popcol if M[r,x] and x not in set(sel.tolist())][:100-len(sel)]
        sel=np.concatenate([sel,np.array(extra,dtype=np.int32)])
    rows.append(sel)
A=np.stack(rows)
out=pd.DataFrame({'uid':np.repeat(uids,100).astype(np.int64),'item_id':items[A.ravel()].astype(np.int64),'rank':np.tile(np.arange(1,101),nU).astype(np.int64)})
out.to_parquet('/app/solution/submission.parquet',index=False)
print('saved',out.shape)
PYEOF
python safety.py

python - <<'EOF'
s=open('feats.py').read()
s=s.replace("""    UU=np.asarray(([email protected]).todense(),dtype=np.float32)
    np.fill_diagonal(UU,0.0)
    [email protected](Xn.todense(),dtype=np.float32)
    del UU
    rss('post-cf'); print('cf done',time.time()-t0, S.shape)""","""    rss('post-cf')""")
s=s.replace("""    Xlu=sparse.diags(1.0/(np.asarray(Xl.sum(1)).ravel()+1e-6)).astype(np.float32)@Xl
    UUl=np.asarray(([email protected]).todense(),dtype=np.float32)
    np.fill_diagonal(UUl,0.0)
    [email protected](Xln.todense(),dtype=np.float32)
    del UUl
    print('cf2 done',time.time()-t0)""","""    Xlu=sparse.diags(1.0/(np.asarray(Xl.sum(1)).ravel()+1e-6)).astype(np.float32)@Xl
    print('cf2 done',time.time()-t0)""")
s=s.replace("S=S,Sl=Sl,","Xn=Xn.tocsr(),Xr=Xr.tocsr(),Xln=Xln.tocsr(),Xlu=Xlu.tocsr(),")
# rewrite assemble
a=s.index("def assemble(")
s=s[:a]+'''def assemble(B, eligible_mask, chunk=800):
    global FEATNAMES
    t0=time.time()
    nU=B['nU']; F=B['F']; pu=B['pu']; pi=B['pi']
    Xn=B['Xn']; Xr=B['Xr']; Xln=B['Xln']; Xlu=B['Xlu']
    XnT=Xn.T.tocsr(); XlnT=Xln.T.tocsr()
    P=len(pu); ne=B['n_extra']
    starts=np.searchsorted(pu,np.arange(nU)); ends=np.searchsorted(pu,np.arange(nU),side='right')
    cf_h=np.zeros(P,dtype=np.float32); cfl_h=np.zeros(P,dtype=np.float32)
    eu_l=[]; ei_l=[]; ecf_l=[]; ecfl_l=[]
    for s0 in range(0,nU,chunk):
        s1=min(s0+chunk,nU); n=s1-s0
        UU=np.asarray((Xr[s0:s1]@XnT).todense(),dtype=np.float32)
        UU[np.arange(n),np.arange(s0,s1)]=0.0
        Sc=UU@Xn
        Sc=np.asarray(Sc.todense() if sparse.issparse(Sc) else Sc,dtype=np.float32)
        UL=np.asarray((Xlu[s0:s1]@XlnT).todense(),dtype=np.float32)
        UL[np.arange(n),np.arange(s0,s1)]=0.0
        SL=UL@Xln
        SL=np.asarray(SL.todense() if sparse.issparse(SL) else SL,dtype=np.float32)
        del UU,UL
        for r in range(s0,s1):
            sl=slice(starts[r],ends[r])
            cf_h[sl]=Sc[r-s0,pi[sl]]; cfl_h[sl]=SL[r-s0,pi[sl]]
        bl=Sc+0.5*SL
        bl[~eligible_mask[s0:s1]]=-1e9
        for r in range(s0,s1):
            bl[r-s0,pi[starts[r]:ends[r]]]=-1e9
        idx=np.argpartition(-bl,ne,axis=1)[:,:ne]
        rr=np.repeat(np.arange(n),ne); cc=idx.ravel()
        good=bl[rr,cc]>-1e8
        eu_l.append((rr[good]+s0).astype(np.int32)); ei_l.append(cc[good].astype(np.int32))
        ecf_l.append(Sc[rr[good],cc[good]]); ecfl_l.append(SL[rr[good],cc[good]])
        del Sc,SL,bl
    eu=np.concatenate(eu_l); ei=np.concatenate(ei_l)
    ecf=np.concatenate(ecf_l); ecfl=np.concatenate(ecfl_l)
    del eu_l,ei_l,ecf_l,ecfl_l; gc.collect()
    cu=np.concatenate([pu,eu]); ci=np.concatenate([pi,ei])
    ish=np.concatenate([np.ones(P,dtype=np.float32),np.zeros(len(eu),dtype=np.float32)])
    E=len(cu)
    cols={}
    for k,v in F.items(): cols[k]=np.concatenate([v.astype(np.float32),np.zeros(E-P,dtype=np.float32)])
    for k in ['last_age','last_strong_age','last_org_age','first_age']:
        cols[k]=np.where(ish>0,np.minimum(cols[k],999.0),999.0).astype(np.float32)
    cols['is_hist']=ish
    cols['cf']=np.concatenate([cf_h,ecf]); cols['cf_l']=np.concatenate([cfl_h,ecfl])
    for nm,arr in [('u_cnt',B['ucnt']),('u_cnt7',B['ucnt7']),('u_cnt_s',B['ucnt_s']),('u_dist',B['udist']),('u_org',B['uorg'])]:
        cols[nm]=arr[cu].astype(np.float32)
    for nm,arr in [('i_cnt',B['icnt']),('i_cnt7',B['icnt7']),('i_cnt_s',B['icnt_s']),('i_cnt_os',B['icnt_os']),
                   ('i_users',B['iusers']),('i_rep',B['irepeat']),('i_str',B['istrength']),('i_org',B['iorg'])]:
        cols[nm]=arr[ci].astype(np.float32)
    cols['share']=cols['n_all']/np.maximum(cols['u_cnt'],1)
    cols['share_s']=cols['n_strong']/np.maximum(cols['u_cnt_s'],1)
    cols['share7']=cols['c7']/np.maximum(cols['u_cnt7'],1)
    cols['i_pop_rank']=np.log1p(ci.astype(np.float32))
    ck=cu.astype(np.int64)*NI+ci
    order=np.argsort(ck); cks=ck[order]
    def statefeat(k_arr,ts_arr):
        f=np.zeros(E,dtype=np.float32); fa=np.full(E,999.0,dtype=np.float32)
        if len(k_arr)==0: return f,fa
        p=np.searchsorted(cks,k_arr)
        ok=(p<E)&(cks[np.minimum(p,E-1)]==k_arr)
        pos=order[p[ok]]
        f[pos]=1.0; fa[pos]=((B['cutoff']-ts_arr[ok])/DAY).astype(np.float32)
        return f,fa
    cols['liked'],cols['liked_age']=statefeat(B['lk_k'],B['lk_ts'])
    cols['disliked'],cols['disliked_age']=statefeat(B['ds_k'],B['ds_ts'])
    names=sorted(cols.keys()); FEATNAMES=names
    X=np.column_stack([cols[n] for n in names]).astype(np.float32)
    del cols; gc.collect()
    print('assemble',time.time()-t0,X.shape,flush=True)
    return cu,ci,X,names
'''
open('feats.py','w').write(s)
print('ok')
EOF
echo done

cat > pipeline.py <<'PYEOF'
import numpy as np, pandas as pd, pickle, time, gc, sys
from feats import build_features, assemble, NI, DAY
t0=time.time()
c=pickle.load(open('/app/solution/cache.pkl','rb'))
uids=c['uids']; items=c['items']; nU=len(uids)
Mval=np.unpackbits(np.load('/task/data/validation_eligible_candidates.npz')['packed_eligible'],axis=1,bitorder='little')[:,:NI].astype(bool)
Mtest=np.unpackbits(np.load('/task/data/test_eligible_candidates.npz')['packed_eligible'],axis=1,bitorder='little')[:,:NI].astype(bool)
TC=24786800; WIN=604800; GAP=1800

def labels(splits, lo, hi):
    L=pd.concat([c[s+'_listens'] for s in splits],ignore_index=True)
    L=L[(L.timestamp>=lo)&(L.timestamp<hi)&(L.is_organic==1)&(L.played_ratio_pct>=80)]
    K=pd.concat([c[s+'_likes'] for s in splits],ignore_index=True)
    K=K[(K.timestamp>=lo)&(K.timestamp<hi)&(K.is_organic==1)]
    return np.unique(L.u.values.astype(np.int64)*NI+L.i.values), np.unique(K.u.values.astype(np.int64)*NI+K.i.values)

def grades(cu,ci,k1,k3):
    ck=cu.astype(np.int64)*NI+ci
    g=np.zeros(len(ck),dtype=np.int8); g[np.isin(ck,k1)]=1; g[np.isin(ck,k3)]=3
    return g

stage=sys.argv[1]
if stage=='fold1':
    T1=TC-WIN-GAP
    B=build_features(c,['train'],T1); cu,ci,X,names=assemble(B,Mval); del B; gc.collect()
    k1,k3=labels(['train'],T1+GAP,T1+GAP+WIN)
    y=grades(cu,ci,k1,k3)
    np.savez('/app/solution/fold1.npz',cu=cu,ci=ci,X=X,y=y); pickle.dump(names,open('/app/solution/names.pkl','wb'))
    print('fold1',X.shape,(y>0).mean(),(y==3).sum(),'recall-proxy',(y>0).sum(),time.time()-t0)
elif stage=='fold2':
    B=build_features(c,['train'],TC); cu,ci,X,names=assemble(B,Mval); del B; gc.collect()
    k1,k3=labels(['val'],TC+GAP,TC+GAP+WIN)
    y=grades(cu,ci,k1,k3)
    np.savez('/app/solution/fold2.npz',cu=cu,ci=ci,X=X,y=y)
    print('fold2',X.shape,(y>0).sum(),(y==3).sum(),time.time()-t0)
elif stage=='test':
    B=build_features(c,['train','val'],TC+GAP+WIN); cu,ci,X,names=assemble(B,Mtest); del B; gc.collect()
    np.savez('/app/solution/foldt.npz',cu=cu,ci=ci,X=X)
    print('test',X.shape,time.time()-t0)
PYEOF
(timeout 400 python pipeline.py fold1; echo "E1=$?") 2>&1 | tail -8

timeout 400 python pipeline.py fold2 2>&1|tail -3 && timeout 400 python pipeline.py test 2>&1|tail -3

cat > train_rank.py <<'PYEOF'
import numpy as np, pandas as pd, pickle, time
from sklearn.ensemble import HistGradientBoostingRegressor
NI=30000
t0=time.time()
c=pickle.load(open('/app/solution/cache.pkl','rb')); uids=c['uids']; items=c['items']; nU=len(uids)
names=pickle.load(open('/app/solution/names.pkl','rb'))
f1=np.load('/app/solution/fold1.npz'); f2=np.load('/app/solution/fold2.npz')
def gain(y): return (2.0**y-1.0).astype(np.float32)
m=HistGradientBoostingRegressor(max_iter=300,learning_rate=0.08,max_leaf_nodes=63,min_samples_leaf=50,l2_regularization=1.0,early_stopping=False,random_state=0)
m.fit(f1['X'],gain(f1['y'])); print('fit',time.time()-t0,flush=True)
p=m.predict(f2['X'])
def topk(cu,ci,sc,k=100):
    o=np.lexsort((-sc,cu)); cu,ci,sc=cu[o],ci[o],sc[o]
    st=np.searchsorted(cu,np.arange(nU)); en=np.searchsorted(cu,np.arange(nU),side='right')
    return [ci[st[r]:en[r]][:k] for r in range(nU)]
def write(rows,path,M):
    out=[]
    for r,sel in enumerate(rows):
        sel=list(dict.fromkeys(int(x) for x in sel))[:100]
        if len(sel)<100:
            extra=[x for x in range(NI) if M[r,x] and x not in set(sel)][:100-len(sel)]
            sel=sel+extra
        out.append(sel)
    A=np.array(out,dtype=np.int64)
    pd.DataFrame({'uid':np.repeat(uids,100).astype(np.int64),'item_id':items[A.ravel()].astype(np.int64),
                  'rank':np.tile(np.arange(1,101),nU).astype(np.int64)}).to_parquet(path,index=False)
Mval=np.unpackbits(np.load('/task/data/validation_eligible_candidates.npz')['packed_eligible'],axis=1,bitorder='little')[:,:NI].astype(bool)
write(topk(f2['cu'],f2['ci'],p),'/app/solution/validation.parquet',Mval)
print('val written',time.time()-t0,flush=True)
PYEOF
timeout 500 python train_rank.py && python /task/evaluate_public.py /app/solution/validation.parquet

cat >> train_rank.py <<'PYEOF'
# refit on both folds for the hidden test
X=np.vstack([f1['X'],f2['X']]); Y=np.concatenate([gain(f1['y']),gain(f2['y'])])
m2=HistGradientBoostingRegressor(max_iter=300,learning_rate=0.08,max_leaf_nodes=63,min_samples_leaf=50,l2_regularization=1.0,early_stopping=False,random_state=0)
m2.fit(X,Y); print('fit2',time.time()-t0,flush=True)
ft=np.load('/app/solution/foldt.npz')
pt=m2.predict(ft['X'])
Mtest=np.unpackbits(np.load('/task/data/test_eligible_candidates.npz')['packed_eligible'],axis=1,bitorder='little')[:,:NI].astype(bool)
write(topk(ft['cu'],ft['ci'],pt),'/app/solution/submission_model.parquet',Mtest)
print('test written',time.time()-t0,flush=True)
PYEOF
nohup python train_rank.py > train.log 2>&1 &
echo started

sleep 115; cat /app/solution/train.log

sleep 60; cat /app/solution/train.log; ls -la /app/solution/*.parquet

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

pkill -f train_rank.py; python - <<'EOF'
s=open('/app/solution/train_rank.py').read()
s=s.replace("def topk(cu,ci,sc,k=100):\n    o=","def topk(cu,ci,sc,M,k=100):\n    ok=M[cu,ci]; cu,ci,sc=cu[ok],ci[ok],sc[ok]\n    o=")
s=s.replace("write(topk(f2['cu'],f2['ci'],p),'/app/solution/validation.parquet',Mval)","np.save('/app/solution/pval.npy',p)\nwrite(topk(f2['cu'],f2['ci'],p,Mval),'/app/solution/validation.parquet',Mval)")
s=s.replace("write(topk(ft['cu'],ft['ci'],pt),'/app/solution/submission_model.parquet',Mtest)","np.save('/app/solution/ptest.npy',pt)\nwrite(topk(ft['cu'],ft['ci'],pt,Mtest),'/app/solution/submission.parquet',Mtest)")
s=s.replace("Mval=np.unpackbits","import os\nMval=np.unpackbits")
open('/app/solution/train_rank.py','w').write(s)
EOF
nohup python train_rank.py > train.log 2>&1 & echo started

sleep 5 && (pgrep -af train_rank.py || echo "not running"); cat train.log

grep -n "topk(" train_rank.py | head; nohup python train_rank.py > train.log 2>&1 < /dev/null & sleep 3; echo "pid check:"; pgrep -f "python train_rank.py" | head -1

kill 8495 2>/dev/null; sleep 1; python - <<'EOF'
p='/app/solution/train_rank.py'
s=open(p).read()
s=s.replace("def topk(cu,ci,sc,k=100):\n    o=","def topk(cu,ci,sc,M,k=100):\n    ok=M[cu,ci]; cu,ci,sc=cu[ok],ci[ok],sc[ok]\n    o=")
s=s.replace("write(topk(f2['cu'],f2['ci'],p),'/app/solution/validation.parquet',Mval)","np.save('/app/solution/pval.npy',p)\nwrite(topk(f2['cu'],f2['ci'],p,Mval),'/app/solution/validation.parquet',Mval)")
s=s.replace("write(topk(ft['cu'],ft['ci'],pt),'/app/solution/submission_model.parquet',Mtest)","np.save('/app/solution/ptest.npy',pt)\nwrite(topk(ft['cu'],ft['ci'],pt,Mtest),'/app/solution/submission_model.parquet',Mtest)")
open(p,'w').write(s)
print(s[s.index('def topk'):s.index('def write')])
print([l for l in s.split('\n') if 'write(topk' in l or 'np.save' in l])
EOF

nohup python train_rank.py > train.log 2>&1 < /dev/null & sleep 2; pgrep -f "python train_rank.py" | head -1; echo launched

sleep 170; cat /app/solution/train.log

sleep 60; cat /app/solution/train.log; ls -la /app/solution/*.parquet

pgrep -af "python train_rank" || echo NOTRUNNING; wc -c /app/solution/train.log; free -g | head -2

setsid nohup python train_rank.py > train.log 2>&1 < /dev/null & disown; sleep 3; pgrep -f "python train_rank" | head -1

sleep 110; cat /app/solution/train.log; pgrep -f "python train_rank" >/dev/null && echo RUNNING

sleep 60; cat /app/solution/train.log; python /task/evaluate_public.py /app/solution/validation.parquet 2>&1 | tail -6

python - <<'EOF'
import numpy as np, pandas as pd, pickle
NI=30000
c=pickle.load(open('/app/solution/cache.pkl','rb'))
f2=np.load('/app/solution/fold2.npz')
M=np.unpackbits(np.load('/task/data/validation_eligible_candidates.npz')['packed_eligible'],axis=1,bitorder='little')[:,:NI].astype(bool)
cu,ci,y,X=f2['cu'],f2['ci'],f2['y'],f2['X']
ok=M[cu,ci]
print('cands eligible frac',ok.mean())
print('rel total',(y>0).sum(),'rel & eligible',((y>0)&ok).sum())
names=pickle.load(open('/app/solution/names.pkl','rb'))
ih=X[:,names.index('is_hist')]
print('eligible rel that are hist:', ih[(y>0)&ok].mean())
print('hist cand eligible frac', ok[ih>0].mean())
# how many eligible relevant per user
import collections
uu=cu[(y>0)&ok]
print('users with >=1 eligible rel in candidates', len(np.unique(uu)))
EOF

cat > disc.py <<'PYEOF'
import numpy as np, pandas as pd, pickle, time, gc, sys
from scipy import sparse
NI=30000; DAY=86400.0
t0=time.time()
c=pickle.load(open('/app/solution/cache.pkl','rb'))
uids=c['uids']; items=c['items']; nU=len(uids)

def signals(splits, cutoff):
    L=pd.concat([c[s+'_listens'] for s in splits],ignore_index=True); L=L[L.timestamp<cutoff]
    u=L.u.values.astype(np.int32); i=L.i.values.astype(np.int32)
    age=((cutoff-L.timestamp.values.astype(np.float64))/DAY).astype(np.float32)
    strong=(L.played_ratio_pct.values>=80); org=(L.is_organic.values==1)
    key=u.astype(np.int64)*NI+i
    up,code=np.unique(key,return_inverse=True); P=len(up)
    pu=(up//NI).astype(np.int32); pi=(up%NI).astype(np.int32)
    ns=np.bincount(code,weights=strong.astype(np.float64),minlength=P).astype(np.float32)
    d30=np.bincount(code,weights=np.exp(-age/30.0),minlength=P).astype(np.float32)
    d7=np.bincount(code,weights=np.exp(-age/7.0),minlength=P).astype(np.float32)
    ic=np.bincount(i,weights=(strong&org).astype(np.float64),minlength=NI).astype(np.float32)
    ic7=np.bincount(i,weights=((age<7)&strong).astype(np.float64),minlength=NI).astype(np.float32)
    ic30=np.bincount(i,weights=((age<30)&strong).astype(np.float64),minlength=NI).astype(np.float32)
    iu=np.bincount(pi,minlength=NI).astype(np.float32)
    LK=pd.concat([c[s+'_likes'] for s in splits],ignore_index=True); LK=LK[LK.timestamp<cutoff]
    ilike=np.bincount(LK.i.values.astype(np.int32),minlength=NI).astype(np.float32)
    DS=pd.concat([c[s+'_dislikes'] for s in splits],ignore_index=True); DS=DS[DS.timestamp<cutoff]
    idis=np.bincount(DS.i.values.astype(np.int32),minlength=NI).astype(np.float32)
    sel=ns>0
    # binary/implicit matrices
    def mk(vals,rows,cols):
        X=sparse.csr_matrix((vals.astype(np.float32),(rows,cols)),shape=(nU,NI),dtype=np.float32)
        return X
    Xb=mk(np.ones(sel.sum(),np.float32),pu[sel],pi[sel])          # binary interaction
    Xr=mk(d30[sel]+0.5*d7[sel],pu[sel],pi[sel])                   # recency profile
    Xl=mk(np.ones(len(LK),np.float32),LK.u.values.astype(np.int32),LK.i.values.astype(np.int32)); Xl.sum_duplicates(); Xl.data[:]=1.0
    return dict(Xb=Xb,Xr=Xr,Xl=Xl,ic=ic,ic7=ic7,ic30=ic30,iu=iu,ilike=ilike,idis=idis,
                pu=pu[sel],pi=pi[sel],Xh=mk(np.ones(sel.sum(),np.float32),pu[sel],pi[sel]))

def normcols(X, alpha=0.5):
    n=np.asarray(X.multiply(X).sum(0)).ravel()
    w=1.0/np.maximum(n,1e-6)**alpha
    return X.multiply(sparse.csr_matrix(w)).tocsr().astype(np.float32)

def build_scorers(S):
    Xb=S['Xb']; Xr=S['Xr']; Xl=S['Xl']
    Bn=normcols(Xb,0.5); Ln=normcols(Xl,0.5)
    Rn=sparse.diags(1.0/(np.asarray(Xr.sum(1)).ravel()+1e-6)).astype(np.float32)@Xr
    Bu=sparse.diags(1.0/np.sqrt(np.asarray(Xb.sum(1)).ravel()+1e-6)).astype(np.float32)@Bn
    Lu=sparse.diags(1.0/np.sqrt(np.asarray(Xl.sum(1)).ravel()+1e-6)).astype(np.float32)@Ln
    return dict(Bn=Bn.tocsr(),Ln=Ln.tocsr(),Rn=Rn.tocsr(),Bu=Bu.tocsr(),Lu=Lu.tocsr(),
                BnT=Bn.T.tocsr(),LnT=Ln.T.tocsr(),Xh=S['Xh'].tocsr())

def chunk_scores(SC,S,s0,s1,topn_users=300):
    n=s1-s0
    # user-user CF on listens
    UU=np.asarray((SC['Bu'][s0:s1]@SC['BnT']).todense(),dtype=np.float32)
    UU[np.arange(n),np.arange(s0,s1)]=0.0
    if topn_users<UU.shape[1]:
        th=np.partition(UU,-topn_users,axis=1)[:,-topn_users][:,None]
        UU=np.where(UU>=th,UU,0.0)
    A=np.asarray((sparse.csr_matrix(UU)@SC['Bn']).todense(),dtype=np.float32)
    del UU
    UL=np.asarray((SC['Lu'][s0:s1]@SC['LnT']).todense(),dtype=np.float32)
    UL[np.arange(n),np.arange(s0,s1)]=0.0
    if topn_users<UL.shape[1]:
        th=np.partition(UL,-topn_users,axis=1)[:,-topn_users][:,None]
        UL=np.where(UL>=th,UL,0.0)
    Bm=np.asarray((sparse.csr_matrix(UL)@SC['Ln']).todense(),dtype=np.float32)
    del UL
    # item-item: recency profile x item cooc  (Rn @ (Bn^T Bn)) computed as (Rn@Bn^T)@Bn
    RU=np.asarray((SC['Rn'][s0:s1]@SC['BnT']).todense(),dtype=np.float32)
    RU[np.arange(n),np.arange(s0,s1)]=0.0
    C=np.asarray((sparse.csr_matrix(RU)@SC['Bn']).todense(),dtype=np.float32)
    del RU
    # cross: listen-profile -> likes (what similar-listening users LIKE)
    D=np.asarray((sparse.csr_matrix(np.asarray((SC['Bu'][s0:s1]@SC['BnT']).todense(),dtype=np.float32))@SC['Ln']).todense(),dtype=np.float32)
    return A,Bm,C,D
PYEOF
echo ok

cat > gen.py <<'PYEOF'
import numpy as np, pandas as pd, pickle, time, gc, sys
from scipy import sparse
from disc import *
stage=sys.argv[1]
TC=24786800; WIN=604800; GAP=1800
if stage=='val':
    S=signals(['train'],TC); M=np.unpackbits(np.load('/task/data/validation_eligible_candidates.npz')['packed_eligible'],axis=1,bitorder='little')[:,:NI].astype(bool)
else:
    S=signals(['train','val'],TC+GAP+WIN); M=np.unpackbits(np.load('/task/data/test_eligible_candidates.npz')['packed_eligible'],axis=1,bitorder='little')[:,:NI].astype(bool)
SC=build_scorers(S); print('scorers',time.time()-t0,flush=True)
lp=np.log1p(S['ic']); lp7=np.log1p(S['ic7']); lp30=np.log1p(S['ic30'])
llike=np.log1p(S['ilike']); ldis=np.log1p(S['idis']); liu=np.log1p(S['iu'])
KEEP=400
US=[];IS=[];FS=[]
CH=800
for s0 in range(0,nU,CH):
    s1=min(s0+CH,nU); n=s1-s0
    A,B,C,D=chunk_scores(SC,S,s0,s1)
    em=M[s0:s1]
    for X in (A,B,C,D): X[~em]=0.0
    def nz(X):
        m=X.max(1,keepdims=True); return X/np.maximum(m,1e-9)
    An,Bn_,Cn,Dn=nz(A),nz(B),nz(C),nz(D)
    base=An+0.6*Bn_+0.8*Cn+0.6*Dn+0.15*(lp/max(lp.max(),1e-9))[None,:]
    base[~em]=-1e9
    idx=np.argpartition(-base,KEEP,axis=1)[:,:KEEP]
    rr=np.repeat(np.arange(n),KEEP); cc=idx.ravel()
    good=base[rr,cc]>-1e8
    rr,cc=rr[good],cc[good]
    US.append((rr+s0).astype(np.int32)); IS.append(cc.astype(np.int32))
    FS.append(np.column_stack([An[rr,cc],Bn_[rr,cc],Cn[rr,cc],Dn[rr,cc],
                               A[rr,cc],B[rr,cc],C[rr,cc],D[rr,cc],
                               lp[cc],lp7[cc],lp30[cc],llike[cc],ldis[cc],liu[cc]]).astype(np.float32))
    del A,B,C,D,An,Bn_,Cn,Dn,base
    if s0%3200==0: print(s0,time.time()-t0,flush=True)
np.savez('/app/solution/cand_%s.npz'%stage,u=np.concatenate(US),i=np.concatenate(IS),F=np.vstack(FS))
print('done',stage,time.time()-t0)
PYEOF
setsid nohup python gen.py val > gen_val.log 2>&1 < /dev/null & disown; sleep 45; cat gen_val.log

cat > tune.py <<'PYEOF'
import numpy as np, pandas as pd, pickle, time
NI=30000
c=pickle.load(open('/app/solution/cache.pkl','rb'))
uids=c['uids']; items=c['items']; nU=len(uids)
M=np.unpackbits(np.load('/task/data/validation_eligible_candidates.npz')['packed_eligible'],axis=1,bitorder='little')[:,:NI].astype(bool)
vl=c['val_listens']; vk=c['val_likes']
s=vl[(vl.is_organic==1)&(vl.played_ratio_pct>=80)][['u','i']].drop_duplicates()
k=vk[vk.is_organic==1][['u','i']].drop_duplicates()
G={}
for uu,ii in s.values.tolist():
    if M[uu,ii]: G.setdefault(uu,{})[ii]=1
for uu,ii in k.values.tolist():
    if M[uu,ii]: G.setdefault(uu,{})[ii]=3
print('users with labels',len(G))
d=np.load('/app/solution/cand_val.npz'); U,I,F=d['u'],d['i'],d['F']
order=np.argsort(U,kind='stable'); U,I,F=U[order],I[order],F[order]
st=np.searchsorted(U,np.arange(nU)); en=np.searchsorted(U,np.arange(nU),side='right')
disc=1.0/np.log2(np.arange(2,12))
idcg={u:float(np.sum((2.0**np.array(sorted(g.values(),reverse=True)[:10])-1)*disc[:len(list(g)[:10])])) for u,g in G.items()}
def evalw(w):
    tot=0.0
    for u,g in G.items():
        a,b=st[u],en[u]
        if a==b: continue
        sc=F[a:b]@w
        kk=min(10,b-a)
        top=np.argpartition(-sc,kk-1)[:kk]
        top=top[np.argsort(-sc[top])]
        its=I[a:b][top]
        gains=np.array([g.get(int(x),0) for x in its],dtype=np.float64)
        tot+=float(np.sum((2.0**gains-1)*disc[:len(gains)]))/idcg[u]
    return tot/len(G)
names=['An','Bn','Cn','Dn','A','B','C','D','lp','lp7','lp30','llike','ldis','liu']
base=np.zeros(14); 
for nm,v in [('An',1.0),('Bn',0.6),('Cn',0.8),('Dn',0.6),('lp',0.05)]: base[names.index(nm)]=v
print('base',round(evalw(base),5),flush=True)
best=base.copy(); bs=evalw(best)
import itertools, random
rng=np.random.default_rng(0)
grid={'An':[0,0.5,1,1.5],'Bn':[0,0.3,0.6,1.0],'Cn':[0,0.4,0.8,1.5],'Dn':[0,0.3,0.6,1.0],'lp':[0,0.05,0.15,0.3],'lp7':[0,0.05,0.15],'llike':[0,0.05,0.15],'ldis':[0,-0.05,-0.15],'liu':[0,-0.05,0.05]}
for rounds in range(3):
    for nm,vals in grid.items():
        j=names.index(nm); cur=best[j]; bv=cur
        for v in vals:
            best[j]=v; sc=evalw(best)
            if sc>bs: bs=sc; bv=v
        best[j]=bv
    print('round',rounds,round(bs,5),dict(zip(names,np.round(best,3))),flush=True)
np.save('/app/solution/wbest.npy',best)
print('BEST',bs)
PYEOF
sleep 100; cat gen_val.log | tail -3

setsid nohup python gen.py test > gen_test.log 2>&1 < /dev/null & disown; sleep 2; timeout 115 python tune.py 2>&1 | tail -20

timeout 110 python - <<'EOF' 2>&1 | tail -12
exec(open('/app/solution/tune.py').read().split("names=['An'")[0])
names=['An','Bn','Cn','Dn','A','B','C','D','lp','lp7','lp30','llike','ldis','liu']
def W(d):
    w=np.zeros(14)
    for k,v in d.items(): w[names.index(k)]=v
    return w
# pool recall
pool={}
for u,g in G.items():
    a,b=st[u],en[u]; s_=set(I[a:b].tolist())
    pool[u]=sum(1 for x in g if x in s_)/len(g)
print('pool recall',np.mean(list(pool.values())))
for nm,d in [('pop',{'lp':1.0}),('pop7',{'lp7':1.0}),('cfA',{'An':1.0}),('cfB',{'Bn':1.0}),('cfC',{'Cn':1.0}),('cfD',{'Dn':1.0}),
             ('best',{'Bn':0.3,'Dn':0.3,'lp':0.05,'lp7':0.15,'ldis':-0.15,'liu':0.05}),
             ('b2',{'Bn':0.3,'Dn':0.3,'lp':0.1,'lp7':0.3,'ldis':-0.15,'liu':0.05}),
             ('b3',{'An':0.2,'Bn':0.3,'Cn':0.2,'Dn':0.3,'lp':0.05,'lp7':0.15,'ldis':-0.15,'liu':0.05})]:
    print(nm, round(evalw(W(d)),5), flush=True)
EOF
tail -2 gen_test.log

pkill -f "gen.py test"; cat > final.py <<'PYEOF'
import numpy as np, pandas as pd, pickle, time, sys
from scipy import sparse
from disc import *
TC=24786800; WIN=604800; GAP=1800
W=dict(An=0.0,Bn=0.3,Cn=0.0,Dn=0.3,lp=0.05,lp7=0.15,ldis=-0.15,liu=0.05)
def run(stage,out):
    if stage=='val':
        S=signals(['train'],TC); M=np.unpackbits(np.load('/task/data/validation_eligible_candidates.npz')['packed_eligible'],axis=1,bitorder='little')[:,:NI].astype(bool)
    else:
        S=signals(['train','val'],TC+GAP+WIN); M=np.unpackbits(np.load('/task/data/test_eligible_candidates.npz')['packed_eligible'],axis=1,bitorder='little')[:,:NI].astype(bool)
    SC=build_scorers(S)
    lp=np.log1p(S['ic']); lp7=np.log1p(S['ic7']); ldis=np.log1p(S['idis']); liu=np.log1p(S['iu'])
    ipart=(W['lp']*lp+W['lp7']*lp7+W['ldis']*ldis+W['liu']*liu).astype(np.float32)
    top=np.zeros((nU,100),dtype=np.int32)
    CH=800
    for s0 in range(0,nU,CH):
        s1=min(s0+CH,nU); n=s1-s0
        A,B,C,D=chunk_scores(SC,S,s0,s1)
        em=M[s0:s1]
        for X in (A,B,C,D): X[~em]=0.0
        def nz(X):
            m=X.max(1,keepdims=True); return X/np.maximum(m,1e-9)
        sc=W['An']*nz(A)+W['Bn']*nz(B)+W['Cn']*nz(C)+W['Dn']*nz(D)+ipart[None,:]
        sc[~em]=-1e9
        idx=np.argpartition(-sc,100,axis=1)[:,:100]
        vals=np.take_along_axis(sc,idx,1)
        o=np.argsort(-vals,axis=1)
        top[s0:s1]=np.take_along_axis(idx,o,1)
        del A,B,C,D,sc
        print(stage,s0,time.time()-t0,flush=True)
    assert M[np.repeat(np.arange(nU),100),top.ravel()].all()
    pd.DataFrame({'uid':np.repeat(uids,100).astype(np.int64),'item_id':items[top.ravel()].astype(np.int64),
                  'rank':np.tile(np.arange(1,101),nU).astype(np.int64)}).to_parquet(out,index=False)
    print('WROTE',out,time.time()-t0,flush=True)
run('val','/app/solution/validation.parquet')
run('test','/app/solution/submission_new.parquet')
PYEOF
setsid nohup python final.py > final.log 2>&1 < /dev/null & disown; sleep 3; echo started

pgrep -f "python final.py" >/dev/null && echo RUNNING || (setsid nohup python final.py > final.log 2>&1 < /dev/null & disown; sleep 3; echo relaunched); tail -2 final.log 2>/dev/null

sleep 115; tail -3 /app/solution/final.log

ls -la /app/solution/ | tail -20; pgrep -af "python final.py" | head -2

/app/solution/final.py

"""Score every eligible candidate with the tuned discovery blend and write rankings."""
import numpy as np, pandas as pd, time
from disc import *  # noqa: F401,F403  (signals, build_scorers, chunk_scores, NI, nU, uids, items, t0)

TC = 24786800
WIN = 604800
GAP = 1800
W = dict(An=0.0, Bn=0.3, Cn=0.0, Dn=0.3, lp=0.05, lp7=0.15, ldis=-0.15, liu=0.05)


def run(stage, out):
    if stage == 'val':
        S = signals(['train'], TC)
        pack = np.load('/task/data/validation_eligible_candidates.npz')['packed_eligible']
    else:
        S = signals(['train', 'val'], TC + GAP + WIN)
        pack = np.load('/task/data/test_eligible_candidates.npz')['packed_eligible']
    M = np.unpackbits(pack, axis=1, bitorder='little')[:, :NI].astype(bool)
    SC = build_scorers(S)
    lp = np.log1p(S['ic']); lp7 = np.log1p(S['ic7'])
    ldis = np.log1p(S['idis']); liu = np.log1p(S['iu'])
    ipart = (W['lp'] * lp + W['lp7'] * lp7 + W['ldis'] * ldis + W['liu'] * liu).astype(np.float32)
    top = np.zeros((nU, 100), dtype=np.int32)
    CH = 800
    for s0 in range(0, nU, CH):
        s1 = min(s0 + CH, nU)
        A, B, C, D = chunk_scores(SC, S, s0, s1)
        em = M[s0:s1]
        for X in (A, B, C, D):
            X[~em] = 0.0

        def nz(X):
            return X / np.maximum(X.max(1, keepdims=True), 1e-9)

        sc = W['An'] * nz(A) + W['Bn'] * nz(B) + W['Cn'] * nz(C) + W['Dn'] * nz(D) + ipart[None, :]
        sc[~em] = -1e9
        idx = np.argpartition(-sc, 100, axis=1)[:, :100]
        vals = np.take_along_axis(sc, idx, 1)
        top[s0:s1] = np.take_along_axis(idx, np.argsort(-vals, axis=1), 1)
        del A, B, C, D, sc
        print(stage, s0, round(time.time() - t0, 1), flush=True)
    assert M[np.repeat(np.arange(nU), 100), top.ravel()].all()
    pd.DataFrame({
        'uid': np.repeat(uids, 100).astype(np.int64),
        'item_id': items[top.ravel()].astype(np.int64),
        'rank': np.tile(np.arange(1, 101), nU).astype(np.int64),
    }).to_parquet(out, index=False)
    print('WROTE', out, round(time.time() - t0, 1), flush=True)


if __name__ == '__main__':
    run('val', '/app/solution/validation.parquet')
    run('test', '/app/solution/submission_new.parquet')

rm -f /app/solution/fold1.npz /app/solution/fold2.npz /app/solution/foldt.npz /app/solution/pval.npy; setsid nohup python /app/solution/final.py > /app/solution/final.log 2>&1 < /dev/null & disown; sleep 20; tail -3 /app/solution/final.log