fraud-detection

A real-time fraud detection pipeline built on a modern stream processing stack. Synthetic (and real) transaction data flows through Apache Kafka, is windowed by Apache Flink, scored by an XGBoost ML model served via FastAPI, and routed by a Spring Boot orchestrator — with a React dashboard for human-in-the-loop analyst review and a weekly MLOps retraining loop.

---

Architecture

TransactionProducer (Java)
        │
        ▼
Kafka Topic: raw-transactions
        │
        ▼
FraudStreamingJob (Apache Flink)
  • Groups by userId
  • 1-minute tumbling window
  • Aggregates: sum(amount), count(txns)
        │
        ▼
Redis  user:{userId}:features
  { total_spent_1min, tx_count_1min }
        │
        ▼ (velocity features)
fraud-orchestrator (Spring Boot :8080)
  POST /api/v1/process-transaction
  ├── Fetch Redis velocity features
  ├── Call ML API POST /predict
  ├── Rule engine (thresholds: 70 / 90)
  ├── Persist → PostgreSQL (transactions table)
  ├── CHALLENGE → WebSocket push
  └── Async Kafka re-publish
        │
        ├── APPROVED  → return to caller
        ├── DECLINED  → return to caller
        └── CHALLENGE → fraud-dashboard (React :3000)
                              │
                      Analyst clicks Approve / Confirm Fraud
                              │
                              ▼
                      POST /api/analyst/decision
                              │
                              ▼
                       PostgreSQL (analyst label)
                              │
                     [Sunday 2 AM cron]
                              ▼
                   retrain_from_db.py
                   Champion/challenger AUC gate
                              │
                   POST /admin/reload-model  ←── FastAPI hot-swap

---

Project Structure

fraud-detection/
├── README.md
│
├── fraud-pipeline/                     # Java (Maven) — Kafka producer + Flink streaming job
│   ├── pom.xml
│   └── src/main/java/com/fraud/
│       ├── TransactionProducer.java    # Publishes synthetic {userId, amount, timestamp} to Kafka
│       ├── FraudStreamingJob.java      # Flink job: window → aggregate → write to Redis
│       ├── TransactionInfo.java        # POJO for Flink serialization
│       └── RedisSinkFunction.java      # Flink sink: writes windowed aggregates to Redis
│
├── ml-model/                           # Python — XGBoost model + FastAPI inference server
│   ├── requirements.txt
│   ├── train.py                        # Train XGBoost on creditcard.csv, save fraud_model.json
│   ├── ml_api.py                       # FastAPI: GET /health, POST /predict, POST /admin/reload-model
│   ├── seed_fraud_test.py              # Populate dashboard with scored test transactions
│   ├── retrain_from_db.py              # Weekly MLOps: pull analyst labels → retrain → champion/challenger gate
│   ├── fraud_model.json                # Serialized trained model (XGBoost native format)
│   └── data/creditcard.csv             # Kaggle Credit Card Fraud dataset
│
├── fraud-orchestrator/                 # Java (Spring Boot) — decision engine + API
│   ├── pom.xml
│   └── src/main/java/com/fraud/orchestrator/
│       ├── OrchestratorApplication.java
│       ├── config/
│       │   ├── AppConfig.java          # CORS, JedisPool, RestTemplate, KafkaProducer beans
│       │   └── WebSocketConfig.java    # STOMP broker, SockJS endpoint /ws-fraud-alert
│       ├── controller/
│       │   ├── TransactionController.java  # POST /api/v1/process-transaction
│       │   └── AnalystController.java      # POST /api/analyst/decision, GET /api/analyst/queue
│       ├── service/
│       │   ├── TransactionService.java     # JPA persistence + analyst decision validation
│       │   ├── MlApiService.java           # FastAPI client (graceful degradation on failure)
│       │   ├── RedisService.java           # Velocity feature lookup
│       │   ├── KafkaPublisherService.java  # Async fire-and-forget Kafka publish
│       │   └── RetrainingScheduler.java    # @Scheduled cron: Sunday 2 AM retraining
│       ├── model/
│       │   ├── TransactionRecord.java      # JPA entity: transactions table
│       │   ├── TransactionRequest.java     # Incoming request DTO (supports real v-features)
│       │   ├── MlPredictRequest.java       # FastAPI payload (velocity proxy or real features)
│       │   ├── MlPredictResponse.java      # FastAPI response (risk_score, is_fraud)
│       │   ├── DecisionResponse.java       # Outgoing decision DTO
│       │   └── AnalystDecisionRequest.java # Analyst review DTO
│       └── repository/
│           └── TransactionRepository.java  # Spring Data JPA (findByStatusOrderByMlRiskScoreDesc)
│
└── fraud-dashboard/                    # React (Vite) — analyst review UI
    ├── package.json
    ├── vite.config.js
    ├── index.html
    └── src/
        ├── main.jsx
        ├── App.jsx
        └── components/
            └── FraudDashboard.jsx      # Live WebSocket queue, approve/confirm buttons, toast alerts

---

Components

fraud-pipeline — Stream Processing (Java / Flink)

TransactionProducer.java — Generates synthetic transactions at 2/sec and publishes to the raw-transactions Kafka topic:

{ "userId": "user_<1-100>", "amount": 5.00-1500.00, "timestamp": <epoch ms> }

FraudStreamingJob.java — Flink streaming job:

1. Reads from raw-transactions (Kafka)
2. Groups by userId
3. Applies a 1-minute tumbling processing-time window
4. Reduces: sums amount, counts transactions
5. Writes to Redis hash user:{userId}:features (total_spent_1min, tx_count_1min)

---

ml-model — ML Inference (Python / XGBoost + FastAPI)

