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:

ApproximateTemporalGraphSample

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

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

Parameters:
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:

  • nobs (float | None)

  • suff_stat (TemporalGraphGrammarStatistics)

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 (TemporalGraphGrammarStatistics)

Return type:

TemporalGraphGrammarAccumulator

value()[source]

Return serialized sufficient statistics for estimation or merging.

Return type:

TemporalGraphGrammarStatistics

from_value(x)[source]

Restore accumulator state from serialized sufficient statistics.

Parameters:

x (TemporalGraphGrammarStatistics)

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(directed=False)[source]

Bases: DataSequenceEncoder

Pass-through encoder for dynamic graph sequence observations.

Parameters:

directed (bool)

seq_encode(x)[source]

Validate and own every adjacency in a temporal graph batch.

Parameters:

x (Sequence[Sequence[ndarray]])

Return type:

Sequence[Sequence[ndarray]]

row_count(x)[source]

Return the number of temporal sequences in an encoded payload.

Parameters:

x (Any)

Return type:

int

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 final node index, and one edge-record group per transition containing one record per added edge in canonical _edge_diff order. 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]

Validate and encode a batch of event-aligned attributed observations.

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 event-aligned attributed graph encoder.

Return type:

LabeledTemporalGraphGrammarDataEncoder

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, *, structure, node_encoder=None, edge_encoder=None, name=None)[source]

Bases: ParameterEstimator

Estimator for attributed temporal graph grammars.

Parameters:

  • structure_estimator (Any)

  • node_estimator (Any)

  • edge_estimator (Any)

  • structure (TemporalGraphGrammarDistribution)

  • node_encoder (Any)

  • edge_encoder (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:

  • nobs (float | None)

  • suff_stat (LabeledTemporalGraphGrammarStatistics)

Return type:

LabeledTemporalGraphGrammarDistribution

class LabeledTemporalGraphGrammarAccumulator(structure_acc, node_acc, edge_acc, node_encoder, edge_encoder, factory)[source]

Bases: SequenceEncodableStatisticAccumulator

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

Parameters:

  • structure_acc (Any)

  • node_acc (Any)

  • edge_acc (Any)

  • node_encoder (Any)

  • edge_encoder (Any)

  • factory (LabeledTemporalGraphGrammarAccumulatorFactory)

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 (LabeledTemporalGraphGrammarStatistics)

Return type:

LabeledTemporalGraphGrammarAccumulator

value()[source]

Return serialized attributed-graph sufficient statistics.

Return type:

LabeledTemporalGraphGrammarStatistics

from_value(x)[source]

Restore accumulator state from serialized sufficient statistics.

Parameters:

x (LabeledTemporalGraphGrammarStatistics)

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:

LabeledTemporalGraphGrammarDataEncoder

class LabeledTemporalGraphGrammarAccumulatorFactory(structure, structure_factory, node_factory, edge_factory, node_encoder, edge_encoder)[source]

Bases: StatisticAccumulatorFactory

Factory for attributed temporal graph grammar accumulators.

Parameters:

  • structure (TemporalGraphGrammarDistribution)

  • structure_factory (Any)

  • node_factory (Any)

  • edge_factory (Any)

  • node_encoder (Any)

  • edge_encoder (Any)

make()[source]

Create a fresh attributed temporal graph grammar accumulator.

Return type:

LabeledTemporalGraphGrammarAccumulator

class LabeledTemporalGraphGrammarDataEncoder(structure, has_node_dist, has_edge_dist)[source]

Bases: DataSequenceEncoder

Validate event alignment while retaining the attributed observation structure.

Parameters:

  • structure (TemporalGraphGrammarDistribution)

  • has_node_dist (bool)

  • has_edge_dist (bool)

seq_encode(x)[source]

Encode the iid observation sequence x for vectorized evaluation.

Parameters:

x (Sequence[tuple])

Return type:

tuple[tuple, …]

row_count(x)[source]

Return the number of observations represented by x.

The default covers simple array/list encodings and structurally aligned tuples/dicts. Encoders with flattened, ragged, or otherwise specialized layouts must override this method so metadata wrappers can verify counts instead of trusting a caller assertion.

Parameters:

x (Any)

Return type:

int

class LabeledTemporalGraphGrammarStatistics(schema_version, structure, node, edge, node_weight, edge_weight)[source]

Bases: NamedTuple

Parameters:
schema_version: int

Alias for field number 0

structure: Any

Alias for field number 1

node: Any

Alias for field number 2

edge: Any

Alias for field number 3

node_weight: float

Alias for field number 4

edge_weight: float

Alias for field number 5

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]

