Agent trace

/app:
total 12
drwxr-xr-x 3 agent agent 4096 Oct  3 08:59 .
drwxr-xr-x 1 root  root  4096 Oct  3 09:02 ..
drwxr-xr-x 2 agent agent 4096 Oct  3 08:59 solution

/app/solution:
total 8
drwxr-xr-x 2 agent agent 4096 Oct  3 08:59 .
drwxr-xr-x 3 agent agent 4096 Oct  3 08:59 ..

/task:
total 40
dr-xr-xr-x 1 root root 4096 Oct  3 08:59 .
drwxr-xr-x 1 root root 4096 Oct  3 09:02 ..
-r--r--r-- 1 root root    0 Oct  3 08:57 .keep
dr-xr-xr-x 1 root root 4096 Oct  3 08:57 current_model
dr-xr-xr-x 1 root root 4096 Oct  3 08:58 data
-r--r--r-- 1 root root 5774 Oct  3 08:57 evaluate_public.py
-r--r--r-- 1 root root 8265 Oct  3 08:57 ranking_evaluation.py
-r--r--r-- 1 root root 1295 Oct  3 08:57 requirements.txt

/task/current_model:
total 104
dr-xr-xr-x 1 root root  4096 Oct  3 08:57 .
dr-xr-xr-x 1 root root  4096 Oct  3 08:59 ..
-r--r--r-- 1 root root 80058 Oct  3 08:57 train_current_model.py
-r--r--r-- 1 root root 15985 Oct  3 08:57 two_tower_model.py

/task/data:
total 347392
dr-xr-xr-x 1 root root      4096 Oct  3 08:58 .
dr-xr-xr-x 1 root root      4096 Oct  3 08:59 ..
-r--r--r-- 1 root root    349352 Oct  3 08:57 candidate_catalog.parquet
-r--r--r-- 1 root root     84127 Oct  3 08:57 target_users.parquet
-r--r--r-- 1 root root   7736341 Oct  3 08:57 test_eligible_candidates.npz
-r--r--r-- 1 root root   1092329 Oct  3 08:58 test_preference_state.parquet
-r--r--r-- 1 root root    857213 Oct  3 08:57 train_dislikes.parquet
-r--r--r-- 1 root root   5374652 Oct  3 08:57 train_likes.parquet
-r--r--r-- 1 root root 319183255 Oct  3 08:57 train_listens.parquet
-r--r--r-- 1 root root    155988 Oct  3 08:57 train_undislikes.parquet
-r--r--r-- 1 root root   1855885 Oct  3 08:57 train_unlikes.parquet
-r--r--r-- 1 root root     36571 Oct  3 08:57 val_dislikes.parquet
-r--r--r-- 1 root root    298484 Oct  3 08:57 val_likes.parquet
-r--r--r-- 1 root root   9947609 Oct  3 08:57 val_listens.parquet
-r--r--r-- 1 root root      7845 Oct  3 08:57 val_undislikes.parquet
-r--r--r-- 1 root root     72536 Oct  3 08:57 val_unlikes.parquet
-r--r--r-- 1 root root   7573821 Oct  3 08:57 validation_eligible_candidates.npz
-r--r--r-- 1 root root   1061761 Oct  3 08:58 validation_preference_state.parquet
8
               total        used        free      shared  buff/cache   available
Mem:              15           1           0           0          13          14
Swap:              0           0           0
2.8.0+cpu 2.2.6 2.3.3

<persisted-output>
Output too large (29.3KB). Full output saved to: /logs/agent/sessions/projects/-app/47c3e013-b2c2-46e8-8ac3-4197c05494a2/tool-results/b6kyfntm8.txt

Preview (first 2KB):
"""Shared public-data primitives for the Yambda two-tower baseline."""
from __future__ import annotations

import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Iterator

import numpy as np
import pandas as pd
import pyarrow.parquet as pq
import torch
from torch import nn


DAY = 86_400
TRAIN_END = 24_786_800
VALIDATION_END = 25_393_400
LISTEN_POSITIVE_THRESHOLD = 50
PROFILE_LOOKBACK_DAYS = 120
LIKE_ITEMS = 48
DISLIKE_ITEMS = 24
TOP_N = 100


@dataclass(frozen=True)
class TowerConfig:
    sequence_length: int = 100
    embedding_dim: int = 192
    hidden_dim: int = 192
    user_hidden_1: int = 512
    user_hidden_2: int = 256
    item_hidden_1: int = 512
    item_hidden_2: int = 256
    dropout: float = 0.05
    epochs: int = 1
    learning_rate: float = 1.0e-3
    weight_decay: float = 1.0e-5
    gradient_clip: float = 2.0
    uniform_negatives: int = 24
    inbatch_negatives: int = 8
    inbatch_correction_power: float = 0.5
    inbatch_correction_min: float = 0.25
    inbatch_correction_max: float = 4.0
    seed: int = 29


@dataclass(frozen=True)
class ListenEvents:
    timestamps: np.ndarray
    items: np.ndarray
    organic: np.ndarray
    completion_pct: np.ndarray


