mixle.stats.trees.spanning_tree module

Create, estimate, sample, and enumerate a weighted spanning-tree distribution over a labeled graph.

Defines the SpanningTreeDistribution, SpanningTreeEnumerator, SpanningTreeSampler, SpanningTreeAccumulatorFactory, SpanningTreeAccumulator, SpanningTreeEstimator, and the SpanningTreeDataEncoder classes for use with mixle.

Data type: a spanning tree of n labeled nodes given as a sequence of n-1 undirected edges, each an (i, j) pair (e.g. [(0, 1), (1, 2), (1, 3)]). Unlike ChowLiuTree (a tree-structured distribution over vectors), this is a distribution over the tree STRUCTURES themselves.

Each undirected edge has a positive weight w[i, j]. A spanning tree T has probability

p(T) = prod_{(i,j) in T} w[i, j] / Z, Z = sum over all spanning trees of prod w[e],

and by the Matrix-Tree theorem Z equals any first cofactor of the weighted graph Laplacian L = diag(W 1) - W, i.e. det(L[1:, 1:]). Sampling uses Wilson’s loop-erased-random-walk algorithm, which draws exactly from this weighted uniform-spanning-tree law. Estimation matches empirical or smoothed edge frequencies to the model edge marginals (an exponential family over trees, fit by projected gradient ascent on the log-weights); the per-edge marginal w[i,j] * R_eff(i,j) is read from the Laplacian pseudoinverse. Exact finite enumeration scans all positive-edge subsets of size n-1, keeps the spanning trees, and sorts them by fitted probability.

class SpanningTreeDistribution(weights, name=None, keys=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Weighted spanning-tree distribution over n labeled nodes with symmetric positive edge weights.

Data type: a sequence of n-1 undirected edges (i, j) forming a spanning tree of 0,…,n-1.

Parameters:
classmethod compute_capabilities()[source]
density(x)[source]

Return the probability of a spanning tree x (a sequence of edges).

Parameters:

x (Sequence[Sequence[int]])

Return type:

float

log_density(x)[source]

Return the log-probability of a spanning tree x (a sequence of n-1 edges).

Parameters:

x (Sequence[Sequence[int]])

Return type:

float

seq_log_density(x)[source]

Return vectorized log-probabilities for a sequence of canonical edge arrays.

Parameters:

x (Sequence[ndarray])

Return type:

ndarray

sampler(seed=None)[source]

Return a sampler for drawing spanning trees from this distribution.

Parameters:

seed (int | None)

Return type:

SpanningTreeSampler

enumerator(max_edge_subsets=_DEFAULT_MAX_ENUMERATION_SUBSETS)[source]

Return an exact finite enumerator over all supported spanning trees in probability order.

Parameters:

max_edge_subsets (int | None)

Return type:

SpanningTreeEnumerator

estimator(pseudo_count=None)[source]

Return an estimator that keeps the node count fixed at this distribution’s n.

Parameters:

pseudo_count (float | None)

Return type:

SpanningTreeEstimator

dist_to_encoder()[source]

Return the data encoder used by this distribution for vectorized methods.

Return type:

SpanningTreeDataEncoder

class SpanningTreeEnumerator(dist, max_edge_subsets=_DEFAULT_MAX_ENUMERATION_SUBSETS)[source]

Bases: DistributionEnumerator

Enumerate supported spanning trees in descending probability order, lazily.

A tree’s probability is the product of its edge weights, so descending probability is increasing total edge cost under cost = -log(weights) (zero-weight edges become +inf, i.e. absent). Gabow’s k-best spanning-tree algorithm streams the trees in that order from one constrained-MST oracle per node, without scanning the exponential set of edge subsets.

Parameters:

  • dist (SpanningTreeDistribution)

  • max_edge_subsets (int | None)

class SpanningTreeSampler(dist, seed=None)[source]

Bases: DistributionSampler

Draw iid spanning trees via Wilson’s loop-erased-random-walk algorithm.

Parameters:

  • dist (SpanningTreeDistribution)

  • seed (int | None)

sample(size=None)[source]

Draw spanning trees (each a sorted edge list); a single tree when size is None.

Parameters:

size (int | None)

Return type:

list[tuple[int, int]] | list[list[tuple[int, int]]]

class SpanningTreeAccumulator(dim, keys=None)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulate the weighted edge-appearance counts (the sufficient statistic for the tree weights).

Parameters:
update(x, weight, estimate)[source]
Parameters:
Return type:

None

initialize(x, weight, rng)[source]
Parameters:
Return type:

None

seq_update(x, weights, estimate)[source]
Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]
Parameters:
Return type:

None

combine(suff_stat)[source]
Parameters:

suff_stat (tuple[float, ndarray])

Return type:

SpanningTreeAccumulator

value()[source]
Return type:

tuple[float, ndarray]

from_value(x)[source]
Parameters:

x (tuple[float, ndarray])

Return type:

SpanningTreeAccumulator

key_merge(stats_dict)[source]

Pool this accumulator’s statistics into stats_dict under its merge key.

The structural default implements the common single-key pattern: store the accumulator under self.keys the first time the key is seen, else combine into the one already there. Accumulators with several named keys (e.g. an HMM’s init/trans/state keys) or a non-accumulator stats payload override this. A keys of None (the default) is a no-op.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

key_replace(stats_dict)[source]

Replace this accumulator’s statistics from the pooled stats_dict entry (see key_merge).

Parameters:

stats_dict (dict[str, Any])

Return type:

None

acc_to_encoder()[source]
Return type:

SpanningTreeDataEncoder

class SpanningTreeAccumulatorFactory(dim, keys=None)[source]

Bases: StatisticAccumulatorFactory

Factory for SpanningTreeAccumulator.

Parameters:
make()[source]
Return type:

SpanningTreeAccumulator

class SpanningTreeEstimator(dim, pseudo_count=None, max_steps=500, learning_rate=1.0, tol=1.0e-7, name=None, keys=None)[source]

Bases: ParameterEstimator

Estimate edge weights by matching empirical or smoothed tree edge marginals.

Parameters:
accumulator_factory()[source]
Return type:

SpanningTreeAccumulatorFactory

estimate(nobs, suff_stat)[source]
Parameters:
Return type:

SpanningTreeDistribution

class SpanningTreeDataEncoder(dim=None)[source]

Bases: DataSequenceEncoder

Encode a sequence of spanning trees (edge lists) into per-observation canonical edge arrays.

Parameters:

dim (int | None)

seq_encode(x)[source]

Encode the iid observation sequence x for vectorized evaluation.

Parameters:

x (Sequence[Sequence[Sequence[int]]])

Return type:

list[ndarray]