oversampleqa.typed_validator

oversampleqa.typed_validator

Typed validator with runtime validation and async support.

PydanticValidationConfig

Bases: BaseModel

Runtime validation for configuration parameters.

Source code in src/oversampleqa/typed_validator.py

class PydanticValidationConfig(BaseModel):
    """Runtime validation for configuration parameters."""

    hidden_ratio: float = Field(default=0.1, gt=0.0, lt=1.0)
    metric: str = Field(default="hassanat")
    return_details: bool = Field(default=False)
    random_state: int | None = Field(default=None)

    @field_validator("metric")
    def validate_metric(cls, value: str) -> str:
        """Validate that the metric is supported.

        Accepts registered plugin metrics as well as built-ins. Checking the
        built-in table alone rejected a plugin metric at config construction --
        before any validation ran -- even though ``distance_matrix`` would
        compute it.

        Args:
            value: Metric name.

        Returns:
            The validated metric name.

        Raises:
            ValueError: If the metric is neither built in nor registered.
        """
        resolve_metric(value)
        return value

    @field_validator("random_state")
    def validate_random_state(cls, value: int | None) -> int | None:
        """Validate random_state bounds when provided.

        Args:
            value: Optional random state.

        Returns:
            The validated random state.
        """
        if value is not None and not (0 <= value < 2**31):
            raise ValueError("random_state must be between 0 and 2**31 - 1")
        return value

validate_metric(value)

Validate that the metric is supported.

Accepts registered plugin metrics as well as built-ins. Checking the built-in table alone rejected a plugin metric at config construction -- before any validation ran -- even though distance_matrix would compute it.

Parameters:

Name	Type	Description	Default
value	str	Metric name.	required

Returns:

Type	Description
str	The validated metric name.

Raises:

Type	Description
ValueError	If the metric is neither built in nor registered.

Source code in src/oversampleqa/typed_validator.py

@field_validator("metric")
def validate_metric(cls, value: str) -> str:
    """Validate that the metric is supported.

    Accepts registered plugin metrics as well as built-ins. Checking the
    built-in table alone rejected a plugin metric at config construction --
    before any validation ran -- even though ``distance_matrix`` would
    compute it.

    Args:
        value: Metric name.

    Returns:
        The validated metric name.

    Raises:
        ValueError: If the metric is neither built in nor registered.
    """
    resolve_metric(value)
    return value

validate_random_state(value)

Validate random_state bounds when provided.

Parameters:

Name	Type	Description	Default
value	int | None	Optional random state.	required

Returns:

Type	Description
int | None	The validated random state.

Source code in src/oversampleqa/typed_validator.py

@field_validator("random_state")
def validate_random_state(cls, value: int | None) -> int | None:
    """Validate random_state bounds when provided.

    Args:
        value: Optional random state.

    Returns:
        The validated random state.
    """
    if value is not None and not (0 <= value < 2**31):
        raise ValueError("random_state must be between 0 and 2**31 - 1")
    return value

TypedValidator

Bases: BaseValidator[ValidationResult]

Type-safe validator wrapper with runtime validation.

Source code in src/oversampleqa/typed_validator.py