@dataclass(frozen=True)
class ExplicitEvents:
    timestamps: np.ndarray
    items: np.ndarray
    channel: np.ndarray
    active: np.ndarray
    organic_like: np.ndarray


def log(message: str) -> None:
    print(message, file=sys.stderr, flush=True)


def build_item_features(catalog: pd.DataFrame) -> np.ndarray:
    """Return four standardized features built only from the public catalog."""
    positives = catalog.positive_listens.to_numpy(dtype=np.float64, copy=False)
    organic = catalog.organic_positive_listens.to_numpy(dtype=np.float64, copy=False)
    popularity_rank = catalog.popularity_rank.to_numpy(dtype=np.float64, copy=False)
    features = np.column_stack(
        [
            np.log1p(positives),
            np.log1p(organic),
...
</persisted-output>

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")
2164 train_current_model.py
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],
176	    active_dislikes: dict[int, int],
177	    n_items: int,
178	    liked_states: list[np.ndarray],
179	    disliked_states: list[np.ndarray],
180	    state_counts: list[tuple[int, int]],
181	) -> int:
182	    liked_states.append(base.pack_active_items(active_likes, base.LIKE_ITEMS, n_items))
183	    disliked_states.append(
184	        base.pack_active_items(active_dislikes, base.DISLIKE_ITEMS, n_items)
185	    )
186	    state_counts.append((len(active_likes), len(active_dislikes)))
187	    return len(liked_states) - 1
188	
189	
190	def causal_activity_features(
191	    timestamps: np.ndarray,
192	    items: np.ndarray,
193	    organic: np.ndarray,
194	    completion: np.ndarray,
195	) -> np.ndarray:
196	    """Compute the first four reviewed user features at every event cutoff."""
197	    length = len(items)
198	    result = np.zeros((length, 4), dtype=np.float16)
199	    if not length:
200	        return result
201	    left_rows = np.searchsorted(
202	        timestamps,
203	        timestamps - base.PROFILE_LOOKBACK_DAYS * base.DAY,
204	        side="left",
205	    )
206	    rows = np.arange(length, dtype=np.int64)
207	    counts = rows - left_rows + 1
208	    organic_sum = np.cumsum(organic.astype(np.float64))
209	    completion_sum = np.cumsum(np.minimum(completion, 100).astype(np.float64))
210	    before_organic = np.where(
211	        left_rows > 0, organic_sum[np.maximum(left_rows - 1, 0)], 0.0
212	    )
213	    before_completion = np.where(
214	        left_rows > 0, completion_sum[np.maximum(left_rows - 1, 0)], 0.0
215	    )
216	
217	    unique_counts = np.empty(length, dtype=np.int32)
218	    frequencies: dict[int, int] = {}
219	    left = 0
220	    for right, raw_item in enumerate(items):
221	        item = int(raw_item)
222	        frequencies[item] = frequencies.get(item, 0) + 1
223	        required_left = int(left_rows[right])
224	        while left < required_left:
225	            old_item = int(items[left])
226	            remaining = frequencies[old_item] - 1
227	            if remaining:
228	                frequencies[old_item] = remaining
229	            else:
230	                del frequencies[old_item]
231	            left += 1
232	        unique_counts[right] = len(frequencies)
233	
234	    result[:, 0] = (np.log1p(counts) / 8.0).astype(np.float16)
235	    result[:, 1] = (np.log1p(unique_counts) / 6.0).astype(np.float16)
236	    result[:, 2] = (
237	        (organic_sum - before_organic) / np.maximum(counts, 1)
238	    ).astype(np.float16)
239	    result[:, 3] = (
240	        (completion_sum - before_completion)
241	        / np.maximum(counts, 1)
242	        / 100.0
243	    ).astype(np.float16)
244	    return result
245	
246	
247	def build_sequence_parts(
248	    listens: base.ListenEvents,
249	    explicit: base.ExplicitEvents,
250	    n_items: int,
251	    liked_states: list[np.ndarray],
252	    disliked_states: list[np.ndarray],
253	    state_counts: list[tuple[int, int]],
254	) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray] | None:
255	    qualifying = listens.completion_pct >= base.LISTEN_POSITIVE_THRESHOLD
256	    timestamps = listens.timestamps[qualifying].astype(np.int64, copy=False)
257	    items = listens.items[qualifying].astype(np.uint16, copy=False)
258	    organic = listens.organic[qualifying].astype(np.float32, copy=False)
259	    completion = listens.completion_pct[qualifying].astype(np.float32, copy=False)
260	    if not len(items):
261	        return None
262	
263	    event_features = np.zeros((len(items), 3), dtype=np.float16)
264	    event_features[:, 0] = organic.astype(np.float16)
265	    event_features[:, 1] = (
266	        np.minimum(completion, 100.0) / 100.0
267	    ).astype(np.float16)
268	    if len(items) > 1:
269	        gaps = np.maximum(0, timestamps[1:] - timestamps[:-1])
270	        event_features[1:, 2] = np.exp(
271	            -gaps.astype(np.float64) / (60.0 * base.DAY)
272	        ).astype(np.float16)
273	
274	    activity = causal_activity_features(
275	        timestamps, items, organic, completion
276	    )
277	    states = np.zeros(len(items), dtype=np.uint32)
278	    active_likes: dict[int, int] = {}
279	    active_dislikes: dict[int, int] = {}
280	    explicit_position = 0
281	    current_state = 0
282	    for row, timestamp in enumerate(timestamps):
283	        changed = False
284	        while (
285	            explicit_position < len(explicit.timestamps)
286	            and int(explicit.timestamps[explicit_position]) < int(timestamp)
287	        ):
288	            item = int(explicit.items[explicit_position])
289	            event_timestamp = int(explicit.timestamps[explicit_position])
290	            destination = (
291	                active_likes
292	                if int(explicit.channel[explicit_position]) == 0
293	                else active_dislikes
294	            )
295	            if int(explicit.active[explicit_position]) == 1:
296	                destination[item] = event_timestamp
297	            else:
298	                destination.pop(item, None)
299	            explicit_position += 1
300	            changed = True
301	        if changed:
302	            current_state = append_state_snapshot(
303	                active_likes,
304	                active_dislikes,
305	                n_items,
306	                liked_states,
307	                disliked_states,
308	                state_counts,
309	            )
310	        states[row] = np.uint32(current_state)
311	    return timestamps, items.copy(), event_features, activity, states
312	
313	
314	def final_user_features(
315	    timestamps: np.ndarray,
316	    items: np.ndarray,
317	    event_features: np.ndarray,
318	    history_end: int,
319	    like_count: int,
320	    dislike_count: int,
321	) -> np.ndarray:
322	    start = int(
323	        np.searchsorted(
324	            timestamps,
325	            history_end - base.PROFILE_LOOKBACK_DAYS * base.DAY,
326	            side="left",
327	        )
328	    )
329	    result = np.zeros(6, dtype=np.float32)
330	    recent_items = items[start:]
331	    if len(recent_items):
332	        result[0] = np.log1p(len(recent_items)) / 8.0
333	        result[1] = np.log1p(len(np.unique(recent_items))) / 6.0
334	        result[2] = float(np.mean(event_features[start:, 0]))
335	        result[3] = float(np.mean(event_features[start:, 1]))
336	    result[4] = np.log1p(like_count) / 6.0
337	    result[5] = np.log1p(dislike_count) / 4.0
338	    return result
339	
340	

