#!/usr/bin/env python
import json
import logging
import os
import sys
from subprocess import check_call
import boto3
# ============================================================================
# PACKAGE INSTALLATION CONFIGURATION
# ============================================================================
# Control which PyPI source to use via environment variable
# Set USE_SECURE_PYPI=true to use secure CodeArtifact PyPI
# Set USE_SECURE_PYPI=false or leave unset to use public PyPI
USE_SECURE_PYPI = os.environ.get("USE_SECURE_PYPI", "false").lower() == "true"
# Logging setup
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def _get_secure_pypi_access_token() -> str:
"""
Get CodeArtifact access token for secure PyPI.
Returns:
str: Authorization token for CodeArtifact
Raises:
Exception: If token retrieval fails
"""
try:
os.environ["AWS_STS_REGIONAL_ENDPOINTS"] = "regional"
sts = boto3.client("sts", region_name="us-east-1")
caller_identity = sts.get_caller_identity()
assumed_role_object = sts.assume_role(
RoleArn=f"arn:aws:iam::{os.environ.get('SECURE_PYPI_ROLE_ACCOUNT', '123456789012')}:role/SecurePyPIReadRole_"
+ caller_identity["Account"],
RoleSessionName="SecurePypiReadRole",
)
credentials = assumed_role_object["Credentials"]
code_artifact_client = boto3.client(
"codeartifact",
aws_access_key_id=credentials["AccessKeyId"],
aws_secret_access_key=credentials["SecretAccessKey"],
aws_session_token=credentials["SessionToken"],
region_name="us-west-2",
)
token = code_artifact_client.get_authorization_token(
domain=os.environ.get("SECURE_PYPI_DOMAIN", "amazon"),
domainOwner=os.environ.get("SECURE_PYPI_DOMAIN_OWNER", "123456789012"),
)["authorizationToken"]
logger.info("Successfully retrieved secure PyPI access token")
return token
except Exception as e:
logger.error(f"Failed to retrieve secure PyPI access token: {e}")
raise
[docs]
def install_packages_from_public_pypi(packages: list) -> None:
"""
Install packages from standard public PyPI.
Args:
packages: List of package specifications (e.g., ["pandas==1.5.0", "numpy"])
"""
logger.info(f"Installing {len(packages)} packages from public PyPI")
logger.info(f"Packages: {packages}")
try:
check_call([sys.executable, "-m", "pip", "install", *packages])
logger.info("✓ Successfully installed packages from public PyPI")
except Exception as e:
logger.error(f"✗ Failed to install packages from public PyPI: {e}")
raise
[docs]
def install_packages_from_secure_pypi(packages: list) -> None:
"""
Install packages from secure CodeArtifact PyPI.
Args:
packages: List of package specifications (e.g., ["pandas==1.5.0", "numpy"])
"""
logger.info(f"Installing {len(packages)} packages from secure PyPI")
logger.info(f"Packages: {packages}")
try:
token = _get_secure_pypi_access_token()
index_url = f"https://aws:{token}@{os.environ.get('SECURE_PYPI_DOMAIN', 'amazon')}-{os.environ.get('SECURE_PYPI_DOMAIN_OWNER', '123456789012')}.d.codeartifact.us-west-2.amazonaws.com/pypi/{os.environ.get('SECURE_PYPI_REPOSITORY', 'secure-pypi')}/simple/"
check_call(
[
sys.executable,
"-m",
"pip",
"install",
"--index-url",
index_url,
*packages,
]
)
logger.info("✓ Successfully installed packages from secure PyPI")
except Exception as e:
logger.error(f"✗ Failed to install packages from secure PyPI: {e}")
raise
[docs]
def install_packages(packages: list, use_secure: bool = USE_SECURE_PYPI) -> None:
"""
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.
Args:
packages: List of package specifications (e.g., ["pandas==1.5.0", "numpy"])
use_secure: 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"])
"""
logger.info("=" * 70)
logger.info("PACKAGE INSTALLATION")
logger.info("=" * 70)
logger.info(f"PyPI Source: {'SECURE (CodeArtifact)' if use_secure else 'PUBLIC'}")
logger.info(
f"Environment Variable USE_SECURE_PYPI: {os.environ.get('USE_SECURE_PYPI', 'not set')}"
)
logger.info(f"Number of packages: {len(packages)}")
logger.info("=" * 70)
try:
if use_secure:
install_packages_from_secure_pypi(packages)
else:
install_packages_from_public_pypi(packages)
logger.info("=" * 70)
logger.info("✓ PACKAGE INSTALLATION COMPLETED SUCCESSFULLY")
logger.info("=" * 70)
except Exception as e:
logger.error("=" * 70)
logger.error("✗ PACKAGE INSTALLATION FAILED")
logger.error("=" * 70)
raise
# ============================================================================
# INSTALL REQUIRED PACKAGES
# ============================================================================
# Define required packages for this script
required_packages = [
"matplotlib==3.7.0",
]
# Install packages using unified installation function
install_packages(required_packages)
print("***********************Package Installation Complete*********************")
import argparse
import logging
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple, Union
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from scipy import stats
from scipy.stats import pearsonr, spearmanr
from sklearn.metrics import (
average_precision_score,
f1_score,
precision_recall_curve,
precision_score,
recall_score,
roc_auc_score,
roc_curve,
)
# Configure logging
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
# Container path constants - aligned with script contract
CONTAINER_PATHS = {
"EVAL_DATA_DIR": "/opt/ml/processing/input/eval_data",
"OUTPUT_METRICS_DIR": "/opt/ml/processing/output/metrics",
"OUTPUT_PLOTS_DIR": "/opt/ml/processing/output/plots",
}
def _detect_file_format(file_path: str) -> str:
"""
Detect file format based on extension.
Args:
file_path: Path to file
Returns:
Format string: 'csv', 'tsv', 'parquet', or 'json'
"""
file_path_lower = file_path.lower()
if file_path_lower.endswith((".parquet", ".pq")):
return "parquet"
elif file_path_lower.endswith(".tsv"):
return "tsv"
elif file_path_lower.endswith(".json"):
return "json"
elif file_path_lower.endswith(".csv"):
return "csv"
else:
# Default to CSV for unknown extensions
return "csv"
[docs]
def detect_and_load_predictions(
input_dir: str, preferred_format: str = None
) -> pd.DataFrame:
"""
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.
"""
# Determine order of formats to try
formats_to_try = []
if preferred_format:
formats_to_try.append(preferred_format)
# Add other formats as fallback
for fmt in ["parquet", "csv", "tsv", "json"]:
if fmt not in formats_to_try:
formats_to_try.append(fmt)
# Try each format in order
for fmt in formats_to_try:
file_path = os.path.join(input_dir, f"predictions.{fmt}")
if os.path.exists(file_path):
detected_format = _detect_file_format(file_path)
logger.info(
f"Loading predictions from {file_path} (format: {detected_format})"
)
if detected_format == "parquet":
return pd.read_parquet(file_path)
elif detected_format == "tsv":
return pd.read_csv(file_path, sep="\t")
elif detected_format == "json":
return pd.read_json(file_path)
else: # csv or default
return pd.read_csv(file_path)
# Also try eval_predictions.csv from xgboost_model_eval output
eval_pred_path = os.path.join(input_dir, "eval_predictions.csv")
if os.path.exists(eval_pred_path):
logger.info(f"Loading predictions from {eval_pred_path} (format: csv)")
return pd.read_csv(eval_pred_path)
raise FileNotFoundError(
"No predictions file found in supported formats (csv, tsv, parquet, json)"
)
[docs]
def parse_score_fields(environ_vars: Dict[str, str]) -> List[str]:
"""
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
"""
# Check for SCORE_FIELDS first (multi-task)
score_fields_str = environ_vars.get("SCORE_FIELDS", "").strip()
if score_fields_str:
score_fields = [
field.strip() for field in score_fields_str.split(",") if field.strip()
]
if not score_fields:
raise ValueError("SCORE_FIELDS is empty after parsing")
logger.info(
f"Multi-task mode: Found {len(score_fields)} score fields: {score_fields}"
)
return score_fields
# Fall back to SCORE_FIELD (single-task, backward compatible)
score_field = environ_vars.get("SCORE_FIELD", "").strip()
if score_field:
logger.info(f"Single-task mode: Using score field: {score_field}")
return [score_field]
# Default
default_field = "prob_class_1"
logger.warning(
f"Neither SCORE_FIELD nor SCORE_FIELDS provided, using default: {default_field}"
)
return [default_field]
[docs]
def parse_previous_score_fields(
environ_vars: Dict[str, str], score_fields: List[str]
) -> List[str]:
"""
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
Args:
environ_vars: Environment variables dictionary
score_fields: List of score field names (for validation)
Returns:
List of previous score field names (empty if no comparison)
"""
is_multitask = len(score_fields) > 1
# Check for PREVIOUS_SCORE_FIELDS first (multi-task)
prev_score_fields_str = environ_vars.get("PREVIOUS_SCORE_FIELDS", "").strip()
if prev_score_fields_str:
prev_score_fields = [
field.strip() for field in prev_score_fields_str.split(",") if field.strip()
]
if len(prev_score_fields) != len(score_fields):
raise ValueError(
f"PREVIOUS_SCORE_FIELDS length ({len(prev_score_fields)}) must match "
f"SCORE_FIELDS length ({len(score_fields)}). "
f"Score fields: {score_fields}, Previous score fields: {prev_score_fields}"
)
logger.info(
f"Multi-task comparison mode: Found {len(prev_score_fields)} previous score fields: {prev_score_fields}"
)
return prev_score_fields
# Fall back to PREVIOUS_SCORE_FIELD (single-task, backward compatible)
prev_score_field = environ_vars.get("PREVIOUS_SCORE_FIELD", "").strip()
if prev_score_field:
if is_multitask:
logger.warning(
f"Multi-task mode detected but only PREVIOUS_SCORE_FIELD provided. "
f"Use PREVIOUS_SCORE_FIELDS for multi-task comparison."
)
return [] # Return empty to disable comparison
logger.info(
f"Single-task comparison mode: Using previous score field: {prev_score_field}"
)
return [prev_score_field]
# No comparison mode
return []
[docs]
def parse_task_label_fields(
environ_vars: Dict[str, str], score_fields: List[str]
) -> List[str]:
"""
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)
Args:
environ_vars: Environment variables dictionary
score_fields: List of score field names
Returns:
List of label field names, one per score field
"""
is_multitask = len(score_fields) > 1
# Option 1: Explicit TASK_LABEL_NAMES (preferred for multi-task)
task_labels_str = environ_vars.get("TASK_LABEL_NAMES", "").strip()
if task_labels_str:
task_labels = [
field.strip() for field in task_labels_str.split(",") if field.strip()
]
if len(task_labels) != len(score_fields):
raise ValueError(
f"TASK_LABEL_NAMES length ({len(task_labels)}) must match "
f"SCORE_FIELDS length ({len(score_fields)}). "
f"Score fields: {score_fields}, Label fields: {task_labels}"
)
logger.info(f"Using explicit task label fields: {task_labels}")
return task_labels
# Option 2: Infer from score_fields for multi-task
if is_multitask:
task_labels = []
for score_field in score_fields:
# Standard naming: task_prob → task (remove _prob suffix)
if score_field.endswith("_prob"):
label_field = score_field.replace("_prob", "") # isFraud_prob → isFraud
elif score_field.endswith("_score"):
label_field = score_field.replace("_score", "_label")
else:
# Fallback: append _true
logger.warning(
f"Score field '{score_field}' doesn't follow standard naming. "
f"Inferring label as '{score_field}_true'"
)
label_field = f"{score_field}_true"
task_labels.append(label_field)
logger.info(
f"Inferred {len(task_labels)} task label fields from score fields: "
f"{dict(zip(score_fields, task_labels))}"
)
return task_labels
# Option 3: Single-task - use LABEL_FIELD (backward compatibility)
label_field = environ_vars.get("LABEL_FIELD", "label")
logger.info(f"Single-task mode: Using label field: {label_field}")
return [label_field]
[docs]
def validate_prediction_columns(
df: pd.DataFrame,
score_fields: List[str],
label_fields: List[str],
id_field: str,
) -> Dict[str, Any]:
"""
Validate that all required columns exist in DataFrame.
Args:
df: Input DataFrame
score_fields: List of score column names
label_fields: List of label column names
id_field: ID column name
Returns:
Validation report dictionary
Raises:
ValueError: If critical columns are missing
"""
validation_report = {
"is_valid": True,
"errors": [],
"warnings": [],
"data_summary": {},
}
# Check ID field
if id_field not in df.columns:
validation_report["errors"].append(f"ID field '{id_field}' not found")
validation_report["is_valid"] = False
# Check score fields
missing_scores = [f for f in score_fields if f not in df.columns]
if missing_scores:
validation_report["errors"].append(f"Missing score fields: {missing_scores}")
validation_report["is_valid"] = False
# Check label fields
missing_labels = [f for f in label_fields if f not in df.columns]
if missing_labels:
validation_report["errors"].append(f"Missing label fields: {missing_labels}")
validation_report["is_valid"] = False
# Generate data summary for multi-task. Include prediction_columns + has_amount_data so the
# shape matches the single-task summary (validate_prediction_data) — the report-writer
# (generate_text_summary) reads data_summary['prediction_columns'] unconditionally, so the
# multi-task path must populate it too (its analog is the score columns; multi-task carries no
# amount field, so has_amount_data is False).
validation_report["data_summary"] = {
"total_records": len(df),
"score_columns": score_fields,
"label_columns": label_fields,
"prediction_columns": score_fields,
"has_amount_data": False,
}
# Log results
if not validation_report["is_valid"]:
logger.error("Column validation failed:")
for error in validation_report["errors"]:
logger.error(f" - {error}")
logger.info(f"Available columns: {df.columns.tolist()}")
return validation_report
[docs]
def validate_prediction_data(
df: pd.DataFrame, id_field: str, label_field: str, amount_field: str = None
) -> Dict[str, Any]:
"""
Validate prediction data schema and return validation report.
Legacy function for backward compatibility with single-task.
"""
validation_report = {
"is_valid": True,
"errors": [],
"warnings": [],
"data_summary": {},
}
# Check required columns
required_cols = [id_field, label_field]
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
validation_report["errors"].append(f"Missing required columns: {missing_cols}")
validation_report["is_valid"] = False
# Check prediction probability columns
prob_cols = [col for col in df.columns if col.startswith("prob_class_")]
if not prob_cols:
validation_report["errors"].append("No prediction probability columns found")
validation_report["is_valid"] = False
# Check amount column if specified
if amount_field and amount_field not in df.columns:
validation_report["warnings"].append(
f"Amount field '{amount_field}' not found - dollar recall will be skipped"
)
# Generate data summary
validation_report["data_summary"] = {
"total_records": len(df),
"prediction_columns": prob_cols,
"has_amount_data": amount_field in df.columns if amount_field else False,
"label_distribution": (
df[label_field].value_counts().to_dict()
if label_field in df.columns
else {}
),
}
return validation_report
[docs]
def compute_standard_metrics(
y_true: np.ndarray, y_prob: np.ndarray, is_binary: bool = True
) -> Dict[str, float]:
"""
Compute comprehensive standard ML metrics matching xgboost_model_eval.py.
Supports both binary and multiclass classification with full metric coverage.
"""
metrics = {}
if is_binary:
# Binary classification metrics - matching compute_metrics_binary()
y_score = y_prob[:, 1] if y_prob.shape[1] > 1 else y_prob
# Core metrics (exact match with original)
metrics["auc_roc"] = roc_auc_score(y_true, y_score)
metrics["average_precision"] = average_precision_score(y_true, y_score)
metrics["f1_score"] = f1_score(y_true, y_score > 0.5)
# Precision-Recall curve analysis
precision, recall, thresholds = precision_recall_curve(y_true, y_score)
metrics["precision_at_threshold_0.5"] = precision_score(
y_true, (y_score > 0.5).astype(int)
)
metrics["recall_at_threshold_0.5"] = recall_score(
y_true, (y_score > 0.5).astype(int)
)
# Threshold-based metrics (matching original)
for threshold in [0.3, 0.5, 0.7]:
y_pred = (y_score >= threshold).astype(int)
metrics[f"f1_score_at_{threshold}"] = f1_score(y_true, y_pred)
metrics[f"precision_at_{threshold}"] = precision_score(y_true, y_pred)
metrics[f"recall_at_{threshold}"] = recall_score(y_true, y_pred)
# Additional analysis metrics
metrics["max_f1_score"] = np.max(
2 * precision * recall / (precision + recall + 1e-8)
)
# ROC curve analysis
fpr, tpr, roc_thresholds = roc_curve(y_true, y_score)
metrics["optimal_threshold"] = roc_thresholds[np.argmax(tpr - fpr)]
else:
# Multiclass classification metrics - matching compute_metrics_multiclass()
n_classes = y_prob.shape[1]
# Per-class metrics (exact match with original)
for i in range(n_classes):
y_true_bin = (y_true == i).astype(int)
y_score = y_prob[:, i]
metrics[f"auc_roc_class_{i}"] = roc_auc_score(y_true_bin, y_score)
metrics[f"average_precision_class_{i}"] = average_precision_score(
y_true_bin, y_score
)
metrics[f"f1_score_class_{i}"] = f1_score(y_true_bin, y_score > 0.5)
# Micro and macro averages (exact match with original)
metrics["auc_roc_micro"] = roc_auc_score(
y_true, y_prob, multi_class="ovr", average="micro"
)
metrics["auc_roc_macro"] = roc_auc_score(
y_true, y_prob, multi_class="ovr", average="macro"
)
metrics["average_precision_micro"] = average_precision_score(
y_true, y_prob, average="micro"
)
metrics["average_precision_macro"] = average_precision_score(
y_true, y_prob, average="macro"
)
y_pred = np.argmax(y_prob, axis=1)
metrics["f1_score_micro"] = f1_score(y_true, y_pred, average="micro")
metrics["f1_score_macro"] = f1_score(y_true, y_pred, average="macro")
# Class distribution metrics (matching original)
unique, counts = np.unique(y_true, return_counts=True)
for cls, count in zip(unique, counts):
metrics[f"class_{cls}_count"] = int(count)
metrics[f"class_{cls}_ratio"] = float(count) / len(y_true)
return metrics
[docs]
def calculate_count_recall(scores, labels, amounts, cutoff=0.1):
"""
Calculate count recall - imported from evaluation.py
"""
assert len(scores) == len(labels), "Input lengths don't match!"
threshold = np.quantile(scores, 1 - cutoff)
abuse_order_total = len(labels[labels == 1])
abuse_order_above_threshold = len(labels[(labels == 1) & (scores >= threshold)])
order_count_recall = abuse_order_above_threshold / abuse_order_total
return order_count_recall
[docs]
def calculate_dollar_recall(scores, labels, amounts, fpr=0.1):
"""
Calculate dollar recall - imported from evaluation.py
"""
assert len(scores) == len(labels) == len(amounts), "Input lengths don't match!"
threshold = np.quantile(scores[labels == 0], 1 - fpr)
abuse_amount_total = amounts[labels == 1].sum()
abuse_amount_above_threshold = amounts[(labels == 1) & (scores > threshold)].sum()
dollar_recall = abuse_amount_above_threshold / abuse_amount_total
return dollar_recall
[docs]
def compute_domain_metrics(
scores: np.ndarray,
labels: np.ndarray,
amounts: np.ndarray = None,
compute_dollar_recall: bool = True,
compute_count_recall: bool = True,
dollar_recall_fpr: float = 0.1,
count_recall_cutoff: float = 0.1,
) -> Dict[str, float]:
"""
Compute domain-specific metrics including dollar and count recall.
Integrates functions from evaluation.py for business impact analysis.
"""
domain_metrics = {}
if compute_count_recall:
# Count recall - percentage of abuse orders caught
count_recall = calculate_count_recall(
scores=scores,
labels=labels,
amounts=amounts, # Not used but required by function signature
cutoff=count_recall_cutoff,
)
domain_metrics["count_recall"] = count_recall
domain_metrics["count_recall_cutoff"] = count_recall_cutoff
if compute_dollar_recall and amounts is not None:
# Dollar recall - percentage of abuse dollar amount caught
dollar_recall = calculate_dollar_recall(
scores=scores, labels=labels, amounts=amounts, fpr=dollar_recall_fpr
)
domain_metrics["dollar_recall"] = dollar_recall
domain_metrics["dollar_recall_fpr"] = dollar_recall_fpr
# Additional amount-based analysis
domain_metrics["total_abuse_amount"] = amounts[labels == 1].sum()
domain_metrics["average_abuse_amount"] = amounts[labels == 1].mean()
domain_metrics["total_legitimate_amount"] = amounts[labels == 0].sum()
domain_metrics["amount_ratio_abuse_to_total"] = (
amounts[labels == 1].sum() / amounts.sum()
)
return domain_metrics
[docs]
def plot_and_save_roc_curve(
y_true: np.ndarray, y_score: np.ndarray, output_dir: str, prefix: str = ""
) -> str:
"""
Plot ROC curve and save as JPG (exact match with xgboost_model_eval.py).
"""
fpr, tpr, _ = roc_curve(y_true, y_score)
auc = roc_auc_score(y_true, y_score)
plt.figure()
plt.plot(fpr, tpr, label=f"ROC curve (AUC = {auc:.2f})")
plt.plot([0, 1], [0, 1], "k--", label="Random")
plt.xlabel("False Positive Rate")
plt.ylabel("True Positive Rate")
plt.title("ROC Curve")
plt.legend(loc="lower right")
out_path = os.path.join(output_dir, f"{prefix}roc_curve.jpg")
plt.savefig(out_path, format="jpg", dpi=300, bbox_inches="tight")
plt.close()
logger.info(f"Saved ROC curve to {out_path}")
return out_path
[docs]
def plot_and_save_pr_curve(
y_true: np.ndarray, y_score: np.ndarray, output_dir: str, prefix: str = ""
) -> str:
"""
Plot Precision-Recall curve and save as JPG (exact match with xgboost_model_eval.py).
"""
precision, recall, _ = precision_recall_curve(y_true, y_score)
ap = average_precision_score(y_true, y_score)
plt.figure()
plt.plot(recall, precision, label=f"PR curve (AP = {ap:.2f})")
plt.xlabel("Recall")
plt.ylabel("Precision")
plt.title("Precision-Recall Curve")
plt.legend(loc="lower left")
out_path = os.path.join(output_dir, f"{prefix}pr_curve.jpg")
plt.savefig(out_path, format="jpg", dpi=300, bbox_inches="tight")
plt.close()
logger.info(f"Saved PR curve to {out_path}")
return out_path
[docs]
def generate_recommendations(metrics: Dict[str, float]) -> List[str]:
"""
Generate actionable recommendations based on performance analysis.
"""
recommendations = []
auc = metrics.get("auc_roc", 0)
if auc < 0.75:
recommendations.append(
"Consider feature engineering or model architecture improvements"
)
recommendations.append("Investigate data quality and label accuracy")
dollar_recall = metrics.get("dollar_recall")
count_recall = metrics.get("count_recall")
if dollar_recall and count_recall and dollar_recall < 0.6:
recommendations.append("Focus on improving detection of high-value abuse cases")
recommendations.append(
"Consider amount-weighted loss functions during training"
)
if count_recall and count_recall < 0.7:
recommendations.append(
"Consider lowering decision threshold to catch more abuse cases"
)
recommendations.append("Evaluate if additional features could improve recall")
# F1 score analysis
max_f1 = metrics.get("max_f1_score", 0)
if max_f1 < 0.6:
recommendations.append(
"Model shows poor precision-recall balance - consider class balancing techniques"
)
return recommendations
[docs]
def generate_comprehensive_report(
standard_metrics: Dict[str, float],
domain_metrics: Dict[str, float],
plot_paths: Dict[str, str],
validation_report: Dict[str, Any],
output_dir: str,
) -> Dict[str, str]:
"""
Generate comprehensive metrics report with insights and recommendations.
"""
# Combine all metrics
all_metrics = {**standard_metrics, **domain_metrics}
# Generate JSON report
json_report = {
"timestamp": datetime.utcnow().isoformat(),
"data_summary": validation_report["data_summary"],
"standard_metrics": standard_metrics,
"domain_metrics": domain_metrics,
"visualizations": plot_paths,
"performance_insights": generate_performance_insights(all_metrics),
"recommendations": generate_recommendations(all_metrics),
}
json_path = os.path.join(output_dir, "metrics_report.json")
with open(json_path, "w") as f:
json.dump(json_report, f, indent=2)
# Generate text summary
text_summary = generate_text_summary(json_report)
text_path = os.path.join(output_dir, "metrics_summary.txt")
with open(text_path, "w") as f:
f.write(text_summary)
return {"json_report": json_path, "text_summary": text_path}
[docs]
def compute_comparison_metrics(
y_true: np.ndarray,
y_new_score: np.ndarray,
y_prev_score: np.ndarray,
is_binary: bool = True,
) -> Dict[str, float]:
"""
Compute comparison metrics between new model and previous model scores.
Identical to xgboost_model_eval.py implementation.
"""
logger.info("Computing model comparison metrics")
comparison_metrics = {}
# Basic correlation metrics - with error handling for scipy compatibility
try:
pearson_corr, pearson_p = pearsonr(y_new_score, y_prev_score)
spearman_corr, spearman_p = spearmanr(y_new_score, y_prev_score)
except (TypeError, AttributeError) as e:
logger.warning(
f"SciPy correlation computation failed: {e}. Using fallback numpy correlation."
)
# Fallback to numpy correlation
pearson_corr = float(np.corrcoef(y_new_score, y_prev_score)[0, 1])
pearson_p = 0.0 # p-value not available with numpy
spearman_corr = pearson_corr # Use Pearson as fallback
spearman_p = 0.0
comparison_metrics.update(
{
"pearson_correlation": pearson_corr,
"pearson_p_value": pearson_p,
"spearman_correlation": spearman_corr,
"spearman_p_value": spearman_p,
}
)
# Performance comparison metrics
if is_binary:
# Binary classification comparison
new_auc = roc_auc_score(y_true, y_new_score)
prev_auc = roc_auc_score(y_true, y_prev_score)
new_ap = average_precision_score(y_true, y_new_score)
prev_ap = average_precision_score(y_true, y_prev_score)
# Delta metrics
comparison_metrics.update(
{
"new_model_auc": new_auc,
"previous_model_auc": prev_auc,
"auc_delta": new_auc - prev_auc,
"auc_lift_percent": (
((new_auc - prev_auc) / prev_auc) * 100 if prev_auc > 0 else 0
),
"new_model_ap": new_ap,
"previous_model_ap": prev_ap,
"ap_delta": new_ap - prev_ap,
"ap_lift_percent": (
((new_ap - prev_ap) / prev_ap) * 100 if prev_ap > 0 else 0
),
}
)
# F1 score comparison at different thresholds
for threshold in [0.3, 0.5, 0.7]:
new_f1 = f1_score(y_true, (y_new_score >= threshold).astype(int))
prev_f1 = f1_score(y_true, (y_prev_score >= threshold).astype(int))
comparison_metrics[f"new_model_f1_at_{threshold}"] = new_f1
comparison_metrics[f"previous_model_f1_at_{threshold}"] = prev_f1
comparison_metrics[f"f1_delta_at_{threshold}"] = new_f1 - prev_f1
# Score distribution comparison
comparison_metrics.update(
{
"new_score_mean": float(np.mean(y_new_score)),
"previous_score_mean": float(np.mean(y_prev_score)),
"new_score_std": float(np.std(y_new_score)),
"previous_score_std": float(np.std(y_prev_score)),
"score_mean_delta": float(np.mean(y_new_score) - np.mean(y_prev_score)),
}
)
# Agreement metrics
if is_binary:
for threshold in [0.3, 0.5, 0.7]:
new_pred = (y_new_score >= threshold).astype(int)
prev_pred = (y_prev_score >= threshold).astype(int)
agreement = np.mean(new_pred == prev_pred)
comparison_metrics[f"prediction_agreement_at_{threshold}"] = agreement
logger.info(
f"Comparison metrics computed: AUC delta={comparison_metrics.get('auc_delta', 'N/A'):.4f}, "
f"Correlation={comparison_metrics.get('pearson_correlation', 'N/A'):.4f}"
)
return comparison_metrics
[docs]
def plot_comparison_roc_curves(
y_true: np.ndarray,
y_new_score: np.ndarray,
y_prev_score: np.ndarray,
output_dir: str,
) -> str:
"""
Plot side-by-side ROC curves comparing new and previous models.
Identical to xgboost_model_eval.py implementation.
"""
logger.info("Creating comparison ROC curves")
# Calculate ROC curves for both models
fpr_new, tpr_new, _ = roc_curve(y_true, y_new_score)
fpr_prev, tpr_prev, _ = roc_curve(y_true, y_prev_score)
auc_new = roc_auc_score(y_true, y_new_score)
auc_prev = roc_auc_score(y_true, y_prev_score)
plt.figure(figsize=(10, 6))
# Plot both ROC curves
plt.plot(
fpr_new, tpr_new, "b-", linewidth=2, label=f"New Model (AUC = {auc_new:.3f})"
)
plt.plot(
fpr_prev,
tpr_prev,
"r--",
linewidth=2,
label=f"Previous Model (AUC = {auc_prev:.3f})",
)
plt.plot([0, 1], [0, 1], "k:", alpha=0.6, label="Random")
plt.xlabel("False Positive Rate")
plt.ylabel("True Positive Rate")
plt.title(f"ROC Curve Comparison (Δ AUC = {auc_new - auc_prev:+.3f})")
plt.legend(loc="lower right")
plt.grid(True, alpha=0.3)
out_path = os.path.join(output_dir, "comparison_roc_curves.jpg")
plt.savefig(out_path, format="jpg", dpi=150, bbox_inches="tight")
plt.close()
logger.info(f"Saved comparison ROC curves to {out_path}")
return out_path
[docs]
def plot_comparison_pr_curves(
y_true: np.ndarray,
y_new_score: np.ndarray,
y_prev_score: np.ndarray,
output_dir: str,
) -> str:
"""
Plot side-by-side Precision-Recall curves comparing new and previous models.
Identical to xgboost_model_eval.py implementation.
"""
logger.info("Creating comparison PR curves")
# Calculate PR curves for both models
precision_new, recall_new, _ = precision_recall_curve(y_true, y_new_score)
precision_prev, recall_prev, _ = precision_recall_curve(y_true, y_prev_score)
ap_new = average_precision_score(y_true, y_new_score)
ap_prev = average_precision_score(y_true, y_prev_score)
plt.figure(figsize=(10, 6))
# Plot both PR curves
plt.plot(
recall_new,
precision_new,
"b-",
linewidth=2,
label=f"New Model (AP = {ap_new:.3f})",
)
plt.plot(
recall_prev,
precision_prev,
"r--",
linewidth=2,
label=f"Previous Model (AP = {ap_prev:.3f})",
)
# Add baseline (random classifier)
baseline = np.mean(y_true)
plt.axhline(
y=baseline,
color="k",
linestyle=":",
alpha=0.6,
label=f"Random (AP = {baseline:.3f})",
)
plt.xlabel("Recall")
plt.ylabel("Precision")
plt.title(f"Precision-Recall Curve Comparison (Δ AP = {ap_new - ap_prev:+.3f})")
plt.legend(loc="lower left")
plt.grid(True, alpha=0.3)
out_path = os.path.join(output_dir, "comparison_pr_curves.jpg")
plt.savefig(out_path, format="jpg", dpi=150, bbox_inches="tight")
plt.close()
logger.info(f"Saved comparison PR curves to {out_path}")
return out_path
[docs]
def plot_score_scatter(
y_new_score: np.ndarray,
y_prev_score: np.ndarray,
y_true: np.ndarray,
output_dir: str,
) -> str:
"""
Plot scatter plot of new vs previous model scores, colored by true labels.
Identical to xgboost_model_eval.py implementation.
"""
logger.info("Creating score scatter plot")
plt.figure(figsize=(10, 8))
# Separate positive and negative examples
pos_mask = y_true == 1
neg_mask = y_true == 0
# Plot negative examples
plt.scatter(
y_prev_score[neg_mask],
y_new_score[neg_mask],
c="lightcoral",
alpha=0.6,
s=20,
label="Negative (0)",
)
# Plot positive examples
plt.scatter(
y_prev_score[pos_mask],
y_new_score[pos_mask],
c="lightblue",
alpha=0.6,
s=20,
label="Positive (1)",
)
# Add diagonal line (perfect correlation)
min_score = min(np.min(y_prev_score), np.min(y_new_score))
max_score = max(np.max(y_prev_score), np.max(y_new_score))
plt.plot(
[min_score, max_score],
[min_score, max_score],
"k--",
alpha=0.8,
label="Perfect Correlation",
)
# Calculate and display correlation with error handling for SciPy compatibility
try:
correlation = pearsonr(y_new_score, y_prev_score)[0]
except (TypeError, AttributeError) as e:
logger.warning(f"SciPy pearsonr failed: {e}. Using numpy correlation.")
correlation = float(np.corrcoef(y_new_score, y_prev_score)[0, 1])
plt.xlabel("Previous Model Score")
plt.ylabel("New Model Score")
plt.title(f"Model Score Comparison (Correlation = {correlation:.3f})")
plt.legend()
plt.grid(True, alpha=0.3)
# Add correlation text box
textstr = f"Pearson r = {correlation:.3f}"
props = dict(boxstyle="round", facecolor="wheat", alpha=0.8)
plt.text(
0.05,
0.95,
textstr,
transform=plt.gca().transAxes,
fontsize=12,
verticalalignment="top",
bbox=props,
)
out_path = os.path.join(output_dir, "score_scatter_plot.jpg")
plt.savefig(out_path, format="jpg", dpi=150, bbox_inches="tight")
plt.close()
logger.info(f"Saved score scatter plot to {out_path}")
return out_path
[docs]
def plot_score_distributions(
y_new_score: np.ndarray,
y_prev_score: np.ndarray,
y_true: np.ndarray,
output_dir: str,
) -> str:
"""
Plot score distributions for both models, separated by true labels.
Identical to xgboost_model_eval.py implementation.
"""
logger.info("Creating score distribution plots")
# Set matplotlib backend explicitly for headless environments
import matplotlib
matplotlib.use("Agg")
# Create figure and axes with comprehensive error handling
fig = None
axes = None
try:
# First attempt: standard subplots
result = plt.subplots(2, 2, figsize=(15, 10))
if isinstance(result, tuple) and len(result) == 2:
fig, axes = result
# Ensure axes is always a 2D array
if hasattr(axes, "ndim") and axes.ndim == 1:
axes = axes.reshape(2, 2)
else:
raise ValueError("subplots returned unexpected format")
except Exception as e:
logger.warning(f"Standard subplots failed: {e}. Using fallback approach.")
# Fallback approach: create figure and individual subplots
try:
fig = plt.figure(figsize=(15, 10))
axes = []
for i in range(4):
ax = fig.add_subplot(2, 2, i + 1)
axes.append(ax)
axes = np.array(axes).reshape(2, 2)
except Exception as e2:
logger.error(
f"Fallback subplot creation also failed: {e2}. Creating minimal plot."
)
# Final fallback: create a simple single plot to satisfy test expectations
fig = plt.figure(figsize=(8, 6))
ax = fig.add_subplot(1, 1, 1)
ax.text(0.5, 0.5, "Plot generation failed", ha="center", va="center")
ax.set_title("Score Distributions (Error)")
# Continue to save this minimal plot
out_path = os.path.join(output_dir, "score_distributions.jpg")
plt.savefig(out_path, format="jpg", dpi=150, bbox_inches="tight")
plt.close()
logger.info(f"Saved minimal error plot to {out_path}")
return out_path
# Separate positive and negative examples
pos_mask = y_true == 1
neg_mask = y_true == 0
# Plot 1: New model score distributions
axes[0, 0].hist(
y_new_score[neg_mask],
bins=30,
alpha=0.7,
color="lightcoral",
label="Negative (0)",
density=True,
)
axes[0, 0].hist(
y_new_score[pos_mask],
bins=30,
alpha=0.7,
color="lightblue",
label="Positive (1)",
density=True,
)
axes[0, 0].set_title("New Model Score Distribution")
axes[0, 0].set_xlabel("Score")
axes[0, 0].set_ylabel("Density")
axes[0, 0].legend()
axes[0, 0].grid(True, alpha=0.3)
# Plot 2: Previous model score distributions
axes[0, 1].hist(
y_prev_score[neg_mask],
bins=30,
alpha=0.7,
color="lightcoral",
label="Negative (0)",
density=True,
)
axes[0, 1].hist(
y_prev_score[pos_mask],
bins=30,
alpha=0.7,
color="lightblue",
label="Positive (1)",
density=True,
)
axes[0, 1].set_title("Previous Model Score Distribution")
axes[0, 1].set_xlabel("Score")
axes[0, 1].set_ylabel("Density")
axes[0, 1].legend()
axes[0, 1].grid(True, alpha=0.3)
# Plot 3: Score difference distribution
score_diff = y_new_score - y_prev_score
axes[1, 0].hist(
score_diff[neg_mask],
bins=30,
alpha=0.7,
color="lightcoral",
label="Negative (0)",
density=True,
)
axes[1, 0].hist(
score_diff[pos_mask],
bins=30,
alpha=0.7,
color="lightblue",
label="Positive (1)",
density=True,
)
axes[1, 0].axvline(x=0, color="black", linestyle="--", alpha=0.8)
axes[1, 0].set_title("Score Difference Distribution (New - Previous)")
axes[1, 0].set_xlabel("Score Difference")
axes[1, 0].set_ylabel("Density")
axes[1, 0].legend()
axes[1, 0].grid(True, alpha=0.3)
# Plot 4: Box plots comparing both models
box_data = [
y_prev_score[neg_mask],
y_new_score[neg_mask],
y_prev_score[pos_mask],
y_new_score[pos_mask],
]
box_labels = ["Prev (Neg)", "New (Neg)", "Prev (Pos)", "New (Pos)"]
box_colors = ["lightcoral", "lightcoral", "lightblue", "lightblue"]
# Set tick labels AFTER plotting rather than via a boxplot kwarg: matplotlib renamed the
# `labels` kwarg to `tick_labels` in 3.9 and removed `labels` in 3.11, so passing it raises
# TypeError. set_xticklabels is stable across versions.
bp = axes[1, 1].boxplot(box_data, patch_artist=True)
axes[1, 1].set_xticklabels(box_labels)
for patch, color in zip(bp["boxes"], box_colors):
patch.set_facecolor(color)
patch.set_alpha(0.7)
axes[1, 1].set_title("Score Distribution Comparison")
axes[1, 1].set_ylabel("Score")
axes[1, 1].grid(True, alpha=0.3)
plt.tight_layout()
out_path = os.path.join(output_dir, "score_distributions.jpg")
plt.savefig(out_path, format="jpg", dpi=150, bbox_inches="tight")
plt.close()
logger.info(f"Saved score distribution plots to {out_path}")
return out_path
[docs]
def generate_text_summary(json_report: Dict[str, Any]) -> str:
"""
Generate a human-readable text summary from the JSON report.
"""
summary = []
summary.append("MODEL METRICS COMPUTATION REPORT")
summary.append("=" * 50)
summary.append(f"Generated: {json_report['timestamp']}")
summary.append("")
# Data summary. Use .get() defaults so the report never crashes on a summary variant that omits
# a key (the single-task and multi-task validators build slightly different shapes).
data_summary = json_report["data_summary"]
summary.append("DATA SUMMARY")
summary.append("-" * 20)
summary.append(f"Total Records: {data_summary.get('total_records', 'N/A')}")
summary.append(
f"Prediction Columns: {', '.join(data_summary.get('prediction_columns', []))}"
)
summary.append(f"Has Amount Data: {data_summary.get('has_amount_data', False)}")
summary.append("")
# Standard metrics
standard_metrics = json_report["standard_metrics"]
summary.append("STANDARD ML METRICS")
summary.append("-" * 20)
for name, value in standard_metrics.items():
if isinstance(value, (int, float)):
summary.append(f"{name}: {value:.4f}")
else:
summary.append(f"{name}: {value}")
summary.append("")
# Domain metrics
domain_metrics = json_report["domain_metrics"]
if domain_metrics:
summary.append("DOMAIN-SPECIFIC METRICS")
summary.append("-" * 25)
for name, value in domain_metrics.items():
if isinstance(value, (int, float)):
summary.append(f"{name}: {value:.4f}")
else:
summary.append(f"{name}: {value}")
summary.append("")
# Performance insights
insights = json_report["performance_insights"]
if insights:
summary.append("PERFORMANCE INSIGHTS")
summary.append("-" * 20)
for insight in insights:
summary.append(f"• {insight}")
summary.append("")
# Recommendations
recommendations = json_report["recommendations"]
if recommendations:
summary.append("RECOMMENDATIONS")
summary.append("-" * 15)
for rec in recommendations:
summary.append(f"• {rec}")
summary.append("")
# Visualizations
visualizations = json_report["visualizations"]
if visualizations:
summary.append("GENERATED VISUALIZATIONS")
summary.append("-" * 25)
for name, path in visualizations.items():
summary.append(f"• {name}: {os.path.basename(path)}")
return "\n".join(summary)
[docs]
def log_metrics_summary(
metrics: Dict[str, Union[int, float, str]], is_binary: bool = True
) -> None:
"""
Log a nicely formatted summary of metrics for easy visibility in logs.
"""
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
logger.info("=" * 80)
logger.info(f"METRICS SUMMARY - {timestamp}")
logger.info("=" * 80)
# Log each metric with a consistent format
for name, value in metrics.items():
# Format numeric values to 4 decimal places
if isinstance(value, (int, float)):
formatted_value = f"{value:.4f}"
else:
formatted_value = str(value)
# Add a special prefix for easy searching in logs
logger.info(f"METRIC: {name.ljust(25)} = {formatted_value}")
# Highlight key metrics based on task type
logger.info("=" * 80)
logger.info("KEY PERFORMANCE METRICS")
logger.info("=" * 80)
if is_binary:
logger.info(
f"METRIC_KEY: AUC-ROC = {metrics.get('auc_roc', 'N/A'):.4f}"
)
logger.info(
f"METRIC_KEY: Average Precision = {metrics.get('average_precision', 'N/A'):.4f}"
)
logger.info(
f"METRIC_KEY: F1 Score = {metrics.get('f1_score', 'N/A'):.4f}"
)
else:
logger.info(
f"METRIC_KEY: Macro AUC-ROC = {metrics.get('auc_roc_macro', 'N/A'):.4f}"
)
logger.info(
f"METRIC_KEY: Micro AUC-ROC = {metrics.get('auc_roc_micro', 'N/A'):.4f}"
)
ap_macro = metrics.get("average_precision_macro", "N/A")
if isinstance(ap_macro, (int, float)):
logger.info(f"METRIC_KEY: Macro Average Precision = {ap_macro:.4f}")
else:
logger.info(f"METRIC_KEY: Macro Average Precision = {ap_macro}")
logger.info(
f"METRIC_KEY: Macro F1 = {metrics.get('f1_score_macro', 'N/A'):.4f}"
)
logger.info(
f"METRIC_KEY: Micro F1 = {metrics.get('f1_score_micro', 'N/A'):.4f}"
)
logger.info("=" * 80)
[docs]
def save_metrics(
metrics: Dict[str, Union[int, float, str]], output_metrics_dir: str
) -> None:
"""
Save computed metrics as a JSON file (matching xgboost_model_eval.py).
"""
# Convert numpy types to Python native types for JSON serialization
serializable_metrics = {}
for key, value in metrics.items():
if isinstance(value, np.bool_):
serializable_metrics[key] = bool(value)
elif isinstance(value, (np.integer, np.int64, np.int32)):
serializable_metrics[key] = int(value)
elif isinstance(value, (np.floating, np.float64, np.float32)):
serializable_metrics[key] = float(value)
elif isinstance(value, np.ndarray):
serializable_metrics[key] = value.tolist()
else:
serializable_metrics[key] = value
out_path = os.path.join(output_metrics_dir, "metrics.json")
with open(out_path, "w") as f:
json.dump(serializable_metrics, f, indent=2)
logger.info(f"Saved metrics to {out_path}")
# Also create a plain text summary for easy viewing
summary_path = os.path.join(output_metrics_dir, "metrics_summary.txt")
with open(summary_path, "w") as f:
f.write("METRICS SUMMARY\n")
f.write("=" * 50 + "\n")
# Write key metrics at the top
if "auc_roc" in metrics: # Binary classification
f.write(f"AUC-ROC: {metrics['auc_roc']:.4f}\n")
if "average_precision" in metrics:
f.write(f"Average Precision: {metrics['average_precision']:.4f}\n")
if "f1_score" in metrics:
f.write(f"F1 Score: {metrics['f1_score']:.4f}\n")
else: # Multiclass classification
f.write(f"AUC-ROC (Macro): {metrics.get('auc_roc_macro', 'N/A'):.4f}\n")
f.write(f"AUC-ROC (Micro): {metrics.get('auc_roc_micro', 'N/A'):.4f}\n")
f.write(f"F1 Score (Macro): {metrics.get('f1_score_macro', 'N/A'):.4f}\n")
f.write("=" * 50 + "\n\n")
# Write all metrics
f.write("ALL METRICS\n")
f.write("=" * 50 + "\n")
for name, value in sorted(metrics.items()):
if isinstance(value, (int, float)):
f.write(f"{name}: {value:.6f}\n")
else:
f.write(f"{name}: {value}\n")
logger.info(f"Saved metrics summary to {summary_path}")
[docs]
def compute_multitask_metrics(
df: pd.DataFrame,
score_fields: List[str],
label_fields: List[str],
amounts: np.ndarray = None,
environ_vars: Dict[str, str] = None,
) -> Dict[str, Any]:
"""
Compute per-task and aggregate metrics for multi-task predictions.
Pattern matching lightgbmmt_model_eval.py
Args:
df: DataFrame with predictions
score_fields: List of score column names
label_fields: List of label column names
amounts: Optional array of transaction amounts
environ_vars: Environment variables for domain metrics
Returns:
Dictionary with per-task and aggregate metrics
"""
logger.info("Computing multi-task metrics")
metrics = {}
# Per-task metrics
auc_rocs = []
aps = []
f1s = []
for score_field, label_field in zip(score_fields, label_fields):
logger.info(f"Computing metrics for task: {label_field}")
y_true = df[label_field].values
y_score = df[score_field].values
# Reshape for binary classification
y_prob = np.column_stack([1 - y_score, y_score]) # [prob_class_0, prob_class_1]
# Compute standard metrics
task_metrics = compute_standard_metrics(y_true, y_prob, is_binary=True)
# Store with task prefix
metrics[f"task_{label_field}"] = task_metrics
# Collect for aggregation
auc_rocs.append(task_metrics["auc_roc"])
aps.append(task_metrics["average_precision"])
f1s.append(task_metrics["f1_score"])
logger.info(
f"Task {label_field}: AUC={task_metrics['auc_roc']:.4f}, "
f"AP={task_metrics['average_precision']:.4f}, "
f"F1={task_metrics['f1_score']:.4f}"
)
# Aggregate metrics (matching lightgbmmt_model_eval.py pattern)
if auc_rocs:
metrics["aggregate"] = {
"mean_auc_roc": float(np.mean(auc_rocs)),
"median_auc_roc": float(np.median(auc_rocs)),
"mean_average_precision": float(np.mean(aps)),
"median_average_precision": float(np.median(aps)),
"mean_f1_score": float(np.mean(f1s)),
"median_f1_score": float(np.median(f1s)),
}
logger.info("Aggregate Metrics:")
logger.info(f" Mean AUC-ROC: {metrics['aggregate']['mean_auc_roc']:.4f}")
logger.info(f" Mean AP: {metrics['aggregate']['mean_average_precision']:.4f}")
logger.info(f" Mean F1: {metrics['aggregate']['mean_f1_score']:.4f}")
# Domain metrics per task (if amounts provided)
if amounts is not None and environ_vars:
compute_dollar = (
environ_vars.get("COMPUTE_DOLLAR_RECALL", "true").lower() == "true"
)
compute_count = (
environ_vars.get("COMPUTE_COUNT_RECALL", "true").lower() == "true"
)
if compute_dollar or compute_count:
domain_metrics = compute_multitask_domain_metrics(
df, score_fields, label_fields, amounts, environ_vars
)
metrics.update(domain_metrics)
return metrics
[docs]
def compute_multitask_domain_metrics(
df: pd.DataFrame,
score_fields: List[str],
label_fields: List[str],
amounts: np.ndarray,
environ_vars: Dict[str, str],
) -> Dict[str, Any]:
"""
Compute domain-specific metrics (dollar/count recall) for each task.
Args:
df: DataFrame with predictions
score_fields: List of score column names
label_fields: List of label column names
amounts: Array of transaction amounts
environ_vars: Environment variables
Returns:
Dictionary with per-task domain metrics
"""
domain_metrics = {}
compute_dollar = environ_vars.get("COMPUTE_DOLLAR_RECALL", "true").lower() == "true"
compute_count = environ_vars.get("COMPUTE_COUNT_RECALL", "true").lower() == "true"
dollar_fpr = float(environ_vars.get("DOLLAR_RECALL_FPR", "0.1"))
count_cutoff = float(environ_vars.get("COUNT_RECALL_CUTOFF", "0.1"))
for score_field, label_field in zip(score_fields, label_fields):
y_true = df[label_field].values
y_score = df[score_field].values
# Compute domain metrics for this task
task_domain = compute_domain_metrics(
scores=y_score,
labels=y_true,
amounts=amounts,
compute_dollar_recall=compute_dollar,
compute_count_recall=compute_count,
dollar_recall_fpr=dollar_fpr,
count_recall_cutoff=count_cutoff,
)
# Store with task prefix
for metric_name, value in task_domain.items():
domain_metrics[f"task_{label_field}_{metric_name}"] = value
return domain_metrics
[docs]
def generate_multitask_visualizations(
df: pd.DataFrame,
score_fields: List[str],
label_fields: List[str],
output_dir: str,
) -> Dict[str, str]:
"""
Generate per-task ROC and PR curves.
Pattern matching lightgbmmt_model_eval.py
Args:
df: DataFrame with predictions
score_fields: List of score column names
label_fields: List of label column names
output_dir: Output directory for plots
Returns:
Dictionary of plot file paths
"""
logger.info("Generating multi-task visualizations")
plot_paths = {}
for score_field, label_field in zip(score_fields, label_fields):
logger.info(f"Generating plots for task: {label_field}")
y_true = df[label_field].values
y_score = df[score_field].values
# Skip if only one class present
if len(np.unique(y_true)) < 2:
logger.warning(
f"Task {label_field}: Only one class present, skipping plots"
)
continue
# ROC curve
plot_paths[f"task_{label_field}_roc"] = plot_and_save_roc_curve(
y_true, y_score, output_dir, prefix=f"task_{label_field}_"
)
# PR curve
plot_paths[f"task_{label_field}_pr"] = plot_and_save_pr_curve(
y_true, y_score, output_dir, prefix=f"task_{label_field}_"
)
logger.info(f"Generated {len(plot_paths)} visualization plots")
return plot_paths
[docs]
def compute_multitask_comparison_metrics(
df: pd.DataFrame,
score_fields: List[str],
label_fields: List[str],
prev_score_fields: List[str],
) -> Dict[str, Any]:
"""
Compute comparison metrics for multi-task predictions.
Pattern matching single-task comparison but per-task.
Args:
df: DataFrame with predictions
score_fields: List of current score column names
label_fields: List of label column names
prev_score_fields: List of previous score column names
Returns:
Dictionary with per-task and aggregate comparison metrics
"""
logger.info("Computing multi-task comparison metrics")
comparison_metrics = {}
# Per-task comparison metrics
auc_deltas = []
ap_deltas = []
correlations = []
for score_field, label_field, prev_score_field in zip(
score_fields, label_fields, prev_score_fields
):
logger.info(f"Computing comparison for task: {label_field}")
y_true = df[label_field].values
y_new_score = df[score_field].values
y_prev_score = df[prev_score_field].values
# Compute task-specific comparison metrics
task_comp = compute_comparison_metrics(
y_true, y_new_score, y_prev_score, is_binary=True
)
# Store with task prefix
for metric_name, value in task_comp.items():
comparison_metrics[f"task_{label_field}_{metric_name}"] = value
# Collect for aggregation
auc_deltas.append(task_comp.get("auc_delta", 0))
ap_deltas.append(task_comp.get("ap_delta", 0))
correlations.append(task_comp.get("pearson_correlation", 0))
logger.info(
f"Task {label_field}: AUC delta={task_comp.get('auc_delta', 0):.4f}, "
f"Correlation={task_comp.get('pearson_correlation', 0):.4f}"
)
# Aggregate comparison metrics
if auc_deltas:
comparison_metrics["aggregate_comparison"] = {
"mean_auc_delta": float(np.mean(auc_deltas)),
"median_auc_delta": float(np.median(auc_deltas)),
"mean_ap_delta": float(np.mean(ap_deltas)),
"median_ap_delta": float(np.median(ap_deltas)),
"mean_correlation": float(np.mean(correlations)),
"median_correlation": float(np.median(correlations)),
}
logger.info("Aggregate Comparison Metrics:")
logger.info(
f" Mean AUC Delta: {comparison_metrics['aggregate_comparison']['mean_auc_delta']:.4f}"
)
logger.info(
f" Mean Correlation: {comparison_metrics['aggregate_comparison']['mean_correlation']:.4f}"
)
return comparison_metrics
[docs]
def generate_multitask_comparison_plots(
df: pd.DataFrame,
score_fields: List[str],
label_fields: List[str],
prev_score_fields: List[str],
output_dir: str,
) -> Dict[str, str]:
"""
Generate per-task comparison visualizations.
Pattern matching single-task comparison but per-task.
Args:
df: DataFrame with predictions
score_fields: List of current score column names
label_fields: List of label column names
prev_score_fields: List of previous score column names
output_dir: Output directory for plots
Returns:
Dictionary of plot file paths
"""
logger.info("Generating multi-task comparison visualizations")
plot_paths = {}
for score_field, label_field, prev_score_field in zip(
score_fields, label_fields, prev_score_fields
):
logger.info(f"Generating comparison plots for task: {label_field}")
y_true = df[label_field].values
y_new_score = df[score_field].values
y_prev_score = df[prev_score_field].values
# Skip if only one class present
if len(np.unique(y_true)) < 2:
logger.warning(
f"Task {label_field}: Only one class present, skipping comparison plots"
)
continue
# Generate per-task comparison plots with task prefix
prefix = f"task_{label_field}_"
# ROC curve comparison
plot_paths[f"task_{label_field}_comparison_roc"] = plot_comparison_roc_curves(
y_true, y_new_score, y_prev_score, output_dir
)
# Rename to include task prefix
old_path = plot_paths[f"task_{label_field}_comparison_roc"]
new_path = os.path.join(output_dir, f"{prefix}comparison_roc_curves.jpg")
if os.path.exists(old_path):
os.rename(old_path, new_path)
plot_paths[f"task_{label_field}_comparison_roc"] = new_path
# PR curve comparison
plot_paths[f"task_{label_field}_comparison_pr"] = plot_comparison_pr_curves(
y_true, y_new_score, y_prev_score, output_dir
)
# Rename to include task prefix
old_path = plot_paths[f"task_{label_field}_comparison_pr"]
new_path = os.path.join(output_dir, f"{prefix}comparison_pr_curves.jpg")
if os.path.exists(old_path):
os.rename(old_path, new_path)
plot_paths[f"task_{label_field}_comparison_pr"] = new_path
# Score scatter plot
plot_paths[f"task_{label_field}_score_scatter"] = plot_score_scatter(
y_new_score, y_prev_score, y_true, output_dir
)
# Rename to include task prefix
old_path = plot_paths[f"task_{label_field}_score_scatter"]
new_path = os.path.join(output_dir, f"{prefix}score_scatter_plot.jpg")
if os.path.exists(old_path):
os.rename(old_path, new_path)
plot_paths[f"task_{label_field}_score_scatter"] = new_path
# Score distributions
plot_paths[f"task_{label_field}_score_distributions"] = (
plot_score_distributions(y_new_score, y_prev_score, y_true, output_dir)
)
# Rename to include task prefix
old_path = plot_paths[f"task_{label_field}_score_distributions"]
new_path = os.path.join(output_dir, f"{prefix}score_distributions.jpg")
if os.path.exists(old_path):
os.rename(old_path, new_path)
plot_paths[f"task_{label_field}_score_distributions"] = new_path
logger.info(f"Generated {len(plot_paths)} multi-task comparison plots")
return plot_paths
[docs]
def create_health_check_file(output_path: str) -> str:
"""Create a health check file to signal script completion."""
health_path = output_path
with open(health_path, "w") as f:
f.write(f"healthy: {datetime.now().isoformat()}")
return health_path
[docs]
def main(
input_paths: Dict[str, str],
output_paths: Dict[str, str],
environ_vars: Dict[str, str],
job_args: argparse.Namespace,
) -> None:
"""
Main entry point for Model Metrics Computation script.
Loads prediction data, computes metrics, generates visualizations, and saves results.
Args:
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
"""
# Extract paths from parameters - using contract-defined logical names
eval_data_dir = input_paths.get("eval_output")
output_metrics_dir = output_paths.get("metrics_output")
# plots_output is an optional contract output; when absent, co-locate plots with the metrics
# dir (a literal var default, not an undeclared-alias fallback) so makedirs never sees None.
output_plots_dir = output_paths.get("plots_output", output_metrics_dir)
# Extract environment variables
id_field = environ_vars.get("ID_FIELD", "id")
label_field = environ_vars.get("LABEL_FIELD", "label")
amount_field = environ_vars.get("AMOUNT_FIELD", None)
input_format = environ_vars.get("INPUT_FORMAT", "auto")
compute_dollar_recall = (
environ_vars.get("COMPUTE_DOLLAR_RECALL", "true").lower() == "true"
)
compute_count_recall = (
environ_vars.get("COMPUTE_COUNT_RECALL", "true").lower() == "true"
)
dollar_recall_fpr = float(environ_vars.get("DOLLAR_RECALL_FPR", "0.1"))
count_recall_cutoff = float(environ_vars.get("COUNT_RECALL_CUTOFF", "0.1"))
generate_plots = environ_vars.get("GENERATE_PLOTS", "true").lower() == "true"
# Extract comparison mode environment variables
comparison_mode = environ_vars.get("COMPARISON_MODE", "false").lower() == "true"
previous_score_field = environ_vars.get("PREVIOUS_SCORE_FIELD", "")
comparison_metrics = environ_vars.get("COMPARISON_METRICS", "all")
statistical_tests = environ_vars.get("STATISTICAL_TESTS", "true").lower() == "true"
comparison_plots = environ_vars.get("COMPARISON_PLOTS", "true").lower() == "true"
# Guard rail: If PREVIOUS_SCORE_FIELD is empty, disable comparison mode
if comparison_mode and (
not previous_score_field or previous_score_field.strip() == ""
):
logger.warning(
"COMPARISON_MODE is enabled but PREVIOUS_SCORE_FIELD is empty. Disabling comparison mode."
)
comparison_mode = False
logger.info(f"Comparison mode: {comparison_mode}")
if comparison_mode:
logger.info(f"Previous score field: {previous_score_field}")
logger.info(f"Comparison metrics: {comparison_metrics}")
logger.info(f"Statistical tests: {statistical_tests}")
logger.info(f"Comparison plots: {comparison_plots}")
# Log job info
logger.info("Running model metrics computation")
# Ensure output directories exist
os.makedirs(output_metrics_dir, exist_ok=True)
os.makedirs(output_plots_dir, exist_ok=True)
logger.info("Starting model metrics computation script")
# ===== NEW: Multi-Task Detection =====
# Step 1: Parse score fields (determines single-task vs multi-task)
score_fields = parse_score_fields(environ_vars)
is_multitask = len(score_fields) > 1
logger.info(f"Detected mode: {'multi-task' if is_multitask else 'single-task'}")
logger.info(f"Score fields: {score_fields}")
# Step 2: Parse label fields (infer if needed)
label_fields = parse_task_label_fields(environ_vars, score_fields)
logger.info(f"Label fields: {label_fields}")
# Step 2b: Parse previous score fields (for comparison mode)
prev_score_fields = parse_previous_score_fields(environ_vars, score_fields)
if prev_score_fields:
logger.info(f"Previous score fields: {prev_score_fields}")
logger.info(
f"Comparison mode will be enabled for {'multi-task' if is_multitask else 'single-task'}"
)
# Step 3: Load and validate prediction data
df = detect_and_load_predictions(
eval_data_dir, preferred_format=input_format if input_format != "auto" else None
)
# Step 4: Validate columns exist
validation_report = validate_prediction_columns(
df, score_fields, label_fields, id_field
)
if not validation_report["is_valid"]:
logger.error("Data validation failed:")
for error in validation_report["errors"]:
logger.error(f" - {error}")
raise ValueError("Input data validation failed")
# Step 4b: Validate previous score fields if in comparison mode
if prev_score_fields:
missing_prev = [f for f in prev_score_fields if f not in df.columns]
if missing_prev:
logger.warning(
f"Comparison mode requested but previous score fields missing: {missing_prev}. "
f"Disabling comparison mode."
)
prev_score_fields = [] # Disable comparison
else:
logger.info(
f"Validated previous score fields exist in data: {prev_score_fields}"
)
# Step 5: Get amounts if available
amounts = (
df[amount_field].values if amount_field and amount_field in df.columns else None
)
# ===== ROUTING: Multi-Task vs Single-Task =====
if is_multitask:
logger.info(
f"Running multi-task metrics computation for {len(score_fields)} tasks"
)
# Compute multi-task metrics
all_metrics = compute_multitask_metrics(
df, score_fields, label_fields, amounts, environ_vars
)
# Generate multi-task visualizations
if generate_plots:
plot_paths = generate_multitask_visualizations(
df, score_fields, label_fields, output_plots_dir
)
else:
plot_paths = {}
# Multi-task comparison mode
if prev_score_fields:
logger.info(
f"Enabling multi-task comparison mode for {len(prev_score_fields)} tasks"
)
# Compute multi-task comparison metrics
mt_comp_metrics = compute_multitask_comparison_metrics(
df, score_fields, label_fields, prev_score_fields
)
all_metrics.update(mt_comp_metrics)
# Generate multi-task comparison plots
if generate_plots and comparison_plots:
logger.info("Generating multi-task comparison visualizations")
mt_comp_plots = generate_multitask_comparison_plots(
df, score_fields, label_fields, prev_score_fields, output_plots_dir
)
plot_paths.update(mt_comp_plots)
logger.info(
f"Generated {len(mt_comp_plots)} multi-task comparison plots"
)
# Extract standard and domain metrics for reporting
standard_metrics = all_metrics.get("aggregate", {})
domain_metrics = {
k: v
for k, v in all_metrics.items()
if k.startswith("task_") and k not in ["aggregate"]
}
else:
# Single-task mode (EXISTING CODE PATH - unchanged)
logger.info("Running single-task metrics computation")
# Extract single task data
label_field = label_fields[0]
score_field = score_fields[0]
# Use legacy validation for backward compatibility
legacy_validation = validate_prediction_data(
df, id_field, label_field, amount_field
)
# Log warnings
for warning in legacy_validation.get("warnings", []):
logger.warning(warning)
y_true = df[label_field].values
# Detect probability columns (original logic)
prob_cols = [col for col in df.columns if col.startswith("prob_class_")]
if not prob_cols:
# Fallback: create binary prob columns from score field
logger.info(
f"No prob_class_* columns found, using {score_field} to create binary probabilities"
)
prob_cols = ["prob_class_0", "prob_class_1"]
df["prob_class_0"] = 1 - df[score_field]
df["prob_class_1"] = df[score_field]
y_prob = df[prob_cols].values
is_binary = y_prob.shape[1] == 2
scores = y_prob[:, 1] if is_binary else np.max(y_prob, axis=1)
logger.info(
f"Computing metrics for {'binary' if is_binary else 'multiclass'} classification"
)
logger.info(f"Data shape: {df.shape}, Predictions shape: {y_prob.shape}")
# Compute standard metrics (EXISTING)
standard_metrics = compute_standard_metrics(y_true, y_prob, is_binary=is_binary)
log_metrics_summary(standard_metrics, is_binary=is_binary)
# Compute domain metrics (EXISTING)
domain_metrics = compute_domain_metrics(
scores=scores,
labels=y_true,
amounts=amounts,
compute_dollar_recall=compute_dollar_recall and amounts is not None,
compute_count_recall=compute_count_recall,
dollar_recall_fpr=dollar_recall_fpr,
count_recall_cutoff=count_recall_cutoff,
)
if domain_metrics:
logger.info("Domain-specific metrics computed:")
for name, value in domain_metrics.items():
if isinstance(value, (int, float)):
logger.info(f" {name}: {value:.4f}")
else:
logger.info(f" {name}: {value}")
# Generate visualizations (EXISTING)
plot_paths = {}
if generate_plots:
logger.info("Generating performance visualizations")
plot_paths = generate_performance_visualizations(
y_true, y_prob, standard_metrics, output_plots_dir, is_binary=is_binary
)
logger.info(f"Generated {len(plot_paths)} visualization plots")
# Combine metrics
all_metrics = {**standard_metrics, **domain_metrics}
# Check for comparison mode
previous_scores = None
if comparison_mode:
if previous_score_field in df.columns:
previous_scores = df[previous_score_field].values
logger.info(
f"Found previous model scores in column '{previous_score_field}' with {len(previous_scores)} values"
)
else:
logger.warning(
f"Comparison mode enabled but column '{previous_score_field}' not found in data. Proceeding with standard evaluation."
)
comparison_mode = False
# Combine all metrics for saving
all_metrics = {**standard_metrics, **domain_metrics}
# Add comparison metrics if comparison mode is enabled
if comparison_mode and previous_scores is not None:
logger.info("Computing comparison metrics")
# Compute comparison metrics
if comparison_metrics in ["all", "basic"]:
comp_metrics = compute_comparison_metrics(
y_true, scores, previous_scores, is_binary=is_binary
)
all_metrics.update(comp_metrics)
# Perform statistical tests
if statistical_tests:
stat_results = perform_statistical_tests(
y_true, scores, previous_scores, is_binary=is_binary
)
all_metrics.update(stat_results)
# Generate comparison plots
if comparison_plots and generate_plots:
logger.info("Generating comparison visualizations")
comparison_plot_paths = {}
if is_binary:
comparison_plot_paths["comparison_roc_curves"] = (
plot_comparison_roc_curves(
y_true, scores, previous_scores, output_plots_dir
)
)
comparison_plot_paths["comparison_pr_curves"] = (
plot_comparison_pr_curves(
y_true, scores, previous_scores, output_plots_dir
)
)
comparison_plot_paths["score_scatter_plot"] = plot_score_scatter(
scores, previous_scores, y_true, output_plots_dir
)
comparison_plot_paths["score_distributions"] = plot_score_distributions(
scores, previous_scores, y_true, output_plots_dir
)
# Add comparison plots to existing plot paths
plot_paths.update(comparison_plot_paths)
logger.info(
f"Generated {len(comparison_plot_paths)} comparison visualization plots"
)
# Save metrics in original format (matching xgboost_model_eval.py)
save_metrics(all_metrics, output_metrics_dir)
# Generate comprehensive report
report_paths = generate_comprehensive_report(
standard_metrics,
domain_metrics,
plot_paths,
validation_report,
output_metrics_dir,
)
logger.info(f"Generated comprehensive report: {report_paths['json_report']}")
logger.info(f"Generated text summary: {report_paths['text_summary']}")
logger.info("Model metrics computation script complete")
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--job_type", type=str, required=True)
args = parser.parse_args()
# Set up paths using contract-defined paths only
input_paths = {
"processed_data": CONTAINER_PATHS["EVAL_DATA_DIR"],
}
output_paths = {
"metrics_output": CONTAINER_PATHS["OUTPUT_METRICS_DIR"],
"plots_output": CONTAINER_PATHS["OUTPUT_PLOTS_DIR"],
}
# Collect environment variables
environ_vars = {
# Basic field configuration
"ID_FIELD": os.environ.get("ID_FIELD", "id"),
"LABEL_FIELD": os.environ.get("LABEL_FIELD", "label"),
"AMOUNT_FIELD": os.environ.get("AMOUNT_FIELD", None),
"INPUT_FORMAT": os.environ.get("INPUT_FORMAT", "auto"),
# Multi-task configuration (NEW)
"SCORE_FIELDS": os.environ.get(
"SCORE_FIELDS", ""
), # Comma-separated score fields for multi-task
"SCORE_FIELD": os.environ.get(
"SCORE_FIELD", ""
), # Single score field for backward compatibility
"TASK_LABEL_NAMES": os.environ.get(
"TASK_LABEL_NAMES", ""
), # Optional explicit task labels
"PREVIOUS_SCORE_FIELDS": os.environ.get(
"PREVIOUS_SCORE_FIELDS", ""
), # Comma-separated previous score fields for multi-task comparison
# Domain metrics configuration
"COMPUTE_DOLLAR_RECALL": os.environ.get("COMPUTE_DOLLAR_RECALL", "true"),
"COMPUTE_COUNT_RECALL": os.environ.get("COMPUTE_COUNT_RECALL", "true"),
"DOLLAR_RECALL_FPR": os.environ.get("DOLLAR_RECALL_FPR", "0.1"),
"COUNT_RECALL_CUTOFF": os.environ.get("COUNT_RECALL_CUTOFF", "0.1"),
# Visualization configuration
"GENERATE_PLOTS": os.environ.get("GENERATE_PLOTS", "true"),
# Comparison mode configuration
"COMPARISON_MODE": os.environ.get("COMPARISON_MODE", "false"),
"PREVIOUS_SCORE_FIELD": os.environ.get("PREVIOUS_SCORE_FIELD", ""),
"COMPARISON_METRICS": os.environ.get("COMPARISON_METRICS", "all"),
"STATISTICAL_TESTS": os.environ.get("STATISTICAL_TESTS", "true"),
"COMPARISON_PLOTS": os.environ.get("COMPARISON_PLOTS", "true"),
}
try:
# Call main function with testability parameters
main(input_paths, output_paths, environ_vars, args)
# Signal success
success_path = os.path.join(output_paths["metrics_output"], "_SUCCESS")
Path(success_path).touch()
logger.info(f"Created success marker: {success_path}")
# Create health check file
health_path = os.path.join(output_paths["metrics_output"], "_HEALTH")
create_health_check_file(health_path)
logger.info(f"Created health check file: {health_path}")
sys.exit(0)
except Exception as e:
# Log error and create failure marker
logger.exception(f"Script failed with error: {e}")
failure_path = os.path.join(
output_paths.get("metrics_output", "/tmp"), "_FAILURE"
)
os.makedirs(os.path.dirname(failure_path), exist_ok=True)
with open(failure_path, "w") as f:
f.write(f"Error: {str(e)}")
sys.exit(1)