Skip to content

Data, Metrics, And Utilities

Data Cleaning

matgpr.data_cleaning

normalize_column_names(df)

Return a copy with consistent lowercase snake_case column names.

Use this as an early cleaning step so later functions can refer to stable feature names. Non-alphanumeric runs are converted to underscores, leading and trailing underscores are removed, and duplicate generated names receive a numeric suffix.

Source code in matgpr/data_cleaning.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def normalize_column_names(df: pd.DataFrame) -> pd.DataFrame:
    """Return a copy with consistent lowercase snake_case column names.

    Use this as an early cleaning step so later functions can refer to stable
    feature names. Non-alphanumeric runs are converted to underscores, leading
    and trailing underscores are removed, and duplicate generated names receive
    a numeric suffix.
    """
    out = df.copy()
    seen: dict[str, int] = {}
    columns: list[str] = []

    for column in out.columns:
        name = re.sub(r"[^0-9a-zA-Z]+", "_", str(column).strip().lower())
        name = re.sub(r"_+", "_", name).strip("_") or "column"

        count = seen.get(name, 0)
        seen[name] = count + 1
        columns.append(name if count == 0 else f"{name}_{count + 1}")

    out.columns = columns
    return out

replace_missing_placeholders(df, placeholders=('', 'nan', 'none', 'null', 'na', 'n/a', '-'))

Replace common string placeholders for missing values with np.nan.

Only object/string-like columns are inspected. Numeric columns are left unchanged so valid numeric zeros or negative values cannot be altered by accident.

Source code in matgpr/data_cleaning.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def replace_missing_placeholders(
    df: pd.DataFrame,
    placeholders: Iterable[str] = ("", "nan", "none", "null", "na", "n/a", "-"),
) -> pd.DataFrame:
    """Replace common string placeholders for missing values with ``np.nan``.

    Only object/string-like columns are inspected. Numeric columns are left
    unchanged so valid numeric zeros or negative values cannot be altered by
    accident.
    """
    out = df.copy()
    normalized = {str(value).strip().lower() for value in placeholders}

    for column in out.select_dtypes(include=["object", "string"]).columns:
        text = out[column].astype("string").str.strip().str.lower()
        mask = text.isin(normalized).fillna(False)
        out.loc[mask, column] = np.nan

    return out

drop_duplicate_rows(df, subset=None, *, reset_index=True)

Remove duplicate rows.

Parameters:

Name Type Description Default
df DataFrame

Input dataframe.

 required
subset Iterable[str] | None

Optional columns used to identify duplicates. When omitted, every column is considered.

 None
reset_index bool

If True, return a compact RangeIndex. Set False to preserve material IDs or other meaningful row indices.

 True
Source code in matgpr/data_cleaning.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def drop_duplicate_rows(
    df: pd.DataFrame,
    subset: Iterable[str] | None = None,
    *,
    reset_index: bool = True,
) -> pd.DataFrame:
    """Remove duplicate rows.

    Parameters
    ----------
    df
        Input dataframe.
    subset
        Optional columns used to identify duplicates. When omitted, every
        column is considered.
    reset_index
        If ``True``, return a compact ``RangeIndex``. Set ``False`` to preserve
        material IDs or other meaningful row indices.
    """
    subset_columns = None if subset is None else list(subset)
    out = df.drop_duplicates(subset=subset_columns)
    return out.reset_index(drop=True) if reset_index else out.copy()

drop_columns_by_missing_fraction(df, max_missing_fraction=0.5, *, return_dropped=False)

Drop columns whose missing fraction is above max_missing_fraction.

max_missing_fraction=0.5 keeps columns with up to 50 percent missing values and removes columns with more missing data than that.

Source code in matgpr/data_cleaning.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def drop_columns_by_missing_fraction(
    df: pd.DataFrame,
    max_missing_fraction: float = 0.5,
    *,
    return_dropped: bool = False,
) -> pd.DataFrame | tuple[pd.DataFrame, list[str]]:
    """Drop columns whose missing fraction is above ``max_missing_fraction``.

    ``max_missing_fraction=0.5`` keeps columns with up to 50 percent missing
    values and removes columns with more missing data than that.
    """
    if not 0 <= max_missing_fraction <= 1:
        raise ValueError("max_missing_fraction must be between 0 and 1")

    keep = df.isna().mean() <= max_missing_fraction
    cleaned = df.loc[:, keep].copy()
    if return_dropped:
        return cleaned, df.columns[~keep].tolist()
    return cleaned

impute_missing_values(df, *, strategy='median', columns=None, fill_value=None, return_imputer=False)

Impute missing values in selected dataframe columns.

Parameters:

Name Type Description Default
df DataFrame

Input dataframe.

 required
strategy str

Any sklearn.impute.SimpleImputer strategy: "mean", "median", "most_frequent", or "constant".

 'median'
columns Iterable[str] | None

Columns to impute. If omitted, numeric columns are selected.

 None
fill_value object | None

Replacement value used only when strategy="constant".

 None
return_imputer bool

If True, also return the fitted SimpleImputer so the same transformation can be applied to validation or test data.

 False