class TypedValidator(BaseValidator[ValidationResult]):
    """Type-safe validator wrapper with runtime validation."""

    def __init__(self, mode: ValidationMode = ValidationMode.STANDARD) -> None:
        self.mode = mode

    @overload
    def validate(
        self,
        X: FloatArray,
        y: IntArray,
        minority_label: int,
        oversampler: OversamplerProtocol,
        config: ValidationConfig,
    ) -> ValidationResult:
        """Validate using a prebuilt ValidationConfig.

        Args:
            X: Feature matrix.
            y: Target labels.
            minority_label: Minority class label.
            oversampler: Oversampler instance.
            config: ValidationConfig.

        Returns:
            ValidationResult.
        """

    @overload
    def validate(
        self,
        X: FloatArray,
        y: IntArray,
        minority_label: int,
        oversampler: OversamplerProtocol,
        *,
        hidden_ratio: float = 0.1,
        metric: str = "hassanat",
        return_details: bool = False,
        random_state: int | None = None,
    ) -> ValidationResult:
        """Validate using keyword configuration parameters.

        Args:
            X: Feature matrix.
            y: Target labels.
            minority_label: Minority class label.
            oversampler: Oversampler instance.
            hidden_ratio: Fraction of majority to hide.
            metric: Distance metric name.
            return_details: Whether to include distance matrices.
            random_state: Optional random seed.

        Returns:
            ValidationResult.
        """

    def validate(
        self,
        X: FloatArray,
        y: IntArray,
        minority_label: int,
        oversampler: OversamplerProtocol,
        config: ValidationConfig | None = None,
        **kwargs: Any,
    ) -> ValidationResult:
        """Validate oversampling with typed configuration.

        Args:
            X: Feature matrix.
            y: Target labels.
            minority_label: Minority class label.
            oversampler: Oversampler instance.
            config: ValidationConfig, or None to build from kwargs.
            **kwargs: ValidationConfig fields when ``config`` is None.

        Returns:
            ValidationResult with error rate and optional details.
        """
        if config is None:
            parsed = PydanticValidationConfig(**kwargs)
            metric_name = cast(MetricName, parsed.metric)
            config = ValidationConfig(
                hidden_ratio=parsed.hidden_ratio,
                metric=metric_name,
                return_details=parsed.return_details,
                random_state=parsed.random_state,
            )

        self._validate_inputs(X, y, minority_label, oversampler, config)

        if self.mode == ValidationMode.MEMORY_EFFICIENT:
            return self._validate_standard(X, y, minority_label, oversampler, config)
        if self.mode == ValidationMode.PARALLEL:
            # Future: add parallel implementation
            return self._validate_standard(X, y, minority_label, oversampler, config)
        if self.mode == ValidationMode.ASYNC:
            # Deliberately raises rather than driving a loop.
            #
            # This used to call get_event_loop().run_until_complete(), which
            # throws if a loop is already running -- so it broke in Jupyter and
            # in any async host, exactly where someone would reach for it.
            #
            # Driving it correctly would not help either: the work is CPU-bound
            # NumPy, so asyncio buys no concurrency at all. It only moves the
            # call onto an executor thread and waits for it. Callers who want
            # a coroutine should await validate_async directly; callers who want
            # parallelism want processes, not an event loop.
            raise ConfigurationError(
                "ValidationMode.ASYNC cannot be driven from a synchronous call. "
                "Validation is CPU-bound NumPy, so asyncio provides no "
                "concurrency for it. Await `validate_async(...)` directly from "
                "async code, or use ValidationMode.STANDARD here and parallelise "
                "across repeats or datasets instead."
            )
        return self._validate_standard(X, y, minority_label, oversampler, config)

    async def validate_async(
        self,
        X: FloatArray,
        y: IntArray,
        minority_label: int,
        oversampler: OversamplerProtocol,
        config: ValidationConfig,
    ) -> ValidationResult:
        """Async wrapper around validate using an executor.

        Args:
            X: Feature matrix.
            y: Target labels.
            minority_label: Minority class label.
            oversampler: Oversampler instance.
            config: ValidationConfig.

        Returns:
            ValidationResult.
        """
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(
            None, self.validate, X, y, minority_label, oversampler, config
        )

    def _validate_inputs(
        self,
        X: FloatArray,
        y: IntArray,
        minority_label: int,
        oversampler: OversamplerProtocol,
        config: ValidationConfig,
    ) -> None:
        """Validate input arrays and configuration.

        Args:
            X: Feature matrix.
            y: Target labels.
            minority_label: Minority class label.
            oversampler: Oversampler instance.
            config: ValidationConfig.
        """
        if not isinstance(X, np.ndarray) or not np.issubdtype(X.dtype, np.floating):
            raise ValidationError("X must be a floating-point numpy array")
        if not isinstance(y, np.ndarray) or not np.issubdtype(y.dtype, np.integer):
            raise ValidationError("y must be an integer numpy array")
        if X.shape[0] != y.shape[0]:
            raise ValidationError("X and y must have the same number of rows")
        if minority_label not in y:
            raise ValidationError(f"minority_label {minority_label} not present in y")
        if not hasattr(oversampler, "fit_resample"):
            raise ValidationError("oversampler must implement fit_resample")
        # Built-ins *and* registered plugins. Checking _METRICS alone rejected a
        # plugin metric that distance_matrix would happily have computed.
        try:
            resolve_metric(config.metric)
        except ValueError as exc:
            raise MetricError(str(exc)) from None

    def _validate_standard(
        self,
        X: FloatArray,
        y: IntArray,
        minority_label: int,
        oversampler: OversamplerProtocol,
        config: ValidationConfig,
    ) -> ValidationResult:
        """Execute the standard validation pipeline.

        Args:
            X: Feature matrix.
            y: Target labels.
            minority_label: Minority class label.
            oversampler: Oversampler instance.
            config: ValidationConfig.

        Returns:
            ValidationResult.
        """
        from .validator import validate_oversampling

        try:
            if config.return_details:
                details = validate_oversampling(
                    X,
                    y,
                    minority_label,
                    oversampler,
                    hidden_ratio=config.hidden_ratio,
                    metric=config.metric,
                    return_details=True,
                    reference=config.reference,
                    random_state=config.random_state,
                    n_repeats=config.n_repeats,
                )
                # return_details=True always yields ValidationDetails; the
                # runtime check narrows the union without suppressing the type.
                if not isinstance(details, ValidationDetails):
                    raise TypeError(
                        "validate_oversampling(return_details=True) must return "
                        f"ValidationDetails, got {type(details).__name__}"
                    )
                n_synthetic = details.n_synthetic
                ci = self._wilson_confidence_interval(
                    details.error_rate, max(n_synthetic, 1)
                )
                return ValidationResult(
                    error_rate=details.error_rate,
                    n_errors=details.n_errors,
                    n_synthetic=n_synthetic,
                    confidence_interval=ci,
                    metadata={
                        "distance_matrices": {
                            "hidden": details.dist_hidden,
                            "minority": details.dist_min,
                        },
                        "n_ties": details.n_ties,
                        "duplication_rate": details.duplication_rate,
                        "reference": details.reference,
                        "random_state": config.random_state,
                        "n_repeats": details.n_repeats,
                        "rates": details.rates,
                        "std": details.std,
                        "repeat_interval": details.interval,
                    },
                )
            error_rate = validate_oversampling(
                X,
                y,
                minority_label,
                oversampler,
                hidden_ratio=config.hidden_ratio,
                metric=config.metric,
                return_details=False,
                reference=config.reference,
                random_state=config.random_state,
                n_repeats=config.n_repeats,
            )
            if isinstance(error_rate, ValidationDetails):  # pragma: no cover
                raise ValidationError(
                    "validate_oversampling(return_details=False) must return a float"
                )
            ci = self._wilson_confidence_interval(error_rate, len(y))
            return ValidationResult(
                error_rate=error_rate,
                n_errors=0,
                n_synthetic=0,
                confidence_interval=ci,
                metadata={},
            )
        except Exception as exc:  # pragma: no cover - defensive
            raise ValidationError(f"Validation failed: {exc}") from exc

    @staticmethod
    def _wilson_confidence_interval(
        rate: float, n: int, z: float = 1.96
    ) -> tuple[float, float]:
        """Compute a Wilson score interval for a binomial proportion.

        Replaces the previous Wald interval, which is unreliable exactly where
        this package spends most of its time. Wald is symmetric around the
        estimate, so near ``rate = 0`` it produces a degenerate zero-width
        interval (its standard error vanishes) and can extend below zero. Error
        rates near zero are the common case here. Wilson stays inside ``[0, 1]``
        and keeps sensible width at the boundaries.

        .. warning::

           This assumes **independent Bernoulli trials**. Synthetic points
           interpolated from shared parent points are not independent, so a real
           interval is wider than this one. Treat it as a lower bound on
           uncertainty. Proper inference is not yet implemented; see
           :func:`~oversampleqa.validate_oversampling`'s ``n_repeats``, which
           measures the variability of the hold-out split instead.

        Args:
            rate: Estimated proportion.
            n: Sample size.
            z: Z-score for the confidence level.

        Returns:
            Lower and upper confidence bounds, both within ``[0, 1]``.
        """
        if n <= 0:
            return (0.0, 1.0)
        denominator = 1.0 + z**2 / n
        centre = (rate + z**2 / (2 * n)) / denominator
        margin = z * math.sqrt(rate * (1 - rate) / n + z**2 / (4 * n**2)) / denominator
        return (max(0.0, centre - margin), min(1.0, centre + margin))