341	def prepare_data(
342	    public: Path, stage: str, config: Config
343	) -> tuple[SequenceData, CandidateSet]:
344	    data = public / "data"
345	    catalog = pd.read_parquet(data / "candidate_catalog.parquet").sort_values(
346	        "popularity_rank"
347	    )
348	    item_ids = catalog.item_id.to_numpy(dtype=np.int64, copy=True)
349	    item_index = pd.Index(item_ids)
350	    n_items = len(item_ids)
351	    if n_items >= np.iinfo(np.uint16).max:
352	        raise RuntimeError("uint16 item storage is too small for this catalog")
353	
354	    if stage == "validation":
355	        prefixes = ("train",)
356	        history_end = base.TRAIN_END
357	        candidate_name = "validation_eligible_candidates.npz"
358	        state_name = "validation_preference_state.parquet"
359	    else:
360	        prefixes = ("train", "val")
361	        history_end = base.VALIDATION_END
362	        candidate_name = "test_eligible_candidates.npz"
363	        state_name = "test_preference_state.parquet"
364	    eligibility = CandidateSet.read(data / candidate_name)
365	    if not np.array_equal(eligibility.item_ids, item_ids):
366	        raise RuntimeError("catalog and candidate artifact item order differ")
367	
368	    explicit_by_user = base.load_explicit_events(data, prefixes, item_index)
369	    inference_likes, inference_dislikes, like_counts, dislike_counts = (
370	        base.load_inference_state(
371	            data / state_name,
372	            item_index,
373	            eligibility.uids,
374	            n_items,
375	        )
376	    )
377	    inference_rows = {int(uid): row for row, uid in enumerate(eligibility.uids)}
378	    inference_user_features = np.zeros((len(eligibility.uids), 6), dtype=np.float32)
379	    inference_sequence_rows = np.full(len(eligibility.uids), -1, dtype=np.int32)
380	
381	    # Snapshot zero is the shared empty explicit state.
382	    liked_states = [np.full(base.LIKE_ITEMS, n_items, dtype=np.uint16)]
383	    disliked_states = [
384	        np.full(base.DISLIKE_ITEMS, n_items, dtype=np.uint16)
385	    ]
386	    state_counts: list[tuple[int, int]] = [(0, 0)]
387	    user_ids: list[int] = []
388	    offsets = [0]
389	    timestamp_parts: list[np.ndarray] = []
390	    item_parts: list[np.ndarray] = []
391	    event_parts: list[np.ndarray] = []
392	    activity_parts: list[np.ndarray] = []
393	    state_parts: list[np.ndarray] = []
394	
395	    listen_paths = [data / f"{prefix}_listens.parquet" for prefix in prefixes]
396	    started = time.monotonic()
397	    for seen_users, (uid, listens) in enumerate(
398	        base.merge_listen_streams(listen_paths, item_index), start=1
399	    ):
400	        parts = build_sequence_parts(
401	            listens,
402	            explicit_by_user.get(uid, base.empty_explicit_events()),
403	            n_items,
404	            liked_states,
405	            disliked_states,
406	            state_counts,
407	        )
408	        if parts is None:
409	            continue
410	        sequence_row = len(user_ids)
411	        user_ids.append(uid)
412	        timestamp_parts.append(parts[0].astype(np.uint32, copy=False))
413	        item_parts.append(parts[1])
414	        event_parts.append(parts[2])
415	        activity_parts.append(parts[3])
416	        state_parts.append(parts[4])
417	        offsets.append(offsets[-1] + len(parts[1]))
418	        inference_row = inference_rows.get(uid)
419	        if inference_row is not None:
420	            inference_sequence_rows[inference_row] = sequence_row
421	            inference_user_features[inference_row] = final_user_features(
422	                parts[0],
423	                parts[1],
424	                parts[2],
425	                history_end,
426	                int(like_counts[inference_row]),
427	                int(dislike_counts[inference_row]),
428	            )
429	        if seen_users % 1_000 == 0:
430	            log(
431	                f"prepared users={seen_users:,} tokens={offsets[-1]:,} "
432	                f"states={len(liked_states):,} "
433	                f"elapsed={time.monotonic() - started:.1f}s"
434	            )
435	
436	    missing = np.flatnonzero(inference_sequence_rows < 0)
437	    if len(missing):
438	        inference_user_features[missing, 4] = np.log1p(like_counts[missing]) / 6.0
439	        inference_user_features[missing, 5] = (
440	            np.log1p(dislike_counts[missing]) / 4.0
441	        )
442	        log(f"target users without qualifying catalog sequence={len(missing):,}")
443	
444	    prepared = SequenceData(
445	        item_ids=item_ids,
446	        item_features=base.build_item_features(catalog),
447	        user_ids=np.asarray(user_ids, dtype=np.int64),
448	        offsets=np.asarray(offsets, dtype=np.int64),
449	        sequence_timestamps=np.concatenate(timestamp_parts),
450	        sequence_items=np.concatenate(item_parts),
451	        event_features=np.concatenate(event_parts),
452	        activity_features=np.concatenate(activity_parts),
453	        state_ids=np.concatenate(state_parts),
454	        liked_states=np.stack(liked_states),
455	        disliked_states=np.stack(disliked_states),
456	        state_counts=np.asarray(state_counts, dtype=np.int32),
457	        inference_user_rows=inference_sequence_rows,
458	        inference_liked_items=inference_likes,
459	        inference_disliked_items=inference_dislikes,
460	        inference_user_features=inference_user_features,
461	    )
462	    log(
463	        f"prepared stage={stage} sequence_users={len(prepared.user_ids):,} "
464	        f"tokens={len(prepared.sequence_items):,} "
465	        f"next_item_targets={prepared.target_count:,} "
466	        f"states={len(prepared.liked_states):,}"
467	    )
468	    return prepared, eligibility
469	
470	
471	def cache_metadata(stage: str, config: Config) -> dict[str, object]:
472	    return {
473	        "cache_version": CACHE_VERSION,
474	        "stage": stage,
475	        "positive_listen_threshold": base.LISTEN_POSITIVE_THRESHOLD,
476	        "profile_lookback_days": base.PROFILE_LOOKBACK_DAYS,
477	        "event_time_feature": "exp-negative-gap-over-60-days",
478	    }
479	
480	
481	def save_cache(
482	    path: Path, prepared: SequenceData, stage: str, config: Config
483	) -> None:
484	    path.parent.mkdir(parents=True, exist_ok=True)
485	    np.savez(
486	        path,
487	        metadata=np.asarray(json.dumps(cache_metadata(stage, config), sort_keys=True)),
488	        **prepared.__dict__,
489	    )
490	    log(f"saved public-derived full-sequence cache: {path}")
491	
492	
493	def load_cache(path: Path, stage: str, config: Config) -> SequenceData:
494	    with np.load(path, allow_pickle=False) as artifact:
495	        actual = json.loads(str(artifact["metadata"].item()))
496	        expected = cache_metadata(stage, config)
497	        if actual != expected:
498	            raise RuntimeError(f"sequence cache differs: {actual} != {expected}")
499	        values = {
500	            field: artifact[field].copy()
501	            for field in SequenceData.__dataclass_fields__
502	        }
503	    prepared = SequenceData(**values)
504	    log(
505	        f"loaded full-sequence cache: tokens={len(prepared.sequence_items):,} "
506	        f"targets={prepared.target_count:,}"
507	    )
508	    return prepared
509	
510	
511	def load_or_prepare(
512	    public: Path,
513	    stage: str,
514	    config: Config,
515	    cache: Path | None,
516	) -> tuple[SequenceData, CandidateSet]:
517	    candidate_name = (
518	        "validation_eligible_candidates.npz"
519	        if stage == "validation"
520	        else "test_eligible_candidates.npz"
521	    )
522	    eligibility = CandidateSet.read(public / "data" / candidate_name)
523	    if cache is not None and cache.is_file():
524	        prepared = load_cache(cache, stage, config)
525	        if not np.array_equal(prepared.item_ids, eligibility.item_ids):
526	            raise RuntimeError("cached catalog differs from candidate artifact")
527	        return prepared, eligibility
528	    prepared, built_eligibility = prepare_data(public, stage, config)
529	    if cache is not None:
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,

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,

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]:
1489	    return {
1490	        name: value.detach().cpu().clone()
1491	        for name, value in model.state_dict().items()
1492	    }
1493	
1494	
1495	def save_epoch_checkpoint(
1496	    checkpoint_dir: Path,
1497	    epoch: int,
1498	    model_state: dict[str, torch.Tensor],
1499	    config: Config,
1500	    diagnostics: dict[str, float],
1501	    optimizer: torch.optim.Optimizer,
1502	    generator: torch.Generator,
1503	) -> Path:
1504	    checkpoint_dir.mkdir(parents=True, exist_ok=True)
1505	    path = checkpoint_dir / f"epoch-{epoch:02d}.pt"
1506	    torch.save(
1507	        {
1508	            "epoch": epoch,
1509	            "config": asdict(config),
1510	            "diagnostics": diagnostics,
1511	            "model_state": model_state,
1512	            "optimizer_state": optimizer.state_dict(),
1513	            "generator_state": generator.get_state(),
1514	        },
1515	        path,
1516	    )
1517	    log(f"saved epoch checkpoint: {path}")
1518	    return path
1519	
1520	
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__":

