"""End-to-end smoke test. Runs the entire pipeline on a small synthetic dataset and verifies: 1. Data loads (synthetic fallback works) 2. Labels build 3. Split works 4. FeatureBuilder fits and transforms consistently 5. Each model fits and scores 6. Ensemble stacks and calibrates 7. Metrics compute 8. Artifacts save & load 9. ScoringService produces a decision If this exits 0, you are safe to commit. Run with: python scripts/smoke_test.py """ from __future__ import annotations import sys import tempfile import traceback from pathlib import Path # Add repo root to path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) def step(name: str): print(f"\n=== {name} ===", flush=True) def fail(msg: str, exc: Exception | None = None) -> None: print(f"\n❌ FAIL: {msg}") if exc is not None: traceback.print_exc() sys.exit(1) def main() -> None: from src.data import LendingClubLoader, build_fraud_labels, LabelConfig from src.data.splitter import time_based_split, stratified_split from src.features import FeatureBuilder from src.models import ( XGBFraudModel, LGBMFraudModel, IsolationForestFraudModel, StackingFraudEnsemble, ) from src.evaluation import binary_classification_metrics from src.evaluation.business import cost_sensitive_evaluation, CostMatrix from src.utils.io import save_joblib, load_joblib from src.api.service import ScoringService from src.api.schemas import LoanApplication from datetime import date # 1. Load synthetic data step("1. Load synthetic data") try: df = LendingClubLoader._synthetic(n=3000, seed=42) assert len(df) == 3000 print(f" loaded {len(df):,} rows, {df.shape[1]} cols") except Exception as e: fail("Synthetic loader broke", e) # 2. Build fraud labels step("2. Build fraud labels") try: df = build_fraud_labels(df, LabelConfig()) rate = df["is_fraud"].mean() print(f" fraud rate: {rate:.2%}") assert 0.01 < rate < 0.6, f"Fraud rate {rate:.2%} outside expected range" except Exception as e: fail("Label builder broke", e) # 3. Split step("3. Time-based split") try: train, val, test = stratified_split(df, "is_fraud", 0.15, 0.15, random_seed=1) print(f" train={len(train)} val={len(val)} test={len(test)}") except Exception as e: fail("Splitter broke", e) drop = ["is_fraud", "rule_fpd", "rule_income_anomaly", "rule_debt_inconsist", "rule_address_ring", "n_anomalies", "loan_status", "last_pymnt_d", "id"] y_train, y_val, y_test = train["is_fraud"], val["is_fraud"], test["is_fraud"] X_train_raw = train.drop(columns=drop) X_val_raw = val.drop(columns=drop) X_test_raw = test.drop(columns=drop) # 4. Feature builder step("4. FeatureBuilder fit+transform") try: builder = FeatureBuilder(use_velocity=True, use_graph=True) X_train = builder.fit_transform(X_train_raw, y_train) X_val = builder.transform(X_val_raw) X_test = builder.transform(X_test_raw) print(f" produced {X_train.shape[1]} features") assert X_train.shape[1] == X_val.shape[1] == X_test.shape[1] assert not X_train.isna().any().any(), "NaNs in transformed training set" except Exception as e: fail("FeatureBuilder broke", e) # 5. Train each model step("5. Train base models") try: xgb = XGBFraudModel(params={"n_estimators": 50, "max_depth": 4}).fit( X_train, y_train, eval_set=[(X_val, y_val)] ) lgb = LGBMFraudModel(params={"n_estimators": 50, "num_leaves": 15}).fit( X_train, y_train, eval_set=[(X_val, y_val)] ) iso = IsolationForestFraudModel(params={"n_estimators": 50}).fit(X_train) print(f" XGB AUC = {binary_classification_metrics(y_test, xgb.predict_proba(X_test))['roc_auc']:.3f}") print(f" LGB AUC = {binary_classification_metrics(y_test, lgb.predict_proba(X_test))['roc_auc']:.3f}") print(f" IF AUC = {binary_classification_metrics(y_test, iso.predict_proba(X_test))['roc_auc']:.3f}") except Exception as e: fail("Model training broke", e) # 6. Stacking ensemble step("6. Stacking ensemble") try: ens = StackingFraudEnsemble( base_models=[xgb, lgb, iso], n_folds=3, calibration="isotonic", ).fit(X_train, y_train, X_val=X_val, y_val=y_val) proba = ens.predict_proba(X_test) metrics = binary_classification_metrics(y_test, proba) print(f" ensemble: AUC={metrics['roc_auc']:.3f}, " f"PR-AUC={metrics['pr_auc']:.3f}, KS={metrics['ks']:.3f}") assert 0 <= proba.min() and proba.max() <= 1 except Exception as e: fail("Ensemble broke", e) # 7. Cost sweep step("7. Cost-sensitive evaluation") try: cost_df = cost_sensitive_evaluation(y_test, proba, CostMatrix()) best = cost_df.loc[cost_df["total_cost"].idxmin()] print(f" optimal threshold = {best['threshold']:.2f}, " f"cost/applicant = ${best['cost_per_applicant']:.2f}") except Exception as e: fail("Cost evaluation broke", e) # 8. Save + reload artifacts step("8. Save & load artifacts") try: with tempfile.TemporaryDirectory() as tmp: tmp_path = Path(tmp) save_joblib(builder, tmp_path / "feature_builder.joblib") save_joblib(ens, tmp_path / "model_ensemble.joblib") builder2 = load_joblib(tmp_path / "feature_builder.joblib") ens2 = load_joblib(tmp_path / "model_ensemble.joblib") import numpy as np # (a) Builder roundtrip - re-loaded builder transforms the SAME # full raw frame to identical features (velocity/graph are # batch-context dependent, so we must transform the same input). Xb1 = builder.transform(X_test_raw) Xb2 = builder2.transform(X_test_raw) assert np.allclose(Xb1.values, Xb2.values, atol=1e-6), ( "Re-loaded FeatureBuilder produces different features" ) # (b) Model roundtrip - re-loaded ensemble scores identically on # identical features. p1 = ens.predict_proba(X_test) p2 = ens2.predict_proba(X_test) assert np.allclose(p1, p2, atol=1e-6), "Re-loaded model produces different scores" print(" serialization roundtrip OK") # 9. Scoring service step("9. ScoringService end-to-end") svc = ScoringService(artifacts_dir=tmp_path) svc.load() assert svc.is_ready app = LoanApplication( loan_amnt=15000, term=36, int_rate=13.5, installment=509.0, grade="B", sub_grade="B3", emp_length=4.0, home_ownership="RENT", annual_inc=60000, verification_status="Verified", purpose="debt_consolidation", zip_code="941xx", addr_state="CA", dti=18.0, revol_util=45.0, revol_bal=5000, open_acc=8, total_acc=15, issue_d=date.today(), ) result = svc.score_one(app) print(f" score = {result.fraud_score:.4f}, decision = {result.decision}") assert 0 <= result.fraud_score <= 1 assert result.decision in ("APPROVE", "REVIEW", "DECLINE") except Exception as e: fail("Serialization or scoring service broke", e) print("\n✅ ALL SMOKE CHECKS PASSED - safe to commit.\n") if __name__ == "__main__": main()