cursus.steps.scripts.risk_table_mapping

Risk Table Mapping Processing Script

This script creates and applies risk tables for categorical features based on target variable correlation, and handles missing value imputation for numeric features. It supports both training mode (fit and transform) and inference mode (transform only).

collect_risk_table_statistics_pass1(all_shards, signature_columns, cat_field_list, label_name, smooth_factor, count_threshold, max_unique_threshold, log_func)

Pass 1: Collect risk table statistics from training shards.

Memory-efficient incremental crosstab aggregation: - For each categorical field:

• Accumulate counts per (category, label) combination

• Compute risk scores after all shards processed

This function processes shards sequentially to build risk tables incrementally, avoiding loading the entire dataset into memory.

Parameters:

• all_shards (List[Path]) – List of training shard paths

• signature_columns (List[str] | None) – Optional column names for CSV/TSV

• cat_field_list (List[str]) – List of categorical field names

• label_name (str) – Target variable name

• smooth_factor (float) – Smoothing factor for risk calculation

• count_threshold (int) – Minimum count threshold

• max_unique_threshold (int) – Max unique values for validation

• log_func (Callable) – Logging function

Returns:

Dictionary of risk tables (same format as OfflineBinning.risk_tables) Format: {field: {varName, type, mode, default_bin, bins: {category: risk}}}

Return type:

Dict[str, Any]

process_shard_end_to_end_risk_mapping(args)

Process single shard: read → apply risk tables → write.

Stateless per-shard processing using global risk_tables from Pass 1. Preserves 1:1 shard mapping (input shard number → output shard number).

This function is designed to be called in parallel via multiprocessing.Pool. Each worker processes one shard completely independently.

Parameters:

• args (tuple) – Tuple of (shard_path, shard_num, global_context, output_base, signature_columns, output_format)

• contain (global_context must)

• "risk_tables" (-) – Dictionary of risk tables from Pass 1

• "split_name" (-) – Which split this shard belongs to (“train”, “val”, “test”, etc.)

Returns:

Statistics dict with row count for this split Format: {“train”: 1000} or {“val”: 200} or {“validation”: 500}

Return type:

Dict[str, int]

Example

Input: train/part-00042.csv (1000 rows) Output: train/part-00042.csv (1000 rows, risk-mapped) Return: {“train”: 1000}

process_streaming_mode_risk_mapping(input_dir, output_dir, signature_columns, job_type, hyperparams, environ_vars, max_workers, model_artifacts_input_dir=None, model_artifacts_output_dir=None, log_func=None)

Streaming mode for risk table mapping with train/val/test subdirectories.

Two-pass architecture: - Pass 1 (Sequential): Collect risk table statistics from training shards only - Pass 2 (Parallel): Apply risk tables per split in parallel

Auto-detects output format from input shards (mirrors batch mode behavior).

Input structure (training mode):

input_dir/

train/part-00000.csv, part-00001.csv, … val/part-00000.csv, part-00001.csv, … test/part-00000.csv, part-00001.csv, …

Output structure (training mode):

output_dir/

train/part-00000.csv, part-00001.csv, … (risk-mapped, same format) val/part-00000.csv, part-00001.csv, … (risk-mapped, same format) test/part-00000.csv, part-00001.csv, … (risk-mapped, same format)

Parameters:

• input_dir (str) – Base input directory

• output_dir (str) – Base output directory

• signature_columns (List[str] | None) – Optional column names for CSV/TSV

• job_type (str) – ‘training’, ‘validation’, ‘testing’, ‘calibration’

• hyperparams (Dict[str, Any]) – Hyperparameters dict (contains cat_field_list, etc.)

• environ_vars (Dict[str, str]) – Environment variables dict

• max_workers (int | None) – Number of parallel workers

• model_artifacts_input_dir (str | None) – Input model artifacts directory

• model_artifacts_output_dir (str | None) – Output model artifacts directory

• log_func (Callable | None) – Logging function (defaults to print)

