Build a UPI Fraud Detection API

Train an XGBoost model and serve real-time fraud scores via a FastAPI REST API.

Introduction

30 Second Summary

Every time you send money through a payment app, a system silently checks whether that transaction looks suspicious. If the system catches something off, it can stop the transfer before your money disappears.

In this project, you will build that system yourself. You will train a machine learning model to detect fraudulent UPI transactions and serve its predictions through a real-time API.

What You'll Build

A production-style fraud detection API that scores UPI transactions in real time using XGBoost and FastAPI.

By the end of this project, you'll have:

  • A synthetic dataset generator that creates 10,000+ UPI transactions with embedded fraud patterns.
  • A trained XGBoost fraud detection model that scores transactions with measurable precision, recall, and ROC-AUC metrics.
  • A running FastAPI server with endpoints that accept transaction data and return real-time risk scores with ALLOW, FLAG, or BLOCK decisions.
  • A working blacklist system that automatically blocks UPI IDs after repeated fraud complaints.
  • Secret Mission: Add a fraud analytics endpoint that returns aggregated statistics including fraud rate, top blocked UPIs, and risk score distribution.

Are there any prerequisites?

You need basic Python knowledge (variables, functions, loops). No machine learning or API experience is required.

Generate Synthetic UPI Transaction Data

Your environment is ready with all packages installed. Now it's time to create the data your fraud detection model will learn from.

Real fraud datasets have anonymized features like V1 through V28 that tell you nothing about what's happening. You'll generate synthetic UPI transactions with named, meaningful features instead. This makes it easy to see exactly why a transaction looks suspicious.

In this step, get ready to:
  • Build a data generator that creates 10,000 realistic UPI transactions.
  • Inject fraud patterns that mimic real scam behavior.
  • Save the dataset to CSV and verify the class distribution.
Build the legitimate transaction generator

Your fraud model needs to learn what normal transactions look like before it can spot abnormal ones. You'll start by generating 9,500 legitimate transactions with features that mimic everyday UPI payments.

  • In VS Code, open generate_data.py from the Explorer sidebar.
  • Add the following code to define the imports, function signature, and legitimate transaction data:
import pandas as pd
import numpy as np

def generate_upi_transactions(n_samples=10000, fraud_ratio=0.05, seed=42):
    np.random.seed(seed)
    n_fraud = int(n_samples * fraud_ratio)
    n_legit = n_samples - n_fraud

    # Generate legitimate transactions
    legit_data = {
        "sender_upi": [f"user{np.random.randint(1, 500)}@upi" for _ in range(n_legit)],
        "receiver_upi": [f"merchant{np.random.randint(1, 200)}@upi" for _ in range(n_legit)],
        "amount": np.random.exponential(scale=500, size=n_legit).clip(10, 5000).round(2),
        "transaction_hour": np.random.choice(range(8, 22), size=n_legit),
        "transaction_frequency": np.random.poisson(lam=2, size=n_legit),
        "complaint_count": np.zeros(n_legit, dtype=int),
        "is_new_device": np.random.choice([0, 1], size=n_legit, p=[0.95, 0.05]),
        "is_new_ip": np.random.choice([0, 1], size=n_legit, p=[0.93, 0.07]),
        "distance_km": np.random.exponential(scale=5, size=n_legit).clip(0, 50).round(1),
        "receiver_account_age_days": np.random.randint(30, 1000, size=n_legit),
        "amount_deviation": np.random.normal(loc=0, scale=0.3, size=n_legit).clip(-1, 1).round(3),
        "fraud_label": np.zeros(n_legit, dtype=int),
    }

What does this code do?

This code creates a function that generates realistic-looking UPI payment data. The seed=42 makes results reproducible so you get the same data each time you run it.

Each feature captures something a fraud system would measure in real life:

  • amount uses an exponential distribution (most payments are small, few are large).
  • transaction_hour ranges from 8am to 10pm for legitimate users.
  • is_new_device is 0 (same device) 95% of the time for normal users.
  • amount_deviation measures how far this payment is from the sender's usual amounts.

Why synthetic data instead of a real dataset?

Real fraud datasets (like the Kaggle Credit Card dataset) have anonymized features named V1 through V28. You can't tell what any feature actually means. Synthetic data with named features like complaint_count and distance_km makes it clear why the model flags a transaction.

  • Save generate_data.py.

You won't see a change yet. The next chunk adds the fraud patterns that make this data useful for training.

Inject fraud patterns into the dataset

Real fraudsters exhibit specific behaviors: they transact at odd hours, use new devices, target brand-new accounts, and move fast. You'll inject these patterns into 500 fraudulent transactions so the model can learn to recognize them.

  • In generate_data.py, find the closing brace } of the legit_data dictionary. Add the following code directly below it:
    # Generate fraudulent transactions with suspicious patterns
    fraud_data = {
        "sender_upi": [f"user{np.random.randint(1, 500)}@upi" for _ in range(n_fraud)],
        "receiver_upi": [f"scam{np.random.randint(1, 30)}@upi" for _ in range(n_fraud)],
        "amount": np.random.exponential(scale=3000, size=n_fraud).clip(1000, 50000).round(2),
        "transaction_hour": np.random.choice([0, 1, 2, 3, 4, 5, 23], size=n_fraud),
        "transaction_frequency": np.random.poisson(lam=8, size=n_fraud) + 5,
        "complaint_count": np.random.randint(2, 15, size=n_fraud),
        "is_new_device": np.random.choice([0, 1], size=n_fraud, p=[0.3, 0.7]),
        "is_new_ip": np.random.choice([0, 1], size=n_fraud, p=[0.25, 0.75]),
        "distance_km": np.random.exponential(scale=50, size=n_fraud).clip(10, 500).round(1),
        "receiver_account_age_days": np.random.randint(1, 20, size=n_fraud),
        "amount_deviation": np.random.normal(loc=2.5, scale=0.8, size=n_fraud).clip(1, 5).round(3),
        "fraud_label": np.ones(n_fraud, dtype=int),
    }

What makes these transactions look fraudulent?

Compare the fraud data to the legitimate data and notice the differences:

  • Amounts are much larger (scale=3000 vs scale=500) and clipped between 1,000 and 50,000.
  • Transactions happen at midnight to 5am and 11pm instead of business hours.
  • Receiver accounts are less than 20 days old (brand-new accounts created for scams).
  • 70% use a new device and 75% have a new IP address.
  • Transaction frequency is much higher (poisson(8) + 5 vs poisson(2)).

These are the exact patterns the model will learn to flag during training.

  • Save generate_data.py.

You won't see output yet. The next chunk combines both datasets and adds the script runner.

Combine the data and run the generator