validate(X, y, minority_label, oversampler, config=None, **kwargs)

validate(
    X: FloatArray,
    y: IntArray,
    minority_label: int,
    oversampler: OversamplerProtocol,
    config: ValidationConfig,
) -> ValidationResult

validate(
    X: FloatArray,
    y: IntArray,
    minority_label: int,
    oversampler: OversamplerProtocol,
    *,
    hidden_ratio: float = 0.1,
    metric: str = "hassanat",
    return_details: bool = False,
    random_state: int | None = None,
) -> ValidationResult

Validate oversampling with typed configuration.

Parameters:

Name	Type	Description	Default
X	FloatArray	Feature matrix.	required
y	IntArray	Target labels.	required
minority_label	int	Minority class label.	required
oversampler	OversamplerProtocol	Oversampler instance.	required
config	ValidationConfig | None	ValidationConfig, or None to build from kwargs.	None
**kwargs	Any	ValidationConfig fields when config is None.	{}

Returns:

Type	Description
ValidationResult	ValidationResult with error rate and optional details.

Source code in src/oversampleqa/typed_validator.py

def validate(
    self,
    X: FloatArray,
    y: IntArray,
    minority_label: int,
    oversampler: OversamplerProtocol,
    config: ValidationConfig | None = None,
    **kwargs: Any,
) -> ValidationResult:
    """Validate oversampling with typed configuration.

    Args:
        X: Feature matrix.
        y: Target labels.
        minority_label: Minority class label.
        oversampler: Oversampler instance.
        config: ValidationConfig, or None to build from kwargs.
        **kwargs: ValidationConfig fields when ``config`` is None.

    Returns:
        ValidationResult with error rate and optional details.
    """
    if config is None:
        parsed = PydanticValidationConfig(**kwargs)
        metric_name = cast(MetricName, parsed.metric)
        config = ValidationConfig(
            hidden_ratio=parsed.hidden_ratio,
            metric=metric_name,
            return_details=parsed.return_details,
            random_state=parsed.random_state,
        )

    self._validate_inputs(X, y, minority_label, oversampler, config)

    if self.mode == ValidationMode.MEMORY_EFFICIENT:
        return self._validate_standard(X, y, minority_label, oversampler, config)
    if self.mode == ValidationMode.PARALLEL:
        # Future: add parallel implementation
        return self._validate_standard(X, y, minority_label, oversampler, config)
    if self.mode == ValidationMode.ASYNC:
        # Deliberately raises rather than driving a loop.
        #
        # This used to call get_event_loop().run_until_complete(), which
        # throws if a loop is already running -- so it broke in Jupyter and
        # in any async host, exactly where someone would reach for it.
        #
        # Driving it correctly would not help either: the work is CPU-bound
        # NumPy, so asyncio buys no concurrency at all. It only moves the
        # call onto an executor thread and waits for it. Callers who want
        # a coroutine should await validate_async directly; callers who want
        # parallelism want processes, not an event loop.
        raise ConfigurationError(
            "ValidationMode.ASYNC cannot be driven from a synchronous call. "
            "Validation is CPU-bound NumPy, so asyncio provides no "
            "concurrency for it. Await `validate_async(...)` directly from "
            "async code, or use ValidationMode.STANDARD here and parallelise "
            "across repeats or datasets instead."
        )
    return self._validate_standard(X, y, minority_label, oversampler, config)

