30.4 Building an Anomaly Discovery Pipeline

1. The Three-Method Ensemble Strategy

Imagine a telescope survey flags a faint transient. Isolation Forest scores it as the night's top anomaly, yet LOF ranks it as perfectly ordinary, and an autoencoder reconstructs it with almost zero error. Is that transient a genuine astrophysical event or an instrumental glitch? The answer depends on which detector you trust. If you can only trust one, you will be wrong far too often. The solution is deliberate diversity: combine three methods chosen for complementary strengths so that no single blind spot dictates the outcome.

1. Isolation Forest: a tree-based method that detects anomalies through ease of isolation. It makes no density assumptions, scales linearly, and handles high-dimensional data naturally. Its blind spot is anomalies near cluster boundaries.
2. LOF (Local Outlier Factor): a density-based method that detects local anomalies by comparing each point's density to its neighbors'. It excels at finding anomalies in datasets with clusters of varying density. Its blind spot is global anomalies in uniformly distributed data.
3. Autoencoder reconstruction error: a deep method that detects anomalies through learned data representations. It captures structural and nonlinear patterns invisible to distance-based methods. Its blind spot is anomalies that the network can reconstruct well (when the bottleneck is too wide).

Each method scores anomalies on a different scale. To combine them meaningfully, we need score normalization. The standard approach is to convert each method's raw scores to a common scale using one of three strategies:

Rank normalization: replace each score with its rank divided by the number of samples. This is the most robust option because it is invariant to the score distribution shape. A point ranked 5th out of 1,000 by Isolation Forest and 3rd out of 1,000 by LOF receives normalized scores of 0.005 and 0.003 regardless of the raw score magnitudes.

Rank normalization converts each detector's raw anomaly scores into a uniform distribution on \([0, 1]\) by replacing each score with its fractional position in the sorted order. This step is essential because raw scores from different detectors live on incomparable scales: Isolation Forest might output values near \(-0.5\), LOF might output values near 1.2, and an autoencoder might output reconstruction errors near 0.01. Averaging these raw numbers lets whichever detector produces the largest magnitudes dominate the ensemble. The procedure is straightforward: sort the \(n\) scores, assign each its rank (1 through \(n\)), then divide by \(n - 1\). The least anomalous point maps to 0 and the most anomalous maps to 1. Prefer rank normalization over min-max or z-score alternatives when score distributions are skewed, heavy-tailed, or contain extreme outliers, the typical case for anomaly scores. Reserve z-score normalization for the rare scenario where all detectors produce approximately Gaussian score distributions.

Alternative Normalization Strategies

Min-max normalization: scale scores to \([0, 1]\) using \(\hat{s} = (s - s_{\min}) / (s_{\max} - s_{\min})\). Simple but sensitive to extreme outliers in the score distribution.

Z-score normalization: transform to zero mean and unit variance. Assumes roughly Gaussian score distributions, which is often violated for anomaly scores (they are typically right-skewed). In short: normalize scores by rank so that every detector gets an equal vote, then let consensus, not any single method's confidence, decide what deserves your attention.

Before diving into code, Figure 30.6 illustrates the end-to-end data flow: raw input passes through standardization, fans out to three independent detectors, and converges through rank normalization and averaging into a single consensus-scored anomaly report.

2. The Complete Pipeline Implementation

With the ensemble strategy and its scoring mechanics established, the next step is to wire these three detectors into a single, reusable pipeline that handles everything from raw data ingestion through consensus ranking. The implementation below also introduces consensus counting, a complementary signal that records how many of the three detectors independently flag each point, providing a measure of cross-method agreement beyond the averaged score.

The pipeline below processes a dataset through three detectors, normalizes and combines scores, and produces a ranked anomaly report. It is designed to be domain-agnostic: swap in your scientific data and adjust the contamination estimate, and the pipeline handles the rest.

import numpy as np
from dataclasses import dataclass, field
from typing import Optional

from sklearn.ensemble import IsolationForest
from sklearn.neighbors import LocalOutlierFactor
from sklearn.preprocessing import StandardScaler
from pyod.models.auto_encoder import AutoEncoder

