Full Data Preprocessing Pipeline

import os
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
import category_encoders as ce
from scipy.linalg import block_diag
DATA_OUTPUT_DIR = "example_graph_data"
# Customers dataset
df_customers = pd.DataFrame(
    {
        "customer_id": ["C001", "C002", "C003"],
        "age": [25, 40, 35],
        "gender": ["Male", "Female", "Other"],
        "loyalty_points": [150, 300, 225],
    }
)
# Merchants dataset
df_merchants = pd.DataFrame(
    {
        "merchant_id": ['M001', 'M002', 'M003'],
        "merchant_rating": ["Poor", "Average", "Excellent"],
        "merchant_category": ["Grocery", "Electronics", "Apparel"],
        "merchant_location": ["New York", "90001", "Chicago"],  # City or ZIP code
    }
)
# Transactions dataset
df_transactions = pd.DataFrame(
    {
        "transaction_id": ['T001', 'T002', 'T003', 'T004', 'T005'],
        "customer_id": ['C002', 'C003', 'C001', 'C001', 'C003'],
        "merchant_id": ['M002', 'M003', 'M001', 'M003', 'M003'],
        "amount": [250.0, 175.5, 320.75, 85.99, 1000.0],
        "transaction_timestamp": [
            "2025-03-10 14:23:00",
            "2025-03-10 09:45:00",
            "2025-03-08 16:30:00",
            "2025-03-01 06:30:00",
            "2025-03-11 02:30:00",
        ],
        "transaction_type": ["online", "in-store", "online", "online", "in-store"],
        "is_fraud": [1, 0, 1, 0, 1],
    }
)
# ------------------------------------------------
# Step 2.1: Preprocess Customers
# ------------------------------------------------
preprocessed_customer_columns = []
preprocessed_merchant_columns = []
preprocessed_transaction_columns = []
# Binary encode customer_id (fit encoder and save for reuse)
encoder_customer_id = ce.BinaryEncoder(cols=["customer_id"]).fit(
    df_customers["customer_id"]
)
transformed_column_names = encoder_customer_id.get_params()["mapping"][0]["mapping"].columns
df_customers[transformed_column_names] = encoder_customer_id.transform(df_customers["customer_id"])
preprocessed_customer_columns.extend(transformed_column_names)
# Binary encode gender
encoder_gender = ce.BinaryEncoder(cols=["gender"]).fit(df_customers["gender"])
transformed_column_names = encoder_gender.get_params()["mapping"][0]["mapping"].columns
df_customers[transformed_column_names] = encoder_gender.transform(df_customers["gender"])
preprocessed_customer_columns.extend(transformed_column_names)
# Scale numerical features: age and loyalty_points
scaler_customers = StandardScaler()
df_customers[["age", "loyalty_points"]] = scaler_customers.fit_transform(
    df_customers[["age", "loyalty_points"]]
)
preprocessed_customer_columns.extend(["age", "loyalty_points"])
print("\nPreprocessed Customers Data:")
print(df_customers)
# ------------------------------------------------
# Step 2.2: Preprocess Merchants
# ------------------------------------------------
# Binary encode merchant_id (fit encoder and save for reuse)
encoder_merchant_id = ce.BinaryEncoder(cols=["merchant_id"]).fit(
    df_merchants["merchant_id"]
)
transformed_column_names = encoder_merchant_id.get_params()["mapping"][0]["mapping"].columns
df_merchants[transformed_column_names] = encoder_merchant_id.transform(df_merchants["merchant_id"])
preprocessed_merchant_columns.extend(transformed_column_names)
# Ordinal encode merchant_rating
rating_mapping = {"Poor": 1, "Average": 2, "Good": 3, "Very Good": 4, "Excellent": 5}
df_merchants["merchant_rating_numeric"] = df_merchants["merchant_rating"].map(
    rating_mapping
)
preprocessed_merchant_columns.extend(["merchant_rating_numeric"])
# Binary encode merchant_category
encoder_merchant_cat = ce.BinaryEncoder(cols=["merchant_category"]).fit(
    df_merchants["merchant_category"]
)
transformed_column_names = encoder_merchant_cat.get_params()["mapping"][0]["mapping"].columns
df_merchants[transformed_column_names] = encoder_merchant_cat.transform(df_merchants["merchant_category"])
preprocessed_merchant_columns.extend(transformed_column_names)
# Binary encode merchant_location
encoder_merchant_location = ce.BinaryEncoder(cols=["merchant_location"]).fit(
    df_merchants["merchant_location"]
)
transformed_column_names = encoder_merchant_location.get_params()["mapping"][0]["mapping"].columns
df_merchants[transformed_column_names] = encoder_merchant_location.transform(
    df_merchants["merchant_location"]
)
preprocessed_merchant_columns.extend(transformed_column_names)
print("\nPreprocessed Merchants Data:")
print(df_merchants)
# ------------------------------------------------
# Step 2.3: Preprocess Transactions
# ------------------------------------------------
# Process transaction_timestamp: convert to datetime and extract features
df_transactions["transaction_timestamp"] = pd.to_datetime(
    df_transactions["transaction_timestamp"]
)
df_transactions["hour"] = df_transactions["transaction_timestamp"].dt.hour
df_transactions["day_of_week"] = df_transactions["transaction_timestamp"].dt.dayofweek
df_transactions["month"] = df_transactions["transaction_timestamp"].dt.month
preprocessed_transaction_columns.extend(["hour", "day_of_week", "month"])
# Scale the 'amount' column
scaler_amount = StandardScaler()
df_transactions["amount_scaled"] = scaler_amount.fit_transform(
    df_transactions[["amount"]]
)
# Binary encode 'transaction_type'
encoder_tx_type = ce.BinaryEncoder(cols=["transaction_type"]).fit(
    df_transactions["transaction_type"]
)
transformed_column_names = encoder_tx_type.get_params()["mapping"][0]["mapping"].columns
df_transactions[transformed_column_names] = encoder_tx_type.transform(df_transactions["transaction_type"])
preprocessed_transaction_columns.extend(transformed_column_names)
print("\nPreprocessed Transactions Data:")
print(df_transactions)
# ------------------------------------------------
# Step 3.1: Generate Graph nodes and edges
# ------------------------------------------------
## Graph Generation
# - Each node in the graph must have a unique ID.
# - The IDs must be contiguous and start from 0.
# In the following code, merchant nodes are assigned unique IDs starting from 0.
# After renumbering the merchant nodes, customer nodes receive the subsequent IDs,
# and finally, transaction nodes are assigned their IDs in sequence.
COL_MERCHANT_ID = "merchant_id"
COL_TRANSACTION_ID = "transaction_id"
COL_CUSTOMER_ID = "customer_id"
merchant_id_to_zero_based_id = dict(
    (key, idx) for idx, key in enumerate(df_merchants[COL_MERCHANT_ID].unique())
)
df_merchants[COL_MERCHANT_ID] = df_merchants[COL_MERCHANT_ID].map(
    merchant_id_to_zero_based_id
)
nr_unique_merchants = len(df_merchants[COL_MERCHANT_ID].unique())
customer_id_to_zero_based_id = dict(
    (key, idx + nr_unique_merchants)
    for idx, key in enumerate(df_customers[COL_CUSTOMER_ID].unique())
)
df_customers[COL_CUSTOMER_ID] = df_customers[COL_CUSTOMER_ID].map(
    customer_id_to_zero_based_id
)
nr_unique_customers = len(df_customers[COL_CUSTOMER_ID].unique())
transaction_id_to_zero_based_id = dict(
    (key, idx + nr_unique_merchants + nr_unique_customers)
    for idx, key in enumerate(df_transactions[COL_TRANSACTION_ID].unique())
)
df_transactions[COL_TRANSACTION_ID] = df_transactions[COL_TRANSACTION_ID].map(
    transaction_id_to_zero_based_id
)
df_transactions[COL_MERCHANT_ID] = df_transactions[COL_MERCHANT_ID].map(
    merchant_id_to_zero_based_id
)
df_transactions[COL_CUSTOMER_ID] = df_transactions[COL_CUSTOMER_ID].map(
    customer_id_to_zero_based_id
)
# Create Edge List
# This step generates an edge list by connecting customer, merchant, and transaction IDs,
# thereby establishing the relationships between these nodes in the graph.
# Create edges from CUSTOMER_ID to TRANSACTION_ID
edges_customer_to_transaction = df_transactions[
    [COL_CUSTOMER_ID, COL_TRANSACTION_ID]
].rename(columns={COL_CUSTOMER_ID: "src", COL_TRANSACTION_ID: "dst"})
# Create edges from TRANSACTION_ID to MERCHANT_ID
edges_transaction_to_merchant = df_transactions[
    [COL_TRANSACTION_ID, COL_MERCHANT_ID]
].rename(columns={COL_TRANSACTION_ID: "src", COL_MERCHANT_ID: "dst"})
# Combine all edges
edges = pd.concat(
    [
        edges_customer_to_transaction,
        edges_transaction_to_merchant,
    ],
    ignore_index=True,
)
# If needed, each connection can be represented in both directions.
is_bidirectional=False
if is_bidirectional:
    # Create reverse edges from TRANSACTION_ID to CUSTOMER_ID
    edges_transaction_to_customer = df_transactions[
        [COL_CUSTOMER_ID, COL_TRANSACTION_ID]
    ].rename(columns={COL_CUSTOMER_ID: "dst", COL_TRANSACTION_ID: "src"})
    # Create reverse edges from MERCHANT_ID to TRANSACTION_ID
    edges_merchant_to_transaction = df_transactions[
        [COL_TRANSACTION_ID, COL_MERCHANT_ID]
    ].rename(columns={COL_TRANSACTION_ID: "dst", COL_MERCHANT_ID: "src"})
    # Combine all edges
    edges = pd.concat(
        [
            edges_customer_to_transaction,
            edges_transaction_to_customer,
            edges_transaction_to_merchant,
            edges_merchant_to_transaction,
        ],
        ignore_index=True,
    )
