"""
Configuration for Risk Table Mapping Processing Step.
This module defines the configuration class for the risk table mapping processing step,
which is responsible for creating and applying risk tables for categorical features.
After Phase 6 refactor: Hyperparameters are now embedded in the source directory,
eliminating the need for S3 upload logic and hyperparameters_s3_uri field.
"""
from typing import List, Dict, Optional
from pydantic import Field, model_validator, field_validator, PrivateAttr
import logging
from .config_processing_step_base import ProcessingStepConfigBase
logger = logging.getLogger(__name__)
[docs]
class RiskTableMappingConfig(ProcessingStepConfigBase):
"""
Configuration for Risk Table Mapping Processing Step.
This class extends ProcessingStepConfigBase to include specific fields
for risk table mapping, including categorical fields and job type.
Source Directory Integration:
After Phase 6 refactor, hyperparameters are embedded in the source directory
structure and no longer require separate S3 upload. The expected source
directory structure is:
source_dir/
├── risk_table_mapping.py # Main script (entry point)
└── hyperparams/ # Hyperparameters directory
└── hyperparameters.json # Generated hyperparameters file
The hyperparameters.json file is automatically generated from the configuration
fields (cat_field_list, label_name, smooth_factor, count_threshold) and
embedded in the source directory for runtime access.
"""
# Script settings
processing_entry_point: str = Field(
default="risk_table_mapping.py",
description="Script for risk table mapping (embedded in source directory)",
)
# Job type for the processing script
job_type: str = Field(
default="training",
description="Type of job to perform. One of 'training', 'validation', 'testing', 'calibration'",
)
# Risk table mapping hyperparameters (embedded in source directory)
cat_field_list: List[str] = Field(
default=[],
description="List of categorical fields to apply risk table mapping to (embedded in hyperparams/hyperparameters.json)",
)
label_name: str = Field(
default="target",
description="Name of the target/label column (embedded in hyperparams/hyperparameters.json)",
)
smooth_factor: float = Field(
default=0.01,
description="Smoothing factor for risk table calculation (embedded in hyperparams/hyperparameters.json)",
)
count_threshold: int = Field(
default=5,
description="Minimum count threshold for risk table calculation (embedded in hyperparams/hyperparameters.json)",
)
max_unique_threshold: int = Field(
default=100,
description="Maximum unique values threshold for categorical field validation (embedded in hyperparams/hyperparameters.json)",
)
# Streaming mode configuration
enable_true_streaming: bool = Field(
default=False,
description="Enable memory-efficient streaming mode for large datasets. When enabled, processes data in shards instead of loading entire dataset into memory. Provides 3-5× speed improvement and 70-80% memory reduction.",
)
max_workers: int = Field(
default=0,
description="Number of parallel workers for streaming mode (0 = auto-detect based on CPU count, capped at 8). Only applies when enable_true_streaming=True.",
)
# ===== Derived Fields (Tier 3) =====
# These are fields calculated from other fields
# They are private with public read-only property access
_environment_variables: Optional[Dict[str, str]] = PrivateAttr(default=None)
# ===== Properties for Derived Fields =====
@property
def environment_variables(self) -> Dict[str, str]:
"""
Get environment variables for the risk table mapping script.
Returns:
Dictionary of environment variables
"""
if self._environment_variables is None:
self._environment_variables = {
"SMOOTH_FACTOR": str(self.smooth_factor),
"COUNT_THRESHOLD": str(self.count_threshold),
"MAX_UNIQUE_THRESHOLD": str(self.max_unique_threshold),
# Streaming mode configuration
"ENABLE_TRUE_STREAMING": str(self.enable_true_streaming).lower(),
"MAX_WORKERS": str(self.max_workers),
}
return self._environment_variables
[docs]
@field_validator("max_workers")
@classmethod
def validate_max_workers(cls, v: int) -> int:
"""Ensure max_workers is 0 (auto-detect) or a positive integer."""
if not isinstance(v, int) or v < 0:
raise ValueError(
"max_workers must be 0 (auto-detect) or a positive integer"
)
return v
[docs]
@field_validator("job_type")
@classmethod
def validate_job_type(cls, v: str) -> str:
"""Validate job type is one of the allowed values."""
if not v.replace("_", "").isalnum() or v != v.lower():
raise ValueError(
f"job_type must be lowercase alphanumeric (with underscores), got {v}"
)
return v.lower()
[docs]
@model_validator(mode="after")
def validate_risk_table_config(self) -> "RiskTableMappingConfig":
"""
Validate risk table mapping configuration.
After Phase 6 refactor: Simplified validation focusing on core configuration
without S3 upload logic or external hyperparameters handling.
"""
# Validate label_name is provided
if not self.label_name:
raise ValueError("label_name must be provided for risk table mapping")
# For training job type, validate cat_field_list
if self.job_type == "training" and not self.cat_field_list:
logger.warning(
"cat_field_list is empty for training job. Risk table mapping will validate fields at runtime."
)
# Validate numeric parameters
if self.smooth_factor < 0:
raise ValueError("smooth_factor must be non-negative")
if self.count_threshold < 0:
raise ValueError("count_threshold must be non-negative")
return self
[docs]
def get_job_arguments(self) -> Optional[List[str]]:
"""CLI args — config is the single source (FZ 31e1d3h)."""
return self._job_type_arg()