@dataclass
class AnomalyReport:
    """Container for anomaly detection results."""
    scores: np.ndarray              # combined anomaly scores
    rankings: np.ndarray            # indices sorted by anomaly score (descending)
    per_method_scores: dict         # method_name -> normalized scores
    labels: np.ndarray              # binary labels: 1 = anomaly, 0 = normal
    consensus_counts: np.ndarray    # how many methods agree each point is anomalous
    threshold: float                # score threshold used for labeling
    metadata: dict = field(default_factory=dict)


class AnomalyDiscoveryPipeline:
    """Three-method ensemble anomaly detection pipeline.

    Combines Isolation Forest, LOF, and Autoencoder with
    rank-based score normalization and consensus voting.
    """

    def __init__(self, contamination: float = 0.05, random_state: int = 42):
        self.contamination = contamination
        self.random_state = random_state
        self.scaler = StandardScaler()
        self.detectors = {}
        self.is_fitted = False

    def _build_detectors(self, n_features: int):
        """Initialize detectors with appropriate hyperparameters."""
        self.detectors = {
            'isolation_forest': IsolationForest(
                n_estimators=200,
                contamination=self.contamination,
                random_state=self.random_state,
                max_samples='auto',
            ),
            'lof': LocalOutlierFactor(
                n_neighbors=20,
                contamination=self.contamination,
                novelty=False,  # outlier detection mode
            ),
            'autoencoder': AutoEncoder(
                hidden_neurons=[n_features, 64, 32, 16, 32, 64, n_features],
                epochs=100,
                contamination=self.contamination,
                preprocessing=True,
                verbose=0,
            ),
        }

    def fit_score(self, X: np.ndarray) -> AnomalyReport:
        """Fit all detectors and produce anomaly report.

        Args:
            X: (n_samples, n_features) data matrix
        Returns:
            AnomalyReport with combined scores and rankings
        """
        n_samples, n_features = X.shape
        self._build_detectors(n_features)

        # Standardize features
        X_scaled = self.scaler.fit_transform(X)

        # Run each detector and collect raw scores
        raw_scores = {}

        # Isolation Forest: score_samples returns negative scores
        iso = self.detectors['isolation_forest']
        iso.fit(X_scaled)
        raw_scores['isolation_forest'] = -iso.score_samples(X_scaled)

        # LOF: negative_outlier_factor_ gives negative scores
        lof = self.detectors['lof']
        lof.fit_predict(X_scaled)
        raw_scores['lof'] = -lof.negative_outlier_factor_

        # Autoencoder: decision_function returns per-sample anomaly
        # scores (here, reconstruction error: how poorly the
        # autoencoder reproduces each input after compression).
        ae = self.detectors['autoencoder']
        ae.fit(X_scaled)
        raw_scores['autoencoder'] = ae.decision_function(X_scaled)

        # Rank-normalize each method's scores to [0, 1]
        normalized_scores = {}
        for name, scores in raw_scores.items():
            # Double argsort: the first argsort returns the indices that
            # would sort the array; applying argsort again maps each
            # element to its position (rank) in the sorted order.
            ranks = np.argsort(np.argsort(scores)).astype(float)
            normalized_scores[name] = ranks / (n_samples - 1)

        # Combine: average of normalized scores
        combined = np.mean(
            [normalized_scores[m] for m in normalized_scores], axis=0
        )

        # Consensus counting: for each detector independently, find the
        # top contamination-fraction of points and mark them anomalous.
        # Then count how many of the three detectors agree on each point.
        # A point with consensus 3/3 was flagged by every method.
        n_anomalies = max(1, int(n_samples * self.contamination))
        consensus = np.zeros(n_samples, dtype=int)
        for name, scores in raw_scores.items():
            top_k = np.argsort(scores)[-n_anomalies:]
            consensus[top_k] += 1

        # Threshold: top contamination fraction by combined score
        threshold = np.percentile(combined, 100 * (1 - self.contamination))
        labels = (combined >= threshold).astype(int)

        # Ranking: most anomalous first
        rankings = np.argsort(combined)[::-1]

        self.is_fitted = True
        return AnomalyReport(
            scores=combined,
            rankings=rankings,
            per_method_scores=normalized_scores,
            labels=labels,
            consensus_counts=consensus,
            threshold=threshold,
            metadata={
                'n_samples': n_samples,
                'n_features': n_features,
                'contamination': self.contamination,
                'n_flagged': labels.sum(),
            }
        )

    def report_summary(self, report: AnomalyReport,
                       top_k: int = 20) -> str:
        """Generate a human-readable summary of the anomaly report."""
        lines = [
            "=" * 60,
            "ANOMALY DISCOVERY REPORT",
            "=" * 60,
            f"Dataset: {report.metadata['n_samples']} samples, "
            f"{report.metadata['n_features']} features",
            f"Contamination: {report.metadata['contamination']:.1%}",
            f"Total flagged: {report.metadata['n_flagged']}",
            f"Score threshold: {report.threshold:.4f}",
            "",
            f"Top {top_k} anomalies:",
            "-" * 60,
            f"{'Rank':>4}  {'Index':>6}  {'Score':>8}  "
            f"{'Consensus':>9}  {'Methods Agreeing'}",
            "-" * 60,
        ]
        for rank, idx in enumerate(report.rankings[:top_k], 1):
            methods = []
            for name in report.per_method_scores:
                if report.per_method_scores[name][idx] >= (1 - self.contamination):
                    methods.append(name[:3].upper())
            lines.append(
                f"{rank:>4}  {idx:>6}  {report.scores[idx]:>8.4f}  "
                f"{report.consensus_counts[idx]:>9}/3  "
                f"{'  '.join(methods)}"
            )
        return "\n".join(lines)

