2. Generate credit card transactions data
Generate credit card transactions data and send to kafka topic.
Inspiration of this example was taken from here.
Prerequisites
Install Faker if you don’t have it installed already. We use Faker to generate synthetic data simulating credit cards and financial transactions.
#!pip install Faker
Imports
from collections import defaultdict
from faker import Faker
import pandas as pd
import numpy as np
import datetime
import hashlib
import random
import math
import os
from hops import hdfs
from hops import pandas_helper as pandas
# Seed for Reproducibility
faker = Faker()
faker.seed_locale('en_US', 0)
SEED = 123
random.seed(SEED)
np.random.seed(SEED)
faker.seed_instance(SEED)
Constants
TOTAL_UNIQUE_TRANSACTIONS = 43200
TOTAL_UNIQUE_USERS = 100
START_DATE = '2021-04-01 00:00:00'
END_DATE = '2021-04-02 00:01:00'
DATE_FORMAT = '%Y-%m-%d %H:%M:%S'
# Change this according to your settings
KAFKA_BROKER_ADDRES = "broker.kafka.service.consul:9091"
KAFKA_TOPIC_NAME = "credit_card_transactions"
Generate Transactions
Generate Unique Credit Card Numbers
Credit card numbers are uniquely assigned to users. Since, there are 10K users, we would want to generate 10K unique card numbers.
def generate_unique_credit_card_numbers(n: int) -> list:
cc_ids = set()
for _ in range(n):
cc_id = faker.credit_card_number(card_type='visa')
cc_ids.add(cc_id)
return list(cc_ids)
credit_card_numbers = generate_unique_credit_card_numbers(TOTAL_UNIQUE_USERS)
assert len(credit_card_numbers) == TOTAL_UNIQUE_USERS
assert len(credit_card_numbers[0]) == 16 # validate if generated number is 16-digit
# inspect random sample of credit card numbers
random.sample(credit_card_numbers, 5)
['4333652104565302',
'4858129660270572',
'4106727807825537',
'4997591057565538',
'4762714231199452']
Generate Time Series
def generate_timestamps(n: int) -> list:
start = datetime.datetime.strptime(START_DATE, DATE_FORMAT)
end = datetime.datetime.strptime(END_DATE, DATE_FORMAT)
timestamps = list()
for _ in range(n):
timestamp = faker.date_time_between(start_date=start, end_date=end, tzinfo=None).strftime(DATE_FORMAT)
timestamps.append(timestamp)
timestamps = sorted(timestamps)
return timestamps
timestamps = generate_timestamps(TOTAL_UNIQUE_TRANSACTIONS)
assert len(timestamps) == TOTAL_UNIQUE_TRANSACTIONS
# inspect random sample of timestamps
random.sample(timestamps, 5)
['2021-04-02 04:55:07',
'2021-04-01 11:41:20',
'2021-04-01 04:11:55',
'2021-04-02 17:30:30',
'2021-04-03 10:31:47']
Generate Random Transaction Amounts
The transaction amounts are presumed to follow Pareto distribution, as it is logical for consumers to make many more smaller purchases than large ones. The break down of the distribution is shown in the table below.
Percentage | Range (Amount in $) |
---|---|
5\% | 0.01 to 1 |
7.5\% | 1 to 10 |
52.5\% | 10 to 100 |
25\% | 100 to 1000 |
10\% | 1000 to 10000 |
def get_random_transaction_amount(start: float, end: float) -> float:
amt = round(np.random.uniform(start, end), 2)
return amt
distribution_percentages = {0.05: (0.01, 1.01),
0.075: (1, 11.01),
0.525: (10, 100.01),
0.25: (100, 1000.01),
0.10: (1000, 10000.01)}
amounts = []
for percentage, span in distribution_percentages.items():
n = int(TOTAL_UNIQUE_TRANSACTIONS * percentage)
start, end = span
for _ in range(n):
amounts.append(get_random_transaction_amount(start, end+1))
random.shuffle(amounts)
assert len(amounts) == TOTAL_UNIQUE_TRANSACTIONS
# inspect random sample of transaction amounts
random.sample(amounts, 5)
[96.24, 11.8, 120.07, 994.69, 27.7]
Generate Credit Card Transactions
def generate_transaction_id(timestamp: str, credit_card_number: str, transaction_amount: float) -> str:
hashable = f'{timestamp}{credit_card_number}{transaction_amount}'
hexdigest = hashlib.md5(hashable.encode('utf-8')).hexdigest()
return hexdigest
transactions = []
for timestamp, amount in zip(timestamps, amounts):
credit_card_number = random.choice(credit_card_numbers)
transaction_id = generate_transaction_id(timestamp, credit_card_number, amount)
transactions.append({'tid': transaction_id,
'datetime': timestamp,
'cc_num': credit_card_number,
'amount': amount,
'fraud_label': 0})
assert len(transactions) == TOTAL_UNIQUE_TRANSACTIONS
# inspect random sample of credit card transactions
random.sample(transactions, 1)
[{'tid': '619ff51c702f6fa77fcda03e1a20e0c5',
'datetime': '2021-04-03 18:07:17',
'cc_num': '4446288810992896',
'amount': 786.9,
'fraud_label': 0}]
Log financial transaction into Kafka
from hops import kafka
from hops import tls
from hops import hdfs
import json
from confluent_kafka import Producer
config = {
"bootstrap.servers": KAFKA_BROKER_ADDRES,
"security.protocol": kafka.get_security_protocol(),
"ssl.ca.location": tls.get_ca_chain_location(),
"ssl.certificate.location": tls.get_client_certificate_location(),
"ssl.key.location": tls.get_client_key_location(),
"group.id": "1"
}
producer = Producer(config)
i = 0
for transaction in transactions:
if "fraud_label" in transaction:
transaction.pop("fraud_label")
if i % 4000 == 0:
print(json.dumps(transaction))
producer.produce(KAFKA_TOPIC_NAME, json.dumps(transaction))
producer.flush()
i += 1
{"tid": "60acdc1fa703e872cee995498dbea49a", "datetime": "2021-04-01 00:01:32", "cc_num": "4867010117638802", "amount": 24.71}
{"tid": "6b2b9833759eae5aac955b17fa3fc254", "datetime": "2021-04-01 13:08:45", "cc_num": "4564139086560436", "amount": 51.46}
{"tid": "f52e0dc51edb2bf663b35027198807e2", "datetime": "2021-04-02 02:30:36", "cc_num": "4638396144844325", "amount": 525.45}
{"tid": "354460b3ec66dab3e9c6df3dc8b06107", "datetime": "2021-04-02 16:07:29", "cc_num": "4460285888258185", "amount": 28.56}
{"tid": "afaa0bd2d5a226d5427ad8a3e6885ad0", "datetime": "2021-04-03 05:37:13", "cc_num": "4032763187099525", "amount": 16.6}
{"tid": "0fcd3a747ecbd93f3b2808fca28f0548", "datetime": "2021-04-03 18:46:16", "cc_num": "4650661577010550", "amount": 431.74}
Simulate fraudulent transactions
NOTE: Before polluting the
credit_card_transactions
topic with fraudulent transactions, you can run the notebooks3_stream-ingestion
,4_create_training_dataset
and5_model_training
to create a Training Dataset with the original transactions and train an autoencoder that learns these patterns. Once the model is served, you can create fraudulent transactions by running the code below.
Create Attack Transaction Chains
FRAUD_RATIO = 0.0025 # percentage of transactions that are fraudulent
NUMBER_OF_FRAUDULENT_TRANSACTIONS = int(FRAUD_RATIO * TOTAL_UNIQUE_TRANSACTIONS)
ATTACK_CHAIN_LENGTHS = [3, 4, 5, 6, 7, 8, 9, 10]
visited = set()
chains = defaultdict(list)
def size(chains: dict) -> int:
counts = {key: len(values)+1 for (key, values) in chains.items()}
return sum(counts.values())
def create_attack_chain(i: int):
chain_length = random.choice(ATTACK_CHAIN_LENGTHS)
for j in range(1, chain_length):
if i+j not in visited:
if size(chains) == NUMBER_OF_FRAUDULENT_TRANSACTIONS:
break
chains[i].append(i+j)
visited.add(i+j)
while size(chains) < NUMBER_OF_FRAUDULENT_TRANSACTIONS:
i = random.choice(range(TOTAL_UNIQUE_TRANSACTIONS))
if i not in visited:
create_attack_chain(i)
visited.add(i)
assert size(chains) == NUMBER_OF_FRAUDULENT_TRANSACTIONS
Modify Transactions with Fraud Chain Attacks
def generate_timestamps_for_fraud_attacks(timestamp: str, chain_length: int) -> list:
timestamps = []
timestamp = datetime.datetime.strptime(timestamp, DATE_FORMAT)
for _ in range(chain_length):
# interval in seconds between fraudulent attacks
delta = random.randint(30, 120)
current = timestamp + datetime.timedelta(seconds=delta)
timestamps.append(current.strftime(DATE_FORMAT))
timestamp = current
return timestamps
def generate_amounts_for_fraud_attacks(chain_length: int) -> list:
amounts = []
for percentage, span in distribution_percentages.items():
n = math.ceil(chain_length * percentage)
start, end = span
for _ in range(n):
amounts.append(get_random_transaction_amount(start, end+1))
return amounts[:chain_length]
for key, chain in chains.items():
transaction = transactions[key]
timestamp = transaction['datetime']
cc_num = transaction['cc_num']
amount = transaction['amount']
transaction['fraud_label'] = 1
inject_timestamps = generate_timestamps_for_fraud_attacks(timestamp, len(chain))
inject_amounts = generate_amounts_for_fraud_attacks(len(chain))
random.shuffle(inject_amounts)
for i, idx in enumerate(chain):
original_transaction = transactions[idx]
inject_timestamp = inject_timestamps[i]
original_transaction['datetime'] = inject_timestamp
original_transaction['fraud_label'] = 1
original_transaction['cc_num'] = cc_num
original_transaction['amount'] = inject_amounts[i]
original_transaction['tid'] = generate_transaction_id(inject_timestamp, cc_num, amount)
transactions[idx] = original_transaction
Log fraudulent transactions into Kafka
i = 0
for transaction in transactions:
if i % 2000 == 0:
print(json.dumps(transaction))
producer.produce(KAFKA_TOPIC_NAME, json.dumps(transaction))
producer.flush()
i += 1
{"tid": "60acdc1fa703e872cee995498dbea49a", "datetime": "2021-04-01 00:01:32", "cc_num": "4867010117638802", "amount": 24.71}
{"tid": "f52e0dc51edb2bf663b35027198807e2", "datetime": "2021-04-02 02:30:36", "cc_num": "4638396144844325", "amount": 525.45}
{"tid": "afaa0bd2d5a226d5427ad8a3e6885ad0", "datetime": "2021-04-03 05:37:13", "cc_num": "4032763187099525", "amount": 16.6}