With both legitimate and fraudulent data defined, you need to merge them into a single shuffled dataset. Shuffling ensures the model doesn't learn the order of rows during training.

  • In generate_data.py, add the following code below the closing brace } of the fraud_data dictionary:
    legit_df = pd.DataFrame(legit_data)
    fraud_df = pd.DataFrame(fraud_data)
    df = pd.concat([legit_df, fraud_df], ignore_index=True).sample(frac=1, random_state=seed).reset_index(drop=True)

    return df


if __name__ == "__main__":
    df = generate_upi_transactions()
    df.to_csv("upi_transactions.csv", index=False)
    print(f"Generated {len(df)} transactions")
    print(f"Fraud: {df['fraud_label'].sum()} ({df['fraud_label'].mean()*100:.1f}%)")
    print(f"Legitimate: {(1 - df['fraud_label']).sum().astype(int)}")
    print(f"\nSample:\n{df.head()}")

What does this code do?

  • pd.concat merges the legitimate and fraud DataFrames into one.
  • .sample(frac=1) shuffles all rows randomly so fraud isn't grouped at the end.
  • The if __name__ == "__main__" block runs only when you execute the script directly. It saves the CSV and prints summary stats.
  • Save generate_data.py.
  • Run the script by entering this command in your terminal:
python generate_data.py

You should see output showing 10,000 total transactions with 500 fraud cases (5.0%) and 9,500 legitimate cases, followed by a sample of the first 5 rows.

  • Confirm upi_transactions.csv was created by running:
ls

You should see upi_transactions.csv listed alongside your other project files.

Not seeing the expected output?

  • If you see IndentationError, make sure the legit_data and fraud_data dictionaries are indented inside the function (4 spaces).
  • If you see ModuleNotFoundError: No module named 'pandas', your virtual environment may not be activated. Run .\venv\Scripts\activate (Windows) or source venv/bin/activate (macOS) and try again.
  • If the fraud count is not exactly 500, check that your function signature uses fraud_ratio=0.05 and n_samples=10000.

Why is my generate_data.py script throwing an error?

✔️ Awesome, I've got everything!

Great. Double check you've saved your file.

ⓧ I'd like to double check the full code

Here's the complete generate_data.py file for comparison:

import pandas as pd
import numpy as np

def generate_upi_transactions(n_samples=10000, fraud_ratio=0.05, seed=42):
    np.random.seed(seed)
    n_fraud = int(n_samples * fraud_ratio)
    n_legit = n_samples - n_fraud

    # Generate legitimate transactions
    legit_data = {
        "sender_upi": [f"user{np.random.randint(1, 500)}@upi" for _ in range(n_legit)],
        "receiver_upi": [f"merchant{np.random.randint(1, 200)}@upi" for _ in range(n_legit)],
        "amount": np.random.exponential(scale=500, size=n_legit).clip(10, 5000).round(2),
        "transaction_hour": np.random.choice(range(8, 22), size=n_legit),
        "transaction_frequency": np.random.poisson(lam=2, size=n_legit),
        "complaint_count": np.zeros(n_legit, dtype=int),
        "is_new_device": np.random.choice([0, 1], size=n_legit, p=[0.95, 0.05]),
        "is_new_ip": np.random.choice([0, 1], size=n_legit, p=[0.93, 0.07]),
        "distance_km": np.random.exponential(scale=5, size=n_legit).clip(0, 50).round(1),
        "receiver_account_age_days": np.random.randint(30, 1000, size=n_legit),
        "amount_deviation": np.random.normal(loc=0, scale=0.3, size=n_legit).clip(-1, 1).round(3),
        "fraud_label": np.zeros(n_legit, dtype=int),
    }

    # Generate fraudulent transactions with suspicious patterns
    fraud_data = {
        "sender_upi": [f"user{np.random.randint(1, 500)}@upi" for _ in range(n_fraud)],
        "receiver_upi": [f"scam{np.random.randint(1, 30)}@upi" for _ in range(n_fraud)],
        "amount": np.random.exponential(scale=3000, size=n_fraud).clip(1000, 50000).round(2),
        "transaction_hour": np.random.choice([0, 1, 2, 3, 4, 5, 23], size=n_fraud),
        "transaction_frequency": np.random.poisson(lam=8, size=n_fraud) + 5,
        "complaint_count": np.random.randint(2, 15, size=n_fraud),
        "is_new_device": np.random.choice([0, 1], size=n_fraud, p=[0.3, 0.7]),
        "is_new_ip": np.random.choice([0, 1], size=n_fraud, p=[0.25, 0.75]),
        "distance_km": np.random.exponential(scale=50, size=n_fraud).clip(10, 500).round(1),
        "receiver_account_age_days": np.random.randint(1, 20, size=n_fraud),
        "amount_deviation": np.random.normal(loc=2.5, scale=0.8, size=n_fraud).clip(1, 5).round(3),
        "fraud_label": np.ones(n_fraud, dtype=int),
    }

    legit_df = pd.DataFrame(legit_data)
    fraud_df = pd.DataFrame(fraud_data)
    df = pd.concat([legit_df, fraud_df], ignore_index=True).sample(frac=1, random_state=seed).reset_index(drop=True)

    return df


if __name__ == "__main__":
    df = generate_upi_transactions()
    df.to_csv("upi_transactions.csv", index=False)
    print(f"Generated {len(df)} transactions")
    print(f"Fraud: {df['fraud_label'].sum()} ({df['fraud_label'].mean()*100:.1f}%)")
    print(f"Legitimate: {(1 - df['fraud_label']).sum().astype(int)}")
    print(f"\nSample:\n{df.head()}")

You now have 10,000 labeled transactions ready for training. Next, you'll feed this data into an XGBoost classifier that learns to tell fraud apart from legitimate payments.

Train the XGBoost Fraud Detection Model

You now have 10,000 synthetic UPI transactions sitting in a CSV file. The next challenge is teaching a machine to spot the 5% that are fraudulent.

In this step, you will train an XGBoost classifier to detect fraud patterns in your data. The model will learn which combinations of features (high amounts, new devices, rapid transactions) signal fraud, even though fraudulent transactions are rare.

In this step, get ready to:
  • Load the dataset and split it into training and testing sets.
  • Train an XGBoost classifier tuned for imbalanced fraud data.
  • Evaluate model performance and save the trained model to disk.
Load and split the dataset

Before a model can learn, you need to separate your data into inputs (features) and the answer you want it to predict (the fraud label). You also need a held-out test set that the model never sees during training, so you can measure its real-world accuracy.

  • Create a new file called train_model.py in your upi-fraud-detection folder.
  • Add the imports and feature column list by pasting this code:
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, roc_auc_score
from xgboost import XGBClassifier
import joblib

# The 9 numeric features the model will learn from
FEATURE_COLUMNS = [
    "amount",
    "transaction_hour",
    "transaction_frequency",
    "complaint_count",
    "is_new_device",
    "is_new_ip",
    "distance_km",
    "receiver_account_age_days",
    "amount_deviation",
]

What does this code do?