3. Running the Pipeline on Scientific Data

The following code runs the pipeline on a realistic synthetic dataset that mimics a common scientific scenario: a large population of normal observations, a small population of known anomalies (instrument errors), and an even smaller population of genuinely novel observations (potential discoveries). The pipeline's job is to rank the genuine novelties above the instrument errors.

# Generate a realistic scientific dataset
rng = np.random.default_rng(42)
n_features = 20

# Population 1: Normal observations (95%)
n_normal = 950
X_normal = rng.multivariate_normal(
    mean=np.zeros(n_features),
    cov=np.eye(n_features) * 0.5 + 0.1,  # slight correlation
    size=n_normal
)

# Population 2: Instrument errors (3%) - random spikes in single features
n_errors = 30
X_errors = rng.multivariate_normal(
    mean=np.zeros(n_features),
    cov=np.eye(n_features) * 0.5 + 0.1,
    size=n_errors
)
# Inject random spikes (single-feature anomalies)
for i in range(n_errors):
    spike_feat = rng.integers(0, n_features)
    X_errors[i, spike_feat] += rng.choice([-1, 1]) * rng.uniform(5, 10)

# Population 3: Genuine novelties (2%) - coherent multi-feature pattern
n_novel = 20
novel_direction = rng.standard_normal(n_features)
novel_direction /= np.linalg.norm(novel_direction)
X_novel = rng.multivariate_normal(
    mean=novel_direction * 4.0,  # shifted in a specific direction
    cov=np.eye(n_features) * 0.3,
    size=n_novel
)

# Combine
X = np.vstack([X_normal, X_errors, X_novel])
labels = np.array(
    ['normal'] * n_normal + ['error'] * n_errors + ['novel'] * n_novel
)

# Run the pipeline
pipeline = AnomalyDiscoveryPipeline(contamination=0.05, random_state=42)
report = pipeline.fit_score(X)

# Print summary
print(pipeline.report_summary(report, top_k=15))

# Detailed evaluation: how well does the pipeline separate categories?
top_50_indices = report.rankings[:50]
top_50_labels = labels[top_50_indices]
print(f"\nCategory breakdown in top 50 anomalies:")
for cat in ['novel', 'error', 'normal']:
    count = (top_50_labels == cat).sum()
    total = (labels == cat).sum()
    print(f"  {cat}: {count}/{total} ({count/total:.0%} recall)")

# Consensus analysis: do genuine novelties get higher consensus?
for cat in ['novel', 'error', 'normal']:
    mask = labels == cat
    mean_consensus = report.consensus_counts[mask].mean()
    mean_score = report.scores[mask].mean()
    print(f"  {cat}: mean consensus={mean_consensus:.2f}/3, "
          f"mean score={mean_score:.4f}")

4. Ground-Truth Validation

A ranked list of anomalies is only as useful as your confidence that the rankings reflect reality, so the natural next question after running the pipeline is: how do we know these detections are correct?

Evaluating anomaly detectors is fundamentally challenging because ground truth is almost never available in the quantities needed for standard supervised evaluation. In most scientific settings, we face one of three scenarios:

