Estimators

matgpr.estimators

MissingValueReport dataclass

Summary of estimator-level missing-value handling.

The fit-time report is attached to fitted estimators as missing_report_. Prediction-time reports are returned by summarize_prediction_missing_values without mutating estimator state. The report records the selected policy, row counts, feature-level missing counts, and any imputation settings used by the estimator.

Source code in matgpr/estimators.py

@dataclass(frozen=True)
class MissingValueReport:
    """Summary of estimator-level missing-value handling.

    The fit-time report is attached to fitted estimators as
    ``missing_report_``. Prediction-time reports are returned by
    ``summarize_prediction_missing_values`` without mutating estimator state.
    The report records the selected policy, row counts, feature-level missing
    counts, and any imputation settings used by the estimator.
    """

    stage: str
    policy: str
    input_rows: int
    output_rows: int
    dropped_rows: int
    rows_with_missing_features: int
    rows_with_missing_target: int
    feature_missing_counts: dict[str, int]
    imputed_features: tuple[str, ...] = ()
    imputation_strategy: str | None = None

    def to_dict(self) -> dict[str, object]:
        """Return the report as a plain dictionary for notebook display."""
        return asdict(self)

to_dict()

Return the report as a plain dictionary for notebook display.

Source code in matgpr/estimators.py

def to_dict(self) -> dict[str, object]:
    """Return the report as a plain dictionary for notebook display."""
    return asdict(self)

MatGPRRegressor

Bases: RegressorMixin, BaseEstimator

Scikit-learn-style wrapper around exact GPyTorch GPR.

This estimator exposes the familiar fit, predict, score, get_params, and set_params interface while reusing the lower-level :func:matgpr.fit_gpytorch_gpr implementation. It is intended for notebook workflows, cross-validation utilities, and users who prefer a standard estimator object over separate train/predict helper functions.

Source code in matgpr/estimators.py