The imports pull in your ML toolkit: scikit-learn for splitting data and evaluation metrics, XGBoost for the model itself, and joblib for saving the trained model to disk.

FEATURE_COLUMNS lists the 9 numeric columns the model will use as inputs. Text fields like sender_upi are excluded because the model needs numbers to learn patterns.

  • Below the FEATURE_COLUMNS list, add the function that loads and splits the data:

def train_fraud_model(data_path="upi_transactions.csv"):
    df = pd.read_csv(data_path)

    # Separate features (X) from the target label (y)
    X = df[FEATURE_COLUMNS]
    y = df["fraud_label"]

    # Split into 80% training, 20% testing
    # stratify=y keeps the same fraud ratio in both sets
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )

What does stratify do?

Your dataset is 5% fraud and 95% legitimate. Without stratify=y, random splitting could accidentally put most fraud cases in training and almost none in testing.

Stratification guarantees both sets maintain that 5%/95% ratio. This gives you a fair test of how the model handles fraud in the real world.

  • Save train_model.py.

You will not see output yet. The function is defined but not called. The next chunk adds the model training logic.

Configure and train the XGBoost model

Fraud is only 5% of your data. If the model simply guessed "not fraud" every time, it would be 95% accurate but completely useless. The scale_pos_weight parameter tells XGBoost to pay extra attention to the rare fraud cases during training.

  • Add the model configuration and training code inside the train_fraud_model() function, directly below the train_test_split call:
    # Calculate scale_pos_weight for class imbalance
    neg_count = (y_train == 0).sum()
    pos_count = (y_train == 1).sum()
    scale_weight = neg_count / pos_count

    model = XGBClassifier(
        n_estimators=200,
        max_depth=6,
        learning_rate=0.1,
        scale_pos_weight=scale_weight,
        objective="binary:logistic",
        eval_metric="aucpr",
        random_state=42,
    )

    model.fit(X_train, y_train)

What do these parameters mean?

  • scale_pos_weight is calculated as negatives divided by positives (about 19). This makes the model treat one fraud sample as 19 times more important than one legitimate sample.
  • n_estimators=200 builds 200 decision trees that each correct the previous tree's mistakes.
  • objective="binary:logistic" outputs a probability between 0 and 1, which you will later map to a 0-100 risk score.
  • eval_metric="aucpr" uses precision-recall area under the curve, which is more informative than accuracy for imbalanced datasets.

Why XGBoost instead of Random Forest?

XGBoost provides native scale_pos_weight for handling class imbalance without needing extra dependencies like SMOTE. It also produces better-calibrated probabilities, which matters when you map predictions to a 0-100 risk score.

  • Save train_model.py.

You will not see output yet. The next chunk adds evaluation and runs the script.

Evaluate and save the trained model

Training a model is not useful if you cannot measure how well it works. You need metrics that go beyond simple accuracy, because with imbalanced data, accuracy is misleading.

  • Add the evaluation, saving, and entry point code below model.fit(X_train, y_train) inside the function:
    # Evaluate
    y_pred = model.predict(X_test)
    y_proba = model.predict_proba(X_test)[:, 1]

    print("Classification Report:")
    print(classification_report(y_test, y_pred))
    print(f"ROC-AUC Score: {roc_auc_score(y_test, y_proba):.4f}")

    # Save model
    joblib.dump(model, "fraud_model.joblib")
    print("\nModel saved to fraud_model.joblib")


if __name__ == "__main__":
    train_fraud_model()

What do these metrics tell you?

  • Precision answers: of all transactions the model flagged as fraud, how many actually were fraud?
  • Recall answers: of all actual fraud cases, how many did the model catch?
  • ROC-AUC measures the model's ability to separate fraud from legitimate across all possible thresholds. A score of 1.0 is perfect. A score of 0.5 is random guessing.
  • joblib.dump() saves the trained model to a file so your API can load it later without retraining.
  • Save train_model.py.
  • Train the model by running this command in your terminal:
python train_model.py

You should see a classification report with precision, recall, and F1 scores for both classes (0 and 1), followed by a ROC-AUC score and a confirmation that the model was saved to fraud_model.joblib.

  • Confirm the model file was created by running:
ls fraud_model.joblib

You should see fraud_model.joblib listed in the output.

Seeing a FileNotFoundError?

  • Make sure you ran python generate_data.py in the previous step. The training script looks for upi_transactions.csv in the same folder.
  • Check that your terminal is in the upi-fraud-detection folder, not a subfolder.

Help me debug my train_model.py script

✔️ Awesome, I've got everything!

Great. Double check you have saved train_model.py and that fraud_model.joblib exists in your project folder.

ⓧ I'd like to double check the full code

Here is the complete train_model.py file for comparison:

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, roc_auc_score
from xgboost import XGBClassifier
import joblib

FEATURE_COLUMNS = [
    "amount",
    "transaction_hour",
    "transaction_frequency",
    "complaint_count",
    "is_new_device",
    "is_new_ip",
    "distance_km",
    "receiver_account_age_days",
    "amount_deviation",
]


def train_fraud_model(data_path="upi_transactions.csv"):
    df = pd.read_csv(data_path)

    X = df[FEATURE_COLUMNS]
    y = df["fraud_label"]

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )

    # Calculate scale_pos_weight for class imbalance
    neg_count = (y_train == 0).sum()
    pos_count = (y_train == 1).sum()
    scale_weight = neg_count / pos_count

    model = XGBClassifier(
        n_estimators=200,
        max_depth=6,
        learning_rate=0.1,
        scale_pos_weight=scale_weight,
        objective="binary:logistic",
        eval_metric="aucpr",
        random_state=42,
    )

    model.fit(X_train, y_train)

    # Evaluate
    y_pred = model.predict(X_test)
    y_proba = model.predict_proba(X_test)[:, 1]

    print("Classification Report:")
    print(classification_report(y_test, y_pred))
    print(f"ROC-AUC Score: {roc_auc_score(y_test, y_proba):.4f}")

    # Save model
    joblib.dump(model, "fraud_model.joblib")
    print("\nModel saved to fraud_model.joblib")


if __name__ == "__main__":
    train_fraud_model()

Your fraud detection model is trained and saved. Next up, you will wrap it in a FastAPI application so it can score transactions in real time through HTTP requests.

Build the FastAPI Fraud Scoring API

Your trained model can detect fraud, but it lives as a file on disk. To use it in a real system, other applications need a way to send transaction data and get a risk score back instantly.

In this step, you will wrap your model in a FastAPI application that accepts transaction data over HTTP and returns a fraud risk assessment. This is how production fraud systems work. They sit between the payment request and the payment processor, scoring every transaction in milliseconds.

In this step, get ready to:
  • Define Pydantic schemas for transaction input and API responses.
  • Build the fraud scoring engine that loads and runs the trained model.
  • Create the FastAPI application with a /predict endpoint for real-time scoring.
Define the Pydantic schemas