validate_async(X, y, minority_label, oversampler, config) async

Async wrapper around validate using an executor.

Parameters:

Name	Type	Description	Default
X	FloatArray	Feature matrix.	required
y	IntArray	Target labels.	required
minority_label	int	Minority class label.	required
oversampler	OversamplerProtocol	Oversampler instance.	required
config	ValidationConfig	ValidationConfig.	required

Returns:

Type	Description
ValidationResult	ValidationResult.

Source code in src/oversampleqa/typed_validator.py

async def validate_async(
    self,
    X: FloatArray,
    y: IntArray,
    minority_label: int,
    oversampler: OversamplerProtocol,
    config: ValidationConfig,
) -> ValidationResult:
    """Async wrapper around validate using an executor.

    Args:
        X: Feature matrix.
        y: Target labels.
        minority_label: Minority class label.
        oversampler: Oversampler instance.
        config: ValidationConfig.

    Returns:
        ValidationResult.
    """
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(
        None, self.validate, X, y, minority_label, oversampler, config
    )

ServiceRegistry

Minimal dependency injection container.

Source code in src/oversampleqa/typed_validator.py

class ServiceRegistry:
    """Minimal dependency injection container."""

    def __init__(self) -> None:
        self._services: dict[type, Any] = {}

    def register(self, service_type: type, implementation: Any) -> None:
        """Register a service implementation by type.

        Args:
            service_type: Key type.
            implementation: Service implementation instance.
        """
        self._services[service_type] = implementation

    def get(self, service_type: type) -> Any:
        """Retrieve a registered service implementation.

        Args:
            service_type: Key type.

        Returns:
            Registered service implementation.
        """
        if service_type not in self._services:
            raise ConfigurationError(f"Service {service_type} not registered")
        return self._services[service_type]

