🔑 Mode Complementarity

The rings.complementarity module measures the alignment between node features and graph structure by comparing their induced metric spaces. It pairs graph metrics (diffusion, heat-kernel, resistance, shortest-path) with matrix-norm comparators and a functor that handles disconnected graphs component-wise.

Functor

Core Functor for facilitating Mode Complementarity computation.

class rings.complementarity.functor.ComplementarityFunctor(feature_metric, graph_metric, comparator, n_jobs: int = 1, use_edge_information: bool = False, normalize_diameters: bool = True, edge_attr: str | None = None, **kwargs)

A functor for computing complementarity between graph structure and node features.

This class computes complementarity by comparing the metric spaces derived from graph structure and node features using a specified comparator. It quantifies how well node features align with the graph structure, with lower values indicating stronger alignment.

For disconnected graphs, the functor processes each connected component separately and aggregates the results using a weighted average, where component sizes serve as weights. This approach ensures that larger components have proportionally more influence on the final complementarity score.

Parameters:

• feature_metric (callable) – The metric function used to compute distances between node features.

• graph_metric (callable) – The metric function used to compute distances in the graph structure.

• comparator (class) – A comparator class that implements a __call__ method to compare metric spaces.

• n_jobs (int, default=1) – Number of jobs to run in parallel. If 1, no parallelism is used.

• use_edge_information (bool, default=False) – Whether to use edge attributes in graph metric computation.

• normalize_diameters (bool, default=True) – Whether to normalize the diameters of metric spaces before comparison.

• **kwargs (dict) – Additional arguments passed to the comparator and metric functions.

Examples

>>> import numpy as np
>>> import torch
>>> from torch_geometric.data import Data
>>> from rings.complementarity.comparator import MatrixNormComparator
>>> from rings.complementarity.metrics import standard_feature_metrics, shortest_path_distance
>>>
>>> # Create a simple graph where node features correspond to positions in the graph
>>> # A path graph: 0 -- 1 -- 2 -- 3 with features encoding their positions
>>> edge_index = torch.tensor([[0, 1, 1, 2, 2, 3], [1, 0, 2, 1, 3, 2]], dtype=torch.long)
>>> # Node features encode positions: node 0 is at position 0, node 1 at position 1, etc.
>>> x = torch.tensor([[0], [1], [2], [3]], dtype=torch.float)
>>> data = Data(x=x, edge_index=edge_index)
>>>
>>> # Create functor with simple metrics
>>> functor = ComplementarityFunctor(
...     feature_metric='euclidean',  # Use Euclidean distance for features
...     graph_metric='shortest_path_distance',  # Use shortest path for graph
...     comparator=MatrixNormComparator,  # Compare using matrix norm
...     n_jobs=1,
...     normalize_diameters=True,  # Normalize distances for fair comparison
... )
>>>
>>> # Compute complementarity for the graph
>>> result = functor([data])
>>> print(f"Complementarity score: {result['complementarity'].item():.4f}")
Complementarity score: 0.0000
>>>
>>> # The score is 0, indicating perfect alignment between features and structure
>>> # (node features perfectly correspond to their positions in the path)
>>>
>>> # Now create a graph with misaligned features
>>> x_misaligned = torch.tensor([[3], [1], [2], [0]], dtype=torch.float)  # Swap 0 and 3
>>> data_misaligned = Data(x=x_misaligned, edge_index=edge_index)
>>>
>>> # Compute complementarity for the misaligned graph
>>> result = functor([data_misaligned])
>>> print(f"Complementarity score: {result['complementarity'].item():.4f}")
Complementarity score: 0.5000
>>>
>>> # The score is higher, indicating weaker alignment between features and structure
>>> # (node features no longer match their positions in the path)

__init__(feature_metric, graph_metric, comparator, n_jobs: int = 1, use_edge_information: bool = False, normalize_diameters: bool = True, edge_attr: str | None = None, **kwargs)

Initialize internal Module state, shared by both nn.Module and ScriptModule.

forward(batch, as_dataframe: bool = True) → Dict[str, Any] | DataFrame | List[Dict[str, Any]]