Validate and encode homophily observations.

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 validated homophily encoder.

Return type:

HomophilyTemporalGraphGrammarDataEncoder

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:

  • nobs (float | None)

  • suff_stat (HomophilyTemporalGraphGrammarStatistics)

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 (HomophilyTemporalGraphGrammarStatistics)

Return type:

HomophilyTemporalGraphGrammarAccumulator

value()[source]

Return serialized homophily sufficient statistics.

Return type:

HomophilyTemporalGraphGrammarStatistics

from_value(x)[source]

Restore accumulator state from serialized sufficient statistics.

Parameters:

x (HomophilyTemporalGraphGrammarStatistics)

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:

HomophilyTemporalGraphGrammarDataEncoder

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 HomophilyTemporalGraphGrammarDataEncoder(motif, num_types)[source]

Bases: DataSequenceEncoder

Validate homophily observations and retain their graph/type pairing.

Parameters:

  • motif (CommonNeighbourMotif)

  • num_types (int)

seq_encode(x)[source]

Encode the iid observation sequence x for vectorized evaluation.

Parameters:

x (Sequence[tuple])

Return type:

tuple[tuple, …]

row_count(x)[source]

Return the number of observations represented by x.

The default covers simple array/list encodings and structurally aligned tuples/dicts. Encoders with flattened, ragged, or otherwise specialized layouts must override this method so metadata wrappers can verify counts instead of trusting a caller assertion.

Parameters:

x (Any)

Return type:

int

class HomophilyTemporalGraphGrammarStatistics(schema_version, bins, num_types, edge_counts, type_counts, nodes, steps)[source]

Bases: NamedTuple

Parameters:
schema_version: int

Alias for field number 0

bins: tuple[int, ...]

Alias for field number 1

num_types: int

Alias for field number 2

edge_counts: ndarray

Alias for field number 3

type_counts: ndarray

Alias for field number 4

nodes: float

Alias for field number 5

steps: float

Alias for field number 6

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]

Validate and encode identity-tracked graph observations.

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 validated identity-tracked graph encoder.

Return type:

ChurningTemporalGraphGrammarDataEncoder

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, edit_grammar=None, name=None)[source]

Bases: ParameterEstimator

Estimator for identity-tracked churning temporal graph grammars.

Parameters:

  • edit_estimator (Any)

  • edit_grammar (TemporalGraphGrammarDistribution | None)

  • 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:

  • nobs (float | None)

  • suff_stat (ChurningTemporalGraphGrammarStatistics)

Return type:

ChurningTemporalGraphGrammarDistribution

class ChurningTemporalGraphGrammarAccumulator(edit_acc, directed, factory)[source]

Bases: SequenceEncodableStatisticAccumulator

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

Parameters:

  • edit_acc (Any)

  • directed (bool)

  • factory (ChurningTemporalGraphGrammarAccumulatorFactory)

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 (ChurningTemporalGraphGrammarStatistics)

Return type:

ChurningTemporalGraphGrammarAccumulator

value()[source]

Return serialized churning sufficient statistics.

Return type:

ChurningTemporalGraphGrammarStatistics

from_value(x)[source]

Restore accumulator state from serialized sufficient statistics.

Parameters:

x (ChurningTemporalGraphGrammarStatistics)

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:

ChurningTemporalGraphGrammarDataEncoder

class ChurningTemporalGraphGrammarAccumulatorFactory(edit_factory, directed)[source]

Bases: StatisticAccumulatorFactory

Factory for churning temporal graph grammar accumulators.

Parameters:

  • edit_factory (Any)

  • directed (bool)

make()[source]

Create a fresh churning accumulator.

Return type:

ChurningTemporalGraphGrammarAccumulator

class ChurningTemporalGraphGrammarDataEncoder(directed)[source]

Bases: DataSequenceEncoder

Validate and encode identity-tracked temporal graph sequences.

Parameters:

directed (bool)

seq_encode(x)[source]

Encode the iid observation sequence x for vectorized evaluation.

Parameters:

x (Sequence[Any])

Return type:

tuple[tuple, …]

row_count(x)[source]

Return the number of observations represented by x.

The default covers simple array/list encodings and structurally aligned tuples/dicts. Encoders with flattened, ragged, or otherwise specialized layouts must override this method so metadata wrappers can verify counts instead of trusting a caller assertion.

Parameters:

x (Any)

Return type:

int

class ChurningTemporalGraphGrammarStatistics(schema_version, directed, edit, removed, steps)[source]

Bases: NamedTuple

Parameters:
schema_version: int

Alias for field number 0

directed: bool

Alias for field number 1

edit: Any

Alias for field number 2

removed: float

