Spaces:
Sleeping
Sleeping
| import json | |
| import pathlib | |
| # IMPORTANT: pandas must be imported BEFORE torch on Windows. | |
| # They share an OpenMP/MKL DLL and whichever loads first wins; if torch | |
| # loads first, importing pandas afterwards crashes the process silently | |
| # (exit code 127). Keep pandas above torch. | |
| import pandas as pd | |
| import joblib | |
| import torch | |
| from src.data.preprocess import preprocess_data | |
| from src.models.model import FraudMLP | |
| # --- locate the artifacts/ folder at the project root --- | |
| # this file is src/app/inference.py, so go up 2 levels to reach the root | |
| ROOT = pathlib.Path(__file__).resolve().parents[2] | |
| ARTIFACTS = ROOT / "artifacts" | |
| # --- load the three saved artifacts ONCE when this module is imported --- | |
| with open(ARTIFACTS / "metadata.json") as f: | |
| _metadata = json.load(f) | |
| INPUT_DIM = _metadata["input_dim"] | |
| FEATURE_COLUMNS = _metadata["feature_columns"] | |
| _scaler = joblib.load(ARTIFACTS / "scaler.joblib") | |
| _model = FraudMLP(input_dim=INPUT_DIM) | |
| _model.load_state_dict(torch.load(ARTIFACTS / "model.pt", map_location="cpu")) | |
| _model.eval() # eval mode: BatchNorm/Dropout behave correctly for inference | |
| def predict(transaction: dict) -> dict: | |
| """Score a single raw transaction for fraud. | |
| transaction: a dict with the raw fields (step, type, amount, the 4 balances) | |
| returns: {"fraud_probability": float, "is_fraud": int} | |
| """ | |
| # 1. one-row DataFrame from the raw input | |
| df = pd.DataFrame([transaction]) | |
| # 2. run the SAME preprocessing used in training | |
| processed = preprocess_data(df) | |
| # 3. guard: preprocess keeps only TRANSFER/CASH_OUT — anything else -> empty | |
| if processed.empty: | |
| raise ValueError("type must be TRANSFER or CASH_OUT") | |
| # 4. force the columns to match training exactly (adds missing type_* as 0, | |
| # drops isFraud if present, orders columns like training) | |
| X = processed.reindex(columns=FEATURE_COLUMNS, fill_value=0) | |
| # 5. scale with the TRAINING scaler — transform only, never fit | |
| X_scaled = _scaler.transform(X) | |
| # 6. run the model | |
| tensor = torch.tensor(X_scaled, dtype=torch.float32) | |
| with torch.inference_mode(): | |
| logit = _model(tensor) | |
| prob = torch.sigmoid(logit).item() | |
| return { | |
| "fraud_probability": prob, | |
| "is_fraud": int(prob > 0.5), | |
| } | |