Compute complementarity for a batch of graphs in PyTorch Geometric format.

Parameters:

• batch (list) – A batch of graphs in PyTorch Geometric format.

• as_dataframe (bool, default=True) – If True, return results as a pandas DataFrame, otherwise as a dictionary with tensor values or a list of dictionaries.

Returns:

Results of complementarity computation for each graph in the batch. If as_dataframe=True, returns a pandas DataFrame. Otherwise, returns either a dictionary with tensor values or a list of dictionaries.

Return type:

Union[Dict[str, Any], pd.DataFrame, List[Dict[str, Any]]]

Examples

>>> from rings.complementarity.comparator import MatrixNormComparator
>>> import torch_geometric.datasets as datasets
>>> dataset = datasets.TUDataset(root='/tmp/ENZYMES', name='ENZYMES')
>>> functor = ComplementarityFunctor(
...     feature_metric=lambda x, y: np.linalg.norm(x - y),
...     graph_metric=lambda x, y: abs(x - y),
...     comparator=MatrixNormComparator,
...     n_jobs=4,
...     normalize_diameters=True
... )
>>> # Get first 5 graphs as a batch
>>> batch = [dataset[i] for i in range(5)]
>>> results = functor.forward(batch)
>>> print(results)

property invalid_data

Return a dictionary with NaN score and other standardized fields.

Returns:

Dictionary with keys: - score: NaN (invalid data) - pvalue: None - pvalue_adjusted: None - significant: None - method: The norm used

Return type:

dict

Comparators

This module contains classes that compare metric spaces (i.e. pairwise distance matrices).

class rings.complementarity.comparator.MatrixNormComparator(**kwargs)

A class to compare matrices based on specified norms and calculate complementarity.

This class computes complementarity between two distance matrices by calculating their difference using a matrix norm. It provides a quantitative measure of how similar two metric spaces are.

Parameters:

**kwargs (dict) –

Keyword arguments to specify the norm to be used. Supported norms are:

• normstr, default=”L11”

  The matrix norm to use. Options include: - “L11”: The L1,1 norm (sum of absolute values) - “frobenius”: The Frobenius norm (square root of sum of squared values)

• Additional parameters are ignored.

norm

The norm to be used for comparison.

Type:

str

__call__(x, y, \*\*kwargs)

Compare two matrices using the specified norm.

L11_norm(M)

Calculate the L1,1 norm of a matrix.

frobenius_norm(M)

Calculate the Frobenius norm of a matrix.

invalid_data()

Return a dictionary with NaN score and other standardized fields.

Examples

>>> import numpy as np
>>> from rings.complementarity.comparator import MatrixNormComparator
>>>
>>> # Create two distance matrices
>>> D1 = np.array([[0, 1, 2], [1, 0, 1], [2, 1, 0]])
>>> D2 = np.array([[0, 2, 4], [2, 0, 2], [4, 2, 0]])  # D1 scaled by 2
>>>
>>> # Compare using L11 norm (default)
>>> comparator1 = MatrixNormComparator()
>>> result1 = comparator1(D1, D2)
>>> print(f"L11 complementarity: {result1['score']}")
L11 complementarity: 0.6666666666666666
>>>
>>> # Compare using Frobenius norm
>>> comparator2 = MatrixNormComparator(norm="frobenius")
>>> result2 = comparator2(D1, D2)
>>> print(f"Frobenius complementarity: {result2['score']}")
Frobenius complementarity: 0.5773502691896257

__init__(**kwargs)

Initialize the MatrixNormComparator with the specified norm. :param **kwargs: Keyword arguments to specify the norm to be used. Supported norms are:

• “L11” (default)

• “frobenius”

static L11_norm(M)

Calculate the L1,1 norm of a matrix. :param M: The matrix for which to calculate the L1,1 norm. :type M: numpy.ndarray

Returns:

The L1,1 norm of the matrix.

Return type:

float

static frobenius_norm(M)

Calculate the Frobenius norm of a matrix. :param M: The matrix for which to calculate the Frobenius norm. :type M: numpy.ndarray

