Real-time Fraud Scoring with XGBoost

Solution for the use case Fraud Detection for E-Commerce Payments.

AI tools: XGBoost, scikit-learn, FastAPI, Redis, Apache Kafka (confluent-kafka), Elasticsearch
Language: Python
Published: 2025-07-09

Approach

Scores every card-not-present payment with an XGBoost model before it is authorized. The core is point-in-time feature engineering: for each transaction the service looks up the user's recent transactions and running profile in a feature store and computes velocity and behavior features such as transactions in the last 10 minutes, amount compared with the user's average, new device and foreign country.

The same feature code builds the training set (by replaying history in time order) and serves live requests, so training and serving cannot drift apart.

Each decision is approve, review (step-up verification) or block, with the three features that pushed the score up the most (XGBoost SHAP contributions) as reasons.

The review threshold is chosen on a later time period for a target false-positive rate, not hard-coded.

Optional adapters: Redis as a shared feature store, an Apache Kafka worker that scores a transaction topic, and Elasticsearch for indexing decisions.

The example trains and tests on a synthetic transaction stream. Train on your own labeled chargebacks before using it.

Implementation steps

Define the transaction event (user, amount, country, device, merchant category, timestamp) and a feature store interface with recent(), profile() and record().
Write the feature function once: velocity counts for 10 minutes and 1 hour, amount sum, amount relative to the user's mean, foreign country, new device, distinct countries in 24 hours, time since the last transaction.
Build the training set by replaying labeled transactions in time order through an in-memory store, so each row only sees its past.
Split by time (train, threshold period, test period) and train an XGBClassifier with scaleposweight for the class imbalance.
Pick the review threshold on the threshold period for a target false-positive rate and report precision, recall and false-positive rate on the test period.
Serve the model with FastAPI: POST /score returns score, action and reasons from SHAP contributions, and records the transaction in the store after scoring.
For several workers, switch to the Redis feature store (sorted set of recent transactions plus a profile hash per user).
Optionally run the Kafka worker, which scores a transactions topic, publishes decisions to a second topic and indexes them in Elasticsearch.
Retrain regularly on confirmed fraud and chargebacks, and watch the review rate and score distribution for drift.

Code (Python)

"""Real-time scoring: features from the store, XGBoost score, decision, reasons."""

from __future__ import annotations

import time
from dataclasses import asdict, dataclass

import numpy as np
import xgboost as xgb

from .features import FEATURE_NAMES, FeatureStore, Transaction, features_for


@dataclass
class Decision:
    transaction_id: str
    score: float
    action: str  # "approve" | "review" | "block"
    reasons: list[str]
    latency_ms: float

    def as_dict(self) -> dict:
        return asdict(self)


class Scorer:
    def __init__(self, model: xgb.XGBClassifier, store: FeatureStore,
                 review_threshold: float, block_threshold: float = 0.9) -> None:
        if not 0 <= review_threshold <= block_threshold <= 1:
            raise ValueError("expected 0 <= review_threshold <= block_threshold <= 1")
        self.booster = model.get_booster()
        self.booster.set_param({"nthread": 1})  # one row at a time: threads only add overhead
        self.store = store
        self.review_threshold = review_threshold
        self.block_threshold = block_threshold

    def score(self, txn: Transaction) -> Decision:
        started = time.perf_counter()
        row = np.asarray([features_for(txn, self.store)], dtype=np.float32)
        matrix = xgb.DMatrix(row, feature_names=FEATURE_NAMES)
        score = float(self.booster.predict(matrix)[0])
        # SHAP-style contributions: which features pushed the score up the most.
        contribs = self.booster.predict(matrix, pred_contribs=True)[0][:-1]
        top = np.argsort(contribs)[::-1][:3]
        reasons = [FEATURE_NAMES[i] for i in top if contribs[i] > 0]

        if score >= self.block_threshold:
            action = "block"
        elif score >= self.review_threshold:
            action = "review"
        else:
            action = "approve"
        # Record after scoring, so the transaction never sees itself in its features.
        self.store.record(txn)
        return Decision(txn.transaction_id, round(score, 4), action, reasons,
                        round((time.perf_counter() - started) * 1000, 3))

Performance

Targets, not measured results:

Scoring latency in the low milliseconds per transaction for the model and in-memory features; with Redis add one network round trip.

Target false-positive rate below 0.5 % at the review threshold (configurable with --max-fpr), with recall as high as possible at that rate.

The tests check ROC AUC above 0.95 on the synthetic stream. Synthetic fraud is much easier to separate than real fraud, so these numbers say nothing about production quality. Measure on your own labeled data.

Lessons learned

The checkout or payment service calls one HTTP endpoint and gets back a score, an action and short reasons such as txncount10m, newdevice or foreigncountry. Legitimate customers are approved without extra steps; risky payments trigger step-up verification (for example 3-D Secure) instead of a hard decline, and only very high scores are blocked. Fraud analysts can search the decisions in Elasticsearch.