mixle.stats.graphs.temporal_graph_grammar module

Temporal (dynamic) graph grammar – a distribution over graph SEQUENCES you can score, fit, and sample.

A dynamic graph is observed as a sequence of adjacency snapshots [A_0, A_1, ..., A_T] (binary, undirected; nodes may be appended and edges added over time – a growth process). The model is a Markov chain over graphs whose transition kernel is a stochastic motif-edit grammar:

given G_{t-1}, draw a number of new edges, and produce each by firing a grammar rule – “add an edge that creates motif m” – where the rule is chosen from the motif distribution w and an anchor is chosen uniformly among the non-edges of G_{t-1} that instantiate that motif.

So the grammar EDITS the graph over time, and its rule weights ARE the motif distribution it imposes. The default motif family bins a candidate edge by how many triangles it would close (its number of common neighbours: 0 = a bridge, 1, 2, 3+), i.e. a learnable triadic-closure profile; a custom mutually-exclusive motif partition can be supplied instead. Because the bins are mutually exclusive each added edge has a single motif, so scoring and fitting are exact (no per-edge latent – the VRG/HRG grammars marginalise over derivations; here the derivation is read off the snapshots).

Edges both FORM and DISSOLVE: an ADD grammar (motif_weights / edge_rate) draws new edges by the motif each would create, and a separate REMOVE grammar (remove_weights / edge_remove_rate) deletes existing edges by the motif each is part of (so e.g. growth can favour triadic closure while decay favours bridges – ties in dense neighbourhoods persist). Removal defaults off, so the constructor is backward compatible and a pure-growth grammar still scores a deletion as -inf.

Adjacencies may be dense ndarray or scipy.sparse – scoring and fitting never form the n*n bin matrix (they touch only the changed edges and the wedge structure of A @ A), so a 200k-node graph scores in a fraction of a second where a dense adjacency would need hundreds of GB. (Sampling stays dense/moderate-scale.) LabeledTemporalGraphGrammarDistribution attaches node attributes (location, name, age, …) and edge attributes (communication counts, channel, …) as ordinary mixle distributions scored as emissions on top of the topology – the whole thing fits jointly with the full distribution machinery (mixtures, every leaf family, the numba fusion).

Graphs may be directed (directed=True): the adjacency is asymmetric (i->j and j->i are distinct edges), the candidate space is the full off-diagonal, and A @ A counts transitive i->k->j paths – a directed triadic-closure profile. Weighted edges are just an edge attribute: put a weight distribution (Poisson volume, Gaussian strength, …) in the labeled model’s edge_dist, so a directed + weighted + attributed dynamic graph is a directed structure composed with node/edge emission models.

Nodes also LEAVE: ChurningTemporalGraphGrammarDistribution tracks stable node identities (each snapshot is (adjacency, node_ids)) so a transition can remove nodes – those whose id disappears, their edges vanishing with them – before running the edit grammar on the surviving subgraph. Node churn is a thin wrapper: identity alignment + a node-removal Poisson term on top of all the motif/edge machinery.

The dynamics carry a HIDDEN REGIME: LatentTemporalGraphGrammarDistribution is an HMM whose emission models ARE the edit grammars – a latent Markov state z_t selects which of K grammars governs transition t, so the graph switches phases over time (bursty growth/densification, then fragmentation/decay – dynamics a single grammar cannot produce). The sequence likelihood marginalises the regime path by the forward algorithm; EM (Baum-Welch) runs forward-backward then a per-regime weighted M-step reusing each grammar’s accumulator; decode (Viterbi) recovers the active regime at each step.

This is the temporal counterpart of the static vertex-/hyperedge-replacement grammars in this package. Scope: undirected or directed, binary topology (attribute models carry weights/labels); edges add+remove, nodes join+leave; dense or sparse scoring, dense or scalable (rejection) sampling; optional hidden regime. Sparse-path churn and directed scalable sampling are the remaining extensions.

class CommonNeighbourMotif(bins=(0, 1, 2, 3), directed=False)[source]

Bases: object

A motif rule keyed by how many common neighbours a candidate edge has (triangles it would close).

bins is an increasing list of thresholds; bin b covers common-neighbour counts in [bins[b], bins[b+1]) with the last bin open-ended. The default [0, 1, 2, 3] gives the interpretable {bridge, closes-1, closes-2, closes-3+} partition. A non-edge falls in exactly one bin, so the motifs partition every candidate edge.

Parameters:

  • bins (Sequence[int])

  • directed (bool)

property num_motifs: int

Number of mutually exclusive motif bins.

assign(adj, on_edges=False)[source]

Motif bin of every candidate pair (and -1 on non-candidates / diagonal).

The common-neighbour count of a pair (i, j) is (A @ A)[i, j] – for a non-edge, how many triangles adding it would CLOSE; for an existing edge, how many triangles it is PART of. Binning by self.bins gives its motif. With on_edges=False the candidates are the non-edges (addition); with on_edges=True they are the existing edges (removal). Non-candidates and the diagonal -> -1.

Parameters:
Return type:

ndarray

counts_and_binner(adj, on_edges)[source]

Return (candidate_counts[M], lookup(i, j) -> motif index) WITHOUT forming the n*n bin matrix.

Sparse-scalable: only the existing edges (O(m)) and the non-edges that close a triangle (O(wedges) = A @ A’s nonzeros) are ever enumerated; the bridge count (cn=0 non-edges) is the analytic remainder pairs - edges - wedge_non_edges. The lookup reads (A @ A)[i, j] for the handful of observed edges. (For graphs with mega-hubs the wedge set itself is large – the documented limit.)

