From daf263fc047e32429bc16e0d4200bead64aabe38 Mon Sep 17 00:00:00 2001 From: Samarth Uday Date: Sat, 29 Aug 2026 21:47:31 +0530 Subject: [PATCH] fix: Use calibration/validation splits correctly, fix walk-forward block bug Two genuine methodology issues survived the prior integration-fix branch, both confirmed by reading the actual code (not just the bug report) before changing anything: 1. train_model.py fit the probability calibrator AND selected the alert threshold from the same calibration split -- the separate `validation` dataframe from temporal_split was computed and logged but never actually used for anything. Fixed: calibrator now fits only on `calibration`; threshold selection and the calibration-quality diagnostic (Brier/log-loss, calibration_curve.png) now use the true held-out `validation` split. The baseline logistic model's calibrator is likewise fit on `calibration` (matching the main model) rather than on validation. feature_ablation.py had the same unused `validation` parameter, but with no natural role there (ablation only measures top-K ranking quality, no threshold is chosen) -- removed rather than forcing an artificial use. 2. walk_forward_backtest.py indexed into the *raw* unique-timestamp array using literal offsets (train_end = 4 + window*2, etc). On a dataset with 7.75M unique timestamps this meant every window only ever used the first ~14 timestamps of the entire dataset -- which is why it previously failed with "no valid walk-forward windows were available" on the real SAML-D run. Fixed: timestamps are now split into 14 contiguous chronological blocks via np.array_split first, and the same expanding-window indices are applied to block boundaries instead. Verified against the real 9.5M-row dataset: now produces 4 real windows spanning the full timeline (PR-AUC 0.938 -> 0.990 as the training window expands), instead of raising. Extracted the boundary logic into window_boundaries() and added 3 regression tests (tests/test_walk_forward.py) so this can't silently regress back to a same-bug-different-shape variant. Also, two smaller items from the same review: 3. unseen_entity_evaluation.py only reported per-subgroup top-K metrics (each subgroup gets its own top 0.5%). Added metrics_at_threshold() to src/evaluation/metrics.py and a second table applying the artifact's actual fixed decision_threshold to both subgroups -- answers "what does the current production policy do to unseen accounts," distinct from "what's the best possible ranking within this subgroup alone." 4. run_experiments.py never ran generate_report_figures.py, so the documented single-command workflow didn't actually produce the figures the dashboard/README reference. Wired it in, and added the unseen-entity experiment behind an explicit --generalization flag rather than forcing the research extension into every run. generate_report_figures.py and unseen_entity_evaluation.py now also accept --features/--artifact so they honor custom paths passed to run_experiments.py instead of silently falling back to defaults. Status: all 63 existing tests + 3 new tests pass, ruff clean. Walk-forward fix independently verified against the real dataset (above). The calibration/validation split fix is verified by tests and only changes the saved artifact's decision_threshold/validation_probability_quantiles and calibration diagnostics (Brier/log-loss/calibration_curve.png) -- it cannot change test-set PR-AUC/ROC-AUC, since those are computed from the test split independently of where the threshold was chosen. Retraining to refresh the committed artifact and docs/assets/ diagnostics with the corrected methodology is a separate follow-up, not yet done. Co-Authored-By: Claude Sonnet 5 --- scripts/feature_ablation.py | 17 ++++----- scripts/generate_report_figures.py | 14 +++++-- scripts/run_experiments.py | 9 +++++ scripts/train_model.py | 40 ++++++++++--------- scripts/unseen_entity_evaluation.py | 57 +++++++++++++++++++++++++--- scripts/walk_forward_backtest.py | 59 ++++++++++++++++++++--------- src/evaluation/metrics.py | 36 ++++++++++++++++++ tests/test_walk_forward.py | 40 +++++++++++++++++++ 8 files changed, 219 insertions(+), 53 deletions(-) create mode 100644 tests/test_walk_forward.py diff --git a/scripts/feature_ablation.py b/scripts/feature_ablation.py index d47ed9c..7434222 100644 --- a/scripts/feature_ablation.py +++ b/scripts/feature_ablation.py @@ -28,23 +28,23 @@ ALERT_RATE = 0.005 -def evaluate_feature_set(name, features, train, calibration, validation, test, config): +def evaluate_feature_set(name, features, train, calibration, test, config): logger.info(f"Evaluating feature set: '{name}' ({len(features)} features)") ( preprocessor, model, - X_validation_processed, - y_validation, + X_calibration_processed, + y_calibration, ) = fit_model(train, calibration, features, config=config) logger.debug(f"Model trained for '{name}'") - validation_probabilities = model.predict_proba( - X_validation_processed + calibration_probabilities = model.predict_proba( + X_calibration_processed )[:, 1] calibrator = ProbabilityCalibrator().fit( - validation_probabilities, - y_validation, + calibration_probabilities, + y_calibration, ) test_probabilities = calibrator.predict( @@ -82,7 +82,7 @@ def main(): logger.info(f"Loaded {len(df):,} transactions") logger.info("Performing temporal split...") - train, calibration, validation, test = temporal_split(df) + train, calibration, _validation, test = temporal_split(df) config = get_config(fast=args.fast) if args.fast: @@ -95,7 +95,6 @@ def main(): features, train, calibration, - validation, test, config, ) diff --git a/scripts/generate_report_figures.py b/scripts/generate_report_figures.py index da4c692..116a564 100644 --- a/scripts/generate_report_figures.py +++ b/scripts/generate_report_figures.py @@ -1,5 +1,6 @@ """Generate report figures (ROC, PR, feature importance, ablation, typology) from existing artifacts.""" +import argparse import csv import sys from pathlib import Path @@ -25,16 +26,21 @@ def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--features", type=Path, default=FEATURE_PATH) + parser.add_argument("--artifact", type=Path, default=ARTIFACT_PATH) + args = parser.parse_args() + FIGURES_DIR.mkdir(parents=True, exist_ok=True) - logger.info(f"Loading artifact from {ARTIFACT_PATH}...") - artifact = joblib.load(ARTIFACT_PATH) + logger.info(f"Loading artifact from {args.artifact}...") + artifact = joblib.load(args.artifact) preprocessor = artifact["preprocessor"] model = artifact["model"] calibrator = artifact["calibrator"] - logger.info(f"Loading features from {FEATURE_PATH}...") - df = pd.read_parquet(FEATURE_PATH) + logger.info(f"Loading features from {args.features}...") + df = pd.read_parquet(args.features) logger.info("Performing temporal split to recover test set...") _, _, _, test = temporal_split(df) diff --git a/scripts/run_experiments.py b/scripts/run_experiments.py index 2d7aff1..2b48e49 100644 --- a/scripts/run_experiments.py +++ b/scripts/run_experiments.py @@ -17,6 +17,11 @@ def main(): parser.add_argument("--features", type=Path, default=PROJECT_ROOT / "data/processed/transactions_features.parquet") parser.add_argument("--artifact", type=Path, default=PROJECT_ROOT / "artifacts/risk_model.joblib") parser.add_argument("--fast", action="store_true") + parser.add_argument( + "--generalization", + action="store_true", + help="Also run the unseen-entity generalization experiment (research extension, not part of the core workflow).", + ) args = parser.parse_args() python = sys.executable @@ -31,6 +36,10 @@ def main(): run(train) run(ablation) run(walk_forward) + run([python, "scripts/generate_report_figures.py", "--features", str(args.features), "--artifact", str(args.artifact)]) + + if args.generalization: + run([python, "scripts/unseen_entity_evaluation.py", "--features", str(args.features), "--artifact", str(args.artifact)]) if __name__ == "__main__": diff --git a/scripts/train_model.py b/scripts/train_model.py index 43a93bd..73cc7d9 100644 --- a/scripts/train_model.py +++ b/scripts/train_model.py @@ -105,8 +105,8 @@ def main(): ( preprocessor, model, - X_validation_processed, - y_validation, + X_calibration_processed, + y_calibration, ) = fit_model( train, calibration, @@ -114,12 +114,12 @@ def main(): ) # --------------------------- - # CALIBRATION + # CALIBRATION (fit calibrator on the calibration split) # --------------------------- raw_calibration_prob = ( model.predict_proba( - X_validation_processed + X_calibration_processed )[:, 1] ) @@ -127,22 +127,28 @@ def main(): calibrator.fit( raw_calibration_prob, - y_validation, + y_calibration, ) - calibrated_validation_prob = ( - calibrator.predict(raw_calibration_prob) - ) + # --------------------------- + # VALIDATION (independent set: calibration quality check + threshold selection) + # --------------------------- + + X_validation_processed = preprocessor.transform(validation[MODEL_FEATURES]) + y_validation = validation[TARGET] + + raw_validation_prob = model.predict_proba(X_validation_processed)[:, 1] + calibrated_validation_prob = calibrator.predict(raw_validation_prob) calibration_report = { "raw_brier_score": float( - evaluate_model(y_validation, raw_calibration_prob)["brier_score"] + evaluate_model(y_validation, raw_validation_prob)["brier_score"] ), "calibrated_brier_score": float( evaluate_model(y_validation, calibrated_validation_prob)["brier_score"] ), "raw_log_loss": float( - evaluate_model(y_validation, raw_calibration_prob)["log_loss"] + evaluate_model(y_validation, raw_validation_prob)["log_loss"] ), "calibrated_log_loss": float( evaluate_model(y_validation, calibrated_validation_prob)["log_loss"] @@ -154,7 +160,7 @@ def main(): figure, axis = plt.subplots(figsize=(6, 6)) for probabilities, label in ( - (raw_calibration_prob, "Raw XGBoost"), + (raw_validation_prob, "Raw XGBoost"), (calibrated_validation_prob, "Calibrated XGBoost"), ): observed, predicted = calibration_curve( @@ -176,11 +182,11 @@ def main(): # Use a realistic operational alert capacity alert_rate = 0.005 - calibration_alerts = top_k_alert_mask( + validation_alerts = top_k_alert_mask( calibrated_validation_prob, alert_rate, ) - threshold = float(calibrated_validation_prob[calibration_alerts].min()) + threshold = float(calibrated_validation_prob[validation_alerts].min()) logger.info( f"Threshold for {alert_rate:.2%} alert rate: {threshold:.6f}" @@ -251,12 +257,12 @@ def main(): logger.info("Training logistic baseline for comparison...") baseline = build_logistic_baseline() baseline.fit(preprocessor.transform(train[MODEL_FEATURES]), train[TARGET]) - baseline_validation_prob = baseline.predict_proba( - X_validation_processed + baseline_calibration_prob = baseline.predict_proba( + X_calibration_processed )[:, 1] baseline_calibrator = ProbabilityCalibrator().fit( - baseline_validation_prob, - y_validation, + baseline_calibration_prob, + y_calibration, ) baseline_test_prob = baseline_calibrator.predict( baseline.predict_proba(X_test_processed)[:, 1] diff --git a/scripts/unseen_entity_evaluation.py b/scripts/unseen_entity_evaluation.py index 269ba81..72d1829 100644 --- a/scripts/unseen_entity_evaluation.py +++ b/scripts/unseen_entity_evaluation.py @@ -7,6 +7,7 @@ and evaluates it on three partitions of the same held-out test set. """ +import argparse import json import sys from pathlib import Path @@ -19,7 +20,11 @@ PROJECT_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROJECT_ROOT)) -from src.evaluation.metrics import evaluate_model, precision_recall_at_alert_rate +from src.evaluation.metrics import ( + evaluate_model, + metrics_at_threshold, + precision_recall_at_alert_rate, +) from src.logging_config import setup_logging from src.models.train import MODEL_FEATURES, TARGET, temporal_split @@ -32,6 +37,19 @@ ALERT_RATE = 0.005 +def summarize_fixed_threshold( + mask: np.ndarray, y_test: np.ndarray, calibrated_prob: np.ndarray, threshold: float, label: str +) -> dict: + """What the artifact's actual production threshold does to this subgroup, + as opposed to summarize()'s per-subgroup top-K (best ranking within the + subgroup alone).""" + subset_y = y_test[mask] + subset_prob = calibrated_prob[mask] + if len(subset_y) == 0: + return {"label": label, "metrics": None} + return {"label": label, "metrics": metrics_at_threshold(subset_y, subset_prob, threshold)} + + def summarize(mask: np.ndarray, y_test: np.ndarray, calibrated_prob: np.ndarray, label: str) -> dict: subset_y = y_test[mask] subset_prob = calibrated_prob[mask] @@ -65,14 +83,19 @@ def summarize(mask: np.ndarray, y_test: np.ndarray, calibrated_prob: np.ndarray, def main(): - logger.info(f"Loading artifact from {ARTIFACT_PATH}...") - artifact = joblib.load(ARTIFACT_PATH) + parser = argparse.ArgumentParser() + parser.add_argument("--features", type=Path, default=FEATURE_PATH) + parser.add_argument("--artifact", type=Path, default=ARTIFACT_PATH) + args = parser.parse_args() + + logger.info(f"Loading artifact from {args.artifact}...") + artifact = joblib.load(args.artifact) preprocessor = artifact["preprocessor"] model = artifact["model"] calibrator = artifact["calibrator"] - logger.info(f"Loading features from {FEATURE_PATH}...") - df = pd.read_parquet(FEATURE_PATH) + logger.info(f"Loading features from {args.features}...") + df = pd.read_parquet(args.features) train, _, _, test = temporal_split(df) @@ -99,16 +122,29 @@ def main(): seen_result = summarize(both_seen, y_test, calibrated_prob, "Both sender and receiver seen during training") unseen_result = summarize(unseen_entity, y_test, calibrated_prob, "At least one party unseen during training") + decision_threshold = artifact["decision_threshold"] + fixed_threshold_seen = summarize_fixed_threshold( + both_seen, y_test, calibrated_prob, decision_threshold, "Both parties seen" + ) + fixed_threshold_unseen = summarize_fixed_threshold( + unseen_entity, y_test, calibrated_prob, decision_threshold, "Unseen entity" + ) + results = { "alert_rate": ALERT_RATE, "accounts_seen_in_training": len(seen_accounts), "standard_out_of_time": standard, "both_parties_seen": seen_result, "unseen_entity": unseen_result, + "fixed_production_threshold": { + "decision_threshold": decision_threshold, + "both_parties_seen": fixed_threshold_seen, + "unseen_entity": fixed_threshold_unseen, + }, } logger.info("=" * 70) - logger.info("UNSEEN-ENTITY GENERALIZATION RESULTS") + logger.info("UNSEEN-ENTITY GENERALIZATION RESULTS (per-subgroup top-K)") logger.info("=" * 70) for key in ("standard_out_of_time", "both_parties_seen", "unseen_entity"): r = results[key] @@ -119,6 +155,15 @@ def main(): for metric_key, value in r["metrics"].items(): logger.info(f" {metric_key:30s}: {value:.6f}") + logger.info("=" * 70) + logger.info(f"SAME FIXED PRODUCTION THRESHOLD ({decision_threshold:.6f}) APPLIED TO BOTH SUBGROUPS") + logger.info("=" * 70) + for r in (fixed_threshold_seen, fixed_threshold_unseen): + logger.info(f"\n{r['label']}:") + if r["metrics"]: + for metric_key, value in r["metrics"].items(): + logger.info(f" {metric_key:30s}: {value:.6f}") + OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True) OUTPUT_PATH.write_text(json.dumps(results, indent=2) + "\n") logger.info(f"\nResults written to {OUTPUT_PATH.relative_to(PROJECT_ROOT)}") diff --git a/scripts/walk_forward_backtest.py b/scripts/walk_forward_backtest.py index 84e9044..77cad31 100644 --- a/scripts/walk_forward_backtest.py +++ b/scripts/walk_forward_backtest.py @@ -3,6 +3,7 @@ import sys from pathlib import Path +import numpy as np import pandas as pd PROJECT_ROOT = Path(__file__).resolve().parents[1] @@ -17,31 +18,55 @@ logger = setup_logging(__name__) -def run_backtest(frame, config): - logger.info(f"Starting walk-forward backtest on {len(frame):,} transactions") - ordered = frame.sort_values("timestamp").reset_index(drop=True) - timestamps = ordered["timestamp"].drop_duplicates().sort_values().tolist() - block_count = 14 +def window_boundaries(timestamps, block_count=14, window_count=4): + """Expanding-window (train/calibration/test) cutoffs spanning the full timestamp range. + + Splits the full range into `block_count` contiguous chronological blocks and + walks 4 expanding windows across them, each cutoff being the first timestamp + of its block. The last window's test cutoff is None (extends to the end). + """ + timestamps = np.asarray(timestamps) if len(timestamps) < block_count: - raise ValueError("walk-forward backtesting requires at least 14 unique timestamps") + raise ValueError(f"requires at least {block_count} unique timestamps") - logger.info(f"Found {len(timestamps)} unique timestamps, creating rolling windows...") - results = [] - for window in range(4): + blocks = np.array_split(timestamps, block_count) + block_starts = [block[0] for block in blocks] + + boundaries = [] + for window in range(window_count): train_end = 4 + window * 2 calibration_end = train_end + 2 test_end = calibration_end + 2 - if test_end > len(timestamps): + if test_end > block_count: continue - train = ordered[ordered["timestamp"] < timestamps[train_end]] + train_cutoff = block_starts[train_end] + calibration_cutoff = block_starts[calibration_end] + test_cutoff = block_starts[test_end] if test_end < block_count else None + boundaries.append((train_cutoff, calibration_cutoff, test_cutoff)) + return boundaries + + +def run_backtest(frame, config): + logger.info(f"Starting walk-forward backtest on {len(frame):,} transactions") + ordered = frame.sort_values("timestamp").reset_index(drop=True) + timestamps = ordered["timestamp"].drop_duplicates().sort_values().to_numpy() + boundaries = window_boundaries(timestamps) + + logger.info(f"Found {len(timestamps)} unique timestamps, creating rolling windows...") + results = [] + for window, (train_cutoff, calibration_cutoff, test_cutoff) in enumerate(boundaries): + train = ordered[ordered["timestamp"] < train_cutoff] calibration = ordered[ - (ordered["timestamp"] >= timestamps[train_end]) - & (ordered["timestamp"] < timestamps[calibration_end]) - ] - test = ordered[ - (ordered["timestamp"] >= timestamps[calibration_end]) - & (ordered["timestamp"] < timestamps[test_end]) + (ordered["timestamp"] >= train_cutoff) + & (ordered["timestamp"] < calibration_cutoff) ] + if test_cutoff is not None: + test = ordered[ + (ordered["timestamp"] >= calibration_cutoff) + & (ordered["timestamp"] < test_cutoff) + ] + else: + test = ordered[ordered["timestamp"] >= calibration_cutoff] if calibration[TARGET].nunique() < 2: logger.debug(f"Window {window + 1}: Skipping (insufficient target variance in calibration)") continue diff --git a/src/evaluation/metrics.py b/src/evaluation/metrics.py index 9ed1242..008be28 100644 --- a/src/evaluation/metrics.py +++ b/src/evaluation/metrics.py @@ -86,6 +86,42 @@ def precision_recall_at_alert_rate( } +def metrics_at_threshold( + y_true: Union[np.ndarray, list], + probabilities: Union[np.ndarray, list], + threshold: float, +) -> Dict[str, float]: + """Precision/recall/lift/alert-rate for a fixed probability threshold. + + Unlike precision_recall_at_alert_rate (which selects the top-K for each + population independently), this applies one threshold decided elsewhere + (e.g. the artifact's production decision_threshold) -- the question is + "what does the existing policy do to this subgroup," not "what is the + best top-K for this subgroup." + """ + y_true = np.asarray(y_true) + probabilities = np.asarray(probabilities) + predicted = probabilities >= threshold + + tp = np.sum((predicted == 1) & (y_true == 1)) + fp = np.sum((predicted == 1) & (y_true == 0)) + fn = np.sum((predicted == 0) & (y_true == 1)) + + precision = tp / (tp + fp) if tp + fp > 0 else 0 + recall = tp / (tp + fn) if tp + fn > 0 else 0 + base_rate = y_true.mean() + lift = precision / base_rate if base_rate > 0 else 0 + + return { + "threshold": float(threshold), + "precision": float(precision), + "recall": float(recall), + "lift": float(lift), + "alert_rate": float(predicted.mean()), + "alerts": int(predicted.sum()), + } + + def expected_decision_cost( y_true: Union[np.ndarray, list], probabilities: Union[np.ndarray, list], diff --git a/tests/test_walk_forward.py b/tests/test_walk_forward.py new file mode 100644 index 0000000..a826d1c --- /dev/null +++ b/tests/test_walk_forward.py @@ -0,0 +1,40 @@ +import numpy as np +import pandas as pd + +from scripts.walk_forward_backtest import window_boundaries + + +def test_window_boundaries_span_full_timestamp_range(): + timestamps = pd.date_range("2022-01-01", periods=1_000, freq="h").to_numpy() + + boundaries = window_boundaries(timestamps) + + assert len(boundaries) == 4 + # The last window's test cutoff is open-ended and must reach near the end + # of the range -- not stop at the 14th unique timestamp. + last_train_cutoff = boundaries[-1][0] + assert last_train_cutoff > timestamps[len(timestamps) // 2] + + +def test_window_boundaries_are_expanding_and_ordered(): + timestamps = pd.date_range("2022-01-01", periods=1_000, freq="h").to_numpy() + + boundaries = window_boundaries(timestamps) + + train_cutoffs = [b[0] for b in boundaries] + assert train_cutoffs == sorted(train_cutoffs) + assert len(set(train_cutoffs)) == len(train_cutoffs) + + for train_cutoff, calibration_cutoff, test_cutoff in boundaries: + assert train_cutoff < calibration_cutoff + if test_cutoff is not None: + assert calibration_cutoff < test_cutoff + + assert boundaries[-1][2] is None + + +def test_window_boundaries_requires_minimum_timestamps(): + import pytest + + with pytest.raises(ValueError): + window_boundaries(np.arange(5))