UnsupervisedPipeline

The deployable artefact an unsupervised run produces: preprocessor + fitted detector + result in one joblib file, with a single predict(X) surface. Both detectors build one via to_pipeline(), and a run with outputs_dir= persists it under <outputs_dir>/deploy/unsupervised_pipeline.joblib.

See the unsupervised guide for the workflow; this page is the API surface.

In short

• One file holds the preprocessor, the fitted detector and the result; score new rows with predict(X).
• Read applied_overrides to see every place the run departed from your config.
• A pipeline whose serves_rows_independently is False (it contains UMAP) refuses to predict.
• load() refuses an artefact with a different version stamp: re-fit and re-persist it.

Reading a run honestly

Three properties exist specifically so a consumer cannot mistake what happened:

• applied_overrides — every place the engine departed from the requested configuration (a skipped or clamped pre-reduction, a PCA-for-UMAP substitution, a derived dbscan_eps, a subsampled silhouette, a resolved contamination…). It is derived from the fitted objects, not self-reported by the code paths that make those decisions, so a downgrade cannot forget to register itself.
• serves_rows_independently — False when the transform chain contains a component whose output depends on batch composition (UMAP). Such a pipeline refuses to predict, because a row's answer would depend on which rows were scored beside it.
• fit_frame — row count, columns and a content fingerprint of the frame the model was fitted on, so a read surface can refuse to join fit-time row positions into a dataset that has since changed.

Each applied_overrides entry describes the knob that actually governed this algorithm: the density algorithms report that they discovered the cluster count (they never read n_clusters), GMM reports gmm_n_components, and only a fit whose auto_k_score_name is set is described as having had its k chosen by a sweep.

The MLflow payload, result.json and report.txt carry the same honesty fields as the result objects: separation (the anomaly metric — anomaly_ratio is a param, since in mode="single" it restates contamination), score_degeneracy and caveats, and for clustering verdict / verdict_reason, metric_coverage (the silhouette is printed as 0.62 (on 14% of rows)) and birch_threshold_used. Score percentiles are namespaced by the detector that produced them, because they are in that detector's own units.

Version gate

Artefacts are version-stamped and load() raises StaleArtifactError (a ValueError) for anything whose stamp is not the current one. There is no migration path and no compatibility shim: re-fit and re-persist. The stamp is read before the pickled object graph is reconstructed, so a stale artefact fails with the version message rather than an AttributeError naming an internal class.

library_versions records the third-party versions the pickled estimators were written by; a differing major is refused at load, and anything else warns.

UnsupervisedPipeline

UnsupervisedPipeline(
    *,
    detector: Detector,
    embedding: Optional[Embedding2D] = None,
    embedding_coords: Optional[ndarray] = None,
    metadata: Optional[dict[str, Any]] = None,
    version: str = _FORMAT_VERSION,
    created_at: Optional[str] = None,
    config: Optional[dict[str, Any]] = None,
    X_fit: Optional[Union[DataFrame, ndarray]] = None
)

Composite artifact bundling preprocessor + detector + result.

Use one of the factories (from_anomaly_detector / from_clustering_detector) when wiring an already-fit detector, or instantiate directly when assembling a multi-detector workflow (e.g. cluster-then-flag-anomalies-per-cluster).

The artifact exposes a unified predict(X) surface that returns a DataFrame with the input rows plus the task-appropriate output columns (Anomaly/Anomaly_Score for anomaly tasks; Cluster for clustering tasks). See plan §7.1.

supports_inductive_predict property

supports_inductive_predict: bool

True when the bundled detector can score unseen rows.

Anomaly detectors are always inductive; clustering pipelines are inductive only for KMeans / MiniBatchKMeans / GMM / Birch. Transductive clustering pipelines (DBSCAN / HDBSCAN / Agglomerative) have no out-of-sample predict and so must not be logged as a servable MLflow model.

A pipeline whose serving chain is not row-independent (:attr:serves_rows_independently) is equally unservable, for the same reason expressed differently: it has a predict, but the answer it gives a row depends on which rows were scored alongside it.

batch_dependent_serving_components property

batch_dependent_serving_components: tuple[str, ...]

Attribute paths of serving-chain stages that transform batch-wise.

Empty tuple ⇒ every stage is per-row deterministic. Recomputed on each access rather than cached at construction, because callers legitimately swap a reducer onto a fitted detector after the pipeline was built. See :func:parleml_ml4.unsupervised.pipeline_serving.batch_dependent_components for what the scan covers and what it cannot reach.

serves_rows_independently property

serves_rows_independently: bool

True when predict / explain_rows are a function of each row.

The serving contract this pipeline is expected to honour: scoring a row alone, in a chunk of 5 000, or in a chunk of 20 000 must give the same answer. False means some stage of the transform chain reads the batch — see :attr:batch_dependent_serving_components for which.

applied_overrides property