Alias for field number 3

steps: float

Alias for field number 4

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]

Validate and encode latent-regime graph observations.

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 validated 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:

  • nobs (float | None)

  • suff_stat (LatentTemporalGraphGrammarStatistics)

Return type:

LatentTemporalGraphGrammarDistribution

class LatentTemporalGraphGrammarAccumulator(k, state_accs, bins, directed, factory)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulator for latent-regime graph grammar EM sufficient statistics.

Parameters:

  • k (int)

  • state_accs (Sequence[Any])

  • bins (tuple[int, ...])

  • directed (bool)

  • factory (LatentTemporalGraphGrammarAccumulatorFactory)

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 (LatentTemporalGraphGrammarStatistics)

Return type:

LatentTemporalGraphGrammarAccumulator

value()[source]

Return serialized latent-regime sufficient statistics.

Return type:

LatentTemporalGraphGrammarStatistics

from_value(x)[source]

Restore accumulator state from serialized sufficient statistics.

Parameters:

x (LatentTemporalGraphGrammarStatistics)

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, bins, directed)[source]

Bases: StatisticAccumulatorFactory

Factory for latent temporal graph grammar accumulators.

Parameters:

  • k (int)

  • state_factories (Sequence[Any])

  • bins (tuple[int, ...])

  • directed (bool)

make()[source]

Create a fresh latent-regime accumulator.

Return type:

LatentTemporalGraphGrammarAccumulator

class LatentTemporalGraphGrammarStatistics(schema_version, bins, directed, num_states, init_counts, trans_counts, state_values, accepted_weight, rejected_weight, transition_weight)[source]

Bases: NamedTuple

Parameters:
schema_version: int

Alias for field number 0

bins: tuple[int, ...]

Alias for field number 1

directed: bool

Alias for field number 2

num_states: int

Alias for field number 3

init_counts: ndarray

Alias for field number 4

trans_counts: ndarray

Alias for field number 5

state_values: tuple[Any, ...]

Alias for field number 6

accepted_weight: float

Alias for field number 7

rejected_weight: float

Alias for field number 8

transition_weight: float

Alias for field number 9

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]

Validate and encode attributed latent-regime observations.

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 event-aligned latent attributed encoder.

Return type:

LatentAttributedTemporalGraphGrammarDataEncoder

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, *, structures, node_encoders=None, edge_encoders=None, pseudo_count=None, name=None)[source]

Bases: ParameterEstimator

Parameters:

  • structure_estimators (Sequence[Any])

  • node_estimators (Sequence[Any] | None)

  • edge_estimators (Sequence[Any] | None)

  • structures (Sequence[TemporalGraphGrammarDistribution])

  • node_encoders (Sequence[Any] | None)

  • edge_encoders (Sequence[Any] | None)

  • pseudo_count (float | None)

  • name (str | None)

accumulator_factory()[source]

Return the accumulator factory used to collect this estimator’s sufficient statistics.

Return type:

LatentAttributedTemporalGraphGrammarAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate a distribution from accumulated sufficient statistics.

Parameters:

  • nobs (float | None)

  • suff_stat (LatentAttributedTemporalGraphGrammarStatistics)

Return type:

LatentAttributedTemporalGraphGrammarDistribution

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

Bases: SequenceEncodableStatisticAccumulator

Fail-closed EM accumulator with event-aligned attribute statistics.

Parameters:

  • struct_accs (Sequence[Any])

  • node_accs (Sequence[Any] | None)

  • edge_accs (Sequence[Any] | None)

  • factory (LatentAttributedTemporalGraphGrammarAccumulatorFactory)

update(x, weight, estimate)[source]

Accumulate one weighted observation under an optional current estimate.

Parameters:
Return type:

None

initialize(x, weight, rng)[source]

Initialize sufficient statistics from one weighted observation.

Parameters:
Return type:

None

seq_update(x, weights, estimate)[source]

Accumulate weighted sufficient statistics from sequence-encoded observations.

Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize sufficient statistics from sequence-encoded observations.

Parameters:
Return type:

None

combine(suff_stat)[source]

Merge serialized sufficient statistics into this accumulator.

Parameters:

suff_stat (LatentAttributedTemporalGraphGrammarStatistics)

Return type:

LatentAttributedTemporalGraphGrammarAccumulator

value()[source]

Return this accumulator’s serialized sufficient statistics.

Return type:

LatentAttributedTemporalGraphGrammarStatistics

from_value(x)[source]

Restore this accumulator from serialized sufficient statistics.

Parameters:

x (LatentAttributedTemporalGraphGrammarStatistics)

Return type:

