Spaces:
Running
Running
| from __future__ import annotations | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| from typing import Any | |
| import numpy as np | |
| import pandas as pd | |
| try: | |
| import mlflow | |
| _MLFLOW = True | |
| except ImportError: | |
| mlflow = None | |
| _MLFLOW = False | |
| from ..data import LendingClubLoader, build_fraud_labels, LabelConfig | |
| from ..data.splitter import time_based_split, stratified_split | |
| from ..features import FeatureBuilder | |
| from ..models import XGBFraudModel, LGBMFraudModel, CatBoostFraudModel, IsolationForestFraudModel, AutoencoderFraudModel, StackingFraudEnsemble | |
| from ..evaluation import binary_classification_metrics | |
| from ..evaluation.business import cost_sensitive_evaluation, CostMatrix | |
| from ..utils.config import load_config | |
| from ..utils.io import save_joblib, save_json, ensure_dir | |
| from ..utils.logging import get_logger | |
| log = get_logger(__name__) | |
| class Trainer: | |
| config: dict | |
| feature_builder_: FeatureBuilder | None = None | |
| base_models_: dict | None = None | |
| ensemble_: StackingFraudEnsemble | None = None | |
| def run(self, mlflow_run_name: str | None=None) -> dict: | |
| cfg = self.config | |
| seed = cfg['project']['random_seed'] | |
| np.random.seed(seed) | |
| loader = LendingClubLoader(raw_path=Path(cfg['paths']['data_raw']) / cfg['data']['raw_filename'], sample_size=cfg['data'].get('sample_size'), random_seed=seed) | |
| df = loader.load() | |
| label_cfg = LabelConfig.from_dict(cfg['labels']) | |
| df = build_fraud_labels(df, label_cfg) | |
| if cfg['data']['split_strategy'] == 'time': | |
| (train, val, test) = time_based_split(df, cfg['data']['date_col'], cfg['data']['val_size'], cfg['data']['test_size']) | |
| else: | |
| (train, val, test) = stratified_split(df, cfg['data']['target_col'], cfg['data']['val_size'], cfg['data']['test_size'], random_seed=seed) | |
| y_train = train['is_fraud'] | |
| y_val = val['is_fraud'] | |
| y_test = test['is_fraud'] | |
| drop_cols = ['is_fraud', 'rule_fpd', 'rule_income_anomaly', 'rule_debt_inconsist', 'rule_address_ring', 'n_anomalies', 'loan_status', 'last_pymnt_d', 'id'] | |
| X_train_raw = train.drop(columns=[c for c in drop_cols if c in train.columns]) | |
| X_val_raw = val.drop(columns=[c for c in drop_cols if c in val.columns]) | |
| X_test_raw = test.drop(columns=[c for c in drop_cols if c in test.columns]) | |
| self.feature_builder_ = FeatureBuilder() | |
| X_train = self.feature_builder_.fit_transform(X_train_raw, y_train) | |
| X_val = self.feature_builder_.transform(X_val_raw) | |
| X_test = self.feature_builder_.transform(X_test_raw) | |
| self.base_models_ = {} | |
| mcfg = cfg['models'] | |
| if mcfg['xgboost']['enabled']: | |
| m = XGBFraudModel(params=mcfg['xgboost']['params'], monotonic_constraints=mcfg['xgboost'].get('monotonic_constraints')) | |
| m.fit(X_train, y_train, eval_set=[(X_val, y_val)]) | |
| self.base_models_['xgboost'] = m | |
| if mcfg['lightgbm']['enabled']: | |
| m = LGBMFraudModel(params=mcfg['lightgbm']['params']) | |
| m.fit(X_train, y_train, eval_set=[(X_val, y_val)]) | |
| self.base_models_['lightgbm'] = m | |
| if mcfg['catboost']['enabled']: | |
| try: | |
| m = CatBoostFraudModel(params=mcfg['catboost']['params']) | |
| m.fit(X_train, y_train, eval_set=[(X_val, y_val)]) | |
| self.base_models_['catboost'] = m | |
| except ImportError: | |
| log.warning('CatBoost not installed - skipping.') | |
| if mcfg['isolation_forest']['enabled']: | |
| m = IsolationForestFraudModel(params=mcfg['isolation_forest']['params']) | |
| m.fit(X_train) | |
| self.base_models_['isolation_forest'] = m | |
| if mcfg['autoencoder']['enabled']: | |
| m = AutoencoderFraudModel(params=mcfg['autoencoder']['params']) | |
| m.fit(X_train, y_train) | |
| self.base_models_['autoencoder'] = m | |
| if mcfg['ensemble']['enabled'] and len(self.base_models_) >= 2: | |
| self.ensemble_ = StackingFraudEnsemble(base_models=list(self.base_models_.values()), n_folds=3, calibration=mcfg['ensemble']['calibration'], random_seed=seed) | |
| self.ensemble_.fit(X_train, y_train, X_val=X_val, y_val=y_val) | |
| results: dict[str, Any] = {'per_model': {}, 'ensemble': None} | |
| for (name, m) in self.base_models_.items(): | |
| proba = m.predict_proba(X_test) | |
| results['per_model'][name] = binary_classification_metrics(y_test, proba) | |
| if self.ensemble_ is not None: | |
| proba = self.ensemble_.predict_proba(X_test) | |
| results['ensemble'] = binary_classification_metrics(y_test, proba) | |
| cost_cfg = cfg['evaluation']['cost_matrix'] | |
| cost = CostMatrix(false_negative=cost_cfg['false_negative'], false_positive=cost_cfg['false_positive']) | |
| cost_df = cost_sensitive_evaluation(y_test, proba, cost=cost) | |
| best_t = cost_df.loc[cost_df['total_cost'].idxmin()] | |
| results['best_threshold'] = float(best_t['threshold']) | |
| results['expected_cost_per_applicant'] = float(best_t['cost_per_applicant']) | |
| artifacts_dir = ensure_dir(Path(cfg['paths']['artifacts'])) | |
| save_joblib(self.feature_builder_, artifacts_dir / 'feature_builder.joblib') | |
| for (name, m) in self.base_models_.items(): | |
| save_joblib(m, artifacts_dir / f'model_{name}.joblib') | |
| if self.ensemble_ is not None: | |
| save_joblib(self.ensemble_, artifacts_dir / 'model_ensemble.joblib') | |
| save_json(results, artifacts_dir / 'results.json') | |
| if self.ensemble_ is not None: | |
| test_proba = self.ensemble_.predict_proba(X_test) | |
| else: | |
| first_model = next(iter(self.base_models_.values())) | |
| test_proba = first_model.predict_proba(X_test) | |
| pd.DataFrame({'y_true': y_test.to_numpy(), 'proba': test_proba}).to_csv(artifacts_dir / 'test_predictions.csv', index=False) | |
| try: | |
| from ..monitoring.drift import DriftMonitor | |
| monitor = DriftMonitor(psi_alert=cfg['monitoring']['drift']['psi_alert_threshold']).fit(X_train) | |
| save_joblib(monitor, artifacts_dir / 'drift_monitor.joblib') | |
| except Exception as exc: | |
| log.warning(f'Could not build drift monitor: {exc}') | |
| if _MLFLOW and mlflow is not None: | |
| import os as _os | |
| _os.environ.setdefault('MLFLOW_ALLOW_FILE_STORE', 'true') | |
| try: | |
| self._log_mlflow(results, cfg, mlflow_run_name) | |
| except Exception as exc: | |
| log.warning(f'MLflow logging skipped (non-fatal): {exc}') | |
| log.info('Training pipeline complete.') | |
| log.info(f"Ensemble metrics: {results['ensemble']}") | |
| return results | |
| def _log_mlflow(self, results: dict, cfg: dict, run_name: str | None) -> None: | |
| mlflow.set_tracking_uri(cfg['paths']['mlflow_uri']) | |
| mlflow.set_experiment(cfg['project']['name']) | |
| with mlflow.start_run(run_name=run_name or 'training-run'): | |
| mlflow.log_params({'random_seed': cfg['project']['random_seed']}) | |
| mlflow.log_param('sample_size', cfg['data'].get('sample_size')) | |
| mlflow.log_param('split_strategy', cfg['data']['split_strategy']) | |
| for (name, metrics) in results['per_model'].items(): | |
| for (k, v) in metrics.items(): | |
| if isinstance(v, (int, float)): | |
| mlflow.log_metric(f'{name}_{k}', v) | |
| if results['ensemble']: | |
| for (k, v) in results['ensemble'].items(): | |
| if isinstance(v, (int, float)): | |
| mlflow.log_metric(f'ensemble_{k}', v) | |
| if 'best_threshold' in results: | |
| mlflow.log_metric('best_threshold', results['best_threshold']) | |
| mlflow.log_metric('expected_cost_per_applicant', results['expected_cost_per_applicant']) | |
| mlflow.log_artifacts(cfg['paths']['artifacts'], 'model_artifacts') | |
| def train_from_config(config_path: str | Path='config/config.yaml') -> dict: | |
| cfg = load_config(config_path) | |
| return Trainer(config=cfg).run() | |