train.py — Trains an XGBClassifier on the Kaggle Credit Card Fraud dataset:

• Features: V1–V28 (PCA-anonymized) + Amount + Time (30 total)
• scale_pos_weight handles severe class imbalance
• Saves to fraud_model.json (XGBoost native format)

ml_api.py — FastAPI inference server (port 8000):

Endpoint	Method	Description
/health	GET	Liveness check
/predict	POST	Score a transaction → {risk_score: 0-100, is_fraud: bool}
/admin/reload-model	POST	Hot-swap model without restart

retrain_from_db.py — Weekly MLOps job:

1. Queries PostgreSQL for analyst-verified labels (CONFIRMED_FRAUD / MANUAL_APPROVED)
2. Merges with original Kaggle training data
3. Trains a candidate model and computes ROC-AUC
4. Champion/challenger gate: only deploys if candidate AUC exceeds current

---

fraud-orchestrator — Decision Engine (Java / Spring Boot)

REST API on port 8080. The critical path targets <50ms end-to-end.

POST /api/v1/process-transaction

{
  "userId": "user_42",
  "amount": 349.99,
  "merchantId": "Amazon",
  "timestamp": 1718000000000,
  "vFeatures": [...]   // optional: 28 real PCA features from card network
}

Response:

{ "transactionId": "...", "status": "APPROVED|CHALLENGE|DECLINED", "riskScore": 45.2, "reason": "..." }

Rule engine thresholds (configurable via application.properties):

• risk > 90 → DECLINED (auto-block)
• 70 < risk ≤ 90 → CHALLENGE (analyst queue + WebSocket push)
• risk ≤ 70 → APPROVED

ML feature paths:

• With real v-features: passed directly to model (accurate scoring)
• Without v-features: velocity proxy via Redis (V1 = totalSpent/10000, V2 = txCount/100)

POST /api/analyst/decision — Submit analyst review:

{ "transactionId": "...", "decision": "MANUAL_APPROVED|CONFIRMED_FRAUD", "analystId": "analyst_01" }

GET /api/analyst/queue — Return all CHALLENGE transactions sorted by risk score descending.

---

fraud-dashboard — Analyst UI (React / Vite)

Dark-theme SPA on port 3000:

• Connects to STOMP WebSocket at /ws-fraud-alert
• Subscribes to /topic/suspicious-transactions for live CHALLENGE alerts
• Shows pending queue sorted by risk score; highlights new arrivals for 3 seconds
• Toast notifications auto-dismiss after 5 seconds
• Approve → MANUAL_APPROVED; Confirm Fraud → CONFIRMED_FRAUD
• Stats cards: pending count, highest risk score, WebSocket status

---

Tech Stack

Layer	Technology	Version
Message Queue	Apache Kafka (kafka-clients)	3.7.0
Stream Processing	Apache Flink	2.2.0
Kafka-Flink Bridge	flink-connector-kafka	4.0.1-2.0
State / Cache	Redis (Jedis)	5.1.0
ML Model	XGBoost (XGBClassifier)	≥ 2.0.0
ML API	FastAPI + Uvicorn	≥ 0.111.0
Orchestrator	Spring Boot	3.3.0
WebSocket	STOMP + SockJS	—
Database	PostgreSQL	—
ORM	Spring Data JPA / Hibernate	—
Dashboard	React 18 + Vite	18.3.0 / 5.3.0
HTTP Client (frontend)	Axios	1.7.0
JSON Processing	Jackson (jackson-databind)	2.17.1
Build Tool (Java)	Maven	—
Language (pipeline)	Java	—
Language (ML)	Python	3.x

---

Getting Started

Prerequisites

• Java (JDK 17+)
• Maven
• Python 3.x + pip
• Apache Kafka running on localhost:9092
• Redis on localhost:6379
• PostgreSQL on localhost:5432 (database: fraud_db, user: fraud_user, password: fraud_pass)
• Node.js + npm (for dashboard)

---

1. Train the ML Model

cd ml-model
python3 -m venv venv && source venv/bin/activate
pip install -r requirements.txt
python train.py          # produces fraud_model.json

2. Start the ML API

cd ml-model
uvicorn ml_api:app --host 0.0.0.0 --port 8000
# Docs: http://localhost:8000/docs

3. Build & Start the Orchestrator

cd fraud-orchestrator
mvn clean package
mvn spring-boot:run
# API: http://localhost:8080

4. Build & Run the Flink Streaming Job

cd fraud-pipeline
mvn clean package
# Submit to a running Flink cluster, or run locally:
mvn exec:java -Dexec.mainClass="com.fraud.FraudStreamingJob"

5. Start the Transaction Producer

cd fraud-pipeline
mvn exec:java -Dexec.mainClass="com.fraud.TransactionProducer"

6. Start the Dashboard

cd fraud-dashboard
npm install
npm run dev
# UI: http://localhost:3000

7. Seed Test Data (optional)

cd ml-model
python seed_fraud_test.py   # populates dashboard with pre-scored challenge transactions

---

Status

• Kafka transaction producer (synthetic data)
• Flink stream processing pipeline (windowed velocity aggregation → Redis)
• Redis state management (per-user 1-min velocity features)
• XGBoost model training (Kaggle Credit Card Fraud dataset)
• FastAPI ML inference server (/predict, /health, /admin/reload-model)
• Spring Boot orchestrator (rule engine, PostgreSQL persistence, WebSocket push)
• Human-in-the-loop analyst review (approve / confirm fraud)
• React analyst dashboard (live WebSocket alerts, queue management)
• Weekly MLOps retraining loop (champion/challenger AUC gate, hot-swap)
• Production deployment / containerisation (Docker Compose)
• Alerting / external notification sink (email, PagerDuty, etc.)