Returns:

Dictionary with total row counts per split Format: {“train”: 100000, “val”: 20000, “test”: 30000}

Return type:

Dict[str, int]

find_split_shards(input_dir, split_name, log_func)

Find all input shards in a specific split subdirectory.

Used when input data is organized as: input_dir/

train/part-00000.csv, part-00001.csv, … val/part-00000.csv, part-00001.csv, … test/part-00000.csv, part-00001.csv, …

Parameters:

• input_dir (str) – Base input directory

• split_name (str) – Split subdirectory name (“train”, “val”, “test”, etc.)

• log_func (Callable) – Logging function

Returns:

Sorted list of shard paths from the split subdirectory

Raises:

RuntimeError – If split subdirectory or shards not found

Return type:

List[Path]

extract_shard_number(shard_path)

Extract shard number from filename like part-00042.csv.

Handles various formats: - part-00042.csv → 42 - part-00042.csv.gz → 42 - part-00042.parquet → 42 - part-00042.snappy.parquet → 42

Parameters:

shard_path (Path) – Path to shard file

Returns:

Integer shard number

Raises:

ValueError – If shard number cannot be extracted

Return type:

int

Example

>>> extract_shard_number(Path("part-00042.csv"))
42
>>> extract_shard_number(Path("part-00001.csv.gz"))
1

write_shard_file(df, output_path, output_format)

Write a DataFrame to a shard file in the specified format.

Creates parent directories if needed.

Parameters:

• df (DataFrame) – DataFrame to write

• output_path (Path) – Full path for output file (including filename)

• output_format (str) – Format to write (‘csv’, ‘tsv’, or ‘parquet’)

Raises:

ValueError – If output_format is not supported

detect_shard_format(shard_path)

Auto-detect output format from input shard filename.

Mirrors batch mode’s format preservation behavior.

Parameters:

shard_path (Path) – Path to a shard file

Returns:

‘csv’, ‘tsv’, or ‘parquet’

Return type:

Format string

Example

>>> detect_shard_format(Path("part-00001.csv"))
'csv'
>>> detect_shard_format(Path("part-00001.parquet"))
'parquet'

load_json_config(config_path)

Loads a JSON configuration file.

Parameters:

config_path (str) – Path to the JSON configuration file

Returns:

Dict containing the loaded JSON configuration

Raises:

• FileNotFoundError – If the configuration file doesn’t exist

• PermissionError – If the configuration file can’t be accessed due to permissions

• json.JSONDecodeError – If the configuration file contains invalid JSON

• Exception – For other unexpected errors

Return type:

Dict[str, Any]

validate_categorical_fields(df, cat_field_list, max_unique_threshold=100)

Validate that fields in cat_field_list are suitable for risk mapping.

Parameters:

• df (DataFrame) – DataFrame containing the data

• cat_field_list (List[str]) – List of categorical field names

• max_unique_threshold (int) – Maximum unique values threshold for suitability (default: 100)

Returns:

List of valid fields for risk mapping

Return type:

List[str]

class OfflineBinning(cat_field_list, target_field)

Bases: object

A class to create risk tables for categorical features.

Risk tables map categorical values to numerical risk scores based on their correlation with the target variable.

fit(df, smooth_factor=0, count_threshold=0)

Fits the risk tables based on the provided dataframe.

transform(df)

Transforms the dataframe using the fitted risk tables.

load_risk_tables(risk_tables)

Loads pre-existing risk tables.

load_split_data(job_type, input_dir)

Load data according to job_type with automatic format detection.

For ‘training’: Loads data from train, test, and val subdirectories For others: Loads single job_type split

Returns:

Dictionary with DataFrames and detected format stored in ‘_format’ key

Return type:

Dict[str, DataFrame]

save_output_data(job_type, output_dir, data_dict)

Save processed data according to job_type, preserving input format.

For ‘training’: Saves data to train, test, and val subdirectories For others: Saves to single job_type directory

