Core-Periphery Algorithms

Core-periphery detection algorithms.

class asunder.base.algorithms.core_periphery.UnionFind(n)

Bases: object

Union-Find with path compression and union by rank.

Parameters:

n (int) – Number of elements.

find(i)

Find operation identifies which set a particular element belongs to.

Parameters:

i (int) – Any element i.

Returns:

Representative element from the set containing i.

Return type:

int

union(i, j)

Union operation merges the sets that two elements belong to (by rank).

Parameters:

  • i (int) – Any element i.

  • j (int) – Any element j.

Return type:

None

components()

Computes the components in each block and returns said blocks and a component map.

Return type:

tuple[list[list[int]], dict[int, int]]

Returns:

  • blocks (list[list[int]]) – Groups of nodes, each of which is referred to as a block.

  • comp_map (dict[int, int]]) – Maps nodes to the block they belong to.

asunder.base.algorithms.core_periphery.normalized_BE_score(A, labels)

Compute normalized Borgatti-Everett Pearson score.

Parameters:

  • A (np.ndarray of int or float, shape (N, N)) – Graph adjacency/weight matrix.

  • labels (np.ndarray of int, shape (N,)) – Reflects the group that each node n belongs to.

Returns:

BE score.

Return type:

float

asunder.base.algorithms.core_periphery.find_core(A, labels)

Choose the best orientation for binary core-periphery labels.

Parameters:

  • A (np.ndarray of int or float, shape (N, N)) – Graph adjacency/weight matrix.

  • labels (np.ndarray of int, shape (N,)) – Original orientation of binary core-periphery labels.

Returns:

Best orientation of binary core-periphery labels.

Return type:

np.ndarray of int, shape (N,)

asunder.base.algorithms.core_periphery.find_core_advanced(A, labels)

Select the best core-periphery split induced by unique label values. Used to recover best core-periphery structure from a multi-label graph split.

Parameters:

  • A (np.ndarray of int or float, shape (N, N)) – Graph adjacency / weight matrix.

  • labels (np.ndarray of int, shape (N,)) – Original node labels including two or more unique groups.

Return type:

tuple[ndarray, float]

Returns:

  • best_labels (np.ndarray of int, shape (N,)) – Best core-periphery split

  • best_score (float) – Best BE score.

class asunder.base.algorithms.core_periphery.EnhancedGeneticBE(A, must_links=None, pop_size=50, generations=100, init_mut_rate=0.1, elitism_size=2, tournament_size=3, seed=42)

Bases: object

Binary genetic search for Borgatti-Everett objective with must-link blocks. Includes:

  • Adaptive Mutation Rate: mutation probability decreases as the population converges or as fitness increases

  • Elitism: the top‑𝑘 individuals are carried over unchanged into the next generation (Goldberg 1989)

  • Multi‑Point Crossover: instead of a single cut, two‑point crossover swaps two segments between parents

  • Must‑link constraints

Parameters:

  • A (np.ndarray of int or float, shape (N, N)) – Graph adjacency / weight matrix.

  • must_links (list[tuple[int, int]] | None) – Node pairs that must be linked because the edges between them cannot connect nodes in different communities.

  • pop_size (int) – Size of the population.

  • generations (int) – Number of generations.

  • init_mut_rate (float) – Initial mutation rate.

  • elitism_size (int) – Number of individuals carried unchanged into next generation.

  • tournament_size (int) – Number of individuals per tournament.

  • seed (int | None) – Random seed value.

_enforce_blocks(z)

Hard repair: force each must‑link block to use uniform labels.

Parameters:

z (np.ndarray of int, shape (N,)) – 1D graph partition.

Return type:

None

_initialize_population()

Random population initialization with block repair.

Returns:

population – Entire population.

Return type:

list[np.ndarray], shape (N, self.pop_size)

_fitness(z)

Pearson correlation between adjacency A, and the core-periphery pattern matrix, D = (z_i OR z_j).

Parameters:

z (np.ndarray of int, shape (N,)) – Core membership vector which is 1 if node i is in the core and 0 otherwise.

Returns:

Pearson correlation score.

Return type:

float

_tournament(pop, fits)

Random tournament-based selection.

Parameters:

  • pop (list[np.ndarray]) – Entire population.

  • fits (list[float]) – Fitness score for each individual in population.

Returns:

Individual tournament winner.

Return type:

np.ndarray of int, shape (N,)

_multi_point_crossover(p1, p2)

Two‑point crossover (DEAP style): Randomly selects two crossover points and swaps the genetic material between these points in the two parent individuals.

Parameters:

  • p1 (np.ndarray of int, shape (N,)) – Tournament 1 winner (parent 1).

  • p2 (np.ndarray of int, shape (N,)) – Tournament 2 winner (parent 2).