No labels at all: the most common case. We can only evaluate using internal quality metrics (agreement between methods, stability under resampling) or by inspecting the top-ranked anomalies manually. Internal evaluation is necessary but insufficient; it cannot distinguish between a detector that finds genuinely interesting anomalies and one that consistently flags artifacts.

Partial labels: some anomalies are known (from previous experiments, known failure modes, or expert annotation of a subset). We can compute precision and recall on the labeled subset, but we must be careful about selection bias: the known anomalies may not be representative of the anomalies we want to discover.

Synthetic injection: we inject known anomalies into real data and evaluate detection. This is the most controlled approach but requires domain knowledge to create realistic synthetic anomalies. If the injected anomalies are too obvious, the evaluation is meaningless. If they are too subtle, even perfect detectors will miss them.

from sklearn.metrics import (
    roc_auc_score, average_precision_score,
    precision_recall_curve
)

def evaluate_anomaly_detector(scores: np.ndarray,
                              true_labels: np.ndarray,
                              positive_label: str = 'anomaly') -> dict:
    """Evaluate anomaly detection performance with multiple metrics.

    Args:
        scores: continuous anomaly scores (higher = more anomalous)
        true_labels: array of string labels
        positive_label: which label counts as a true anomaly
    Returns:
        dict of evaluation metrics
    """
    # Convert to binary: 1 = anomaly, 0 = normal
    binary_labels = (true_labels != 'normal').astype(int)

    # Core metrics
    auroc = roc_auc_score(binary_labels, scores)
    avg_precision = average_precision_score(binary_labels, scores)

    # Precision at various recall levels
    precision, recall, thresholds = precision_recall_curve(
        binary_labels, scores
    )

    # Precision@k: precision in top-k ranked items
    anomaly_rate = binary_labels.mean()
    results = {
        'auroc': auroc,
        'avg_precision': avg_precision,
        'anomaly_rate': anomaly_rate,
    }

    # P@k for various k values
    for k in [10, 20, 50, 100]:
        if k <= len(scores):
            top_k = np.argsort(scores)[-k:]
            p_at_k = binary_labels[top_k].mean()
            results[f'precision@{k}'] = p_at_k

    return results


def stability_analysis(X: np.ndarray, pipeline_cls,
                       n_bootstrap: int = 10, **kwargs) -> dict:
    """Assess ranking stability via bootstrap resampling.

    Bootstrap resampling draws repeated random samples (with replacement)
    from the dataset, each the same size as the original. By running the
    pipeline on each resample and comparing the resulting rankings, we
    measure how sensitive the detector is to small perturbations in the
    data. Stable detectors produce similar rankings across bootstrap
    samples. Unstable detectors flag different anomalies each time.
    """
    n_samples = X.shape[0]
    all_rankings = np.zeros((n_bootstrap, n_samples))
    rng = np.random.default_rng(42)

    for b in range(n_bootstrap):
        # Bootstrap sample (with replacement)
        indices = rng.choice(n_samples, size=n_samples, replace=True)
        X_boot = X[indices]

        pipeline = pipeline_cls(**kwargs)
        report = pipeline.fit_score(X_boot)

        # Map back to original indices
        for rank, boot_idx in enumerate(report.rankings):
            orig_idx = indices[boot_idx]
            all_rankings[b, orig_idx] = rank

    # Compute stability metrics
    mean_rank = all_rankings.mean(axis=0)
    rank_std = all_rankings.std(axis=0)

    # Points with low mean rank AND low std are stably anomalous
    return {
        'mean_rank': mean_rank,
        'rank_std': rank_std,
        'stable_top_20': np.argsort(mean_rank)[:20],
        'mean_top20_std': rank_std[np.argsort(mean_rank)[:20]].mean(),
    }


# Evaluate on our labeled dataset
eval_results = evaluate_anomaly_detector(
    report.scores, labels, positive_label='novel'
)
print("Evaluation Results:")
for metric, value in eval_results.items():
    print(f"  {metric}: {value:.4f}")

# Evaluate per-method performance
print("\nPer-method AUROC:")
binary_labels = (labels != 'normal').astype(int)
for name, norm_scores in report.per_method_scores.items():
    method_auroc = roc_auc_score(binary_labels, norm_scores)
    print(f"  {name}: {method_auroc:.4f}")
