In this project a streaming risk processor was developed based on Apache Flink (Python API). The processor has four responsabilities:
- Generate features related for each transaction received (the features will be aggregated at customer_id level);
- assign each transaction to an experiment using experiment assigners;
- apply risk rules based on rules executors that will use the features generated to decide if a transaction must be approved or not;
- maintain a the feature store up-to-date;
- persist data in different sinks: MinIO, Kafka and Postgres.
Besides the risk processor an analtyics processor was built to read the processed transactions and generate data to be analyzed in a dashboard tool (Apache Superset).
The tools used to develop this project were:
- Apache Flink: used as streaming processor and is the application heart and brain;
- Postgres: used as relational database to store every data generated during the application execution;
- Apache Kafka: as the event streaming platform;
- MinIO: used as object storage (that is compatible to the AWS S3 api) to store data to create a data lake;
- Apache Superset: to develop a monitoring tool to analyze metrics related to the risk processor (like approval rate, total amount approved, number of customers transacting, etc);
- Kafdrop: to monitor Kafka broker.
The following figure depicts a high level overview of the risk processor architecture:
And the following one depicts a high level overview of the analytics processor architecture.
Two fake datasets were developed to use within this project. The first one is a customers dataset that have information about each customer and has the following schema:
CREATE TABLE IF NOT EXISTS customers (
customer_id BIGINT PRIMARY KEY,
created_at BIGINT,
name VARCHAR,
ssn VARCHAR,
date_of_birth VARCHAR,
email VARCHAR,
latitude VARCHAR,
longitude VARCHAR
);The customer_id field is an unique identifier that identifies each customer.
The second dataset in a transactional dataset that have the transactions between two customers (from_customer_id to to_customer_id). The schema used for this dataset can be viewed bellow:
CREATE TABLE IF NOT EXISTS transactions (
transaction_id BIGINT PRIMARY KEY,
from_customer_id BIGINT,
to_customer_id BIGINT,
created_at BIGINT,
amount FLOAT,
latitude FLOAT,
longitude FLOAT
);Each transaction in this dataset is sent to a Kafka topic called transactions, respecting the created_at order.
The data generation scripts can be found under the scripts folder and are named generate_customers.py and generate_transactions.py.
these scripts are based in Faker library to generate fake data and also have some logic to avoid data incositencies like a customer sending/receiving money to another customer before the customer creation.
The features generated are basically velocity based features: this features count ou sum the total attempted transactions/amount in the past X hours.
velocity features are useful to limit the exposure in a certain time period.
The velocity features are associated with a customer_id and were calculated for windows of 1, 3, 6, 24 and 48 hours.
The features are calculated in two classes FeaturesProcessor and FeatureStoreProcessor the main difference is that the output of the FeaturesProcessor is used during risk analysis and the output of the FeatureStoreProcessor is sent to the Postgres database, using an upsert operation: the table will only have the most recent data for each customer_id.
Experiment assigners are functions that receive the processor generated data and can assign each transaction/customer to an experiment and to a bucket (control and treatment).
Every assigner is saved in the experiments table with BYTEA type and an active flag (active equals to true means that the experiment are running).
Every experimetn assigner is a Python function that receives only one argument, that will contain the following data (example):
{
"transaction": {
"transaction_id": 99,
"from_customer_id": 7,
"to_customer_id": 220,
"created_at": 1704693776,
"amount": 767.4847412109375,
"latitude": 54.468841552734375,
"longitude": 97.77935028076172
},
"features": {
"feature:velocity_count_last_1h": 1,
"feature:velocity_count_of_credits_last_1h": 1,
"feature:velocity_count_of_debits_last_1h": 0,
"feature:velocity_amount_last_1h": 767.4847412109375,
"feature:velocity_amount_of_credits_last_1h": 767.4847412109375,
"feature:velocity_amount_of_debits_last_1h": 0,
"feature:velocity_count_last_2h": 1,
"feature:velocity_count_of_credits_last_2h": 1,
"feature:velocity_count_of_debits_last_2h": 0,
"feature:velocity_amount_last_2h": 767.4847412109375,
"feature:velocity_amount_of_credits_last_2h": 767.4847412109375,
"feature:velocity_amount_of_debits_last_2h": 0,
"feature:velocity_count_last_3h": 1,
"feature:velocity_count_of_credits_last_3h": 1,
"feature:velocity_count_of_debits_last_3h": 0,
"feature:velocity_amount_last_3h": 767.4847412109375,
"feature:velocity_amount_of_credits_last_3h": 767.4847412109375,
"feature:velocity_amount_of_debits_last_3h": 0,
"feature:velocity_count_last_4h": 1,
"feature:velocity_count_of_credits_last_4h": 1,
"feature:velocity_count_of_debits_last_4h": 0,
"feature:velocity_amount_last_4h": 767.4847412109375,
"feature:velocity_amount_of_credits_last_4h": 767.4847412109375,
"feature:velocity_amount_of_debits_last_4h": 0,
"feature:velocity_count_last_5h": 1,
"feature:velocity_count_of_credits_last_5h": 1,
"feature:velocity_count_of_debits_last_5h": 0,
"feature:velocity_amount_last_5h": 767.4847412109375,
"feature:velocity_amount_of_credits_last_5h": 767.4847412109375,
"feature:velocity_amount_of_debits_last_5h": 0,
"feature:velocity_count_last_24h": 2,
"feature:velocity_count_of_credits_last_24h": 1,
"feature:velocity_count_of_debits_last_24h": 1,
"feature:velocity_amount_last_24h": 1042.85498046875,
"feature:velocity_amount_of_credits_last_24h": 767.4847412109375,
"feature:velocity_amount_of_debits_last_24h": 275.3702392578125,
"feature:velocity_count_last_48h": 3,
"feature:velocity_count_of_credits_last_48h": 2,
"feature:velocity_count_of_debits_last_48h": 1,
"feature:velocity_amount_last_48h": 1265.9140930175781,
"feature:velocity_amount_of_credits_last_48h": 990.5438537597656,
"feature:velocity_amount_of_debits_last_48h": 275.3702392578125
}
} Based on this data the assingment can be done.
The following code is an example of experiment assigner:
def experiment__understand_impact_of_backstop_1h_thresholds_increase(value: dict[str, dict[str, Any]]) -> dict[str, str]:
eligible_condition = value['transaction']['from_customer_id'] % 4 <= 1
treatment_condition = value['transaction']['from_customer_id'] % 4 == 0
experiment_start_date, experiment_end_date = 1706745600, 1709251200
active = experiment_start_date <= value['transaction']['created_at'] <= experiment_end_date
if eligible_condition and active:
ab_group = 'treatment' if treatment_condition else 'control'
return {'ab_group': ab_group, 'level': 'customer', 'group': 'eligible'}The name of the function is used as experiment name.
More experiment assigners can be found under the scripts/generate_experiments.py file.
The rules that will be applied to each transaction are saved in the rules table with BYTEA type and an active flag (active equals to true means that the rules are running). During processing these rules will be retrieved from the database and will be applied to every transction, returning an object with the rule name and decision (approve or block).
As the experiment assigner the rules are Python functions that receive only one argument. The following code is an example of rule:
def backstop__amount_of_transactions_last_24_hour(value: dict[str, dict[str, Any]]) -> str:
if value['features']['feature:velocity_amount_last_24h'] >= 10000:
return 'block'The argument is the same argument received by the experiment assigner, an object containing transaction data and all the features calculated by the processor.
Rules can and must be used together with experiments: the experiment will assign each customer/transaction to a control or treatment group and based on this group the rule action can change. An example can be seen in the following code:
def backstop__count_of_transactions_last_1_hour(value: dict[str, dict[str, Any]]) -> str:
experiment_name = 'experiment__understand_impact_of_backstop_1h_thresholds_increase'
experiment_selected = list(filter(lambda x: x['experiment_name'] == experiment_name, value['experiments']))
eligible = experiment_selected[0]['group'] if experiment_selected != [] else 'not_eligible'
if eligible:
ab_group = experiment_selected[0]['ab_group'] if experiment_selected != [] else 'control'
if (value['features']['feature:velocity_count_last_1h'] >= 10) and (ab_group == 'treatment'):
return 'block'
elif (value['features']['feature:velocity_count_last_1h'] >= 5) and (not (ab_group == 'treatment')):
return 'block'
else:
if value['features']['feature:velocity_count_last_1h'] >= 5:
return 'block'If the transactions is eligible to the experiment and is in the treatment group a different backstop threshold will be applied to define if the transaction must be approved or not.
More rules can be found under the scripts/generate_rules.py file.
To run the project the first thing is to install everything using uv.
After the installation a make setup command will:
- generate a customers.csv (containing customer inforation) and transactions.parquet (containing every transaction) files;
- run the application (this will run every service in the docker-compose.yml file);
- load the pre-definied experiments to the database;
- load the pre-definied rules to the database;
- send both risk and analytics jobs to Apache Flink job manager (that is running in a isolated docker container);
- start to stream the transactions dataset to the Apache Kafka broker (to a topic called transactions).
Two applications will be available to understand what is happening during the job execution:
- the Kafdrop application is running in localhost:19000 and allows the users to see the data in the broker;
- the Flink dashboard is running in localhost:8081 and allows the usrs to see both jobs running (risk and analytics processor);
- the Minio browser is available in the localhost:9001.
To access the Postgres database a software like PGAdmin or DBeaver can be used.
To clear the data the command make clear can be used.
├── Dockerfile.flink \\ Flink dockefile
├── Dockerfile.postgres \\ Postgres dockefile
├── Dockerfile.superset \\ Superset dockefile
├── Makefile \\ Makefile with commands to run the project
├── README.md
├── analytics \\ Folder with analytics service
│ ├── aggregations
│ │ ├── __init__.py
│ │ └── metrics.py \\ Python code that generated the aggregations used to create the visualizations
│ ├── data
│ │ ├── __init__.py
│ │ ├── sinks.py \\ Data sink definition
│ │ └── sources.py \\ Data source definition
│ └── main.py
├── config.sh \\ .sh file to configure Superset
├── data \\ Folder to store all the data generated
├── data_lake \\ Folder to store MinIO data
│ └── raw \\ Bucket name
│ └── artifacts
├── docker-compose.yml \\ Docker compose configuration
├── init.sql \\ Database initialiation
├── jars \\ Folder to store necessary .jar files
├── notebooks \\ Folder to store project notebooks
├── processor \\ Folder with risk service
│ ├── data \\ Data definitions
│ │ ├── __init__.py
│ │ ├── schema.py \\ Schemas used by the processor
│ │ ├── sinks.py \\ Data sinks definitions
│ │ ├── sources.py \\ Data source definitions
│ │ └── utils.py \\ Utils
│ ├── database \\ Database manipulation files
│ │ ├── __init__.py
│ │ └── query.py \\ Define the function used to interact with the database
│ ├── features \\ Folder to store features definitions
│ │ ├── __init__.py
│ │ └── velocity.py \\ Velocity features definition
│ ├── input \\ Folder to store input data definition
│ │ ├── __init__.py
│ │ └── transactions.py \\ Definition of input data
│ ├── main.py \\ Main code
│ └── processors \\ Folder to store processors used inside the risk processor
│ ├── __init__.py
│ ├── experiments.py \\ Experiment assigner
│ ├── features.py \\ Features calculation
│ └── rules.py \\ Rules engine
├── pyproject.toml
├── scripts \\ Folder to store scripts necessary to run the project
│ ├── clear_kafka.py \\ Utils to clear Kafka broker
│ ├── create_topics.py \\ Utils to create topics
│ ├── delete_topics.py \\ Utils to delete topics
│ ├── generate_customers.py \\ Code to generate customers data
│ ├── generate_experiments.py \\ Code to generate experiment assigners and save in the database
│ ├── generate_rules.py \\ Code to generate rules and save in the database
│ ├── generate_transactions.py \\ Code to generate transactions
│ └── producer.py \\ Code used to produce data to Kafka
├── superset \\ Folder to store Superset artifacts
│ └── superset.db
└── uv.lock