Returns:

The Frobenius norm of the matrix.

Return type:

float

property invalid_data

Return a dictionary with NaN score and other standardized fields.

Returns:

Dictionary with keys: - score: NaN (invalid data) - pvalue: None - pvalue_adjusted: None - significant: None - method: The norm used

Return type:

dict

rings.complementarity.comparator.L11MatrixNormComparator(**kwargs)

Factory function that returns a MatrixNormComparator with L11 norm.

This is a convenience function that creates a MatrixNormComparator pre-configured to use the L1,1 norm (sum of absolute values).

Parameters:

**kwargs (dict) – Additional keyword arguments to pass to the MatrixNormComparator constructor.

Returns:

A comparator configured to use the L11 norm.

Return type:

MatrixNormComparator

Examples

>>> import numpy as np
>>> from rings.complementarity.comparator import L11MatrixNormComparator
>>>
>>> # Create two distance matrices
>>> D1 = np.array([[0, 1], [1, 0]])
>>> D2 = np.array([[0, 2], [2, 0]])
>>>
>>> # Create comparator and compare matrices
>>> comparator = L11MatrixNormComparator()
>>> result = comparator(D1, D2)
>>> print(f"Complementarity score: {result['score']}")
Complementarity score: 0.5

rings.complementarity.comparator.FrobeniusMatrixNormComparator(**kwargs)

Factory function that returns a MatrixNormComparator with Frobenius norm.

This is a convenience function that creates a MatrixNormComparator pre-configured to use the Frobenius norm (square root of sum of squared values).

Parameters:

**kwargs (dict) – Additional keyword arguments to pass to the MatrixNormComparator constructor.

Returns:

A comparator configured to use the Frobenius norm.

Return type:

MatrixNormComparator

Examples

>>> import numpy as np
>>> from rings.complementarity.comparator import FrobeniusMatrixNormComparator
>>>
>>> # Create two distance matrices
>>> D1 = np.array([[0, 1], [1, 0]])
>>> D2 = np.array([[0, 2], [2, 0]])
>>>
>>> # Create comparator and compare matrices
>>> comparator = FrobeniusMatrixNormComparator()
>>> result = comparator(D1, D2)
>>> print(f"Complementarity score: {result['score']}")
Complementarity score: 0.5

Metrics

Methods for lifting attributes and graphs to metric spaces.

rings.complementarity.metrics.lift_attributes(X, metric, n_jobs, **kwargs)

Lift node attributes to a metric space.

This function transforms node attributes into a pairwise distance matrix, effectively “lifting” them into a metric space. It supports standard metrics from scikit-learn as well as custom metrics defined in this module.

Parameters:

• X (array-like of shape (n_samples, n_features)) – Node feature matrix where each row represents a node’s features.

• metric (str) – Name of the metric to use. Can be any metric supported by scikit-learn’s pairwise_distances or a custom metric defined in this module.

• n_jobs (int) – Number of jobs to run in parallel for distance computation. -1 means using all processors.

• **kwargs (dict) – Additional keyword arguments to pass to the metric function.

Returns:

Distance matrix representing pairwise distances between node features.

Return type:

ndarray of shape (n_samples, n_samples)

Raises:

RuntimeError – If the specified metric is not supported.

Examples

>>> import numpy as np
>>> from rings.complementarity.metrics import lift_attributes
>>>
>>> # Node feature matrix: 3 nodes with 2 features each
>>> X = np.array([[1, 2], [3, 4], [5, 6]])
>>>
>>> # Using scikit-learn's euclidean metric
>>> D_euclidean = lift_attributes(X, metric="euclidean", n_jobs=1)
>>> print(D_euclidean)
[[0.         2.82842712 5.65685425]
 [2.82842712 0.         2.82842712]
 [5.65685425 2.82842712 0.        ]]
>>>
>>> # Using custom metric
>>> D_custom = lift_attributes(X, metric="standard_feature_metrics", n_jobs=1, metric="manhattan")
>>> print(D_custom)
[[0. 4. 8.]
 [4. 0. 4.]
 [8. 4. 0.]]