Return type:

tuple[ndarray, ndarray]

Returns:

  • c1 (np.ndarray of int, shape (N,)) – Child 1 from parents.

  • c2 (np.ndarray of int, shape (N,)) – Child 2 from parents.

_adaptive_mutation(z, gen)

Adaptive mutation of input individual z. mutation_rate = init_mut_rate * (1 - gen / total_gens)

Parameters:

  • z (np.ndarray of int, shape (N,)) – Input individual.

  • gen (int) – Current generation.

Returns:

mutated individual.

Return type:

np.ndarray of int, shape (N,)

run()

Run genetic algorithm with elitism.

Return type:

tuple[dict[int, int], float]

Returns:

  • labels (dict[int, int]) – Binary Core-Periphery assignments.

  • best_fit (float) – Best fitness score, corresponding to returned labels.

class asunder.base.algorithms.core_periphery.FullContinuousGeneticBE(A, must_links=None, nonlinear_nodes=None, pop_size=50, generations=100, crossover_rate=0.8, mutation_rate=0.1, tournament_size=3, gene_init_scale=1.0, seed=42)

Bases: object

Continuous genetic search for BE objective with block constraints.

Parameters:

  • A (np.ndarray of int or float, shape (N, N)) – Graph adjacency / weight matrix.

  • must_links (list[tuple[int, int]] | None) – Node pairs that must be linked because the edges between them cannot connect nodes in different communities.

  • nonlinear_nodes (list[int] | None) – Nodes that correspond to nonlinear constraints and so, should be merged.

  • pop_size (int) – Size of the population.

  • generations (int) – Number of generations.

  • crossover_rate (float) – crossover rate below which parents are blended into children.

  • mutation_rate (float) – Fixed mutation rate.

  • tournament_size (int) – Number of individuals per tournament.

  • gene_init_scale (float) – gene initialization scale.

  • seed (int | None) – Random seed value.

_enforce_blocks(c)

For each block, set all c_i to the block‐mean.

Parameters:

c (np.ndarray of int, shape (N,)) – 1D graph partition.

Return type:

None

_init_population()

Initialize population with real vectors in [0,scale], one per individual.

Returns:

population – Entire population.

Return type:

list[np.ndarray]

_fitness(c)

Pearson correlation score between A_ij and D_ij = c_i * c_j for i<j.

Parameters:

c (np.ndarray of int, shape (N,)) – Core membership vector which is 1 if node i is in the core and 0 otherwise.

Returns:

Pearson correlation score.

Return type:

float

_tournament(pop, fits)

Select one parent by tournament.

Parameters:

  • pop (list[np.ndarray]) – Entire population.

  • fits (list[float]) – Fitness score for each individual in population.

Returns:

Individual tournament winner.

Return type:

np.ndarray of int, shape (N,)

_crossover(p1, p2)

Blend crossover: c’ = alpha * p1 + (1-alpha)*p2.

Parameters:

  • p1 (np.ndarray of int, shape (N,)) – Tournament 1 winner (parent 1).

  • p2 (np.ndarray of int, shape (N,)) – Tournament 2 winner (parent 2).

Return type:

tuple[ndarray, ndarray]

Returns:

  • c1 (np.ndarray of int, shape (N,)) – Child 1 from parents.

  • c2 (np.ndarray of int, shape (N,)) – Child 2 from parents.

_mutate(c)

Gaussian mutation around each gene.

Parameters:

c (np.ndarray of int, shape (N,)) – Input individual.

Returns:

mutated individual.

Return type:

np.ndarray of int, shape (N,)

run()

Execute GA and return continuous coreness and best fitness.

Return type:

tuple[dict[int, float], float]

Returns:

  • coreness (dict[int, float]) – Continuous coreness score for each node.

  • best_fit (float) – Best fitness score, corresponding to returned coreness.

_expand_blocks(g)

Expand block-level genome into full individual genome.

Parameters:

g (np.ndarray of float, shape (n_blocks,)) – Block-level genome g (len = n_blocks)

Returns:

c – Full individual genome c (len = N)

Return type:

np.ndarray of float, shape (N,)

_init_population_blocks()

Initialize population in block space.

Returns:

Entire population in block space.

Return type:

list[np.ndarray]

run_de(F=0.7, CR=0.9)

Run Differential Evolution (DE/rand/1/bin) using a block-level genome.

Return type:

tuple[dict[int, float], float]

Returns:

  • coreness (dict[int, float]) – Continuous coreness score for each node.

  • best_fit (float) – Best fitness score, corresponding to returned coreness.

Parameters:

  • F (float)

  • CR (float)

asunder.base.algorithms.core_periphery.upper_triangle(mat)

Return upper-triangle index arrays.

Parameters:

