Skip to content

gmm

sleap.qc.gmm

Gaussian Mixture Model for anomaly detection.

Classes:

Name Description
GMMDetector

Anomaly detection using GMM likelihood.
ZScoreDetector

Fallback detector using simple z-score thresholding.

GMMDetector

Anomaly detection using GMM likelihood.

Fits a mixture model to clean data and flags low-likelihood instances.

Attributes:

Name Type Description
n_components

Number of Gaussian components.
covariance_type

Covariance type for GMM.
percentile_threshold

Percentile below which instances are anomalies.
model Optional[GaussianMixture]

Fitted GaussianMixture model.
scaler Optional[StandardScaler]

Fitted StandardScaler for feature normalization.
log_likelihood_threshold Optional[float]

Threshold for anomaly detection.
feature_names Optional[list[str]]

Optional list of feature names.

Methods:

Name Description
__init__

Initialize detector.
fit

Fit the GMM on clean data.
score

Score a single instance.
score_batch

Score multiple instances.
Source code in sleap/qc/gmm.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
class GMMDetector:
    """Anomaly detection using GMM likelihood.

    Fits a mixture model to clean data and flags low-likelihood instances.

    Attributes:
        n_components: Number of Gaussian components.
        covariance_type: Covariance type for GMM.
        percentile_threshold: Percentile below which instances are anomalies.
        model: Fitted GaussianMixture model.
        scaler: Fitted StandardScaler for feature normalization.
        log_likelihood_threshold: Threshold for anomaly detection.
        feature_names: Optional list of feature names.
    """

    def __init__(
        self,
        n_components: int = 5,
        covariance_type: str = "full",
        percentile_threshold: float = 5.0,
    ):
        """Initialize detector.

        Args:
            n_components: Number of Gaussian components.
            covariance_type: Covariance type ("full", "tied", "diag", "spherical").
            percentile_threshold: Percentile below which instances are anomalies.
        """
        self.n_components = n_components
        self.covariance_type = covariance_type
        self.percentile_threshold = percentile_threshold

        self.model: Optional[GaussianMixture] = None
        self.scaler: Optional[StandardScaler] = None
        self.log_likelihood_threshold: Optional[float] = None
        self.train_log_likelihoods: Optional[np.ndarray] = None  # For percentiles
        self.feature_names: Optional[list[str]] = None

    def fit(
        self,
        feature_matrix: np.ndarray,
        feature_names: Optional[list[str]] = None,
    ) -> "GMMDetector":
        """Fit the GMM on clean data.

        Args:
            feature_matrix: (N_instances, N_features) array of clean instances.
            feature_names: Optional list of feature names.

        Returns:
            Self for chaining.
        """
        self.feature_names = feature_names

        # Remove instances with NaN
        valid_mask = ~np.isnan(feature_matrix).any(axis=1)
        valid_features = feature_matrix[valid_mask]

        if len(valid_features) < 2:
            raise ValueError("Need at least 2 valid instances to fit GMM")

        # Determine number of components (at most n_samples // 10)
        n_components = min(self.n_components, len(valid_features) // 10)
        if n_components < 1:
            n_components = 1

        # Standardize features
        self.scaler = StandardScaler()
        scaled = self.scaler.fit_transform(valid_features)

        # Fit GMM
        self.model = GaussianMixture(
            n_components=n_components,
            covariance_type=self.covariance_type,
            random_state=42,
        )
        self.model.fit(scaled)

        # Compute threshold and store training log-likelihoods for percentile scoring
        self.train_log_likelihoods = self.model.score_samples(scaled)
        self.log_likelihood_threshold = np.percentile(
            self.train_log_likelihoods, self.percentile_threshold
        )

        return self

    def score(self, feature_vector: np.ndarray) -> dict[str, float]:
        """Score a single instance.

        Args:
            feature_vector: (N_features,) array.

        Returns:
            Dictionary with:
            - log_likelihood: log-likelihood under the model
            - is_anomaly: True if below threshold
            - normalized_score: score normalized to ~0-1 (higher = more anomalous)
            - component_probs: probability of belonging to each component
        """
        if self.model is None or self.scaler is None:
            raise ValueError("Model not fitted. Call fit() first.")

        if np.isnan(feature_vector).any():
            return {
                "log_likelihood": np.nan,
                "is_anomaly": False,
                "normalized_score": np.nan,
                "component_probs": [],
            }

        # Scale and score
        scaled = self.scaler.transform(feature_vector.reshape(1, -1))
        log_likelihood = self.model.score_samples(scaled)[0]
        component_probs = self.model.predict_proba(scaled)[0]

        is_anomaly = log_likelihood < self.log_likelihood_threshold

        # Normalize: convert log-likelihood to percentile-based score
        # Score = 1 - percentile (so low log_ll = high score = anomalous)
        percentile = (self.train_log_likelihoods < log_likelihood).sum() / len(
            self.train_log_likelihoods
        )
        normalized = 1.0 - percentile

        return {
            "log_likelihood": float(log_likelihood),
            "is_anomaly": is_anomaly,
            "normalized_score": float(normalized),
            "component_probs": component_probs.tolist(),
        }

    def score_batch(self, feature_matrix: np.ndarray) -> np.ndarray:
        """Score multiple instances.

        Args:
            feature_matrix: (N_instances, N_features) array.

        Returns:
            (N_instances,) array of normalized anomaly scores (0-1).
        """
        if self.model is None or self.scaler is None:
            raise ValueError("Model not fitted. Call fit() first.")

        scores = np.full(len(feature_matrix), np.nan)

        # Find valid instances
        valid_mask = ~np.isnan(feature_matrix).any(axis=1)
        if valid_mask.sum() == 0:
            return scores

        valid_features = feature_matrix[valid_mask]
        scaled = self.scaler.transform(valid_features)

        log_likelihoods = self.model.score_samples(scaled)

        # Normalize to 0-1 (higher = more anomalous) using percentile-based scoring
        # For each log_ll, compute its percentile in training distribution
        # Score = 1 - percentile (so low log_ll = high score)
        normalized = np.zeros(len(log_likelihoods))
        for i, ll in enumerate(log_likelihoods):
            percentile = (self.train_log_likelihoods < ll).sum() / len(
                self.train_log_likelihoods
            )
            normalized[i] = 1.0 - percentile

        scores[valid_mask] = normalized
        return scores

__init__(n_components=5, covariance_type='full', percentile_threshold=5.0)

Initialize detector.

Parameters:

Name Type Description Default
n_components int

Number of Gaussian components.

 5
covariance_type str

Covariance type ("full", "tied", "diag", "spherical").

 'full'
percentile_threshold float

Percentile below which instances are anomalies.

 5.0
Source code in sleap/qc/gmm.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    n_components: int = 5,
    covariance_type: str = "full",
    percentile_threshold: float = 5.0,
):
    """Initialize detector.

    Args:
        n_components: Number of Gaussian components.
        covariance_type: Covariance type ("full", "tied", "diag", "spherical").
        percentile_threshold: Percentile below which instances are anomalies.
    """
    self.n_components = n_components
    self.covariance_type = covariance_type
    self.percentile_threshold = percentile_threshold

    self.model: Optional[GaussianMixture] = None
    self.scaler: Optional[StandardScaler] = None
    self.log_likelihood_threshold: Optional[float] = None
    self.train_log_likelihoods: Optional[np.ndarray] = None  # For percentiles
    self.feature_names: Optional[list[str]] = None