rings.complementarity.metrics.lift_graph(G, metric, **kwargs)

Lift graph structure to a metric space.

This function transforms a graph structure into a pairwise distance matrix using a specified graph metric, effectively “lifting” the graph into a metric space.

Parameters:

• G (networkx.Graph) – The input graph whose structure will be lifted to a metric space.

• metric (str) – Name of the graph metric to use (e.g., “shortest_path_distance”, “diffusion_distance”, “resistance_distance”).

• **kwargs (dict) – Additional keyword arguments to pass to the metric function.

Returns:

Distance matrix representing pairwise distances between nodes according to the specified graph metric.

Return type:

ndarray of shape (n_nodes, n_nodes)

Raises:

RuntimeError – If the specified metric is not supported.

Examples

>>> import networkx as nx
>>> from rings.complementarity.metrics import lift_graph
>>>
>>> # Create a simple graph: path graph with 3 nodes (0-1-2)
>>> G = nx.path_graph(3)
>>>
>>> # Lift using shortest path distance
>>> D_sp = lift_graph(G, metric="shortest_path_distance")
>>> print(D_sp)
[[0. 1. 2.]
 [1. 0. 1.]
 [2. 1. 0.]]
>>>
>>> # Lift using diffusion distance
>>> D_diff = lift_graph(G, metric="diffusion_distance", num_steps=1)
>>> # Result will be a diffusion-based distance matrix

rings.complementarity.metrics.standard_feature_metrics(X, **kwargs)

Calculate pairwise distances between node features.

This function calculates a distance matrix between node features using scikit-learn’s pairwise_distances function. All metrics supported by scikit-learn can be used.

Parameters:

• X (array_like of shape (n_samples, n_features)) – Node feature matrix, with rows indicating nodes and columns corresponding to feature dimensions.

• **kwargs (dict) –

  Additional keyword arguments including:

  • metricstr, default=”euclidean”

    The metric to use for distance calculation. Options include: “euclidean”, “manhattan”, “cosine”, “jaccard”, etc.

  • n_jobsint, default=None

    Number of jobs to run in parallel. None means 1 unless in a joblib.parallel_backend context.

Returns:

Matrix of pairwise distances between node features.

Return type:

ndarray of shape (n_samples, n_samples)

Examples

>>> import numpy as np
>>> from rings.complementarity.metrics import standard_feature_metrics
>>>
>>> # Node features for 3 nodes
>>> X = np.array([[1, 0], [0, 1], [1, 1]])
>>>
>>> # Calculate Euclidean distances
>>> D_eucl = standard_feature_metrics(X, metric="euclidean")
>>> print(D_eucl)
[[0.         1.41421356 1.        ]
 [1.41421356 0.         1.        ]
 [1.         1.         0.        ]]
>>>
>>> # Calculate Manhattan distances
>>> D_manh = standard_feature_metrics(X, metric="manhattan")
>>> print(D_manh)
[[0. 2. 1.]
 [2. 0. 1.]
 [1. 1. 0.]]

rings.complementarity.metrics.diffusion_distance(G, **kwargs)

Calculate diffusion distance between vertices of a graph.

The diffusion distance measures how similar two nodes are in terms of their connectivity patterns in the graph. It is based on a diffusion process where heat (or probability) spreads throughout the graph over time.

Parameters:

• G (networkx.Graph) – Input graph. All attributes of the graph will be ignored in the subsequent calculations.

• num_steps (int, default=1) – Number of steps for the diffusion operator that is used for the distance calculations.

• norm (bool, default=True) – Whether to normalize the Laplacian.

• symmetric (bool, default=True) – Whether to normalize the Laplacian symmetrically.

• n_jobs (int, optional) – Number of jobs for parallel distance computation.

Returns:

Matrix of pairwise diffusion distances between nodes.

Return type:

ndarray of shape (n_nodes, n_nodes)

Notes

This implementation computes diffusion distances by: 1. Computing the graph Laplacian matrix (normalized or unnormalized) 2. Computing the matrix Ψ = L^t where L is the Laplacian and t is num_steps 3. Computing pairwise Euclidean distances between rows of Ψ