Source code in matgpr/data_cleaning.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def impute_missing_values(
    df: pd.DataFrame,
    *,
    strategy: str = "median",
    columns: Iterable[str] | None = None,
    fill_value: object | None = None,
    return_imputer: bool = False,
) -> pd.DataFrame | tuple[pd.DataFrame, SimpleImputer]:
    """Impute missing values in selected dataframe columns.

    Parameters
    ----------
    df
        Input dataframe.
    strategy
        Any ``sklearn.impute.SimpleImputer`` strategy: ``"mean"``,
        ``"median"``, ``"most_frequent"``, or ``"constant"``.
    columns
        Columns to impute. If omitted, numeric columns are selected.
    fill_value
        Replacement value used only when ``strategy="constant"``.
    return_imputer
        If ``True``, also return the fitted ``SimpleImputer`` so the same
        transformation can be applied to validation or test data.
    """
    out = df.copy()

    if columns is None:
        columns = out.select_dtypes(include="number").columns

    selected_columns = list(columns)
    if not selected_columns:
        raise ValueError("No columns selected for imputation")

    imputer = SimpleImputer(strategy=strategy, fill_value=fill_value)
    out[selected_columns] = imputer.fit_transform(out[selected_columns])
    if return_imputer:
        return out, imputer
    return out

filter_iqr_outliers(df, columns, *, factor=1.5, drop_if='any', reset_index=True)

Remove rows with IQR outliers in selected numeric columns.

Missing values are preserved by the filter. Use imputation before or after this step depending on whether missing values should influence the computed quartiles.

drop_if="any" keeps the historical behavior: a row is removed if it is an outlier in at least one selected column. drop_if="all" removes a row only when it is an outlier in every selected column.

Source code in matgpr/data_cleaning.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def filter_iqr_outliers(
    df: pd.DataFrame,
    columns: Iterable[str],
    *,
    factor: float = 1.5,
    drop_if: Literal["any", "all"] = "any",
    reset_index: bool = True,
) -> pd.DataFrame:
    """Remove rows with IQR outliers in selected numeric columns.

    Missing values are preserved by the filter. Use imputation before or after
    this step depending on whether missing values should influence the computed
    quartiles.

    ``drop_if="any"`` keeps the historical behavior: a row is removed if it is
    an outlier in at least one selected column. ``drop_if="all"`` removes a row
    only when it is an outlier in every selected column.
    """
    if factor <= 0:
        raise ValueError("factor must be positive")
    if drop_if not in {"any", "all"}:
        raise ValueError("drop_if must be either 'any' or 'all'")

    selected_columns = list(columns)
    if not selected_columns:
        raise ValueError("At least one column is required")

    outlier_masks: list[pd.Series] = []

    for column in selected_columns:
        q1 = df[column].quantile(0.25)
        q3 = df[column].quantile(0.75)
        iqr = q3 - q1
        if pd.isna(q1) or pd.isna(q3) or not np.isfinite(iqr) or iqr == 0:
            continue
        lower = q1 - factor * iqr
        upper = q3 + factor * iqr
        outlier_masks.append(~(df[column].between(lower, upper) | df[column].isna()))

    if not outlier_masks:
        return df.reset_index(drop=True) if reset_index else df.copy()

    outlier_table = pd.concat(outlier_masks, axis=1)
    if drop_if == "any":
        keep = ~outlier_table.any(axis=1)
    else:
        keep = ~outlier_table.all(axis=1)

    filtered = df.loc[keep]
    return filtered.reset_index(drop=True) if reset_index else filtered.copy()

Data Splitting

matgpr.data_splitting

separate_features_target(df, target_column)

Split a dataframe into feature columns X and target vector y.

This is the first modeling split: it separates the column to predict from the columns used as model inputs.

Source code in matgpr/data_splitting.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
def separate_features_target(
    df: pd.DataFrame,
    target_column: str,
) -> tuple[pd.DataFrame, pd.Series]:
    """Split a dataframe into feature columns ``X`` and target vector ``y``.

    This is the first modeling split: it separates the column to predict from
    the columns used as model inputs.
    """
    if target_column not in df.columns:
        raise KeyError(f"target_column '{target_column}' not found in dataframe")

    return df.drop(columns=[target_column]), df[target_column]

split_train_test(X, y, *, test_size=0.2, random_state=42, shuffle=True, stratify=None)

Split features and targets into train/test sets.

Parameters mirror sklearn.model_selection.train_test_split so the returned values are X_train, X_test, y_train, y_test.

Source code in matgpr/data_splitting.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def split_train_test(
    X,
    y,
    *,
    test_size: float = 0.2,
    random_state: int = 42,
    shuffle: bool = True,
    stratify=None,
):
    """Split features and targets into train/test sets.

    Parameters mirror ``sklearn.model_selection.train_test_split`` so the
    returned values are ``X_train, X_test, y_train, y_test``.
    """
    return train_test_split(
        X,
        y,
        test_size=test_size,
        random_state=random_state,
        shuffle=shuffle,
        stratify=stratify,
    )

Preprocessing

matgpr.preprocessing

identify_feature_types(X)

Identify numeric and categorical feature columns in a dataframe.

Returns:

Type Description
(numeric_features, categorical_features)

Lists that can be passed directly into build_preprocessor.
Source code in matgpr/preprocessing.py 
18
19
20
21
22
23
24
25
26
27
28
def identify_feature_types(X: pd.DataFrame) -> tuple[list[str], list[str]]:
    """Identify numeric and categorical feature columns in a dataframe.

    Returns
    -------
    numeric_features, categorical_features
        Lists that can be passed directly into ``build_preprocessor``.
    """
    numeric_features = X.select_dtypes(include="number").columns.tolist()
    categorical_features = [column for column in X.columns if column not in numeric_features]
    return numeric_features, categorical_features

