{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"name":"python","version":"3.12.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"}},"nbformat_minor":4,"nbformat":4,"cells":[{"id":"8399b63a-2473-46d1-8527-6524cd8f01f1","cell_type":"markdown","source":"# H&M Two-Stage Recommendation Demo\n\nThis single Kaggle notebook demonstrates a compact train/validation version of the two-stage pipeline used in the repo:\n\n1. Load raw H&M CSV files with Polars.\n2. Encode `customer_id` and `article_id` to integer keys inside the notebook.\n3. Build 1st-stage candidates for train cutoffs and a validation cutoff.\n4. Label candidate rows with the next 7 days of purchases.\n5. Evaluate 1st-stage Recall/MAP at `@12, @24, @48, @96, @120`.\n6. Train a small GPU XGBoost ranking 2nd-stage reranker.\n7. Evaluate 2nd-stage train and valid metrics at the same cutoffs.\n\nValidation protocol for new experiments:\n\n- Train example cutoffs: `2020-09-01`, `2020-09-08`\n- Validation cutoff: `2020-09-15`\n- Validation label window: `2020-09-16` through `2020-09-22`\n- All validation candidates/features use only `t_dat <= 2020-09-15`\n\nThe notebook is intentionally smaller than production experiments. It samples customers so it can run inside Kaggle notebooks as a teaching/demo workflow. Increase the sample sizes for a fuller run.","metadata":{}},{"id":"f18f6b33-6bee-42f9-9feb-573074a78cab","cell_type":"code","source":"try:\n    import polars as pl\nexcept ImportError:\n    %pip -q install 'polars[gpu]'\n    import polars as pl\n\ntry:\n    import xgboost as xgb\nexcept ImportError:\n    %pip -q install xgboost\n    import xgboost as xgb\n\ntry:\n    import matplotlib.pyplot as plt\nexcept ImportError:\n    %pip -q install matplotlib\n    import matplotlib.pyplot as plt\n\nimport math\nimport random\nfrom collections import defaultdict\nfrom datetime import date, timedelta\nfrom pathlib import Path\n\nimport numpy as np\n\nSEED = 42\nUSE_POLARS_GPU = True\n\n\ndef collect_pl(lf: pl.LazyFrame) -> pl.DataFrame:\n    \"\"\"Use Polars GPU engine when available, then fall back to CPU.\"\"\"\n    global USE_POLARS_GPU\n    if USE_POLARS_GPU:\n        try:\n            return lf.collect(engine=\"gpu\")\n        except Exception as exc:\n            USE_POLARS_GPU = False\n            print(f\"Polars GPU unavailable; falling back to CPU. Reason: {type(exc).__name__}: {exc}\")\n    return lf.collect()\n\n\npl.__version__","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:33:29.521771Z","iopub.execute_input":"2026-06-29T03:33:29.521962Z","iopub.status.idle":"2026-06-29T03:33:32.814252Z","shell.execute_reply.started":"2026-06-29T03:33:29.521940Z","shell.execute_reply":"2026-06-29T03:33:32.813599Z"}},"outputs":[],"execution_count":null},{"id":"6688cb2b-1966-44f6-bc86-ad124dd1efe0","cell_type":"markdown","source":"## Parameters\n\nFor the demo we use two train cutoffs and one validation cutoff. The new validation protocol uses the final fully observed local week as labels.","metadata":{}},{"id":"63c49877-369f-4d70-a511-3516aa84cfac","cell_type":"code","source":"# Kaggle path first, then local repo paths for either notebook-dir or repo-root execution.\nKAGGLE_DATA_DIR = Path(\"/kaggle/input/competitions/h-and-m-personalized-fashion-recommendations\")\nLOCAL_DATA_DIRS = [Path(\"../data/raw\"), Path(\"data/raw\")]\nDATA_DIR = next((path for path in [KAGGLE_DATA_DIR, *LOCAL_DATA_DIRS] if path.exists()), LOCAL_DATA_DIRS[0])\n\nTRANSACTIONS_FILE = DATA_DIR / \"transactions_train.csv\"\nCUSTOMERS_FILE = DATA_DIR / \"customers.csv\"\nARTICLES_FILE = DATA_DIR / \"articles.csv\"\nSAMPLE_SUBMISSION_FILE = DATA_DIR / \"sample_submission.csv\"\n\nassert TRANSACTIONS_FILE.exists(), f\"Missing {TRANSACTIONS_FILE}\"\nassert CUSTOMERS_FILE.exists(), f\"Missing {CUSTOMERS_FILE}\"\nassert ARTICLES_FILE.exists(), f\"Missing {ARTICLES_FILE}\"\nassert SAMPLE_SUBMISSION_FILE.exists(), f\"Missing {SAMPLE_SUBMISSION_FILE}\"\n\n# Train/valid split. Each cutoff uses the next 7 days as labels.\nTRAIN_CUTOFFS = [date(2020, 9, 1), date(2020, 9, 8)]\nVALIDATION_CUTOFF = date(2020, 9, 15)\nSUBMISSION_CUTOFF = date(2020, 9, 22)\nLABEL_DAYS = 7\nEVAL_CUTOFFS = [12, 24, 48, 96, 120]\n\n# Keep train/valid small for Kaggle demo runtime. Set to None for all labeled customers.\nMAX_TRAIN_CUSTOMERS_PER_CUTOFF = 5_000\nMAX_VALID_CUSTOMERS = 5_000\n\n# Submission defaults to all sample_submission customers. Set to a small number for a smoke test only.\nMAX_SUBMISSION_CUSTOMERS = None\nSUBMISSION_CHUNK_SIZE = 20_000\nSUBMISSION_OUTPUT = Path(\"submission.csv\")\n\nCANDIDATE_LIMIT = 120\nHISTORY_DAYS = 45\nHISTORY_TOP_N = 24\nGLOBAL_POPULARITY_DAYS = [7, 14, 30, 60]\nGLOBAL_TOP_N = 50\nAGE_POPULARITY_DAYS = 30\nAGE_TOP_N = 40\nCATEGORY_HISTORY_DAYS = 120\nCATEGORY_POPULARITY_DAYS = 45\nCATEGORY_TOP_N = 16\nCATEGORY_PREF_TOP_N = 2\nCATEGORY_COLS_RAW = [\"section_name\", \"garment_group_name\", \"index_group_name\"]\n\nSOURCE_WEIGHTS = {\n    \"history\": 4.0,\n    \"category\": 2.0,\n    \"age14\": 1.4,\n    \"global7\": 2.0,\n    \"global14\": 1.6,\n    \"global30\": 1.2,\n    \"global60\": 0.8,\n}\nRANK_WEIGHT = 2.0\nCOUNT_WEIGHT = 0.08\nRECENCY_WEIGHT = 1.5\nMISSING_RANK = 999\n\nMAX_HISTORY_DAYS = max(\n    HISTORY_DAYS,\n    max(GLOBAL_POPULARITY_DAYS),\n    AGE_POPULARITY_DAYS,\n    CATEGORY_HISTORY_DAYS,\n    CATEGORY_POPULARITY_DAYS,\n)\nEARLIEST_NEEDED_DATE = min(TRAIN_CUTOFFS + [VALIDATION_CUTOFF]) - timedelta(days=MAX_HISTORY_DAYS)\nLABEL_END = max(VALIDATION_CUTOFF + timedelta(days=LABEL_DAYS), SUBMISSION_CUTOFF)\n\nTRAIN_CUTOFFS, VALIDATION_CUTOFF, SUBMISSION_CUTOFF, LABEL_END, EARLIEST_NEEDED_DATE\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:35:41.103743Z","iopub.execute_input":"2026-06-29T03:35:41.104491Z","iopub.status.idle":"2026-06-29T03:35:41.120625Z","shell.execute_reply.started":"2026-06-29T03:35:41.104458Z","shell.execute_reply":"2026-06-29T03:35:41.119907Z"}},"outputs":[],"execution_count":null},{"id":"702e0250-cc3a-417e-b0c9-ffd60ee9c8d5","cell_type":"markdown","source":"## Load and encode raw data\n\nProduction experiments use shared encoded parquet files. This standalone Kaggle demo builds a small in-notebook encoding from the raw CSV files so the full flow is visible.","metadata":{}},{"id":"7052986e-db26-4477-91f6-a72b3b46e5a8","cell_type":"code","source":"transactions_raw = collect_pl(\n    pl.scan_csv(\n        TRANSACTIONS_FILE,\n        schema_overrides={\"t_dat\": pl.Utf8, \"customer_id\": pl.Utf8, \"article_id\": pl.Utf8},\n    )\n    .select([\"t_dat\", \"customer_id\", \"article_id\"])\n    .with_columns(pl.col(\"t_dat\").str.strptime(pl.Date, strict=False))\n    .filter((pl.col(\"t_dat\") > pl.lit(EARLIEST_NEEDED_DATE)) & (pl.col(\"t_dat\") <= pl.lit(LABEL_END)))\n)\n\ncustomers_raw = collect_pl(\n    pl.scan_csv(\n        CUSTOMERS_FILE,\n        schema_overrides={\"customer_id\": pl.Utf8, \"age\": pl.Float32},\n    ).select([\"customer_id\", \"age\"])\n)\n\narticles_raw = collect_pl(\n    pl.scan_csv(\n        ARTICLES_FILE,\n        schema_overrides={\"article_id\": pl.Utf8, **{col: pl.Utf8 for col in CATEGORY_COLS_RAW}},\n    )\n    .select([\"article_id\", *CATEGORY_COLS_RAW])\n    .with_columns([pl.col(col).fill_null(\"unknown\") for col in CATEGORY_COLS_RAW])\n)\n\nsample_submission_raw = collect_pl(\n    pl.scan_csv(\n        SAMPLE_SUBMISSION_FILE,\n        schema_overrides={\"customer_id\": pl.Utf8, \"prediction\": pl.Utf8},\n    ).select([\"customer_id\", \"prediction\"])\n)\n\ncustomer_map = collect_pl(\n    pl.concat([transactions_raw.select(\"customer_id\"), customers_raw.select(\"customer_id\"), sample_submission_raw.select(\"customer_id\")])\n    .lazy()\n    .unique()\n    .sort(\"customer_id\")\n    .with_row_index(\"customer_idx\", offset=1)\n    .select([\"customer_idx\", \"customer_id\"])\n)\n\narticle_map = collect_pl(\n    pl.concat([transactions_raw.select(\"article_id\"), articles_raw.select(\"article_id\")])\n    .lazy()\n    .unique()\n    .sort(\"article_id\")\n    .with_row_index(\"article_idx\", offset=1)\n    .select([\"article_idx\", \"article_id\"])\n)\n\ntransactions = transactions_raw.join(customer_map, on=\"customer_id\", how=\"left\").join(article_map, on=\"article_id\", how=\"left\")\ntransactions = transactions.select([\"t_dat\", \"customer_idx\", \"article_idx\"])\n\ncustomers = customers_raw.join(customer_map, on=\"customer_id\", how=\"left\").select([\"customer_idx\", \"age\"])\narticles = articles_raw.join(article_map, on=\"article_id\", how=\"left\").select([\"article_idx\", *CATEGORY_COLS_RAW])\n\nsample_submission_customers = sample_submission_raw.join(customer_map, on=\"customer_id\", how=\"left\").select([\"customer_id\", \"customer_idx\"])\nassert sample_submission_customers[\"customer_idx\"].null_count() == 0\n\ntransactions.shape, customers.shape, articles.shape, sample_submission_customers.shape","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:35:42.152860Z","iopub.execute_input":"2026-06-29T03:35:42.153638Z","iopub.status.idle":"2026-06-29T03:36:33.870400Z","shell.execute_reply.started":"2026-06-29T03:35:42.153605Z","shell.execute_reply":"2026-06-29T03:36:33.869625Z"}},"outputs":[],"execution_count":null},{"id":"e11da742-fec1-40d3-8dcf-af78df962894","cell_type":"code","source":"def add_age_bucket(customers_df: pl.DataFrame) -> pl.DataFrame:\n    return customers_df.with_columns(\n        pl.when(pl.col(\"age\").is_null())\n        .then(pl.lit(\"unknown\"))\n        .when(pl.col(\"age\") < 20)\n        .then(pl.lit(\"<20\"))\n        .when(pl.col(\"age\") < 30)\n        .then(pl.lit(\"20s\"))\n        .when(pl.col(\"age\") < 40)\n        .then(pl.lit(\"30s\"))\n        .when(pl.col(\"age\") < 50)\n        .then(pl.lit(\"40s\"))\n        .when(pl.col(\"age\") < 60)\n        .then(pl.lit(\"50s\"))\n        .otherwise(pl.lit(\"60+\"))\n        .alias(\"age_bucket\")\n    )\n\n\ncustomers = add_age_bucket(customers)\nage_bucket_values = sorted(str(x) for x in customers[\"age_bucket\"].unique().to_list())\nage_bucket_map = {bucket: i + 1 for i, bucket in enumerate(age_bucket_values)}\ncustomer_age_bucket = {\n    int(row[\"customer_idx\"]): str(row[\"age_bucket\"])\n    for row in customers.select([\"customer_idx\", \"age_bucket\"]).iter_rows(named=True)\n}\ncustomer_age_bucket_idx = {customer: age_bucket_map.get(bucket, 0) for customer, bucket in customer_age_bucket.items()}\n\ncategory_value_maps = {}\nfor col in CATEGORY_COLS_RAW:\n    values = sorted(str(x) for x in articles[col].unique().to_list())\n    category_value_maps[col] = {value: i + 1 for i, value in enumerate(values)}\n\narticle_feature_map = {}\nfor row in articles.iter_rows(named=True):\n    article_idx = int(row[\"article_idx\"])\n    article_feature_map[article_idx] = {\n        f\"article_{col}_idx\": category_value_maps[col].get(str(row[col]), 0)\n        for col in CATEGORY_COLS_RAW\n    }\n\nARTICLE_FEATURE_COLUMNS = [f\"article_{col}_idx\" for col in CATEGORY_COLS_RAW]\nARTICLE_FEATURE_COLUMNS[:3], len(article_feature_map)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:36:33.871883Z","iopub.execute_input":"2026-06-29T03:36:33.872673Z","iopub.status.idle":"2026-06-29T03:36:35.526466Z","shell.execute_reply.started":"2026-06-29T03:36:33.872638Z","shell.execute_reply":"2026-06-29T03:36:35.525620Z"}},"outputs":[],"execution_count":null},{"id":"637c8c2d-efe4-4e96-b520-ea0d09c32279","cell_type":"markdown","source":"## 1st-stage candidate generation helpers\n\nThe 1st stage is deliberately transparent: multiple candidate sources contribute source flags, source ranks, and a weighted linear `candidate_score`. The 2nd-stage model will receive these same source/rank features.","metadata":{}},{"id":"5bf39c49-fd52-44ef-a84a-d316f0bfdf95","cell_type":"code","source":"def sorted_count_frame(df: pl.DataFrame, group_cols: list[str]) -> pl.DataFrame:\n    return collect_pl(\n        df.lazy()\n        .group_by(group_cols)\n        .agg(pl.len().alias(\"cnt\"))\n        .sort(group_cols[:-1] + [\"cnt\", group_cols[-1]], descending=[False] * (len(group_cols) - 1) + [True, False])\n    )\n\n\ndef actuals_for_window(transactions_df: pl.DataFrame, cutoff: date, label_days: int) -> dict[int, list[int]]:\n    label_end = cutoff + timedelta(days=label_days)\n    grouped = collect_pl(\n        transactions_df.lazy()\n        .filter((pl.col(\"t_dat\") > pl.lit(cutoff)) & (pl.col(\"t_dat\") <= pl.lit(label_end)))\n        .select([\"customer_idx\", \"article_idx\"])\n        .unique()\n        .group_by(\"customer_idx\")\n        .agg(pl.col(\"article_idx\").sort().alias(\"actual_articles\"))\n    )\n    return {\n        int(row[\"customer_idx\"]): [int(article) for article in row[\"actual_articles\"]]\n        for row in grouped.iter_rows(named=True)\n    }\n\n\ndef sample_customers(customers_with_labels: list[int], max_customers: int | None, seed: int) -> list[int]:\n    customers_with_labels = sorted(customers_with_labels)\n    if max_customers is None or len(customers_with_labels) <= max_customers:\n        return customers_with_labels\n    rng = random.Random(seed)\n    return sorted(rng.sample(customers_with_labels, max_customers))\n\n\ndef build_top_list(transactions_df: pl.DataFrame, cutoff: date, days: int, top_n: int) -> list[tuple[int, int]]:\n    start = cutoff - timedelta(days=days)\n    counts = sorted_count_frame(\n        transactions_df.filter((pl.col(\"t_dat\") > start) & (pl.col(\"t_dat\") <= cutoff)).select([\"article_idx\"]),\n        [\"article_idx\"],\n    )\n    return [(int(row[\"article_idx\"]), int(row[\"cnt\"])) for row in counts.head(top_n).iter_rows(named=True)]\n\n\ndef build_customer_history(transactions_df: pl.DataFrame, cutoff: date, days: int, top_n: int) -> dict[int, list[tuple[int, int, int]]]:\n    start = cutoff - timedelta(days=days)\n    hist = collect_pl(\n        transactions_df.lazy()\n        .filter((pl.col(\"t_dat\") > pl.lit(start)) & (pl.col(\"t_dat\") <= pl.lit(cutoff)))\n        .group_by([\"customer_idx\", \"article_idx\"])\n        .agg([\n            pl.len().alias(\"cnt\"),\n            pl.col(\"t_dat\").max().alias(\"last_purchase\"),\n        ])\n        .with_columns((pl.lit(cutoff) - pl.col(\"last_purchase\")).dt.total_days().cast(pl.Int32).alias(\"recency_days\"))\n        .sort([\"customer_idx\", \"recency_days\", \"cnt\", \"article_idx\"], descending=[False, False, True, False])\n    )\n    out = defaultdict(list)\n    for row in hist.iter_rows(named=True):\n        customer = int(row[\"customer_idx\"])\n        if len(out[customer]) < top_n:\n            out[customer].append((int(row[\"article_idx\"]), int(row[\"cnt\"]), int(row[\"recency_days\"])))\n    return dict(out)\n\n\ndef build_age_top(transactions_df: pl.DataFrame, customers_df: pl.DataFrame, cutoff: date, days: int, top_n: int) -> dict[str, list[tuple[int, int]]]:\n    start = cutoff - timedelta(days=days)\n    recent = collect_pl(\n        transactions_df.lazy()\n        .filter((pl.col(\"t_dat\") > pl.lit(start)) & (pl.col(\"t_dat\") <= pl.lit(cutoff)))\n        .select([\"customer_idx\", \"article_idx\"])\n        .join(customers_df.lazy().select([\"customer_idx\", \"age_bucket\"]), on=\"customer_idx\", how=\"left\")\n        .with_columns(pl.col(\"age_bucket\").fill_null(\"unknown\"))\n    )\n    counts = sorted_count_frame(recent, [\"age_bucket\", \"article_idx\"])\n    out = defaultdict(list)\n    for row in counts.iter_rows(named=True):\n        bucket = str(row[\"age_bucket\"])\n        if len(out[bucket]) < top_n:\n            out[bucket].append((int(row[\"article_idx\"]), int(row[\"cnt\"])))\n    return dict(out)\n\n\ndef build_category_top(transactions_df: pl.DataFrame, articles_df: pl.DataFrame, cutoff: date, days: int, category_cols: list[str], top_n: int) -> dict[tuple[str, str], list[tuple[int, int]]]:\n    start = cutoff - timedelta(days=days)\n    recent = collect_pl(\n        transactions_df.lazy()\n        .filter((pl.col(\"t_dat\") > pl.lit(start)) & (pl.col(\"t_dat\") <= pl.lit(cutoff)))\n        .select([\"article_idx\"])\n        .join(articles_df.lazy(), on=\"article_idx\", how=\"left\")\n    )\n    category_top = {}\n    for col in category_cols:\n        counts = sorted_count_frame(recent.select([col, \"article_idx\"]).filter(pl.col(col).is_not_null()), [col, \"article_idx\"])\n        out = defaultdict(list)\n        for row in counts.iter_rows(named=True):\n            value = str(row[col])\n            if len(out[value]) < top_n:\n                out[value].append((int(row[\"article_idx\"]), int(row[\"cnt\"])))\n        for value, items in out.items():\n            category_top[(col, value)] = items\n    return category_top\n\n\ndef build_category_preferences(transactions_df: pl.DataFrame, articles_df: pl.DataFrame, cutoff: date, days: int, category_cols: list[str], top_n: int) -> dict[int, list[tuple[str, str]]]:\n    start = cutoff - timedelta(days=days)\n    recent = collect_pl(\n        transactions_df.lazy()\n        .filter((pl.col(\"t_dat\") > pl.lit(start)) & (pl.col(\"t_dat\") <= pl.lit(cutoff)))\n        .select([\"customer_idx\", \"article_idx\"])\n        .join(articles_df.lazy(), on=\"article_idx\", how=\"left\")\n    )\n    prefs = defaultdict(list)\n    for col in category_cols:\n        counts = collect_pl(\n            recent.lazy()\n            .select([\"customer_idx\", col])\n            .filter(pl.col(col).is_not_null())\n            .group_by([\"customer_idx\", col])\n            .agg(pl.len().alias(\"cnt\"))\n            .sort([\"customer_idx\", \"cnt\", col], descending=[False, True, False])\n        )\n        seen_per_col = defaultdict(int)\n        for row in counts.iter_rows(named=True):\n            customer = int(row[\"customer_idx\"])\n            if seen_per_col[(customer, col)] < top_n:\n                prefs[customer].append((col, str(row[col])))\n                seen_per_col[(customer, col)] += 1\n    return dict(prefs)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:36:35.527490Z","iopub.execute_input":"2026-06-29T03:36:35.527703Z","iopub.status.idle":"2026-06-29T03:36:35.548526Z","shell.execute_reply.started":"2026-06-29T03:36:35.527683Z","shell.execute_reply":"2026-06-29T03:36:35.547590Z"}},"outputs":[],"execution_count":null},{"id":"5e08d5a1-39ff-4a61-aee7-67e26127cfe3","cell_type":"code","source":"SOURCE_COLUMNS = [\n    \"source_history\",\n    \"source_age14\",\n    \"source_global7\",\n    \"source_global14\",\n    \"source_global30\",\n    \"source_global60\",\n    \"source_category\",\n]\nRANK_COLUMNS = [\n    \"history_rank\",\n    \"age_rank\",\n    \"global7_rank\",\n    \"global14_rank\",\n    \"global30_rank\",\n    \"global60_rank\",\n    \"category_rank\",\n]\n\n\ndef new_candidate_features() -> dict:\n    return {\n        \"candidate_score\": 0.0,\n        **{col: 0 for col in SOURCE_COLUMNS},\n        **{col: MISSING_RANK for col in RANK_COLUMNS},\n    }\n\n\ndef score_piece(source: str, rank: int, count: int, recency_days: int | None = None) -> float:\n    score = SOURCE_WEIGHTS[source] + RANK_WEIGHT / max(rank, 1) + COUNT_WEIGHT * math.log1p(max(count, 0))\n    if recency_days is not None:\n        score += RECENCY_WEIGHT / (1.0 + max(recency_days, 0))\n    return score\n\n\ndef add_candidate(candidates: dict[int, dict], article_idx: int, source: str, rank: int, count: int, recency_days: int | None = None) -> None:\n    if article_idx not in candidates:\n        candidates[article_idx] = new_candidate_features()\n    row = candidates[article_idx]\n    row[f\"source_{source}\"] = 1\n    rank_col = \"age_rank\" if source == \"age14\" else f\"{source}_rank\"\n    row[rank_col] = min(row[rank_col], rank)\n    row[\"candidate_score\"] += score_piece(source, rank, count, recency_days)\n\n\ndef build_cutoff_components(cutoff: date) -> dict:\n    return {\n        \"history\": build_customer_history(transactions, cutoff, HISTORY_DAYS, HISTORY_TOP_N),\n        \"global_top\": {days: build_top_list(transactions, cutoff, days, GLOBAL_TOP_N) for days in GLOBAL_POPULARITY_DAYS},\n        \"age_top\": build_age_top(transactions, customers, cutoff, AGE_POPULARITY_DAYS, AGE_TOP_N),\n        \"category_top\": build_category_top(transactions, articles, cutoff, CATEGORY_POPULARITY_DAYS, CATEGORY_COLS_RAW, CATEGORY_TOP_N),\n        \"category_preferences\": build_category_preferences(transactions, articles, cutoff, CATEGORY_HISTORY_DAYS, CATEGORY_COLS_RAW, CATEGORY_PREF_TOP_N),\n    }\n\n\ndef candidate_map_for_customer(customer_idx: int, components: dict) -> dict[int, dict]:\n    candidates = {}\n\n    for rank, (article_idx, cnt, recency_days) in enumerate(components[\"history\"].get(customer_idx, []), start=1):\n        add_candidate(candidates, article_idx, \"history\", rank, cnt, recency_days)\n\n    bucket = customer_age_bucket.get(customer_idx, \"unknown\")\n    for rank, (article_idx, cnt) in enumerate(components[\"age_top\"].get(bucket, []), start=1):\n        add_candidate(candidates, article_idx, \"age14\", rank, cnt)\n\n    for days, top_items in components[\"global_top\"].items():\n        source = f\"global{days}\"\n        for rank, (article_idx, cnt) in enumerate(top_items, start=1):\n            add_candidate(candidates, article_idx, source, rank, cnt)\n\n    category_rank = 1\n    for key in components[\"category_preferences\"].get(customer_idx, []):\n        for article_idx, cnt in components[\"category_top\"].get(key, []):\n            add_candidate(candidates, article_idx, \"category\", category_rank, cnt)\n            category_rank += 1\n\n    return candidates\n\n\ndef ordered_candidates(candidates: dict[int, dict]) -> list[tuple[int, dict]]:\n    return sorted(\n        candidates.items(),\n        key=lambda item: (\n            -float(item[1][\"candidate_score\"]),\n            min(int(item[1][rank_col]) for rank_col in RANK_COLUMNS),\n            int(item[0]),\n        ),\n    )[:CANDIDATE_LIMIT]","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:36:35.550456Z","iopub.execute_input":"2026-06-29T03:36:35.550690Z","iopub.status.idle":"2026-06-29T03:36:35.569841Z","shell.execute_reply.started":"2026-06-29T03:36:35.550670Z","shell.execute_reply":"2026-06-29T03:36:35.568997Z"}},"outputs":[],"execution_count":null},{"id":"107cfef0-942c-431c-bd79-c261634f3acc","cell_type":"markdown","source":"## Metrics helpers\n\n`Recall customer mean` averages per-customer recall. `Recall item` counts all hit labels over all labels. `MAP@k` follows the competition's customer-mean AP style.","metadata":{}},{"id":"a456c9e1-1660-4f0e-8cbb-a39961910466","cell_type":"code","source":"def apk_from_labels(labels: list[int], actual_count: int, k: int) -> float:\n    if actual_count <= 0:\n        return 0.0\n    score = 0.0\n    hits = 0\n    for rank, label in enumerate(labels[:k], start=1):\n        if int(label) == 1:\n            hits += 1\n            score += hits / rank\n    return score / min(actual_count, k)\n\n\ndef metrics_from_ranked_frame(frame: pl.DataFrame, rank_col: str, stage: str, split: str) -> pl.DataFrame:\n    grouped = (\n        frame.sort([\"cutoff\", \"customer_idx\", rank_col])\n        .group_by([\"cutoff\", \"customer_idx\"], maintain_order=True)\n        .agg([\n            pl.col(\"label\").alias(\"labels\"),\n            pl.first(\"actual_count\").alias(\"actual_count\"),\n        ])\n    )\n    rows = []\n    groups = grouped.height\n    total_labels = int(grouped[\"actual_count\"].sum())\n    for k in EVAL_CUTOFFS:\n        customer_recalls = []\n        ap_values = []\n        total_hits = 0\n        for row in grouped.iter_rows(named=True):\n            labels = [int(x) for x in row[\"labels\"]]\n            actual_count = int(row[\"actual_count\"])\n            hits = sum(labels[:k])\n            total_hits += hits\n            customer_recalls.append(hits / actual_count if actual_count else 0.0)\n            ap_values.append(apk_from_labels(labels, actual_count, k))\n        rows.append({\n            \"stage\": stage,\n            \"split\": split,\n            \"k\": k,\n            \"groups\": groups,\n            \"labels\": total_labels,\n            \"recall_customer_mean\": float(np.mean(customer_recalls)) if customer_recalls else 0.0,\n            \"recall_item\": float(total_hits / total_labels) if total_labels else 0.0,\n            \"map\": float(np.mean(ap_values)) if ap_values else 0.0,\n        })\n    return pl.DataFrame(rows)\n\n\ndef print_metrics(metrics: pl.DataFrame, title: str) -> None:\n    print(title)\n    print(metrics.select([\"stage\", \"split\", \"k\", \"recall_customer_mean\", \"recall_item\", \"map\"]))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:36:35.570932Z","iopub.execute_input":"2026-06-29T03:36:35.571380Z","iopub.status.idle":"2026-06-29T03:36:35.589270Z","shell.execute_reply.started":"2026-06-29T03:36:35.571325Z","shell.execute_reply":"2026-06-29T03:36:35.588299Z"}},"outputs":[],"execution_count":null},{"id":"2c0903fa-1a77-48a8-8e4b-26a150024045","cell_type":"markdown","source":"## Build train and validation candidate rows\n\nEach candidate row has:\n\n- group keys: `split`, `cutoff`, `customer_idx`\n- candidate keys: `article_idx`, `candidate_order`\n- source/rank/score features from the 1st stage\n- simple article/customer features\n- `label`: whether the candidate article was purchased in the next 7 days\n- `actual_count`: denominator for recall/MAP when true items are absent from candidates","metadata":{}},{"id":"02bf7f4f-ef54-4f86-b692-e8534a70e908","cell_type":"code","source":"def build_candidates_for_cutoff(split: str, cutoff: date, max_customers: int | None, seed: int) -> pl.DataFrame:\n    print(f\"Building {split} candidates for cutoff={cutoff}...\")\n    actuals_all = actuals_for_window(transactions, cutoff, LABEL_DAYS)\n    selected_customers = sample_customers(list(actuals_all), max_customers, seed)\n    actuals = {customer: actuals_all[customer] for customer in selected_customers}\n    components = build_cutoff_components(cutoff)\n\n    candidate_rows = []\n    for n, customer_idx in enumerate(selected_customers, start=1):\n        candidates = candidate_map_for_customer(customer_idx, components)\n        ordered = ordered_candidates(candidates)\n        actual_set = set(actuals[customer_idx])\n        actual_count = len(actual_set)\n        age_idx = customer_age_bucket_idx.get(customer_idx, 0)\n\n        for order, (article_idx, features) in enumerate(ordered, start=1):\n            row = {\n                \"split\": split,\n                \"cutoff\": cutoff.isoformat(),\n                \"customer_idx\": customer_idx,\n                \"article_idx\": article_idx,\n                \"candidate_order\": order,\n                \"label\": int(article_idx in actual_set),\n                \"actual_count\": actual_count,\n                \"age_bucket_idx\": age_idx,\n                **features,\n                **article_feature_map.get(article_idx, {col: 0 for col in ARTICLE_FEATURE_COLUMNS}),\n            }\n            candidate_rows.append(row)\n\n        if n % 1000 == 0:\n            print(f\"  customers={n:,}/{len(selected_customers):,}; rows={len(candidate_rows):,}\")\n\n    frame = pl.DataFrame(candidate_rows)\n    int_cols = [\"customer_idx\", \"article_idx\", \"candidate_order\", \"label\", \"actual_count\", \"age_bucket_idx\", *SOURCE_COLUMNS, *RANK_COLUMNS, *ARTICLE_FEATURE_COLUMNS]\n    return frame.with_columns([pl.col(col).cast(pl.Int32) for col in int_cols] + [pl.col(\"candidate_score\").cast(pl.Float32)])\n\n\ntrain_frames = []\nfor i, cutoff in enumerate(TRAIN_CUTOFFS):\n    train_frames.append(build_candidates_for_cutoff(\"train\", cutoff, MAX_TRAIN_CUSTOMERS_PER_CUTOFF, SEED + i))\ntrain_candidates = pl.concat(train_frames, how=\"vertical_relaxed\")\nvalid_candidates = build_candidates_for_cutoff(\"valid\", VALIDATION_CUTOFF, MAX_VALID_CUSTOMERS, SEED + 999)\n\ntrain_candidates.shape, valid_candidates.shape","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:36:35.590476Z","iopub.execute_input":"2026-06-29T03:36:35.591326Z","iopub.status.idle":"2026-06-29T03:37:40.868496Z","shell.execute_reply.started":"2026-06-29T03:36:35.591294Z","shell.execute_reply":"2026-06-29T03:37:40.867698Z"}},"outputs":[],"execution_count":null},{"id":"1fa42a88-9888-4258-b7fe-f3462c215cc1","cell_type":"code","source":"stage1_train_metrics = metrics_from_ranked_frame(train_candidates, \"candidate_order\", \"stage1\", \"train\")\nstage1_valid_metrics = metrics_from_ranked_frame(valid_candidates, \"candidate_order\", \"stage1\", \"valid\")\n\nprint_metrics(stage1_train_metrics, \"1st-stage train metrics\")\nprint_metrics(stage1_valid_metrics, \"1st-stage valid metrics\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:37:40.869565Z","iopub.execute_input":"2026-06-29T03:37:40.869822Z","iopub.status.idle":"2026-06-29T03:37:42.226519Z","shell.execute_reply.started":"2026-06-29T03:37:40.869794Z","shell.execute_reply":"2026-06-29T03:37:42.225659Z"}},"outputs":[],"execution_count":null},{"id":"b5598c9f-dd52-4b5b-a0ac-7d5950f17508","cell_type":"markdown","source":"## Train a small GPU XGBoost 2nd-stage reranker\n\nFor speed, this demo trains on positive groups only: customers whose generated candidate set contains at least one positive label. Production experiments use the same idea for efficient ranking training. XGBoost is configured with `device=\"cuda\"` and falls back to CPU if the Kaggle session has no GPU.","metadata":{}},{"id":"301f7377-495a-4de2-86d0-40578e9fb49f","cell_type":"code","source":"FEATURE_COLUMNS = [\n    \"candidate_score\",\n    \"candidate_order\",\n    \"age_bucket_idx\",\n    *SOURCE_COLUMNS,\n    *RANK_COLUMNS,\n    *ARTICLE_FEATURE_COLUMNS,\n]\n\n\ndef filter_positive_groups(frame: pl.DataFrame) -> pl.DataFrame:\n    positive_groups = (\n        frame.group_by([\"split\", \"cutoff\", \"customer_idx\"])\n        .agg(pl.col(\"label\").sum().alias(\"positive_labels_in_candidates\"))\n        .filter(pl.col(\"positive_labels_in_candidates\") > 0)\n        .select([\"split\", \"cutoff\", \"customer_idx\"])\n    )\n    return frame.join(positive_groups, on=[\"split\", \"cutoff\", \"customer_idx\"], how=\"inner\")\n\n\ndef xgb_arrays(frame: pl.DataFrame, positive_groups_only: bool) -> tuple[pl.DataFrame, np.ndarray, np.ndarray, list[int]]:\n    work = filter_positive_groups(frame) if positive_groups_only else frame\n    work = work.sort([\"split\", \"cutoff\", \"customer_idx\", \"candidate_order\"])\n    group_sizes = (\n        work.group_by([\"split\", \"cutoff\", \"customer_idx\"], maintain_order=True)\n        .agg(pl.len().alias(\"group_size\"))[\"group_size\"]\n        .to_list()\n    )\n    x = work.select(FEATURE_COLUMNS).fill_null(0).to_numpy().astype(np.float32)\n    y = work[\"label\"].to_numpy().astype(np.float32)\n    return work, x, y, [int(size) for size in group_sizes]\n\n\ntrain_rank_frame, x_train, y_train, train_group = xgb_arrays(train_candidates, positive_groups_only=True)\nvalid_rank_frame, x_valid, y_valid, valid_group = xgb_arrays(valid_candidates, positive_groups_only=False)\n\nprint(f\"train rows used for ranker: {x_train.shape[0]:,}; groups={len(train_group):,}; positives={int(y_train.sum()):,}\")\nprint(f\"valid rows for monitoring: {x_valid.shape[0]:,}; groups={len(valid_group):,}; positives={int(y_valid.sum()):,}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:37:42.227662Z","iopub.execute_input":"2026-06-29T03:37:42.228277Z","iopub.status.idle":"2026-06-29T03:37:42.444962Z","shell.execute_reply.started":"2026-06-29T03:37:42.228251Z","shell.execute_reply":"2026-06-29T03:37:42.444064Z"}},"outputs":[],"execution_count":null},{"id":"6229cbd5-104b-4c2c-a3ce-8f5c14316154","cell_type":"code","source":"NUM_BOOST_ROUND = 200\n\n\ndef make_xgb_dmatrix(x: np.ndarray, y: np.ndarray | None, group: list[int] | None = None) -> xgb.DMatrix:\n    dmat = xgb.DMatrix(x, label=y, feature_names=FEATURE_COLUMNS)\n    if group is not None:\n        dmat.set_group(group)\n    return dmat\n\n\ndef train_xgb_ranker(\n    x_train: np.ndarray,\n    y_train: np.ndarray,\n    train_group: list[int],\n    x_valid: np.ndarray | None = None,\n    y_valid: np.ndarray | None = None,\n    valid_group: list[int] | None = None,\n    num_boost_round: int = NUM_BOOST_ROUND,\n) -> xgb.Booster:\n    params = {\n        \"objective\": \"rank:ndcg\",\n        \"eval_metric\": \"ndcg@12\",\n        \"tree_method\": \"hist\",\n        \"device\": \"cuda\",\n        \"learning_rate\": 0.05,\n        \"max_depth\": 6,\n        \"min_child_weight\": 20,\n        \"subsample\": 0.9,\n        \"colsample_bytree\": 0.9,\n        \"lambda\": 1.0,\n        \"seed\": SEED,\n        \"verbosity\": 1,\n    }\n    dtrain = make_xgb_dmatrix(x_train, y_train, train_group)\n    evals = [(dtrain, \"train\")]\n    if x_valid is not None:\n        dvalid = make_xgb_dmatrix(x_valid, y_valid, valid_group)\n        evals.append((dvalid, \"valid\"))\n    try:\n        return xgb.train(params, dtrain, num_boost_round=num_boost_round, evals=evals, verbose_eval=25)\n    except xgb.core.XGBoostError as exc:\n        print(f\"XGBoost GPU training failed; falling back to CPU. Reason: {type(exc).__name__}: {exc}\")\n        params[\"device\"] = \"cpu\"\n        return xgb.train(params, dtrain, num_boost_round=num_boost_round, evals=evals, verbose_eval=25)\n\n\nmodel = train_xgb_ranker(x_train, y_train, train_group, x_valid, y_valid, valid_group)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:37:42.445921Z","iopub.execute_input":"2026-06-29T03:37:42.446152Z","iopub.status.idle":"2026-06-29T03:37:47.465615Z","shell.execute_reply.started":"2026-06-29T03:37:42.446132Z","shell.execute_reply":"2026-06-29T03:37:47.465099Z"}},"outputs":[],"execution_count":null},{"id":"03a5c926-cbd1-4f3f-a9c0-068a7015a31f","cell_type":"markdown","source":"## Score train and valid with the 2nd-stage model\n\nThis produces a full top120 reranked order, so we can compare 1st-stage and 2nd-stage metrics at the same cutoffs.","metadata":{}},{"id":"5fae1786-fca6-4973-baee-1895cd1e482e","cell_type":"code","source":"USE_XGB_GPU_INFERENCE = True\n\n\ndef predict_xgb(model: xgb.Booster, x: np.ndarray) -> np.ndarray:\n    \"\"\"Predict with GPU when CuPy is available; fall back to standard DMatrix prediction.\"\"\"\n    global USE_XGB_GPU_INFERENCE\n    if USE_XGB_GPU_INFERENCE:\n        try:\n            import cupy as cp\n            return cp.asnumpy(model.inplace_predict(cp.asarray(x)))\n        except Exception as exc:\n            USE_XGB_GPU_INFERENCE = False\n            print(f\"XGBoost GPU inference unavailable; falling back to DMatrix prediction. Reason: {type(exc).__name__}: {exc}\")\n    dmat = xgb.DMatrix(x, feature_names=FEATURE_COLUMNS)\n    return model.predict(dmat, iteration_range=(0, model.num_boosted_rounds()))\n\n\ndef score_candidates(model: xgb.Booster, frame: pl.DataFrame) -> pl.DataFrame:\n    work = frame.sort([\"split\", \"cutoff\", \"customer_idx\", \"candidate_order\"])\n    x = work.select(FEATURE_COLUMNS).fill_null(0).to_numpy().astype(np.float32)\n    scores = predict_xgb(model, x)\n    scored = work.with_columns(pl.Series(\"stage2_score\", scores).cast(pl.Float32))\n    return (\n        scored.sort([\"split\", \"cutoff\", \"customer_idx\", \"stage2_score\", \"candidate_order\"], descending=[False, False, False, True, False])\n        .with_columns((pl.int_range(pl.len()).over([\"split\", \"cutoff\", \"customer_idx\"]) + 1).cast(pl.Int32).alias(\"stage2_order\"))\n    )\n\n\ntrain_scored = score_candidates(model, train_candidates)\nvalid_scored = score_candidates(model, valid_candidates)\n\nstage2_train_metrics = metrics_from_ranked_frame(train_scored, \"stage2_order\", \"stage2\", \"train\")\nstage2_valid_metrics = metrics_from_ranked_frame(valid_scored, \"stage2_order\", \"stage2\", \"valid\")\n\nall_metrics = pl.concat([stage1_train_metrics, stage1_valid_metrics, stage2_train_metrics, stage2_valid_metrics])\nall_metrics.select([\"stage\", \"split\", \"k\", \"recall_customer_mean\", \"recall_item\", \"map\"])","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:37:47.467088Z","iopub.execute_input":"2026-06-29T03:37:47.467279Z","iopub.status.idle":"2026-06-29T03:37:51.118526Z","shell.execute_reply.started":"2026-06-29T03:37:47.467259Z","shell.execute_reply":"2026-06-29T03:37:51.117661Z"}},"outputs":[],"execution_count":null},{"id":"ca4775dc-b109-41ea-8655-4896b8919ae4","cell_type":"code","source":"valid_plot = all_metrics.filter(pl.col(\"split\") == \"valid\").sort([\"stage\", \"k\"])\n\nfig, axes = plt.subplots(1, 2, figsize=(12, 4.2))\nfor stage, color in [(\"stage1\", \"tab:blue\"), (\"stage2\", \"tab:orange\")]:\n    part = valid_plot.filter(pl.col(\"stage\") == stage)\n    axes[0].plot(part[\"k\"].to_list(), part[\"recall_customer_mean\"].to_list(), marker=\"o\", label=stage, color=color)\n    axes[1].plot(part[\"k\"].to_list(), part[\"map\"].to_list(), marker=\"o\", label=stage, color=color)\n\naxes[0].set_title(\"Valid recall customer mean\")\naxes[0].set_xlabel(\"k\")\naxes[0].set_ylabel(\"Recall\")\naxes[0].set_xticks(EVAL_CUTOFFS)\naxes[0].grid(True, alpha=0.25)\naxes[0].legend()\n\naxes[1].set_title(\"Valid MAP\")\naxes[1].set_xlabel(\"k\")\naxes[1].set_ylabel(\"MAP\")\naxes[1].set_xticks(EVAL_CUTOFFS)\naxes[1].grid(True, alpha=0.25)\naxes[1].legend()\n\nfig.tight_layout()\nplt.show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:37:51.119695Z","iopub.execute_input":"2026-06-29T03:37:51.120062Z","iopub.status.idle":"2026-06-29T03:37:51.458949Z","shell.execute_reply.started":"2026-06-29T03:37:51.120039Z","shell.execute_reply":"2026-06-29T03:37:51.458295Z"}},"outputs":[],"execution_count":null},{"id":"ab99e28a-ae7f-4ed8-8399-4d8f0c441808","cell_type":"code","source":"gain = model.get_score(importance_type=\"gain\")\nweight = model.get_score(importance_type=\"weight\")\nimportance = pl.DataFrame([\n    {\n        \"feature\": feature,\n        \"gain\": float(gain.get(feature, 0.0)),\n        \"split\": int(weight.get(feature, 0)),\n    }\n    for feature in FEATURE_COLUMNS\n]).sort(\"gain\", descending=True)\n\nimportance.head(20)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:37:51.459736Z","iopub.execute_input":"2026-06-29T03:37:51.459994Z","iopub.status.idle":"2026-06-29T03:37:51.468320Z","shell.execute_reply.started":"2026-06-29T03:37:51.459972Z","shell.execute_reply":"2026-06-29T03:37:51.467740Z"}},"outputs":[],"execution_count":null},{"id":"b0f55ee2-71d3-4f5a-9d6a-feb7d74ceb94","cell_type":"markdown","source":"## What to notice\n\n- The 1st stage controls candidate recall: if the true item is absent from top120, the 2nd stage cannot recover it.\n- The 2nd stage can improve MAP@12 by reordering the same candidate rows.\n- Source flags, source ranks, and `candidate_score` are useful 2nd-stage features because they explain why a row entered the candidate set.\n- This notebook is a demo, not a production experiment. Production runs should use the shared encoded parquet files, persisted feature store, more train cutoffs, and full customer groups.","metadata":{}},{"id":"74438f7b-98d8-408a-96c8-9680cf913289","cell_type":"markdown","source":"## Train the final demo model\n\nAfter checking validation metrics, we train one final demo reranker on the labeled train and validation candidate rows. This is still a small demo model because the earlier cells sample customers. Production submissions should use the full training recipe from the repo.","metadata":{}},{"id":"0b79ce04-2e80-409b-b7e6-8f178311a506","cell_type":"code","source":"final_labeled_candidates = pl.concat([train_candidates, valid_candidates], how=\"vertical_relaxed\")\nfinal_rank_frame, x_final, y_final, final_group = xgb_arrays(final_labeled_candidates, positive_groups_only=True)\n\nfinal_model = train_xgb_ranker(\n    x_final,\n    y_final,\n    final_group,\n    num_boost_round=model.num_boosted_rounds(),\n)\n\nprint(f\"final demo model rows={x_final.shape[0]:,}; groups={len(final_group):,}; positives={int(y_final.sum()):,}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:38:13.545808Z","iopub.execute_input":"2026-06-29T03:38:13.546624Z","iopub.status.idle":"2026-06-29T03:38:18.910763Z","shell.execute_reply.started":"2026-06-29T03:38:13.546593Z","shell.execute_reply":"2026-06-29T03:38:18.910062Z"}},"outputs":[],"execution_count":null},{"id":"7945e4c8-4fc6-46a3-8c2b-62e63a42c4f5","cell_type":"markdown","source":"## Create a Kaggle submission\n\nThis section uses all `sample_submission.csv` customers by default. It builds candidates at `cutoff=2020-09-22`, scores each chunk with the final 2nd-stage model, converts `article_idx` back to the original string `article_id`, and writes `submission.csv`.\n\nFor a quick smoke test, set `MAX_SUBMISSION_CUSTOMERS` to a small number. For an actual Kaggle submission, keep `MAX_SUBMISSION_CUSTOMERS = None`.","metadata":{}},{"id":"797e0139-6af1-4df9-87ca-610cf0418bfd","cell_type":"code","source":"def clone_candidate_map(candidate_map: dict[int, dict]) -> dict[int, dict]:\n    return {int(article_idx): features.copy() for article_idx, features in candidate_map.items()}\n\n\ndef build_static_candidates_for_age_bucket(bucket: str, components: dict) -> dict[int, dict]:\n    candidates = {}\n\n    for rank, (article_idx, cnt) in enumerate(components[\"age_top\"].get(bucket, []), start=1):\n        add_candidate(candidates, article_idx, \"age14\", rank, cnt)\n\n    for days, top_items in components[\"global_top\"].items():\n        source = f\"global{days}\"\n        for rank, (article_idx, cnt) in enumerate(top_items, start=1):\n            add_candidate(candidates, article_idx, source, rank, cnt)\n\n    return candidates\n\n\ndef add_personal_candidates(candidates: dict[int, dict], customer_idx: int, components: dict) -> None:\n    for rank, (article_idx, cnt, recency_days) in enumerate(components[\"history\"].get(customer_idx, []), start=1):\n        add_candidate(candidates, article_idx, \"history\", rank, cnt, recency_days)\n\n    category_rank = 1\n    for key in components[\"category_preferences\"].get(customer_idx, []):\n        for article_idx, cnt in components[\"category_top\"].get(key, []):\n            add_candidate(candidates, article_idx, \"category\", category_rank, cnt)\n            category_rank += 1\n\n\ndef make_submission_ordered_candidate_getter(components: dict):\n    buckets = sorted(set(age_bucket_values) | set(customer_age_bucket.values()) | {\"unknown\"})\n    base_by_bucket = {bucket: build_static_candidates_for_age_bucket(bucket, components) for bucket in buckets}\n    static_ordered_by_bucket = {bucket: ordered_candidates(base) for bucket, base in base_by_bucket.items()}\n\n    def get_ordered(customer_idx: int) -> list[tuple[int, dict]]:\n        bucket = customer_age_bucket.get(customer_idx, \"unknown\")\n        history = components[\"history\"].get(customer_idx, [])\n        category_preferences = components[\"category_preferences\"].get(customer_idx, [])\n\n        # Many submission customers have no recent personal signal. Reuse the static age+global list.\n        if not history and not category_preferences:\n            return static_ordered_by_bucket[bucket]\n\n        candidates = clone_candidate_map(base_by_bucket[bucket])\n        add_personal_candidates(candidates, customer_idx, components)\n        return ordered_candidates(candidates)\n\n    return get_ordered\n\n\ndef build_submission_feature_arrays_for_customers(\n    customer_ids: list[int],\n    get_ordered_candidates,\n) -> tuple[np.ndarray, np.ndarray, list[int], list[int]]:\n    expected_feature_columns = [\n        \"candidate_score\",\n        \"candidate_order\",\n        \"age_bucket_idx\",\n        *SOURCE_COLUMNS,\n        *RANK_COLUMNS,\n        *ARTICLE_FEATURE_COLUMNS,\n    ]\n    assert FEATURE_COLUMNS == expected_feature_columns\n\n    max_rows = max(len(customer_ids) * CANDIDATE_LIMIT, 1)\n    x = np.empty((max_rows, len(FEATURE_COLUMNS)), dtype=np.float32)\n    article_idx_values = np.empty(max_rows, dtype=np.int32)\n    group_customers = []\n    group_sizes = []\n    article_default = {col: 0 for col in ARTICLE_FEATURE_COLUMNS}\n    row_pos = 0\n\n    for customer_idx in customer_ids:\n        ordered = get_ordered_candidates(customer_idx)\n        age_idx = customer_age_bucket_idx.get(customer_idx, 0)\n        group_customers.append(int(customer_idx))\n        group_sizes.append(len(ordered))\n\n        for order, (article_idx, features) in enumerate(ordered, start=1):\n            article_features = article_feature_map.get(article_idx, article_default)\n            col_pos = 0\n            x[row_pos, col_pos] = features[\"candidate_score\"]; col_pos += 1\n            x[row_pos, col_pos] = order; col_pos += 1\n            x[row_pos, col_pos] = age_idx; col_pos += 1\n            for col in SOURCE_COLUMNS:\n                x[row_pos, col_pos] = features[col]; col_pos += 1\n            for col in RANK_COLUMNS:\n                x[row_pos, col_pos] = features[col]; col_pos += 1\n            for col in ARTICLE_FEATURE_COLUMNS:\n                x[row_pos, col_pos] = article_features[col]; col_pos += 1\n            article_idx_values[row_pos] = int(article_idx)\n            row_pos += 1\n\n    return x[:row_pos], article_idx_values[:row_pos], group_customers, group_sizes\n\n\ndef top12_prediction_strings_from_arrays(\n    model: xgb.Booster,\n    x: np.ndarray,\n    article_idx_values: np.ndarray,\n    group_customers: list[int],\n    group_sizes: list[int],\n    article_idx_to_id: dict[int, str],\n    fallback_prediction: str,\n) -> pl.DataFrame:\n    scores = predict_xgb(model, x) if len(x) else np.array([], dtype=np.float32)\n    out_rows = []\n    offset = 0\n\n    for customer_idx, size in zip(group_customers, group_sizes):\n        end = offset + size\n        if size == 0:\n            prediction = fallback_prediction\n        else:\n            local_scores = scores[offset:end]\n            local_articles = article_idx_values[offset:end]\n            order = np.argsort(-local_scores, kind=\"stable\")[:12]\n            prediction = \" \".join(article_idx_to_id.get(int(article_idx), \"\") for article_idx in local_articles[order])\n        out_rows.append({\"customer_idx\": int(customer_idx), \"prediction\": prediction})\n        offset = end\n\n    return pl.DataFrame(out_rows)\n\n\narticle_idx_to_id = {int(row[\"article_idx\"]): str(row[\"article_id\"]) for row in article_map.iter_rows(named=True)}\nsubmission_base_customers = sample_submission_customers\nif MAX_SUBMISSION_CUSTOMERS is not None:\n    print(\"WARNING: MAX_SUBMISSION_CUSTOMERS is set. The output is only a smoke-test subset, not a valid full Kaggle submission.\")\n    submission_base_customers = sample_submission_customers.head(MAX_SUBMISSION_CUSTOMERS)\nsubmission_customer_ids = [int(x) for x in submission_base_customers[\"customer_idx\"].to_list()]\n\nsubmission_components = build_cutoff_components(SUBMISSION_CUTOFF)\nget_submission_ordered_candidates = make_submission_ordered_candidate_getter(submission_components)\nglobal_fallback = \" \".join(article_idx_to_id[int(article_idx)] for article_idx, _ in submission_components[\"global_top\"][7][:12])\nprediction_frames = []\n\nfor start in range(0, len(submission_customer_ids), SUBMISSION_CHUNK_SIZE):\n    chunk_customers = submission_customer_ids[start:start + SUBMISSION_CHUNK_SIZE]\n    x_chunk, article_idx_chunk, chunk_group_customers, chunk_group_sizes = build_submission_feature_arrays_for_customers(\n        chunk_customers, get_submission_ordered_candidates\n    )\n    prediction_frames.append(\n        top12_prediction_strings_from_arrays(\n            final_model,\n            x_chunk,\n            article_idx_chunk,\n            chunk_group_customers,\n            chunk_group_sizes,\n            article_idx_to_id,\n            global_fallback,\n        )\n    )\n    print(f\"scored submission customers {start + len(chunk_customers):,}/{len(submission_customer_ids):,}\")\n\nsubmission_pred = pl.concat(prediction_frames, how=\"vertical_relaxed\") if prediction_frames else pl.DataFrame({\"customer_idx\": [], \"prediction\": []})\nsubmission = (\n    submission_base_customers.join(submission_pred, on=\"customer_idx\", how=\"left\")\n    .with_columns(pl.col(\"prediction\").fill_null(global_fallback))\n    .select([\"customer_id\", \"prediction\"])\n)\n\nsubmission.write_csv(SUBMISSION_OUTPUT)\nprint(f\"wrote {SUBMISSION_OUTPUT} rows={submission.height:,}\")\nsubmission.head()\n\n\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:38:21.807090Z","iopub.execute_input":"2026-06-29T03:38:21.807718Z","iopub.status.idle":"2026-06-29T03:39:14.012172Z","shell.execute_reply.started":"2026-06-29T03:38:21.807691Z","shell.execute_reply":"2026-06-29T03:39:14.011195Z"}},"outputs":[],"execution_count":null},{"id":"273c5dab-2962-4e54-b930-154cea68fcb9","cell_type":"code","source":"# The output file should match Kaggle's required columns.\nsubmission.select([pl.len().alias(\"rows\"), pl.col(\"prediction\").str.split(\" \").list.len().min().alias(\"min_pred_len\"), pl.col(\"prediction\").str.split(\" \").list.len().max().alias(\"max_pred_len\")])","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-29T03:39:14.013187Z","iopub.status.idle":"2026-06-29T03:39:14.013513Z","shell.execute_reply.started":"2026-06-29T03:39:14.013316Z","shell.execute_reply":"2026-06-29T03:39:14.013331Z"}},"outputs":[],"execution_count":null},{"id":"e9b3335d-b9d3-448a-b874-077d9640ff1a","cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}