Spaces:
Running
Running
File size: 1,631 Bytes
42029e4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 | """Batch-score a CSV of applications with a trained ensemble.
Usage:
python scripts/score.py --input data/raw/new_apps.csv --output out.csv
"""
from __future__ import annotations
import sys
from pathlib import Path
import click
import pandas as pd
sys.path.insert(0, str(Path(__file__).parent.parent))
from src.utils.io import load_joblib # noqa: E402
from src.utils.logging import setup_logging, get_logger # noqa: E402
log = get_logger(__name__)
@click.command()
@click.option("--input", "input_path", required=True, type=click.Path(exists=True))
@click.option("--output", "output_path", required=True, type=click.Path())
@click.option("--artifacts", default="artifacts", show_default=True)
@click.option("--threshold", default=0.5, show_default=True)
def main(input_path: str, output_path: str, artifacts: str, threshold: float) -> None:
setup_logging()
artifacts_dir = Path(artifacts)
builder = load_joblib(artifacts_dir / "feature_builder.joblib")
model_path = artifacts_dir / "model_ensemble.joblib"
if not model_path.exists():
model_path = artifacts_dir / "model_xgboost.joblib"
model = load_joblib(model_path)
df = pd.read_csv(input_path)
log.info(f"Scoring {len(df):,} applications from {input_path}")
X = builder.transform(df)
proba = model.predict_proba(X)
df_out = df.copy()
df_out["fraud_score"] = proba
df_out["fraud_decision"] = (proba >= threshold).astype(int)
df_out.to_csv(output_path, index=False)
log.info(f"Wrote {output_path}. Fraud flag rate: {df_out['fraud_decision'].mean():.2%}")
if __name__ == "__main__":
main()
|