Feature/holdout evaluation

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:

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.")
+
+# COMMAND ----------
+
 # MAGIC %md
 # MAGIC ## Recap
 # MAGIC - Batch scoring loads `@champion`; promotion is an alias swap, so this job never changes.

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

solution/feature_engineering/notebooks/build_features.py

@@ -141,6 +141,59 @@ 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 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 - **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 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: 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 = (
+    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/15/15 train/validation/test, grouped by client (deterministic hash).")
+
+# COMMAND ----------
+
 # MAGIC %md
 # MAGIC ## 3. Register the on-demand feature functions
 # MAGIC