如何从相关矩阵中提取100%依赖的事件组合？

我有一个事件自相关的相关矩阵：

• 当事件A与B的相关系数为1时，A发生则B必然发生；
• 若事件G与A、B的相关系数均为1，则三者必然同时发生。

需要获取所有内部元素两两相关系数为1的事件组合列表（即100%依赖的事件组）。

已通过pandas .corr()生成相关矩阵并转为numpy数组，示例矩阵如下：

import numpy as np

matrix = np.array([
    [1, -0.5, -0.5, 1, 1, -0.5, -0.5, -0.5, -0.5],
    [-0.5, 1, 1, -0.5, -0.5, 1, 1, 1, 1], 
    [-0.5, 1, 1, -0.5, -0.5, 1, 1, 1, 1],
    [1, -0.5, -0.5, 1, 1, -0.5, -0.5, -0.5, -0.5],  
    [1, -0.5, -0.5, 1, 1, -0.5, -0.5, -0.5, -0.5],  
    [-0.5, 1, 1, -0.5, -0.5, 1, 1, 1, 1], 
    [-0.5, 1, 1, -0.5, -0.5, 1, 1, 1, 1], 
    [-0.5, 1, 1, -0.5, -0.5, 1, 1, 1, 1], 
    [-0.5, 1, 1, -0.5, -0.5, 1, 1, 1, 1],
])

示例中，事件索引[1,2,5,6,7,8]两两相关系数为1，[0,3,4]同理，期望输出：

result = [[1,2,5,6,7,8], [0,3,4]]

之前尝试获取所有值为1的坐标并空间对比但未成功，需要更简便的实现方法。

解决方案 #

这个问题本质是图的连通分量查找：把每个事件看作图的节点，当两个事件的相关系数为1时，节点间存在一条边。我们需要找出图中所有的连通分量，每个分量就是一组两两100%依赖的事件。

方法1：使用scipy的连通分量算法 #

这是最简便的实现方式，利用scipy内置的连通分量函数：

import numpy as np
from scipy.sparse import csr_matrix
from scipy.sparse.csgraph import connected_components

# 构建邻接矩阵：仅保留相关系数为1的边（排除自身到自身的连接）
adj_matrix = (matrix == 1).astype(int)
np.fill_diagonal(adj_matrix, 0)  # 去掉自环，不影响连通分量结果

# 查找连通分量
n_components, labels = connected_components(csgraph=csr_matrix(adj_matrix), directed=False, return_labels=True)

# 按标签分组得到结果
result = []
for label in range(n_components):
    group = np.where(labels == label)[0].tolist()
    result.append(group)

print(result)

输出结果：

[[0, 3, 4], [1, 2, 5, 6, 7, 8]]

方法2：手动实现并查集（Union-Find）算法 #

如果不想依赖scipy，可以手动实现并查集处理连通性：

import numpy as np

class UnionFind:
    def __init__(self, size):
        self.parent = list(range(size))
    
    def find(self, x):
        if self.parent[x] != x:
            self.parent[x] = self.find(self.parent[x])
        return self.parent[x]
    
    def union(self, x, y):
        x_root = self.find(x)
        y_root = self.find(y)
        if x_root != y_root:
            self.parent[y_root] = x_root

# 初始化并查集
n_events = matrix.shape[0]
uf = UnionFind(n_events)

# 遍历矩阵，合并相关系数为1的事件对（仅遍历上三角避免重复）
for i in range(n_events):
    for j in range(i+1, n_events):
        if matrix[i][j] == 1:
            uf.union(i, j)

# 分组结果
groups = {}
for idx in range(n_events):
    root = uf.find(idx)
    if root not in groups:
        groups[root] = []
    groups[root].append(idx)

result = list(groups.values())
print(result)

输出结果同样符合预期。

内容的提问来源于stack exchange，提问作者MaxS.