FastAPI uses Pydantic models to validate incoming request data and shape outgoing responses. Each model defines the exact fields, types, and default values that the API expects or returns.

  • Open app/schemas.py in VS Code and paste the following code:
from pydantic import BaseModel


# Input schema for the /predict endpoint
class TransactionInput(BaseModel):
    sender_upi: str
    receiver_upi: str
    amount: float
    transaction_hour: int
    transaction_frequency: int
    complaint_count: int = 0
    is_new_device: int = 0
    is_new_ip: int = 0
    distance_km: float = 0.0
    receiver_account_age_days: int = 365
    amount_deviation: float = 0.0


# Output schema for the /predict endpoint
class PredictionResponse(BaseModel):
    risk_score: int
    decision: str
    reason_codes: list[str]
    receiver_upi: str

What does this code do?

TransactionInput defines every field the API expects when scoring a transaction. Fields with default values (like complaint_count = 0) are optional in the request. The caller only needs to provide sender_upi, receiver_upi, amount, transaction_hour, and transaction_frequency.

PredictionResponse defines what the API sends back: a 0-100 risk score, a decision string (ALLOW, FLAG, or BLOCK), a list of reason codes explaining why, and the receiver UPI that was checked.

Your API will also need schemas for the complaint and blacklist system you will build in the next step. Adding them now keeps all schemas in one place.

  • Add the following classes below the PredictionResponse class in the same file:
# Input schema for the /report-fraud endpoint
class FraudReportInput(BaseModel):
    upi_id: str
    reason: str


# Output schema for the /report-fraud endpoint
class FraudReportResponse(BaseModel):
    upi_id: str
    total_complaints: int
    is_blocked: bool
    message: str


# Schema for blacklist entries
class BlacklistEntry(BaseModel):
    upi_id: str
    complaint_count: int

What does this code do?

FraudReportInput accepts a UPI ID and a reason string when someone reports fraud. FraudReportResponse tells the caller how many complaints that UPI has and whether it crossed the blocking threshold.

BlacklistEntry is a simple schema for listing blocked UPI IDs. You will use these schemas in Step 5 when you add the complaint endpoints.

  • Save app/schemas.py.

Seeing a red underline on list[str]?

The list[str] type annotation requires Python 3.9+. If VS Code shows an error, confirm you are using Python 3.11 or later in your virtual environment.

Help me fix my Python version for type annotations.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before moving on.

ⓧ I'd like to double check the full code

from pydantic import BaseModel


class TransactionInput(BaseModel):
    sender_upi: str
    receiver_upi: str
    amount: float
    transaction_hour: int
    transaction_frequency: int
    complaint_count: int = 0
    is_new_device: int = 0
    is_new_ip: int = 0
    distance_km: float = 0.0
    receiver_account_age_days: int = 365
    amount_deviation: float = 0.0


class PredictionResponse(BaseModel):
    risk_score: int
    decision: str
    reason_codes: list[str]
    receiver_upi: str


class FraudReportInput(BaseModel):
    upi_id: str
    reason: str


class FraudReportResponse(BaseModel):
    upi_id: str
    total_complaints: int
    is_blocked: bool
    message: str


class BlacklistEntry(BaseModel):
    upi_id: str
    complaint_count: int
Build the fraud scoring engine

The fraud engine is the bridge between raw transaction data and your trained model. It loads the model from disk, converts incoming transaction fields into the feature array the model expects, and translates the raw probability into a human-readable risk score with decision logic.

  • Open app/fraud_engine.py and paste the imports and feature column list:
import numpy as np
import joblib

# Same feature order the model was trained on
FEATURE_COLUMNS = [
    "amount",
    "transaction_hour",
    "transaction_frequency",
    "complaint_count",
    "is_new_device",
    "is_new_ip",
    "distance_km",
    "receiver_account_age_days",
    "amount_deviation",
]

Why define FEATURE_COLUMNS again?

The model expects features in the exact same order it was trained on. This list mirrors the one in train_model.py. If the order is wrong, the model will assign feature importance to the wrong values and produce garbage scores.

  • Below the FEATURE_COLUMNS list, add the FraudEngine class:
class FraudEngine:
    def __init__(self, model_path="fraud_model.joblib"):
        self.model = joblib.load(model_path)

    def score_transaction(self, features: dict) -> dict:
        """Score a transaction and return risk assessment."""
        feature_array = np.array(
            [[features[col] for col in FEATURE_COLUMNS]]
        )
        fraud_probability = self.model.predict_proba(feature_array)[0][1]
        risk_score = int(fraud_probability * 100)

        decision = self._get_decision(risk_score)
        reason_codes = self._get_reason_codes(features, risk_score)

        return {
            "risk_score": risk_score,
            "decision": decision,
            "reason_codes": reason_codes,
        }

    def _get_decision(self, risk_score: int) -> str:
        if risk_score < 30:
            return "ALLOW"
        elif risk_score < 70:
            return "FLAG"
        else:
            return "BLOCK"

    def _get_reason_codes(self, features: dict, risk_score: int) -> list[str]:
        codes = []
        if features["complaint_count"] >= 3:
            codes.append("MULTIPLE_FRAUD_COMPLAINTS")
        if features["amount_deviation"] > 2.0:
            codes.append("UNUSUAL_TRANSACTION_AMOUNT")
        if features["transaction_frequency"] > 8:
            codes.append("RAPID_TRANSACTION_ACTIVITY")
        if features["is_new_device"] == 1:
            codes.append("DEVICE_MISMATCH")
        if features["distance_km"] > 100:
            codes.append("SUSPICIOUS_LOCATION_CHANGE")
        if features["receiver_account_age_days"] < 7:
            codes.append("HIGH_RISK_RECEIVER")
        if features["is_new_ip"] == 1 and features["is_new_device"] == 1:
            codes.append("ACCOUNT_TAKEOVER_RISK")
        if not codes and risk_score >= 30:
            codes.append("ELEVATED_RISK_PATTERN")
        return codes


# Create a module-level instance so it loads the model once at startup
fraud_engine = FraudEngine()

What does this code do?

The constructor loads the trained model from disk using joblib. This happens once when the module is imported, not on every request.

score_transaction() converts the features dictionary into a numpy array in the correct column order, calls predict_proba() to get the fraud probability (0.0 to 1.0), then multiplies by 100 to get a 0-100 risk score.

_get_decision() maps the score to a three-tier decision. Below 30 is ALLOW, 30 to 69 is FLAG for manual review, and 70 or above is BLOCK. _get_reason_codes() checks individual feature thresholds to explain why a transaction was flagged.

  • Save app/fraud_engine.py.

Seeing a FileNotFoundError for fraud_model.joblib?

The fraud engine expects fraud_model.joblib in the project root (the upi-fraud-detection folder). Make sure you ran train_model.py in the previous step and that the file exists.

Help me find my fraud_model.joblib file.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before moving on.

ⓧ I'd like to double check the full code

