Python做风险控制|找出形成环状投资的公司

article/2025/8/22 17:22:54

大家好，我是小小明。

今天我将带大家利用python找到关系数据的环。先说下需求和背景：

需求描述

某投资机构需要考虑各公司的投资风险，手上一份各公司投资方向的数据，主要字段是投资者和被投资者。

而有部分公司并不是真的投资，而是买通一大堆作为僚机的公司，最终投资的钱还是会变成投资自己，于是我们就需要找出这样的环状结果，把处于这个环的所有公司都找出来。

例如，ABCDEFG这7个公司的投资关系如下：

那么很明显，ABD和ABCD构成了环，我们最终的目标就是找出这两个环并输出。

原始数据形式

首先，读取测试数据（已脱敏）：

数值编码

为了提升后续程序处理的速度，我们给每个公司创建一个唯一的数值编码：

n = df.shape[0]
codes, labels = pd.factorize(np.r_[df.values[:, 0], df.values[:, 1]])
print(codes[:10])
print(labels[:10])

[0 1 2 3 4 1 5 6 7 8]
['GD1' 'GD2' 'GD3' 'GD4' 'GD5' 'GD6' 'GD7' 'GD8' 'GD9' 'GD10']

此时两列数据已合并到一列进行了统一的编码，codes前n个数据是第一列的编码，后n个数据是第二列的编码。

而labels就是对应的原始标签，我们可以随时通过labels和对应的编码查询到原始的数据。

设计数据结构存储图

下面我们设计一个数据结构来存储图，考虑到引用关系，我想到了直接用链表存储图，通过巧妙的处理，它自动就能形成环状的结构，完美的原样表达我们想表达的结构：

class Node:def __init__(self, data, children=None):self.in_degree = 0self.data = dataif children is None:children = set()self.children = childrendef __repr__(self):return f"{self.data}->{self.out_degree}"@propertydef out_degree(self):return len(self.children)

咱们先一次性把所有可能的节点一次性找出来并存储起来：

nodes_cache = [Node(code) for code in range(len(labels))]

然后我们就可以根据编码后的边数据来构造每个节点的子节点信息：

for s, d in zip(codes[:n], codes[n:]):nodes_cache[s].children.add(nodes_cache[d])

然后处理每个节点的入度信息（被多少个公司投资）：

for parent in nodes_cache:for node in parent.children:node.in_degree += 1

找出所有入度为0的顶点

为了减少重复查找的概率，我们从所有被被投资的公司开始找起，首先找出所有入度为0的节点保存起来：

vertexs = []
for node in nodes_cache:if node.in_degree == 0:vertexs.append(node)

注意：由于都是边数据，不可能存在入度和出度都为0的离群点，所以无需做多余的判断。

回溯算法找出环

可能大部分读者都打算使用图论中的算法来查找，但本人只找到判断是否存在环算法，有些说能够实现找出具体的环却看不到具体的实现代码。

所以本人按照自己的思维去完成这个找环的过程：

def get_ciyle(vertexs, max_lenth=10, max_count=None):"""max_lenth:路径超过指定个数的环都忽略max_count：每个顶点至少要找出多少个环，传None表示全部找出"""def dfs(node):visited.append(node.data)if len(visited) <= max_lenth:for child_node in node.children:if child_node.data in visited:ciyle_data = visited[visited.index(child_node.data):]i = ciyle_data.index(min(ciyle_data))result.add(tuple(ciyle_data[i:]+ciyle_data[:i]))if max_count and len(result) >= max_count:# 超过最大限度则结束returncontinuedfs(child_node)visited.pop()total = set()for node in vertexs:result = set()visited = []dfs(node)total = total.union(result)return totalciyle_codes = get_ciyle(vertexs)