build_scaler(name='standard')

Create a scikit-learn feature scaler by name.

Parameters:

Name Type Description Default
name str

"standard" for zero-mean/unit-variance scaling, "minmax" for range scaling, "robust" for median/IQR scaling, or "none" / "passthrough" to leave features unchanged in a pipeline.

 'standard'
Source code in matgpr/preprocessing.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def build_scaler(name: str = "standard"):
    """Create a scikit-learn feature scaler by name.

    Parameters
    ----------
    name
        ``"standard"`` for zero-mean/unit-variance scaling, ``"minmax"`` for
        range scaling, ``"robust"`` for median/IQR scaling, or ``"none"`` /
        ``"passthrough"`` to leave features unchanged in a pipeline.
    """
    normalized = name.lower()

    if normalized == "standard":
        return StandardScaler()
    if normalized == "minmax":
        return MinMaxScaler()
    if normalized == "robust":
        return RobustScaler()
    if normalized in {"none", "passthrough"}:
        return "passthrough"

    raise ValueError("name must be one of: standard, minmax, robust, none")

build_preprocessor(*, numeric_features=None, categorical_features=None, scaler='standard', numeric_imputation='median', categorical_imputation='most_frequent')

Build a reusable scikit-learn preprocessing transformer.

Fit this transformer only on training data, then reuse it for validation, test, or prediction data. Numeric columns are imputed and optionally scaled; categorical columns are imputed and one-hot encoded.

Source code in matgpr/preprocessing.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def build_preprocessor(
    *,
    numeric_features: Iterable[str] | None = None,
    categorical_features: Iterable[str] | None = None,
    scaler: str = "standard",
    numeric_imputation: str = "median",
    categorical_imputation: str = "most_frequent",
) -> ColumnTransformer:
    """Build a reusable scikit-learn preprocessing transformer.

    Fit this transformer only on training data, then reuse it for validation,
    test, or prediction data. Numeric columns are imputed and optionally scaled;
    categorical columns are imputed and one-hot encoded.
    """
    transformers = []
    numeric_features = list(numeric_features or [])
    categorical_features = list(categorical_features or [])

    if numeric_features:
        numeric_pipeline = Pipeline(
            steps=[
                ("imputer", SimpleImputer(strategy=numeric_imputation)),
                ("scaler", build_scaler(scaler)),
            ]
        )
        transformers.append(("numeric", numeric_pipeline, numeric_features))

    if categorical_features:
        categorical_pipeline = Pipeline(
            steps=[
                ("imputer", SimpleImputer(strategy=categorical_imputation)),
                ("onehot", OneHotEncoder(handle_unknown="ignore", sparse_output=False)),
            ]
        )
        transformers.append(("categorical", categorical_pipeline, categorical_features))

    if not transformers:
        raise ValueError("Provide at least one numeric or categorical feature")

    return ColumnTransformer(transformers=transformers, remainder="drop")

Metrics

The numerical metric helpers keep scientifically undefined values as NaN. Use the JSON-safe companion helpers when metrics must be serialized through a strict JSON encoder or API response.

matgpr.metrics

regression_metrics(y_true, y_pred, *, prefix='')

Calculate common regression metrics for model analysis.

The returned dictionary contains R2, RMSE, MAE, and Pearson r. Use prefix to distinguish train/test metrics when storing experiment results, for example prefix="test" gives test_R2.

Source code in matgpr/metrics.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def regression_metrics(
    y_true,
    y_pred,
    *,
    prefix: str = "",
) -> dict[str, float]:
    """Calculate common regression metrics for model analysis.

    The returned dictionary contains R2, RMSE, MAE, and Pearson r. Use
    ``prefix`` to distinguish train/test metrics when storing experiment
    results, for example ``prefix="test"`` gives ``test_R2``.
    """
    y_true = _as_1d_float_array(y_true, "y_true")
    y_pred = _as_1d_float_array(y_pred, "y_pred")

    if y_true.shape[0] != y_pred.shape[0]:
        raise ValueError("y_true and y_pred must have the same length")

    rmse = float(np.sqrt(mean_squared_error(y_true, y_pred)))
    mae = float(mean_absolute_error(y_true, y_pred))
    r2 = _r2_score(y_true, y_pred)
    r = _pearson_r(y_true, y_pred)

    metric_prefix = f"{prefix}_" if prefix else ""
    return {
        f"{metric_prefix}R2": r2,
        f"{metric_prefix}RMSE": rmse,
        f"{metric_prefix}MAE": mae,
        f"{metric_prefix}r": r,
    }

train_test_regression_metrics(y_train_true, y_train_pred, y_test_true, y_test_pred)

Calculate regression metrics for both train and test predictions.

Source code in matgpr/metrics.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def train_test_regression_metrics(
    y_train_true,
    y_train_pred,
    y_test_true,
    y_test_pred,
) -> dict[str, float]:
    """Calculate regression metrics for both train and test predictions."""
    metrics = {}
    metrics.update(
        regression_metrics(
            y_train_true,
            y_train_pred,
            prefix="train",
        )
    )
    metrics.update(
        regression_metrics(
            y_test_true,
            y_test_pred,
            prefix="test",
        )
    )
    return metrics

json_safe_metrics(metrics, *, include_warnings=True, warning_key='metric_warnings')

Return a strict-JSON-safe copy of a metric dictionary.