class MatGPRRegressor(RegressorMixin, BaseEstimator):
    """Scikit-learn-style wrapper around exact GPyTorch GPR.

    This estimator exposes the familiar ``fit``, ``predict``, ``score``,
    ``get_params``, and ``set_params`` interface while reusing the lower-level
    :func:`matgpr.fit_gpytorch_gpr` implementation. It is intended for
    notebook workflows, cross-validation utilities, and users who prefer a
    standard estimator object over separate train/predict helper functions.
    """

    def __init__(
        self,
        *,
        kernel: str = "matern",
        ard: bool = True,
        lr: float = 0.01,
        training_iter: int = 1000,
        initial_noise: float | None = 0.1,
        standardize_y: bool = True,
        missing: str = "error",
        imputation_strategy: str = "median",
        imputation_fill_value: object | None = None,
        device: str = "cpu",
        dtype: str | torch.dtype = "float64",
        verbose: bool = False,
        log_every: int = 100,
        include_observation_noise: bool = True,
        random_state: int | None = None,
    ):
        self.kernel = kernel
        self.ard = ard
        self.lr = lr
        self.training_iter = training_iter
        self.initial_noise = initial_noise
        self.standardize_y = standardize_y
        self.missing = missing
        self.imputation_strategy = imputation_strategy
        self.imputation_fill_value = imputation_fill_value
        self.device = device
        self.dtype = dtype
        self.verbose = verbose
        self.log_every = log_every
        self.include_observation_noise = include_observation_noise
        self.random_state = random_state

    def fit(self, X, y):
        """Fit the Gaussian Process Regressor.

        Parameters
        ----------
        X
            Numeric feature matrix. Dataframes are accepted; string column
            names are stored in ``feature_names_in_`` for prediction-time
            consistency checks.
        y
            One-dimensional target values.

        Notes
        -----
        Missing-value handling is controlled by ``missing``. ``"error"``
        rejects missing data, ``"drop"`` removes incomplete training rows, and
        ``"impute"`` learns a numeric feature imputer from the training data.
        Missing targets are never imputed; they are either rejected or dropped.
        """
        X_checked, y_checked = _validate_fit_input(self, X, y)

        _seed_torch(self.random_state)
        dtype = _resolve_torch_dtype(self.dtype)
        mean_module = self._build_mean_module(X_checked=X_checked, y_checked=y_checked)

        self.result_ = fit_gpytorch_gpr(
            X_checked,
            y_checked,
            kernel=self.kernel,
            ard=self.ard,
            mean_module=mean_module,
            lr=self.lr,
            training_iter=self.training_iter,
            initial_noise=self.initial_noise,
            standardize_y=self.standardize_y,
            device=self.device,
            dtype=dtype,
            verbose=self.verbose,
            log_every=self.log_every,
        )
        self.model_ = self.result_.model
        self.likelihood_ = self.result_.likelihood
        self.loss_history_ = list(self.result_.loss_history)
        self.target_mean_ = self.result_.target_mean
        self.target_std_ = self.result_.target_std
        self.mean_module_ = self.result_.model.mean_module
        self._after_fit()
        return self

    def predict(
        self,
        X,
        *,
        return_std: bool = False,
        include_observation_noise: bool | None = None,
    ):
        """Predict target values for new samples.

        Parameters
        ----------
        X
            Numeric feature matrix with the same columns used during fitting.
        return_std
            If ``True``, return ``(mean, std)`` like scikit-learn's
            ``GaussianProcessRegressor``.
        include_observation_noise
            If ``True``, predictive uncertainty includes fitted observation
            noise. If omitted, the estimator-level setting is used.
        """
        prediction = self.predict_distribution(
            X,
            return_std=return_std,
            include_observation_noise=include_observation_noise,
        )
        if return_std:
            return prediction.mean, prediction.std
        return prediction.mean

    def predict_distribution(
        self,
        X,
        *,
        return_std: bool = True,
        confidence_level: float | None = None,
        include_observation_noise: bool | None = None,
    ) -> GPyTorchPrediction:
        """Return predictive mean, uncertainty, and optional confidence bounds."""
        check_is_fitted(self, "result_")
        X_checked = self._validate_prediction_input(X)
        if include_observation_noise is None:
            include_observation_noise = self.include_observation_noise

        return self.result_.predict(
            X_checked,
            return_std=return_std,
            confidence_level=confidence_level,
            include_observation_noise=include_observation_noise,
        )

    def predict_interval(
        self,
        X,
        *,
        confidence_level: float = 0.95,
        include_observation_noise: bool | None = None,
    ) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
        """Return predictive mean and confidence interval bounds."""
        prediction = self.predict_distribution(
            X,
            confidence_level=confidence_level,
            include_observation_noise=include_observation_noise,
        )
        return prediction.mean, prediction.lower, prediction.upper

    def summarize_prediction_missing_values(self, X) -> MissingValueReport:
        """Summarize missing values in prediction features without predicting.

        This method is useful for auditing candidate pools before calling
        :meth:`predict`. It does not mutate the estimator, which keeps
        prediction calls compatible with scikit-learn conventions.
        """
        check_is_fitted(self, "result_")
        policy = _validate_missing_policy(self.missing)
        X_checked = _validate_prediction_features(self, X, allow_nan=True)
        return _make_missing_report(
            self,
            stage="predict",
            policy=policy,
            X_checked=X_checked,
            y_checked=None,
            output_rows=X_checked.shape[0],
            imputed_features=_imputed_feature_names(self, X_checked) if policy == "impute" else (),
            imputation_strategy=_validate_imputation_strategy(self.imputation_strategy)
            if policy == "impute"
            else None,
        )

    def score(self, X, y, sample_weight=None) -> float:
        """Return the coefficient of determination, R2."""
        return r2_score(y, self.predict(X), sample_weight=sample_weight)

    def _build_mean_module(self, *, X_checked: np.ndarray, y_checked: np.ndarray):
        return None

    def _after_fit(self) -> None:
        return None

    def _validate_prediction_input(self, X) -> np.ndarray:
        return _validate_predict_input(self, X)

fit(X, y)

Fit the Gaussian Process Regressor.

Parameters:

Name	Type	Description	Default
X		Numeric feature matrix. Dataframes are accepted; string column names are stored in feature_names_in_ for prediction-time consistency checks.	required
y		One-dimensional target values.	required

Notes

Missing-value handling is controlled by missing. "error" rejects missing data, "drop" removes incomplete training rows, and "impute" learns a numeric feature imputer from the training data. Missing targets are never imputed; they are either rejected or dropped.

Source code in matgpr/estimators.py