ensemble_auroc = roc_auc_score(binary_labels, report.scores)
print(f"  ensemble: {ensemble_auroc:.4f}")

Key Insight: Precision@k Matters More Than AUROC for Discovery

In scientific anomaly detection, we rarely care about classifying every point correctly. What matters is whether the top-ranked anomalies are worth investigating. Precision@k (the fraction of true anomalies among the top \(k\) ranked candidates) is the metric that directly measures this. A detector with Area Under the Receiver Operating Characteristic curve (AUROC) 0.95 but precision@20 of 0.30 ranks many false positives above true anomalies, wasting experimental resources on dead ends. A detector with AUROC 0.85 but precision@20 of 0.80 delivers actionable results despite lower overall discrimination. Always report precision@k alongside AUROC, where \(k\) matches the number of anomalies your lab or team can realistically investigate in a given cycle. This connects directly to the experimental design considerations in Chapter 46.

5. Integration with the Discovery Workbench

Validation confirms that the pipeline's detections are trustworthy, but individual runs remain isolated analyses unless the results flow into a broader system that tracks, triages, and acts on them over time.

The anomaly pipeline becomes most valuable when integrated into a larger scientific workflow. In the Discovery Workbench architecture from Chapter 6, the anomaly pipeline is a discovery module that connects to data sources upstream and to investigation workflows downstream. The integration points are:

• Data ingestion: the pipeline reads from the Workbench's data store (a feature matrix or embedding store populated by the representation learning modules of Chapter 26).
• Drift monitoring: the pipeline includes a drift detector (Section 30.3) that fires alerts when the incoming data distribution shifts, triggering pipeline retraining or investigation.
• Anomaly registry: detected anomalies are stored in a structured registry with their scores, consensus counts, and metadata, forming an audit trail for the provenance system of Chapter 47.
• Hypothesis generation: top anomalies are forwarded to the hypothesis generation module (Chapter 39) for automated investigation.

from dataclasses import dataclass
from datetime import datetime
import json

@dataclass
class AnomalyRecord:
    """A single anomaly entry in the discovery registry."""
    sample_id: str
    anomaly_score: float
    consensus_count: int
    per_method_scores: dict
    detection_timestamp: str
    status: str = 'pending_review'  # pending_review | confirmed | dismissed
    investigation_notes: str = ''
    related_hypothesis_id: str = ''


class AnomalyRegistry:
    """Registry for tracking detected anomalies through investigation.

    Integrates with the Discovery Workbench's provenance system.
    """

    def __init__(self):
        self.records: dict[str, AnomalyRecord] = {}
        self.run_metadata: list[dict] = []

    def register_run(self, report: AnomalyReport,
                     sample_ids: list[str],
                     dataset_name: str) -> str:
        """Register a pipeline run and its top anomalies."""
        run_id = f"anomaly_run_{datetime.now().strftime('%Y%m%d_%H%M%S')}"

        self.run_metadata.append({
            'run_id': run_id,
            'dataset': dataset_name,
            'timestamp': datetime.now().isoformat(),
            'n_samples': report.metadata['n_samples'],
            'n_flagged': report.metadata['n_flagged'],
            'contamination': report.metadata['contamination'],
        })

        # Register each flagged anomaly
        for idx in report.rankings[:report.metadata['n_flagged']]:
            record = AnomalyRecord(
                sample_id=sample_ids[idx],
                anomaly_score=float(report.scores[idx]),
                consensus_count=int(report.consensus_counts[idx]),
                per_method_scores={
                    name: float(scores[idx])
                    for name, scores in report.per_method_scores.items()
                },
                detection_timestamp=datetime.now().isoformat(),
            )
            self.records[sample_ids[idx]] = record

        return run_id

    def update_status(self, sample_id: str, status: str,
                      notes: str = ''):
        """Update the investigation status of an anomaly."""
        if sample_id in self.records:
            self.records[sample_id].status = status
            self.records[sample_id].investigation_notes = notes

    def get_pending(self) -> list[AnomalyRecord]:
        """Get all anomalies pending review, sorted by score."""
        pending = [r for r in self.records.values()
                   if r.status == 'pending_review']
        return sorted(pending, key=lambda r: r.anomaly_score, reverse=True)

    def summary_stats(self) -> dict:
        """Summary statistics for the registry."""
        statuses = [r.status for r in self.records.values()]
        return {
            'total': len(self.records),
            'pending': statuses.count('pending_review'),
            'confirmed': statuses.count('confirmed'),
            'dismissed': statuses.count('dismissed'),
            'confirmation_rate': (
                statuses.count('confirmed') /
                max(1, statuses.count('confirmed') + statuses.count('dismissed'))
            ),
        }

    def export_json(self, filepath: str):
        """Export registry to JSON for provenance tracking."""
        data = {
            'runs': self.run_metadata,
            'records': {
                sid: {
                    'anomaly_score': r.anomaly_score,
                    'consensus': r.consensus_count,
                    'status': r.status,
                    'notes': r.investigation_notes,
                }
                for sid, r in self.records.items()
            }
        }
        with open(filepath, 'w') as f:
            json.dump(data, f, indent=2)


