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)