If the graph has isolated nodes, the function returns NaN.

Examples

>>> import networkx as nx
>>> import numpy as np
>>> from rings.complementarity.metrics import diffusion_distance
>>>
>>> # Create a simple path graph: 0-1-2-3
>>> G = nx.path_graph(4)
>>>
>>> # Calculate diffusion distances
>>> D = diffusion_distance(G, num_steps=1)
>>>
>>> # Verify properties of the distance matrix
>>> np.testing.assert_almost_equal(np.diag(D), np.zeros(4))  # Zero diagonal
>>> np.testing.assert_almost_equal(D, D.T)  # Symmetric matrix

rings.complementarity.metrics.heat_kernel_distance(G, **kwargs)

Calculate heat kernel distance between vertices of a graph.

The heat kernel distance is a measure of node similarity based on the heat diffusion process on the graph. It captures how heat flows through the graph structure over time, providing a rich metric for comparing nodes.

Parameters:

• G (networkx.Graph) – Input graph. All attributes of the graph will be ignored in the subsequent calculations.

• num_steps (int, default=1) – Number of time steps for the diffusion process. Controls how far the heat diffuses through the graph.

• norm (bool, default=True) – Whether to normalize the Laplacian.

• symmetric (bool, default=True) – Whether to normalize the Laplacian symmetrically.

• n_jobs (int, optional) – Number of jobs for parallel distance computation.

Returns:

Matrix of pairwise heat kernel distances between nodes.

Return type:

ndarray of shape (n_nodes, n_nodes)

Notes

This implementation computes heat kernel distances by: 1. Computing the graph Laplacian matrix 2. Computing eigenvalues and eigenvectors of the Laplacian 3. Scaling eigenvectors by e^(-t * eigenvalues) 4. Computing pairwise Euclidean distances between rows of the scaled eigenvectors

If complex numbers are encountered, the real part is used and a warning is issued.

Examples

>>> import networkx as nx
>>> import numpy as np
>>> from rings.complementarity.metrics import heat_kernel_distance
>>>
>>> # Create a simple grid graph
>>> G = nx.grid_2d_graph(2, 2)  # 2x2 grid
>>> G = nx.convert_node_labels_to_integers(G)  # Relabel as integers
>>>
>>> # Calculate heat kernel distances
>>> D = heat_kernel_distance(G, num_steps=0.1)
>>>
>>> # Verify properties
>>> assert D.shape == (4, 4)  # 4 nodes in 2x2 grid
>>> np.testing.assert_almost_equal(np.diag(D), np.zeros(4))  # Zero diagonal
>>> np.testing.assert_almost_equal(D, D.T)  # Symmetric matrix

rings.complementarity.metrics.resistance_distance(G, **kwargs)

Calculate resistance distance between vertices of a graph.

The resistance distance between two nodes is the effective resistance when the graph is viewed as an electrical network with edges as resistors. It is also equal to the commute time between nodes in a random walk (up to a scaling factor).

Parameters:

• G (networkx.Graph) – Input graph. Graph attributes are ignored in the calculations.

• weight (str or None, optional) – The edge attribute that holds the numerical value used as a weight. If set to None, the graph is treated as unweighted. For compatibility with other RINGS metrics, this should be consistent with other metrics.

• **kwargs (dict) – Additional keyword arguments. Only required for compatibility with the metric API.

Returns:

Matrix of pairwise resistance distances between nodes.

Return type:

ndarray of shape (n_nodes, n_nodes)

Notes

This function uses NetworkX’s built-in resistance_distance function. If the graph is not connected, the function will raise a NetworkXError, which this wrapper catches and returns NaN.

For unweighted graphs, the resistance distance is directly related to the commute time and the pseudo-inverse of the graph Laplacian.

Examples

