"""Huawei TechArena 2026 Phase 1 - submission_v6: single self-contained end-to-end pipeline (raw CSV -> features -> model -> CSV).
Laptop-budget config (as v4_lite) + low-rank history-centroid features: 2-component GBDT blend.
  base : LightGBM binary, lr 0.05, 63 leaves, min_data 200, ff 0.7, bagging 0.8/1, l2 10; 2 seeds (42, 43), 400 rounds
  slow : lr 0.03, ff 0.55, 95 leaves, l2 30, 1 seed (42), 500 rounds
  blend: mean of the two component logits (base = seed-mean raw score), sigmoid.
Training rows: folds f1..f6 = first event per user in [Ck, Ck+1d), Ck = Jan21..Jan26, all stats frozen at Ck. Eval stats frozen at Jan27.
Features: v3 features (train-only global/item stats + each row's own history), v2 features, similarity extras (extra.py,
extra2.py), time/gap features dropped ('notime'). NO aggregates over evaluation rows (item frequency among eval candidates,
eval co-occurrence, ...): global/item tables come from training data only; each eval row only contributes its own history to
its own user features; within-event (same 10 candidates) z-scores/ranks are used. rowid/event_id are never features.
feat_lr (lrfeat2.py): per cutoff C, binary user x item matrix of each user's last 40 distinct pre-C history clicks (train events only),
columns / sqrt(pop+5), exact svds; item vectors = components 2..K+1 (popularity component dropped) * s, K in 4,6,8,10,12,16;
cosine(candidate, centroid of the row's own history) (+ minus 0.5/1.0 x centroid of the user's own past skipped impressions),
within-event z and gap-to-max. Eval fitted at Jan27 on train only.
extra.py item-side vectors / user clusters use train-only user vectors (leak-hygiene fix, 23 Oct 17:10).
Local validation, same recipe at cutoff Jan25 / Jan26: AUC 0.80185 / 0.81192 (v4_lite 0.77860 / 0.79429).

Usage:  python solution_v6.py --raw DIR_WITH_train.csv_AND_evaluation.csv [--work WORKDIR] [--threads 4]
                                   [--out submission_v6.csv] [--from STAGE] [--to STAGE]
Needs: python 3.11+, polars, numpy, scipy, scikit-learn, lightgbm (tested: polars 1.x, lightgbm 4.7). ~10 GB RAM, ~25 GB disk.
The module sources are embedded below (SRC) and written to WORKDIR at start-up."""
import os, sys, time, json, argparse, subprocess