Scientific metrics such as R2, Pearson r, or calibration correlations can be undefined for legitimate edge cases, including one-sample splits or constant arrays. The numerical metric helpers keep those values as NaN. This helper converts non-finite scalar values to None so the result can be serialized with json.dumps(..., allow_nan=False).

Source code in matgpr/metrics.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def json_safe_metrics(
    metrics: Mapping[str, Any],
    *,
    include_warnings: bool = True,
    warning_key: str = "metric_warnings",
) -> dict[str, Any]:
    """Return a strict-JSON-safe copy of a metric dictionary.

    Scientific metrics such as R2, Pearson r, or calibration correlations can
    be undefined for legitimate edge cases, including one-sample splits or
    constant arrays. The numerical metric helpers keep those values as
    ``NaN``. This helper converts non-finite scalar values to ``None`` so the
    result can be serialized with ``json.dumps(..., allow_nan=False)``.
    """
    if include_warnings and warning_key in metrics:
        raise ValueError(f"warning_key {warning_key!r} conflicts with a metric key")

    safe: dict[str, Any] = {}
    warnings: list[str] = []
    for key, value in metrics.items():
        safe_value, value_warnings = _to_json_safe_value(value, path=str(key))
        safe[str(key)] = safe_value
        warnings.extend(value_warnings)

    if include_warnings and warnings:
        safe[warning_key] = warnings
    return safe

json_safe_regression_metrics(y_true, y_pred, *, prefix='', include_warnings=True, warning_key='metric_warnings')

Calculate regression metrics and return a strict-JSON-safe dictionary.

Source code in matgpr/metrics.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def json_safe_regression_metrics(
    y_true,
    y_pred,
    *,
    prefix: str = "",
    include_warnings: bool = True,
    warning_key: str = "metric_warnings",
) -> dict[str, Any]:
    """Calculate regression metrics and return a strict-JSON-safe dictionary."""
    return json_safe_metrics(
        regression_metrics(y_true, y_pred, prefix=prefix),
        include_warnings=include_warnings,
        warning_key=warning_key,
    )

json_safe_train_test_regression_metrics(y_train_true, y_train_pred, y_test_true, y_test_pred, *, include_warnings=True, warning_key='metric_warnings')

Calculate train/test regression metrics with non-finite values as None.

Source code in matgpr/metrics.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def json_safe_train_test_regression_metrics(
    y_train_true,
    y_train_pred,
    y_test_true,
    y_test_pred,
    *,
    include_warnings: bool = True,
    warning_key: str = "metric_warnings",
) -> dict[str, Any]:
    """Calculate train/test regression metrics with non-finite values as ``None``."""
    return json_safe_metrics(
        train_test_regression_metrics(
            y_train_true,
            y_train_pred,
            y_test_true,
            y_test_pred,
        ),
        include_warnings=include_warnings,
        warning_key=warning_key,
    )

PCA

matgpr.pca

fit_pca(X, n_components=None, *, scale=False)

Fit PCA and return component scores, the fitted PCA, and optional scaler.

DataFrames are reduced to numeric columns before PCA. Set scale=True to fit a StandardScaler before PCA; the fitted scaler is returned so new data can be transformed consistently.

Source code in matgpr/pca.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def fit_pca(
    X,
    n_components: int | float | None = None,
    *,
    scale: bool = False,
) -> tuple[pd.DataFrame, PCA, StandardScaler | None]:
    """Fit PCA and return component scores, the fitted PCA, and optional scaler.

    DataFrames are reduced to numeric columns before PCA. Set ``scale=True`` to
    fit a ``StandardScaler`` before PCA; the fitted scaler is returned so new
    data can be transformed consistently.
    """
    X_values, feature_names = _numeric_values(X, context="PCA")
    scaler = StandardScaler() if scale else None

    if scaler is not None:
        X_values = scaler.fit_transform(X_values)

    pca = PCA(n_components=n_components)
    scores = pca.fit_transform(X_values)
    if feature_names is not None:
        # Recorded so ``transform_pca`` can reject new data whose numeric
        # columns differ from training, including a different column order.
        # Deliberately not scikit-learn's ``feature_names_in_``: that would make
        # ``pca.transform`` warn for the plain-array inputs this module accepts.
        pca.matgpr_feature_names_in_ = tuple(feature_names)
    return _scores_dataframe(scores), pca, scaler

summarize_pca(pca)

Return explained and cumulative variance for a fitted PCA model.

Source code in matgpr/pca.py 
44
45
46
47
48
49
50
51
52
53
def summarize_pca(pca: PCA) -> pd.DataFrame:
    """Return explained and cumulative variance for a fitted PCA model."""
    explained = pca.explained_variance_ratio_
    return pd.DataFrame(
        {
            "component": [f"PC{i + 1}" for i in range(len(explained))],
            "explained_variance_ratio": explained,
            "cumulative_variance_ratio": np.cumsum(explained),
        }
    )

transform_pca(X, pca, *, scaler=None, imputer=None)

Transform new data using an already-fitted PCA model.

Pass the same scaler and imputer used for training data so new data follows the identical preprocessing path.

When pca was fitted on a dataframe, the numeric columns of X must match the training columns exactly, including their order. A mismatch raises rather than silently producing scores from misaligned features.

