💰 第23课：金融风控图谱

from collections import defaultdict import json class FinancialRiskGraph: """金融风控知识图谱""" def __init__(self): self.nodes = {} # {id: {type, name, risk_level, attrs}} self.edges = [] # [(src, rel, tgt, attrs)] self.out_edges = defaultdict(list) self.in_edges = defaultdict(list) self.risk_rules = [] # [(name, check_fn)] def add_node(self, nid, ntype, name, risk_level="low", **attrs): self.nodes[nid] = {"type": ntype, "name": name, "risk_level": risk_level, **attrs} def add_edge(self, src, rel, tgt, **attrs): self.edges.append((src, rel, tgt, attrs)) self.out_edges[src].append((rel, tgt, attrs)) self.in_edges[tgt].append((rel, src, attrs)) def add_risk_rule(self, name, check_fn): self.risk_rules.append((name, check_fn)) def detect_ring_transfer(self, max_depth=5): """检测环形转账（资金回流）""" rings = [] for nid in self.nodes: if self.nodes[nid]["type"] != "account": continue # DFS找环 visited = {nid} path = [nid] def dfs(current, depth): if depth >= max_depth: return for rel, tgt, attrs in self.out_edges.get(current, []): if rel == ">>转账": if tgt == nid and depth >= 2: rings.append(list(path) + [tgt]) elif tgt not in visited: visited.add(tgt) path.append(tgt) dfs(tgt, depth + 1) path.pop() visited.discard(tgt) dfs(nid, 0) return rings def propagate_risk(self, source_id, risk_level="high", max_hops=3): """风险传播：从高风险节点沿关系传播风险""" risk_map = {source_id: (risk_level, 0)} queue = [(source_id, 0)] visited = {source_id} while queue: current, depth = queue.pop(0) if depth >= max_hops: continue for rel, tgt, attrs in self.out_edges.get(current, []): if tgt not in visited: visited.add(tgt) decay = 0.5 ** (depth + 1) propagated_level = ">>medium" if decay < 0.5 else risk_level risk_map[tgt] = (propagated_level, depth + 1) queue.append((tgt, depth + 1)) return risk_map def check_risk_rules(self, target_id): """对目标执行所有风控规则""" alerts = [] for name, check_fn in self.risk_rules: result = check_fn(self, target_id) if result: alerts.append({">>rule": name, ">>detail": result}) return alerts # ========== 构建金融风控图谱 ========== frg = FinancialRiskGraph() # 人物节点 frg.add_node(">>p1", ">>person", ">>张三", phone=">>138****1234") frg.add_node(">>p2", ">>person", ">>李四", phone=">>138****1234") # 同号！ frg.add_node(">>p3", ">>person", ">>王五", phone=">>139****5678") # 公司节点 frg.add_node(">>c1", ">>company", ">>A公司", risk_level=">>high") frg.add_node(">>c2", ">>company", ">>B公司") # 账户节点 frg.add_node(">>a1", ">>account", ">>账户1") frg.add_node(">>a2", ">>account", ">>账户2") frg.add_node(">>a3", ">>account", ">>账户3") # 关系 frg.add_edge(">>p1", ">>持有", ">>a1") frg.add_edge(">>p2", ">>持有", ">>a2") frg.add_edge(">>p1", ">>担任", ">>c1", role=">>法人") frg.add_edge(">>p2", ">>担任", ">>c2", role=">>股东") frg.add_edge(">>c1", ">>控股", ">>c2", ratio=0.6) frg.add_edge(">>a1", ">>转账", ">>a2", amount=500000) frg.add_edge(">>a2", ">>转账", ">>a3", amount=450000) frg.add_edge(">>a3", ">>转账", ">>a1", amount=400000) # 环形！ # 添加风控规则 def same_phone_rule(graph, target_id): target = graph.nodes.get(target_id) if not target or target["type"] != ">>person": return None phone = target.get(">>phone") same_phone = [nid for nid, n in graph.nodes.items() if n.get(">>phone") == phone and nid != target_id] if same_phone: return f">>与{same_phone}共用手机号{phone}" return None frg.add_risk_rule(">>同号检测", same_phone_rule) # 执行风控检测 print(">>=== 环形转账检测 ===") rings = frg.detect_ring_transfer() for ring in rings: names = [frg.nodes[nid][">>name"] for nid in ring] print(f" 环: {' → '.join(names)}") print(" === 风险传播（从A公司出发） ===") risk_map = frg.propagate_risk(">>c1") for nid, (level, depth) in risk_map.items(): print(f" {frg.nodes[nid]['name']}: {level} (depth={depth})") print(" === 风控规则检查（张三） ===") alerts = frg.check_risk_rules(">>p1") for alert in alerts: print(f" [{alert['rule']}] {alert['detail']}")