started

#!/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.

    Args:
        eligibility: Validation-cutoff candidate membership service.
        targets: Complete set of target user IDs.

    Returns:
        A nested ``uid -> item_id -> grade`` mapping. Users without an eligible
        relevant event are omitted.
    """
    listens = pd.read_parquet(DATA / "val_listens.parquet")
    likes = pd.read_parquet(DATA / "val_likes.parquet")
    strong = listens[
        listens.uid.isin(targets)
        & (listens.is_organic == 1)
        & (listens.played_ratio_pct >= LISTEN_STRONG)
    ][["uid", "item_id"]].drop_duplicates()
    organic_likes = likes[
        likes.uid.isin(targets) & (likes.is_organic == 1)
    ][["uid", "item_id"]].drop_duplicates()

    graded: dict[int, dict[int, int]] = {}
    for raw_uid, raw_item in strong.itertuples(index=False, name=None):
        uid, item = int(raw_uid), int(raw_item)
        if eligibility.is_eligible(uid, item):
            graded.setdefault(uid, {})[item] = 1
    for raw_uid, raw_item in organic_likes.itertuples(index=False, name=None):
        uid, item = int(raw_uid), int(raw_item)
        if eligibility.is_eligible(uid, item):
            graded.setdefault(uid, {})[item] = 3
    return graded


def load_submission(
    path: Path, eligibility: CandidateSet, targets: set[int]
) -> dict[int, list[int]]:
    """Load and validate a public validation ranking.

    Files ending in ``.parquet`` are read as parquet; every other suffix is
    read as CSV. The required columns must be integer-valued, cover exactly the
    target-user set, contain 100 rows and ranks 1 through 100 per user, avoid
    duplicate user-item pairs, and contain only cutoff-eligible items.

    Args:
        path: Ranking file to read.
        eligibility: Validation-cutoff candidate membership service.
        targets: Exact set of users the ranking must cover.

    Returns:
        A ``uid -> ranked item IDs`` mapping ordered by ascending rank.

    Raises:
        ValueError: If the ranking violates the required schema or invariants.
    """
    frame = validate_ranking(read_ranking(path), targets)
    if not all(
        eligibility.is_eligible(uid, item)
        for uid, item in frame[["uid", "item_id"]].itertuples(index=False, name=None)
    ):
        raise RankingValidationError(
            "submission contains an item rejected by eligibility service"
        )
    return ranked_items(frame)


def score(
    ranked: dict[int, list[int]],
    graded: dict[int, dict[int, int]],
) -> dict[str, float | int]:
    """Aggregate per-user NDCG@10 over users with public relevance labels.

    Args:
        ranked: Ranked item IDs for every target user.
        graded: Eligible validation relevance keyed by user and item.

    Returns:
        The six-decimal ``quality_ndcg@10`` value and number of scored users.
    """
    return score_rankings(ranked, graded)


def main() -> None:
    """Run the public evaluator CLI and print its JSON score report."""
    targets_frame = pd.read_parquet(DATA / "target_users.parquet")
    targets = set(targets_frame.uid.astype(int))
    eligibility = CandidateSet(DATA / "validation_eligible_candidates.npz")
    graded = validation_truth(eligibility, targets)

    if len(sys.argv) != 2:
        raise SystemExit(
            "usage: evaluate_public.py <validation-ranking.parquet|csv>"
        )
    path = Path(sys.argv[1])
    ranked = load_submission(path, eligibility, targets)
    print(json.dumps({"model": str(path), "scores": score(ranked, graded)}, indent=2))


if __name__ == "__main__":
    main()
=====
"""Shared public primitives for Yambda discovery-ranking evaluation.