Source code in matgpr/pca.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def transform_pca(
    X,
    pca: PCA,
    *,
    scaler: StandardScaler | None = None,
    imputer=None,
) -> pd.DataFrame:
    """Transform new data using an already-fitted PCA model.

    Pass the same scaler and imputer used for training data so new data follows
    the identical preprocessing path.

    When ``pca`` was fitted on a dataframe, the numeric columns of ``X`` must
    match the training columns exactly, including their order. A mismatch
    raises rather than silently producing scores from misaligned features.
    """
    original_index = X.index if isinstance(X, pd.DataFrame) else None
    X_values, feature_names = _numeric_values(X, context="PCA transform")
    _validate_transform_features(pca, feature_names, X_values.shape[1])

    if imputer is not None:
        X_values = imputer.transform(X_values)
    if scaler is not None:
        X_values = scaler.transform(X_values)

    scores = pca.transform(X_values)
    return _scores_dataframe(scores, index=original_index)

Reporting

matgpr.reporting

summarize_missingness(df)

Return missing counts and fractions by column.

Source code in matgpr/reporting.py 
16
17
18
19
20
21
22
23
24
25
def summarize_missingness(df: pd.DataFrame) -> pd.DataFrame:
    """Return missing counts and fractions by column."""
    report = pd.DataFrame(
        {
            "missing_count": df.isna().sum(),
            "missing_fraction": df.isna().mean(),
            "dtype": df.dtypes.astype(str),
        }
    )
    return report.sort_values("missing_fraction", ascending=False)

summarize_numeric_columns(df)

Return descriptive statistics for numeric columns.

Source code in matgpr/reporting.py 
28
29
30
def summarize_numeric_columns(df: pd.DataFrame) -> pd.DataFrame:
    """Return descriptive statistics for numeric columns."""
    return df.describe().T

decompose_multifidelity_prediction(prediction, *, y_true=None, sample_labels=None, model_name='multi_fidelity_gpr', split='prediction')

Return a per-sample component table for a multi-fidelity prediction.

Parameters:

Name Type Description Default
prediction

Object with a mean attribute and optional multi-fidelity component attributes such as low_fidelity_mean, correction_mean, rho, intercept, std, low_fidelity_std, and correction_std. :class:matgpr.MultiFidelityGPRPrediction follows this convention.

 required
y_true

Optional measured high-fidelity values. When supplied, signed and absolute prediction errors are added.

 None
sample_labels Sequence[object] | None

Optional sample identifiers. If omitted, integer positions are used.

 None
model_name str

Labels stored in the returned dataframe for reporting and grouping.

 'multi_fidelity_gpr'
split str

Labels stored in the returned dataframe for reporting and grouping.

 'multi_fidelity_gpr'

Returns:

Type Description
DataFrame

Per-sample rows with the high-fidelity prediction and any available low-fidelity, correction, and uncertainty components.
Source code in matgpr/reporting.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def decompose_multifidelity_prediction(
    prediction,
    *,
    y_true=None,
    sample_labels: Sequence[object] | None = None,
    model_name: str = "multi_fidelity_gpr",
    split: str = "prediction",
) -> pd.DataFrame:
    """Return a per-sample component table for a multi-fidelity prediction.

    Parameters
    ----------
    prediction
        Object with a ``mean`` attribute and optional multi-fidelity component
        attributes such as ``low_fidelity_mean``, ``correction_mean``, ``rho``,
        ``intercept``, ``std``, ``low_fidelity_std``, and ``correction_std``.
        :class:`matgpr.MultiFidelityGPRPrediction` follows this convention.
    y_true
        Optional measured high-fidelity values. When supplied, signed and
        absolute prediction errors are added.
    sample_labels
        Optional sample identifiers. If omitted, integer positions are used.
    model_name, split
        Labels stored in the returned dataframe for reporting and grouping.

    Returns
    -------
    pandas.DataFrame
        Per-sample rows with the high-fidelity prediction and any available
        low-fidelity, correction, and uncertainty components.
    """
    if not hasattr(prediction, "mean"):
        raise ValueError("prediction must provide a mean attribute")
    y_pred = _to_1d_numeric(getattr(prediction, "mean"), "prediction.mean")
    n_samples = y_pred.shape[0]
    labels = _resolve_sample_labels(sample_labels, n_samples)
    frame = pd.DataFrame(
        {
            "model": str(model_name),
            "split": str(split),
            "sample_position": np.arange(n_samples, dtype=int),
            "sample_label": labels,
            "y_pred": y_pred,
        }
    )

    y_true_array = _optional_1d_numeric(y_true, "y_true")
    if y_true_array is not None:
        _validate_same_length(y_true_array, n_samples, "y_true")
        frame["y_true"] = y_true_array

    for attribute, column in (
        ("std", "y_std"),
        ("lower", "y_lower"),
        ("upper", "y_upper"),
        ("low_fidelity_mean", "low_fidelity_pred"),
        ("low_fidelity_std", "low_fidelity_std"),
        ("scaled_low_fidelity_mean", "scaled_low_fidelity_pred"),
        ("scaled_low_fidelity_std", "scaled_low_fidelity_std"),
        ("correction_mean", "correction_pred"),
        ("correction_std", "correction_std"),
        ("discrepancy_mean", "discrepancy_pred"),
        ("discrepancy_std", "discrepancy_std"),
    ):
        values = _optional_prediction_attribute(prediction, attribute)
        if values is not None:
            array = _to_1d_numeric(values, f"prediction.{attribute}")
            _validate_same_length(array, n_samples, f"prediction.{attribute}")
            frame[column] = array

    rho = _optional_scalar_attribute(prediction, "rho")
    intercept = _optional_scalar_attribute(prediction, "intercept")
    if rho is not None:
        frame["rho"] = rho
    if intercept is not None:
        frame["intercept"] = intercept

    return _with_multifidelity_derived_columns(frame)