register(service_type, implementation)

Register a service implementation by type.

Parameters:

Name	Type	Description	Default
service_type	type	Key type.	required
implementation	Any	Service implementation instance.	required

Source code in src/oversampleqa/typed_validator.py

def register(self, service_type: type, implementation: Any) -> None:
    """Register a service implementation by type.

    Args:
        service_type: Key type.
        implementation: Service implementation instance.
    """
    self._services[service_type] = implementation

get(service_type)

Retrieve a registered service implementation.

Parameters:

Name	Type	Description	Default
service_type	type	Key type.	required

Returns:

Type	Description
Any	Registered service implementation.

Source code in src/oversampleqa/typed_validator.py

def get(self, service_type: type) -> Any:
    """Retrieve a registered service implementation.

    Args:
        service_type: Key type.

    Returns:
        Registered service implementation.
    """
    if service_type not in self._services:
        raise ConfigurationError(f"Service {service_type} not registered")
    return self._services[service_type]

validation_session(config) async

Async context manager that yields a TypedValidator.

Parameters:

Name	Type	Description	Default
config	ValidationConfig	ValidationConfig (reserved for future use).	required

Yields:

Type	Description
AsyncIterator[TypedValidator]	TypedValidator instance.

Source code in src/oversampleqa/typed_validator.py

@asynccontextmanager
async def validation_session(config: ValidationConfig) -> AsyncIterator[TypedValidator]:
    """Async context manager that yields a TypedValidator.

    Args:
        config: ValidationConfig (reserved for future use).

    Yields:
        TypedValidator instance.
    """
    validator = TypedValidator()
    try:
        yield validator
    finally:
        # No `return` here. A bare return inside `finally` swallows whatever
        # exception was in flight, so a failure inside the session body
        # disappeared silently and the caller saw a clean exit.
        logger.debug("validation_session closed")