From 2d029480cd2e509a9142229de88954c3ac011680 Mon Sep 17 00:00:00 2001 From: popchanovska Date: Fri, 5 Jun 2026 14:56:51 +0200 Subject: [PATCH 1/9] Solve warning in example_compat_conversion --- examples/io_and_data/example_compat_conversion.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/examples/io_and_data/example_compat_conversion.py b/examples/io_and_data/example_compat_conversion.py index d28cc2f2c..a51d77cdd 100644 --- a/examples/io_and_data/example_compat_conversion.py +++ b/examples/io_and_data/example_compat_conversion.py @@ -10,7 +10,7 @@ from py3plex.compat import convert, to_ir from py3plex.compat.schema import infer_schema from py3plex.io.schema import MultiLayerGraph, Node, Layer, Edge - +from py3plex.compat.exceptions import CompatibilityError def create_sample_graph(): """Create a sample multilayer network.""" @@ -82,7 +82,6 @@ def main(): print("\n3. Error Handling (Strict Mode)") print("-" * 60) try: - from py3plex.compat.exceptions import CompatibilityError matrix = convert(graph, "scipy_sparse", strict=True) print(f" Conversion succeeded") except CompatibilityError as e: From d66b49f300beeeb5745090812f5a21c2ec78585c Mon Sep 17 00:00:00 2001 From: popchanovska Date: Sun, 7 Jun 2026 17:56:01 +0200 Subject: [PATCH 2/9] Add mdl_metric simple implementation --- .../autocommunity_executor.py | 5 +- .../multilayer_quality_metrics.py | 153 +++++++++++++++++- 2 files changed, 153 insertions(+), 5 deletions(-) diff --git a/py3plex/algorithms/community_detection/autocommunity_executor.py b/py3plex/algorithms/community_detection/autocommunity_executor.py index c9c8502a6..4f5458c3d 100644 --- a/py3plex/algorithms/community_detection/autocommunity_executor.py +++ b/py3plex/algorithms/community_detection/autocommunity_executor.py @@ -455,6 +455,7 @@ def _evaluate_algorithms( from py3plex.algorithms.community_detection.multilayer_quality_metrics import ( replica_consistency, layer_entropy, + mdl_score, ) import inspect @@ -519,9 +520,7 @@ def _evaluate_algorithms( value = 0.0 elif metric_name == "mdl": - # Description length (if available) - # Placeholder for now - value = 0.0 + value = mdl_score(partition, network) elif metric_name == "replica_consistency": # Multilayer coherence metric diff --git a/py3plex/algorithms/community_detection/multilayer_quality_metrics.py b/py3plex/algorithms/community_detection/multilayer_quality_metrics.py index eb37f24fc..8f673e700 100644 --- a/py3plex/algorithms/community_detection/multilayer_quality_metrics.py +++ b/py3plex/algorithms/community_detection/multilayer_quality_metrics.py @@ -9,6 +9,8 @@ layers are assigned to the same community (multilayer coherence) - layer_entropy: Measures the balance of community sizes within each layer, averaged across layers (degeneracy guardrail) +- mdl_score: Minimum Description Length based on SBM two-part description + length (lower is better) """ from __future__ import annotations @@ -318,8 +320,155 @@ def layer_entropy( return 0.0 mean_entropy = float(np.mean(layer_entropies)) - + # Apply clipping clipped_entropy = np.clip(mean_entropy, clip[0], clip[1]) - + return float(clipped_entropy) + + +def _mdl_single_layer( + layer_partition: Dict[Any, int], + layer_edges: List[tuple], + directed: bool = False, +) -> float: + """Compute SBM two-part description length for a single flat graph layer. + + Args: + layer_partition: Dict mapping node -> community_id for this layer only. + layer_edges: List of (u, v) edge pairs within this layer. + directed: Whether the underlying graph is directed. Controls the + maximum-edge normalization for block pairs. Defaults to False + (undirected), preserving backward-compatible behavior for callers + that do not pass the flag. + + Returns: + Description length in bits for this layer. + """ + communities: Dict[int, List[Any]] = defaultdict(list) + for node, comm in layer_partition.items(): + communities[comm].append(node) + + k = len(communities) + n = len(layer_partition) + + if k == 0 or n == 0: + return 0.0 + + # Count edges between/within each block pair + # Self-loops are skipped: they do not fit the n_r*(n_r-1) pair model + edge_counts: Dict[tuple, int] = defaultdict(int) + for u, v in layer_edges: + if u == v: + continue + r = layer_partition.get(u) + s = layer_partition.get(v) + if r is None or s is None: + continue + edge_counts[(min(r, s), max(r, s))] += 1 + + comm_ids = list(communities.keys()) + + # Part 1: model cost - n * log2(k) bits to encode node assignments + model_cost = n * np.log2(k) if k > 1 else 0.0 + + # Part 2: data cost - binary entropy per block pair + data_cost = 0.0 + for i, r in enumerate(comm_ids): + n_r = len(communities[r]) + for j, s in enumerate(comm_ids): + if j < i: + continue + n_s = len(communities[s]) + + # Maximum possible edges between blocks r and s. + # Directed graphs allow both (u->v) and (v->u): the off-diagonal + # capacity doubles and the on-diagonal uses n_r*(n_r-1). + if r == s: + m_rs = n_r * (n_r - 1) if directed else n_r * (n_r - 1) / 2 + else: + m_rs = 2 * n_r * n_s if directed else n_r * n_s + if m_rs == 0: + continue + + e_rs = edge_counts.get((min(r, s), max(r, s)), 0) + + # Clamp to [0, 1]. Parallel/directional edges can push p above 1, + # which makes log2(1 - p) return nan and silently corrupt the score. + p = min(e_rs / m_rs, 1.0) + + # Binary entropy H(p); 0 when p == 0 or p == 1 + if 0.0 < p < 1.0: + data_cost += m_rs * (-p * np.log2(p) - (1.0 - p) * np.log2(1.0 - p)) + + return model_cost + data_cost + + +def mdl_score( + partition: Dict[Any, int], + network: Any, +) -> float: + """Compute MDL (Minimum Description Length) score for a multilayer partition. + + Computes the SBM two-part description length independently per layer and + sums the results. Operating per layer avoids the flattening problem where + pooling all layers into core_network inflates block sizes with + cross-layer node-layer pairs that share no edges, corrupting the edge + density estimates that drive the data cost. + + Per-layer computation: + - Part 1 (model cost): n_ℓ * log2(k_ℓ) bits to encode the community + assignment of each node in layer ℓ, where n_ℓ is the number of + node-layer pairs in that layer and k_ℓ is the number of distinct + communities present in that layer. + - Part 2 (data cost): for every block pair (r, s) within layer ℓ, + the binary entropy of the observed intra-layer edge density times the + maximum possible intra-layer edges between those blocks. + + Total MDL = Σ_ℓ (model_cost_ℓ + data_cost_ℓ) + + Lower score is better. + + Args: + partition: Dict mapping (node_id, layer) -> community_id. + network: Multilayer network with a core_network NetworkX graph. + + Returns: + Total description length in bits (float). Returns 0.0 for empty + inputs. + + Examples: + >>> # Perfect 2-community split on a clique pair + >>> partition = {('A', 'L'): 0, ('B', 'L'): 0, ('C', 'L'): 1, ('D', 'L'): 1} + >>> score = mdl_score(partition, net) + >>> assert score >= 0.0 + """ + if not partition: + return 0.0 + + G = network.core_network + directed = G.is_directed() + + # Collect layers and per-layer node assignments + layer_partitions: Dict[Any, Dict[Any, int]] = defaultdict(dict) + for node, comm in partition.items(): + if isinstance(node, tuple) and len(node) >= 2: + layer = node[1] + else: + layer = None + layer_partitions[layer][node] = comm + + # Pre-index edges by layer: only keep intra-layer edges + layer_edges: Dict[Any, List[tuple]] = defaultdict(list) + for u, v in G.edges(): + u_layer = u[1] if isinstance(u, tuple) and len(u) >= 2 else None + v_layer = v[1] if isinstance(v, tuple) and len(v) >= 2 else None + if u_layer == v_layer: + layer_edges[u_layer].append((u, v)) + + # Sum description length across layers + total = 0.0 + for layer, lpartition in layer_partitions.items(): + total += _mdl_single_layer(lpartition, layer_edges[layer], directed=directed) + + return total From 6fd7d92eb2c9888b37e688cccd6d0d180c9ec11e Mon Sep 17 00:00:00 2001 From: popchanovska Date: Mon, 6 Jul 2026 17:29:42 +0200 Subject: [PATCH 3/9] Resolve comment #1: Partial partitions can be unfairly rewarded. --- .../multilayer_quality_metrics.py | 122 +++++++++++------- 1 file changed, 77 insertions(+), 45 deletions(-) diff --git a/py3plex/algorithms/community_detection/multilayer_quality_metrics.py b/py3plex/algorithms/community_detection/multilayer_quality_metrics.py index 8f673e700..d4a42e066 100644 --- a/py3plex/algorithms/community_detection/multilayer_quality_metrics.py +++ b/py3plex/algorithms/community_detection/multilayer_quality_metrics.py @@ -328,7 +328,7 @@ def layer_entropy( def _mdl_single_layer( - layer_partition: Dict[Any, int], + layer_partition: Dict[Any, Any], layer_edges: List[tuple], directed: bool = False, ) -> float: @@ -336,6 +336,8 @@ def _mdl_single_layer( Args: layer_partition: Dict mapping node -> community_id for this layer only. + Community ids need only be hashable (e.g. singleton sentinel + objects for unassigned nodes are supported). layer_edges: List of (u, v) edge pairs within this layer. directed: Whether the underlying graph is directed. Controls the maximum-edge normalization for block pairs. Defaults to False @@ -345,7 +347,7 @@ def _mdl_single_layer( Returns: Description length in bits for this layer. """ - communities: Dict[int, List[Any]] = defaultdict(list) + communities: Dict[Any, List[Any]] = defaultdict(list) for node, comm in layer_partition.items(): communities[comm].append(node) @@ -355,9 +357,12 @@ def _mdl_single_layer( if k == 0 or n == 0: return 0.0 - # Count edges between/within each block pair + # Count edges between/within each block pair. Community ids are only + # required to be hashable (not orderable) -- singleton sentinel ids used + # for unassigned nodes are plain objects, so pairs are keyed with a + # frozenset rather than a min/max tuple. # Self-loops are skipped: they do not fit the n_r*(n_r-1) pair model - edge_counts: Dict[tuple, int] = defaultdict(int) + edge_counts: Dict[frozenset, int] = defaultdict(int) for u, v in layer_edges: if u == v: continue @@ -365,41 +370,44 @@ def _mdl_single_layer( s = layer_partition.get(v) if r is None or s is None: continue - edge_counts[(min(r, s), max(r, s))] += 1 - - comm_ids = list(communities.keys()) + edge_counts[frozenset((r, s))] += 1 # Part 1: model cost - n * log2(k) bits to encode node assignments model_cost = n * np.log2(k) if k > 1 else 0.0 - # Part 2: data cost - binary entropy per block pair + # Part 2: data cost - binary entropy per block pair. Block pairs with no + # observed edges contribute exactly 0 (H(p=0) == 0), so it suffices to + # iterate over the pairs that actually have edges rather than every pair + # of communities. This keeps the cost O(|E|) instead of O(k^2), which + # matters once unassigned nodes are represented as singleton communities + # (k can then be as large as n). data_cost = 0.0 - for i, r in enumerate(comm_ids): + for pair, e_rs in edge_counts.items(): + if len(pair) == 1: + (r,) = pair + s = r + else: + r, s = tuple(pair) n_r = len(communities[r]) - for j, s in enumerate(comm_ids): - if j < i: - continue - n_s = len(communities[s]) - - # Maximum possible edges between blocks r and s. - # Directed graphs allow both (u->v) and (v->u): the off-diagonal - # capacity doubles and the on-diagonal uses n_r*(n_r-1). - if r == s: - m_rs = n_r * (n_r - 1) if directed else n_r * (n_r - 1) / 2 - else: - m_rs = 2 * n_r * n_s if directed else n_r * n_s - if m_rs == 0: - continue - - e_rs = edge_counts.get((min(r, s), max(r, s)), 0) - - # Clamp to [0, 1]. Parallel/directional edges can push p above 1, - # which makes log2(1 - p) return nan and silently corrupt the score. - p = min(e_rs / m_rs, 1.0) - - # Binary entropy H(p); 0 when p == 0 or p == 1 - if 0.0 < p < 1.0: - data_cost += m_rs * (-p * np.log2(p) - (1.0 - p) * np.log2(1.0 - p)) + n_s = len(communities[s]) + + # Maximum possible edges between blocks r and s. + # Directed graphs allow both (u->v) and (v->u): the off-diagonal + # capacity doubles and the on-diagonal uses n_r*(n_r-1). + if r == s: + m_rs = n_r * (n_r - 1) if directed else n_r * (n_r - 1) / 2 + else: + m_rs = 2 * n_r * n_s if directed else n_r * n_s + if m_rs == 0: + continue + + # Clamp to [0, 1]. Parallel/directional edges can push p above 1, + # which makes log2(1 - p) return nan and silently corrupt the score. + p = min(e_rs / m_rs, 1.0) + + # Binary entropy H(p); 0 when p == 0 or p == 1 + if 0.0 < p < 1.0: + data_cost += m_rs * (-p * np.log2(p) - (1.0 - p) * np.log2(1.0 - p)) return model_cost + data_cost @@ -429,6 +437,13 @@ def mdl_score( Lower score is better. + Partial partitions: every node present in the network is accounted for, + not just the ones covered by `partition`. A node the partition omits is + scored as its own singleton community, so both n_ℓ (model cost) and its + incident edges (data cost) still count. Without this, an algorithm could + lower its MDL score simply by leaving hard nodes and their edges out of + the partition. + Args: partition: Dict mapping (node_id, layer) -> community_id. network: Multilayer network with a core_network NetworkX graph. @@ -443,24 +458,41 @@ def mdl_score( >>> score = mdl_score(partition, net) >>> assert score >= 0.0 """ - if not partition: + G = getattr(network, "core_network", None) + node_set = set(G.nodes()) if G is not None else set() + all_nodes = node_set | set(partition.keys()) + if not all_nodes: return 0.0 - G = network.core_network - directed = G.is_directed() - - # Collect layers and per-layer node assignments - layer_partitions: Dict[Any, Dict[Any, int]] = defaultdict(dict) - for node, comm in partition.items(): - if isinstance(node, tuple) and len(node) >= 2: - layer = node[1] + directed = G.is_directed() if G is not None else False + + # Collect layers and per-layer node assignments. Every node in the + # network is included -- not just the ones covered by `partition` -- so + # that an algorithm returning a partial partition cannot shrink n_ℓ or + # skip edges just by omitting nodes. Nodes missing from `partition` get a + # unique sentinel object as their "community", i.e. each is treated as + # its own singleton community. + layer_partitions: Dict[Any, Dict[Any, Any]] = defaultdict(dict) + n_unassigned = 0 + for node in all_nodes: + layer = node[1] if isinstance(node, tuple) and len(node) >= 2 else None + if node in partition: + layer_partitions[layer][node] = partition[node] else: - layer = None - layer_partitions[layer][node] = comm + layer_partitions[layer][node] = object() + n_unassigned += 1 + + if n_unassigned: + warnings.warn( + f"Partition covers {len(partition)} of {len(all_nodes)} nodes; " + f"{n_unassigned} unassigned node(s) are scored as singleton " + "communities so they cannot be omitted to lower the MDL score.", + stacklevel=2 + ) # Pre-index edges by layer: only keep intra-layer edges layer_edges: Dict[Any, List[tuple]] = defaultdict(list) - for u, v in G.edges(): + for u, v in (G.edges() if G is not None else []): u_layer = u[1] if isinstance(u, tuple) and len(u) >= 2 else None v_layer = v[1] if isinstance(v, tuple) and len(v) >= 2 else None if u_layer == v_layer: From 17c7dbbb498810c941de5655e6e3d948da1cc5d8 Mon Sep 17 00:00:00 2001 From: popchanovska Date: Mon, 6 Jul 2026 17:47:09 +0200 Subject: [PATCH 4/9] Resolve comment #2: Inter-layer edges are ignored. --- .../multilayer_quality_metrics.py | 80 +++++++++++++++---- 1 file changed, 65 insertions(+), 15 deletions(-) diff --git a/py3plex/algorithms/community_detection/multilayer_quality_metrics.py b/py3plex/algorithms/community_detection/multilayer_quality_metrics.py index d4a42e066..27b9e79d2 100644 --- a/py3plex/algorithms/community_detection/multilayer_quality_metrics.py +++ b/py3plex/algorithms/community_detection/multilayer_quality_metrics.py @@ -331,21 +331,36 @@ def _mdl_single_layer( layer_partition: Dict[Any, Any], layer_edges: List[tuple], directed: bool = False, + include_model_cost: bool = True, ) -> float: - """Compute SBM two-part description length for a single flat graph layer. + """Compute SBM two-part description length for a flat edge set. + + Despite the name, this is used for two things in `mdl_score`: the + per-layer intra-layer term (with `include_model_cost=True`, using a + single layer's local node -> community assignment), and the global + inter-layer term (with `include_model_cost=False`, using the full + across-all-layers node-layer -> community assignment, restricted to + edges that cross layers). See `mdl_score` for why model cost is only + charged once, per layer. Args: - layer_partition: Dict mapping node -> community_id for this layer only. + layer_partition: Dict mapping node -> community_id. Community ids need only be hashable (e.g. singleton sentinel objects for unassigned nodes are supported). - layer_edges: List of (u, v) edge pairs within this layer. + layer_edges: List of (u, v) edge pairs to account for. Both + endpoints must be keys of `layer_partition`. directed: Whether the underlying graph is directed. Controls the maximum-edge normalization for block pairs. Defaults to False (undirected), preserving backward-compatible behavior for callers that do not pass the flag. + include_model_cost: Whether to add the n * log2(k) model-cost term. + Set to False when `layer_partition`/`layer_edges` describe a + block structure whose assignment cost was already charged + elsewhere (e.g. the inter-layer term reuses assignments already + paid for by the per-layer model cost). Returns: - Description length in bits for this layer. + Description length in bits. """ communities: Dict[Any, List[Any]] = defaultdict(list) for node, comm in layer_partition.items(): @@ -373,7 +388,7 @@ def _mdl_single_layer( edge_counts[frozenset((r, s))] += 1 # Part 1: model cost - n * log2(k) bits to encode node assignments - model_cost = n * np.log2(k) if k > 1 else 0.0 + model_cost = n * np.log2(k) if (include_model_cost and k > 1) else 0.0 # Part 2: data cost - binary entropy per block pair. Block pairs with no # observed edges contribute exactly 0 (H(p=0) == 0), so it suffices to @@ -419,12 +434,13 @@ def mdl_score( """Compute MDL (Minimum Description Length) score for a multilayer partition. Computes the SBM two-part description length independently per layer and - sums the results. Operating per layer avoids the flattening problem where - pooling all layers into core_network inflates block sizes with - cross-layer node-layer pairs that share no edges, corrupting the edge - density estimates that drive the data cost. + sums the results, then adds one more term for inter-layer (cross-layer) + edges. Operating per layer for the intra-layer term avoids the flattening + problem where pooling all layers into core_network inflates block sizes + with cross-layer node-layer pairs that share no edges, corrupting the + edge density estimates that drive the data cost. - Per-layer computation: + Per-layer computation (intra-layer edges only): - Part 1 (model cost): n_ℓ * log2(k_ℓ) bits to encode the community assignment of each node in layer ℓ, where n_ℓ is the number of node-layer pairs in that layer and k_ℓ is the number of distinct @@ -433,9 +449,27 @@ def mdl_score( the binary entropy of the observed intra-layer edge density times the maximum possible intra-layer edges between those blocks. - Total MDL = Σ_ℓ (model_cost_ℓ + data_cost_ℓ) - - Lower score is better. + Inter-layer computation (edges whose endpoints are in different layers, + e.g. multiplex coupling edges or general cross-layer edges): + - No additional model cost is charged -- every node-layer pair's + community assignment was already paid for by the per-layer model + cost above. + - Data cost only: block sizes N_r/N_s are the *global* community + sizes (summed across all layers), and the block pair (r, s) is + scored using the observed inter-layer edge density between them, + the same binary-entropy formula as the intra-layer case. This + slightly overstates the space of possible inter-layer edges (it + does not subtract same-layer node pairs from N_r * N_s, since + inter-layer edges in py3plex are not restricted to same-node + couplings and can connect arbitrary node-layer pairs), which is a + conservative approximation rather than an exact block model. + + Total MDL = Σ_ℓ (model_cost_ℓ + data_cost_ℓ) + data_cost_inter_layer + + Lower score is better. Ignoring inter-layer edges entirely would let two + partitions with identical intra-layer structure but very different (e.g. + incoherent vs. replica-consistent) cross-layer coupling score identically, + which is misleading for a *multilayer* MDL metric. Partial partitions: every node present in the network is accounted for, not just the ones covered by `partition`. A node the partition omits is @@ -490,17 +524,33 @@ def mdl_score( stacklevel=2 ) - # Pre-index edges by layer: only keep intra-layer edges + # Split edges into intra-layer (same layer on both ends) and inter-layer + # (endpoints in different layers, e.g. multiplex coupling edges). layer_edges: Dict[Any, List[tuple]] = defaultdict(list) + inter_layer_edges: List[tuple] = [] for u, v in (G.edges() if G is not None else []): u_layer = u[1] if isinstance(u, tuple) and len(u) >= 2 else None v_layer = v[1] if isinstance(v, tuple) and len(v) >= 2 else None if u_layer == v_layer: layer_edges[u_layer].append((u, v)) + else: + inter_layer_edges.append((u, v)) - # Sum description length across layers + # Sum description length across layers (intra-layer model + data cost) total = 0.0 + global_partition: Dict[Any, Any] = {} for layer, lpartition in layer_partitions.items(): total += _mdl_single_layer(lpartition, layer_edges[layer], directed=directed) + global_partition.update(lpartition) + + # Add the inter-layer data cost: community assignments were already paid + # for above, so only the observed cross-layer edge density is charged. + if inter_layer_edges: + total += _mdl_single_layer( + global_partition, + inter_layer_edges, + directed=directed, + include_model_cost=False, + ) return total From da59bc09d9de3a8a2cab67398ced4942f3e5d84a Mon Sep 17 00:00:00 2001 From: popchanovska Date: Thu, 9 Jul 2026 15:11:55 +0200 Subject: [PATCH 5/9] Resolve comment #3: Parallel/weighted edges are not properly modeled. --- .../multilayer_quality_metrics.py | 80 ++++++++++++++++++- 1 file changed, 76 insertions(+), 4 deletions(-) diff --git a/py3plex/algorithms/community_detection/multilayer_quality_metrics.py b/py3plex/algorithms/community_detection/multilayer_quality_metrics.py index 27b9e79d2..ddbf9a9e5 100644 --- a/py3plex/algorithms/community_detection/multilayer_quality_metrics.py +++ b/py3plex/algorithms/community_detection/multilayer_quality_metrics.py @@ -20,6 +20,7 @@ from collections import defaultdict from typing import Any, Dict, Iterable, List, Optional, Tuple +import networkx as nx import numpy as np @@ -327,6 +328,32 @@ def layer_entropy( return float(clipped_entropy) +def _dedupe_node_pairs(edges: List[tuple], directed: bool) -> List[tuple]: + """Collapse parallel/multi-edges into a single edge per node pair. + + py3plex's default `core_network` container is a `nx.MultiGraph` / + `nx.MultiDiGraph`, which can hold more than one edge between the same + node pair. The block-pair capacity `m_rs` computed in + `_mdl_single_layer` counts *distinct* node pairs, so leaving multi-edges + un-collapsed lets the raw edge count `e_rs` exceed `m_rs` -- silently + clamped to a density of 1.0, which scores e.g. a 2-edge and a 200-edge + block identically. Deduplicating here is a deliberate, explicit collapse + (one of "reject / collapse / weighted likelihood") that keeps the + Bernoulli block model well-defined for multigraph inputs, at the cost of + discarding multiplicity (and, since `_mdl_single_layer` never looks at + edge weight, weight magnitude too -- see `mdl_score`'s warning). + """ + seen: set = set() + deduped = [] + for u, v in edges: + key = (u, v) if directed else frozenset((u, v)) + if key in seen: + continue + seen.add(key) + deduped.append((u, v)) + return deduped + + def _mdl_single_layer( layer_partition: Dict[Any, Any], layer_edges: List[tuple], @@ -348,7 +375,10 @@ def _mdl_single_layer( Community ids need only be hashable (e.g. singleton sentinel objects for unassigned nodes are supported). layer_edges: List of (u, v) edge pairs to account for. Both - endpoints must be keys of `layer_partition`. + endpoints must be keys of `layer_partition`. May contain + parallel/multi-edges (e.g. from a MultiGraph); these are + collapsed to a single edge per node pair before scoring, see + `_dedupe_node_pairs`. directed: Whether the underlying graph is directed. Controls the maximum-edge normalization for block pairs. Defaults to False (undirected), preserving backward-compatible behavior for callers @@ -372,6 +402,10 @@ def _mdl_single_layer( if k == 0 or n == 0: return 0.0 + # Collapse parallel/multi-edges before block-pair counting, so e_rs can + # never exceed the simple-graph capacity m_rs computed below. + layer_edges = _dedupe_node_pairs(layer_edges, directed) + # Count edges between/within each block pair. Community ids are only # required to be hashable (not orderable) -- singleton sentinel ids used # for unassigned nodes are plain objects, so pairs are keyed with a @@ -416,8 +450,11 @@ def _mdl_single_layer( if m_rs == 0: continue - # Clamp to [0, 1]. Parallel/directional edges can push p above 1, - # which makes log2(1 - p) return nan and silently corrupt the score. + # e_rs <= m_rs is now guaranteed by the parallel-edge collapse above + # (each node pair contributes at most one edge), so this clamp is a + # defensive fallback rather than the primary safeguard -- without it, + # p > 1 would make log2(1 - p) return nan and silently corrupt the + # score. p = min(e_rs / m_rs, 1.0) # Binary entropy H(p); 0 when p == 0 or p == 1 @@ -478,6 +515,17 @@ def mdl_score( lower its MDL score simply by leaving hard nodes and their edges out of the partition. + Parallel/weighted edges: the block model above is Bernoulli (edge + present or absent between a node pair), which is only well-defined for + simple graphs. py3plex's default network container is a + `nx.MultiGraph`/`nx.MultiDiGraph`, so if the same node pair carries more + than one edge, it is collapsed to a single edge before scoring (see + `_dedupe_node_pairs`) rather than left to silently overflow the + simple-graph capacity and clamp to a density of 1.0. Edge `weight` + attributes are likewise ignored -- every edge counts as one, regardless + of magnitude. Both cases emit a `UserWarning` since the description + length then cannot distinguish a lightly- from a heavily-connected block. + Args: partition: Dict mapping (node_id, layer) -> community_id. network: Multilayer network with a core_network NetworkX graph. @@ -499,6 +547,30 @@ def mdl_score( return 0.0 directed = G.is_directed() if G is not None else False + raw_edges: List[tuple] = list(G.edges()) if G is not None else [] + + # Detect parallel edges (possible on py3plex's default MultiGraph/ + # MultiDiGraph container) and non-unit edge weights. Neither is modeled + # by the Bernoulli block structure below -- multi-edges are collapsed to + # a single edge per node pair (see `_dedupe_node_pairs`) and weight + # magnitude is ignored entirely -- so warn rather than silently scoring + # a lightly- and heavily-connected block identically. + if isinstance(G, (nx.MultiGraph, nx.MultiDiGraph)): + has_parallel_edges = len(_dedupe_node_pairs(raw_edges, directed)) < len(raw_edges) + else: + has_parallel_edges = False + has_weighted_edges = G is not None and any( + data.get("weight", 1) != 1 for _, _, data in G.edges(data=True) + ) + if has_parallel_edges or has_weighted_edges: + warnings.warn( + "Network has parallel edges and/or non-unit edge weights; " + "mdl_score models a simple Bernoulli block structure, so " + "parallel edges are collapsed to one edge per node pair and " + "weight magnitude is ignored. The resulting description length " + "cannot distinguish a lightly- from a heavily-connected block.", + stacklevel=2, + ) # Collect layers and per-layer node assignments. Every node in the # network is included -- not just the ones covered by `partition` -- so @@ -528,7 +600,7 @@ def mdl_score( # (endpoints in different layers, e.g. multiplex coupling edges). layer_edges: Dict[Any, List[tuple]] = defaultdict(list) inter_layer_edges: List[tuple] = [] - for u, v in (G.edges() if G is not None else []): + for u, v in raw_edges: u_layer = u[1] if isinstance(u, tuple) and len(u) >= 2 else None v_layer = v[1] if isinstance(v, tuple) and len(v) >= 2 else None if u_layer == v_layer: From 690ec525ce7222deb1f4dd4a051cfa682f964532 Mon Sep 17 00:00:00 2001 From: popchanovska Date: Thu, 9 Jul 2026 15:50:29 +0200 Subject: [PATCH 6/9] Note for comment 4: Tested on 5000-node graph --- .../multilayer_quality_metrics.py | 28 ++++++++++++ tests/test_multilayer_quality_metrics.py | 44 +++++++++++++++++++ 2 files changed, 72 insertions(+) diff --git a/py3plex/algorithms/community_detection/multilayer_quality_metrics.py b/py3plex/algorithms/community_detection/multilayer_quality_metrics.py index ddbf9a9e5..ff37fdb6c 100644 --- a/py3plex/algorithms/community_detection/multilayer_quality_metrics.py +++ b/py3plex/algorithms/community_detection/multilayer_quality_metrics.py @@ -391,6 +391,21 @@ def _mdl_single_layer( Returns: Description length in bits. + + Notes: + - Complexity: O(n + |E|), not O(k^2). The data cost loop iterates + only over `edge_counts` (block pairs with at least one observed + edge), not over all k^2 pairs of communities -- block pairs with + no edges contribute exactly 0 (H(p=0) == 0), so they're skipped + rather than enumerated. This matters once heavily-fragmented + partitions (e.g. many singleton communities from unassigned nodes, + see `mdl_score`) push k up toward n: a naive double loop over + communities would be O(k^2) regardless of how sparse the edges + are, whereas this stays tied to the actual edge count. The + remaining cost still scales with |E| itself, which is inherent to + reading every edge once and not specific to fragmentation -- a + dense graph is the same O(|E|) cost whether it has 2 communities + or n of them. """ communities: Dict[Any, List[Any]] = defaultdict(list) for node, comm in layer_partition.items(): @@ -534,6 +549,19 @@ def mdl_score( Total description length in bits (float). Returns 0.0 for empty inputs. + Notes: + - Complexity: O(N + |E|) total across layers, where N is the number + of node-layer pairs and E the edges (intra- plus inter-layer) -- + see `_mdl_single_layer`'s Notes for why fragmentation (many + singleton communities, e.g. from unassigned nodes) doesn't push + this toward O(k^2). On expected AutoCommunity graph sizes (up to + ~1e5 nodes / ~1e6 edges, matching the practical_limits declared + for candidate algorithms like leiden_multilayer), this is expected + to stay well under a second; see + `tests/test_multilayer_quality_metrics.py::TestPerformance::test_no_quadratic_blowup_mdl_fragmented` + for a regression guard against a future reintroduction of a + quadratic-in-communities loop. + Examples: >>> # Perfect 2-community split on a clique pair >>> partition = {('A', 'L'): 0, ('B', 'L'): 0, ('C', 'L'): 1, ('D', 'L'): 1} diff --git a/tests/test_multilayer_quality_metrics.py b/tests/test_multilayer_quality_metrics.py index 9e31e3b6c..3bc81a6cb 100644 --- a/tests/test_multilayer_quality_metrics.py +++ b/tests/test_multilayer_quality_metrics.py @@ -11,6 +11,7 @@ import pytest import numpy as np +import networkx as nx from collections import defaultdict from py3plex.core import multinet @@ -18,6 +19,7 @@ replica_consistency, layer_entropy, iter_layered_assignments, + mdl_score, ) @@ -528,6 +530,48 @@ def test_no_quadratic_blowup_entropy(self): assert elapsed < 1.0 assert 0.0 <= h <= 1.0 + def test_no_quadratic_blowup_mdl_fragmented(self): + """mdl_score should stay near-linear even with many singleton + communities, not quadratic in the number of communities per layer. + + _mdl_single_layer's data cost loop iterates over observed block + pairs (`edge_counts`), not over all k^2 community pairs, so cost + tracks O(n + |E|) rather than O(k^2). Extreme fragmentation (every + node its own community, as an AutoCommunity algorithm might produce + for a partial/degenerate partition) maximizes k without blowing up + |E| -- this is exactly the scenario a regression to a k^2 nested + loop over communities would fail on. + """ + n = 5000 + nodes = [(f"node_{i}", "layer_0") for i in range(n)] + + # Sparse, deterministic edge set (a cycle): O(n) edges, and since + # every node is its own community, each edge lands on a distinct + # community pair -- the worst case for the number of block pairs + # relative to node count. + G = nx.Graph() + G.add_nodes_from(nodes) + for i in range(n): + G.add_edge(nodes[i], nodes[(i + 1) % n]) + + net = multinet.multi_layer_network(directed=False) + net.core_network = G + + # Every node in its own singleton community (maximal fragmentation). + partition = {node: i for i, node in enumerate(nodes)} + + # Partition covers every node on a plain (non-multi, unweighted) + # graph, so none of mdl_score's warning paths (partial partition, + # parallel/weighted edges) should fire here. + import time + start = time.time() + score = mdl_score(partition, net) + elapsed = time.time() - start + + # Should complete in reasonable time (< 2 seconds) despite k == n. + assert elapsed < 2.0 + assert score >= 0.0 + class TestIntegrationWithAutoCommunity: """Test integration with AutoCommunity.""" From 95b245f12ff58ee688c978bcd71b327f19a961f6 Mon Sep 17 00:00:00 2001 From: popchanovska Date: Thu, 9 Jul 2026 16:04:38 +0200 Subject: [PATCH 7/9] Note for comment 4: Adda 2M-node graph test --- tests/test_multilayer_quality_metrics.py | 37 ++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/tests/test_multilayer_quality_metrics.py b/tests/test_multilayer_quality_metrics.py index 3bc81a6cb..16adfa59f 100644 --- a/tests/test_multilayer_quality_metrics.py +++ b/tests/test_multilayer_quality_metrics.py @@ -572,6 +572,43 @@ def test_no_quadratic_blowup_mdl_fragmented(self): assert elapsed < 2.0 assert score >= 0.0 + @pytest.mark.slow + def test_no_quadratic_blowup_mdl_large_scale(self): + """mdl_score should stay near-linear at ~2M nodes, not just at the + smaller scale in test_no_quadratic_blowup_mdl_fragmented. + + Marked slow: builds a 2,000,000-node graph and runs the full + mdl_score computation. Measured ~8s / ~3GB peak RSS on a dev + machine; a regression to a quadratic-in-communities loop would + take minutes rather than tens of seconds at this scale. Skip via + `pytest -m "not slow"` for a fast local/CI run. + """ + n = 2_000_000 + nodes = [(f"node_{i}", "layer_0") for i in range(n)] + + G = nx.Graph() + G.add_nodes_from(nodes) + for i in range(n): + G.add_edge(nodes[i], nodes[(i + 1) % n]) + + net = multinet.multi_layer_network(directed=False) + net.core_network = G + + # Worst case for _mdl_single_layer's block-pair counting: every + # node is its own singleton community, maximizing k without + # increasing |E|. + partition = {node: i for i, node in enumerate(nodes)} + + import time + start = time.time() + score = mdl_score(partition, net) + elapsed = time.time() - start + + # Generous bound to absorb slower/loaded CI hardware while still + # catching a real regression to quadratic-in-communities behavior. + assert elapsed < 60.0 + assert score >= 0.0 + class TestIntegrationWithAutoCommunity: """Test integration with AutoCommunity.""" From addff89ed902a3f7b49fd83ef689640c83875825 Mon Sep 17 00:00:00 2001 From: popchanovska Date: Fri, 10 Jul 2026 15:12:06 +0200 Subject: [PATCH 8/9] Comment 5: Add tests for MDL metric. --- .../multilayer_quality_metrics.py | 5 +- tests/test_multilayer_quality_metrics.py | 85 +------------------ 2 files changed, 6 insertions(+), 84 deletions(-) diff --git a/py3plex/algorithms/community_detection/multilayer_quality_metrics.py b/py3plex/algorithms/community_detection/multilayer_quality_metrics.py index ff37fdb6c..c9c908c49 100644 --- a/py3plex/algorithms/community_detection/multilayer_quality_metrics.py +++ b/py3plex/algorithms/community_detection/multilayer_quality_metrics.py @@ -558,8 +558,9 @@ def mdl_score( ~1e5 nodes / ~1e6 edges, matching the practical_limits declared for candidate algorithms like leiden_multilayer), this is expected to stay well under a second; see - `tests/test_multilayer_quality_metrics.py::TestPerformance::test_no_quadratic_blowup_mdl_fragmented` - for a regression guard against a future reintroduction of a + `tests/test_mdl_score.py::TestPerformance::test_no_quadratic_blowup_mdl_fragmented` + (and `..._large_scale` for a slow-marked ~2M-node check) for + regression guards against a future reintroduction of a quadratic-in-communities loop. Examples: diff --git a/tests/test_multilayer_quality_metrics.py b/tests/test_multilayer_quality_metrics.py index 16adfa59f..1b61e19a2 100644 --- a/tests/test_multilayer_quality_metrics.py +++ b/tests/test_multilayer_quality_metrics.py @@ -1,17 +1,18 @@ """Tests for multilayer-specific community quality metrics. -Tests cover: +Covers replica_consistency and layer_entropy: - Determinism (repeated calls yield identical results) - Correctness (known partitions yield expected metric values) - Edge cases (single-layer nodes, single community, fragmentation) - Hard-to-game behavior (giant cluster vs reasonable partition) - Performance (no quadratic blowups) - Integration with AutoCommunity + +mdl_score has its own dedicated suite in test_mdl_score.py. """ import pytest import numpy as np -import networkx as nx from collections import defaultdict from py3plex.core import multinet @@ -19,7 +20,6 @@ replica_consistency, layer_entropy, iter_layered_assignments, - mdl_score, ) @@ -530,85 +530,6 @@ def test_no_quadratic_blowup_entropy(self): assert elapsed < 1.0 assert 0.0 <= h <= 1.0 - def test_no_quadratic_blowup_mdl_fragmented(self): - """mdl_score should stay near-linear even with many singleton - communities, not quadratic in the number of communities per layer. - - _mdl_single_layer's data cost loop iterates over observed block - pairs (`edge_counts`), not over all k^2 community pairs, so cost - tracks O(n + |E|) rather than O(k^2). Extreme fragmentation (every - node its own community, as an AutoCommunity algorithm might produce - for a partial/degenerate partition) maximizes k without blowing up - |E| -- this is exactly the scenario a regression to a k^2 nested - loop over communities would fail on. - """ - n = 5000 - nodes = [(f"node_{i}", "layer_0") for i in range(n)] - - # Sparse, deterministic edge set (a cycle): O(n) edges, and since - # every node is its own community, each edge lands on a distinct - # community pair -- the worst case for the number of block pairs - # relative to node count. - G = nx.Graph() - G.add_nodes_from(nodes) - for i in range(n): - G.add_edge(nodes[i], nodes[(i + 1) % n]) - - net = multinet.multi_layer_network(directed=False) - net.core_network = G - - # Every node in its own singleton community (maximal fragmentation). - partition = {node: i for i, node in enumerate(nodes)} - - # Partition covers every node on a plain (non-multi, unweighted) - # graph, so none of mdl_score's warning paths (partial partition, - # parallel/weighted edges) should fire here. - import time - start = time.time() - score = mdl_score(partition, net) - elapsed = time.time() - start - - # Should complete in reasonable time (< 2 seconds) despite k == n. - assert elapsed < 2.0 - assert score >= 0.0 - - @pytest.mark.slow - def test_no_quadratic_blowup_mdl_large_scale(self): - """mdl_score should stay near-linear at ~2M nodes, not just at the - smaller scale in test_no_quadratic_blowup_mdl_fragmented. - - Marked slow: builds a 2,000,000-node graph and runs the full - mdl_score computation. Measured ~8s / ~3GB peak RSS on a dev - machine; a regression to a quadratic-in-communities loop would - take minutes rather than tens of seconds at this scale. Skip via - `pytest -m "not slow"` for a fast local/CI run. - """ - n = 2_000_000 - nodes = [(f"node_{i}", "layer_0") for i in range(n)] - - G = nx.Graph() - G.add_nodes_from(nodes) - for i in range(n): - G.add_edge(nodes[i], nodes[(i + 1) % n]) - - net = multinet.multi_layer_network(directed=False) - net.core_network = G - - # Worst case for _mdl_single_layer's block-pair counting: every - # node is its own singleton community, maximizing k without - # increasing |E|. - partition = {node: i for i, node in enumerate(nodes)} - - import time - start = time.time() - score = mdl_score(partition, net) - elapsed = time.time() - start - - # Generous bound to absorb slower/loaded CI hardware while still - # catching a real regression to quadratic-in-communities behavior. - assert elapsed < 60.0 - assert score >= 0.0 - class TestIntegrationWithAutoCommunity: """Test integration with AutoCommunity.""" From dc673e44fe1abec7ca3ae53cc0a15ca01dfc22eb Mon Sep 17 00:00:00 2001 From: popchanovska Date: Fri, 10 Jul 2026 15:17:06 +0200 Subject: [PATCH 9/9] Comment 5: Add tests for MDL metric pt.2 --- tests/test_mdl_score.py | 597 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 597 insertions(+) create mode 100644 tests/test_mdl_score.py diff --git a/tests/test_mdl_score.py b/tests/test_mdl_score.py new file mode 100644 index 000000000..1d13efb98 --- /dev/null +++ b/tests/test_mdl_score.py @@ -0,0 +1,597 @@ +"""Tests for mdl_score (Minimum Description Length quality metric). + +mdl_score computes an SBM two-part description length for a multilayer +partition (lower is better). + +- partial partitions: omitting hard nodes from a partition + must not lower the score -- see TestMissingPartitionNodes. +- inter-layer edges: cross-layer/coupling edges must be + scored, not ignored -- see TestInterLayerEdges. +- parallel/weighted edges: multi-edges and edge weight + magnitude are collapsed/ignored explicitly (not silently clamped), and + disclosed via a warning -- see TestParallelAndWeightedEdges. +- performance: the data-cost loop is O(n + |E|), not + quadratic in the number of communities, even under extreme + fragmentation -- see TestPerformance. +- baseline coverage: empty partitions, single-layer graphs, + multi-layer graphs, directed graphs, singleton communities, missing + partition nodes, inter-layer edges, plus additional cases (self-loops, + determinism, score quality, duck-typed network inputs) judged necessary + for a metric this easy to silently break. +""" + +import time +import warnings + +import networkx as nx +import numpy as np +import pytest + +from py3plex.core import multinet +from py3plex.algorithms.community_detection.multilayer_quality_metrics import ( + mdl_score, +) + + +def _net(G, directed=False): + """Build a multi_layer_network wrapping a prebuilt networkx graph.""" + net = multinet.multi_layer_network(directed=directed) + net.core_network = G + return net + + +class TestEmptyPartitions: + + def test_empty_partition_and_empty_network(self): + """No partition, no network content = 0.0, no warning.""" + net = multinet.multi_layer_network(directed=False) + assert mdl_score({}, net) == 0.0 + + def test_empty_partition_nonempty_network_warns(self): + """Empty partition but the network has nodes: every node becomes + its own singleton community and a partial-partition warning fires. + """ + G = nx.Graph() + G.add_edge(("A", "L"), ("B", "L")) + net = _net(G) + + with pytest.warns(UserWarning, match="Partition covers 0 of"): + score = mdl_score({}, net) + assert score >= 0.0 + + def test_empty_partition_none_core_network(self): + """A network whose core_network was never populated (stays None) + combined with an empty partition returns 0.0.""" + net = multinet.multi_layer_network(directed=False) + assert net.core_network is None + assert mdl_score({}, net) == 0.0 + + +class TestSingleAndMultiLayerGraphs: + """Single-layer and multi-layer graphs.""" + + def test_single_layer_graph(self): + """Perfect 2-community split on one layer scores a known value.""" + G = nx.Graph() + G.add_edge(("A", "L"), ("B", "L")) + G.add_edge(("C", "L"), ("D", "L")) + net = _net(G) + partition = {("A", "L"): 0, ("B", "L"): 0, ("C", "L"): 1, ("D", "L"): 1} + + score = mdl_score(partition, net) + assert score == pytest.approx(4.0, abs=1e-6) + + def test_multi_layer_independent_layers_sum_exactly(self): + """Two structurally identical layers with no coupling edges score + exactly 2x a single layer's score: layers are scored independently + and summed, with no spurious inter-layer term when there are no + cross-layer edges (contrast with TestInterLayerEdges, where coupling + edges are present and do add a term). + """ + G1 = nx.Graph() + G1.add_edge(("A", "L1"), ("B", "L1")) + G1.add_edge(("C", "L1"), ("D", "L1")) + p1 = {("A", "L1"): 0, ("B", "L1"): 0, ("C", "L1"): 1, ("D", "L1"): 1} + single_score = mdl_score(p1, _net(G1)) + + G2 = nx.Graph() + partition2 = {} + for layer in ("L1", "L2"): + G2.add_edge(("A", layer), ("B", layer)) + G2.add_edge(("C", layer), ("D", layer)) + partition2[("A", layer)] = 0 + partition2[("B", layer)] = 0 + partition2[("C", layer)] = 1 + partition2[("D", layer)] = 1 + multi_score = mdl_score(partition2, _net(G2)) + + assert multi_score == pytest.approx(2 * single_score, rel=1e-9) + + def test_multi_layer_different_partitions_per_layer(self): + """Layers may have completely different community counts/labels + without crashing or interfering with each other's scoring.""" + G = nx.Graph() + G.add_edge(("A", "L1"), ("B", "L1")) + G.add_edge(("C", "L2"), ("D", "L2")) + G.add_edge(("D", "L2"), ("E", "L2")) + net = _net(G) + partition = { + ("A", "L1"): 0, ("B", "L1"): 0, + ("C", "L2"): 0, ("D", "L2"): 1, ("E", "L2"): 2, + } + + score = mdl_score(partition, net) + assert np.isfinite(score) + assert score >= 0.0 + + +class TestDirectedGraphs: + """Directed graphs.""" + + def test_directed_graph_uses_directed_capacity(self): + """Directed and undirected scores differ on identical + edges/partition, confirming the directed capacity branch (doubled + off-diagonal, n*(n-1) on-diagonal) is actually used.""" + partition = {("A", "L"): 0, ("B", "L"): 0, ("C", "L"): 1, ("D", "L"): 1} + + Gd = nx.DiGraph() + Gd.add_edge(("A", "L"), ("B", "L")) + Gd.add_edge(("C", "L"), ("D", "L")) + directed_score = mdl_score(partition, _net(Gd, directed=True)) + + Gu = nx.Graph() + Gu.add_edge(("A", "L"), ("B", "L")) + Gu.add_edge(("C", "L"), ("D", "L")) + undirected_score = mdl_score(partition, _net(Gu, directed=False)) + + assert directed_score == pytest.approx(8.0, abs=1e-6) + assert undirected_score == pytest.approx(4.0, abs=1e-6) + assert directed_score != undirected_score + + def test_directed_multigraph_parallel_edges(self): + """A directed MultiDiGraph with parallel directed edges collapses + correctly (ordered dedup key, not frozenset) and warns without + crashing or producing nan.""" + G = nx.MultiDiGraph() + for _ in range(5): + G.add_edge(("A", "L"), ("B", "L")) + G.add_edge(("C", "L"), ("D", "L")) + net = _net(G, directed=True) + partition = {("A", "L"): 0, ("B", "L"): 0, ("C", "L"): 1, ("D", "L"): 1} + + with pytest.warns(UserWarning): + score = mdl_score(partition, net) + assert score == pytest.approx(8.0, abs=1e-6) + assert np.isfinite(score) + + +class TestSingletonCommunities: + """Singleton communities.""" + + def test_singleton_communities_no_crash(self): + """Every node in its own community should not crash and stays + finite even with fully fragmented input.""" + G = nx.Graph() + G.add_edge(("A", "L"), ("B", "L")) + G.add_edge(("B", "L"), ("C", "L")) + G.add_edge(("C", "L"), ("D", "L")) + net = _net(G) + partition = {("A", "L"): 0, ("B", "L"): 1, ("C", "L"): 2, ("D", "L"): 3} + + score = mdl_score(partition, net) + assert np.isfinite(score) + assert score == pytest.approx(8.0, abs=1e-6) + + def test_non_orderable_community_ids(self): + """Community ids need only be hashable, not orderable -- using + plain (non-orderable) objects as labels must not crash the + frozenset-keyed block-pair counting.""" + G = nx.Graph() + G.add_edge(("A", "L"), ("B", "L")) + net = _net(G) + c0, c1 = object(), object() + partition = {("A", "L"): c0, ("B", "L"): c1} + + score = mdl_score(partition, net) + assert np.isfinite(score) + assert score >= 0.0 + + +class TestMissingPartitionNodes: + """Regression: partial partitions can't be gamed by + omitting hard nodes.""" + + def test_missing_partition_nodes_cannot_lower_score(self): + """Omitting hard nodes from the partition must not lower the score + below what the full partition gets.""" + G = nx.Graph() + for u, v in [("A", "B"), ("B", "C"), ("C", "A"), + ("D", "E"), ("E", "F"), ("F", "D"), ("A", "D")]: + G.add_edge((u, "L"), (v, "L")) + net = _net(G) + + full = {("A", "L"): 0, ("B", "L"): 0, ("C", "L"): 0, + ("D", "L"): 1, ("E", "L"): 1, ("F", "L"): 1} + partial = {k: v for k, v in full.items() if k != ("F", "L")} + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + full_score = mdl_score(full, net) + partial_score = mdl_score(partial, net) + + assert partial_score >= full_score + + def test_missing_partition_nodes_warns_with_counts(self): + """Partial partitions emit exactly one warning naming the exact + coverage counts.""" + G = nx.Graph() + G.add_edge(("A", "L"), ("B", "L")) + G.add_edge(("B", "L"), ("C", "L")) + net = _net(G) + partial = {("A", "L"): 0, ("B", "L"): 0} # "C" omitted + + with pytest.warns(UserWarning, match=r"Partition covers 2 of 3"): + mdl_score(partial, net) + + def test_fully_covered_partition_does_not_warn_about_coverage(self): + """No partial-partition warning when every network node is + assigned.""" + G = nx.Graph() + G.add_edge(("A", "L"), ("B", "L")) + net = _net(G) + full = {("A", "L"): 0, ("B", "L"): 1} + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + mdl_score(full, net) + assert not any("Partition covers" in str(w.message) for w in caught) + + +class TestInterLayerEdges: + """Regression: inter-layer (coupling) edges are scored, + not ignored.""" + + def test_inter_layer_edges_penalize_replica_inconsistency(self): + """With identical intra-layer structure, a replica-consistent + partition must score lower than an inconsistent one once coupling + edges are present.""" + G = nx.Graph() + for layer in ("L1", "L2"): + G.add_edge(("A", layer), ("B", layer)) + G.add_edge(("C", layer), ("D", layer)) + for node in ("A", "B", "C", "D"): + G.add_edge((node, "L1"), (node, "L2")) + net = _net(G) + + consistent = {} + for layer in ("L1", "L2"): + consistent[("A", layer)] = 0 + consistent[("B", layer)] = 0 + consistent[("C", layer)] = 1 + consistent[("D", layer)] = 1 + inconsistent = dict(consistent) + inconsistent[("A", "L2")] = 1 + inconsistent[("B", "L2")] = 1 + inconsistent[("C", "L2")] = 0 + inconsistent[("D", "L2")] = 0 + + consistent_score = mdl_score(consistent, net) + inconsistent_score = mdl_score(inconsistent, net) + assert consistent_score < inconsistent_score + + def test_inter_layer_edges_add_cost_over_no_coupling(self): + """Adding coupling edges on top of an otherwise-identical network + cannot decrease the score: the inter-layer term is purely additive + (zero when there are no cross-layer edges, a non-negative binary + entropy term otherwise) and never revisits the intra-layer terms. + """ + G_no_coupling = nx.Graph() + for layer in ("L1", "L2"): + G_no_coupling.add_edge(("A", layer), ("B", layer)) + partition = { + ("A", "L1"): 0, ("B", "L1"): 0, + ("A", "L2"): 0, ("B", "L2"): 0, + } + score_no_coupling = mdl_score(partition, _net(G_no_coupling)) + + G_coupling = nx.Graph(G_no_coupling) + G_coupling.add_edge(("A", "L1"), ("A", "L2")) + G_coupling.add_edge(("B", "L1"), ("B", "L2")) + score_with_coupling = mdl_score(partition, _net(G_coupling)) + + assert score_with_coupling >= score_no_coupling + + def test_no_inter_layer_edges_means_no_inter_layer_term(self): + """Two disconnected single-node-pair layers (no coupling at all) + must score identically to summing each layer's score alone -- + i.e. the inter-layer branch is skipped entirely, not charged as + zero-cost work that could hide a latent bug.""" + G = nx.Graph() + G.add_edge(("A", "L1"), ("B", "L1")) + G.add_edge(("C", "L2"), ("D", "L2")) + net = _net(G) + partition = { + ("A", "L1"): 0, ("B", "L1"): 0, + ("C", "L2"): 0, ("D", "L2"): 0, + } + + score = mdl_score(partition, net) + + G1 = nx.Graph() + G1.add_edge(("A", "L1"), ("B", "L1")) + score1 = mdl_score({("A", "L1"): 0, ("B", "L1"): 0}, _net(G1)) + + G2 = nx.Graph() + G2.add_edge(("C", "L2"), ("D", "L2")) + score2 = mdl_score({("C", "L2"): 0, ("D", "L2"): 0}, _net(G2)) + + assert score == pytest.approx(score1 + score2, rel=1e-9) + + +class TestParallelAndWeightedEdges: + """Regression: parallel/weighted edges are collapsed + explicitly and disclosed via a warning, never silently clamped.""" + + def test_parallel_edges_collapse_to_same_score_as_simple_graph(self): + """A MultiGraph with 10 parallel edges between the same pair scores + identically to a simple graph with 1 edge there -- multiplicity is + discarded by design, not silently corrupted via clamping to + density 1.0.""" + partition = {("A", "L"): 0, ("B", "L"): 0, ("C", "L"): 1, ("D", "L"): 1} + + Gm = nx.MultiGraph() + for _ in range(10): + Gm.add_edge(("A", "L"), ("B", "L")) + Gm.add_edge(("C", "L"), ("D", "L")) + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + multigraph_score = mdl_score(partition, _net(Gm)) + + Gs = nx.Graph() + Gs.add_edge(("A", "L"), ("B", "L")) + Gs.add_edge(("C", "L"), ("D", "L")) + simple_score = mdl_score(partition, _net(Gs)) + + assert np.isfinite(multigraph_score) # never nan from p > 1 + assert multigraph_score == simple_score + + def test_parallel_edges_warn(self): + """A MultiGraph with actual parallel edges triggers the + disclosure warning.""" + G = nx.MultiGraph() + G.add_edge(("A", "L"), ("B", "L")) + G.add_edge(("A", "L"), ("B", "L")) + net = _net(G) + partition = {("A", "L"): 0, ("B", "L"): 1} + + with pytest.warns(UserWarning, match="parallel edges"): + mdl_score(partition, net) + + def test_weighted_edges_ignored_matches_unweighted_score(self): + """Edge weight magnitude does not affect the score -- only + presence/absence per node pair is modeled.""" + partition = {("A", "L"): 0, ("B", "L"): 0, ("C", "L"): 1, ("D", "L"): 1} + + Gw = nx.Graph() + Gw.add_edge(("A", "L"), ("B", "L"), weight=100) + Gw.add_edge(("C", "L"), ("D", "L"), weight=0.5) + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + weighted_score = mdl_score(partition, _net(Gw)) + + Gs = nx.Graph() + Gs.add_edge(("A", "L"), ("B", "L")) + Gs.add_edge(("C", "L"), ("D", "L")) + unweighted_score = mdl_score(partition, _net(Gs)) + + assert weighted_score == unweighted_score + + def test_weighted_edges_warn(self): + """Non-unit edge weights trigger the disclosure warning.""" + G = nx.Graph() + G.add_edge(("A", "L"), ("B", "L"), weight=5) + net = _net(G) + partition = {("A", "L"): 0, ("B", "L"): 1} + + with pytest.warns(UserWarning, match="weight"): + mdl_score(partition, net) + + def test_unit_weight_does_not_warn(self): + """An explicit weight=1 is not 'non-unit' and should not warn -- + this metric's notion of 'weighted' is narrower than + network.capabilities()'s ('any weight key present'), and that's + intentional: weight=1 everywhere is indistinguishable from + unweighted under this metric's Bernoulli model.""" + G = nx.Graph() + G.add_edge(("A", "L"), ("B", "L"), weight=1) + net = _net(G) + partition = {("A", "L"): 0, ("B", "L"): 1} + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + mdl_score(partition, net) + assert len(caught) == 0 + + def test_plain_simple_graph_never_warns(self): + """No false-positive warnings for an ordinary unweighted, + non-multi graph -- the common case.""" + G = nx.Graph() + G.add_edge(("A", "L"), ("B", "L")) + G.add_edge(("C", "L"), ("D", "L")) + net = _net(G) + partition = {("A", "L"): 0, ("B", "L"): 0, ("C", "L"): 1, ("D", "L"): 1} + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + mdl_score(partition, net) + assert len(caught) == 0 + + def test_multigraph_container_without_actual_parallel_edges_does_not_warn(self): + """A MultiGraph-typed container with no actual duplicate edges is + the common case (py3plex's default container type) and must not + false-positive the warning.""" + G = nx.MultiGraph() + G.add_edge(("A", "L"), ("B", "L")) + G.add_edge(("C", "L"), ("D", "L")) + net = _net(G) + partition = {("A", "L"): 0, ("B", "L"): 0, ("C", "L"): 1, ("D", "L"): 1} + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + mdl_score(partition, net) + assert len(caught) == 0 + + +class TestSelfLoops: + """Self-loops must not crash and must not be counted -- they don't fit + the n_r*(n_r-1) pair capacity model.""" + + def test_self_loop_ignored(self): + G_no_loop = nx.Graph() + G_no_loop.add_edge(("A", "L"), ("B", "L")) + G_no_loop.add_edge(("C", "L"), ("D", "L")) + partition = {("A", "L"): 0, ("B", "L"): 0, ("C", "L"): 1, ("D", "L"): 1} + score_no_loop = mdl_score(partition, _net(G_no_loop)) + + G_with_loop = nx.Graph(G_no_loop) + G_with_loop.add_edge(("A", "L"), ("A", "L")) + score_with_loop = mdl_score(partition, _net(G_with_loop)) + + assert score_with_loop == score_no_loop + + +class TestDeterminism: + """Repeated calls must be bit-identical -- + this is a pure computation with no randomness.""" + + def test_repeated_calls_identical(self): + G = nx.Graph() + G.add_edge(("A", "L"), ("B", "L")) + G.add_edge(("C", "L"), ("D", "L")) + net = _net(G) + partition = {("A", "L"): 0, ("B", "L"): 0, ("C", "L"): 1, ("D", "L"): 1} + + scores = [mdl_score(partition, net) for _ in range(5)] + assert len(set(scores)) == 1 + + +class TestScoreQuality: + """Sanity check that mdl_score actually + rewards better partitions -- otherwise none of the other correctness + guarantees above matter.""" + + def test_true_structure_scores_lower_than_giant_community(self): + G = nx.Graph() + G.add_edge(("A", "L"), ("B", "L")) + G.add_edge(("C", "L"), ("D", "L")) + net = _net(G) + + true_partition = {("A", "L"): 0, ("B", "L"): 0, ("C", "L"): 1, ("D", "L"): 1} + giant_partition = {("A", "L"): 0, ("B", "L"): 0, ("C", "L"): 0, ("D", "L"): 0} + + assert mdl_score(true_partition, net) < mdl_score(giant_partition, net) + + +class TestNetworkInputVariants: + """Duck-typed / partially-populated network + objects, since mdl_score's `network` parameter is typed `Any`.""" + + def test_network_object_without_core_network_attribute(self): + """A duck-typed network object with no core_network attribute at + all falls back gracefully via getattr's default.""" + class BareNetwork: + pass + + partition = {("A", "L"): 0, ("B", "L"): 1} + score = mdl_score(partition, BareNetwork()) + assert score == pytest.approx(2.0, abs=1e-6) + + def test_network_with_none_core_network(self): + """A real multi_layer_network whose core_network was never + populated behaves the same as the duck-typed case above.""" + net = multinet.multi_layer_network(directed=False) + assert net.core_network is None + partition = {("A", "L"): 0, ("B", "L"): 1} + score = mdl_score(partition, net) + assert score == pytest.approx(2.0, abs=1e-6) + + +class TestPerformance: + """Regression: the data-cost loop is O(n + |E|), not + quadratic in the number of communities, even under extreme + fragmentation (many singleton communities).""" + + def test_no_quadratic_blowup_mdl_fragmented(self): + """mdl_score should stay near-linear even with many singleton + communities, not quadratic in the number of communities per layer. + + _mdl_single_layer's data cost loop iterates over observed block + pairs (`edge_counts`), not over all k^2 community pairs, so cost + tracks O(n + |E|) rather than O(k^2). Extreme fragmentation (every + node its own community, as an AutoCommunity algorithm might produce + for a partial/degenerate partition) maximizes k without blowing up + |E| -- this is exactly the scenario a regression to a k^2 nested + loop over communities would fail on. + """ + n = 5000 + nodes = [(f"node_{i}", "layer_0") for i in range(n)] + + # Sparse, deterministic edge set (a cycle): O(n) edges, and since + # every node is its own community, each edge lands on a distinct + # community pair -- the worst case for the number of block pairs + # relative to node count. + G = nx.Graph() + G.add_nodes_from(nodes) + for i in range(n): + G.add_edge(nodes[i], nodes[(i + 1) % n]) + + net = _net(G) + + # Every node in its own singleton community (maximal fragmentation). + partition = {node: i for i, node in enumerate(nodes)} + + # Partition covers every node on a plain (non-multi, unweighted) + # graph, so none of mdl_score's warning paths (partial partition, + # parallel/weighted edges) should fire here. + start = time.time() + score = mdl_score(partition, net) + elapsed = time.time() - start + + # Should complete in reasonable time despite k == n. + assert elapsed < 2.0 + assert score >= 0.0 + + @pytest.mark.slow + def test_no_quadratic_blowup_mdl_large_scale(self): + """mdl_score should stay near-linear at ~2M nodes, not just at the + smaller scale in test_no_quadratic_blowup_mdl_fragmented. + + Marked slow: builds a 2,000,000-node graph and runs the full + mdl_score computation. Measured ~8s / ~3GB peak RSS on a dev + machine; a regression to a quadratic-in-communities loop would + take minutes rather than tens of seconds at this scale. Skip via + `pytest -m "not slow"` for a fast local/CI run. + """ + n = 2_000_000 + nodes = [(f"node_{i}", "layer_0") for i in range(n)] + + G = nx.Graph() + G.add_nodes_from(nodes) + for i in range(n): + G.add_edge(nodes[i], nodes[(i + 1) % n]) + + net = _net(G) + + # Worst case for _mdl_single_layer's block-pair counting: every + # node is its own singleton community, maximizing k without + # increasing |E|. + partition = {node: i for i, node in enumerate(nodes)} + + start = time.time() + score = mdl_score(partition, net) + elapsed = time.time() - start + + # Generous bound to absorb slower/loaded CI hardware while still + # catching a real regression to quadratic-in-communities behavior. + assert elapsed < 60.0 + assert score >= 0.0 \ No newline at end of file