summarize_multifidelity_components(components, *, group_by=None)

Summarize multi-fidelity prediction components for reports.

components can be either a per-sample dataframe, such as the predictions table returned by multifidelity_learning_curve with store_predictions=True, or a prediction object accepted by :func:decompose_multifidelity_prediction.

The summary reports low-fidelity and correction contribution statistics, uncertainty component statistics, reconstruction residuals, and prediction error metrics when y_true is available.

Source code in matgpr/reporting.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def summarize_multifidelity_components(
    components,
    *,
    group_by: Sequence[str] | str | None = None,
) -> pd.DataFrame:
    """Summarize multi-fidelity prediction components for reports.

    ``components`` can be either a per-sample dataframe, such as the
    ``predictions`` table returned by ``multifidelity_learning_curve`` with
    ``store_predictions=True``, or a prediction object accepted by
    :func:`decompose_multifidelity_prediction`.

    The summary reports low-fidelity and correction contribution statistics,
    uncertainty component statistics, reconstruction residuals, and prediction
    error metrics when ``y_true`` is available.
    """
    frame = _component_frame_from_input(components)
    frame = _with_multifidelity_derived_columns(frame)
    group_columns = _resolve_group_columns(frame, group_by)
    rows = []

    for key_values, group in _iter_groups(frame, group_columns):
        row = dict(key_values)
        row["n_samples"] = int(group.shape[0])
        for column in (
            "y_true",
            "y_pred",
            "low_fidelity_input",
            "low_fidelity_pred",
            "scaled_low_fidelity_pred",
            "correction_pred",
            "discrepancy_pred",
            "reconstructed_y_pred",
            "y_std",
            "low_fidelity_std",
            "scaled_low_fidelity_std",
            "correction_std",
            "discrepancy_std",
            "low_fidelity_variance_fraction",
            "correction_variance_fraction",
            "discrepancy_variance_fraction",
        ):
            _add_numeric_column_summary(row, group, column)

        if "correction_pred" in group:
            row["mean_abs_correction_pred"] = _finite_abs_mean(group["correction_pred"])
        if "discrepancy_pred" in group:
            row["mean_abs_discrepancy_pred"] = _finite_abs_mean(group["discrepancy_pred"])
        if "scaled_low_fidelity_pred" in group:
            row["mean_abs_scaled_low_fidelity_pred"] = _finite_abs_mean(
                group["scaled_low_fidelity_pred"]
            )
        if "component_residual" in group:
            row["max_abs_component_residual"] = _finite_abs_max(group["component_residual"])
        if "signed_error" in group:
            signed_error = _finite_array(group["signed_error"])
            if signed_error.size:
                row["bias"] = float(np.mean(signed_error))
                row["RMSE"] = float(np.sqrt(np.mean(signed_error**2)))
                row["MAE"] = float(np.mean(np.abs(signed_error)))
        rows.append(row)

    return pd.DataFrame(rows)

Artifact IO

log_experiment_result appends experiment rows to a schema-safe CSV log. When new runs add or omit metric columns, the existing file is realigned to the union of columns and remains readable by pandas.read_csv.

matgpr.io_utils

save_artifact(artifact, path)

Save a fitted preprocessor, model, or full pipeline with joblib.

Source code in matgpr/io_utils.py 
17
18
19
20
21
def save_artifact(artifact, path: str | Path) -> None:
    """Save a fitted preprocessor, model, or full pipeline with joblib."""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    joblib.dump(artifact, path)

load_artifact(path)

Load a joblib artifact saved by save_artifact.

Source code in matgpr/io_utils.py 
24
25
26
def load_artifact(path: str | Path):
    """Load a joblib artifact saved by ``save_artifact``."""
    return joblib.load(path)

log_experiment_result(metrics, *, metadata=None, path='results.csv')

Append one experiment result row to a schema-safe CSV file.

metadata is written before metric columns and is intended for values such as model name, kernel, train size, random state, or feature set.

If the CSV already exists, the file is read, aligned to the union of old and new columns, and rewritten. Existing column order is preserved and new columns are appended at the end. This keeps experiment logs readable when later runs add or omit metrics.

Source code in matgpr/io_utils.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def log_experiment_result(
    metrics: Mapping[str, Any],
    *,
    metadata: Mapping[str, Any] | None = None,
    path: str | Path = "results.csv",
) -> None:
    """Append one experiment result row to a schema-safe CSV file.

    ``metadata`` is written before metric columns and is intended for values
    such as model name, kernel, train size, random state, or feature set.

    If the CSV already exists, the file is read, aligned to the union of old
    and new columns, and rewritten. Existing column order is preserved and new
    columns are appended at the end. This keeps experiment logs readable when
    later runs add or omit metrics.
    """
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)

    row_data: dict[str, Any] = {}
    if metadata is not None:
        _update_row_data(row_data, metadata, source_name="metadata")
    _update_row_data(row_data, metrics, source_name="metrics")

    row = pd.DataFrame([row_data])
    if not path.exists() or path.stat().st_size == 0:
        row.to_csv(path, index=False)
        return

    try:
        existing = pd.read_csv(path)
    except pd.errors.EmptyDataError:
        row.to_csv(path, index=False)
        return

    columns = _merged_columns(existing.columns, row.columns)
    combined = pd.concat(
        [
            existing.reindex(columns=columns),
            row.reindex(columns=columns),
        ],
        ignore_index=True,
    )
    combined.to_csv(path, index=False)

