cursus.steps.scripts.model_metrics_computation

check_call(cmd, *a, **k)

install_packages_from_public_pypi(packages)

Install packages from standard public PyPI.

Parameters:

packages (list) – List of package specifications (e.g., [“pandas==1.5.0”, “numpy”])

install_packages_from_secure_pypi(packages)

Install packages from secure CodeArtifact PyPI.

Parameters:

packages (list) – List of package specifications (e.g., [“pandas==1.5.0”, “numpy”])

install_packages(packages, use_secure=False)

Install packages from PyPI source based on configuration.

This is the main installation function that delegates to either public or secure PyPI based on the USE_SECURE_PYPI environment variable.

Parameters:

• packages (list) – List of package specifications (e.g., [“pandas==1.5.0”, “numpy”])

• use_secure (bool) – If True, use secure CodeArtifact PyPI; if False, use public PyPI. Defaults to USE_SECURE_PYPI environment variable.

Environment Variables:

USE_SECURE_PYPI: Set to “true” to use secure PyPI, “false” for public PyPI

Example

# Install from public PyPI (default) install_packages([“pandas==1.5.0”, “numpy”])

# Install from secure PyPI os.environ[“USE_SECURE_PYPI”] = “true” install_packages([“pandas==1.5.0”, “numpy”])

detect_and_load_predictions(input_dir, preferred_format=None)

Auto-detect and load predictions file in CSV, TSV, Parquet, or JSON format. Supports intelligent format detection and graceful fallback. Aligned with format preservation pattern used across cursus framework.

parse_score_fields(environ_vars)

Parse SCORE_FIELD or SCORE_FIELDS from environment variables. Pattern matching model_calibration.py

Priority: 1. SCORE_FIELDS (multi-task) - comma-separated list 2. SCORE_FIELD (single-task) - backward compatible 3. Default: “prob_class_1”

Returns:

List of score field names

Return type:

List[str]

parse_previous_score_fields(environ_vars, score_fields)

Parse PREVIOUS_SCORE_FIELDS or PREVIOUS_SCORE_FIELD from environment variables.

Priority: 1. PREVIOUS_SCORE_FIELDS (multi-task) - comma-separated list 2. PREVIOUS_SCORE_FIELD (single-task) - backward compatible 3. Empty list if not in comparison mode

Parameters:

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

• score_fields (List[str]) – List of score field names (for validation)

Returns:

List of previous score field names (empty if no comparison)

Return type:

List[str]

parse_task_label_fields(environ_vars, score_fields)

Parse TASK_LABEL_NAMES or infer from score_fields. Pattern matching model_calibration.py

Priority: 1. Explicit TASK_LABEL_NAMES (preferred for multi-task) 2. Infer from score field names (_prob → removes suffix) 3. Single LABEL_FIELD (backward compatibility)

Parameters:

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

• score_fields (List[str]) – List of score field names

Returns:

List of label field names, one per score field

Return type:

List[str]

validate_prediction_columns(df, score_fields, label_fields, id_field)

Validate that all required columns exist in DataFrame.

Parameters:

• df (DataFrame) – Input DataFrame

• score_fields (List[str]) – List of score column names

• label_fields (List[str]) – List of label column names

• id_field (str) – ID column name

Returns:

Validation report dictionary

Raises:

ValueError – If critical columns are missing

Return type:

Dict[str, Any]

validate_prediction_data(df, id_field, label_field, amount_field=None)

Validate prediction data schema and return validation report. Legacy function for backward compatibility with single-task.

compute_standard_metrics(y_true, y_prob, is_binary=True)

Compute comprehensive standard ML metrics matching xgboost_model_eval.py. Supports both binary and multiclass classification with full metric coverage.

calculate_count_recall(scores, labels, amounts, cutoff=0.1)

Calculate count recall - imported from evaluation.py

calculate_dollar_recall(scores, labels, amounts, fpr=0.1)

Calculate dollar recall - imported from evaluation.py

compute_domain_metrics(scores, labels, amounts=None, compute_dollar_recall=True, compute_count_recall=True, dollar_recall_fpr=0.1, count_recall_cutoff=0.1)

Compute domain-specific metrics including dollar and count recall. Integrates functions from evaluation.py for business impact analysis.

plot_and_save_roc_curve(y_true, y_score, output_dir, prefix='')

Plot ROC curve and save as JPG (exact match with xgboost_model_eval.py).

plot_and_save_pr_curve(y_true, y_score, output_dir, prefix='')

Plot Precision-Recall curve and save as JPG (exact match with xgboost_model_eval.py).

generate_performance_visualizations(y_true, y_prob, metrics, output_dir, is_binary=True)

Generate comprehensive performance visualizations matching xgboost_model_eval.py. Returns dictionary of plot file paths.