fit(feature_matrix, feature_names=None)

Fit the GMM on clean data.

Parameters:

Name Type Description Default
feature_matrix ndarray

(N_instances, N_features) array of clean instances.

 required
feature_names Optional[list[str]]

Optional list of feature names.

 None

Returns:

Type Description
'GMMDetector'

Self for chaining.
Source code in sleap/qc/gmm.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def fit(
    self,
    feature_matrix: np.ndarray,
    feature_names: Optional[list[str]] = None,
) -> "GMMDetector":
    """Fit the GMM on clean data.

    Args:
        feature_matrix: (N_instances, N_features) array of clean instances.
        feature_names: Optional list of feature names.

    Returns:
        Self for chaining.
    """
    self.feature_names = feature_names

    # Remove instances with NaN
    valid_mask = ~np.isnan(feature_matrix).any(axis=1)
    valid_features = feature_matrix[valid_mask]

    if len(valid_features) < 2:
        raise ValueError("Need at least 2 valid instances to fit GMM")

    # Determine number of components (at most n_samples // 10)
    n_components = min(self.n_components, len(valid_features) // 10)
    if n_components < 1:
        n_components = 1

    # Standardize features
    self.scaler = StandardScaler()
    scaled = self.scaler.fit_transform(valid_features)

    # Fit GMM
    self.model = GaussianMixture(
        n_components=n_components,
        covariance_type=self.covariance_type,
        random_state=42,
    )
    self.model.fit(scaled)

    # Compute threshold and store training log-likelihoods for percentile scoring
    self.train_log_likelihoods = self.model.score_samples(scaled)
    self.log_likelihood_threshold = np.percentile(
        self.train_log_likelihoods, self.percentile_threshold
    )

    return self

score(feature_vector)

Score a single instance.

Parameters:

Name Type Description Default
feature_vector ndarray

(N_features,) array.

 required

Returns:

Type Description
dict[str, float]

Dictionary with: - log_likelihood: log-likelihood under the model - is_anomaly: True if below threshold - normalized_score: score normalized to ~0-1 (higher = more anomalous) - component_probs: probability of belonging to each component
Source code in sleap/qc/gmm.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def score(self, feature_vector: np.ndarray) -> dict[str, float]:
    """Score a single instance.

    Args:
        feature_vector: (N_features,) array.

    Returns:
        Dictionary with:
        - log_likelihood: log-likelihood under the model
        - is_anomaly: True if below threshold
        - normalized_score: score normalized to ~0-1 (higher = more anomalous)
        - component_probs: probability of belonging to each component
    """
    if self.model is None or self.scaler is None:
        raise ValueError("Model not fitted. Call fit() first.")

    if np.isnan(feature_vector).any():
        return {
            "log_likelihood": np.nan,
            "is_anomaly": False,
            "normalized_score": np.nan,
            "component_probs": [],
        }

    # Scale and score
    scaled = self.scaler.transform(feature_vector.reshape(1, -1))
    log_likelihood = self.model.score_samples(scaled)[0]
    component_probs = self.model.predict_proba(scaled)[0]

    is_anomaly = log_likelihood < self.log_likelihood_threshold

    # Normalize: convert log-likelihood to percentile-based score
    # Score = 1 - percentile (so low log_ll = high score = anomalous)
    percentile = (self.train_log_likelihoods < log_likelihood).sum() / len(
        self.train_log_likelihoods
    )
    normalized = 1.0 - percentile

    return {
        "log_likelihood": float(log_likelihood),
        "is_anomaly": is_anomaly,
        "normalized_score": float(normalized),
        "component_probs": component_probs.tolist(),
    }

score_batch(feature_matrix)

Score multiple instances.

Parameters:

Name Type Description Default
feature_matrix ndarray

(N_instances, N_features) array.

 required

Returns:

Type Description
ndarray

(N_instances,) array of normalized anomaly scores (0-1).
Source code in sleap/qc/gmm.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def score_batch(self, feature_matrix: np.ndarray) -> np.ndarray:
    """Score multiple instances.

    Args:
        feature_matrix: (N_instances, N_features) array.

    Returns:
        (N_instances,) array of normalized anomaly scores (0-1).
    """
    if self.model is None or self.scaler is None:
        raise ValueError("Model not fitted. Call fit() first.")

    scores = np.full(len(feature_matrix), np.nan)

    # Find valid instances
    valid_mask = ~np.isnan(feature_matrix).any(axis=1)
    if valid_mask.sum() == 0:
        return scores

    valid_features = feature_matrix[valid_mask]
    scaled = self.scaler.transform(valid_features)

    log_likelihoods = self.model.score_samples(scaled)

    # Normalize to 0-1 (higher = more anomalous) using percentile-based scoring
    # For each log_ll, compute its percentile in training distribution
    # Score = 1 - percentile (so low log_ll = high score)
    normalized = np.zeros(len(log_likelihoods))
    for i, ll in enumerate(log_likelihoods):
        percentile = (self.train_log_likelihoods < ll).sum() / len(
            self.train_log_likelihoods
        )
        normalized[i] = 1.0 - percentile

    scores[valid_mask] = normalized
    return scores

ZScoreDetector

Fallback detector using simple z-score thresholding.

Used when there are too few samples for GMM fitting.

Methods:

Name Description
__init__

Initialize detector.
fit

Compute mean and std from reference data.
score_batch

Score instances by max z-score.
Source code in sleap/qc/gmm.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
class ZScoreDetector:
    """Fallback detector using simple z-score thresholding.

    Used when there are too few samples for GMM fitting.
    """

    def __init__(self, threshold: float = 3.0):
        """Initialize detector.

        Args:
            threshold: Z-score threshold for flagging.
        """
        self.threshold = threshold
        self.means: Optional[np.ndarray] = None
        self.stds: Optional[np.ndarray] = None

    def fit(self, feature_matrix: np.ndarray) -> "ZScoreDetector":
        """Compute mean and std from reference data.

        Args:
            feature_matrix: (N_instances, N_features) array.

        Returns:
            Self for chaining.
        """
        valid_mask = ~np.isnan(feature_matrix).any(axis=1)
        valid_features = feature_matrix[valid_mask]

        self.means = np.mean(valid_features, axis=0)
        self.stds = np.std(valid_features, axis=0)
        # Avoid division by zero
        self.stds = np.maximum(self.stds, 1e-6)

        return self

    def score_batch(self, feature_matrix: np.ndarray) -> np.ndarray:
        """Score instances by max z-score.

        Args:
            feature_matrix: (N_instances, N_features) array.

        Returns:
            (N_instances,) array of normalized anomaly scores (0-1).
        """
        if self.means is None or self.stds is None:
            raise ValueError("Model not fitted. Call fit() first.")

        scores = np.full(len(feature_matrix), np.nan)
        valid_mask = ~np.isnan(feature_matrix).any(axis=1)

        if valid_mask.sum() == 0:
            return scores

        valid_features = feature_matrix[valid_mask]

        # Compute z-scores
        zscores = np.abs((valid_features - self.means) / self.stds)
        max_zscores = np.max(zscores, axis=1)

        # Normalize to 0-1 using sigmoid around threshold
        diff = max_zscores - self.threshold
        normalized = 1 / (1 + np.exp(-diff))

        scores[valid_mask] = normalized
        return scores

__init__(threshold=3.0)

Initialize detector.

Parameters:

Name Type Description Default
threshold float

Z-score threshold for flagging.

 3.0
Source code in sleap/qc/gmm.py 
187
188
189
190
191
192
193
194
195
def __init__(self, threshold: float = 3.0):
    """Initialize detector.

    Args:
        threshold: Z-score threshold for flagging.
    """
    self.threshold = threshold
    self.means: Optional[np.ndarray] = None
    self.stds: Optional[np.ndarray] = None

fit(feature_matrix)

Compute mean and std from reference data.

Parameters:

Name Type Description Default
feature_matrix ndarray

(N_instances, N_features) array.

 required

Returns:

Type Description
'ZScoreDetector'

Self for chaining.
Source code in sleap/qc/gmm.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def fit(self, feature_matrix: np.ndarray) -> "ZScoreDetector":
    """Compute mean and std from reference data.

    Args:
        feature_matrix: (N_instances, N_features) array.

    Returns:
        Self for chaining.
    """
    valid_mask = ~np.isnan(feature_matrix).any(axis=1)
    valid_features = feature_matrix[valid_mask]

    self.means = np.mean(valid_features, axis=0)
    self.stds = np.std(valid_features, axis=0)
    # Avoid division by zero
    self.stds = np.maximum(self.stds, 1e-6)

    return self

score_batch(feature_matrix)

Score instances by max z-score.

Parameters:

Name Type Description Default
feature_matrix ndarray

(N_instances, N_features) array.

 required

Returns:

Type Description
ndarray

(N_instances,) array of normalized anomaly scores (0-1).
Source code in sleap/qc/gmm.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def score_batch(self, feature_matrix: np.ndarray) -> np.ndarray:
    """Score instances by max z-score.

    Args:
        feature_matrix: (N_instances, N_features) array.

    Returns:
        (N_instances,) array of normalized anomaly scores (0-1).
    """
    if self.means is None or self.stds is None:
        raise ValueError("Model not fitted. Call fit() first.")

    scores = np.full(len(feature_matrix), np.nan)
    valid_mask = ~np.isnan(feature_matrix).any(axis=1)

    if valid_mask.sum() == 0:
        return scores

    valid_features = feature_matrix[valid_mask]

    # Compute z-scores
    zscores = np.abs((valid_features - self.means) / self.stds)
    max_zscores = np.max(zscores, axis=1)

    # Normalize to 0-1 using sigmoid around threshold
    diff = max_zscores - self.threshold
    normalized = 1 / (1 + np.exp(-diff))

    scores[valid_mask] = normalized
    return scores