import numpy as np
import joblib

FEATURE_COLUMNS = [
    "amount",
    "transaction_hour",
    "transaction_frequency",
    "complaint_count",
    "is_new_device",
    "is_new_ip",
    "distance_km",
    "receiver_account_age_days",
    "amount_deviation",
]


class FraudEngine:
    def __init__(self, model_path="fraud_model.joblib"):
        self.model = joblib.load(model_path)

    def score_transaction(self, features: dict) -> dict:
        """Score a transaction and return risk assessment."""
        feature_array = np.array(
            [[features[col] for col in FEATURE_COLUMNS]]
        )
        fraud_probability = self.model.predict_proba(feature_array)[0][1]
        risk_score = int(fraud_probability * 100)

        decision = self._get_decision(risk_score)
        reason_codes = self._get_reason_codes(features, risk_score)

        return {
            "risk_score": risk_score,
            "decision": decision,
            "reason_codes": reason_codes,
        }

    def _get_decision(self, risk_score: int) -> str:
        if risk_score < 30:
            return "ALLOW"
        elif risk_score < 70:
            return "FLAG"
        else:
            return "BLOCK"

    def _get_reason_codes(self, features: dict, risk_score: int) -> list[str]:
        codes = []
        if features["complaint_count"] >= 3:
            codes.append("MULTIPLE_FRAUD_COMPLAINTS")
        if features["amount_deviation"] > 2.0:
            codes.append("UNUSUAL_TRANSACTION_AMOUNT")
        if features["transaction_frequency"] > 8:
            codes.append("RAPID_TRANSACTION_ACTIVITY")
        if features["is_new_device"] == 1:
            codes.append("DEVICE_MISMATCH")
        if features["distance_km"] > 100:
            codes.append("SUSPICIOUS_LOCATION_CHANGE")
        if features["receiver_account_age_days"] < 7:
            codes.append("HIGH_RISK_RECEIVER")
        if features["is_new_ip"] == 1 and features["is_new_device"] == 1:
            codes.append("ACCOUNT_TAKEOVER_RISK")
        if not codes and risk_score >= 30:
            codes.append("ELEVATED_RISK_PATTERN")
        return codes


fraud_engine = FraudEngine()
Create the FastAPI application

The /predict endpoint needs to check whether a receiver UPI is blacklisted before running the model. This means your main app depends on a blacklist module. You will build out the full complaint system in the next step, but the module needs to exist now for the import to work.

  • Open app/blacklist.py and paste the following code:
class BlacklistManager:
    def __init__(self, block_threshold=3):
        self.complaints: dict[str, int] = {}
        self.block_threshold = block_threshold

    def report(self, upi_id: str) -> tuple[int, bool]:
        """Report a UPI ID. Returns (total_complaints, newly_blocked)."""
        if upi_id not in self.complaints:
            self.complaints[upi_id] = 0
        self.complaints[upi_id] += 1
        count = self.complaints[upi_id]
        newly_blocked = count == self.block_threshold
        return count, newly_blocked

    def is_blocked(self, upi_id: str) -> bool:
        return self.complaints.get(upi_id, 0) >= self.block_threshold

    def get_all_blocked(self) -> list[dict]:
        return [
            {"upi_id": upi_id, "complaint_count": count}
            for upi_id, count in self.complaints.items()
            if count >= self.block_threshold
        ]


# Module-level instance shared across all endpoints
blacklist_manager = BlacklistManager()

What does this code do?

BlacklistManager stores complaint counts in a plain Python dictionary. When a UPI ID receives 3 or more complaints, is_blocked() returns True. The /predict endpoint will use this to immediately block transactions to known scam accounts.

You will add the endpoints that call report() and get_all_blocked() in the next step. For now, the module just needs to exist so that app/main.py can import it.

  • Save app/blacklist.py.

Now you can wire everything together in the main application file.

  • Open app/main.py and paste the imports and app setup:
from fastapi import FastAPI
from app.schemas import (
    TransactionInput,
    PredictionResponse,
    FraudReportInput,
    FraudReportResponse,
    BlacklistEntry,
)
from app.fraud_engine import fraud_engine
from app.blacklist import blacklist_manager

# Create the FastAPI application instance
app = FastAPI(
    title="UPI Fraud Detection API",
    description="Real-time fraud scoring for UPI transactions",
    version="1.0.0",
)


@app.get("/health")
def health_check():
    return {"status": "healthy", "model_loaded": True}

What does this code do?

The imports pull in your schemas, the fraud engine (which loads the model), and the blacklist manager. FastAPI() creates the application with metadata that appears in the auto-generated documentation.

The /health endpoint is a simple GET route that confirms the server is running and the model loaded successfully. Production APIs always have a health check for monitoring tools to ping.

  • Below the health_check function, add the /predict endpoint:
@app.post("/predict", response_model=PredictionResponse)
def predict_fraud(transaction: TransactionInput):
    # Check blacklist first
    if blacklist_manager.is_blocked(transaction.receiver_upi):
        return PredictionResponse(
            risk_score=100,
            decision="BLOCK",
            reason_codes=["BLACKLISTED_UPI"],
            receiver_upi=transaction.receiver_upi,
        )

    features = {
        "amount": transaction.amount,
        "transaction_hour": transaction.transaction_hour,
        "transaction_frequency": transaction.transaction_frequency,
        "complaint_count": transaction.complaint_count,
        "is_new_device": transaction.is_new_device,
        "is_new_ip": transaction.is_new_ip,
        "distance_km": transaction.distance_km,
        "receiver_account_age_days": transaction.receiver_account_age_days,
        "amount_deviation": transaction.amount_deviation,
    }

    result = fraud_engine.score_transaction(features)

    return PredictionResponse(
        risk_score=result["risk_score"],
        decision=result["decision"],
        reason_codes=result["reason_codes"],
        receiver_upi=transaction.receiver_upi,
    )

What does this code do?

The /predict endpoint first checks if the receiver UPI is on the blacklist. If it is, the endpoint immediately returns a BLOCK decision with a risk score of 100. No model inference needed for known bad actors.

If the receiver is not blacklisted, the endpoint extracts the transaction features into a dictionary and passes them to the fraud engine. The engine runs the model, calculates the risk score, and returns the full assessment.

  • Save app/main.py.

Seeing import errors in VS Code?

VS Code may show red underlines on the from app.schemas imports. This is because VS Code does not know the working directory yet. These will resolve when you start the server from the project root.

Help me fix import resolution in VS Code.

✔️ Awesome, I've got everything!

Great. Make sure you have saved all files before starting the server.

ⓧ I'd like to double check the full code

from fastapi import FastAPI
from app.schemas import (
    TransactionInput,
    PredictionResponse,
    FraudReportInput,
    FraudReportResponse,
    BlacklistEntry,
)
from app.fraud_engine import fraud_engine
from app.blacklist import blacklist_manager

