Preprocessing Pipeline

omnirec.preprocess.base.Preprocessor()

Bases: ABC, Generic[T, U]

Source code in src\omnirec\preprocess\base.py

def __init__(self) -> None:
    self._trace_component: str
    self._trace_params: dict[str, Any]

    if not hasattr(self, "_trace_component"):
        self._trace_component = type(self).__name__
    if not hasattr(self, "_trace_params"):
        self._trace_params = {}
    super().__init__()

validation_rules() -> list[ValidationRule]

Return the input validation rules for this preprocessing step.

Override this hook to declare requirements using Pandera schemas or DataFrameModel classes wrapped in ValidationRule objects. The default implementation returns an empty list, so validation is optional.

process() checks each rule against every input DataFrame before calling _process(): the raw frame, all train/validation/test frames, or those frames in every fold. Rules run in list order. Each rule's on_error setting controls whether a failure raises PreprocessorValidationError (RAISE, the default), logs a warning and continues (WARN), or logs a warning and skips this entire preprocessing step (SKIP). A skipped step returns the original dataset without adding a trace.

Validation checks inputs; the validated or coerced DataFrames returned by Pandera are not assigned back to the dataset.

Returns:

Type	Description
list[ValidationRule]	list[ValidationRule]: Rules to check before transforming the dataset.

Source code in src\omnirec\preprocess\base.py

def validation_rules(self) -> list[ValidationRule]:
    """Return the input validation rules for this preprocessing step.

    Override this hook to declare requirements using Pandera schemas or
    DataFrameModel classes wrapped in ``ValidationRule`` objects. The default
    implementation returns an empty list, so validation is optional.

    ``process()`` checks each rule against every input DataFrame before
    calling ``_process()``: the raw frame, all train/validation/test frames,
    or those frames in every fold. Rules run in list order. Each rule's
    ``on_error`` setting controls whether a failure raises
    ``PreprocessorValidationError`` (RAISE, the default), logs a warning and
    continues (WARN), or logs a warning and skips this entire preprocessing step (SKIP).
    A skipped step returns the original dataset without adding a trace.

    Validation checks inputs; the validated or coerced DataFrames returned
    by Pandera are not assigned back to the dataset.

    Returns:
        list[ValidationRule]: Rules to check before transforming the dataset.
    """
    return []

process(dataset: RecSysDataSet[T]) -> RecSysDataSet[U]

Validates and processes a dataset, recording execution metadata in the lineage.

This is the public entry point for running a preprocessor. It checks validation_rules() before delegating the transformation to _process(), then records timing, dataset shapes, and a trace.

Parameters:

Name	Type	Description	Default
dataset	RecSysDataSet[T]	The dataset to process.	required

Returns:

Type	Description
RecSysDataSet[U]	RecSysDataSet[U]: The processed dataset with an appended trace entry, or the original dataset without a new trace if validation skips this step. A skipped dataset retains its original data variant.

Raises:

Type	Description
PreprocessorValidationError	An input fails a rule with RAISE mode.

Source code in src\omnirec\preprocess\base.py

@final
def process(self, dataset: RecSysDataSet[T]) -> RecSysDataSet[U]:
    """Validates and processes a dataset, recording execution metadata in the lineage.

    This is the public entry point for running a preprocessor. It checks
    ``validation_rules()`` before delegating the transformation to
    ``_process()``, then records timing, dataset shapes, and a trace.

    Args:
        dataset (RecSysDataSet[T]): The dataset to process.

    Returns:
        RecSysDataSet[U]: The processed dataset with an appended trace entry,
            or the original dataset without a new trace if validation skips
            this step. A skipped dataset retains its original data variant.

    Raises:
        PreprocessorValidationError: An input fails a rule with RAISE mode.
    """
    for rule in self.validation_rules():
        for name, df in dataset.iter_dataframes():
            is_valid, msg = rule.is_valid(df, name, dataset._meta.name)
            if not is_valid:
                match rule.on_error:
                    case ValidationFailureMode.RAISE:
                        raise PreprocessorValidationError(msg)
                    case ValidationFailureMode.WARN:
                        self.logger.warning(f"{msg}\nContinuing anyway.")
                    case ValidationFailureMode.SKIP:
                        self.logger.warning(
                            f"{msg}\nSkipping {type(self).__name__}"
                        )
                        # HACK: This might not be the best way here:
                        return cast(RecSysDataSet[U], dataset)

    before_rows = dataset.num_interactions()
    before_columns = dataset.num_columns()
    start_time = perf_counter()
    new_ds = self._process(dataset)
    runtime = perf_counter() - start_time

    trace = Trace(
        component=self._trace_component,
        params=self._trace_params.copy(),
        executed_at=datetime.now(UTC),
        runtime=runtime,
        before_rows=before_rows,
        before_columns=before_columns,
        after_rows=new_ds.num_interactions(),
        after_columns=new_ds.num_columns(),
    )
    new_ds._lineage.append(trace)
    return new_ds