def fit(self, X, y):
    """Fit the Gaussian Process Regressor.

    Parameters
    ----------
    X
        Numeric feature matrix. Dataframes are accepted; string column
        names are stored in ``feature_names_in_`` for prediction-time
        consistency checks.
    y
        One-dimensional target values.

    Notes
    -----
    Missing-value handling is controlled by ``missing``. ``"error"``
    rejects missing data, ``"drop"`` removes incomplete training rows, and
    ``"impute"`` learns a numeric feature imputer from the training data.
    Missing targets are never imputed; they are either rejected or dropped.
    """
    X_checked, y_checked = _validate_fit_input(self, X, y)

    _seed_torch(self.random_state)
    dtype = _resolve_torch_dtype(self.dtype)
    mean_module = self._build_mean_module(X_checked=X_checked, y_checked=y_checked)

    self.result_ = fit_gpytorch_gpr(
        X_checked,
        y_checked,
        kernel=self.kernel,
        ard=self.ard,
        mean_module=mean_module,
        lr=self.lr,
        training_iter=self.training_iter,
        initial_noise=self.initial_noise,
        standardize_y=self.standardize_y,
        device=self.device,
        dtype=dtype,
        verbose=self.verbose,
        log_every=self.log_every,
    )
    self.model_ = self.result_.model
    self.likelihood_ = self.result_.likelihood
    self.loss_history_ = list(self.result_.loss_history)
    self.target_mean_ = self.result_.target_mean
    self.target_std_ = self.result_.target_std
    self.mean_module_ = self.result_.model.mean_module
    self._after_fit()
    return self

predict(X, *, return_std=False, include_observation_noise=None)

Predict target values for new samples.

Parameters:

Name	Type	Description	Default
X		Numeric feature matrix with the same columns used during fitting.	required
return_std	bool	If True, return (mean, std) like scikit-learn's GaussianProcessRegressor.	False
include_observation_noise	bool | None	If True, predictive uncertainty includes fitted observation noise. If omitted, the estimator-level setting is used.	None

Source code in matgpr/estimators.py

def predict(
    self,
    X,
    *,
    return_std: bool = False,
    include_observation_noise: bool | None = None,
):
    """Predict target values for new samples.

    Parameters
    ----------
    X
        Numeric feature matrix with the same columns used during fitting.
    return_std
        If ``True``, return ``(mean, std)`` like scikit-learn's
        ``GaussianProcessRegressor``.
    include_observation_noise
        If ``True``, predictive uncertainty includes fitted observation
        noise. If omitted, the estimator-level setting is used.
    """
    prediction = self.predict_distribution(
        X,
        return_std=return_std,
        include_observation_noise=include_observation_noise,
    )
    if return_std:
        return prediction.mean, prediction.std
    return prediction.mean

predict_distribution(X, *, return_std=True, confidence_level=None, include_observation_noise=None)

Return predictive mean, uncertainty, and optional confidence bounds.

Source code in matgpr/estimators.py

def predict_distribution(
    self,
    X,
    *,
    return_std: bool = True,
    confidence_level: float | None = None,
    include_observation_noise: bool | None = None,
) -> GPyTorchPrediction:
    """Return predictive mean, uncertainty, and optional confidence bounds."""
    check_is_fitted(self, "result_")
    X_checked = self._validate_prediction_input(X)
    if include_observation_noise is None:
        include_observation_noise = self.include_observation_noise

    return self.result_.predict(
        X_checked,
        return_std=return_std,
        confidence_level=confidence_level,
        include_observation_noise=include_observation_noise,
    )

predict_interval(X, *, confidence_level=0.95, include_observation_noise=None)

Return predictive mean and confidence interval bounds.

Source code in matgpr/estimators.py