app = FastAPI(
    title="UPI Fraud Detection API",
    description="Real-time fraud scoring for UPI transactions",
    version="1.0.0",
)


@app.get("/health")
def health_check():
    return {"status": "healthy", "model_loaded": True}


@app.post("/predict", response_model=PredictionResponse)
def predict_fraud(transaction: TransactionInput):
    # Check blacklist first
    if blacklist_manager.is_blocked(transaction.receiver_upi):
        return PredictionResponse(
            risk_score=100,
            decision="BLOCK",
            reason_codes=["BLACKLISTED_UPI"],
            receiver_upi=transaction.receiver_upi,
        )

    features = {
        "amount": transaction.amount,
        "transaction_hour": transaction.transaction_hour,
        "transaction_frequency": transaction.transaction_frequency,
        "complaint_count": transaction.complaint_count,
        "is_new_device": transaction.is_new_device,
        "is_new_ip": transaction.is_new_ip,
        "distance_km": transaction.distance_km,
        "receiver_account_age_days": transaction.receiver_account_age_days,
        "amount_deviation": transaction.amount_deviation,
    }

    result = fraud_engine.score_transaction(features)

    return PredictionResponse(
        risk_score=result["risk_score"],
        decision=result["decision"],
        reason_codes=result["reason_codes"],
        receiver_upi=transaction.receiver_upi,
    )

Time to start the server and confirm everything works together.

  • Make sure your terminal is in the upi-fraud-detection folder with your virtual environment activated.
  • Start the FastAPI development server by running:
fastapi dev app/main.py
  • You should see output showing the server started at http://127.0.0.1:8000.
  • Open your browser and navigate to http://127.0.0.1:8000/docs to see the interactive Swagger documentation.

You should see a page titled UPI Fraud Detection API with your /health and /predict endpoints listed.

What is Swagger?

FastAPI automatically generates interactive API documentation at the /docs path. You can test every endpoint directly in the browser by clicking Try it out, filling in the request body, and clicking Execute. No separate API testing tool needed.

Server not starting?

If you see ModuleNotFoundError: No module named 'app', make sure you are running the command from the upi-fraud-detection folder (the one that contains the app/ directory).

If you see FileNotFoundError: fraud_model.joblib, the model file is not in the project root. Run ls to check that fraud_model.joblib is in the same folder as your app/ directory.

Help me debug my FastAPI server startup error.

Your fraud detection API is live. Next up, you will add the complaint and blacklist endpoints so that reported scam accounts get automatically blocked.

Add the Blacklist and Complaint System

Your ML model can now score transactions for fraud risk. However, real payment systems also need a way for users to report fraud after it happens.

In this step, you will build a complaint-driven blacklist that automatically blocks UPI IDs after repeated reports. Once a UPI ID is blocked, the /predict endpoint will skip the model entirely and return an instant BLOCK decision.

In this step, get ready to:
  • Build an in-memory blacklist manager that tracks complaints and auto-blocks after 3 reports.
  • Add a POST /report-fraud endpoint for submitting fraud complaints.
  • Add a GET /blacklist endpoint and verify the full blacklist integration works.
Build the BlacklistManager class

The blacklist manager tracks how many fraud complaints each UPI ID receives. Once a UPI ID crosses the threshold (3 complaints), every future transaction involving that ID gets blocked instantly.

  • In VS Code, open app/blacklist.py (currently empty) and paste the following code:
class BlacklistManager:
    def __init__(self, block_threshold=3):
        self.complaints: dict[str, int] = {}
        self.block_threshold = block_threshold

    def report(self, upi_id: str) -> tuple[int, bool]:
        """Report a UPI ID. Returns (total_complaints, newly_blocked)."""
        if upi_id not in self.complaints:
            self.complaints[upi_id] = 0
        self.complaints[upi_id] += 1
        count = self.complaints[upi_id]
        newly_blocked = count == self.block_threshold
        return count, newly_blocked

    def is_blocked(self, upi_id: str) -> bool:
        return self.complaints.get(upi_id, 0) >= self.block_threshold

    def get_all_blocked(self) -> list[dict]:
        return [
            {"upi_id": upi_id, "complaint_count": count}
            for upi_id, count in self.complaints.items()
            if count >= self.block_threshold
        ]


blacklist_manager = BlacklistManager()

What does this code do?

The BlacklistManager uses a Python dictionary to count complaints per UPI ID. The key methods are:

  • report() increments the complaint counter and returns whether the UPI just crossed the blocking threshold.
  • is_blocked() checks if a UPI ID has 3 or more complaints.
  • get_all_blocked() returns all blocked UPI IDs with their complaint counts for the /blacklist endpoint.

The module-level blacklist_manager instance is a singleton that persists in memory as long as the server is running.

  • Save app/blacklist.py.

Why an in-memory dictionary instead of a database?

A production system would store complaints in a database like PostgreSQL. For this project, the in-memory dictionary demonstrates the blacklist pattern without adding database setup time.

The tradeoff is that all complaint data resets when you restart the server. The pattern itself (count complaints, compare to threshold, block) is identical to what you would build with a real database.

Add the /report-fraud endpoint

Users need a way to report fraudulent UPI IDs. This endpoint accepts a complaint, increments the counter, and tells the caller whether the reported UPI just got blocked.

  • In app/main.py, scroll to the bottom of the file (after the predict_fraud function) and add the following endpoint:
@app.post("/report-fraud", response_model=FraudReportResponse)
def report_fraud(report: FraudReportInput):
    total_complaints, newly_blocked = blacklist_manager.report(report.upi_id)

    if newly_blocked:
        message = f"UPI ID {report.upi_id} has been BLOCKED after {total_complaints} complaints."
    elif blacklist_manager.is_blocked(report.upi_id):
        message = f"UPI ID {report.upi_id} is already blocked. Complaint recorded."
    else:
        message = f"Complaint recorded. {blacklist_manager.block_threshold - total_complaints} more needed to block."

    return FraudReportResponse(
        upi_id=report.upi_id,
        total_complaints=total_complaints,
        is_blocked=blacklist_manager.is_blocked(report.upi_id),
        message=message,
    )

What does this endpoint do?

The endpoint calls blacklist_manager.report() to increment the complaint counter. It then builds a response message with one of three outcomes:

  • The UPI was just blocked (hit the 3-complaint threshold on this request).
  • The UPI was already blocked from a previous report.
  • The complaint was recorded but more reports are needed before blocking.
  • Save app/main.py.

Seeing a red underline on blacklist_manager?

Make sure you saved app/blacklist.py in the previous substep. VS Code cannot resolve the import until the file has content.

help me fix an import error with blacklist_manager in my FastAPI app

Add the /blacklist endpoint and verify the integration

The final piece is an endpoint that returns all currently blocked UPI IDs. This gives administrators visibility into which accounts have been blacklisted.

  • In app/main.py, add this endpoint below the report_fraud function you just wrote:
