Louvain method in Python
Greedy modularity optimisation with hierarchical aggregation of communities.
- File
louvain_method.py, 91 lines- Level
- I0 not checked yet
- Toolchain
- Python 3.14, pytest 9
- Test framework
- pytest
- Run the tests
python3 -m pytest
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/louvain-method/. The authors are listed in prov/GC71E9.jsonl.
from fractions import Fraction
def _one_level(
k: int, adj: list[dict[int, Fraction]], loops: list[Fraction], total: Fraction
) -> tuple[list[int], bool]:
"""Local moving on a weighted graph with k vertices; returns (community of each vertex, moved)."""
degree = [loops[v] * 2 + sum(adj[v].values()) for v in range(k)]
comm = list(range(k)) # @line L3
tot = degree[:]
moved_any = False
improved = True
while improved: # @line L4
improved = False # @line L5
for v in range(k): # @line L6
old = comm[v] # @line L7
tot[old] -= degree[v]
links: dict[int, Fraction] = {}
for w, x in adj[v].items():
links[comm[w]] = links.get(comm[w], Fraction(0)) + x
best = old # @line L8
best_gain = links.get(old, Fraction(0)) - tot[old] * degree[v] / (2 * total)
for c in sorted(links):
gain = links[c] - tot[c] * degree[v] / (2 * total)
if gain > best_gain:
best, best_gain = c, gain
comm[v] = best # @line L9
tot[best] += degree[v]
if best != old:
improved = True
moved_any = True
return comm, moved_any # @line L10
def louvain_method(
n: int, edges: list[tuple[int, int, int]]
) -> tuple[list[list[int]], list[list[list[int]]]]:
"""Return (communities, levels) for an undirected weighted graph on 0 .. n-1.
edges holds triples (u, v, w) with integer weight w > 0; a pair given twice
adds its weights, and u == v is a self-loop. Each level runs local moving
(vertices in index order, a vertex joins the neighbouring community of
largest positive modularity gain, ties to the smallest community id, and
stays when its own community is among the best), then contracts each
community to one vertex. levels lists the partition of the original
vertices after each level that changed something; communities is the last.
Arithmetic is exact. Time O(n + m) per sweep of local moving.
"""
adj: list[dict[int, Fraction]] = [{} for _ in range(n)]
loops = [Fraction(0)] * n
total = Fraction(0)
for u, v, w in edges:
total += w
if u == v:
loops[u] += w
else:
adj[u][v] = adj[u].get(v, Fraction(0)) + w
adj[v][u] = adj[v].get(u, Fraction(0)) + w
parts = [[v] for v in range(n)] # @line L1
levels: list[list[list[int]]] = []
if total == 0:
return parts, levels
while True: # @line L2
k = len(parts)
comm, moved = _one_level(k, adj, loops, total)
if not moved: # @line L11
return sorted(sorted(p) for p in parts), levels
index: dict[int, int] = {} # @line L12
for v in range(k):
if comm[v] not in index:
index[comm[v]] = len(index)
new_parts: list[list[int]] = [[] for _ in index]
for v in range(k):
new_parts[index[comm[v]]] += parts[v]
new_adj: list[dict[int, Fraction]] = [{} for _ in index] # @line L13
new_loops = [Fraction(0)] * len(index)
for v in range(k):
a = index[comm[v]]
new_loops[a] += loops[v]
for w, x in adj[v].items():
b = index[comm[w]]
if a == b:
if v < w:
new_loops[a] += x
else:
new_adj[a][b] = new_adj[a].get(b, Fraction(0)) + x
parts, adj, loops = new_parts, new_adj, new_loops
levels.append(sorted(sorted(p) for p in parts))
test_louvain_method.py 145 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/louvain-method/. The authors are listed in prov/GC71E9.jsonl.
import random
from fractions import Fraction
from itertools import combinations
from louvain_method import louvain_method
def modularity(n, edges, groups):
"""Weighted modularity from the definition, over ordered pairs of vertices."""
a = [[Fraction(0)] * n for _ in range(n)]
for u, v, w in edges:
if u == v:
a[u][u] += 2 * w
else:
a[u][v] += w
a[v][u] += w
k = [sum(row) for row in a]
two_w = sum(k)
label = {}
for i, g in enumerate(groups):
for v in g:
label[v] = i
q = Fraction(0)
for i in range(n):
for j in range(n):
if label[i] == label[j]:
q += a[i][j] - k[i] * k[j] / two_w
return q / two_w
def reference(n, edges):
"""Louvain with blocks of original vertices, each move scored by a full modularity computation."""
if sum(w for _, _, w in edges) == 0:
return [[v] for v in range(n)], []
blocks = [[v] for v in range(n)]
linked = set()
for u, v, _ in edges:
linked.add((u, v))
linked.add((v, u))
levels = []
while True:
k = len(blocks)
comm = list(range(k))
moved = False
improved = True
while improved:
improved = False
for b in range(k):
old = comm[b]
near = {
comm[c]
for c in range(k)
if c != b and any((x, y) in linked for x in blocks[b] for y in blocks[c])
}
def score(target):
trial = comm[:]
trial[b] = target
groups = {}
for c in range(k):
groups.setdefault(trial[c], []).extend(blocks[c])
return modularity(n, edges, list(groups.values()))
best, best_q = old, score(old)
for c in sorted(near):
q = score(c)
if q > best_q:
best, best_q = c, q
comm[b] = best
if best != old:
improved = moved = True
if not moved:
return sorted(sorted(p) for p in blocks), levels
order = []
for b in range(k):
if comm[b] not in order:
order.append(comm[b])
blocks = [sum((blocks[b] for b in range(k) if comm[b] == c), []) for c in order]
levels.append(sorted(sorted(p) for p in blocks))
def all_partitions(items):
if not items:
yield []
return
first, rest = items[0], items[1:]
for p in all_partitions(rest):
for i in range(len(p)):
yield p[:i] + [[first] + p[i]] + p[i + 1 :]
yield [[first]] + p
def random_weighted(rng, n, p):
edges = [(u, v, rng.choice([1, 1, 1, 2, 3])) for u, v in combinations(range(n), 2) if rng.random() < p]
if n and rng.random() < 0.2:
v = rng.randrange(n)
edges.append((v, v, rng.choice([1, 2])))
return edges
def test_example():
edges = [(u, v, 1) for u, v in [(0, 1), (0, 2), (1, 2), (1, 3), (2, 3), (3, 4), (4, 5), (4, 6), (5, 6)]]
comms, levels = louvain_method(7, edges)
assert comms == [[0, 1, 2, 3], [4, 5, 6]]
assert levels == [[[0, 1, 2, 3], [4, 5, 6]]]
assert modularity(7, edges, comms) == Fraction(59, 162)
def test_edge_cases():
assert louvain_method(0, []) == ([], [])
assert louvain_method(3, []) == ([[0], [1], [2]], [])
assert louvain_method(2, [(0, 1, 1)])[0] == [[0, 1]]
assert louvain_method(4, [(0, 1, 1), (2, 3, 1)])[0] == [[0, 1], [2, 3]]
def test_random_against_reference():
rng = random.Random(20081001)
for _ in range(500):
n = rng.randint(0, 8)
edges = random_weighted(rng, n, rng.choice([0.2, 0.4, 0.6]))
assert louvain_method(n, edges) == reference(n, edges)
def test_never_above_optimum():
rng = random.Random(10008)
for _ in range(150):
n = rng.randint(1, 7)
edges = random_weighted(rng, n, 0.45)
if not edges:
continue
comms, levels = louvain_method(n, edges)
q = modularity(n, edges, comms)
best = max(modularity(n, edges, p) for p in all_partitions(list(range(n))))
assert q <= best
qs = [modularity(n, edges, [[v] for v in range(n)])] + [modularity(n, edges, p) for p in levels]
assert all(x < y for x, y in zip(qs, qs[1:]))
if __name__ == "__main__":
test_example()
test_edge_cases()
test_random_against_reference()
test_never_above_optimum()
print("ok")
What the next level needs
- the tests did not run in this build