Real-Time Fraud Detection — Documentation

Overview

The Real-Time Fraud Detection System scores transactions as they occur and separates high-risk activity from normal behaviour. It is designed to deliver sub-20ms inference for payment networks and insurance claims pipelines.

Architecture

The system consists of ingestion, feature enrichment, anomaly scoring, and review orchestration modules.

fraud_pipeline/
├── ingestion/
│   ├── stream.py            # Kafka / Kinesis adapters
│   └── validator.py         # schema and field checks
├── features/
│   ├── transactional.py     # velocity and location features
│   ├── identity.py          # customer profile signals
│   └── temporal.py          # session-level aggregation
├── models/
│   ├── autoencoder.py       # reconstruction anomaly scoring
│   ├── isolation_forest.py  # unsupervised anomaly detection
│   └── ensemble.py          # risk scoring and blending
├── api/
│   ├── app.py               # FastAPI endpoint
│   └── schema.py            # request / response models
└── triage/
    ├── case_router.py       # review queue assignment
    └── monitor.py           # drift and latency monitoring

Data Schema

Expected transaction fields for scoring:

transaction_id        str      Unique transaction identifier
customer_id           str      Customer account identifier
amount                float    Transaction value
timestamp             str      ISO timestamp of event
merchant_category     str      Merchant / claim category
location              str      Country or branch code
device_type           str      Web / mobile / POS
previous_tx_count     int      Transactions in prior 24h
avg_tx_amount_7d      float    Avg amount in last 7 days
is_high_risk_country  int      High-risk origin flag

Installation

git clone https://github.com/chirchirp/fraud-detection-system.git
cd fraud-detection-system

python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

pip install -r requirements.txt

cp .env.example .env
# Set message bus, database, and API configuration

Usage

Start the API

uvicorn fraud_pipeline.api.app:app --host 0.0.0.0 --port 8000

Score a transaction

import httpx

payload = {
    "transaction_id": "TX-1001",
    "customer_id": "CUST-042",
    "amount": 125.40,
    "timestamp": "2025-05-01T14:22:00Z",
    "merchant_category": "retail",
    "location": "KE",
    "device_type": "mobile",
    "previous_tx_count": 4,
    "avg_tx_amount_7d": 82.60,
    "is_high_risk_country": 0
}

response = httpx.post("http://localhost:8000/predict", json=payload)
print(response.json())

API Reference

POST /predict — Real-time transaction scoring.

POST /predict/batch — Batch scoring of multiple transactions.

GET /health — Live system health check.

Performance

Deployment

The fraud detection API is ready for container-based deployment with secure messaging and monitoring.

docker build -t fraud-detection:latest .
docker run -p 8000:8000 --env-file .env fraud-detection:latest

# Deploy to AWS ECS / Azure Container Apps with secure event bus