@app.get("/blacklist", response_model=list[BlacklistEntry])
def get_blacklist():
    blocked = blacklist_manager.get_all_blocked()
    return [BlacklistEntry(**entry) for entry in blocked]

What does this endpoint do?

This calls get_all_blocked() to retrieve every UPI ID with 3+ complaints. The **entry syntax unpacks each dictionary into keyword arguments that the BlacklistEntry Pydantic model expects.

Your /predict endpoint already checks the blacklist before running the model. Look at the top of the predict_fraud function to see this in action:

# Check blacklist first
if blacklist_manager.is_blocked(transaction.receiver_upi):
    return PredictionResponse(
        risk_score=100,
        decision="BLOCK",
        reason_codes=["BLACKLISTED_UPI"],
        receiver_upi=transaction.receiver_upi,
    )

Why check the blacklist before running the model?

If a receiver UPI is already blacklisted, there is no point scoring the transaction with the ML model. Returning immediately saves computation and guarantees a BLOCK decision regardless of the other features.

  • Save app/main.py.
  • Verify the server starts by running this command in your terminal:
fastapi dev app/main.py
  • You should see output showing the server started at http://127.0.0.1:8000 with no import errors.
  • Press Ctrl+C to stop the server for now.

Seeing an ImportError or ModuleNotFoundError?

Check that app/__init__.py exists (it can be empty). Without it, Python does not recognize app/ as a package.

Also confirm you are running the command from the upi-fraud-detection folder, not from inside the app/ subfolder.

help me fix this ImportError when starting my FastAPI server

✔️ Awesome, I've got everything!

Make sure you have saved both app/blacklist.py and app/main.py.

ⓧ I'd like to double check the full code

Here is the complete app/blacklist.py:

class BlacklistManager:
    def __init__(self, block_threshold=3):
        self.complaints: dict[str, int] = {}
        self.block_threshold = block_threshold

    def report(self, upi_id: str) -> tuple[int, bool]:
        """Report a UPI ID. Returns (total_complaints, newly_blocked)."""
        if upi_id not in self.complaints:
            self.complaints[upi_id] = 0
        self.complaints[upi_id] += 1
        count = self.complaints[upi_id]
        newly_blocked = count == self.block_threshold
        return count, newly_blocked

    def is_blocked(self, upi_id: str) -> bool:
        return self.complaints.get(upi_id, 0) >= self.block_threshold

    def get_all_blocked(self) -> list[dict]:
        return [
            {"upi_id": upi_id, "complaint_count": count}
            for upi_id, count in self.complaints.items()
            if count >= self.block_threshold
        ]


blacklist_manager = BlacklistManager()

Here is the complete app/main.py:

from fastapi import FastAPI
from app.schemas import (
    TransactionInput,
    PredictionResponse,
    FraudReportInput,
    FraudReportResponse,
    BlacklistEntry,
)
from app.fraud_engine import fraud_engine
from app.blacklist import blacklist_manager

app = FastAPI(
    title="UPI Fraud Detection API",
    description="Real-time fraud scoring for UPI transactions",
    version="1.0.0",
)


@app.get("/health")
def health_check():
    return {"status": "healthy", "model_loaded": True}


@app.post("/predict", response_model=PredictionResponse)
def predict_fraud(transaction: TransactionInput):
    # Check blacklist first
    if blacklist_manager.is_blocked(transaction.receiver_upi):
        return PredictionResponse(
            risk_score=100,
            decision="BLOCK",
            reason_codes=["BLACKLISTED_UPI"],
            receiver_upi=transaction.receiver_upi,
        )

    features = {
        "amount": transaction.amount,
        "transaction_hour": transaction.transaction_hour,
        "transaction_frequency": transaction.transaction_frequency,
        "complaint_count": transaction.complaint_count,
        "is_new_device": transaction.is_new_device,
        "is_new_ip": transaction.is_new_ip,
        "distance_km": transaction.distance_km,
        "receiver_account_age_days": transaction.receiver_account_age_days,
        "amount_deviation": transaction.amount_deviation,
    }

    result = fraud_engine.score_transaction(features)

    return PredictionResponse(
        risk_score=result["risk_score"],
        decision=result["decision"],
        reason_codes=result["reason_codes"],
        receiver_upi=transaction.receiver_upi,
    )


@app.post("/report-fraud", response_model=FraudReportResponse)
def report_fraud(report: FraudReportInput):
    total_complaints, newly_blocked = blacklist_manager.report(report.upi_id)

    if newly_blocked:
        message = f"UPI ID {report.upi_id} has been BLOCKED after {total_complaints} complaints."
    elif blacklist_manager.is_blocked(report.upi_id):
        message = f"UPI ID {report.upi_id} is already blocked. Complaint recorded."
    else:
        message = f"Complaint recorded. {blacklist_manager.block_threshold - total_complaints} more needed to block."

    return FraudReportResponse(
        upi_id=report.upi_id,
        total_complaints=total_complaints,
        is_blocked=blacklist_manager.is_blocked(report.upi_id),
        message=message,
    )


@app.get("/blacklist", response_model=list[BlacklistEntry])
def get_blacklist():
    blocked = blacklist_manager.get_all_blocked()
    return [BlacklistEntry(**entry) for entry in blocked]

Your API now has a complete fraud response loop. The model scores transactions, users can report fraud, and repeat offenders get automatically blocked. Next up, you will test the full workflow end-to-end through the interactive Swagger docs.

Test the Complete Fraud Detection Workflow

Your fraud detection model is trained. Your API endpoints are built. Now it is time to prove the whole system works end-to-end.

In this step, you will start the FastAPI server and use its built-in Swagger documentation to send test requests. You will simulate a legitimate transaction, a suspicious transaction, and a blacklist-triggered block.

In this step, get ready to:
  • Start the FastAPI development server and open the interactive Swagger UI.
  • Test the /predict endpoint with both legitimate and suspicious transactions.
  • Trigger the blacklist by reporting a fraudulent UPI ID 3 times, then verify it auto-blocks.
Start the server and test a legitimate transaction

FastAPI includes an auto-generated interactive documentation page called Swagger UI. You can test every endpoint directly from your browser without writing any client code.

  • Start the FastAPI development server by running this command:
fastapi dev app/main.py

You should see output showing the server is running at http://127.0.0.1:8000.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.

Also confirm you are running the command from inside your upi-fraud-detection folder, where fraud_model.joblib lives.

My FastAPI server won't start. How do I fix this?

  • Open your browser and navigate to http://127.0.0.1:8000/docs.
  • You should see the Swagger UI page with all four endpoints listed: /health, /predict, /report-fraud, and /blacklist.

What is Swagger UI?

Swagger UI is an interactive API documentation page that FastAPI generates automatically from your code. Each endpoint shows its expected inputs and lets you send real requests by clicking Try it out.