mat (csr_matrix, shape (N, N)) – Input matrix.

Returns:

Row and column indices of elements in the upper triangle.

Return type:

tuple[np.ndarray, np.ndarray]

asunder.base.algorithms.core_periphery.corr_upper(A_vals, D_vals)

Compute Pearson correlation between two equal-length vectors.

Parameters:

  • A_vals (np.ndarray) – Values from the upper triangle of the adjacency / weight matrix.

  • D_vals (np.ndarray) – Values from the upper triangle of the pattern matrix.

Returns:

Pearson correlation score.

Return type:

float

asunder.base.algorithms.core_periphery.detect_continuous_KL(A_csr, must_links, nonlinear_nodes, max_iter=100, seed=42)

Continuous- BE coreness via constrained block updates.

Parameters:

  • A_csr (csr_matrix, shape (N, N)) – Input adjacency / weight matrix as a csr matrix.

  • must_links (list[tuple[int, int]]) – Node pairs that must be linked because the edges between them cannot connect nodes in different communities.

  • nonlinear_nodes (list[int] | None) – Nodes that correspond to nonlinear constraints and so, should be merged.

  • max_iter (int) – Maximum number of iterations to run.

  • seed (int | None) – Random seed value.

Return type:

tuple[ndarray, float]

Returns:

  • c (np.ndarray of float, shape (N,)) – Continuous coreness score for each node.

  • best_r (float) – Pearson correlation achieved.

asunder.base.algorithms.core_periphery.spectral_continuous_cp_detection(A_csr, must_links, nonlinear_nodes, normalize=True)

Block-aggregated continuous coreness using principal eigenvector.

  1. Builds blocks from must_links + nonlinear_nodes.

  2. Aggregates A into a block‐adjacency B of size b×b.

  3. Computes principal eigenvector v of B -> one coreness per block.

  4. Expands v to node‐level c[i] = v[block(i)] and normalizes if requested.

  5. Computes Q = corr( A_flat, (c⌣c)_flat ) as the continuous BE score.

Parameters:

  • A_csr (csr_matrix) – Input adjacency / weight matrix as a csr matrix.

  • must_links (list[tuple[int, int]]) – Node pairs that must be linked because the edges between them cannot connect nodes in different communities.

  • nonlinear_nodes (list[int] | None) – Nodes that correspond to nonlinear constraints and so, should be merged.

  • normalize (bool) – Enable/disable normalization.

Return type:

tuple[ndarray, float]

Returns:

  • c (np.ndarray of float, shape (N,)) – Continuous coreness vector.

  • Q (float) – Pearson‐r fit of A vs. c·cᵀ

asunder.base.algorithms.core_periphery._to_undirected_graph(adj)

Convert adjacency input into an undirected NetworkX Graph.

Supported inputs are NumPy 2D arrays, SciPy sparse matrices, networkx.Graph objects, and networkx.DiGraph objects. Edge weights are preserved if present, and diagonal entries are ignored.

Parameters:

adj (AdjLike) – Adjacency / weight matrix-like array.

Returns:

NetworkX graph.

Return type:

nx.Graph

asunder.base.algorithms.core_periphery._core_mask_from_partition(core_periphery, core_is=0)

Produce a boolean mask of shape (N,) where True indicates a core node.

Parameters:

  • core_periphery (np.ndarray | list[int]) –

    Accepted inputs:

    1D vector length N with values in {0,1,True,False} (1/True => core). 2D partition matrix (N,2) that is one-hot.

  • core_is (int | None) – Index of the ‘core’ column (default 0).

Returns:

Boolean core mask.

Return type:

np.ndarray of bool, shape (N,)

asunder.base.algorithms.core_periphery.partition_periphery_components(adj, core_periphery, *, core_is=0)

Build community labels and assignment matrix after collapsing all core nodes into a single community (id=0) and splitting the periphery into connected components once core nodes are removed.

Parameters:

  • adj (AdjLike) – Graph adjacency / weight matrix. Treated as undirected for component detection.

  • core_periphery (np.ndarray | list[int]) –

    Either:

    • length-N vector, where nonzero/True => core, 0/False => periphery, or

    • (N,2) one-hot partition matrix [core_col, periphery_col].

  • core_is ({0,1} or None, default=0) – If a (N,2) matrix is provided, index of the core column. Ignored for 1D vector input.

Return type:

tuple[ndarray, dict[str, Any]]

Returns:

  • labels (np.ndarray, shape (N,)) – Community id for each node. 0 => all core nodes; 1..K => periphery components.

  • info (dict) –

    Metadata including:

    • ’n_core’, ‘n_periphery’

    • ’components’: list of sets with node indices for each periphery component (in order)

    • ’community_node_indices’: list where entry j contains indices for community j