LatentAttributedTemporalGraphGrammarAccumulator

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 whose key does not cover their whole state (several named keys, a wrapped child accumulator, a non-accumulator payload) override this – and ONLY those: as of 2026-07-14 every whole-state family uses this default, after a sweep found sixteen hand-rolled copies that had drifted into three broken shapes. If you are tempted to reimplement this per family, don’t; extend the base or override with super() calls. A keys of None (the default) is a no-op.

Why this exact shape is correct – and the drifted shapes are not: the POOL must live in the dict. Combining INTO the dict-held accumulator (below) mutates the pool in place, so after all sites merge, the dict entry holds everyone’s statistics and key_replace hands that pool to every site. The observed broken variants each violate one half: pulling the dict’s stats into self without writing back (the dict keeps the FIRST site only – later sites’ data silently discarded); merging only when the key is already present (nobody ever inserts: both passes are no-ops and tying does nothing); and replacing by pushing self INTO the dict (the LAST site overwrites the pool). The keyed-protocol sweep (keyed_protocol_sweep_test.py) enforces this behaviorally for every catalog family.

Parameters:

stats_dict (dict)

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)

Return type:

None

acc_to_encoder()[source]

Return a sequence encoder compatible with this accumulator.

Return type:

LatentAttributedTemporalGraphGrammarDataEncoder

class LatentAttributedTemporalGraphGrammarAccumulatorFactory(structures, struct_factories, node_factories, edge_factories, node_encoders, edge_encoders)[source]

Bases: StatisticAccumulatorFactory

Parameters:

  • structures (Sequence[TemporalGraphGrammarDistribution])

  • struct_factories (Sequence[Any])

  • node_factories (Sequence[Any] | None)

  • edge_factories (Sequence[Any] | None)

  • node_encoders (Sequence[Any] | None)

  • edge_encoders (Sequence[Any] | None)

make()[source]

Create a fresh accumulator instance.

Return type:

LatentAttributedTemporalGraphGrammarAccumulator

class LatentAttributedTemporalGraphGrammarDataEncoder(structure, has_node_dists, has_edge_dists)[source]

Bases: DataSequenceEncoder

Parameters:

  • structure (TemporalGraphGrammarDistribution)

  • has_node_dists (bool)

  • has_edge_dists (bool)

seq_encode(x)[source]

Encode the iid observation sequence x for vectorized evaluation.

Parameters:

x (Sequence[Any])

Return type:

tuple[tuple, …]

row_count(x)[source]

Return the number of observations represented by x.

The default covers simple array/list encodings and structurally aligned tuples/dicts. Encoders with flattened, ragged, or otherwise specialized layouts must override this method so metadata wrappers can verify counts instead of trusting a caller assertion.

Parameters:

x (Any)

Return type:

int

class LatentAttributedTemporalGraphGrammarStatistics(schema_version, bins, directed, num_states, init_counts, trans_counts, structure_values, node_values, edge_values, node_weights, edge_weights, accepted_weight, rejected_weight, transition_weight)[source]

Bases: NamedTuple

Parameters:
schema_version: int

Alias for field number 0

bins: tuple[int, ...]

Alias for field number 1

directed: bool

Alias for field number 2

num_states: int

Alias for field number 3

init_counts: ndarray

Alias for field number 4

trans_counts: ndarray

Alias for field number 5

structure_values: tuple[Any, ...]

Alias for field number 6

node_values: tuple[Any, ...] | None

Alias for field number 7

edge_values: tuple[Any, ...] | None

Alias for field number 8

node_weights: ndarray

Alias for field number 9

edge_weights: ndarray

Alias for field number 10

accepted_weight: float

Alias for field number 11

rejected_weight: float

Alias for field number 12

transition_weight: float

Alias for field number 13

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]

Validate and encode latent churning observations.

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 validated identity-tracked graph encoder.

Return type:

ChurningTemporalGraphGrammarDataEncoder

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, *, states=None, pseudo_count=None, name=None)[source]

Bases: ParameterEstimator

Parameters:

  • state_estimators (Sequence[Any])

  • states (Sequence[TemporalGraphGrammarDistribution] | None)

  • pseudo_count (float | None)

  • name (str | None)

accumulator_factory()[source]

Return the accumulator factory used to collect this estimator’s sufficient statistics.

Return type:

LatentChurningTemporalGraphGrammarAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate a distribution from accumulated sufficient statistics.

Parameters:

  • nobs (float | None)

  • suff_stat (LatentChurningTemporalGraphGrammarStatistics)

Return type:

LatentChurningTemporalGraphGrammarDistribution

class LatentChurningTemporalGraphGrammarAccumulator(state_accs, factory)[source]