applied_overrides: list[dict[str, Any]]

Every place the fitted run differs from the config as written.

Each entry is {field, requested, effective, reason}, derived from the fitted objects rather than reported by the sites that made the decisions — see :func:parleml_ml4.unsupervised.pipeline_provenance.derive_applied_overrides.

Best-effort and total: it is rendered into result.json and report.txt, and an artefact with unusual state must degrade to a shorter list, never to a failed export.

text_embedding_records

text_embedding_records() -> 'tuple[EmbeddingRecord, ...]'

What this pipeline's text embeddings came from — its external dependencies.

One record (a local model's pinned revision, or a provider's connection, model and fingerprint) when the preprocessor embedded text at fit, else empty.

from_anomaly_detector classmethod

from_anomaly_detector(
    detector: AnomalyDetector,
    *,
    compute_embedding: bool = False,
    X_for_embedding: Optional[
        Union[DataFrame, ndarray]
    ] = None,
    metadata: Optional[dict[str, Any]] = None,
    X_fit: Optional[Union[DataFrame, ndarray]] = None
) -> "UnsupervisedPipeline"

Bundle a fitted :class:AnomalyDetector into a deployable pipeline.

X_fit is the frame the detector was fitted on; it is fingerprinted (never stored) so a read surface can detect that the dataset it is joining fit-time row positions into is no longer that frame (PIP-7). Defaults to X_for_embedding, which is the same frame whenever the caller asked for an embedding — pass it explicitly to keep the guard when compute_embedding=False.

from_clustering_detector classmethod

from_clustering_detector(
    detector: ClusteringDetector,
    *,
    compute_embedding: bool = False,
    X_for_embedding: Optional[
        Union[DataFrame, ndarray]
    ] = None,
    metadata: Optional[dict[str, Any]] = None,
    X_fit: Optional[Union[DataFrame, ndarray]] = None
) -> "UnsupervisedPipeline"

Bundle a fitted :class:ClusteringDetector into a deployable pipeline.

See :meth:from_anomaly_detector for X_fit.

fit_frame_mismatch

fit_frame_mismatch(
    df: Optional[DataFrame],
) -> Optional[str]

Why df is not the frame this run was fitted on — or None.

Everything a run reports positionally — top_k_indices, labels, the per-row silhouette, the score time-series — indexes the fit frame by position, so a dataset re-ingested under the same id leaves iloc[idx] succeeding against rows that have nothing to do with those scores (PIP-7). Returns a human-readable reason, or None when the frame still lines up (never a false alarm). See :func:parleml_ml4.unsupervised.pipeline_provenance.fit_frame_mismatch.

set_operating_threshold

set_operating_threshold(value: Optional[float]) -> None

Bake an anomaly operating threshold into the bundled detector.

Anomaly only — relabels future :meth:predict / :meth:explain_rows calls at decision_function(X) >= value (None reverts to the fitted threshold). Raises NotImplementedError for clustering pipelines, which have no scalar operating point.

predict

predict(X: Any) -> DataFrame

Score / cluster X and return the input frame enriched with the task-appropriate output column(s).

• Anomaly tasks add Anomaly (0/1) and Anomaly_Score cols.
• Clustering tasks add a Cluster column (int). For transductive algorithms (DBSCAN / HDBSCAN / Agglomerative) the underlying detector raises NotImplementedError — callers see the same error pointing at the inductive alternatives.

Accepts a DataFrame (preferred — preserves original columns) or any sklearn-style array; see :meth:coerce_input for exactly how non-frames are named.

The input's index is preserved verbatim — string, duplicated and non-monotonic indices all round-trip, and the output columns are assigned positionally, so no label alignment happens behind your back.

Raises NotImplementedError when the serving chain is not row independent (see :attr:serves_rows_independently).

explain_rows

explain_rows(
    X: Any, *, top_k_features: int = 3
) -> list[dict[str, Any]]

Per-row reason codes for new rows (inductive runs only).

Delegates to the bundled detector:

• Anomaly — label / score + top contributing features.
• Inductive clustering — assigned cluster + centroid distance.
• Transductive clustering — raises NotImplementedError (same contract as :meth:predict).

The shape is uniform across tasks (task-irrelevant fields are None / []) so the backend can serve one wire model.

The index field is positional, not a label. It is the 0-based position of the row within X — deliberately, because that is the only join key that is always well defined: a scoring frame routinely arrives with a string index (a parquet slice), a duplicated index (a groupby().head()) or a shuffled one (a filtered dataset), and a label join would then attach a reason code to the wrong row, or to several rows, without erroring. Entries are returned in input order, so reasons[i] describes X.iloc[i] and reasons[i]["index"] == i; callers that need the source label should read X.index[r["index"]]. Pinned by test_explain_rows_index_is_positional_under_a_non_range_index.

Raises NotImplementedError when the serving chain is not row independent (see :attr:serves_rows_independently).