generate_performance_insights(metrics)

Generate actionable performance insights based on metrics.

generate_recommendations(metrics)

Generate actionable recommendations based on performance analysis.

generate_comprehensive_report(standard_metrics, domain_metrics, plot_paths, validation_report, output_dir)

Generate comprehensive metrics report with insights and recommendations.

compute_comparison_metrics(y_true, y_new_score, y_prev_score, is_binary=True)

Compute comparison metrics between new model and previous model scores. Identical to xgboost_model_eval.py implementation.

perform_statistical_tests(y_true, y_new_score, y_prev_score, is_binary=True)

Perform statistical significance tests comparing model performances. Identical to xgboost_model_eval.py implementation.

plot_comparison_roc_curves(y_true, y_new_score, y_prev_score, output_dir)

Plot side-by-side ROC curves comparing new and previous models. Identical to xgboost_model_eval.py implementation.

plot_comparison_pr_curves(y_true, y_new_score, y_prev_score, output_dir)

Plot side-by-side Precision-Recall curves comparing new and previous models. Identical to xgboost_model_eval.py implementation.

plot_score_scatter(y_new_score, y_prev_score, y_true, output_dir)

Plot scatter plot of new vs previous model scores, colored by true labels. Identical to xgboost_model_eval.py implementation.

plot_score_distributions(y_new_score, y_prev_score, y_true, output_dir)

Plot score distributions for both models, separated by true labels. Identical to xgboost_model_eval.py implementation.

generate_text_summary(json_report)

Generate a human-readable text summary from the JSON report.

log_metrics_summary(metrics, is_binary=True)

Log a nicely formatted summary of metrics for easy visibility in logs.

save_metrics(metrics, output_metrics_dir)

Save computed metrics as a JSON file (matching xgboost_model_eval.py).

compute_multitask_metrics(df, score_fields, label_fields, amounts=None, environ_vars=None)

Compute per-task and aggregate metrics for multi-task predictions. Pattern matching lightgbmmt_model_eval.py

Parameters:

• df (DataFrame) – DataFrame with predictions

• score_fields (List[str]) – List of score column names

• label_fields (List[str]) – List of label column names

• amounts (ndarray) – Optional array of transaction amounts

• environ_vars (Dict[str, str]) – Environment variables for domain metrics

Returns:

Dictionary with per-task and aggregate metrics

Return type:

Dict[str, Any]

compute_multitask_domain_metrics(df, score_fields, label_fields, amounts, environ_vars)

Compute domain-specific metrics (dollar/count recall) for each task.

Parameters:

• df (DataFrame) – DataFrame with predictions

• score_fields (List[str]) – List of score column names

• label_fields (List[str]) – List of label column names

• amounts (ndarray) – Array of transaction amounts

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

Returns:

Dictionary with per-task domain metrics

Return type:

Dict[str, Any]

generate_multitask_visualizations(df, score_fields, label_fields, output_dir)

Generate per-task ROC and PR curves. Pattern matching lightgbmmt_model_eval.py

Parameters:

• df (DataFrame) – DataFrame with predictions

• score_fields (List[str]) – List of score column names

• label_fields (List[str]) – List of label column names

• output_dir (str) – Output directory for plots

Returns:

Dictionary of plot file paths

Return type:

Dict[str, str]

compute_multitask_comparison_metrics(df, score_fields, label_fields, prev_score_fields)

Compute comparison metrics for multi-task predictions. Pattern matching single-task comparison but per-task.

Parameters:

• df (DataFrame) – DataFrame with predictions

• score_fields (List[str]) – List of current score column names

• label_fields (List[str]) – List of label column names

• prev_score_fields (List[str]) – List of previous score column names

Returns:

Dictionary with per-task and aggregate comparison metrics

Return type:

Dict[str, Any]

generate_multitask_comparison_plots(df, score_fields, label_fields, prev_score_fields, output_dir)

Generate per-task comparison visualizations. Pattern matching single-task comparison but per-task.

Parameters:

• df (DataFrame) – DataFrame with predictions

• score_fields (List[str]) – List of current score column names

• label_fields (List[str]) – List of label column names

• prev_score_fields (List[str]) – List of previous score column names

• output_dir (str) – Output directory for plots

Returns:

Dictionary of plot file paths

Return type:

Dict[str, str]

create_health_check_file(output_path)

Create a health check file to signal script completion.

main(input_paths, output_paths, environ_vars, job_args)

Main entry point for Model Metrics Computation script. Loads prediction data, computes metrics, generates visualizations, and saves results.

Parameters:

• input_paths (Dict[str, str]) – Dictionary of input paths

• output_paths (Dict[str, str]) – Dictionary of output paths

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

• job_args (argparse.Namespace) – Command line arguments