A missing-aware Pipeline from fit to reconstruction¶
This example uses the 0.5 API. Run its standalone script from a matching checkout using the installation instructions. The input has 36 samples and two original variables: a categorical region and a continuous size. NaNs are missing. A Boolean mask also excludes arbitrary placeholder values, including strings in the continuous column.
The Pipeline consumes samples × variables throughout. ObservationMasker
replaces excluded dense values with NaNs before AutoEncoder learns categories
and scaling. The encoder expands the region into indicator features and
standardizes size. VBPCA receives that samples × encoded-features table.
No transpose or raw mask is passed across feature expansion.
Row validation and capacity search¶
Enable sklearn metadata routing with
config_context(enable_metadata_routing=True). Pass the original-variable
mask as observation_mask=mask for Pipeline/GridSearchCV fitting, or through
params={"observation_mask": mask} for cross_validate. Each validation split
slices data and mask by sample, then learns categories and scaling on training
rows alone. The search compares capacities zero and one and refits the selected
Pipeline on all training rows.
The default unsupervised scorer averages observed-entry Gaussian marginal log densities across validation rows; larger values are preferred. It uses fitted loading means and noise, integrating the latent scores. It does not integrate posterior loading uncertainty. This rowwise target differs from the cellwise PRMS/Brier targets in the component-selection guide.
A budget of eight fitting iterations keeps this an API example. The result does not establish posterior convergence, predictive calibration or a recommended capacity for other data.
New rows and inverse reconstruction¶
The four new rows are separate from the fitting table and have their own mask.
Call search.best_estimator_.transform(new_data, observation_mask=new_mask):
the search wrapper's transform method does not route mask metadata. These
are conditional projections under fixed fitted parameters; joint training
variational means remain in the model's scores_. A completely missing new
row receives zero latent mean.
fitted[1:].inverse_transform(projected) reverses the fitted model and encoder,
returning region labels and continuous sizes in original units. The leading
masker has no inverse, so this explicitly uses the numerical/encoding portion
of the fitted Pipeline. Reconstruction predicts all original variables; it
does not restore hidden placeholders, recover the input mask or guarantee
equality with observed values. No fitting occurs during this operation.
Names and output tables¶
The default run needs NumPy, SciPy and sklearn, without table dependencies.
For named input and pandas output, install the data extra; for this mixed
Pipeline with Polars output, install both data and polars extras:
python scripts/example_masked_pipeline.py --output pandas
python scripts/example_masked_pipeline.py --output polars
Pandas retains the new-row index. Polars preserves row order; the raw object
masking step stays in pandas mode and subsequent numeric outputs use Polars.
Use get_feature_names_out() to inspect encoded features and latent output
names. Inverse reconstruction returns an array in original variable order,
regardless of the configured transform output container.
Executed source¶
The source below is the same standalone script exercised on the minimum and current supported sklearn versions, including both optional table modes.
"""Fit, search, project and reconstruct a missing-aware mixed-data Pipeline."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
import numpy as np
from sklearn import config_context
from sklearn.model_selection import GridSearchCV, KFold, cross_validate
from sklearn.pipeline import Pipeline
from vbpca_py import VBPCA, AutoEncoder, ObservationMasker
def run_example(output="default"):
"""Run a bounded API example; eight iterations do not establish convergence.
Returns:
Input tables, fitted search, fold estimators and reconstructed new rows.
Raises:
RuntimeError: Scores or reconstruction violate the demonstrated contract.
"""
rng = np.random.default_rng(23)
raw = np.empty((36, 2), dtype=object)
raw[:, 0] = np.tile(["west", "east", "north"], 12)
raw[:, 1] = rng.normal(100, 5, len(raw))
mask = rng.random(raw.shape) > 0.1
raw[4, 1], raw[5, 0] = np.nan, np.nan
mask[4, 1], mask[5, 0] = True, True
data = np.where(mask, raw, "unobserved placeholder")
new_data = np.array(
[
["west", 90.0],
["unobserved placeholder", np.nan],
["north", "unobserved placeholder"],
["east", 115.0],
],
dtype=object,
)
new_mask = np.array([[True, True], [False, True], [True, False], [True, True]])
if output != "default":
import pandas as pd
data = pd.DataFrame(
data,
columns=["region", "size"],
index=[f"train-{i}" for i in range(len(raw))],
)
new_data = pd.DataFrame(
new_data,
columns=data.columns,
index=[f"new-{i}" for i in range(len(new_data))],
)
pipeline = Pipeline([
("observe", ObservationMasker()),
(
"encode",
AutoEncoder(column_types=["categorical", "continuous"], binary="both"),
),
(
"model",
VBPCA(
1,
maxiters=8,
niter_broadprior=0,
verbose=0,
random_state=23,
num_cpu=1,
runtime_tuning="off",
),
),
])
if output != "default":
pipeline.set_output(transform=output)
if output == "polars":
# Raw object observations stay in pandas; numerical output uses Polars.
pipeline.named_steps["observe"].set_output(transform="pandas")
splitter = KFold(n_splits=3, shuffle=True, random_state=23)
with config_context(enable_metadata_routing=True):
result = cross_validate(
pipeline,
data,
cv=splitter,
params={"observation_mask": mask},
error_score="raise",
return_estimator=True,
)
search = GridSearchCV(
pipeline, {"model__n_components": [0, 1]}, cv=splitter, error_score="raise"
).fit(data, observation_mask=mask)
fitted = search.best_estimator_
projected = fitted.transform(new_data, observation_mask=new_mask)
# The masker has no inverse. Reverse the fitted model and encoder only.
reconstructed = fitted[1:].inverse_transform(projected)
if not np.isfinite(result["test_score"]).all():
raise RuntimeError("The synthetic CV example returned a nonfinite score")
if projected.shape != (len(new_data), search.best_params_["model__n_components"]):
raise RuntimeError("The synthetic projection has the wrong shape")
if reconstructed.shape != new_data.shape:
raise RuntimeError("Reconstruction must use the original variable layout")
if not np.isfinite(reconstructed[:, 1].astype(float)).all():
raise RuntimeError("Reconstructed continuous values must be finite")
return {
"data": data,
"mask": mask,
"new_data": new_data,
"new_mask": new_mask,
"pipeline": pipeline,
"splitter": splitter,
"cv": result,
"search": search,
"projected": projected,
"reconstructed": reconstructed,
}
def export_gallery(result, output_dir):
"""Render learned fold offsets and original-unit reconstruction summaries."""
import matplotlib.pyplot as plt
fitted = result["search"].best_estimator_
encoder = fitted.named_steps["encode"]
schema = encoder.encoding_schema_
schema.validate(len(encoder.get_feature_names_out()), strict_numeric=True)
offsets = [
fold.named_steps["encode"].encoding_schema_.variables[1].offset[0]
for fold in result["cv"]["estimator"]
]
fig, ax = plt.subplots(figsize=(6, 3.5))
ax.bar(range(len(offsets)), offsets)
ax.set(
xlabel="Row-validation fold",
ylabel="Training-only size offset (input units)",
xticks=range(len(offsets)),
title="Preprocessing is fitted on each training split",
)
fig.tight_layout()
output_dir.mkdir(parents=True, exist_ok=True)
fig.savefig(output_dir / "pipeline.png", dpi=130)
plt.close(fig)
summary = {
"training_shape": list(result["data"].shape),
"new_rows": len(result["new_data"]),
"encoded_features": encoder.get_feature_names_out().tolist(),
"schema_kinds": [v.kind for v in schema.variables],
"fold_training_offsets": offsets,
"selected_capacity": result["search"].best_params_["model__n_components"],
"decoded_size_input_units": result["reconstructed"][:, 1]
.astype(float)
.tolist(),
}
(output_dir / "summary.json").write_text(json.dumps(summary, indent=2) + "\n")
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--output", choices=["default", "pandas", "polars"], default="default"
)
parser.add_argument("--output-dir", type=Path)
args = parser.parse_args()
result = run_example(args.output)
fitted = result["search"].best_estimator_
print(
"Encoded features:",
fitted.named_steps["encode"].get_feature_names_out().tolist(),
)
print("Selected capacity:", result["search"].best_params_["model__n_components"])
print("Conditional score shape:", result["projected"].shape)
print("Reconstructed rows (region, size):")
print(result["reconstructed"])
print("Masked Pipeline, row CV, GridSearchCV and new-row projection passed")
if args.output_dir is not None:
export_gallery(result, args.output_dir)
if __name__ == "__main__":
main()