"""Semi-supervised learning: do the 4,058 unlabeled weathered spectra help?"""
from __future__ import annotations
from typing import Any
import numpy as np
from sklearn.base import BaseEstimator, clone
from sklearn.decomposition import PCA
from sklearn.metrics import accuracy_score, f1_score, matthews_corrcoef
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.semi_supervised import LabelSpreading, SelfTrainingClassifier
from .calibration import make_base_pipeline
from .data import loso_splits
[docs]
def self_train(
estimator: BaseEstimator,
X_lab: np.ndarray,
y_lab: np.ndarray,
X_unlab: np.ndarray,
threshold: float = 0.90,
max_iter: int = 3,
) -> SelfTrainingClassifier:
base = make_base_pipeline(estimator, needs_proba=True)
model = SelfTrainingClassifier(base, threshold=threshold, max_iter=max_iter)
X = np.vstack([X_lab, X_unlab])
y = np.concatenate([y_lab, np.full(len(X_unlab), -1)])
return model.fit(X, y)
[docs]
def label_spread(
X_lab: np.ndarray,
y_lab: np.ndarray,
X_unlab: np.ndarray,
n_neighbors: int = 10,
pca_components: int = 50,
random_state: int = 42,
) -> tuple[LabelSpreading, Pipeline]:
"""LabelSpreading with a sparse kNN kernel on PCA-reduced features.
Returns the fitted spreader plus the scaler+PCA embedding used, so test
data can be transformed into the same space for inductive prediction.
"""
embed = Pipeline([
("scaler", StandardScaler()),
("pca", PCA(n_components=pca_components, random_state=random_state)),
])
Z = embed.fit_transform(np.vstack([X_lab, X_unlab]))
y = np.concatenate([y_lab, np.full(len(X_unlab), -1)])
spreader = LabelSpreading(kernel="knn", n_neighbors=n_neighbors, max_iter=50)
spreader.fit(Z, y)
return spreader, embed
def _fold_metrics(y_true: np.ndarray, y_pred: np.ndarray) -> dict[str, float]:
return {
"accuracy": float(accuracy_score(y_true, y_pred)),
"macro_f1": float(f1_score(y_true, y_pred, average="macro", zero_division=0)),
"mcc": float(matthews_corrcoef(y_true, y_pred)),
}
[docs]
def ssl_loso_experiment(
estimator: BaseEstimator,
X_lab: np.ndarray,
y_lab: np.ndarray,
study: np.ndarray,
X_unlab: np.ndarray,
threshold: float = 0.90,
max_iter: int = 3,
n_neighbors: int = 10,
pca_components: int = 50,
random_state: int = 42,
) -> list[dict[str, Any]]:
"""Per LOSO fold: supervised-only vs self-training vs label-spreading,
all evaluated on the held-out study."""
rows: list[dict[str, Any]] = []
for held, train_idx, test_idx in loso_splits(study):
X_tr, y_tr = X_lab[train_idx], y_lab[train_idx]
X_te, y_te = X_lab[test_idx], y_lab[test_idx]
supervised = make_base_pipeline(clone(estimator), needs_proba=True).fit(X_tr, y_tr)
rows.append({"fold": held, "method": "supervised",
**_fold_metrics(y_te, supervised.predict(X_te))})
st = self_train(estimator, X_tr, y_tr, X_unlab, threshold, max_iter)
rows.append({"fold": held, "method": "self_training",
**_fold_metrics(y_te, st.predict(X_te)),
"pseudo_labeled": int(np.sum(st.labeled_iter_ > 0))})
spreader, embed = label_spread(
X_tr, y_tr, X_unlab,
n_neighbors=n_neighbors, pca_components=pca_components,
random_state=random_state,
)
rows.append({"fold": held, "method": "label_spreading",
**_fold_metrics(y_te, spreader.predict(embed.transform(X_te)))})
return rows