Do you want to try out this notebook? Get a free account (no credit-card reqd) at hopsworks.ai. You can also install open-source Hopsworks or view tutorial videos here.
2. Generate credit card transactions data and send to kafka topic
Generate credit card transactions data and send to kafka topic.
Inspiration of this example was taken from here.
Prerequisites
#!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 = 5400
TOTAL_UNIQUE_USERS = 100
START_DATE = '2021-04-01 00:00:00'
END_DATE = '2021-04-04 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
Using the random credit card numbers, timestamps and transaction amounts generated in the above steps,
we can generate random credit card transactions by combining them. The transaction id for the transaction is the md5
hash of the above mentioned entities.
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}]
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]
Create Transaction Chains
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
Save labels separatelly
transaction_labels = []
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:
transaction_label = {}
poped_label = transaction.pop("fraud_label")
transaction_label["tid"]=transaction["tid"]
transaction_label["fraud_label"]=poped_label
transaction_labels.append(transaction_label)
if i % 1000 == 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}
Save labels as separate file
labelsDF = pd.DataFrame.from_records(transaction_labels)
pandas.write_csv(hdfs.project_path() + "/Resources/transaction_labels.csv", labelsDF, index=False)