process_data(data_dict, cat_field_list, label_name, job_type, risk_tables_dict=None, smooth_factor=0.01, count_threshold=5, max_unique_threshold=100)

Core data processing logic for risk table mapping.

Parameters:

• data_dict (Dict[str, DataFrame]) – Dictionary of dataframes keyed by split name

• cat_field_list (List[str]) – List of categorical field names

• label_name (str) – Target column name

• job_type (str) – Type of job (training, validation, testing, calibration)

• risk_tables_dict (Dict | None) – Pre-existing risk tables (for non-training jobs)

• smooth_factor (float) – Smoothing factor for risk tables (default: 0.01)

• count_threshold (int) – Minimum count threshold (default: 5)

• max_unique_threshold (int) – Maximum unique values threshold (default: 100)

Returns:

• Dictionary of transformed dataframes

• OfflineBinning instance with fitted risk tables

Return type:

Tuple containing

save_artifacts(binner, hyperparams, output_path)

Save risk table artifacts to the specified output path.

Output format matches XGBoost training’s risk_table_map.pkl format: A dictionary mapping variable names to their risk table dictionaries. Each risk table dict contains ‘bins’ and ‘default_bin’ keys.

Parameters:

• binner (OfflineBinning) – OfflineBinning instance with fitted risk tables

• hyperparams (Dict[str, Any]) – Hyperparameter dictionary

• output_path (Path) – Path to save artifacts to

copy_existing_artifacts(src_dir, dst_dir)

Copy all existing model artifacts from previous processing steps.

This enables the parameter accumulator pattern where each step: 1. Copies artifacts from previous steps 2. Adds its own artifacts 3. Passes all artifacts to the next step

Parameters:

• src_dir (str) – Source directory containing existing artifacts

• dst_dir (str) – Destination directory to copy artifacts to

load_risk_tables(risk_table_path)

Load risk tables from a pickle file.

Parameters:

risk_table_path (Path) – Path to the risk tables file

Returns:

Dictionary of risk tables

Return type:

Dict

internal_main(job_type, input_dir, output_dir, hyperparams, environ_vars=None, model_artifacts_input_dir=None, model_artifacts_output_dir=None, load_data_func=<function load_split_data>, save_data_func=<function save_output_data>)

Main logic for risk table mapping with dual-mode support (batch/streaming).

Parameters:

• job_type (str) – Type of job (training, validation, testing, calibration)

• input_dir (str) – Input directory for data

• output_dir (str) – Output directory for processed data

• hyperparams (Dict[str, Any]) – Hyperparameters dictionary loaded from hyperparameters.json

• environ_vars (Dict[str, str] | None) – Environment variables dictionary

• model_artifacts_input_dir (str | None) – Directory containing model artifacts from previous steps

• model_artifacts_output_dir (str | None) – Directory to save model artifacts for next steps

• load_data_func (Callable) – Function to load data (for dependency injection in tests)

• save_data_func (Callable) – Function to save data (for dependency injection in tests)

Returns:

• Dictionary of transformed dataframes

• OfflineBinning instance with fitted risk tables

Return type:

Tuple containing

main(input_paths, output_paths, environ_vars, job_args=None)

Standardized main entry point for risk table mapping script.

Parameters:

• input_paths (Dict[str, str]) – Dictionary of input paths with logical names - “data_input”: Input data directory - “model_artifacts_input”: Model artifacts from previous steps (standardized) - “hyperparameters_s3_uri”: Path to hyperparameters (optional)

• output_paths (Dict[str, str]) – Dictionary of output paths with logical names - “processed_data”: Output directory for processed data - “model_artifacts_output”: Model artifacts output for next steps (standardized)

• environ_vars (Dict[str, str]) – Dictionary of environment variables

• job_args (Namespace | None) – Command line arguments (optional)

Returns:

• Dictionary of transformed dataframes

• OfflineBinning instance with fitted risk tables

Return type:

Tuple containing