cursus.steps.scripts.active_sample_selection

Active Sample Selection Script for Semi-Supervised and Active Learning.

This script implements intelligent sample selection from model predictions for: 1. Semi-Supervised Learning (SSL): High-confidence samples for pseudo-labeling 2. Active Learning (AL): Uncertain/diverse samples for human labeling

Supports multiple strategies: - SSL: confidence_threshold, top_k_per_class - AL: uncertainty (margin/entropy/least_confidence), diversity (k-center), BADGE

Author: Cursus Framework Date: 2025-11-17

load_dataframe_with_format(file_path)

Load DataFrame and detect its format.

Parameters:

file_path (Path) – Path to the file

Returns:

Tuple of (DataFrame, format_string)

Return type:

Tuple[DataFrame, str]

save_dataframe_with_format(df, output_path, format_str)

Save DataFrame in specified format.

Parameters:

• df (DataFrame) – DataFrame to save

• output_path (Path) – Base output path (without extension)

• format_str (str) – Format to save in (‘csv’, ‘tsv’, or ‘parquet’)

Returns:

Path to saved file

Return type:

Path

load_inference_data(inference_data_dir, id_field='id')

Load inference data from various upstream sources with format detection.

Supports inference outputs from: - XGBoost/LightGBM/PyTorch model inference - Bedrock batch processing / Bedrock processing - Label ruleset execution

Parameters:

• inference_data_dir (str) – Path to inference output data

• id_field (str) – Name of ID column

Returns:

Tuple of (DataFrame, format_string) where format is ‘csv’, ‘tsv’, or ‘parquet’

Raises:

• FileNotFoundError – If no data files found

• ValueError – If ID field not found

Return type:

Tuple[DataFrame, str]

extract_score_columns(df, score_field=None, score_prefix='prob_class_')

Extract score columns from inference data.

Priority: 1. If SCORE_FIELD specified, use that single column 2. Otherwise, use SCORE_FIELD_PREFIX to find all matching columns 3. Fall back to auto-detection if prefix doesn’t match

Parameters:

• df (DataFrame) – DataFrame with inference data

• score_field (str | None) – Single score column name

• score_prefix (str) – Prefix for finding multiple score columns

Returns:

List of score column names

Raises:

ValueError – If no valid score columns found

Return type:

List[str]

normalize_scores_to_probabilities(df, score_cols)

Normalize various score formats to probability distributions.

Parameters:

• df (DataFrame) – DataFrame with score columns

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

Returns:

DataFrame with normalized prob_class_* columns

Return type:

DataFrame

class ConfidenceThresholdSampler(confidence_threshold=0.9, max_samples=0, random_seed=42)

Bases: object

Simple confidence-based selection for SSL pipelines.

select_batch(probabilities, indices=None)

Select samples where max probability exceeds threshold.

Returns:

Tuple of (selected_indices, confidence_scores)

Return type:

Tuple[ndarray, ndarray]

class TopKPerClassSampler(k_per_class=100, random_seed=42)

Bases: object

Balanced selection ensuring representation across classes.

select_batch(probabilities, indices=None)

Select top-k most confident samples per predicted class.

class UncertaintySampler(strategy='margin', random_seed=42)

Bases: object

Uncertainty-based sampling strategies.

compute_scores(probabilities)

Compute uncertainty scores (higher = more uncertain).

select_batch(probabilities, batch_size, indices=None)

Select batch of most uncertain samples.

class DiversitySampler(metric='euclidean', random_seed=42)

Bases: object

Core-set diversity sampling using k-center algorithm.

select_batch(embeddings, batch_size, indices=None)

Select batch using farthest-first algorithm.

class BADGESampler(metric='euclidean', random_seed=42)

Bases: object

BADGE: Batch Active learning by Diverse Gradient Embeddings.

compute_gradient_embeddings(features, probabilities)

Compute gradient embeddings for BADGE.

select_batch(features, probabilities, batch_size, indices=None)

Select batch using BADGE algorithm.

select_samples(df, strategy, batch_size, strategy_config, id_field='id')

Main selection function coordinating all sampling strategies.

Parameters:

• df (DataFrame) – DataFrame with processed data and predictions

• strategy (str) – Sampling strategy name

• batch_size (int) – Number of samples to select

• strategy_config (Dict[str, Any]) – Strategy-specific configuration

• id_field (str) – Name of ID column

Returns:

DataFrame with selected samples including selection metadata

Return type:

DataFrame

save_selected_samples(selected_df, output_dir, output_format='csv')

Save selected samples using format preservation.

Parameters:

• selected_df (DataFrame) – DataFrame with selected samples

• output_dir (str) – Output directory path

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

Returns:

Path to saved file

Return type:

str

save_selection_metadata(metadata, metadata_dir)

Save selection metadata to separate output channel.

Parameters:

• metadata (Dict[str, Any]) – Metadata dictionary

• metadata_dir (str) – Metadata output directory path

Returns:

Path to saved metadata file

Return type:

str

validate_strategy_for_use_case(strategy, use_case)

Validate strategy compatibility with use case.

main(input_paths, output_paths, environ_vars, job_args)

Main function for active sample selection.

Parameters:

• input_paths (Dict[str, str]) – Input data paths

• output_paths (Dict[str, str]) – Output data paths

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

• job_args (Namespace) – Command-line arguments