{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"name":"python","version":"3.12.12","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":[{"cell_type":"markdown","source":"# Spark-Based Cross-Domain Customer Churn Benchmark for Journal Submission\n\nThis is the final journal-facing Kaggle notebook for the Spark-based customer churn benchmark.\n\n**Purpose**\n\nThe notebook builds a reproducible Big Data + Spark MLlib benchmark for customer churn transportability across public datasets and business domains.\n\n**Main design**\n\n1. Resolve the mounted public churn datasets in Kaggle.\n2. Use Spark/PySpark for raw source profiling, schema checks, missingness profiling, target prevalence checks, and harmonized artifact generation.\n3. Harmonize 12 public churn datasets into a shared churn feature ontology.\n4. Use Spark MLlib class-weighted logistic regression as the main scalable and interpretable modeling baseline.\n5. Evaluate four protocols: within-dataset, pooled random split, leave-one-dataset-out, and leave-one-domain-out.\n6. Export manuscript-ready CSV tables, summary text, and high-resolution figures in PNG, PDF, and SVG formats.\n\n**Journal-facing framing**\n\nDescribe the work as a **Spark-based Big Data and ML benchmark for churn transportability under dataset and domain shift**. Do not describe it using internal notebook version labels.\n\n**Output package**\n\nThe final small export file is created at:\n\n```text\n/kaggle/working/spark_churn_paper_only_results.zip\n```\n","metadata":{}},{"cell_type":"code","source":"# ============================================================\n# 1. Imports, configuration, and output folders\n# ============================================================\n\nimport os\nimport re\nimport json\nimport math\nimport time\nimport shutil\nimport hashlib\nimport warnings\nimport subprocess\nimport gc\nfrom pathlib import Path\nfrom datetime import datetime\nfrom collections import defaultdict\n\nimport numpy as np\nimport pandas as pd\n\nwarnings.filterwarnings(\"ignore\")\n\nfrom IPython.display import display\n\nimport matplotlib.pyplot as plt\n\ntry:\n    from scipy.stats import wilcoxon\n    SCIPY_AVAILABLE = True\nexcept Exception:\n    SCIPY_AVAILABLE = False\n\nRANDOM_SEED = 42\nnp.random.seed(RANDOM_SEED)\n\n# Runtime controls. For the full journal run, keep RUN_QUICK_TEST=False.\nRUN_QUICK_TEST = False\nRESOURCE_SAFE_LIGHT_KKBOX = True  # Avoid full KKBox transaction aggregation in low-memory Kaggle runs.\nMAX_ROWS_PER_DATASET_FOR_MODELING = 60000\nTRAIN_FRACTION = 0.60\nVALIDATION_FRACTION = 0.20\nTEST_FRACTION = 0.20\n\n# Spark MLlib model controls.\nRUN_LOGISTIC_REGRESSION = True\nRUN_RANDOM_FOREST = False  # Kaggle-safe default; set True only for optional RF rerun after LR completes.\nRUN_GBT = False      # Optional; set True only after LR/RF complete successfully.\nRF_NUM_TREES = 8   # Resource-safe journal run; enough for stable Spark MLlib baseline.\nRF_MAX_DEPTH = 4\nGBT_MAX_ITER = 30\n\n# Output folders use journal-safe names.\nOUTPUT_ROOT = \"/kaggle/working/spark_churn_journal_benchmark\"\nDIRS = {\n    \"root\": OUTPUT_ROOT,\n    \"registry\": f\"{OUTPUT_ROOT}/registry\",\n    \"profiles\": f\"{OUTPUT_ROOT}/bigdata_profiles\",\n    \"harmonized\": f\"{OUTPUT_ROOT}/harmonized_data\",\n    \"results\": f\"{OUTPUT_ROOT}/results\",\n    \"figures\": f\"{OUTPUT_ROOT}/figures\",\n    \"tables\": f\"{OUTPUT_ROOT}/tables\",\n    \"logs\": f\"{OUTPUT_ROOT}/logs\",\n}\nfor d in DIRS.values():\n    os.makedirs(d, exist_ok=True)\n\nprint(\"Output root:\", OUTPUT_ROOT)\nprint(\"Full journal run:\", not RUN_QUICK_TEST)\nprint(\"Modeling cap per dataset:\", MAX_ROWS_PER_DATASET_FOR_MODELING)\nprint(\"Random seed:\", RANDOM_SEED)\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T06:49:19.901585Z","iopub.execute_input":"2026-06-28T06:49:19.901860Z","iopub.status.idle":"2026-06-28T06:49:20.737708Z","shell.execute_reply.started":"2026-06-28T06:49:19.901828Z","shell.execute_reply":"2026-06-28T06:49:20.736508Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 2. Spark setup\n# ============================================================\n\ntry:\n    import pyspark\n    from pyspark.sql import SparkSession\n    from pyspark.sql import functions as F\n    from pyspark.sql import types as T\n    from pyspark.sql.window import Window\n    from pyspark import StorageLevel\n\n    from pyspark.ml import Pipeline\n    from pyspark.ml.feature import (\n        Imputer,\n        StringIndexer,\n        OneHotEncoder,\n        VectorAssembler,\n        StandardScaler,\n    )\n    from pyspark.ml.classification import (\n        LogisticRegression,\n        RandomForestClassifier,\n        GBTClassifier,\n    )\n    from pyspark.ml.evaluation import BinaryClassificationEvaluator\n    from pyspark.ml.functions import vector_to_array\n\n    SPARK_AVAILABLE = True\nexcept Exception as e:\n    SPARK_AVAILABLE = False\n    raise RuntimeError(\n        \"PySpark is required for this journal notebook. In Kaggle, the audit output showed PySpark 4.0.2 is available. \"\n        f\"Import error: {repr(e)}\"\n    )\n\nspark = (\n    SparkSession.builder\n    .appName(\"Spark_Cross_Domain_Churn_Journal_Benchmark\")\n    .master(\"local[*]\")\n    .config(\"spark.sql.shuffle.partitions\", \"16\")\n    .config(\"spark.default.parallelism\", \"16\")\n    .config(\"spark.driver.memory\", \"10g\")\n    .config(\"spark.memory.fraction\", \"0.70\")\n    .config(\"spark.memory.storageFraction\", \"0.25\")\n    .config(\"spark.sql.adaptive.enabled\", \"true\")\n    .config(\"spark.sql.execution.arrow.pyspark.enabled\", \"false\")\n    .getOrCreate()\n)\nspark.sparkContext.setLogLevel(\"WARN\")\n\nruntime_info = pd.DataFrame([{\n    \"timestamp_utc\": datetime.utcnow().isoformat(timespec=\"seconds\") + \"Z\",\n    \"spark_available\": True,\n    \"spark_version\": spark.version,\n    \"spark_master\": spark.sparkContext.master,\n    \"output_root\": OUTPUT_ROOT,\n    \"max_rows_per_dataset_for_modeling\": MAX_ROWS_PER_DATASET_FOR_MODELING,\n    \"run_quick_test\": RUN_QUICK_TEST,\n}])\nruntime_info.to_csv(f\"{DIRS['logs']}/runtime_info.csv\", index=False)\ndisplay(runtime_info)\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T06:49:20.739407Z","iopub.execute_input":"2026-06-28T06:49:20.739796Z","iopub.status.idle":"2026-06-28T06:49:31.828181Z","shell.execute_reply.started":"2026-06-28T06:49:20.739771Z","shell.execute_reply":"2026-06-28T06:49:31.827180Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 3. Input path resolver for Kaggle-mounted datasets\n# ============================================================\n\nINPUT_ROOTS = [\"/kaggle/input/datasets\", \"/kaggle/input/competitions\", \"/kaggle/input\"]\n\n\ndef find_input_file(fragment, required=True, prefer_contains=None):\n    \"\"\"Find a file under Kaggle input roots using a stable filename/path fragment.\"\"\"\n    fragment_l = fragment.lower().replace(\"\\\\\", \"/\")\n    matches = []\n    for input_root in INPUT_ROOTS:\n        if not os.path.exists(input_root):\n            continue\n        for root, dirs, files in os.walk(input_root):\n            for f in files:\n                full = os.path.join(root, f)\n                rel = full.replace(\"\\\\\", \"/\")\n                if fragment_l in rel.lower():\n                    matches.append(full)\n    if prefer_contains:\n        pref = prefer_contains.lower()\n        preferred = [m for m in matches if pref in m.lower()]\n        if preferred:\n            matches = preferred\n    matches = sorted(set(matches))\n    if not matches:\n        if required:\n            raise FileNotFoundError(f\"Could not find input file matching fragment: {fragment}\")\n        return None\n    if len(matches) > 1:\n        print(f\"Multiple matches for {fragment}; using first:\")\n        for m in matches[:10]:\n            print(\"  \", m)\n    return matches[0]\n\nPATHS = {\n    \"telco_ibm\": find_input_file(\"WA_Fn-UseC_-Telco-Customer-Churn.csv\"),\n    \"bank_original\": find_input_file(\"Bank_Customer_Churn_Prediction.csv\"),\n    \"rees46\": find_input_file(\"rees46_customer_model.csv\"),\n    \"rees46_dictionary\": find_input_file(\"rees46_data_dictionary.csv\"),\n\n    \"kkbox_train_archive\": find_input_file(\"train_v2.csv.7z\"),\n    \"kkbox_members_archive\": find_input_file(\"members_v3.csv.7z\"),\n    \"kkbox_transactions_archive\": find_input_file(\"transactions_v2.csv.7z\"),\n\n    \"bank_creditcard\": find_input_file(\"credit_card_churn.csv\"),\n    \"subscription_generic_train\": find_input_file(\"customer_churn_dataset-training-master.csv\"),\n    \"subscription_generic_test\": find_input_file(\"customer_churn_dataset-testing-master.csv\"),\n    \"telco_orange_train\": find_input_file(\"churn-bigml-80.csv\"),\n    \"telco_orange_test\": find_input_file(\"churn-bigml-20.csv\"),\n    \"insurance_auto\": find_input_file(\"autoinsurance_churn.csv\"),\n\n    \"cell2cell_train\": find_input_file(\"cell2celltrain.csv\"),\n    \"cell2cell_holdout\": find_input_file(\"cell2cellholdout.csv\"),\n    \"ecommerce_shriyash_large\": find_input_file(\"ecommerce_customer_data_large.csv\"),\n    \"ecommerce_dhairyajeet\": find_input_file(\"ecommerce_customer_churn_dataset.csv\"),\n    \"insurance_k123_train\": find_input_file(\"Insurance_Churn_ParticipantsData/Train.csv\"),\n}\n\npath_rows = []\nfor k, p in PATHS.items():\n    path_rows.append({\n        \"file_key\": k,\n        \"path\": p,\n        \"exists\": bool(p and os.path.exists(p)),\n        \"size_mb\": round(os.path.getsize(p) / (1024 * 1024), 3) if p and os.path.exists(p) else np.nan,\n    })\npath_check = pd.DataFrame(path_rows)\npath_check.to_csv(f\"{DIRS['registry']}/input_file_registry.csv\", index=False)\ndisplay(path_check)\nassert path_check[\"exists\"].all(), \"Some required input files are missing. Check the Kaggle Add Input panel.\"\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T06:49:31.829274Z","iopub.execute_input":"2026-06-28T06:49:31.829495Z","iopub.status.idle":"2026-06-28T06:49:32.142269Z","shell.execute_reply.started":"2026-06-28T06:49:31.829472Z","shell.execute_reply":"2026-06-28T06:49:32.141038Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 4. Extract KKBox archives used by the subscription-domain dataset\n# ============================================================\n\nKKBOX_DIR = \"/kaggle/working/kkbox_extracted_for_spark_churn\"\nos.makedirs(KKBOX_DIR, exist_ok=True)\n\n\ndef extract_7z_if_needed(archive_path, out_dir, expected_filename):\n    expected_path = os.path.join(out_dir, expected_filename)\n    if os.path.exists(expected_path):\n        print(\"Already extracted:\", expected_path)\n        return expected_path\n    print(\"Extracting:\", archive_path)\n    cmd = [\"7z\", \"x\", \"-y\", archive_path, f\"-o{out_dir}\"]\n    proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)\n    if proc.returncode != 0:\n        print(proc.stdout[-2000:])\n        print(proc.stderr[-2000:])\n        raise RuntimeError(\"7z extraction failed. In Kaggle, ensure the competition files are attached and the 7z command is available.\")\n    if not os.path.exists(expected_path):\n        # Search recursively in case the archive created subfolders.\n        for root, dirs, files in os.walk(out_dir):\n            if expected_filename in files:\n                return os.path.join(root, expected_filename)\n        raise FileNotFoundError(f\"Expected extracted file not found: {expected_filename}\")\n    return expected_path\n\nPATHS[\"kkbox_train\"] = extract_7z_if_needed(PATHS[\"kkbox_train_archive\"], KKBOX_DIR, \"train_v2.csv\")\nPATHS[\"kkbox_members\"] = extract_7z_if_needed(PATHS[\"kkbox_members_archive\"], KKBOX_DIR, \"members_v3.csv\")\nPATHS[\"kkbox_transactions\"] = extract_7z_if_needed(PATHS[\"kkbox_transactions_archive\"], KKBOX_DIR, \"transactions_v2.csv\")\n\nkkbox_paths = pd.DataFrame([{k: PATHS[k]} for k in [\"kkbox_train\", \"kkbox_members\", \"kkbox_transactions\"]])\ndisplay(kkbox_paths)\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T06:49:32.144421Z","iopub.execute_input":"2026-06-28T06:49:32.144662Z","iopub.status.idle":"2026-06-28T06:49:54.207825Z","shell.execute_reply.started":"2026-06-28T06:49:32.144641Z","shell.execute_reply":"2026-06-28T06:49:54.207039Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 5. Shared churn ontology and Spark helper functions\n# ============================================================\n\nCOMMON_COLUMNS = [\n    \"customer_id\",\n    \"dataset_id\",\n    \"domain_id\",\n    \"tenure_months\",\n    \"age\",\n    \"gender\",\n    \"engagement_score\",\n    \"payment_score\",\n    \"usage_score\",\n    \"support_score\",\n    \"transaction_count\",\n    \"avg_payment_amount\",\n    \"total_payment_amount\",\n    \"service_count\",\n    \"auto_renew_count\",\n    \"cancel_count\",\n    \"recent_activity_score\",\n    \"churn\",\n]\n\nNUMERIC_FEATURES = [\n    \"tenure_months\",\n    \"age\",\n    \"engagement_score\",\n    \"payment_score\",\n    \"usage_score\",\n    \"support_score\",\n    \"transaction_count\",\n    \"avg_payment_amount\",\n    \"total_payment_amount\",\n    \"service_count\",\n    \"auto_renew_count\",\n    \"cancel_count\",\n    \"recent_activity_score\",\n]\nCATEGORICAL_FEATURES = [\"gender\"]\nFEATURE_COLUMNS = NUMERIC_FEATURES + CATEGORICAL_FEATURES\n\nCHURN_ONTOLOGY = {\n    \"tenure_months\": \"customer relationship duration\",\n    \"age\": \"customer age or age proxy\",\n    \"gender\": \"portable demographic category when available\",\n    \"engagement_score\": \"activity or interaction intensity\",\n    \"payment_score\": \"payment, balance, price, or payment-delay proxy\",\n    \"usage_score\": \"usage amount or usage intensity\",\n    \"support_score\": \"support calls, tickets, contacts, or complaint burden\",\n    \"transaction_count\": \"transaction, call, purchase, login, or renewal frequency\",\n    \"avg_payment_amount\": \"average monetary amount\",\n    \"total_payment_amount\": \"total monetary amount or customer value proxy\",\n    \"service_count\": \"breadth of products, services, plans, or relationships\",\n    \"auto_renew_count\": \"renewal, long-contract, or auto-renew proxy\",\n    \"cancel_count\": \"safe cancellation-risk or complaint proxy, not direct post-outcome target\",\n    \"recent_activity_score\": \"recency or recent inactivity proxy\",\n}\n\nwith open(f\"{DIRS['registry']}/shared_churn_ontology.json\", \"w\") as f:\n    json.dump({\"features\": FEATURE_COLUMNS, \"target\": \"churn\", \"ontology\": CHURN_ONTOLOGY}, f, indent=2)\n\n\ndef spark_read_csv(path, infer_schema=True):\n    reader = (\n        spark.read\n        .option(\"header\", True)\n        .option(\"multiLine\", False)\n        .option(\"quote\", '\"')\n        .option(\"escape\", '\"')\n        .option(\"mode\", \"PERMISSIVE\")\n    )\n    if infer_schema:\n        reader = reader.option(\"inferSchema\", True).option(\"samplingRatio\", 0.20)\n    return reader.csv(path)\n\n\ndef clean_numeric_expr(expr):\n    \"\"\"Convert messy numeric strings to double without triggering Spark 4 ANSI cast errors.\n\n    Spark 4 runs in ANSI mode by default; direct `.cast(\"double\")` fails on blank\n    strings such as ``\"\"``. This helper first removes non-numeric characters, validates\n    the cleaned string with a numeric regex, and only casts valid values. Invalid or\n    blank values become NULL and can then be imputed downstream.\n    \"\"\"\n    cleaned = F.regexp_replace(F.trim(expr.cast(\"string\")), r\"[^0-9eE+\\-.]\", \"\")\n    valid_number = cleaned.rlike(r\"^[+-]?(?:\\d+(?:\\.\\d*)?|\\.\\d+)(?:[eE][+-]?\\d+)?$\")\n    return F.when(valid_number, cleaned.cast(\"double\")).otherwise(F.lit(None).cast(\"double\"))\n\n\ndef num(df, col_name, default=None):\n    if col_name in df.columns:\n        out = clean_numeric_expr(F.col(col_name))\n    else:\n        out = F.lit(None).cast(\"double\")\n    if default is not None:\n        out = F.coalesce(out, F.lit(float(default)))\n    return out\n\n\ndef str_expr(df, col_name, default=\"unknown\"):\n    if col_name in df.columns:\n        return F.coalesce(F.col(col_name).cast(\"string\"), F.lit(default))\n    return F.lit(default)\n\n\ndef yes_no(df, col_name):\n    if col_name not in df.columns:\n        return F.lit(0.0)\n    x = F.lower(F.trim(F.col(col_name).cast(\"string\")))\n    return F.when(x.isin(\"yes\", \"y\", \"true\", \"1\"), F.lit(1.0)).otherwise(F.lit(0.0))\n\n\ndef bool_or_one(df, col_name):\n    if col_name not in df.columns:\n        return F.lit(0.0)\n    x = F.lower(F.trim(F.col(col_name).cast(\"string\")))\n    return F.when(x.isin(\"true\", \"yes\", \"y\", \"1\"), F.lit(1.0)).otherwise(F.lit(0.0))\n\n\ndef contains_attrited(df, col_name):\n    x = F.lower(F.trim(F.col(col_name).cast(\"string\")))\n    return F.when(x.contains(\"attrited\"), F.lit(1.0)).otherwise(F.lit(0.0))\n\n\ndef safe_divide_expr(a, b):\n    return F.when((b.isNotNull()) & (F.abs(b) > F.lit(1e-12)), a / b).otherwise(F.lit(None).cast(\"double\"))\n\n\ndef sum_nums(df, columns):\n    expr = F.lit(0.0)\n    for c in columns:\n        expr = expr + num(df, c, default=0.0)\n    return expr\n\n\ndef count_yes(df, columns):\n    expr = F.lit(0.0)\n    for c in columns:\n        expr = expr + yes_no(df, c)\n    return expr\n\n\ndef normalize_gender_expr(expr):\n    x = F.lower(F.trim(expr.cast(\"string\")))\n    return (\n        F.when(x.isin(\"m\", \"male\"), F.lit(\"male\"))\n        .when(x.isin(\"f\", \"female\"), F.lit(\"female\"))\n        .otherwise(F.lit(\"unknown\"))\n    )\n\n\ndef finalize_spark_common(df, dataset_id, domain_id):\n    out = df\n    if \"customer_id\" not in out.columns:\n        out = out.withColumn(\"customer_id\", F.monotonically_increasing_id().cast(\"string\"))\n    if \"gender\" not in out.columns:\n        out = out.withColumn(\"gender\", F.lit(\"unknown\"))\n\n    out = out.withColumn(\"dataset_id\", F.lit(dataset_id))\n    out = out.withColumn(\"domain_id\", F.lit(domain_id))\n    out = out.withColumn(\"gender\", normalize_gender_expr(F.col(\"gender\")))\n\n    for c in NUMERIC_FEATURES:\n        if c not in out.columns:\n            out = out.withColumn(c, F.lit(None).cast(\"double\"))\n        else:\n            out = out.withColumn(c, clean_numeric_expr(F.col(c)))\n\n    out = out.withColumn(\"churn\", clean_numeric_expr(F.col(\"churn\")))\n    out = out.filter(F.col(\"churn\").isin(0.0, 1.0))\n    out = out.select(*COMMON_COLUMNS)\n    return out\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T06:49:54.208682Z","iopub.execute_input":"2026-06-28T06:49:54.208955Z","iopub.status.idle":"2026-06-28T06:49:54.330475Z","shell.execute_reply.started":"2026-06-28T06:49:54.208930Z","shell.execute_reply":"2026-06-28T06:49:54.329665Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 6. Big Data source profiling with Spark\n# ============================================================\n\nSOURCE_PROFILE_CONFIG = [\n    (\"telco_ibm\", \"telecom\", PATHS[\"telco_ibm\"], \"Churn\", \"yes\"),\n    (\"telco_orange_train\", \"telecom\", PATHS[\"telco_orange_train\"], \"Churn\", \"bool\"),\n    (\"telco_orange_test\", \"telecom\", PATHS[\"telco_orange_test\"], \"Churn\", \"bool\"),\n    (\"cell2cell_train\", \"telecom\", PATHS[\"cell2cell_train\"], \"Churn\", \"yes\"),\n    (\"cell2cell_holdout\", \"telecom\", PATHS[\"cell2cell_holdout\"], \"Churn\", \"yes\"),\n    (\"bank_original\", \"banking\", PATHS[\"bank_original\"], \"churn\", \"numeric\"),\n    (\"bank_creditcard\", \"banking\", PATHS[\"bank_creditcard\"], \"Attrition_Flag\", \"attrited\"),\n    (\"ecommerce_rees46\", \"ecommerce\", PATHS[\"rees46\"], \"target_event\", \"numeric\"),\n    (\"ecommerce_shriyash_large\", \"ecommerce\", PATHS[\"ecommerce_shriyash_large\"], \"Churn\", \"numeric\"),\n    (\"ecommerce_dhairyajeet\", \"ecommerce\", PATHS[\"ecommerce_dhairyajeet\"], \"Churned\", \"numeric\"),\n    (\"subscription_generic_train\", \"subscription\", PATHS[\"subscription_generic_train\"], \"Churn\", \"numeric\"),\n    (\"subscription_generic_test\", \"subscription\", PATHS[\"subscription_generic_test\"], \"Churn\", \"numeric\"),\n    (\"subscription_kkbox_train\", \"subscription\", PATHS[\"kkbox_train\"], \"is_churn\", \"numeric\"),\n    (\"subscription_kkbox_members\", \"subscription\", PATHS[\"kkbox_members\"], None, None),\n    (\"subscription_kkbox_transactions\", \"subscription\", PATHS[\"kkbox_transactions\"], None, None),\n    (\"insurance_auto\", \"insurance\", PATHS[\"insurance_auto\"], \"Churn\", \"numeric\"),\n    (\"insurance_k123_train\", \"insurance\", PATHS[\"insurance_k123_train\"], \"labels\", \"numeric\"),\n]\n\n\ndef positive_expr_for_profile(df, target_col, target_type):\n    if target_col is None or target_col not in df.columns:\n        return None\n    if target_type == \"yes\":\n        return yes_no(df, target_col)\n    if target_type == \"bool\":\n        return bool_or_one(df, target_col)\n    if target_type == \"attrited\":\n        return contains_attrited(df, target_col)\n    if target_type == \"numeric\":\n        return F.when(num(df, target_col) == 1.0, F.lit(1.0)).otherwise(F.lit(0.0))\n    return None\n\nsource_profile_rows = []\nmissingness_rows = []\nschema_rows = []\n\nt0 = time.time()\nfor source_key, domain_id, path, target_col, target_type in SOURCE_PROFILE_CONFIG:\n    print(\"Profiling:\", source_key)\n    sdf = spark_read_csv(path)\n    n_rows = sdf.count()\n    n_cols = len(sdf.columns)\n    size_mb = os.path.getsize(path) / (1024 * 1024) if os.path.exists(path) else np.nan\n\n    prof = {\n        \"source_key\": source_key,\n        \"domain_id\": domain_id,\n        \"path\": path,\n        \"file_name\": os.path.basename(path),\n        \"source_size_mb\": round(size_mb, 3),\n        \"spark_rows\": int(n_rows),\n        \"spark_columns\": int(n_cols),\n        \"target_column\": target_col,\n    }\n    pos_expr = positive_expr_for_profile(sdf, target_col, target_type)\n    if pos_expr is not None:\n        tmp = sdf.withColumn(\"_target_profile\", pos_expr)\n        stats = tmp.agg(\n            F.count(\"_target_profile\").alias(\"target_non_null\"),\n            F.sum(\"_target_profile\").alias(\"positive_count\"),\n            F.avg(\"_target_profile\").alias(\"positive_rate\"),\n        ).collect()[0].asDict()\n        prof.update(stats)\n    source_profile_rows.append(prof)\n\n    for col_name, dtype in sdf.dtypes:\n        schema_rows.append({\n            \"source_key\": source_key,\n            \"domain_id\": domain_id,\n            \"column_name\": col_name,\n            \"spark_dtype\": dtype,\n        })\n\n    # Missingness profiling: full scan for moderate files and selected columns for very wide files.\n    cols_for_missing = sdf.columns[:120]\n    agg_exprs = []\n    for col_name in cols_for_missing:\n        agg_exprs.append(\n            F.sum(F.when(F.col(col_name).isNull() | (F.trim(F.col(col_name).cast(\"string\")) == \"\"), 1).otherwise(0)).alias(col_name)\n        )\n    miss = sdf.agg(*agg_exprs).collect()[0].asDict() if agg_exprs else {}\n    for col_name, miss_count in miss.items():\n        missingness_rows.append({\n            \"source_key\": source_key,\n            \"domain_id\": domain_id,\n            \"column_name\": col_name,\n            \"missing_or_blank_count\": int(miss_count),\n            \"missing_or_blank_rate\": float(miss_count) / max(n_rows, 1),\n        })\n\nsource_profile = pd.DataFrame(source_profile_rows)\nschema_profile = pd.DataFrame(schema_rows)\nmissingness_profile = pd.DataFrame(missingness_rows)\n\nsource_profile.to_csv(f\"{DIRS['profiles']}/spark_source_profile.csv\", index=False)\nschema_profile.to_csv(f\"{DIRS['profiles']}/spark_schema_profile.csv\", index=False)\nmissingness_profile.to_csv(f\"{DIRS['profiles']}/spark_missingness_profile.csv\", index=False)\n\ndomain_source_summary = (\n    source_profile.groupby(\"domain_id\", as_index=False)\n    .agg(\n        source_files=(\"source_key\", \"count\"),\n        spark_rows=(\"spark_rows\", \"sum\"),\n        source_size_mb=(\"source_size_mb\", \"sum\"),\n    )\n)\ndomain_source_summary.to_csv(f\"{DIRS['profiles']}/domain_source_summary.csv\", index=False)\n\nprint(f\"Spark profiling completed in {(time.time() - t0):.1f} seconds\")\ndisplay(source_profile)\ndisplay(domain_source_summary)\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T06:49:54.331421Z","iopub.execute_input":"2026-06-28T06:49:54.331659Z","iopub.status.idle":"2026-06-28T06:51:02.499892Z","shell.execute_reply.started":"2026-06-28T06:49:54.331634Z","shell.execute_reply":"2026-06-28T06:51:02.499100Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 7. Dataset registry, label cards, and leakage-control notes\n# ============================================================\n\nbenchmark_registry = pd.DataFrame([\n    {\"dataset_id\": \"telco_ibm\", \"domain_id\": \"telecom\", \"source_keys\": \"telco_ibm\", \"label_semantics\": \"contract/account churn\", \"label_confidence\": \"high\", \"quality_tier\": \"A\"},\n    {\"dataset_id\": \"telco_orange\", \"domain_id\": \"telecom\", \"source_keys\": \"telco_orange_train; telco_orange_test\", \"label_semantics\": \"telecom churn flag\", \"label_confidence\": \"high\", \"quality_tier\": \"A\"},\n    {\"dataset_id\": \"cell2cell_train_holdout\", \"domain_id\": \"telecom\", \"source_keys\": \"cell2cell_train; cell2cell_holdout\", \"label_semantics\": \"telecom churn flag\", \"label_confidence\": \"high\", \"quality_tier\": \"A\"},\n    {\"dataset_id\": \"bank_original\", \"domain_id\": \"banking\", \"source_keys\": \"bank_original\", \"label_semantics\": \"bank customer attrition\", \"label_confidence\": \"high\", \"quality_tier\": \"A\"},\n    {\"dataset_id\": \"bank_creditcard\", \"domain_id\": \"banking\", \"source_keys\": \"bank_creditcard\", \"label_semantics\": \"credit-card customer attrition\", \"label_confidence\": \"high\", \"quality_tier\": \"A\"},\n    {\"dataset_id\": \"ecommerce_rees46\", \"domain_id\": \"ecommerce\", \"source_keys\": \"ecommerce_rees46\", \"label_semantics\": \"e-commerce future event/inactivity proxy\", \"label_confidence\": \"medium\", \"quality_tier\": \"B\"},\n    {\"dataset_id\": \"ecommerce_shriyash_large\", \"domain_id\": \"ecommerce\", \"source_keys\": \"ecommerce_shriyash_large\", \"label_semantics\": \"e-commerce churn flag\", \"label_confidence\": \"high\", \"quality_tier\": \"A\"},\n    {\"dataset_id\": \"ecommerce_dhairyajeet\", \"domain_id\": \"ecommerce\", \"source_keys\": \"ecommerce_dhairyajeet\", \"label_semantics\": \"e-commerce churn flag\", \"label_confidence\": \"high\", \"quality_tier\": \"A\"},\n    {\"dataset_id\": \"subscription_kkbox\", \"domain_id\": \"subscription\", \"source_keys\": \"subscription_kkbox_train; subscription_kkbox_members; subscription_kkbox_transactions\", \"label_semantics\": \"subscription non-renewal/churn\", \"label_confidence\": \"high\", \"quality_tier\": \"A\"},\n    {\"dataset_id\": \"subscription_generic\", \"domain_id\": \"subscription\", \"source_keys\": \"subscription_generic_train; subscription_generic_test\", \"label_semantics\": \"subscription churn flag\", \"label_confidence\": \"high\", \"quality_tier\": \"A\"},\n    {\"dataset_id\": \"insurance_auto\", \"domain_id\": \"insurance\", \"source_keys\": \"insurance_auto\", \"label_semantics\": \"auto-insurance churn flag\", \"label_confidence\": \"high\", \"quality_tier\": \"A\"},\n    {\"dataset_id\": \"insurance_k123vinod\", \"domain_id\": \"insurance\", \"source_keys\": \"insurance_k123_train\", \"label_semantics\": \"anonymized insurance churn label\", \"label_confidence\": \"medium\", \"quality_tier\": \"B\"},\n])\nbenchmark_registry.to_csv(f\"{DIRS['registry']}/benchmark_dataset_registry.csv\", index=False)\n\nleakage_control = pd.DataFrame([\n    {\"dataset_id\": \"cell2cell_train_holdout\", \"excluded_source_or_feature\": \"retention-intervention variables\", \"reason\": \"Retention-team contacts/offers can encode late-stage intervention information.\"},\n    {\"dataset_id\": \"insurance_auto\", \"excluded_source_or_feature\": \"termination.csv and post-event date fields\", \"reason\": \"Post-outcome termination information is not valid at prediction time.\"},\n    {\"dataset_id\": \"bank_creditcard\", \"excluded_source_or_feature\": \"Naive_Bayes_Classifier_* columns\", \"reason\": \"Provided model scores are not raw customer predictors.\"},\n    {\"dataset_id\": \"subscription_kkbox\", \"excluded_source_or_feature\": \"transactions after 2017-02-28\", \"reason\": \"Future transactions after the label cutoff can leak the churn month.\"},\n])\nleakage_control.to_csv(f\"{DIRS['registry']}/leakage_control_notes.csv\", index=False)\n\ndisplay(benchmark_registry)\ndisplay(leakage_control)\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T06:51:02.501642Z","iopub.execute_input":"2026-06-28T06:51:02.501914Z","iopub.status.idle":"2026-06-28T06:51:02.528192Z","shell.execute_reply.started":"2026-06-28T06:51:02.501884Z","shell.execute_reply":"2026-06-28T06:51:02.527111Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 8. Spark harmonization adapters for the 12 benchmark datasets\n# ============================================================\n\n\ndef harmonize_telco_ibm(path):\n    df = spark_read_csv(path)\n    service_cols = [\"PhoneService\", \"MultipleLines\", \"InternetService\", \"OnlineSecurity\", \"OnlineBackup\", \"DeviceProtection\", \"TechSupport\", \"StreamingTV\", \"StreamingMovies\"]\n    total_charges = num(df, \"TotalCharges\")\n    out = df.select(\n        F.col(\"customerID\").cast(\"string\").alias(\"customer_id\"),\n        num(df, \"tenure\").alias(\"tenure_months\"),\n        num(df, \"SeniorCitizen\").alias(\"age\"),\n        str_expr(df, \"gender\").alias(\"gender\"),\n        count_yes(df, service_cols).alias(\"service_count\"),\n        count_yes(df, service_cols).alias(\"engagement_score\"),\n        num(df, \"MonthlyCharges\").alias(\"payment_score\"),\n        total_charges.alias(\"usage_score\"),\n        yes_no(df, \"TechSupport\").alias(\"support_score\"),\n        num(df, \"tenure\").alias(\"transaction_count\"),\n        num(df, \"MonthlyCharges\").alias(\"avg_payment_amount\"),\n        total_charges.alias(\"total_payment_amount\"),\n        F.when(F.lower(F.trim(F.col(\"Contract\").cast(\"string\"))) != \"month-to-month\", 1.0).otherwise(0.0).alias(\"auto_renew_count\"),\n        F.lit(0.0).alias(\"cancel_count\"),\n        num(df, \"tenure\").alias(\"recent_activity_score\"),\n        yes_no(df, \"Churn\").alias(\"churn\"),\n    )\n    return finalize_spark_common(out, \"telco_ibm\", \"telecom\")\n\n\ndef harmonize_telco_orange(path_train, path_test):\n    df = spark_read_csv(path_train).unionByName(spark_read_csv(path_test), allowMissingColumns=True)\n    call_cols = [\"Total day calls\", \"Total eve calls\", \"Total night calls\", \"Total intl calls\"]\n    minute_cols = [\"Total day minutes\", \"Total eve minutes\", \"Total night minutes\", \"Total intl minutes\"]\n    charge_cols = [\"Total day charge\", \"Total eve charge\", \"Total night charge\", \"Total intl charge\"]\n    total_payment = sum_nums(df, charge_cols)\n    out = df.select(\n        F.concat(F.lit(\"telco_orange_\"), F.monotonically_increasing_id().cast(\"string\")).alias(\"customer_id\"),\n        num(df, \"Account length\").alias(\"tenure_months\"),\n        F.lit(None).cast(\"double\").alias(\"age\"),\n        F.lit(\"unknown\").alias(\"gender\"),\n        sum_nums(df, call_cols).alias(\"transaction_count\"),\n        sum_nums(df, minute_cols).alias(\"usage_score\"),\n        total_payment.alias(\"total_payment_amount\"),\n        safe_divide_expr(total_payment, num(df, \"Account length\")).alias(\"avg_payment_amount\"),\n        total_payment.alias(\"payment_score\"),\n        sum_nums(df, call_cols).alias(\"engagement_score\"),\n        num(df, \"Customer service calls\", default=0).alias(\"support_score\"),\n        (yes_no(df, \"International plan\") + yes_no(df, \"Voice mail plan\")).alias(\"service_count\"),\n        F.lit(0.0).alias(\"auto_renew_count\"),\n        num(df, \"Customer service calls\", default=0).alias(\"cancel_count\"),\n        num(df, \"Account length\").alias(\"recent_activity_score\"),\n        bool_or_one(df, \"Churn\").alias(\"churn\"),\n    )\n    return finalize_spark_common(out, \"telco_orange\", \"telecom\")\n\n\ndef harmonize_cell2cell(path_train, path_holdout):\n    # The public Cell2Cell holdout file contains no usable churn positives in this run\n    # and should not enter supervised modeling. It remains in the raw Spark profiling\n    # table as a source artifact, but the harmonized supervised benchmark uses the\n    # labeled training file only. This matches the manuscript-scale supervised sample.\n    df = spark_read_csv(path_train)\n    df = df.filter(F.col(\"Churn\").isNotNull())\n    out = df.select(\n        F.col(\"CustomerID\").cast(\"string\").alias(\"customer_id\"),\n        num(df, \"MonthsInService\").alias(\"tenure_months\"),\n        num(df, \"AgeHH1\").alias(\"age\"),\n        F.lit(\"unknown\").alias(\"gender\"),\n        (num(df, \"MonthlyMinutes\", 0) + num(df, \"ReceivedCalls\", 0) + num(df, \"OutboundCalls\", 0)).alias(\"engagement_score\"),\n        num(df, \"MonthlyRevenue\", 0).alias(\"payment_score\"),\n        num(df, \"MonthlyMinutes\", 0).alias(\"usage_score\"),\n        num(df, \"CustomerCareCalls\", 0).alias(\"support_score\"),\n        (num(df, \"PeakCallsInOut\", 0) + num(df, \"OffPeakCallsInOut\", 0) + num(df, \"InboundCalls\", 0) + num(df, \"OutboundCalls\", 0)).alias(\"transaction_count\"),\n        num(df, \"MonthlyRevenue\", 0).alias(\"avg_payment_amount\"),\n        (num(df, \"MonthlyRevenue\", 0) * num(df, \"MonthsInService\", 0)).alias(\"total_payment_amount\"),\n        (num(df, \"ActiveSubs\", 0) + num(df, \"Handsets\", 0)).alias(\"service_count\"),\n        F.lit(0.0).alias(\"auto_renew_count\"),\n        num(df, \"DroppedBlockedCalls\", 0).alias(\"cancel_count\"),\n        num(df, \"CurrentEquipmentDays\", 0).alias(\"recent_activity_score\"),\n        yes_no(df, \"Churn\").alias(\"churn\"),\n    )\n    return finalize_spark_common(out, \"cell2cell_train_holdout\", \"telecom\")\n\n\ndef harmonize_bank_original(path):\n    df = spark_read_csv(path)\n    out = df.select(\n        F.col(\"customer_id\").cast(\"string\").alias(\"customer_id\"),\n        num(df, \"tenure\").alias(\"tenure_months\"),\n        num(df, \"age\").alias(\"age\"),\n        str_expr(df, \"gender\").alias(\"gender\"),\n        num(df, \"active_member\", 0).alias(\"engagement_score\"),\n        num(df, \"balance\", 0).alias(\"payment_score\"),\n        num(df, \"estimated_salary\", 0).alias(\"usage_score\"),\n        F.lit(0.0).alias(\"support_score\"),\n        num(df, \"products_number\", 0).alias(\"transaction_count\"),\n        num(df, \"balance\", 0).alias(\"avg_payment_amount\"),\n        num(df, \"balance\", 0).alias(\"total_payment_amount\"),\n        num(df, \"products_number\", 0).alias(\"service_count\"),\n        num(df, \"credit_card\", 0).alias(\"auto_renew_count\"),\n        F.lit(0.0).alias(\"cancel_count\"),\n        num(df, \"active_member\", 0).alias(\"recent_activity_score\"),\n        num(df, \"churn\").alias(\"churn\"),\n    )\n    return finalize_spark_common(out, \"bank_original\", \"banking\")\n\n\ndef harmonize_bank_creditcard(path):\n    df = spark_read_csv(path)\n    avg_payment = safe_divide_expr(num(df, \"Total_Trans_Amt\", 0), num(df, \"Total_Trans_Ct\"))\n    out = df.select(\n        F.col(\"CLIENTNUM\").cast(\"string\").alias(\"customer_id\"),\n        num(df, \"Months_on_book\").alias(\"tenure_months\"),\n        num(df, \"Customer_Age\").alias(\"age\"),\n        str_expr(df, \"Gender\").alias(\"gender\"),\n        num(df, \"Total_Trans_Ct\", 0).alias(\"engagement_score\"),\n        num(df, \"Credit_Limit\", 0).alias(\"payment_score\"),\n        num(df, \"Total_Trans_Amt\", 0).alias(\"usage_score\"),\n        num(df, \"Contacts_Count_12_mon\", 0).alias(\"support_score\"),\n        num(df, \"Total_Trans_Ct\", 0).alias(\"transaction_count\"),\n        avg_payment.alias(\"avg_payment_amount\"),\n        num(df, \"Total_Trans_Amt\", 0).alias(\"total_payment_amount\"),\n        num(df, \"Total_Relationship_Count\", 0).alias(\"service_count\"),\n        F.lit(0.0).alias(\"auto_renew_count\"),\n        num(df, \"Contacts_Count_12_mon\", 0).alias(\"cancel_count\"),\n        num(df, \"Months_Inactive_12_mon\", 0).alias(\"recent_activity_score\"),\n        contains_attrited(df, \"Attrition_Flag\").alias(\"churn\"),\n    )\n    return finalize_spark_common(out, \"bank_creditcard\", \"banking\")\n\n\ndef harmonize_rees46(path):\n    df = spark_read_csv(path)\n    out = df.select(\n        F.col(\"user_id\").cast(\"string\").alias(\"customer_id\"),\n        num(df, \"user_rec\").alias(\"tenure_months\"),\n        F.lit(None).cast(\"double\").alias(\"age\"),\n        F.lit(\"unknown\").alias(\"gender\"),\n        (num(df, \"click_count_sum\", 0) + num(df, \"view_count_sum\", 0) + num(df, \"cart_count_sum\", 0)).alias(\"engagement_score\"),\n        num(df, \"purchase_revenue_sum\", 0).alias(\"payment_score\"),\n        num(df, \"session_number_sum\", 0).alias(\"usage_score\"),\n        F.lit(0.0).alias(\"support_score\"),\n        num(df, \"purchase_count_sum\", 0).alias(\"transaction_count\"),\n        num(df, \"purchase_revenue_mean\", 0).alias(\"avg_payment_amount\"),\n        num(df, \"purchase_revenue_sum\", 0).alias(\"total_payment_amount\"),\n        num(df, \"session_number_mean\", 0).alias(\"service_count\"),\n        F.lit(0.0).alias(\"auto_renew_count\"),\n        F.lit(0.0).alias(\"cancel_count\"),\n        num(df, \"session_recency_mean\", 0).alias(\"recent_activity_score\"),\n        num(df, \"target_event\").alias(\"churn\"),\n    )\n    return finalize_spark_common(out, \"ecommerce_rees46\", \"ecommerce\")\n\n\ndef harmonize_ecommerce_shriyash(path):\n    df = spark_read_csv(path)\n    purchase_date = F.to_timestamp(str_expr(df, \"Purchase Date\"))\n    try:\n        max_date = df.select(F.max(purchase_date).alias(\"max_purchase_date\")).collect()[0][\"max_purchase_date\"]\n    except Exception:\n        max_date = None\n    recent_expr = F.datediff(F.lit(max_date), purchase_date) if max_date is not None else F.lit(None).cast(\"double\")\n    out = df.select(\n        F.col(\"Customer ID\").cast(\"string\").alias(\"customer_id\"),\n        F.lit(None).cast(\"double\").alias(\"tenure_months\"),\n        num(df, \"Customer Age\").alias(\"age\"),\n        str_expr(df, \"Gender\").alias(\"gender\"),\n        num(df, \"Quantity\", 0).alias(\"engagement_score\"),\n        num(df, \"Product Price\", 0).alias(\"payment_score\"),\n        num(df, \"Total Purchase Amount\", 0).alias(\"usage_score\"),\n        num(df, \"Returns\", 0).alias(\"support_score\"),\n        num(df, \"Quantity\", 0).alias(\"transaction_count\"),\n        num(df, \"Product Price\", 0).alias(\"avg_payment_amount\"),\n        num(df, \"Total Purchase Amount\", 0).alias(\"total_payment_amount\"),\n        F.lit(1.0).alias(\"service_count\"),\n        F.lit(0.0).alias(\"auto_renew_count\"),\n        num(df, \"Returns\", 0).alias(\"cancel_count\"),\n        recent_expr.alias(\"recent_activity_score\"),\n        num(df, \"Churn\").alias(\"churn\"),\n    )\n    return finalize_spark_common(out, \"ecommerce_shriyash_large\", \"ecommerce\")\n\n\ndef harmonize_ecommerce_dhairyajeet(path):\n    df = spark_read_csv(path)\n    out = df.select(\n        F.concat(F.lit(\"ecommerce_dhairyajeet_\"), F.monotonically_increasing_id().cast(\"string\")).alias(\"customer_id\"),\n        (num(df, \"Membership_Years\") * F.lit(12.0)).alias(\"tenure_months\"),\n        num(df, \"Age\").alias(\"age\"),\n        str_expr(df, \"Gender\").alias(\"gender\"),\n        (num(df, \"Login_Frequency\", 0) + num(df, \"Email_Open_Rate\", 0) + num(df, \"Social_Media_Engagement_Score\", 0)).alias(\"engagement_score\"),\n        num(df, \"Average_Order_Value\", 0).alias(\"payment_score\"),\n        (num(df, \"Session_Duration_Avg\", 0) + num(df, \"Mobile_App_Usage\", 0)).alias(\"usage_score\"),\n        num(df, \"Customer_Service_Calls\", 0).alias(\"support_score\"),\n        num(df, \"Total_Purchases\", 0).alias(\"transaction_count\"),\n        num(df, \"Average_Order_Value\", 0).alias(\"avg_payment_amount\"),\n        num(df, \"Lifetime_Value\", 0).alias(\"total_payment_amount\"),\n        num(df, \"Payment_Method_Diversity\", 0).alias(\"service_count\"),\n        F.lit(0.0).alias(\"auto_renew_count\"),\n        (num(df, \"Returns_Rate\", 0) + num(df, \"Cart_Abandonment_Rate\", 0)).alias(\"cancel_count\"),\n        num(df, \"Days_Since_Last_Purchase\").alias(\"recent_activity_score\"),\n        num(df, \"Churned\").alias(\"churn\"),\n    )\n    return finalize_spark_common(out, \"ecommerce_dhairyajeet\", \"ecommerce\")\n\n\ndef harmonize_kkbox(train_path, members_path, transactions_path=None, cutoff_yyyymmdd=20170228):\n    \"\"\"Resource-safe KKBox harmonization for Kaggle Spark MLlib runs.\n\n    The raw KKBox transaction table is still included in the Spark Big Data profiling\n    layer. For the supervised ML table, this low-memory implementation joins the\n    churn labels with member metadata and avoids a full transaction-level groupBy\n    over 1.4M records, which caused Spark executor OOM in Kaggle. Transaction-based\n    enrichment can be enabled later on a larger Spark cluster.\n    \"\"\"\n    train = spark_read_csv(train_path).select(\"msno\", \"is_churn\").dropDuplicates([\"msno\"])\n    members = (\n        spark_read_csv(members_path)\n        .select(\"msno\", \"city\", \"bd\", \"gender\", \"registered_via\", \"registration_init_time\")\n        .dropDuplicates([\"msno\"])\n    )\n\n    df = train.join(members, on=\"msno\", how=\"left\")\n    registration_date = F.to_date(F.col(\"registration_init_time\").cast(\"string\"), \"yyyyMMdd\")\n    cutoff_date = F.to_date(F.lit(str(cutoff_yyyymmdd)), \"yyyyMMdd\")\n    bd = num(df, \"bd\")\n    age_clean = F.when((bd >= 5.0) & (bd <= 100.0), bd).otherwise(F.lit(None).cast(\"double\"))\n\n    tenure_expr = F.datediff(cutoff_date, registration_date) / F.lit(30.44)\n    registered_via_num = num(df, \"registered_via\", 0)\n    city_num = num(df, \"city\", 0)\n\n    out = df.select(\n        F.col(\"msno\").cast(\"string\").alias(\"customer_id\"),\n        tenure_expr.alias(\"tenure_months\"),\n        age_clean.alias(\"age\"),\n        str_expr(df, \"gender\").alias(\"gender\"),\n\n        # Metadata-derived safe proxies. These are not as rich as transaction aggregates,\n        # but they allow a complete Spark MLlib cross-domain run on Kaggle memory.\n        registered_via_num.alias(\"transaction_count\"),\n        registered_via_num.alias(\"engagement_score\"),\n        F.lit(0.0).alias(\"payment_score\"),\n        tenure_expr.alias(\"usage_score\"),\n        F.lit(0.0).alias(\"support_score\"),\n        F.lit(0.0).alias(\"avg_payment_amount\"),\n        F.lit(0.0).alias(\"total_payment_amount\"),\n        city_num.alias(\"service_count\"),\n        F.lit(0.0).alias(\"auto_renew_count\"),\n        F.lit(0.0).alias(\"cancel_count\"),\n        tenure_expr.alias(\"recent_activity_score\"),\n        num(df, \"is_churn\").alias(\"churn\"),\n    )\n    return finalize_spark_common(out, \"subscription_kkbox\", \"subscription\")\n\n\ndef harmonize_subscription_generic(train_path, test_path):\n    df = spark_read_csv(train_path).unionByName(spark_read_csv(test_path), allowMissingColumns=True)\n    out = df.select(\n        F.col(\"CustomerID\").cast(\"string\").alias(\"customer_id\"),\n        num(df, \"Tenure\").alias(\"tenure_months\"),\n        num(df, \"Age\").alias(\"age\"),\n        str_expr(df, \"Gender\").alias(\"gender\"),\n        num(df, \"Usage Frequency\", 0).alias(\"engagement_score\"),\n        num(df, \"Payment Delay\", 0).alias(\"payment_score\"),\n        num(df, \"Usage Frequency\", 0).alias(\"usage_score\"),\n        num(df, \"Support Calls\", 0).alias(\"support_score\"),\n        num(df, \"Usage Frequency\", 0).alias(\"transaction_count\"),\n        safe_divide_expr(num(df, \"Total Spend\", 0), num(df, \"Tenure\", 0) + F.lit(1.0)).alias(\"avg_payment_amount\"),\n        num(df, \"Total Spend\", 0).alias(\"total_payment_amount\"),\n        F.when(F.col(\"Subscription Type\") == \"Basic\", 1.0).when(F.col(\"Subscription Type\") == \"Standard\", 2.0).when(F.col(\"Subscription Type\") == \"Premium\", 3.0).otherwise(1.0).alias(\"service_count\"),\n        F.when(F.lower(F.trim(F.col(\"Contract Length\").cast(\"string\"))) != \"monthly\", 1.0).otherwise(0.0).alias(\"auto_renew_count\"),\n        num(df, \"Payment Delay\", 0).alias(\"cancel_count\"),\n        num(df, \"Last Interaction\").alias(\"recent_activity_score\"),\n        num(df, \"Churn\").alias(\"churn\"),\n    )\n    return finalize_spark_common(out, \"subscription_generic\", \"subscription\")\n\n\ndef harmonize_insurance_auto(path):\n    df = spark_read_csv(path)\n    total_payment = num(df, \"curr_ann_amt\", 0) * (num(df, \"days_tenure\", 0) / F.lit(365.25))\n    out = df.select(\n        F.col(\"individual_id\").cast(\"string\").alias(\"customer_id\"),\n        (num(df, \"days_tenure\") / F.lit(30.44)).alias(\"tenure_months\"),\n        num(df, \"age_in_years\").alias(\"age\"),\n        F.lit(\"unknown\").alias(\"gender\"),\n        (num(df, \"home_owner\", 0) + num(df, \"college_degree\", 0) + num(df, \"good_credit\", 0)).alias(\"engagement_score\"),\n        num(df, \"curr_ann_amt\", 0).alias(\"payment_score\"),\n        num(df, \"income\", 0).alias(\"usage_score\"),\n        F.lit(0.0).alias(\"support_score\"),\n        num(df, \"length_of_residence\", 0).alias(\"transaction_count\"),\n        num(df, \"curr_ann_amt\", 0).alias(\"avg_payment_amount\"),\n        total_payment.alias(\"total_payment_amount\"),\n        (num(df, \"has_children\", 0) + num(df, \"home_owner\", 0) + num(df, \"college_degree\", 0) + num(df, \"good_credit\", 0)).alias(\"service_count\"),\n        F.lit(0.0).alias(\"auto_renew_count\"),\n        F.lit(0.0).alias(\"cancel_count\"),\n        num(df, \"days_tenure\", 0).alias(\"recent_activity_score\"),\n        num(df, \"Churn\").alias(\"churn\"),\n    )\n    return finalize_spark_common(out, \"insurance_auto\", \"insurance\")\n\n\ndef harmonize_insurance_k123(path):\n    df = spark_read_csv(path)\n    out = df.select(\n        F.concat(F.lit(\"insurance_k123vinod_\"), F.monotonically_increasing_id().cast(\"string\")).alias(\"customer_id\"),\n        num(df, \"feature_0\").alias(\"tenure_months\"),\n        num(df, \"feature_2\").alias(\"age\"),\n        F.lit(\"unknown\").alias(\"gender\"),\n        (num(df, \"feature_3\", 0) + num(df, \"feature_7\", 0)).alias(\"engagement_score\"),\n        num(df, \"feature_1\", 0).alias(\"payment_score\"),\n        num(df, \"feature_5\", 0).alias(\"usage_score\"),\n        num(df, \"feature_6\", 0).alias(\"support_score\"),\n        num(df, \"feature_14\", 0).alias(\"transaction_count\"),\n        num(df, \"feature_1\", 0).alias(\"avg_payment_amount\"),\n        num(df, \"feature_4\", 0).alias(\"total_payment_amount\"),\n        num(df, \"feature_15\", 0).alias(\"service_count\"),\n        num(df, \"feature_10\", 0).alias(\"auto_renew_count\"),\n        num(df, \"feature_11\", 0).alias(\"cancel_count\"),\n        num(df, \"feature_3\", 0).alias(\"recent_activity_score\"),\n        num(df, \"labels\").alias(\"churn\"),\n    )\n    return finalize_spark_common(out, \"insurance_k123vinod\", \"insurance\")\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T06:51:02.529579Z","iopub.execute_input":"2026-06-28T06:51:02.530717Z","iopub.status.idle":"2026-06-28T06:51:02.590234Z","shell.execute_reply.started":"2026-06-28T06:51:02.530659Z","shell.execute_reply":"2026-06-28T06:51:02.589434Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 9. Build resource-safe capped harmonized Spark benchmark tables\n# ============================================================\n\n# In the previous run, Spark failed while materializing a full uncapped union of all\n# harmonized datasets. The failure was not a data error; it was a Kaggle memory limit.\n# This cell writes a capped supervised table per dataset, then reloads only the capped\n# modeling benchmark. The Big Data profiling layer still reports the full raw-source\n# scale from all source files.\n\nCAPPED_DATA_DIR = Path(DIRS[\"harmonized\"]) / \"capped_supervised_tables\"\nif CAPPED_DATA_DIR.exists():\n    shutil.rmtree(CAPPED_DATA_DIR)\nCAPPED_DATA_DIR.mkdir(parents=True, exist_ok=True)\n\n\ndef allocate_class_targets(class_count_rows, max_n):\n    \"\"\"Return exact class-level target counts that sum to max_n when possible.\"\"\"\n    rows = [(float(r[\"churn\"]), int(r[\"class_count\"])) for r in class_count_rows]\n    total = sum(c for _, c in rows)\n    if total <= max_n:\n        return {label: count for label, count in rows}\n    raw = [(label, count, (count * max_n / total)) for label, count in rows]\n    base = {label: int(math.floor(x)) for label, count, x in raw}\n    # Keep any present class represented when possible.\n    for label, count, x in raw:\n        if count > 0 and base[label] == 0 and max_n >= len(raw):\n            base[label] = 1\n    remaining = max_n - sum(base.values())\n    ranked = sorted(raw, key=lambda z: (z[2] - math.floor(z[2])), reverse=True)\n    j = 0\n    while remaining > 0 and ranked:\n        label, count, x = ranked[j % len(ranked)]\n        if base[label] < count:\n            base[label] += 1\n            remaining -= 1\n        j += 1\n        if j > 10000:\n            break\n    # If representation adjustment overshot, remove from largest classes.\n    while remaining < 0:\n        label = max(base, key=lambda k: base[k])\n        base[label] -= 1\n        remaining += 1\n    return base\n\n\ndef cap_one_dataset(df, dataset_id, max_n):\n    \"\"\"Cap one dataset exactly where possible using a memory-safe pre-sampling stage.\"\"\"\n    base = (\n        df.select(*COMMON_COLUMNS)\n        .withColumn(\"churn\", clean_numeric_expr(F.col(\"churn\")).cast(\"double\"))\n        .filter(F.col(\"churn\").isin(0.0, 1.0))\n        .repartition(8)\n    )\n\n    class_rows = base.groupBy(\"churn\").agg(F.count(\"*\").alias(\"class_count\")).collect()\n    total_n = sum(int(r[\"class_count\"]) for r in class_rows)\n    if total_n == 0:\n        raise ValueError(f\"No valid supervised rows for {dataset_id}\")\n\n    targets = allocate_class_targets(class_rows, min(int(max_n), int(total_n)))\n    target_pdf = pd.DataFrame([\n        {\"churn\": float(label), \"target_class_n\": int(tn), \"class_count\": int(dict((float(r[\"churn\"]), int(r[\"class_count\"])) for r in class_rows)[float(label)])}\n        for label, tn in targets.items()\n    ])\n\n    target_sdf = spark.createDataFrame(target_pdf)\n    joined = base.join(F.broadcast(target_sdf), on=\"churn\", how=\"inner\")\n\n    # Pre-sample to avoid sorting millions of rows. The buffer is deliberately large\n    # to preserve exact class targets with high probability.\n    sampled = (\n        joined\n        .withColumn(\"sample_fraction\", F.least(F.lit(1.0), (F.col(\"target_class_n\") / F.col(\"class_count\")) * F.lit(2.5) + F.lit(0.05)))\n        .withColumn(\"u_presample\", F.rand(RANDOM_SEED + 17))\n        .filter(F.col(\"u_presample\") <= F.col(\"sample_fraction\"))\n    )\n\n    w_class = Window.partitionBy(\"churn\").orderBy(F.rand(RANDOM_SEED + 31))\n    capped = (\n        sampled\n        .withColumn(\"rn_class\", F.row_number().over(w_class))\n        .filter(F.col(\"rn_class\") <= F.col(\"target_class_n\"))\n        .drop(\"target_class_n\", \"class_count\", \"sample_fraction\", \"u_presample\", \"rn_class\")\n    )\n\n    # Final exact dataset cap in case class arithmetic produced a boundary issue.\n    w_dataset = Window.partitionBy(\"dataset_id\").orderBy(F.rand(RANDOM_SEED + 43))\n    capped = (\n        capped\n        .withColumn(\"rn_dataset\", F.row_number().over(w_dataset))\n        .filter(F.col(\"rn_dataset\") <= F.lit(int(max_n)))\n        .drop(\"rn_dataset\")\n    )\n    return capped\n\n\nDATASET_BUILDERS = [\n    (\"telco_ibm\", lambda: harmonize_telco_ibm(PATHS[\"telco_ibm\"])),\n    (\"telco_orange\", lambda: harmonize_telco_orange(PATHS[\"telco_orange_train\"], PATHS[\"telco_orange_test\"])),\n    (\"cell2cell_train_holdout\", lambda: harmonize_cell2cell(PATHS[\"cell2cell_train\"], PATHS[\"cell2cell_holdout\"])),\n    (\"bank_original\", lambda: harmonize_bank_original(PATHS[\"bank_original\"])),\n    (\"bank_creditcard\", lambda: harmonize_bank_creditcard(PATHS[\"bank_creditcard\"])),\n    (\"ecommerce_rees46\", lambda: harmonize_rees46(PATHS[\"rees46\"])),\n    (\"ecommerce_shriyash_large\", lambda: harmonize_ecommerce_shriyash(PATHS[\"ecommerce_shriyash_large\"])),\n    (\"ecommerce_dhairyajeet\", lambda: harmonize_ecommerce_dhairyajeet(PATHS[\"ecommerce_dhairyajeet\"])),\n    (\"subscription_kkbox\", lambda: harmonize_kkbox(PATHS[\"kkbox_train\"], PATHS[\"kkbox_members\"], PATHS.get(\"kkbox_transactions\"))),\n    (\"subscription_generic\", lambda: harmonize_subscription_generic(PATHS[\"subscription_generic_train\"], PATHS[\"subscription_generic_test\"])),\n    (\"insurance_auto\", lambda: harmonize_insurance_auto(PATHS[\"insurance_auto\"])),\n    (\"insurance_k123vinod\", lambda: harmonize_insurance_k123(PATHS[\"insurance_k123_train\"])),\n]\n\ncapped_paths = []\nmodeling_build_rows = []\n\nfor i, (dataset_id, builder) in enumerate(DATASET_BUILDERS, start=1):\n    print(f\"[{i}/{len(DATASET_BUILDERS)}] Harmonizing and capping: {dataset_id}\")\n    t0 = time.time()\n    raw_harmonized = builder()\n    capped = cap_one_dataset(raw_harmonized, dataset_id, MAX_ROWS_PER_DATASET_FOR_MODELING)\n    out_path = CAPPED_DATA_DIR / dataset_id\n    capped.coalesce(2).write.mode(\"overwrite\").parquet(str(out_path))\n    capped_read = spark.read.parquet(str(out_path))\n    row = capped_read.agg(\n        F.count(\"*\").alias(\"modeling_rows\"),\n        F.avg(\"churn\").alias(\"positive_rate\")\n    ).collect()[0].asDict()\n    domain = capped_read.select(\"domain_id\").first()[\"domain_id\"]\n    modeling_build_rows.append({\n        \"dataset_id\": dataset_id,\n        \"domain_id\": domain,\n        \"modeling_rows\": int(row[\"modeling_rows\"]),\n        \"positive_rate\": float(row[\"positive_rate\"]),\n        \"runtime_seconds\": time.time() - t0,\n        \"path\": str(out_path),\n    })\n    capped_paths.append(str(out_path))\n    spark.catalog.clearCache()\n    gc.collect()\n\nmodeling_build_profile = pd.DataFrame(modeling_build_rows)\nmodeling_build_profile.to_csv(f\"{DIRS['registry']}/modeling_build_profile.csv\", index=False)\ndisplay(modeling_build_profile)\n\nprint(\"Capped supervised tables written to:\", CAPPED_DATA_DIR)\nprint(\"Total capped modeling rows:\", int(modeling_build_profile[\"modeling_rows\"].sum()))\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T06:51:02.591080Z","iopub.execute_input":"2026-06-28T06:51:02.591276Z","iopub.status.idle":"2026-06-28T06:55:24.028794Z","shell.execute_reply.started":"2026-06-28T06:51:02.591255Z","shell.execute_reply":"2026-06-28T06:55:24.027913Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 10. Load capped modeling table and create deterministic splits\n# ============================================================\n\nmodeling_data = spark.read.parquet(*capped_paths)\nmodeling_data = (\n    modeling_data\n    .select(*COMMON_COLUMNS)\n    .withColumn(\"label\", F.col(\"churn\").cast(\"double\"))\n    .filter(F.col(\"label\").isin(0.0, 1.0))\n    .withColumn(\"split_rand\", F.rand(RANDOM_SEED))\n    .withColumn(\n        \"split\",\n        F.when(F.col(\"split_rand\") < F.lit(TRAIN_FRACTION), F.lit(\"train\"))\n         .when(F.col(\"split_rand\") < F.lit(TRAIN_FRACTION + VALIDATION_FRACTION), F.lit(\"validation\"))\n         .otherwise(F.lit(\"test\"))\n    )\n    .drop(\"split_rand\")\n    .repartition(16, \"dataset_id\")\n    .persist(StorageLevel.MEMORY_AND_DISK)\n)\n\nmodeling_count = modeling_data.count()\nprint(\"Modeling rows:\", modeling_count)\n\nmodeling_summary = (\n    modeling_data.groupBy(\"dataset_id\", \"domain_id\")\n    .agg(\n        F.count(\"*\").alias(\"modeling_rows\"),\n        F.avg(\"label\").alias(\"positive_rate\"),\n        F.sum(F.when(F.col(\"split\") == \"train\", 1).otherwise(0)).alias(\"train_rows\"),\n        F.sum(F.when(F.col(\"split\") == \"validation\", 1).otherwise(0)).alias(\"validation_rows\"),\n        F.sum(F.when(F.col(\"split\") == \"test\", 1).otherwise(0)).alias(\"test_rows\"),\n    )\n    .orderBy(\"domain_id\", \"dataset_id\")\n)\nmodeling_summary_pd = modeling_summary.toPandas()\nmodeling_summary_pd.to_csv(f\"{DIRS['registry']}/modeling_summary.csv\", index=False)\ndisplay(modeling_summary_pd)\n\ndomain_modeling_summary = (\n    modeling_data.groupBy(\"domain_id\")\n    .agg(\n        F.count(\"*\").alias(\"modeling_rows\"),\n        F.avg(\"label\").alias(\"positive_rate\"),\n        F.countDistinct(\"dataset_id\").alias(\"datasets\"),\n    )\n    .orderBy(\"domain_id\")\n)\ndomain_modeling_summary_pd = domain_modeling_summary.toPandas()\ndomain_modeling_summary_pd.to_csv(f\"{DIRS['registry']}/domain_modeling_summary.csv\", index=False)\ndisplay(domain_modeling_summary_pd)\n\n# Export the compact modeling table for reproducibility. This is intentionally capped\n# and small enough to reload in later cells without rebuilding raw harmonization.\nFINAL_MODELING_PARQUET = str(Path(DIRS[\"harmonized\"]) / \"supervised_modeling_table_capped.parquet\")\nmodeling_data.write.mode(\"overwrite\").parquet(FINAL_MODELING_PARQUET)\nprint(\"Saved capped modeling table:\", FINAL_MODELING_PARQUET)\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T06:55:24.031026Z","iopub.execute_input":"2026-06-28T06:55:24.031272Z","iopub.status.idle":"2026-06-28T06:55:29.573984Z","shell.execute_reply.started":"2026-06-28T06:55:24.031250Z","shell.execute_reply":"2026-06-28T06:55:29.573190Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 11. Spark MLlib model, weighting, and metric helpers\n# ============================================================\n\n\n\ndef prepare_for_modeling(df):\n    \"\"\"Defensive feature cleaning before Spark ML fitting.\n\n    This version is Spark-4/ANSI-safe: blank strings and malformed numeric values\n    are converted to NULL before imputation, rather than being directly cast to double.\n    \"\"\"\n    out = df.fillna({\"gender\": \"unknown\"})\n    for c in NUMERIC_FEATURES + [\"label\", \"churn\"]:\n        if c not in out.columns:\n            out = out.withColumn(c, F.lit(0.0))\n        else:\n            cleaned = clean_numeric_expr(F.col(c))\n            out = out.withColumn(c, F.coalesce(cleaned, F.lit(0.0)).cast(\"double\"))\n    out = out.withColumn(\"gender\", F.coalesce(F.col(\"gender\").cast(\"string\"), F.lit(\"unknown\")))\n    return out\n\n\ndef add_sample_weights(train_df, weighting_mode=\"row_weighted\"):\n    \"\"\"Add sample weights using precomputed aggregation tables.\n\n    This avoids unnecessary cached joins staying in memory during repeated\n    leave-one-dataset/domain experiments.\n    \"\"\"\n    total_n = train_df.count()\n    label_counts = train_df.groupBy(\"label\").agg(F.count(\"*\").alias(\"label_n\"))\n    label_weights = label_counts.withColumn(\n        \"class_weight\",\n        F.lit(float(total_n)) / (F.lit(2.0) * F.col(\"label_n\"))\n    )\n    out = train_df.join(F.broadcast(label_weights.select(\"label\", \"class_weight\")), on=\"label\", how=\"left\")\n\n    if weighting_mode == \"dataset_balanced\":\n        dataset_counts_local = train_df.groupBy(\"dataset_id\").agg(F.count(\"*\").alias(\"dataset_n\"))\n        n_datasets = dataset_counts_local.count()\n        dataset_weights = dataset_counts_local.withColumn(\n            \"dataset_weight\",\n            F.lit(float(total_n)) / (F.lit(float(max(n_datasets, 1))) * F.col(\"dataset_n\"))\n        )\n        out = out.join(F.broadcast(dataset_weights.select(\"dataset_id\", \"dataset_weight\")), on=\"dataset_id\", how=\"left\")\n        out = out.withColumn(\"sample_weight\", F.col(\"class_weight\") * F.col(\"dataset_weight\"))\n    else:\n        out = out.withColumn(\"sample_weight\", F.col(\"class_weight\"))\n\n    return out.drop(\"class_weight\", \"dataset_weight\")\n\n\ndef make_preprocess_stages(scale_for_linear=False):\n    imputed_cols = [f\"{c}_imputed\" for c in NUMERIC_FEATURES]\n    imputer = Imputer(inputCols=NUMERIC_FEATURES, outputCols=imputed_cols).setStrategy(\"median\")\n    indexer = StringIndexer(inputCol=\"gender\", outputCol=\"gender_index\", handleInvalid=\"keep\")\n    encoder = OneHotEncoder(inputCols=[\"gender_index\"], outputCols=[\"gender_ohe\"], dropLast=False)\n    try:\n        encoder = encoder.setHandleInvalid(\"keep\")\n    except Exception:\n        pass\n    assembler_input = imputed_cols + [\"gender_ohe\"]\n    assembler = VectorAssembler(inputCols=assembler_input, outputCol=\"features_raw\", handleInvalid=\"keep\")\n    if scale_for_linear:\n        scaler = StandardScaler(inputCol=\"features_raw\", outputCol=\"features\", withMean=False, withStd=True)\n        return [imputer, indexer, encoder, assembler, scaler]\n    else:\n        assembler_final = VectorAssembler(inputCols=assembler_input, outputCol=\"features\", handleInvalid=\"keep\")\n        return [imputer, indexer, encoder, assembler_final]\n\n\ndef make_model_pipeline(model_name):\n    if model_name == \"spark_logistic_regression_weighted\":\n        lr = LogisticRegression(\n            labelCol=\"label\",\n            featuresCol=\"features\",\n            weightCol=\"sample_weight\",\n            probabilityCol=\"probability\",\n            rawPredictionCol=\"rawPrediction\",\n            predictionCol=\"prediction\",\n            maxIter=60,\n            regParam=0.01,\n            elasticNetParam=0.0,\n            standardization=True,\n        )\n        return Pipeline(stages=make_preprocess_stages(scale_for_linear=True) + [lr])\n\n    if model_name == \"spark_random_forest_weighted\":\n        rf = RandomForestClassifier(\n            labelCol=\"label\",\n            featuresCol=\"features\",\n            weightCol=\"sample_weight\",\n            probabilityCol=\"probability\",\n            rawPredictionCol=\"rawPrediction\",\n            predictionCol=\"prediction\",\n            numTrees=RF_NUM_TREES,\n            maxDepth=RF_MAX_DEPTH,\n            subsamplingRate=0.80,\n            featureSubsetStrategy=\"sqrt\",\n            seed=RANDOM_SEED,\n        )\n        return Pipeline(stages=make_preprocess_stages(scale_for_linear=False) + [rf])\n\n    if model_name == \"spark_gradient_boosted_trees_weighted\":\n        gbt = GBTClassifier(\n            labelCol=\"label\",\n            featuresCol=\"features\",\n            weightCol=\"sample_weight\",\n            probabilityCol=\"probability\",\n            rawPredictionCol=\"rawPrediction\",\n            predictionCol=\"prediction\",\n            maxIter=GBT_MAX_ITER,\n            maxDepth=5,\n            seed=RANDOM_SEED,\n        )\n        return Pipeline(stages=make_preprocess_stages(scale_for_linear=False) + [gbt])\n\n    raise ValueError(f\"Unknown model_name: {model_name}\")\n\n\ndef attach_score(pred_df):\n    return pred_df.withColumn(\"score\", vector_to_array(F.col(\"probability\"))[1].cast(\"double\"))\n\n\ndef safe_auc(pred_df, metric_name):\n    evaluator = BinaryClassificationEvaluator(\n        labelCol=\"label\",\n        rawPredictionCol=\"rawPrediction\",\n        metricName=metric_name,\n    )\n    try:\n        return float(evaluator.evaluate(pred_df))\n    except Exception:\n        return float(\"nan\")\n\n\ndef threshold_metrics(pred_df, threshold=0.5):\n    tmp = pred_df.withColumn(\"pred_label\", F.when(F.col(\"score\") >= threshold, 1.0).otherwise(0.0))\n    row = tmp.agg(\n        F.sum(F.when((F.col(\"label\") == 1.0) & (F.col(\"pred_label\") == 1.0), 1).otherwise(0)).alias(\"tp\"),\n        F.sum(F.when((F.col(\"label\") == 0.0) & (F.col(\"pred_label\") == 0.0), 1).otherwise(0)).alias(\"tn\"),\n        F.sum(F.when((F.col(\"label\") == 0.0) & (F.col(\"pred_label\") == 1.0), 1).otherwise(0)).alias(\"fp\"),\n        F.sum(F.when((F.col(\"label\") == 1.0) & (F.col(\"pred_label\") == 0.0), 1).otherwise(0)).alias(\"fn\"),\n    ).collect()[0].asDict()\n    tp, tn, fp, fn = [float(row[k]) for k in [\"tp\", \"tn\", \"fp\", \"fn\"]]\n    precision = tp / max(tp + fp, 1.0)\n    recall = tp / max(tp + fn, 1.0)\n    specificity = tn / max(tn + fp, 1.0)\n    f1 = 2 * precision * recall / max(precision + recall, 1e-12)\n    balanced_accuracy = 0.5 * (recall + specificity)\n    return {\"f1\": f1, \"balanced_accuracy\": balanced_accuracy, \"precision\": precision, \"recall\": recall, \"specificity\": specificity}\n\n\ndef calibration_and_business_metrics(pred_df, k=0.10, n_bins=10):\n    \"\"\"Calibration, lift, and profit metrics with explicit cache cleanup.\"\"\"\n    cached = pred_df.select(\"label\", \"score\").filter(\n        F.col(\"label\").isNotNull() & F.col(\"score\").isNotNull()\n    ).persist(StorageLevel.MEMORY_AND_DISK)\n    try:\n        n = cached.count()\n        if n == 0:\n            return {\n                \"brier\": np.nan, \"log_loss\": np.nan, \"ece_10\": np.nan,\n                \"lift_at_10\": np.nan, \"profit_index_at_10\": np.nan,\n                \"positive_rate\": np.nan,\n            }\n\n        eps = F.lit(1e-6)\n        clipped = F.least(F.greatest(F.col(\"score\"), eps), F.lit(1.0) - eps)\n        base = cached.withColumn(\"score_clipped\", clipped)\n\n        agg = base.agg(\n            F.avg(\"label\").alias(\"positive_rate\"),\n            F.avg((F.col(\"score\") - F.col(\"label\")) * (F.col(\"score\") - F.col(\"label\"))).alias(\"brier\"),\n            F.avg(\n                -(\n                    F.col(\"label\") * F.log(F.col(\"score_clipped\"))\n                    + (F.lit(1.0) - F.col(\"label\")) * F.log(F.lit(1.0) - F.col(\"score_clipped\"))\n                )\n            ).alias(\"log_loss\"),\n        ).collect()[0].asDict()\n\n        binned = base.withColumn(\n            \"cal_bin\",\n            F.least(F.floor(F.col(\"score\") * F.lit(n_bins)).cast(\"int\"), F.lit(n_bins - 1)),\n        )\n        bin_table = binned.groupBy(\"cal_bin\").agg(\n            F.count(\"*\").alias(\"bin_n\"),\n            F.avg(\"label\").alias(\"bin_acc\"),\n            F.avg(\"score\").alias(\"bin_conf\"),\n        )\n        ece = bin_table.withColumn(\n            \"weighted_abs_gap\",\n            (F.col(\"bin_n\") / F.lit(float(n))) * F.abs(F.col(\"bin_acc\") - F.col(\"bin_conf\")),\n        ).agg(F.sum(\"weighted_abs_gap\")).collect()[0][0]\n\n        top_n = max(int(math.ceil(n * k)), 1)\n        w = Window.orderBy(F.desc(\"score\"))\n        ranked = base.withColumn(\"rank_desc\", F.row_number().over(w)).filter(F.col(\"rank_desc\") <= top_n)\n        top_pos_rate = ranked.agg(F.avg(\"label\")).collect()[0][0]\n        positive_rate = float(agg[\"positive_rate\"] or 0.0)\n        lift = float(top_pos_rate / positive_rate) if positive_rate > 0 else np.nan\n\n        retention_value_index = 100.0\n        intervention_cost_index = 10.0\n        success_rate = 0.20\n        top_summary = ranked.agg(\n            F.sum(\"label\").alias(\"top_positives\"),\n            F.count(\"*\").alias(\"contacted\")\n        ).collect()[0].asDict()\n        profit_index = (\n            (float(top_summary[\"top_positives\"] or 0.0) * retention_value_index * success_rate)\n            - (float(top_summary[\"contacted\"] or 0.0) * intervention_cost_index)\n        ) / max(n, 1)\n\n        return {\n            \"brier\": float(agg[\"brier\"]),\n            \"log_loss\": float(agg[\"log_loss\"]),\n            \"ece_10\": float(ece) if ece is not None else np.nan,\n            \"lift_at_10\": lift,\n            \"profit_index_at_10\": float(profit_index),\n            \"positive_rate\": positive_rate,\n        }\n    finally:\n        cached.unpersist(blocking=False)\n\n\ndef evaluate_dummy_prior(test_df, prior):\n    pred = test_df.select(\"label\").withColumn(\"score\", F.lit(float(prior))).persist(StorageLevel.MEMORY_AND_DISK)\n    try:\n        metrics = calibration_and_business_metrics(pred, k=0.10)\n        metrics.update(threshold_metrics(pred))\n        # For a constant score, ROC-AUC is 0.5 if both classes are present; PR-AUC equals prevalence.\n        metrics[\"roc_auc\"] = 0.5\n        metrics[\"pr_auc\"] = metrics.get(\"positive_rate\", np.nan)\n        return metrics\n    finally:\n        pred.unpersist(blocking=False)\n\n\ndef evaluate_model_predictions(pred_df):\n    pred = attach_score(pred_df).persist(StorageLevel.MEMORY_AND_DISK)\n    try:\n        metrics = calibration_and_business_metrics(pred, k=0.10)\n        metrics.update(threshold_metrics(pred))\n        metrics[\"roc_auc\"] = safe_auc(pred, \"areaUnderROC\")\n        metrics[\"pr_auc\"] = safe_auc(pred, \"areaUnderPR\")\n        return metrics\n    finally:\n        pred.unpersist(blocking=False)\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T06:55:29.574826Z","iopub.execute_input":"2026-06-28T06:55:29.575024Z","iopub.status.idle":"2026-06-28T06:55:29.602299Z","shell.execute_reply.started":"2026-06-28T06:55:29.575003Z","shell.execute_reply":"2026-06-28T06:55:29.601446Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 12. Experiment runner\n# ============================================================\n\nMODEL_NAMES = []\nif RUN_LOGISTIC_REGRESSION:\n    MODEL_NAMES.append(\"spark_logistic_regression_weighted\")\nif RUN_RANDOM_FOREST:\n    MODEL_NAMES.append(\"spark_random_forest_weighted\")\nif RUN_GBT:\n    MODEL_NAMES.append(\"spark_gradient_boosted_trees_weighted\")\n\nall_dataset_ids = [r[\"dataset_id\"] for r in modeling_data.select(\"dataset_id\").distinct().orderBy(\"dataset_id\").collect()]\nall_domain_ids = [r[\"domain_id\"] for r in modeling_data.select(\"domain_id\").distinct().orderBy(\"domain_id\").collect()]\nprint(\"Datasets:\", all_dataset_ids)\nprint(\"Domains:\", all_domain_ids)\nprint(\"Models:\", MODEL_NAMES)\n\n\ndef base_result_record(experiment, target_dataset, target_domain, model_name, train_df, test_df, weighting_mode):\n    return {\n        \"experiment\": experiment,\n        \"target_dataset\": target_dataset,\n        \"target_domain\": target_domain,\n        \"model\": model_name,\n        \"training_weighting\": weighting_mode,\n        \"n_train\": int(train_df.count()),\n        \"n_test\": int(test_df.count()),\n        \"train_positive_rate\": float(train_df.agg(F.avg(\"label\")).collect()[0][0]),\n        \"test_positive_rate\": float(test_df.agg(F.avg(\"label\")).collect()[0][0]),\n    }\n\n\ndef run_single_experiment(experiment, target_dataset, target_domain, train_df, test_df, weighting_mode=\"row_weighted\"):\n    rows = []\n    train_df = prepare_for_modeling(train_df.filter(F.col(\"label\").isin(0.0, 1.0))).persist(StorageLevel.MEMORY_AND_DISK)\n    test_df = prepare_for_modeling(test_df.filter(F.col(\"label\").isin(0.0, 1.0))).persist(StorageLevel.MEMORY_AND_DISK)\n    train_n = train_df.count()\n    test_n = test_df.count()\n\n    if train_n == 0 or test_n == 0:\n        print(\"Skipping empty experiment:\", experiment, target_dataset, target_domain)\n        return rows\n\n    # Dummy prior baseline.\n    prior = float(train_df.agg(F.avg(\"label\")).collect()[0][0])\n    dummy_metrics = evaluate_dummy_prior(test_df, prior)\n    rec = base_result_record(experiment, target_dataset, target_domain, \"dummy_prior\", train_df, test_df, weighting_mode)\n    rec.update(dummy_metrics)\n    rec[\"runtime_seconds\"] = 0.0\n    rows.append(rec)\n\n    weighted_train = add_sample_weights(train_df, weighting_mode=weighting_mode).persist(StorageLevel.MEMORY_AND_DISK)\n    _ = weighted_train.count()\n\n    for model_name in MODEL_NAMES:\n        print(f\"Running {experiment} | target={target_dataset or target_domain} | model={model_name} | train={train_n} | test={test_n}\")\n        t0 = time.time()\n        try:\n            pipeline = make_model_pipeline(model_name)\n            fitted = pipeline.fit(weighted_train)\n            pred = fitted.transform(test_df)\n            metrics = evaluate_model_predictions(pred)\n            rec = base_result_record(experiment, target_dataset, target_domain, model_name, train_df, test_df, weighting_mode)\n            rec.update(metrics)\n            rec[\"runtime_seconds\"] = time.time() - t0\n            rows.append(rec)\n\n            # Keep compact feature importance for Spark tree/linear models.\n            fi = extract_spark_feature_importance(fitted, experiment, target_dataset, target_domain, model_name)\n            if fi is not None and len(fi) > 0:\n                feature_importance_frames.append(fi)\n        except Exception as e:\n            rec = base_result_record(experiment, target_dataset, target_domain, model_name, train_df, test_df, weighting_mode)\n            rec.update({\"error\": repr(e), \"runtime_seconds\": time.time() - t0})\n            rows.append(rec)\n            print(\"ERROR:\", repr(e))\n\n    weighted_train.unpersist(blocking=False)\n    train_df.unpersist(blocking=False)\n    test_df.unpersist(blocking=False)\n    spark.catalog.clearCache()\n    gc.collect()\n    return rows\n\n\ndef get_feature_names_from_fitted_pipeline(fitted_pipeline):\n    # Numeric imputed columns are first; then gender one-hot columns.\n    names = [f\"{c}_imputed\" for c in NUMERIC_FEATURES]\n    try:\n        indexer_model = fitted_pipeline.stages[1]\n        labels = list(indexer_model.labels)\n        names.extend([f\"gender={x}\" for x in labels])\n        names.append(\"gender=unknown_or_invalid\")\n    except Exception:\n        names.append(\"gender_ohe\")\n    return names\n\n\ndef extract_spark_feature_importance(fitted_pipeline, experiment, target_dataset, target_domain, model_name):\n    try:\n        final_model = fitted_pipeline.stages[-1]\n        feature_names = get_feature_names_from_fitted_pipeline(fitted_pipeline)\n        if hasattr(final_model, \"featureImportances\"):\n            importances = final_model.featureImportances.toArray()\n        elif hasattr(final_model, \"coefficients\"):\n            importances = np.abs(np.array(final_model.coefficients.toArray(), dtype=float))\n        else:\n            return None\n        if len(feature_names) < len(importances):\n            feature_names = feature_names + [f\"extra_feature_{i}\" for i in range(len(importances) - len(feature_names))]\n        feature_names = feature_names[:len(importances)]\n        out = pd.DataFrame({\n            \"experiment\": experiment,\n            \"target_dataset\": target_dataset,\n            \"target_domain\": target_domain,\n            \"model\": model_name,\n            \"feature\": feature_names,\n            \"importance\": importances,\n        })\n        total = out[\"importance\"].sum()\n        out[\"importance_norm\"] = out[\"importance\"] / total if total > 0 else 0.0\n        return out.sort_values(\"importance_norm\", ascending=False)\n    except Exception:\n        return None\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T06:55:29.603183Z","iopub.execute_input":"2026-06-28T06:55:29.603464Z","iopub.status.idle":"2026-06-28T06:55:30.387699Z","shell.execute_reply.started":"2026-06-28T06:55:29.603431Z","shell.execute_reply":"2026-06-28T06:55:30.386901Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 13. Run within, pooled, leave-one-dataset-out, and leave-one-domain-out experiments\n# ============================================================\n\nresults_rows = []\nfeature_importance_frames = []\n\ndef save_partial_results():\n    pd.DataFrame(results_rows).to_csv(f\"{DIRS['results']}/spark_ml_all_results_partial.csv\", index=False)\n    if len(feature_importance_frames) > 0:\n        pd.concat(feature_importance_frames, ignore_index=True).to_csv(\n            f\"{DIRS['results']}/spark_feature_importance_partial.csv\", index=False\n        )\n\ndef append_experiment_rows(rows):\n    results_rows.extend(rows)\n    save_partial_results()\n    spark.catalog.clearCache()\n    gc.collect()\n\n# 13.1 Within-dataset evaluation.\nfor i, dataset_id in enumerate(all_dataset_ids, start=1):\n    print(f\"[within {i}/{len(all_dataset_ids)}] {dataset_id}\")\n    df = modeling_data.filter(F.col(\"dataset_id\") == dataset_id)\n    domain_id = df.select(\"domain_id\").first()[\"domain_id\"]\n    train_df = df.filter(F.col(\"split\").isin(\"train\", \"validation\"))\n    test_df = df.filter(F.col(\"split\") == \"test\")\n    append_experiment_rows(\n        run_single_experiment(\n            experiment=\"within_dataset\",\n            target_dataset=dataset_id,\n            target_domain=domain_id,\n            train_df=train_df,\n            test_df=test_df,\n            weighting_mode=\"row_weighted\",\n        )\n    )\n\n# 13.2 Pooled random split, row-weighted.\nprint(\"[pooled] row-weighted\")\npooled_train = modeling_data.filter(F.col(\"split\").isin(\"train\", \"validation\"))\npooled_test = modeling_data.filter(F.col(\"split\") == \"test\")\nappend_experiment_rows(\n    run_single_experiment(\n        experiment=\"pooled_random_split_row_weighted\",\n        target_dataset=\"mixed\",\n        target_domain=\"mixed\",\n        train_df=pooled_train,\n        test_df=pooled_test,\n        weighting_mode=\"row_weighted\",\n    )\n)\n\n# 13.3 Pooled random split, dataset-balanced.\nprint(\"[pooled] dataset-balanced\")\nappend_experiment_rows(\n    run_single_experiment(\n        experiment=\"pooled_random_split_dataset_balanced\",\n        target_dataset=\"mixed\",\n        target_domain=\"mixed\",\n        train_df=pooled_train,\n        test_df=pooled_test,\n        weighting_mode=\"dataset_balanced\",\n    )\n)\n\n# 13.4 Leave-one-dataset-out.\nfor i, dataset_id in enumerate(all_dataset_ids, start=1):\n    print(f\"[LODO {i}/{len(all_dataset_ids)}] {dataset_id}\")\n    target_domain = modeling_data.filter(F.col(\"dataset_id\") == dataset_id).select(\"domain_id\").first()[\"domain_id\"]\n    train_df = modeling_data.filter(F.col(\"dataset_id\") != dataset_id)\n    test_df = modeling_data.filter(F.col(\"dataset_id\") == dataset_id)\n    append_experiment_rows(\n        run_single_experiment(\n            experiment=\"leave_one_dataset_out\",\n            target_dataset=dataset_id,\n            target_domain=target_domain,\n            train_df=train_df,\n            test_df=test_df,\n            weighting_mode=\"dataset_balanced\",\n        )\n    )\n\n# 13.5 Leave-one-domain-out.\nfor i, domain_id in enumerate(all_domain_ids, start=1):\n    print(f\"[LODOm {i}/{len(all_domain_ids)}] {domain_id}\")\n    train_df = modeling_data.filter(F.col(\"domain_id\") != domain_id)\n    test_df = modeling_data.filter(F.col(\"domain_id\") == domain_id)\n    append_experiment_rows(\n        run_single_experiment(\n            experiment=\"leave_one_domain_out\",\n            target_dataset=\"heldout_domain\",\n            target_domain=domain_id,\n            train_df=train_df,\n            test_df=test_df,\n            weighting_mode=\"dataset_balanced\",\n        )\n    )\n\nall_results = pd.DataFrame(results_rows)\nall_results.to_csv(f\"{DIRS['results']}/spark_ml_all_results.csv\", index=False)\n\ndisplay(all_results.sort_values([\"experiment\", \"target_domain\", \"target_dataset\", \"model\"]).head(50))\nprint(\"Saved results:\", f\"{DIRS['results']}/spark_ml_all_results.csv\")\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T06:55:30.388826Z","iopub.execute_input":"2026-06-28T06:55:30.389168Z","iopub.status.idle":"2026-06-28T07:07:20.909598Z","shell.execute_reply.started":"2026-06-28T06:55:30.389134Z","shell.execute_reply":"2026-06-28T07:07:20.908875Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 14. Best-model summary, transfer degradation, and statistical tests\n# ============================================================\n\nresults = pd.read_csv(f\"{DIRS['results']}/spark_ml_all_results.csv\")\nresults_valid = results[results[\"roc_auc\"].notna()].copy()\nnon_dummy = results_valid[results_valid[\"model\"] != \"dummy_prior\"].copy()\n\nbest_by_group = (\n    non_dummy.sort_values([\"experiment\", \"target_dataset\", \"target_domain\", \"roc_auc\"], ascending=[True, True, True, False])\n    .groupby([\"experiment\", \"target_dataset\", \"target_domain\"], as_index=False)\n    .head(1)\n    .reset_index(drop=True)\n)\nbest_by_group.to_csv(f\"{DIRS['results']}/spark_ml_best_results.csv\", index=False)\n\n# Headline protocol means.\nheadline_rows = []\n\ndef add_headline(protocol_name, value, note):\n    headline_rows.append({\"protocol\": protocol_name, \"roc_auc\": value, \"note\": note})\n\nwithin_mean = best_by_group.loc[best_by_group[\"experiment\"] == \"within_dataset\", \"roc_auc\"].mean()\nlodo_mean = best_by_group.loc[best_by_group[\"experiment\"] == \"leave_one_dataset_out\", \"roc_auc\"].mean()\ndomain_out_mean = best_by_group.loc[best_by_group[\"experiment\"] == \"leave_one_domain_out\", \"roc_auc\"].mean()\npooled_row = best_by_group.loc[best_by_group[\"experiment\"] == \"pooled_random_split_row_weighted\", \"roc_auc\"].max()\npooled_bal = best_by_group.loc[best_by_group[\"experiment\"] == \"pooled_random_split_dataset_balanced\", \"roc_auc\"].max()\n\nadd_headline(\"within_dataset_mean\", within_mean, \"Mean of the best non-dummy model within each dataset.\")\nadd_headline(\"pooled_random_split_row_weighted\", pooled_row, \"Optimistic pooled split using row-proportional training weights.\")\nadd_headline(\"pooled_random_split_dataset_balanced\", pooled_bal, \"Optimistic pooled split using dataset-balanced training weights.\")\nadd_headline(\"leave_one_dataset_out_mean\", lodo_mean, \"Mean performance when each complete dataset is unseen during training.\")\nadd_headline(\"leave_one_domain_out_mean\", domain_out_mean, \"Mean performance when each complete business domain is unseen during training.\")\n\nheadline = pd.DataFrame(headline_rows)\nheadline.to_csv(f\"{DIRS['results']}/manuscript_headline_results.csv\", index=False)\n\n# Transfer degradation: within vs LODO by dataset.\nwithin = best_by_group[best_by_group[\"experiment\"] == \"within_dataset\"][[\"target_dataset\", \"target_domain\", \"roc_auc\", \"ece_10\", \"model\"]].rename(columns={\"roc_auc\": \"within_roc_auc\", \"ece_10\": \"within_ece\", \"model\": \"within_model\"})\nlodo = best_by_group[best_by_group[\"experiment\"] == \"leave_one_dataset_out\"][[\"target_dataset\", \"target_domain\", \"roc_auc\", \"ece_10\", \"model\"]].rename(columns={\"roc_auc\": \"lodo_roc_auc\", \"ece_10\": \"lodo_ece\", \"model\": \"lodo_model\"})\ntransfer = within.merge(lodo, on=[\"target_dataset\", \"target_domain\"], how=\"inner\")\ntransfer[\"roc_auc_drop\"] = transfer[\"lodo_roc_auc\"] - transfer[\"within_roc_auc\"]\ntransfer.to_csv(f\"{DIRS['results']}/dataset_transfer_degradation.csv\", index=False)\n\n# Domain transfer degradation.\nwithin_domain = (\n    within.groupby(\"target_domain\", as_index=False)\n    .agg(within_domain_mean_roc_auc=(\"within_roc_auc\", \"mean\"))\n    .rename(columns={\"target_domain\": \"domain_id\"})\n)\ndomain_out = best_by_group[best_by_group[\"experiment\"] == \"leave_one_domain_out\"][[\"target_domain\", \"roc_auc\", \"ece_10\", \"model\"]].rename(columns={\"target_domain\": \"domain_id\", \"roc_auc\": \"domain_out_roc_auc\", \"ece_10\": \"domain_out_ece\", \"model\": \"domain_out_model\"})\ndomain_transfer = within_domain.merge(domain_out, on=\"domain_id\", how=\"inner\")\ndomain_transfer[\"roc_auc_drop\"] = domain_transfer[\"domain_out_roc_auc\"] - domain_transfer[\"within_domain_mean_roc_auc\"]\ndomain_transfer.to_csv(f\"{DIRS['results']}/domain_transfer_degradation.csv\", index=False)\n\n# Statistical test on dataset-level transfer drops.\nstat_rows = []\nif len(transfer) > 1:\n    diffs = transfer[\"roc_auc_drop\"].dropna().values\n    rec = {\n        \"test\": \"dataset_level_lodo_minus_within_roc_auc\",\n        \"n\": int(len(diffs)),\n        \"mean_drop\": float(np.mean(diffs)),\n        \"median_drop\": float(np.median(diffs)),\n        \"std_drop\": float(np.std(diffs)),\n    }\n    if SCIPY_AVAILABLE:\n        try:\n            w = wilcoxon(diffs)\n            rec[\"wilcoxon_statistic\"] = float(w.statistic)\n            rec[\"wilcoxon_pvalue\"] = float(w.pvalue)\n        except Exception:\n            rec[\"wilcoxon_statistic\"] = np.nan\n            rec[\"wilcoxon_pvalue\"] = np.nan\n    stat_rows.append(rec)\n\nstat_tests = pd.DataFrame(stat_rows)\nstat_tests.to_csv(f\"{DIRS['results']}/statistical_tests.csv\", index=False)\n\ndisplay(headline)\ndisplay(best_by_group)\ndisplay(transfer)\ndisplay(domain_transfer)\ndisplay(stat_tests)\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T07:07:20.910535Z","iopub.execute_input":"2026-06-28T07:07:20.910907Z","iopub.status.idle":"2026-06-28T07:07:21.008179Z","shell.execute_reply.started":"2026-06-28T07:07:20.910874Z","shell.execute_reply":"2026-06-28T07:07:21.007445Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 15. Feature importance and stability outputs\n# ============================================================\n\nif len(feature_importance_frames) > 0:\n    feature_importance = pd.concat(feature_importance_frames, ignore_index=True)\n    feature_importance[\"feature_family\"] = feature_importance[\"feature\"].str.replace(\"_imputed\", \"\", regex=False)\n    feature_importance[\"feature_family\"] = feature_importance[\"feature_family\"].str.replace(r\"gender=.*\", \"gender\", regex=True)\n    feature_importance.to_csv(f\"{DIRS['results']}/spark_feature_importance_all.csv\", index=False)\n\n    stability = (\n        feature_importance.groupby(\"feature_family\", as_index=False)\n        .agg(\n            appearances=(\"importance_norm\", \"count\"),\n            mean_importance=(\"importance_norm\", \"mean\"),\n            median_importance=(\"importance_norm\", \"median\"),\n            max_importance=(\"importance_norm\", \"max\"),\n        )\n    )\n    stability[\"stability_score\"] = stability[\"mean_importance\"] * np.log1p(stability[\"appearances\"])\n    stability = stability.sort_values(\"stability_score\", ascending=False)\n    stability.to_csv(f\"{DIRS['results']}/feature_stability_summary.csv\", index=False)\nelse:\n    feature_importance = pd.DataFrame()\n    stability = pd.DataFrame()\n    feature_importance.to_csv(f\"{DIRS['results']}/spark_feature_importance_all.csv\", index=False)\n    stability.to_csv(f\"{DIRS['results']}/feature_stability_summary.csv\", index=False)\n\nprint(\"Feature importance rows:\", len(feature_importance))\ndisplay(stability.head(20))\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T07:07:21.009086Z","iopub.execute_input":"2026-06-28T07:07:21.009311Z","iopub.status.idle":"2026-06-28T07:07:21.042582Z","shell.execute_reply.started":"2026-06-28T07:07:21.009290Z","shell.execute_reply":"2026-06-28T07:07:21.041668Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 16. Reconstruct paper profile tables for plotting and summary\n# ============================================================\n# This cell makes the figure section robust. It creates `target_profile`\n# and `modeling_profile` from saved output files, so the plot cells work\n# even if the notebook kernel was restarted after the ML run.\n\nfrom pathlib import Path\n\n# Ensure DIRS exists if this cell is run after a kernel restart and the config cell was not rerun.\nif \"DIRS\" not in globals():\n    OUTPUT_ROOT = \"/kaggle/working/spark_churn_journal_benchmark\"\n    DIRS = {\n        \"root\": OUTPUT_ROOT,\n        \"registry\": f\"{OUTPUT_ROOT}/registry\",\n        \"profiles\": f\"{OUTPUT_ROOT}/bigdata_profiles\",\n        \"harmonized\": f\"{OUTPUT_ROOT}/harmonized_data\",\n        \"results\": f\"{OUTPUT_ROOT}/results\",\n        \"figures\": f\"{OUTPUT_ROOT}/figures\",\n        \"tables\": f\"{OUTPUT_ROOT}/tables\",\n        \"logs\": f\"{OUTPUT_ROOT}/logs\",\n    }\n    for d in DIRS.values():\n        os.makedirs(d, exist_ok=True)\n\n# Load Spark source profile if the in-memory variable is missing.\nif \"source_profile\" not in globals():\n    source_profile_path = Path(DIRS[\"profiles\"]) / \"spark_source_profile.csv\"\n    if not source_profile_path.exists():\n        raise FileNotFoundError(\n            f\"Missing {source_profile_path}. Run the Spark profiling cell before the figure section.\"\n        )\n    source_profile = pd.read_csv(source_profile_path)\n\n# Build the final supervised benchmark target profile from source-level Spark profiling.\n# Important: cell2cell holdout is kept in raw profiling but excluded from supervised benchmark rows.\nprofile_dataset_map = {\n    \"telco_ibm\": \"telco_ibm\",\n    \"telco_orange_train\": \"telco_orange\",\n    \"telco_orange_test\": \"telco_orange\",\n    \"cell2cell_train\": \"cell2cell_train_holdout\",\n    # Do not map cell2cell_holdout: it has no supervised labels for this benchmark.\n    \"bank_original\": \"bank_original\",\n    \"bank_creditcard\": \"bank_creditcard\",\n    \"ecommerce_rees46\": \"ecommerce_rees46\",\n    \"ecommerce_shriyash_large\": \"ecommerce_shriyash_large\",\n    \"ecommerce_dhairyajeet\": \"ecommerce_dhairyajeet\",\n    \"subscription_generic_train\": \"subscription_generic\",\n    \"subscription_generic_test\": \"subscription_generic\",\n    \"subscription_kkbox_train\": \"subscription_kkbox\",\n    \"insurance_auto\": \"insurance_auto\",\n    \"insurance_k123_train\": \"insurance_k123vinod\",\n}\n\nsp = source_profile.copy()\nsp[\"dataset_id\"] = sp[\"source_key\"].map(profile_dataset_map)\nsp = sp[sp[\"dataset_id\"].notna()].copy()\n\nfor c in [\"spark_rows\", \"positive_count\", \"positive_rate\"]:\n    if c in sp.columns:\n        sp[c] = pd.to_numeric(sp[c], errors=\"coerce\")\n\n# Convert source-level rows into dataset-level target profile.\ntarget_profile = (\n    sp.groupby([\"dataset_id\", \"domain_id\"], as_index=False)\n    .agg(\n        full_rows=(\"spark_rows\", \"sum\"),\n        positive_count=(\"positive_count\", \"sum\"),\n    )\n)\ntarget_profile[\"positive_count\"] = target_profile[\"positive_count\"].fillna(0).round().astype(int)\ntarget_profile[\"full_rows\"] = target_profile[\"full_rows\"].round().astype(int)\ntarget_profile[\"positive_rate\"] = target_profile[\"positive_count\"] / target_profile[\"full_rows\"].clip(lower=1)\ntarget_profile = target_profile.sort_values([\"domain_id\", \"dataset_id\"]).reset_index(drop=True)\n\n# Load or reconstruct modeling profile.\nmodeling_profile_path_candidates = [\n    Path(DIRS[\"tables\"]) / \"modeling_sample_profile.csv\",\n    Path(DIRS[\"registry\"]) / \"modeling_summary.csv\",\n    Path(DIRS[\"registry\"]) / \"modeling_build_profile.csv\",\n]\nmodeling_profile = None\nfor p in modeling_profile_path_candidates:\n    if p.exists():\n        tmp = pd.read_csv(p)\n        if \"dataset_id\" in tmp.columns and \"modeling_rows\" in tmp.columns:\n            modeling_profile = tmp.copy()\n            break\n\nif modeling_profile is None:\n    raise FileNotFoundError(\n        \"Could not find a modeling profile with dataset_id and modeling_rows. \"\n        \"Run the capped modeling table / split creation cell before plotting.\"\n    )\n\n# Normalize modeling profile columns.\nif \"domain_id\" not in modeling_profile.columns:\n    modeling_profile = modeling_profile.merge(target_profile[[\"dataset_id\", \"domain_id\"]], on=\"dataset_id\", how=\"left\")\nif \"positive_rate\" not in modeling_profile.columns and \"label_positive_rate\" in modeling_profile.columns:\n    modeling_profile = modeling_profile.rename(columns={\"label_positive_rate\": \"positive_rate\"})\nif \"positive_count\" not in modeling_profile.columns and \"positive_rate\" in modeling_profile.columns:\n    modeling_profile[\"positive_count\"] = (pd.to_numeric(modeling_profile[\"modeling_rows\"], errors=\"coerce\") * pd.to_numeric(modeling_profile[\"positive_rate\"], errors=\"coerce\")).round().astype(\"Int64\")\n\nmodeling_profile[\"modeling_rows\"] = pd.to_numeric(modeling_profile[\"modeling_rows\"], errors=\"coerce\").fillna(0).round().astype(int)\nmodeling_profile = modeling_profile.sort_values([\"domain_id\", \"dataset_id\"]).reset_index(drop=True)\n\n# Save clean tables for manuscript use.\ntarget_profile.to_csv(f\"{DIRS['tables']}/target_distribution_full.csv\", index=False)\nmodeling_profile.to_csv(f\"{DIRS['tables']}/modeling_sample_profile.csv\", index=False)\n\nprint(\"Target profile ready:\", target_profile.shape)\nprint(\"Modeling profile ready:\", modeling_profile.shape)\nprint(\"Full benchmark rows:\", f\"{int(target_profile['full_rows'].sum()):,}\")\nprint(\"Modeling rows:\", f\"{int(modeling_profile['modeling_rows'].sum()):,}\")\ndisplay(target_profile)\ndisplay(modeling_profile[[c for c in [\"dataset_id\", \"domain_id\", \"modeling_rows\", \"positive_rate\"] if c in modeling_profile.columns]])\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T07:07:21.043465Z","iopub.execute_input":"2026-06-28T07:07:21.043709Z","iopub.status.idle":"2026-06-28T07:07:21.086521Z","shell.execute_reply.started":"2026-06-28T07:07:21.043689Z","shell.execute_reply":"2026-06-28T07:07:21.085731Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 17. Journal-ready figures for manuscript and supplementary material\n# ============================================================\n\nplt.rcParams.update({\n    \"figure.dpi\": 160,\n    \"savefig.dpi\": 600,\n    \"font.family\": \"DejaVu Serif\",\n    \"font.size\": 10,\n    \"axes.spines.top\": False,\n    \"axes.spines.right\": False,\n    \"axes.grid\": True,\n    \"grid.alpha\": 0.25,\n    \"grid.linestyle\": \"--\",\n})\n\nfigure_catalog = []\n\n\ndef pretty_label(x):\n    return str(x).replace(\"_\", \" \").title()\n\n\ndef savefig_all(name, caption):\n    \"\"\"Save the current Matplotlib figure in paper-friendly formats.\"\"\"\n    for ext in [\"png\", \"pdf\", \"svg\"]:\n        plt.savefig(f\"{DIRS['figures']}/{name}.{ext}\", bbox_inches=\"tight\")\n    figure_catalog.append({\n        \"figure_file_stem\": name,\n        \"caption_suggestion\": caption,\n        \"formats\": \"png;pdf;svg\",\n    })\n    print(\"Saved figure:\", name)\n\n# ------------------------------------------------------------\n# Figure 1: dataset scale and churn prevalence.\n# ------------------------------------------------------------\nd = target_profile.merge(modeling_profile[[\"dataset_id\", \"modeling_rows\"]], on=\"dataset_id\", how=\"left\")\nd = d.sort_values(\"full_rows\")\nplt.figure(figsize=(9.5, 6.2))\nplt.barh(d[\"dataset_id\"].map(pretty_label), d[\"full_rows\"])\nplt.xscale(\"log\")\nplt.xlabel(\"Full rows, log scale\")\nplt.title(\"Benchmark scale and churn prevalence across datasets\")\nfor i, (_, r) in enumerate(d.iterrows()):\n    plt.text(r[\"full_rows\"], i, f\"  {100*r['positive_rate']:.1f}% churn\", va=\"center\", fontsize=8)\nsavefig_all(\n    \"figure_01_dataset_scale_and_prevalence\",\n    \"Dataset scale and churn prevalence across the harmonized benchmark. Full rows are plotted on a logarithmic scale, with churn prevalence shown beside each dataset.\"\n)\nplt.show()\n\n# ------------------------------------------------------------\n# Figure 2: domain composition by full records.\n# ------------------------------------------------------------\ndomain_full = (\n    target_profile.groupby(\"domain_id\", as_index=False)\n    .agg(\n        full_rows=(\"full_rows\", \"sum\"),\n        datasets=(\"dataset_id\", \"nunique\"),\n        positive_count=(\"positive_count\", \"sum\"),\n    )\n)\ndomain_full[\"positive_rate\"] = domain_full[\"positive_count\"] / domain_full[\"full_rows\"]\ndomain_full = domain_full.sort_values(\"full_rows\")\nplt.figure(figsize=(8.5, 5.2))\nplt.barh(domain_full[\"domain_id\"].map(pretty_label), domain_full[\"full_rows\"])\nplt.xscale(\"log\")\nplt.xlabel(\"Full rows, log scale\")\nplt.title(\"Benchmark composition by business domain\")\nfor i, (_, r) in enumerate(domain_full.iterrows()):\n    plt.text(r[\"full_rows\"], i, f\"  {int(r['datasets'])} datasets, {100*r['positive_rate']:.1f}% churn\", va=\"center\", fontsize=8)\nsavefig_all(\n    \"figure_02_domain_composition\",\n    \"Domain-level composition of the benchmark, showing full row counts, number of datasets, and average churn prevalence by domain.\"\n)\nplt.show()\n\n# ------------------------------------------------------------\n# Figure 3: protocol-level ROC-AUC summary.\n# ------------------------------------------------------------\nh = headline.copy()\nprotocol_order = [\n    \"within_dataset_mean\",\n    \"pooled_random_split_dataset_balanced\",\n    \"pooled_random_split_row_weighted\",\n    \"leave_one_dataset_out_mean\",\n    \"leave_one_domain_out_mean\",\n]\nh[\"order\"] = h[\"protocol\"].map({p: i for i, p in enumerate(protocol_order)})\nh = h.sort_values(\"order\")\nplt.figure(figsize=(9.5, 4.8))\nplt.barh(h[\"protocol\"].map(pretty_label), h[\"roc_auc\"])\nplt.axvline(0.5, linestyle=\"--\")\nplt.xlabel(\"ROC-AUC\")\nplt.title(\"Performance falls under stricter external-generalization protocols\")\nfor i, (_, r) in enumerate(h.iterrows()):\n    plt.text(r[\"roc_auc\"], i, f\" {r['roc_auc']:.3f}\", va=\"center\", fontsize=9)\nsavefig_all(\n    \"figure_03_protocol_roc_auc_summary\",\n    \"Headline ROC-AUC across evaluation protocols. Within-dataset and pooled testing are stronger than leave-one-dataset-out and leave-one-domain-out testing.\"\n)\nplt.show()\n\n# ------------------------------------------------------------\n# Figure 4: leave-one-dataset-out transfer degradation.\n# ------------------------------------------------------------\nif len(transfer) > 0:\n    t = transfer.sort_values(\"roc_auc_drop\")\n    plt.figure(figsize=(9.5, 6.2))\n    plt.barh(t[\"target_dataset\"].map(pretty_label), t[\"roc_auc_drop\"])\n    plt.axvline(0, linestyle=\"--\")\n    plt.xlabel(\"LODO ROC-AUC minus within-dataset ROC-AUC\")\n    plt.title(\"Leave-one-dataset-out transfer degradation\")\n    for i, (_, r) in enumerate(t.iterrows()):\n        plt.text(r[\"roc_auc_drop\"], i, f\" {r['roc_auc_drop']:.3f}\", va=\"center\", fontsize=8)\n    savefig_all(\n        \"figure_04_dataset_transfer_degradation\",\n        \"Dataset-level transfer degradation, computed as leave-one-dataset-out ROC-AUC minus within-dataset ROC-AUC. Negative values indicate weaker transportability to unseen datasets.\"\n    )\n    plt.show()\n\n# ------------------------------------------------------------\n# Figure 5: leave-one-domain-out transfer degradation.\n# ------------------------------------------------------------\nif len(domain_transfer) > 0:\n    dt = domain_transfer.sort_values(\"roc_auc_drop\")\n    plt.figure(figsize=(8.5, 4.8))\n    plt.barh(dt[\"domain_id\"].map(pretty_label), dt[\"roc_auc_drop\"])\n    plt.axvline(0, linestyle=\"--\")\n    plt.xlabel(\"Domain-out ROC-AUC minus within-domain mean ROC-AUC\")\n    plt.title(\"Leave-one-domain-out transfer degradation\")\n    for i, (_, r) in enumerate(dt.iterrows()):\n        plt.text(r[\"roc_auc_drop\"], i, f\" {r['roc_auc_drop']:.3f}\", va=\"center\", fontsize=9)\n    savefig_all(\n        \"figure_05_domain_transfer_degradation\",\n        \"Domain-level transfer degradation, computed as leave-one-domain-out ROC-AUC minus the mean within-domain ROC-AUC.\"\n    )\n    plt.show()\n\n# ------------------------------------------------------------\n# Figure 6: within-dataset ROC-AUC versus LODO ROC-AUC.\n# Journal-clean version: use compact dataset codes to avoid label overlap.\n# ------------------------------------------------------------\nif len(transfer) > 0:\n    short_label_map = {\n        \"bank_creditcard\": \"B-CC\",\n        \"bank_original\": \"B-Orig\",\n        \"cell2cell_train_holdout\": \"T-C2C\",\n        \"ecommerce_dhairyajeet\": \"E-Dhair\",\n        \"ecommerce_rees46\": \"E-Rees\",\n        \"ecommerce_shriyash_large\": \"E-Shri\",\n        \"insurance_auto\": \"I-Auto\",\n        \"insurance_k123vinod\": \"I-K123\",\n        \"subscription_generic\": \"S-Gen\",\n        \"subscription_kkbox\": \"S-KKBox\",\n        \"telco_ibm\": \"T-IBM\",\n        \"telco_orange\": \"T-Orange\",\n    }\n\n    dataset_label_key = pd.DataFrame([\n        {\"dataset_code\": code, \"dataset_id\": ds, \"dataset_name\": pretty_label(ds)}\n        for ds, code in short_label_map.items()\n    ]).sort_values(\"dataset_code\")\n    dataset_label_key.to_csv(f\"{DIRS['figures']}/figure_06_dataset_label_key.csv\", index=False)\n\n    plot_df = transfer.copy()\n    plot_df[\"short_label\"] = plot_df[\"target_dataset\"].map(short_label_map).fillna(plot_df[\"target_dataset\"])\n    plot_df = plot_df.sort_values(\"within_roc_auc\").reset_index(drop=True)\n\n    plt.figure(figsize=(8.8, 6.4))\n    plt.scatter(plot_df[\"within_roc_auc\"], plot_df[\"lodo_roc_auc\"], s=45, alpha=0.85)\n\n    min_axis = min(plot_df[\"within_roc_auc\"].min(), plot_df[\"lodo_roc_auc\"].min()) - 0.03\n    max_axis = max(plot_df[\"within_roc_auc\"].max(), plot_df[\"lodo_roc_auc\"].max()) + 0.03\n\n    plt.plot([min_axis, max_axis], [min_axis, max_axis], linestyle=\"--\", linewidth=1.1)\n    plt.axhline(0.5, linestyle=\":\", linewidth=1.2)\n    plt.axvline(0.5, linestyle=\":\", linewidth=1.2)\n\n    # Manual offsets reduce overlap in dense regions. Dataset codes are explained\n    # in figure_06_dataset_label_key.csv and in the manuscript caption.\n    offsets = {\n        \"B-CC\": (6, 8),\n        \"B-Orig\": (-36, 8),\n        \"T-C2C\": (-40, -12),\n        \"E-Dhair\": (6, 8),\n        \"E-Rees\": (6, -13),\n        \"E-Shri\": (6, 7),\n        \"I-Auto\": (6, -13),\n        \"I-K123\": (6, 7),\n        \"S-Gen\": (6, 5),\n        \"S-KKBox\": (6, 7),\n        \"T-IBM\": (6, -13),\n        \"T-Orange\": (6, -15),\n    }\n\n    for _, r in plot_df.iterrows():\n        label = r[\"short_label\"]\n        dx, dy = offsets.get(label, (6, 6))\n        plt.annotate(\n            label,\n            xy=(r[\"within_roc_auc\"], r[\"lodo_roc_auc\"]),\n            xytext=(dx, dy),\n            textcoords=\"offset points\",\n            fontsize=8,\n            ha=\"left\" if dx >= 0 else \"right\",\n            va=\"center\",\n            bbox=dict(boxstyle=\"round,pad=0.15\", fc=\"white\", ec=\"none\", alpha=0.75),\n        )\n\n    plt.xlabel(\"Within-dataset ROC-AUC\")\n    plt.ylabel(\"Leave-one-dataset-out ROC-AUC\")\n    plt.title(\"Within-dataset success does not guarantee dataset transfer\")\n    plt.xlim(min_axis, max_axis)\n    plt.ylim(min_axis, max_axis)\n\n    savefig_all(\n        \"figure_06_within_vs_lodo_scatter\",\n        \"Within-dataset ROC-AUC versus leave-one-dataset-out ROC-AUC for each dataset. Compact labels are used to avoid overlap: B-CC=bank credit card, B-Orig=bank original, T-C2C=Cell2Cell, T-IBM=IBM Telco, T-Orange=Orange Telecom, E-Dhair=Dhairyajeet e-commerce, E-Rees=REES46, E-Shri=Shriyash e-commerce, S-Gen=generic subscription, S-KKBox=KKBox, I-Auto=auto insurance, and I-K123=K123Vinod insurance.\"\n    )\n\n    # Optional duplicate clean filename for easy manual replacement in Overleaf.\n    for ext in [\"png\", \"pdf\", \"svg\"]:\n        plt.savefig(f\"{DIRS['figures']}/figure_06_within_vs_lodo_scatter_clean.{ext}\", bbox_inches=\"tight\")\n\n    plt.show()\n\n# ------------------------------------------------------------\n# Figure 7: calibration error by protocol.\n# ------------------------------------------------------------\nif \"ece_10\" in best_by_group.columns:\n    cal = best_by_group.groupby(\"experiment\", as_index=False)[\"ece_10\"].mean()\n    cal[\"order\"] = cal[\"experiment\"].map({\n        \"within_dataset\": 0,\n        \"pooled_random_split_dataset_balanced\": 1,\n        \"pooled_random_split_row_weighted\": 2,\n        \"leave_one_dataset_out\": 3,\n        \"leave_one_domain_out\": 4,\n    })\n    cal = cal.sort_values(\"order\")\n    plt.figure(figsize=(9, 4.8))\n    plt.barh(cal[\"experiment\"].map(pretty_label), cal[\"ece_10\"])\n    plt.xlabel(\"Mean expected calibration error\")\n    plt.title(\"Calibration error differs across validation protocols\")\n    for i, (_, r) in enumerate(cal.iterrows()):\n        plt.text(r[\"ece_10\"], i, f\" {r['ece_10']:.3f}\", va=\"center\", fontsize=9)\n    savefig_all(\n        \"figure_07_calibration_error_by_protocol\",\n        \"Mean expected calibration error by evaluation protocol. The figure reports probability reliability in addition to discrimination.\"\n    )\n    plt.show()\n\n# ------------------------------------------------------------\n# Figure 8: PR-AUC by protocol.\n# ------------------------------------------------------------\nif \"pr_auc\" in best_by_group.columns:\n    pr = best_by_group.groupby(\"experiment\", as_index=False)[\"pr_auc\"].mean()\n    pr[\"order\"] = pr[\"experiment\"].map({\n        \"within_dataset\": 0,\n        \"pooled_random_split_dataset_balanced\": 1,\n        \"pooled_random_split_row_weighted\": 2,\n        \"leave_one_dataset_out\": 3,\n        \"leave_one_domain_out\": 4,\n    })\n    pr = pr.sort_values(\"order\")\n    plt.figure(figsize=(9, 4.8))\n    plt.barh(pr[\"experiment\"].map(pretty_label), pr[\"pr_auc\"])\n    plt.xlabel(\"Mean PR-AUC\")\n    plt.title(\"Precision-recall performance by evaluation protocol\")\n    for i, (_, r) in enumerate(pr.iterrows()):\n        plt.text(r[\"pr_auc\"], i, f\" {r['pr_auc']:.3f}\", va=\"center\", fontsize=9)\n    savefig_all(\n        \"figure_08_pr_auc_by_protocol\",\n        \"Mean PR-AUC by evaluation protocol. PR-AUC is included because churn prevalence varies substantially across datasets.\"\n    )\n    plt.show()\n\n# ------------------------------------------------------------\n# Figure 9: Lift@10 by protocol.\n# ------------------------------------------------------------\nif \"lift_at_10\" in best_by_group.columns:\n    lift = best_by_group.groupby(\"experiment\", as_index=False)[\"lift_at_10\"].mean()\n    lift[\"order\"] = lift[\"experiment\"].map({\n        \"within_dataset\": 0,\n        \"pooled_random_split_dataset_balanced\": 1,\n        \"pooled_random_split_row_weighted\": 2,\n        \"leave_one_dataset_out\": 3,\n        \"leave_one_domain_out\": 4,\n    })\n    lift = lift.sort_values(\"order\")\n    plt.figure(figsize=(9, 4.8))\n    plt.barh(lift[\"experiment\"].map(pretty_label), lift[\"lift_at_10\"])\n    plt.axvline(1.0, linestyle=\"--\")\n    plt.xlabel(\"Mean Lift@10\")\n    plt.title(\"Top-decile churn targeting value by protocol\")\n    for i, (_, r) in enumerate(lift.iterrows()):\n        plt.text(r[\"lift_at_10\"], i, f\" {r['lift_at_10']:.2f}\", va=\"center\", fontsize=9)\n    savefig_all(\n        \"figure_09_lift_at_10_by_protocol\",\n        \"Mean Lift@10 by evaluation protocol. Values above 1 indicate enrichment of churners in the top-scored decile relative to random targeting.\"\n    )\n    plt.show()\n\n# ------------------------------------------------------------\n# Figure 10: feature stability.\n# ------------------------------------------------------------\nif len(stability) > 0:\n    s = stability.head(12).sort_values(\"stability_score\")\n    plt.figure(figsize=(8.8, 5.8))\n    plt.barh(s[\"feature_family\"].map(pretty_label), s[\"stability_score\"])\n    plt.xlabel(\"Feature stability score\")\n    plt.title(\"Most stable harmonized churn feature families\")\n    savefig_all(\n        \"figure_10_feature_stability\",\n        \"Most stable harmonized churn feature families, summarized from Spark MLlib logistic-regression coefficient magnitudes across experiments.\"\n    )\n    plt.show()\n\n# ------------------------------------------------------------\n# Figure 11: held-out domain ROC-AUC.\n# ------------------------------------------------------------\nif len(domain_transfer) > 0:\n    dt_auc = domain_transfer.sort_values(\"domain_out_roc_auc\")\n    plt.figure(figsize=(8.5, 4.8))\n    plt.barh(dt_auc[\"domain_id\"].map(pretty_label), dt_auc[\"domain_out_roc_auc\"])\n    plt.axvline(0.5, linestyle=\"--\")\n    plt.xlabel(\"Leave-one-domain-out ROC-AUC\")\n    plt.title(\"Unseen-domain performance remains weak to moderate\")\n    for i, (_, r) in enumerate(dt_auc.iterrows()):\n        plt.text(r[\"domain_out_roc_auc\"], i, f\" {r['domain_out_roc_auc']:.3f}\", va=\"center\", fontsize=9)\n    savefig_all(\n        \"figure_11_domain_out_roc_auc\",\n        \"ROC-AUC for each held-out business domain under leave-one-domain-out evaluation.\"\n    )\n    plt.show()\n\nfigure_catalog_df = pd.DataFrame(figure_catalog)\nfigure_catalog_df.to_csv(f\"{DIRS['figures']}/figure_catalog.csv\", index=False)\ndisplay(figure_catalog_df)\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T07:07:21.087481Z","iopub.execute_input":"2026-06-28T07:07:21.087709Z","iopub.status.idle":"2026-06-28T07:07:37.205308Z","shell.execute_reply.started":"2026-06-28T07:07:21.087681Z","shell.execute_reply":"2026-06-28T07:07:37.204432Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# 18. Manuscript-ready summary text and final paper-only ZIP export\n# ============================================================\n\nsummary = {\n    \"benchmark_datasets\": int(target_profile[\"dataset_id\"].nunique()),\n    \"business_domains\": int(target_profile[\"domain_id\"].nunique()),\n    \"full_harmonized_rows\": int(target_profile[\"full_rows\"].sum()),\n    \"modeling_rows\": int(modeling_profile[\"modeling_rows\"].sum()),\n    \"source_files_profiled_with_spark\": int(source_profile.shape[0]),\n    \"spark_observed_source_rows\": int(source_profile[\"spark_rows\"].sum()),\n    \"source_size_mb\": float(source_profile[\"source_size_mb\"].sum()),\n    \"within_dataset_mean_roc_auc\": float(within_mean) if not pd.isna(within_mean) else None,\n    \"pooled_random_split_dataset_balanced_roc_auc\": float(pooled_bal) if not pd.isna(pooled_bal) else None,\n    \"pooled_random_split_row_weighted_roc_auc\": float(pooled_row) if not pd.isna(pooled_row) else None,\n    \"leave_one_dataset_out_mean_roc_auc\": float(lodo_mean) if not pd.isna(lodo_mean) else None,\n    \"leave_one_domain_out_mean_roc_auc\": float(domain_out_mean) if not pd.isna(domain_out_mean) else None,\n    \"mean_dataset_transfer_drop\": float(transfer[\"roc_auc_drop\"].mean()) if len(transfer) else None,\n    \"mean_domain_transfer_drop\": float(domain_transfer[\"roc_auc_drop\"].mean()) if len(domain_transfer) else None,\n    \"main_ml_framework\": \"Spark MLlib\",\n    \"big_data_framework\": \"Apache Spark / PySpark\",\n    \"main_model\": \"Class-weighted Spark MLlib logistic regression\",\n}\n\nwith open(f\"{DIRS['results']}/manuscript_summary.json\", \"w\", encoding=\"utf-8\") as f:\n    json.dump(summary, f, indent=2)\n\nsummary_text = f\"\"\"\nSpark-based cross-domain churn benchmark summary\n================================================\n\nThe final benchmark contains {summary['benchmark_datasets']} harmonized public churn datasets from {summary['business_domains']} business domains.\nSpark profiled {summary['source_files_profiled_with_spark']} source files, covering {summary['spark_observed_source_rows']:,} source rows and {summary['source_size_mb']:.1f} MB of source data.\nThe full harmonized registry contains {summary['full_harmonized_rows']:,} rows, and the controlled modeling layer contains {summary['modeling_rows']:,} rows after stratified dataset-level capping.\n\nMain Spark MLlib model:\n- {summary['main_model']}\n\nHeadline ROC-AUC results:\n- Within-dataset mean: {summary['within_dataset_mean_roc_auc']:.6f}\n- Pooled random split, dataset-balanced: {summary['pooled_random_split_dataset_balanced_roc_auc']:.6f}\n- Pooled random split, row-weighted: {summary['pooled_random_split_row_weighted_roc_auc']:.6f}\n- Leave-one-dataset-out mean: {summary['leave_one_dataset_out_mean_roc_auc']:.6f}\n- Leave-one-domain-out mean: {summary['leave_one_domain_out_mean_roc_auc']:.6f}\n\nTransfer degradation:\n- Mean dataset-level transfer drop: {summary['mean_dataset_transfer_drop']:.6f}\n- Mean domain-level transfer drop: {summary['mean_domain_transfer_drop']:.6f}\n\nManuscript claim:\nA Spark-based, multi-source churn benchmark shows that strong within-dataset and pooled performance does not guarantee reliable transportability to unseen datasets or unseen business domains. External validation, label auditing, leakage control, calibration analysis, and business-aware ranking metrics are therefore necessary before deploying churn models in new operational settings.\n\"\"\".strip()\n\nwith open(f\"{DIRS['results']}/manuscript_summary.txt\", \"w\", encoding=\"utf-8\") as f:\n    f.write(summary_text + \"\\n\")\n\nprint(summary_text)\n\n# Full output export can be very large because it includes Parquet files.\n# The paper-only export below excludes harmonized Parquet data and is the file to upload or archive for paper drafting.\npaper_export_dir = Path(\"/kaggle/working/spark_churn_paper_only_results\")\nif paper_export_dir.exists():\n    shutil.rmtree(paper_export_dir)\npaper_export_dir.mkdir(parents=True, exist_ok=True)\n\nfor folder in [\"results\", \"tables\", \"figures\", \"registry\", \"bigdata_profiles\", \"logs\"]:\n    src_dir = Path(OUTPUT_ROOT) / folder\n    dst_dir = paper_export_dir / folder\n    if src_dir.exists():\n        shutil.copytree(\n            src_dir,\n            dst_dir,\n            ignore=shutil.ignore_patterns(\"*.parquet\", \"*.crc\", \"_SUCCESS\", \"part-*\", \"*.snappy\", \"*.bin\")\n        )\n\nzip_path = shutil.make_archive(\"/kaggle/working/spark_churn_paper_only_results\", \"zip\", paper_export_dir)\nprint(\"\\nPaper-only output ZIP:\", zip_path)\n\nprint(\"\\nKey files:\")\nfor p in [\n    f\"{DIRS['results']}/spark_ml_all_results.csv\",\n    f\"{DIRS['results']}/spark_ml_best_results.csv\",\n    f\"{DIRS['results']}/manuscript_headline_results.csv\",\n    f\"{DIRS['results']}/dataset_transfer_degradation.csv\",\n    f\"{DIRS['results']}/domain_transfer_degradation.csv\",\n    f\"{DIRS['results']}/feature_stability_summary.csv\",\n    f\"{DIRS['results']}/statistical_tests.csv\",\n    f\"{DIRS['results']}/manuscript_summary.txt\",\n    f\"{DIRS['figures']}/figure_catalog.csv\",\n    zip_path,\n]:\n    print(\"-\", p)\n","metadata":{"execution":{"iopub.status.busy":"2026-06-28T07:07:37.206324Z","iopub.execute_input":"2026-06-28T07:07:37.207003Z","iopub.status.idle":"2026-06-28T07:07:37.425600Z","shell.execute_reply.started":"2026-06-28T07:07:37.206966Z","shell.execute_reply":"2026-06-28T07:07:37.424801Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Run instructions\n\nRun the notebook from top to bottom in Kaggle after attaching the same public churn datasets used in the previous successful run. The final paper-only result package will be saved as:\n\n```text\n/kaggle/working/spark_churn_paper_only_results.zip\n```\n\nUse the CSV files in `results/`, `tables/`, and `registry/` for manuscript tables. Use the PNG/PDF/SVG files in `figures/` for manuscript figures. The file `figures/figure_catalog.csv` contains suggested figure captions.\n","metadata":{}},{"cell_type":"markdown","source":"","metadata":{}},{"cell_type":"markdown","source":"## Figure 06 alternative: paired horizontal dumbbell plot","metadata":{}},{"cell_type":"code","source":"# ============================================================\n# CELL 6: Clean Figure 6 + auto-update paper-only ZIP\n# Purpose:\n# - Replace overlapping scatter labels with a clean dumbbell plot\n# - Save PNG/PDF/SVG\n# - Overwrite old Figure 6 filenames so LaTeX still works\n# - Recreate spark_churn_paper_only_results.zip automatically\n# ============================================================\n\nimport os\nimport shutil\nfrom pathlib import Path\n\nimport pandas as pd\nimport matplotlib.pyplot as plt\n\n# -----------------------------\n# 1) Paths\n# -----------------------------\nOUTPUT_ROOT = Path(\"/kaggle/working/spark_churn_journal_benchmark\")\nRESULTS_DIR = OUTPUT_ROOT / \"results\"\nFIG_DIR = OUTPUT_ROOT / \"figures\"\nTABLES_DIR = OUTPUT_ROOT / \"tables\"\nREGISTRY_DIR = OUTPUT_ROOT / \"registry\"\nBIGDATA_DIR = OUTPUT_ROOT / \"bigdata_profiles\"\nLOGS_DIR = OUTPUT_ROOT / \"logs\"\n\nFIG_DIR.mkdir(parents=True, exist_ok=True)\n\ndataset_transfer_path = RESULTS_DIR / \"dataset_transfer_degradation.csv\"\n\nif not dataset_transfer_path.exists():\n    raise FileNotFoundError(\n        f\"Missing file: {dataset_transfer_path}\\n\"\n        \"Run the main notebook first until dataset_transfer_degradation.csv is created.\"\n    )\n\n# -----------------------------\n# 2) Load transfer results\n# -----------------------------\ndf = pd.read_csv(dataset_transfer_path)\n\nprint(\"Loaded:\", dataset_transfer_path)\nprint(\"Columns found:\")\nprint(list(df.columns))\ndisplay(df.head())\n\n# -----------------------------\n# 3) Robust column detection\n# -----------------------------\ndataset_candidates = [\n    \"dataset_id\", \"dataset\", \"target\", \"target_dataset\",\n    \"heldout_dataset\", \"held_out_dataset\", \"test_dataset\",\n    \"dataset_name\", \"source_dataset\"\n]\n\nwithin_candidates = [\n    \"within_roc_auc\", \"within_auc\", \"within_dataset_roc_auc\",\n    \"within_dataset_auc\", \"mean_within_roc_auc\"\n]\n\nlodo_candidates = [\n    \"lodo_roc_auc\", \"leave_one_dataset_out_roc_auc\",\n    \"leave_one_dataset_out_auc\", \"lodo_auc\",\n    \"external_roc_auc\", \"transfer_roc_auc\"\n]\n\ndataset_col = next((c for c in dataset_candidates if c in df.columns), None)\nwithin_col = next((c for c in within_candidates if c in df.columns), None)\nlodo_col = next((c for c in lodo_candidates if c in df.columns), None)\n\n# Fallback inference\nif dataset_col is None:\n    possible = [c for c in df.columns if \"dataset\" in c.lower() or \"target\" in c.lower()]\n    dataset_col = possible[0] if possible else None\n\nif within_col is None:\n    possible = [c for c in df.columns if \"within\" in c.lower() and \"auc\" in c.lower()]\n    within_col = possible[0] if possible else None\n\nif lodo_col is None:\n    possible = [\n        c for c in df.columns\n        if (\"lodo\" in c.lower() or \"leave\" in c.lower() or \"external\" in c.lower())\n        and \"auc\" in c.lower()\n    ]\n    lodo_col = possible[0] if possible else None\n\nprint(\"\\nDetected columns:\")\nprint(\"Dataset column:\", dataset_col)\nprint(\"Within column:\", within_col)\nprint(\"LODO column:\", lodo_col)\n\nif dataset_col is None or within_col is None or lodo_col is None:\n    raise ValueError(\n        \"Could not detect required columns automatically. \"\n        \"Check printed column list above.\"\n    )\n\n# -----------------------------\n# 4) Clean dataset labels\n# -----------------------------\nlabel_map = {\n    \"bank_creditcard\": \"Bank credit card\",\n    \"bank_original\": \"Bank original\",\n    \"cell2cell_train_holdout\": \"Cell2Cell telecom\",\n    \"ecommerce_dhairyajeet\": \"E-commerce Dhairyajeet\",\n    \"ecommerce_rees46\": \"E-commerce REES46\",\n    \"ecommerce_shriyash_large\": \"E-commerce Shriyash\",\n    \"insurance_auto\": \"Auto insurance\",\n    \"insurance_k123vinod\": \"K123Vinod insurance\",\n    \"subscription_generic\": \"Generic subscription\",\n    \"subscription_kkbox\": \"KKBox subscription\",\n    \"telco_ibm\": \"IBM Telco\",\n    \"telco_orange\": \"Orange Telecom\",\n}\n\ndf[\"dataset_label\"] = (\n    df[dataset_col]\n    .astype(str)\n    .map(label_map)\n    .fillna(df[dataset_col].astype(str))\n)\n\ndf[within_col] = pd.to_numeric(df[within_col], errors=\"coerce\")\ndf[lodo_col] = pd.to_numeric(df[lodo_col], errors=\"coerce\")\ndf = df.dropna(subset=[within_col, lodo_col]).copy()\n\ndf[\"transfer_drop\"] = df[lodo_col] - df[within_col]\ndf = df.sort_values(\"transfer_drop\", ascending=True).reset_index(drop=True)\n\n# Save label key for paper/supplement\nlabel_key = pd.DataFrame({\n    \"dataset_id\": df[dataset_col].astype(str),\n    \"display_label\": df[\"dataset_label\"]\n})\nlabel_key.to_csv(FIG_DIR / \"figure_06_dataset_label_key.csv\", index=False)\n\n# -----------------------------\n# 5) Plot clean dumbbell figure\n# -----------------------------\nplt.rcParams.update({\n    \"font.size\": 10,\n    \"axes.titlesize\": 12,\n    \"axes.labelsize\": 10,\n    \"xtick.labelsize\": 9,\n    \"ytick.labelsize\": 8.5,\n    \"legend.fontsize\": 9,\n})\n\nfig, ax = plt.subplots(figsize=(9.8, 7.2), dpi=220)\n\ny = list(range(len(df)))\n\n# Connecting lines\nfor i, row in df.iterrows():\n    ax.plot(\n        [row[within_col], row[lodo_col]],\n        [i, i],\n        linewidth=1.4,\n        alpha=0.75\n    )\n\n# Points\nax.scatter(\n    df[within_col],\n    y,\n    s=50,\n    label=\"Within-dataset ROC-AUC\",\n    alpha=0.9,\n    zorder=3\n)\n\nax.scatter(\n    df[lodo_col],\n    y,\n    s=50,\n    label=\"Leave-one-dataset-out ROC-AUC\",\n    alpha=0.9,\n    zorder=3\n)\n\n# Random-ranking reference\nax.axvline(0.5, linestyle=\":\", linewidth=1.3)\n\nax.set_yticks(y)\nax.set_yticklabels(df[\"dataset_label\"])\nax.set_xlabel(\"ROC-AUC\")\nax.set_ylabel(\"Held-out dataset\")\nax.set_title(\"Dataset transfer gap between within-dataset and leave-one-dataset-out testing\")\nax.set_xlim(0.25, 1.0)\nax.grid(axis=\"x\", linestyle=\"--\", linewidth=0.5, alpha=0.45)\n\n# Legend outside bottom to avoid covering graph\nax.legend(\n    loc=\"upper center\",\n    bbox_to_anchor=(0.5, -0.12),\n    ncol=2,\n    frameon=True\n)\n\nfig.tight_layout(rect=[0, 0.06, 1, 1])\n\n# Save with new clean name\nfor ext in [\"png\", \"pdf\", \"svg\"]:\n    fig.savefig(\n        FIG_DIR / f\"figure_06_within_vs_lodo_dumbbell.{ext}\",\n        bbox_inches=\"tight\",\n        dpi=300\n    )\n\n# Overwrite old figure name so existing LaTeX automatically uses clean plot\nfor ext in [\"png\", \"pdf\", \"svg\"]:\n    fig.savefig(\n        FIG_DIR / f\"figure_06_within_vs_lodo_scatter.{ext}\",\n        bbox_inches=\"tight\",\n        dpi=300\n    )\n\nplt.show()\n\nprint(\"\\nSaved clean Figure 6:\")\nprint(FIG_DIR / \"figure_06_within_vs_lodo_dumbbell.png\")\nprint(FIG_DIR / \"figure_06_within_vs_lodo_dumbbell.pdf\")\nprint(FIG_DIR / \"figure_06_within_vs_lodo_dumbbell.svg\")\nprint(\"\\nAlso overwritten old Figure 6 filenames:\")\nprint(FIG_DIR / \"figure_06_within_vs_lodo_scatter.png\")\nprint(FIG_DIR / \"figure_06_within_vs_lodo_scatter.pdf\")\nprint(FIG_DIR / \"figure_06_within_vs_lodo_scatter.svg\")\n\n# -----------------------------\n# 6) Update figure catalog\n# -----------------------------\ncatalog_path = FIG_DIR / \"figure_catalog.csv\"\n\nnew_row = {\n    \"figure\": \"Figure 6\",\n    \"file_png\": \"figure_06_within_vs_lodo_scatter.png\",\n    \"file_pdf\": \"figure_06_within_vs_lodo_scatter.pdf\",\n    \"file_svg\": \"figure_06_within_vs_lodo_scatter.svg\",\n    \"title\": \"Dataset transfer gap between within-dataset and leave-one-dataset-out testing\",\n    \"caption\": (\n        \"Within-dataset and leave-one-dataset-out ROC-AUC by held-out dataset. \"\n        \"Each horizontal segment connects the conventional within-dataset score to the corresponding \"\n        \"external-dataset score. Most segments move leftward under leave-one-dataset-out testing, \"\n        \"showing that local predictive success does not reliably transfer to unseen datasets.\"\n    )\n}\n\nif catalog_path.exists():\n    catalog = pd.read_csv(catalog_path)\n    # Remove older Figure 6 row if present\n    if \"figure\" in catalog.columns:\n        catalog = catalog[catalog[\"figure\"].astype(str) != \"Figure 6\"].copy()\n    catalog = pd.concat([catalog, pd.DataFrame([new_row])], ignore_index=True)\nelse:\n    catalog = pd.DataFrame([new_row])\n\ncatalog.to_csv(catalog_path, index=False)\n\nprint(\"\\nUpdated figure catalog:\")\nprint(catalog_path)\n\n# -----------------------------\n# 7) Recreate small paper-only ZIP automatically\n# -----------------------------\nSMALL_EXPORT = Path(\"/kaggle/working/spark_churn_paper_only_results\")\n\nif SMALL_EXPORT.exists():\n    shutil.rmtree(SMALL_EXPORT)\n\nSMALL_EXPORT.mkdir(parents=True, exist_ok=True)\n\nfolders_to_copy = [\n    \"results\",\n    \"tables\",\n    \"figures\",\n    \"registry\",\n    \"bigdata_profiles\",\n    \"logs\",\n]\n\nfor folder in folders_to_copy:\n    src = OUTPUT_ROOT / folder\n    dst = SMALL_EXPORT / folder\n    if src.exists():\n        shutil.copytree(\n            src,\n            dst,\n            ignore=shutil.ignore_patterns(\n                \"*.parquet\", \"*.crc\", \"_SUCCESS\",\n                \"part-*\", \"*.snappy\", \"*.bin\"\n            )\n        )\n\n# Copy top-level small files if any\nfor ext in [\"*.csv\", \"*.json\", \"*.txt\", \"*.md\"]:\n    for f in OUTPUT_ROOT.glob(ext):\n        shutil.copy2(f, SMALL_EXPORT / f.name)\n\nzip_path = shutil.make_archive(\n    \"/kaggle/working/spark_churn_paper_only_results\",\n    \"zip\",\n    SMALL_EXPORT\n)\n\nprint(\"\\nAuto-updated small paper-only ZIP:\")\nprint(zip_path)\n\nprint(\"\\nFiles inside updated ZIP source folder:\")\ntotal_mb = 0\nfor p in sorted(SMALL_EXPORT.rglob(\"*\")):\n    if p.is_file():\n        mb = p.stat().st_size / (1024 * 1024)\n        total_mb += mb\n        print(f\"{mb:8.3f} MB | {p}\")\n\nprint(f\"\\nTotal small export size: {total_mb:.2f} MB\")\nprint(\"\\nUse this updated zip:\")\nprint(\"/kaggle/working/spark_churn_paper_only_results.zip\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-28T07:29:11.224210Z","iopub.execute_input":"2026-06-28T07:29:11.224504Z","iopub.status.idle":"2026-06-28T07:29:13.179099Z","shell.execute_reply.started":"2026-06-28T07:29:11.224478Z","shell.execute_reply":"2026-06-28T07:29:13.178217Z"}},"outputs":[],"execution_count":null}]}