Agent trace

current_model:
total 104
dr-xr-xr-x 1 root root  4096 Sep 26 07:10 .
dr-xr-xr-x 1 root root  4096 Sep 26 07:12 ..
-r--r--r-- 1 root root 80058 Sep 26 07:10 train_current_model.py
-r--r--r-- 1 root root 15985 Sep 26 07:10 two_tower_model.py

data:
total 347392
dr-xr-xr-x 1 root root      4096 Sep 26 07:11 .
dr-xr-xr-x 1 root root      4096 Sep 26 07:12 ..
-r--r--r-- 1 root root    349352 Sep 26 07:10 candidate_catalog.parquet
-r--r--r-- 1 root root     84127 Sep 26 07:10 target_users.parquet
-r--r--r-- 1 root root   7736341 Sep 26 07:10 test_eligible_candidates.npz
-r--r--r-- 1 root root   1092329 Sep 26 07:11 test_preference_state.parquet
-r--r--r-- 1 root root    857213 Sep 26 07:10 train_dislikes.parquet
-r--r--r-- 1 root root   5374652 Sep 26 07:10 train_likes.parquet
-r--r--r-- 1 root root 319183255 Sep 26 07:10 train_listens.parquet
-r--r--r-- 1 root root    155988 Sep 26 07:10 train_undislikes.parquet
-r--r--r-- 1 root root   1855885 Sep 26 07:10 train_unlikes.parquet
-r--r--r-- 1 root root     36571 Sep 26 07:10 val_dislikes.parquet
-r--r--r-- 1 root root    298484 Sep 26 07:10 val_likes.parquet
-r--r--r-- 1 root root   9947609 Sep 26 07:10 val_listens.parquet
-r--r--r-- 1 root root      7845 Sep 26 07:10 val_undislikes.parquet
-r--r--r-- 1 root root     72536 Sep 26 07:10 val_unlikes.parquet
-r--r--r-- 1 root root   7573821 Sep 26 07:10 validation_eligible_candidates.npz
-r--r--r-- 1 root root   1061761 Sep 26 07:11 validation_preference_state.parquet
  2164 current_model/train_current_model.py
   433 current_model/two_tower_model.py
   157 evaluate_public.py
  2754 total
8
               total        used        free      shared  buff/cache   available
Mem:              15           1           0           0          13          13
Swap:              0           0           0
Shell cwd was reset to /app

started

1	"""Shared public-data primitives for the Yambda two-tower baseline."""
2	from __future__ import annotations
3	
4	import sys
5	from dataclasses import dataclass
6	from pathlib import Path
7	from typing import Iterator
8	
9	import numpy as np
10	import pandas as pd
11	import pyarrow.parquet as pq
12	import torch
13	from torch import nn
14	
15	
16	DAY = 86_400
17	TRAIN_END = 24_786_800
18	VALIDATION_END = 25_393_400
19	LISTEN_POSITIVE_THRESHOLD = 50
20	PROFILE_LOOKBACK_DAYS = 120
21	LIKE_ITEMS = 48
22	DISLIKE_ITEMS = 24
23	TOP_N = 100
24	
25	
26	@dataclass(frozen=True)
27	class TowerConfig:
28	    sequence_length: int = 100
29	    embedding_dim: int = 192
30	    hidden_dim: int = 192
31	    user_hidden_1: int = 512
32	    user_hidden_2: int = 256
33	    item_hidden_1: int = 512
34	    item_hidden_2: int = 256
35	    dropout: float = 0.05
36	    epochs: int = 1
37	    learning_rate: float = 1.0e-3
38	    weight_decay: float = 1.0e-5
39	    gradient_clip: float = 2.0
40	    uniform_negatives: int = 24
41	    inbatch_negatives: int = 8
42	    inbatch_correction_power: float = 0.5
43	    inbatch_correction_min: float = 0.25
44	    inbatch_correction_max: float = 4.0
45	    seed: int = 29
46	
47	
48	@dataclass(frozen=True)
49	class ListenEvents:
50	    timestamps: np.ndarray
51	    items: np.ndarray
52	    organic: np.ndarray
53	    completion_pct: np.ndarray
54	
55	
56	@dataclass(frozen=True)
57	class ExplicitEvents:
58	    timestamps: np.ndarray
59	    items: np.ndarray
60	    channel: np.ndarray
61	    active: np.ndarray
62	    organic_like: np.ndarray
63	
64	
65	def log(message: str) -> None:
66	    print(message, file=sys.stderr, flush=True)
67	
68	
69	def build_item_features(catalog: pd.DataFrame) -> np.ndarray:
70	    """Return four standardized features built only from the public catalog."""
71	    positives = catalog.positive_listens.to_numpy(dtype=np.float64, copy=False)
72	    organic = catalog.organic_positive_listens.to_numpy(dtype=np.float64, copy=False)
73	    popularity_rank = catalog.popularity_rank.to_numpy(dtype=np.float64, copy=False)
74	    features = np.column_stack(
75	        [
76	            np.log1p(positives),
77	            np.log1p(organic),
78	            organic / np.maximum(positives, 1.0),
79	            1.0 - (popularity_rank - 1.0) / max(len(catalog) - 1, 1),
80	        ]
81	    )
82	    means = features.mean(axis=0, keepdims=True)
83	    scales = np.maximum(features.std(axis=0, keepdims=True), 1.0e-6)
84	    standardized = ((features - means) / scales).astype(np.float32)
85	    if not np.isfinite(standardized).all():
86	        raise RuntimeError("catalog features contain non-finite values")
87	    return standardized
88	
89	
90	def _combine_listen_chunks(parts: list[ListenEvents]) -> ListenEvents:
91	    if len(parts) == 1:
92	        return parts[0]
93	    timestamps = np.concatenate([part.timestamps for part in parts])
94	    items = np.concatenate([part.items for part in parts])
95	    organic = np.concatenate([part.organic for part in parts])
96	    completion = np.concatenate([part.completion_pct for part in parts])
97	    order = np.argsort(timestamps, kind="stable")
98	    return ListenEvents(
99	        timestamps=timestamps[order],
100	        items=items[order],
101	        organic=organic[order],
102	        completion_pct=completion[order],
103	    )
104	
105	
106	def iter_listen_users(
107	    path: Path, item_index: pd.Index
108	) -> Iterator[tuple[int, ListenEvents]]:
109	    """Stream one uid-sorted public listen file one user at a time."""
110	    parquet = pq.ParquetFile(path)
111	    pending_uid: int | None = None
112	    pending: list[ListenEvents] = []
113	    previous_uid = -1
114	    for batch in parquet.iter_batches(
115	        batch_size=750_000,
116	        columns=["uid", "timestamp", "item_id", "is_organic", "played_ratio_pct"],
117	    ):
118	        frame = batch.to_pandas()
119	        mapped = item_index.get_indexer(frame.item_id.to_numpy(dtype=np.int64))
120	        keep = mapped >= 0
121	        if not np.any(keep):
122	            continue
123	        uids = frame.uid.to_numpy(dtype=np.int64, copy=False)[keep]
124	        timestamps = frame.timestamp.to_numpy(dtype=np.int64, copy=False)[keep]
125	        items = mapped[keep].astype(np.int32, copy=False)
126	        organic = frame.is_organic.to_numpy(dtype=np.uint8, copy=False)[keep]
127	        completion = frame.played_ratio_pct.to_numpy(dtype=np.uint16, copy=False)[keep]
128	        if np.any(uids[1:] < uids[:-1]):
129	            raise RuntimeError(f"{path.name} is not sorted by uid")
130	        boundaries = np.r_[
131	            0,
132	            np.flatnonzero(uids[1:] != uids[:-1]) + 1,
133	            len(uids),
134	        ]
135	        for start, end in zip(boundaries[:-1], boundaries[1:]):
136	            uid = int(uids[start])
137	            part = ListenEvents(
138	                timestamps[start:end].copy(),
139	                items[start:end].copy(),
140	                organic[start:end].copy(),
141	                completion[start:end].copy(),
142	            )
143	            if pending_uid is None:
144	                pending_uid = uid
145	            elif uid != pending_uid:
146	                if pending_uid < previous_uid:
147	                    raise RuntimeError(f"{path.name} user order regressed")
148	                yield pending_uid, _combine_listen_chunks(pending)
149	                previous_uid = pending_uid
150	                pending_uid = uid
151	                pending = []
152	            pending.append(part)
153	    if pending_uid is not None:
154	        yield pending_uid, _combine_listen_chunks(pending)
155	
156	
157	def merge_listen_streams(
158	    paths: list[Path], item_index: pd.Index
159	) -> Iterator[tuple[int, ListenEvents]]:
160	    """Merge uid-sorted public listen files without materializing them."""
161	    streams = [iter(iter_listen_users(path, item_index)) for path in paths]
162	    heads: list[tuple[int, ListenEvents] | None] = [
163	        next(stream, None) for stream in streams
164	    ]
165	    while any(head is not None for head in heads):
166	        uid = min(head[0] for head in heads if head is not None)
167	        parts: list[ListenEvents] = []
168	        for position, head in enumerate(heads):
169	            if head is not None and head[0] == uid:
170	                parts.append(head[1])
171	                heads[position] = next(streams[position], None)
172	        yield uid, _combine_listen_chunks(parts)
173	
174	
175	def load_explicit_events(
176	    data: Path,
177	    prefixes: tuple[str, ...],
178	    item_index: pd.Index,
179	) -> dict[int, ExplicitEvents]:
180	    """Load public explicit actions as chronological state changes."""
181	    specifications = (
182	        ("likes", 0, 1, 0),
183	        ("unlikes", 0, 0, 1),
184	        ("dislikes", 1, 1, 0),
185	        ("undislikes", 1, 0, 1),
186	    )
187	    parts: list[pd.DataFrame] = []
188	    for family, channel, active, tie_priority in specifications:
189	        for prefix in prefixes:
190	            frame = pd.read_parquet(
191	                data / f"{prefix}_{family}.parquet",
192	                columns=["uid", "timestamp", "item_id", "is_organic"],
193	            )
194	            mapped = item_index.get_indexer(
195	                frame.item_id.to_numpy(dtype=np.int64, copy=False)
196	            )
197	            keep = mapped >= 0
198	            if not np.any(keep):
199	                continue
200	            retained = frame.loc[keep, ["uid", "timestamp", "is_organic"]].copy()
201	            retained["item"] = mapped[keep].astype(np.int32, copy=False)
202	            retained["channel"] = np.int8(channel)
203	            retained["active"] = np.int8(active)
204	            retained["tie_priority"] = np.int8(tie_priority)
205	            retained["organic_like"] = np.int8(
206	                family == "likes"
207	            ) * retained.is_organic.astype(np.int8)
208	            parts.append(retained.drop(columns="is_organic"))
209	    if not parts:
210	        return {}
211	    events = pd.concat(parts, ignore_index=True)
212	    events = events.sort_values(
213	        ["uid", "timestamp", "tie_priority"], kind="mergesort"
214	    ).reset_index(drop=True)
215	    result: dict[int, ExplicitEvents] = {}
216	    for uid, rows in events.groupby("uid", sort=False):
217	        result[int(uid)] = ExplicitEvents(
218	            timestamps=rows.timestamp.to_numpy(dtype=np.int64, copy=True),
219	            items=rows.item.to_numpy(dtype=np.int32, copy=True),
220	            channel=rows.channel.to_numpy(dtype=np.int8, copy=True),
221	            active=rows.active.to_numpy(dtype=np.int8, copy=True),
222	            organic_like=rows.organic_like.to_numpy(dtype=np.int8, copy=True),
223	        )
224	    log(f"loaded explicit events: rows={len(events):,} users={len(result):,}")
225	    return result
226	
227	
228	def empty_explicit_events() -> ExplicitEvents:
229	    return ExplicitEvents(
230	        timestamps=np.empty(0, dtype=np.int64),
231	        items=np.empty(0, dtype=np.int32),
232	        channel=np.empty(0, dtype=np.int8),
233	        active=np.empty(0, dtype=np.int8),
234	        organic_like=np.empty(0, dtype=np.int8),
235	    )
236	
237	
238	def pack_active_items(
239	    active: dict[int, int], width: int, padding_item: int
240	) -> np.ndarray:
241	    packed = np.full(width, padding_item, dtype=np.uint16)
242	    if not active:
243	        return packed
244	    ordered = sorted(active.items(), key=lambda pair: (-pair[1], pair[0]))[:width]
245	    packed[: len(ordered)] = np.fromiter(
246	        (item for item, _ in ordered), dtype=np.uint16, count=len(ordered)
247	    )
248	    return packed
249	
250	
251	def load_inference_state(
252	    path: Path,
253	    item_index: pd.Index,
254	    uids: np.ndarray,
255	    n_items: int,
256	) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
257	    """Pack the task-provided active explicit state."""
258	    state = pd.read_parquet(path, columns=["uid", "item_id", "state"])
259	    mapped = item_index.get_indexer(state.item_id.to_numpy(dtype=np.int64))
260	    keep = mapped >= 0
261	    state = state.loc[keep, ["uid", "state"]].copy()
262	    state["item"] = mapped[keep].astype(np.int32, copy=False)
263	    user_rows = pd.Index(uids)
264	    liked = np.full((len(uids), LIKE_ITEMS), n_items, dtype=np.uint16)
265	    disliked = np.full((len(uids), DISLIKE_ITEMS), n_items, dtype=np.uint16)
266	    like_counts = np.zeros(len(uids), dtype=np.int32)
267	    dislike_counts = np.zeros(len(uids), dtype=np.int32)
268	    for value, width, destination, counts in (
269	        ("liked", LIKE_ITEMS, liked, like_counts),
270	        ("disliked", DISLIKE_ITEMS, disliked, dislike_counts),
271	    ):
272	        selected = state[state.state == value]
273	        for uid, rows in selected.groupby("uid", sort=False):
274	            row = int(user_rows.get_indexer([int(uid)])[0])
275	            if row < 0:
276	                continue
277	            items = np.sort(rows.item.unique())
278	            counts[row] = len(items)
279	            retained = items[:width]
280	            destination[row, : len(retained)] = retained.astype(np.uint16)
281	    return liked, disliked, like_counts, dislike_counts
282	
283	
284	class SequentialTwoTower(nn.Module):
285	    """Order-sensitive user tower and independent catalog-aware item tower."""
286	
287	    def __init__(self, item_features: np.ndarray, config: TowerConfig) -> None:
288	        super().__init__()
289	        n_items, n_item_features = item_features.shape
290	        dim = config.embedding_dim
291	        self.n_items = n_items
292	        self.context_embedding = nn.Embedding(
293	            n_items + 1, dim, padding_idx=n_items
294	        )
295	        self.event_projection = nn.Sequential(
296	            nn.Linear(dim + 3, config.hidden_dim),
297	            nn.GELU(),
298	        )
299	        self.sequence_encoder = nn.GRU(
300	            input_size=config.hidden_dim,
301	            hidden_size=dim,
302	            num_layers=1,
303	            batch_first=True,
304	        )
305	        self.user_tower = nn.Sequential(
306	            nn.Linear(dim * 3 + 6, config.user_hidden_1),
307	            nn.GELU(),
308	            nn.Dropout(config.dropout),
309	            nn.Linear(config.user_hidden_1, config.user_hidden_2),
310	            nn.GELU(),
311	            nn.Dropout(config.dropout),
312	            nn.Linear(config.user_hidden_2, dim),
313	        )
314	        self.candidate_embedding = nn.Embedding(n_items, dim)
315	        self.register_buffer(
316	            "item_features", torch.from_numpy(item_features), persistent=True
317	        )
318	        self.item_tower = nn.Sequential(
319	            nn.Linear(dim + n_item_features, config.item_hidden_1),
320	            nn.GELU(),
321	            nn.Dropout(config.dropout),
322	            nn.Linear(config.item_hidden_1, config.item_hidden_2),
323	            nn.GELU(),
324	            nn.Dropout(config.dropout),
325	            nn.Linear(config.item_hidden_2, dim),
326	        )
327	        nn.init.normal_(self.context_embedding.weight, std=0.02)
328	        nn.init.normal_(self.candidate_embedding.weight, std=0.02)
329	        with torch.no_grad():
330	            self.context_embedding.weight[n_items].zero_()
331	
332	    def _mean_context_channel(self, items: torch.Tensor) -> torch.Tensor:
333	        mask = items.ne(self.n_items)
334	        vectors = self.context_embedding(items)
335	        return (vectors * mask.unsqueeze(-1)).sum(dim=1) / mask.sum(
336	            dim=1, keepdim=True
337	        ).clamp_min(1)
338	
339	    def encode_items(self, items: torch.Tensor) -> torch.Tensor:
340	        inputs = torch.cat(
341	            [self.candidate_embedding(items), self.item_features[items]], dim=1
342	        )
343	        return self.item_tower(inputs)
344	
345	
346	def invalid_negative_mask(
347	    negatives: torch.Tensor,
348	    targets: torch.Tensor,
349	    context_items: torch.Tensor,
350	    liked_items: torch.Tensor,
351	) -> torch.Tensor:
352	    """Identify targets, known positives, active likes, and row duplicates."""
353	    invalid = negatives.eq(targets[:, None])
354	    invalid |= negatives[:, :, None].eq(context_items[:, None, :]).any(dim=2)
355	    invalid |= negatives[:, :, None].eq(liked_items[:, None, :]).any(dim=2)
356	    for column in range(1, negatives.shape[1]):
357	        invalid[:, column] |= negatives[:, column, None].eq(
358	            negatives[:, :column]
359	        ).any(dim=1)
360	    return invalid
361	
362	
363	def draw_filtered_mixed_negatives(
364	    targets: torch.Tensor,
365	    context_items: torch.Tensor,
366	    liked_items: torch.Tensor,
367	    n_items: int,
368	    uniform_count: int,
369	    inbatch_count: int,
370	    item_correction: torch.Tensor,
371	    generator: torch.Generator,
372	    device: torch.device,
373	) -> tuple[torch.Tensor, torch.Tensor]:
374	    """Draw uniform and unique-target in-batch negatives."""
375	    batch_size = len(targets)
376	    uniform = torch.randint(
377	        n_items,
378	        (batch_size, uniform_count),
379	        generator=generator,
380	        device="cpu",
381	    )
382	    inbatch = torch.empty((batch_size, 0), dtype=torch.long, device="cpu")
383	    corrected_count = 0
384	    if inbatch_count:
385	        unique_targets = torch.unique(targets.detach().cpu())
386	        corrected_count = min(inbatch_count, max(len(unique_targets) - 1, 0))
387	        if corrected_count:
388	            priorities = torch.rand(
389	                (batch_size, len(unique_targets)), generator=generator
390	            )
391	            priorities.masked_fill_(
392	                unique_targets[None, :].eq(targets.detach().cpu()[:, None]),
393	                -1.0,
394	            )
395	            columns = torch.topk(priorities, k=corrected_count, dim=1).indices
396	            inbatch = unique_targets[columns]
397	        if corrected_count < inbatch_count:
398	            filler = torch.randint(
399	                n_items,
400	                (batch_size, inbatch_count - corrected_count),
401	                generator=generator,
402	                device="cpu",
403	            )
404	            inbatch = torch.cat([inbatch, filler], dim=1)
405	
406	    negatives_cpu = torch.cat([uniform, inbatch], dim=1)
407	    correction_cpu = torch.ones_like(negatives_cpu, dtype=torch.float32)
408	    if corrected_count:
409	        correction_cpu[
410	            :, uniform_count : uniform_count + corrected_count
411	        ] = item_correction[inbatch[:, :corrected_count]]
412	    negatives = negatives_cpu.to(device)
413	    correction = correction_cpu.to(device)
414	    for _ in range(32):
415	        invalid = invalid_negative_mask(
416	            negatives, targets, context_items, liked_items
417	        )
418	        if not bool(invalid.any()):
419	            return negatives, correction
420	        replacements = torch.randint(
421	            n_items,
422	            (int(invalid.sum().detach().cpu()),),
423	            generator=generator,
424	            device="cpu",
425	        ).to(device)
426	        negatives[invalid] = replacements
427	        correction[invalid] = 1.0
428	    invalid = invalid_negative_mask(
429	        negatives, targets, context_items, liked_items
430	    )
431	    if bool(invalid.any()):
432	        raise RuntimeError("mixed negative resampling failed false-negative filter")
433	    return negatives, correction
434	