def predict_interval(
    self,
    X,
    *,
    confidence_level: float = 0.95,
    include_observation_noise: bool | None = None,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Return predictive mean and confidence interval bounds."""
    prediction = self.predict_distribution(
        X,
        confidence_level=confidence_level,
        include_observation_noise=include_observation_noise,
    )
    return prediction.mean, prediction.lower, prediction.upper

summarize_prediction_missing_values(X)

Summarize missing values in prediction features without predicting.

This method is useful for auditing candidate pools before calling :meth:predict. It does not mutate the estimator, which keeps prediction calls compatible with scikit-learn conventions.

Source code in matgpr/estimators.py

def summarize_prediction_missing_values(self, X) -> MissingValueReport:
    """Summarize missing values in prediction features without predicting.

    This method is useful for auditing candidate pools before calling
    :meth:`predict`. It does not mutate the estimator, which keeps
    prediction calls compatible with scikit-learn conventions.
    """
    check_is_fitted(self, "result_")
    policy = _validate_missing_policy(self.missing)
    X_checked = _validate_prediction_features(self, X, allow_nan=True)
    return _make_missing_report(
        self,
        stage="predict",
        policy=policy,
        X_checked=X_checked,
        y_checked=None,
        output_rows=X_checked.shape[0],
        imputed_features=_imputed_feature_names(self, X_checked) if policy == "impute" else (),
        imputation_strategy=_validate_imputation_strategy(self.imputation_strategy)
        if policy == "impute"
        else None,
    )

score(X, y, sample_weight=None)

Return the coefficient of determination, R2.

Source code in matgpr/estimators.py

def score(self, X, y, sample_weight=None) -> float:
    """Return the coefficient of determination, R2."""
    return r2_score(y, self.predict(X), sample_weight=sample_weight)

PhysicsInformedGPRRegressor

Bases: MatGPRRegressor

Scikit-learn-style physics-informed Gaussian Process Regressor.

Physics enters through a :class:matgpr.PhysicsInformedMean module. The GP then learns the residual between the physics equation and the data.

Source code in matgpr/estimators.py

class PhysicsInformedGPRRegressor(MatGPRRegressor):
    """Scikit-learn-style physics-informed Gaussian Process Regressor.

    Physics enters through a :class:`matgpr.PhysicsInformedMean` module. The GP
    then learns the residual between the physics equation and the data.
    """

    def __init__(
        self,
        *,
        equation: PhysicsEquation | None = None,
        feature_indices: Mapping[str, int] | None = None,
        physics_features: Sequence[str] | None = None,
        learnable_parameters: Mapping[str, float] | None = None,
        positive_parameters: Sequence[str] = (),
        fixed_parameters: Mapping[str, float] | None = None,
        feature_means: Mapping[str, float] | None = None,
        feature_stds: Mapping[str, float] | None = None,
        kernel: str = "matern",
        ard: bool = True,
        lr: float = 0.01,
        training_iter: int = 1000,
        initial_noise: float | None = 0.1,
        standardize_y: bool = True,
        missing: str = "error",
        imputation_strategy: str = "median",
        imputation_fill_value: object | None = None,
        device: str = "cpu",
        dtype: str | torch.dtype = "float64",
        verbose: bool = False,
        log_every: int = 100,
        include_observation_noise: bool = True,
        random_state: int | None = None,
    ):
        super().__init__(
            kernel=kernel,
            ard=ard,
            lr=lr,
            training_iter=training_iter,
            initial_noise=initial_noise,
            standardize_y=standardize_y,
            missing=missing,
            imputation_strategy=imputation_strategy,
            imputation_fill_value=imputation_fill_value,
            device=device,
            dtype=dtype,
            verbose=verbose,
            log_every=log_every,
            include_observation_noise=include_observation_noise,
            random_state=random_state,
        )
        self.equation = equation
        self.feature_indices = feature_indices
        self.physics_features = physics_features
        self.learnable_parameters = learnable_parameters
        self.positive_parameters = positive_parameters
        self.fixed_parameters = fixed_parameters
        self.feature_means = feature_means
        self.feature_stds = feature_stds

    def _build_mean_module(self, *, X_checked: np.ndarray, y_checked: np.ndarray):
        if self.equation is None:
            raise ValueError("equation must be provided before fitting")

        feature_indices = self._resolve_physics_feature_indices()
        return PhysicsInformedMean(
            equation=self.equation,
            feature_indices=feature_indices,
            learnable_parameters=self.learnable_parameters,
            positive_parameters=self.positive_parameters,
            fixed_parameters=self.fixed_parameters,
            feature_means=self.feature_means,
            feature_stds=self.feature_stds,
        )

    def _after_fit(self) -> None:
        if hasattr(self.mean_module_, "current_parameter_values"):
            self.learned_physics_parameters_ = self.mean_module_.current_parameter_values()

    def _resolve_physics_feature_indices(self) -> dict[str, int]:
        if self.feature_indices is not None and self.physics_features is not None:
            raise ValueError("Use either feature_indices or physics_features, not both")

        if self.feature_indices is not None:
            return dict(self.feature_indices)

        if self.physics_features is None:
            return {}

        if not hasattr(self, "feature_names_in_"):
            raise ValueError("physics_features requires fitting with a dataframe that has columns")

        fitted_names = list(self.feature_names_in_)
        missing = [name for name in self.physics_features if name not in fitted_names]
        if missing:
            raise ValueError(f"physics_features were not found in X columns: {missing}")
        return {name: fitted_names.index(name) for name in self.physics_features}

MultitaskGPRRegressor

Bases: MatGPRRegressor

Scikit-learn-style wrapper around exact multitask GPyTorch GPR.

This estimator models complete multi-output target matrices with a shared input-space kernel and a learned task covariance. It is intended for materials datasets where each row has multiple related properties observed, such as strength and ductility for the same alloy or permeability and selectivity for the same polymer.

Source code in matgpr/estimators.py

class MultitaskGPRRegressor(MatGPRRegressor):
    """Scikit-learn-style wrapper around exact multitask GPyTorch GPR.

    This estimator models complete multi-output target matrices with a shared
    input-space kernel and a learned task covariance. It is intended for
    materials datasets where each row has multiple related properties observed,
    such as strength and ductility for the same alloy or permeability and
    selectivity for the same polymer.
    """

    def __init__(
        self,
        *,
        task_names: Sequence[str] | None = None,
        task_covar_rank: int = 1,
        kernel: str = "matern",
        ard: bool = True,
        lr: float = 0.01,
        training_iter: int = 1000,
        initial_noise: float | None = 0.1,
        initial_task_noises: Sequence[float] | None = None,
        standardize_y: bool = True,
        missing: str = "error",
        imputation_strategy: str = "median",
        imputation_fill_value: object | None = None,
        device: str = "cpu",
        dtype: str | torch.dtype = "float64",
        verbose: bool = False,
        log_every: int = 100,
        include_observation_noise: bool = True,
        random_state: int | None = None,
    ):
        super().__init__(
            kernel=kernel,
            ard=ard,
            lr=lr,
            training_iter=training_iter,
            initial_noise=initial_noise,
            standardize_y=standardize_y,
            missing=missing,
            imputation_strategy=imputation_strategy,
            imputation_fill_value=imputation_fill_value,
            device=device,
            dtype=dtype,
            verbose=verbose,
            log_every=log_every,
            include_observation_noise=include_observation_noise,
            random_state=random_state,
        )
        self.task_names = task_names
        self.task_covar_rank = task_covar_rank
        self.initial_task_noises = initial_task_noises

    def fit(self, X, y):
        """Fit the multitask Gaussian Process Regressor.

        Parameters
        ----------
        X
            Numeric feature matrix. Dataframes are accepted; string column
            names are stored in ``feature_names_in_`` for prediction-time
            consistency checks.
        y
            Two-dimensional numeric target matrix with shape
            ``(n_samples, n_tasks)``. Dataframes are accepted; string target
            columns are used as task names when ``task_names`` is not supplied.

        Notes
        -----
        This first multitask estimator expects complete target observations.
        Missing targets are rejected or row-dropped according to ``missing``;
        target values are never imputed.
        """
        X_checked, y_checked, task_names = _validate_multitask_fit_input(self, X, y)

        _seed_torch(self.random_state)
        dtype = _resolve_torch_dtype(self.dtype)

        self.result_ = fit_multitask_gpytorch_gpr(
            X_checked,
            y_checked,
            task_names=task_names,
            task_covar_rank=self.task_covar_rank,
            kernel=self.kernel,
            ard=self.ard,
            lr=self.lr,
            training_iter=self.training_iter,
            initial_noise=self.initial_noise,
            initial_task_noises=self.initial_task_noises,
            standardize_y=self.standardize_y,
            device=self.device,
            dtype=dtype,
            verbose=self.verbose,
            log_every=self.log_every,
        )
        self.model_ = self.result_.model
        self.likelihood_ = self.result_.likelihood
        self.loss_history_ = list(self.result_.loss_history)
        self.target_mean_ = self.result_.target_mean.copy()
        self.target_std_ = self.result_.target_std.copy()
        self.task_names_ = self.result_.task_names
        self.n_tasks_in_ = self.result_.num_tasks
        self.n_outputs_ = self.result_.num_tasks
        return self

    def predict_distribution(
        self,
        X,
        *,
        return_std: bool = True,
        confidence_level: float | None = None,
        include_observation_noise: bool | None = None,
    ) -> MultitaskGPyTorchPrediction:
        """Return per-task predictive means, uncertainties, and intervals."""
        check_is_fitted(self, "result_")
        X_checked = self._validate_prediction_input(X)
        if include_observation_noise is None:
            include_observation_noise = self.include_observation_noise

        return self.result_.predict(
            X_checked,
            return_std=return_std,
            confidence_level=confidence_level,
            include_observation_noise=include_observation_noise,
        )

fit(X, y)

Fit the multitask Gaussian Process Regressor.

Parameters:

Name	Type	Description	Default
X		Numeric feature matrix. Dataframes are accepted; string column names are stored in feature_names_in_ for prediction-time consistency checks.	required
y		Two-dimensional numeric target matrix with shape (n_samples, n_tasks). Dataframes are accepted; string target columns are used as task names when task_names is not supplied.	required

Notes

This first multitask estimator expects complete target observations. Missing targets are rejected or row-dropped according to missing; target values are never imputed.

Source code in matgpr/estimators.py

def fit(self, X, y):
    """Fit the multitask Gaussian Process Regressor.

    Parameters
    ----------
    X
        Numeric feature matrix. Dataframes are accepted; string column
        names are stored in ``feature_names_in_`` for prediction-time
        consistency checks.
    y
        Two-dimensional numeric target matrix with shape
        ``(n_samples, n_tasks)``. Dataframes are accepted; string target
        columns are used as task names when ``task_names`` is not supplied.

    Notes
    -----
    This first multitask estimator expects complete target observations.
    Missing targets are rejected or row-dropped according to ``missing``;
    target values are never imputed.
    """
    X_checked, y_checked, task_names = _validate_multitask_fit_input(self, X, y)

    _seed_torch(self.random_state)
    dtype = _resolve_torch_dtype(self.dtype)

    self.result_ = fit_multitask_gpytorch_gpr(
        X_checked,
        y_checked,
        task_names=task_names,
        task_covar_rank=self.task_covar_rank,
        kernel=self.kernel,
        ard=self.ard,
        lr=self.lr,
        training_iter=self.training_iter,
        initial_noise=self.initial_noise,
        initial_task_noises=self.initial_task_noises,
        standardize_y=self.standardize_y,
        device=self.device,
        dtype=dtype,
        verbose=self.verbose,
        log_every=self.log_every,
    )
    self.model_ = self.result_.model
    self.likelihood_ = self.result_.likelihood
    self.loss_history_ = list(self.result_.loss_history)
    self.target_mean_ = self.result_.target_mean.copy()
    self.target_std_ = self.result_.target_std.copy()
    self.task_names_ = self.result_.task_names
    self.n_tasks_in_ = self.result_.num_tasks
    self.n_outputs_ = self.result_.num_tasks
    return self

predict_distribution(X, *, return_std=True, confidence_level=None, include_observation_noise=None)

Return per-task predictive means, uncertainties, and intervals.

Source code in matgpr/estimators.py

def predict_distribution(
    self,
    X,
    *,
    return_std: bool = True,
    confidence_level: float | None = None,
    include_observation_noise: bool | None = None,
) -> MultitaskGPyTorchPrediction:
    """Return per-task predictive means, uncertainties, and intervals."""
    check_is_fitted(self, "result_")
    X_checked = self._validate_prediction_input(X)
    if include_observation_noise is None:
        include_observation_noise = self.include_observation_noise

    return self.result_.predict(
        X_checked,
        return_std=return_std,
        confidence_level=confidence_level,
        include_observation_noise=include_observation_noise,
    )

SparseMultitaskGPRRegressor

Bases: MatGPRRegressor

Scikit-learn-style sparse multitask GPyTorch GPR.

This estimator accepts a multitask target matrix with NaN entries for unobserved task values. Each finite target value becomes one observed (sample, task) training pair, so the model can learn from incomplete materials-property matrices without discarding partially observed rows.

Source code in matgpr/estimators.py

class SparseMultitaskGPRRegressor(MatGPRRegressor):
    """Scikit-learn-style sparse multitask GPyTorch GPR.

    This estimator accepts a multitask target matrix with ``NaN`` entries for
    unobserved task values. Each finite target value becomes one observed
    ``(sample, task)`` training pair, so the model can learn from incomplete
    materials-property matrices without discarding partially observed rows.
    """

    def __init__(
        self,
        *,
        task_names: Sequence[str] | None = None,
        task_covar_rank: int = 1,
        kernel: str = "matern",
        ard: bool = True,
        lr: float = 0.01,
        training_iter: int = 1000,
        initial_noise: float | None = 0.1,
        initial_task_noises: Sequence[float] | Mapping[str, float] | None = None,
        known_noise_variance=None,
        noise_mode: str = "shared",
        noise_lower_bound: float = 1e-4,
        standardize_y: bool = True,
        min_observations_per_task: int = 2,
        missing: str = "error",
        imputation_strategy: str = "median",
        imputation_fill_value: object | None = None,
        device: str = "cpu",
        dtype: str | torch.dtype = "float64",
        verbose: bool = False,
        log_every: int = 100,
        include_observation_noise: bool = True,
        random_state: int | None = None,
    ):
        super().__init__(
            kernel=kernel,
            ard=ard,
            lr=lr,
            training_iter=training_iter,
            initial_noise=initial_noise,
            standardize_y=standardize_y,
            missing=missing,
            imputation_strategy=imputation_strategy,
            imputation_fill_value=imputation_fill_value,
            device=device,
            dtype=dtype,
            verbose=verbose,
            log_every=log_every,
            include_observation_noise=include_observation_noise,
            random_state=random_state,
        )
        self.task_names = task_names
        self.task_covar_rank = task_covar_rank
        self.initial_task_noises = initial_task_noises
        self.known_noise_variance = known_noise_variance
        self.noise_mode = noise_mode
        self.noise_lower_bound = noise_lower_bound
        self.min_observations_per_task = min_observations_per_task

    def fit(self, X, y):
        """Fit sparse multitask GPR from a target matrix with ``NaN`` gaps.

        ``NaN`` target values mean "task not observed" and are not imputed.
        Missing feature values still follow the estimator-level ``missing``
        policy: reject, drop feature-incomplete rows, or impute features.
        """
        X_checked, y_checked, task_names, known_noise_variance = (
            _validate_sparse_multitask_fit_input(
                self,
                X,
                y,
            )
        )

        _seed_torch(self.random_state)
        dtype = _resolve_torch_dtype(self.dtype)

        self.result_ = fit_sparse_multitask_gpytorch_gpr(
            X_checked,
            y_checked,
            task_names=task_names,
            task_covar_rank=self.task_covar_rank,
            kernel=self.kernel,
            ard=self.ard,
            lr=self.lr,
            training_iter=self.training_iter,
            initial_noise=self.initial_noise,
            initial_task_noises=self.initial_task_noises,
            known_noise_variance=known_noise_variance,
            noise_mode=self.noise_mode,
            noise_lower_bound=self.noise_lower_bound,
            standardize_y=self.standardize_y,
            min_observations_per_task=self.min_observations_per_task,
            device=self.device,
            dtype=dtype,
            verbose=self.verbose,
            log_every=self.log_every,
        )
        self.model_ = self.result_.model
        self.likelihood_ = self.result_.likelihood
        self.loss_history_ = list(self.result_.loss_history)
        self.target_mean_ = self.result_.target_mean.copy()
        self.target_std_ = self.result_.target_std.copy()
        self.noise_mode_ = self.result_.noise_mode
        self.standardized_task_noise_variance_ = (
            self.result_.standardized_task_noise_variance.copy()
        )
        self.task_noise_variance_ = self.result_.task_noise_variance.copy()
        self.task_noise_std_ = self.result_.task_noise_std.copy()
        self.standardized_observation_noise_variance_ = (
            None
            if self.result_.standardized_observation_noise_variance is None
            else self.result_.standardized_observation_noise_variance.copy()
        )
        self.observation_noise_variance_ = (
            None
            if self.result_.observation_noise_variance is None
            else self.result_.observation_noise_variance.copy()
        )
        self.task_names_ = self.result_.task_names
        self.n_tasks_in_ = self.result_.num_tasks
        self.n_outputs_ = self.result_.num_tasks
        self.observation_data_ = self.result_.observation_data
        self.observed_target_mask_ = self.observation_data_.observed_mask.copy()
        self.task_observation_counts_ = self.observation_data_.task_observation_counts.copy()
        return self

    def predict_distribution(
        self,
        X,
        *,
        return_std: bool = True,
        confidence_level: float | None = None,
        include_observation_noise: bool | None = None,
        prediction_noise_variance=None,
    ) -> MultitaskGPyTorchPrediction:
        """Return per-task predictive means, uncertainties, and intervals."""
        check_is_fitted(self, "result_")
        X_checked = self._validate_prediction_input(X)
        if include_observation_noise is None:
            include_observation_noise = self.include_observation_noise

        return self.result_.predict(
            X_checked,
            return_std=return_std,
            confidence_level=confidence_level,
            include_observation_noise=include_observation_noise,
            prediction_noise_variance=prediction_noise_variance,
        )

    def score(self, X, y, sample_weight=None) -> float:
        """Return R2 over observed target entries only."""
        y_true = np.asarray(y, dtype=float)
        if y_true.ndim != 2:
            raise ValueError("Sparse multitask score expects a 2D target matrix")
        y_pred = np.asarray(self.predict(X), dtype=float)
        if y_true.shape != y_pred.shape:
            raise ValueError("y and predictions must have the same shape")
        observed_mask = np.isfinite(y_true)
        if not observed_mask.any():
            raise ValueError("Cannot score sparse multitask data with no observed targets")

        entry_weights = None
        if sample_weight is not None:
            weights = np.asarray(sample_weight, dtype=float).reshape(-1)
            if weights.shape[0] != y_true.shape[0]:
                raise ValueError("sample_weight must contain one value per sample")
            entry_weights = np.broadcast_to(weights.reshape(-1, 1), y_true.shape)[observed_mask]

        return r2_score(
            y_true[observed_mask],
            y_pred[observed_mask],
            sample_weight=entry_weights,
        )

fit(X, y)

Fit sparse multitask GPR from a target matrix with NaN gaps.

NaN target values mean "task not observed" and are not imputed. Missing feature values still follow the estimator-level missing policy: reject, drop feature-incomplete rows, or impute features.

Source code in matgpr/estimators.py

def fit(self, X, y):
    """Fit sparse multitask GPR from a target matrix with ``NaN`` gaps.

    ``NaN`` target values mean "task not observed" and are not imputed.
    Missing feature values still follow the estimator-level ``missing``
    policy: reject, drop feature-incomplete rows, or impute features.
    """
    X_checked, y_checked, task_names, known_noise_variance = (
        _validate_sparse_multitask_fit_input(
            self,
            X,
            y,
        )
    )

    _seed_torch(self.random_state)
    dtype = _resolve_torch_dtype(self.dtype)

    self.result_ = fit_sparse_multitask_gpytorch_gpr(
        X_checked,
        y_checked,
        task_names=task_names,
        task_covar_rank=self.task_covar_rank,
        kernel=self.kernel,
        ard=self.ard,
        lr=self.lr,
        training_iter=self.training_iter,
        initial_noise=self.initial_noise,
        initial_task_noises=self.initial_task_noises,
        known_noise_variance=known_noise_variance,
        noise_mode=self.noise_mode,
        noise_lower_bound=self.noise_lower_bound,
        standardize_y=self.standardize_y,
        min_observations_per_task=self.min_observations_per_task,
        device=self.device,
        dtype=dtype,
        verbose=self.verbose,
        log_every=self.log_every,
    )
    self.model_ = self.result_.model
    self.likelihood_ = self.result_.likelihood
    self.loss_history_ = list(self.result_.loss_history)
    self.target_mean_ = self.result_.target_mean.copy()
    self.target_std_ = self.result_.target_std.copy()
    self.noise_mode_ = self.result_.noise_mode
    self.standardized_task_noise_variance_ = (
        self.result_.standardized_task_noise_variance.copy()
    )
    self.task_noise_variance_ = self.result_.task_noise_variance.copy()
    self.task_noise_std_ = self.result_.task_noise_std.copy()
    self.standardized_observation_noise_variance_ = (
        None
        if self.result_.standardized_observation_noise_variance is None
        else self.result_.standardized_observation_noise_variance.copy()
    )
    self.observation_noise_variance_ = (
        None
        if self.result_.observation_noise_variance is None
        else self.result_.observation_noise_variance.copy()
    )
    self.task_names_ = self.result_.task_names
    self.n_tasks_in_ = self.result_.num_tasks
    self.n_outputs_ = self.result_.num_tasks
    self.observation_data_ = self.result_.observation_data
    self.observed_target_mask_ = self.observation_data_.observed_mask.copy()
    self.task_observation_counts_ = self.observation_data_.task_observation_counts.copy()
    return self

predict_distribution(X, *, return_std=True, confidence_level=None, include_observation_noise=None, prediction_noise_variance=None)

Return per-task predictive means, uncertainties, and intervals.

Source code in matgpr/estimators.py

def predict_distribution(
    self,
    X,
    *,
    return_std: bool = True,
    confidence_level: float | None = None,
    include_observation_noise: bool | None = None,
    prediction_noise_variance=None,
) -> MultitaskGPyTorchPrediction:
    """Return per-task predictive means, uncertainties, and intervals."""
    check_is_fitted(self, "result_")
    X_checked = self._validate_prediction_input(X)
    if include_observation_noise is None:
        include_observation_noise = self.include_observation_noise

    return self.result_.predict(
        X_checked,
        return_std=return_std,
        confidence_level=confidence_level,
        include_observation_noise=include_observation_noise,
        prediction_noise_variance=prediction_noise_variance,
    )

score(X, y, sample_weight=None)

Return R2 over observed target entries only.

Source code in matgpr/estimators.py

def score(self, X, y, sample_weight=None) -> float:
    """Return R2 over observed target entries only."""
    y_true = np.asarray(y, dtype=float)
    if y_true.ndim != 2:
        raise ValueError("Sparse multitask score expects a 2D target matrix")
    y_pred = np.asarray(self.predict(X), dtype=float)
    if y_true.shape != y_pred.shape:
        raise ValueError("y and predictions must have the same shape")
    observed_mask = np.isfinite(y_true)
    if not observed_mask.any():
        raise ValueError("Cannot score sparse multitask data with no observed targets")

    entry_weights = None
    if sample_weight is not None:
        weights = np.asarray(sample_weight, dtype=float).reshape(-1)
        if weights.shape[0] != y_true.shape[0]:
            raise ValueError("sample_weight must contain one value per sample")
        entry_weights = np.broadcast_to(weights.reshape(-1, 1), y_true.shape)[observed_mask]

    return r2_score(
        y_true[observed_mask],
        y_pred[observed_mask],
        sample_weight=entry_weights,
    )