Reference / API reference / dorothy / dorothy.analysis / dorothy.analysis.knn_anomaly

dorothy.analysis.knn_anomaly

k-NN anomaly detection for stellar spectra.

A star is an anomaly when its k-th nearest training neighbour in embedding space is farther than a percentile of the same distance over the training set.

The steps: 1. Extract embeddings from the trained model (see Predictor.get_embeddings). 2. L2-normalize the embeddings to unit vectors. 3. Find the k nearest training embeddings by brute-force search. 4. Take the Euclidean distance to the k-th neighbour. 5. Flag the stars whose distance exceeds the percentile threshold.

Examples

>>> detector = AnomalyDetector.from_predictor(predictor)
>>> detector.fit(X_train, k=10, percentile=99)
>>> result = detector.detect(X_test)
>>> anomaly_indices = np.where(result.is_anomaly)[0]

Functions

l2_normalize(X)

Normalize each row of X to unit L2 norm.

Classes

AnomalyDetector(predictor[, layer_index])	k-NN anomaly detector for stellar spectra.
AnomalyResult(distances, is_anomaly, ...)	Result of AnomalyDetector.detect.

dorothy.analysis.knn_anomaly.l2_normalize(X)

Normalize each row of X to unit L2 norm.

Each row is divided by its norm plus 1e-10, so a zero row stays zero.

Parameters:

X (NDArray[float32]) – Array of shape (n_samples, n_features).

Returns:

Array of shape (n_samples, n_features).

Return type:

NDArray[float32]

class dorothy.analysis.knn_anomaly.AnomalyResult(distances, is_anomaly, threshold, k, percentile)

Bases: object

Result of AnomalyDetector.detect.

distances

Distance to the k-th nearest training neighbour, shape (n_samples,). The embeddings are unit vectors, so it is at most 2.

is_anomaly

True where distances exceeds threshold, shape (n_samples,).

threshold

Distance threshold.

k

Number of neighbours.

percentile

Percentile of the training distances that gave the threshold.

property n_samples: int

Number of samples evaluated.

property n_anomalies: int

Number of detected anomalies.

property anomaly_fraction: float

Fraction of samples flagged as anomalies.

get_anomaly_indices()

Indices of the anomalous samples, shape (n_anomalies,).

class dorothy.analysis.knn_anomaly.AnomalyDetector(predictor, layer_index=-2)

Bases: object

k-NN anomaly detector for stellar spectra.

predictor

The predictor that extracts embeddings.

layer_index

Layer of the embeddings.

threshold

Distance threshold; None before fit.

k

Number of neighbours; None before fit.

percentile

Percentile of the training distances that gave the threshold; None before fit.

train_embeddings

Unit-norm training embeddings, shape (n_train, embedding_dim); None before fit.

Examples

>>> detector = AnomalyDetector.from_predictor(predictor)
>>> detector.fit(X_train, k=10, percentile=99)
>>> result = detector.detect(X_test)

classmethod from_predictor(predictor, layer_index=-2)

Create an anomaly detector from a predictor.

Parameters:

• predictor (Predictor) – Trained predictor that extracts the embeddings.

• layer_index (int) – Layer of the embeddings, as in __init__.

Returns:

An unfitted detector.

Return type:

AnomalyDetector

fit(X_train, k=10, percentile=99.0, batch_size=1024)

Fit the detector on training spectra.

The threshold is the given percentile of the distance of each training star to its k-th nearest other training star. The brute-force search takes O(n_train^2 * embedding_dim) time.

Parameters:

• X_train (NDArray[float32]) – Training spectra, shape (n_train, 3, n_wavelengths).

• k (int) – Number of nearest neighbours.

• percentile (float) – Percentile of the training distances, in [0, 100].

• batch_size (int) – Batch size for the embeddings.

Returns:

The detector.

Raises:

ValueError – If k is not an integer in [1, n_train - 1] or percentile is not in [0, 100].

Return type:

AnomalyDetector

detect(X, batch_size=1024)

Flag the anomalies among new spectra.

The brute-force search takes O(n_samples * n_train * embedding_dim) time.

Parameters:

• X (NDArray[float32]) – Spectra, shape (n_samples, 3, n_wavelengths).

• batch_size (int) – Batch size for the embeddings.

Returns:

Distances and anomaly flags.

Raises:

RuntimeError – If the detector is not fitted.

Return type:

AnomalyResult

get_training_distances()

Distance of each training star to its k-th nearest other training star.

This takes O(n_train^2 * embedding_dim) time.

Returns:

Distances, shape (n_train,).

Raises:

RuntimeError – If the detector is not fitted.

Return type:

NDArray[float32]

save(path)

Pickle the fitted state to a file.

The state is the embeddings, threshold, k, percentile and layer index. The predictor is not saved.

Parameters:

path (str | Path) – Output file.

Raises:

RuntimeError – If the detector is not fitted.

load_state(path)

Load a state written by save, replacing the fitted state.

The predictor comes from __init__. Unpickling runs code, so load only files that you trust.

Parameters:

path (str | Path) – File written by save.

Returns:

The detector.

Return type:

AnomalyDetector