print('Edges:')
print(edges)
## Create Node feature
#  Each node's unique identifier corresponds to its row index in the node feature (e.g. node.csv) file.
#  The order of ids must be maintained in the node feature (e.g. `node.csv`) file.
#   1. Merchant nodes: starting from 0.
#   2. Customer nodes: receiving subsequent IDs after merchant nodes.
#   3. Transaction nodes: assigned in sequence after customer nodes.
#
# Create the block matrix with correct node ordering
# The block matrix is constructed by aligning the nodes in the same order as defined as defined earlier.
# This ensures that merchant nodes, customer nodes, and transaction nodes are correctly positioned in the matrix.
M = df_merchants[preprocessed_merchant_columns].values
C = df_customers[preprocessed_customer_columns].values
T = df_transactions[preprocessed_transaction_columns].values
node_feature_df = pd.DataFrame(
    block_diag(M, C, T),
    columns=preprocessed_merchant_columns
    + preprocessed_customer_columns
    + preprocessed_transaction_columns,
)
print('Node feature:')
print(node_feature_df)
## Create Node label
# The node label file must contain the same number of rows as the node feature file (e.g., node.csv).
# For nodes without a label, assign a default value of zero.
# Initialize with all zeros
node_label_df = pd.DataFrame(
    np.zeros(len(node_feature_df), dtype=int), columns=["is_fraud"]
)
# Copy the label of transaction nodes to corresponding indices
node_label_df.iloc[df_transactions[COL_TRANSACTION_ID].values, 0] = df_transactions[
    "is_fraud"
].values
print('Node label:')
print(node_label_df)
# ---------------------------------------------------------------
# Step 3.2: Write the files under the proper directory structure
# ---------------------------------------------------------------
#  data_root
#    ├── edges
#    │   └── node_to_node.csv
#    └── nodes
#        ├── node.csv
#        └── node_label.csv
os.makedirs(os.path.join(DATA_OUTPUT_DIR, "edges"), exist_ok=True)
os.makedirs(os.path.join(DATA_OUTPUT_DIR, "nodes"), exist_ok=True)
node_feature_df.to_csv(os.path.join(DATA_OUTPUT_DIR, "nodes", "node.csv"), index=False)
node_label_df.to_csv(
    os.path.join(DATA_OUTPUT_DIR, "nodes", "node_label.csv"), index=False
)
edges.to_csv(os.path.join(DATA_OUTPUT_DIR, "edges", "node_to_node.csv"), index=False)