Skip to content

StudentTMixture API

StudentTMixture is a robust alternative to GaussianMixture (Peel & McLachlan, 2000): each component is a multivariate Student-t with mean, scale matrix, and degrees of freedom, fit by ECM. See the Other Mixture Families user guide page for when to reach for it over GaussianMixture.

StudentTMixture

Bases: BaseMixture

A multivariate Student-t Mixture Model, fit by ECM (Expectation Conditional Maximization; Peel & McLachlan, 2000).

Each component has a mean mu_k, scale matrix Sigma_k (covariances_), and degrees of freedom nu_k (dof_). The extra per-sample weight :math:u_{ik} = (\nu_k + d) / (\nu_k + \delta_{ik}) downweights points far from a component's center (small u) relative to a Gaussian fit, which is what gives the t-mixture its robustness to outliers.

Shares its entire EM/CEM/SEM/SAEM/online-EM loop, n_init best-run selection, predict/predict_proba/score/bic/aic/icl, and generic sample with every other BaseMixture subclass -- see that class. Gibbs-sampling mode (n_components=None) and supervised fitting (fit(X, labels=...)) are intentionally Gaussian-only; neither is available here.

Parameters:

Name Type Description Default
n_components int

Number of mixture components.

required
n_features int

Dimensionality of the data. Inferred from X at fit time if None.

None
covariance_type str

One of 'full', 'diag', 'spherical', 'tied_full' -- the GPCM parameterizations (Phase 3) and Gibbs-only axis-aligned-tied types are not supported here.

'full'
dof (per_component, shared)

'per_component' fits one nu_k per component (closed-form bisection+Newton root-find each M-step); 'shared' fits one nu for every component (the bracketed digamma statistic is summed over k before solving); a float fixes nu at that value for every component for the whole fit (no root-find at all -- e.g. dof=1e4 recovers an (approximately) Gaussian fit, used by the parity test against GaussianMixture).

'per_component'
dof_init float

Initial nu_k for every component, before the first M-step.

5.0
dof_bounds tuple of float

(lo, hi) clamp applied to every fitted nu_k.

(0.1, 1000.0)
scale_update (ecm, ecme)

'ecm': :math:\Sigma_k = \sum_i \tau_{ik} u_{ik} (x_i-\mu_k)(x_i-\mu_k)^T / \sum_i \tau_{ik} (the standard ECM update). 'ecme': divide by :math:\sum_i \tau_{ik} u_{ik} instead, which converges faster (Liu & Rubin, 1995) but is a different fixed point in general.

'ecm'
init_means see `BaseMixture`.
'kmeans'
init_weights see `BaseMixture`.
'kmeans'
init_covariances str or Tensor

