变量关系太复杂?用 Python 绘制正负相关网络图(含代码)

引言

在科研数据分析中，我们经常会遇到很多变量同时存在的情况。例如岩土工程监测中可能包含位移、沉降、应变、孔压、降雨、荷载、损伤指标、安全系数和失效概率等变量；材料试验中也可能同时记录强度、能量、裂纹、刚度退化和残余响应等指标。面对这类多变量数据，最常见的做法是计算相关系数矩阵，然后绘制相关性热力图。相关性热力图虽然直接，但当变量数量增多后，矩阵图会变得拥挤。读者需要在大量色块中寻找关系，很难快速判断哪些变量处在关系中心，哪些变量之间存在强正相关，哪些变量之间存在强负相关，也不容易看出变量是否可以划分为几个功能模块。为了解决这个问题，本文用 Python 生成一组科研变量数据，并基于相关系数构建变量关系网络。每个节点代表一个变量，节点颜色表示变量所属类别，节点大小表示该变量在网络中的连接强度；连线表示变量之间的强相关关系，其中橙色实线表示正相关，蓝色虚线表示负相关。通过这种方式，相关性矩阵可以转化为更直观的网络结构图。

import osimport numpy as npimport matplotlib.pyplot as pltimport networkx as nxfrom matplotlib.patches import Ellipsefrom matplotlib.lines import Line2Dnp.random.seed(2026)plt.rcParams["font.family"] = "Times New Roman"plt.rcParams["axes.unicode_minus"] = Falseplt.rcParams["figure.dpi"] = 160plt.rcParams["savefig.dpi"] = 600out_dir = "python_network_positive_negative_fixed"os.makedirs(out_dir, exist_ok=True)modules = {    "Deformation": [        "Mean disp.",        "Max disp.",        "Disp. rate",        "Settlement",        "Strain peak"    ],    "Environment": [        "Rainfall",        "Pore pressure",        "Water content",        "Temperature",        "Load level"    ],    "Damage": [        "Damage index",        "Crack density",        "Stiffness loss",        "Plastic zone",        "Acoustic energy"    ],    "Safety": [        "Safety factor",        "Stability index",        "Strength reserve",        "Risk index",        "Failure prob."    ]}feature_names = []feature_module = {}for module_name, vars_ in modules.items():    for var in vars_:        feature_names.append(var)        feature_module[var] = module_namen_features = len(feature_names)n_samples = 700module_colors = {    "Deformation": "#4C72B0",    "Environment": "#55A868",    "Damage": "#C44E52",    "Safety": "#8172B2"}#随机生成带正负相关结构的数据latent_env = np.random.normal(0, 1.0, n_samples)latent_deform = np.random.normal(0, 1.0, n_samples)latent_damage = np.random.normal(0, 1.0, n_samples)latent_risk = (    0.45 * latent_env    + 0.55 * latent_deform    + 0.65 * latent_damage    + np.random.normal(0, 0.55, n_samples))latent_deform = 0.55 * latent_risk + 0.65 * latent_deformlatent_damage = 0.60 * latent_risk + 0.60 * latent_damageX = np.zeros((n_samples, n_features))idx = {name: i for i, name in enumerate(feature_names)}# Deformation：与风险正相关for name in modules["Deformation"]:    X[:, idx[name]] = (        np.random.uniform(0.75, 1.10) * latent_deform        + 0.35 * latent_risk        + np.random.normal(0, 0.55, n_samples)    )# EnvironmentX[:, idx["Rainfall"]] = 0.85 * latent_env + 0.35 * latent_risk + np.random.normal(0, 0.60, n_samples)X[:, idx["Pore pressure"]] = 0.80 * latent_env + 0.45 * latent_risk + np.random.normal(0, 0.55, n_samples)X[:, idx["Water content"]] = 0.75 * latent_env + 0.35 * latent_damage + np.random.normal(0, 0.58, n_samples)X[:, idx["Temperature"]] = -0.35 * latent_env + np.random.normal(0, 0.80, n_samples)X[:, idx["Load level"]] = 0.65 * latent_env + 0.35 * latent_deform + np.random.normal(0, 0.60, n_samples)# Damagefor name in modules["Damage"]:    X[:, idx[name]] = (        np.random.uniform(0.70, 1.15) * latent_damage        + 0.40 * latent_risk        + 0.20 * latent_deform        + np.random.normal(0, 0.55, n_samples)    )# Safety：混合正负相关X[:, idx["Safety factor"]] = -0.95 * latent_risk - 0.35 * latent_damage + np.random.normal(0, 0.55, n_samples)X[:, idx["Stability index"]] = -0.85 * latent_risk - 0.30 * latent_deform + np.random.normal(0, 0.58, n_samples)X[:, idx["Strength reserve"]] = -0.80 * latent_risk - 0.30 * latent_damage + np.random.normal(0, 0.60, n_samples)X[:, idx["Risk index"]] = 0.90 * latent_risk + 0.30 * latent_damage + np.random.normal(0, 0.55, n_samples)X[:, idx["Failure prob."]] = 0.85 * latent_risk + 0.35 * latent_damage + np.random.normal(0, 0.56, n_samples)X = (X - X.mean(axis=0)) / X.std(axis=0)corr = np.corrcoef(X, rowvar=False)threshold = 0.55top_pos_per_node = 2top_neg_per_node = 2G = nx.Graph()for name in feature_names:    G.add_node(name, module=feature_module[name])candidate_pos = []candidate_neg = []for i in range(n_features):    for j in range(i + 1, n_features):        r = corr[i, j]        if r >= threshold:            candidate_pos.append((feature_names[i], feature_names[j], r))        if r <= -threshold:            candidate_neg.append((feature_names[i], feature_names[j], r))selected_edges = {}for node in feature_names:    pos_related = []    neg_related = []    for u, v, r in candidate_pos:        if u == node or v == node:            pos_related.append((u, v, r))    for u, v, r in candidate_neg:        if u == node or v == node:            neg_related.append((u, v, r))    pos_related = sorted(pos_related, key=lambda x: abs(x[2]), reverse=True)[:top_pos_per_node]    neg_related = sorted(neg_related, key=lambda x: abs(x[2]), reverse=True)[:top_neg_per_node]    for u, v, r in pos_related + neg_related:        key = tuple(sorted([u, v]))        selected_edges[key] = rfor (u, v), r in selected_edges.items():    G.add_edge(        u,        v,        corr=r,        weight=abs(r),        sign="positive" if r > 0 else "negative"    )weighted_degree = {}for node in G.nodes():    total = 0.0    for _, _, data in G.edges(node, data=True):        total += data["weight"]    weighted_degree[node] = totaldeg_values = np.array(list(weighted_degree.values()))deg_min = deg_values.min()deg_max = deg_values.max()node_sizes = {}for node, value in weighted_degree.items():    if deg_max > deg_min:        node_sizes[node] = 520 + 1350 * (value - deg_min) / (deg_max - deg_min)    else:        node_sizes[node] = 900node_colors = [module_colors[G.nodes[n]["module"]] for n in G.nodes()]node_size_list = [node_sizes[n] for n in G.nodes()]#图1：手动规整布局，不再用 spring_layoutmodule_centers_fig1 = {    "Deformation": (-1.6,  0.95),    "Environment": ( 1.6,  0.95),    "Damage":      (-1.6, -0.95),    "Safety":      ( 1.6, -0.95)}radius_fig1 = 0.65pos1 = {}# 每个模块内均匀分布在小圆上，并稍微旋转一下避免完全一致module_angle_offset = {    "Deformation": 0.25,    "Environment": 0.60,    "Damage": 0.05,    "Safety": 0.45}for module_name, vars_ in modules.items():    cx, cy = module_centers_fig1[module_name]    n_var = len(vars_)    angles = np.linspace(0, 2 * np.pi, n_var, endpoint=False) + module_angle_offset[module_name]    for k, var in enumerate(vars_):        rr = radius_fig1 * (0.95 + 0.08 * (k % 2))  # 轻微交错        x = cx + rr * np.cos(angles[k])        y = cy + rr * np.sin(angles[k])        pos1[var] = np.array([x, y])fig, ax = plt.subplots(figsize=(11.0, 8.0))pos_edges = [(u, v) for u, v, d in G.edges(data=True) if d["sign"] == "positive"]neg_edges = [(u, v) for u, v, d in G.edges(data=True) if d["sign"] == "negative"]pos_widths = [1.0 + 4.6 * G[u][v]["weight"] for u, v in pos_edges]neg_widths = [1.0 + 4.6 * G[u][v]["weight"] for u, v in neg_edges]nx.draw_networkx_edges(    G,    pos1,    edgelist=pos_edges,    width=pos_widths,    edge_color="#D55E00",    alpha=0.38,    ax=ax)nx.draw_networkx_edges(    G,    pos1,    edgelist=neg_edges,    width=neg_widths,    edge_color="#0072B2",    alpha=0.60,    style="dashed",    ax=ax)nx.draw_networkx_nodes(    G,    pos1,    node_color=node_colors,    node_size=node_size_list,    edgecolors="white",    linewidths=1.3,    alpha=0.98,    ax=ax)nx.draw_networkx_labels(    G,    pos1,    font_size=9.0,    font_family="Times New Roman",    font_weight="bold",    ax=ax)ax.set_title(    "Positive and Negative Correlation Network",    fontsize=18,    pad=16)ax.text(    0.02,    0.98,    f"Edges are shown when |r| ≥ {threshold}\n"    f"Top {top_pos_per_node} positive and top {top_neg_per_node} negative links per node are retained",    transform=ax.transAxes,    ha="left",    va="top",    fontsize=11,    bbox=dict(        boxstyle="round,pad=0.35",        facecolor="white",        edgecolor="#999999",        alpha=0.90    ))legend_elements = []for module_name, color in module_colors.items():    legend_elements.append(        Line2D(            [0],            [0],            marker="o",            color="w",            label=module_name,            markerfacecolor=color,            markeredgecolor="white",            markersize=11        )    )legend_elements.append(    Line2D(        [0],        [0],        color="#D55E00",        lw=3,        label="Positive correlation"    ))legend_elements.append(    Line2D(        [0],        [0],        color="#0072B2",        lw=3,        linestyle="--",        label="Negative correlation"    ))ax.legend(    handles=legend_elements,    loc="lower left",    frameon=True,    fontsize=10)ax.axis("off")ax.set_xlim(-3.0, 3.0)ax.set_ylim(-2.3, 2.3)plt.tight_layout()plt.savefig(    os.path.join(out_dir, "01_positive_negative_correlation_network_fixed.png"),    bbox_inches="tight",    facecolor="white")# 图2：模块社群图module_centers = {    "Deformation": (-3.4,  2.2),    "Environment": ( 3.4,  2.2),    "Damage":      (-3.4, -2.2),    "Safety":      ( 3.4, -2.2)}module_radius = 0.95pos2 = {}for module_name, vars_ in modules.items():    cx, cy = module_centers[module_name]    n_var = len(vars_)    angles = np.linspace(0, 2 * np.pi, n_var, endpoint=False)    for k, var in enumerate(vars_):        pos2[var] = np.array([            cx + module_radius * np.cos(angles[k]),            cy + module_radius * np.sin(angles[k])        ])within_edges = []between_edges = []for u, v, d in G.edges(data=True):    if G.nodes[u]["module"] == G.nodes[v]["module"]:        within_edges.append((u, v))    else:        between_edges.append((u, v))between_edges_sorted = sorted(    between_edges,    key=lambda e: G[e[0]][e[1]]["weight"],    reverse=True)between_edges_keep = between_edges_sorted[:12]within_pos = [(u, v) for u, v in within_edges if G[u][v]["sign"] == "positive"]within_neg = [(u, v) for u, v in within_edges if G[u][v]["sign"] == "negative"]between_pos = [(u, v) for u, v in between_edges_keep if G[u][v]["sign"] == "positive"]between_neg = [(u, v) for u, v in between_edges_keep if G[u][v]["sign"] == "negative"]fig, ax = plt.subplots(figsize=(11.2, 8.2))for module_name, (cx, cy) in module_centers.items():    ellipse = Ellipse(        xy=(cx, cy),        width=3.8,        height=2.8,        angle=0,        facecolor=module_colors[module_name],        edgecolor=module_colors[module_name],        alpha=0.12,        linewidth=2.0,        zorder=0    )    ax.add_patch(ellipse)    ax.text(        cx,        cy + 1.65,        module_name,        ha="center",        va="center",        fontsize=15,        weight="bold",        color=module_colors[module_name]    )nx.draw_networkx_edges(    G,    pos2,    edgelist=between_pos,    width=[1.0 + 4.2 * G[u][v]["weight"] for u, v in between_pos],    edge_color="#D55E00",    alpha=0.28,    ax=ax)nx.draw_networkx_edges(    G,    pos2,    edgelist=between_neg,    width=[1.0 + 4.2 * G[u][v]["weight"] for u, v in between_neg],    edge_color="#0072B2",    alpha=0.48,    style="dashed",    ax=ax)nx.draw_networkx_edges(    G,    pos2,    edgelist=within_pos,    width=[1.2 + 4.6 * G[u][v]["weight"] for u, v in within_pos],    edge_color="#D55E00",    alpha=0.52,    ax=ax)nx.draw_networkx_edges(    G,    pos2,    edgelist=within_neg,    width=[1.2 + 4.6 * G[u][v]["weight"] for u, v in within_neg],    edge_color="#0072B2",    alpha=0.62,    style="dashed",    ax=ax)nx.draw_networkx_nodes(    G,    pos2,    node_color=node_colors,    node_size=node_size_list,    edgecolors="white",    linewidths=1.4,    alpha=0.98,    ax=ax)nx.draw_networkx_labels(    G,    pos2,    font_size=9.1,    font_family="Times New Roman",    font_weight="bold",    ax=ax)ax.set_title(    "Modular Network with Positive and Negative Couplings",    fontsize=18,    pad=18)ax.legend(    handles=legend_elements,    loc="upper center",    bbox_to_anchor=(0.5, 0.98),   # 居中，放在坐标轴上方一点    ncol=3,                       # 一行放 3 个，可按需要改成 2 或 5    frameon=True,    fontsize=10)ax.axis("off")ax.set_xlim(-5.4, 5.4)ax.set_ylim(-4.2, 4.2)plt.tight_layout()plt.savefig(    os.path.join(out_dir, "02_modular_positive_negative_network_fixed.png"),    bbox_inches="tight",    facecolor="white")

特别声明：

以上代码与文案均为网上资料整合而成，仅供广大同行们参考学习，如有侵权请联系删除。