Optional Dependencies

matgpr.optional_dependencies

OptionalDependency dataclass

Metadata for an optional backend dependency.

Parameters:

Name Type Description Default
import_name str

Python import path used to import the package.

 required
package_name str

Package name users should install.

 required
extra str

matgpr optional dependency extra that includes the package.

 required
purpose str

Short user-facing description of why the dependency is needed.

 required
Source code in matgpr/optional_dependencies.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@dataclass(frozen=True)
class OptionalDependency:
    """Metadata for an optional backend dependency.

    Parameters
    ----------
    import_name
        Python import path used to import the package.
    package_name
        Package name users should install.
    extra
        `matgpr` optional dependency extra that includes the package.
    purpose
        Short user-facing description of why the dependency is needed.
    """

    import_name: str
    package_name: str
    extra: str
    purpose: str

require_optional_dependency(dependency, *, purpose=None, extra=None, package_name=None)

Import an optional dependency or raise a clear installation message.

Parameters:

Name Type Description Default
dependency str | OptionalDependency

Registered dependency name, Python import name, or explicit OptionalDependency metadata.

 required
purpose str | None

Optional purpose override for custom wrappers.

 None
extra str | None

Optional matgpr extra override for custom wrappers.

 None
package_name str | None

Optional install package name override for custom wrappers.

 None
Source code in matgpr/optional_dependencies.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def require_optional_dependency(
    dependency: str | OptionalDependency,
    *,
    purpose: str | None = None,
    extra: str | None = None,
    package_name: str | None = None,
) -> ModuleType:
    """Import an optional dependency or raise a clear installation message.

    Parameters
    ----------
    dependency
        Registered dependency name, Python import name, or explicit
        `OptionalDependency` metadata.
    purpose
        Optional purpose override for custom wrappers.
    extra
        Optional `matgpr` extra override for custom wrappers.
    package_name
        Optional install package name override for custom wrappers.
    """
    metadata = _resolve_optional_dependency(
        dependency,
        purpose=purpose,
        extra=extra,
        package_name=package_name,
    )
    try:
        return importlib.import_module(metadata.import_name)
    except ImportError as exc:
        raise ImportError(_missing_dependency_message(metadata)) from exc

is_optional_dependency_available(dependency)

Return True when an optional dependency can be imported.

Source code in matgpr/optional_dependencies.py 
135
136
137
138
139
140
141
142
def is_optional_dependency_available(dependency: str | OptionalDependency) -> bool:
    """Return `True` when an optional dependency can be imported."""
    metadata = _resolve_optional_dependency(dependency)
    try:
        importlib.import_module(metadata.import_name)
    except ImportError:
        return False
    return True

list_optional_dependencies()

Return unique optional dependency metadata records.

Source code in matgpr/optional_dependencies.py 
145
146
147
148
149
150
def list_optional_dependencies() -> tuple[OptionalDependency, ...]:
    """Return unique optional dependency metadata records."""
    unique: dict[tuple[str, str], OptionalDependency] = {}
    for metadata in OPTIONAL_DEPENDENCIES.values():
        unique[(metadata.import_name, metadata.extra)] = metadata
    return tuple(unique.values())

Scikit-Learn GPR Helpers

matgpr.sklearn_gpr

build_sklearn_gpr_kernel(name='matern', *, n_features=None, constant_value=1.0, constant_value_bounds=(0.001, 1000.0), length_scale=1.0, length_scale_bounds=(1e-05, 100000.0), noise_level=1e-06, noise_level_bounds=(1e-10, 0.1), nu=2.5, metric=None, normalize=True, feature_scales=None, validate_nonnegative=True)

Create a common scikit-learn Gaussian-process kernel.

"ard_rbf" and "ard_matern" use one length scale per feature, so n_features is required for those options.

