Harness context codex item completed
{
"type": "codex_item_completed",
"item": {
"type": "CommandExecution",
"id": "exec-e757c407-6867-4c02-bdf2-0b4ffa102707",
"process_id": "22665",
"command": [
"/bin/sh",
"-lc",
"rg -n \"negative|inbatch|loss|sample|argparse|stage|Config\\(\" /task/current_model/train_current_model.py /task/current_model/two_tower_model.py && sed -n '260,760p' /task/current_model/train_current_model.py && sed -n '300,520p' /task/current_model/two_tower_model.py"
],
"cwd": "file:///app",
"parsed_cmd": [
{
"type": "search",
"cmd": "rg -n \"negative|inbatch|loss|sample|argparse|stage|Config\\\\(\" /task/current_model/train_current_model.py /task/current_model/two_tower_model.py",
"query": "negative|inbatch|loss|sample|argparse|stage|Config\\(",
"path": "train_current_model.py"
},
{
"type": "read",
"cmd": "sed -n '260,760p' /task/current_model/train_current_model.py",
"name": "train_current_model.py",
"path": "/task/current_model/train_current_model.py"
},
{
"type": "read",
"cmd": "sed -n '300,520p' /task/current_model/two_tower_model.py",
"name": "two_tower_model.py",
"path": "/task/current_model/two_tower_model.py"
}
],
"source": "unified_exec_startup",
"status": "completed",
"stdout": "/task/current_model/two_tower_model.py:40: uniform_negatives: int = 24\n/task/current_model/two_tower_model.py:41: inbatch_negatives: int = 8\n/task/current_model/two_tower_model.py:42: inbatch_correction_power: float = 0.5\n/task/current_model/two_tower_model.py:43: inbatch_correction_min: float = 0.25\n/task/current_model/two_tower_model.py:44: inbatch_correction_max: float = 4.0\n/task/current_model/two_tower_model.py:346:def invalid_negative_mask(\n/task/current_model/two_tower_model.py:347: negatives: torch.Tensor,\n/task/current_model/two_tower_model.py:353: invalid = negatives.eq(targets[:, None])\n/task/current_model/two_tower_model.py:354: invalid |= negatives[:, :, None].eq(context_items[:, None, :]).any(dim=2)\n/task/current_model/two_tower_model.py:355: invalid |= negatives[:, :, None].eq(liked_items[:, None, :]).any(dim=2)\n/task/current_model/two_tower_model.py:356: for column in range(1, negatives.shape[1]):\n/task/current_model/two_tower_model.py:357: invalid[:, column] |= negatives[:, column, None].eq(\n/task/current_model/two_tower_model.py:358: negatives[:, :column]\n/task/current_model/two_tower_model.py:363:def draw_filtered_mixed_negatives(\n/task/current_model/two_tower_model.py:369: inbatch_count: int,\n/task/current_model/two_tower_model.py:374: \"\"\"Draw uniform and unique-target in-batch negatives.\"\"\"\n/task/current_model/two_tower_model.py:382: inbatch = torch.empty((batch_size, 0), dtype=torch.long, device=\"cpu\")\n/task/current_model/two_tower_model.py:384: if inbatch_count:\n/task/current_model/two_tower_model.py:386: corrected_count = min(inbatch_count, max(len(unique_targets) - 1, 0))\n/task/current_model/two_tower_model.py:396: inbatch = unique_targets[columns]\n/task/current_model/two_tower_model.py:397: if corrected_count < inbatch_count:\n/task/current_model/two_tower_model.py:400: (batch_size, inbatch_count - corrected_count),\n/task/current_model/two_tower_model.py:404: inbatch = torch.cat([inbatch, filler], dim=1)\n/task/current_model/two_tower_model.py:406: negatives_cpu = torch.cat([uniform, inbatch], dim=1)\n/task/current_model/two_tower_model.py:407: correction_cpu = torch.ones_like(negatives_cpu, dtype=torch.float32)\n/task/current_model/two_tower_model.py:411: ] = item_correction[inbatch[:, :corrected_count]]\n/task/current_model/two_tower_model.py:412: negatives = negatives_cpu.to(device)\n/task/current_model/two_tower_model.py:415: invalid = invalid_negative_mask(\n/task/current_model/two_tower_model.py:416: negatives, targets, context_items, liked_items\n/task/current_model/two_tower_model.py:419: return negatives, correction\n/task/current_model/two_tower_model.py:426: negatives[invalid] = replacements\n/task/current_model/two_tower_model.py:428: invalid = invalid_negative_mask(\n/task/current_model/two_tower_model.py:429: negatives, targets, context_items, liked_items\n/task/current_model/two_tower_model.py:432: raise RuntimeError(\"mixed negative resampling failed false-negative filter\")\n/task/current_model/two_tower_model.py:433: return negatives, correction\n/task/current_model/train_current_model.py:6:loss. The fast default retains each user's most recent 128 training targets and\n/task/current_model/train_current_model.py:9:computes all of its timestep losses and carries the detached GRU hidden state\n/task/current_model/train_current_model.py:13:normalization, raw dot-product retrieval, sample-weighted binary cross entropy,\n/task/current_model/train_current_model.py:14:and 24 uniform plus eight propensity-corrected unique in-batch negatives. Every\n/task/current_model/train_current_model.py:20:import argparse\n/task/current_model/train_current_model.py:55: loss_microbatch_size: int = 65_536\n/task/current_model/train_current_model.py:56: uniform_negatives: int = 24\n/task/current_model/train_current_model.py:57: inbatch_negatives: int = 8\n/task/current_model/train_current_model.py:58: inbatch_correction_power: float = 0.5\n/task/current_model/train_current_model.py:59: inbatch_correction_min: float = 0.25\n/task/current_model/train_current_model.py:60: inbatch_correction_max: float = 4.0\n/task/current_model/train_current_model.py:61: negative_filter_scope: str = \"full-training-history\"\n/task/current_model/train_current_model.py:152: return base.TowerConfig(\n/task/current_model/train_current_model.py:165: uniform_negatives=config.uniform_negatives,\n/task/current_model/train_current_model.py:166: inbatch_negatives=config.inbatch_negatives,\n/task/current_model/train_current_model.py:167: inbatch_correction_power=config.inbatch_correction_power,\n/task/current_model/train_current_model.py:168: inbatch_correction_min=config.inbatch_correction_min,\n/task/current_model/train_current_model.py:169: inbatch_correction_max=config.inbatch_correction_max,\n/task/current_model/train_current_model.py:342: public: Path, stage: str, config: Config\n/task/current_model/train_current_model.py:354: if stage == \"validation\":\n/task/current_model/train_current_model.py:463: f\"prepared stage={stage} sequence_users={len(prepared.user_ids):,} \"\n/task/current_model/train_current_model.py:471:def cache_metadata(stage: str, config: Config) -> dict[str, object]:\n/task/current_model/train_current_model.py:474: \"stage\": stage,\n/task/current_model/train_current_model.py:477: \"event_time_feature\": \"exp-negative-gap-over-60-days\",\n/task/current_model/train_current_model.py:482: path: Path, prepared: SequenceData, stage: str, config: Config\n/task/current_model/train_current_model.py:487: metadata=np.asarray(json.dumps(cache_metadata(stage, config), sort_keys=True)),\n/task/current_model/train_current_model.py:493:def load_cache(path: Path, stage: str, config: Config) -> SequenceData:\n/task/current_model/train_current_model.py:496: expected = cache_metadata(stage, config)\n/task/current_model/train_current_model.py:513: stage: str,\n/task/current_model/train_current_model.py:519: if stage == \"validation\"\n/task/current_model/train_current_model.py:524: prepared = load_cache(cache, stage, config)\n/task/current_model/train_current_model.py:528: prepared, built_eligibility = prepare_data(public, stage, config)\n/task/current_model/train_current_model.py:530: save_cache(cache, prepared, stage, config)\n/task/current_model/train_current_model.py:562: config.inbatch_correction_power,\n/task/current_model/train_current_model.py:567: config.inbatch_correction_min,\n/task/current_model/train_current_model.py:568: config.inbatch_correction_max,\n/task/current_model/train_current_model.py:572: f\"inbatch correction: min={correction.min():.3f} \"\n/task/current_model/train_current_model.py:603: f\"full-history false-negative filter: users={len(prepared.user_ids):,} \"\n/task/current_model/train_current_model.py:609:def invalid_full_history_negatives(\n/task/current_model/train_current_model.py:610: negatives: np.ndarray,\n/task/current_model/train_current_model.py:616: \"\"\"Return row-level invalid negatives using the complete training prefix.\"\"\"\n/task/current_model/train_current_model.py:617: byte_columns = negatives >> 3\n/task/current_model/train_current_model.py:618: bit_columns = negatives & 7\n/task/current_model/train_current_model.py:621: invalid |= negatives == targets[:, None]\n/task/current_model/train_current_model.py:622: invalid |= (negatives[:, :, None] == liked_items[:, None, :]).any(axis=2)\n/task/current_model/train_current_model.py:623: for column in range(1, negatives.shape[1]):\n/task/current_model/train_current_model.py:625: negatives[:, column, None] == negatives[:, :column]\n/task/current_model/train_current_model.py:630:def draw_full_history_filtered_negatives(\n/task/current_model/train_current_model.py:641: \"\"\"Draw mixed negatives and reject all known public-training positives.\"\"\"\n/task/current_model/train_current_model.py:647: (batch_size, config.uniform_negatives),\n/task/current_model/train_current_model.py:651: inbatch = torch.empty((batch_size, 0), dtype=torch.long)\n/task/current_model/train_current_model.py:653: if config.inbatch_negatives:\n/task/current_model/train_current_model.py:656: config.inbatch_negatives, max(len(unique_targets) - 1, 0)\n/task/current_model/train_current_model.py:671: inbatch = unique_targets[columns]\n/task/current_model/train_current_model.py:672: if corrected_count < config.inbatch_negatives:\n/task/current_model/train_current_model.py:675: (batch_size, config.inbatch_negatives - corrected_count),\n/task/current_model/train_current_model.py:679: inbatch = torch.cat([inbatch, filler], dim=1)\n/task/current_model/train_current_model.py:681: negatives = torch.cat([uniform, inbatch], dim=1).numpy()\n/task/current_model/train_current_model.py:682: correction = np.ones_like(negatives, dtype=np.float32)\n/task/current_model/train_current_model.py:685: :, config.uniform_negatives : config.uniform_negatives + corrected_count\n/task/current_model/train_current_model.py:686: ] = item_correction[inbatch[:, :corrected_count]].numpy()\n/task/current_model/train_current_model.py:688: invalid = invalid_full_history_negatives(\n/task/current_model/train_current_model.py:689: negatives,\n/task/current_model/train_current_model.py:697: torch.from_numpy(negatives).to(device),\n/task/current_model/train_current_model.py:706: negatives[invalid] = replacements\n/task/current_model/train_current_model.py:709: invalid = invalid_full_history_negatives(\n/task/current_model/train_current_model.py:710: negatives,\n/task/current_model/train_current_model.py:717: raise RuntimeError(\"full-history negative resampling failed\")\n/task/current_model/train_current_model.py:719: torch.from_numpy(negatives).to(device),\n/task/current_model/train_current_model.py:724:def draw_training_negatives(\n/task/current_model/train_current_model.py:736: if config.negative_filter_scope == \"full-training-history\":\n/task/current_model/train_current_model.py:738: raise RuntimeError(\"full-history negative filter was not built\")\n/task/current_model/train_current_model.py:739: return draw_full_history_filtered_negatives(\n/task/current_model/train_current_model.py:750: return base.draw_filtered_mixed_negatives(\n/task/current_model/train_current_model.py:755: config.uniform_negatives,\n/task/current_model/train_current_model.py:756: config.inbatch_negatives,\n/task/current_model/train_current_model.py:816: negatives: torch.Tensor,\n/task/current_model/train_current_model.py:818: all_items = torch.cat([positives, negatives.reshape(-1)])\n/task/current_model/train_current_model.py:822: negative_vectors = encoded[inverse[len(positives) :]].reshape(\n/task/current_model/train_current_model.py:823: len(positives), negatives.shape[1], -1\n/task/current_model/train_current_model.py:825: return positive_vectors, negative_vectors\n/task/current_model/train_current_model.py:845:def corrected_negative_loss(\n/task/current_model/train_current_model.py:851: total = config.uniform_negatives + config.inbatch_negatives\n/task/current_model/train_current_model.py:853: if config.uniform_negatives:\n/task/current_model/train_current_model.py:854: result += config.uniform_negatives * raw[\n/task/current_model/train_current_model.py:855: :, : config.uniform_negatives\n/task/current_model/train_current_model.py:857: if config.inbatch_negatives:\n/task/current_model/train_current_model.py:858: selected = slice(config.uniform_negatives, total)\n/task/current_model/train_current_model.py:860: result += config.inbatch_negatives * (\n/task/current_model/train_current_model.py:880: raise RuntimeError(\"chronological target partition produced negative counts\")\n/task/current_model/train_current_model.py:915: public: Path, stage: str, item_ids: np.ndarray\n/task/current_model/train_current_model.py:918: prefixes = (\"train\",) if stage == \"validation\" else (\"train\", \"val\")\n/task/current_model/train_current_model.py:957: stage: str,\n/task/current_model/train_current_model.py:978: public, stage, prepared.item_ids\n/task/current_model/train_current_model.py:981: base.TRAIN_END if stage == \"validation\" else base.VALIDATION_END\n/task/current_model/train_current_model.py:1119: negatives: torch.Tensor,\n/task/current_model/train_current_model.py:1120: negative_weights: torch.Tensor,\n/task/current_model/train_current_model.py:1121: sample_weights: np.ndarray,\n/task/current_model/train_current_model.py:1127: total_weight = float(sample_weights.sum())\n/task/current_model/train_current_model.py:1129: raise RuntimeError(\"target block has no positive sample weight\")\n/task/current_model/train_current_model.py:1132: \"loss_sum\": 0.0,\n/task/current_model/train_current_model.py:1134: \"negative_sum\": 0.0,\n/task/current_model/train_current_model.py:1138: for begin in range(0, target_count, config.loss_microbatch_size):\n/task/current_model/train_current_model.py:1139: end = min(begin + config.loss_microbatch_size, target_count)\n/task/current_model/train_current_model.py:1162: positive_vectors, negative_vectors = candidate_vectors(\n/task/current_model/train_current_model.py:1165: negatives[begin:end],\n/task/current_model/train_current_model.py:1168: negative_logits = torch.einsum(\n/task/current_model/train_current_model.py:1169: \"bd,bkd->bk\", query_vectors, negative_vectors\n/task/current_model/train_current_model.py:1171: positive_loss = F.softplus(-positive_logits)\n/task/current_model/train_current_model.py:1172: negative_loss = corrected_negative_loss(\n/task/current_model/train_current_model.py:1173: negative_logits, negative_weights[begin:end], config\n/task/current_model/train_current_model.py:1176: sample_weights[begin:end].astype(np.float32, copy=False)\n/task/current_model/train_current_model.py:1178: loss_sum = (\n/task/current_model/train_current_model.py:1179: selected_weights * (positive_loss + negative_loss) / 2.0\n/task/current_model/train_current_model.py:1181: if not bool(torch.isfinite(loss_sum)):\n/task/current_model/train_current_model.py:1182: raise RuntimeError(\"dense next-item training produced non-finite loss\")\n/task/current_model/train_current_model.py:1184: (loss_sum / total_weight).backward()\n/task/current_model/train_current_model.py:1188: statistics[\"loss_sum\"] += float(loss_sum.detach().cpu())\n/task/current_model/train_current_model.py:1190: statistics[\"negative_sum\"] += float(negative_logits.detach().sum().cpu())\n/task/current_model/train_current_model.py:1192: (positive_logits.detach()[:, None] > negative_logits.detach()).sum().cpu()\n/task/current_model/train_current_model.py:1227: negative_count = config.uniform_negatives + config.inbatch_negatives\n/task/current_model/train_current_model.py:1228: total_loss = 0.0\n/task/current_model/train_current_model.py:1232: total_negative = 0.0\n/task/current_model/train_current_model.py:1293: negatives, negative_weights = draw_training_negatives(\n/task/current_model/train_current_model.py:1314: negatives,\n/task/current_model/train_current_model.py:1315: negative_weights,\n/task/current_model/train_current_model.py:1327: total_loss += statistics[\"loss_sum\"]\n/task/current_model/train_current_model.py:1331: total_negative += statistics[\"negative_sum\"]\n/task/current_model/train_current_model.py:1337: batch_bce = statistics[\"loss_sum\"] / max(\n/task/current_model/train_current_model.py:1340: running_bce = total_loss / max(total_weight, 1.0e-12)\n/task/current_model/train_current_model.py:1351: \"loss\": total_loss / max(total_weight, 1.0e-12),\n/task/current_model/train_current_model.py:1353: \"negative_logit\": total_negative / max(total_targets * negative_count, 1),\n/task/current_model/train_current_model.py:1355: / max(total_targets * negative_count, 1),\n/task/current_model/train_current_model.py:1388: total_loss = 0.0\n/task/current_model/train_current_model.py:1449: negatives, negative_weights = draw_training_negatives(\n/task/current_model/train_current_model.py:1469: negatives,\n/task/current_model/train_current_model.py:1470: negative_weights,\n/task/current_model/train_current_model.py:1475: total_loss += statistics[\"loss_sum\"]\n/task/current_model/train_current_model.py:1481: \"validation_loss\": total_loss / max(total_weight, 1.0e-12),\n/task/current_model/train_current_model.py:1523: stage: str,\n/task/current_model/train_current_model.py:1560: stage,\n/task/current_model/train_current_model.py:1567: if config.negative_filter_scope == \"full-training-history\"\n/task/current_model/train_current_model.py:1571: best_loss = float(\"inf\")\n/task/current_model/train_current_model.py:1601: \"negative_filter_scope\", \"last-100\"\n/task/current_model/train_current_model.py:1629: stored_filter_scope != config.negative_filter_scope\n/task/current_model/train_current_model.py:1657: prior.pop(\"validation_loss\", None)\n/task/current_model/train_current_model.py:1660: f\"{stored_filter_scope}->{config.negative_filter_scope}, \"\n/task/current_model/train_current_model.py:1666: if \"validation_loss\" not in prior:\n/task/current_model/train_current_model.py:1682: best_loss = float(prior[\"validation_loss\"])\n/task/current_model/train_current_model.py:1687: f\"loss={best_loss:.6f}\"\n/task/current_model/train_current_model.py:1731: diagnostics[\"validation_loss\"]\n/task/current_model/train_current_model.py:1732: < best_loss - config.early_stopping_min_delta\n/task/current_model/train_current_model.py:1735: best_loss = diagnostics[\"validation_loss\"]\n/task/current_model/train_current_model.py:1754: f\"epoch={epoch}/{config.epochs} train_loss={diagnostics['loss']:.5f} \"\n/task/current_model/train_current_model.py:1755: f\"validation_loss={diagnostics['validation_loss']:.5f} \"\n/task/current_model/train_current_model.py:1766: f\"best_epoch={best_epoch} best_validation_loss={best_loss:.6f}\"\n/task/current_model/train_current_model.py:1776: f\"loss={best_loss:.6f}\"\n/task/current_model/train_current_model.py:1923: parser = argparse.ArgumentParser(description=__doc__)\n/task/current_model/train_current_model.py:1925: \"--stage\", choices=(\"validation\", \"submission\"), default=\"validation\"\n/task/current_model/train_current_model.py:1939: \"--loss-microbatch-size\",\n/task/current_model/train_current_model.py:1941: default=Config.loss_microbatch_size,\n/task/current_model/train_current_model.py:1944: \"--negative-filter-scope\",\n/task/current_model/train_current_model.py:1946: default=Config.negative_filter_scope,\n/task/current_model/train_current_model.py:2012: or args.loss_microbatch_size < 1\n/task/current_model/train_current_model.py:2028: config = Config(\n/task/current_model/train_current_model.py:2033: loss_microbatch_size=args.loss_microbatch_size,\n/task/current_model/train_current_model.py:2034: negative_filter_scope=args.negative_filter_scope,\n/task/current_model/train_current_model.py:2057: args.stage,\n/task/current_model/train_current_model.py:2071: \"stage\": args.stage,\n/task/current_model/train_current_model.py:2095: args.stage,\n/task/current_model/train_current_model.py:2115: \"stage\": args.stage,\n/task/current_model/train_current_model.py:2126: \"sequence_encoder\": \"one-layer-GRU-dense-timestep-loss\",\n/task/current_model/train_current_model.py:2129: \"loss_execution\": \"gradient-accumulated-microbatches\",\n/task/current_model/train_current_model.py:2135: \"objective\": \"sample-weighted-BCE-at-every-timestep\",\n/task/current_model/train_current_model.py:2136: \"negative_sampling\": (\n/task/current_model/train_current_model.py:2137: \"24-uniform-plus-8-propensity-corrected-inbatch\"\n/task/current_model/train_current_model.py:2139: \"false_negative_filter\": config.negative_filter_scope,\n if not len(items):\n return None\n\n event_features = np.zeros((len(items), 3), dtype=np.float16)\n event_features[:, 0] = organic.astype(np.float16)\n event_features[:, 1] = (\n np.minimum(completion, 100.0) / 100.0\n ).astype(np.float16)\n if len(items) > 1:\n gaps = np.maximum(0, timestamps[1:] - timestamps[:-1])\n event_features[1:, 2] = np.exp(\n -gaps.astype(np.float64) / (60.0 * base.DAY)\n ).astype(np.float16)\n\n activity = causal_activity_features(\n timestamps, items, organic, completion\n )\n states = np.zeros(len(items), dtype=np.uint32)\n active_likes: dict[int, int] = {}\n active_dislikes: dict[int, int] = {}\n explicit_position = 0\n current_state = 0\n for row, timestamp in enumerate(timestamps):\n changed = False\n while (\n explicit_position < len(explicit.timestamps)\n and int(explicit.timestamps[explicit_position]) < int(timestamp)\n ):\n item = int(explicit.items[explicit_position])\n event_timestamp = int(explicit.timestamps[explicit_position])\n destination = (\n active_likes\n if int(explicit.channel[explicit_position]) == 0\n else active_dislikes\n )\n if int(explicit.active[explicit_position]) == 1:\n destination[item] = event_timestamp\n else:\n destination.pop(item, None)\n explicit_position += 1\n changed = True\n if changed:\n current_state = append_state_snapshot(\n active_likes,\n active_dislikes,\n n_items,\n liked_states,\n disliked_states,\n state_counts,\n )\n states[row] = np.uint32(current_state)\n return timestamps, items.copy(), event_features, activity, states\n\n\ndef final_user_features(\n timestamps: np.ndarray,\n items: np.ndarray,\n event_features: np.ndarray,\n history_end: int,\n like_count: int,\n dislike_count: int,\n) -> np.ndarray:\n start = int(\n np.searchsorted(\n timestamps,\n history_end - base.PROFILE_LOOKBACK_DAYS * base.DAY,\n side=\"left\",\n )\n )\n result = np.zeros(6, dtype=np.float32)\n recent_items = items[start:]\n if len(recent_items):\n result[0] = np.log1p(len(recent_items)) / 8.0\n result[1] = np.log1p(len(np.unique(recent_items))) / 6.0\n result[2] = float(np.mean(event_features[start:, 0]))\n result[3] = float(np.mean(event_features[start:, 1]))\n result[4] = np.log1p(like_count) / 6.0\n result[5] = np.log1p(dislike_count) / 4.0\n return result\n\n\ndef prepare_data(\n public: Path, stage: str, config: Config\n) -> tuple[SequenceData, CandidateSet]:\n data = public / \"data\"\n catalog = pd.read_parquet(data / \"candidate_catalog.parquet\").sort_values(\n \"popularity_rank\"\n )\n item_ids = catalog.item_id.to_numpy(dtype=np.int64, copy=True)\n item_index = pd.Index(item_ids)\n n_items = len(item_ids)\n if n_items >= np.iinfo(np.uint16).max:\n raise RuntimeError(\"uint16 item storage is too small for this catalog\")\n\n if stage == \"validation\":\n prefixes = (\"train\",)\n history_end = base.TRAIN_END\n candidate_name = \"validation_eligible_candidates.npz\"\n state_name = \"validation_preference_state.parquet\"\n else:\n prefixes = (\"train\", \"val\")\n history_end = base.VALIDATION_END\n candidate_name = \"test_eligible_candidates.npz\"\n state_name = \"test_preference_state.parquet\"\n eligibility = CandidateSet.read(data / candidate_name)\n if not np.array_equal(eligibility.item_ids, item_ids):\n raise RuntimeError(\"catalog and candidate artifact item order differ\")\n\n explicit_by_user = base.load_explicit_events(data, prefixes, item_index)\n inference_likes, inference_dislikes, like_counts, dislike_counts = (\n base.load_inference_state(\n data / state_name,\n item_index,\n eligibility.uids,\n n_items,\n )\n )\n inference_rows = {int(uid): row for row, uid in enumerate(eligibility.uids)}\n inference_user_features = np.zeros((len(eligibility.uids), 6), dtype=np.float32)\n inference_sequence_rows = np.full(len(eligibility.uids), -1, dtype=np.int32)\n\n # Snapshot zero is the shared empty explicit state.\n liked_states = [np.full(base.LIKE_ITEMS, n_items, dtype=np.uint16)]\n disliked_states = [\n np.full(base.DISLIKE_ITEMS, n_items, dtype=np.uint16)\n ]\n state_counts: list[tuple[int, int]] = [(0, 0)]\n user_ids: list[int] = []\n offsets = [0]\n timestamp_parts: list[np.ndarray] = []\n item_parts: list[np.ndarray] = []\n event_parts: list[np.ndarray] = []\n activity_parts: list[np.ndarray] = []\n state_parts: list[np.ndarray] = []\n\n listen_paths = [data / f\"{prefix}_listens.parquet\" for prefix in prefixes]\n started = time.monotonic()\n for seen_users, (uid, listens) in enumerate(\n base.merge_listen_streams(listen_paths, item_index), start=1\n ):\n parts = build_sequence_parts(\n listens,\n explicit_by_user.get(uid, base.empty_explicit_events()),\n n_items,\n liked_states,\n disliked_states,\n state_counts,\n )\n if parts is None:\n continue\n sequence_row = len(user_ids)\n user_ids.append(uid)\n timestamp_parts.append(parts[0].astype(np.uint32, copy=False))\n item_parts.append(parts[1])\n event_parts.append(parts[2])\n activity_parts.append(parts[3])\n state_parts.append(parts[4])\n offsets.append(offsets[-1] + len(parts[1]))\n inference_row = inference_rows.get(uid)\n if inference_row is not None:\n inference_sequence_rows[inference_row] = sequence_row\n inference_user_features[inference_row] = final_user_features(\n parts[0],\n parts[1],\n parts[2],\n history_end,\n int(like_counts[inference_row]),\n int(dislike_counts[inference_row]),\n )\n if seen_users % 1_000 == 0:\n log(\n f\"prepared users={seen_users:,} tokens={offsets[-1]:,} \"\n f\"states={len(liked_states):,} \"\n f\"elapsed={time.monotonic() - started:.1f}s\"\n )\n\n missing = np.flatnonzero(inference_sequence_rows < 0)\n if len(missing):\n inference_user_features[missing, 4] = np.log1p(like_counts[missing]) / 6.0\n inference_user_features[missing, 5] = (\n np.log1p(dislike_counts[missing]) / 4.0\n )\n log(f\"target users without qualifying catalog sequence={len(missing):,}\")\n\n prepared = SequenceData(\n item_ids=item_ids,\n item_features=base.build_item_features(catalog),\n user_ids=np.asarray(user_ids, dtype=np.int64),\n offsets=np.asarray(offsets, dtype=np.int64),\n sequence_timestamps=np.concatenate(timestamp_parts),\n sequence_items=np.concatenate(item_parts),\n event_features=np.concatenate(event_parts),\n activity_features=np.concatenate(activity_parts),\n state_ids=np.concatenate(state_parts),\n liked_states=np.stack(liked_states),\n disliked_states=np.stack(disliked_states),\n state_counts=np.asarray(state_counts, dtype=np.int32),\n inference_user_rows=inference_sequence_rows,\n inference_liked_items=inference_likes,\n inference_disliked_items=inference_dislikes,\n inference_user_features=inference_user_features,\n )\n log(\n f\"prepared stage={stage} sequence_users={len(prepared.user_ids):,} \"\n f\"tokens={len(prepared.sequence_items):,} \"\n f\"next_item_targets={prepared.target_count:,} \"\n f\"states={len(prepared.liked_states):,}\"\n )\n return prepared, eligibility\n\n\ndef cache_metadata(stage: str, config: Config) -> dict[str, object]:\n return {\n \"cache_version\": CACHE_VERSION,\n \"stage\": stage,\n \"positive_listen_threshold\": base.LISTEN_POSITIVE_THRESHOLD,\n \"profile_lookback_days\": base.PROFILE_LOOKBACK_DAYS,\n \"event_time_feature\": \"exp-negative-gap-over-60-days\",\n }\n\n\ndef save_cache(\n path: Path, prepared: SequenceData, stage: str, config: Config\n) -> None:\n path.parent.mkdir(parents=True, exist_ok=True)\n np.savez(\n path,\n metadata=np.asarray(json.dumps(cache_metadata(stage, config), sort_keys=True)),\n **prepared.__dict__,\n )\n log(f\"saved public-derived full-sequence cache: {path}\")\n\n\ndef load_cache(path: Path, stage: str, config: Config) -> SequenceData:\n with np.load(path, allow_pickle=False) as artifact:\n actual = json.loads(str(artifact[\"metadata\"].item()))\n expected = cache_metadata(stage, config)\n if actual != expected:\n raise RuntimeError(f\"sequence cache differs: {actual} != {expected}\")\n values = {\n field: artifact[field].copy()\n for field in SequenceData.__dataclass_fields__\n }\n prepared = SequenceData(**values)\n log(\n f\"loaded full-sequence cache: tokens={len(prepared.sequence_items):,} \"\n f\"targets={prepared.target_count:,}\"\n )\n return prepared\n\n\ndef load_or_prepare(\n public: Path,\n stage: str,\n config: Config,\n cache: Path | None,\n) -> tuple[SequenceData, CandidateSet]:\n candidate_name = (\n \"validation_eligible_candidates.npz\"\n if stage == \"validation\"\n else \"test_eligible_candidates.npz\"\n )\n eligibility = CandidateSet.read(public / \"data\" / candidate_name)\n if cache is not None and cache.is_file():\n prepared = load_cache(cache, stage, config)\n if not np.array_equal(prepared.item_ids, eligibility.item_ids):\n raise RuntimeError(\"cached catalog differs from candidate artifact\")\n return prepared, eligibility\n prepared, built_eligibility = prepare_data(public, stage, config)\n if cache is not None:\n save_cache(cache, prepared, stage, config)\n return prepared, built_eligibility\n\n\ndef build_item_correction(\n prepared: SequenceData,\n config: Config,\n target_starts: np.ndarray,\n target_counts: np.ndarray,\n) -> torch.Tensor:\n \"\"\"Estimate in-batch propensities from the selected training targets.\"\"\"\n counts = np.zeros(len(prepared.item_ids), dtype=np.float64)\n for user_row, count in enumerate(target_counts):\n if not count:\n continue\n offset = int(prepared.offsets[user_row])\n target_start = int(target_starts[user_row])\n counts += np.bincount(\n prepared.sequence_items[\n offset\n + target_start\n + 1 : offset\n + target_start\n + int(count)\n + 1\n ].astype(np.int64),\n minlength=len(prepared.item_ids),\n )\n total = float(counts.sum())\n probabilities = (counts + 1.0) / (total + len(counts))\n raw = np.power(\n (1.0 / len(counts)) / probabilities,\n config.inbatch_correction_power,\n )\n normalization = float(np.sum(counts * raw) / max(total, 1.0))\n correction = np.clip(\n raw / max(normalization, 1.0e-12),\n config.inbatch_correction_min,\n config.inbatch_correction_max,\n ).astype(np.float32)\n mean_on_targets = float(np.sum(counts * correction) / max(total, 1.0))\n log(\n f\"inbatch correction: min={correction.min():.3f} \"\n f\"mean_on_targets={mean_on_targets:.3f} max={correction.max():.3f}\"\n )\n return torch.from_numpy(correction)\n\n\ndef build_known_positive_filter(\n prepared: SequenceData, training_counts: np.ndarray\n) -> np.ndarray:\n \"\"\"Pack each user's complete public-training positive-item set.\"\"\"\n width = (len(prepared.item_ids) + 7) // 8\n packed = np.zeros((len(prepared.user_ids), width), dtype=np.uint8)\n retained = 0\n for user_row, target_count in enumerate(training_counts):\n if not target_count:\n continue\n offset = int(prepared.offsets[user_row])\n # T adjacent targets consume sequence positions 0 through T. The\n # chronological BCE suffix starts at T + 1 and is intentionally absent.\n items = np.unique(\n prepared.sequence_items[\n offset : offset + int(target_count) + 1\n ].astype(np.int64)\n )\n byte_columns = items >> 3\n bit_values = np.left_shift(\n np.uint8(1), (items & 7).astype(np.uint8)\n )\n np.bitwise_or.at(packed[user_row], byte_columns, bit_values)\n retained += len(items)\n log(\n f\"full-history false-negative filter: users={len(prepared.user_ids):,} \"\n f\"unique_user_items={retained:,} bytes={packed.nbytes:,}\"\n )\n return packed\n\n\ndef invalid_full_history_negatives(\n negatives: np.ndarray,\n targets: np.ndarray,\n user_rows: np.ndarray,\n known_positive_packed: np.ndarray,\n liked_items: np.ndarray,\n) -> np.ndarray:\n \"\"\"Return row-level invalid negatives using the complete training prefix.\"\"\"\n byte_columns = negatives >> 3\n bit_columns = negatives & 7\n known = known_positive_packed[user_rows[:, None], byte_columns]\n invalid = ((known >> bit_columns) & 1).astype(bool)\n invalid |= negatives == targets[:, None]\n invalid |= (negatives[:, :, None] == liked_items[:, None, :]).any(axis=2)\n for column in range(1, negatives.shape[1]):\n invalid[:, column] |= (\n negatives[:, column, None] == negatives[:, :column]\n ).any(axis=1)\n return invalid\n\n\ndef draw_full_history_filtered_negatives(\n targets: torch.Tensor,\n user_rows: np.ndarray,\n liked_items: torch.Tensor,\n known_positive_packed: np.ndarray,\n n_items: int,\n config: Config,\n item_correction: torch.Tensor,\n generator: torch.Generator,\n device: torch.device,\n) -> tuple[torch.Tensor, torch.Tensor]:\n \"\"\"Draw mixed negatives and reject all known public-training positives.\"\"\"\n targets_cpu = targets.detach().cpu().numpy().astype(np.int64, copy=False)\n liked_cpu = liked_items.detach().cpu().numpy().astype(np.int64, copy=False)\n batch_size = len(targets_cpu)\n uniform = torch.randint(\n n_items,\n (batch_size, config.uniform_negatives),\n generator=generator,\n device=\"cpu\",\n )\n inbatch = torch.empty((batch_size, 0), dtype=torch.long)\n corrected_count = 0\n if config.inbatch_negatives:\n unique_targets = torch.unique(targets.detach().cpu())\n corrected_count = min(\n config.inbatch_negatives, max(len(unique_targets) - 1, 0)\n )\n if corrected_count:\n priorities = torch.rand(\n (batch_size, len(unique_targets)), generator=generator\n )\n priorities.masked_fill_(\n unique_targets[None, :].eq(\n torch.from_numpy(targets_cpu)[:, None]\n ),\n -1.0,\n )\n columns = torch.topk(\n priorities, k=corrected_count, dim=1\n ).indices\n inbatch = unique_targets[columns]\n if corrected_count < config.inbatch_negatives:\n filler = torch.randint(\n n_items,\n (batch_size, config.inbatch_negatives - corrected_count),\n generator=generator,\n device=\"cpu\",\n )\n inbatch = torch.cat([inbatch, filler], dim=1)\n\n negatives = torch.cat([uniform, inbatch], dim=1).numpy()\n correction = np.ones_like(negatives, dtype=np.float32)\n if corrected_count:\n correction[\n :, config.uniform_negatives : config.uniform_negatives + corrected_count\n ] = item_correction[inbatch[:, :corrected_count]].numpy()\n for _ in range(32):\n invalid = invalid_full_history_negatives(\n negatives,\n targets_cpu,\n user_rows,\n known_positive_packed,\n liked_cpu,\n )\n if not np.any(invalid):\n return (\n torch.from_numpy(negatives).to(device),\n torch.from_numpy(correction).to(device),\n )\n replacements = torch.randint(\n n_items,\n (int(invalid.sum()),),\n generator=generator,\n device=\"cpu\",\n ).numpy()\n negatives[invalid] = replacements\n # Rejected in-batch items become ordinary uniform draws.\n correction[invalid] = 1.0\n invalid = invalid_full_history_negatives(\n negatives,\n targets_cpu,\n user_rows,\n known_positive_packed,\n liked_cpu,\n )\n if np.any(invalid):\n raise RuntimeError(\"full-history negative resampling failed\")\n return (\n torch.from_numpy(negatives).to(device),\n torch.from_numpy(correction).to(device),\n )\n\n\ndef draw_training_negatives(\n positive_items: torch.Tensor,\n known_history: torch.Tensor,\n liked_items: torch.Tensor,\n user_rows: np.ndarray,\n known_positive_packed: np.ndarray | None,\n prepared: SequenceData,\n config: Config,\n item_correction: torch.Tensor,\n generator: torch.Generator,\n device: torch.device,\n) -> tuple[torch.Tensor, torch.Tensor]:\n if config.negative_filter_scope == \"full-training-history\":\n if known_positive_packed is None:\n raise RuntimeError(\"full-history negative filter was not built\")\n return draw_full_history_filtered_negatives(\n positive_items,\n user_rows,\n liked_items,\n known_positive_packed,\n len(prepared.item_ids),\n config,\n item_correction,\n generator,\n device,\n )\n return base.draw_filtered_mixed_negatives(\n positive_items,\n known_history,\n liked_items,\n len(prepared.item_ids),\n config.uniform_negatives,\n config.inbatch_negatives,\n item_correction,\n generator,\n device,\n )\n input_size=config.hidden_dim,\n hidden_size=dim,\n num_layers=1,\n batch_first=True,\n )\n self.user_tower = nn.Sequential(\n nn.Linear(dim * 3 + 6, config.user_hidden_1),\n nn.GELU(),\n nn.Dropout(config.dropout),\n nn.Linear(config.user_hidden_1, config.user_hidden_2),\n nn.GELU(),\n nn.Dropout(config.dropout),\n nn.Linear(config.user_hidden_2, dim),\n )\n self.candidate_embedding = nn.Embedding(n_items, dim)\n self.register_buffer(\n \"item_features\", torch.from_numpy(item_features), persistent=True\n )\n self.item_tower = nn.Sequential(\n nn.Linear(dim + n_item_features, config.item_hidden_1),\n nn.GELU(),\n nn.Dropout(config.dropout),\n nn.Linear(config.item_hidden_1, config.item_hidden_2),\n nn.GELU(),\n nn.Dropout(config.dropout),\n nn.Linear(config.item_hidden_2, dim),\n )\n nn.init.normal_(self.context_embedding.weight, std=0.02)\n nn.init.normal_(self.candidate_embedding.weight, std=0.02)\n with torch.no_grad():\n self.context_embedding.weight[n_items].zero_()\n\n def _mean_context_channel(self, items: torch.Tensor) -> torch.Tensor:\n mask = items.ne(self.n_items)\n vectors = self.context_embedding(items)\n return (vectors * mask.unsqueeze(-1)).sum(dim=1) / mask.sum(\n dim=1, keepdim=True\n ).clamp_min(1)\n\n def encode_items(self, items: torch.Tensor) -> torch.Tensor:\n inputs = torch.cat(\n [self.candidate_embedding(items), self.item_features[items]], dim=1\n )\n return self.item_tower(inputs)\n\n\ndef invalid_negative_mask(\n negatives: torch.Tensor,\n targets: torch.Tensor,\n context_items: torch.Tensor,\n liked_items: torch.Tensor,\n) -> torch.Tensor:\n \"\"\"Identify targets, known positives, active likes, and row duplicates.\"\"\"\n invalid = negatives.eq(targets[:, None])\n invalid |= negatives[:, :, None].eq(context_items[:, None, :]).any(dim=2)\n invalid |= negatives[:, :, None].eq(liked_items[:, None, :]).any(dim=2)\n for column in range(1, negatives.shape[1]):\n invalid[:, column] |= negatives[:, column, None].eq(\n negatives[:, :column]\n ).any(dim=1)\n return invalid\n\n\ndef draw_filtered_mixed_negatives(\n targets: torch.Tensor,\n context_items: torch.Tensor,\n liked_items: torch.Tensor,\n n_items: int,\n uniform_count: int,\n inbatch_count: int,\n item_correction: torch.Tensor,\n generator: torch.Generator,\n device: torch.device,\n) -> tuple[torch.Tensor, torch.Tensor]:\n \"\"\"Draw uniform and unique-target in-batch negatives.\"\"\"\n batch_size = len(targets)\n uniform = torch.randint(\n n_items,\n (batch_size, uniform_count),\n generator=generator,\n device=\"cpu\",\n )\n inbatch = torch.empty((batch_size, 0), dtype=torch.long, device=\"cpu\")\n corrected_count = 0\n if inbatch_count:\n unique_targets = torch.unique(targets.detach().cpu())\n corrected_count = min(inbatch_count, max(len(unique_targets) - 1, 0))\n if corrected_count:\n priorities = torch.rand(\n (batch_size, len(unique_targets)), generator=generator\n )\n priorities.masked_fill_(\n unique_targets[None, :].eq(targets.detach().cpu()[:, None]),\n -1.0,\n )\n columns = torch.topk(priorities, k=corrected_count, dim=1).indices\n inbatch = unique_targets[columns]\n if corrected_count < inbatch_count:\n filler = torch.randint(\n n_items,\n (batch_size, inbatch_count - corrected_count),\n generator=generator,\n device=\"cpu\",\n )\n inbatch = torch.cat([inbatch, filler], dim=1)\n\n negatives_cpu = torch.cat([uniform, inbatch], dim=1)\n correction_cpu = torch.ones_like(negatives_cpu, dtype=torch.float32)\n if corrected_count:\n correction_cpu[\n :, uniform_count : uniform_count + corrected_count\n ] = item_correction[inbatch[:, :corrected_count]]\n negatives = negatives_cpu.to(device)\n correction = correction_cpu.to(device)\n for _ in range(32):\n invalid = invalid_negative_mask(\n negatives, targets, context_items, liked_items\n )\n if not bool(invalid.any()):\n return negatives, correction\n replacements = torch.randint(\n n_items,\n (int(invalid.sum().detach().cpu()),),\n generator=generator,\n device=\"cpu\",\n ).to(device)\n negatives[invalid] = replacements\n correction[invalid] = 1.0\n invalid = invalid_negative_mask(\n negatives, targets, context_items, liked_items\n )\n if bool(invalid.any()):\n raise RuntimeError(\"mixed negative resampling failed false-negative filter\")\n return negatives, correction\n",
"stderr": "",
"aggregated_output": "/task/current_model/two_tower_model.py:40: uniform_negatives: int = 24\n/task/current_model/two_tower_model.py:41: inbatch_negatives: int = 8\n/task/current_model/two_tower_model.py:42: inbatch_correction_power: float = 0.5\n/task/current_model/two_tower_model.py:43: inbatch_correction_min: float = 0.25\n/task/current_model/two_tower_model.py:44: inbatch_correction_max: float = 4.0\n/task/current_model/two_tower_model.py:346:def invalid_negative_mask(\n/task/current_model/two_tower_model.py:347: negatives: torch.Tensor,\n/task/current_model/two_tower_model.py:353: invalid = negatives.eq(targets[:, None])\n/task/current_model/two_tower_model.py:354: invalid |= negatives[:, :, None].eq(context_items[:, None, :]).any(dim=2)\n/task/current_model/two_tower_model.py:355: invalid |= negatives[:, :, None].eq(liked_items[:, None, :]).any(dim=2)\n/task/current_model/two_tower_model.py:356: for column in range(1, negatives.shape[1]):\n/task/current_model/two_tower_model.py:357: invalid[:, column] |= negatives[:, column, None].eq(\n/task/current_model/two_tower_model.py:358: negatives[:, :column]\n/task/current_model/two_tower_model.py:363:def draw_filtered_mixed_negatives(\n/task/current_model/two_tower_model.py:369: inbatch_count: int,\n/task/current_model/two_tower_model.py:374: \"\"\"Draw uniform and unique-target in-batch negatives.\"\"\"\n/task/current_model/two_tower_model.py:382: inbatch = torch.empty((batch_size, 0), dtype=torch.long, device=\"cpu\")\n/task/current_model/two_tower_model.py:384: if inbatch_count:\n/task/current_model/two_tower_model.py:386: corrected_count = min(inbatch_count, max(len(unique_targets) - 1, 0))\n/task/current_model/two_tower_model.py:396: inbatch = unique_targets[columns]\n/task/current_model/two_tower_model.py:397: if corrected_count < inbatch_count:\n/task/current_model/two_tower_model.py:400: (batch_size, inbatch_count - corrected_count),\n/task/current_model/two_tower_model.py:404: inbatch = torch.cat([inbatch, filler], dim=1)\n/task/current_model/two_tower_model.py:406: negatives_cpu = torch.cat([uniform, inbatch], dim=1)\n/task/current_model/two_tower_model.py:407: correction_cpu = torch.ones_like(negatives_cpu, dtype=torch.float32)\n/task/current_model/two_tower_model.py:411: ] = item_correction[inbatch[:, :corrected_count]]\n/task/current_model/two_tower_model.py:412: negatives = negatives_cpu.to(device)\n/task/current_model/two_tower_model.py:415: invalid = invalid_negative_mask(\n/task/current_model/two_tower_model.py:416: negatives, targets, context_items, liked_items\n/task/current_model/two_tower_model.py:419: return negatives, correction\n/task/current_model/two_tower_model.py:426: negatives[invalid] = replacements\n/task/current_model/two_tower_model.py:428: invalid = invalid_negative_mask(\n/task/current_model/two_tower_model.py:429: negatives, targets, context_items, liked_items\n/task/current_model/two_tower_model.py:432: raise RuntimeError(\"mixed negative resampling failed false-negative filter\")\n/task/current_model/two_tower_model.py:433: return negatives, correction\n/task/current_model/train_current_model.py:6:loss. The fast default retains each user's most recent 128 training targets and\n/task/current_model/train_current_model.py:9:computes all of its timestep losses and carries the detached GRU hidden state\n/task/current_model/train_current_model.py:13:normalization, raw dot-product retrieval, sample-weighted binary cross entropy,\n/task/current_model/train_current_model.py:14:and 24 uniform plus eight propensity-corrected unique in-batch negatives. Every\n/task/current_model/train_current_model.py:20:import argparse\n/task/current_model/train_current_model.py:55: loss_microbatch_size: int = 65_536\n/task/current_model/train_current_model.py:56: uniform_negatives: int = 24\n/task/current_model/train_current_model.py:57: inbatch_negatives: int = 8\n/task/current_model/train_current_model.py:58: inbatch_correction_power: float = 0.5\n/task/current_model/train_current_model.py:59: inbatch_correction_min: float = 0.25\n/task/current_model/train_current_model.py:60: inbatch_correction_max: float = 4.0\n/task/current_model/train_current_model.py:61: negative_filter_scope: str = \"full-training-history\"\n/task/current_model/train_current_model.py:152: return base.TowerConfig(\n/task/current_model/train_current_model.py:165: uniform_negatives=config.uniform_negatives,\n/task/current_model/train_current_model.py:166: inbatch_negatives=config.inbatch_negatives,\n/task/current_model/train_current_model.py:167: inbatch_correction_power=config.inbatch_correction_power,\n/task/current_model/train_current_model.py:168: inbatch_correction_min=config.inbatch_correction_min,\n/task/current_model/train_current_model.py:169: inbatch_correction_max=config.inbatch_correction_max,\n/task/current_model/train_current_model.py:342: public: Path, stage: str, config: Config\n/task/current_model/train_current_model.py:354: if stage == \"validation\":\n/task/current_model/train_current_model.py:463: f\"prepared stage={stage} sequence_users={len(prepared.user_ids):,} \"\n/task/current_model/train_current_model.py:471:def cache_metadata(stage: str, config: Config) -> dict[str, object]:\n/task/current_model/train_current_model.py:474: \"stage\": stage,\n/task/current_model/train_current_model.py:477: \"event_time_feature\": \"exp-negative-gap-over-60-days\",\n/task/current_model/train_current_model.py:482: path: Path, prepared: SequenceData, stage: str, config: Config\n/task/current_model/train_current_model.py:487: metadata=np.asarray(json.dumps(cache_metadata(stage, config), sort_keys=True)),\n/task/current_model/train_current_model.py:493:def load_cache(path: Path, stage: str, config: Config) -> SequenceData:\n/task/current_model/train_current_model.py:496: expected = cache_metadata(stage, config)\n/task/current_model/train_current_model.py:513: stage: str,\n/task/current_model/train_current_model.py:519: if stage == \"validation\"\n/task/current_model/train_current_model.py:524: prepared = load_cache(cache, stage, config)\n/task/current_model/train_current_model.py:528: prepared, built_eligibility = prepare_data(public, stage, config)\n/task/current_model/train_current_model.py:530: save_cache(cache, prepared, stage, config)\n/task/current_model/train_current_model.py:562: config.inbatch_correction_power,\n/task/current_model/train_current_model.py:567: config.inbatch_correction_min,\n/task/current_model/train_current_model.py:568: config.inbatch_correction_max,\n/task/current_model/train_current_model.py:572: f\"inbatch correction: min={correction.min():.3f} \"\n/task/current_model/train_current_model.py:603: f\"full-history false-negative filter: users={len(prepared.user_ids):,} \"\n/task/current_model/train_current_model.py:609:def invalid_full_history_negatives(\n/task/current_model/train_current_model.py:610: negatives: np.ndarray,\n/task/current_model/train_current_model.py:616: \"\"\"Return row-level invalid negatives using the complete training prefix.\"\"\"\n/task/current_model/train_current_model.py:617: byte_columns = negatives >> 3\n/task/current_model/train_current_model.py:618: bit_columns = negatives & 7\n/task/current_model/train_current_model.py:621: invalid |= negatives == targets[:, None]\n/task/current_model/train_current_model.py:622: invalid |= (negatives[:, :, None] == liked_items[:, None, :]).any(axis=2)\n/task/current_model/train_current_model.py:623: for column in range(1, negatives.shape[1]):\n/task/current_model/train_current_model.py:625: negatives[:, column, None] == negatives[:, :column]\n/task/current_model/train_current_model.py:630:def draw_full_history_filtered_negatives(\n/task/current_model/train_current_model.py:641: \"\"\"Draw mixed negatives and reject all known public-training positives.\"\"\"\n/task/current_model/train_current_model.py:647: (batch_size, config.uniform_negatives),\n/task/current_model/train_current_model.py:651: inbatch = torch.empty((batch_size, 0), dtype=torch.long)\n/task/current_model/train_current_model.py:653: if config.inbatch_negatives:\n/task/current_model/train_current_model.py:656: config.inbatch_negatives, max(len(unique_targets) - 1, 0)\n/task/current_model/train_current_model.py:671: inbatch = unique_targets[columns]\n/task/current_model/train_current_model.py:672: if corrected_count < config.inbatch_negatives:\n/task/current_model/train_current_model.py:675: (batch_size, config.inbatch_negatives - corrected_count),\n/task/current_model/train_current_model.py:679: inbatch = torch.cat([inbatch, filler], dim=1)\n/task/current_model/train_current_model.py:681: negatives = torch.cat([uniform, inbatch], dim=1).numpy()\n/task/current_model/train_current_model.py:682: correction = np.ones_like(negatives, dtype=np.float32)\n/task/current_model/train_current_model.py:685: :, config.uniform_negatives : config.uniform_negatives + corrected_count\n/task/current_model/train_current_model.py:686: ] = item_correction[inbatch[:, :corrected_count]].numpy()\n/task/current_model/train_current_model.py:688: invalid = invalid_full_history_negatives(\n/task/current_model/train_current_model.py:689: negatives,\n/task/current_model/train_current_model.py:697: torch.from_numpy(negatives).to(device),\n/task/current_model/train_current_model.py:706: negatives[invalid] = replacements\n/task/current_model/train_current_model.py:709: invalid = invalid_full_history_negatives(\n/task/current_model/train_current_model.py:710: negatives,\n/task/current_model/train_current_model.py:717: raise RuntimeError(\"full-history negative resampling failed\")\n/task/current_model/train_current_model.py:719: torch.from_numpy(negatives).to(device),\n/task/current_model/train_current_model.py:724:def draw_training_negatives(\n/task/current_model/train_current_model.py:736: if config.negative_filter_scope == \"full-training-history\":\n/task/current_model/train_current_model.py:738: raise RuntimeError(\"full-history negative filter was not built\")\n/task/current_model/train_current_model.py:739: return draw_full_history_filtered_negatives(\n/task/current_model/train_current_model.py:750: return base.draw_filtered_mixed_negatives(\n/task/current_model/train_current_model.py:755: config.uniform_negatives,\n/task/current_model/train_current_model.py:756: config.inbatch_negatives,\n/task/current_model/train_current_model.py:816: negatives: torch.Tensor,\n/task/current_model/train_current_model.py:818: all_items = torch.cat([positives, negatives.reshape(-1)])\n/task/current_model/train_current_model.py:822: negative_vectors = encoded[inverse[len(positives) :]].reshape(\n/task/current_model/train_current_model.py:823: len(positives), negatives.shape[1], -1\n/task/current_model/train_current_model.py:825: return positive_vectors, negative_vectors\n/task/current_model/train_current_model.py:845:def corrected_negative_loss(\n/task/current_model/train_current_model.py:851: total = config.uniform_negatives + config.inbatch_negatives\n/task/current_model/train_current_model.py:853: if config.uniform_negatives:\n/task/current_model/train_current_model.py:854: result += config.uniform_negatives * raw[\n/task/current_model/train_current_model.py:855: :, : config.uniform_negatives\n/task/current_model/train_current_model.py:857: if config.inbatch_negatives:\n/task/current_model/train_current_model.py:858: selected = slice(config.uniform_negatives, total)\n/task/current_model/train_current_model.py:860: result += config.inbatch_negatives * (\n/task/current_model/train_current_model.py:880: raise RuntimeError(\"chronological target partition produced negative counts\")\n/task/current_model/train_current_model.py:915: public: Path, stage: str, item_ids: np.ndarray\n/task/current_model/train_current_model.py:918: prefixes = (\"train\",) if stage == \"validation\" else (\"train\", \"val\")\n/task/current_model/train_current_model.py:957: stage: str,\n/task/current_model/train_current_model.py:978: public, stage, prepared.item_ids\n/task/current_model/train_current_model.py:981: base.TRAIN_END if stage == \"validation\" else base.VALIDATION_END\n/task/current_model/train_current_model.py:1119: negatives: torch.Tensor,\n/task/current_model/train_current_model.py:1120: negative_weights: torch.Tensor,\n/task/current_model/train_current_model.py:1121: sample_weights: np.ndarray,\n/task/current_model/train_current_model.py:1127: total_weight = float(sample_weights.sum())\n/task/current_model/train_current_model.py:1129: raise RuntimeError(\"target block has no positive sample weight\")\n/task/current_model/train_current_model.py:1132: \"loss_sum\": 0.0,\n/task/current_model/train_current_model.py:1134: \"negative_sum\": 0.0,\n/task/current_model/train_current_model.py:1138: for begin in range(0, target_count, config.loss_microbatch_size):\n/task/current_model/train_current_model.py:1139: end = min(begin + config.loss_microbatch_size, target_count)\n/task/current_model/train_current_model.py:1162: positive_vectors, negative_vectors = candidate_vectors(\n/task/current_model/train_current_model.py:1165: negatives[begin:end],\n/task/current_model/train_current_model.py:1168: negative_logits = torch.einsum(\n/task/current_model/train_current_model.py:1169: \"bd,bkd->bk\", query_vectors, negative_vectors\n/task/current_model/train_current_model.py:1171: positive_loss = F.softplus(-positive_logits)\n/task/current_model/train_current_model.py:1172: negative_loss = corrected_negative_loss(\n/task/current_model/train_current_model.py:1173: negative_logits, negative_weights[begin:end], config\n/task/current_model/train_current_model.py:1176: sample_weights[begin:end].astype(np.float32, copy=False)\n/task/current_model/train_current_model.py:1178: loss_sum = (\n/task/current_model/train_current_model.py:1179: selected_weights * (positive_loss + negative_loss) / 2.0\n/task/current_model/train_current_model.py:1181: if not bool(torch.isfinite(loss_sum)):\n/task/current_model/train_current_model.py:1182: raise RuntimeError(\"dense next-item training produced non-finite loss\")\n/task/current_model/train_current_model.py:1184: (loss_sum / total_weight).backward()\n/task/current_model/train_current_model.py:1188: statistics[\"loss_sum\"] += float(loss_sum.detach().cpu())\n/task/current_model/train_current_model.py:1190: statistics[\"negative_sum\"] += float(negative_logits.detach().sum().cpu())\n/task/current_model/train_current_model.py:1192: (positive_logits.detach()[:, None] > negative_logits.detach()).sum().cpu()\n/task/current_model/train_current_model.py:1227: negative_count = config.uniform_negatives + config.inbatch_negatives\n/task/current_model/train_current_model.py:1228: total_loss = 0.0\n/task/current_model/train_current_model.py:1232: total_negative = 0.0\n/task/current_model/train_current_model.py:1293: negatives, negative_weights = draw_training_negatives(\n/task/current_model/train_current_model.py:1314: negatives,\n/task/current_model/train_current_model.py:1315: negative_weights,\n/task/current_model/train_current_model.py:1327: total_loss += statistics[\"loss_sum\"]\n/task/current_model/train_current_model.py:1331: total_negative += statistics[\"negative_sum\"]\n/task/current_model/train_current_model.py:1337: batch_bce = statistics[\"loss_sum\"] / max(\n/task/current_model/train_current_model.py:1340: running_bce = total_loss / max(total_weight, 1.0e-12)\n/task/current_model/train_current_model.py:1351: \"loss\": total_loss / max(total_weight, 1.0e-12),\n/task/current_model/train_current_model.py:1353: \"negative_logit\": total_negative / max(total_targets * negative_count, 1),\n/task/current_model/train_current_model.py:1355: / max(total_targets * negative_count, 1),\n/task/current_model/train_current_model.py:1388: total_loss = 0.0\n/task/current_model/train_current_model.py:1449: negatives, negative_weights = draw_training_negatives(\n/task/current_model/train_current_model.py:1469: negatives,\n/task/current_model/train_current_model.py:1470: negative_weights,\n/task/current_model/train_current_model.py:1475: total_loss += statistics[\"loss_sum\"]\n/task/current_model/train_current_model.py:1481: \"validation_loss\": total_loss / max(total_weight, 1.0e-12),\n/task/current_model/train_current_model.py:1523: stage: str,\n/task/current_model/train_current_model.py:1560: stage,\n/task/current_model/train_current_model.py:1567: if config.negative_filter_scope == \"full-training-history\"\n/task/current_model/train_current_model.py:1571: best_loss = float(\"inf\")\n/task/current_model/train_current_model.py:1601: \"negative_filter_scope\", \"last-100\"\n/task/current_model/train_current_model.py:1629: stored_filter_scope != config.negative_filter_scope\n/task/current_model/train_current_model.py:1657: prior.pop(\"validation_loss\", None)\n/task/current_model/train_current_model.py:1660: f\"{stored_filter_scope}->{config.negative_filter_scope}, \"\n/task/current_model/train_current_model.py:1666: if \"validation_loss\" not in prior:\n/task/current_model/train_current_model.py:1682: best_loss = float(prior[\"validation_loss\"])\n/task/current_model/train_current_model.py:1687: f\"loss={best_loss:.6f}\"\n/task/current_model/train_current_model.py:1731: diagnostics[\"validation_loss\"]\n/task/current_model/train_current_model.py:1732: < best_loss - config.early_stopping_min_delta\n/task/current_model/train_current_model.py:1735: best_loss = diagnostics[\"validation_loss\"]\n/task/current_model/train_current_model.py:1754: f\"epoch={epoch}/{config.epochs} train_loss={diagnostics['loss']:.5f} \"\n/task/current_model/train_current_model.py:1755: f\"validation_loss={diagnostics['validation_loss']:.5f} \"\n/task/current_model/train_current_model.py:1766: f\"best_epoch={best_epoch} best_validation_loss={best_loss:.6f}\"\n/task/current_model/train_current_model.py:1776: f\"loss={best_loss:.6f}\"\n/task/current_model/train_current_model.py:1923: parser = argparse.ArgumentParser(description=__doc__)\n/task/current_model/train_current_model.py:1925: \"--stage\", choices=(\"validation\", \"submission\"), default=\"validation\"\n/task/current_model/train_current_model.py:1939: \"--loss-microbatch-size\",\n/task/current_model/train_current_model.py:1941: default=Config.loss_microbatch_size,\n/task/current_model/train_current_model.py:1944: \"--negative-filter-scope\",\n/task/current_model/train_current_model.py:1946: default=Config.negative_filter_scope,\n/task/current_model/train_current_model.py:2012: or args.loss_microbatch_size < 1\n/task/current_model/train_current_model.py:2028: config = Config(\n/task/current_model/train_current_model.py:2033: loss_microbatch_size=args.loss_microbatch_size,\n/task/current_model/train_current_model.py:2034: negative_filter_scope=args.negative_filter_scope,\n/task/current_model/train_current_model.py:2057: args.stage,\n/task/current_model/train_current_model.py:2071: \"stage\": args.stage,\n/task/current_model/train_current_model.py:2095: args.stage,\n/task/current_model/train_current_model.py:2115: \"stage\": args.stage,\n/task/current_model/train_current_model.py:2126: \"sequence_encoder\": \"one-layer-GRU-dense-timestep-loss\",\n/task/current_model/train_current_model.py:2129: \"loss_execution\": \"gradient-accumulated-microbatches\",\n/task/current_model/train_current_model.py:2135: \"objective\": \"sample-weighted-BCE-at-every-timestep\",\n/task/current_model/train_current_model.py:2136: \"negative_sampling\": (\n/task/current_model/train_current_model.py:2137: \"24-uniform-plus-8-propensity-corrected-inbatch\"\n/task/current_model/train_current_model.py:2139: \"false_negative_filter\": config.negative_filter_scope,\n if not len(items):\n return None\n\n event_features = np.zeros((len(items), 3), dtype=np.float16)\n event_features[:, 0] = organic.astype(np.float16)\n event_features[:, 1] = (\n np.minimum(completion, 100.0) / 100.0\n ).astype(np.float16)\n if len(items) > 1:\n gaps = np.maximum(0, timestamps[1:] - timestamps[:-1])\n event_features[1:, 2] = np.exp(\n -gaps.astype(np.float64) / (60.0 * base.DAY)\n ).astype(np.float16)\n\n activity = causal_activity_features(\n timestamps, items, organic, completion\n )\n states = np.zeros(len(items), dtype=np.uint32)\n active_likes: dict[int, int] = {}\n active_dislikes: dict[int, int] = {}\n explicit_position = 0\n current_state = 0\n for row, timestamp in enumerate(timestamps):\n changed = False\n while (\n explicit_position < len(explicit.timestamps)\n and int(explicit.timestamps[explicit_position]) < int(timestamp)\n ):\n item = int(explicit.items[explicit_position])\n event_timestamp = int(explicit.timestamps[explicit_position])\n destination = (\n active_likes\n if int(explicit.channel[explicit_position]) == 0\n else active_dislikes\n )\n if int(explicit.active[explicit_position]) == 1:\n destination[item] = event_timestamp\n else:\n destination.pop(item, None)\n explicit_position += 1\n changed = True\n if changed:\n current_state = append_state_snapshot(\n active_likes,\n active_dislikes,\n n_items,\n liked_states,\n disliked_states,\n state_counts,\n )\n states[row] = np.uint32(current_state)\n return timestamps, items.copy(), event_features, activity, states\n\n\ndef final_user_features(\n timestamps: np.ndarray,\n items: np.ndarray,\n event_features: np.ndarray,\n history_end: int,\n like_count: int,\n dislike_count: int,\n) -> np.ndarray:\n start = int(\n np.searchsorted(\n timestamps,\n history_end - base.PROFILE_LOOKBACK_DAYS * base.DAY,\n side=\"left\",\n )\n )\n result = np.zeros(6, dtype=np.float32)\n recent_items = items[start:]\n if len(recent_items):\n result[0] = np.log1p(len(recent_items)) / 8.0\n result[1] = np.log1p(len(np.unique(recent_items))) / 6.0\n result[2] = float(np.mean(event_features[start:, 0]))\n result[3] = float(np.mean(event_features[start:, 1]))\n result[4] = np.log1p(like_count) / 6.0\n result[5] = np.log1p(dislike_count) / 4.0\n return result\n\n\ndef prepare_data(\n public: Path, stage: str, config: Config\n) -> tuple[SequenceData, CandidateSet]:\n data = public / \"data\"\n catalog = pd.read_parquet(data / \"candidate_catalog.parquet\").sort_values(\n \"popularity_rank\"\n )\n item_ids = catalog.item_id.to_numpy(dtype=np.int64, copy=True)\n item_index = pd.Index(item_ids)\n n_items = len(item_ids)\n if n_items >= np.iinfo(np.uint16).max:\n raise RuntimeError(\"uint16 item storage is too small for this catalog\")\n\n if stage == \"validation\":\n prefixes = (\"train\",)\n history_end = base.TRAIN_END\n candidate_name = \"validation_eligible_candidates.npz\"\n state_name = \"validation_preference_state.parquet\"\n else:\n prefixes = (\"train\", \"val\")\n history_end = base.VALIDATION_END\n candidate_name = \"test_eligible_candidates.npz\"\n state_name = \"test_preference_state.parquet\"\n eligibility = CandidateSet.read(data / candidate_name)\n if not np.array_equal(eligibility.item_ids, item_ids):\n raise RuntimeError(\"catalog and candidate artifact item order differ\")\n\n explicit_by_user = base.load_explicit_events(data, prefixes, item_index)\n inference_likes, inference_dislikes, like_counts, dislike_counts = (\n base.load_inference_state(\n data / state_name,\n item_index,\n eligibility.uids,\n n_items,\n )\n )\n inference_rows = {int(uid): row for row, uid in enumerate(eligibility.uids)}\n inference_user_features = np.zeros((len(eligibility.uids), 6), dtype=np.float32)\n inference_sequence_rows = np.full(len(eligibility.uids), -1, dtype=np.int32)\n\n # Snapshot zero is the shared empty explicit state.\n liked_states = [np.full(base.LIKE_ITEMS, n_items, dtype=np.uint16)]\n disliked_states = [\n np.full(base.DISLIKE_ITEMS, n_items, dtype=np.uint16)\n ]\n state_counts: list[tuple[int, int]] = [(0, 0)]\n user_ids: list[int] = []\n offsets = [0]\n timestamp_parts: list[np.ndarray] = []\n item_parts: list[np.ndarray] = []\n event_parts: list[np.ndarray] = []\n activity_parts: list[np.ndarray] = []\n state_parts: list[np.ndarray] = []\n\n listen_paths = [data / f\"{prefix}_listens.parquet\" for prefix in prefixes]\n started = time.monotonic()\n for seen_users, (uid, listens) in enumerate(\n base.merge_listen_streams(listen_paths, item_index), start=1\n ):\n parts = build_sequence_parts(\n listens,\n explicit_by_user.get(uid, base.empty_explicit_events()),\n n_items,\n liked_states,\n disliked_states,\n state_counts,\n )\n if parts is None:\n continue\n sequence_row = len(user_ids)\n user_ids.append(uid)\n timestamp_parts.append(parts[0].astype(np.uint32, copy=False))\n item_parts.append(parts[1])\n event_parts.append(parts[2])\n activity_parts.append(parts[3])\n state_parts.append(parts[4])\n offsets.append(offsets[-1] + len(parts[1]))\n inference_row = inference_rows.get(uid)\n if inference_row is not None:\n inference_sequence_rows[inference_row] = sequence_row\n inference_user_features[inference_row] = final_user_features(\n parts[0],\n parts[1],\n parts[2],\n history_end,\n int(like_counts[inference_row]),\n int(dislike_counts[inference_row]),\n )\n if seen_users % 1_000 == 0:\n log(\n f\"prepared users={seen_users:,} tokens={offsets[-1]:,} \"\n f\"states={len(liked_states):,} \"\n f\"elapsed={time.monotonic() - started:.1f}s\"\n )\n\n missing = np.flatnonzero(inference_sequence_rows < 0)\n if len(missing):\n inference_user_features[missing, 4] = np.log1p(like_counts[missing]) / 6.0\n inference_user_features[missing, 5] = (\n np.log1p(dislike_counts[missing]) / 4.0\n )\n log(f\"target users without qualifying catalog sequence={len(missing):,}\")\n\n prepared = SequenceData(\n item_ids=item_ids,\n item_features=base.build_item_features(catalog),\n user_ids=np.asarray(user_ids, dtype=np.int64),\n offsets=np.asarray(offsets, dtype=np.int64),\n sequence_timestamps=np.concatenate(timestamp_parts),\n sequence_items=np.concatenate(item_parts),\n event_features=np.concatenate(event_parts),\n activity_features=np.concatenate(activity_parts),\n state_ids=np.concatenate(state_parts),\n liked_states=np.stack(liked_states),\n disliked_states=np.stack(disliked_states),\n state_counts=np.asarray(state_counts, dtype=np.int32),\n inference_user_rows=inference_sequence_rows,\n inference_liked_items=inference_likes,\n inference_disliked_items=inference_dislikes,\n inference_user_features=inference_user_features,\n )\n log(\n f\"prepared stage={stage} sequence_users={len(prepared.user_ids):,} \"\n f\"tokens={len(prepared.sequence_items):,} \"\n f\"next_item_targets={prepared.target_count:,} \"\n f\"states={len(prepared.liked_states):,}\"\n )\n return prepared, eligibility\n\n\ndef cache_metadata(stage: str, config: Config) -> dict[str, object]:\n return {\n \"cache_version\": CACHE_VERSION,\n \"stage\": stage,\n \"positive_listen_threshold\": base.LISTEN_POSITIVE_THRESHOLD,\n \"profile_lookback_days\": base.PROFILE_LOOKBACK_DAYS,\n \"event_time_feature\": \"exp-negative-gap-over-60-days\",\n }\n\n\ndef save_cache(\n path: Path, prepared: SequenceData, stage: str, config: Config\n) -> None:\n path.parent.mkdir(parents=True, exist_ok=True)\n np.savez(\n path,\n metadata=np.asarray(json.dumps(cache_metadata(stage, config), sort_keys=True)),\n **prepared.__dict__,\n )\n log(f\"saved public-derived full-sequence cache: {path}\")\n\n\ndef load_cache(path: Path, stage: str, config: Config) -> SequenceData:\n with np.load(path, allow_pickle=False) as artifact:\n actual = json.loads(str(artifact[\"metadata\"].item()))\n expected = cache_metadata(stage, config)\n if actual != expected:\n raise RuntimeError(f\"sequence cache differs: {actual} != {expected}\")\n values = {\n field: artifact[field].copy()\n for field in SequenceData.__dataclass_fields__\n }\n prepared = SequenceData(**values)\n log(\n f\"loaded full-sequence cache: tokens={len(prepared.sequence_items):,} \"\n f\"targets={prepared.target_count:,}\"\n )\n return prepared\n\n\ndef load_or_prepare(\n public: Path,\n stage: str,\n config: Config,\n cache: Path | None,\n) -> tuple[SequenceData, CandidateSet]:\n candidate_name = (\n \"validation_eligible_candidates.npz\"\n if stage == \"validation\"\n else \"test_eligible_candidates.npz\"\n )\n eligibility = CandidateSet.read(public / \"data\" / candidate_name)\n if cache is not None and cache.is_file():\n prepared = load_cache(cache, stage, config)\n if not np.array_equal(prepared.item_ids, eligibility.item_ids):\n raise RuntimeError(\"cached catalog differs from candidate artifact\")\n return prepared, eligibility\n prepared, built_eligibility = prepare_data(public, stage, config)\n if cache is not None:\n save_cache(cache, prepared, stage, config)\n return prepared, built_eligibility\n\n\ndef build_item_correction(\n prepared: SequenceData,\n config: Config,\n target_starts: np.ndarray,\n target_counts: np.ndarray,\n) -> torch.Tensor:\n \"\"\"Estimate in-batch propensities from the selected training targets.\"\"\"\n counts = np.zeros(len(prepared.item_ids), dtype=np.float64)\n for user_row, count in enumerate(target_counts):\n if not count:\n continue\n offset = int(prepared.offsets[user_row])\n target_start = int(target_starts[user_row])\n counts += np.bincount(\n prepared.sequence_items[\n offset\n + target_start\n + 1 : offset\n + target_start\n + int(count)\n + 1\n ].astype(np.int64),\n minlength=len(prepared.item_ids),\n )\n total = float(counts.sum())\n probabilities = (counts + 1.0) / (total + len(counts))\n raw = np.power(\n (1.0 / len(counts)) / probabilities,\n config.inbatch_correction_power,\n )\n normalization = float(np.sum(counts * raw) / max(total, 1.0))\n correction = np.clip(\n raw / max(normalization, 1.0e-12),\n config.inbatch_correction_min,\n config.inbatch_correction_max,\n ).astype(np.float32)\n mean_on_targets = float(np.sum(counts * correction) / max(total, 1.0))\n log(\n f\"inbatch correction: min={correction.min():.3f} \"\n f\"mean_on_targets={mean_on_targets:.3f} max={correction.max():.3f}\"\n )\n return torch.from_numpy(correction)\n\n\ndef build_known_positive_filter(\n prepared: SequenceData, training_counts: np.ndarray\n) -> np.ndarray:\n \"\"\"Pack each user's complete public-training positive-item set.\"\"\"\n width = (len(prepared.item_ids) + 7) // 8\n packed = np.zeros((len(prepared.user_ids), width), dtype=np.uint8)\n retained = 0\n for user_row, target_count in enumerate(training_counts):\n if not target_count:\n continue\n offset = int(prepared.offsets[user_row])\n # T adjacent targets consume sequence positions 0 through T. The\n # chronological BCE suffix starts at T + 1 and is intentionally absent.\n items = np.unique(\n prepared.sequence_items[\n offset : offset + int(target_count) + 1\n ].astype(np.int64)\n )\n byte_columns = items >> 3\n bit_values = np.left_shift(\n np.uint8(1), (items & 7).astype(np.uint8)\n )\n np.bitwise_or.at(packed[user_row], byte_columns, bit_values)\n retained += len(items)\n log(\n f\"full-history false-negative filter: users={len(prepared.user_ids):,} \"\n f\"unique_user_items={retained:,} bytes={packed.nbytes:,}\"\n )\n return packed\n\n\ndef invalid_full_history_negatives(\n negatives: np.ndarray,\n targets: np.ndarray,\n user_rows: np.ndarray,\n known_positive_packed: np.ndarray,\n liked_items: np.ndarray,\n) -> np.ndarray:\n \"\"\"Return row-level invalid negatives using the complete training prefix.\"\"\"\n byte_columns = negatives >> 3\n bit_columns = negatives & 7\n known = known_positive_packed[user_rows[:, None], byte_columns]\n invalid = ((known >> bit_columns) & 1).astype(bool)\n invalid |= negatives == targets[:, None]\n invalid |= (negatives[:, :, None] == liked_items[:, None, :]).any(axis=2)\n for column in range(1, negatives.shape[1]):\n invalid[:, column] |= (\n negatives[:, column, None] == negatives[:, :column]\n ).any(axis=1)\n return invalid\n\n\ndef draw_full_history_filtered_negatives(\n targets: torch.Tensor,\n user_rows: np.ndarray,\n liked_items: torch.Tensor,\n known_positive_packed: np.ndarray,\n n_items: int,\n config: Config,\n item_correction: torch.Tensor,\n generator: torch.Generator,\n device: torch.device,\n) -> tuple[torch.Tensor, torch.Tensor]:\n \"\"\"Draw mixed negatives and reject all known public-training positives.\"\"\"\n targets_cpu = targets.detach().cpu().numpy().astype(np.int64, copy=False)\n liked_cpu = liked_items.detach().cpu().numpy().astype(np.int64, copy=False)\n batch_size = len(targets_cpu)\n uniform = torch.randint(\n n_items,\n (batch_size, config.uniform_negatives),\n generator=generator,\n device=\"cpu\",\n )\n inbatch = torch.empty((batch_size, 0), dtype=torch.long)\n corrected_count = 0\n if config.inbatch_negatives:\n unique_targets = torch.unique(targets.detach().cpu())\n corrected_count = min(\n config.inbatch_negatives, max(len(unique_targets) - 1, 0)\n )\n if corrected_count:\n priorities = torch.rand(\n (batch_size, len(unique_targets)), generator=generator\n )\n priorities.masked_fill_(\n unique_targets[None, :].eq(\n torch.from_numpy(targets_cpu)[:, None]\n ),\n -1.0,\n )\n columns = torch.topk(\n priorities, k=corrected_count, dim=1\n ).indices\n inbatch = unique_targets[columns]\n if corrected_count < config.inbatch_negatives:\n filler = torch.randint(\n n_items,\n (batch_size, config.inbatch_negatives - corrected_count),\n generator=generator,\n device=\"cpu\",\n )\n inbatch = torch.cat([inbatch, filler], dim=1)\n\n negatives = torch.cat([uniform, inbatch], dim=1).numpy()\n correction = np.ones_like(negatives, dtype=np.float32)\n if corrected_count:\n correction[\n :, config.uniform_negatives : config.uniform_negatives + corrected_count\n ] = item_correction[inbatch[:, :corrected_count]].numpy()\n for _ in range(32):\n invalid = invalid_full_history_negatives(\n negatives,\n targets_cpu,\n user_rows,\n known_positive_packed,\n liked_cpu,\n )\n if not np.any(invalid):\n return (\n torch.from_numpy(negatives).to(device),\n torch.from_numpy(correction).to(device),\n )\n replacements = torch.randint(\n n_items,\n (int(invalid.sum()),),\n generator=generator,\n device=\"cpu\",\n ).numpy()\n negatives[invalid] = replacements\n # Rejected in-batch items become ordinary uniform draws.\n correction[invalid] = 1.0\n invalid = invalid_full_history_negatives(\n negatives,\n targets_cpu,\n user_rows,\n known_positive_packed,\n liked_cpu,\n )\n if np.any(invalid):\n raise RuntimeError(\"full-history negative resampling failed\")\n return (\n torch.from_numpy(negatives).to(device),\n torch.from_numpy(correction).to(device),\n )\n\n\ndef draw_training_negatives(\n positive_items: torch.Tensor,\n known_history: torch.Tensor,\n liked_items: torch.Tensor,\n user_rows: np.ndarray,\n known_positive_packed: np.ndarray | None,\n prepared: SequenceData,\n config: Config,\n item_correction: torch.Tensor,\n generator: torch.Generator,\n device: torch.device,\n) -> tuple[torch.Tensor, torch.Tensor]:\n if config.negative_filter_scope == \"full-training-history\":\n if known_positive_packed is None:\n raise RuntimeError(\"full-history negative filter was not built\")\n return draw_full_history_filtered_negatives(\n positive_items,\n user_rows,\n liked_items,\n known_positive_packed,\n len(prepared.item_ids),\n config,\n item_correction,\n generator,\n device,\n )\n return base.draw_filtered_mixed_negatives(\n positive_items,\n known_history,\n liked_items,\n len(prepared.item_ids),\n config.uniform_negatives,\n config.inbatch_negatives,\n item_correction,\n generator,\n device,\n )\n input_size=config.hidden_dim,\n hidden_size=dim,\n num_layers=1,\n batch_first=True,\n )\n self.user_tower = nn.Sequential(\n nn.Linear(dim * 3 + 6, config.user_hidden_1),\n nn.GELU(),\n nn.Dropout(config.dropout),\n nn.Linear(config.user_hidden_1, config.user_hidden_2),\n nn.GELU(),\n nn.Dropout(config.dropout),\n nn.Linear(config.user_hidden_2, dim),\n )\n self.candidate_embedding = nn.Embedding(n_items, dim)\n self.register_buffer(\n \"item_features\", torch.from_numpy(item_features), persistent=True\n )\n self.item_tower = nn.Sequential(\n nn.Linear(dim + n_item_features, config.item_hidden_1),\n nn.GELU(),\n nn.Dropout(config.dropout),\n nn.Linear(config.item_hidden_1, config.item_hidden_2),\n nn.GELU(),\n nn.Dropout(config.dropout),\n nn.Linear(config.item_hidden_2, dim),\n )\n nn.init.normal_(self.context_embedding.weight, std=0.02)\n nn.init.normal_(self.candidate_embedding.weight, std=0.02)\n with torch.no_grad():\n self.context_embedding.weight[n_items].zero_()\n\n def _mean_context_channel(self, items: torch.Tensor) -> torch.Tensor:\n mask = items.ne(self.n_items)\n vectors = self.context_embedding(items)\n return (vectors * mask.unsqueeze(-1)).sum(dim=1) / mask.sum(\n dim=1, keepdim=True\n ).clamp_min(1)\n\n def encode_items(self, items: torch.Tensor) -> torch.Tensor:\n inputs = torch.cat(\n [self.candidate_embedding(items), self.item_features[items]], dim=1\n )\n return self.item_tower(inputs)\n\n\ndef invalid_negative_mask(\n negatives: torch.Tensor,\n targets: torch.Tensor,\n context_items: torch.Tensor,\n liked_items: torch.Tensor,\n) -> torch.Tensor:\n \"\"\"Identify targets, known positives, active likes, and row duplicates.\"\"\"\n invalid = negatives.eq(targets[:, None])\n invalid |= negatives[:, :, None].eq(context_items[:, None, :]).any(dim=2)\n invalid |= negatives[:, :, None].eq(liked_items[:, None, :]).any(dim=2)\n for column in range(1, negatives.shape[1]):\n invalid[:, column] |= negatives[:, column, None].eq(\n negatives[:, :column]\n ).any(dim=1)\n return invalid\n\n\ndef draw_filtered_mixed_negatives(\n targets: torch.Tensor,\n context_items: torch.Tensor,\n liked_items: torch.Tensor,\n n_items: int,\n uniform_count: int,\n inbatch_count: int,\n item_correction: torch.Tensor,\n generator: torch.Generator,\n device: torch.device,\n) -> tuple[torch.Tensor, torch.Tensor]:\n \"\"\"Draw uniform and unique-target in-batch negatives.\"\"\"\n batch_size = len(targets)\n uniform = torch.randint(\n n_items,\n (batch_size, uniform_count),\n generator=generator,\n device=\"cpu\",\n )\n inbatch = torch.empty((batch_size, 0), dtype=torch.long, device=\"cpu\")\n corrected_count = 0\n if inbatch_count:\n unique_targets = torch.unique(targets.detach().cpu())\n corrected_count = min(inbatch_count, max(len(unique_targets) - 1, 0))\n if corrected_count:\n priorities = torch.rand(\n (batch_size, len(unique_targets)), generator=generator\n )\n priorities.masked_fill_(\n unique_targets[None, :].eq(targets.detach().cpu()[:, None]),\n -1.0,\n )\n columns = torch.topk(priorities, k=corrected_count, dim=1).indices\n inbatch = unique_targets[columns]\n if corrected_count < inbatch_count:\n filler = torch.randint(\n n_items,\n (batch_size, inbatch_count - corrected_count),\n generator=generator,\n device=\"cpu\",\n )\n inbatch = torch.cat([inbatch, filler], dim=1)\n\n negatives_cpu = torch.cat([uniform, inbatch], dim=1)\n correction_cpu = torch.ones_like(negatives_cpu, dtype=torch.float32)\n if corrected_count:\n correction_cpu[\n :, uniform_count : uniform_count + corrected_count\n ] = item_correction[inbatch[:, :corrected_count]]\n negatives = negatives_cpu.to(device)\n correction = correction_cpu.to(device)\n for _ in range(32):\n invalid = invalid_negative_mask(\n negatives, targets, context_items, liked_items\n )\n if not bool(invalid.any()):\n return negatives, correction\n replacements = torch.randint(\n n_items,\n (int(invalid.sum().detach().cpu()),),\n generator=generator,\n device=\"cpu\",\n ).to(device)\n negatives[invalid] = replacements\n correction[invalid] = 1.0\n invalid = invalid_negative_mask(\n negatives, targets, context_items, liked_items\n )\n if bool(invalid.any()):\n raise RuntimeError(\"mixed negative resampling failed false-negative filter\")\n return negatives, correction\n",
"exit_code": 0,
"duration": {
"secs": 0,
"nanos": 3850
},
"formatted_output": "Warning: truncated output (original token count: 10978)\nTotal output lines: 847\n\n/task/current_model/two_tower_model.py:40: uniform_negatives: int = 24\n/task/current_model/two_tower_model.py:41: inbatch_negatives: int = 8\n/task/current_model/two_tower_model.py:42: inbatch_correction_power: float = 0.5\n/task/current_model/two_tower_model.py:43: inbatch_correction_min: float = 0.25\n/task/current_model/two_tower_model.py:44: inbatch_correction_max: float = 4.0\n/task/current_model/two_tower_model.py:346:def invalid_negative_mask(\n/task/current_model/two_tower_model.py:347: negatives: torch.Tensor,\n/task/current_model/two_tower_model.py:353: invalid = negatives.eq(targets[:, None])\n/task/current_model/two_tower_model.py:354: invalid |= negatives[:, :, None].eq(context_items[:, None, :]).any(dim=2)\n/task/current_model/two_tower_model.py:355: invalid |= negatives[:, :, None].eq(liked_items[:, None, :]).any(dim=2)\n/task/current_model/two_tower_model.py:356: for column in range(1, negatives.shape[1]):\n/task/current_model/two_tower_model.py:357: invalid[:, column] |= negatives[:, column, None].eq(\n/task/current_model/two_tower_model.py:358: negatives[:, :column]\n/task/current_model/two_tower_model.py:363:def draw_filtered_mixed_negatives(\n/task/current_model/two_tower_model.py:369: inbatch_count: int,\n/task/current_model/two_tower_model.py:374: \"\"\"Draw uniform and unique-target in-batch negatives.\"\"\"\n/task/current_model/two_tower_model.py:382: inbatch = torch.empty((batch_size, 0), dtype=torch.long, device=\"cpu\")\n/task/current_model/two_tower_model.py:384: if inbatch_count:\n/task/current_model/two_tower_model.py:386: corrected_count = min(inbatch_count, max(len(unique_targets) - 1, 0))\n/task/current_model/two_tower_model.py:396: inbatch = unique_targets[columns]\n/task/current_model/two_tower_model.py:397: if corrected_count < inbatch_count:\n/task/current_model/two_tower_model.py:400: (batch_size, inbatch_count - corrected_count),\n/task/current_model/two_tower_model.py:404: inbatch = torch.cat([inbatch, filler], dim=1)\n/task/current_model/two_tower_model.py:406: negatives_cpu = torch.cat([uniform, inbatch], dim=1)\n/task/current_model/two_tower_model.py:407: correction_cpu = torch.ones_like(negatives_cpu, dtype=torch.float32)\n/task/current_model/two_tower_model.py:411: ] = item_correction[inbatch[:, :corrected_count]]\n/task/current_model/two_tower_model.py:412: negatives = negatives_cpu.to(device)\n/task/current_model/two_tower_model.py:415: invalid = invalid_negative_mask(\n/task/current_model/two_tower_model.py:416: negatives, targets, context_items, liked_items\n/task/current_model/two_tower_model.py:419: return negatives, correction\n/task/current_model/two_tower_model.py:426: negatives[invalid] = replacements\n/task/current_model/two_tower_model.py:428: invalid = invalid_negative_mask(\n/task/current_model/two_tower_model.py:429: negatives, targets, context_items, liked_items\n/task/current_model/two_tower_model.py:432: raise RuntimeError(\"mixed negative resampling failed false-negative filter\")\n/task/current_model/two_tower_model.py:433: return negatives, correction\n/task/current_model/train_current_model.py:6:loss. The fast default retains each user's most recent 128 training targets and\n/task/current_model/train_current_model.py:9:computes all of its timestep losses and carries the detached GRU hidden state\n/task/current_model/train_current_model.py:13:normalization, raw dot-product retrieval, sample-weighted binary cross entropy,\n/task/current_model/train_current_model.py:14:and 24 uniform plus eight propensity-corrected unique in-batch negatives. Every\n/task/current_model/train_current_model.py:20:import argparse\n/task/current_model/train_current_model.py:55: loss_microbatch_size: int = 65_536\n/task/current_model/train_current_model.py:56: uniform_negatives: int = 24\n/task/current_model/train_current_model.py:57: inbatch_negatives: int = 8\n/task/current_model/train_current_model.py:58: inbatch_correction_power: float = 0.5\n/task/current_model/train_current_model.py:59: inbatch_correction_min: float = 0.25\n/task/current_model/train_current_model.py:60: inbatch_correction_max: float = 4.0\n/task/current_model/train_current_model.py:61: negative_filter_scope: str = \"full-training-history\"\n/task/current_model/train_current_model.py:152: return base.TowerConfig(\n/task/current_model/train_current_model.py:165: uniform_negatives=config.uniform_negatives,\n/task/current_model/train_current_model.py:166: inbatch_negatives=config.inbatch_negatives,\n/task/current_model/train_current_model.py:167: inbatch_correction_power=config.inbatch_correction_power,\n/task/current_model/train_current_model.py:168: inbatch_correction_min=config.inbatch_correction_min,\n/task/current_model/train_current_model.py:169: inbatch_correction_max=config.inbatch_correction_max,\n/task/current_model/train_current_model.py:342: public: Path, stage: str, config: Config\n/task/current_model/train_current_model.py:354: if stage == \"validation\":\n/task/current_model/train_current_model.py:463: f\"prepared stage={stage} sequence_users={len(prepared.user_ids):,} \"\n/task/current_model/train_current_model.py:471:def cache_metadata(stage: str, config: Config) -> dict[str, object]:\n/task/current_model/train_current_model.py:474: \"stage\": stage,\n/task/current_model/train_current_model.py:477: \"event_time_feature\": \"exp-negative-gap-over-60-days\",\n/task/current_model/train_current_model.py:482: path: Path, prepared: SequenceData, stage: str, config: Config\n/task/current_model/train_current_model.py:487: metadata=np.asarray(json.dumps(cache_metadata(stage, config), sort_keys=True)),\n/task/current_model/train_current_model.py:493:def load_cache(path: Path, stage: str, config: Config) -> SequenceData:\n/task/current_model/train_current_model.py:496: expected = cache_metadata(stage, config)\n/task/current_model/train_current_model.py:513: stage: str,\n/task/current_model/train_current_model.py:519: if stage == \"validation\"\n/task/current_model/train_current_model.py:524: prepared = load_cache(cache, stage, config)\n/task/current_model/train_current_model.py:528: prepared, built_eligibility = prepare_data(public, stage, config)\n/task/current_model/train_current_model.py:530: save_cache(cache, prepared, stage, config)\n/task/current_model/train_current_model.py:562: config.inbatch_correction_power,\n/task/current_model/train_current_model.py:567: config.inbatch_correction_min,\n/task/current_model/train_current_model.py:568: config.inbatch_correction_max,\n/task/current_model/train_current_model.py:572: f\"inbatch correction: min={correction.min():.3f} \"\n/task/current_model/train_current_model.py:603: f\"full-history false-negative filter: users={len(prepared.user_ids):,} \"\n/task/current_model/train_current_model.py:609:def invalid_full_history_negatives(\n/task/current_model/train_current_model.py:610: negatives: np.ndarray,\n/task/current_model/train_current_model.py:616: \"\"\"Return row-level invalid negatives using the complete training prefix.\"\"\"\n/task/current_model/train_current_model.py:617: byte_columns = negatives >> 3\n/task/current_model/train_current_model.py:618: bit_columns = negatives & 7\n/task/current_model/train_current_model.py:621: invalid |= negatives == targets[:, None]\n/task/current_model/train_current_model.py:622: invalid |= (negatives[:, :, None] == liked_items[:, None, :]).any(axis=2)\n/task/current_model/train_current_model.py:623: for column in range(1, negatives.shape[1]):\n/task/current_model/train_current_model.py:625: negatives[:, column, None] == negatives[:, :column]\n/task/current_model/train_current_model.py:630:def draw_full_history_filtered_negatives(\n/task/current_model/train_current_model.py:641: \"\"\"Draw mixed negatives and reject all known public-training positives.\"\"\"\n/task/current_model/train_current_model.py:647: (batch_size, config.uniform_negatives),\n/task/current_model/train_current_model.py:651: inbatch = torch.empty((batch_size, 0), dtype=torch.long)\n/task/current_model/train_current_model.py:653: if config.inbatch_negatives:\n/task/current_model/train_current_model.py:656: config.inbatch_negatives, max(len(unique_targets) - 1, 0)\n/task/current_model/train_current_model.py:671: inbatch = unique_targets[columns]\n/task/current_model/train_current_model.py:672: if corrected_count < config.inbatch_negatives:\n/task/current_model/train_current_model.py:675: (batch_size, config.inbatch_negatives - corrected_count),\n/task/current_model/train_current_model.py:679: inbatch = torch.cat([inbatch, filler], dim=1)\n/task/current_model/train_current_model.py:681: negatives = torch.cat([uniform, inbatch], dim=1).numpy()\n/task/current_model/train_current_model.py:682: correction = np.ones_like(negatives, dtype=np.float32)\n/task/current_model/train_current_model.py:685: :, config.uniform_negatives : config.uniform_negatives + corrected_count\n/task/current_model/train_current_model.py:686: ] = item_correction[inbatch[:, :corrected_count]].numpy()\n/task/current_model/train_current_model.py:688: invalid = invalid_full_history_negatives(\n/task/current_model/train_current_model.py:689: negatives,\n/task/current_model/train_current_model.py:697: torch.from_numpy(negatives).to(device),\n/task/current_model/train_current_model.py:706: negatives[invalid] = replacements\n/task/current_model/train_current_model.py:709: invalid = invalid_full_history_negatives(\n/task/current_model/train_current_model.py:710: negatives,\n/task/current_model/train_current_model.py:717: raise RuntimeError(\"full-history negative resampling failed\")\n/task/current_model/train_current_model.py:719: torch.from_numpy(negatives).to(device),\n/task/current_model/train_current_model.py:724:def draw_training_negatives(\n/task/current_model/train_current_model.py:736: if config.negative_filter_scope == \"full-training-history\":\n/task/current_model/train_current_model.py:738: raise RuntimeError(\"full-history negative filter was not built\")\n/task/current_model/train_current_model.py:739: return draw_full_history_filtered_negatives(\n/task/current_model/train_current_model.py:750: return base.draw_filtered_mixed_negatives(\n/task/current_model/train_current_model.py:755: config.uniform_negatives,\n/task/current_model/train_current_model.py:756: config.inbatch_negatives,\n/task/current_model/train_current_model.py:816: negatives: torch.Tensor,\n/task/current_model/train_current_model.py:818: all_items = torch.cat([positives, negatives.reshape(-1)])\n/task/current_model/train_current_model.py:822: negative_vectors = encoded[inverse[len(positives) :]].reshape(\n/task/current_model/train_current_model.py:823: len(positives), negatives.shape[1], -1\n/task/current_model/train_current_model.py:825: return positive_vectors, negative_vectors\n/task/current_model/train_current_model.py:845:def corrected_negative_loss(\n/task/current_model/train_current_model.py:851: total = config.uniform_negatives + config.inbatch_negatives\n/task/current_model/train_current_model.py:853: if config.uniform_negatives:\n/task/current_model/train_current_model.py:854: result += config.uniform_negatives * raw[\n/task/current_model/train_current_model.py:855: :, : config.uniform_negatives\n/task/current_model/train_current_model.py:857: if config.inbatch_negatives:\n/task/current_model/train_current_model.py:858: selected = slice(config.uniform_negatives, total)\n/task/current_model/train_current_model.py:860: result += config.inbatch_negatives * (\n/task/current_model/train_current_model.py:880: raise RuntimeError(\"chronological target partition produced negative counts\")\n/task/current_model/train_current_model.py:915: public: Path, stage: str, item_ids: np.ndarray\n/task/current_model/train_current_model.py:918: prefixes = (\"train\",) if stage == \"validation\" else (\"train\", \"val\")\n/task/current_model/train_current_model.py:957: stage: str,\n/task/current_model/train_current_model.py:978: public, stage, prepared.item_ids\n/task/current_model/train_current_model.py:981: base.TRAIN_END if stage == \"validation\" else base.VALIDATION_END\n/task/current_model/train_current_model.py:1119: negatives: torch.Tensor,\n/task/current_model/train_current_model.py:1120: negative_weights: torch.Tensor,\n/task/current_model/train_current_model.py:1121: sample_weights: np.ndarray,\n/task/current_model/train_current_model.py:1127: total_weight = float(sample_weights.sum())\n/task/current_model/train_current_model.py:1129: raise RuntimeError(\"target block has no positive sample weight\")\n/task/current_model/train_current_model.py:1132: \"loss_sum\": 0.0,\n/task/current_model/train_current_model.py:1134: \"negative_sum\": 0.0,\n/task/current_model/train_current_model.py:1138: for begin in range(0, target_count, config.loss_microbatch_size):\n/task/current_model/train_current_model.py:1139: end = min(begin + config.loss_microbatch_size, target_count)\n/task/current_model/train_current_model.py:1162: positive_vectors, negative_vectors = candidate_vectors(\n/task/current_model/train_current_model.py:1165: negatives[begin:end],\n/task/current_model/train_current_model.py:1168: negative_logits = torch.einsum(\n/task/current_model/train_current_model.py:1169: \"bd,bkd->bk\", query_vectors, negative_vectors\n/task/current_model/train_current_model.py:1171: positive_loss = F.softplus(-positive_logits)\n/task/current_model/train_current_model.py:1172: negative_loss = corrected_negative_loss(\n/task/current_model/train_current_model.py:1173: negative_logits, negative_weights[begin:end], config\n/task/current_model/train_current_model.py:1176: sample_weights[begin:end].astype(np.float32, copy=False)\n/task/current_model/train_current_model.py:1178: loss_sum = (\n/task/current_model/train_current_model.py:1179: selected_weights * (positive_loss + negative_loss) / 2.0\n/task/current_model/train_current_model.py:1181: if not bool(torch.isfinite(loss_sum)):\n/task/current_model/train_current_model.py:1182: raise RuntimeError(\"dense next-item training produced non-finite loss\")\n/task/current_model/train_current_model.py:1184: (loss_sum / total_weight).backward()\n/task/current_model/train_current_model.py:1188: statistics[\"loss_sum\"] += float(loss_sum.detach().cpu())\n/task/current_model/train_current_model.py:1190: statistics[\"negative_sum\"] += float(negative_logits.detach().sum().cpu())\n/task/current_model/train_current_model.py:1192: (positive_logits.detach()[:, None] > negative_logits.detach()).sum().cpu()\n/task/current_model/train_current_model.py:1227: negative_count = config.uniform_negatives + config.inbatch_negatives\n/task/current_model/train_current_model.py:1228: total_loss = 0.0\n/task/current_model/train_current_model.py:1232: total_negative = 0.0\n/task/current_model/train_current_model.py:1293: negatives, negative_weights = draw_training_negatives(\n/task/current_model/train_current_model.py:1314: negatives,\n/task/current_model/train_current_model.py:1315: negative_weights,\n/task/current_model/train_current_model.py:1327: total_loss += statistics[\"loss_sum\"]\n/task/current_model/train_current_model.py:1331: total_negative += statistics[\"negative_sum\"]\n/task/current_model/train_current_model.py:1337: batch_bce = statistics[\"loss_sum\"] / max(\n/task/current_model/train_current_model.py:1340: running_bce = total_loss / max(total_weight, 1.0e-12)\n/task/current_model/train_current_model.py:1351: \"loss\": total_loss / max(total_weight, 1.0e-12),\n/task/current_model/train_current_model.py:1353: \"negative_logit\": total_negative / max(total_targets * negative_count, 1),\n/task/current_model/train_current_model.py:1355: / max(total_targets * negative_count, 1),\n/task/current_model/train_current_model.py:1388: total_loss = 0.0\n/task/current_model/train_current_model.py:1449: negatives, negative_weights = draw_training_negatives(\n/task/current_model/train_current_model.py:1469: negatives,\n/task/current_model/train_current_model.py:1470: negative_weights,\n/task/current_model/train_current_model.py:1475: total_loss += statistics[\"loss_sum\"]\n/task/current_model/train_current_model.py:1481: \"validation_loss\": total_loss / max(total_weight, 1.0e-12),\n/task/current_model/train_current_model.py:1523: stage: str,\n/task/current_model/train_current_model.py:1560: stage,\n/task/current_model/train_current_model.py:1567: if config.negative_filter_scope == \"full-training-history\"\n/task/current_model/train_current_model.py:1571: best_loss = float(\"inf\")\n/task/current_model/train_current_model.py:1601: \"negative_filter_scope\", \"last-100\"\n/task/current_model/train_current_model.py:1629: stored_filter_scope != config.negative_filter_scope\n/task/current_model/train_current_model.py:1657: prior.pop(\"validation_loss\", None)\n/task/current_model/train_current_model.py:1660: f\"{stored_filter_scope}->{config.negative_filter_scope}, \"\n/task/current_model/train_current_model.py:1666: if \"validation_loss\" not in prior:\n/task/current_model/train_current_model.py:1682: best_loss = float(prior[\"validation_loss\"])\n/task/current_model/train_current_model.py:1687: f\"loss={best_loss:.6f}\"\n/task/current_model/train_current_model.py:1731: diagnostics[\"validation_loss\"]\n/task/current_model/train_current_model.py:1732: < best_loss - config.early_stopping_min_delta\n/task/current_model/train_current_model.py:1735: best_loss = diagnostics[\"validation_loss\"]\n/task/current_model/train_current_model.py:1754: f\"epoch={epoch}/{config.epochs} train_loss={diagnostics['loss']:.5f} \"\n/task/current_model/train_current_model.py:1755: f\"validation_loss={diagnostics['validation_loss']:.5f} \"\n/task/current_model/train_current_model.py:1766: f\"best_epoch={best_epoch} best_validation_loss={best_loss:.6f}\"\n/task/current_model/train_current_model.py:1776: f\"loss={best_loss:.6f}\"\n/task/current_model/train_current_model.py:1923: parser = argparse.ArgumentParser(description=__doc__)\n/task/current_model/train_current_model.py:1925: \"--stage\", choices=(\"validation\", \"submission\"), default=\"validation\"\n/task/current_model/train_current_model.py:1939: \"--loss-microbatch-size\",\n/task/current_model/train_current_model.py:1941: default=Config.loss_microbatch_size,\n/task/current_model/train_current_model.py:1944: \"--negative-filter-scope\",\n/task/current_model/train_current_model.py:1946: default=Config.negative_filter_scope,\n/task/current_model/train_current_model.py:2012: or args.loss_microbatch_size < 1\n/task/current_model/train_current_model.py:2028: config = Config(\n/task/current_model/train_current_model.py:2033: loss_microbatch_size=args.loss_microbatch_size,\n/task/current_model/train_current_model.py:2034: negative_filter_scope=args.negative_filter_scope,\n/task/current_model/train_current_model.py:2057: args.stage,\n/task/current_model/train_current_model.py:2071: \"stage\": args.stage,\n/task/current_model/train_current_model.py:2095: args.stage,\n/task/current_model/train_current_model.py:2115…978 tokens truncated…(\"uint16 item storage is too small for this catalog\")\n\n if stage == \"validation\":\n prefixes = (\"train\",)\n history_end = base.TRAIN_END\n candidate_name = \"validation_eligible_candidates.npz\"\n state_name = \"validation_preference_state.parquet\"\n else:\n prefixes = (\"train\", \"val\")\n history_end = base.VALIDATION_END\n candidate_name = \"test_eligible_candidates.npz\"\n state_name = \"test_preference_state.parquet\"\n eligibility = CandidateSet.read(data / candidate_name)\n if not np.array_equal(eligibility.item_ids, item_ids):\n raise RuntimeError(\"catalog and candidate artifact item order differ\")\n\n explicit_by_user = base.load_explicit_events(data, prefixes, item_index)\n inference_likes, inference_dislikes, like_counts, dislike_counts = (\n base.load_inference_state(\n data / state_name,\n item_index,\n eligibility.uids,\n n_items,\n )\n )\n inference_rows = {int(uid): row for row, uid in enumerate(eligibility.uids)}\n inference_user_features = np.zeros((len(eligibility.uids), 6), dtype=np.float32)\n inference_sequence_rows = np.full(len(eligibility.uids), -1, dtype=np.int32)\n\n # Snapshot zero is the shared empty explicit state.\n liked_states = [np.full(base.LIKE_ITEMS, n_items, dtype=np.uint16)]\n disliked_states = [\n np.full(base.DISLIKE_ITEMS, n_items, dtype=np.uint16)\n ]\n state_counts: list[tuple[int, int]] = [(0, 0)]\n user_ids: list[int] = []\n offsets = [0]\n timestamp_parts: list[np.ndarray] = []\n item_parts: list[np.ndarray] = []\n event_parts: list[np.ndarray] = []\n activity_parts: list[np.ndarray] = []\n state_parts: list[np.ndarray] = []\n\n listen_paths = [data / f\"{prefix}_listens.parquet\" for prefix in prefixes]\n started = time.monotonic()\n for seen_users, (uid, listens) in enumerate(\n base.merge_listen_streams(listen_paths, item_index), start=1\n ):\n parts = build_sequence_parts(\n listens,\n explicit_by_user.get(uid, base.empty_explicit_events()),\n n_items,\n liked_states,\n disliked_states,\n state_counts,\n )\n if parts is None:\n continue\n sequence_row = len(user_ids)\n user_ids.append(uid)\n timestamp_parts.append(parts[0].astype(np.uint32, copy=False))\n item_parts.append(parts[1])\n event_parts.append(parts[2])\n activity_parts.append(parts[3])\n state_parts.append(parts[4])\n offsets.append(offsets[-1] + len(parts[1]))\n inference_row = inference_rows.get(uid)\n if inference_row is not None:\n inference_sequence_rows[inference_row] = sequence_row\n inference_user_features[inference_row] = final_user_features(\n parts[0],\n parts[1],\n parts[2],\n history_end,\n int(like_counts[inference_row]),\n int(dislike_counts[inference_row]),\n )\n if seen_users % 1_000 == 0:\n log(\n f\"prepared users={seen_users:,} tokens={offsets[-1]:,} \"\n f\"states={len(liked_states):,} \"\n f\"elapsed={time.monotonic() - started:.1f}s\"\n )\n\n missing = np.flatnonzero(inference_sequence_rows < 0)\n if len(missing):\n inference_user_features[missing, 4] = np.log1p(like_counts[missing]) / 6.0\n inference_user_features[missing, 5] = (\n np.log1p(dislike_counts[missing]) / 4.0\n )\n log(f\"target users without qualifying catalog sequence={len(missing):,}\")\n\n prepared = SequenceData(\n item_ids=item_ids,\n item_features=base.build_item_features(catalog),\n user_ids=np.asarray(user_ids, dtype=np.int64),\n offsets=np.asarray(offsets, dtype=np.int64),\n sequence_timestamps=np.concatenate(timestamp_parts),\n sequence_items=np.concatenate(item_parts),\n event_features=np.concatenate(event_parts),\n activity_features=np.concatenate(activity_parts),\n state_ids=np.concatenate(state_parts),\n liked_states=np.stack(liked_states),\n disliked_states=np.stack(disliked_states),\n state_counts=np.asarray(state_counts, dtype=np.int32),\n inference_user_rows=inference_sequence_rows,\n inference_liked_items=inference_likes,\n inference_disliked_items=inference_dislikes,\n inference_user_features=inference_user_features,\n )\n log(\n f\"prepared stage={stage} sequence_users={len(prepared.user_ids):,} \"\n f\"tokens={len(prepared.sequence_items):,} \"\n f\"next_item_targets={prepared.target_count:,} \"\n f\"states={len(prepared.liked_states):,}\"\n )\n return prepared, eligibility\n\n\ndef cache_metadata(stage: str, config: Config) -> dict[str, object]:\n return {\n \"cache_version\": CACHE_VERSION,\n \"stage\": stage,\n \"positive_listen_threshold\": base.LISTEN_POSITIVE_THRESHOLD,\n \"profile_lookback_days\": base.PROFILE_LOOKBACK_DAYS,\n \"event_time_feature\": \"exp-negative-gap-over-60-days\",\n }\n\n\ndef save_cache(\n path: Path, prepared: SequenceData, stage: str, config: Config\n) -> None:\n path.parent.mkdir(parents=True, exist_ok=True)\n np.savez(\n path,\n metadata=np.asarray(json.dumps(cache_metadata(stage, config), sort_keys=True)),\n **prepared.__dict__,\n )\n log(f\"saved public-derived full-sequence cache: {path}\")\n\n\ndef load_cache(path: Path, stage: str, config: Config) -> SequenceData:\n with np.load(path, allow_pickle=False) as artifact:\n actual = json.loads(str(artifact[\"metadata\"].item()))\n expected = cache_metadata(stage, config)\n if actual != expected:\n raise RuntimeError(f\"sequence cache differs: {actual} != {expected}\")\n values = {\n field: artifact[field].copy()\n for field in SequenceData.__dataclass_fields__\n }\n prepared = SequenceData(**values)\n log(\n f\"loaded full-sequence cache: tokens={len(prepared.sequence_items):,} \"\n f\"targets={prepared.target_count:,}\"\n )\n return prepared\n\n\ndef load_or_prepare(\n public: Path,\n stage: str,\n config: Config,\n cache: Path | None,\n) -> tuple[SequenceData, CandidateSet]:\n candidate_name = (\n \"validation_eligible_candidates.npz\"\n if stage == \"validation\"\n else \"test_eligible_candidates.npz\"\n )\n eligibility = CandidateSet.read(public / \"data\" / candidate_name)\n if cache is not None and cache.is_file():\n prepared = load_cache(cache, stage, config)\n if not np.array_equal(prepared.item_ids, eligibility.item_ids):\n raise RuntimeError(\"cached catalog differs from candidate artifact\")\n return prepared, eligibility\n prepared, built_eligibility = prepare_data(public, stage, config)\n if cache is not None:\n save_cache(cache, prepared, stage, config)\n return prepared, built_eligibility\n\n\ndef build_item_correction(\n prepared: SequenceData,\n config: Config,\n target_starts: np.ndarray,\n target_counts: np.ndarray,\n) -> torch.Tensor:\n \"\"\"Estimate in-batch propensities from the selected training targets.\"\"\"\n counts = np.zeros(len(prepared.item_ids), dtype=np.float64)\n for user_row, count in enumerate(target_counts):\n if not count:\n continue\n offset = int(prepared.offsets[user_row])\n target_start = int(target_starts[user_row])\n counts += np.bincount(\n prepared.sequence_items[\n offset\n + target_start\n + 1 : offset\n + target_start\n + int(count)\n + 1\n ].astype(np.int64),\n minlength=len(prepared.item_ids),\n )\n total = float(counts.sum())\n probabilities = (counts + 1.0) / (total + len(counts))\n raw = np.power(\n (1.0 / len(counts)) / probabilities,\n config.inbatch_correction_power,\n )\n normalization = float(np.sum(counts * raw) / max(total, 1.0))\n correction = np.clip(\n raw / max(normalization, 1.0e-12),\n config.inbatch_correction_min,\n config.inbatch_correction_max,\n ).astype(np.float32)\n mean_on_targets = float(np.sum(counts * correction) / max(total, 1.0))\n log(\n f\"inbatch correction: min={correction.min():.3f} \"\n f\"mean_on_targets={mean_on_targets:.3f} max={correction.max():.3f}\"\n )\n return torch.from_numpy(correction)\n\n\ndef build_known_positive_filter(\n prepared: SequenceData, training_counts: np.ndarray\n) -> np.ndarray:\n \"\"\"Pack each user's complete public-training positive-item set.\"\"\"\n width = (len(prepared.item_ids) + 7) // 8\n packed = np.zeros((len(prepared.user_ids), width), dtype=np.uint8)\n retained = 0\n for user_row, target_count in enumerate(training_counts):\n if not target_count:\n continue\n offset = int(prepared.offsets[user_row])\n # T adjacent targets consume sequence positions 0 through T. The\n # chronological BCE suffix starts at T + 1 and is intentionally absent.\n items = np.unique(\n prepared.sequence_items[\n offset : offset + int(target_count) + 1\n ].astype(np.int64)\n )\n byte_columns = items >> 3\n bit_values = np.left_shift(\n np.uint8(1), (items & 7).astype(np.uint8)\n )\n np.bitwise_or.at(packed[user_row], byte_columns, bit_values)\n retained += len(items)\n log(\n f\"full-history false-negative filter: users={len(prepared.user_ids):,} \"\n f\"unique_user_items={retained:,} bytes={packed.nbytes:,}\"\n )\n return packed\n\n\ndef invalid_full_history_negatives(\n negatives: np.ndarray,\n targets: np.ndarray,\n user_rows: np.ndarray,\n known_positive_packed: np.ndarray,\n liked_items: np.ndarray,\n) -> np.ndarray:\n \"\"\"Return row-level invalid negatives using the complete training prefix.\"\"\"\n byte_columns = negatives >> 3\n bit_columns = negatives & 7\n known = known_positive_packed[user_rows[:, None], byte_columns]\n invalid = ((known >> bit_columns) & 1).astype(bool)\n invalid |= negatives == targets[:, None]\n invalid |= (negatives[:, :, None] == liked_items[:, None, :]).any(axis=2)\n for column in range(1, negatives.shape[1]):\n invalid[:, column] |= (\n negatives[:, column, None] == negatives[:, :column]\n ).any(axis=1)\n return invalid\n\n\ndef draw_full_history_filtered_negatives(\n targets: torch.Tensor,\n user_rows: np.ndarray,\n liked_items: torch.Tensor,\n known_positive_packed: np.ndarray,\n n_items: int,\n config: Config,\n item_correction: torch.Tensor,\n generator: torch.Generator,\n device: torch.device,\n) -> tuple[torch.Tensor, torch.Tensor]:\n \"\"\"Draw mixed negatives and reject all known public-training positives.\"\"\"\n targets_cpu = targets.detach().cpu().numpy().astype(np.int64, copy=False)\n liked_cpu = liked_items.detach().cpu().numpy().astype(np.int64, copy=False)\n batch_size = len(targets_cpu)\n uniform = torch.randint(\n n_items,\n (batch_size, config.uniform_negatives),\n generator=generator,\n device=\"cpu\",\n )\n inbatch = torch.empty((batch_size, 0), dtype=torch.long)\n corrected_count = 0\n if config.inbatch_negatives:\n unique_targets = torch.unique(targets.detach().cpu())\n corrected_count = min(\n config.inbatch_negatives, max(len(unique_targets) - 1, 0)\n )\n if corrected_count:\n priorities = torch.rand(\n (batch_size, len(unique_targets)), generator=generator\n )\n priorities.masked_fill_(\n unique_targets[None, :].eq(\n torch.from_numpy(targets_cpu)[:, None]\n ),\n -1.0,\n )\n columns = torch.topk(\n priorities, k=corrected_count, dim=1\n ).indices\n inbatch = unique_targets[columns]\n if corrected_count < config.inbatch_negatives:\n filler = torch.randint(\n n_items,\n (batch_size, config.inbatch_negatives - corrected_count),\n generator=generator,\n device=\"cpu\",\n )\n inbatch = torch.cat([inbatch, filler], dim=1)\n\n negatives = torch.cat([uniform, inbatch], dim=1).numpy()\n correction = np.ones_like(negatives, dtype=np.float32)\n if corrected_count:\n correction[\n :, config.uniform_negatives : config.uniform_negatives + corrected_count\n ] = item_correction[inbatch[:, :corrected_count]].numpy()\n for _ in range(32):\n invalid = invalid_full_history_negatives(\n negatives,\n targets_cpu,\n user_rows,\n known_positive_packed,\n liked_cpu,\n )\n if not np.any(invalid):\n return (\n torch.from_numpy(negatives).to(device),\n torch.from_numpy(correction).to(device),\n )\n replacements = torch.randint(\n n_items,\n (int(invalid.sum()),),\n generator=generator,\n device=\"cpu\",\n ).numpy()\n negatives[invalid] = replacements\n # Rejected in-batch items become ordinary uniform draws.\n correction[invalid] = 1.0\n invalid = invalid_full_history_negatives(\n negatives,\n targets_cpu,\n user_rows,\n known_positive_packed,\n liked_cpu,\n )\n if np.any(invalid):\n raise RuntimeError(\"full-history negative resampling failed\")\n return (\n torch.from_numpy(negatives).to(device),\n torch.from_numpy(correction).to(device),\n )\n\n\ndef draw_training_negatives(\n positive_items: torch.Tensor,\n known_history: torch.Tensor,\n liked_items: torch.Tensor,\n user_rows: np.ndarray,\n known_positive_packed: np.ndarray | None,\n prepared: SequenceData,\n config: Config,\n item_correction: torch.Tensor,\n generator: torch.Generator,\n device: torch.device,\n) -> tuple[torch.Tensor, torch.Tensor]:\n if config.negative_filter_scope == \"full-training-history\":\n if known_positive_packed is None:\n raise RuntimeError(\"full-history negative filter was not built\")\n return draw_full_history_filtered_negatives(\n positive_items,\n user_rows,\n liked_items,\n known_positive_packed,\n len(prepared.item_ids),\n config,\n item_correction,\n generator,\n device,\n )\n return base.draw_filtered_mixed_negatives(\n positive_items,\n known_history,\n liked_items,\n len(prepared.item_ids),\n config.uniform_negatives,\n config.inbatch_negatives,\n item_correction,\n generator,\n device,\n )\n input_size=config.hidden_dim,\n hidden_size=dim,\n num_layers=1,\n batch_first=True,\n )\n self.user_tower = nn.Sequential(\n nn.Linear(dim * 3 + 6, config.user_hidden_1),\n nn.GELU(),\n nn.Dropout(config.dropout),\n nn.Linear(config.user_hidden_1, config.user_hidden_2),\n nn.GELU(),\n nn.Dropout(config.dropout),\n nn.Linear(config.user_hidden_2, dim),\n )\n self.candidate_embedding = nn.Embedding(n_items, dim)\n self.register_buffer(\n \"item_features\", torch.from_numpy(item_features), persistent=True\n )\n self.item_tower = nn.Sequential(\n nn.Linear(dim + n_item_features, config.item_hidden_1),\n nn.GELU(),\n nn.Dropout(config.dropout),\n nn.Linear(config.item_hidden_1, config.item_hidden_2),\n nn.GELU(),\n nn.Dropout(config.dropout),\n nn.Linear(config.item_hidden_2, dim),\n )\n nn.init.normal_(self.context_embedding.weight, std=0.02)\n nn.init.normal_(self.candidate_embedding.weight, std=0.02)\n with torch.no_grad():\n self.context_embedding.weight[n_items].zero_()\n\n def _mean_context_channel(self, items: torch.Tensor) -> torch.Tensor:\n mask = items.ne(self.n_items)\n vectors = self.context_embedding(items)\n return (vectors * mask.unsqueeze(-1)).sum(dim=1) / mask.sum(\n dim=1, keepdim=True\n ).clamp_min(1)\n\n def encode_items(self, items: torch.Tensor) -> torch.Tensor:\n inputs = torch.cat(\n [self.candidate_embedding(items), self.item_features[items]], dim=1\n )\n return self.item_tower(inputs)\n\n\ndef invalid_negative_mask(\n negatives: torch.Tensor,\n targets: torch.Tensor,\n context_items: torch.Tensor,\n liked_items: torch.Tensor,\n) -> torch.Tensor:\n \"\"\"Identify targets, known positives, active likes, and row duplicates.\"\"\"\n invalid = negatives.eq(targets[:, None])\n invalid |= negatives[:, :, None].eq(context_items[:, None, :]).any(dim=2)\n invalid |= negatives[:, :, None].eq(liked_items[:, None, :]).any(dim=2)\n for column in range(1, negatives.shape[1]):\n invalid[:, column] |= negatives[:, column, None].eq(\n negatives[:, :column]\n ).any(dim=1)\n return invalid\n\n\ndef draw_filtered_mixed_negatives(\n targets: torch.Tensor,\n context_items: torch.Tensor,\n liked_items: torch.Tensor,\n n_items: int,\n uniform_count: int,\n inbatch_count: int,\n item_correction: torch.Tensor,\n generator: torch.Generator,\n device: torch.device,\n) -> tuple[torch.Tensor, torch.Tensor]:\n \"\"\"Draw uniform and unique-target in-batch negatives.\"\"\"\n batch_size = len(targets)\n uniform = torch.randint(\n n_items,\n (batch_size, uniform_count),\n generator=generator,\n device=\"cpu\",\n )\n inbatch = torch.empty((batch_size, 0), dtype=torch.long, device=\"cpu\")\n corrected_count = 0\n if inbatch_count:\n unique_targets = torch.unique(targets.detach().cpu())\n corrected_count = min(inbatch_count, max(len(unique_targets) - 1, 0))\n if corrected_count:\n priorities = torch.rand(\n (batch_size, len(unique_targets)), generator=generator\n )\n priorities.masked_fill_(\n unique_targets[None, :].eq(targets.detach().cpu()[:, None]),\n -1.0,\n )\n columns = torch.topk(priorities, k=corrected_count, dim=1).indices\n inbatch = unique_targets[columns]\n if corrected_count < inbatch_count:\n filler = torch.randint(\n n_items,\n (batch_size, inbatch_count - corrected_count),\n generator=generator,\n device=\"cpu\",\n )\n inbatch = torch.cat([inbatch, filler], dim=1)\n\n negatives_cpu = torch.cat([uniform, inbatch], dim=1)\n correction_cpu = torch.ones_like(negatives_cpu, dtype=torch.float32)\n if corrected_count:\n correction_cpu[\n :, uniform_count : uniform_count + corrected_count\n ] = item_correction[inbatch[:, :corrected_count]]\n negatives = negatives_cpu.to(device)\n correction = correction_cpu.to(device)\n for _ in range(32):\n invalid = invalid_negative_mask(\n negatives, targets, context_items, liked_items\n )\n if not bool(invalid.any()):\n return negatives, correction\n replacements = torch.randint(\n n_items,\n (int(invalid.sum().detach().cpu()),),\n generator=generator,\n device=\"cpu\",\n ).to(device)\n negatives[invalid] = replacements\n correction[invalid] = 1.0\n invalid = invalid_negative_mask(\n negatives, targets, context_items, liked_items\n )\n if bool(invalid.any()):\n raise RuntimeError(\"mixed negative resampling failed false-negative filter\")\n return negatives, correction\n"
}
}