# Usage example
registry = AnomalyRegistry()
sample_ids = [f"sample_{i:04d}" for i in range(len(X))]
run_id = registry.register_run(report, sample_ids, "protein_activity_screen")

# Simulate expert review
pending = registry.get_pending()
print(f"Anomalies pending review: {len(pending)}")
print(f"Top anomaly: {pending[0].sample_id}, "
      f"score={pending[0].anomaly_score:.4f}, "
      f"consensus={pending[0].consensus_count}/3")

# Expert reviews top anomalies
for record in pending[:5]:
    idx = int(record.sample_id.split('_')[1])
    true_label = labels[idx]
    if true_label == 'novel':
        registry.update_status(record.sample_id, 'confirmed',
                              'Novel pattern confirmed by domain expert')
    elif true_label == 'error':
        registry.update_status(record.sample_id, 'dismissed',
                              'Instrument artifact')

print(f"\nRegistry summary: {registry.summary_stats()}")

from pyod.models.suod import SUOD
from pyod.models.iforest import IForest
from pyod.models.lof import LOF
from pyod.models.auto_encoder import AutoEncoder

# Define base detectors
base_estimators = [
    IForest(n_estimators=200, contamination=0.05),
    LOF(n_neighbors=20, contamination=0.05),
    AutoEncoder(hidden_neurons=[20, 64, 32, 16, 32, 64, 20],
                epochs=100, contamination=0.05),
]

# SUOD handles parallelization, normalization, and combination
ensemble = SUOD(base_estimators=base_estimators, n_jobs=-1,
                combination='average', verbose=False)
ensemble.fit(X)
scores = ensemble.decision_function(X)
labels_pred = ensemble.predict(X)
print(f"SUOD flagged {labels_pred.sum()} anomalies")

6. Putting It All Together: The Recipe

Here is the complete recipe for deploying an anomaly discovery pipeline in a scientific setting, from data preparation through investigation. Follow these steps in order; each builds on the previous.

1. Characterize your data: dimensionality, sample size, expected contamination rate, known anomaly types. This determines method selection and hyperparameters. For tabular data under 100 features, the Isolation Forest + LOF + Autoencoder combination used in this section is a strong default. For image or sequence data, replace LOF with a domain-appropriate deep method (e.g., a convolutional autoencoder or a temporal model). For very small datasets (under 200 samples), drop the autoencoder (which needs sufficient training data) and substitute a kernel-based method such as One-Class SVM.
2. Preprocess: standardize features, handle missing values, apply domain-specific transformations (log-transform counts, normalize spectra). If dimensionality exceeds 100, apply Principal Component Analysis (PCA) or use embeddings from a pretrained model (Chapter 26).
3. Run the ensemble: execute 3 to 5 detectors from different algorithmic families. Use rank normalization and average combination.
4. Assess consensus: count how many methods flag each anomaly. Prioritize high-consensus anomalies for investigation.
5. Validate stability: run bootstrap stability analysis. Anomalies that appear consistently across bootstrap samples are more trustworthy than those that appear sporadically.
6. Evaluate with available labels: if any labeled anomalies exist, compute precision@k to calibrate expectations. If not, use synthetic injection to establish a baseline.
7. Monitor for drift: deploy a drift detector on incoming data. When drift is detected, investigate before retraining (the drift may be the discovery).
8. Track through investigation: register anomalies, record expert reviews, and maintain the confirmation rate. A sustained low confirmation rate signals that the pipeline needs recalibration.