PayGuard is an end-to-end, near-real-time fraud-monitoring demonstration. It simulates card transactions, publishes them to Kafka, processes the streams in Databricks using a medallion architecture, produces alert tables, sends email notifications, and visualizes operational metrics in a Databricks Lakeview dashboard.

This is a demo project using generated data. It is not a production fraud-decisioning system.

Python transaction simulator                 Fraud watchlist CSV
  normal / high-value / watchlist traffic          |
                 |                                 | one JSON file per record
                 v                                 v
       Confluent Cloud Kafka                 Unity Catalog volume
                 |                                 |
                 +----------> Databricks Bronze <---+
                                      |
                                      v
                              Databricks Silver
                       transactions | customers | watchlist
                                      |
                 +--------------------+--------------------+
                 v                    v                    v
      Watchlist-match alerts   High-value alerts    Windowed volumes
                 |                    |                    |
                 +------> email notification      Lakeview dashboard

The kafka_producer application generates deterministic synthetic customers and merchants from RANDOM_SEED, persists them to kafka_producer/data/, and publishes JSON transactions to Kafka. It uses SASL/SSL and reads connection details from kafka_producer/.env.

Available producer entry points:

The simulator has rules for high-value transactions, impossible travel, new devices, high-risk or blacklisted merchants, international usage, velocity, and card-testing patterns. These scores are used by the simulator but are not written to the Kafka payload used by the pipeline.

payguard.bronze.transactions reads credit_card_transactions from Kafka, retaining the raw JSON payload and Kafka metadata (topic, partition, offset, timestamp) with an ingestion timestamp.

payguard.bronze.fraud_watchlist uses Databricks Auto Loader to read JSON files from:

/Volumes/payguard/source/fraud_watchlist/source_data/

The watchlist generator notebook converts the repository CSV into one JSON file per record so it can be ingested incrementally.

payguard.silver.transactions parses the JSON payload into typed fields and preserves Kafka lineage. It drops records missing transaction, customer, card, or merchant identifiers, and records an expectation for positive amounts.

payguard.silver.fraud_watchlist normalizes IDs, risk levels, and actions; converts effective_from to a timestamp; and retains source-file lineage.

payguard.silver.customers cleans a Bronze customer stream. The Bronze customer source is expected to be provisioned separately; the included PostgreSQL scripts provide the customer seed and incremental data.

The two alert notifier modules use foreachBatch and Gmail SMTP credentials stored in Databricks Secrets to email alerts.

• Python 3.10+ and pip
• A Kafka-compatible cluster (the current code is configured for Confluent Cloud SASL/SSL)
• A Databricks workspace with Unity Catalog, access to the required volume paths, and permissions to create/read the payguard catalog schemas and tables
• A Databricks pipeline environment supporting pyspark.pipelines
• A configured Gmail App Password if email notifications will be enabled
• PostgreSQL if you use the included customer bootstrap scripts

From the repository root:

cd kafka_producer
python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install -r requirements.txt

Create kafka_producer/.env with your own Kafka connection details:

BOOTSTRAP_SERVERS=your-cluster-host:9092
API_KEY=your-kafka-api-key
API_SECRET=your-kafka-api-secret
TOPIC_NAME=credit_card_transactions
TRANSACTIONS_PER_SECOND=5
FRAUD_PERCENTAGE=0.08
TOTAL_CUSTOMERS=1000
TOTAL_MERCHANTS=200
RANDOM_SEED=42

Start normal traffic:

python producer_normal.py

Run the specialised producers in separate terminals when testing each alert scenario.

1. Create the payguard catalog/schemas and Unity Catalog volumes referenced by the code.
2. Put Kafka credentials in the Databricks secret scope payguard-scope under kafka_connection_details as JSON with bootstrap_servers, api_key, api_secret, and topic keys.
3. Put the Gmail App Password in payguard-scope under the key expected by the notifier modules.
4. Load the historic customer dataset into the Bronze customer table expected by payguard.silver.customers; then apply the incremental SQL script when needed.
5. Upload or generate fraud-watchlist JSON files in the configured volume path.
6. Deploy the Bronze, Silver, Gold, and alert Python modules as one Databricks pipeline. Keep dependent tables in the same pipeline or ensure upstream tables are already available.
7. Import dashboard/PayGuard Fraud Detection Monitoring.lvdash.json, map its datasets to the deployed tables, and refresh the dashboard.

The 01_kafka_streaming_test.py and 04_Autoloader_test.py notebooks are useful for validating connectivity and ingestion before deploying the full pipeline.

The Lakeview dashboard includes alert and transaction counters, alert trends, high-risk activity, top customers and merchants, payment-channel and international/domestic splits, a country heatmap, and windowed transaction-volume views.

Before creating or pushing a new remote repository:

• Rotate all Kafka API keys, passwords, Gmail App Passwords, and Databricks tokens that have appeared in local files or notebook history.
• Remove hard-coded credentials from 01_kafka_streaming_test.py and 02_Setup_Secret_Scope.py. Store secrets only in Databricks Secrets or local untracked environment files.
• Do not commit kafka_producer/.env, .venv/, generated data/, Databricks personal-access tokens, or real cardholder data. Add them to .gitignore.
• The included SQL and generated data contain email addresses and card-like numbers. Treat them as demo data and review them before publishing the repository.

• Table names transaciton_count_by_minute* retain the source spelling for compatibility.
• The customer Bronze ingestion definition is not included in this repository, although the Silver customer pipeline requires payguard.bronze.customers.
• The pipeline code uses fixed catalog, secret-scope, and Unity Catalog volume names. Parameterize these before deploying to another workspace.
• There are no automated tests or CI workflows in the repository at present.