Now test the /predict endpoint with a normal, legitimate-looking transaction.

  • Click on the POST /predict row to expand it.
  • Click Try it out on the right.
  • Replace the request body with this JSON for a legitimate transaction:
{
  "sender_upi": "user42@upi",
  "receiver_upi": "merchant10@upi",
  "amount": 250.0,
  "transaction_hour": 14,
  "transaction_frequency": 2,
  "complaint_count": 0,
  "is_new_device": 0,
  "is_new_ip": 0,
  "distance_km": 3.5,
  "receiver_account_age_days": 500,
  "amount_deviation": 0.1
}

Why these values?

This transaction mimics a normal purchase. The amount is low (250), it is during business hours (14:00), the device and IP are familiar, and the receiver account is well-established (500 days old). Every signal points to a safe transaction.

  • Click Execute.
  • Scroll down to the response body. You should see a risk_score below 30, a decision of ALLOW, and an empty or minimal reason_codes list.

Not seeing ALLOW?

Double-check that all the values in your JSON match the example above. If complaint_count or is_new_device accidentally have high values, the model will flag the transaction.

My legitimate transaction is not returning ALLOW. Help me figure out why.

Test a suspicious transaction and report fraud

A legitimate transaction passed through safely. Now send a transaction with every red flag you can think of to see how the model reacts.

  • In the same POST /predict section, replace the request body with this suspicious transaction:
{
  "sender_upi": "user99@upi",
  "receiver_upi": "scam5@upi",
  "amount": 15000.0,
  "transaction_hour": 2,
  "transaction_frequency": 12,
  "complaint_count": 5,
  "is_new_device": 1,
  "is_new_ip": 1,
  "distance_km": 200.0,
  "receiver_account_age_days": 3,
  "amount_deviation": 3.5
}

What makes this transaction suspicious?

This transaction has a high amount (15,000), happens at 2 AM, comes from a new device and new IP, targets a 3-day-old receiver account, and the amount deviates 3.5x from the sender's average. The receiver already has 5 prior complaints.

These are the exact patterns you injected into the fraud data during generation. The trained model learned to recognize this combination.

  • Click Execute.
  • You should see a high risk_score (likely 70+), a decision of FLAG or BLOCK, and multiple reason_codes like MULTIPLE_FRAUD_COMPLAINTS, DEVICE_MISMATCH, and HIGH_RISK_RECEIVER.

The model caught the fraud. Now simulate what happens when multiple users report this receiver as fraudulent.

  • Scroll down to the POST /report-fraud section and click Try it out.
  • Enter this request body:
{
  "upi_id": "scam5@upi",
  "reason": "Received money request from unknown person"
}
  • Click Execute.
  • The response should show total_complaints: 1 and is_blocked: false with a message saying 2 more complaints are needed to block.
  • Click Execute two more times (for a total of 3 reports).
  • On the third execution, the response should show total_complaints: 3, is_blocked: true, and a message confirming the UPI ID has been BLOCKED.

What just happened?

The blacklist manager counts complaints per UPI ID. Once a UPI ID hits 3 complaints (the block_threshold you set in app/blacklist.py), it is automatically blocked. Any future transaction involving that receiver will be instantly rejected without even running the ML model.

Verify the blacklist blocks all future transactions

The ultimate test: send a completely normal transaction to the now-blocked receiver. The blacklist should override the model and reject it immediately.

  • Scroll back to the POST /predict section.
  • Replace the request body with this safe-looking transaction that targets the blocked receiver:
{
  "sender_upi": "user1@upi",
  "receiver_upi": "scam5@upi",
  "amount": 100.0,
  "transaction_hour": 12,
  "transaction_frequency": 1,
  "complaint_count": 0,
  "is_new_device": 0,
  "is_new_ip": 0,
  "distance_km": 1.0,
  "receiver_account_age_days": 365,
  "amount_deviation": 0.0
}
  • Click Execute.
  • The response should show risk_score: 100, decision: "BLOCK", and reason_codes: ["BLACKLISTED_UPI"].

Even though every feature in this transaction looks perfectly safe, the blacklist check fires first and blocks it instantly.

  • Finally, scroll to the GET /blacklist section and click Try it out, then Execute.
  • You should see scam5@upi in the response with complaint_count: 3.

Why check the blacklist before the model?

In production fraud systems, known-bad actors are blocked immediately without wasting compute on ML inference. The blacklist acts as a fast-path rejection layer that sits in front of the more expensive model scoring.

Your fraud detection system is fully operational. You have confirmed that legitimate transactions pass through, suspicious transactions get flagged, and reported scammers get permanently blocked. Next up: a secret mission to add analytics monitoring.

Secret mission

Build a Fraud Analytics Endpoint

Your fraud detection API scores transactions in real time. But how do you know if the system is actually working well? Build an analytics endpoint that tracks every prediction and reports aggregated fraud intelligence.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Python virtual environment (venv folder) with installed packages.
  • Generated dataset file (upi_transactions.csv).
  • Trained model file (fraud_model.joblib).
  • FastAPI development server (running on port 8000).
  • Project folder with all source code files.

Keep everything running

No action needed. Choose this if you are still actively building or want to experiment with the API endpoints.

  • Your FastAPI server stays running at http://127.0.0.1:8000.
  • All files and the virtual environment remain on your Desktop.
  • You can keep testing endpoints in the Swagger docs at /docs.

Pause - I'll come back to this later

Shut down the running server to free up your terminal. Your files stay on disk so you can pick up where you left off.

  • Stop the FastAPI server by pressing Ctrl+C in the terminal where it is running.
  • Deactivate the virtual environment by running deactivate in your terminal.
  • When you return, reactivate the virtual environment and start the server again with fastapi dev app/main.py.

Delete - I don't want to use this again

Remove all project resources and start fresh. This permanently deletes your trained model, generated data, and source code.

  • Stop the FastAPI server by pressing Ctrl+C in the terminal.
  • Deactivate the virtual environment by running deactivate.
  • Delete the entire project folder from your Desktop.
  • On macOS, drag the folder to the Trash. On Windows, right-click the folder and select Delete.

Nice Work!

Nice Work!

You just built a real-time UPI fraud detection system from scratch. You took an ML model from raw data generation all the way to a production-ready API that scores transactions and blocks fraudsters automatically.

Here's what you accomplished:

  • Trained an XGBoost classifier on synthetic UPI transaction data with scale_pos_weight to handle class imbalance, achieving measurable precision, recall, and ROC-AUC scores.
  • Built a FastAPI REST API that accepts UPI transaction data and returns real-time risk scores (0-100) with ALLOW, FLAG, or BLOCK decisions and detailed reason codes.
  • Implemented a complaint-driven blacklist system that tracks fraud reports per UPI ID and automatically blocks repeat offenders after 3 complaints.
  • Built a fraud analytics endpoint that returns aggregated statistics including fraud rate, top blocked UPIs, and risk score distribution.

Ready to quiz yourself?