Both the solver-visible validation evaluator and the private submission judge
import this module from the generated task's ``public`` directory. It owns the
stage-independent contract: reading a ranking, validating its shape, converting
it to per-user rank order, looking up packed candidate eligibility, and
computing graded NDCG@10.

The module deliberately contains no event-derived labels, private artifact
paths, or pass thresholds. Each evaluator supplies its own relevance data and
per-cutoff eligibility checks around these common operations.
"""
from __future__ import annotations

from pathlib import Path

import numpy as np
import pandas as pd


__all__ = [
    "CandidateSet",
    "QUALITY_K",
    "RankingValidationError",
    "REQUIRED_COLUMNS",
    "TOP_N",
    "ndcg_at_k",
    "ranked_items",
    "read_ranking",
    "score_rankings",
    "validate_ranking",
]

TOP_N = 100
QUALITY_K = 10
REQUIRED_COLUMNS = ("uid", "item_id", "rank")


class RankingValidationError(ValueError):
    """Report a ranking-schema or structural-contract violation."""


class CandidateSet:
    """Provide membership lookups over one packed candidate-set archive.

    The archive contains aligned ``uids``, ``item_ids``, and
    ``packed_eligible`` arrays. Candidate column ``c`` is represented by bit
    ``c % 8`` of byte ``c // 8`` using little-endian bit order. Arrays are
    copied into memory so they remain valid after the archive is closed.
    """

    def __init__(self, path: Path):
        """Load an eligibility archive and build ID-to-position indexes.

        Args:
            path: Path to a ``.npz`` archive with the candidate-set schema.
        """
        with np.load(path, allow_pickle=False) as artifact:
            self.uids = artifact["uids"].astype(np.int64, copy=True)
            self.item_ids = artifact["item_ids"].astype(np.int64, copy=True)
            self.packed = artifact["packed_eligible"].astype(np.uint8, copy=True)
        self.uid_row = {int(uid): row for row, uid in enumerate(self.uids)}
        self.item_column = {
            int(item): column for column, item in enumerate(self.item_ids)
        }

    def is_eligible(self, uid: int, item_id: int) -> bool:
        """Return whether ``item_id`` is eligible for ``uid``.

        IDs absent from the target-user or candidate-item arrays are treated as
        ineligible rather than raising an exception.

        Args:
            uid: Target user ID to query.
            item_id: Candidate item ID to query.

        Returns:
            ``True`` only when the corresponding packed eligibility bit is set.
        """
        row = self.uid_row.get(int(uid))

   item_id  positive_listens  organic_positive_listens  popularity_rank
0  5862961             26615                     20620                1
1  6901374             25536                     17347                2
2  5635052             24691                     17194                3
3  3542184             24075                     17699                4
4  9378983             24036                     18195                5
['item_id', 'positive_listens', 'organic_positive_listens', 'popularity_rank']
            item_id  ...  popularity_rank
count  3.000000e+04  ...     30000.000000
mean   4.698278e+06  ...     15000.500000
std    2.700274e+06  ...      8660.398374
min    2.060000e+02  ...         1.000000
25%    2.382368e+06  ...      7500.750000
50%    4.687992e+06  ...     15000.500000
75%    7.040711e+06  ...     22500.250000
max    9.390294e+06  ...     30000.000000

[8 rows x 4 columns]
target users 8456
eligible per user: mean 29067.306291390727 min 24657
users with relevant 4263
relevant items total 37769 grade3 4417
relevant item popularity rank quantiles [  428.   1733.   5622.  13374.  21680.2]
item_ids in popularity order? True
popularity baseline {'quality_ndcg@10': 0.01071, 'quality_users': 4263}
organic popularity baseline {'quality_ndcg@10': 3.2e-05, 'quality_users': 4263}

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

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
train_listens 43484159
val_listens 1501697
train_likes 814831
val_likes 37795
train users 9238

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

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 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

Exit code 1
Traceback (most recent call last):
  File "<stdin>", line 244, in <module>
PermissionError: [Errno 13] Permission denied: 'train_current_model.py'

total 116
drwxr-xr-x 1 agent agent  4096 Oct  3 09:04 .
drwxr-xr-x 1 agent agent  4096 Oct  3 08:59 ..
drwxr-xr-x 2 agent agent  4096 Oct  3 09:06 cache
drwxr-xr-x 2 agent agent  4096 Oct  3 09:05 runs
-r--r--r-- 1 agent agent 80058 Oct  3 09:04 train_current_model.py
-r--r--r-- 1 agent agent 15985 Oct  3 09:04 two_tower_model.py
-rw-r--r-- 1 agent agent 80058 Oct  3 09:04 train_current_model.py

ok
62:    loss_objective: str = "sampled-softmax"
580:def build_negative_sampler(
672:def score_target_microbatches_softmax(
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

   8149 agent     20   0 7104764   4.5g 150120 R 393.3  29.4  21:39.51 python
   8058 agent     20   0 5604140 257312 127636 S   6.7   1.6   0:03.79 claude
      1 root      20   0    2592   1716   1612 S   0.0   0.0   0:00.01 sh
      7 root      20   0    2920   1612   1508 S   0.0   0.0   0:00.00 sleep
   8051 agent     20   0    4480   3368   3020 S   0.0   0.0   0:00.01 bash
   8059 agent     20   0    2924   1704   1592 S   0.0   0.0   0:00.02 tee
   8145 agent     20   0    4492   1752   1380 S   0.0   0.0   0:00.00 bash
   8323 agent     20   0    4648   3480   2940 S   0.0   0.0   0:00.00 bash
ok
syntax ok

Command did not complete within its 120s timeout and was moved to the background (ID: brgd9v0dj). Output is being written to: /tmp/claude-1000/-app/47c3e013-b2c2-46e8-8ac3-4197c05494a2/tasks/brgd9v0dj.output. You will be notified when it completes. If it is still running after 30m in the background, it will be stopped and you will be notified. To check interim output, use Read on that file path.

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,558,741 bytes=34,256,250
negative sampler: power=0.75 shared=2048 mass_top1000=0.125 inbatch_softmax=True logq_weight=1.0
epoch=1/1 train_loss=7.86972 validation_loss=7.36741 pair_accuracy=0.472 train_targets=36,027 validation_targets=17,721 train_seconds=39.8 validation_seconds=6.8
restored best validation-BCE checkpoint: epoch=1 loss=7.367406
---
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

/bin/bash: line 3:  8389 Killed                  nohup python train_current_model.py --stage validation --public-dir /task --prepared-cache cache/validation.npz --output runs/improved_validation.parquet --threads 4 > runs/improved_val.json 2> runs/improved_val.log
epoch=1/1 batches=250 targets=987,417 batch_bce=0.547529 running_bce=0.541614 elapsed=416.3s

               total        used        free      shared  buff/cache   available
Mem:              15           2           6           0           6          12
8331 0.0 0.001688GB 600 python train_current_model.py --stage validation --public-dir /task
8332 10.5 1.68154GB train_current_model.py --stage validation --public-dir /task --prepared-cache cache/validation.npz
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
negative sampler: power=0.75 shared=2048 mass_top1000=0.178 inbatch_softmax=True logq_weight=1.0
epoch=1/1 train_loss=7.86972 validation_loss=7.36741 pair_accuracy=0.472 train_targets=36,027 validation_targets=17,721 train_seconds=39.8 validation_seconds=6.8
restored best validation-BCE checkpoint: epoch=1 loss=7.367406

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
epoch=1/1 batches=250 targets=987,417 batch_bce=0.547529 running_bce=0.541614 elapsed=416.3s
---

ls: cannot access 'production_validation.parquet': No such file or directory
runs/:
total 140
drwxr-xr-x 2 agent agent  4096 Oct  3 09:12 .
drwxr-xr-x 1 agent agent  4096 Oct  3 09:04 ..
-rw-r--r-- 1 agent agent  1874 Oct  3 09:04 analyze.py
-rw-r--r-- 1 agent agent     0 Oct  3 09:04 baseline.json
-rw-r--r-- 1 agent agent  1329 Oct  3 09:12 baseline.log
-r--r--r-- 1 agent agent 80058 Oct  3 09:04 baseline_train.py
-rw-r--r-- 1 agent agent  9542 Oct  3 09:08 helper.py

8589934592
8589934592
low 0
high 0
max 5791
oom 6
oom_kill 2
oom_group_kill 0
sock_throttled 0
-rw-r--r-- 1 agent agent   3179 Oct  3 09:13 runs/smoke.json
-rw-r--r-- 1 agent agent    951 Oct  3 09:13 runs/smoke.log
-rw-r--r-- 1 agent agent 929529 Oct  3 09:13 runs/smoke.parquet
ranked users=8,192/8,456
ranked users=8,456/8,456
0

{
  "model": "runs/smoke.parquet",
  "scores": {
    "quality_ndcg@10": 0.021115,
    "quality_users": 4263
  }
}
started

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
negative sampler: power=0.75 shared=2048 mass_top1000=0.178 inbatch_softmax=True logq_weight=1.0
3.11482 GB RSS

Exit code 143
Command timed out after 2m 0s

epoch=1/1 train_loss=8.20848 validation_loss=7.35754 pair_accuracy=0.579 train_targets=1,052,953 validation_targets=130,103 train_seconds=142.7 validation_seconds=10.0
restored best validation-BCE checkpoint: epoch=1 loss=7.357538
3.69726 GB RSS
Sat Oct  3 09:16:57 UTC 2026

ranked users=8,456/8,456
runs/improved_validation.parquet
{
  "model": "runs/improved_validation.parquet",
  "scores": {
    "quality_ndcg@10": 0.0276,
    "quality_users": 4263
  }
}

started
Sat Oct  3 09:17:56 UTC 2026

full-history false-negative filter: users=9,153 unique_user_items=5,682,910 bytes=34,323,750
negative sampler: power=0.75 shared=2048 mass_top1000=0.178 inbatch_softmax=True logq_weight=1.0
Sat Oct  3 09:19:43 UTC 2026

restored best validation-BCE checkpoint: epoch=1 loss=7.417665
restored best validation-BCE checkpoint: epoch=1 loss=7.417665
Sat Oct  3 09:21:35 UTC 2026

ranked users=8,456/8,456
-rw-r--r-- 1 agent agent 1151492 Oct  3 09:22 /app/solution/submission.parquet
submission OK: 845600 rows 8456 users
Sat Oct  3 09:22:40 UTC 2026

started
Sat Oct  3 09:22:45 UTC 2026

  File "/app/solution/runs/baseline_train.py", line 34, in <module>
    import two_tower_model as base
ModuleNotFoundError: No module named 'two_tower_model'
0

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 Oct  3 09:23:04 UTC 2026

full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250
Sat Oct  3 09:24:25 UTC 2026

full-history false-negative filter: users=9,135 unique_user_items=5,524,275 bytes=34,256,250
Sat Oct  3 09:26:25 UTC 2026

restored best validation-BCE checkpoint: epoch=1 loss=0.496636
Sat Oct  3 09:28:22 UTC 2026

ranked users=8,456/8,456
production_validation.parquet
{
  "model": "production_validation.parquet",
  "scores": {
    "quality_ndcg@10": 0.02128,
    "quality_users": 4263
  }
}
"seconds": 220.89775895599996
Sat Oct  3 09:29:18 UTC 2026

started
Sat Oct  3 09:29:24 UTC 2026

Sat Oct  3 09:31:24 UTC 2026

epoch=1/2 batches=250 targets=987,417 batch_bce=7.765315 running_bce=8.219107 elapsed=133.3s
epoch=1/2 train_loss=8.20848 validation_loss=7.35754 pair_accuracy=0.579 train_targets=1,052,953 validation_targets=130,103 train_seconds=144.3 validation_secon
Sat Oct  3 09:33:06 UTC 2026

Exit code 143
Command timed out after 2m 0s

epoch=2/2 train_loss=7.78724 validation_loss=7.14654 pair_accuracy=0.690 train_targets=1,052,953 validation_targets=130,
restored best validation-BCE checkpoint: epoch=2 loss=7.146540
Sat Oct  3 09:35:12 UTC 2026

ranked users=8,456/8,456
runs/improved_e2_validation.parquet
{
  "model": "runs/improved_e2_validation.parquet",
  "scores": {
    "quality_ndcg@10": 0.019932,
    "quality_users": 4263
  }
}
Sat Oct  3 09:36:14 UTC 2026

ok
Sat Oct  3 09:36:26 UTC 2026

total 2480
drwxr-xr-x 1 agent agent    4096 Oct  3 09:29 .
drwxr-xr-x 1 agent agent    4096 Oct  3 08:59 ..
-rw-r--r-- 1 agent agent    4291 Oct  3 09:36 README.md
drwxr-xr-x 2 agent agent    4096 Oct  3 09:06 cache
-rw-r--r-- 1 agent agent 1244230 Oct  3 09:28 production_validation.parquet
drwxr-xr-x 2 agent agent    4096 Oct  3 09:36 runs
-rw-r--r-- 1 agent agent 1151492 Oct  3 09:22 submission.parquet
-rw-r--r-- 1 agent agent   93610 Oct  3 09:08 train_current_model.py
-rw-r--r-- 1 agent agent   15985 Oct  3 09:04 two_tower_model.py
Sat Oct  3 09:38:21 UTC 2026

epoch=1/1 train_loss=8.07126 validation_loss=7.23489 pair_accuracy=0.611 train_targets=1,052,953 val
restored best validation-BCE checkpoint: epoch=1 loss=7.2348
Sat Oct  3 09:40:03 UTC 2026

ranked users=8,456/8,456
runs/improved_lq05_validation.parquet
{
  "model": "runs/improved_lq05_validation.parquet",
  "scores": {
    "quality_ndcg@10": 0.019907,
    "quality_users": 4263
  }
}
Sat Oct  3 09:41:10 UTC 2026

submission OK: 845600 rows 8456 users
submission sampled-softmax 1 1.0 2048 2048-shared-popularity^0.75-plus-inbatch-targets-logq 253.806
regular-file
87