Initial Sigma_k: 'empirical' (per-component empirical covariance from a nearest-mean hard assignment), 'global' (the whole dataset's own covariance, shared as the starting point for every component), 'eye', or an explicit tensor.

'empirical'
max_iter int
1000
tol int
1000
reg_covar int
1000
n_init int
1000
random_state int
1000
warm_start int
1000
algorithm int
1000
batch_size Optional[int]
None
temperature Optional[int]
None
convergence Optional[int]
None
step_size Optional[int]
None
sem_burn_in Optional[int]
None
sem_estimate see `BaseMixture`.
'best'
verbose see `BaseMixture`.
'best'
verbose_interval see `BaseMixture`.
'best'
device see `BaseMixture`.
'best'

Attributes:

Name Type Description
weights_ (Tensor, shape(n_components))
means_ (Tensor, shape(n_components, n_features))
covariances_ Tensor

Shape depends on covariance_type, same convention as GaussianMixture.

dof_ (Tensor, shape(n_components))

Fitted nu_k -- always one value per component, even when dof='shared' or a fixed float (all entries equal in that case).

Source code in tgmm/mixtures/student_t.py
 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
179
180
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
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
class StudentTMixture(BaseMixture):
    r"""
    A multivariate Student-t Mixture Model, fit by ECM (Expectation
    Conditional Maximization; Peel & McLachlan, 2000).

    Each component has a mean `mu_k`, scale matrix `Sigma_k`
    (`covariances_`), and degrees of freedom `nu_k` (`dof_`). The extra
    per-sample weight :math:`u_{ik} = (\nu_k + d) / (\nu_k + \delta_{ik})`
    downweights points far from a component's center (small `u`) relative
    to a Gaussian fit, which is what gives the t-mixture its robustness to
    outliers.

    Shares its entire EM/CEM/SEM/SAEM/online-EM loop, `n_init` best-run
    selection, `predict`/`predict_proba`/`score`/`bic`/`aic`/`icl`, and
    generic `sample` with every other `BaseMixture` subclass -- see that
    class. Gibbs-sampling mode (`n_components=None`) and supervised fitting
    (`fit(X, labels=...)`) are intentionally Gaussian-only; neither is
    available here.

    Parameters
    ----------
    n_components : int
        Number of mixture components.
    n_features : int, optional
        Dimensionality of the data. Inferred from `X` at fit time if `None`.
    covariance_type : str, default='full'
        One of `'full'`, `'diag'`, `'spherical'`, `'tied_full'` -- the
        GPCM parameterizations (Phase 3) and Gibbs-only axis-aligned-tied
        types are not supported here.
    dof : {'per_component', 'shared'} or float, default='per_component'
        `'per_component'` fits one `nu_k` per component (closed-form
        bisection+Newton root-find each M-step); `'shared'` fits one `nu`
        for every component (the bracketed digamma statistic is summed
        over `k` before solving); a float fixes `nu` at that value for
        every component for the whole fit (no root-find at all -- e.g.
        `dof=1e4` recovers an (approximately) Gaussian fit, used by the
        parity test against `GaussianMixture`).
    dof_init : float, default=5.0
        Initial `nu_k` for every component, before the first M-step.
    dof_bounds : tuple of float, default=(0.1, 1000.0)
        `(lo, hi)` clamp applied to every fitted `nu_k`.
    scale_update : {'ecm', 'ecme'}, default='ecm'
        `'ecm'`: :math:`\Sigma_k = \sum_i \tau_{ik} u_{ik} (x_i-\mu_k)(x_i-\mu_k)^T / \sum_i \tau_{ik}`
        (the standard ECM update). `'ecme'`: divide by
        :math:`\sum_i \tau_{ik} u_{ik}` instead, which converges faster
        (Liu & Rubin, 1995) but is a different fixed point in general.
    init_means, init_weights : see `BaseMixture`.
    init_covariances : str or torch.Tensor, default='empirical'
        Initial `Sigma_k`: `'empirical'` (per-component empirical
        covariance from a nearest-mean hard assignment), `'global'` (the
        whole dataset's own covariance, shared as the starting point for
        every component), `'eye'`, or an explicit tensor.
    max_iter, tol, reg_covar, n_init, random_state, warm_start, algorithm,
    batch_size, temperature, convergence, step_size, sem_burn_in,
    sem_estimate, verbose, verbose_interval, device : see `BaseMixture`.

    Attributes
    ----------
    weights_ : torch.Tensor, shape (n_components,)
    means_ : torch.Tensor, shape (n_components, n_features)
    covariances_ : torch.Tensor
        Shape depends on `covariance_type`, same convention as
        `GaussianMixture`.
    dof_ : torch.Tensor, shape (n_components,)
        Fitted `nu_k` -- always one value per component, even when
        `dof='shared'` or a fixed float (all entries equal in that case).
    """

    def __init__(
        self,
        n_components: int,
        n_features: Optional[int] = None,
        covariance_type: str = 'full',
        dof: Union[str, float] = 'per_component',
        dof_init: float = 5.0,
        dof_bounds: Tuple[float, float] = (0.1, 1000.0),
        scale_update: str = 'ecm',
        max_iter: int = 1000,
        tol: float = 1e-4,
        reg_covar: float = 1e-6,
        n_init: int = 1,
        init_means='kmeans',
        init_weights='uniform',
        init_covariances='empirical',
        random_state: Optional[int] = None,
        warm_start: bool = False,
        algorithm: str = 'em',
        batch_size: Optional[int] = None,
        temperature: Optional[Tuple[float, int]] = None,
        convergence: str = 'relative',
        step_size=None,
        sem_burn_in: Optional[int] = None,
        sem_estimate: str = 'best',
        components_per_class=1,
        unlabeled_weight: float = 1.0,
        label_confidence: float = 1.0,
        verbose: bool = False,
        verbose_interval: int = 10,
        device: Optional[str] = None,
    ):
        super().__init__(
            n_components=n_components, n_features=n_features, max_iter=max_iter, tol=tol,
            reg_covar=reg_covar, n_init=n_init, init_means=init_means, init_weights=init_weights,
            random_state=random_state, warm_start=warm_start, algorithm=algorithm,
            batch_size=batch_size, temperature=temperature, convergence=convergence,
            step_size=step_size, sem_burn_in=sem_burn_in, sem_estimate=sem_estimate,
            components_per_class=components_per_class, unlabeled_weight=unlabeled_weight,
            label_confidence=label_confidence,
            verbose=verbose, verbose_interval=verbose_interval, device=device,
        )

        if covariance_type not in _VALID_COVARIANCE_TYPES:
            raise ValueError(f"covariance_type must be one of {_VALID_COVARIANCE_TYPES}, got {covariance_type!r}")
        self.covariance_type = covariance_type
        self.init_covariances = init_covariances

        if not (isinstance(dof, (int, float)) or dof in ('per_component', 'shared')):
            raise ValueError(f"dof must be 'per_component', 'shared', or a positive float, got {dof!r}")
        if isinstance(dof, (int, float)) and dof <= 0:
            raise ValueError(f"dof must be > 0, got {dof}.")
        self.dof = dof
        self.dof_init = dof_init
        if not (0.0 < dof_bounds[0] < dof_bounds[1]):
            raise ValueError(f"dof_bounds must be (lo, hi) with 0 < lo < hi, got {dof_bounds!r}")
        self.dof_bounds = dof_bounds

        if scale_update not in _VALID_SCALE_UPDATES:
            raise ValueError(f"scale_update must be one of {_VALID_SCALE_UPDATES}, got {scale_update!r}")
        self.scale_update = scale_update

        self.dof_ = None

    # ===================================================================
    # Parameter allocation
    # ===================================================================
    def _allocate_parameters(self, X: Optional[torch.Tensor] = None, set_random_state: bool = True):
        if set_random_state and self.random_state is not None:
            torch.manual_seed(self.random_state)
            if torch.cuda.is_available():
                torch.cuda.manual_seed_all(self.random_state)

        # ---- means ----
        if isinstance(self.init_means, torch.Tensor):
            if self.init_means.shape != (self.n_components, self.n_features):
                raise ValueError(
                    f"init_means tensor must be shape ({self.n_components}, {self.n_features}), "
                    f"got {self.init_means.shape}."
                )
            self.means_ = self.init_means.to(device=self.device, dtype=self.dtype)
        elif isinstance(self.init_means, str):
            if X is None:
                self.means_ = torch.randn(self.n_components, self.n_features, device=self.device, dtype=self.dtype)
            else:
                X_cpu = X.cpu()
                method = self.init_means.lower()
                if method == 'kmeans':
                    self.means_ = GMMInitializer.kmeans(X_cpu, self.n_components).to(self.device)
                elif method == 'kpp':
                    self.means_ = GMMInitializer.kpp(X_cpu, self.n_components).to(self.device)
                elif method == 'points':
                    self.means_ = GMMInitializer.points(X_cpu, self.n_components).to(self.device)
                elif method == 'maxdist':
                    self.means_ = GMMInitializer.maxdist(X_cpu, self.n_components).to(self.device)
                elif method == 'random':
                    self.means_ = GMMInitializer.random(X_cpu, self.n_components).to(self.device)
                else:
                    raise ValueError(
                        f"Unsupported init_means method: {method!r}. "
                        "Supported: 'kmeans', 'kpp', 'points', 'maxdist', 'random'."
                    )
        else:
            raise TypeError(f"init_means must be a string or torch.Tensor, got {type(self.init_means)}")
        self.means_ = self.means_.to(dtype=self.dtype)

        # ---- weights ----
        if isinstance(self.init_weights, torch.Tensor):
            if self.init_weights.shape != (self.n_components,):
                raise ValueError(f"init_weights tensor must be shape ({self.n_components},), got {self.init_weights.shape}.")
            weights = self.init_weights.to(device=self.device, dtype=self.dtype)
            if torch.sum(weights) < 1e-20:
                raise ValueError("Initial weights must sum to > 0.")
            self.weights_ = weights / torch.sum(weights)
        elif isinstance(self.init_weights, str):
            method = self.init_weights.lower()
            if method in ('uniform', 'equal'):
                self.weights_ = GMMInitializer.init_weights_uniform(self.n_components, self.device, dtype=self.dtype)
            elif method == 'random':
                self.weights_ = GMMInitializer.init_weights_random(self.n_components, self.device, dtype=self.dtype)
            elif method == 'kmeans':
                if X is None:
                    warnings.warn("'kmeans' weight initialization requires data. Falling back to uniform weights.", UserWarning)
                    self.weights_ = GMMInitializer.init_weights_uniform(self.n_components, self.device, dtype=self.dtype)
                else:
                    self.weights_ = GMMInitializer.init_weights_from_clusters(X.to(self.means_.device), self.means_).to(self.device)
            else:
                raise ValueError(f"Unsupported init_weights method: {method!r}. Supported: 'uniform', 'random', 'kmeans'.")
        else:
            raise TypeError(f"init_weights must be a string or torch.Tensor, got {type(self.init_weights)}")

        # ---- scale (covariances_) ----
        if isinstance(self.init_covariances, torch.Tensor):
            self.covariances_ = self.init_covariances.to(device=self.device, dtype=self.dtype)
        elif isinstance(self.init_covariances, str):
            method = self.init_covariances.lower()
            if method == 'empirical':
                if X is None:
                    raise ValueError("init_covariances='empirical' requires data (X).")
                self.covariances_ = GMMInitializer.init_covariances_empirical(
                    X, self.means_, self.covariance_type, self.reg_covar
                ).to(device=self.device, dtype=self.dtype)
            elif method == 'global':
                if X is None:
                    raise ValueError("init_covariances='global' requires data (X).")
                self.covariances_ = GMMInitializer.init_covariances_global(
                    X, self.n_components, self.covariance_type, self.reg_covar
                ).to(device=self.device, dtype=self.dtype)
            elif method == 'eye':
                self.covariances_ = GMMInitializer.init_covariances_eye(
                    self.n_components, self.n_features, self.covariance_type, self.reg_covar,
                    self.device, dtype=self.dtype,
                )
            else:
                raise ValueError(f"Unsupported init_covariances method: {method!r}. Supported: 'empirical', 'global', 'eye'.")
        else:
            raise TypeError(f"init_covariances must be a string or torch.Tensor, got {type(self.init_covariances)}")

        # ---- degrees of freedom ----
        if isinstance(self.dof, (int, float)):
            dof_value = float(self.dof)
        else:
            dof_value = float(self.dof_init)
        self.dof_ = torch.full((self.n_components,), dof_value, device=self.device, dtype=self.dtype)

        # A fresh allocation starts a fresh fit's bookkeeping -- without
        # this, n_init > 1 could leak a prior run's converged_=True onto a
        # later run that never actually converged (see GaussianMixture's
        # own _allocate_parameters, which does the same reset).
        self.fitted_ = False
        self.converged_ = False
        self.n_iter_ = 0
        self.lower_bound_ = -float("inf")

    # ===================================================================
    # Snapshotting: dof_ is the one extra fitted tensor beyond
    # weights_/means_/covariances_ -- see BaseMixture._get_param_snapshot.
    # ===================================================================
    def _get_param_snapshot(self) -> dict:
        snapshot = super()._get_param_snapshot()
        snapshot['dof_'] = self.dof_.clone()
        return snapshot

    def _is_tied_covariance(self) -> bool:
        return self.covariance_type == 'tied_full'

    # ===================================================================
    # E-step hook
    # ===================================================================
    def _component_log_prob(self, X: torch.Tensor) -> torch.Tensor:
        r"""
        .. math::
            \log t(x|\mu_k,\Sigma_k,\nu_k) = \ln\Gamma\!\left(\tfrac{\nu_k+d}{2}\right)
                - \ln\Gamma\!\left(\tfrac{\nu_k}{2}\right) - \tfrac{d}{2}\ln(\nu_k\pi)
                - \tfrac12 \ln|\Sigma_k| - \tfrac{\nu_k+d}{2}\ln\!\left(1 + \tfrac{\delta_{ik}}{\nu_k}\right)
        """
        delta, logdet = _mahalanobis_and_logdet(X, self.means_, self.covariances_, self.covariance_type)
        d = self.n_features
        nu = self.dof_  # (K,)

        const = (
            torch.lgamma((nu + d) / 2) - torch.lgamma(nu / 2)
            - (d / 2.0) * torch.log(nu * math.pi) - 0.5 * logdet
        )  # (K,)
        tail = -((nu + d) / 2).unsqueeze(0) * torch.log1p(delta / nu.unsqueeze(0))  # (n, K)
        return const.unsqueeze(0) + tail

    # ===================================================================
    # M-step hooks
    # ===================================================================
    def _compute_stats(self, X: torch.Tensor, resp: torch.Tensor, batch_size: Optional[int] = None) -> SufficientStats:
        r"""
        Accumulates the ECM statistics: :math:`s1 = \sum_i \tau_{ik} u_{ik} x_i`,
        :math:`s2 = \sum_i \tau_{ik} u_{ik} (x_i - \bar y_k)(x_i - \bar y_k)^T`
        (centered on :math:`\bar y_k = s1_k / \sum_i \tau_{ik} u_{ik}`, this
        E-step's own u-weighted mean estimate), plus `aux['tau_u_sum']`
        (:math:`\sum_i \tau_{ik} u_{ik}`) and `aux['log_u_minus_u']`
        (:math:`\sum_i \tau_{ik}(\log u_{ik} - u_{ik})`), the statistic the
        `nu_k` root-find needs. `nk` keeps its usual meaning,
        :math:`\sum_i \tau_{ik}`.

        `batch_size` is accepted for interface parity with
        `BaseMixture._m_step`/online-EM but not separately chunked here
        (unlike `GaussianMixture._compute_stats`): Student-t's per-sample
        `u_ik` already requires `delta`, itself an `(n, K)` tensor, in
        memory at once, so chunking rows would save only the final
        reduction, not the dominant cost.
        """
        n_samples, d = X.shape
        delta, _ = _mahalanobis_and_logdet(X, self.means_, self.covariances_, self.covariance_type)
        nu = self.dof_  # (K,)
        u = (nu.unsqueeze(0) + d) / (nu.unsqueeze(0) + delta)  # (n, K)

        tau = resp
        tau_u = tau * u

        nk = tau.sum(dim=0)
        m_k = tau_u.sum(dim=0)  # sum_i tau_ik u_ik
        s1 = tau_u.t() @ X  # (K, d)
        ybar = s1 / (m_k + 1e-20).unsqueeze(1)

        log_u_minus_u = (tau * (torch.log(u) - u)).sum(dim=0)  # (K,)

        base_type = self.covariance_type.replace('tied_', '', 1)
        diff = X.unsqueeze(1) - ybar.unsqueeze(0)  # (n, K, d)
        if base_type == 'full':
            s2 = torch.einsum('nk,nkd,nke->kde', tau_u, diff, diff)
        elif base_type == 'diag':
            s2 = torch.einsum('nk,nkd->kd', tau_u, diff.pow(2))
        elif base_type == 'spherical':
            s2 = torch.einsum('nk,nk->k', tau_u, diff.pow(2).sum(dim=2))
        else:
            raise ValueError(f"Unsupported covariance type: {self.covariance_type}")

        aux = {'tau_u_sum': m_k, 'log_u_minus_u': log_u_minus_u}
        return SufficientStats(nk=nk, s1=s1, s2=s2, n_samples=n_samples, aux=aux)

    def _m_step_from_stats(self, stats: SufficientStats):
        n_samples = stats.n_samples
        nk = stats.nk + 1e-20  # sum_i tau_ik
        m_k = stats.aux['tau_u_sum'] + 1e-20  # sum_i tau_ik u_ik

        self.weights_ = stats.nk / n_samples
        self.weights_.clamp_(min=1e-20)

        self.means_ = stats.s1 / m_k.unsqueeze(1)

        denom = nk if self.scale_update == 'ecm' else m_k
        base_type = self.covariance_type.replace('tied_', '', 1)

        if self.covariance_type.startswith('tied_'):
            pooled_scatter = stats.s2.sum(dim=0)
            pooled_denom = denom.sum()
            cov = pooled_scatter / pooled_denom
            if base_type == 'full':
                cov = cov + self.reg_covar * torch.eye(self.n_features, device=self.device, dtype=self.dtype)
            else:
                cov = cov + self.reg_covar
            self.covariances_ = cov
        else:
            if base_type == 'full':
                cov = stats.s2 / denom.view(-1, 1, 1)
                eye = torch.eye(self.n_features, device=self.device, dtype=self.dtype).unsqueeze(0)
                cov = cov + self.reg_covar * eye
            elif base_type == 'diag':
                cov = stats.s2 / denom.unsqueeze(1) + self.reg_covar
            else:  # spherical
                cov = stats.s2 / denom + self.reg_covar
            self.covariances_ = cov

        self._update_dof(stats, nk)

    def _update_dof(self, stats: SufficientStats, nk: torch.Tensor):
        d = self.n_features
        if isinstance(self.dof, (int, float)):
            return  # fixed dof: never updated

        log_u_minus_u = stats.aux['log_u_minus_u']
        if self.dof == 'per_component':
            c = log_u_minus_u / nk
            self.dof_ = self._solve_dof(c, d)
        else:  # 'shared'
            c_shared = log_u_minus_u.sum() / nk.sum()
            nu_shared = self._solve_dof(c_shared.unsqueeze(0), d)
            self.dof_ = nu_shared.expand(self.n_components).clone()

    def _solve_dof(self, c: torch.Tensor, d: int) -> torch.Tensor:
        r"""
        Solve, elementwise over `c`'s shape,

        .. math::
            1 - \psi(\nu/2) + \ln(\nu/2) + c + \psi\!\left(\tfrac{\nu+d}{2}\right)
                - \ln\!\left(\tfrac{\nu+d}{2}\right) = 0

        for :math:`\nu`, by bisection within `self.dof_bounds` followed by
        a few Newton steps on :math:`\ln\nu` (using `torch.special.digamma`/
        `polygamma` for the derivative). Bisection alone would already be
        accurate enough in practice; the Newton polish tightens it per the
        plan's "bisection then Newton" spec.
        """
        def f(nu):
            return (
                1.0 - torch.special.digamma(nu / 2) + torch.log(nu / 2) + c
                + torch.special.digamma((nu + d) / 2) - torch.log((nu + d) / 2)
            )

        lo = torch.full_like(c, self.dof_bounds[0])
        hi = torch.full_like(c, self.dof_bounds[1])
        f_lo = f(lo)

        for _ in range(60):
            mid = 0.5 * (lo + hi)
            f_mid = f(mid)
            same_sign = torch.sign(f_mid) == torch.sign(f_lo)
            lo = torch.where(same_sign, mid, lo)
            hi = torch.where(same_sign, hi, mid)
            f_lo = torch.where(same_sign, f_mid, f_lo)

        nu = 0.5 * (lo + hi)
        log_nu = torch.log(nu)
        log_lo, log_hi = math.log(self.dof_bounds[0]), math.log(self.dof_bounds[1])
        for _ in range(4):
            nu_cur = torch.exp(log_nu)
            f_val = f(nu_cur)
            fprime = (
                -0.5 * torch.special.polygamma(1, nu_cur / 2) + 1.0 / nu_cur
                + 0.5 * torch.special.polygamma(1, (nu_cur + d) / 2) - 1.0 / (nu_cur + d)
            )
            d_f_d_lognu = nu_cur * fprime
            step = f_val / (d_f_d_lognu - torch.sign(d_f_d_lognu) * 1e-20 - 1e-20)
            log_nu = torch.clamp(log_nu - step, log_lo, log_hi)

        return torch.exp(log_nu).clamp(self.dof_bounds[0], self.dof_bounds[1])

    # ===================================================================
    # Model complexity / sampling
    # ===================================================================
    def _n_parameters(self) -> int:
        K, d = self.n_components, self.n_features
        if self.covariance_type == 'full':
            cov_params = K * d * (d + 1) / 2.0
        elif self.covariance_type == 'diag':
            cov_params = K * d
        elif self.covariance_type == 'spherical':
            cov_params = K
        else:  # tied_full
            cov_params = d * (d + 1) / 2.0

        if self.dof == 'per_component':
            dof_params = K
        elif self.dof == 'shared':
            dof_params = 1
        else:
            dof_params = 0

        mean_params = K * d
        weight_params = K - 1
        return int(cov_params + mean_params + weight_params + dof_params)

    def _sample_component(self, k: int, n: int) -> torch.Tensor:
        d = self.n_features
        mean = self.means_[k]
        if self.covariance_type == 'full':
            cov = self.covariances_[k]
        elif self.covariance_type == 'diag':
            cov = torch.diag_embed(self.covariances_[k])
        elif self.covariance_type == 'spherical':
            cov = torch.eye(d, device=self.device, dtype=self.dtype) * self.covariances_[k]
        else:  # tied_full
            cov = self.covariances_

        L = torch.linalg.cholesky(cov)
        z = torch.randn(n, d, device=self.device, dtype=self.dtype) @ L.t()

        nu = self.dof_[k]
        gamma = torch.distributions.Gamma(nu / 2, nu / 2)
        g = gamma.sample((n,)).to(device=self.device, dtype=self.dtype)
        return mean.unsqueeze(0) + z / torch.sqrt(g).unsqueeze(-1)

    def sample_weights(self, X: torch.Tensor) -> torch.Tensor:
        r"""
        The robustness weight :math:`u_{i\hat z_i}` at each sample's
        predicted component :math:`\hat z_i = \arg\max_k \tau_{ik}`. Small
        values flag likely outliers -- points far enough from their
        assigned component's center that the t-mixture downweighted them
        relative to what a Gaussian fit would have done.

        Parameters
        ----------
        X : torch.Tensor, shape (n_samples, n_features)

        Returns
        -------
        torch.Tensor, shape (n_samples,)
        """
        X = X.to(self.device)
        resp, _ = self._e_step(X)
        z_hat = torch.argmax(resp, dim=1)
        delta, _ = _mahalanobis_and_logdet(X, self.means_, self.covariances_, self.covariance_type)
        u = (self.dof_.unsqueeze(0) + self.n_features) / (self.dof_.unsqueeze(0) + delta)
        return u[torch.arange(X.shape[0], device=self.device), z_hat]

    # ===================================================================
    # Persistence -- independent of GaussianMixture's, see the design doc.
    # ===================================================================
    def to_dict(self) -> dict:
        return {
            'n_components': self.n_components,
            'n_features': self.n_features,
            'covariance_type': self.covariance_type,
            'dof': self.dof,
            'dof_init': self.dof_init,
            'dof_bounds': self.dof_bounds,
            'scale_update': self.scale_update,
            'max_iter': self.max_iter,
            'tol': self.tol,
            'reg_covar': self.reg_covar,
            'n_init': self.n_init,
            'init_means': self.init_means,
            'init_weights': self.init_weights,
            'init_covariances': self.init_covariances,
            'random_state': self.random_state,
            'warm_start': self.warm_start,
            'algorithm': self.algorithm,
            'batch_size': self.batch_size,
            'temperature': self.temperature,
            'convergence': self.convergence,
            'step_size': self.step_size,
            'sem_burn_in': self.sem_burn_in,
            'sem_estimate': self.sem_estimate,
            'verbose': self.verbose,
            'verbose_interval': self.verbose_interval,
            'weights_': self.weights_,
            'means_': self.means_,
            'covariances_': self.covariances_,
            'dof_': self.dof_,
            'fitted_': self.fitted_,
            'converged_': self.converged_,
            'n_iter_': self.n_iter_,
            'lower_bound_': self.lower_bound_,
            'n_components_': self.n_components_,
        }

    def save(self, filepath: str):
        r"""Save this model to `filepath` (see `load`)."""
        torch.save(self.to_dict(), filepath)

    @classmethod
    def load(cls, filepath: str, device: Optional[str] = None) -> "StudentTMixture":
        r"""Load a model previously saved with `save`."""
        if device is None:
            state = torch.load(filepath, weights_only=False)
        else:
            state = torch.load(filepath, map_location=device, weights_only=False)

        model = cls(
            n_components=state['n_components'], n_features=state['n_features'],
            covariance_type=state['covariance_type'], dof=state['dof'], dof_init=state['dof_init'],
            dof_bounds=state['dof_bounds'], scale_update=state['scale_update'],
            max_iter=state['max_iter'], tol=state['tol'], reg_covar=state['reg_covar'],
            n_init=state['n_init'], init_means=state['init_means'], init_weights=state['init_weights'],
            init_covariances=state['init_covariances'], random_state=state['random_state'],
            warm_start=state['warm_start'], algorithm=state['algorithm'], batch_size=state['batch_size'],
            temperature=state['temperature'], convergence=state['convergence'], step_size=state['step_size'],
            sem_burn_in=state['sem_burn_in'], sem_estimate=state['sem_estimate'],
            verbose=state['verbose'], verbose_interval=state['verbose_interval'], device=device,
        )
        model.weights_ = state['weights_']
        model.means_ = state['means_']
        model.covariances_ = state['covariances_']
        model.dof_ = state['dof_']
        model.fitted_ = state['fitted_']
        model.converged_ = state['converged_']
        model.n_iter_ = state['n_iter_']
        model.lower_bound_ = state['lower_bound_']
        model.n_components_ = state['n_components_']
        return model

    def save_state_dict(self) -> dict:
        r"""Alias for `to_dict` -- parity with `GaussianMixture`'s naming."""
        return self.to_dict()

    def load_state_dict(self, state_dict: dict):
        r"""In-place counterpart to `load`: restores this instance's fitted state from `state_dict`."""
        self.weights_ = state_dict['weights_']
        self.means_ = state_dict['means_']
        self.covariances_ = state_dict['covariances_']
        self.dof_ = state_dict['dof_']
        self.fitted_ = state_dict['fitted_']
        self.converged_ = state_dict['converged_']
        self.n_iter_ = state_dict['n_iter_']
        self.lower_bound_ = state_dict['lower_bound_']
        self.n_components_ = state_dict['n_components_']

Methods:

sample_weights(X)

The robustness weight :math:u_{i\hat z_i} at each sample's predicted component :math:\hat z_i = \arg\max_k \tau_{ik}. Small values flag likely outliers -- points far enough from their assigned component's center that the t-mixture downweighted them relative to what a Gaussian fit would have done.

Parameters:

Name Type Description Default
X (Tensor, shape(n_samples, n_features))
required

Returns:

Type Description
(Tensor, shape(n_samples))
Source code in tgmm/mixtures/student_t.py
def sample_weights(self, X: torch.Tensor) -> torch.Tensor:
    r"""
    The robustness weight :math:`u_{i\hat z_i}` at each sample's
    predicted component :math:`\hat z_i = \arg\max_k \tau_{ik}`. Small
    values flag likely outliers -- points far enough from their
    assigned component's center that the t-mixture downweighted them
    relative to what a Gaussian fit would have done.

    Parameters
    ----------
    X : torch.Tensor, shape (n_samples, n_features)

    Returns
    -------
    torch.Tensor, shape (n_samples,)
    """
    X = X.to(self.device)
    resp, _ = self._e_step(X)
    z_hat = torch.argmax(resp, dim=1)
    delta, _ = _mahalanobis_and_logdet(X, self.means_, self.covariances_, self.covariance_type)
    u = (self.dof_.unsqueeze(0) + self.n_features) / (self.dof_.unsqueeze(0) + delta)
    return u[torch.arange(X.shape[0], device=self.device), z_hat]

save(filepath)

Save this model to filepath (see load).

Source code in tgmm/mixtures/student_t.py
def save(self, filepath: str):
    r"""Save this model to `filepath` (see `load`)."""
    torch.save(self.to_dict(), filepath)

load(filepath, device=None) classmethod

Load a model previously saved with save.

Source code in tgmm/mixtures/student_t.py
@classmethod
def load(cls, filepath: str, device: Optional[str] = None) -> "StudentTMixture":
    r"""Load a model previously saved with `save`."""
    if device is None:
        state = torch.load(filepath, weights_only=False)
    else:
        state = torch.load(filepath, map_location=device, weights_only=False)

    model = cls(
        n_components=state['n_components'], n_features=state['n_features'],
        covariance_type=state['covariance_type'], dof=state['dof'], dof_init=state['dof_init'],
        dof_bounds=state['dof_bounds'], scale_update=state['scale_update'],
        max_iter=state['max_iter'], tol=state['tol'], reg_covar=state['reg_covar'],
        n_init=state['n_init'], init_means=state['init_means'], init_weights=state['init_weights'],
        init_covariances=state['init_covariances'], random_state=state['random_state'],
        warm_start=state['warm_start'], algorithm=state['algorithm'], batch_size=state['batch_size'],
        temperature=state['temperature'], convergence=state['convergence'], step_size=state['step_size'],
        sem_burn_in=state['sem_burn_in'], sem_estimate=state['sem_estimate'],
        verbose=state['verbose'], verbose_interval=state['verbose_interval'], device=device,
    )
    model.weights_ = state['weights_']
    model.means_ = state['means_']
    model.covariances_ = state['covariances_']
    model.dof_ = state['dof_']
    model.fitted_ = state['fitted_']
    model.converged_ = state['converged_']
    model.n_iter_ = state['n_iter_']
    model.lower_bound_ = state['lower_bound_']
    model.n_components_ = state['n_components_']
    return model

save_state_dict()

Alias for to_dict -- parity with GaussianMixture's naming.

Source code in tgmm/mixtures/student_t.py
def save_state_dict(self) -> dict:
    r"""Alias for `to_dict` -- parity with `GaussianMixture`'s naming."""
    return self.to_dict()

load_state_dict(state_dict)

In-place counterpart to load: restores this instance's fitted state from state_dict.

Source code in tgmm/mixtures/student_t.py
def load_state_dict(self, state_dict: dict):
    r"""In-place counterpart to `load`: restores this instance's fitted state from `state_dict`."""
    self.weights_ = state_dict['weights_']
    self.means_ = state_dict['means_']
    self.covariances_ = state_dict['covariances_']
    self.dof_ = state_dict['dof_']
    self.fitted_ = state_dict['fitted_']
    self.converged_ = state_dict['converged_']
    self.n_iter_ = state_dict['n_iter_']
    self.lower_bound_ = state_dict['lower_bound_']
    self.n_components_ = state_dict['n_components_']