Parameters:
Return type:

tuple

class TemporalGraphGrammarDistribution(motif_weights, edge_rate=1.0, node_rate=0.0, remove_weights=None, edge_remove_rate=0.0, motif=None, directed=False, name=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Distribution over dynamic graphs (sequences of adjacency snapshots) under a motif-edit grammar.

Parameters:

  • motif_weights (Sequence[float])

  • edge_rate (float)

  • node_rate (float)

  • remove_weights (Sequence[float] | None)

  • edge_remove_rate (float)

  • motif (CommonNeighbourMotif | None)

  • directed (bool)

  • name (str | None)

transition_components(prev, cur)[source]

The PARAMETER-INDEPENDENT decomposition of a transition: (new_nodes, add_bins, add_cand, rem_bins, rem_cand, valid). Depends only on the graph pair and the motif, NOT on the grammar’s weights/rates – so K regimes sharing a motif can compute the (expensive A@A) decomposition ONCE and score it K times via score_components(). valid is False for an impossible node removal.

Parameters:
Return type:

tuple

score_components(components)[source]

Score a precomputed transition_components() decomposition under THIS grammar’s parameters.

Parameters:

components (tuple)

Return type:

float

log_density(x)[source]

Log-density of one dynamic graph: the sum of transition log-densities over the snapshot chain.

x is a sequence of binary adjacency matrices – dense ndarray or scipy.sparse (large graphs). The initial graph is taken as given (its marginal is not modelled, matching how the static grammars treat their start symbol).

Parameters:

x (Sequence[ndarray])

Return type:

float

seq_encode(x)[source]

Return dynamic graph sequences unchanged for sequence scoring.

Parameters:

x (Sequence[Sequence[ndarray]])

Return type:

Sequence[Sequence[ndarray]]

seq_log_density(x)[source]

Score a batch of dynamic graph sequences.

Parameters:

x (Sequence[Sequence[ndarray]])

Return type:

ndarray

sampler(seed=None)[source]

Return a sampler for dynamic graph sequences.

Parameters:

seed (int | None)

Return type:

TemporalGraphGrammarSampler

estimator(pseudo_count=None)[source]

Return the closed-form estimator for this motif grammar.

Parameters:

pseudo_count (float | None)

Return type:

TemporalGraphGrammarEstimator

dist_to_encoder()[source]

Return the pass-through data encoder for graph sequences.

Return type:

TemporalGraphGrammarDataEncoder

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

Bases: DistributionSampler

Sampler for dynamic graph sequences generated by a motif-edit grammar.

Parameters:

  • dist (TemporalGraphGrammarDistribution)

  • seed (int | None)

sample_one(num_steps=10, seed_graph=None, n_init=5)[source]

Run a derivation: from a seed graph, apply num_steps of grammar-sampled edits.

Parameters:
Return type:

list[ndarray]

sample_one_scalable(num_steps=10, seed_edges=None, n_init=5, max_reject=64)[source]

Sample a dynamic graph for a LARGE sparse graph – never materialises the n*n adjacency.

The dense sample_one() is exact but O(n^2) in space (the full bin matrix). This path keeps the graph as an edge set and emits scipy.sparse snapshots, costing O(edges + wedges) per step:

  • triangle-closing motifs (cn>=1) are exactly the wedge non-edges – the nonzeros of A @ A that aren’t edges – so they are enumerated directly from the wedge structure, never the full pair grid;

  • bridges (cn=0) dominate a sparse graph and can’t be enumerated, so they are rejection-sampled: draw random pairs and accept those that are neither an edge nor a wedge (acceptance ~ 1 when the graph is sparse). max_reject attempts per bridge bound the loop; a shortfall is explicit capping (same realized-rate semantics as the dense sampler when a motif’s anchors run out).

Growth+removal; undirected or directed (directed: ordered i->j edges, full off-diagonal candidates, A@A = transitive i->k->j wedges). Returns a list of csr_array snapshots. The realized motif distribution matches the weights, so a model fit on these snapshots recovers the grammar.

Parameters:
Return type:

list

sample(size=None, *, num_steps=10, n_init=5)[source]

Draw one sequence or a list of sequences from the grammar.

Parameters:

  • size (int | None)

  • num_steps (int)

  • n_init (int)

Return type:

Any

class TemporalGraphGrammarEstimator(motif=None, pseudo_count=None, name=None)[source]

Bases: ParameterEstimator

Learn the motif distribution (rule weights) + edge/node rates from observed dynamic graphs.

Parameters:

  • motif (CommonNeighbourMotif | None)

  • pseudo_count (float | None)

  • name (str | None)

accumulator_factory()[source]

Return the accumulator factory used by the estimator.

Return type:

TemporalGraphGrammarAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate grammar weights and rates from sufficient statistics.

Parameters:
Return type:

TemporalGraphGrammarDistribution

class TemporalGraphGrammarAccumulator(motif)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulate per-motif edge counts + step/edge/node totals – the exact sufficient statistics.

Parameters:

motif (CommonNeighbourMotif)

update(x, weight, estimate)[source]

Accumulate sufficient statistics from one dynamic graph sequence.

Parameters:
Return type:

None

seq_update(x, weights, estimate)[source]

Accumulate weighted sufficient statistics from a batch of sequences.

Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize sufficient statistics from a weighted batch.

Parameters:
Return type:

None

initialize(x, weight, rng)[source]

Initialize sufficient statistics from one weighted sequence.

Parameters:
Return type:

None

combine(suff_stat)[source]

Merge serialized sufficient statistics into this accumulator.

Parameters:

suff_stat (tuple)

Return type:

TemporalGraphGrammarAccumulator

value()[source]

Return serialized sufficient statistics for estimation or merging.

Return type:

tuple

from_value(x)[source]

Restore accumulator state from serialized sufficient statistics.

Parameters:

x (tuple)

Return type:

TemporalGraphGrammarAccumulator

key_merge(stats_dict)[source]

Merge keyed sufficient statistics; unused for this accumulator.

Parameters:

stats_dict (dict)

Return type:

None

key_replace(stats_dict)[source]

Replace keyed sufficient statistics; unused for this accumulator.

Parameters:

stats_dict (dict)

Return type:

None

acc_to_encoder()[source]

Return the encoder associated with this accumulator.

Return type:

TemporalGraphGrammarDataEncoder

class TemporalGraphGrammarAccumulatorFactory(motif)[source]

Bases: StatisticAccumulatorFactory

Factory for temporal graph grammar accumulators.

Parameters:

motif (CommonNeighbourMotif)

make()[source]

Create a fresh temporal graph grammar accumulator.

Return type:

TemporalGraphGrammarAccumulator

class TemporalGraphGrammarDataEncoder[source]

Bases: DataSequenceEncoder

Pass-through encoder for dynamic graph sequence observations.

seq_encode(x)[source]

Return graph sequences unchanged.

Parameters:

x (Sequence[Sequence[ndarray]])

Return type:

Sequence[Sequence[ndarray]]

class LabeledTemporalGraphGrammarDistribution(structure, node_dist=None, edge_dist=None, name=None)[source]

Bases: SequenceEncodableProbabilityDistribution

A dynamic graph whose nodes and edges carry attributes.

Composes a structural TemporalGraphGrammarDistribution (the topology over time) with two ordinary mixle distributions: node_dist over per-node attribute records (location, name, age, … – typically a CompositeDistribution of leaves or a mixture) and edge_dist over per-edge attribute records (communication counts, channel, weight, …). An observation is (snapshots, node_features, edge_features): the adjacency chain, one attribute record per node, and one per added edge. The likelihood factorises – structure x node attributes x edge attributes – so the attribute models are fit (and scored) with the full mixle distribution machinery (mixtures, fusion, all leaf families).

Parameters:

  • structure (TemporalGraphGrammarDistribution)

  • node_dist (SequenceEncodableProbabilityDistribution | None)

  • edge_dist (SequenceEncodableProbabilityDistribution | None)

  • name (str | None)

log_density(x)[source]

Score one attributed dynamic graph observation.

Parameters:

x (tuple)

Return type:

float

seq_encode(x)[source]

Return attributed dynamic graph observations unchanged.

Parameters:

x (Sequence[tuple])

Return type:

Sequence[tuple]

seq_log_density(x)[source]

Score a batch of attributed dynamic graph observations.

Parameters:

x (Sequence[tuple])

Return type:

ndarray

sampler(seed=None)[source]

Return a sampler for attributed dynamic graph observations.

Parameters:

seed (int | None)

Return type:

LabeledTemporalGraphGrammarSampler

estimator(**kw)[source]

Return the estimator for structure, node attributes, and edge attributes.

Parameters:

kw (Any)

Return type:

LabeledTemporalGraphGrammarEstimator

dist_to_encoder()[source]

Return the pass-through graph encoder.

Return type:

TemporalGraphGrammarDataEncoder

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

Bases: DistributionSampler

Sampler for attributed dynamic graph observations.

Parameters:

  • dist (LabeledTemporalGraphGrammarDistribution)

  • seed (int | None)

sample_one(**kw)[source]

Draw one attributed dynamic graph observation.

Parameters:

kw (Any)

Return type:

tuple

sample(size=None, **kw)[source]

Draw one observation or a list of observations.

Parameters:
Return type:

Any

class LabeledTemporalGraphGrammarEstimator(structure_estimator, node_estimator=None, edge_estimator=None, name=None)[source]

Bases: ParameterEstimator

Estimator for attributed temporal graph grammars.

Parameters:

  • structure_estimator (Any)

  • node_estimator (Any)

  • edge_estimator (Any)

  • name (str | None)

accumulator_factory()[source]

Return the accumulator factory used by this estimator.

Return type:

LabeledTemporalGraphGrammarAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate structure and attribute distributions from sufficient statistics.

Parameters:
Return type:

LabeledTemporalGraphGrammarDistribution

class LabeledTemporalGraphGrammarAccumulator(structure_acc, node_acc, edge_acc)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulator for structure, node-attribute, and edge-attribute sufficient statistics.

Parameters:

  • structure_acc (Any)

  • node_acc (Any)

  • edge_acc (Any)

update(x, weight, estimate)[source]

Accumulate sufficient statistics from one attributed graph observation.

Parameters:
Return type:

None

seq_update(x, weights, estimate)[source]

Accumulate weighted sufficient statistics from a batch.

Parameters:
Return type:

None

initialize(x, weight, rng)[source]

Initialize sufficient statistics from one weighted observation.

Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize sufficient statistics from a weighted batch.

Parameters:
Return type:

None

combine(suff_stat)[source]

Merge serialized attributed-graph sufficient statistics.

Parameters:

suff_stat (tuple)

Return type:

LabeledTemporalGraphGrammarAccumulator

value()[source]

Return serialized attributed-graph sufficient statistics.

Return type:

tuple

from_value(x)[source]

Restore accumulator state from serialized sufficient statistics.

Parameters:

x (tuple)

Return type:

LabeledTemporalGraphGrammarAccumulator

key_merge(stats_dict)[source]

Merge keyed sufficient statistics; unused for this accumulator.

Parameters:

stats_dict (dict)

Return type:

None

key_replace(stats_dict)[source]

Replace keyed sufficient statistics; unused for this accumulator.

Parameters:

stats_dict (dict)

Return type:

None

acc_to_encoder()[source]

Return the encoder associated with this accumulator.

Return type:

TemporalGraphGrammarDataEncoder

class LabeledTemporalGraphGrammarAccumulatorFactory(structure_factory, node_factory, edge_factory)[source]

Bases: StatisticAccumulatorFactory

Factory for attributed temporal graph grammar accumulators.

Parameters:

  • structure_factory (Any)

  • node_factory (Any)

  • edge_factory (Any)

make()[source]

Create a fresh attributed temporal graph grammar accumulator.

Return type:

LabeledTemporalGraphGrammarAccumulator

class HomophilyTemporalGraphGrammarDistribution(rate, type_weights, node_rate=0.0, motif=None, name=None)[source]

Bases: SequenceEncodableProbabilityDistribution

A growth grammar whose edge formation depends on node ATTRIBUTES, not just structure (homophily).

Each node carries a categorical type (community / location-bucket / …). The per-step number of new edges of motif m between an (unordered) type pair (a, b) is Poisson(rate[m, a, b]), placed uniformly among the candidate non-edges of that motif and type pair. Making rate[m, a, a] larger than rate[m, a, b] is homophily (“similar nodes connect more”); the rate tensor is the learnable coupling between attributes and topology. New nodes draw their type from type_weights.

Observation: (snapshots, node_types) – the adjacency chain plus an int type per node. Exact and closed-form: the rate tensor is just edge counts per (motif, type-pair) over steps, and the type distribution is node-type counts. (Phase: growth-only, dense; add+remove and sparse compose with the machinery above and are the natural extensions.)

Parameters:

  • rate (np.ndarray)

  • type_weights (Sequence[float])

  • node_rate (float)

  • motif (CommonNeighbourMotif | None)

  • name (str | None)

log_density(x)[source]

Score one homophily dynamic graph observation.

Parameters:

x (tuple)

Return type:

float

seq_encode(x)[source]

Return homophily observations unchanged for sequence scoring.

Parameters:

x (Sequence[tuple])

Return type:

Sequence[tuple]

seq_log_density(x)[source]

Score a batch of homophily dynamic graph observations.

Parameters:

x (Sequence[tuple])

Return type:

ndarray

sampler(seed=None)[source]

Return a sampler for homophily dynamic graph observations.

Parameters:

seed (int | None)

Return type:

HomophilyTemporalGraphGrammarSampler

estimator(pseudo_count=None)[source]

Return the estimator for homophily rates and node-type weights.

Parameters:

pseudo_count (float | None)

Return type:

HomophilyTemporalGraphGrammarEstimator

dist_to_encoder()[source]

Return the pass-through graph encoder.

Return type:

TemporalGraphGrammarDataEncoder

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

Bases: DistributionSampler

Sampler for homophily temporal graph observations.

Parameters:

  • dist (HomophilyTemporalGraphGrammarDistribution)

  • seed (int | None)

sample_one(num_steps=8, seed_graph=None, n_init=8)[source]

Draw one homophily dynamic graph observation.

Parameters:
Return type:

tuple

sample(size=None, **kw)[source]

Draw one observation or a list of observations.

Parameters:
Return type:

Any

class HomophilyTemporalGraphGrammarEstimator(M, K, motif=None, pseudo_count=None, name=None)[source]

Bases: ParameterEstimator

Estimator for homophily temporal graph grammars.

Parameters:

  • M (int)

  • K (int)

  • motif (CommonNeighbourMotif | None)

  • pseudo_count (float | None)

  • name (str | None)

accumulator_factory()[source]

Return the accumulator factory used by this estimator.

Return type:

HomophilyTemporalGraphGrammarAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate homophily rates, type weights, and node rate from sufficient statistics.

Parameters:
Return type:

HomophilyTemporalGraphGrammarDistribution

class HomophilyTemporalGraphGrammarAccumulator(M, K, motif)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulator for homophily edge-rate and node-type sufficient statistics.

Parameters:

  • M (int)

  • K (int)

  • motif (CommonNeighbourMotif)

update(x, weight, estimate)[source]

Accumulate sufficient statistics from one homophily observation.

Parameters:
Return type:

None

seq_update(x, weights, estimate)[source]

Accumulate weighted sufficient statistics from a batch.

Parameters:
Return type:

None

initialize(x, weight, rng)[source]

Initialize sufficient statistics from one weighted observation.

Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize sufficient statistics from a weighted batch.

Parameters:
Return type:

None

combine(suff_stat)[source]

Merge serialized homophily sufficient statistics.

Parameters:

suff_stat (tuple)

Return type:

HomophilyTemporalGraphGrammarAccumulator

value()[source]

Return serialized homophily sufficient statistics.

Return type:

tuple

from_value(x)[source]

Restore accumulator state from serialized sufficient statistics.

Parameters:

x (tuple)

Return type:

HomophilyTemporalGraphGrammarAccumulator

key_merge(stats_dict)[source]

Merge keyed sufficient statistics; unused for this accumulator.

Parameters:

stats_dict (dict)

Return type:

None

key_replace(stats_dict)[source]

Replace keyed sufficient statistics; unused for this accumulator.

Parameters:

stats_dict (dict)

Return type:

None

acc_to_encoder()[source]

Return the encoder associated with this accumulator.

Return type:

TemporalGraphGrammarDataEncoder

class HomophilyTemporalGraphGrammarAccumulatorFactory(M, K, motif)[source]

Bases: StatisticAccumulatorFactory

Factory for homophily temporal graph grammar accumulators.

Parameters:

  • M (int)

  • K (int)

  • motif (CommonNeighbourMotif)

make()[source]

Create a fresh homophily accumulator.

Return type:

HomophilyTemporalGraphGrammarAccumulator

class ChurningTemporalGraphGrammarDistribution(edit_grammar, node_remove_rate=0.0, name=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Dynamic graph where nodes both JOIN and LEAVE, tracked by stable identity.

Each snapshot is (adjacency, node_ids) – node_ids[i] is the persistent identity of row i. A transition first removes nodes (those whose id disappears; count ~ Poisson(node_remove_rate), chosen uniformly, their edges vanishing with them), then runs the wrapped edit grammar on the surviving subgraph (which also appends new nodes + adds/removes edges). So churn is a thin wrapper: identity alignment + a node-removal Poisson term on top of all the existing motif/edge machinery. Scoring and fitting accept dense or scipy.sparse adjacencies (the id alignment slices either); the sampler is dense.

Parameters:

  • edit_grammar (TemporalGraphGrammarDistribution)

  • node_remove_rate (float)

  • name (str | None)

log_density(x)[source]

Score one identity-tracked churning graph sequence.

Parameters:

x (Sequence[tuple])

Return type:

float

seq_encode(x)[source]

Return churning graph observations unchanged for scoring.

Parameters:

x (Sequence[Any])

Return type:

Sequence[Any]

seq_log_density(x)[source]

Score a batch of churning graph observations.

Parameters:

x (Sequence[Any])

Return type:

ndarray

sampler(seed=None)[source]

Return a sampler for churning graph observations.

Parameters:

seed (int | None)

Return type:

ChurningTemporalGraphGrammarSampler

estimator(pseudo_count=None)[source]

Return the estimator for the wrapped edit grammar and node churn rate.

Parameters:

pseudo_count (float | None)

Return type:

ChurningTemporalGraphGrammarEstimator

dist_to_encoder()[source]

Return the pass-through graph encoder.

Return type:

TemporalGraphGrammarDataEncoder

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

Bases: DistributionSampler

Sampler for identity-tracked churning temporal graph observations.

Parameters:

  • dist (ChurningTemporalGraphGrammarDistribution)

  • seed (int | None)

sample_one(num_steps=10, seed_graph=None, n_init=8)[source]

Draw one churning graph sequence.

Parameters:
Return type:

list

sample(size=None, **kw)[source]

Draw one sequence or a list of sequences.

Parameters:
Return type:

Any

class ChurningTemporalGraphGrammarEstimator(edit_estimator, name=None)[source]

Bases: ParameterEstimator

Estimator for identity-tracked churning temporal graph grammars.

Parameters:

  • edit_estimator (Any)

  • name (str | None)

accumulator_factory()[source]

Return the accumulator factory used by this estimator.

Return type:

ChurningTemporalGraphGrammarAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate the wrapped edit grammar and node-removal rate.

Parameters:
Return type:

ChurningTemporalGraphGrammarDistribution

class ChurningTemporalGraphGrammarAccumulator(edit_acc)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulator for churning graph grammar and node-removal sufficient statistics.

Parameters:

edit_acc (Any)

update(x, weight, estimate)[source]

Accumulate sufficient statistics from one churning graph sequence.

Parameters:
Return type:

None

seq_update(x, weights, estimate)[source]

Accumulate weighted sufficient statistics from a batch.

Parameters:
Return type:

None

initialize(x, weight, rng)[source]

Initialize sufficient statistics from one weighted sequence.

Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize sufficient statistics from a weighted batch.

Parameters:
Return type:

None

combine(suff_stat)[source]

Merge serialized churning sufficient statistics.

Parameters:

suff_stat (tuple)

Return type:

ChurningTemporalGraphGrammarAccumulator

value()[source]

Return serialized churning sufficient statistics.

Return type:

tuple

from_value(x)[source]

Restore accumulator state from serialized sufficient statistics.

Parameters:

x (tuple)

Return type:

ChurningTemporalGraphGrammarAccumulator

key_merge(stats_dict)[source]

Merge keyed sufficient statistics; unused for this accumulator.

Parameters:

stats_dict (dict)

Return type:

None

key_replace(stats_dict)[source]

Replace keyed sufficient statistics; unused for this accumulator.

Parameters:

stats_dict (dict)

Return type:

None

acc_to_encoder()[source]

Return the encoder associated with this accumulator.

Return type:

TemporalGraphGrammarDataEncoder

class ChurningTemporalGraphGrammarAccumulatorFactory(edit_factory)[source]

Bases: StatisticAccumulatorFactory

Factory for churning temporal graph grammar accumulators.

Parameters:

edit_factory (Any)

make()[source]

Create a fresh churning accumulator.

Return type:

ChurningTemporalGraphGrammarAccumulator

class LatentTemporalGraphGrammarDistribution(states, initial_probs=None, transition_matrix=None, name=None)[source]

Bases: SequenceEncodableProbabilityDistribution

A dynamic graph whose edit grammar is governed by a hidden, time-evolving REGIME.

A latent state z_t (a Markov chain: initial_probs pi, transition_matrix A) selects which of K edit grammars governs transition t. So the graph can switch regimes over time – e.g. a bursty growth / densification phase, then a fragmentation / decay phase – dynamics a single grammar cannot express. The sequence likelihood marginalises the regime path by the forward algorithm; emissions are the per- transition edit log-densities of each regime’s grammar, so this is an HMM whose emission models are the graph-edit grammars and EM reuses each grammar’s weighted accumulator for the M-step.

Observation = a plain list of adjacency snapshots (same as the base grammar) – the regime is latent. decode returns the most likely regime active at each transition (Viterbi).

Parameters:

  • states (Sequence[TemporalGraphGrammarDistribution])

  • initial_probs (Sequence[float] | None)

  • transition_matrix (Sequence[Sequence[float]] | None)

  • name (str | None)

log_density(x)[source]

Score one dynamic graph sequence with regimes marginalized out.

Parameters:

x (Sequence[Any])

Return type:

float

decode(x)[source]

Viterbi: the most likely regime governing each transition.

Parameters:

x (Sequence[Any])

Return type:

list

seq_encode(x)[source]

Return latent-regime graph observations unchanged for scoring.

Parameters:

x (Sequence[Any])

Return type:

Sequence[Any]

seq_log_density(x)[source]

Score a batch of latent-regime graph observations.

Parameters:

x (Sequence[Any])

Return type:

ndarray

sampler(seed=None)[source]

Return a sampler for latent-regime graph sequences.

Parameters:

seed (int | None)

Return type:

LatentTemporalGraphGrammarSampler

estimator(pseudo_count=None)[source]

Return the Baum-Welch estimator for the latent-regime grammar.

Parameters:

pseudo_count (float | None)

Return type:

LatentTemporalGraphGrammarEstimator

dist_to_encoder()[source]

Return the pass-through graph encoder.

Return type:

TemporalGraphGrammarDataEncoder

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

Bases: DistributionSampler

Sampler for regime-switching temporal graph grammars.

Parameters:

  • dist (LatentTemporalGraphGrammarDistribution)

  • seed (int | None)

sample_one(num_steps=10, seed_graph=None, n_init=5)[source]

Draw one latent-regime graph sequence.

Parameters:
Return type:

list

sample(size=None, **kw)[source]

Draw one sequence or a list of sequences.

Parameters:
Return type:

Any

class LatentTemporalGraphGrammarEstimator(state_estimators, pseudo_count=None, name=None)[source]

Bases: ParameterEstimator

EM (Baum-Welch) for the regime-switching grammar: forward-backward E-step, per-regime weighted M-step.

Parameters:

  • state_estimators (Sequence[Any])

  • pseudo_count (float | None)

  • name (str | None)

accumulator_factory()[source]

Return the accumulator factory used by this estimator.

Return type:

LatentTemporalGraphGrammarAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate regime priors, transitions, and state grammars from EM statistics.

Parameters:
Return type:

LatentTemporalGraphGrammarDistribution

class LatentTemporalGraphGrammarAccumulator(k, state_accs)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulator for latent-regime graph grammar EM sufficient statistics.

Parameters:

  • k (int)

  • state_accs (Sequence[Any])

update(x, weight, estimate)[source]

Accumulate posterior-weighted sufficient statistics for one sequence.

Parameters:
Return type:

None

initialize(x, weight, rng)[source]

Initialize latent-regime sufficient statistics with random soft assignments.

Parameters:
Return type:

None

seq_update(x, weights, estimate)[source]

Accumulate posterior-weighted sufficient statistics from a batch.

Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize sufficient statistics from a weighted batch.

Parameters:
Return type:

None

combine(suff_stat)[source]

Merge serialized latent-regime sufficient statistics.

Parameters:

suff_stat (tuple)

Return type:

LatentTemporalGraphGrammarAccumulator

value()[source]

Return serialized latent-regime sufficient statistics.

Return type:

tuple

from_value(x)[source]

Restore accumulator state from serialized sufficient statistics.

Parameters:

x (tuple)

Return type:

LatentTemporalGraphGrammarAccumulator

key_merge(stats_dict)[source]

Merge keyed sufficient statistics; unused for this accumulator.

Parameters:

stats_dict (dict)

Return type:

None

key_replace(stats_dict)[source]

Replace keyed sufficient statistics; unused for this accumulator.

Parameters:

stats_dict (dict)

Return type:

None

acc_to_encoder()[source]

Return the encoder associated with this accumulator.

Return type:

TemporalGraphGrammarDataEncoder

class LatentTemporalGraphGrammarAccumulatorFactory(k, state_factories)[source]

Bases: StatisticAccumulatorFactory

Factory for latent temporal graph grammar accumulators.

Parameters:

  • k (int)

  • state_factories (Sequence[Any])

make()[source]

Create a fresh latent-regime accumulator.

Return type:

LatentTemporalGraphGrammarAccumulator

class LatentAttributedTemporalGraphGrammarDistribution(structures, node_dists=None, edge_dists=None, initial_probs=None, transition_matrix=None, name=None)[source]

Bases: SequenceEncodableProbabilityDistribution

A regime-switching dynamic graph where the hidden regime drives the STRUCTURE and the ATTRIBUTES.

Each of K regimes carries a full edit grammar plus (optionally) a node-attribute distribution and an edge-attribute distribution, all switched by one latent Markov state z_t. So a single regime change can densify the topology AND spike communication volume / shift node properties together – e.g. an “active” phase with bursty triadic closure and high message counts, vs a “quiet” phase. The per-transition emission under regime k is structure_k(transition) + node_attrs_k(nodes added this step) + edge_attrs_k(edges added this step); the sequence likelihood marginalises the regime path by the forward algorithm and EM does forward-backward + a per-regime weighted M-step over each piece.

Observation = (snapshots, node_features, edge_features) where node_features[t] / edge_features[t] are the attribute records of the nodes / edges that appear at transition t (lists, length = #transitions).

Parameters:

  • structures (Sequence[TemporalGraphGrammarDistribution])

  • node_dists (Sequence[Any] | None)

  • edge_dists (Sequence[Any] | None)

  • initial_probs (Sequence[float] | None)

  • transition_matrix (Sequence[Sequence[float]] | None)

  • name (str | None)

log_density(x)[source]

Score one attributed graph sequence with regimes marginalized out.

Parameters:

x (tuple)

Return type:

float

decode(x)[source]

Viterbi: the most likely regime governing each transition (jointly explaining structure+attrs).

Parameters:

x (tuple)

Return type:

list

seq_encode(x)[source]

Return attributed latent-regime observations unchanged for scoring.

Parameters:

x (Sequence[Any])

Return type:

Sequence[Any]

seq_log_density(x)[source]

Score a batch of attributed latent-regime observations.

Parameters:

x (Sequence[Any])

Return type:

ndarray

sampler(seed=None)[source]

Return a sampler for attributed latent-regime graph sequences.

Parameters:

seed (int | None)

Return type:

LatentAttributedTemporalGraphGrammarSampler

estimator(pseudo_count=None)[source]

Return the EM estimator for structure and attribute regime models.

Parameters:

pseudo_count (float | None)

Return type:

LatentAttributedTemporalGraphGrammarEstimator

dist_to_encoder()[source]

Return the pass-through graph encoder.

Return type:

TemporalGraphGrammarDataEncoder

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

Bases: DistributionSampler

Sampler for regime-switching attributed temporal graph grammars.

Parameters:

  • dist (LatentAttributedTemporalGraphGrammarDistribution)

  • seed (int | None)

sample_one(num_steps=10, seed_graph=None, n_init=5)[source]

Draw one attributed latent-regime graph observation.

Parameters:
Return type:

tuple

sample(size=None, **kw)[source]

Draw one observation or a list of observations.

Parameters:
Return type:

Any

class LatentAttributedTemporalGraphGrammarEstimator(structure_estimators, node_estimators=None, edge_estimators=None, pseudo_count=None, name=None)[source]

Bases: ParameterEstimator

EM for the regime-switching attributed grammar: forward-backward E-step, per-regime weighted M-step over structure + node attrs + edge attrs together.

Parameters:

  • structure_estimators (Sequence[Any])

  • node_estimators (Any)

  • edge_estimators (Any)

  • pseudo_count (float | None)

  • name (str | None)

accumulator_factory()[source]

Return the accumulator factory used by this estimator.

Return type:

LatentAttributedTemporalGraphGrammarAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate regime priors, transitions, structures, and attribute models.

Parameters:
Return type:

LatentAttributedTemporalGraphGrammarDistribution

class LatentAttributedTemporalGraphGrammarAccumulator(k, struct_accs, node_accs, edge_accs)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulator for attributed latent-regime graph grammar EM statistics.

Parameters:

  • k (int)

  • struct_accs (Sequence[Any])

  • node_accs (Any)

  • edge_accs (Any)

update(x, weight, estimate)[source]

Accumulate posterior-weighted statistics for one attributed sequence.

Parameters:
Return type:

None

initialize(x, weight, rng)[source]

Initialize attributed latent-regime statistics with random soft assignments.

Parameters:
Return type:

None

seq_update(x, weights, estimate)[source]

Accumulate posterior-weighted statistics from a batch.

Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize sufficient statistics from a weighted batch.

Parameters:
Return type:

None

combine(suff_stat)[source]

Merge serialized attributed latent-regime sufficient statistics.

Parameters:

suff_stat (tuple)

Return type:

LatentAttributedTemporalGraphGrammarAccumulator

value()[source]

Return serialized attributed latent-regime sufficient statistics.

Return type:

tuple

from_value(x)[source]

Restore accumulator state from serialized sufficient statistics.

Parameters:

x (tuple)

Return type:

LatentAttributedTemporalGraphGrammarAccumulator

key_merge(stats_dict)[source]

Merge keyed sufficient statistics; unused for this accumulator.

Parameters:

stats_dict (dict)

Return type:

None

key_replace(stats_dict)[source]

Replace keyed sufficient statistics; unused for this accumulator.

Parameters:

stats_dict (dict)

Return type:

None

acc_to_encoder()[source]

Return the encoder associated with this accumulator.

Return type:

TemporalGraphGrammarDataEncoder

class LatentAttributedTemporalGraphGrammarAccumulatorFactory(k, struct_factories, node_factories, edge_factories)[source]

Bases: StatisticAccumulatorFactory

Factory for attributed latent-regime graph grammar accumulators.

Parameters:

  • k (int)

  • struct_factories (Sequence[Any])

  • node_factories (Any)

  • edge_factories (Any)

make()[source]

Create a fresh attributed latent-regime accumulator.

Return type:

LatentAttributedTemporalGraphGrammarAccumulator

class LatentChurningTemporalGraphGrammarDistribution(states, node_remove_rates=None, initial_probs=None, transition_matrix=None, name=None)[source]

Bases: SequenceEncodableProbabilityDistribution

A regime-switching dynamic graph where the hidden regime also governs NODE TURNOVER.

Combines the latent regime HMM with identity-tracked node churn: each of K regimes carries a full edit grammar AND its own node-removal rate, so the graph can switch between e.g. a stable phase (slow turnover, triadic growth) and a churn phase (fast member departure, fragmentation). Each snapshot is (adjacency, node_ids); per transition the active regime first removes nodes (those whose id disappears) then edits the surviving subgraph, and the per-transition emission under regime k is node_removal_k(#removed) + grammar_k(edit on the aligned surviving subgraph). The sequence likelihood marginalises the regime path by the forward algorithm; EM does forward-backward then a per-regime weighted M-step over both the grammar and the turnover rate. decode recovers the active regime.

Observation = (snapshots, node_ids) where node_ids is a list of per-snapshot id arrays. Dense.

Parameters:

  • states (Sequence[TemporalGraphGrammarDistribution])

  • node_remove_rates (Sequence[float] | None)

  • initial_probs (Sequence[float] | None)

  • transition_matrix (Sequence[Sequence[float]] | None)

  • name (str | None)

log_density(x)[source]

Score one identity-tracked sequence with regimes marginalized out.

Parameters:

x (tuple)

Return type:

float

decode(x)[source]

Return the most likely churn/edit regime for each transition.

Parameters:

x (tuple)

Return type:

list

seq_encode(x)[source]

Return latent churning observations unchanged for scoring.

Parameters:

x (Sequence[Any])

Return type:

Sequence[Any]

seq_log_density(x)[source]

Score a batch of latent churning graph observations.

Parameters:

x (Sequence[Any])

Return type:

ndarray

sampler(seed=None)[source]

Return a sampler for latent churning graph sequences.

Parameters:

seed (int | None)

Return type:

LatentChurningTemporalGraphGrammarSampler

estimator(pseudo_count=None)[source]

Return the EM estimator for regime-specific edit grammars and churn rates.

Parameters:

pseudo_count (float | None)

Return type:

LatentChurningTemporalGraphGrammarEstimator

dist_to_encoder()[source]

Return the pass-through graph encoder.

Return type:

TemporalGraphGrammarDataEncoder

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

Bases: DistributionSampler

Sampler for regime-switching churning temporal graph grammars.

Parameters:

  • dist (LatentChurningTemporalGraphGrammarDistribution)

  • seed (int | None)

sample_one(num_steps=10, seed_graph=None, n_init=8)[source]

Draw one latent churning graph sequence.

Parameters:
Return type:

tuple

sample(size=None, **kw)[source]

Draw one sequence or a list of sequences.

Parameters:
Return type:

Any

class LatentChurningTemporalGraphGrammarEstimator(state_estimators, pseudo_count=None, name=None)[source]

Bases: ParameterEstimator

Estimator for regime-switching churning temporal graph grammars.

Parameters:

  • state_estimators (Sequence[Any])

  • pseudo_count (float | None)

  • name (str | None)

accumulator_factory()[source]

Return the accumulator factory used by this estimator.

Return type:

LatentChurningTemporalGraphGrammarAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate regime priors, transitions, grammars, and churn rates.

Parameters:
Return type:

LatentChurningTemporalGraphGrammarDistribution

class LatentChurningTemporalGraphGrammarAccumulator(k, state_accs)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulator for latent churning graph grammar EM sufficient statistics.

Parameters:

  • k (int)

  • state_accs (Sequence[Any])

update(x, weight, estimate)[source]

Accumulate posterior-weighted statistics for one churning sequence.

Parameters:
Return type:

None

initialize(x, weight, rng)[source]

Initialize latent churning statistics with random soft assignments.

Parameters:
Return type:

None

seq_update(x, weights, estimate)[source]

Accumulate posterior-weighted statistics from a batch.

Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize sufficient statistics from a weighted batch.

Parameters:
Return type:

None

combine(suff_stat)[source]

Merge serialized latent churning sufficient statistics.

Parameters:

suff_stat (tuple)

Return type:

LatentChurningTemporalGraphGrammarAccumulator

value()[source]

Return serialized latent churning sufficient statistics.

Return type:

tuple

from_value(x)[source]

Restore accumulator state from serialized sufficient statistics.

Parameters:

x (tuple)

Return type:

LatentChurningTemporalGraphGrammarAccumulator

key_merge(stats_dict)[source]

Merge keyed sufficient statistics; unused for this accumulator.

Parameters:

stats_dict (dict)

Return type:

None

key_replace(stats_dict)[source]

Replace keyed sufficient statistics; unused for this accumulator.

Parameters:

stats_dict (dict)

Return type:

None

acc_to_encoder()[source]

Return the encoder associated with this accumulator.

Return type:

TemporalGraphGrammarDataEncoder

class LatentChurningTemporalGraphGrammarAccumulatorFactory(k, state_factories)[source]

Bases: StatisticAccumulatorFactory

Factory for latent churning temporal graph grammar accumulators.

Parameters:

  • k (int)

  • state_factories (Sequence[Any])

make()[source]

Create a fresh latent churning accumulator.

Return type:

LatentChurningTemporalGraphGrammarAccumulator

regime_moment_init(estimator, proto, data, k, seed=None)[source]

Seed a regime-switching grammar EM by clustering observed per-transition edit signatures.

Returns an initial distribution whose regimes are the k-means clusters of the (identifiable) transition signatures. Because the derivation is observed, these signatures are sufficient statistics that separate the regimes, so this avoids the local optima of random-restart EM. proto is any distribution of the target class (used only for its motif/attribute structure); estimator produces the fitted result.

Parameters:
Return type:

Any