Real-Time Fraud Detection with Confluent Cloud + Azure AI

Payment rail fraud doesn't wait — and neither should your detection pipeline. This project demonstrates how Confluent Cloud and Azure AI can be used together to detect fraud in real time across high-volume payment streams, the moment a suspicious pattern emerges.

Using ShadowTraffic to generate realistic synthetic transaction data and Confluent Cloud Flink SQL to analyze it, you'll build a progressively richer fraud detection pipeline — from simple threshold rules all the way to AI-generated analyst case notes powered by Azure OpenAI GPT-4o.

The default runtime path uses plain JSON values on Kafka. In Confluent Cloud Flink, that plain path is read-only and uses typed views over inferred topic tables. The previous JSON + Schema Registry path is still supported as an opt-in mode behind --registry, and it remains the write-capable Flink path for persistent sink topics.

This is a hands-on workshop and demonstration environment. No production data required.

Traditional batch-based fraud detection finds problems hours or days after the damage is done. Real-time streaming changes that equation:

• Confluent Cloud Kafka ingests authorization and settlement events the moment they occur — at any scale, with guaranteed ordering and exactly-once delivery
• Confluent Cloud Flink SQL joins streams across time windows, enriches events with merchant context, and applies detection logic — all without managing infrastructure
• Azure OpenAI GPT-4o goes beyond rule-based detection to reason about each transaction holistically and generate narrative case notes that analysts can act on immediately

Three coordinated Kafka streams form the foundation of the demo:

Injected fraud pattern: 10% of settlements have a settled_amount greater than requested_amount by $5-$50 — simulating over-settlement fraud where a merchant or intermediary inflates the charge after authorization.

From there, you'll work through progressively more powerful detection approaches:

1. Basic stream join — flag any transaction where settled_amount > requested_amount
2. Merchant enrichment — add risk scores and category context via a temporal lookup join
3. Timing analysis — measure auth-to-settlement delay and classify FRAUD vs NORMAL
4. Built-in anomaly detection — ML_DETECT_ANOMALIES_ROBUST with zero external dependencies
5. Azure OpenAI integration — AI_COMPLETE with GPT-4o for fraud reasoning and case note generation
6. Fraud alert pipeline — sink enriched AI assessments to a fraud_alerts Kafka topic

Instructors: See instructor/README.md for the one-shot setup script that stands up the complete demo environment.

cp .env.example .env

Open .env and fill in the values below — see docs/configuration.md for the complete reference.

Confluent Cloud:

CONFLUENT_CLOUD_BROKER=pkc-xxxxx.region.azure.confluent.cloud:9092

# Pre-composed JAAS string — paste the full line including the trailing semicolon
KAFKA_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username='<API_KEY>' password='<API_SECRET>';

# Optional: only required for ./start_demo.sh --registry
SCHEMA_REGISTRY_URL=https://psrc-xxxxx.region.azure.confluent.cloud

# Schema Registry key:secret pair
SCHEMA_REGISTRY_USER_INFO=<SR_API_KEY>:<SR_API_SECRET>

ShadowTraffic license:

LICENSE_ID=your-license-id
LICENSE_EMAIL=your@email.com
LICENSE_ORGANIZATION=Your Org
LICENSE_EDITION=ShadowTraffic Enterprise
LICENSE_EXPIRATION=YYYY-MM-DD
LICENSE_SIGNATURE=your-license-signature

./start_demo.sh

This starts the default plain JSON path:

1. shadowtraffic-merchant — loads 50 merchants into merchant_metadata (one-time)
2. shadowtraffic-transactions — streams correlated auth + settlement events continuously at one message every 3 seconds per topic

To start the legacy registry-backed mode instead:

./start_demo.sh --registry

# Authorization events
confluent kafka topic consume txn_auth --from-beginning --max-messages 5

# Settlement events (allow ~2 minutes after auths appear)
confluent kafka topic consume txn_settlement --from-beginning --max-messages 5

# Merchant enrichment data
confluent kafka topic consume merchant_metadata --from-beginning --max-messages 5

Navigate to Flink in your Confluent Cloud environment. Run the matching bootstrap first, then work through the files in data_discovery/flink_sql/ in order, replacing <catalog>.<database> with your environment and cluster names.