_process(dataset: RecSysDataSet[T]) -> RecSysDataSet[U] abstractmethod

Implementation hook for transforming a dataset.

Subclasses should override this method with the preprocessing logic itself. Callers should use process() rather than invoking this method directly.

Parameters:

Name	Type	Description	Default
dataset	RecSysDataSet[T]	The dataset to transform.	required

Returns:

Type	Description
RecSysDataSet[U]	RecSysDataSet[U]: The transformed dataset.

Source code in src\omnirec\preprocess\base.py

@abstractmethod
def _process(self, dataset: RecSysDataSet[T]) -> RecSysDataSet[U]:
    """Implementation hook for transforming a dataset.

    Subclasses should override this method with the preprocessing logic itself.
    Callers should use ``process()`` rather than invoking this method directly.

    Args:
        dataset (RecSysDataSet[T]): The dataset to transform.

    Returns:
        RecSysDataSet[U]: The transformed dataset.
    """
    pass

Input Validation

omnirec.preprocess.validation.ValidationRule(schema: pa.DataFrameSchema | type[pa.DataFrameModel], on_error: ValidationFailureMode = ValidationFailureMode.RAISE, message: str | None = None) dataclass

A Pandera input schema and its failure policy.

Parameters:

Name	Type	Description	Default
schema	DataFrameSchema | type[DataFrameModel]	A Pandera DataFrameSchema instance or DataFrameModel class.	required
on_error	ValidationFailureMode	How the preprocessor handles a failed check. Defaults to RAISE.	RAISE
message	str | None	Optional replacement for the generated failure summary. The dataset and DataFrame names are still included for context.	None

is_valid(df: pd.DataFrame, df_name: str = 'Unnamed', dataset_name: str = 'UnnamedDataset') -> tuple[bool, str | None]

Check a DataFrame without replacing it with Pandera's validated output.

Parameters:

Name	Type	Description	Default
df	DataFrame	Input DataFrame to check using lazy validation.	required
df_name	str	Frame name, such as raw, train, or fold_0/test.	'Unnamed'
dataset_name	str	Dataset name used in error messages.	'UnnamedDataset'

Returns:

Type	Description
tuple[bool, str | None]	tuple[bool, str | None]: True and None on success, or False and a failure summary on a Pandera SchemaErrors exception. This method does not apply the failure policy; the preprocessor does that.

Source code in src\omnirec\preprocess\validation.py

def is_valid(
    self,
    df: pd.DataFrame,
    df_name: str = "Unnamed",
    dataset_name: str = "UnnamedDataset",
) -> tuple[bool, str | None]:
    """Check a DataFrame without replacing it with Pandera's validated output.

    Args:
        df: Input DataFrame to check using lazy validation.
        df_name: Frame name, such as raw, train, or fold_0/test.
        dataset_name: Dataset name used in error messages.

    Returns:
        tuple[bool, str | None]: True and None on success, or False and a
            failure summary on a Pandera SchemaErrors exception. This method
            does not apply the failure policy; the preprocessor does that.
    """
    context = f"{dataset_name}/{df_name}"
    try:
        self.schema.validate(df, lazy=True)
    except pa.errors.SchemaErrors as e:
        if self.message is not None:
            msg = f"{context}: {self.message}"
        else:
            fc: pd.DataFrame = e.failure_cases

            if fc.empty:
                msg = f"{context}: schema validation failed."
            else:
                issues = (
                    fc.assign(
                        issue_target=fc["column"]
                        .where(fc["column"].notna(), fc["failure_case"])
                        .fillna("<schema>")
                    )
                    .groupby(
                        ["schema_context", "issue_target", "check"],
                        dropna=False,
                    )
                    .size()
                    .reset_index(name="count")
                )

                first = issues.iloc[0]
                col = first["issue_target"]
                check = first["check"]
                count = int(first["count"])
                more = len(issues) - 1

                msg = f"{context}: '{col}' failed {check} ({count} failing rows)"
                if more:
                    msg += f"\n +{more} more issue(s)"

        return False, msg

    return True, None