Source code in matgpr/sklearn_gpr.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def build_sklearn_gpr_kernel(
    name: str = "matern",
    *,
    n_features: int | None = None,
    constant_value: float = 1.0,
    constant_value_bounds: tuple[float, float] | str = (1e-3, 1e3),
    length_scale: float | np.ndarray = 1.0,
    length_scale_bounds: tuple[float, float] | str = (1e-5, 1e5),
    noise_level: float = 1e-6,
    noise_level_bounds: tuple[float, float] | str = (1e-10, 1e-1),
    nu: float = 2.5,
    metric: str | None = None,
    normalize: bool = True,
    feature_scales=None,
    validate_nonnegative: bool = True,
):
    """Create a common scikit-learn Gaussian-process kernel.

    ``"ard_rbf"`` and ``"ard_matern"`` use one length scale per feature, so
    ``n_features`` is required for those options.
    """
    normalized = _normalize_kernel_name(name)

    if normalized == "rbf":
        return _build_constant_scaled_kernel(
            RBF(length_scale=length_scale, length_scale_bounds=length_scale_bounds),
            constant_value=constant_value,
            constant_value_bounds=constant_value_bounds,
            noise_level=noise_level,
            noise_level_bounds=noise_level_bounds,
        )
    if normalized in {"matern", "matern25", "matern52"}:
        return _build_constant_scaled_kernel(
            Matern(length_scale=length_scale, length_scale_bounds=length_scale_bounds, nu=nu),
            constant_value=constant_value,
            constant_value_bounds=constant_value_bounds,
            noise_level=noise_level,
            noise_level_bounds=noise_level_bounds,
        )
    if normalized == "ardrbf":
        n_features = _validate_n_features(n_features, name="ard_rbf")
        return _build_constant_scaled_kernel(
            RBF(
                length_scale=np.ones(n_features),
                length_scale_bounds=length_scale_bounds,
            ),
            constant_value=constant_value,
            constant_value_bounds=constant_value_bounds,
            noise_level=noise_level,
            noise_level_bounds=noise_level_bounds,
        )
    if normalized == "ardmatern":
        n_features = _validate_n_features(n_features, name="ard_matern")
        return _build_constant_scaled_kernel(
            Matern(
                length_scale=np.ones(n_features),
                length_scale_bounds=length_scale_bounds,
                nu=nu,
            ),
            constant_value=constant_value,
            constant_value_bounds=constant_value_bounds,
            noise_level=noise_level,
            noise_level_bounds=noise_level_bounds,
        )
    if normalized == "tanimoto":
        return build_tanimoto_gpr_kernel(
            constant_value=constant_value,
            constant_value_bounds=constant_value_bounds,
            noise_level=noise_level,
            noise_level_bounds=noise_level_bounds,
            validate_nonnegative=validate_nonnegative,
        )
    if normalized in {"elementfraction", "composition"}:
        return build_element_fraction_gpr_kernel(
            constant_value=constant_value,
            constant_value_bounds=constant_value_bounds,
            length_scale=float(length_scale),
            length_scale_bounds=length_scale_bounds,
            noise_level=noise_level,
            noise_level_bounds=noise_level_bounds,
            metric="l1" if metric is None else metric,
            normalize=normalize,
        )
    if normalized in {"structure", "structurefeatures"}:
        return build_structure_gpr_kernel(
            constant_value=constant_value,
            constant_value_bounds=constant_value_bounds,
            length_scale=float(length_scale),
            length_scale_bounds=length_scale_bounds,
            noise_level=noise_level,
            noise_level_bounds=noise_level_bounds,
            metric="l2" if metric is None else metric,
            feature_scales=feature_scales,
        )

    raise ValueError(
        "name must be one of: rbf, matern, ard_rbf, ard_matern, tanimoto, "
        "element_fraction, composition, structure, structure_features"
    )

build_sklearn_gpr_model(*, kernel='matern', n_features=None, alpha=1e-08, normalize_y=True, n_restarts_optimizer=5, random_state=42, **kernel_kwargs)

Build a scikit-learn Gaussian Process Regressor.

This returns an unfitted model. Put it inside a scikit-learn Pipeline with build_preprocessor when raw dataframe columns need preprocessing. alpha may be a scalar or one variance per training row, which is useful when appending soft virtual physics observations.

Source code in matgpr/sklearn_gpr.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def build_sklearn_gpr_model(
    *,
    kernel: str = "matern",
    n_features: int | None = None,
    alpha: float | Sequence[float] | np.ndarray = 1e-8,
    normalize_y: bool = True,
    n_restarts_optimizer: int = 5,
    random_state: int | None = 42,
    **kernel_kwargs,
) -> GaussianProcessRegressor:
    """Build a scikit-learn Gaussian Process Regressor.

    This returns an unfitted model. Put it inside a scikit-learn Pipeline with
    ``build_preprocessor`` when raw dataframe columns need preprocessing.
    ``alpha`` may be a scalar or one variance per training row, which is useful
    when appending soft virtual physics observations.
    """
    kernel_object = build_sklearn_gpr_kernel(
        kernel,
        n_features=n_features,
        **kernel_kwargs,
    )
    return GaussianProcessRegressor(
        kernel=kernel_object,
        alpha=alpha,
        normalize_y=normalize_y,
        n_restarts_optimizer=n_restarts_optimizer,
        random_state=random_state,
    )

Build a grid search over useful scikit-learn GPR kernels and settings.

Source code in matgpr/sklearn_gpr.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def build_sklearn_gpr_grid_search(
    *,
    n_features: int,
    cv: int = 5,
    scoring: str = "neg_root_mean_squared_error",
    random_state: int | None = 42,
    n_jobs: int = 1,
) -> GridSearchCV:
    """Build a grid search over useful scikit-learn GPR kernels and settings."""
    _validate_n_features(n_features, name="grid search")
    kernels = [
        build_sklearn_gpr_kernel("rbf"),
        build_sklearn_gpr_kernel("matern"),
        build_sklearn_gpr_kernel("matern", nu=1.5),
        build_sklearn_gpr_kernel("ard_rbf", n_features=n_features),
        build_sklearn_gpr_kernel("ard_matern", n_features=n_features),
    ]

    model = GaussianProcessRegressor(
        normalize_y=True,
        random_state=random_state,
    )

    param_grid = {
        "kernel": kernels,
        "alpha": [1e-10, 1e-8, 1e-6],
        "n_restarts_optimizer": [0, 2],
    }

    return GridSearchCV(
        estimator=model,
        param_grid=param_grid,
        scoring=scoring,
        cv=cv,
        n_jobs=n_jobs,
        refit=True,
        return_train_score=True,
    )