coerce_input

coerce_input(X: Any) -> DataFrame

Normalise any sklearn-style input into the frame the detector expects.

Public because it is part of the serving contract, not an implementation detail. An MLflow endpoint hands the pyfunc wrapper whatever the caller posted — routinely a bare ndarray, since _mlflow_signature is best-effort and a model logged without a signature does no coercion of its own. pd.DataFrame(arr) names those columns 0..n-1, which the preprocessor rejects with transform missing numeric columns; naming them from preprocessor.source_columns_ (when the widths agree) is what makes the array work. Both doors into this model — :meth:predict and :class:~parleml_ml4.unsupervised.pyfunc.UnsupervisedPyfunc — must go through exactly this function or they answer the same input differently (PIP-9).

A DataFrame is returned as-is (no copy). A Series and a 1-D array both mean one row, because that is what a caller scoring a single record actually holds: df.iloc[0] returns a Series whose index is the column names. Reading it as a column instead — which is what Series.to_frame() does, and what this used to do — produced an n×1 frame whose one column is named 0 and whose index is the feature names, so the preprocessor rejected it with a missing-columns error that named every feature the caller had in fact supplied. Anything neither 1-D nor 2-D is rejected here, where the caller can see what it handed over, rather than deeper in the preprocessor.

save

save(path: Union[str, PathLike]) -> Path

Persist the pipeline to path via joblib (zlib level 3), atomically.

The write is atomic (temp file + os.replace) because this path routinely overwrites an artefact that is already deployed and possibly being read: the detector writes it at the end of fit and the runner immediately re-saves the enriched pipeline over the same name (see :data:parleml_ml4.run_layout.UNSUPERVISED_ARTEFACT_NAME), and a committed operating threshold re-persists it again later. A torn write there would leave the serving model corrupt, and a reader mid-load would see half a file. See :func:parleml_ml4.atomic_io.atomic_joblib_dump.

Fit-time caches are stripped on the way out — see :meth:__getstate__.

load classmethod

load(path: Union[str, PathLike]) -> 'UnsupervisedPipeline'

Read an artefact written by :meth:save, or refuse it with a reason.

The gate runs first and entirely off a peek at the pickle stream, so a stale or environment-incompatible artefact fails with an actionable message instead of an AttributeError on some internal name — see :func:parleml_ml4.unsupervised.pipeline_gate.enforce_artifact_gate.

log_mlflow_run

log_mlflow_run(
    mlops: Any,
    *,
    X_fit: Optional[Union[DataFrame, ndarray]] = None,
    plots: Optional[PlotBuilders] = None,
    **log_kwargs: Any
) -> Optional[str]

Drive one MLflow run for this pipeline and return the run id.

Reuses :meth:log_to_mlflow (params / metrics / supplemental artifacts) via the shared engine-agnostic runner and, when the detector is inductive, additionally logs the servable pyfunc model under the canonical model path + optional Unity Catalog registration. Transductive clustering pipelines log a run with artifacts only (no scorer). No-op when MLOps logging is disabled. X_fit also supplies the model signature's input example.

log_to_mlflow

log_to_mlflow(
    logger: "BaseLogger",
    *,
    artifact_filename: str = "unsupervised_pipeline.joblib",
    log_findings: bool = True,
    log_cluster_profiles: bool = True,
    log_report: bool = True,
    log_result_json: bool = True,
    log_top_k_csv: bool = True,
    X_fit: Optional[Union[DataFrame, ndarray]] = None,
    plots: Optional[PlotBuilders] = None
) -> None

Log params / metrics / artifacts via a parleml BaseLogger.

Mirrors plan §10 — emits the standard 5-artifact bundle expected by the supervised UI (pipeline.joblib, result.json, report.txt, plots/*.html, top_k_anomalies.csv).

:param logger: any concrete :class:parleml_ml4.loggers.base_logger.BaseLogger. :param artifact_filename: leaf filename used for the pipeline joblib upload. :param log_findings: when True, write unsupervised_findings.json (anomaly mode). :param log_cluster_profiles: when True, write cluster_profiles.csv (clustering mode). :param log_report: when True, persist a textual summary of the served result as report.txt; anomaly auto mode appends ADEngine.report(state, format='text'). :param log_result_json: when True, write result.json — a lightweight numpy-free view of the result. :param log_top_k_csv: when True (anomaly only), write top_k_anomalies.csv with the top-K rows. Its row_index values are positions into the fit frame, so X_fit has to be that frame for the joined columns to mean anything. :param X_fit: the frame the detector was fitted on; its raw columns ride along on top_k_anomalies.csv. :param plots: (artifact name, zero-arg figure builder) pairs, each written as plots/<name>.html. ml4 draws no figures itself; parleml_mli.unsupervised.registry.mlflow_plots builds the standard set. None logs no plots.

Next: Unsupervised results · Unsupervised guide