SRC = {'prep.py': "import polars as pl, time, sys\nD='@@WORK@@'\ndef hexid(e, n=15):  # opaque id -> int64 via first 15 hex chars after prefix\n    return e.str.slice(4, n).str.to_integer(base=16)\ndef prep(name, has_label):\n    t=time.time()\n    lf=pl.scan_csv(f'{D}/{name}.csv', schema_overrides={'history_item_ids':pl.Utf8,'history_timestamps':pl.Utf8})\n    cols=[pl.col('rowid'), hexid(pl.col('event_id')).alias('eid'), hexid(pl.col('item_id')).alias('item')]\n    if has_label: cols.append(pl.col('clicked').cast(pl.Int8))\n    lf.select(cols).sink_parquet(f'{D}/{name}_rows.parquet')\n    print(name,'rows done',time.time()-t,flush=True)\n    ev=(lf.select('event_id','user_id','timestamp','time_bucket','day_type','history_item_ids','history_timestamps')\n          .unique(subset='event_id',keep='any')\n          .select(hexid(pl.col('event_id')).alias('eid'), hexid(pl.col('user_id')).alias('user'),\n                  pl.col('timestamp').cast(pl.Int64).alias('ts'), pl.col('time_bucket').cast(pl.Int8).alias('tb'),\n                  pl.col('day_type').cast(pl.Int8).alias('dt'),\n                  pl.col('history_item_ids').str.json_decode(pl.List(pl.Utf8)).list.eval(pl.element().str.slice(4,15).str.to_integer(base=16)).alias('h_items'),\n                  pl.col('history_timestamps').str.json_decode(pl.List(pl.Int64)).alias('h_ts')))\n    ev.collect(engine='streaming').sort('ts').write_parquet(f'{D}/{name}_events.parquet')\n    print(name,'events done',time.time()-t,flush=True)\nprep('evaluation',False)\nprep('train',True)\n", 'v3fold.py': '"""v3 features (v2work/v3feat_ne.py = v3/features.py with global/item stats from train data only) for all fold specs.\nUsage: python v3fold.py NAME... -> v2work/feat_v3/NAME.parquet\n  f1..f6: first event/user in [Ck, Ck+1d); val25/val26: [C, end of train); eval: Jan27 cutoff\n  e1..e4: [Ck, Jan25) (eval-like staleness, for cutoff-25 validation); E1..E6: [Ck, Jan27) (eval-like, final fit)"""\nimport sys, os\nsys.path.insert(0, \'@@WORK@@\'); sys.path.insert(1, \'@@WORK@@/v3\')\nfrom v3feat_ne import load, build, log\nfrom split import J20, DAY, first_events\nTE, TR, EE, ER, NI, NU = load(\'@@WORK@@/v3/data\')\nOUT = \'@@WORK@@/feat_v3\'; os.makedirs(OUT, exist_ok=True)\nSPEC = {**{f\'f{k}\': (J20 + k * DAY, DAY) for k in range(1, 7)}, \'val25\': (J20 + 5 * DAY, None), \'val26\': (J20 + 6 * DAY, None),\n        **{f\'e{k}\': (J20 + k * DAY, (5 - k) * DAY) for k in range(1, 5)}, **{f\'E{k}\': (J20 + k * DAY, (7 - k) * DAY) for k in range(1, 7)}, **{f\'d{k}\': (J20 + k * DAY, (6 - k) * DAY) for k in range(1, 6)}}\nfor n in sys.argv[1:]:\n    if n == \'eval\': R = build(J20 + 7 * DAY, EE, ER, TE, TR, NI, NU)\n    else:\n        C, W = SPEC[n]; T_ev = first_events(TE, C, W)\n        R = build(C, T_ev, TR.join(T_ev.select(\'eid\'), on=\'eid\', how=\'semi\'), TE, TR, NI, NU)\n    R.write_parquet(f\'{OUT}/{n}.parquet\'); log(\'wrote\', n, R.shape)\n', 'v3feat_ne.py': '"""Feature builder (as-of-cutoff, leakage-safe) for TechArena CTR.\n\nbuild(C, T_ev, T_rows, TE, TR) returns one feature row per target candidate.\nEvery statistic uses ONLY:\n  * train events/rows with ts < C (fit window: impressions + labels), and\n  * click histories of events with ts < C, plus the target event\'s OWN history (strictly before its ts).\nSince target events are each user\'s first event at/after C, their histories contain only clicks < C,\nso nothing observed after a target row\'s own event time is ever joined.\n\nFeature groups (column prefixes, used for ablation):\n  hist_  candidate vs the user\'s click log / own history / past impressions of that user-item pair\n  it_    item popularity & CTR (full fit window, last 24h, last 6h, trend)\n  co_    item-item co-click (user co-occurrence cosine) and sequential transitions vs user\'s last 5/20 clicks\n  us_    user activity / CTR / recency / cold-start flags\n  ev_    within-event rank / z-score / event max & mean of the main features\n  tm_    time_bucket, day_type, hour, staleness since cutoff\n  sv_    truncated SVD (tf-idf, row-normalised user x item click matrix as of C): user/history/item cosines\n  mf_    logistic matrix factorisation score fitted on fit-window impressions\n"""\nimport time\nimport numpy as np\nimport polars as pl\nfrom scipy import sparse\nfrom sklearn.utils.extmath import randomized_svd\n\nJ20 = 1737331200\nDAY = 86400\nG_PRIOR = 0.0629\nK_ITEM = 20          # pseudo-impressions for item CTR shrinkage\nK_USER = 5           # pseudo-events for user CTR shrinkage\nTOPK = 200           # co-click neighbours kept per item\nID_COLS = [\'rowid\', \'eid\', \'u\', \'ii\', \'ts\', \'clicked\', \'ord\']\n_t0 = time.time()\n\ndef log(*a):\n    print(time.strftime(\'%H:%M:%S\'), *a, flush=True)\n\ndef load(D):\n    TE = pl.read_parquet(f\'{D}/ev_train.parquet\')\n    TR = pl.read_parquet(f\'{D}/rows_train.parquet\')\n    EE = pl.read_parquet(f\'{D}/ev_eval.parquet\')\n    ER = pl.read_parquet(f\'{D}/rows_eval.parquet\')\n    NI = pl.read_parquet(f\'{D}/items.parquet\').height\n    NU = pl.read_parquet(f\'{D}/users.parquet\').height\n    return TE, TR, EE, ER, NI, NU\n\ndef _hist_long(E):\n    return (E.select(\'u\', \'h\', \'h_ts\').explode([\'h\', \'h_ts\']).drop_nulls(\'h\')\n             .select(\'u\', pl.col(\'h\').cast(pl.Int32).alias(\'ii\'), pl.col(\'h_ts\').cast(pl.Int64).alias(\'ts\')))\n\ndef click_log(C, P_ev, P_rows, T_ev):\n    """Exact (user,item,ts) click log as of C: fit-window clicks + warm-up clicks seen in pre-C / own histories."""\n    win = P_rows.filter(pl.col(\'clicked\') == 1).select(\'u\', pl.col(\'ii\').cast(pl.Int32), \'ts\')\n    hist = pl.concat([_hist_long(P_ev), _hist_long(T_ev)]).filter(pl.col(\'ts\') < C)\n    CL = pl.concat([win, hist]).unique().sort(\'u\', \'ts\', \'ii\')\n    return CL\n\ndef cooc_tables(CL, NI):\n    """User co-occurrence cosine (top-K per item) + lag-weighted transition probabilities prev->next."""\n    us = CL.select(\'u\', \'ii\').unique()\n    pop = us.group_by(\'ii\').agg(pl.len().alias(\'pop\'))\n    pairs = (us.join(us, on=\'u\', suffix=\'_b\').filter(pl.col(\'ii\') != pl.col(\'ii_b\'))\n               .group_by(\'ii\', \'ii_b\').agg(pl.len().cast(pl.Int32).alias(\'co\')))\n    co = (pairs.join(pop, on=\'ii\').join(pop.rename({\'ii\': \'ii_b\', \'pop\': \'pop_b\'}), on=\'ii_b\')\n               .with_columns((pl.col(\'co\') / (pl.col(\'pop\') * pl.col(\'pop_b\')).sqrt()).cast(pl.Float32).alias(\'cos\'))\n               .select(pl.col(\'ii_b\').alias(\'j\'), \'ii\', \'cos\')       # j = history item, ii = candidate\n               .sort(\'cos\', descending=True).group_by(\'j\').head(TOPK))\n    seq = CL.select(\'u\', \'ii\')  # already sorted by u, ts\n    parts = []\n    for lag in (1, 2, 3):\n        parts.append(seq.with_columns(pl.col(\'ii\').shift(lag).over(\'u\').alias(\'j\')).drop_nulls(\'j\')\n                        .select(pl.col(\'j\').cast(pl.Int32), \'ii\', pl.lit(1.0 / lag).alias(\'w\')))\n    trn = pl.concat(parts).filter(pl.col(\'j\') != pl.col(\'ii\')).group_by(\'j\', \'ii\').agg(pl.col(\'w\').sum())\n    tot = trn.group_by(\'j\').agg(pl.col(\'w\').sum().alias(\'tot\'))\n    trn = (trn.join(tot, on=\'j\').with_columns((pl.col(\'w\') / (pl.col(\'tot\') + 5.0)).cast(pl.Float32).alias(\'trp\'))\n              .select(\'j\', \'ii\', \'trp\').sort(\'trp\', descending=True).group_by(\'j\').head(TOPK))\n    return co, trn\n\ndef fit_mf(P_rows, NU, NI, dim=24, epochs=6, lr=0.05, l2=1e-4, seed=0, batch=200_000):\n    """Logistic MF  logit = b + bu + bi + <pu, qi>  fitted with Adagrad on fit-window impressions (numpy)."""\n    rng = np.random.default_rng(seed)\n    u = P_rows[\'u\'].to_numpy().astype(np.int64); i = P_rows[\'ii\'].to_numpy().astype(np.int64)\n    y = P_rows[\'clicked\'].to_numpy().astype(np.float32)\n    P = (rng.standard_normal((NU, dim)) * 0.01).astype(np.float32); Q = (rng.standard_normal((NI, dim)) * 0.01).astype(np.float32)\n    bu = np.zeros(NU, np.float32); bi = np.zeros(NI, np.float32)\n    b = np.float32(np.log(y.mean() / (1 - y.mean())))\n    gP = np.full_like(P, 1e-6); gQ = np.full_like(Q, 1e-6); gbu = np.full_like(bu, 1e-6); gbi = np.full_like(bi, 1e-6)\n    n = len(y)\n    for ep in range(epochs):\n        perm = rng.permutation(n)\n        for s in range(0, n, batch):\n            idx = perm[s:s + batch]; ub, ib, yb = u[idx], i[idx], y[idx]\n            pu, qi = P[ub], Q[ib]\n            z = b + bu[ub] + bi[ib] + (pu * qi).sum(1)\n            g = (1 / (1 + np.exp(-z)) - yb).astype(np.float32)          # dL/dz\n            m = len(idx)\n            Su = sparse.csr_matrix((np.ones(m, np.float32), (ub, np.arange(m))), shape=(NU, m))\n            Si = sparse.csr_matrix((np.ones(m, np.float32), (ib, np.arange(m))), shape=(NI, m))\n            dP = Su @ (g[:, None] * qi) + l2 * P * (np.asarray(Su.sum(1)).ravel() > 0)[:, None]\n            dQ = Si @ (g[:, None] * pu) + l2 * Q * (np.asarray(Si.sum(1)).ravel() > 0)[:, None]\n            dbu = Su @ g; dbi = Si @ g\n            gP += dP * dP; gQ += dQ * dQ; gbu += dbu * dbu; gbi += dbi * dbi\n            P -= lr * dP / np.sqrt(gP); Q -= lr * dQ / np.sqrt(gQ)\n            bu -= lr * dbu / np.sqrt(gbu); bi -= lr * dbi / np.sqrt(gbi)\n    return dict(b=b, bu=bu, bi=bi, P=P, Q=Q)\n\ndef build(C, T_ev, T_rows, TE, TR, NI, NU, with_mf=True, sv_mode=\'std\'):\n    t0 = time.time()\n    P_ev = TE.filter(pl.col(\'ts\') < C)\n    P_rows = TR.join(P_ev.select(\'eid\', \'u\', \'ts\'), on=\'eid\', how=\'inner\')\n    g = float(P_rows[\'clicked\'].mean())\n    ndays = (C - J20) / DAY\n    CL = click_log(C, P_ev, P_rows, T_ev)          # per-user stats: train data + each target row\'s OWN history\n    CLg = click_log(C, P_ev, P_rows, T_ev.head(0))  # item-level / global stats: train data only (no target-event histories)\n\n    # ---------------- item ----------------\n    def ictr(c, n):\n        return (pl.col(c) + K_ITEM * g) / (pl.col(n) + K_ITEM)\n    it = P_rows.group_by(\'ii\').agg(\n        pl.len().alias(\'it_imp\'), pl.col(\'clicked\').sum().cast(pl.Int32).alias(\'it_clk\'),\n        (pl.col(\'ts\') >= C - DAY).sum().alias(\'it_imp24\'), pl.col(\'clicked\').filter(pl.col(\'ts\') >= C - DAY).sum().cast(pl.Int32).alias(\'it_clk24\'),\n        (pl.col(\'ts\') >= C - 6 * 3600).sum().alias(\'it_imp6\'), pl.col(\'clicked\').filter(pl.col(\'ts\') >= C - 6 * 3600).sum().cast(pl.Int32).alias(\'it_clk6\'))\n    it = it.with_columns(ictr(\'it_clk\', \'it_imp\').alias(\'it_ctr\'), ictr(\'it_clk24\', \'it_imp24\').alias(\'it_ctr24\'),\n                         ictr(\'it_clk6\', \'it_imp6\').alias(\'it_ctr6\'))\n    it = it.with_columns((pl.col(\'it_ctr24\') / pl.col(\'it_ctr\')).alias(\'it_trend24\'), (pl.col(\'it_ctr6\') / pl.col(\'it_ctr\')).alias(\'it_trend6\'),\n                         (pl.col(\'it_imp24\') / (pl.col(\'it_imp\') / ndays)).alias(\'it_imptrend24\'),\n                         (pl.col(\'it_imp\') / ndays).alias(\'it_imp_rate\'))\n    ip = CLg.group_by(\'ii\').agg(pl.len().alias(\'it_nclk_all\'), pl.col(\'u\').n_unique().alias(\'it_nusers\'),\n                               (pl.col(\'ts\') < J20).sum().alias(\'it_nclk_warm\'),\n                               ((pl.col(\'ts\') - C) / (2 * DAY)).exp().sum().cast(pl.Float32).alias(\'it_clk_dec2d\'))\n\n    # ---------------- user ----------------\n    ue = P_rows.group_by(\'u\').agg(pl.col(\'eid\').n_unique().alias(\'us_nev\'), pl.col(\'clicked\').sum().cast(pl.Int32).alias(\'us_nclk_win\'),\n                                  pl.col(\'ts\').max().alias(\'_u_last_ev\'))\n    ue = ue.with_columns(((pl.col(\'us_nclk_win\') + K_USER * 0.629) / (pl.col(\'us_nev\') + K_USER)).alias(\'us_ctr\'))\n    uc = CL.group_by(\'u\').agg(pl.len().alias(\'us_nclk_all\'), (pl.col(\'ts\') < J20).sum().alias(\'us_nclk_warm\'),\n                              pl.col(\'ii\').n_unique().alias(\'us_nuniq\'), pl.col(\'ts\').max().alias(\'_u_last_clk\'),\n                              (pl.col(\'ts\') >= C - DAY).sum().alias(\'us_nclk24\'))\n    uc = uc.with_columns((1 - pl.col(\'us_nuniq\') / pl.col(\'us_nclk_all\')).alias(\'us_repeat_rate\'))\n\n    # ---------------- user x item ----------------\n    uic = CL.group_by(\'u\', \'ii\').agg(pl.len().alias(\'hist_n_all\'), (pl.col(\'ts\') < J20).sum().alias(\'hist_n_warm\'),\n                                     (pl.col(\'ts\') >= J20).sum().alias(\'hist_n_win\'), pl.col(\'ts\').max().alias(\'_ui_last_clk\'))\n    uii = P_rows.group_by(\'u\', \'ii\').agg(pl.len().alias(\'hist_imp\'), pl.col(\'clicked\').sum().cast(pl.Int32).alias(\'_ui_clk\'),\n                                         pl.col(\'ts\').max().alias(\'_ui_last_imp\'))\n    uii = uii.with_columns((pl.col(\'hist_imp\') - pl.col(\'_ui_clk\')).alias(\'hist_noclk\'))\n\n    # ---------------- assemble ----------------\n    R = (T_rows.select([c for c in (\'rowid\', \'eid\', \'ii\', \'clicked\', \'ord\') if c in T_rows.columns])\n         .join(T_ev.select(\'eid\', \'u\', \'ts\', \'tb\', \'dt\', \'h\', \'h_ts\'), on=\'eid\', how=\'left\').with_row_index(\'r\'))\n    R = (R.join(it, on=\'ii\', how=\'left\').join(ip, on=\'ii\', how=\'left\').join(ue, on=\'u\', how=\'left\').join(uc, on=\'u\', how=\'left\')\n          .join(uic, on=[\'u\', \'ii\'], how=\'left\').join(uii, on=[\'u\', \'ii\'], how=\'left\'))\n    R = R.with_columns(\n        pl.col(\'h\').list.len().alias(\'us_hlen\'),\n        (pl.col(\'h\').list.len() == 0).cast(pl.Int8).alias(\'us_nohist\'),\n        pl.col(\'us_nev\').is_null().cast(pl.Int8).alias(\'us_cold\'),\n        (pl.col(\'ts\') - pl.col(\'_u_last_ev\')).alias(\'us_since_ev\'),\n        (pl.col(\'ts\') - pl.col(\'_u_last_clk\')).alias(\'us_since_clk\'),\n        (pl.col(\'ts\') - pl.col(\'_ui_last_clk\')).alias(\'hist_since_clk\'),\n        (pl.col(\'ts\') - pl.col(\'_ui_last_imp\')).alias(\'hist_since_imp\'),\n        (pl.col(\'ts\') - C).alias(\'tm_stale\'), ((pl.col(\'ts\') // 3600) % 24).cast(pl.Int8).alias(\'tm_hour\'),\n        pl.col(\'tb\').alias(\'tm_tb\'), pl.col(\'dt\').alias(\'tm_dt\'))\n    # position (from end, 1 = most recent) of candidate in own history\n    hx = (R.select(\'r\', \'ii\', \'h\').with_columns(pl.int_ranges(pl.col(\'h\').list.len(), 0, -1).alias(\'pos\'))\n            .explode([\'h\', \'pos\']).drop_nulls(\'h\').filter(pl.col(\'h\') == pl.col(\'ii\'))\n            .group_by(\'r\').agg(pl.col(\'pos\').min().alias(\'hist_pos\'), pl.len().alias(\'hist_n_own\')))\n    R = R.join(hx, on=\'r\', how=\'left\')\n    R = R.with_columns(pl.col(\'hist_since_clk\').fill_null(30 * DAY), pl.col(\'hist_since_imp\').fill_null(30 * DAY),\n                       pl.col(\'hist_pos\').fill_null(99), pl.col(\'hist_n_own\').fill_null(0), pl.col(\'us_since_ev\').fill_null(30 * DAY), pl.col(\'us_since_clk\').fill_null(30 * DAY))\n    R = R.with_columns((pl.col(\'hist_n_all\').fill_null(0) > 0).cast(pl.Int8).alias(\'hist_in\'))\n\n    # ---------------- co-click ----------------\n    co, trn = cooc_tables(CLg, NI)\n    last20 = CL.group_by(\'u\', maintain_order=True).agg(pl.col(\'ii\').tail(20).alias(\'l20\'))\n    T_u = R.select(\'r\', \'u\', \'ii\').join(last20, on=\'u\', how=\'left\')\n    X = (T_u.with_columns(pl.int_ranges(pl.col(\'l20\').list.len(), 0, -1).alias(\'pos\')).explode([\'l20\', \'pos\'])\n            .drop_nulls(\'l20\').rename({\'l20\': \'j\'}).with_columns(pl.col(\'j\').cast(pl.Int32))\n            .join(co, on=[\'j\', \'ii\'], how=\'left\').join(trn, on=[\'j\', \'ii\'], how=\'left\')\n            .with_columns(pl.col(\'cos\').fill_null(0), pl.col(\'trp\').fill_null(0)))\n    l5 = pl.col(\'pos\') <= 5\n    agg = X.group_by(\'r\').agg(\n        pl.col(\'cos\').max().alias(\'co_cos_max20\'), pl.col(\'cos\').mean().alias(\'co_cos_mean20\'),\n        pl.col(\'cos\').filter(l5).max().alias(\'co_cos_max5\'), pl.col(\'cos\').filter(l5).mean().alias(\'co_cos_mean5\'),\n        pl.col(\'cos\').filter(pl.col(\'pos\') == 1).first().alias(\'co_cos_last\'),\n        pl.col(\'trp\').max().alias(\'co_tr_max20\'), pl.col(\'trp\').mean().alias(\'co_tr_mean20\'),\n        pl.col(\'trp\').filter(l5).max().alias(\'co_tr_max5\'), pl.col(\'trp\').filter(l5).mean().alias(\'co_tr_mean5\'),\n        pl.col(\'trp\').filter(pl.col(\'pos\') == 1).first().alias(\'co_tr_last\'),\n        (pl.col(\'cos\') / pl.col(\'pos\').cast(pl.Float32).sqrt()).sum().alias(\'co_cos_rw\'), pl.col(\'cos\').sum().alias(\'co_cos_sum20\'))\n    R = R.join(agg, on=\'r\', how=\'left\').sort(\'r\')\n\n    # ---------------- SVD embeddings ----------------\n    # sv_mode (leakage checks only): \'std\' = click log as of C; \'notgt\' = drop target events\' own histories;\n    # \'lag1d\' = only clicks/impressions with ts < C - 1 day\n    if sv_mode == \'std\':\n        CLsv, NEGsrc = CLg, P_rows\n    elif sv_mode == \'notgt\':\n        CLsv = pl.concat([P_rows.filter(pl.col(\'clicked\') == 1).select(\'u\', pl.col(\'ii\').cast(pl.Int32), \'ts\'),\n                          _hist_long(P_ev).filter(pl.col(\'ts\') < C)]).unique()\n        NEGsrc = P_rows\n    elif sv_mode == \'lag1d\':\n        CLsv, NEGsrc = CL.filter(pl.col(\'ts\') < C - DAY), P_rows.filter(pl.col(\'ts\') < C - DAY)\n    else:\n        raise ValueError(sv_mode)\n    assert (CLsv[\'ts\'].max() or 0) < C and (NEGsrc[\'ts\'].max() or 0) < C\n    cnt = CLsv.group_by(\'u\', \'ii\').agg(pl.len().alias(\'n\'))\n    Xc = sparse.csr_matrix((np.ones(cnt.height, np.float32), (cnt[\'u\'].to_numpy(), cnt[\'ii\'].to_numpy())), shape=(NU, NI))\n    df_ = np.asarray(Xc.sum(0)).ravel(); idf = np.log((NU + 1) / (df_ + 1)).astype(np.float32)\n    Xc = Xc @ sparse.diags(idf); rn = np.sqrt(np.asarray(Xc.multiply(Xc).sum(1)).ravel()) + 1e-6; Xc = sparse.diags(1 / rn) @ Xc\n    # user-side matrix incl. each target user\'s own history (fold-in onto item factors learned from train-only clicks)\n    cnt = CL.filter(pl.col(\'ts\') < C).group_by(\'u\', \'ii\').agg(pl.len().alias(\'n\')) if sv_mode == \'std\' else cnt\n    Xu = sparse.csr_matrix((np.ones(cnt.height, np.float32), (cnt[\'u\'].to_numpy(), cnt[\'ii\'].to_numpy())), shape=(NU, NI)) @ sparse.diags(idf)\n    Xu = sparse.diags(1 / (np.sqrt(np.asarray(Xu.multiply(Xu).sum(1)).ravel()) + 1e-6)) @ Xu\n    uu = R[\'u\'].to_numpy().astype(np.int64); iic = R[\'ii\'].to_numpy().astype(np.int64)\n    ex = X.select(\'r\', \'j\', \'pos\').unique([\'r\', \'j\', \'pos\'])\n    # per-event history (last 20 clicks) is shared by all candidates: take rows of one candidate per (r) anyway\n    er_r = ex[\'r\'].to_numpy(); er_j = ex[\'j\'].to_numpy().astype(np.int64); er_w = (1.0 / np.sqrt(ex[\'pos\'].to_numpy())).astype(np.float32)\n    er_l5 = ex[\'pos\'].to_numpy() <= 5\n    nR = R.height\n    svcols = {}\n    for k in (16, 32, 64):\n        U, S, Vt = randomized_svd(Xc, k, random_state=0, n_iter=5)\n        V = (Vt.T * np.sqrt(S)).astype(np.float32); Uf = np.asarray(Xu @ (Vt.T / np.sqrt(S))).astype(np.float32)  # fold-in = U*sqrt(S) for train users\n        Vn = V / (np.linalg.norm(V, axis=1, keepdims=True) + 1e-9); Un = Uf / (np.linalg.norm(Uf, axis=1, keepdims=True) + 1e-9)\n        cand = Vn[iic]\n        svcols[f\'sv_k{k}_ucos\'] = (Un[uu] * cand).sum(1)\n        svcols[f\'sv_k{k}_udot\'] = (Uf[uu] * V[iic]).sum(1)\n        hv = np.zeros((nR, k), np.float32); np.add.at(hv, er_r, Vn[er_j] * er_w[:, None])\n        hn = np.linalg.norm(hv, axis=1); svcols[f\'sv_k{k}_hcos\'] = (hv * cand).sum(1) / (hn + 1e-9)\n        if k == 32:\n            svcols[\'sv_k32_hnorm\'] = hn\n            sims = (Vn[er_j] * Vn[iic[er_r]]).sum(1)\n            mx = np.full(nR, -1.0, np.float32); np.maximum.at(mx, er_r, sims)\n            mx5 = np.full(nR, -1.0, np.float32); np.maximum.at(mx5, er_r[er_l5], sims[er_l5])\n            svcols[\'sv_k32_imax20\'] = mx; svcols[\'sv_k32_imax5\'] = mx5\n        if k == 64:\n            # click centroid vs non-clicked-impression centroid (impressions frozen at C: P_rows only)\n            Sc = sparse.csr_matrix((np.ones(cnt.height, np.float32), (cnt[\'u\'].to_numpy(), cnt[\'ii\'].to_numpy())), shape=(NU, NI))\n            neg = NEGsrc.filter(pl.col(\'clicked\') == 0)\n            Sn = sparse.csr_matrix((np.ones(neg.height, np.float32), (neg[\'u\'].to_numpy(), neg[\'ii\'].to_numpy())), shape=(NU, NI))\n            nC = np.asarray(Sc @ Vn); nC /= np.linalg.norm(nC, axis=1, keepdims=True) + 1e-9\n            nN = np.asarray(Sn @ Vn); nN /= np.linalg.norm(nN, axis=1, keepdims=True) + 1e-9\n            cc = (nC[uu] * cand).sum(1); cn = (nN[uu] * cand).sum(1)\n            svcols[\'sv_k64_ccos\'] = cc; svcols[\'sv_k64_ncos\'] = cn\n            for a, nm in ((0.25, \'025\'), (0.5, \'05\'), (1.0, \'1\')):\n                M = nC - a * nN; M /= np.linalg.norm(M, axis=1, keepdims=True) + 1e-9\n                svcols[f\'sv_k64_con{nm}\'] = (M[uu] * cand).sum(1)\n    R = R.with_columns([pl.Series(c, v.astype(np.float32)) for c, v in svcols.items()])\n\n    # ---------------- MF ----------------\n    if with_mf:\n        mf = fit_mf(P_rows, NU, NI)\n        uu = R[\'u\'].to_numpy().astype(np.int64); ii = R[\'ii\'].to_numpy().astype(np.int64)\n        dot = (mf[\'P\'][uu] * mf[\'Q\'][ii]).sum(1)\n        R = R.with_columns(pl.Series(\'mf_score\', mf[\'b\'] + mf[\'bu\'][uu] + mf[\'bi\'][ii] + dot),\n                           pl.Series(\'mf_dot\', dot), pl.Series(\'mf_bi\', mf[\'bi\'][ii]))\n\n    fill0 = [c for c in R.columns if c.startswith((\'it_imp\', \'it_clk\', \'it_n\', \'us_n\', \'hist_n\', \'hist_imp\', \'hist_noclk\', \'co_\')) and R[c].dtype != pl.List]\n    R = R.with_columns([pl.col(c).fill_null(0) for c in fill0])\n    R = R.with_columns([pl.col(c).fill_null(g) for c in (\'it_ctr\', \'it_ctr24\', \'it_ctr6\')] +\n                       [pl.col(c).fill_null(1.0) for c in (\'it_trend24\', \'it_trend6\', \'it_imptrend24\')] +\n                       [pl.col(\'us_ctr\').fill_null(g), pl.col(\'it_imp_rate\').fill_null(0), pl.col(\'us_repeat_rate\').fill_null(0)])\n\n    # ---------------- event context ----------------\n    main = [\'it_ctr\', \'it_ctr24\', \'it_imp\', \'it_nclk_all\', \'it_clk_dec2d\', \'hist_n_all\', \'hist_noclk\',\n            \'co_cos_max20\', \'co_cos_rw\', \'co_tr_max5\', \'co_tr_last\',\n            \'sv_k16_ucos\', \'sv_k32_ucos\', \'sv_k64_ucos\', \'sv_k32_hcos\', \'sv_k64_hcos\', \'sv_k32_imax20\', \'sv_k64_con05\', \'sv_k64_con025\', \'sv_k64_ccos\'] + ([\'mf_score\', \'mf_dot\'] if with_mf else [])\n    exprs = []\n    for c in main:\n        x = pl.col(c).cast(pl.Float64)\n        exprs += [x.rank(\'average\').over(\'eid\').alias(f\'ev_{c}_rk\'),\n                  ((x - x.mean().over(\'eid\')) / (x.std().over(\'eid\') + 1e-9)).alias(f\'ev_{c}_z\'),\n                  x.max().over(\'eid\').alias(f\'ev_{c}_max\'), x.mean().over(\'eid\').alias(f\'ev_{c}_mean\')]\n    R = R.with_columns(exprs)\n    R = R.drop([c for c in R.columns if c.startswith(\'_\')] + [\'h\', \'h_ts\', \'tb\', \'dt\']).sort(\'r\').drop(\'r\')\n    log(f\'build C={C} rows={R.height} cols={R.width} {time.time()-t0:.0f}s\')\n    return R\n\nGROUPS = [\'it\', \'hist\', \'us\', \'co\', \'ev\', \'tm\', \'sv\', \'mf\']\n\ndef feature_cols(df, groups):\n    cols = []\n    for c in df.columns:\n        if c in ID_COLS: continue\n        gname = c.split(\'_\')[0]\n        if gname == \'ev\':\n            src = c[3:].split(\'_\')[0]  # ev_<src group>_...\n            if \'ev\' in groups and src in groups: cols.append(c)\n        elif gname in groups:\n            cols.append(c)\n    return cols\n', 'build_folds.py': "import polars as pl, sys, os\nsys.path.insert(0,'@@WORK@@')\nfrom feats import load, first_events, build, log\nD='@@WORK@@'; os.makedirs(f'{D}/feat',exist_ok=True)\nTE,TR,EE,ER,NI=load(D)\nDAY=86400; J20=1737331200\nEND=J20+7*DAY; VC=J20+4*DAY\n# validation: V = first event/user in [Jan24,Jan27); training folds end before Jan24\nspecs=[(VC,END-VC,'V')]+[(J20+k*DAY,VC-(J20+k*DAY),f'g{k}') for k in (1,2,3)]\nif len(sys.argv)>1 and sys.argv[1]=='c25':\n    specs=[(J20+5*DAY,2*DAY,'val25')]+[(J20+k*DAY,DAY,f'h{k}') for k in (1,2,3,4)]\nif len(sys.argv)>1 and sys.argv[1]=='c26':\n    specs=[(J20+6*DAY,DAY,'val26')]+[(J20+k*DAY,DAY,f'h{k}') for k in (5,)]+[('eval',None,'eval')]\nif len(sys.argv)>1 and sys.argv[1]=='eft':\n    specs=[(J20+k*DAY,(J20+5*DAY)-(J20+k*DAY),f'e{k}') for k in (1,2,3,4)]+[(J20+k*DAY,END-(J20+k*DAY),f'E{k}') for k in (1,2,3,4,5,6)]\nif len(sys.argv)>1 and sys.argv[1]=='dft':\n    specs=[(J20+k*DAY,(J20+6*DAY)-(J20+k*DAY),f'd{k}') for k in (1,2,3,4,5)]\nif len(sys.argv)>1 and sys.argv[1]=='final':\n    specs=[(J20+k*DAY,END-(J20+k*DAY),f'F{k}') for k in (1,2,3,4,5,6)]+[('eval',None,'eval')]\nif len(sys.argv)>1 and sys.argv[1]=='lite':   # v4_lite: only what the final fit needs (h1..h5 + val26 = f1..f6, eval)\n    specs=[(J20+k*DAY,DAY,f'h{k}') for k in (1,2,3,4,5)]+[(J20+6*DAY,DAY,'val26'),('eval',None,'eval')]\nfor C,W,name in specs:\n    out=f'{D}/feat/{name}.parquet'\n    if os.path.exists(out): continue\n    if C=='eval':\n        C=J20+7*DAY; T_ev=EE; T_rows=ER\n    else:\n        T_ev=first_events(TE,C,W); T_rows=TR.join(T_ev.select('eid'),on='eid')\n    R=build(C,T_ev,T_rows,TE,TR,NI)\n    R.drop('h','h_ts').write_parquet(out)\n    log('wrote',name,R.shape)\n", 'feats.py': '"""Feature builder for TechArena CTR. All stats for a cutoff C use only events with ts < C\n(plus the target events\' own histories, which by construction contain only clicks < their ts)."""\nimport polars as pl, numpy as np, time\nfrom scipy import sparse\nfrom sklearn.utils.extmath import randomized_svd\n\nDAY = 86400\nK_SMOOTH = 20\n\ndef log(*a):\n    print(time.strftime(\'%H:%M:%S\'), *a, flush=True)\n\ndef load(D):\n    TE = pl.read_parquet(f\'{D}/train_events.parquet\')\n    TR = pl.read_parquet(f\'{D}/train_rows.parquet\')\n    EE = pl.read_parquet(f\'{D}/evaluation_events.parquet\')\n    ER = pl.read_parquet(f\'{D}/evaluation_rows.parquet\')\n    # global item index\n    items = pl.concat([TR[\'item\'], ER[\'item\'], TE[\'h_items\'].explode().drop_nulls(), EE[\'h_items\'].explode().drop_nulls()]).unique().sort()\n    imap = pl.DataFrame({\'item\': items, \'ii\': pl.arange(0, len(items), eager=True).cast(pl.Int32)})\n    def mapi(df, col=\'item\'):\n        return df.join(imap.rename({\'item\': col, \'ii\': col + \'_ii\'}), on=col, how=\'left\')\n    TR = mapi(TR).drop(\'item\').rename({\'item_ii\': \'ii\'})\n    ER = mapi(ER).drop(\'item\').rename({\'item_ii\': \'ii\'})\n    def map_hist(E):\n        ex = E.select(\'eid\', \'h_items\').explode(\'h_items\').with_row_index(\'r\')\n        ex = ex.join(imap.rename({\'item\': \'h_items\'}), on=\'h_items\', how=\'left\').sort(\'r\')\n        h = ex.group_by(\'eid\', maintain_order=True).agg(pl.col(\'ii\').drop_nulls().alias(\'h\'))\n        return E.drop(\'h_items\').join(h, on=\'eid\', how=\'left\').with_columns(pl.col(\'h\').fill_null([]))\n    TE = map_hist(TE); EE = map_hist(EE)\n    return TE, TR, EE, ER, len(items)\n\ndef first_events(TE, C, W):\n    """first event per user in [C, C+W)"""\n    return (TE.filter((pl.col(\'ts\') >= C) & (pl.col(\'ts\') < C + W)).sort(\'ts\')\n              .group_by(\'user\', maintain_order=True).first())\n\ndef build(C, T_ev, T_rows, TE, TR, NI, svd_k=64, with_label=True):\n    t0 = time.time()\n    P_ev = TE.filter(pl.col(\'ts\') < C)\n    P_rows = TR.join(P_ev.select(\'eid\', \'user\', \'ts\', \'tb\'), on=\'eid\', how=\'inner\')\n    g = P_rows[\'clicked\'].mean() if len(P_rows) else 0.063\n    ndays = max((C - P_ev[\'ts\'].min()) / DAY, 1e-3) if len(P_ev) else 1e-3\n\n    # ---------- click pool (user, item, ts) ----------\n    def hist_clicks(E):\n        return (E.select(\'user\', \'h\', \'h_ts\').explode([\'h\', \'h_ts\']).drop_nulls()\n                 .rename({\'h\': \'ii\', \'h_ts\': \'ts\'}))\n    pool = pl.concat([hist_clicks(P_ev), hist_clicks(T_ev),\n                      P_rows.filter(pl.col(\'clicked\') == 1).select(\'user\', \'ii\', \'ts\')]).unique()\n    pool = pool.with_columns(pl.col(\'ii\').cast(pl.Int32), pl.col(\'ts\').cast(pl.Int64))\n    # global/item-level pool: train data only (no target-event histories, so nothing aggregates across eval rows);\n    # `pool` (with each target row\'s own history) is used only for that user\'s own features\n    poolg = pl.concat([hist_clicks(P_ev), P_rows.filter(pl.col(\'clicked\') == 1).select(\'user\', \'ii\', \'ts\')]).unique() \\\n              .with_columns(pl.col(\'ii\').cast(pl.Int32), pl.col(\'ts\').cast(pl.Int64))\n    log(\'pool\', len(pool))\n\n    # ---------- item stats ----------\n    it = P_rows.group_by(\'ii\').agg(pl.len().alias(\'i_imp\'), pl.col(\'clicked\').sum().alias(\'i_clk\'),\n                                   (pl.col(\'ts\') >= C - DAY).sum().alias(\'i_imp1d\'),\n                                   pl.col(\'clicked\').filter(pl.col(\'ts\') >= C - DAY).sum().alias(\'i_clk1d\'))\n    it = it.with_columns(((pl.col(\'i_clk\') + K_SMOOTH * g) / (pl.col(\'i_imp\') + K_SMOOTH)).alias(\'i_ctr\'),\n                         ((pl.col(\'i_clk1d\') + K_SMOOTH * g) / (pl.col(\'i_imp1d\') + K_SMOOTH)).alias(\'i_ctr1d\'),\n                         (pl.col(\'i_imp\') / ndays).alias(\'i_imp_rate\'))\n    ip = poolg.group_by(\'ii\').agg(pl.len().alias(\'ip_n\'), pl.col(\'user\').n_unique().alias(\'ip_users\'),\n                                 (pl.col(\'ts\') >= C - DAY).sum().alias(\'ip_1d\'),\n                                 (pl.col(\'ts\') >= C - 3 * DAY).sum().alias(\'ip_3d\'),\n                                 ((pl.col(\'ts\') - C) / DAY).exp().sum().alias(\'ip_dec1\'),\n                                 ((pl.col(\'ts\') - C) / (3 * DAY)).exp().sum().alias(\'ip_dec3\'),\n                                 (pl.col(\'ts\') < 1737331200).sum().alias(\'ip_warm\'))\n    itb = P_rows.group_by(\'ii\', \'tb\').agg(pl.len().alias(\'n\'), pl.col(\'clicked\').sum().alias(\'c\'))\n    itb = itb.with_columns(((pl.col(\'c\') + K_SMOOTH * g) / (pl.col(\'n\') + K_SMOOTH)).alias(\'i_ctr_tb\')).select(\'ii\', \'tb\', \'i_ctr_tb\')\n\n    # ---------- user stats ----------\n    ue = P_rows.group_by(\'user\').agg(pl.col(\'eid\').n_unique().alias(\'u_ev\'), pl.col(\'clicked\').sum().alias(\'u_clk\'),\n                                     pl.col(\'ts\').max().alias(\'u_last_ev\'))\n    ue = ue.with_columns(((pl.col(\'u_clk\') + 5 * g * 10) / (pl.col(\'u_ev\') + 5)).alias(\'u_ctr\'))\n    up = pool.group_by(\'user\').agg(pl.len().alias(\'up_n\'), pl.col(\'ts\').max().alias(\'up_last\'),\n                                   pl.col(\'ii\').n_unique().alias(\'up_uniq\'),\n                                   (pl.col(\'ts\') >= C - DAY).sum().alias(\'up_1d\'))\n    # user-item prior impressions\n    ui = P_rows.group_by(\'user\', \'ii\').agg(pl.len().alias(\'ui_imp\'), pl.col(\'clicked\').sum().alias(\'ui_clk\'),\n                                           pl.col(\'ts\').max().alias(\'ui_last_imp\'))\n    uip = pool.group_by(\'user\', \'ii\').agg(pl.len().alias(\'uip_n\'), pl.col(\'ts\').max().alias(\'uip_last\'))\n\n    # ---------- co-occurrence (user sets) & transitions ----------\n    us = poolg.select(\'user\', \'ii\').unique()\n    ipop = us.group_by(\'ii\').agg(pl.len().alias(\'pop\'))\n    pairs = us.join(us, on=\'user\', suffix=\'_b\').filter(pl.col(\'ii\') != pl.col(\'ii_b\'))\n    co = pairs.group_by(\'ii\', \'ii_b\').agg(pl.len().alias(\'co\'))\n    del pairs\n    co = (co.join(ipop, on=\'ii\').join(ipop.rename({\'ii\': \'ii_b\', \'pop\': \'pop_b\'}), on=\'ii_b\')\n            .with_columns((pl.col(\'co\') / (pl.col(\'pop\') * pl.col(\'pop_b\')).sqrt()).cast(pl.Float32).alias(\'cos\'))\n            .select(pl.col(\'ii\').cast(pl.Int32), pl.col(\'ii_b\').cast(pl.Int32), pl.col(\'co\').cast(pl.Int32), \'cos\'))\n    log(\'cooc pairs\', len(co))\n    seq = poolg.sort(\'user\', \'ts\')\n    tr_pairs = []\n    for lag in (1, 2, 3):\n        tr_pairs.append(seq.with_columns(pl.col(\'ii\').shift(lag).over(\'user\').alias(\'prev\'))\n                           .drop_nulls(\'prev\').select(pl.col(\'prev\').cast(pl.Int32), \'ii\', pl.lit(1.0 / lag).alias(\'w\')))\n    trn = pl.concat(tr_pairs).group_by(\'prev\', \'ii\').agg(pl.col(\'w\').sum().alias(\'trn\'))\n    trn_out = trn.group_by(\'prev\').agg(pl.col(\'trn\').sum().alias(\'trn_tot\'))\n    trn = trn.join(trn_out, on=\'prev\').with_columns((pl.col(\'trn\') / (pl.col(\'trn_tot\') + 5)).cast(pl.Float32).alias(\'trp\')).select(\'prev\', \'ii\', \'trn\', \'trp\')\n\n    # ---------- SVD embeddings of user x item click matrix ----------\n    uidx = us.select(\'user\').unique().with_row_index(\'ur\')\n    m = us.join(uidx, on=\'user\')\n    df_ = np.bincount(m[\'ii\'].to_numpy(), minlength=NI)\n    idf = np.log((len(uidx) + 1) / (df_ + 1)).astype(np.float32)\n    X = sparse.csr_matrix((idf[m[\'ii\'].to_numpy()], (m[\'ur\'].to_numpy(), m[\'ii\'].to_numpy())), shape=(len(uidx), NI))\n    rn = np.sqrt(np.asarray(X.multiply(X).sum(1)).ravel()) + 1e-6\n    X = sparse.diags(1 / rn) @ X\n    U, S, Vt = randomized_svd(X, svd_k, random_state=0, n_iter=4)\n    V = Vt.T.astype(np.float32)  # unscaled singular vectors (best in emb_exp)\n    V /= (np.linalg.norm(V, axis=1, keepdims=True) + 1e-6)\n    V32 = V[:, :32] / (np.linalg.norm(V[:, :32], axis=1, keepdims=True) + 1e-6)\n    # fold-in of every target user\'s own pool (train clicks + own history) onto the train-only factors\n    uidx = pool.select(\'user\').unique().with_row_index(\'ur\')\n    mu = pool.select(\'user\', \'ii\').unique().join(uidx, on=\'user\')\n    Xu = sparse.csr_matrix((idf[mu[\'ii\'].to_numpy()], (mu[\'ur\'].to_numpy(), mu[\'ii\'].to_numpy())), shape=(len(uidx), NI))\n    Xu = sparse.diags(1 / (np.sqrt(np.asarray(Xu.multiply(Xu).sum(1)).ravel()) + 1e-6)) @ Xu\n    U = np.asarray(Xu @ (Vt.T / S)).astype(np.float32)\n    S = np.ones_like(S)\n    log(\'svd done\')\n\n    # ---------- assemble target rows ----------\n    R = T_rows.join(T_ev, on=\'eid\', how=\'left\').with_row_index(\'r\')\n    R = R.with_columns((pl.col(\'ts\') - C).alias(\'stale\'), ((pl.col(\'ts\') // 3600) % 24).cast(pl.Int8).alias(\'hour\'))\n    R = (R.join(it, on=\'ii\', how=\'left\').join(ip, on=\'ii\', how=\'left\').join(itb, on=[\'ii\', \'tb\'], how=\'left\')\n          .join(ue, on=\'user\', how=\'left\').join(up, on=\'user\', how=\'left\')\n          .join(ui, on=[\'user\', \'ii\'], how=\'left\').join(uip, on=[\'user\', \'ii\'], how=\'left\').sort(\'r\'))\n    R = R.with_columns(\n        pl.col(\'h\').list.len().alias(\'h_len\'),\n        (pl.col(\'ts\') - pl.col(\'h_ts\').list.last()).alias(\'h_since_last\'),\n        (pl.col(\'ts\') - pl.col(\'h_ts\').list.first()).alias(\'h_since_first\'),\n        pl.col(\'h\').list.n_unique().alias(\'h_uniq\'),\n        pl.col(\'h_ts\').list.eval((pl.element() >= pl.element().max() - DAY).sum()).list.first().alias(\'h_1d_before_last\'),\n        (pl.col(\'ts\') - pl.col(\'u_last_ev\')).alias(\'u_since_ev\'),\n        (pl.col(\'ts\') - pl.col(\'up_last\')).alias(\'up_since\'),\n        (pl.col(\'ts\') - pl.col(\'ui_last_imp\')).alias(\'ui_since_imp\'),\n        (pl.col(\'ts\') - pl.col(\'uip_last\')).alias(\'uip_since\'),\n        (pl.col(\'ui_imp\') - pl.col(\'ui_clk\')).alias(\'ui_noclk\'),\n        (pl.col(\'ip_n\') / (pl.col(\'i_imp\') + 1)).alias(\'ip_per_imp\'),\n        (pl.col(\'i_ctr_tb\') - pl.col(\'i_ctr\')).alias(\'i_ctr_tb_diff\'),\n    )\n    R = R.with_columns((pl.col(\'h_since_first\') / (pl.col(\'h_len\') - 1).clip(1)).alias(\'h_gap_mean\'),\n                       (pl.col(\'h_uniq\') / pl.col(\'h_len\').clip(1)).alias(\'h_uniq_ratio\'))\n\n    # pairwise candidate x history\n    PR = (R.select(\'r\', \'ii\', \'h\', \'h_ts\', \'ts\').with_columns(pl.int_ranges(pl.col(\'h\').list.len(), 0, -1).alias(\'pos\'))\n            .explode([\'h\', \'h_ts\', \'pos\']).drop_nulls(\'h\')\n            .with_columns(pl.col(\'h\').cast(pl.Int32)))\n    PR = PR.join(co.rename({\'ii\': \'ii\', \'ii_b\': \'h\'}), on=[\'ii\', \'h\'], how=\'left\').with_columns(pl.col(\'cos\').fill_null(0), pl.col(\'co\').fill_null(0))\n    PR = PR.with_columns((0.85 ** (pl.col(\'pos\') - 1)).alias(\'rw\'), (pl.col(\'h\') == pl.col(\'ii\')).alias(\'same\'))\n    agg = PR.group_by(\'r\').agg(\n        pl.col(\'cos\').sum().alias(\'co_sum\'), pl.col(\'cos\').max().alias(\'co_max\'), pl.col(\'cos\').mean().alias(\'co_mean\'),\n        (pl.col(\'cos\') * pl.col(\'rw\')).sum().alias(\'co_rw\'), pl.col(\'co\').sum().alias(\'co_cnt\'),\n        pl.col(\'cos\').filter(pl.col(\'pos\') <= 3).sum().alias(\'co_last3\'),\n        pl.col(\'same\').sum().alias(\'inh_cnt\'), pl.col(\'pos\').filter(pl.col(\'same\')).min().alias(\'inh_pos\'),\n        (pl.col(\'ts\') - pl.col(\'h_ts\')).filter(pl.col(\'same\')).min().alias(\'inh_since\'),\n        (pl.col(\'cos\') > 0).sum().alias(\'co_nz\'))\n    R = R.join(agg, on=\'r\', how=\'left\')\n    # transitions from last 3 history items\n    TRp = (R.select(\'r\', \'ii\', pl.col(\'h\').list.tail(3).alias(\'h3\')).with_columns(pl.int_ranges(pl.col(\'h3\').list.len(), 0, -1).alias(\'pos\'))\n             .explode([\'h3\', \'pos\']).drop_nulls(\'h3\').rename({\'h3\': \'prev\'}).with_columns(pl.col(\'prev\').cast(pl.Int32))\n             .join(trn, on=[\'prev\', \'ii\'], how=\'left\').with_columns(pl.col(\'trn\').fill_null(0), pl.col(\'trp\').fill_null(0)))\n    tagg = TRp.group_by(\'r\').agg(pl.col(\'trn\').filter(pl.col(\'pos\') == 1).sum().alias(\'tr_last\'),\n                                 pl.col(\'trp\').filter(pl.col(\'pos\') == 1).sum().alias(\'trp_last\'),\n                                 pl.col(\'trn\').sum().alias(\'tr_last3\'), pl.col(\'trp\').sum().alias(\'trp_last3\'))\n    R = R.join(tagg, on=\'r\', how=\'left\').sort(\'r\')\n    # SVD cosine: recency weighted mean of history vectors\n    hv = np.zeros((len(R), V.shape[1]), np.float32)\n    ex = PR.select(\'r\', \'h\', \'rw\')\n    np.add.at(hv, ex[\'r\'].to_numpy(), V[ex[\'h\'].to_numpy()] * ex[\'rw\'].to_numpy()[:, None].astype(np.float32))\n    hvn = hv / (np.linalg.norm(hv, axis=1, keepdims=True) + 1e-6)\n    cand = V[R[\'ii\'].to_numpy()]\n    svd_cos = (hvn * cand).sum(1)\n    # user factor from SVD (pool-based, covers clicks beyond last 20)\n    uf = np.zeros((len(R), V.shape[1]), np.float32)\n    um = R.select(\'user\').join(uidx, on=\'user\', how=\'left\')[\'ur\'].to_numpy()\n    ok = ~np.isnan(um.astype(float)) if um.dtype.kind == \'f\' else np.ones(len(um), bool)\n    umi = np.nan_to_num(um.astype(float), nan=0).astype(np.int64)\n    ufv = (U[umi] * np.sqrt(S)).astype(np.float32); ufv[~ok] = 0\n    ufv /= (np.linalg.norm(ufv, axis=1, keepdims=True) + 1e-6)\n    svd_ucos = (ufv * cand).sum(1)\n    hv32 = np.zeros((len(R), 32), np.float32)\n    np.add.at(hv32, ex[\'r\'].to_numpy(), V32[ex[\'h\'].to_numpy()] * ex[\'rw\'].to_numpy()[:, None].astype(np.float32))\n    svd32_cos = (hv32 / (np.linalg.norm(hv32, axis=1, keepdims=True) + 1e-6) * V32[R[\'ii\'].to_numpy()]).sum(1)\n    extra_svd = {}\n    for kk in (16, 24, 48):\n        Vk = V[:, :kk] / (np.linalg.norm(V[:, :kk], axis=1, keepdims=True) + 1e-6)\n        hk = np.zeros((len(R), kk), np.float32)\n        np.add.at(hk, ex[\'r\'].to_numpy(), Vk[ex[\'h\'].to_numpy()] * ex[\'rw\'].to_numpy()[:, None].astype(np.float32))\n        extra_svd[f\'svd{kk}_cos\'] = (hk / (np.linalg.norm(hk, axis=1, keepdims=True) + 1e-6) * Vk[R[\'ii\'].to_numpy()]).sum(1)\n    # pairwise svd similarity: max over history, and with the most recent item\n    pr_r = PR[\'r\'].to_numpy(); pr_c = PR[\'ii\'].to_numpy(); pr_h = PR[\'h\'].to_numpy()\n    sims = np.empty(len(PR), np.float32)\n    for k in range(0, len(PR), 2_000_000):\n        sl = slice(k, k + 2_000_000); sims[sl] = (V[pr_c[sl]] * V[pr_h[sl]]).sum(1)\n    sp = pl.DataFrame({\'r\': pr_r, \'sim\': sims, \'pos\': PR[\'pos\'].to_numpy(), \'same\': PR[\'same\'].to_numpy()}).filter(~pl.col(\'same\'))\n    sagg = sp.group_by(\'r\').agg(pl.col(\'sim\').max().alias(\'svd_maxsim\'), pl.col(\'sim\').filter(pl.col(\'pos\') == 1).max().alias(\'svd_last\'),\n                                pl.col(\'sim\').top_k(3).mean().alias(\'svd_top3\'))\n    R = R.with_columns(pl.Series(\'svd_cos\', svd_cos), pl.Series(\'svd_ucos\', svd_ucos), pl.Series(\'svd_hnorm\', np.linalg.norm(hv, axis=1)),\n                       pl.Series(\'svd32_cos\', svd32_cos), *[pl.Series(k, v) for k, v in extra_svd.items()])\n    R = R.join(sagg, on=\'r\', how=\'left\').sort(\'r\')\n\n    fill0 = [\'i_imp\', \'i_clk\', \'i_imp1d\', \'i_clk1d\', \'ip_n\', \'ip_users\', \'ip_1d\', \'ip_3d\', \'ip_dec1\', \'ip_dec3\', \'ip_warm\',\n             \'u_ev\', \'u_clk\', \'up_n\', \'up_uniq\', \'up_1d\', \'ui_imp\', \'ui_clk\', \'ui_noclk\', \'uip_n\', \'co_sum\', \'co_max\', \'co_mean\', \'co_rw\',\n             \'co_cnt\', \'co_last3\', \'inh_cnt\', \'co_nz\', \'tr_last\', \'trp_last\', \'tr_last3\', \'trp_last3\']\n    R = R.with_columns([pl.col(c).fill_null(0) for c in fill0])\n    R = R.with_columns(pl.col(\'i_ctr\').fill_null(g), pl.col(\'i_ctr1d\').fill_null(g), pl.col(\'i_ctr_tb\').fill_null(g), pl.col(\'u_ctr\').fill_null(g * 10))\n    # within-event relative features\n    rel = [\'i_ctr\', \'ip_dec1\', \'ip_n\', \'co_sum\', \'co_rw\', \'co_max\', \'svd_cos\', \'svd_ucos\', \'svd32_cos\', \'svd16_cos\', \'svd24_cos\', \'svd48_cos\', \'svd_maxsim\', \'svd_top3\', \'tr_last3\', \'uip_n\', \'i_ctr_tb\', \'ui_noclk\']\n    R = R.with_columns([pl.col(c).rank(\'average\').over(\'eid\').alias(c + \'_rk\') for c in rel] +\n                       [(pl.col(c) - pl.col(c).mean().over(\'eid\')).alias(c + \'_dm\') for c in rel] +\n                       [(pl.col(c) / (pl.col(c).max().over(\'eid\') + 1e-9)).alias(c + \'_rmax\') for c in [\'co_sum\', \'svd_cos\', \'ip_dec1\', \'co_rw\']])\n    R = R.with_columns(pl.col(\'co_rw\').max().over(\'eid\').alias(\'ev_co_max\'), pl.col(\'svd_cos\').max().over(\'eid\').alias(\'ev_svd_max\'),\n                       pl.col(\'svd_cos\').mean().over(\'eid\').alias(\'ev_svd_mean\'),\n                       (pl.col(\'inh_cnt\') > 0).sum().over(\'eid\').alias(\'ev_n_inh\'), pl.col(\'i_ctr\').mean().over(\'eid\').alias(\'ev_ictr_mean\'),\n                       pl.col(\'ip_dec1\').mean().over(\'eid\').alias(\'ev_ipdec_mean\'))\n    log(f\'build C={C} rows={len(R)} {time.time()-t0:.0f}s\')\n    starter = dict(g=g)\n    return R\n\nDROP = [\'r\', \'rowid\', \'eid\', \'user\', \'ii\', \'h\', \'h_ts\', \'ts\', \'clicked\', \'u_last_ev\', \'up_last\', \'ui_last_imp\', \'uip_last\', \'fold\']\n', 'extra.py': '"""Extra leak-free features on v3 data (all stats from data with ts < C).\nUsage: python extra.py NAME...   (names: f1..f6, val25, val26, eval)  -> v2work/feat_x/NAME.parquet\n - con variants: k=32/128, recency-weighted (tau=2d) click/non-click centroids, item-side con (users who clicked i vs skipped i)\n - user-cluster item CTR (kmeans on user SVD vectors, 64 & 256 clusters), smoothed toward item CTR\n - within-event z-score and rank for each"""\nimport sys, os, time, numpy as np, polars as pl\nfrom scipy import sparse\nfrom sklearn.utils.extmath import randomized_svd\nfrom sklearn.cluster import MiniBatchKMeans\nsys.path.insert(0, \'@@WORK@@/v3\')\nfrom features import load, click_log\nfrom split import J20, DAY, first_events\nOUT = os.environ.get(\'X_OUT\', \'@@WORK@@/feat_x\'); os.makedirs(OUT, exist_ok=True)\nTE, TR, EE, ER, NI, NU = load(\'@@WORK@@/v3/data\')\nSPEC = {**{f\'f{k}\': (J20 + k * DAY, DAY) for k in range(1, 7)}, \'val25\': (J20 + 5 * DAY, None), \'val26\': (J20 + 6 * DAY, None), \'eval\': (J20 + 7 * DAY, \'eval\'),\n        **{f\'e{k}\': (J20 + k * DAY, (5 - k) * DAY) for k in range(1, 5)}, **{f\'E{k}\': (J20 + k * DAY, (7 - k) * DAY) for k in range(1, 7)}, **{f\'d{k}\': (J20 + k * DAY, (6 - k) * DAY) for k in range(1, 6)}}\ndef nrm(M): return M / (np.linalg.norm(M, axis=1, keepdims=True) + 1e-9)\ndef csr(u, i, w): return sparse.csr_matrix((w.astype(np.float32), (u, i)), shape=(NU, NI))\ndef build(C, T_ev, T_rows):\n    P_ev = TE.filter(pl.col(\'ts\') < C)\n    P_rows = TR.join(P_ev.select(\'eid\', \'u\', \'ts\'), on=\'eid\', how=\'inner\')\n    CL = click_log(C, P_ev, P_rows, T_ev)            # user-level: train data + each target row\'s own history\n    CLg = click_log(C, P_ev, P_rows, T_ev.head(0))   # global/item-level: train data only\n    assert CL[\'ts\'].max() < C and P_rows[\'ts\'].max() < C\n    R = T_rows.join(T_ev.select(\'eid\', \'u\'), on=\'eid\', how=\'inner\').select(\'rowid\', \'eid\', \'u\', \'ii\')\n    uu = R[\'u\'].to_numpy().astype(np.int64); ii = R[\'ii\'].to_numpy().astype(np.int64)\n    cg = CLg.group_by(\'u\', \'ii\').agg(pl.len().alias(\'n\'))\n    Xc = csr(cg[\'u\'].to_numpy(), cg[\'ii\'].to_numpy(), np.ones(cg.height)); Scg = Xc.copy()\n    cnt = CL.group_by(\'u\', \'ii\').agg(pl.len().alias(\'n\'), pl.col(\'ts\').max().alias(\'lt\'))\n    cu = cnt[\'u\'].to_numpy(); ci = cnt[\'ii\'].to_numpy()\n    df_ = np.asarray(Xc.sum(0)).ravel(); idf = np.log((NU + 1) / (df_ + 1)).astype(np.float32)\n    Xc = Xc @ sparse.diags(idf); Xc = sparse.diags(1 / (np.sqrt(np.asarray(Xc.multiply(Xc).sum(1)).ravel()) + 1e-6)) @ Xc\n    Xu = csr(cu, ci, np.ones(cnt.height)) @ sparse.diags(idf); Xu = sparse.diags(1 / (np.sqrt(np.asarray(Xu.multiply(Xu).sum(1)).ravel()) + 1e-6)) @ Xu\n    neg = P_rows.filter(pl.col(\'clicked\') == 0)\n    nu_ = neg[\'u\'].to_numpy(); ni_ = neg[\'ii\'].to_numpy()\n    Sc = csr(cu, ci, np.ones(cnt.height)); Sn = csr(nu_, ni_, np.ones(neg.height))\n    rc = np.exp(-(C - cnt[\'lt\'].to_numpy()) / (2 * DAY)); rn = np.exp(-(C - neg[\'ts\'].to_numpy()) / (2 * DAY))\n    Scr = csr(cu, ci, rc); Snr = csr(nu_, ni_, rn)\n    F = {}\n    for k in (32, 64, 128):\n        U, S, Vt = randomized_svd(Xc, k, random_state=0, n_iter=5)\n        Vn = nrm(Vt.T * np.sqrt(S)); Un = nrm(np.asarray(Xu @ (Vt.T / np.sqrt(S)))); cand = Vn[ii]  # user fold-in\n        nC = nrm(np.asarray(Sc @ Vn)); nN = nrm(np.asarray(Sn @ Vn))\n        F[f\'x_con05_k{k}\'] = (nrm(nC - 0.5 * nN)[uu] * cand).sum(1)\n        if k == 64:\n            nCr = nrm(np.asarray(Scr @ Vn)); nNr = nrm(np.asarray(Snr @ Vn))\n            for a, nm in ((0.5, \'05\'), (1.0, \'1\')):\n                F[f\'x_conr{nm}_k64\'] = (nrm(nCr - a * nNr)[uu] * cand).sum(1)\n            F[\'x_crcos_k64\'] = (nCr[uu] * cand).sum(1)\n            # item side: centroid of users who clicked item minus users who skipped it, vs target user vector\n            # item side / clusters: user vectors from TRAIN-ONLY clicks (Ug), so no target (eval) histories enter item-level stats\n            Ug = nrm(np.asarray(Xc @ (Vt.T / np.sqrt(S))))\n            iC = nrm(np.asarray(Scg.T @ Ug)); iN = nrm(np.asarray(Sn.T @ Ug))\n            F[\'x_icon05_k64\'] = (nrm(iC - 0.5 * iN)[ii] * Un[uu]).sum(1)\n            F[\'x_icos_k64\'] = (iC[ii] * Un[uu]).sum(1)\n            # user-cluster CTR\n            act = np.unique(np.concatenate([P_rows[\'u\'].to_numpy(), cg[\'u\'].to_numpy()]))\n            for nc in (64, 256):\n                km = MiniBatchKMeans(nc, random_state=0, batch_size=4096, n_init=3).fit(Ug[act])\n                lab = km.predict(Ug).astype(np.int32)          # train users\' clusters (train-only vectors) -> cluster CTR tables\n                labt = km.predict(Un[uu]).astype(np.int32)     # target row\'s cluster from its own user vector (own history allowed)\n                g = float(P_rows[\'clicked\'].mean())\n                pr = P_rows.select(\'ii\', \'clicked\', pl.Series(\'cl\', lab[P_rows[\'u\'].to_numpy()]))\n                ict = pr.group_by(\'ii\').agg(((pl.col(\'clicked\').sum() + 20 * g) / (pl.len() + 20)).alias(\'ictr\'))\n                cct = pr.group_by(\'cl\', \'ii\').agg(pl.col(\'clicked\').sum().alias(\'c\'), pl.len().alias(\'n\'))\n                q = pl.DataFrame({\'cl\': labt, \'ii\': ii.astype(np.int32), \'r\': np.arange(len(uu))}).join(cct, on=[\'cl\', \'ii\'], how=\'left\') \\\n                      .join(ict, on=\'ii\', how=\'left\').sort(\'r\').with_columns(pl.col(\'ictr\').fill_null(g), pl.col(\'c\').fill_null(0), pl.col(\'n\').fill_null(0))\n                F[f\'x_clctr{nc}\'] = ((q[\'c\'] + 10 * q[\'ictr\']) / (q[\'n\'] + 10)).to_numpy()\n                F[f\'x_clctr{nc}_lift\'] = F[f\'x_clctr{nc}\'] / q[\'ictr\'].to_numpy()\n                F[f\'x_climp{nc}\'] = q[\'n\'].to_numpy().astype(np.float32)\n    R = R.with_columns([pl.Series(c, v.astype(np.float32)) for c, v in F.items()])\n    zc = [c for c in F if not c.startswith(\'x_climp\')]\n    R = R.with_columns([((pl.col(c) - pl.col(c).mean().over(\'eid\')) / (pl.col(c).std().over(\'eid\') + 1e-6)).alias(c + \'_z\') for c in zc] +\n                       [pl.col(c).rank(\'average\').over(\'eid\').alias(c + \'_rk\') for c in zc])\n    return R.drop(\'eid\', \'u\', \'ii\')\nfor name in sys.argv[1:]:\n    t0 = time.time(); C, W = SPEC[name]\n    if W == \'eval\': R = build(C, EE, ER)\n    else:\n        T_ev = first_events(TE, C, W); R = build(C, T_ev, TR.join(T_ev.select(\'eid\'), on=\'eid\', how=\'semi\'))\n    R.write_parquet(f\'{OUT}/{name}.parquet\'); print(name, R.shape, f\'{time.time()-t0:.0f}s\', flush=True)\n', 'extra2.py': '"""Extra features #2 (leak-free; everything from ts < C):\n - ridge \'taste vector\' per user on frozen SVD k64 item vectors (clicks y=1 incl. pre-C click log, skipped impressions y=0 or -0.5)\n - user state: time since last event/click, event-level click rate, recent-event click streaks, events in last 24h before C\nUsage: python extra2.py NAME... -> v2work/feat_x2/NAME.parquet"""\nimport sys, os, time, numpy as np, polars as pl\nfrom scipy import sparse\nfrom sklearn.utils.extmath import randomized_svd\nsys.path.insert(0, \'@@WORK@@/v3\')\nfrom features import load, click_log\nfrom split import J20, DAY, first_events\nOUT = \'@@WORK@@/feat_x2\'; os.makedirs(OUT, exist_ok=True)\nTE, TR, EE, ER, NI, NU = load(\'@@WORK@@/v3/data\')\nSPEC = {**{f\'f{k}\': (J20 + k * DAY, DAY) for k in range(1, 7)}, \'val25\': (J20 + 5 * DAY, None), \'val26\': (J20 + 6 * DAY, None),\n        \'val22\': (J20 + 2 * DAY, None), \'eval\': (J20 + 7 * DAY, \'eval\'),\n        **{f\'e{k}\': (J20 + k * DAY, (5 - k) * DAY) for k in range(1, 5)}, **{f\'E{k}\': (J20 + k * DAY, (7 - k) * DAY) for k in range(1, 7)}, **{f\'d{k}\': (J20 + k * DAY, (6 - k) * DAY) for k in range(1, 6)}}\ndef nrm(M): return M / (np.linalg.norm(M, axis=1, keepdims=True) + 1e-9)\ndef ridge_users(Wm, Ym, V, lam):\n    """per-user solve (V_I^T W V_I + lam I) w = V_I^T (W*y). Wm, Ym: csr users x items (weights, weight*target)."""\n    k = V.shape[1]; A = np.zeros((Wm.shape[0], k, k), np.float32)\n    for a in range(k):\n        cols = Wm @ (V * V[:, a:a + 1])          # (NU, k): sum_i w_ui v_ia v_ib\n        A[:, a, :] = cols\n    A += lam * np.eye(k, dtype=np.float32)[None]\n    b = np.asarray(Ym @ V, dtype=np.float32)\n    return np.linalg.solve(A, b[:, :, None])[:, :, 0]\ndef build(C, T_ev, T_rows):\n    P_ev = TE.filter(pl.col(\'ts\') < C)\n    P_rows = TR.join(P_ev.select(\'eid\', \'u\', \'ts\'), on=\'eid\', how=\'inner\')\n    CL = click_log(C, P_ev, P_rows, T_ev)            # user-level: train data + each target row\'s own history\n    CLg = click_log(C, P_ev, P_rows, T_ev.head(0))   # global/item-level: train data only\n    assert CL[\'ts\'].max() < C and P_rows[\'ts\'].max() < C\n    R = T_rows.join(T_ev.select(\'eid\', \'u\', \'ts\'), on=\'eid\', how=\'inner\').select(\'rowid\', \'eid\', \'u\', \'ii\', \'ts\')\n    uu = R[\'u\'].to_numpy().astype(np.int64); ii = R[\'ii\'].to_numpy().astype(np.int64)\n    # SVD exactly as v3 (tf-idf, row-normalised user x item click matrix)\n    cnt = CLg.group_by(\'u\', \'ii\').agg(pl.len().alias(\'n\'))\n    Xc = sparse.csr_matrix((np.ones(cnt.height, np.float32), (cnt[\'u\'].to_numpy(), cnt[\'ii\'].to_numpy())), shape=(NU, NI))\n    df_ = np.asarray(Xc.sum(0)).ravel(); idf = np.log((NU + 1) / (df_ + 1)).astype(np.float32)\n    Xc = Xc @ sparse.diags(idf); Xc = sparse.diags(1 / (np.sqrt(np.asarray(Xc.multiply(Xc).sum(1)).ravel()) + 1e-6)) @ Xc\n    U, S, Vt = randomized_svd(Xc, 64, random_state=0, n_iter=5); Vn = nrm(Vt.T * np.sqrt(S)).astype(np.float32)\n    # (user,item) labels: clicked anywhere before C -> 1 ; shown-but-never-clicked -> 0\n    pos = CL.select(\'u\', \'ii\').unique().with_columns(pl.lit(1.0).alias(\'y\'))\n    neg = P_rows.filter(pl.col(\'clicked\') == 0).select(\'u\', \'ii\').unique().join(pos.select(\'u\', \'ii\'), on=[\'u\', \'ii\'], how=\'anti\').with_columns(pl.lit(0.0).alias(\'y\'))\n    L = pl.concat([pos, neg])\n    lu = L[\'u\'].to_numpy(); li = L[\'ii\'].to_numpy(); ly = L[\'y\'].to_numpy().astype(np.float32)\n    Wm = sparse.csr_matrix((np.ones(len(ly), np.float32), (lu, li)), shape=(NU, NI))\n    cand = Vn[ii]; F = {}\n    for lam, yneg, nm in ((1.0, 0.0, \'l1\'), (10.0, 0.0, \'l10\'), (3.0, -0.5, \'l3n\')):\n        yy = np.where(ly > 0, 1.0, yneg).astype(np.float32)\n        Ym = sparse.csr_matrix((yy, (lu, li)), shape=(NU, NI))\n        Wu = ridge_users(Wm, Ym, Vn, lam)\n        F[f\'r_{nm}\'] = (Wu[uu] * cand).sum(1)\n    # user state features\n    ue = P_ev.select(\'eid\', \'u\', \'ts\').join(P_rows.group_by(\'eid\').agg(pl.col(\'clicked\').sum().alias(\'nc\')), on=\'eid\', how=\'left\') \\\n             .with_columns(pl.col(\'nc\').fill_null(0)).sort(\'u\', \'ts\')\n    us = ue.group_by(\'u\').agg(pl.col(\'ts\').last().alias(\'last_ev\'), (pl.col(\'nc\') > 0).mean().alias(\'s_evrate\'), pl.len().alias(\'s_nev\'),\n                              pl.col(\'nc\').tail(1).sum().alias(\'s_clk_last1\'), pl.col(\'nc\').tail(3).sum().alias(\'s_clk_last3\'),\n                              (pl.col(\'nc\').tail(5) > 0).sum().alias(\'s_evclk_last5\'), (pl.col(\'ts\') >= C - DAY).sum().alias(\'s_nev24h\'),\n                              pl.col(\'ts\').filter(pl.col(\'nc\') > 0).max().alias(\'last_clkev\'))\n    lc = CL.group_by(\'u\').agg(pl.col(\'ts\').max().alias(\'last_click\'), (pl.col(\'ts\') >= C - DAY).sum().alias(\'s_clk24h\'))\n    E = T_ev.select(\'eid\', \'u\', \'ts\').join(us, on=\'u\', how=\'left\').join(lc, on=\'u\', how=\'left\').with_columns(\n        ((pl.col(\'ts\') - pl.col(\'last_ev\')) / 3600).alias(\'s_since_ev_h\'), ((pl.col(\'ts\') - pl.col(\'last_click\')) / 3600).alias(\'s_since_clk_h\'),\n        ((pl.col(\'ts\') - pl.col(\'last_clkev\')) / 3600).alias(\'s_since_clkev_h\')).drop(\'u\', \'ts\', \'last_ev\', \'last_click\', \'last_clkev\')\n    R = R.with_columns([pl.Series(c, v.astype(np.float32)) for c, v in F.items()])\n    R = R.with_columns([((pl.col(c) - pl.col(c).mean().over(\'eid\')) / (pl.col(c).std().over(\'eid\') + 1e-6)).alias(c + \'_z\') for c in F] +\n                       [(pl.col(c) - pl.col(c).max().over(\'eid\')).alias(c + \'_dmax\') for c in F])\n    R = R.join(E, on=\'eid\', how=\'left\')\n    return R.drop(\'eid\', \'u\', \'ii\', \'ts\')\nfor name in sys.argv[1:]:\n    t0 = time.time(); C, W = SPEC[name]\n    if W == \'eval\': R = build(C, EE, ER)\n    else:\n        T_ev = first_events(TE, C, W); R = build(C, T_ev, TR.join(T_ev.select(\'eid\'), on=\'eid\', how=\'semi\'))\n    R.write_parquet(f\'{OUT}/{name}.parquet\'); print(name, R.shape, f\'{time.time()-t0:.0f}s\', flush=True)\n', 'cl.py': '"""Shared loader + feature-matrix cache (step C). Matrices cached as parquet keyed by hash(table names, feature list, sample)."""\nimport os, sys, hashlib, json, gc, numpy as np, polars as pl\nsys.path.insert(0,\'@@WORK@@\'); sys.path.insert(1,\'@@WORK@@/v3\')\nfrom feats import DROP\nfrom features import feature_cols, GROUPS\nD=\'@@WORK@@\'; P2=f\'{D}/feat\'; P3=f\'{D}/feat_v3\'; CACHE=f\'{D}/cache_mx\'; os.makedirs(CACHE,exist_ok=True)\nV2_OF={\'f1\':\'h1\',\'f2\':\'h2\',\'f3\':\'h3\',\'f4\':\'h4\',\'f5\':\'h5\',\'f6\':\'val26\',\'val25\':\'val25\',\'val26\':\'val26\',\'eval\':\'eval\'}\nV2_OF.update({f\'{p}{k}\':f\'{p}{k}\' for p in \'eEd\' for k in range(1,7)})\nTIME=[\'tm_stale\',\'tm_hour\',\'tm_tb\',\'tm_dt\',\'v2_tb\',\'v2_dt\',\'v2_stale\',\'v2_hour\']\nKEEP=[\'rowid\',\'eid\',\'u\',\'ts\',\'clicked\',\'us_cold\',\'us_nohist\',\'tm_stale\']\ndef load_tab(n, extra=(), variant=\'notime\'):\n    a=pl.read_parquet(f\'{P3}/{n}.parquet\'); b=pl.read_parquet(f\'{P2}/{V2_OF[n]}.parquet\')\n    f2=[c for c in b.columns if c not in DROP]\n    a=a.join(b.select([\'rowid\']+f2).rename({c:\'v2_\'+c for c in f2}),on=\'rowid\',how=\'left\')\n    for sub in (os.environ.get(\'XDIR\',\'feat_x\'),\'feat_x2\')+tuple(extra):\n        a=a.join(pl.read_parquet(f\'{D}/{sub}/{n}.parquet\'),on=\'rowid\',how=\'left\')\n    if \'+\' in variant:\n        raise ValueError(\'mods not used in v6\')\n    return a\ndef featlist(df, variant=\'notime\'):\n    f=feature_cols(df,GROUPS)+[c for c in df.columns if c.startswith((\'v2_\',\'x_\',\'r_\',\'s_\',\'ev2_\',\'rc_\',\'sh_\',\'hs_\',\'lr\'))]\n    if \'notime\' in variant: f=[c for c in f if c not in TIME]\n    if variant.split(\'+\')[0]==\'stale\': f=[c for c in f if c not in TIME or c==\'tm_stale\']\n    return f\ndef usample(df, frac):\n    if frac>=1: return df\n    return df.filter((pl.col(\'u\').hash(seed=7) % 10000) < int(frac*10000))\ndef matrix(n, frac=1.0, variant=\'notime\', extra=()):\n    """returns (X float32, meta DataFrame, feats); cached."""\n    key=hashlib.md5(json.dumps([n,frac,variant,list(extra),[os.path.getmtime(f) for f in (f\'{P3}/{n}.parquet\',f\'{P2}/{V2_OF[n]}.parquet\',f\'{D}/feat_x/{n}.parquet\',f\'{D}/feat_x2/{n}.parquet\')]+[os.path.getmtime(f\'{D}/{x}/{n}.parquet\') for x in extra]]+([os.environ[\'XDIR\']] if os.environ.get(\'XDIR\',\'feat_x\')!=\'feat_x\' else [])).encode()).hexdigest()[:12]\n    fp=f\'{CACHE}/{n}_{key}.parquet\'\n    if os.path.exists(fp):\n        t=pl.read_parquet(fp); feats=[c for c in t.columns if c not in KEEP or c in (\'us_cold\',\'us_nohist\')]\n        fl=json.load(open(fp+\'.json\')); \n    else:\n        t=usample(load_tab(n,extra,variant),frac); fl=featlist(t,variant)\n        t=t.select(list(dict.fromkeys([c for c in KEEP if c in t.columns]+fl)))\n        t=t.with_columns([pl.col(c).cast(pl.Float32) for c in fl if t[c].dtype!=pl.Float32 and c not in KEEP])\n        t.write_parquet(fp); json.dump(fl,open(fp+\'.json\',\'w\'))\n    X=t.select(fl).to_numpy().astype(np.float32); meta=t.select([c for c in KEEP if c in t.columns]); del t; gc.collect()\n    return X, meta, fl\n', 'ens.py': '"""Phase 1 ensemble components. python ens.py CUT TAG [--seeds 0,1] [--folds f|fe] [--stale] [--rc] [--drop re] [--params json] [--rounds 400]\nCUT=25/26: train f1..f(CUT-21) (+e/d folds) full data, predict val{CUT}; CUT=27: train f1..f6 (+E1..E6), predict eval.\nSaves ens/{tag}_c{CUT}_s{seed}.npy with raw scores at rounds 300,400(,500) (columns)."""\nimport sys, re, json, gc, time, argparse, numpy as np, lightgbm as lgb\nfrom cl import matrix\nfrom sklearn.metrics import roc_auc_score\nap=argparse.ArgumentParser(); ap.add_argument(\'cut\',type=int); ap.add_argument(\'tag\'); ap.add_argument(\'--seeds\',default=\'0\')\nap.add_argument(\'--folds\',default=\'f\'); ap.add_argument(\'--stale\',action=\'store_true\'); ap.add_argument(\'--rc\',action=\'store_true\')\nap.add_argument(\'--drop\',default=None); ap.add_argument(\'--params\',default=\'{}\'); ap.add_argument(\'--rounds\',type=int,default=400); ap.add_argument(\'--threads\',type=int,default=5); ap.add_argument(\'--lr\',default=\'\')\na=ap.parse_args(); t0=time.time()\nextra=((\'feat_rc\',) if a.rc else ())+((a.lr,) if a.lr else ()); var=\'stale\' if a.stale else \'notime\'\nefold={25:[f\'e{k}\' for k in range(1,5)],26:[f\'d{k}\' for k in range(1,6)],27:[f\'E{k}\' for k in range(1,7)]}[a.cut]\ntrn=([f\'f{k}\' for k in range(1,a.cut-20)] if \'f\' in a.folds else [])+(efold if \'e\' in a.folds else [])\nXs=[];ys=[];feats=None\nfor n in trn:\n    X,m,f=matrix(n,1.0,variant=var,extra=extra); feats=f; Xs.append(X); ys.append(m[\'clicked\'].to_numpy()); del X,m; gc.collect()\nX=np.concatenate(Xs); del Xs; gc.collect(); y=np.concatenate(ys)\nkeep=[i for i,c in enumerate(feats) if not (a.drop and re.search(a.drop,c))]; fl=[feats[i] for i in keep]\nif len(keep)<len(feats): X=np.ascontiguousarray(X[:,keep])\nP=dict(objective=\'binary\',learning_rate=0.05,num_leaves=63,min_data_in_leaf=200,feature_fraction=0.7,bagging_fraction=0.8,\n       bagging_freq=1,lambda_l2=10.0,verbose=-1,num_threads=a.threads); P.update(json.loads(a.params))\nds=lgb.Dataset(X,y,feature_name=fl,free_raw_data=True).construct(); del X; gc.collect()\nvn=\'eval\' if a.cut==27 else f\'val{a.cut}\'\nXv,mv,fv=matrix(vn,1.0,variant=var,extra=extra); Xv=Xv[:,[fv.index(c) for c in fl]]\nRS=[r for r in (300,400,500) if r<=a.rounds]\nfor s in map(int,a.seeds.split(\',\')):\n    mdl=lgb.train(dict(P,seed=42+s),ds,a.rounds)\n    S=np.stack([mdl.predict(Xv,num_iteration=r,raw_score=True) for r in RS],1)\n    np.save(f\'ens/{a.tag}_c{a.cut}_s{s}.npy\',S.astype(np.float32))\n    if a.cut==27: mdl.save_model(f\'ens/{a.tag}_c27_s{s}.txt\')\n    msg=dict(cut=a.cut,tag=a.tag,seed=s,n=len(y),nf=len(fl),min=round((time.time()-t0)/60,1))\n    if a.cut!=27: msg[\'auc\']={r:round(roc_auc_score(mv[\'clicked\'].to_numpy(),S[:,i]),5) for i,r in enumerate(RS)}\n    print(json.dumps(msg),flush=True); open(\'ens/ens_log.jsonl\',\'a\').write(json.dumps(msg)+\'\\n\')\n', 'write_sub.py': '"""Write + validate submission from eval_pred.parquet. Usage: python write_sub.py OUT.csv"""\nimport sys, csv, math, hashlib, zipfile, os, numpy as np, polars as pl\nout=sys.argv[1]\npr=pl.read_parquet(os.environ.get(\'PRED\',\'@@WORK@@/eval_pred.parquet\'))\nev=pl.read_csv(\'@@WORK@@/evaluation.csv\',columns=[\'rowid\'],schema_overrides={\'rowid\':pl.Utf8})\ndf=ev.join(pr,on=\'rowid\',how=\'left\',maintain_order=\'left\')\np=np.nan_to_num(df[\'p\'].to_numpy().astype(np.float64),nan=0.0629,posinf=1.0,neginf=0.0).clip(0,1)\np[np.isnan(df[\'p\'].to_numpy().astype(float))]=0.0629\nprint(\'missing preds filled:\',int(df[\'p\'].null_count()))\nwith open(out,\'w\',encoding=\'utf-8\',newline=\'\') as f:\n    f.write(\'rowid,predicted_probability\\n\'); f.writelines(f\'{r},{v:.8g}\\n\' for r,v in zip(df[\'rowid\'].to_list(),p))\n# ---- validator\nids=ev[\'rowid\'].to_list(); errs=[]\nwith open(out,newline=\'\',encoding=\'utf-8\') as f:\n    r=csv.reader(f); h=next(r); rows=list(r)\nif h!=[\'rowid\',\'predicted_probability\']: errs.append(\'header\')\nif len(rows)!=529390 or len(ids)!=529390: errs.append(f\'rows {len(rows)}\')\nsid=[x[0] for x in rows]\nif len(set(sid))!=len(sid) or set(sid)!=set(ids): errs.append(\'rowid set mismatch\')\nvals=[float(x[1]) for x in rows]\nif not all(math.isfinite(v) and 0<=v<=1 for v in vals): errs.append(\'bad values\')\nv=np.array(vals); print(f\'rows={len(rows)} unique={len(set(sid))} min={v.min():.4g} mean={v.mean():.4g} max={v.max():.4g} distinct={len(np.unique(v))}\')\nprint(\'VALID\' if not errs else \'INVALID \'+\';\'.join(errs))\nzp=out+\'.zip\'\nwith zipfile.ZipFile(zp,\'w\',zipfile.ZIP_DEFLATED) as z: z.write(out,arcname=os.path.basename(out))\nwith zipfile.ZipFile(zp) as z: print(\'zip contents\',z.namelist())\nprint(\'sha256\',hashlib.sha256(open(out,\'rb\').read()).hexdigest())\n', 'validate.py': '"""Standalone validator: python validate.py SUB.csv"""\nimport sys, csv, math, numpy as np, polars as pl\nout=sys.argv[1]\nids=pl.read_csv(\'@@WORK@@/evaluation.csv\',columns=[\'rowid\'],schema_overrides={\'rowid\':pl.Utf8})[\'rowid\'].to_list()\nraw=open(out,\'rb\').read(); errs=[]\ntry: raw.decode(\'utf-8\')\nexcept Exception: errs.append(\'not utf-8\')\nwith open(out,newline=\'\',encoding=\'utf-8\') as f:\n    r=csv.reader(f); h=next(r); rows=list(r)\nif h!=[\'rowid\',\'predicted_probability\']: errs.append(f\'header {h}\')\nif len(rows)!=529390: errs.append(f\'rows {len(rows)}\')\nsid=[x[0] for x in rows]\nif len(set(sid))!=len(sid): errs.append(\'dup rowids\')\nif set(sid)!=set(ids): errs.append(\'rowid set mismatch\')\nv=np.array([float(x[1]) for x in rows])\nif not (np.isfinite(v).all() and (v>=0).all() and (v<=1).all()): errs.append(\'bad values\')\nprint(f\'rows={len(rows)} unique={len(set(sid))} min={v.min():.4g} mean={v.mean():.5g} max={v.max():.4g} distinct={len(np.unique(v))}\')\nprint(\'VALID\' if not errs else \'INVALID \'+\';\'.join(errs))\n', 'blend_lite.py': '"""v4_lite final blend: equal-weight mean of 2 component logits on eval -> eval_pred.parquet (rowid, p).\nbase = mean raw score of 2 seeds at 400 rounds; slow = 1 seed at 500 rounds."""\nimport glob, numpy as np, polars as pl\nfrom cl import matrix\nCOL = {\'base\': 1, \'slow\': 2}; NS = {\'base\': 2, \'slow\': 1}\n_, m, _ = matrix(\'eval\', 1.0)\nc = {}\nfor t, ci in COL.items():\n    fs = sorted(glob.glob(f\'ens/{t}_c27_s*.npy\')); assert len(fs) == NS[t], (t, fs)\n    c[t] = np.mean([np.load(f)[:, ci] for f in fs], 0); assert len(c[t]) == m.height\nz = np.mean(list(c.values()), 0); p = 1 / (1 + np.exp(-z))\nprint(\'corr base/slow\', round(float(np.corrcoef(c[\'base\'], c[\'slow\'])[0, 1]), 4))\npl.DataFrame({\'rowid\': m[\'rowid\'], \'p\': p}).write_parquet(\'eval_pred.parquet\'); print(\'wrote eval_pred.parquet\', m.height)\n', 'lrfeat2.py': '"""F1 (data_dig_v1): history-centroid cosine in an exact low-rank SVD space. Train-only, rebuilt at each fold cutoff C.\nMatrix: users x items, binary, rows = each user\'s last N distinct history clicks (from train events with ts < C),\ncolumns scaled by 1/sqrt(pop+5); exact svds (ARPACK) k=17; item vectors = comps 2..K+1 (popularity comp dropped) * s.\nFeature: cos(candidate vector, sum of own-history normalised vectors). Within-event _z and _dmax. Eval: fitted at J27 on train only.\nUsage: python lrfeat.py NAME... -> feat_lr/NAME.parquet (rowid, lr{K}_cos/_z/_dmax for K in 10,12,16)"""\nimport sys, os, time, numpy as np, polars as pl, scipy.sparse as sp\nfrom scipy.sparse.linalg import svds\nsys.path.insert(0, \'@@WORK@@/v3\')\nfrom features import load\nfrom split import J20, DAY, first_events\nTE, TR, EE, ER, NI, NU = load(\'@@WORK@@/v3/data\')\nN = int(os.environ.get(\'LR_N\', 40)); OUT = os.environ.get(\'LR_OUT\', \'@@WORK@@/feat_lr2\'); os.makedirs(OUT, exist_ok=True)\nKS = tuple(int(x) for x in os.environ.get(\'LR_KS\', \'10,12,16\').split(\',\'))\nCON = int(os.environ.get(\'LR_CON\', 0))\nSPEC = {**{f\'f{k}\': (J20 + k * DAY, DAY) for k in range(1, 7)}, \'val25\': (J20 + 5 * DAY, None), \'val26\': (J20 + 6 * DAY, None)}\n_V = {}\ndef vecs(C):\n    if C in _V: return _V[C]\n    h = (TE.filter(pl.col(\'ts\') < C).select(\'u\', \'h\', \'h_ts\').explode([\'h\', \'h_ts\']).drop_nulls(\'h\')\n           .select(\'u\', pl.col(\'h\').cast(pl.Int32).alias(\'ii\'), pl.col(\'h_ts\').cast(pl.Int64).alias(\'t\')).unique().sort(\'u\', \'t\', \'ii\'))\n    h = h.with_columns(pl.int_range(pl.len()).over(\'u\').alias(\'p\'), pl.len().over(\'u\').alias(\'L\')).filter(pl.col(\'L\') - pl.col(\'p\') <= N).unique([\'u\', \'ii\'])\n    M = sp.csr_matrix((np.ones(h.height, np.float32), (h[\'u\'].to_numpy(), h[\'ii\'].to_numpy())), shape=(NU, NI))\n    pop = np.asarray(M.sum(0)).ravel(); M = M.multiply(1 / np.sqrt(pop + 5)).tocsr()\n    _, s, vt = svds(M, k=max(max(KS) + 1, 17), random_state=0); o = np.argsort(-s); s, vt = s[o], vt[o]\n    print(time.strftime(\'%T\'), \'C\', C, \'pairs\', h.height, \'sv\', np.round(s[:15], 2), flush=True)\n    _V[C] = (s, vt); return _V[C]\ndef build(C, T_ev, T_rows):\n    s, vt = vecs(C)\n    E = T_ev.select(\'eid\', \'h\').with_row_index(\'k\'); kk = E.select(\'eid\', \'k\')\n    H = E.select(\'k\', \'h\').explode(\'h\').drop_nulls(\'h\'); hk = H[\'k\'].to_numpy(); hi = H[\'h\'].to_numpy().astype(np.int64)\n    R = T_rows.select(\'rowid\', \'eid\', \'ii\').join(kk, on=\'eid\', how=\'left\'); rk = R[\'k\'].to_numpy(); ri = R[\'ii\'].to_numpy().astype(np.int64)\n    F = {}\n    if CON:\n        P_ev = TE.filter(pl.col(\'ts\') < C).select(\'eid\', \'u\')\n        ng = TR.filter(pl.col(\'clicked\') == 0).join(P_ev, on=\'eid\', how=\'inner\').select(\'u\', \'ii\').unique()\n        ue = T_ev.select(\'u\').with_row_index(\'k\').join(ng, on=\'u\', how=\'inner\'); nk = ue[\'k\'].to_numpy(); ni = ue[\'ii\'].to_numpy().astype(np.int64)\n    for K in KS:\n        V = (vt[1:K + 1].T * s[1:K + 1]); V = V / (np.linalg.norm(V, axis=1, keepdims=True) + 1e-9)\n        hc = np.zeros((E.height, K)); np.add.at(hc, hk, V[hi])\n        hcr = hc[rk]; nr = np.linalg.norm(hcr, axis=1)\n        x = np.where(nr > 0, (V[ri] * hcr).sum(1) / (nr + 1e-9), np.nan)\n        F[f\'lr{K}_cos\'] = x.astype(np.float32)\n        if CON:\n            nc = np.zeros((E.height, K)); np.add.at(nc, nk, V[ni])\n            hn = hc / (np.linalg.norm(hc, axis=1, keepdims=True) + 1e-9); nn = nc / (np.linalg.norm(nc, axis=1, keepdims=True) + 1e-9)\n            for al, nm in ((0.5, \'05\'), (1.0, \'1\')):\n                d = (hn - al * nn)[rk]; dn = np.linalg.norm(d, axis=1)\n                F[f\'lr{K}_con{nm}\'] = np.where(dn > 0, (V[ri] * d).sum(1) / (dn + 1e-9), np.nan).astype(np.float32)\n    R = R.select(\'rowid\', \'eid\').with_columns([pl.Series(c, v).fill_nan(None) for c, v in F.items()])\n    R = R.with_columns([((pl.col(c) - pl.col(c).mean().over(\'eid\')) / (pl.col(c).std().over(\'eid\') + 1e-6)).alias(c.replace(\'_cos\', \'_z\') if c.endswith(\'_cos\') else c + \'_z\') for c in F] +\n                       [(pl.col(c) - pl.col(c).max().over(\'eid\')).alias(c.replace(\'_cos\', \'_dmax\') if c.endswith(\'_cos\') else c + \'_dmax\') for c in F])\n    return R.drop(\'eid\')\nfor n in sys.argv[1:]:\n    t0 = time.time()\n    if n == \'eval\': R = build(J20 + 7 * DAY, EE, ER)\n    else:\n        C, W = SPEC[n]; T_ev = first_events(TE, C, W); R = build(C, T_ev, TR.join(T_ev.select(\'eid\'), on=\'eid\', how=\'semi\'))\n    R.write_parquet(f\'{OUT}/{n}.parquet\'); print(n, R.shape, f\'{time.time()-t0:.0f}s\', flush=True)\n', 'v3/prep.py': '"""Parse train/evaluation CSVs into compact parquet tables (dense int ids).\n\nOutputs in v3/data/:\n  rows_train.parquet  : rowid, eid, ii (item idx), clicked\n  rows_eval.parquet   : rowid, eid, ii, ord (original row order)\n  ev_train.parquet / ev_eval.parquet : eid, u (user idx), ts, tb, dt, h (list[int item idx]), h_ts (list[int])\n  items.parquet / users.parquet : id string -> idx\nUsage: python prep.py [DATA_DIR]\n"""\nimport polars as pl, sys, time, os\nDATA = sys.argv[1] if len(sys.argv) > 1 else \'@@RAW@@\'\nOUT = os.path.join(os.path.dirname(os.path.abspath(__file__)), \'data\')\nos.makedirs(OUT, exist_ok=True)\nt0 = time.time()\ndef log(*a): print(f\'[{time.time()-t0:6.1f}s]\', *a, flush=True)\n\ndef scan(name):\n    return pl.scan_csv(f\'{DATA}/{name}.csv\', schema_overrides={\'history_item_ids\': pl.Utf8, \'history_timestamps\': pl.Utf8,\n                       \'rowid\': pl.Utf8, \'event_id\': pl.Utf8, \'user_id\': pl.Utf8, \'item_id\': pl.Utf8})\n\nraw = {}\nfor name, lab in ((\'train\', True), (\'evaluation\', False)):\n    lf = scan(name)\n    cols = [\'rowid\', \'event_id\', \'item_id\'] + ([\'clicked\'] if lab else [])\n    rows = lf.select(cols).collect(engine=\'streaming\')\n    ev = (lf.select(\'event_id\', \'user_id\', \'timestamp\', \'time_bucket\', \'day_type\', \'history_item_ids\', \'history_timestamps\')\n            .unique(subset=\'event_id\', keep=\'any\')\n            .with_columns(pl.col(\'history_item_ids\').str.json_decode(pl.List(pl.Utf8)).alias(\'h_str\'),\n                          pl.col(\'history_timestamps\').str.json_decode(pl.List(pl.Int64)).alias(\'h_ts\'))\n            .drop(\'history_item_ids\', \'history_timestamps\').collect(engine=\'streaming\'))\n    raw[name] = (rows, ev)\n    log(name, rows.shape, ev.shape)\n\ntr_rows, tr_ev = raw[\'train\']; ev_rows, ev_ev = raw[\'evaluation\']\nitems = pl.concat([tr_rows[\'item_id\'], ev_rows[\'item_id\'], tr_ev[\'h_str\'].explode().drop_nulls(),\n                   ev_ev[\'h_str\'].explode().drop_nulls()]).unique().sort()\nitems = pl.DataFrame({\'item_id\': items, \'ii\': pl.arange(0, len(items), eager=True).cast(pl.Int32)})\nusers = pl.concat([tr_ev[\'user_id\'], ev_ev[\'user_id\']]).unique().sort()\nusers = pl.DataFrame({\'user_id\': users, \'u\': pl.arange(0, len(users), eager=True).cast(pl.Int32)})\nevents = pl.concat([tr_ev[\'event_id\'], ev_ev[\'event_id\']]).unique().sort()\nevents = pl.DataFrame({\'event_id\': events, \'eid\': pl.arange(0, len(events), eager=True).cast(pl.Int32)})\nlog(\'items\', len(items), \'users\', len(users), \'events\', len(events))\nitems.write_parquet(f\'{OUT}/items.parquet\'); users.write_parquet(f\'{OUT}/users.parquet\')\n\ndef map_rows(rows, lab):\n    r = rows.with_row_index(\'ord\').join(events, on=\'event_id\', how=\'left\').join(items, on=\'item_id\', how=\'left\')\n    sel = [\'ord\', \'rowid\', \'eid\', \'ii\'] + ([\'clicked\'] if lab else [])\n    r = r.sort(\'ord\').select(sel)\n    if lab: r = r.with_columns(pl.col(\'clicked\').cast(pl.Int8))\n    return r\n\ndef map_ev(ev):\n    e = ev.join(events, on=\'event_id\', how=\'left\').join(users, on=\'user_id\', how=\'left\')\n    ex = e.select(\'eid\', \'h_str\').with_columns(pl.int_ranges(0, pl.col(\'h_str\').list.len()).alias(\'k\')).explode([\'h_str\', \'k\']).drop_nulls(\'h_str\')\n    ex = ex.join(items.rename({\'item_id\': \'h_str\'}), on=\'h_str\', how=\'left\').sort(\'eid\', \'k\')\n    h = ex.group_by(\'eid\', maintain_order=True).agg(pl.col(\'ii\').alias(\'h\'))\n    e = e.join(h, on=\'eid\', how=\'left\').with_columns(pl.col(\'h\').fill_null(pl.lit([], dtype=pl.List(pl.Int32))))\n    return e.select(\'eid\', \'u\', pl.col(\'timestamp\').cast(pl.Int64).alias(\'ts\'), pl.col(\'time_bucket\').cast(pl.Int8).alias(\'tb\'),\n                    pl.col(\'day_type\').cast(pl.Int8).alias(\'dt\'), \'h\', \'h_ts\').sort(\'ts\', \'eid\')\n\nmap_rows(tr_rows, True).drop(\'ord\').write_parquet(f\'{OUT}/rows_train.parquet\')\nmap_rows(ev_rows, False).write_parquet(f\'{OUT}/rows_eval.parquet\')\nmap_ev(tr_ev).write_parquet(f\'{OUT}/ev_train.parquet\')\nmap_ev(ev_ev).write_parquet(f\'{OUT}/ev_eval.parquet\')\nlog(\'done\')\n', 'v3/features.py': '"""Feature builder (as-of-cutoff, leakage-safe) for TechArena CTR.\n\nbuild(C, T_ev, T_rows, TE, TR) returns one feature row per target candidate.\nEvery statistic uses ONLY:\n  * train events/rows with ts < C (fit window: impressions + labels), and\n  * click histories of events with ts < C, plus the target event\'s OWN history (strictly before its ts).\nSince target events are each user\'s first event at/after C, their histories contain only clicks < C,\nso nothing observed after a target row\'s own event time is ever joined.\n\nFeature groups (column prefixes, used for ablation):\n  hist_  candidate vs the user\'s click log / own history / past impressions of that user-item pair\n  it_    item popularity & CTR (full fit window, last 24h, last 6h, trend)\n  co_    item-item co-click (user co-occurrence cosine) and sequential transitions vs user\'s last 5/20 clicks\n  us_    user activity / CTR / recency / cold-start flags\n  ev_    within-event rank / z-score / event max & mean of the main features\n  tm_    time_bucket, day_type, hour, staleness since cutoff\n  sv_    truncated SVD (tf-idf, row-normalised user x item click matrix as of C): user/history/item cosines\n  mf_    logistic matrix factorisation score fitted on fit-window impressions\n"""\nimport time\nimport numpy as np\nimport polars as pl\nfrom scipy import sparse\nfrom sklearn.utils.extmath import randomized_svd\n\nJ20 = 1737331200\nDAY = 86400\nG_PRIOR = 0.0629\nK_ITEM = 20          # pseudo-impressions for item CTR shrinkage\nK_USER = 5           # pseudo-events for user CTR shrinkage\nTOPK = 200           # co-click neighbours kept per item\nID_COLS = [\'rowid\', \'eid\', \'u\', \'ii\', \'ts\', \'clicked\', \'ord\']\n_t0 = time.time()\n\ndef log(*a):\n    print(time.strftime(\'%H:%M:%S\'), *a, flush=True)\n\ndef load(D):\n    TE = pl.read_parquet(f\'{D}/ev_train.parquet\')\n    TR = pl.read_parquet(f\'{D}/rows_train.parquet\')\n    EE = pl.read_parquet(f\'{D}/ev_eval.parquet\')\n    ER = pl.read_parquet(f\'{D}/rows_eval.parquet\')\n    NI = pl.read_parquet(f\'{D}/items.parquet\').height\n    NU = pl.read_parquet(f\'{D}/users.parquet\').height\n    return TE, TR, EE, ER, NI, NU\n\ndef _hist_long(E):\n    return (E.select(\'u\', \'h\', \'h_ts\').explode([\'h\', \'h_ts\']).drop_nulls(\'h\')\n             .select(\'u\', pl.col(\'h\').cast(pl.Int32).alias(\'ii\'), pl.col(\'h_ts\').cast(pl.Int64).alias(\'ts\')))\n\ndef click_log(C, P_ev, P_rows, T_ev):\n    """Exact (user,item,ts) click log as of C: fit-window clicks + warm-up clicks seen in pre-C / own histories."""\n    win = P_rows.filter(pl.col(\'clicked\') == 1).select(\'u\', pl.col(\'ii\').cast(pl.Int32), \'ts\')\n    hist = pl.concat([_hist_long(P_ev), _hist_long(T_ev)]).filter(pl.col(\'ts\') < C)\n    CL = pl.concat([win, hist]).unique().sort(\'u\', \'ts\', \'ii\')\n    return CL\n\ndef cooc_tables(CL, NI):\n    """User co-occurrence cosine (top-K per item) + lag-weighted transition probabilities prev->next."""\n    us = CL.select(\'u\', \'ii\').unique()\n    pop = us.group_by(\'ii\').agg(pl.len().alias(\'pop\'))\n    pairs = (us.join(us, on=\'u\', suffix=\'_b\').filter(pl.col(\'ii\') != pl.col(\'ii_b\'))\n               .group_by(\'ii\', \'ii_b\').agg(pl.len().cast(pl.Int32).alias(\'co\')))\n    co = (pairs.join(pop, on=\'ii\').join(pop.rename({\'ii\': \'ii_b\', \'pop\': \'pop_b\'}), on=\'ii_b\')\n               .with_columns((pl.col(\'co\') / (pl.col(\'pop\') * pl.col(\'pop_b\')).sqrt()).cast(pl.Float32).alias(\'cos\'))\n               .select(pl.col(\'ii_b\').alias(\'j\'), \'ii\', \'cos\')       # j = history item, ii = candidate\n               .sort(\'cos\', descending=True).group_by(\'j\').head(TOPK))\n    seq = CL.select(\'u\', \'ii\')  # already sorted by u, ts\n    parts = []\n    for lag in (1, 2, 3):\n        parts.append(seq.with_columns(pl.col(\'ii\').shift(lag).over(\'u\').alias(\'j\')).drop_nulls(\'j\')\n                        .select(pl.col(\'j\').cast(pl.Int32), \'ii\', pl.lit(1.0 / lag).alias(\'w\')))\n    trn = pl.concat(parts).filter(pl.col(\'j\') != pl.col(\'ii\')).group_by(\'j\', \'ii\').agg(pl.col(\'w\').sum())\n    tot = trn.group_by(\'j\').agg(pl.col(\'w\').sum().alias(\'tot\'))\n    trn = (trn.join(tot, on=\'j\').with_columns((pl.col(\'w\') / (pl.col(\'tot\') + 5.0)).cast(pl.Float32).alias(\'trp\'))\n              .select(\'j\', \'ii\', \'trp\').sort(\'trp\', descending=True).group_by(\'j\').head(TOPK))\n    return co, trn\n\ndef fit_mf(P_rows, NU, NI, dim=24, epochs=6, lr=0.05, l2=1e-4, seed=0, batch=200_000):\n    """Logistic MF  logit = b + bu + bi + <pu, qi>  fitted with Adagrad on fit-window impressions (numpy)."""\n    rng = np.random.default_rng(seed)\n    u = P_rows[\'u\'].to_numpy().astype(np.int64); i = P_rows[\'ii\'].to_numpy().astype(np.int64)\n    y = P_rows[\'clicked\'].to_numpy().astype(np.float32)\n    P = (rng.standard_normal((NU, dim)) * 0.01).astype(np.float32); Q = (rng.standard_normal((NI, dim)) * 0.01).astype(np.float32)\n    bu = np.zeros(NU, np.float32); bi = np.zeros(NI, np.float32)\n    b = np.float32(np.log(y.mean() / (1 - y.mean())))\n    gP = np.full_like(P, 1e-6); gQ = np.full_like(Q, 1e-6); gbu = np.full_like(bu, 1e-6); gbi = np.full_like(bi, 1e-6)\n    n = len(y)\n    for ep in range(epochs):\n        perm = rng.permutation(n)\n        for s in range(0, n, batch):\n            idx = perm[s:s + batch]; ub, ib, yb = u[idx], i[idx], y[idx]\n            pu, qi = P[ub], Q[ib]\n            z = b + bu[ub] + bi[ib] + (pu * qi).sum(1)\n            g = (1 / (1 + np.exp(-z)) - yb).astype(np.float32)          # dL/dz\n            m = len(idx)\n            Su = sparse.csr_matrix((np.ones(m, np.float32), (ub, np.arange(m))), shape=(NU, m))\n            Si = sparse.csr_matrix((np.ones(m, np.float32), (ib, np.arange(m))), shape=(NI, m))\n            dP = Su @ (g[:, None] * qi) + l2 * P * (np.asarray(Su.sum(1)).ravel() > 0)[:, None]\n            dQ = Si @ (g[:, None] * pu) + l2 * Q * (np.asarray(Si.sum(1)).ravel() > 0)[:, None]\n            dbu = Su @ g; dbi = Si @ g\n            gP += dP * dP; gQ += dQ * dQ; gbu += dbu * dbu; gbi += dbi * dbi\n            P -= lr * dP / np.sqrt(gP); Q -= lr * dQ / np.sqrt(gQ)\n            bu -= lr * dbu / np.sqrt(gbu); bi -= lr * dbi / np.sqrt(gbi)\n    return dict(b=b, bu=bu, bi=bi, P=P, Q=Q)\n\ndef build(C, T_ev, T_rows, TE, TR, NI, NU, with_mf=True, sv_mode=\'std\'):\n    t0 = time.time()\n    P_ev = TE.filter(pl.col(\'ts\') < C)\n    P_rows = TR.join(P_ev.select(\'eid\', \'u\', \'ts\'), on=\'eid\', how=\'inner\')\n    g = float(P_rows[\'clicked\'].mean())\n    ndays = (C - J20) / DAY\n    CL = click_log(C, P_ev, P_rows, T_ev)\n\n    # ---------------- item ----------------\n    def ictr(c, n):\n        return (pl.col(c) + K_ITEM * g) / (pl.col(n) + K_ITEM)\n    it = P_rows.group_by(\'ii\').agg(\n        pl.len().alias(\'it_imp\'), pl.col(\'clicked\').sum().cast(pl.Int32).alias(\'it_clk\'),\n        (pl.col(\'ts\') >= C - DAY).sum().alias(\'it_imp24\'), pl.col(\'clicked\').filter(pl.col(\'ts\') >= C - DAY).sum().cast(pl.Int32).alias(\'it_clk24\'),\n        (pl.col(\'ts\') >= C - 6 * 3600).sum().alias(\'it_imp6\'), pl.col(\'clicked\').filter(pl.col(\'ts\') >= C - 6 * 3600).sum().cast(pl.Int32).alias(\'it_clk6\'))\n    it = it.with_columns(ictr(\'it_clk\', \'it_imp\').alias(\'it_ctr\'), ictr(\'it_clk24\', \'it_imp24\').alias(\'it_ctr24\'),\n                         ictr(\'it_clk6\', \'it_imp6\').alias(\'it_ctr6\'))\n    it = it.with_columns((pl.col(\'it_ctr24\') / pl.col(\'it_ctr\')).alias(\'it_trend24\'), (pl.col(\'it_ctr6\') / pl.col(\'it_ctr\')).alias(\'it_trend6\'),\n                         (pl.col(\'it_imp24\') / (pl.col(\'it_imp\') / ndays)).alias(\'it_imptrend24\'),\n                         (pl.col(\'it_imp\') / ndays).alias(\'it_imp_rate\'))\n    ip = CL.group_by(\'ii\').agg(pl.len().alias(\'it_nclk_all\'), pl.col(\'u\').n_unique().alias(\'it_nusers\'),\n                               (pl.col(\'ts\') < J20).sum().alias(\'it_nclk_warm\'),\n                               ((pl.col(\'ts\') - C) / (2 * DAY)).exp().sum().cast(pl.Float32).alias(\'it_clk_dec2d\'))\n\n    # ---------------- user ----------------\n    ue = P_rows.group_by(\'u\').agg(pl.col(\'eid\').n_unique().alias(\'us_nev\'), pl.col(\'clicked\').sum().cast(pl.Int32).alias(\'us_nclk_win\'),\n                                  pl.col(\'ts\').max().alias(\'_u_last_ev\'))\n    ue = ue.with_columns(((pl.col(\'us_nclk_win\') + K_USER * 0.629) / (pl.col(\'us_nev\') + K_USER)).alias(\'us_ctr\'))\n    uc = CL.group_by(\'u\').agg(pl.len().alias(\'us_nclk_all\'), (pl.col(\'ts\') < J20).sum().alias(\'us_nclk_warm\'),\n                              pl.col(\'ii\').n_unique().alias(\'us_nuniq\'), pl.col(\'ts\').max().alias(\'_u_last_clk\'),\n                              (pl.col(\'ts\') >= C - DAY).sum().alias(\'us_nclk24\'))\n    uc = uc.with_columns((1 - pl.col(\'us_nuniq\') / pl.col(\'us_nclk_all\')).alias(\'us_repeat_rate\'))\n\n    # ---------------- user x item ----------------\n    uic = CL.group_by(\'u\', \'ii\').agg(pl.len().alias(\'hist_n_all\'), (pl.col(\'ts\') < J20).sum().alias(\'hist_n_warm\'),\n                                     (pl.col(\'ts\') >= J20).sum().alias(\'hist_n_win\'), pl.col(\'ts\').max().alias(\'_ui_last_clk\'))\n    uii = P_rows.group_by(\'u\', \'ii\').agg(pl.len().alias(\'hist_imp\'), pl.col(\'clicked\').sum().cast(pl.Int32).alias(\'_ui_clk\'),\n                                         pl.col(\'ts\').max().alias(\'_ui_last_imp\'))\n    uii = uii.with_columns((pl.col(\'hist_imp\') - pl.col(\'_ui_clk\')).alias(\'hist_noclk\'))\n\n    # ---------------- assemble ----------------\n    R = (T_rows.select([c for c in (\'rowid\', \'eid\', \'ii\', \'clicked\', \'ord\') if c in T_rows.columns])\n         .join(T_ev.select(\'eid\', \'u\', \'ts\', \'tb\', \'dt\', \'h\', \'h_ts\'), on=\'eid\', how=\'left\').with_row_index(\'r\'))\n    R = (R.join(it, on=\'ii\', how=\'left\').join(ip, on=\'ii\', how=\'left\').join(ue, on=\'u\', how=\'left\').join(uc, on=\'u\', how=\'left\')\n          .join(uic, on=[\'u\', \'ii\'], how=\'left\').join(uii, on=[\'u\', \'ii\'], how=\'left\'))\n    R = R.with_columns(\n        pl.col(\'h\').list.len().alias(\'us_hlen\'),\n        (pl.col(\'h\').list.len() == 0).cast(pl.Int8).alias(\'us_nohist\'),\n        pl.col(\'us_nev\').is_null().cast(pl.Int8).alias(\'us_cold\'),\n        (pl.col(\'ts\') - pl.col(\'_u_last_ev\')).alias(\'us_since_ev\'),\n        (pl.col(\'ts\') - pl.col(\'_u_last_clk\')).alias(\'us_since_clk\'),\n        (pl.col(\'ts\') - pl.col(\'_ui_last_clk\')).alias(\'hist_since_clk\'),\n        (pl.col(\'ts\') - pl.col(\'_ui_last_imp\')).alias(\'hist_since_imp\'),\n        (pl.col(\'ts\') - C).alias(\'tm_stale\'), ((pl.col(\'ts\') // 3600) % 24).cast(pl.Int8).alias(\'tm_hour\'),\n        pl.col(\'tb\').alias(\'tm_tb\'), pl.col(\'dt\').alias(\'tm_dt\'))\n    # position (from end, 1 = most recent) of candidate in own history\n    hx = (R.select(\'r\', \'ii\', \'h\').with_columns(pl.int_ranges(pl.col(\'h\').list.len(), 0, -1).alias(\'pos\'))\n            .explode([\'h\', \'pos\']).drop_nulls(\'h\').filter(pl.col(\'h\') == pl.col(\'ii\'))\n            .group_by(\'r\').agg(pl.col(\'pos\').min().alias(\'hist_pos\'), pl.len().alias(\'hist_n_own\')))\n    R = R.join(hx, on=\'r\', how=\'left\')\n    R = R.with_columns(pl.col(\'hist_since_clk\').fill_null(30 * DAY), pl.col(\'hist_since_imp\').fill_null(30 * DAY),\n                       pl.col(\'hist_pos\').fill_null(99), pl.col(\'hist_n_own\').fill_null(0), pl.col(\'us_since_ev\').fill_null(30 * DAY), pl.col(\'us_since_clk\').fill_null(30 * DAY))\n    R = R.with_columns((pl.col(\'hist_n_all\').fill_null(0) > 0).cast(pl.Int8).alias(\'hist_in\'))\n\n    # ---------------- co-click ----------------\n    co, trn = cooc_tables(CL, NI)\n    last20 = CL.group_by(\'u\', maintain_order=True).agg(pl.col(\'ii\').tail(20).alias(\'l20\'))\n    T_u = R.select(\'r\', \'u\', \'ii\').join(last20, on=\'u\', how=\'left\')\n    X = (T_u.with_columns(pl.int_ranges(pl.col(\'l20\').list.len(), 0, -1).alias(\'pos\')).explode([\'l20\', \'pos\'])\n            .drop_nulls(\'l20\').rename({\'l20\': \'j\'}).with_columns(pl.col(\'j\').cast(pl.Int32))\n            .join(co, on=[\'j\', \'ii\'], how=\'left\').join(trn, on=[\'j\', \'ii\'], how=\'left\')\n            .with_columns(pl.col(\'cos\').fill_null(0), pl.col(\'trp\').fill_null(0)))\n    l5 = pl.col(\'pos\') <= 5\n    agg = X.group_by(\'r\').agg(\n        pl.col(\'cos\').max().alias(\'co_cos_max20\'), pl.col(\'cos\').mean().alias(\'co_cos_mean20\'),\n        pl.col(\'cos\').filter(l5).max().alias(\'co_cos_max5\'), pl.col(\'cos\').filter(l5).mean().alias(\'co_cos_mean5\'),\n        pl.col(\'cos\').filter(pl.col(\'pos\') == 1).first().alias(\'co_cos_last\'),\n        pl.col(\'trp\').max().alias(\'co_tr_max20\'), pl.col(\'trp\').mean().alias(\'co_tr_mean20\'),\n        pl.col(\'trp\').filter(l5).max().alias(\'co_tr_max5\'), pl.col(\'trp\').filter(l5).mean().alias(\'co_tr_mean5\'),\n        pl.col(\'trp\').filter(pl.col(\'pos\') == 1).first().alias(\'co_tr_last\'),\n        (pl.col(\'cos\') / pl.col(\'pos\').cast(pl.Float32).sqrt()).sum().alias(\'co_cos_rw\'), pl.col(\'cos\').sum().alias(\'co_cos_sum20\'))\n    R = R.join(agg, on=\'r\', how=\'left\').sort(\'r\')\n\n    # ---------------- SVD embeddings ----------------\n    # sv_mode (leakage checks only): \'std\' = click log as of C; \'notgt\' = drop target events\' own histories;\n    # \'lag1d\' = only clicks/impressions with ts < C - 1 day\n    if sv_mode == \'std\':\n        CLsv, NEGsrc = CL, P_rows\n    elif sv_mode == \'notgt\':\n        CLsv = pl.concat([P_rows.filter(pl.col(\'clicked\') == 1).select(\'u\', pl.col(\'ii\').cast(pl.Int32), \'ts\'),\n                          _hist_long(P_ev).filter(pl.col(\'ts\') < C)]).unique()\n        NEGsrc = P_rows\n    elif sv_mode == \'lag1d\':\n        CLsv, NEGsrc = CL.filter(pl.col(\'ts\') < C - DAY), P_rows.filter(pl.col(\'ts\') < C - DAY)\n    else:\n        raise ValueError(sv_mode)\n    assert (CLsv[\'ts\'].max() or 0) < C and (NEGsrc[\'ts\'].max() or 0) < C\n    cnt = CLsv.group_by(\'u\', \'ii\').agg(pl.len().alias(\'n\'))\n    Xc = sparse.csr_matrix((np.ones(cnt.height, np.float32), (cnt[\'u\'].to_numpy(), cnt[\'ii\'].to_numpy())), shape=(NU, NI))\n    df_ = np.asarray(Xc.sum(0)).ravel(); idf = np.log((NU + 1) / (df_ + 1)).astype(np.float32)\n    Xc = Xc @ sparse.diags(idf); rn = np.sqrt(np.asarray(Xc.multiply(Xc).sum(1)).ravel()) + 1e-6; Xc = sparse.diags(1 / rn) @ Xc\n    uu = R[\'u\'].to_numpy().astype(np.int64); iic = R[\'ii\'].to_numpy().astype(np.int64)\n    ex = X.select(\'r\', \'j\', \'pos\').unique([\'r\', \'j\', \'pos\'])\n    # per-event history (last 20 clicks) is shared by all candidates: take rows of one candidate per (r) anyway\n    er_r = ex[\'r\'].to_numpy(); er_j = ex[\'j\'].to_numpy().astype(np.int64); er_w = (1.0 / np.sqrt(ex[\'pos\'].to_numpy())).astype(np.float32)\n    er_l5 = ex[\'pos\'].to_numpy() <= 5\n    nR = R.height\n    svcols = {}\n    for k in (16, 32, 64):\n        U, S, Vt = randomized_svd(Xc, k, random_state=0, n_iter=5)\n        V = (Vt.T * np.sqrt(S)).astype(np.float32); Uf = (U * np.sqrt(S)).astype(np.float32)\n        Vn = V / (np.linalg.norm(V, axis=1, keepdims=True) + 1e-9); Un = Uf / (np.linalg.norm(Uf, axis=1, keepdims=True) + 1e-9)\n        cand = Vn[iic]\n        svcols[f\'sv_k{k}_ucos\'] = (Un[uu] * cand).sum(1)\n        svcols[f\'sv_k{k}_udot\'] = (Uf[uu] * V[iic]).sum(1)\n        hv = np.zeros((nR, k), np.float32); np.add.at(hv, er_r, Vn[er_j] * er_w[:, None])\n        hn = np.linalg.norm(hv, axis=1); svcols[f\'sv_k{k}_hcos\'] = (hv * cand).sum(1) / (hn + 1e-9)\n        if k == 32:\n            svcols[\'sv_k32_hnorm\'] = hn\n            sims = (Vn[er_j] * Vn[iic[er_r]]).sum(1)\n            mx = np.full(nR, -1.0, np.float32); np.maximum.at(mx, er_r, sims)\n            mx5 = np.full(nR, -1.0, np.float32); np.maximum.at(mx5, er_r[er_l5], sims[er_l5])\n            svcols[\'sv_k32_imax20\'] = mx; svcols[\'sv_k32_imax5\'] = mx5\n        if k == 64:\n            # click centroid vs non-clicked-impression centroid (impressions frozen at C: P_rows only)\n            Sc = sparse.csr_matrix((np.ones(cnt.height, np.float32), (cnt[\'u\'].to_numpy(), cnt[\'ii\'].to_numpy())), shape=(NU, NI))\n            neg = NEGsrc.filter(pl.col(\'clicked\') == 0)\n            Sn = sparse.csr_matrix((np.ones(neg.height, np.float32), (neg[\'u\'].to_numpy(), neg[\'ii\'].to_numpy())), shape=(NU, NI))\n            nC = np.asarray(Sc @ Vn); nC /= np.linalg.norm(nC, axis=1, keepdims=True) + 1e-9\n            nN = np.asarray(Sn @ Vn); nN /= np.linalg.norm(nN, axis=1, keepdims=True) + 1e-9\n            cc = (nC[uu] * cand).sum(1); cn = (nN[uu] * cand).sum(1)\n            svcols[\'sv_k64_ccos\'] = cc; svcols[\'sv_k64_ncos\'] = cn\n            for a, nm in ((0.25, \'025\'), (0.5, \'05\'), (1.0, \'1\')):\n                M = nC - a * nN; M /= np.linalg.norm(M, axis=1, keepdims=True) + 1e-9\n                svcols[f\'sv_k64_con{nm}\'] = (M[uu] * cand).sum(1)\n    R = R.with_columns([pl.Series(c, v.astype(np.float32)) for c, v in svcols.items()])\n\n    # ---------------- MF ----------------\n    if with_mf:\n        mf = fit_mf(P_rows, NU, NI)\n        uu = R[\'u\'].to_numpy().astype(np.int64); ii = R[\'ii\'].to_numpy().astype(np.int64)\n        dot = (mf[\'P\'][uu] * mf[\'Q\'][ii]).sum(1)\n        R = R.with_columns(pl.Series(\'mf_score\', mf[\'b\'] + mf[\'bu\'][uu] + mf[\'bi\'][ii] + dot),\n                           pl.Series(\'mf_dot\', dot), pl.Series(\'mf_bi\', mf[\'bi\'][ii]))\n\n    fill0 = [c for c in R.columns if c.startswith((\'it_imp\', \'it_clk\', \'it_n\', \'us_n\', \'hist_n\', \'hist_imp\', \'hist_noclk\', \'co_\')) and R[c].dtype != pl.List]\n    R = R.with_columns([pl.col(c).fill_null(0) for c in fill0])\n    R = R.with_columns([pl.col(c).fill_null(g) for c in (\'it_ctr\', \'it_ctr24\', \'it_ctr6\')] +\n                       [pl.col(c).fill_null(1.0) for c in (\'it_trend24\', \'it_trend6\', \'it_imptrend24\')] +\n                       [pl.col(\'us_ctr\').fill_null(g), pl.col(\'it_imp_rate\').fill_null(0), pl.col(\'us_repeat_rate\').fill_null(0)])\n\n    # ---------------- event context ----------------\n    main = [\'it_ctr\', \'it_ctr24\', \'it_imp\', \'it_nclk_all\', \'it_clk_dec2d\', \'hist_n_all\', \'hist_noclk\',\n            \'co_cos_max20\', \'co_cos_rw\', \'co_tr_max5\', \'co_tr_last\',\n            \'sv_k16_ucos\', \'sv_k32_ucos\', \'sv_k64_ucos\', \'sv_k32_hcos\', \'sv_k64_hcos\', \'sv_k32_imax20\', \'sv_k64_con05\', \'sv_k64_con025\', \'sv_k64_ccos\'] + ([\'mf_score\', \'mf_dot\'] if with_mf else [])\n    exprs = []\n    for c in main:\n        x = pl.col(c).cast(pl.Float64)\n        exprs += [x.rank(\'average\').over(\'eid\').alias(f\'ev_{c}_rk\'),\n                  ((x - x.mean().over(\'eid\')) / (x.std().over(\'eid\') + 1e-9)).alias(f\'ev_{c}_z\'),\n                  x.max().over(\'eid\').alias(f\'ev_{c}_max\'), x.mean().over(\'eid\').alias(f\'ev_{c}_mean\')]\n    R = R.with_columns(exprs)\n    R = R.drop([c for c in R.columns if c.startswith(\'_\')] + [\'h\', \'h_ts\', \'tb\', \'dt\']).sort(\'r\').drop(\'r\')\n    log(f\'build C={C} rows={R.height} cols={R.width} {time.time()-t0:.0f}s\')\n    return R\n\nGROUPS = [\'it\', \'hist\', \'us\', \'co\', \'ev\', \'tm\', \'sv\', \'mf\']\n\ndef feature_cols(df, groups):\n    cols = []\n    for c in df.columns:\n        if c in ID_COLS: continue\n        gname = c.split(\'_\')[0]\n        if gname == \'ev\':\n            src = c[3:].split(\'_\')[0]  # ev_<src group>_...\n            if \'ev\' in groups and src in groups: cols.append(c)\n        elif gname in groups:\n            cols.append(c)\n    return cols\n', 'v3/split.py': '"""Time-based splits that mirror the official evaluation.\n\nEvaluation = each user\'s FIRST event at/after the cutoff. So for a cutoff C:\n  fit events  = all train events with ts < C            (labels usable for aggregates)\n  target/val  = for each user, the first train event with C <= ts < C+W (W=None -> until end of train)\nWhole events always stay together (split is by event, rows follow their event).\n"""\nimport polars as pl\n\nJ20 = 1737331200          # 2025-01-20 00:00 UTC (start of train window; earlier clicks = warm-up)\nDAY = 86400\nC_MAIN = J20 + 5 * DAY    # 2025-01-25\nC_SECOND = J20 + 6 * DAY  # 2025-01-26\nC_FINAL = J20 + 7 * DAY   # 2025-01-27 (official cutoff)\n\ndef fit_events(TE: pl.DataFrame, C: int) -> pl.DataFrame:\n    return TE.filter(pl.col(\'ts\') < C)\n\ndef first_events(TE: pl.DataFrame, C: int, W: int | None = None) -> pl.DataFrame:\n    """First event per user in [C, C+W). Ties on ts broken by eid (deterministic)."""\n    m = pl.col(\'ts\') >= C\n    if W is not None:\n        m = m & (pl.col(\'ts\') < C + W)\n    return TE.filter(m).sort(\'ts\', \'eid\').group_by(\'u\', maintain_order=True).first()\n\ndef split(TE: pl.DataFrame, TR: pl.DataFrame, C: int, W: int | None = None):\n    """Return (fit_ev, fit_rows, val_ev, val_rows)."""\n    fe = fit_events(TE, C)\n    ve = first_events(TE, C, W)\n    fr = TR.join(fe.select(\'eid\'), on=\'eid\', how=\'semi\')\n    vr = TR.join(ve.select(\'eid\'), on=\'eid\', how=\'semi\')\n    return fe, fr, ve, vr\n\ndef train_fold_cutoffs(C_val: int):\n    """Daily cutoffs strictly before C_val used to build model-training rows (each fold: first event/user in [Ck, Ck+1d))."""\n    return [J20 + k * DAY for k in range(1, (C_val - J20) // DAY)]\n'}