• Default mode: run data_discovery/flink_sql/00_bootstrap_plain_json.sql, then use analytics files 01-03, data_discovery/flink_sql/04_anomaly_detection_plain.sql, and optionally data_discovery/flink_sql/06_ai_fraud_review_plain.sql after 05. If you need anomaly alerts written to Kafka while keeping plain JSON inputs, also run data_discovery/flink_sql/00_bootstrap_plain_json_alert_sink.sql and then data_discovery/flink_sql/04_anomaly_detection_plain_sink.sql.
• Registry mode: run data_discovery/flink_sql/00_bootstrap_registry.sql, then use the full pipeline files including data_discovery/flink_sql/04_anomaly_detection.sql and data_discovery/flink_sql/06_ai_fraud_pipeline.sql.

shdw_data_fraud/
├── README.md                              ← You are here
├── start_demo.sh                          ← Root launcher (plain JSON by default)
├── docker-compose.yml                     ← Default plain JSON compose file
├── docker-compose.registry.yml            ← Registry-only compose override
├── .env.example                           ← Credential template — copy to .env
├── data_generation/
│   ├── README.md
│   ├── generators/
│   │   ├── merchant_generator_plain.json  ← Default plain JSON merchant generator
│   │   ├── transaction_generator_plain.json ← Default plain JSON auth + settlement generator
│   │   ├── merchant_generator.json        ← Registry-backed merchant generator
│   │   └── transaction_generator.json     ← Registry-backed auth + settlement generator
│   ├── schemas/
│   │   ├── txn_auth.json
│   │   ├── txn_settlement.json
│   │   ├── merchant_metadata.json
│   │   └── fraud_alert.json               ← Output schema for the AI pipeline
│   └── scripts/
│       └── register_schemas.sh            ← Registers schemas to Schema Registry
├── data_discovery/
│   ├── README.md
│   └── flink_sql/
│       ├── 00_bootstrap_plain_json.sql    ← Typed source views over inferred plain JSON topics
│       ├── 00_bootstrap_registry.sql      ← Explicit source/sink tables for registry mode
│       ├── 01_basic_detection.sql         ← Auth-settlement join, flag over-settlements
│       ├── 02_merchant_enrichment.sql     ← Add merchant risk score + tier
│       ├── 03_timing_analysis.sql         ← Settlement delay + FRAUD/NORMAL label
│       ├── 04_anomaly_detection_plain.sql ← Read-only anomaly review for plain mode
│       ├── 04_anomaly_detection.sql       ← Registry-only anomaly pipeline to fraud_alerts
│       ├── 05_azure_openai_setup.sql      ← CREATE CONNECTION + CREATE MODEL
│       ├── 06_ai_fraud_review_plain.sql   ← Read-only AI fraud review for plain mode
│       └── 06_ai_fraud_pipeline.sql       ← Registry-only AI pipeline + INSERT INTO fraud_alerts
├── instructor/
│   ├── README.md                          ← Instructor-only guide
│   └── instructor_setup.sh               ← One-shot demo environment setup
├── utilities/
│   ├── README.md
│   ├── cleanup.sh                         ← Stop containers, delete topics/schemas
│   └── e2e_test.sh                        ← End-to-end validation with timestamped logs
└── docs/
    ├── configuration.md                   ← Full env var + fraud parameter reference
    └── data_specs.md                      ← Field-level schemas + entity distributions

Fraud and settlement behavior are configured directly in data_generation/generators/transaction_generator.json:

"fraud_delta": {
    "_gen": "weightedOneOf",
    "choices": [
        { "weight": 90, "value": 0 },
        { "weight": 10, "value": { "_gen": "uniformDistribution", "bounds": [5.0, 50.0], "decimals": 2 } }
    ]
},
"settlement_status": {
    "_gen": "weightedOneOf",
    "choices": [
        { "value": "COMPLETED", "weight": 95 },
        { "value": "PENDING", "weight": 5 }
    ]
}

To increase the fraud rate, adjust the fraud_delta weights (for example 95/5 for 5% fraud, 80/20 for 20% fraud), then restart:

docker compose restart shadowtraffic-transactions

See docs/configuration.md for full tuning guidance, including amount-range controls in uniformDistribution.bounds.

./utilities/cleanup.sh --all

This stops all Docker containers and deletes the Kafka topics and Schema Registry subjects created by the demo. Run ./utilities/cleanup.sh --status to check what's currently running before cleaning up.

Every piece works together within Confluent Cloud — no ETL pipelines, no separate ML infrastructure, no batch jobs. This is what real-time fraud detection looks like at scale.