>>> import networkx as nx
>>> import numpy as np
>>> from rings.complementarity.metrics import resistance_distance
>>>
>>> # Create a simple path graph: 0-1-2-3
>>> G = nx.path_graph(4)
>>>
>>> # Calculate resistance distances
>>> D = resistance_distance(G)
>>>
>>> # For a path graph, resistance distance equals shortest path distance
>>> sp_distance = nx.floyd_warshall_numpy(G)
>>> np.testing.assert_almost_equal(D, sp_distance)
>>>
>>> # Verify that diagonal elements are zero
>>> np.testing.assert_almost_equal(np.diag(D), np.zeros(4))

rings.complementarity.metrics.shortest_path_distance(G, **kwargs)

Calculate shortest-path distance between all pairs of vertices in a graph.

This function computes the shortest path lengths between all pairs of nodes in the graph using the Floyd-Warshall algorithm. The result is a distance matrix where each element [i,j] represents the shortest path distance from node i to node j.

Parameters:

• G (networkx.Graph) – Input graph.

• weight (str or None, optional) – The edge attribute that holds the numerical value used as a weight. If set to None (default), the graph is treated as unweighted and each edge has weight 1.

• **kwargs (dict) – Additional keyword arguments for compatibility with the metric API. These are not used by this function but allow for consistent interfaces.

Returns:

Matrix of shortest path distances between all pairs of nodes.

Return type:

ndarray of shape (n_nodes, n_nodes)

Notes

The time complexity of the Floyd-Warshall algorithm is O(n³), where n is the number of nodes in the graph. For very large graphs, consider using other algorithms such as Dijkstra’s algorithm on-demand for specific node pairs.

Examples

>>> import networkx as nx
>>> import numpy as np
>>> from rings.complementarity.metrics import shortest_path_distance
>>>
>>> # Create an unweighted graph
>>> G1 = nx.Graph()
>>> G1.add_edges_from([(0, 1), (1, 2), (2, 3), (3, 4), (0, 4)])  # Pentagon
>>> D1 = shortest_path_distance(G1)
>>> print(D1)  # Each entry [i,j] is the shortest path length from i to j
[[0. 1. 2. 2. 1.]
 [1. 0. 1. 2. 2.]
 [2. 1. 0. 1. 2.]
 [2. 2. 1. 0. 1.]
 [1. 2. 2. 1. 0.]]
>>>
>>> # Create a weighted graph
>>> G2 = nx.Graph()
>>> G2.add_weighted_edges_from([(0, 1, 1), (1, 2, 3), (0, 2, 5)])  # Triangle
>>> D2 = shortest_path_distance(G2, weight='weight')
>>> # Path 0->2 should use the direct edge, not 0->1->2, due to weights
>>> assert D2[0, 2] == 4  # Shortest path is 0->1->2 with weight 1+3=4

Utilities

Utility Functions for Mode Complementarity

rings.complementarity.utils.maybe_normalize_diameter(D)

Normalize a distance matrix by its diameter if possible.

If the distance matrix has a non-zero maximum value (diameter), this function normalizes all distances by dividing by that maximum value. This produces a distance matrix with values in [0,1]. If the matrix describes a trivial metric space (e.g., a single point or multiple identical points), it is returned unchanged.

Parameters:

D (array_like) – Square distance matrix, assumed to describe pairwise distances of a finite metric space.

Returns:

The distance matrix, normalized by its diameter (maximum value).

Return type:

array_like

Examples

>>> import numpy as np
>>> from rings.complementarity.utils import maybe_normalize_diameter
>>>
>>> # Example 1: Non-trivial distance matrix
>>> D1 = np.array([[0, 2, 4], [2, 0, 6], [4, 6, 0]])
>>> D1_norm = maybe_normalize_diameter(D1)
>>> print(D1_norm)
[[0.         0.33333333 0.66666667]
 [0.33333333 0.         1.        ]
 [0.66666667 1.         0.        ]]
>>>
>>> # Example 2: Zero matrix (single point space)
>>> D2 = np.zeros((3, 3))
>>> D2_norm = maybe_normalize_diameter(D2)
>>> print(D2_norm)  # Returns unchanged
[[0. 0. 0.]
 [0. 0. 0.]
 [0. 0. 0.]]