STAGES = ['setup', 'prep_v3', 'prep_v2', 'feat_v3', 'feat_v2', 'feat_x', 'feat_x2', 'feat_lr', 'base', 'slow', 'blend', 'write', 'validate']
FOLDS = 'f1 f2 f3 f4 f5 f6 eval'
SLOW = '{"learning_rate":0.03,"feature_fraction":0.55,"num_leaves":95,"lambda_l2":30}'
def main():
    ap = argparse.ArgumentParser(); ap.add_argument('--raw', required=True); ap.add_argument('--work', default='./v6_work')
    ap.add_argument('--threads', type=int, default=4); ap.add_argument('--out', default=None); ap.add_argument('--from', dest='frm', default='setup'); ap.add_argument('--to', default='validate')
    a = ap.parse_args()
    W = os.path.abspath(a.work); RAW = os.path.abspath(a.raw); T = str(a.threads)
    OUT = os.path.abspath(a.out or os.path.join(W, 'submission_v6.csv'))
    env = dict(os.environ, POLARS_MAX_THREADS=T, OMP_NUM_THREADS=T, OPENBLAS_NUM_THREADS=T, MKL_NUM_THREADS=T, NUMEXPR_NUM_THREADS=T, PYTHONUNBUFFERED='1')
    PY = sys.executable
    def sh(cmd, cwd=W): print('>>', cmd, flush=True); subprocess.run(cmd, shell=True, check=True, cwd=cwd, env=env)
    def setup():
        os.makedirs(os.path.join(W, 'v3'), exist_ok=True); os.makedirs(os.path.join(W, 'ens'), exist_ok=True)
        for k, s in SRC.items():
            open(os.path.join(W, k), 'w').write(s.replace('@@WORK@@', W).replace('@@RAW@@', RAW))
        for f in ('train.csv', 'evaluation.csv'):          # v2 prep / writer read the raw CSVs from WORK
            dst = os.path.join(W, f)
            if not os.path.exists(dst): os.symlink(os.path.join(RAW, f), dst)
    run = {'setup': setup,
           'prep_v3': lambda: sh(f'{PY} prep.py {RAW}', os.path.join(W, 'v3')),
           'prep_v2': lambda: sh(f'{PY} prep.py'),
           'feat_v3': lambda: sh(f'{PY} v3fold.py {FOLDS}'),
           'feat_v2': lambda: sh(f'{PY} build_folds.py lite'),
           'feat_x': lambda: sh(f'{PY} extra.py {FOLDS}'),
           'feat_x2': lambda: sh(f'{PY} extra2.py {FOLDS}'),
           'feat_lr': lambda: sh(f'LR_KS=4,6,8,10,12,16 LR_CON=1 {PY} lrfeat2.py {FOLDS}'),
           'base': lambda: sh(f'{PY} ens.py 27 base --seeds 0,1 --threads {T} --lr feat_lr2'),
           'slow': lambda: sh(f"{PY} ens.py 27 slow --seeds 0 --rounds 500 --threads {T} --lr feat_lr2 --params '{SLOW}'"),
           'blend': lambda: sh(f'{PY} blend_lite.py'),
           'write': lambda: sh(f'PRED={W}/eval_pred.parquet {PY} write_sub.py {OUT}'),
           'validate': lambda: sh(f'{PY} validate.py {OUT}')}
    tf = os.path.join(W, 'timings.json'); tim = json.load(open(tf)) if os.path.exists(tf) and a.frm != 'setup' else {}
    t_all = time.time()
    for st in STAGES[STAGES.index(a.frm):STAGES.index(a.to) + 1]:
        t0 = time.time(); run[st](); tim[st] = round(time.time() - t0, 1)
        print(f'[stage {st}] {tim[st]:.0f}s', flush=True)
        os.makedirs(W, exist_ok=True); json.dump(tim, open(tf, 'w'), indent=1)
    print('TOTAL this run %.1f min; all stages %.1f min' % ((time.time() - t_all) / 60, sum(tim.values()) / 60))
    print('submission:', OUT)
if __name__ == '__main__':
    main()