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), }