omnirec.preprocess.validation.ValidationFailureMode

Bases: StrEnum

Action taken by a preprocessor when an input validation rule fails.

Attributes:

Name	Type	Description
RAISE		Raise PreprocessorValidationError and stop processing.
WARN		Log a warning and continue validation and processing.
SKIP		Log a warning and return the original dataset without processing or adding a lineage entry.

omnirec.preprocess.validation.PreprocessorValidationError

Bases: Exception

Raised when a preprocessing input fails a rule with RAISE mode.

Transformations

omnirec.preprocess.subsample.Subsample(sample_size: int | float)

Bases: Preprocessor[RawData, RawData]

Subsamples the dataset to a specified size.

Parameters:

Name	Type	Description	Default
sample_size	int | float	The size of the sample to draw from the dataset. int: The absolute number of interactions to include in the sample. float: The fraction of the dataset to include in the sample (between 0 and 1).	required

Source code in src\omnirec\preprocess\subsample.py

def __init__(self, sample_size: int | float) -> None:
    """Subsamples the dataset to a specified size.

    Args:
        sample_size (int | float): The size of the sample to draw from the dataset.
                                    int: The absolute number of interactions to include in the sample.
                                    float: The fraction of the dataset to include in the sample (between 0 and 1).
    """
    super().__init__()
    self.sample_size = sample_size

omnirec.preprocess.core_pruning.CorePruning(core: int)

Bases: Preprocessor[RawData, RawData]

Prune the dataset to the specified core. Core pruning with a threshold of e.g. 5 means that only users and items with at least 5 interactions are included in the pruned dataset.

Parameters:

Name	Type	Description	Default
core	int	The core threshold for pruning.	required

Source code in src\omnirec\preprocess\core_pruning.py

def __init__(self, core: int) -> None:
    """Prune the dataset to the specified core.
    Core pruning with a threshold of e.g. 5 means that only users and items with at least 5 interactions are included in the pruned dataset.

    Args:
        core (int): The core threshold for pruning.
    """
    super().__init__()
    self.core = core

omnirec.preprocess.feedback_conversion.MakeImplicit(threshold: int | float)

Bases: Preprocessor[RawData, RawData]

Converts explicit feedback to implicit feedback using the specified threshold.

Parameters:

Name	Type	Description	Default
threshold	int | float	The threshold for converting feedback. int: Used directly as the threshold, e.g. 3 -> only interactions with a rating of 3 or higher are included. float: Interpreted as a fraction of the maximum rating, e.g. 0.5 -> only interactions with a rating of at least 50% of the maximum rating are included.	required

Source code in src\omnirec\preprocess\feedback_conversion.py

def __init__(self, threshold: int | float) -> None:
    """Converts explicit feedback to implicit feedback using the specified threshold.

    Args:
        threshold (int | float): The threshold for converting feedback.
                                    int: Used directly as the threshold, e.g. 3 -> only interactions with a rating of 3 or higher are included.
                                    float: Interpreted as a fraction of the maximum rating, e.g. 0.5 -> only interactions with a rating of at least 50% of the maximum rating are included.
    """
    super().__init__()
    self.threshold = threshold

Filtering

omnirec.preprocess.filter.TimeFilter(start: Optional[pd.Timestamp] = None, end: Optional[pd.Timestamp] = None)

Bases: Preprocessor[RawData, RawData]

Filters the interactions by a time range. Only interactions within the specified start and end timestamps are retained.

Parameters:

Name	Type	Description	Default
start	Optional[Timestamp]	The start timestamp for the filter. Defaults to None.	None
end	Optional[Timestamp]	The end timestamp for the filter. Defaults to None.	None

Source code in src\omnirec\preprocess\filter.py

def __init__(
    self, start: Optional[pd.Timestamp] = None, end: Optional[pd.Timestamp] = None
) -> None:
    """Filters the interactions by a time range. Only interactions within the specified start and end timestamps are retained.

    Args:
        start (Optional[pd.Timestamp], optional): The start timestamp for the filter. Defaults to None.
        end (Optional[pd.Timestamp], optional): The end timestamp for the filter. Defaults to None.
    """
    super().__init__()
    self._start = start
    self._end = end

omnirec.preprocess.filter.RatingFilter(lower: Optional[int | float] = None, upper: Optional[int | float] = None)