Bases: SequenceEncodableStatisticAccumulator

Parameters:

  • state_accs (Sequence[Any])

  • factory (LatentChurningTemporalGraphGrammarAccumulatorFactory)

update(x, weight, estimate)[source]

Accumulate one weighted observation under an optional current estimate.

Parameters:
Return type:

None

initialize(x, weight, rng)[source]

Initialize sufficient statistics from one weighted observation.

Parameters:
Return type:

None

seq_update(x, weights, estimate)[source]

Accumulate weighted sufficient statistics from sequence-encoded observations.

Parameters:
Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize sufficient statistics from sequence-encoded observations.

Parameters:
Return type:

None

combine(suff_stat)[source]

Merge serialized sufficient statistics into this accumulator.

Parameters:

suff_stat (LatentChurningTemporalGraphGrammarStatistics)

Return type:

LatentChurningTemporalGraphGrammarAccumulator

value()[source]

Return this accumulator’s serialized sufficient statistics.

Return type:

LatentChurningTemporalGraphGrammarStatistics

from_value(x)[source]

Restore this accumulator from serialized sufficient statistics.

Parameters:

x (LatentChurningTemporalGraphGrammarStatistics)

Return type:

LatentChurningTemporalGraphGrammarAccumulator

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 whose key does not cover their whole state (several named keys, a wrapped child accumulator, a non-accumulator payload) override this – and ONLY those: as of 2026-07-14 every whole-state family uses this default, after a sweep found sixteen hand-rolled copies that had drifted into three broken shapes. If you are tempted to reimplement this per family, don’t; extend the base or override with super() calls. A keys of None (the default) is a no-op.

Why this exact shape is correct – and the drifted shapes are not: the POOL must live in the dict. Combining INTO the dict-held accumulator (below) mutates the pool in place, so after all sites merge, the dict entry holds everyone’s statistics and key_replace hands that pool to every site. The observed broken variants each violate one half: pulling the dict’s stats into self without writing back (the dict keeps the FIRST site only – later sites’ data silently discarded); merging only when the key is already present (nobody ever inserts: both passes are no-ops and tying does nothing); and replacing by pushing self INTO the dict (the LAST site overwrites the pool). The keyed-protocol sweep (keyed_protocol_sweep_test.py) enforces this behaviorally for every catalog family.

Parameters:

stats_dict (dict)

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)

Return type:

None

acc_to_encoder()[source]

Return a sequence encoder compatible with this accumulator.

Return type:

ChurningTemporalGraphGrammarDataEncoder

class LatentChurningTemporalGraphGrammarAccumulatorFactory(states, state_factories)[source]

Bases: StatisticAccumulatorFactory

Parameters:

  • states (Sequence[Any])

  • state_factories (Sequence[Any])

make()[source]

Create a fresh accumulator instance.

Return type:

LatentChurningTemporalGraphGrammarAccumulator

class LatentChurningTemporalGraphGrammarStatistics(schema_version, bins, directed, num_states, init_counts, trans_counts, removed, steps, state_values, accepted_weight, rejected_weight, transition_weight)[source]

Bases: NamedTuple

Parameters:
schema_version: int

Alias for field number 0

bins: tuple[int, ...]

Alias for field number 1

directed: bool

Alias for field number 2

num_states: int

Alias for field number 3

init_counts: ndarray

Alias for field number 4

trans_counts: ndarray

Alias for field number 5

removed: ndarray

Alias for field number 6

steps: ndarray

Alias for field number 7

state_values: tuple[Any, ...]

Alias for field number 8

accepted_weight: float

Alias for field number 9

rejected_weight: float

Alias for field number 10

transition_weight: float

Alias for field number 11

class RegimeMomentInitializationReceipt(num_sequences: 'int', num_transitions: 'int', num_regimes: 'int', feature_dimension: 'int', cluster_counts: 'tuple[int, ...]', structured_feature_maps: 'tuple[str, ...]', seed: 'int | None')[source]

Bases: object

Parameters:

  • num_sequences (int)

  • num_transitions (int)

  • num_regimes (int)

  • feature_dimension (int)

  • cluster_counts (tuple[int, ...])

  • structured_feature_maps (tuple[str, ...])

  • seed (int | None)

class RegimeMomentInitializationResult(model, receipt)[source]

Bases: NamedTuple

Parameters:

  • model (Any)

  • receipt (RegimeMomentInitializationReceipt)

model: Any

Alias for field number 0

receipt: RegimeMomentInitializationReceipt

Alias for field number 1

regime_moment_init(estimator, proto, data, k, seed=None, *, feature_maps=None, return_receipt=False)[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