20:import argparse
41:class Config:
76:class CandidateSet:
82:    def read(cls, path: Path) -> "CandidateSet":
99:class SequenceData:
118:    def lengths(self) -> np.ndarray:
122:    def target_count(self) -> int:
127:class TargetPartition:
136:    def selected_count(self) -> int:
140:def log(message: str) -> None:
144:def seed_everything(seed: int) -> None:
150:def model_config(config: Config) -> base.TowerConfig:
174:def append_state_snapshot(
190:def causal_activity_features(
247:def build_sequence_parts(
314:def final_user_features(
341:def prepare_data(
471:def cache_metadata(stage: str, config: Config) -> dict[str, object]:
481:def save_cache(
493:def load_cache(path: Path, stage: str, config: Config) -> SequenceData:
511:def load_or_prepare(
534:def build_item_correction(
578:def build_known_positive_filter(
609:def invalid_full_history_negatives(
630:def draw_full_history_filtered_negatives(
724:def draw_training_negatives(
763:def event_sequence(
774:def pool_explicit_states(
798:def user_queries(
813:def candidate_vectors(
828:def history_windows(
845:def corrected_negative_loss(
867:def retained_target_window_starts(
888:def chronological_target_partition(
914:def load_organic_like_outcomes(
955:def build_target_weights(
1035:def build_dense_block(
1111:def score_target_microbatches(
1198:def release_accelerator_cache(device: torch.device) -> None:
1203:def run_training_epoch(
1364:def evaluate_validation_bce(
1488:def snapshot_model(model: nn.Module) -> dict[str, torch.Tensor]:
1495:def save_epoch_checkpoint(
1521:def train_model(
1782:def inference_queries(
1861:def rank_candidates(
1912:def write_ranking(frame: pd.DataFrame, output: Path) -> None:
1922:def main() -> int:
1923:    parser = argparse.ArgumentParser(description=__doc__)
1924:    parser.add_argument(
1927:    parser.add_argument("--output", type=Path, required=True)
1928:    parser.add_argument("--public-dir", type=Path, required=True)
1929:    parser.add_argument("--prepared-cache", type=Path)
1930:    parser.add_argument("--device", choices=("cpu", "mps"), default="cpu")
1931:    parser.add_argument("--threads", type=int, default=8)
1932:    parser.add_argument("--bptt-steps", type=int, default=Config.bptt_steps)
1933:    parser.add_argument(
1936:    parser.add_argument("--epochs", type=int, default=Config.epochs)
1937:    parser.add_argument("--learning-rate", type=float, default=Config.learning_rate)
1938:    parser.add_argument(
1943:    parser.add_argument(
1948:    parser.add_argument(
1953:    parser.add_argument(
1958:    parser.add_argument(
1963:    parser.add_argument(
1968:    parser.add_argument(
1973:    parser.add_argument(
1978:    parser.add_argument(
1984:    parser.add_argument(
1990:    parser.add_argument(
1995:    parser.add_argument(
2000:    parser.add_argument("--checkpoint-dir", type=Path)
2001:    parser.add_argument("--resume-checkpoint", type=Path)
2002:    parser.add_argument("--seed", type=int, default=Config.seed)
2003:    parser.add_argument("--ranking-batch-size", type=int, default=64)
2004:    parser.add_argument("--prepare-only", action="store_true")
Shell cwd was reset to /app

1	#!/usr/bin/env python3
2	"""Train the Yambda windowed two-tower baseline.
3	
4	Every catalog listen with at least 50 percent completion is a sequence token.
5	Within each selected chronological window, every adjacent target contributes a
6	loss.  The fast default retains each user's most recent 128 training targets and
7	16 held-out validation targets; passing zero for either cap restores the
8	uncapped behavior for that side of the split.  A 100-step truncated-BPTT block
9	computes all of its timestep losses and carries the detached GRU hidden state
10	into the next block.
11	
12	The model is a pure two-tower retriever: no user-ID embedding, no output
13	normalization, raw dot-product retrieval, sample-weighted binary cross entropy,
14	and 24 uniform plus eight propensity-corrected unique in-batch negatives. Every
15	selected adjacent target retains a nonzero weight. Only files under the supplied
16	public directory are read.
17	"""
18	from __future__ import annotations
19	
20	import argparse
21	import json
22	import random
23	import sys
24	import time
25	from dataclasses import asdict, dataclass
26	from pathlib import Path
27	
28	import numpy as np
29	import pandas as pd
30	import torch
31	from torch import nn
32	from torch.nn import functional as F
33	
34	import two_tower_model as base
35	
36	
37	CACHE_VERSION = 2
38	
39	
40	@dataclass(frozen=True)
41	class Config:
42	    bptt_steps: int = 100
43	    user_batch_size: int = 64
44	    embedding_dim: int = 192
45	    hidden_dim: int = 192
46	    user_hidden_1: int = 512
47	    user_hidden_2: int = 256
48	    item_hidden_1: int = 512
49	    item_hidden_2: int = 256
50	    dropout: float = 0.05
51	    epochs: int = 1
52	    learning_rate: float = 1.0e-3
53	    weight_decay: float = 1.0e-5
54	    gradient_clip: float = 2.0
55	    loss_microbatch_size: int = 65_536
56	    uniform_negatives: int = 24
57	    inbatch_negatives: int = 8
58	    inbatch_correction_power: float = 0.5
59	    inbatch_correction_min: float = 0.25
60	    inbatch_correction_max: float = 4.0
61	    negative_filter_scope: str = "full-training-history"
62	    target_weighting: str = "organic-strong"
63	    ordinary_target_weight: float = 0.25
64	    organic_strong_target_weight: float = 1.0
65	    organic_like_target_weight: float = 3.0
66	    discovery_horizon_days: int = 7
67	    validation_fraction: float = 0.10
68	    max_train_targets_per_user: int = 128
69	    max_validation_targets_per_user: int = 16
70	    early_stopping_patience: int = 1
71	    early_stopping_min_delta: float = 1.0e-4
72	    seed: int = 29
73	
74	
75	@dataclass(frozen=True)
76	class CandidateSet:
77	    uids: np.ndarray
78	    item_ids: np.ndarray
79	    packed: np.ndarray
80	
81	    @classmethod
82	    def read(cls, path: Path) -> "CandidateSet":
83	        with np.load(path, allow_pickle=False) as artifact:
84	            result = cls(
85	                uids=artifact["uids"].astype(np.int64, copy=True),
86	                item_ids=artifact["item_ids"].astype(np.int64, copy=True),
87	                packed=artifact["packed_eligible"].astype(np.uint8, copy=True),
88	            )
89	        expected = (len(result.uids), (len(result.item_ids) + 7) // 8)
90	        if result.packed.shape != expected:
91	            raise ValueError(
92	                f"invalid candidate artifact shape {result.packed.shape}; "
93	                f"expected {expected}"
94	            )
95	        return result
96	
97	
98	@dataclass(frozen=True)
99	class SequenceData:
100	    item_ids: np.ndarray
101	    item_features: np.ndarray
102	    user_ids: np.ndarray
103	    offsets: np.ndarray
104	    sequence_timestamps: np.ndarray
105	    sequence_items: np.ndarray
106	    event_features: np.ndarray
107	    activity_features: np.ndarray
108	    state_ids: np.ndarray
109	    liked_states: np.ndarray
110	    disliked_states: np.ndarray
111	    state_counts: np.ndarray
112	    inference_user_rows: np.ndarray
113	    inference_liked_items: np.ndarray
114	    inference_disliked_items: np.ndarray
115	    inference_user_features: np.ndarray
116	
117	    @property
118	    def lengths(self) -> np.ndarray:
119	        return np.diff(self.offsets)
120	
121	    @property
122	    def target_count(self) -> int:
123	        return int(np.maximum(self.lengths - 1, 0).sum())
124	
125	
126	@dataclass(frozen=True)
127	class TargetPartition:
128	    """A contiguous recent target window and its chronological BCE split."""
129	
130	    window_starts: np.ndarray
131	    training_counts: np.ndarray
132	    validation_counts: np.ndarray
133	    history_training_counts: np.ndarray
134	
135	    @property
136	    def selected_count(self) -> int:
137	        return int(self.training_counts.sum() + self.validation_counts.sum())
138	
139	
140	def log(message: str) -> None:
141	    print(message, file=sys.stderr, flush=True)
142	
143	
144	def seed_everything(seed: int) -> None:
145	    random.seed(seed)
146	    np.random.seed(seed)
147	    torch.manual_seed(seed)
148	
149	
150	def model_config(config: Config) -> base.TowerConfig:
151	    """Map the dense trainer configuration onto the unchanged tower."""
152	    return base.TowerConfig(
153	        sequence_length=config.bptt_steps,
154	        embedding_dim=config.embedding_dim,
155	        hidden_dim=config.hidden_dim,
156	        user_hidden_1=config.user_hidden_1,
157	        user_hidden_2=config.user_hidden_2,
158	        item_hidden_1=config.item_hidden_1,
159	        item_hidden_2=config.item_hidden_2,
160	        dropout=config.dropout,
161	        epochs=config.epochs,
162	        learning_rate=config.learning_rate,
163	        weight_decay=config.weight_decay,
164	        gradient_clip=config.gradient_clip,
165	        uniform_negatives=config.uniform_negatives,
166	        inbatch_negatives=config.inbatch_negatives,
167	        inbatch_correction_power=config.inbatch_correction_power,
168	        inbatch_correction_min=config.inbatch_correction_min,
169	        inbatch_correction_max=config.inbatch_correction_max,
170	        seed=config.seed,
171	    )
172	
173	
174	def append_state_snapshot(
175	    active_likes: dict[int, int],

530	        save_cache(cache, prepared, stage, config)
531	    return prepared, built_eligibility
532	
533	
534	def build_item_correction(
535	    prepared: SequenceData,
536	    config: Config,
537	    target_starts: np.ndarray,
538	    target_counts: np.ndarray,
539	) -> torch.Tensor:
540	    """Estimate in-batch propensities from the selected training targets."""
541	    counts = np.zeros(len(prepared.item_ids), dtype=np.float64)
542	    for user_row, count in enumerate(target_counts):
543	        if not count:
544	            continue
545	        offset = int(prepared.offsets[user_row])
546	        target_start = int(target_starts[user_row])
547	        counts += np.bincount(
548	            prepared.sequence_items[
549	                offset
550	                + target_start
551	                + 1 : offset
552	                + target_start
553	                + int(count)
554	                + 1
555	            ].astype(np.int64),
556	            minlength=len(prepared.item_ids),
557	        )
558	    total = float(counts.sum())
559	    probabilities = (counts + 1.0) / (total + len(counts))
560	    raw = np.power(
561	        (1.0 / len(counts)) / probabilities,
562	        config.inbatch_correction_power,
563	    )
564	    normalization = float(np.sum(counts * raw) / max(total, 1.0))
565	    correction = np.clip(
566	        raw / max(normalization, 1.0e-12),
567	        config.inbatch_correction_min,
568	        config.inbatch_correction_max,
569	    ).astype(np.float32)
570	    mean_on_targets = float(np.sum(counts * correction) / max(total, 1.0))
571	    log(
572	        f"inbatch correction: min={correction.min():.3f} "
573	        f"mean_on_targets={mean_on_targets:.3f} max={correction.max():.3f}"
574	    )
575	    return torch.from_numpy(correction)
576	
577	
578	def build_known_positive_filter(
579	    prepared: SequenceData, training_counts: np.ndarray
580	) -> np.ndarray:
581	    """Pack each user's complete public-training positive-item set."""
582	    width = (len(prepared.item_ids) + 7) // 8
583	    packed = np.zeros((len(prepared.user_ids), width), dtype=np.uint8)
584	    retained = 0
585	    for user_row, target_count in enumerate(training_counts):
586	        if not target_count:
587	            continue
588	        offset = int(prepared.offsets[user_row])
589	        # T adjacent targets consume sequence positions 0 through T.  The
590	        # chronological BCE suffix starts at T + 1 and is intentionally absent.
591	        items = np.unique(
592	            prepared.sequence_items[
593	                offset : offset + int(target_count) + 1
594	            ].astype(np.int64)
595	        )
596	        byte_columns = items >> 3
597	        bit_values = np.left_shift(
598	            np.uint8(1), (items & 7).astype(np.uint8)
599	        )
600	        np.bitwise_or.at(packed[user_row], byte_columns, bit_values)
601	        retained += len(items)
602	    log(
603	        f"full-history false-negative filter: users={len(prepared.user_ids):,} "
604	        f"unique_user_items={retained:,} bytes={packed.nbytes:,}"
605	    )
606	    return packed
607	
608	
609	def invalid_full_history_negatives(
610	    negatives: np.ndarray,
611	    targets: np.ndarray,
612	    user_rows: np.ndarray,
613	    known_positive_packed: np.ndarray,
614	    liked_items: np.ndarray,
615	) -> np.ndarray:
616	    """Return row-level invalid negatives using the complete training prefix."""
617	    byte_columns = negatives >> 3
618	    bit_columns = negatives & 7
619	    known = known_positive_packed[user_rows[:, None], byte_columns]
620	    invalid = ((known >> bit_columns) & 1).astype(bool)
621	    invalid |= negatives == targets[:, None]
622	    invalid |= (negatives[:, :, None] == liked_items[:, None, :]).any(axis=2)
623	    for column in range(1, negatives.shape[1]):
624	        invalid[:, column] |= (
625	            negatives[:, column, None] == negatives[:, :column]
626	        ).any(axis=1)
627	    return invalid
628	
629	
630	def draw_full_history_filtered_negatives(
631	    targets: torch.Tensor,
632	    user_rows: np.ndarray,
633	    liked_items: torch.Tensor,
634	    known_positive_packed: np.ndarray,
635	    n_items: int,
636	    config: Config,
637	    item_correction: torch.Tensor,
638	    generator: torch.Generator,
639	    device: torch.device,
640	) -> tuple[torch.Tensor, torch.Tensor]:
641	    """Draw mixed negatives and reject all known public-training positives."""
642	    targets_cpu = targets.detach().cpu().numpy().astype(np.int64, copy=False)
643	    liked_cpu = liked_items.detach().cpu().numpy().astype(np.int64, copy=False)
644	    batch_size = len(targets_cpu)
645	    uniform = torch.randint(
646	        n_items,
647	        (batch_size, config.uniform_negatives),
648	        generator=generator,
649	        device="cpu",
650	    )
651	    inbatch = torch.empty((batch_size, 0), dtype=torch.long)
652	    corrected_count = 0
653	    if config.inbatch_negatives:
654	        unique_targets = torch.unique(targets.detach().cpu())
655	        corrected_count = min(
656	            config.inbatch_negatives, max(len(unique_targets) - 1, 0)
657	        )
658	        if corrected_count:
659	            priorities = torch.rand(
660	                (batch_size, len(unique_targets)), generator=generator
661	            )
662	            priorities.masked_fill_(
663	                unique_targets[None, :].eq(
664	                    torch.from_numpy(targets_cpu)[:, None]
665	                ),
666	                -1.0,
667	            )
668	            columns = torch.topk(
669	                priorities, k=corrected_count, dim=1
670	            ).indices
671	            inbatch = unique_targets[columns]
672	        if corrected_count < config.inbatch_negatives:
673	            filler = torch.randint(
674	                n_items,
675	                (batch_size, config.inbatch_negatives - corrected_count),
676	                generator=generator,
677	                device="cpu",
678	            )
679	            inbatch = torch.cat([inbatch, filler], dim=1)
680	
681	    negatives = torch.cat([uniform, inbatch], dim=1).numpy()
682	    correction = np.ones_like(negatives, dtype=np.float32)
683	    if corrected_count:
684	        correction[
685	            :, config.uniform_negatives : config.uniform_negatives + corrected_count
686	        ] = item_correction[inbatch[:, :corrected_count]].numpy()
687	    for _ in range(32):
688	        invalid = invalid_full_history_negatives(
689	            negatives,
690	            targets_cpu,
691	            user_rows,
692	            known_positive_packed,
693	            liked_cpu,
694	        )
695	        if not np.any(invalid):
696	            return (
697	                torch.from_numpy(negatives).to(device),
698	                torch.from_numpy(correction).to(device),
699	            )
700	        replacements = torch.randint(
701	            n_items,
702	            (int(invalid.sum()),),
703	            generator=generator,
704	            device="cpu",
705	        ).numpy()
706	        negatives[invalid] = replacements
707	        # Rejected in-batch items become ordinary uniform draws.
708	        correction[invalid] = 1.0
709	    invalid = invalid_full_history_negatives(
710	        negatives,
711	        targets_cpu,
712	        user_rows,
713	        known_positive_packed,
714	        liked_cpu,
715	    )
716	    if np.any(invalid):
717	        raise RuntimeError("full-history negative resampling failed")
718	    return (
719	        torch.from_numpy(negatives).to(device),
720	        torch.from_numpy(correction).to(device),
721	    )
722	
723	
724	def draw_training_negatives(
725	    positive_items: torch.Tensor,
726	    known_history: torch.Tensor,
727	    liked_items: torch.Tensor,
728	    user_rows: np.ndarray,
729	    known_positive_packed: np.ndarray | None,
730	    prepared: SequenceData,
731	    config: Config,
732	    item_correction: torch.Tensor,
733	    generator: torch.Generator,
734	    device: torch.device,
735	) -> tuple[torch.Tensor, torch.Tensor]:
736	    if config.negative_filter_scope == "full-training-history":
737	        if known_positive_packed is None:
738	            raise RuntimeError("full-history negative filter was not built")
739	        return draw_full_history_filtered_negatives(
740	            positive_items,
741	            user_rows,
742	            liked_items,
743	            known_positive_packed,
744	            len(prepared.item_ids),
745	            config,
746	            item_correction,
747	            generator,
748	            device,
749	        )
750	    return base.draw_filtered_mixed_negatives(
751	        positive_items,
752	        known_history,
753	        liked_items,
754	        len(prepared.item_ids),
755	        config.uniform_negatives,
756	        config.inbatch_negatives,
757	        item_correction,
758	        generator,
759	        device,
760	    )
761	
762	
763	def event_sequence(
764	    model: base.SequentialTwoTower,
765	    items: torch.Tensor,
766	    features: torch.Tensor,
767	    hidden: torch.Tensor,
768	) -> tuple[torch.Tensor, torch.Tensor]:
769	    inputs = torch.cat([model.context_embedding(items), features], dim=-1)
770	    projected = model.event_projection(inputs)
771	    return model.sequence_encoder(projected, hidden)
772	
773	
774	def pool_explicit_states(
775	    model: base.SequentialTwoTower,
776	    prepared: SequenceData,
777	    state_ids: np.ndarray,
778	    device: torch.device,
779	) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
780	    unique_states, inverse = np.unique(state_ids, return_inverse=True)
781	    liked_items = torch.from_numpy(
782	        prepared.liked_states[unique_states].astype(np.int64)
783	    ).to(device)
784	    disliked_items = torch.from_numpy(
785	        prepared.disliked_states[unique_states].astype(np.int64)
786	    ).to(device)
787	    liked_vectors = model._mean_context_channel(liked_items)
788	    disliked_vectors = model._mean_context_channel(disliked_items)
789	    inverse_rows = torch.from_numpy(inverse.astype(np.int64)).to(device)
790	    counts = prepared.state_counts[state_ids]
791	    return (
792	        liked_vectors[inverse_rows],
793	        disliked_vectors[inverse_rows],
794	        torch.from_numpy(counts.astype(np.float32)).to(device),
795	    )
796	
797	
798	def user_queries(
799	    model: base.SequentialTwoTower,
800	    sequence_vectors: torch.Tensor,
801	    liked_vectors: torch.Tensor,
802	    disliked_vectors: torch.Tensor,
803	    user_features: torch.Tensor,
804	) -> torch.Tensor:
805	    return model.user_tower(
806	        torch.cat(
807	            [sequence_vectors, liked_vectors, disliked_vectors, user_features],
808	            dim=1,
809	        )
810	    )
811	
812	
813	def candidate_vectors(
814	    model: base.SequentialTwoTower,
815	    positives: torch.Tensor,
816	    negatives: torch.Tensor,
817	) -> tuple[torch.Tensor, torch.Tensor]:
818	    all_items = torch.cat([positives, negatives.reshape(-1)])
819	    unique_items, inverse = torch.unique(all_items, return_inverse=True)
820	    encoded = model.encode_items(unique_items)
821	    positive_vectors = encoded[inverse[: len(positives)]]
822	    negative_vectors = encoded[inverse[len(positives) :]].reshape(
823	        len(positives), negatives.shape[1], -1
824	    )
825	    return positive_vectors, negative_vectors
826	
827	
828	def history_windows(
829	    sequence: np.ndarray,
830	    start: int,
831	    count: int,
832	    width: int,
833	    padding_item: int,
834	) -> np.ndarray:
835	    base_start = max(0, start - width + 1)
836	    segment = sequence[base_start : start + count]
837	    padded = np.concatenate(
838	        [np.full(width - 1, padding_item, dtype=np.uint16), segment]
839	    )
840	    windows = np.lib.stride_tricks.sliding_window_view(padded, width)
841	    offset = start - base_start
842	    return windows[offset : offset + count].copy()
843	
844	
845	def corrected_negative_loss(
846	    logits: torch.Tensor,
847	    correction: torch.Tensor,
848	    config: Config,
849	) -> torch.Tensor:
850	    raw = F.softplus(logits)
851	    total = config.uniform_negatives + config.inbatch_negatives
852	    result = torch.zeros(len(logits), device=logits.device)
853	    if config.uniform_negatives:
854	        result += config.uniform_negatives * raw[
855	            :, : config.uniform_negatives
856	        ].mean(dim=1)
857	    if config.inbatch_negatives:
858	        selected = slice(config.uniform_negatives, total)
859	        weights = correction[:, selected]
860	        result += config.inbatch_negatives * (
861	            (raw[:, selected] * weights).sum(dim=1)
862	            / weights.sum(dim=1).clamp_min(1.0e-6)
863	        )
864	    return result / total
865	
866	
867	def retained_target_window_starts(
868	    total_targets: np.ndarray,
869	    validation_counts: np.ndarray,
870	    training_counts: np.ndarray,
871	) -> np.ndarray:
872	    """Return the discarded prefix before the retained chronological suffix."""
873	    training_region_ends = total_targets - validation_counts
874	    window_starts = training_region_ends - training_counts
875	    if (
876	        np.any(window_starts < 0)
877	        or np.any(training_counts < 0)
878	        or np.any(validation_counts < 0)
879	    ):
880	        raise RuntimeError("chronological target partition produced negative counts")
881	    if not np.array_equal(
882	        window_starts + training_counts + validation_counts, total_targets
883	    ):
884	        raise RuntimeError("chronological target window is not a final contiguous suffix")
885	    return window_starts
886	
887	
888	def chronological_target_partition(
889	    prepared: SequenceData,
890	    validation_fraction: float,
891	    max_train_targets_per_user: int,
892	    max_validation_targets_per_user: int,
893	) -> TargetPartition:
894	    """Select a recent contiguous window with a final out-of-time BCE suffix."""
895	    totals = np.maximum(prepared.lengths - 1, 0).astype(np.int64)
896	    validation = np.floor(totals * validation_fraction).astype(np.int64)
897	    validation[(totals >= 2) & (validation == 0)] = 1
898	    if max_validation_targets_per_user:
899	        validation = np.minimum(validation, max_validation_targets_per_user)
900	
901	    history_training = totals - validation
902	    training = history_training.copy()
903	    if max_train_targets_per_user:
904	        training = np.minimum(training, max_train_targets_per_user)
905	    window_starts = retained_target_window_starts(totals, validation, training)
906	    return TargetPartition(
907	        window_starts=window_starts,
908	        training_counts=training,
909	        validation_counts=validation,
910	        history_training_counts=history_training,
911	    )
912	
913	
914	def load_organic_like_outcomes(
915	    public: Path, stage: str, item_ids: np.ndarray
916	) -> dict[int, dict[int, np.ndarray]]:
917	    """Load mapped organic-like timestamps from public fitting history."""
918	    prefixes = ("train",) if stage == "validation" else ("train", "val")
919	    frames = [
920	        pd.read_parquet(
921	            public / "data" / f"{prefix}_likes.parquet",
922	            columns=["uid", "timestamp", "item_id", "is_organic"],
923	        )
924	        for prefix in prefixes
925	    ]
926	    likes = pd.concat(frames, ignore_index=True)
927	    likes = likes[likes.is_organic == 1].copy()
928	    mapped = pd.Index(item_ids).get_indexer(
929	        likes.item_id.to_numpy(dtype=np.int64)
930	    )
931	    likes = likes.loc[mapped >= 0, ["uid", "timestamp"]].copy()
932	    likes["item"] = mapped[mapped >= 0].astype(np.int32, copy=False)
933	    likes = likes.sort_values(["uid", "item", "timestamp"], kind="stable")
934	    nested: dict[int, dict[int, list[int]]] = {}
935	    for raw_uid, raw_timestamp, raw_item in likes.itertuples(
936	        index=False, name=None
937	    ):
938	        nested.setdefault(int(raw_uid), {}).setdefault(int(raw_item), []).append(
939	            int(raw_timestamp)
940	        )
941	    result = {
942	        uid: {
943	            item: np.asarray(timestamps, dtype=np.int64)
944	            for item, timestamps in items.items()
945	        }
946	        for uid, items in nested.items()
947	    }
948	    log(
949	        f"loaded organic-like outcomes: rows={len(likes):,} "

950	        f"users={len(result):,}"
951	    )
952	    return result
953	
954	
955	def build_target_weights(
956	    public: Path,
957	    stage: str,
958	    prepared: SequenceData,
959	    training_counts: np.ndarray,
960	    config: Config,
961	) -> np.ndarray:
962	    """Build nonzero per-token weights without dropping adjacent targets."""
963	    if config.target_weighting == "uniform":
964	        weights = np.ones(len(prepared.sequence_items), dtype=np.float32)
965	    else:
966	        weights = np.full(
967	            len(prepared.sequence_items),
968	            config.ordinary_target_weight,
969	            dtype=np.float32,
970	        )
971	        organic_strong = (
972	            (prepared.event_features[:, 0] > 0.5)
973	            & (prepared.event_features[:, 1] >= 0.8)
974	        )
975	        weights[organic_strong] = config.organic_strong_target_weight
976	        if config.target_weighting == "discovery-graded":
977	            like_outcomes = load_organic_like_outcomes(
978	                public, stage, prepared.item_ids
979	            )
980	            history_end = (
981	                base.TRAIN_END if stage == "validation" else base.VALIDATION_END
982	            )
983	            horizon = config.discovery_horizon_days * base.DAY
984	            grade_three = 0
985	            for user_row, uid in enumerate(prepared.user_ids):
986	                start = int(prepared.offsets[user_row])
987	                end = int(prepared.offsets[user_row + 1])
988	                if end - start < 2:
989	                    continue
990	                last_training_target = start + int(training_counts[user_row])
991	                first_validation_target = last_training_target + 1
992	                split_timestamp = (
993	                    int(prepared.sequence_timestamps[first_validation_target])
994	                    if first_validation_target < end
995	                    else history_end
996	                )
997	                user_likes = like_outcomes.get(int(uid), {})
998	                seen = {int(prepared.sequence_items[start])}
999	                for position in range(start + 1, end):
1000	                    item = int(prepared.sequence_items[position])
1001	                    novel = item not in seen
1002	                    seen.add(item)
1003	                    if not novel:
1004	                        # Repeats remain training examples, but they cannot be
1005	                        # eligible discovery outcomes at the corresponding
1006	                        # cutoff and therefore keep only the ordinary weight.
1007	                        weights[position] = config.ordinary_target_weight
1008	                        continue
1009	                    timestamps = user_likes.get(item)
1010	                    if timestamps is None:
1011	                        continue
1012	                    timestamp = int(prepared.sequence_timestamps[position])
1013	                    upper = min(
1014	                        timestamp + horizon,
1015	                        split_timestamp
1016	                        if position <= last_training_target
1017	                        else history_end,
1018	                    )
1019	                    outcome = int(np.searchsorted(timestamps, timestamp, side="left"))
1020	                    if outcome < len(timestamps) and int(timestamps[outcome]) < upper:
1021	                        weights[position] = config.organic_like_target_weight
1022	                        grade_three += 1
1023	            log(
1024	                f"grade-3 first-encounter targets={grade_three:,} "
1025	                f"horizon_days={config.discovery_horizon_days}"
1026	            )
1027	    log(
1028	        f"target weighting={config.target_weighting} "
1029	        f"min={weights.min():.3f} mean={weights.mean():.3f} "
1030	        f"max={weights.max():.3f}"
1031	    )
1032	    return weights
1033	
1034	
1035	def build_dense_block(
1036	    prepared: SequenceData,
1037	    target_weights: np.ndarray,
1038	    users: np.ndarray,
1039	    window_starts: np.ndarray,
1040	    process_counts: np.ndarray,
1041	    score_starts: np.ndarray,
1042	    start: int,
1043	    steps: int,
1044	    config: Config,
1045	) -> tuple[np.ndarray, ...]:
1046	    """Build one causal sequence block and mark the target rows to score."""
1047	    batch_size = len(users)
1048	    padding_item = len(prepared.item_ids)
1049	    input_items = np.full((batch_size, steps), padding_item, dtype=np.uint16)
1050	    event_features = np.zeros((batch_size, steps, 3), dtype=np.float16)
1051	    activity_features = np.zeros((batch_size, steps, 4), dtype=np.float16)
1052	    state_ids = np.zeros((batch_size, steps), dtype=np.uint32)
1053	    targets = np.zeros((batch_size, steps), dtype=np.uint16)
1054	    weights = np.ones((batch_size, steps), dtype=np.float32)
1055	    scored = np.zeros((batch_size, steps), dtype=bool)
1056	    histories = np.full(
1057	        (batch_size, steps, config.bptt_steps),
1058	        padding_item,
1059	        dtype=np.uint16,
1060	    )
1061	    processed = np.zeros(batch_size, dtype=np.int32)
1062	
1063	    lengths = prepared.lengths
1064	    for row, user_row in enumerate(users):
1065	        count = min(steps, max(int(process_counts[row]) - start, 0))
1066	        if not count:
1067	            continue
1068	        offset = int(prepared.offsets[user_row])
1069	        window_start = int(window_starts[row])
1070	        absolute_start = window_start + start
1071	        positions = slice(
1072	            offset + absolute_start,
1073	            offset + absolute_start + count,
1074	        )
1075	        target_positions = slice(
1076	            offset + absolute_start + 1,
1077	            offset + absolute_start + count + 1,
1078	        )
1079	        input_items[row, :count] = prepared.sequence_items[positions]
1080	        event_features[row, :count] = prepared.event_features[positions]
1081	        activity_features[row, :count] = prepared.activity_features[positions]
1082	        state_ids[row, :count] = prepared.state_ids[positions]
1083	        targets[row, :count] = prepared.sequence_items[target_positions]
1084	        weights[row, :count] = target_weights[target_positions]
1085	        score_from = min(count, max(int(score_starts[row]) - start, 0))
1086	        scored[row, score_from:count] = True
1087	        processed[row] = count
1088	        user_sequence = prepared.sequence_items[
1089	            offset : offset + int(lengths[user_row])
1090	        ]
1091	        histories[row, :count] = history_windows(
1092	            user_sequence,
1093	            absolute_start,
1094	            count,
1095	            config.bptt_steps,
1096	            padding_item,
1097	        )
1098	    return (
1099	        input_items,
1100	        event_features,
1101	        activity_features,
1102	        state_ids,
1103	        targets,
1104	        scored,
1105	        histories,
1106	        processed,
1107	        weights,
1108	    )
1109	
1110	
1111	def score_target_microbatches(
1112	    model: base.SequentialTwoTower,
1113	    prepared: SequenceData,
1114	    config: Config,
1115	    sequence_vectors: torch.Tensor,
1116	    flat_states: np.ndarray,
1117	    flat_activity: np.ndarray,
1118	    positive_items: torch.Tensor,
1119	    negatives: torch.Tensor,
1120	    negative_weights: torch.Tensor,
1121	    sample_weights: np.ndarray,
1122	    device: torch.device,
1123	    backward: bool,
1124	) -> tuple[dict[str, float], torch.Tensor | None]:
1125	    """Compute one logical block's exact BCE in bounded-memory slices."""
1126	    target_count = len(positive_items)
1127	    total_weight = float(sample_weights.sum())
1128	    if total_weight <= 0:
1129	        raise RuntimeError("target block has no positive sample weight")
1130	    sequence_gradient = torch.zeros_like(sequence_vectors) if backward else None
1131	    statistics = {
1132	        "loss_sum": 0.0,
1133	        "positive_sum": 0.0,
1134	        "negative_sum": 0.0,
1135	        "pair_correct": 0.0,
1136	        "weight_sum": 0.0,
1137	    }
1138	    for begin in range(0, target_count, config.loss_microbatch_size):
1139	        end = min(begin + config.loss_microbatch_size, target_count)
1140	        selected_states = flat_states[begin:end]
1141	        if backward:
1142	            sequence_input = (
1143	                sequence_vectors[begin:end].detach().requires_grad_(True)
1144	            )
1145	        else:
1146	            sequence_input = sequence_vectors[begin:end]
1147	        liked_vectors, disliked_vectors, _ = pool_explicit_states(
1148	            model, prepared, selected_states, device
1149	        )
1150	        counts = prepared.state_counts[selected_states]
1151	        numeric = np.zeros((end - begin, 6), dtype=np.float32)
1152	        numeric[:, :4] = flat_activity[begin:end]
1153	        numeric[:, 4] = np.log1p(counts[:, 0]) / 6.0
1154	        numeric[:, 5] = np.log1p(counts[:, 1]) / 4.0
1155	        query_vectors = user_queries(
1156	            model,
1157	            sequence_input,
1158	            liked_vectors,
1159	            disliked_vectors,
1160	            torch.from_numpy(numeric).to(device),
1161	        )
1162	        positive_vectors, negative_vectors = candidate_vectors(
1163	            model,
1164	            positive_items[begin:end],
1165	            negatives[begin:end],
1166	        )
1167	        positive_logits = (query_vectors * positive_vectors).sum(dim=1)
1168	        negative_logits = torch.einsum(
1169	            "bd,bkd->bk", query_vectors, negative_vectors
1170	        )
1171	        positive_loss = F.softplus(-positive_logits)
1172	        negative_loss = corrected_negative_loss(
1173	            negative_logits, negative_weights[begin:end], config
1174	        )
1175	        selected_weights = torch.from_numpy(
1176	            sample_weights[begin:end].astype(np.float32, copy=False)
1177	        ).to(device)
1178	        loss_sum = (
1179	            selected_weights * (positive_loss + negative_loss) / 2.0
1180	        ).sum()
1181	        if not bool(torch.isfinite(loss_sum)):
1182	            raise RuntimeError("dense next-item training produced non-finite loss")
1183	        if backward:
1184	            (loss_sum / total_weight).backward()
1185	            if sequence_input.grad is None:
1186	                raise RuntimeError("sequence microbatch did not produce a gradient")
1187	            sequence_gradient[begin:end] = sequence_input.grad
1188	        statistics["loss_sum"] += float(loss_sum.detach().cpu())
1189	        statistics["positive_sum"] += float(positive_logits.detach().sum().cpu())
1190	        statistics["negative_sum"] += float(negative_logits.detach().sum().cpu())
1191	        statistics["pair_correct"] += float(
1192	            (positive_logits.detach()[:, None] > negative_logits.detach()).sum().cpu()
1193	        )
1194	        statistics["weight_sum"] += float(selected_weights.sum().detach().cpu())
1195	    return statistics, sequence_gradient
1196	
1197	
1198	def release_accelerator_cache(device: torch.device) -> None:
1199	    if device.type == "mps":
1200	        torch.mps.empty_cache()
1201	
1202	
1203	def run_training_epoch(
1204	    model: base.SequentialTwoTower,
1205	    optimizer: torch.optim.Optimizer,
1206	    prepared: SequenceData,
1207	    config: Config,
1208	    target_weights: np.ndarray,
1209	    window_starts: np.ndarray,
1210	    target_counts: np.ndarray,
1211	    item_correction: torch.Tensor,
1212	    known_positive_packed: np.ndarray | None,
1213	    generator: torch.Generator,
1214	    device: torch.device,
1215	    epoch: int,
1216	) -> dict[str, float]:
1217	    model.train()
1218	    train_users = np.flatnonzero(target_counts > 0)
1219	    sorted_users = train_users[
1220	        np.argsort(target_counts[train_users], kind="stable")
1221	    ]
1222	    groups = [
1223	        sorted_users[start : start + config.user_batch_size]
1224	        for start in range(0, len(sorted_users), config.user_batch_size)
1225	    ]
1226	    group_order = torch.randperm(len(groups), generator=generator).numpy()
1227	    negative_count = config.uniform_negatives + config.inbatch_negatives
1228	    total_loss = 0.0
1229	    total_targets = 0
1230	    total_weight = 0.0
1231	    total_positive = 0.0
1232	    total_negative = 0.0
1233	    total_pair_correct = 0.0
1234	    batches = 0
1235	    started = time.monotonic()
1236	
1237	    for group_number in group_order:
1238	        users = groups[int(group_number)]
1239	        group_counts = target_counts[users]
1240	        hidden = torch.zeros(
1241	            1, len(users), config.embedding_dim, device=device
1242	        )
1243	        max_targets = int(group_counts.max())
1244	        for start in range(0, max_targets, config.bptt_steps):
1245	            steps = min(config.bptt_steps, max_targets - start)
1246	            block = build_dense_block(
1247	                prepared,
1248	                target_weights,
1249	                users,
1250	                window_starts[users],
1251	                group_counts,
1252	                np.zeros(len(users), dtype=np.int64),
1253	                start,
1254	                steps,
1255	                config,
1256	            )
1257	            sequence_outputs, _ = event_sequence(
1258	                model,
1259	                torch.from_numpy(block[0].astype(np.int64)).to(device),
1260	                torch.from_numpy(block[1].astype(np.float32)).to(device),
1261	                hidden,
1262	            )
1263	            hidden = torch.stack(
1264	                [
1265	                    sequence_outputs[row, count - 1]
1266	                    if count
1267	                    else hidden[0, row]
1268	                    for row, count in enumerate(block[7])
1269	                ]
1270	            ).unsqueeze(0).detach()
1271	            flat_scored = block[5].reshape(-1)
1272	            if not np.any(flat_scored):
1273	                continue
1274	            mask = torch.from_numpy(flat_scored).to(device)
1275	            sequence_vectors = sequence_outputs.reshape(
1276	                -1, config.embedding_dim
1277	            )[mask]
1278	            flat_states = block[3].reshape(-1)[flat_scored]
1279	            positive_items = torch.from_numpy(
1280	                block[4].reshape(-1)[flat_scored].astype(np.int64)
1281	            ).to(device)
1282	            known_history = torch.from_numpy(
1283	                block[6].reshape(-1, config.bptt_steps)[flat_scored].astype(
1284	                    np.int64
1285	                )
1286	            ).to(device)
1287	            liked_items = torch.from_numpy(
1288	                prepared.liked_states[flat_states].astype(np.int64)
1289	            ).to(device)
1290	            flat_user_rows = np.broadcast_to(
1291	                users[:, None], block[5].shape
1292	            ).reshape(-1)[flat_scored]
1293	            negatives, negative_weights = draw_training_negatives(
1294	                positive_items,
1295	                known_history,
1296	                liked_items,
1297	                flat_user_rows,
1298	                known_positive_packed,
1299	                prepared,
1300	                config,
1301	                item_correction,
1302	                generator,
1303	                device,
1304	            )
1305	            optimizer.zero_grad(set_to_none=True)
1306	            statistics, sequence_gradient = score_target_microbatches(
1307	                model,
1308	                prepared,
1309	                config,
1310	                sequence_vectors,
1311	                flat_states,
1312	                block[2].reshape(-1, 4)[flat_scored].astype(np.float32),
1313	                positive_items,
1314	                negatives,
1315	                negative_weights,
1316	                block[8].reshape(-1)[flat_scored],
1317	                device,
1318	                backward=True,
1319	            )
1320	            if sequence_gradient is None:
1321	                raise RuntimeError("training block omitted its sequence gradient")
1322	            sequence_vectors.backward(sequence_gradient)
1323	            nn.utils.clip_grad_norm_(model.parameters(), config.gradient_clip)
1324	            optimizer.step()
1325	
1326	            count = len(positive_items)
1327	            total_loss += statistics["loss_sum"]
1328	            total_targets += count
1329	            total_weight += statistics["weight_sum"]
1330	            total_positive += statistics["positive_sum"]
1331	            total_negative += statistics["negative_sum"]
1332	            total_pair_correct += statistics["pair_correct"]
1333	            batches += 1
1334	            if batches % 25 == 0:
1335	                release_accelerator_cache(device)
1336	            if batches % 250 == 0:
1337	                batch_bce = statistics["loss_sum"] / max(
1338	                    statistics["weight_sum"], 1.0e-12
1339	                )
1340	                running_bce = total_loss / max(total_weight, 1.0e-12)
1341	                log(
1342	                    f"epoch={epoch}/{config.epochs} batches={batches:,} "
1343	                    f"targets={total_targets:,} "
1344	                    f"batch_bce={batch_bce:.6f} "
1345	                    f"running_bce={running_bce:.6f} "
1346	                    f"elapsed={time.monotonic() - started:.1f}s"
1347	                )
1348	
1349	    return {
1350	        "epoch": float(epoch),
1351	        "loss": total_loss / max(total_weight, 1.0e-12),
1352	        "positive_logit": total_positive / max(total_targets, 1),
1353	        "negative_logit": total_negative / max(total_targets * negative_count, 1),
1354	        "pair_accuracy": total_pair_correct
1355	        / max(total_targets * negative_count, 1),
1356	        "targets": float(total_targets),
1357	        "target_weight_sum": total_weight,
1358	        "batches": float(batches),
1359	        "seconds": time.monotonic() - started,
1360	    }
1361	
1362	
1363	@torch.no_grad()
1364	def evaluate_validation_bce(
1365	    model: base.SequentialTwoTower,
1366	    prepared: SequenceData,
1367	    config: Config,
1368	    target_weights: np.ndarray,
1369	    window_starts: np.ndarray,

1521	def train_model(
1522	    public: Path,
1523	    stage: str,
1524	    prepared: SequenceData,
1525	    config: Config,
1526	    device: torch.device,
1527	    checkpoint_dir: Path | None,
1528	    resume_checkpoint: Path | None,
1529	) -> tuple[base.SequentialTwoTower, list[dict[str, float]]]:
1530	    partition = chronological_target_partition(
1531	        prepared,
1532	        config.validation_fraction,
1533	        config.max_train_targets_per_user,
1534	        config.max_validation_targets_per_user,
1535	    )
1536	    training_counts = partition.training_counts
1537	    validation_counts = partition.validation_counts
1538	    log(
1539	        f"chronological BCE window: train={int(training_counts.sum()):,} "
1540	        f"validation={int(validation_counts.sum()):,} "
1541	        f"selected={partition.selected_count:,} "
1542	        f"discarded_old_targets={int(partition.window_starts.sum()):,} "
1543	        f"all={prepared.target_count:,}"
1544	    )
1545	    model = base.SequentialTwoTower(prepared.item_features, model_config(config)).to(
1546	        device
1547	    )
1548	    optimizer = torch.optim.AdamW(
1549	        model.parameters(), lr=config.learning_rate, weight_decay=config.weight_decay
1550	    )
1551	    generator = torch.Generator(device="cpu").manual_seed(config.seed + 10_000)
1552	    item_correction = build_item_correction(
1553	        prepared,
1554	        config,
1555	        partition.window_starts,
1556	        training_counts,
1557	    )
1558	    target_weights = build_target_weights(
1559	        public,
1560	        stage,
1561	        prepared,
1562	        partition.history_training_counts,
1563	        config,
1564	    )
1565	    known_positive_packed = (
1566	        build_known_positive_filter(prepared, partition.history_training_counts)
1567	        if config.negative_filter_scope == "full-training-history"
1568	        else None
1569	    )
1570	    history: list[dict[str, float]] = []
1571	    best_loss = float("inf")
1572	    best_epoch = 0
1573	    best_state: dict[str, torch.Tensor] | None = None
1574	    epochs_without_improvement = 0
1575	    start_epoch = 0
1576	
1577	    if resume_checkpoint is not None:
1578	        payload = torch.load(
1579	            resume_checkpoint, map_location=device, weights_only=False
1580	        )
1581	        stored_config = payload.get("config", {})
1582	        architecture_fields = (
1583	            "bptt_steps",
1584	            "embedding_dim",
1585	            "hidden_dim",
1586	            "user_hidden_1",
1587	            "user_hidden_2",
1588	            "item_hidden_1",
1589	            "item_hidden_2",
1590	        )
1591	        mismatches = {
1592	            name: (stored_config.get(name), getattr(config, name))
1593	            for name in architecture_fields
1594	            if stored_config.get(name) != getattr(config, name)
1595	        }
1596	        if mismatches:
1597	            raise RuntimeError(
1598	                f"resume checkpoint architecture differs: {mismatches}"
1599	            )
1600	        stored_filter_scope = stored_config.get(
1601	            "negative_filter_scope", "last-100"
1602	        )
1603	        stored_target_weighting = stored_config.get(
1604	            "target_weighting", "uniform"
1605	        )
1606	        stored_weight_values = (
1607	            stored_config.get("ordinary_target_weight", 1.0),
1608	            stored_config.get("organic_strong_target_weight", 1.0),
1609	            stored_config.get("organic_like_target_weight", 3.0),
1610	            stored_config.get("discovery_horizon_days", 7),
1611	        )
1612	        current_weight_values = (
1613	            config.ordinary_target_weight,
1614	            config.organic_strong_target_weight,
1615	            config.organic_like_target_weight,
1616	            config.discovery_horizon_days,
1617	        )
1618	        stored_window_values = (
1619	            stored_config.get("validation_fraction", 0.10),
1620	            stored_config.get("max_train_targets_per_user", 0),
1621	            stored_config.get("max_validation_targets_per_user", 0),
1622	        )
1623	        current_window_values = (
1624	            config.validation_fraction,
1625	            config.max_train_targets_per_user,
1626	            config.max_validation_targets_per_user,
1627	        )
1628	        objective_changed = (
1629	            stored_filter_scope != config.negative_filter_scope
1630	            or stored_target_weighting != config.target_weighting
1631	            or stored_weight_values != current_weight_values
1632	            or stored_window_values != current_window_values
1633	        )
1634	        model.load_state_dict(payload["model_state"])
1635	        if "optimizer_state" in payload and not objective_changed:
1636	            optimizer.load_state_dict(payload["optimizer_state"])
1637	            for group in optimizer.param_groups:
1638	                group["lr"] = config.learning_rate
1639	                group["weight_decay"] = config.weight_decay
1640	            log(f"restored optimizer state from {resume_checkpoint}")
1641	        else:
1642	            log(
1643	                f"using a fresh AdamW optimizer after loading {resume_checkpoint}"
1644	            )
1645	        if "generator_state" in payload and not objective_changed:
1646	            generator.set_state(payload["generator_state"])
1647	        start_epoch = int(payload["epoch"])
1648	        if start_epoch >= config.epochs:
1649	            raise RuntimeError(
1650	                f"resume epoch {start_epoch} must be below --epochs {config.epochs}"
1651	            )
1652	        prior = {
1653	            str(name): float(value)
1654	            for name, value in payload.get("diagnostics", {}).items()
1655	        }
1656	        if objective_changed:
1657	            prior.pop("validation_loss", None)
1658	            log(
1659	                "BCE objective changed: filter "
1660	                f"{stored_filter_scope}->{config.negative_filter_scope}, "
1661	                "target weighting "
1662	                f"{stored_target_weighting}->{config.target_weighting}, "
1663	                f"target window {stored_window_values}->{current_window_values}; "
1664	                "recomputing the resume checkpoint's validation BCE"
1665	            )
1666	        if "validation_loss" not in prior:
1667	            prior.update(
1668	                evaluate_validation_bce(
1669	                    model,
1670	                    prepared,
1671	                    config,
1672	                    target_weights,
1673	                    partition.window_starts,
1674	                    training_counts,
1675	                    validation_counts,
1676	                    item_correction,
1677	                    known_positive_packed,
1678	                    device,
1679	                )
1680	            )
1681	        history.append(prior)
1682	        best_loss = float(prior["validation_loss"])
1683	        best_epoch = start_epoch
1684	        best_state = snapshot_model(model)
1685	        log(
1686	            f"resumed validation-BCE training: epoch={start_epoch} "
1687	            f"loss={best_loss:.6f}"
1688	        )
1689	
1690	    for epoch in range(start_epoch + 1, config.epochs + 1):
1691	        diagnostics = run_training_epoch(
1692	            model,
1693	            optimizer,
1694	            prepared,
1695	            config,
1696	            target_weights,
1697	            partition.window_starts,
1698	            training_counts,
1699	            item_correction,
1700	            known_positive_packed,
1701	            generator,
1702	            device,
1703	            epoch,
1704	        )
1705	        validation = evaluate_validation_bce(
1706	            model,
1707	            prepared,
1708	            config,
1709	            target_weights,
1710	            partition.window_starts,
1711	            training_counts,
1712	            validation_counts,
1713	            item_correction,
1714	            known_positive_packed,
1715	            device,
1716	        )
1717	        diagnostics.update(validation)
1718	        history.append(diagnostics)
1719	        current_state = snapshot_model(model)
1720	        if checkpoint_dir is not None:
1721	            save_epoch_checkpoint(
1722	                checkpoint_dir,
1723	                epoch,
1724	                current_state,
1725	                config,
1726	                diagnostics,
1727	                optimizer,
1728	                generator,
1729	            )
1730	        improved = (
1731	            diagnostics["validation_loss"]
1732	            < best_loss - config.early_stopping_min_delta
1733	        )
1734	        if improved:
1735	            best_loss = diagnostics["validation_loss"]
1736	            best_epoch = epoch
1737	            best_state = current_state
1738	            epochs_without_improvement = 0
1739	            if checkpoint_dir is not None:
1740	                torch.save(
1741	                    {
1742	                        "epoch": epoch,
1743	                        "config": asdict(config),
1744	                        "diagnostics": diagnostics,
1745	                        "model_state": current_state,
1746	                        "optimizer_state": optimizer.state_dict(),
1747	                        "generator_state": generator.get_state(),
1748	                    },
1749	                    checkpoint_dir / "best.pt",
1750	                )
1751	        else:
1752	            epochs_without_improvement += 1
1753	        log(
1754	            f"epoch={epoch}/{config.epochs} train_loss={diagnostics['loss']:.5f} "
1755	            f"validation_loss={diagnostics['validation_loss']:.5f} "
1756	            f"pair_accuracy={diagnostics['pair_accuracy']:.3f} "
1757	            f"train_targets={int(diagnostics['targets']):,} "
1758	            f"validation_targets={int(diagnostics['validation_targets']):,} "
1759	            f"train_seconds={diagnostics['seconds']:.1f} "
1760	            f"validation_seconds={diagnostics['validation_seconds']:.1f}"
1761	        )
1762	        release_accelerator_cache(device)
1763	        if epochs_without_improvement >= config.early_stopping_patience:
1764	            log(
1765	                f"early stopping after epoch={epoch}; "
1766	                f"best_epoch={best_epoch} best_validation_loss={best_loss:.6f}"
1767	            )
1768	            break
1769	
1770	    if best_state is None:
1771	        raise RuntimeError("training did not produce a valid checkpoint")
1772	    model.load_state_dict(best_state)
1773	    model.to(device).eval()
1774	    log(
1775	        f"restored best validation-BCE checkpoint: epoch={best_epoch} "
1776	        f"loss={best_loss:.6f}"
1777	    )
1778	    return model, history
1779	
1780	
1781	@torch.inference_mode()
1782	def inference_queries(
1783	    model: base.SequentialTwoTower,
1784	    prepared: SequenceData,
1785	    config: Config,
1786	    device: torch.device,
1787	) -> torch.Tensor:
1788	    sequence_rows = prepared.inference_user_rows
1789	    lengths = np.where(
1790	        sequence_rows >= 0, prepared.lengths[np.maximum(sequence_rows, 0)], 0
1791	    )
1792	    ordered = np.argsort(lengths, kind="stable")
1793	    result = torch.zeros(
1794	        len(sequence_rows), config.embedding_dim, dtype=torch.float32
1795	    )
1796	    for group_start in range(0, len(ordered), config.user_batch_size):
1797	        output_rows = ordered[group_start : group_start + config.user_batch_size]
1798	        rows = sequence_rows[output_rows]
1799	        group_lengths = lengths[output_rows]
1800	        hidden = torch.zeros(
1801	            1, len(output_rows), config.embedding_dim, device=device
1802	        )
1803	        for start in range(0, int(group_lengths.max(initial=0)), config.bptt_steps):
1804	            steps = min(
1805	                config.bptt_steps,
1806	                int(group_lengths.max(initial=0)) - start,
1807	            )
1808	            items = np.full(
1809	                (len(output_rows), steps),
1810	                len(prepared.item_ids),
1811	                dtype=np.uint16,
1812	            )
1813	            features = np.zeros((len(output_rows), steps, 3), dtype=np.float16)
1814	            counts = np.zeros(len(output_rows), dtype=np.int32)
1815	            for local_row, sequence_row in enumerate(rows):
1816	                if sequence_row < 0:
1817	                    continue
1818	                count = min(steps, max(int(group_lengths[local_row]) - start, 0))
1819	                if not count:
1820	                    continue
1821	                offset = int(prepared.offsets[sequence_row])
1822	                selected = slice(offset + start, offset + start + count)
1823	                items[local_row, :count] = prepared.sequence_items[selected]
1824	                features[local_row, :count] = prepared.event_features[selected]
1825	                counts[local_row] = count
1826	            outputs, _ = event_sequence(
1827	                model,
1828	                torch.from_numpy(items.astype(np.int64)).to(device),
1829	                torch.from_numpy(features.astype(np.float32)).to(device),
1830	                hidden,
1831	            )
1832	            next_rows: list[torch.Tensor] = []
1833	            for local_row, count in enumerate(counts):
1834	                next_rows.append(
1835	                    outputs[local_row, count - 1]
1836	                    if count
1837	                    else hidden[0, local_row]
1838	                )
1839	            hidden = torch.stack(next_rows).unsqueeze(0)
1840	
1841	        liked_items = torch.from_numpy(
1842	            prepared.inference_liked_items[output_rows].astype(np.int64)
1843	        ).to(device)
1844	        disliked_items = torch.from_numpy(
1845	            prepared.inference_disliked_items[output_rows].astype(np.int64)
1846	        ).to(device)
1847	        queries = user_queries(
1848	            model,
1849	            hidden.squeeze(0),
1850	            model._mean_context_channel(liked_items),
1851	            model._mean_context_channel(disliked_items),
1852	            torch.from_numpy(
1853	                prepared.inference_user_features[output_rows].astype(np.float32)
1854	            ).to(device),
1855	        )
1856	        result[torch.from_numpy(output_rows.astype(np.int64))] = queries.cpu()
1857	    return result
1858	
1859	
1860	@torch.inference_mode()
1861	def rank_candidates(
1862	    model: base.SequentialTwoTower,
1863	    prepared: SequenceData,
1864	    eligibility: CandidateSet,
1865	    config: Config,
1866	    device: torch.device,
1867	    ranking_batch_size: int,
1868	) -> tuple[pd.DataFrame, dict[str, float]]:
1869	    model.eval()
1870	    queries = inference_queries(model, prepared, config, device)
1871	    item_parts: list[torch.Tensor] = []
1872	    for start in range(0, len(prepared.item_ids), 2_048):
1873	        rows = torch.arange(
1874	            start,
1875	            min(start + 2_048, len(prepared.item_ids)),
1876	            dtype=torch.long,
1877	            device=device,
1878	        )
1879	        item_parts.append(model.encode_items(rows).cpu())
1880	    items = torch.cat(item_parts)
1881	    rows: list[tuple[int, int, int]] = []
1882	    for start in range(0, len(eligibility.uids), ranking_batch_size):
1883	        end = min(start + ranking_batch_size, len(eligibility.uids))
1884	        scores = queries[start:end] @ items.T
1885	        eligible = np.unpackbits(
1886	            eligibility.packed[start:end],
1887	            axis=1,
1888	            count=len(prepared.item_ids),
1889	            bitorder="little",
1890	        ).astype(bool, copy=False)
1891	        scores.masked_fill_(~torch.from_numpy(eligible), -torch.inf)
1892	        selected = torch.topk(scores, k=base.TOP_N, dim=1).indices.numpy()
1893	        for local_row, columns in enumerate(selected):
1894	            uid = int(eligibility.uids[start + local_row])
1895	            rows.extend(
1896	                (uid, int(prepared.item_ids[column]), rank)
1897	                for rank, column in enumerate(columns, start=1)
1898	            )
1899	        if end == len(eligibility.uids) or end % 1_024 == 0:
1900	            log(f"ranked users={end:,}/{len(eligibility.uids):,}")
1901	    query_norm = torch.linalg.vector_norm(queries, dim=1).numpy()
1902	    item_norm = torch.linalg.vector_norm(items, dim=1).numpy()
1903	    diagnostics = {
1904	        "query_norm_mean": float(query_norm.mean()),
1905	        "query_norm_std": float(query_norm.std()),
1906	        "item_norm_mean": float(item_norm.mean()),
1907	        "item_norm_std": float(item_norm.std()),
1908	    }
1909	    return pd.DataFrame(rows, columns=["uid", "item_id", "rank"]), diagnostics
1910	
1911	
1912	def write_ranking(frame: pd.DataFrame, output: Path) -> None:
1913	    output.parent.mkdir(parents=True, exist_ok=True)
1914	    if output.suffix == ".parquet":
1915	        frame.to_parquet(output, index=False, compression="zstd")
1916	    elif output.suffix == ".csv":
1917	        frame.to_csv(output, index=False)
1918	    else:
1919	        raise ValueError("--output must end in .parquet or .csv")
1920	
1921	
1922	def main() -> int:
1923	    parser = argparse.ArgumentParser(description=__doc__)
1924	    parser.add_argument(
1925	        "--stage", choices=("validation", "submission"), default="validation"
1926	    )
1927	    parser.add_argument("--output", type=Path, required=True)
1928	    parser.add_argument("--public-dir", type=Path, required=True)
1929	    parser.add_argument("--prepared-cache", type=Path)
1930	    parser.add_argument("--device", choices=("cpu", "mps"), default="cpu")
1931	    parser.add_argument("--threads", type=int, default=8)
1932	    parser.add_argument("--bptt-steps", type=int, default=Config.bptt_steps)
1933	    parser.add_argument(
1934	        "--user-batch-size", type=int, default=Config.user_batch_size
1935	    )
1936	    parser.add_argument("--epochs", type=int, default=Config.epochs)
1937	    parser.add_argument("--learning-rate", type=float, default=Config.learning_rate)
1938	    parser.add_argument(
1939	        "--loss-microbatch-size",
1940	        type=int,
1941	        default=Config.loss_microbatch_size,
1942	    )
1943	    parser.add_argument(
1944	        "--negative-filter-scope",
1945	        choices=("full-training-history", "last-100"),
1946	        default=Config.negative_filter_scope,
1947	    )
1948	    parser.add_argument(
1949	        "--target-weighting",
1950	        choices=("discovery-graded", "organic-strong", "uniform"),
1951	        default=Config.target_weighting,
1952	    )
1953	    parser.add_argument(
1954	        "--ordinary-target-weight",
1955	        type=float,
1956	        default=Config.ordinary_target_weight,
1957	    )
1958	    parser.add_argument(
1959	        "--organic-strong-target-weight",
1960	        type=float,
1961	        default=Config.organic_strong_target_weight,
1962	    )
1963	    parser.add_argument(
1964	        "--organic-like-target-weight",
1965	        type=float,
1966	        default=Config.organic_like_target_weight,
1967	    )
1968	    parser.add_argument(
1969	        "--discovery-horizon-days",
1970	        type=int,
1971	        default=Config.discovery_horizon_days,
1972	    )
1973	    parser.add_argument(
1974	        "--validation-fraction",
1975	        type=float,
1976	        default=Config.validation_fraction,
1977	    )
1978	    parser.add_argument(
1979	        "--max-train-targets-per-user",
1980	        type=int,
1981	        default=Config.max_train_targets_per_user,
1982	        help="retain this many recent pre-validation targets per user; 0 is uncapped",
1983	    )
1984	    parser.add_argument(
1985	        "--max-validation-targets-per-user",
1986	        type=int,
1987	        default=Config.max_validation_targets_per_user,
1988	        help="retain this many final validation targets per user; 0 is uncapped",
1989	    )
1990	    parser.add_argument(
1991	        "--early-stopping-patience",
1992	        type=int,
1993	        default=Config.early_stopping_patience,
1994	    )
1995	    parser.add_argument(
1996	        "--early-stopping-min-delta",
1997	        type=float,
1998	        default=Config.early_stopping_min_delta,
1999	    )
2000	    parser.add_argument("--checkpoint-dir", type=Path)
2001	    parser.add_argument("--resume-checkpoint", type=Path)
2002	    parser.add_argument("--seed", type=int, default=Config.seed)
2003	    parser.add_argument("--ranking-batch-size", type=int, default=64)
2004	    parser.add_argument("--prepare-only", action="store_true")
2005	    args = parser.parse_args()
2006	
2007	    if args.threads < 1 or args.bptt_steps < 2 or args.user_batch_size < 1:
2008	        parser.error("threads, BPTT steps, and user batch size must be positive")
2009	    if (
2010	        args.epochs < 1
2011	        or args.learning_rate <= 0
2012	        or args.loss_microbatch_size < 1
2013	        or args.early_stopping_patience < 1
2014	        or args.early_stopping_min_delta < 0
2015	        or args.ordinary_target_weight <= 0
2016	        or args.organic_strong_target_weight <= 0
2017	        or args.organic_like_target_weight <= 0
2018	        or args.discovery_horizon_days < 1
2019	        or args.max_train_targets_per_user < 0
2020	        or args.max_validation_targets_per_user < 0
2021	    ):
2022	        parser.error("training counts, learning rate, and patience are invalid")
2023	    if not 0.0 < args.validation_fraction < 1.0:
2024	        parser.error("validation fraction must be strictly between zero and one")
2025	    public = args.public_dir.resolve()
2026	    if not (public / "data" / "candidate_catalog.parquet").is_file():
2027	        parser.error(f"public task data not found under {public}")
2028	    config = Config(
2029	        bptt_steps=args.bptt_steps,
2030	        user_batch_size=args.user_batch_size,
2031	        epochs=args.epochs,
2032	        learning_rate=args.learning_rate,
2033	        loss_microbatch_size=args.loss_microbatch_size,
2034	        negative_filter_scope=args.negative_filter_scope,
2035	        target_weighting=args.target_weighting,
2036	        ordinary_target_weight=args.ordinary_target_weight,
2037	        organic_strong_target_weight=args.organic_strong_target_weight,
2038	        organic_like_target_weight=args.organic_like_target_weight,
2039	        discovery_horizon_days=args.discovery_horizon_days,
2040	        validation_fraction=args.validation_fraction,
2041	        max_train_targets_per_user=args.max_train_targets_per_user,
2042	        max_validation_targets_per_user=args.max_validation_targets_per_user,
2043	        early_stopping_patience=args.early_stopping_patience,
2044	        early_stopping_min_delta=args.early_stopping_min_delta,
2045	        seed=args.seed,
2046	    )
2047	    torch.set_num_threads(args.threads)
2048	    torch.set_num_interop_threads(1)
2049	    seed_everything(config.seed)
2050	    device = torch.device(args.device)
2051	    if device.type == "mps" and not torch.backends.mps.is_available():
2052	        parser.error("MPS was requested but is unavailable")
2053	
2054	    started = time.monotonic()
2055	    prepared, eligibility = load_or_prepare(
2056	        public,
2057	        args.stage,
2058	        config,
2059	        args.prepared_cache.resolve() if args.prepared_cache else None,
2060	    )
2061	    target_partition = chronological_target_partition(
2062	        prepared,
2063	        config.validation_fraction,
2064	        config.max_train_targets_per_user,
2065	        config.max_validation_targets_per_user,
2066	    )
2067	    if args.prepare_only:
2068	        print(
2069	            json.dumps(
2070	                {
2071	                    "stage": args.stage,
2072	                    "sequence_users": len(prepared.user_ids),
2073	                    "sequence_tokens": len(prepared.sequence_items),
2074	                    "next_item_targets": prepared.target_count,
2075	                    "selected_training_targets": int(
2076	                        target_partition.training_counts.sum()
2077	                    ),
2078	                    "selected_validation_targets": int(
2079	                        target_partition.validation_counts.sum()
2080	                    ),
2081	                    "discarded_old_targets": int(
2082	                        target_partition.window_starts.sum()
2083	                    ),
2084	                    "target_users": len(eligibility.uids),
2085	                    "runtime_seconds": round(time.monotonic() - started, 3),
2086	                    "config": asdict(config),
2087	                },
2088	                indent=2,
2089	            )
2090	        )
2091	        return 0
2092	
2093	    model, training_history = train_model(
2094	        public,
2095	        args.stage,
2096	        prepared,
2097	        config,
2098	        device,
2099	        args.checkpoint_dir.resolve() if args.checkpoint_dir else None,
2100	        args.resume_checkpoint.resolve() if args.resume_checkpoint else None,
2101	    )
2102	    ranking, embedding_diagnostics = rank_candidates(
2103	        model,
2104	        prepared,
2105	        eligibility,
2106	        config,
2107	        device,
2108	        args.ranking_batch_size,
2109	    )
2110	    write_ranking(ranking, args.output.resolve())
2111	    print(
2112	        json.dumps(
2113	            {
2114	                "baseline": "yambda-windowed-sequence-two-tower-base",
2115	                "stage": args.stage,
2116	                "output": str(args.output.resolve()),
2117	                "rows": len(ranking),
2118	                "users": int(ranking.uid.nunique()),
2119	                "runtime_seconds": round(time.monotonic() - started, 3),
2120	                "public_inputs_only": True,
2121	                "architecture": {
2122	                    "family": "pure-windowed-sequence-two-tower",
2123	                    "sequence_training": (
2124	                        "every-adjacent-target-within-recent-chronological-window"
2125	                    ),
2126	                    "sequence_encoder": "one-layer-GRU-dense-timestep-loss",
2127	                    "bptt_hidden_state": "carried-and-detached",
2128	                    "model_selection": "chronological-validation-BCE-early-stopping",
2129	                    "loss_execution": "gradient-accumulated-microbatches",
2130	                    "user_id_embedding": False,
2131	                    "user_mlp": "582-512-256-192",
2132	                    "item_mlp": "196-512-256-192",
2133	                    "normalization": "none",
2134	                    "score": "raw-dot-product",
2135	                    "objective": "sample-weighted-BCE-at-every-timestep",
2136	                    "negative_sampling": (
2137	                        "24-uniform-plus-8-propensity-corrected-inbatch"
2138	                    ),
2139	                    "false_negative_filter": config.negative_filter_scope,
2140	                    "target_weighting": config.target_weighting,
2141	                },
2142	                "config": asdict(config),
2143	                "sequence_tokens": len(prepared.sequence_items),
2144	                "all_next_item_targets": prepared.target_count,
2145	                "selected_training_targets": int(
2146	                    target_partition.training_counts.sum()
2147	                ),
2148	                "selected_validation_targets": int(
2149	                    target_partition.validation_counts.sum()
2150	                ),
2151	                "discarded_old_targets": int(target_partition.window_starts.sum()),
2152	                "training_history": training_history,
2153	                "embedding_diagnostics": embedding_diagnostics,
2154	                "torch_version": torch.__version__,
2155	                "device": str(device),
2156	            },
2157	            indent=2,
2158	        )
2159	    )
2160	    return 0
2161	
2162	
2163	if __name__ == "__main__":
2164	    raise SystemExit(main())
2165	

loaded explicit events: rows=720,270 users=8,220
prepared users=1,000 tokens=2,090,746 states=35,727 elapsed=4.9s
#!/usr/bin/env python3
"""Score a validation-stage ranking against public discovery labels.

``build_task.py`` publishes this module with the solver-visible task files. It
loads target users, validation eligibility, and validation-window events from
the adjacent ``data`` directory, then evaluates one ranking supplied on the
command line. Stage-independent mechanics come from the adjacent public
``ranking_evaluation.py`` module. The builder installs the same checked-in
module in the private judge, so public and private scoring mechanics cannot
drift. A relevant candidate receives grade 3 for an organic like or grade 1 for
an organic listen with at least 80 percent completion; a like takes precedence
when both events exist.

Before scoring, the ranking must contain exactly 100 distinct eligible items
for every target user, with integer-valued ``uid``, ``item_id``, and ``rank``
columns and ranks 1 through 100 exactly once per user. Extra columns are
ignored. Quality is graded NDCG@10, averaged only over users who have at least
one eligible relevant item in the validation window.

The evaluator prints a JSON object containing the model path and scores. It
does not apply the task's pass threshold; its purpose is model development and
selection. The private judge applies the threshold to the final submission.

Usage::

    python evaluate_public.py <validation-ranking.parquet|csv>
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

import pandas as pd

from ranking_evaluation import (
    CandidateSet,
    RankingValidationError,
    ranked_items,
    read_ranking,
    score_rankings,
    validate_ranking,
)


HERE = Path(__file__).resolve().parent
DATA = HERE / "data"
LISTEN_STRONG = 80


def validation_truth(
    eligibility: CandidateSet, targets: set[int]
) -> dict[int, dict[int, int]]:
    """Build cutoff-correct graded relevance from public validation events.

    Only target-user events for eligible candidates are retained. Organic
    listens completed to at least 80 percent receive grade 1, and organic likes
    receive grade 3. Repeated user-item events are collapsed, and processing
    likes second ensures that grade 3 wins over grade 1.

loaded explicit events: rows=720,270 users=8,220
prepared users=1,000 tokens=2,090,746 states=35,727 elapsed=4.9s
prepared users=2,000 tokens=4,301,266 states=71,242 elapsed=9.9s
prepared users=3,000 tokens=6,486,146 states=109,471 elapsed=15.1s
prepared users=4,000 tokens=8,793,043 states=148,210 elapsed=20.5s
prepared users=5,000 tokens=10,992,205 states=182,596 elapsed=25.3s
prepared users=6,000 tokens=13,061,727 states=217,148 elapsed=30.1s
prepared users=7,000 tokens=15,334,418 states=256,914 elapsed=35.6s
prepared users=8,000 tokens=17,462,672 states=296,703 elapsed=40.8s
prepared users=9,000 tokens=19,568,405 states=331,551 elapsed=45.6s
target users without qualifying catalog sequence=7
prepared stage=validation sequence_users=9,135 tokens=20,056,329 next_item_targets=20,047,194 states=339,133
chronological BCE window: train=1,052,953 validation=130,103 selected=1,183,056 discarded_old_targets=18,864,138 all=20,047,194
inbatch correction: min=0.250 mean_on_targets=1.001 max=4.000
target weighting=organic-strong min=0.250 mean=0.575 max=1.000
full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250

1364	def evaluate_validation_bce(
1365	    model: base.SequentialTwoTower,
1366	    prepared: SequenceData,
1367	    config: Config,
1368	    target_weights: np.ndarray,
1369	    window_starts: np.ndarray,
1370	    training_counts: np.ndarray,
1371	    validation_counts: np.ndarray,
1372	    item_correction: torch.Tensor,
1373	    known_positive_packed: np.ndarray | None,
1374	    device: torch.device,
1375	) -> dict[str, float]:
1376	    """Measure deterministic BCE on each user's chronological target suffix."""
1377	    model.eval()
1378	    totals = training_counts + validation_counts
1379	    users_with_validation = np.flatnonzero(validation_counts > 0)
1380	    ordered = users_with_validation[
1381	        np.argsort(totals[users_with_validation], kind="stable")
1382	    ]
1383	    groups = [
1384	        ordered[start : start + config.user_batch_size]
1385	        for start in range(0, len(ordered), config.user_batch_size)
1386	    ]
1387	    generator = torch.Generator(device="cpu").manual_seed(config.seed + 20_000)
1388	    total_loss = 0.0
1389	    total_targets = 0
1390	    total_weight = 0.0
1391	    started = time.monotonic()
1392	
1393	    for group_number, users in enumerate(groups, start=1):
1394	        group_totals = totals[users]
1395	        group_starts = training_counts[users]
1396	        hidden = torch.zeros(
1397	            1, len(users), config.embedding_dim, device=device
1398	        )
1399	        max_targets = int(group_totals.max())
1400	        for start in range(0, max_targets, config.bptt_steps):
1401	            steps = min(config.bptt_steps, max_targets - start)
1402	            block = build_dense_block(
1403	                prepared,
1404	                target_weights,
1405	                users,
1406	                window_starts[users],
1407	                group_totals,
1408	                group_starts,
1409	                start,
1410	                steps,
1411	                config,
1412	            )
1413	            sequence_outputs, _ = event_sequence(
1414	                model,
1415	                torch.from_numpy(block[0].astype(np.int64)).to(device),
1416	                torch.from_numpy(block[1].astype(np.float32)).to(device),
1417	                hidden,
1418	            )
1419	            hidden = torch.stack(
1420	                [
1421	                    sequence_outputs[row, count - 1]
1422	                    if count
1423	                    else hidden[0, row]
1424	                    for row, count in enumerate(block[7])
1425	                ]
1426	            ).unsqueeze(0)
1427	            flat_scored = block[5].reshape(-1)
1428	            if not np.any(flat_scored):
1429	                continue
1430	            mask = torch.from_numpy(flat_scored).to(device)
1431	            sequence_vectors = sequence_outputs.reshape(
1432	                -1, config.embedding_dim
1433	            )[mask]
1434	            flat_states = block[3].reshape(-1)[flat_scored]
1435	            positive_items = torch.from_numpy(
1436	                block[4].reshape(-1)[flat_scored].astype(np.int64)
1437	            ).to(device)
1438	            known_history = torch.from_numpy(
1439	                block[6].reshape(-1, config.bptt_steps)[flat_scored].astype(
1440	                    np.int64
1441	                )
1442	            ).to(device)
1443	            liked_items = torch.from_numpy(
1444	                prepared.liked_states[flat_states].astype(np.int64)
1445	            ).to(device)
1446	            flat_user_rows = np.broadcast_to(
1447	                users[:, None], block[5].shape
1448	            ).reshape(-1)[flat_scored]
1449	            negatives, negative_weights = draw_training_negatives(
1450	                positive_items,
1451	                known_history,
1452	                liked_items,
1453	                flat_user_rows,
1454	                known_positive_packed,
1455	                prepared,
1456	                config,
1457	                item_correction,
1458	                generator,
1459	                device,
1460	            )
1461	            statistics, _ = score_target_microbatches(
1462	                model,
1463	                prepared,
1464	                config,
1465	                sequence_vectors,
1466	                flat_states,
1467	                block[2].reshape(-1, 4)[flat_scored].astype(np.float32),
1468	                positive_items,
1469	                negatives,
1470	                negative_weights,
1471	                block[8].reshape(-1)[flat_scored],
1472	                device,
1473	                backward=False,
1474	            )
1475	            total_loss += statistics["loss_sum"]
1476	            total_targets += len(positive_items)
1477	            total_weight += statistics["weight_sum"]
1478	        if group_number % 25 == 0:
1479	            release_accelerator_cache(device)
1480	    return {
1481	        "validation_loss": total_loss / max(total_weight, 1.0e-12),
1482	        "validation_targets": float(total_targets),
1483	        "validation_target_weight_sum": total_weight,
1484	        "validation_seconds": time.monotonic() - started,
1485	    }
1486	
1487	
1488	def snapshot_model(model: nn.Module) -> dict[str, torch.Tensor]:

prepared stage=validation sequence_users=9,135 tokens=20,056,329 next_item_targets=20,047,194 states=339,133
chronological BCE window: train=1,052,953 validation=130,103 selected=1,183,056 discarded_old_targets=18,864,138 all=20,047,194
inbatch correction: min=0.250 mean_on_targets=1.001 max=4.000
target weighting=organic-strong min=0.250 mean=0.575 max=1.000
full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250

Traceback (most recent call last):
  File "<stdin>", line 151, in <module>
PermissionError: [Errno 13] Permission denied: 'train_current_model.py'
1198:def release_accelerator_cache(device: torch.device) -> None:

total 112
drwxr-xr-x 1 agent agent  4096 Sep 26 07:18 .
drwxr-xr-x 1 agent agent  4096 Sep 26 07:12 ..
-rw-r--r-- 1 agent agent     0 Sep 26 07:16 base_time.log
-rw-r--r-- 1 agent agent  1159 Sep 26 07:17 base_val.log
-rw-r--r-- 1 agent agent 80058 Sep 26 07:18 train_current_model.py
-rw-r--r-- 1 agent agent 15985 Sep 26 07:18 two_tower_model.py

62:    negative_strategy: str = "catalog-softmax"
1335:def release_accelerator_cache(

            known_history = torch.from_numpy(
                block[6].reshape(-1, config.bptt_steps)[flat_scored].astype(
                    np.int64
                )
            ).to(device)
            liked_items = torch.from_numpy(
                prepared.liked_states[flat_states].astype(np.int64)
            ).to(device)
            flat_user_rows = np.broadcast_to(
                users[:, None], block[5].shape
            ).reshape(-1)[flat_scored]
            optimizer.zero_grad(set_to_none=True)
            if config.negative_strategy == "catalog-softmax":
                statistics, sequence_gradient = score_catalog_softmax(
                    model,
                    prepared,
                    config,
                    sequence_vectors,
                    flat_states,
                    block[2].reshape(-1, 4)[flat_scored].astype(np.float32),
                    positive_items,
                    flat_user_rows,
                    known_positive_packed,
                    block[8].reshape(-1)[flat_scored],
                    device,
                    backward=True,
                )
            else:
                negatives, negative_weights = draw_training_negatives(
                    positive_items,
                    known_history,
                    liked_items,
                    flat_user_rows,
                    known_positive_packed,
                    prepared,
                    config,
                    item_correction,
                    generator,
                    device,
                )
                statistics, sequence_gradient = score_target_microbatches(
                    model,
                    prepared,
                    config,
                    sequence_vectors,
                    flat_states,
                    block[2].reshape(-1, 4)[flat_scored].astype(np.float32),
                    positive_items,
                    negatives,
                    negative_weights,
                    block[8].reshape(-1)[flat_scored],
                    device,
                    backward=True,
                )
            if sequence_gradient is None:
                raise RuntimeError("training block omitted its sequence gradient")
            sequence_vectors.backward(sequence_gradient)
            nn.utils.clip_grad_norm_(model.parameters(), config.gradient_clip)
            optimizer.step()

            count = len(positive_items)
            total_loss += statistics["loss_sum"]
            total_targets += count
            total_weight += statistics["weight_sum"]
            total_positive += statistics["positive_sum"]
            total_negative += statistics["negative_sum"]
            total_pair_correct += statistics["pair_correct"]
            batches += 1
            if batches % 25 == 0:
                release_accelerator_cache(device)
            if batches % 250 == 0:
                batch_bce = statistics["loss_sum"] / max(
                    statistics["weight_sum"], 1.0e-12
                )
                running_bce = total_loss / max(total_weight, 1.0e-12)
                log(
                    f"epoch={epoch}/{config.epochs} batches={batches:,} "
                    f"targets={total_targets:,} "
                    f"batch_bce={batch_bce:.6f} "
                    f"running_bce={running_bce:.6f} "
                    f"elapsed={time.monotonic() - started:.1f}s"

inbatch correction: min=0.250 mean_on_targets=1.001 max=4.000
target weighting=organic-strong min=0.250 mean=0.575 max=1.000
full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250

Sat Sep 26 07:19:37 UTC 2026
agent       8128  0.0  0.0   4492  1996 ?        S    07:16   0:00 /bin/bash -c source /logs/agent/sessions/shell-snapshots/snapshot-bash-1790407017393-1cbxxj.sh 2>/dev/null || true && shopt -u extglob 2>/dev/null || true && { \builtin unalias -- 'unsetenv'; \builtin unset -f -- 'unsetenv'; } >/dev/null 2>&1 || true && eval 'mkdir -p /app/solution && cd /app/solution && (time python /task/current_model/train_current_model.py --stage validation --public-dir /task --output /app/solution/production_validation.parquet > base_val.log 2>&1; python /task/evaluate_public.py /app/solution/production_validation.parquet >> base_val.log 2>&1) > base_time.log 2>&1 & echo started' < /dev/null && pwd -P >| /tmp/claude-53a1-cwd
agent       8130  0.0  0.0   4492  1972 ?        S    07:16   0:00 /bin/bash -c source /logs/agent/sessions/shell-snapshots/snapshot-bash-1790407017393-1cbxxj.sh 2>/dev/null || true && shopt -u extglob 2>/dev/null || true && { \builtin unalias -- 'unsetenv'; \builtin unset -f -- 'unsetenv'; } >/dev/null 2>&1 || true && eval 'mkdir -p /app/solution && cd /app/solution && (time python /task/current_model/train_current_model.py --stage validation --public-dir /task --output /app/solution/production_validation.parquet > base_val.log 2>&1; python /task/evaluate_public.py /app/solution/production_validation.parquet >> base_val.log 2>&1) > base_time.log 2>&1 & echo started' < /dev/null && pwd -P >| /tmp/claude-53a1-cwd
agent       8131  381 27.6 6892168 4422864 ?     Rl   07:16  10:00 python /task/current_model/train_current_model.py --stage validation --public-dir /task --output /app/solution/production_validation.parquet


def cache_metadata(stage: str, config: Config) -> dict[str, object]:
    return {
        "cache_version": CACHE_VERSION,
        "stage": stage,
        "positive_listen_threshold": base.LISTEN_POSITIVE_THRESHOLD,
        "profile_lookback_days": base.PROFILE_LOOKBACK_DAYS,
        "event_time_feature": "exp-negative-gap-over-60-days",
    }


def save_cache(
    path: Path, prepared: SequenceData, stage: str, config: Config
) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    np.savez(
        path,
        metadata=np.asarray(json.dumps(cache_metadata(stage, config), sort_keys=True)),
        **prepared.__dict__,
    )
    log(f"saved public-derived full-sequence cache: {path}")


def load_cache(path: Path, stage: str, config: Config) -> SequenceData:
    with np.load(path, allow_pickle=False) as artifact:
        actual = json.loads(str(artifact["metadata"].item()))
        expected = cache_metadata(stage, config)
        if actual != expected:
            raise RuntimeError(f"sequence cache differs: {actual} != {expected}")
        values = {
            field: artifact[field].copy()
            for field in SequenceData.__dataclass_fields__
        }
    prepared = SequenceData(**values)
    log(
        f"loaded full-sequence cache: tokens={len(prepared.sequence_items):,} "
        f"targets={prepared.target_count:,}"
    )
    return prepared


def load_or_prepare(
    public: Path,
    stage: str,
    config: Config,
    cache: Path | None,
) -> tuple[SequenceData, CandidateSet]:
    candidate_name = (
        "validation_eligible_candidates.npz"
        if stage == "validation"
        else "test_eligible_candidates.npz"
    )
    eligibility = CandidateSet.read(public / "data" / candidate_name)
    if cache is not None and cache.is_file():
        prepared = load_cache(cache, stage, config)
        if not np.array_equal(prepared.item_ids, eligibility.item_ids):
            raise RuntimeError("cached catalog differs from candidate artifact")
        return prepared, eligibility
    prepared, built_eligibility = prepare_data(public, stage, config)
    if cache is not None:
        save_cache(cache, prepared, stage, config)

ok

  File "/opt/task-venv/lib/python3.12/site-packages/pandas/io/common.py", line 882, in get_handle
    handle = open(handle, ioargs.mode)
             ^^^^^^^^^^^^^^^^^^^^^^^^^
FileNotFoundError: [Errno 2] No such file or directory: '/app/solution/production_validation.parquet'
saved public-derived full-sequence cache: /app/solution/cache_validation.npz
chronological BCE window: train=1,052,953 validation=130,103 selected=1,183,056 discarded_old_targets=18,864,138 all=20,047,194
inbatch correction: min=0.250 mean_on_targets=1.001 max=4.000
target weighting=organic-strong min=0.250 mean=0.575 max=1.000
full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250

prepared users=3,000 tokens=6,486,146 states=109,471 elapsed=15.1s
prepared users=4,000 tokens=8,793,043 states=148,210 elapsed=20.5s
prepared users=5,000 tokens=10,992,205 states=182,596 elapsed=25.3s
prepared users=6,000 tokens=13,061,727 states=217,148 elapsed=30.1s
prepared users=7,000 tokens=15,334,418 states=256,914 elapsed=35.6s
prepared users=8,000 tokens=17,462,672 states=296,703 elapsed=40.8s
prepared users=9,000 tokens=19,568,405 states=331,551 elapsed=45.6s
target users without qualifying catalog sequence=7
prepared stage=validation sequence_users=9,135 tokens=20,056,329 next_item_targets=20,047,194 states=339,133
chronological BCE window: train=1,052,953 validation=130,103 selected=1,183,056 discarded_old_targets=18,864,138 all=20,047,194
inbatch correction: min=0.250 mean_on_targets=1.001 max=4.000
target weighting=organic-strong min=0.250 mean=0.575 max=1.000
full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250
Traceback (most recent call last):
FileNotFoundError: [Errno 2] No such file or directory: '/app/solution/production_validation.parquet'
/bin/bash: line 2:  8131 Killed                  python /task/current_model/train_current_model.py --stage validation --public-dir /task --output /app/solution/production_validation.parquet > base_val.log 2>&1

real	4m12.442s
user	18m23.733s
sys	0m57.718s

               total        used        free      shared  buff/cache   available
Mem:           15614        1406        8237           2        6322       14207
Swap:              0           0           0
PID %MEM 0 MB COMMAND  
8057 1.2 188.094 MB claude --verbose --output-format=stream-json
8343 0.0 4.57812 MB ps aux --sort=-rss

loaded explicit events: rows=720,270 users=8,220
prepared users=1,000 tokens=2,090,746 states=35,727 elapsed=7.0s
prepared users=2,000 tokens=4,301,266 states=71,242 elapsed=14.7s
prepared users=3,000 tokens=6,486,146 states=109,471 elapsed=22.1s
prepared users=4,000 tokens=8,793,043 states=148,210 elapsed=29.9s
prepared users=5,000 tokens=10,992,205 states=182,596 elapsed=36.6s
prepared users=6,000 tokens=13,061,727 states=217,148 elapsed=43.5s
prepared users=7,000 tokens=15,334,418 states=256,914 elapsed=51.8s
prepared users=8,000 tokens=17,462,672 states=296,703 elapsed=59.6s
prepared users=9,000 tokens=19,568,405 states=331,551 elapsed=66.3s
target users without qualifying catalog sequence=7
prepared stage=validation sequence_users=9,135 tokens=20,056,329 next_item_targets=20,047,194 states=339,133
saved public-derived full-sequence cache: /app/solution/cache_validation.npz
chronological BCE window: train=1,052,953 validation=130,103 selected=1,183,056 discarded_old_targets=18,864,138 all=20,047,194
inbatch correction: min=0.250 mean_on_targets=1.001 max=4.000
target weighting=organic-strong min=0.250 mean=0.575 max=1.000
full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250

inbatch correction: min=0.250 mean_on_targets=1.001 max=4.000
target weighting=organic-strong min=0.250 mean=0.575 max=1.000
full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250
               total        used        free      shared  buff/cache   available
Mem:           15614        4046        5079           2        6841       11567

inbatch correction: min=0.250 mean_on_targets=1.001 max=4.000
target weighting=organic-strong min=0.250 mean=0.575 max=1.000
full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250
               total        used        free      shared  buff/cache   available
Mem:           15614        3793        5331           2        6841       11820

full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250
epoch=1/3 batches=50 targets=206,457 batch_bce=9.271569 running_bce=9.567764 elapsed=118.4s
Sat Sep 26 07:23:59 UTC 2026

Exit code 144

base_time.log
base_val.log
cache_validation.npz
patch1.py
patch1_fn.txt
patch2.py
patch3.py
sm_val.log
train_current_model.py
two_tower_model.py
3

(Bash completed with no output)

    "seed": 29
  }
loaded explicit events: rows=746,504 users=8,307
target users without qualifying catalog sequence=6
prepared stage=submission sequence_users=9,153 tokens=20,712,971 next_item_targets=20,703,818 states=350,322
saved public-derived full-sequence cache: /app/solution/cache_submission.npz
{
  "stage": "submission",
  "sequence_users": 9153,
  "sequence_tokens": 20712971,

Sat Sep 26 07:25:29 UTC 2026
tail: option used in invalid context -- 1
               total        used        free      shared  buff/cache   available
Mem:           15614        6722        1868           2        7376        8891

full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250
full-history false-negative filter: users=9,153 unique_user_items=5,682,910 bytes=34,323,750
Sat Sep 26 07:28:00 UTC 2026

epoch=1/1 batches=50 targets=206,457 batch_bce=9.271569 running_bce=9.567764 elapsed=185.3s
epoch=1/1 batches=50 targets=196,561 batch_bce=9.393283 running_bce=9.634487 elapsed=177.6s
Sat Sep 26 07:30:03 UTC 2026

(Bash completed with no output)

The file /app/solution/train_current_model.py has been updated successfully.

ok

epoch=1/1 batches=150 targets=596,745 batch_bce=9.298708 running_bce=9.427523 elapsed=543.7s
epoch=1/1 batches=150 targets=597,467 batch_bce=9.308032 running_bce=9.478798 elapsed=538.3s
Sat Sep 26 07:37:16 UTC 2026

  "device": "cpu"
}
  "device": "cpu"
}
Sat Sep 26 07:45:21 UTC 2026

{
  "model": "softmax_validation.parquet",
  "scores": {
    "quality_ndcg@10": 0.018876,
    "quality_users": 4263
  }
}
sm_sub.log:epoch=1/1 train_loss=9.34429 validation_loss=9.17396 pair_accuracy=0.788 train_targets=1,067,606 validation_targets=131,997 train_seconds=973.4 validation_seconds=82.0
sm_sub.log:restored best validation-BCE checkpoint: epoch=1 loss=9.173960
sm_val.log:epoch=1/1 train_loss=9.34329 validation_loss=9.10110 pair_accuracy=0.790 train_targets=1,052,953 validation_targets=130,103 train_seconds=960.0 validation_seconds=81.0
sm_val.log:restored best validation-BCE checkpoint: epoch=1 loss=9.101095

Command did not complete within its 120s timeout and was moved to the background (ID: bak7831ox). Output is being written to: /tmp/claude-1000/-app/64d33cf1-e0cd-4b7e-8d35-24b8783f6856/tasks/bak7831ox.output. You will be notified when it completes. To check interim output, use Read on that file path.
Session cwd remains /app/solution; directory changes made by the backgrounded command do not apply to subsequent commands.

full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250
Sat Sep 26 07:48:35 UTC 2026

full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250
Sat Sep 26 07:50:18 UTC 2026

target weighting=organic-strong min=0.250 mean=0.575 max=1.000
full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250
Sat Sep 26 07:52:21 UTC 2026

PID %CPU %MEM TIME COMMAND  
8914 608 23.5 42:30 python /task/current_model/train_current_model.py --stage
8962 94.7 1.5 6:29 python - 
8057 0.5 1.3 0:12 claude --verbose --output-format=stream-json

full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250
epoch=1/1 batches=250 targets=987,417 batch_bce=0.547529 running_bce=0.541614 elapsed=485.4s
Sat Sep 26 07:53:59 UTC 2026

epoch=1/1 batches=250 targets=987,417 batch_bce=0.547529 running_bce=0.541614 elapsed=485.4s
epoch=1/1 train_loss=0.53984 validation_loss=0.49664 pair_accuracy=0.755 train_targets=1,052,953 validation_targets=130,103 train_seconds=498.3 validation_seconds=18.6
restored best validation-BCE checkpoint: epoch=1 loss=0.496636

ls: cannot access 'production_validation.parquet': No such file or directory
OK submission.parquet: users=8456 rows=845600
Sat Sep 26 07:55:14 UTC 2026

{
  "model": "production_validation.parquet",
  "scores": {
    "quality_ndcg@10": 0.02128,
    "quality_users": 4263
  }
}

Sat Sep 26 07:55:54 UTC 2026

65:    negative_strategy: str = "mixed-bce"

The file /app/solution/train_current_model.py has been updated successfully. (note: the file had been modified on disk since you last read it — the edit applied cleanly, but the file contains other changes not in your context. Read it before edits that depend on surrounding content.)

File created successfully at: /app/solution/README.md (file state is current in your context — no need to Read it back)

restored best validation-BCE checkpoint: epoch=1 loss=0.497949
Sat Sep 26 08:00:13 UTC 2026

Exit code 2
restored best validation-BCE checkpoint: epoch=1 loss=0.497949
ls: cannot access 'production_submission.parquet': No such file or directory

Exit code 2
epoch=1/1 train_loss=0.53835 validation_loss=0.49795 pair_accuracy=0.755 train_targets=1,067,606 validation_targets=131,997 train_seconds=221.9 validation_seconds=15.7
restored best validation-BCE checkpoint: epoch=1 loss=0.497949
ls: cannot access 'production_submission.parquet': No such file or directory

}
-rw-r--r-- 1 agent agent 1212474 Sep 26 08:01 production_submission.parquet
Sat Sep 26 08:02:04 UTC 2026

OK production_submission.parquet: users=8456 rows=845600
OK submission.parquet: users=8456 rows=845600
total 1068532
drwxr-xr-x 1 agent agent      4096 Sep 26 08:01 .
drwxr-xr-x 1 agent agent      4096 Sep 26 07:12 ..
-rw-r--r-- 1 agent agent      2597 Sep 26 07:56 README.md
-rw-r--r-- 1 agent agent      4388 Sep 26 08:01 base_sub.log
-rw-r--r-- 1 agent agent       256 Sep 26 07:21 base_time.log
-rw-r--r-- 1 agent agent      3941 Sep 26 07:55 base_val.log
-rw-r--r-- 1 agent agent 552686232 Sep 26 07:25 cache_submission.npz
-rw-r--r-- 1 agent agent 535225808 Sep 26 07:21 cache_validation.npz
-rw-r--r-- 1 agent agent       940 Sep 26 07:54 check_submission.py
-rw-r--r-- 1 agent agent      2250 Sep 26 07:25 prep_sub.log
-rw-r--r-- 1 agent agent   1212474 Sep 26 08:01 production_submission.parquet
-rw-r--r-- 1 agent agent   1244229 Sep 26 07:55 production_validation.parquet
-rw-r--r-- 1 agent agent      4394 Sep 26 07:45 sm_sub.log
-rw-r--r-- 1 agent agent      4391 Sep 26 07:45 sm_val.log
-rw-r--r-- 1 agent agent   1176020 Sep 26 07:45 softmax_submission.parquet
-rw-r--r-- 1 agent agent   1226922 Sep 26 07:45 softmax_validation.parquet
-rw-r--r-- 1 agent agent   1212474 Sep 26 08:02 submission.parquet
-rw-r--r-- 1 agent agent     88598 Sep 26 07:56 train_current_model.py
-rw-r--r-- 1 agent agent     15985 Sep 26 07:18 two_tower_model.py