Risk Analytics in Python: Building Models Using VS Code, ML Python Libraries, Kedro and Docker
Alexey KhoroshilovTable of contents
Introduction
In today's financial ecosystem, where change is practically unstoppable, risk analytics is an important building block in every decision-making process for banks and fintech companies. During the more than six years of accrued experience in banking with specialized expertise in risk analytics, modeling, and validation, I have been able to witness changes in risk modeling techniques over time and the growing need for productive, scalable, and reproducible methods of building risk models.
In this article, we will review how Python is being utilized in risk analytics, focusing on the model development using templates. This approach not only streamlines the development process but assures consistent, maintainable results for any given project of risk modeling. We'll go through examples taken from various aspects of the industry and present some best practices for different types of professionals working in this exciting field.
Theory
Why Risk Analytics in Finance Matters
Risk in the Financial Sector
The finance industry inherently encompasses a degree of risk. From credit risk to market risk, operational to liquidity risk, financial firms always need to assess and manage various kinds of risks so that stability and profitability can be maintained. It is risk analytics that provides these key toolkits and methodologies for quantifying and analyzing, hence mitigating, these risks effectively.
The Role of Data Science in Risk Management
Data science has totally changed the paradigm for risk management in finance. Financial institutions can now create more accurate prediction models using data from advanced statistical techniques, machine learning algorithms, and big data technologies. This enables them to uncover complex patterns and relationships in the data and apply these to automating various risk assessment processes and providing real-time risk insights. As one such example, today, most banks make use of machine learning models to predict the probability of default of applicants for loans, thus facilitating more correct credit decisioning.
Python: The Go-To Language for Risk Analytics
Why Python?
Python is the de facto language in data science and risk analytics due to its simplicity, versatility, and powerful ecosystem of libraries. Most financial institutions have migrated from legacy systems built around SAS to Python, reaping numerous benefits therein:
Cost-effectiveness: Open-source in nature, reducing licensing costs
Flexibility: Easier integrations with modern data science tools and frameworks
Talent pool: Wider availability of Python developers
Community support: Access to a large community and continuous enhancements
Essential Tools for Risk Analytics
The cornerstones of risk analytics have been the following few Python libraries:
Pandas: Data manipulation and analysis
NumPy: Numerical computing
Scikit-learn: ML algorithms
OptBinning: Data binning
Statsmodels: Statistical modeling
Matplotlib, Seaborn, Plotly: Data visualization
The de facto standard framework for building ML pipelines is Kedro, while containerization has historically been Docker’s domain. For any new risk analytics project, list out all the libraries with their version numbers in a requirements.txt file. This ensures reproducibility across environments.
Building Risk Models Using Templates
The Concept of Model Templates
Model templates are pre-defined structures or frameworks representing the underlying basis from which a risk model is built. A template encapsulates standard practices, best practices in model development, as well as common functionalities. Some of the benefits of using templates include:
Consistency across various models and teams
Less development time
Maintenance and updates easier
Better code quality and readability
Key Components of a Risk Model Template
A typical comprehensive template will contain the following:
Data pre-processing module
A feature engineering framework
A model training and evaluation pipeline
Model performance metrics calculation features
Reporting and visualization functions
Model validation checks
Advanced Techniques in Risk Modeling
Incorporating Macroeconomic Factors
Most credit portfolio stress testing exercises center on macroeconomic modeling. It therefore, becomes necessary to identify the key macroeconomic variables such as GDP growth, unemployment, or inflation. The analyst thereafter constructs a time series model in order to project these indicators and injects this projection into the risk model to determine the magnitude of performance of the portfolio against varied economic conditions. For example, during the COVID-19 pandemic, banks had to rush and alter their stress testing models for economic conditions never seen before. Thus, flexibility in macroeconomic modeling in times of crisis is paramount.
Ensemble Methods for Improved Accuracy
Ensemble methods generally perform better in risk analytics, considering the strengths of various algorithms combined to improve accuracy. Effective ensemble techniques include Random Forests, Gradient Boosting Machines including XGBoost and LightGBM, and stacking for enhanced predictive power. Nevertheless, with the use of the ensemble methods, there is a need to appreciate the trade-off between performance and interpretability. This is particularly so in regulated environments, where explainability often complements accuracy.
Neural Networks for Complex Risk Patterns
Neural networks are ideal for modeling complex and nonlinear relationships that may appear in the data. Therefore, these networks would be best suited for finding complex risk patterns. They work well on high-dimensional feature space and do feature learning automatically. Deep learning models have been applied to perform fraud detection of transactions in real time and achieve much higher detection rates compared to traditional rule-based approaches in the financial world.
Implementing Models in Production
Model Monitoring and Validation
Successful model creation involves manifold key components. For the monitoring of these models effectively and their validation, one needs automatic performance tracking with the threshold definition of model degradation, alerting systems on material performance drops, and retro tests or backtesting. Here's a good practice for robust monitoring: In pursuit of all this, one could look at a dashboard that has real-time data so that any issue can be quickly pinpointed and updated on time.
Integrating Models into Risk Decision Systems
It is important to pay attention to designing modular and scalable code structures, efficiently using data pipelines, and realizing real-time scoring of data in integrating models into production systems to make risk decisions. Besides, the development of API endpoints facilitates seamless model integrations. Examples could be seen in how banks use a real-time credit scoring system that merges different risk models to grant instant lending decisions for online loan applications.
Challenges and Best Practices in Risk Analytics
Data Quality and Availability
Any risk analytics accuracy depends on high-quality data being available. It is best practice to include data validation checks, development of data quality scorecards, putting data governance policies in place, and looking for alternative data sources where traditional data may be in short supply. This can be helped by automation of the checks of data quality within ETL pipelines to find potential issues early.
Model Interpretability and Explainability
While the number of models is increasing and growing in complexity, so is the need for interpretability. Techniques vary from SHAP value calculation over feature importance, Partial Dependence Plots to understand how features have an impact on the outcomes of the model, up to LIME, which tries to explain each prediction model-agnostically, locally. For instance, in the financial sector, regulators demand transparent explanations for credit decisions; thus, interpretable AI is favored.
Regulatory Compliance
There is a lot of documentation for institutions that operate in highly regulated environments in the processes that lead to model development, routine validations, regulatory requirements like the BASEL standards, orderly inventories of models, and version control. Probably the best pragmatic way out for compliance with these requires a model governance framework with a periodic review cycle and broad documentation standards.
Practice
Project Plan
Create an environment for model development using Docker
Create a model template using Kedro
Create a PD model - an example from the Kaggle competition using our template
Creating environment for model development
4. Create folders and files structures on disk:
5. Run Docker instructions:
In ./risk_modeling/docker-compose.yaml:
services:
p310:
build: env/
ports:
- 4141:4141
volumes:
- ./:/docker_disk
tty: true
In ./risk_modeling/env/Dockerfile:
FROM python:3.10-bookworm
COPY requirements.txt /requirements.txt
RUN apt-get update && apt-get upgrade -y && apt-get install cmake -y && apt install libopenblas-dev -y
RUN pip install -r requirements.txt
Installing cmake & libopenblas-dev is necessary for some libraries in requirements
In ./risk_modeling/env/requirements.txt:
pandas==1.4.1
numpy==1.26.3
scikit-learn==1.4.0
scipy==1.11.4
catboost==1.2.2
jupyterlab==4.3.4
ks-metric==0.2.0
lightgbm==3.2.1
matplotlib==3.8.2
matplotlib-inline==0.1.6
missingno==0.5.2
mlxtend==0.21.0
optbinning==0.19.0
optuna==3.5.0
plotly==5.18.0
tqdm==4.66.1
kedro==0.19.10
kedro-viz==10.1.0
kedro-datasets==6.0.0
notebook==7.3.2
openpyxl==3.0.10
6. Build the Docker image:
You can use the system command prompt or terminal in VS Code:
Open the directory risk_modeling. In the folder (or where the file docker-compose.yaml is located) run in terminal:
docker-compose build
Or:
docker-compose -f docker-compose.yaml build
We can check the created docker image using docker command in terminal:
docker images
Next step: we start our development service container from image risk_modeling-p310 with the command in terminal:
docker-compose up
Or:
docker-compose -f docker-compose.yaml up
And finally, attach an IDE (VS Code in our case) to the created Docker container:
The next steps of model development will take place in our container:
Inside container we download extension:
Create a Model Template Using Kedro
Use the following command In the container’s terminal:
kedro new
Set the project name:
risk_model
In the example we will not use PySpark, so include:
1,2,3,4,5,7
We don’t need an example:
no
Now we’ve got a model project structure, with the Kedro template:
./conf: folder for credentials, input parameters and directories (data paths)
./data: data storage folder
./docs: project documentation
./logs: project logs
./notebooks: for code drafts or hypothesis testing using the Jupyter notebook format
./src: main folder with all model codes and pipelines
./tests: folder for testing including writing unit tests
Create a PD Model (Kaggle Competition Example Using Our Template)
We’ve solved a Kaggle task. Banks play a crucial role in market economies. They decide who can get finance and on what terms and can make or break investment decisions. For markets and society to function, individuals and companies need access to credit.
Credit scoring algorithms, which make a guess at the probability of default, are the method banks use to determine whether or not a loan should be granted. This competition requires participants to improve on the state of the art in credit scoring, by predicting the probability that somebody will experience financial distress in the next two years.
Download the Data
Put the unpacked Kaggle data to directory: 01_raw
Set Up Configuration Files
This section specifies saving paths, file types and formats that are obtained during model creation. This includes datasets, binning files, trained model files, tables with statistics, graphs, etc.
In /docker_disk/risk-model/conf/base/catalog.yml:
# https://www.kaggle.com/competitions/GiveMeSomeCredit/data
01_raw_development:
type: pandas.CSVDataset
filepath: data/01_raw/cs-training.csv
01_raw_validation:
type: pandas.CSVDataset
filepath: data/01_raw/cs-test.csv
01_raw_example_to_submit:
type: pandas.CSVDataset
filepath: data/01_raw/sampleEntry.csv
# 01_preprocessing pipeline data: preprocessing stages
## fix index
02_intermediate_development:
type: pickle.PickleDataset
filepath: data/02_intermediate/df_dev.pkl
02_intermediate_validation:
type: pickle.PickleDataset
filepath: data/02_intermediate/df_validation.pkl
## new features
04_feature_development:
type: pickle.PickleDataset
filepath: data/04_feature/df_dev.pkl
04_feature_validation:
type: pickle.PickleDataset
filepath: data/04_feature/df_validation.pkl
## train/test & WOE-binning
05_model_input_X_train:
type: pickle.PickleDataset
filepath: data/05_model_input/X_train.pkl
05_model_input_X_test:
type: pickle.PickleDataset
filepath: data/05_model_input/X_test.pkl
05_model_input_y_train:
type: pickle.PickleDataset
filepath: data/05_model_input/y_train.pkl
05_model_input_y_test:
type: pickle.PickleDataset
filepath: data/05_model_input/y_test.pkl
### long list
05_model_input_X_train_woe:
type: pickle.PickleDataset
filepath: data/05_model_input/X_train_woe.pkl
05_model_input_X_test_woe:
type: pickle.PickleDataset
filepath: data/05_model_input/X_test_woe.pkl
05_model_input_development_woe:
type: pickle.PickleDataset
filepath: data/05_model_input/df_dev_woe.pkl
05_model_input_validation_woe:
type: pickle.PickleDataset
filepath: data/05_model_input/df_validation_woe.pkl
### short list
05_model_input_X_train_woe_short_list:
type: pickle.PickleDataset
filepath: data/05_model_input/X_train_woe_short_list.pkl
05_model_input_X_test_woe_short_list:
type: pickle.PickleDataset
filepath: data/05_model_input/X_test_woe_short_list.pkl
05_model_input_development_woe_short_list:
type: pickle.PickleDataset
filepath: data/05_model_input/df_dev_woe_short_list.pkl
05_model_input_validation_woe_short_list:
type: pickle.PickleDataset
filepath: data/05_model_input/df_validation_woe_short_list.pkl
# 02_modeling pipeline data: modeling stages and final model
## model files
06_models_feature_selection:
type: pickle.PickleDataset
filepath: data/06_models/sfs.pkl
06_models_binning:
type: pickle.PickleDataset
filepath: data/06_models/bp.pkl
06_models_binning_short_list:
type: pickle.PickleDataset
filepath: data/06_models/bp_short_list.pkl
06_models_lr:
type: pickle.PickleDataset
filepath: data/06_models/lr.pkl
06_models_scorecard:
type: pickle.PickleDataset
filepath: data/06_models/scorecard.pkl
## scored files
07_model_output_sample_to_kaggle:
type: pandas.CSVDataset
filepath: data/07_model_output/sampleEntry.csv
07_model_output_development_scored:
type: pickle.PickleDataset
filepath: data/07_model_output/df_dev_scored.pkl
07_model_output_validation_scored:
type: pickle.PickleDataset
filepath: data/07_model_output/df_validation_scored.pkl
# 03_reporting pipeline data: reports
08_reporting_variables_binning_table:
type: pandas.ExcelDataset
filepath: data/08_reporting/table_variables_binning.xlsx
save_args:
index: True
sheet_name: Sheet1
metadata:
kedro-viz:
preview_args:
nrows: 1000
08_reporting_variables_binning_table_short_list:
type: pandas.ExcelDataset
filepath: data/08_reporting/table_variables_binning_short_list.xlsx
save_args:
index: True
sheet_name: Sheet1
metadata:
kedro-viz:
preview_args:
nrows: 1000
08_reporting_variables_summary_table:
type: pandas.ExcelDataset
filepath: data/08_reporting/table_variables_summary.xlsx
save_args:
sheet_name: Sheet1
metadata:
kedro-viz:
preview_args:
nrows: 1000
08_reporting_variables_summary_table_short_list:
type: pandas.ExcelDataset
filepath: data/08_reporting/table_variables_summary_short_list.xlsx
save_args:
sheet_name: Sheet1
metadata:
kedro-viz:
preview_args:
nrows: 1000
08_reporting_model_scorecard_table:
type: pandas.ExcelDataset
filepath: data/08_reporting/table_model_scorecard.xlsx
save_args:
sheet_name: Sheet1
metadata:
kedro-viz:
preview_args:
nrows: 1000
08_reporting_model_feature_correlation:
type: pandas.ExcelDataset
filepath: data/08_reporting/table_model_lr_features_correlation.xlsx
save_args:
sheet_name: Sheet1
metadata:
kedro-viz:
preview_args:
nrows: 1000
08_reporting_model_feature_correlation_plot:
type: matplotlib.MatplotlibWriter
filepath: data/08_reporting/plot_model_feature_correlation.jpeg
08_reporting_model_scorecard_feature_coefs:
type: tracking.JSONDataset
filepath: data/08_reporting/model_scorecard_features_coefs.json
08_reporting_model_scorecard_feature_selection_plot:
type: matplotlib.MatplotlibWriter
filepath: data/08_reporting/plot_model_scorecard_feature_selection.jpeg
08_reporting_statistics_monitoring_train_test:
type: pickle.PickleDataset
filepath: data/08_reporting/scorecard_monitoring_train_test.pkl
08_reporting_statistics_monitoring_dev_valid:
type: pickle.PickleDataset
filepath: data/08_reporting/scorecard_monitoring_dev_valid.pkl
#
08_reporting_model_scorecard_features_psi_detailed_train_test:
type: pandas.ExcelDataset
filepath: data/08_reporting/table_model_scorecard_features_psi_detailed_train_test.xlsx
save_args:
sheet_name: Sheet1
metadata:
kedro-viz:
preview_args:
nrows: 1000
08_reporting_model_scorecard_features_psi_summary_train_test:
type: pandas.ExcelDataset
filepath: data/08_reporting/table_model_scorecard_features_psi_summary_train_test.xlsx
save_args:
sheet_name: Sheet1
metadata:
kedro-viz:
preview_args:
nrows: 1000
08_reporting_model_scorecard_features_psi_detailed_dev_valid:
type: pandas.ExcelDataset
filepath: data/08_reporting/table_model_scorecard_features_psi_detailed_dev_valid.xlsx
save_args:
sheet_name: Sheet1
metadata:
kedro-viz:
preview_args:
nrows: 1000
08_reporting_model_scorecard_features_psi_summary_dev_valid:
type: pandas.ExcelDataset
filepath: data/08_reporting/table_model_scorecard_features_psi_summary_dev_valid.xlsx
save_args:
sheet_name: Sheet1
metadata:
kedro-viz:
preview_args:
nrows: 1000
08_reporting_model_scorecard_psi_summary_train_test:
type: pandas.ExcelDataset
filepath: data/08_reporting/table_model_scorecard_psi_summary_train_test.xlsx
save_args:
sheet_name: Sheet1
metadata:
kedro-viz:
preview_args:
nrows: 1000
08_reporting_model_scorecard_psi_summary_dev_valid:
type: pandas.ExcelDataset
filepath: data/08_reporting/table_model_scorecard_psi_summary_dev_valid.xlsx
save_args:
sheet_name: Sheet1
metadata:
kedro-viz:
preview_args:
nrows: 1000
08_reporting_model_scorecard_psi_summary_train_test_plot:
type: matplotlib.MatplotlibWriter
filepath: data/08_reporting/plot_model_scorecard_psi_summary_train_test.jpeg
08_reporting_model_scorecard_psi_summary_dev_valid_plot:
type: matplotlib.MatplotlibWriter
filepath: data/08_reporting/plot_model_scorecard_psi_summary_dev_valid.jpeg
08_reporting_model_scorecard_statistical_tests:
type: pandas.ExcelDataset
filepath: data/08_reporting/table_model_scorecard_statistical_tests.xlsx
save_args:
sheet_name: Sheet1
metadata:
kedro-viz:
preview_args:
nrows: 1000
08_reporting_model_scorecard_roc_auc_train_test_plot:
type: matplotlib.MatplotlibWriter
filepath: data/08_reporting/plot_model_scorecard_roc_auc_train_test.jpeg
08_reporting_model_scorecard_ks_plot:
type: matplotlib.MatplotlibWriter
filepath: data/08_reporting/plot_model_scorecard_ks_train_test.jpeg
08_reporting_model_scorecard_coefs_plot:
type: matplotlib.MatplotlibWriter
filepath: data/08_reporting/plot_model_scorecard_coefs.jpeg
We will also set the names of the fields from the task file, list global variables, hyperparameters of the model, and set manual binning for those features for which it is needed.
In /docker_disk/risk-model/conf/base/parameters.yml
cols:
col_raw_id: 'Unnamed: 0'
col_final_id: 'Id'
col_final_prediction: 'Prediction'
cols_variables: ['RevolvingUtilizationOfUnsecuredLines',
'age',
'NumberOfTime30-59DaysPastDueNotWorse',
'DebtRatio',
'MonthlyIncome',
'NumberOfOpenCreditLinesAndLoans',
'NumberOfTimes90DaysLate',
'NumberRealEstateLoansOrLines',
'NumberOfTime60-89DaysPastDueNotWorse',
'NumberOfDependents',
'Debt',
'NumberOfTime30+DaysPastDueNotWorse',
'avg_debt_per_credit',
'feat_killer']
cols_categorial: ['feat_killer']
cols_to_manual_binning_object: ['feat_killer']
col_target: 'SeriousDlqin2yrs'
global_parameters:
seed: 1234
test_ratio: 0.33
binning_parameters:
metric: 'woe'
metric_missing: 'empirical'
feature_selection_parameters:
selection_feature_metric: 'roc_auc'
selection_forward: True
manual_binning:
splits: {'feat_killer':
[
[
[ 'flag_bad_utilization=0&flag_bad_dlq=0&flag_bad_noopencreds=0&' ],
[ 'flag_bad_utilization=0&flag_bad_dlq=0&flag_bad_noopencreds=1&',
'flag_bad_utilization=1&flag_bad_dlq=0&flag_bad_noopencreds=0&',
'flag_bad_utilization=1&flag_bad_dlq=0&flag_bad_noopencreds=1&' ],
[ 'flag_bad_utilization=0&flag_bad_dlq=1&flag_bad_noopencreds=0&' ],
[ 'flag_bad_utilization=0&flag_bad_dlq=1&flag_bad_noopencreds=1&',
'flag_bad_utilization=1&flag_bad_dlq=1&flag_bad_noopencreds=0&',
'flag_bad_utilization=1&flag_bad_dlq=1&flag_bad_noopencreds=1&' ]
],
[True, True, True, True]
]
}
monitoring_parameters:
psi_n_bins: 10
psi_method: 'cart'
inplace_y_actual: 0
none_type: 'None'
Create the Pipelines
Next, we code our model in three stages.
In /docker_disk/risk-model/src/risk_model/pipelines:
Create 3 pipelines in 3 folders:
01_preprocessing
02_modeling
03_reporting
In the directories, create files:
init.py
nodes.py
pipeline.py
Pipeline 1: Preprocessing
At the preprocessing stage, we will create additional features that we will use in modeling. For feature preprocessing, we will use WOE binning using the OptBinning library.
In ~/pipelines/01_preprocessing/__init__.py:
from .pipeline import create_pipeline
In ~/pipelines/01_preprocessing/nodes.py:
import numpy as np
import pandas as pd
from typing import Dict, Tuple
from optbinning import BinningProcess
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
import warnings
warnings.filterwarnings('ignore')
def get_index(df:pd.DataFrame, col_to_index:str, index_name:str)->pd.DataFrame:
"""
rebuild indexes
"""
df_indexed = df.rename(columns={col_to_index:index_name}).set_index(index_name)
return df_indexed
def _get_killer_feature(df:pd.DataFrame)->pd.DataFrame:
"""
"""
df_ = df.copy(deep=True)
for x in ['flag_bad_utilization', 'flag_bad_dlq', 'flag_bad_noopencreds']:
df_[x+'_modified'] = x +'='+ df_[x].astype(str) + '&'
df_['feat_killer'] = df_[[x for x in df_ if '_modified' in x]].sum(axis=1)
return df_['feat_killer']
def get_extra_features(df:pd.DataFrame)->pd.DataFrame:
"""
feature engineering
Debt: DebtRatio * MonthlyIncome
NumberOfTime30+DaysPastDueNotWorse: sum('NumberOfTime30-59DaysPastDueNotWorse', 'NumberOfTime60-89DaysPastDueNotWorse', 'NumberOfTimes90DaysLate')
avg_debt_per_credit: Debt / NumberOfOpenCreditLinesAndLoans where NumberOfOpenCreditLinesAndLoans > 0 else NULL
feat_killer: 3 flags of bad clients (flag_bad_utilization, flag_bad_dlq, flag_bad_noopencreds)
"""
df['Debt'] =df['DebtRatio']*df['MonthlyIncome'].fillna(1)
df['NumberOfTime30+DaysPastDueNotWorse'] = df[['NumberOfTime30-59DaysPastDueNotWorse','NumberOfTime60-89DaysPastDueNotWorse', 'NumberOfTimes90DaysLate']].sum(axis=1)
df['avg_debt_per_credit'] = np.where( df['NumberOfOpenCreditLinesAndLoans']>0, df['Debt']/df['NumberOfOpenCreditLinesAndLoans'], np.nan)
df['flag_bad_utilization'] = np.where(df['RevolvingUtilizationOfUnsecuredLines']>1, 1, 0)
df['flag_bad_dlq'] = np.where(df['NumberOfTime30+DaysPastDueNotWorse']>1, 1, 0)
df['flag_bad_noopencreds'] = np.where(df['NumberOfOpenCreditLinesAndLoans']== 0, 1, 0)
df['feat_killer'] = _get_killer_feature(df)
return df
def get_train_test(df: pd.DataFrame, target, test_size:float, random_state:int)->pd.DataFrame:
"""
divide development sample on train/test
"""
X_train, X_test, y_train, y_test = train_test_split(
df[[x for x in df.columns if x!=target ]], df[target], test_size=test_size, random_state=random_state)
return X_train, X_test, y_train, y_test
def get_binning(df_X_train:pd.DataFrame,
series_y_train:pd.Series,
variable_names:list,
categorical_variables:list,
manual_cols_to_bin_splits:dict,
):
"""
binning with manual splits corrections
"""
binning_fit_params={}
if len(manual_cols_to_bin_splits)>0:
for feat in manual_cols_to_bin_splits.keys():
binning_fit_params[feat] = {
'user_splits': np.array(manual_cols_to_bin_splits[feat][0], dtype=object),
'user_splits_fixed': manual_cols_to_bin_splits[feat][1]
}
else:
pass
bp = BinningProcess(variable_names=variable_names,
categorical_variables=categorical_variables,
binning_fit_params=binning_fit_params)
bp.fit(df_X_train[variable_names], series_y_train)
return bp
def get_woe_binned_features(df:pd.DataFrame, bp, metric:str, metric_missing:str ):
"""
transformation to binned
"""
df_woe = bp.transform(df[bp.variable_names], metric=metric, metric_missing=metric_missing )
return df_woe
In ~/pipelines/01_preprocessing/pipeline.py:
from kedro.pipeline import Pipeline, node, pipeline
from .nodes import *
def create_pipeline(**kwargs) -> Pipeline:
return pipeline(
[
node(
func = get_index,
inputs = dict(df = "01_raw_development",
col_to_index ="params:cols.col_raw_id",
index_name = "params:cols.col_final_id"),
outputs="02_intermediate_development",
name="get_index_development",
),
node(
func = get_index,
inputs = dict(df = "01_raw_validation",
col_to_index ="params:cols.col_raw_id",
index_name = "params:cols.col_final_id"),
outputs="02_intermediate_validation",
name="get_index_validation",
),
node(
func = get_extra_features,
inputs = "02_intermediate_development",
outputs="04_feature_development",
name="get_extra_features_development",
),
node(
func = get_extra_features,
inputs = "02_intermediate_validation",
outputs="04_feature_validation",
name="get_extra_features_validation",
),
node(
func = get_train_test,
inputs = dict(df = "04_feature_development",
target="params:cols.col_target",
test_size="params:global_parameters.test_ratio",
random_state="params:global_parameters.seed"),
outputs=["05_model_input_X_train",
"05_model_input_X_test",
"05_model_input_y_train",
"05_model_input_y_test"],
name="get_train_test",
),
node(
func = get_binning,
inputs = dict(df_X_train = "05_model_input_X_train",
series_y_train="05_model_input_y_train",
variable_names="params:cols.cols_variables",
categorical_variables="params:cols.cols_categorial",
manual_cols_to_bin_splits="params:manual_binning.splits"),
outputs="06_models_binning",
name="get_binning",
),
node(
func = get_woe_binned_features,
inputs = dict(df = "05_model_input_X_train",
bp="06_models_binning",
metric="params:binning_parameters.metric",
metric_missing="params:binning_parameters.metric_missing",
),
outputs="05_model_input_X_train_woe",
name="get_woe_binned_features_X_train",
),
node(
func = get_woe_binned_features,
inputs = dict(df = "05_model_input_X_test",
bp="06_models_binning",
metric="params:binning_parameters.metric",
metric_missing="params:binning_parameters.metric_missing",
),
outputs="05_model_input_X_test_woe",
name="get_woe_binned_features_X_test",
),
node(
func = get_woe_binned_features,
inputs = dict(df = "04_feature_development",
bp="06_models_binning",
metric="params:binning_parameters.metric",
metric_missing="params:binning_parameters.metric_missing",
),
outputs="05_model_input_development_woe",
name="get_woe_binned_features_development",
),
node(
func = get_woe_binned_features,
inputs = dict(df = "04_feature_validation",
bp="06_models_binning",
metric="params:binning_parameters.metric",
metric_missing="params:binning_parameters.metric_missing",
),
outputs="05_model_input_validation_woe",
name="get_woe_binned_features_validation",
),
]
)
Pipeline 2: Modeling
At the second stage, we will select binned variables using the forward feature selection method from mlxtend library (https://rasbt.github.io/mlxtend/), optimizing the ROC AUC metric.
After the selection, we will manually adjust the binning (features and group boundaries were indicated in the file: /docker_disk/risk-model/conf/base/parameters.yml ). Next, we will train the model itself on the final dataset.
In ~/pipelines/02_modeling/__init__.py:
from .pipeline import create_pipeline
In ~/pipelines/02_modeling/nodes.py:
import numpy as np
import pandas as pd
from optbinning import BinningProcess
from optbinning import Scorecard
from typing import Dict, Tuple
from mlxtend.feature_selection import SequentialFeatureSelector as SFS
from mlxtend.plotting import plot_sequential_feature_selection as plot_sfs
from mlxtend.evaluate import PredefinedHoldoutSplit
from sklearn.linear_model import LogisticRegression
import warnings
warnings.filterwarnings('ignore')
def feature_selection_logreg_predefined_test( df_dev_woe:pd.DataFrame,
df_dev:pd.DataFrame,
target:str,
X_test:pd.DataFrame,
features_in: list,
seed:int,
selection_forward:bool,
selection_feature_metric:str,
n_jobs=-1
):
"""
feature selection with predefined test sample
"""
test_index = PredefinedHoldoutSplit(X_test.index.to_list())
y_dev = df_dev[target]
np.bool = np.bool_ # old numpy
lr = LogisticRegression(random_state=seed)
sfs = SFS( estimator=lr,
k_features=(1, df_dev_woe[features_in].shape[1]),
forward=selection_forward,
floating=False,
scoring=selection_feature_metric,
cv=test_index)
sfs.fit(df_dev_woe[features_in], y_dev)
return sfs
def get_binning_short_list(df_X_train:pd.DataFrame,
series_y_train:pd.Series,
sfs,
categorical_variables:list,
manual_cols_to_bin_splits:dict,
):
"""
binning with manual splits corrections
"""
variable_names = list(sfs.k_feature_names_)
categorical_variables_in = [x for x in categorical_variables if x in variable_names]
binning_fit_params={}
if len(manual_cols_to_bin_splits)>0:
for feat in manual_cols_to_bin_splits.keys():
binning_fit_params[feat] = {
'user_splits': np.array(manual_cols_to_bin_splits[feat][0], dtype=object),
'user_splits_fixed': manual_cols_to_bin_splits[feat][1]
}
else:
pass
bp_short = BinningProcess(variable_names=variable_names,
categorical_variables=categorical_variables_in,
binning_fit_params=binning_fit_params)
bp_short.fit(df_X_train[bp_short.variable_names], series_y_train)
return bp_short
def get_woe_binned_features_short_list(df:pd.DataFrame, bp_short, metric:str, metric_missing:str ):
"""
transformation to binned
"""
df_woe = bp_short.transform(df[bp_short.variable_names], metric=metric, metric_missing=metric_missing )
return df_woe
def model_scorecard_logreg_sfs(X_train:pd.DataFrame,
y_train:pd.Series,
bp_short,
seed:int):
"""
modeling
"""
lr = LogisticRegression(random_state=seed)
scorecard = Scorecard(binning_process=bp_short,
estimator=lr,
scaling_method=None,
)
scorecard.fit(X_train[list(bp_short.variable_names)], y_train)
return lr, scorecard
In ~/pipelines/02_modeling/pipeline.py:
from kedro.pipeline import Pipeline, node, pipeline
from .nodes import *
def create_pipeline(**kwargs) -> Pipeline:
return pipeline(
[
node(
func = feature_selection_logreg_predefined_test,
inputs = dict( df_dev_woe = "05_model_input_development_woe",
df_dev = "04_feature_development",
target = "params:cols.col_target",
X_test ="05_model_input_X_test_woe",
features_in = "params:cols.cols_variables" ,
seed="params:global_parameters.seed",
selection_forward="params:feature_selection_parameters.selection_forward",
selection_feature_metric ="params:feature_selection_parameters.selection_feature_metric"),
outputs = "06_models_feature_selection",
name = "feature_selection_logreg_predefined_test"
),
node(
func = get_binning_short_list,
inputs = dict(df_X_train = "05_model_input_X_train",
series_y_train = "05_model_input_y_train",
sfs = "06_models_feature_selection",
categorical_variables = "params:cols.cols_categorial",
manual_cols_to_bin_splits = "params:manual_binning.splits"),
outputs = "06_models_binning_short_list",
name = "get_binning_short_list"
),
node(
func = get_woe_binned_features_short_list,
inputs = dict(df = "05_model_input_X_train",
bp_short="06_models_binning_short_list",
metric="params:binning_parameters.metric",
metric_missing="params:binning_parameters.metric_missing",
),
outputs="05_model_input_X_train_woe_short_list",
name="get_woe_binned_features_X_train_short_list",
),
node(
func = get_woe_binned_features_short_list,
inputs = dict(df = "05_model_input_X_test",
bp_short="06_models_binning_short_list",
metric="params:binning_parameters.metric",
metric_missing="params:binning_parameters.metric_missing",
),
outputs="05_model_input_X_test_woe_short_list",
name="get_woe_binned_features_X_test_short_list",
),
node(
func = get_woe_binned_features_short_list,
inputs = dict(df = "04_feature_development",
bp_short="06_models_binning_short_list",
metric="params:binning_parameters.metric",
metric_missing="params:binning_parameters.metric_missing",
),
outputs="05_model_input_development_woe_short_list",
name="get_woe_binned_features_development_short_list",
),
node(
func = get_woe_binned_features_short_list,
inputs = dict(df = "04_feature_validation",
bp_short="06_models_binning_short_list",
metric="params:binning_parameters.metric",
metric_missing="params:binning_parameters.metric_missing",
),
outputs="05_model_input_validation_woe_short_list",
name="get_woe_binned_features_validation_short_list",
),
node(
func = model_scorecard_logreg_sfs,
inputs = dict( X_train = "05_model_input_X_train",
y_train = "05_model_input_y_train",
bp_short = "06_models_binning_short_list",
seed = "params:global_parameters.seed"),
outputs = ["06_models_lr", "06_models_scorecard"],
name = "model_scorecard_logreg_sfs"
),
]
)
Pipeline 3: Reporting
At the last stage, we will score using the final model samples:
Train, Test
Development (Train + Test)
Validation Sample for Kaggle.com
We will also conduct all the necessary validation tests and generate tables and graphs for the model report.
In ~/pipelines/03_reporting/__init__.py:
from .pipeline import create_pipeline
In ~/pipelines/03_reporting/nodes.py:
import numpy as np
import pandas as pd
from typing import Dict, Tuple
import matplotlib.py:plot as plt
from optbinning import Scorecard
from optbinning.scorecard import ScorecardMonitoring
from optbinning.scorecard import plot_auc_roc, plot_ks
from optbinning import BinningProcess
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import roc_curve, roc_auc_score
import seaborn as sns
from mlxtend.feature_selection import SequentialFeatureSelector as SFS
from mlxtend.plotting import plot_sequential_feature_selection as plot_sfs
import matplotlib.ticker as mtick
import plotly.graph_objects as go
def get_scored_sample(df:pd.DataFrame, X_train, X_test, scorecard, metric:str, metric_missing:str):
"""
scoring function (binning included)
"""
df['type_sample'] = np.NaN
df.loc[df.index.intersection(X_train.index), 'type_sample'] = 'train'
df.loc[df.index.intersection(X_test.index), 'type_sample'] = 'test'
df[[x + '_woe' for x in scorecard.binning_process_.variable_names]] = scorecard.binning_process_.transform(df, metric = metric,
metric_missing = metric_missing
)
df['prediction'] = scorecard.predict_proba(df)[:,1]
return df
def get_result_to_kaggle(df:pd.DataFrame, scorecard):
"""
result: upload to Kaggle
"""
df['Probability'] = scorecard.predict_proba(df)[:,1]
return df.reset_index()[['Id', 'Probability']]
def get_table_binning(bp):
"""get binning long_list
"""
t=[]
for feature in list(bp.variable_names):
t.append(bp.get_binned_variable(feature).binning_table.build(show_digits=4))
table=pd.concat(t , keys=list(bp.variable_names))
return table
def get_table_features_quality(bp):
"""get features quality (IV, Gini)
"""
table_info_from_binning = bp.summary()
return table_info_from_binning
def get_table_features_quality_short_list(bp_short, df_dev_scored, target):
"""get features quality (IV, Gini)
"""
table_info_from_binning = bp_short.summary()
table_gini_train_test = pd.DataFrame([])
for type_sampe in ['train', 'test']:
table_info_gini = {}
for feature_woe in [ x for x in df_dev_scored.columns if 'woe' in x]:
table_info_gini[feature_woe] = (2*roc_auc_score(df_dev_scored[df_dev_scored.type_sample == type_sampe][target],
df_dev_scored[df_dev_scored.type_sample == type_sampe][feature_woe]) - 1) * (-1)
table_gini_train_test = pd.concat([table_gini_train_test, pd.DataFrame.from_dict(table_info_gini, orient='index', columns=['gini_{}'.format(type_sampe)])], axis=1)
table_gini_train_test = table_gini_train_test.reset_index().rename(columns = {'index':'name'})
table_gini_train_test['name'] = table_gini_train_test['name'].str.replace('_woe','', regex=True)
table = table_info_from_binning.merge(table_gini_train_test, how='left', on = 'name')
return table
def get_correlation_table_short_list(X_train_woe):
"""get corr table
"""
table_corr = X_train_woe.corr().round(2)
fig = plt.figure(figsize=(15, 10))
fig = sns.heatmap(table_corr, annot=True).get_figure()
return table_corr, fig
def get_table_model_scorecard(model):
"""
get model scorecard info
"""
table = model.table(style='detailed')
return table
def get_table_model_scorecard_feature_coefs(model):
"""
get model scorecard info
"""
weights = {}
weights['const'] = np.round(model.estimator_.intercept_[0],4)
weights['features'] = dict(zip(model.estimator_.feature_names_in_, np.round(model.estimator_.coef_[0],4)))
return weights
def get_plot_feature_selection(sfs):
"""
plot quality depending on the features
"""
plot = plot_sfs(sfs.get_metric_dict(),figsize=(10, 8))[0]
return plot
class ScorecardMonitoringMod(ScorecardMonitoring):
def psi_plot(self, savefig=None):
"""Plot Population Stability Index (PSI).
Parameters
----------
return plot
"""
self._check_is_fitted()
fig, ax1 = plt.subplots()
n_bins = len(self._n_records_a)
indices = np.arange(n_bins)
width = np.min(np.diff(indices))/3
p_records_a = self._n_records_a / self._n_records_a.sum() * 100.0
p_records_e = self._n_records_e / self._n_records_e.sum() * 100.0
p1 = ax1.bar(indices-width, p_records_a, width, color='tab:red',
label="Records Actual", alpha=0.75)
p2 = ax1.bar(indices, p_records_e, width, color='tab:blue',
label="Records Expected", alpha=0.75)
handles = [p1[0], p2[0]]
labels = ['Actual', 'Expected']
ax1.set_xlabel("Bin ID", fontsize=12)
ax1.set_ylabel("Population distribution", fontsize=13)
ax1.yaxis.set_major_formatter(mtick.PercentFormatter())
ax2 = ax1.twinx()
if self._target_dtype == "binary":
metric_label = "Event rate"
elif self._target_dtype == "continuous":
metric_label = "Mean"
ax2.plot(indices, self._metric_a, linestyle="solid", marker="o",
color='tab:red')
ax2.plot(indices, self._metric_e, linestyle="solid", marker="o",
color='tab:blue')
ax2.set_ylabel(metric_label, fontsize=13)
ax2.xaxis.set_major_locator(mtick.MultipleLocator(1))
ax2.set_xlim(-width * 2, n_bins - width * 2)
plt.legend(handles, labels, loc="upper center",
bbox_to_anchor=(0.5, -0.2), ncol=2, fontsize=12)
plt.tight_layout()
if savefig is None:
#plt.show()
pass
else:
if not isinstance(savefig, str):
raise TypeError("savefig must be a string path; got {}."
.format(savefig))
plt.savefig(savefig)
plt.close()
return fig
def get_statistics_monitoring(scorecard, psi_method:str, psi_n_bins:int, target,
X_actual, y_actual, X_expected, y_expected, inplace_y_actual):
"""
file with scorecard metrics and statics (Gini, KS, PSI, etc.)
inplace_y_actual: value
"""
scorecard_monitoring = ScorecardMonitoringMod(scorecard=scorecard, psi_method=psi_method,
psi_n_bins = psi_n_bins, verbose=False)
if (type(y_actual) == str) and (target != None):
y_actual = X_actual[target].fillna(inplace_y_actual)
else:
pass
if (type(y_expected) == str) and (target != None):
y_expected = X_expected[target]
else:
pass
scorecard_monitoring.fit(X_actual, y_actual, X_expected, y_expected)
return scorecard_monitoring
def get_tables_psi_features_detailed(scorecard_monitoring):
"""
get table detailed
"""
return scorecard_monitoring.psi_variable_table(style="detailed")
def get_tables_psi_features_summary(scorecard_monitoring):
"""
get table summary
"""
return scorecard_monitoring.psi_variable_table()
def get_table_psi_scorecard(scorecard_monitoring):
"""
get table psi scorecard
"""
return scorecard_monitoring.psi_table()
def get_plot_psi_scorecard(scorecard_monitoring):
"""
get plot psi_scorecard
"""
return scorecard_monitoring.psi_plot()
def get_table_statistical_tests(scorecard_monitoring):
""" Null hypothesis: actual == expected
Chi-square test - binary target
"""
return scorecard_monitoring.tests_table()
def _calc_plot_auc_roc_mod(y, y_pred, title=None, xlabel=None, ylabel=None,
fname=None, **kwargs):
"""Plot Area Under the Receiver Operating Characteristic Curve (AUC ROC).
"""
fpr, tpr, _ = roc_curve(y, y_pred)
auc_roc = roc_auc_score(y, y_pred)
# Define the plot settings
if title is None:
title = "ROC curve"
if xlabel is None:
xlabel = "False Positive Rate"
if ylabel is None:
ylabel = "True Positive Rate"
plt.plot(fpr, fpr, linestyle="--", color="k", label="Random Model")
plt.plot(fpr, tpr, label="Model (AUC: {:.5f})".format(auc_roc))
plt.title(title, fontdict={"fontsize": 14})
plt.xlabel(xlabel, fontdict={"fontsize": 12})
plt.ylabel(ylabel, fontdict={"fontsize": 12})
plt.legend(loc='lower right')
return plt.gcf()
def get_plot_roc_auc_train_test(scorecard, X_train, y_train, X_test, y_test):
""" plot train test roc_auc plots
"""
train_roc_auc = round(roc_auc_score(y_train, pd.Series(scorecard.predict_proba(X_train)[:, 1])),2)
test_roc_auc = round(roc_auc_score(y_test, pd.Series(scorecard.predict_proba(X_test)[:, 1])),2)
g1 = _calc_plot_auc_roc_mod(y_train, pd.Series(scorecard.predict_proba(X_train)[:, 1]))
g1 = _calc_plot_auc_roc_mod(y_test, scorecard.predict_proba(X_test)[:, 1], title='ROC Train {} Test {}'.format(train_roc_auc, test_roc_auc))
return g1
def get_plot_ks_train_test(scorecard, X_train, y_train, X_test, y_test ):
plt.figure(figsize=(6.4, 4.8))
plt.subplot(121)
plot_ks(y_train, scorecard.predict_proba(X_train)[:, 1], title='KS Train')
plt.subplot(122)
plot_ks(y_test, scorecard.predict_proba(X_test)[:, 1], title='KS Test')
#plt.suptitle('KS')
g1 = plt.gcf()
return g1
def get_plot_coefs_scorecard_model (scorecard_monitoring):
"""
get plot of coefs scorecard
"""
#plt.figure(figsize=(3.2, 2.4))
reg_coef = pd.DataFrame((zip(scorecard_monitoring.scorecard.estimator_.feature_names_in_, scorecard_monitoring.scorecard.estimator_.coef_[0])), columns=['name', 'reg_coef'])
g1 = sns.barplot(round(reg_coef,2), x="reg_coef", y="name", errorbar=None)
g1.bar_label(g1.containers[0], fontsize=8)
return g1.get_figure()
def get_general_statistics_report(scorecard_monitoring):
scorecard_monitoring.system_stability_report()
In ~/pipelines/03_reporting/pipeline.py:
from kedro.pipeline import Pipeline, node, pipeline
from .nodes import *
def create_pipeline(**kwargs) -> Pipeline:
return pipeline(
[
node(
func = get_scored_sample,
inputs = dict( df = "04_feature_development",
X_train = "05_model_input_X_train",
X_test = "05_model_input_X_test",
scorecard = "06_models_scorecard",
metric="params:binning_parameters.metric",
metric_missing="params:binning_parameters.metric_missing",
),
outputs = "07_model_output_development_scored",
name = "get_scored_sample_development"
),
node(
func = get_scored_sample,
inputs = dict( df = "04_feature_validation",
X_train = "05_model_input_X_train",
X_test = "05_model_input_X_test",
scorecard = "06_models_scorecard",
metric="params:binning_parameters.metric",
metric_missing="params:binning_parameters.metric_missing",
),
outputs = "07_model_output_validation_scored",
name = "get_scored_sample_validation"
),
node(
func = get_result_to_kaggle,
inputs = dict( df = "04_feature_validation",
scorecard = "06_models_scorecard"
),
outputs = "07_model_output_sample_to_kaggle",
name = "get_result_to_kaggle"
),
node(
func = get_table_binning,
inputs = dict( bp = "06_models_binning"
),
outputs = "08_reporting_variables_binning_table",
name = "get_table_binning_long_list",
tags="info_table"
),
node(
func = get_table_binning,
inputs = dict( bp = "06_models_binning_short_list"
),
outputs = "08_reporting_variables_binning_table_short_list",
name = "get_table_binning_short_list",
tags="info_table"
),
node(
func = get_table_features_quality,
inputs = dict( bp = "06_models_binning"
),
outputs = "08_reporting_variables_summary_table",
name = "get_table_features_quality_long_list",
tags="info_table"
),
node(
func = get_table_features_quality_short_list,
inputs = dict( bp_short = "06_models_binning_short_list",
df_dev_scored = "07_model_output_development_scored",
target = "params:cols.col_target"
),
outputs = "08_reporting_variables_summary_table_short_list",
name = "get_table_features_quality_short_list",
tags="info_table"
),
node(
func = get_correlation_table_short_list,
inputs = dict( X_train_woe = "05_model_input_X_train_woe_short_list",
),
outputs = ["08_reporting_model_feature_correlation", "08_reporting_model_feature_correlation_plot" ],
name = "get_correlation_table_short_list",
tags="info_plot",
),
node(
func = get_table_model_scorecard,
inputs = dict( model = "06_models_scorecard"
),
outputs = "08_reporting_model_scorecard_table",
name = "get_table_model_scorecard",
tags="info_table"
),
node(
func = get_table_model_scorecard_feature_coefs,
inputs = dict( model = "06_models_scorecard"
),
outputs = "08_reporting_model_scorecard_feature_coefs",
name = "get_table_model_scorecard_feature_coefs",
tags="info_table"
),
node(
func = get_plot_feature_selection,
inputs = dict( sfs = "06_models_feature_selection"
),
outputs = "08_reporting_model_scorecard_feature_selection_plot",
name = "get_plot_feature_selection",
tags="info_plot"
),
node(
func = get_statistics_monitoring,
inputs = dict( scorecard = "06_models_scorecard",
psi_method = "params:monitoring_parameters.psi_method",
psi_n_bins = "params:monitoring_parameters.psi_n_bins",
target = "params:cols.col_target",
X_actual = "05_model_input_X_test",
y_actual = "05_model_input_y_test",
X_expected = "05_model_input_X_train",
y_expected = "05_model_input_y_train",
inplace_y_actual = "params:monitoring_parameters.inplace_y_actual"
),
outputs = "08_reporting_statistics_monitoring_train_test",
name = "get_statistics_monitoring_train_test"
),
node(
func = get_statistics_monitoring,
inputs = dict( scorecard = "06_models_scorecard",
psi_method = "params:monitoring_parameters.psi_method",
psi_n_bins = "params:monitoring_parameters.psi_n_bins",
target = "params:cols.col_target",
X_actual = "04_feature_validation",
y_actual = "params:monitoring_parameters.none_type",
X_expected = "04_feature_development",
y_expected = "params:monitoring_parameters.none_type",
inplace_y_actual = "params:monitoring_parameters.inplace_y_actual"
),
outputs = "08_reporting_statistics_monitoring_dev_valid",
name = "get_statistics_monitoring_dev_valid"
),
node(
func = get_tables_psi_features_detailed,
inputs = dict( scorecard_monitoring = "08_reporting_statistics_monitoring_train_test"
),
outputs = "08_reporting_model_scorecard_features_psi_detailed_train_test",
name = "get_tables_psi_features_detailed_train_test",
tags="info_table"
),
node(
func = get_tables_psi_features_detailed,
inputs = dict( scorecard_monitoring = "08_reporting_statistics_monitoring_dev_valid"
),
outputs = "08_reporting_model_scorecard_features_psi_detailed_dev_valid",
name = "get_tables_psi_features_detailed_dev_valid",
tags="info_table"
),
node(
func = get_tables_psi_features_summary,
inputs = dict( scorecard_monitoring = "08_reporting_statistics_monitoring_train_test"
),
outputs = "08_reporting_model_scorecard_features_psi_summary_train_test",
name = "get_tables_psi_features_summary_train_test",
tags="info_table"
),
node(
func = get_tables_psi_features_summary,
inputs = dict( scorecard_monitoring = "08_reporting_statistics_monitoring_dev_valid"
),
outputs = "08_reporting_model_scorecard_features_psi_summary_dev_valid",
name = "get_tables_psi_features_summary_dev_valid",
tags="info_table"
),
node(
func = get_table_psi_scorecard,
inputs = dict( scorecard_monitoring = "08_reporting_statistics_monitoring_train_test"
),
outputs = "08_reporting_model_scorecard_psi_summary_train_test",
name = "get_table_psi_scorecard_train_test",
tags="info_table"
),
node(
func = get_table_psi_scorecard,
inputs = dict( scorecard_monitoring = "08_reporting_statistics_monitoring_dev_valid"
),
outputs = "08_reporting_model_scorecard_psi_summary_dev_valid",
name = "get_table_psi_scorecard_dev_valid",
tags="info_table"
),
node(
func = get_plot_psi_scorecard,
inputs = dict( scorecard_monitoring = "08_reporting_statistics_monitoring_train_test"
),
outputs = "08_reporting_model_scorecard_psi_summary_train_test_plot",
name = "get_plot_psi_scorecard_train_test",
tags="info_plot"
),
node(
func = get_plot_psi_scorecard,
inputs = dict( scorecard_monitoring = "08_reporting_statistics_monitoring_dev_valid"
),
outputs = "08_reporting_model_scorecard_psi_summary_dev_valid_plot",
name = "get_plot_psi_scorecard_dev_valid",
tags="info_plot"
),
node(
func = get_table_statistical_tests,
inputs = dict( scorecard_monitoring = "08_reporting_statistics_monitoring_train_test"
),
outputs = "08_reporting_model_scorecard_statistical_tests",
name = "get_table_statistical_tests",
tags="info_table"
),
node(
func = get_plot_roc_auc_train_test,
inputs = dict( scorecard = "06_models_scorecard",
X_train = "05_model_input_X_train",
y_train = "05_model_input_y_train",
X_test = "05_model_input_X_test",
y_test = "05_model_input_y_test"
),
outputs = "08_reporting_model_scorecard_roc_auc_train_test_plot",
name = "get_plot_roc_auc_train_test",
tags="info_plot"
),
node(
func = get_plot_ks_train_test,
inputs = dict( scorecard = "06_models_scorecard",
X_train = "05_model_input_X_train",
y_train = "05_model_input_y_train",
X_test = "05_model_input_X_test",
y_test = "05_model_input_y_test"
),
outputs = "08_reporting_model_scorecard_ks_plot",
name = "get_plot_ks_train_test",
tags="info_plot"
),
node(
func = get_plot_coefs_scorecard_model,
inputs = dict( scorecard_monitoring = "08_reporting_statistics_monitoring_train_test",
),
outputs = "08_reporting_model_scorecard_coefs_plot" ,
name = "get_plot_coefs_scorecard_model",
tags="info_plot"
),
node(
func = get_general_statistics_report,
inputs = dict( scorecard_monitoring = "08_reporting_statistics_monitoring_train_test",
),
outputs = None ,
name = "get_general_statistics_report"
),
]
)
Run the Pipelines
To run all the pipelines in /docker_disk/risk-model, open Terminal and run the following command:
kedro run
As an option it is also possible to run partially only one of them:
kedro run --pipeline 01_preprocessing
Or even select the node to run from:
kedro run --pipeline 01_preprocessing --from-nodes get_index_development
After we run all the pipelines, we can check the connections between nodes, pipelines and outputs in one place - in Kedro Viz without checking output folders and files.
Visualization: Kedro Viz
To visualize the pipelines in a Kedro project by showing data, nodes, and the connections between them:
In /docker_disk/risk-model open terminal and run command:
kedro viz run --host=0.0.0.0 --port=4141 --autoreload
Then open the browser and go to http://0.0.0.0:4141/.
Here you can check the total model visualization:
From that window it is possible to access every part of code that was used in the project, check inputs and outputs, parameters, and results.
For better result visualisation it is recommended to use tags in node() in pipeline.py:.
Here we have 2 tags: “info_plot” and “info_table”, so we can get fast access to outputs, filtering by them.
Result
File sampleEntry.csv from ~/data/07_model_output we will upload to Kaggle and submit our task.
In the directory: ~/data/08_reporting we collect all the files that could be used in model documentation:
This template represents the basic skeleton for any risk modeling exercise: data pre-processing, model training, and reporting stages. Remember to modify this template to suit your needs by perhaps incorporating business logic for more advanced feature engineering or different types of models.
Conclusion
Python risk analytics provides some really powerful tools that any financial organization could use in order to understand, manage, and mitigate risk. Using the templates provided along with best practices, we will show how this will enable model development efficiency and consistency, and will rapidly adapt to changing market conditions.
As long as the financial world keeps on changing, so will the techniques and technologies we use in risk analytics. Knowledge of current developments in data science, machine learning, and the Python ecosystem will be fundamental to the risk professional who intends to bring additional value to his organization.
Remember, while templates and tools are essential, the vital ingredients to successful risk analytics include a focus on these technical components in concert with domain expertise, critical thinking, and deep business context. As we continue to push the boundaries of what is possible in risk modeling, let us not forget what we are trying to achieve: make informed decisions that protect and grow our financial institutions.
Subscribe to my newsletter
Read articles from Alexey Khoroshilov directly inside your inbox. Subscribe to the newsletter, and don't miss out.
Written by