From 8ca289d264ef4b94deb1968f81353bff73f8a00b Mon Sep 17 00:00:00 2001 From: Ben Constable Date: Wed, 15 Jul 2026 20:04:14 +0200 Subject: [PATCH 1/7] feat(eval): use a proper, deterministic hold-out split for evaluation The evaluation "holdout" was an independent random sample of the same gold table (orderBy(rand(7)).limit) with a different seed than training. Different seeds do not guarantee disjoint rows, so training data leaked into evaluation and the ROC AUC was optimistic. Implement a proper hold-out following Databricks best practice: - Feature engineering materialises a deterministic train/validation/test split as a Delta table (transactions_split), assigning each transaction by pmod(xxhash64(transaction_id), 100): ~70/15/15. Same id -> same split on every run and in every notebook, so the split is reproducible, leakage-free, versioned. - Training reads only the train + validation rows (join on the split table), disjoint from the test holdout. - Evaluation reads only the test split, ordered deterministically so the candidate and champion are scored on the identical rows. Notebook notes cover the production hardening: hash the entity id to avoid entity leakage, a time-based holdout for temporal fraud data, and point-in-time feature lookups to prevent feature leakage. --- .../model_deployment/notebooks/evaluation.py | 22 +++++++----- .../notebooks/build_features.py | 36 +++++++++++++++++++ solution/training/notebooks/training.py | 16 +++++++-- .../model_deployment/notebooks/evaluation.py | 22 +++++++----- .../notebooks/build_features.py | 36 +++++++++++++++++++ src/training/notebooks/training.py | 16 +++++++-- 6 files changed, 126 insertions(+), 22 deletions(-) diff --git a/solution/deployment/model_deployment/notebooks/evaluation.py b/solution/deployment/model_deployment/notebooks/evaluation.py index db20f06..987c591 100644 --- a/solution/deployment/model_deployment/notebooks/evaluation.py +++ b/solution/deployment/model_deployment/notebooks/evaluation.py @@ -105,16 +105,24 @@ client = MlflowClient() fe = FeatureEngineeringClient() -# Independent evaluation: score the REGISTERED artifact on a fresh, labelled holdout rather -# than trusting the metrics training self-reported. fe.score_batch replays the exact feature -# lookups + on-demand functions recorded at training, and the served model returns a fraud -# probability, so a real ROC AUC can be computed. In production this holdout would be a -# curated, leakage-controlled evaluation table; here it is sampled from the shared gold source. +# Independent evaluation: score the REGISTERED artifact on the held-out `test` split rather than +# trusting the metrics training self-reported. fe.score_batch replays the exact feature lookups + +# on-demand functions recorded at training, and the served model returns a fraud probability, so a +# real ROC AUC can be computed. The test split is the persisted, deterministic holdout that feature +# engineering wrote; it is GUARANTEED disjoint from the rows training used, so this metric is not +# inflated by train/test leakage. gold_table = f"{catalog_name}.fraud_gold.transactions_enriched" +split_table = f"{catalog_name}.{ml_schema}.transactions_split" + +test_ids = ( + spark.read.table(split_table).where(F.col("split") == "test").select("transaction_id") +) eval_spine = ( spark.read.table(gold_table) + .join(test_ids, "transaction_id") # the held-out test split ONLY .select( + "transaction_id", "card_id", "client_id", F.col("amount").cast("double").alias("amount"), @@ -124,11 +132,9 @@ F.col("is_fraud").cast("int").alias("is_fraud"), ) .where(F.col("is_fraud").isNotNull()) - .orderBy(F.rand(7)) # different seed than training, so mostly-unseen rows + .orderBy(F.xxhash64("transaction_id")) # deterministic order so candidate and champion score the SAME rows .limit(50000) ) -# Note: no .cache(), because serverless compute rejects PERSIST. The seeded rand(7) sample is -# deterministic over the immutable gold table, so candidate and champion score the same rows. def score_holdout(version): diff --git a/solution/feature_engineering/notebooks/build_features.py b/solution/feature_engineering/notebooks/build_features.py index 4f97545..3e0148a 100644 --- a/solution/feature_engineering/notebooks/build_features.py +++ b/solution/feature_engineering/notebooks/build_features.py @@ -141,6 +141,42 @@ def publish(name, df, key, description): # COMMAND ---------- +# MAGIC %md +# MAGIC ## 3. Materialise the train / validation / test split +# MAGIC +# MAGIC A proper evaluation needs a hold-out set that is GUARANTEED disjoint from training, and the +# MAGIC same rows on every run. Sampling each notebook independently (even with different seeds) +# MAGIC does not guarantee that: two random samples of the same table overlap. Instead we assign +# MAGIC every transaction to one split by hashing its stable id, and persist the assignment as a +# MAGIC Delta table. Training reads `train`/`validation`; evaluation reads `test`. Because the hash +# MAGIC is deterministic the split is reproducible and leakage-free, and the table is versioned so a +# MAGIC model version is always scored on the identical holdout. +# MAGIC +# MAGIC Production hardening: hashing `transaction_id` is a row-level split. To avoid *entity* +# MAGIC leakage (the same card in both train and test) hash `card_id`/`client_id` instead; and for a +# MAGIC temporal problem like fraud, prefer a time-based holdout (train on older data, test on the +# MAGIC most recent window). A uniform hash keeps the rare fraud rate roughly equal across splits; +# MAGIC stratify by the label if you need it exact. + +# COMMAND ---------- + +split_table = f"{catalog_name}.{ml_schema}.transactions_split" + +# Deterministic bucket 0-99 from a stable key: the same id always lands in the same split, on +# every run and in every notebook, so train and test can never overlap. +bucket = F.pmod(F.xxhash64("transaction_id"), F.lit(100)) +splits = enriched.select( + "transaction_id", + F.when(bucket < 70, "train") + .when(bucket < 85, "validation") + .otherwise("test") + .alias("split"), +) +splits.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(split_table) +print(f"Wrote {split_table}: ~70% train / 15% validation / 15% test (deterministic on transaction_id).") + +# COMMAND ---------- + # MAGIC %md # MAGIC ## 3. Register the on-demand feature functions # MAGIC diff --git a/solution/training/notebooks/training.py b/solution/training/notebooks/training.py index 9ada6f3..30e6b68 100644 --- a/solution/training/notebooks/training.py +++ b/solution/training/notebooks/training.py @@ -159,13 +159,23 @@ schema_fqn = f"{catalog_name}.{ml_schema}" -# Cap the training data at a random sample so the notebook runs in a couple of minutes. -# Fraud is extremely rare, so a random sample (not the first N rows) is what keeps enough -# fraud examples in the set; the seed makes it reproducible run-to-run. +# Train on the train + validation splits only, read from the shared split table that feature +# engineering materialised. This guarantees the rows are disjoint from the `test` holdout that +# evaluation scores, so the evaluation metric is honest (no train/test leakage). Still cap at a +# random sample so the notebook runs in a couple of minutes; fraud is extremely rare, so a random +# sample (not the first N rows) keeps enough fraud examples, and the seed makes it reproducible. SAMPLE_ROWS = 200000 +split_table = f"{catalog_name}.{ml_schema}.transactions_split" +train_ids = ( + spark.read.table(split_table) + .where(F.col("split").isin("train", "validation")) + .select("transaction_id") +) + spine = ( spark.read.table(source_table) + .join(train_ids, "transaction_id") # keep only train/validation rows -> disjoint from test .select( "transaction_id", "card_id", diff --git a/src/deployment/model_deployment/notebooks/evaluation.py b/src/deployment/model_deployment/notebooks/evaluation.py index db20f06..987c591 100644 --- a/src/deployment/model_deployment/notebooks/evaluation.py +++ b/src/deployment/model_deployment/notebooks/evaluation.py @@ -105,16 +105,24 @@ client = MlflowClient() fe = FeatureEngineeringClient() -# Independent evaluation: score the REGISTERED artifact on a fresh, labelled holdout rather -# than trusting the metrics training self-reported. fe.score_batch replays the exact feature -# lookups + on-demand functions recorded at training, and the served model returns a fraud -# probability, so a real ROC AUC can be computed. In production this holdout would be a -# curated, leakage-controlled evaluation table; here it is sampled from the shared gold source. +# Independent evaluation: score the REGISTERED artifact on the held-out `test` split rather than +# trusting the metrics training self-reported. fe.score_batch replays the exact feature lookups + +# on-demand functions recorded at training, and the served model returns a fraud probability, so a +# real ROC AUC can be computed. The test split is the persisted, deterministic holdout that feature +# engineering wrote; it is GUARANTEED disjoint from the rows training used, so this metric is not +# inflated by train/test leakage. gold_table = f"{catalog_name}.fraud_gold.transactions_enriched" +split_table = f"{catalog_name}.{ml_schema}.transactions_split" + +test_ids = ( + spark.read.table(split_table).where(F.col("split") == "test").select("transaction_id") +) eval_spine = ( spark.read.table(gold_table) + .join(test_ids, "transaction_id") # the held-out test split ONLY .select( + "transaction_id", "card_id", "client_id", F.col("amount").cast("double").alias("amount"), @@ -124,11 +132,9 @@ F.col("is_fraud").cast("int").alias("is_fraud"), ) .where(F.col("is_fraud").isNotNull()) - .orderBy(F.rand(7)) # different seed than training, so mostly-unseen rows + .orderBy(F.xxhash64("transaction_id")) # deterministic order so candidate and champion score the SAME rows .limit(50000) ) -# Note: no .cache(), because serverless compute rejects PERSIST. The seeded rand(7) sample is -# deterministic over the immutable gold table, so candidate and champion score the same rows. def score_holdout(version): diff --git a/src/feature_engineering/notebooks/build_features.py b/src/feature_engineering/notebooks/build_features.py index 4d05ede..e7ab8a6 100644 --- a/src/feature_engineering/notebooks/build_features.py +++ b/src/feature_engineering/notebooks/build_features.py @@ -128,6 +128,42 @@ # COMMAND ---------- +# MAGIC %md +# MAGIC ## 3. Materialise the train / validation / test split +# MAGIC +# MAGIC A proper evaluation needs a hold-out set that is GUARANTEED disjoint from training, and the +# MAGIC same rows on every run. Sampling each notebook independently (even with different seeds) +# MAGIC does not guarantee that: two random samples of the same table overlap. Instead we assign +# MAGIC every transaction to one split by hashing its stable id, and persist the assignment as a +# MAGIC Delta table. Training reads `train`/`validation`; evaluation reads `test`. Because the hash +# MAGIC is deterministic the split is reproducible and leakage-free, and the table is versioned so a +# MAGIC model version is always scored on the identical holdout. +# MAGIC +# MAGIC Production hardening: hashing `transaction_id` is a row-level split. To avoid *entity* +# MAGIC leakage (the same card in both train and test) hash `card_id`/`client_id` instead; and for a +# MAGIC temporal problem like fraud, prefer a time-based holdout (train on older data, test on the +# MAGIC most recent window). A uniform hash keeps the rare fraud rate roughly equal across splits; +# MAGIC stratify by the label if you need it exact. + +# COMMAND ---------- + +split_table = f"{catalog_name}.{ml_schema}.transactions_split" + +# Deterministic bucket 0-99 from a stable key: the same id always lands in the same split, on +# every run and in every notebook, so train and test can never overlap. +bucket = F.pmod(F.xxhash64("transaction_id"), F.lit(100)) +splits = enriched.select( + "transaction_id", + F.when(bucket < 70, "train") + .when(bucket < 85, "validation") + .otherwise("test") + .alias("split"), +) +splits.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(split_table) +print(f"Wrote {split_table}: ~70% train / 15% validation / 15% test (deterministic on transaction_id).") + +# COMMAND ---------- + # MAGIC %md # MAGIC ## 3. Register the on-demand feature functions # MAGIC diff --git a/src/training/notebooks/training.py b/src/training/notebooks/training.py index 81cf075..c22e53b 100644 --- a/src/training/notebooks/training.py +++ b/src/training/notebooks/training.py @@ -159,13 +159,23 @@ schema_fqn = f"{catalog_name}.{ml_schema}" -# Cap the training data at a random sample so the notebook runs in a couple of minutes. -# Fraud is extremely rare, so a random sample (not the first N rows) is what keeps enough -# fraud examples in the set; the seed makes it reproducible run-to-run. +# Train on the train + validation splits only, read from the shared split table that feature +# engineering materialised. This guarantees the rows are disjoint from the `test` holdout that +# evaluation scores, so the evaluation metric is honest (no train/test leakage). Still cap at a +# random sample so the notebook runs in a couple of minutes; fraud is extremely rare, so a random +# sample (not the first N rows) keeps enough fraud examples, and the seed makes it reproducible. SAMPLE_ROWS = 200000 +split_table = f"{catalog_name}.{ml_schema}.transactions_split" +train_ids = ( + spark.read.table(split_table) + .where(F.col("split").isin("train", "validation")) + .select("transaction_id") +) + spine = ( spark.read.table(source_table) + .join(train_ids, "transaction_id") # keep only train/validation rows -> disjoint from test .select( "transaction_id", "card_id", From 13b3fe2f52abeae48c135721d2582b834f473535 Mon Sep 17 00:00:00 2001 From: Ben Constable Date: Wed, 15 Jul 2026 20:17:07 +0200 Subject: [PATCH 2/7] feat(eval): group split by customer + stratify by fraud; grouped k-fold in training Make the hold-out representative and leakage-free for an imbalanced, multi-row-per -customer problem: - Split is now GROUPED BY CUSTOMER and STRATIFIED BY FRAUD. Feature engineering assigns whole clients to train/validation/test (never a single transaction), so the same customer can never appear in both train and test (no entity leakage). Assignment uses percent_rank over a hash of client_id WITHIN each fraud/non-fraud stratum, so the fraud rate is identical across splits (representative) and the result is deterministic and versioned. - Training replaces the single train_test_split with StratifiedGroupKFold (n_splits=5, grouped by client_id): each fold stratifies by the rare label and keeps a client in one fold, so no customer is in both the fit and validation of a fold. Metrics are averaged across folds (honest estimate); the registered model is then fit on all train+validation rows. client_id is kept as the CV group and the pyfunc wrapper selects the feature columns so it is not used as a model feature. --- .../notebooks/build_features.py | 59 ++++++---- solution/training/notebooks/training.py | 105 +++++++++++------- .../notebooks/build_features.py | 59 ++++++---- src/training/notebooks/training.py | 101 ++++++++++------- 4 files changed, 200 insertions(+), 124 deletions(-) diff --git a/solution/feature_engineering/notebooks/build_features.py b/solution/feature_engineering/notebooks/build_features.py index 3e0148a..cd7194e 100644 --- a/solution/feature_engineering/notebooks/build_features.py +++ b/solution/feature_engineering/notebooks/build_features.py @@ -144,36 +144,55 @@ def publish(name, df, key, description): # MAGIC %md # MAGIC ## 3. Materialise the train / validation / test split # MAGIC -# MAGIC A proper evaluation needs a hold-out set that is GUARANTEED disjoint from training, and the -# MAGIC same rows on every run. Sampling each notebook independently (even with different seeds) -# MAGIC does not guarantee that: two random samples of the same table overlap. Instead we assign -# MAGIC every transaction to one split by hashing its stable id, and persist the assignment as a -# MAGIC Delta table. Training reads `train`/`validation`; evaluation reads `test`. Because the hash -# MAGIC is deterministic the split is reproducible and leakage-free, and the table is versioned so a -# MAGIC model version is always scored on the identical holdout. +# MAGIC A proper evaluation needs a hold-out set that is representative and GUARANTEED disjoint from +# MAGIC training, on the same rows every run. Two properties matter for fraud: # MAGIC -# MAGIC Production hardening: hashing `transaction_id` is a row-level split. To avoid *entity* -# MAGIC leakage (the same card in both train and test) hash `card_id`/`client_id` instead; and for a -# MAGIC temporal problem like fraud, prefer a time-based holdout (train on older data, test on the -# MAGIC most recent window). A uniform hash keeps the rare fraud rate roughly equal across splits; -# MAGIC stratify by the label if you need it exact. +# MAGIC - **Grouped by customer.** Whole clients are assigned to one split, so the same customer is +# MAGIC never in both train and test. A row-level split would let the model memorise a client seen +# MAGIC in training and then be graded on that same client (entity leakage -> optimistic metrics). +# MAGIC - **Stratified by the label.** Fraud is well under 1%, so we assign clients to 70/15/15 +# MAGIC *within* each fraud / non-fraud stratum, keeping the fraud rate identical across splits so +# MAGIC the test set is representative and has enough positives for a stable metric. +# MAGIC +# MAGIC The assignment is a deterministic rank over a hash of `client_id`, so it is reproducible and +# MAGIC persisted as a versioned Delta table; a model version is always scored on the identical +# MAGIC holdout. Production hardening: for temporal data a time-based holdout (train older, test the +# MAGIC most recent window) is stronger still, and point-in-time feature lookups prevent feature +# MAGIC leakage. # COMMAND ---------- split_table = f"{catalog_name}.{ml_schema}.transactions_split" -# Deterministic bucket 0-99 from a stable key: the same id always lands in the same split, on -# every run and in every notebook, so train and test can never overlap. -bucket = F.pmod(F.xxhash64("transaction_id"), F.lit(100)) -splits = enriched.select( - "transaction_id", - F.when(bucket < 70, "train") - .when(bucket < 85, "validation") +# Group by CUSTOMER and stratify by the (rare) fraud label. Assign whole clients (not rows) to a +# split so no customer ever appears in both train and test. Within each fraud / non-fraud stratum +# rank clients by a hash of client_id and cut at 70/15/15, so every client lands in exactly one +# split (grouped), the fraud-client proportion is identical across splits (stratified), and the +# assignment is deterministic (hash order) and versioned as a Delta table. +from pyspark.sql import Window + +client_label = enriched.groupBy("client_id").agg( + F.max("is_fraud").cast("int").alias("client_has_fraud") +) +strata = Window.partitionBy("client_has_fraud").orderBy(F.xxhash64("client_id")) +client_split = client_label.select( + "client_id", + F.percent_rank().over(strata).alias("pct"), +).select( + "client_id", + F.when(F.col("pct") < 0.70, "train") + .when(F.col("pct") < 0.85, "validation") .otherwise("test") .alias("split"), ) + +splits = ( + enriched.select("transaction_id", "client_id") + .join(client_split, "client_id") + .select("transaction_id", "split") +) splits.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(split_table) -print(f"Wrote {split_table}: ~70% train / 15% validation / 15% test (deterministic on transaction_id).") +print(f"Wrote {split_table}: 70/15/15 train/validation/test, grouped by client, stratified by fraud.") # COMMAND ---------- diff --git a/solution/training/notebooks/training.py b/solution/training/notebooks/training.py index 30e6b68..7c056ec 100644 --- a/solution/training/notebooks/training.py +++ b/solution/training/notebooks/training.py @@ -89,6 +89,7 @@ # COMMAND ---------- import mlflow +import numpy as np from databricks.feature_engineering import ( FeatureEngineeringClient, FeatureFunction, @@ -104,7 +105,7 @@ recall_score, roc_auc_score, ) -from sklearn.model_selection import train_test_split +from sklearn.model_selection import StratifiedGroupKFold # MLflow 3: register to the Unity Catalog model registry (the default in MLflow 3, set # explicitly here so the notebook behaves the same on any runtime). @@ -228,8 +229,9 @@ # HINT: (client_feature_table, lookup_key="client_id", credit_score/yearly_income/current_age). # HINT: fe.create_training_set(df=spine, feature_lookups=feature_lookups + feature_functions, # HINT: label="is_fraud", exclude_columns=[...]). -# HINT: exclude the keys and raw function inputs so the model only sees features: -# HINT: transaction_id, card_id, client_id, mcc, use_chip. +# HINT: exclude the keys and raw function inputs so the model only sees features, but KEEP +# HINT: client_id (it is the cross-validation group; dropped from X before fitting): +# HINT: transaction_id, card_id, mcc, use_chip. feature_lookups = [ FeatureLookup( table_name=card_feature_table, @@ -247,20 +249,19 @@ df=spine, feature_lookups=feature_lookups + feature_functions, label="is_fraud", - exclude_columns=["transaction_id", "card_id", "client_id", "mcc", "use_chip"], + exclude_columns=["transaction_id", "card_id", "mcc", "use_chip"], ) # TODO-END -# Load into pandas and do a standard stratified train/test split (provided: this is ordinary -# scikit-learn, not the MLOps concept this lab is about). +# Load into pandas. Keep client_id as the CROSS-VALIDATION GROUP (dropped from the model +# features below); is_fraud is the label. train_pdf holds only the train+validation splits (the +# join upstream), so the test holdout is never touched here. train_pdf = training_set.load_df().toPandas() -X = train_pdf.drop(columns=["is_fraud"]) +groups = train_pdf["client_id"] # customer id: keeps a client within a single CV fold y = train_pdf["is_fraud"] - -X_train, X_test, y_train, y_test = train_test_split( - X, y, test_size=0.2, stratify=y, random_state=42 -) -print(f"Rows: {len(train_pdf):,} | features: {list(X.columns)} | fraud rate: {y.mean():.4f}") +FEATURE_COLS = [c for c in train_pdf.columns if c not in ("is_fraud", "client_id")] +X = train_pdf[FEATURE_COLS] +print(f"Rows: {len(train_pdf):,} | features: {FEATURE_COLS} | fraud rate: {y.mean():.4f}") # COMMAND ---------- @@ -273,8 +274,10 @@ # MAGIC - Train a baseline random forest on the raw, imbalanced data to establish the problem. # MAGIC - Retrain on balanced data (retain every fraud row, down-sample non-fraud to match) so # MAGIC the model learns the minority class. -# MAGIC - Evaluate on the original, imbalanced test set so the reported metrics reflect -# MAGIC production conditions. +# MAGIC - Estimate generalisation with Stratified GROUP k-fold cross-validation: each fold +# MAGIC stratifies by the rare fraud label and keeps every client in one fold, so no customer +# MAGIC is in both the fit and the validation of a fold. Metrics are averaged across folds and +# MAGIC scored on the untouched (imbalanced) validation side, so they reflect production. # MAGIC # MAGIC `fe.log_model` packages the model with its feature metadata, infers a signature and # MAGIC input example, and registers it to Unity Catalog in a single step (no separate @@ -303,44 +306,60 @@ class FraudProbabilityModel(mlflow.pyfunc.PythonModel): the classifier in a pyfunc keeps fe.log_model's feature resolution intact. """ - def __init__(self, model): + def __init__(self, model, feature_cols): self.model = model + self.feature_cols = feature_cols def predict(self, context, model_input): - return self.model.predict_proba(model_input)[:, 1] + # The serving spine also carries client_id (the lookup key / CV group), which is not a + # model feature, so select the feature columns before predicting. + return self.model.predict_proba(model_input[self.feature_cols])[:, 1] + + +def balance(X_fold, y_fold): + """Down-sample non-fraud to 1:1 within a fold (keep every fraud row).""" + fraud_idx = y_fold[y_fold == 1].index + legit_idx = y_fold[y_fold == 0].sample(n=len(fraud_idx), random_state=42).index + idx = fraud_idx.union(legit_idx) + return X_fold.loc[idx], y_fold.loc[idx] + +PARAMS = {"n_estimators": 100, "random_state": 42} -# Naive random forest (instructor-provided): the imbalance trap. Trained on the raw, -# highly-imbalanced data it just learns to predict the majority class, so accuracy looks -# great while recall is ~0 (it never flags a fraud). Logged for comparison, NOT -# registered. +# Naive baseline (instructor-provided): fit on the raw, imbalanced data with NO balancing to +# show the imbalance trap: near-perfect accuracy but ~0 recall (it never flags a fraud). Scored +# on one grouped fold for illustration; logged for comparison, NOT registered. with mlflow.start_run(run_name="rf_imbalanced"): - naive_params = {"n_estimators": 100, "random_state": 42} - naive = RandomForestClassifier(**naive_params) - naive.fit(X_train, y_train) - naive_metrics = evaluate(naive, X_test, y_test) - mlflow.log_params({**naive_params, "training_data": "imbalanced"}) + tr, va = next(StratifiedGroupKFold(n_splits=5).split(X, y, groups)) + naive = RandomForestClassifier(**PARAMS).fit(X.iloc[tr], y.iloc[tr]) + naive_metrics = evaluate(naive, X.iloc[va], y.iloc[va]) + mlflow.log_params({**PARAMS, "training_data": "imbalanced"}) mlflow.log_metrics(naive_metrics) print("rf_imbalanced:", naive_metrics) # high accuracy, low recall -# Fixed random forest: the model that is tracked and registered. +# Registered model: Stratified GROUP k-fold cross-validation. Each fold stratifies by the rare +# fraud label AND keeps every client in one fold, so no customer is ever in both the fit and the +# validation of a fold (no entity leakage in the estimate). Balance the fit side of each fold, +# score the untouched validation side, and average -> an honest generalisation estimate. The +# registered artifact is then fit on ALL train+validation rows (balanced). with mlflow.start_run(run_name="rf_balanced") as run: - # Balance the training rows: keep every fraud row and randomly sample an equal number of - # non-fraud rows (provided: this is a data-science fix, not the MLOps concept). Evaluation - # still uses the untouched, imbalanced X_test/y_test so metrics reflect reality. - fraud_idx = y_train[y_train == 1].index - legit_idx = y_train[y_train == 0].sample(n=len(fraud_idx), random_state=42).index - bal_idx = fraud_idx.union(legit_idx) - X_bal, y_bal = X_train.loc[bal_idx], y_train.loc[bal_idx] - - params = {"n_estimators": 100, "random_state": 42} - model = RandomForestClassifier(**params) - model.fit(X_bal, y_bal) - metrics = evaluate(model, X_test, y_test) - - mlflow.log_params(params) + sgkf = StratifiedGroupKFold(n_splits=5, shuffle=True, random_state=42) + fold_metrics = [] + for tr, va in sgkf.split(X, y, groups): + X_bal, y_bal = balance(X.iloc[tr], y.iloc[tr]) + clf = RandomForestClassifier(**PARAMS).fit(X_bal, y_bal) + fold_metrics.append(evaluate(clf, X.iloc[va], y.iloc[va])) + # Cross-validated metrics = mean across folds (what we report and gate on). + metrics = {k: float(np.mean([m[k] for m in fold_metrics])) for k in fold_metrics[0]} + + # Final registered model: fit on ALL train+validation rows (balanced). + X_all_bal, y_all_bal = balance(X, y) + model = RandomForestClassifier(**PARAMS).fit(X_all_bal, y_all_bal) + + mlflow.log_params(PARAMS) mlflow.log_param("training_data", "balanced (1:1 down-sampled)") - mlflow.log_param("train_rows", int(len(y_bal))) + mlflow.log_param("cv", "StratifiedGroupKFold(n_splits=5) grouped by client_id") + mlflow.log_param("train_rows", int(len(y_all_bal))) mlflow.log_param("training_table", source_table) mlflow.log_param("card_feature_table", card_feature_table) mlflow.log_param("client_feature_table", client_feature_table) @@ -352,11 +371,11 @@ def predict(self, context, model_input): # TODO-BEGIN: log and register the model to Unity Catalog with its feature metadata # HINT: fe.log_model packages the model with the training_set's feature lookups (so serving # HINT: reproduces the same features) and registers it to UC in a single call. - # HINT: fe.log_model(model=FraudProbabilityModel(model), artifact_path="model", + # HINT: fe.log_model(model=FraudProbabilityModel(model, FEATURE_COLS), artifact_path="model", # HINT: flavor=mlflow.pyfunc, training_set=training_set, # HINT: registered_model_name=uc_model_name, infer_input_example=True) fe.log_model( - model=FraudProbabilityModel(model), + model=FraudProbabilityModel(model, FEATURE_COLS), artifact_path="model", flavor=mlflow.pyfunc, training_set=training_set, diff --git a/src/feature_engineering/notebooks/build_features.py b/src/feature_engineering/notebooks/build_features.py index e7ab8a6..6f8315b 100644 --- a/src/feature_engineering/notebooks/build_features.py +++ b/src/feature_engineering/notebooks/build_features.py @@ -131,36 +131,55 @@ # MAGIC %md # MAGIC ## 3. Materialise the train / validation / test split # MAGIC -# MAGIC A proper evaluation needs a hold-out set that is GUARANTEED disjoint from training, and the -# MAGIC same rows on every run. Sampling each notebook independently (even with different seeds) -# MAGIC does not guarantee that: two random samples of the same table overlap. Instead we assign -# MAGIC every transaction to one split by hashing its stable id, and persist the assignment as a -# MAGIC Delta table. Training reads `train`/`validation`; evaluation reads `test`. Because the hash -# MAGIC is deterministic the split is reproducible and leakage-free, and the table is versioned so a -# MAGIC model version is always scored on the identical holdout. +# MAGIC A proper evaluation needs a hold-out set that is representative and GUARANTEED disjoint from +# MAGIC training, on the same rows every run. Two properties matter for fraud: # MAGIC -# MAGIC Production hardening: hashing `transaction_id` is a row-level split. To avoid *entity* -# MAGIC leakage (the same card in both train and test) hash `card_id`/`client_id` instead; and for a -# MAGIC temporal problem like fraud, prefer a time-based holdout (train on older data, test on the -# MAGIC most recent window). A uniform hash keeps the rare fraud rate roughly equal across splits; -# MAGIC stratify by the label if you need it exact. +# MAGIC - **Grouped by customer.** Whole clients are assigned to one split, so the same customer is +# MAGIC never in both train and test. A row-level split would let the model memorise a client seen +# MAGIC in training and then be graded on that same client (entity leakage -> optimistic metrics). +# MAGIC - **Stratified by the label.** Fraud is well under 1%, so we assign clients to 70/15/15 +# MAGIC *within* each fraud / non-fraud stratum, keeping the fraud rate identical across splits so +# MAGIC the test set is representative and has enough positives for a stable metric. +# MAGIC +# MAGIC The assignment is a deterministic rank over a hash of `client_id`, so it is reproducible and +# MAGIC persisted as a versioned Delta table; a model version is always scored on the identical +# MAGIC holdout. Production hardening: for temporal data a time-based holdout (train older, test the +# MAGIC most recent window) is stronger still, and point-in-time feature lookups prevent feature +# MAGIC leakage. # COMMAND ---------- split_table = f"{catalog_name}.{ml_schema}.transactions_split" -# Deterministic bucket 0-99 from a stable key: the same id always lands in the same split, on -# every run and in every notebook, so train and test can never overlap. -bucket = F.pmod(F.xxhash64("transaction_id"), F.lit(100)) -splits = enriched.select( - "transaction_id", - F.when(bucket < 70, "train") - .when(bucket < 85, "validation") +# Group by CUSTOMER and stratify by the (rare) fraud label. Assign whole clients (not rows) to a +# split so no customer ever appears in both train and test. Within each fraud / non-fraud stratum +# rank clients by a hash of client_id and cut at 70/15/15, so every client lands in exactly one +# split (grouped), the fraud-client proportion is identical across splits (stratified), and the +# assignment is deterministic (hash order) and versioned as a Delta table. +from pyspark.sql import Window + +client_label = enriched.groupBy("client_id").agg( + F.max("is_fraud").cast("int").alias("client_has_fraud") +) +strata = Window.partitionBy("client_has_fraud").orderBy(F.xxhash64("client_id")) +client_split = client_label.select( + "client_id", + F.percent_rank().over(strata).alias("pct"), +).select( + "client_id", + F.when(F.col("pct") < 0.70, "train") + .when(F.col("pct") < 0.85, "validation") .otherwise("test") .alias("split"), ) + +splits = ( + enriched.select("transaction_id", "client_id") + .join(client_split, "client_id") + .select("transaction_id", "split") +) splits.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(split_table) -print(f"Wrote {split_table}: ~70% train / 15% validation / 15% test (deterministic on transaction_id).") +print(f"Wrote {split_table}: 70/15/15 train/validation/test, grouped by client, stratified by fraud.") # COMMAND ---------- diff --git a/src/training/notebooks/training.py b/src/training/notebooks/training.py index c22e53b..51147c9 100644 --- a/src/training/notebooks/training.py +++ b/src/training/notebooks/training.py @@ -89,6 +89,7 @@ # COMMAND ---------- import mlflow +import numpy as np from databricks.feature_engineering import ( FeatureEngineeringClient, FeatureFunction, @@ -104,7 +105,7 @@ recall_score, roc_auc_score, ) -from sklearn.model_selection import train_test_split +from sklearn.model_selection import StratifiedGroupKFold # MLflow 3: register to the Unity Catalog model registry (the default in MLflow 3, set # explicitly here so the notebook behaves the same on any runtime). @@ -229,21 +230,21 @@ # HINT: (client_feature_table, lookup_key="client_id", credit_score/yearly_income/current_age). # HINT: fe.create_training_set(df=spine, feature_lookups=feature_lookups + feature_functions, # HINT: label="is_fraud", exclude_columns=[...]). -# HINT: exclude the keys and raw function inputs so the model only sees features: -# HINT: transaction_id, card_id, client_id, mcc, use_chip. +# HINT: exclude the keys and raw function inputs so the model only sees features, but KEEP +# HINT: client_id (it is the cross-validation group; dropped from X before fitting): +# HINT: transaction_id, card_id, mcc, use_chip. # <-- Your code here # ---------------------------------------------- -# Load into pandas and do a standard stratified train/test split (provided: this is ordinary -# scikit-learn, not the MLOps concept this lab is about). +# Load into pandas. Keep client_id as the CROSS-VALIDATION GROUP (dropped from the model +# features below); is_fraud is the label. train_pdf holds only the train+validation splits (the +# join upstream), so the test holdout is never touched here. train_pdf = training_set.load_df().toPandas() -X = train_pdf.drop(columns=["is_fraud"]) +groups = train_pdf["client_id"] # customer id: keeps a client within a single CV fold y = train_pdf["is_fraud"] - -X_train, X_test, y_train, y_test = train_test_split( - X, y, test_size=0.2, stratify=y, random_state=42 -) -print(f"Rows: {len(train_pdf):,} | features: {list(X.columns)} | fraud rate: {y.mean():.4f}") +FEATURE_COLS = [c for c in train_pdf.columns if c not in ("is_fraud", "client_id")] +X = train_pdf[FEATURE_COLS] +print(f"Rows: {len(train_pdf):,} | features: {FEATURE_COLS} | fraud rate: {y.mean():.4f}") # COMMAND ---------- @@ -256,8 +257,10 @@ # MAGIC - Train a baseline random forest on the raw, imbalanced data to establish the problem. # MAGIC - Retrain on balanced data (retain every fraud row, down-sample non-fraud to match) so # MAGIC the model learns the minority class. -# MAGIC - Evaluate on the original, imbalanced test set so the reported metrics reflect -# MAGIC production conditions. +# MAGIC - Estimate generalisation with Stratified GROUP k-fold cross-validation: each fold +# MAGIC stratifies by the rare fraud label and keeps every client in one fold, so no customer +# MAGIC is in both the fit and the validation of a fold. Metrics are averaged across folds and +# MAGIC scored on the untouched (imbalanced) validation side, so they reflect production. # MAGIC # MAGIC `fe.log_model` packages the model with its feature metadata, infers a signature and # MAGIC input example, and registers it to Unity Catalog in a single step (no separate @@ -286,44 +289,60 @@ class FraudProbabilityModel(mlflow.pyfunc.PythonModel): the classifier in a pyfunc keeps fe.log_model's feature resolution intact. """ - def __init__(self, model): + def __init__(self, model, feature_cols): self.model = model + self.feature_cols = feature_cols def predict(self, context, model_input): - return self.model.predict_proba(model_input)[:, 1] + # The serving spine also carries client_id (the lookup key / CV group), which is not a + # model feature, so select the feature columns before predicting. + return self.model.predict_proba(model_input[self.feature_cols])[:, 1] + + +def balance(X_fold, y_fold): + """Down-sample non-fraud to 1:1 within a fold (keep every fraud row).""" + fraud_idx = y_fold[y_fold == 1].index + legit_idx = y_fold[y_fold == 0].sample(n=len(fraud_idx), random_state=42).index + idx = fraud_idx.union(legit_idx) + return X_fold.loc[idx], y_fold.loc[idx] + +PARAMS = {"n_estimators": 100, "random_state": 42} -# Naive random forest (instructor-provided): the imbalance trap. Trained on the raw, -# highly-imbalanced data it just learns to predict the majority class, so accuracy looks -# great while recall is ~0 (it never flags a fraud). Logged for comparison, NOT -# registered. +# Naive baseline (instructor-provided): fit on the raw, imbalanced data with NO balancing to +# show the imbalance trap: near-perfect accuracy but ~0 recall (it never flags a fraud). Scored +# on one grouped fold for illustration; logged for comparison, NOT registered. with mlflow.start_run(run_name="rf_imbalanced"): - naive_params = {"n_estimators": 100, "random_state": 42} - naive = RandomForestClassifier(**naive_params) - naive.fit(X_train, y_train) - naive_metrics = evaluate(naive, X_test, y_test) - mlflow.log_params({**naive_params, "training_data": "imbalanced"}) + tr, va = next(StratifiedGroupKFold(n_splits=5).split(X, y, groups)) + naive = RandomForestClassifier(**PARAMS).fit(X.iloc[tr], y.iloc[tr]) + naive_metrics = evaluate(naive, X.iloc[va], y.iloc[va]) + mlflow.log_params({**PARAMS, "training_data": "imbalanced"}) mlflow.log_metrics(naive_metrics) print("rf_imbalanced:", naive_metrics) # high accuracy, low recall -# Fixed random forest: the model that is tracked and registered. +# Registered model: Stratified GROUP k-fold cross-validation. Each fold stratifies by the rare +# fraud label AND keeps every client in one fold, so no customer is ever in both the fit and the +# validation of a fold (no entity leakage in the estimate). Balance the fit side of each fold, +# score the untouched validation side, and average -> an honest generalisation estimate. The +# registered artifact is then fit on ALL train+validation rows (balanced). with mlflow.start_run(run_name="rf_balanced") as run: - # Balance the training rows: keep every fraud row and randomly sample an equal number of - # non-fraud rows (provided: this is a data-science fix, not the MLOps concept). Evaluation - # still uses the untouched, imbalanced X_test/y_test so metrics reflect reality. - fraud_idx = y_train[y_train == 1].index - legit_idx = y_train[y_train == 0].sample(n=len(fraud_idx), random_state=42).index - bal_idx = fraud_idx.union(legit_idx) - X_bal, y_bal = X_train.loc[bal_idx], y_train.loc[bal_idx] - - params = {"n_estimators": 100, "random_state": 42} - model = RandomForestClassifier(**params) - model.fit(X_bal, y_bal) - metrics = evaluate(model, X_test, y_test) - - mlflow.log_params(params) + sgkf = StratifiedGroupKFold(n_splits=5, shuffle=True, random_state=42) + fold_metrics = [] + for tr, va in sgkf.split(X, y, groups): + X_bal, y_bal = balance(X.iloc[tr], y.iloc[tr]) + clf = RandomForestClassifier(**PARAMS).fit(X_bal, y_bal) + fold_metrics.append(evaluate(clf, X.iloc[va], y.iloc[va])) + # Cross-validated metrics = mean across folds (what we report and gate on). + metrics = {k: float(np.mean([m[k] for m in fold_metrics])) for k in fold_metrics[0]} + + # Final registered model: fit on ALL train+validation rows (balanced). + X_all_bal, y_all_bal = balance(X, y) + model = RandomForestClassifier(**PARAMS).fit(X_all_bal, y_all_bal) + + mlflow.log_params(PARAMS) mlflow.log_param("training_data", "balanced (1:1 down-sampled)") - mlflow.log_param("train_rows", int(len(y_bal))) + mlflow.log_param("cv", "StratifiedGroupKFold(n_splits=5) grouped by client_id") + mlflow.log_param("train_rows", int(len(y_all_bal))) mlflow.log_param("training_table", source_table) mlflow.log_param("card_feature_table", card_feature_table) mlflow.log_param("client_feature_table", client_feature_table) @@ -337,7 +356,7 @@ def predict(self, context, model_input): # TODO: log and register the model to Unity Catalog with its feature metadata # HINT: fe.log_model packages the model with the training_set's feature lookups (so serving # HINT: reproduces the same features) and registers it to UC in a single call. - # HINT: fe.log_model(model=FraudProbabilityModel(model), artifact_path="model", + # HINT: fe.log_model(model=FraudProbabilityModel(model, FEATURE_COLS), artifact_path="model", # HINT: flavor=mlflow.pyfunc, training_set=training_set, # HINT: registered_model_name=uc_model_name, infer_input_example=True) # <-- Your code here From 5812e08bbefdaea18a16bb2ff16850a6d5ec6bf5 Mon Sep 17 00:00:00 2001 From: Ben Constable Date: Wed, 15 Jul 2026 20:22:34 +0200 Subject: [PATCH 3/7] feat(monitoring): use the validation split as the drift baseline The batch monitor compared each window only against the previous windows, so "drift" meant window-to-window wobble rather than movement away from what the model was validated on. Databricks recommends the model's train/validation data as the reference distribution for an inference monitor. - batch_inference now scores the held-out validation split with the champion and writes it, in the exact schema of fraud_predictions, to fraud_predictions_baseline (overwritten each run so it tracks the champion). - batch_monitor sets baseline_table_name to that table, so drift is measured against the validation distribution. Commented inline to explain why. --- resources/monitoring-resource.yml | 4 ++ .../notebooks/batch_inference.py | 64 +++++++++++++++++++ .../notebooks/batch_inference.py | 64 +++++++++++++++++++ 3 files changed, 132 insertions(+) diff --git a/resources/monitoring-resource.yml b/resources/monitoring-resource.yml index 30a0028..1201426 100644 --- a/resources/monitoring-resource.yml +++ b/resources/monitoring-resource.yml @@ -26,6 +26,10 @@ resources: # model quality (accuracy/precision/recall/f1) as well as drift. batch_monitor: table_name: ${var.catalog_name}.${resources.schemas.fraud_ml.name}.fraud_predictions + # Drift baseline = the validation split scored by the champion (written by batch_inference). + # Databricks recommends the model's train/validation data as the reference distribution, so + # drift is measured against what the model was validated on, not just window-to-window change. + baseline_table_name: ${var.catalog_name}.${resources.schemas.fraud_ml.name}.fraud_predictions_baseline output_schema_name: ${var.catalog_name}.${resources.schemas.fraud_ml.name} assets_dir: /Workspace/Users/${workspace.current_user.userName}/lakehouse_monitoring inference_log: diff --git a/solution/deployment/batch_inference/notebooks/batch_inference.py b/solution/deployment/batch_inference/notebooks/batch_inference.py index d76ae01..eacdf6e 100644 --- a/solution/deployment/batch_inference/notebooks/batch_inference.py +++ b/solution/deployment/batch_inference/notebooks/batch_inference.py @@ -220,6 +220,70 @@ # COMMAND ---------- +# MAGIC %md +# MAGIC ## 4. Write the monitor baseline (the validation split) +# MAGIC +# MAGIC A drift monitor compares the live window against a REFERENCE distribution. Databricks +# MAGIC recommends the data the model was validated on as that reference, so we score the held-out +# MAGIC `validation` split with the same champion and persist it in the SAME schema as the monitored +# MAGIC table. The batch monitor's `baseline_table_name` points at it, so "drift" means the live +# MAGIC predictions/features have moved away from what the model was validated on, not just +# MAGIC window-to-window wobble. Overwritten each run so the baseline tracks the current champion. + +# COMMAND ---------- + +split_table = f"{catalog_name}.{ml_schema}.transactions_split" +val_to_score = ( + spark.read.table(source_table) + .join( + spark.read.table(split_table) + .where(F.col("split") == "validation") + .select("transaction_id"), + "transaction_id", + ) + .select( + "transaction_id", + "card_id", + "client_id", + F.col("amount").cast("double").alias("amount"), + "transaction_hour", + "mcc", + "use_chip", + ) +) +val_scored = fe.score_batch(model_uri=f"models:/{uc_model_name}@{model_alias}", df=val_to_score) + +# Same shaping as the monitored table so the schemas match exactly (the monitor compares them). +baseline = ( + val_scored.withColumn("model_version", F.lit(model_version)) + .withColumn("scored_at", scored_at) + .join(labels, on="transaction_id", how="left") + .select( + "transaction_id", + F.col("prediction").alias("fraud_score"), + (F.col("prediction") >= F.lit(0.5)).cast("int").alias("prediction"), + "model_version", + "scored_at", + "is_fraud", + F.col("amount").cast("double").alias("amount"), + "credit_score", + "credit_limit", + "yearly_income", + "current_age", + "num_cards_issued", + "is_night", + "is_online", + "is_high_risk_mcc", + "amount_to_income_ratio", + "amount_to_credit_limit_ratio", + ) +) +baseline_table = f"{catalog_name}.{ml_schema}.fraud_predictions_baseline" +baseline.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(baseline_table) +print(f"Wrote monitor baseline {baseline_table} from the validation split ({baseline.count():,} rows).") + +# COMMAND ---------- + # MAGIC %md # MAGIC ## Recap # MAGIC - Batch scoring loads `@champion`; promotion is an alias swap, so this job never changes. diff --git a/src/deployment/batch_inference/notebooks/batch_inference.py b/src/deployment/batch_inference/notebooks/batch_inference.py index 036acf8..134df92 100644 --- a/src/deployment/batch_inference/notebooks/batch_inference.py +++ b/src/deployment/batch_inference/notebooks/batch_inference.py @@ -218,6 +218,70 @@ # COMMAND ---------- +# MAGIC %md +# MAGIC ## 4. Write the monitor baseline (the validation split) +# MAGIC +# MAGIC A drift monitor compares the live window against a REFERENCE distribution. Databricks +# MAGIC recommends the data the model was validated on as that reference, so we score the held-out +# MAGIC `validation` split with the same champion and persist it in the SAME schema as the monitored +# MAGIC table. The batch monitor's `baseline_table_name` points at it, so "drift" means the live +# MAGIC predictions/features have moved away from what the model was validated on, not just +# MAGIC window-to-window wobble. Overwritten each run so the baseline tracks the current champion. + +# COMMAND ---------- + +split_table = f"{catalog_name}.{ml_schema}.transactions_split" +val_to_score = ( + spark.read.table(source_table) + .join( + spark.read.table(split_table) + .where(F.col("split") == "validation") + .select("transaction_id"), + "transaction_id", + ) + .select( + "transaction_id", + "card_id", + "client_id", + F.col("amount").cast("double").alias("amount"), + "transaction_hour", + "mcc", + "use_chip", + ) +) +val_scored = fe.score_batch(model_uri=f"models:/{uc_model_name}@{model_alias}", df=val_to_score) + +# Same shaping as the monitored table so the schemas match exactly (the monitor compares them). +baseline = ( + val_scored.withColumn("model_version", F.lit(model_version)) + .withColumn("scored_at", scored_at) + .join(labels, on="transaction_id", how="left") + .select( + "transaction_id", + F.col("prediction").alias("fraud_score"), + (F.col("prediction") >= F.lit(0.5)).cast("int").alias("prediction"), + "model_version", + "scored_at", + "is_fraud", + F.col("amount").cast("double").alias("amount"), + "credit_score", + "credit_limit", + "yearly_income", + "current_age", + "num_cards_issued", + "is_night", + "is_online", + "is_high_risk_mcc", + "amount_to_income_ratio", + "amount_to_credit_limit_ratio", + ) +) +baseline_table = f"{catalog_name}.{ml_schema}.fraud_predictions_baseline" +baseline.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(baseline_table) +print(f"Wrote monitor baseline {baseline_table} from the validation split ({baseline.count():,} rows).") + +# COMMAND ---------- + # MAGIC %md # MAGIC ## Recap # MAGIC - Batch scoring loads `@champion`; promotion is an alias swap, so this job never changes. From e19f732c616bab92944a338f1203706dab8cf269 Mon Sep 17 00:00:00 2001 From: Ben Constable Date: Thu, 16 Jul 2026 10:04:46 +0100 Subject: [PATCH 4/7] fix(training): confusion matrix uses a grouped demo fold, not removed X_test/y_test The CV rewrite removed the single X_test/y_test split, but the confusion-matrix cell still referenced them (NameError at run time). Add a grouped demo fold (first StratifiedGroupKFold split) and plot naive vs a balanced demo model on that held-out fold. Verified end to end on dev: feature engineering, grouped k-fold training, evaluation gate on the test split, and the validation-split baseline all run. --- solution/training/notebooks/training.py | 32 +++++++++++++++++-------- src/training/notebooks/training.py | 32 +++++++++++++++++-------- 2 files changed, 44 insertions(+), 20 deletions(-) diff --git a/solution/training/notebooks/training.py b/solution/training/notebooks/training.py index 7c056ec..8ed19e6 100644 --- a/solution/training/notebooks/training.py +++ b/solution/training/notebooks/training.py @@ -326,17 +326,29 @@ def balance(X_fold, y_fold): PARAMS = {"n_estimators": 100, "random_state": 42} -# Naive baseline (instructor-provided): fit on the raw, imbalanced data with NO balancing to -# show the imbalance trap: near-perfect accuracy but ~0 recall (it never flags a fraud). Scored -# on one grouped fold for illustration; logged for comparison, NOT registered. +# One grouped hold-out fold for the in-notebook naive-vs-balanced comparison and the confusion +# matrices below, so those illustrations are scored on rows the demo models were not fit on. +demo_tr, demo_va = next( + StratifiedGroupKFold(n_splits=5, shuffle=True, random_state=42).split(X, y, groups) +) +X_demo_va, y_demo_va = X.iloc[demo_va], y.iloc[demo_va] + +# Naive baseline (instructor-provided): fit on the raw, imbalanced fold with NO balancing to +# show the imbalance trap: near-perfect accuracy but ~0 recall (it never flags a fraud). Logged +# for comparison, NOT registered. with mlflow.start_run(run_name="rf_imbalanced"): - tr, va = next(StratifiedGroupKFold(n_splits=5).split(X, y, groups)) - naive = RandomForestClassifier(**PARAMS).fit(X.iloc[tr], y.iloc[tr]) - naive_metrics = evaluate(naive, X.iloc[va], y.iloc[va]) + naive = RandomForestClassifier(**PARAMS).fit(X.iloc[demo_tr], y.iloc[demo_tr]) + naive_metrics = evaluate(naive, X_demo_va, y_demo_va) mlflow.log_params({**PARAMS, "training_data": "imbalanced"}) mlflow.log_metrics(naive_metrics) print("rf_imbalanced:", naive_metrics) # high accuracy, low recall +# Balanced model on the SAME demo fold, used only for the confusion-matrix comparison below (the +# registered model is the CV-averaged one fit on all data; this fold model is what we plot so +# both matrices are scored on the held-out demo fold, not on rows the model was fit on). +X_demo_bal, y_demo_bal = balance(X.iloc[demo_tr], y.iloc[demo_tr]) +demo_balanced = RandomForestClassifier(**PARAMS).fit(X_demo_bal, y_demo_bal) + # Registered model: Stratified GROUP k-fold cross-validation. Each fold stratifies by the rare # fraud label AND keeps every client in one fold, so no customer is ever in both the fit and the # validation of a fold (no entity leakage in the estimate). Balance the fit side of each fold, @@ -414,8 +426,8 @@ def balance(X_fold, y_fold): # MAGIC and detects little fraud. # MAGIC - The balanced model recovers true positives at the cost of more false positives. # MAGIC -# MAGIC Both use the same original, imbalanced test set. The figure is logged to the -# MAGIC `rf_balanced` MLflow run so it stays with the model version. +# MAGIC Both are scored on the same held-out demo fold (rows neither demo model was fit on). The +# MAGIC figure is logged to the `rf_balanced` MLflow run so it stays with the model version. # COMMAND ---------- @@ -425,10 +437,10 @@ def balance(X_fold, y_fold): fig, axes = plt.subplots(1, 2, figsize=(11, 4)) for ax, clf, title in ( (axes[0], naive, "rf_imbalanced (naive)"), - (axes[1], model, "rf_balanced (registered)"), + (axes[1], demo_balanced, "rf_balanced (demo fold)"), ): ConfusionMatrixDisplay.from_estimator( - clf, X_test, y_test, display_labels=["legit", "fraud"], cmap="Blues", ax=ax, colorbar=False + clf, X_demo_va, y_demo_va, display_labels=["legit", "fraud"], cmap="Blues", ax=ax, colorbar=False ) ax.set_title(title) fig.tight_layout() diff --git a/src/training/notebooks/training.py b/src/training/notebooks/training.py index 51147c9..50c7f42 100644 --- a/src/training/notebooks/training.py +++ b/src/training/notebooks/training.py @@ -309,17 +309,29 @@ def balance(X_fold, y_fold): PARAMS = {"n_estimators": 100, "random_state": 42} -# Naive baseline (instructor-provided): fit on the raw, imbalanced data with NO balancing to -# show the imbalance trap: near-perfect accuracy but ~0 recall (it never flags a fraud). Scored -# on one grouped fold for illustration; logged for comparison, NOT registered. +# One grouped hold-out fold for the in-notebook naive-vs-balanced comparison and the confusion +# matrices below, so those illustrations are scored on rows the demo models were not fit on. +demo_tr, demo_va = next( + StratifiedGroupKFold(n_splits=5, shuffle=True, random_state=42).split(X, y, groups) +) +X_demo_va, y_demo_va = X.iloc[demo_va], y.iloc[demo_va] + +# Naive baseline (instructor-provided): fit on the raw, imbalanced fold with NO balancing to +# show the imbalance trap: near-perfect accuracy but ~0 recall (it never flags a fraud). Logged +# for comparison, NOT registered. with mlflow.start_run(run_name="rf_imbalanced"): - tr, va = next(StratifiedGroupKFold(n_splits=5).split(X, y, groups)) - naive = RandomForestClassifier(**PARAMS).fit(X.iloc[tr], y.iloc[tr]) - naive_metrics = evaluate(naive, X.iloc[va], y.iloc[va]) + naive = RandomForestClassifier(**PARAMS).fit(X.iloc[demo_tr], y.iloc[demo_tr]) + naive_metrics = evaluate(naive, X_demo_va, y_demo_va) mlflow.log_params({**PARAMS, "training_data": "imbalanced"}) mlflow.log_metrics(naive_metrics) print("rf_imbalanced:", naive_metrics) # high accuracy, low recall +# Balanced model on the SAME demo fold, used only for the confusion-matrix comparison below (the +# registered model is the CV-averaged one fit on all data; this fold model is what we plot so +# both matrices are scored on the held-out demo fold, not on rows the model was fit on). +X_demo_bal, y_demo_bal = balance(X.iloc[demo_tr], y.iloc[demo_tr]) +demo_balanced = RandomForestClassifier(**PARAMS).fit(X_demo_bal, y_demo_bal) + # Registered model: Stratified GROUP k-fold cross-validation. Each fold stratifies by the rare # fraud label AND keeps every client in one fold, so no customer is ever in both the fit and the # validation of a fold (no entity leakage in the estimate). Balance the fit side of each fold, @@ -392,8 +404,8 @@ def balance(X_fold, y_fold): # MAGIC and detects little fraud. # MAGIC - The balanced model recovers true positives at the cost of more false positives. # MAGIC -# MAGIC Both use the same original, imbalanced test set. The figure is logged to the -# MAGIC `rf_balanced` MLflow run so it stays with the model version. +# MAGIC Both are scored on the same held-out demo fold (rows neither demo model was fit on). The +# MAGIC figure is logged to the `rf_balanced` MLflow run so it stays with the model version. # COMMAND ---------- @@ -403,10 +415,10 @@ def balance(X_fold, y_fold): fig, axes = plt.subplots(1, 2, figsize=(11, 4)) for ax, clf, title in ( (axes[0], naive, "rf_imbalanced (naive)"), - (axes[1], model, "rf_balanced (registered)"), + (axes[1], demo_balanced, "rf_balanced (demo fold)"), ): ConfusionMatrixDisplay.from_estimator( - clf, X_test, y_test, display_labels=["legit", "fraud"], cmap="Blues", ax=ax, colorbar=False + clf, X_demo_va, y_demo_va, display_labels=["legit", "fraud"], cmap="Blues", ax=ax, colorbar=False ) ax.set_title(title) fig.tight_layout() From 6c901443873f6e8dc36a28d9d7bb3ca0fd4ac509 Mon Sep 17 00:00:00 2001 From: Ben Constable Date: Thu, 16 Jul 2026 10:12:08 +0100 Subject: [PATCH 5/7] Apply fix for hash collision Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/feature_engineering/notebooks/build_features.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/feature_engineering/notebooks/build_features.py b/src/feature_engineering/notebooks/build_features.py index 6f8315b..8f1e666 100644 --- a/src/feature_engineering/notebooks/build_features.py +++ b/src/feature_engineering/notebooks/build_features.py @@ -161,7 +161,7 @@ client_label = enriched.groupBy("client_id").agg( F.max("is_fraud").cast("int").alias("client_has_fraud") ) -strata = Window.partitionBy("client_has_fraud").orderBy(F.xxhash64("client_id")) +strata = Window.partitionBy("client_has_fraud").orderBy(F.xxhash64("client_id"), F.col("client_id")) client_split = client_label.select( "client_id", F.percent_rank().over(strata).alias("pct"), From f2a8537fec5e0784f39d7ad7651ac458e15a5429 Mon Sep 17 00:00:00 2001 From: Ben Constable Date: Thu, 16 Jul 2026 10:12:39 +0100 Subject: [PATCH 6/7] Remove double spark job Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/deployment/batch_inference/notebooks/batch_inference.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/deployment/batch_inference/notebooks/batch_inference.py b/src/deployment/batch_inference/notebooks/batch_inference.py index 134df92..60cf141 100644 --- a/src/deployment/batch_inference/notebooks/batch_inference.py +++ b/src/deployment/batch_inference/notebooks/batch_inference.py @@ -278,7 +278,7 @@ ) baseline_table = f"{catalog_name}.{ml_schema}.fraud_predictions_baseline" baseline.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(baseline_table) -print(f"Wrote monitor baseline {baseline_table} from the validation split ({baseline.count():,} rows).") +print(f"Wrote monitor baseline {baseline_table} from the validation split.") # COMMAND ---------- From cc06ac1b9a29cd68f889e5ce3c18153c0adcb44e Mon Sep 17 00:00:00 2001 From: Ben Constable Date: Thu, 16 Jul 2026 10:19:20 +0100 Subject: [PATCH 7/7] fix(split,training): use deterministic hash bucket split and guard fold balancing Replace the percent_rank grouped/stratified split with a pure hash bucket (pmod(xxhash64(client_id), 100): <70 train, <85 validation, else test). The per-customer grouping and approximate fraud stratification are preserved, but the cut is population-stable so a client never flips between splits as new data arrives. Guard balance() so a fold with no fraud (or no non-fraud) rows falls back to the fold unchanged, and never samples more non-fraud rows than exist, preventing an empty training set that would fail RandomForestClassifier.fit. Also mirror the earlier src-only double-spark-job fix into the batch_inference solution so regen stays consistent. --- .../notebooks/batch_inference.py | 2 +- .../notebooks/build_features.py | 58 +++++++++---------- solution/training/notebooks/training.py | 9 ++- .../notebooks/build_features.py | 58 +++++++++---------- src/training/notebooks/training.py | 9 ++- 5 files changed, 71 insertions(+), 65 deletions(-) diff --git a/solution/deployment/batch_inference/notebooks/batch_inference.py b/solution/deployment/batch_inference/notebooks/batch_inference.py index eacdf6e..c1fa642 100644 --- a/solution/deployment/batch_inference/notebooks/batch_inference.py +++ b/solution/deployment/batch_inference/notebooks/batch_inference.py @@ -280,7 +280,7 @@ ) baseline_table = f"{catalog_name}.{ml_schema}.fraud_predictions_baseline" baseline.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(baseline_table) -print(f"Wrote monitor baseline {baseline_table} from the validation split ({baseline.count():,} rows).") +print(f"Wrote monitor baseline {baseline_table} from the validation split.") # COMMAND ---------- diff --git a/solution/feature_engineering/notebooks/build_features.py b/solution/feature_engineering/notebooks/build_features.py index cd7194e..f9fc5c4 100644 --- a/solution/feature_engineering/notebooks/build_features.py +++ b/solution/feature_engineering/notebooks/build_features.py @@ -144,46 +144,44 @@ def publish(name, df, key, description): # MAGIC %md # MAGIC ## 3. Materialise the train / validation / test split # MAGIC -# MAGIC A proper evaluation needs a hold-out set that is representative and GUARANTEED disjoint from +# MAGIC A proper evaluation needs a hold-out that is representative and GUARANTEED disjoint from # MAGIC training, on the same rows every run. Two properties matter for fraud: # MAGIC # MAGIC - **Grouped by customer.** Whole clients are assigned to one split, so the same customer is # MAGIC never in both train and test. A row-level split would let the model memorise a client seen # MAGIC in training and then be graded on that same client (entity leakage -> optimistic metrics). -# MAGIC - **Stratified by the label.** Fraud is well under 1%, so we assign clients to 70/15/15 -# MAGIC *within* each fraud / non-fraud stratum, keeping the fraud rate identical across splits so -# MAGIC the test set is representative and has enough positives for a stable metric. +# MAGIC - **Stable and reproducible.** Each client is assigned by a deterministic hash BUCKET of +# MAGIC `client_id` (0-99), so a client always lands in the same split no matter when the job runs +# MAGIC or how much new data has arrived. Nothing flips on recompute. The uniform hash keeps the +# MAGIC rare fraud rate roughly equal across splits (approximately stratified); this trades exact +# MAGIC 70/15/15 proportions for the population stability production needs. # MAGIC -# MAGIC The assignment is a deterministic rank over a hash of `client_id`, so it is reproducible and -# MAGIC persisted as a versioned Delta table; a model version is always scored on the identical -# MAGIC holdout. Production hardening: for temporal data a time-based holdout (train older, test the -# MAGIC most recent window) is stronger still, and point-in-time feature lookups prevent feature -# MAGIC leakage. +# MAGIC The result is persisted as a versioned Delta table, so a model version is always scored on +# MAGIC the identical holdout. Production hardening: for temporal data a time-based holdout (train +# MAGIC older, test the most recent window) is stronger still, and point-in-time feature lookups +# MAGIC prevent feature leakage. # COMMAND ---------- split_table = f"{catalog_name}.{ml_schema}.transactions_split" -# Group by CUSTOMER and stratify by the (rare) fraud label. Assign whole clients (not rows) to a -# split so no customer ever appears in both train and test. Within each fraud / non-fraud stratum -# rank clients by a hash of client_id and cut at 70/15/15, so every client lands in exactly one -# split (grouped), the fraud-client proportion is identical across splits (stratified), and the -# assignment is deterministic (hash order) and versioned as a Delta table. -from pyspark.sql import Window - -client_label = enriched.groupBy("client_id").agg( - F.max("is_fraud").cast("int").alias("client_has_fraud") -) -strata = Window.partitionBy("client_has_fraud").orderBy(F.xxhash64("client_id")) -client_split = client_label.select( - "client_id", - F.percent_rank().over(strata).alias("pct"), -).select( - "client_id", - F.when(F.col("pct") < 0.70, "train") - .when(F.col("pct") < 0.85, "validation") - .otherwise("test") - .alias("split"), +# Group by CUSTOMER: assign whole clients (not rows) to a split so no customer ever appears in +# both train and test. Use a deterministic hash BUCKET of client_id (0-99): the same client always +# lands in the same bucket regardless of when the job runs or how much data has arrived, so the +# split is population-stable (a client never flips train<->test on recompute). The uniform hash +# keeps the rare fraud rate roughly equal across splits (approximately stratified), and the +# assignment is versioned as a Delta table. +bucket = F.pmod(F.xxhash64("client_id"), F.lit(100)) +client_split = ( + enriched.select("client_id") + .distinct() + .select( + "client_id", + F.when(bucket < 70, "train") + .when(bucket < 85, "validation") + .otherwise("test") + .alias("split"), + ) ) splits = ( @@ -192,7 +190,7 @@ def publish(name, df, key, description): .select("transaction_id", "split") ) splits.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(split_table) -print(f"Wrote {split_table}: 70/15/15 train/validation/test, grouped by client, stratified by fraud.") +print(f"Wrote {split_table}: ~70/15/15 train/validation/test, grouped by client (deterministic hash).") # COMMAND ---------- diff --git a/solution/training/notebooks/training.py b/solution/training/notebooks/training.py index 8ed19e6..ddc8518 100644 --- a/solution/training/notebooks/training.py +++ b/solution/training/notebooks/training.py @@ -317,9 +317,14 @@ def predict(self, context, model_input): def balance(X_fold, y_fold): - """Down-sample non-fraud to 1:1 within a fold (keep every fraud row).""" + """Down-sample non-fraud to ~1:1 within a fold (keep every fraud row). Falls back to the fold + unchanged when it has no fraud (nothing to balance) or no non-fraud rows, and never samples + more non-fraud rows than exist, so the fold can never come back empty.""" + legit = y_fold[y_fold == 0] fraud_idx = y_fold[y_fold == 1].index - legit_idx = y_fold[y_fold == 0].sample(n=len(fraud_idx), random_state=42).index + if len(fraud_idx) == 0 or len(legit) == 0: + return X_fold, y_fold + legit_idx = legit.sample(n=min(len(fraud_idx), len(legit)), random_state=42).index idx = fraud_idx.union(legit_idx) return X_fold.loc[idx], y_fold.loc[idx] diff --git a/src/feature_engineering/notebooks/build_features.py b/src/feature_engineering/notebooks/build_features.py index 8f1e666..6b921d4 100644 --- a/src/feature_engineering/notebooks/build_features.py +++ b/src/feature_engineering/notebooks/build_features.py @@ -131,46 +131,44 @@ # MAGIC %md # MAGIC ## 3. Materialise the train / validation / test split # MAGIC -# MAGIC A proper evaluation needs a hold-out set that is representative and GUARANTEED disjoint from +# MAGIC A proper evaluation needs a hold-out that is representative and GUARANTEED disjoint from # MAGIC training, on the same rows every run. Two properties matter for fraud: # MAGIC # MAGIC - **Grouped by customer.** Whole clients are assigned to one split, so the same customer is # MAGIC never in both train and test. A row-level split would let the model memorise a client seen # MAGIC in training and then be graded on that same client (entity leakage -> optimistic metrics). -# MAGIC - **Stratified by the label.** Fraud is well under 1%, so we assign clients to 70/15/15 -# MAGIC *within* each fraud / non-fraud stratum, keeping the fraud rate identical across splits so -# MAGIC the test set is representative and has enough positives for a stable metric. +# MAGIC - **Stable and reproducible.** Each client is assigned by a deterministic hash BUCKET of +# MAGIC `client_id` (0-99), so a client always lands in the same split no matter when the job runs +# MAGIC or how much new data has arrived. Nothing flips on recompute. The uniform hash keeps the +# MAGIC rare fraud rate roughly equal across splits (approximately stratified); this trades exact +# MAGIC 70/15/15 proportions for the population stability production needs. # MAGIC -# MAGIC The assignment is a deterministic rank over a hash of `client_id`, so it is reproducible and -# MAGIC persisted as a versioned Delta table; a model version is always scored on the identical -# MAGIC holdout. Production hardening: for temporal data a time-based holdout (train older, test the -# MAGIC most recent window) is stronger still, and point-in-time feature lookups prevent feature -# MAGIC leakage. +# MAGIC The result is persisted as a versioned Delta table, so a model version is always scored on +# MAGIC the identical holdout. Production hardening: for temporal data a time-based holdout (train +# MAGIC older, test the most recent window) is stronger still, and point-in-time feature lookups +# MAGIC prevent feature leakage. # COMMAND ---------- split_table = f"{catalog_name}.{ml_schema}.transactions_split" -# Group by CUSTOMER and stratify by the (rare) fraud label. Assign whole clients (not rows) to a -# split so no customer ever appears in both train and test. Within each fraud / non-fraud stratum -# rank clients by a hash of client_id and cut at 70/15/15, so every client lands in exactly one -# split (grouped), the fraud-client proportion is identical across splits (stratified), and the -# assignment is deterministic (hash order) and versioned as a Delta table. -from pyspark.sql import Window - -client_label = enriched.groupBy("client_id").agg( - F.max("is_fraud").cast("int").alias("client_has_fraud") -) -strata = Window.partitionBy("client_has_fraud").orderBy(F.xxhash64("client_id"), F.col("client_id")) -client_split = client_label.select( - "client_id", - F.percent_rank().over(strata).alias("pct"), -).select( - "client_id", - F.when(F.col("pct") < 0.70, "train") - .when(F.col("pct") < 0.85, "validation") - .otherwise("test") - .alias("split"), +# Group by CUSTOMER: assign whole clients (not rows) to a split so no customer ever appears in +# both train and test. Use a deterministic hash BUCKET of client_id (0-99): the same client always +# lands in the same bucket regardless of when the job runs or how much data has arrived, so the +# split is population-stable (a client never flips train<->test on recompute). The uniform hash +# keeps the rare fraud rate roughly equal across splits (approximately stratified), and the +# assignment is versioned as a Delta table. +bucket = F.pmod(F.xxhash64("client_id"), F.lit(100)) +client_split = ( + enriched.select("client_id") + .distinct() + .select( + "client_id", + F.when(bucket < 70, "train") + .when(bucket < 85, "validation") + .otherwise("test") + .alias("split"), + ) ) splits = ( @@ -179,7 +177,7 @@ .select("transaction_id", "split") ) splits.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(split_table) -print(f"Wrote {split_table}: 70/15/15 train/validation/test, grouped by client, stratified by fraud.") +print(f"Wrote {split_table}: ~70/15/15 train/validation/test, grouped by client (deterministic hash).") # COMMAND ---------- diff --git a/src/training/notebooks/training.py b/src/training/notebooks/training.py index 50c7f42..adde721 100644 --- a/src/training/notebooks/training.py +++ b/src/training/notebooks/training.py @@ -300,9 +300,14 @@ def predict(self, context, model_input): def balance(X_fold, y_fold): - """Down-sample non-fraud to 1:1 within a fold (keep every fraud row).""" + """Down-sample non-fraud to ~1:1 within a fold (keep every fraud row). Falls back to the fold + unchanged when it has no fraud (nothing to balance) or no non-fraud rows, and never samples + more non-fraud rows than exist, so the fold can never come back empty.""" + legit = y_fold[y_fold == 0] fraud_idx = y_fold[y_fold == 1].index - legit_idx = y_fold[y_fold == 0].sample(n=len(fraud_idx), random_state=42).index + if len(fraud_idx) == 0 or len(legit) == 0: + return X_fold, y_fold + legit_idx = legit.sample(n=min(len(fraud_idx), len(legit)), random_state=42).index idx = fraud_idx.union(legit_idx) return X_fold.loc[idx], y_fold.loc[idx]