Quick Start

Installation

pip install process-improve

PCA Example

import pandas as pd
from process_improve.multivariate.methods import PCA, MCUVScaler

# Load and scale data
X = pd.read_csv("your_data.csv", index_col=0)
scaler = MCUVScaler().fit(X)
X_scaled = scaler.transform(X)

# Fit model
pca = PCA(n_components=3).fit(X_scaled)

# Results
pca.scores_  # Score matrix (N x A)
pca.loadings_  # Loading matrix (K x A)
pca.r2_cumulative_  # Cumulative R² per component

# Diagnostics: the T² limit for the rows the model was fitted to (see below)
pca.detect_outliers(judged_rows="fitted")
pca.score_contributions(X_scaled, component=1)  # per-variable, sums to t1

# Plots
pca.score_plot(judged_rows="fitted")
pca.loading_plot()
pca.spe_plot()
pca.t2_plot(judged_rows="fitted")

The rows a model was fitted to and new rows are judged against different Hotelling’s T² limits: a fitted row helped estimate the centre and the spread of the scores, so it cannot lie as far out as a new row can. judged_rows says which: "fitted" for hotellings_t2_, a score plot or outlier detection, "new" for rows the model has not seen. Left out, it gives the limit for new rows until 3.0 and the one for fitted rows from 3.0, with a FutureWarning until then.

PLS Example

from process_improve.multivariate.methods import PLS, MCUVScaler

# Scale X and Y separately
scaler_x = MCUVScaler().fit(X)
scaler_y = MCUVScaler().fit(Y)
X_scaled = scaler_x.transform(X)
Y_scaled = scaler_y.transform(Y)

# Fit model
pls = PLS(n_components=3).fit(X_scaled, Y_scaled)

# Predict new observations (sklearn-compatible: returns the y_hat array)
y_pred = pls.predict(scaler_x.transform(X_new))

# Predict with full per-row diagnostics (scores, T², SPE, plus y_hat)
result = pls.diagnose(scaler_x.transform(X_new))
result.y_hat  # Predicted Y
result.spe  # SPE diagnostics
result.cumulative_hotellings_t2  # Hotelling's T², one column per component
pls.hotellings_t2_limit(judged_rows="new")  # the limit for these new rows

# Diagnostics
pls.detect_outliers(judged_rows="fitted")
pls.score_contributions(X_scaled, component=1)  # per-variable, sums to t1

Component Selection

Use cross-validation to select the number of PCA components. Pass the raw X, not X_scaled: the selector fits the centring and scaling inside each fold, and warns if it is handed a block that is already scaled.

result = PCA.select_n_components(X, max_components=10)
print(f"Recommended: {result.n_components} components")
print(f"PRESS ratios: {result.press_ratio}")

On-line Monitoring (Adaptive PCA / PLS)

A static model goes stale as soon as the process drifts. AdaptivePCA and AdaptivePLS start from an initial fit, then keep learning as data streams in, one observation at a time.

from process_improve.multivariate import AdaptivePCA

# Seed on a block of known-good ("common cause") data
monitor = AdaptivePCA(n_components=3).fit(X_reference)

# Feed live observations one row at a time
for _, row in X_stream.iterrows():
    result = monitor.update(row.to_numpy())
    if not result.in_control:
        print(f"Out of control: SPE={result.spe:.2f}, T2={result.hotellings_t2:.2f}")

# Drift diagnostics accumulated over the stream
monitor.distance_  # subspace drift, in units of components
monitor.center_shift_  # operating-point migration, in training-SD units

AdaptivePLS does the same for regression and soft sensing, and supports infrequently-sampled responses: the X-space model adapts every step while the regression part waits for the next response.

Optimal & OMARS Designs

Beyond the textbook designs, ask for an OMARS (orthogonal minimally aliased response surface) or an optimal design, then grade its quality:

from process_improve.experiments import Factor, generate_design, evaluate_design

factors = [
    Factor(name="A", low=-1, high=1),
    Factor(name="B", low=-1, high=1),
    Factor(name="C", low=-1, high=1),
]

# OMARS: main effects clear of every second-order term
design = generate_design(factors, design_type="omars")

# A run-budgeted D-optimal design, scored on every quality metric
d_opt = generate_design(factors, design_type="d_optimal", budget=14)
metrics = evaluate_design(d_opt, metric="all")  # D/I/G-efficiency, aliasing, ...

DOE Strategy Example

Plan a multi-stage experimental strategy before running any experiments:

from process_improve.experiments.factor import Factor, Response
from process_improve.experiments.strategy import recommend_strategy

factors = [
    Factor(name="Temperature", low=25, high=40, units="degC"),
    Factor(name="pH", low=5.0, high=7.5),
    Factor(name="Glucose", low=10, high=50, units="g/L"),
    Factor(name="Yeast extract", low=1, high=10, units="g/L"),
    Factor(name="Agitation", low=100, high=400, units="rpm"),
    Factor(name="Aeration", low=0.5, high=2.0, units="vvm"),
    Factor(name="Inoculum", low=2, high=10, units="%v/v"),
]

result = recommend_strategy(
    factors=factors,
    responses=[Response(name="Yield", goal="maximize", units="g/L")],
    budget=40,
    domain="fermentation",
)

for stage in result["stages"]:
    print(
        f"Stage {stage['stage_number']}: {stage['stage_name']} "
        f"({stage['design_type']}, {stage['estimated_runs']} runs)"
    )