Bases: Preprocessor[RawData, RawData]

Filters the interactions by rating values. Only interactions with ratings within the specified lower and upper bounds are retained.

Parameters:

Name	Type	Description	Default
lower	Optional[int | float]	The lower bound for the filter. Defaults to None.	None
upper	Optional[int | float]	The upper bound for the filter. Defaults to None.	None

Source code in src\omnirec\preprocess\filter.py

def __init__(
    self, lower: Optional[int | float] = None, upper: Optional[int | float] = None
) -> None:
    """Filters the interactions by rating values. Only interactions with ratings within the specified lower and upper bounds are retained.

    Args:
        lower (Optional[int  |  float], optional): The lower bound for the filter. Defaults to None.
        upper (Optional[int  |  float], optional): The upper bound for the filter. Defaults to None.
    """
    super().__init__()
    self._lower = lower
    self._upper = upper

omnirec.preprocess.split.RandomCrossValidation(num_folds: int, validation_size: float | int)

Bases: DataSplit[RawData, FoldedData]

Applies random cross-validation to the dataset. Randomly splits the dataset into training, validation, and test sets for each fold.

Parameters:

Name	Type	Description	Default
num_folds	int	The number of folds to use for cross-validation.	required
validation_size	float | int	float: The proportion (between 0 and 1) of the dataset to include in the validation split. int: The absolute number of interactions to include in the validation split.	required

Source code in src\omnirec\preprocess\split.py

def __init__(self, num_folds: int, validation_size: float | int) -> None:
    """Applies random cross-validation to the dataset. Randomly splits the dataset into training, validation, and test sets for each fold.

    Args:
        num_folds (int): The number of folds to use for cross-validation.
        validation_size (float | int): float: The proportion (between 0 and 1) of the dataset to include in the validation split.
                                        int: The absolute number of interactions to include in the validation split.
    """
    super().__init__(validation_size)
    self._num_folds = num_folds

omnirec.preprocess.split.RandomHoldout(validation_size: float | int, test_size: float | int)

Bases: DataSplit[RawData, SplitData]

Applies a random holdout split to the dataset. Randomly splits the dataset into training, validation, and test sets.

Parameters:

Name	Type	Description	Default
validation_size	float | int	float: The proportion (between 0 and 1) of the dataset to include in the validation split. int: The absolute number of interactions to include in the validation split.	required
test_size	float | int	float: The proportion (between 0 and 1) of the dataset to include in the test split. int: The absolute number of interactions to include in the test split.	required

Source code in src\omnirec\preprocess\split.py

def __init__(self, validation_size: float | int, test_size: float | int) -> None:
    """Applies a random holdout split to the dataset. Randomly splits the dataset into training, validation, and test sets.

    Args:
        validation_size (float | int): float: The proportion (between 0 and 1) of the dataset to include in the validation split.
                                        int: The absolute number of interactions to include in the validation split.
        test_size (float | int): float: The proportion (between 0 and 1) of the dataset to include in the test split.
                                    int: The absolute number of interactions to include in the test split.
    """
    super().__init__(validation_size)
    self._test_size = test_size

omnirec.preprocess.split.UserCrossValidation(num_folds: int, validation_size: float | int)

Bases: DataSplit[RawData, FoldedData]

Applies user-based cross-validation to the dataset. Ensures that each user has interactions in the training, validation, and test sets in each fold.

Parameters:

Name	Type	Description	Default
num_folds	int	The number of folds to use for cross-validation.	required
validation_size	float | int	float: The proportion (between 0 and 1) of the dataset to include in the validation split. int: The absolute number of interactions to include in the validation split.	required

Source code in src\omnirec\preprocess\split.py

def __init__(self, num_folds: int, validation_size: float | int) -> None:
    """Applies user-based cross-validation to the dataset. Ensures that each user has interactions in the training, validation, and test sets in each fold.

    Args:
        num_folds (int): The number of folds to use for cross-validation.
        validation_size (float | int): float: The proportion (between 0 and 1) of the dataset to include in the validation split.
                                        int: The absolute number of interactions to include in the validation split.
    """
    super().__init__(validation_size)
    self._num_folds = num_folds

omnirec.preprocess.split.UserHoldout(validation_size: float | int, test_size: float | int)

Bases: DataSplit[RawData, SplitData]

Applies the user holdout split to the dataset. Ensures that each user has interactions in the training, validation, and test sets.

Parameters:

Name	Type	Description	Default
validation_size	float | int	float: The proportion (between 0 and 1) of the dataset to include in the validation split. int: The absolute number of interactions to include in the validation split.	required
test_size	float | int	float: The proportion (between 0 and 1) of the dataset to include in the test split. int: The absolute number of interactions to include in the test split.	required

Source code in src\omnirec\preprocess\split.py

def __init__(self, validation_size: float | int, test_size: float | int) -> None:
    """Applies the user holdout split to the dataset. Ensures that each user has interactions in the training, validation, and test sets.

    Args:
        validation_size (float | int): float: The proportion (between 0 and 1) of the dataset to include in the validation split.
                                        int: The absolute number of interactions to include in the validation split.
        test_size (float | int): float: The proportion (between 0 and 1) of the dataset to include in the test split.
                                    int: The absolute number of interactions to include in the test split.
    """
    super().__init__(validation_size)
    self._test_size = test_size

omnirec.preprocess.split.TimeBasedHoldout(validation: float | int | pd.Timestamp, test: float | int | pd.Timestamp)

Bases: DataSplit[RawData, SplitData]

Applies a time-based hold-out split on a dataset. Splits the dataset into a train, test and validation split based on the timestamp. Can either use proportions, absolute numbers or timestamps as cut-off criteria.

Parameters:

Name	Type	Description	Default
validation	float | int | Timestamp	float: The proportion (between 0 and 1) of newest interactions in the dataset to include in the validation split. int: The absolute number of newest interactions to include in the validation split. pd.Timestamp: The timestamp to use as a cut-off for the validation split. Interactions after this timestamp (newer) are included in the validation split.	required
test	float | int | Timestamp	float: The proportion (between 0 and 1) of newest interactions in the dataset to include in the test split. int: The absolute number of newest interactions to include in the test split. pd.Timestamp: The timestamp to use as a cut-off for the test split. Interactions after this timestamp (newer) are included in the test split.	required

Source code in src\omnirec\preprocess\split.py

def __init__(
    self,
    validation: float | int | pd.Timestamp,
    test: float | int | pd.Timestamp,
) -> None:
    """Applies a time-based hold-out split on a dataset. Splits the dataset into a train, test and validation split based on the timestamp. Can either use proportions, absolute numbers or timestamps as cut-off criteria.

    Args:
        validation (float | int | pd.Timestamp): float: The proportion (between 0 and 1) of newest interactions in the dataset to include in the validation split.
                                                int: The absolute number of newest interactions to include in the validation split.
                                                pd.Timestamp: The timestamp to use as a cut-off for the validation split. Interactions after this timestamp (newer) are included in the validation split.
        test (float | int | pd.Timestamp): float: The proportion (between 0 and 1) of newest interactions in the dataset to include in the test split.
                                            int: The absolute number of newest interactions to include in the test split.
                                            pd.Timestamp: The timestamp to use as a cut-off for the test split. Interactions after this timestamp (newer) are included in the test split.
    """
    super().__init__(0)

    if type(validation) is not type(test):
        self.logger.critical("Validation and test size must be the same type")
        sys.exit(1)

    self._valid_size = validation
    self._test_size = test

omnirec.preprocess.pipe.Pipe(*steps: Unpack[tuple[Unpack[Ts], Preprocessor[Any, T]]])

Bases: Generic[T]

Pipeline for automatically applying sequential preprocessing steps. Takes as input a sequence of Preprocessor objects. If process() is called, each step's process method is called in the order they were provided. Example:

    # Define preprocessing steps
    pipe = Pipe(
        Subsample(0.1),
        MakeImplicit(3),
        CorePruning(5),
        UserCrossValidation(5, 0.1),
    )

    # Apply the steps
    dataset = pipe.process(dataset)

Source code in src\omnirec\preprocess\pipe.py

def __init__(self, *steps: Unpack[tuple[Unpack[Ts], Preprocessor[Any, T]]]) -> None:
    """Pipeline for automatically applying sequential preprocessing steps. Takes as input a sequence of Preprocessor objects.
    If process() is called, each step's process method is called in the order they were provided.
    Example:
        ```Python
            # Define preprocessing steps
            pipe = Pipe(
                Subsample(0.1),
                MakeImplicit(3),
                CorePruning(5),
                UserCrossValidation(5, 0.1),
            )

            # Apply the steps
            dataset = pipe.process(dataset)
        ```
    """
    super().__init__()
    self._steps = steps