如何使用Python为给定数据集绘制Sankey Plot（桑基图）？

首先，你的数据集预处理后如下：

import pandas as pd

data = {
    '1': ['A', 'B', 'C', 'NAN', 'A', 'C', 'NAN', 'C', 'B', 'A'],
    '2': ['B', 'NAN', 'A', 'B', 'C', 'A', 'B', 'NAN', 'A', 'C'],
    '3': ['NAN', 'A', 'B', 'C', 'NAN', 'B', 'A', 'B', 'C', 'A'],
    '4': ['C', 'B', 'NAN', 'A', 'B', 'NAN', 'C', 'A', 'NAN', 'B']
}
df = pd.DataFrame(data)
# 将字符串'NAN'替换为标准缺失值，方便后续处理
df = df.replace('NAN', pd.NA)

步骤1：统计阶段间的转移流量 #

桑基图的核心是节点间的转移次数，我们需要统计相邻阶段（列1→列2、列2→列3、列3→列4）的有效转移（排除任一值为缺失的情况）：

# 定义相邻阶段的列对
stage_pairs = [('1', '2'), ('2', '3'), ('3', '4')]
flow_data = []

for source_col, target_col in stage_pairs:
    # 过滤当前阶段对的缺失数据
    valid_rows = df[[source_col, target_col]].dropna()
    # 统计每个转移组合的出现次数
    count_result = valid_rows.value_counts().reset_index(name='value')
    # 给节点添加阶段标识，避免不同阶段的同名节点混淆
    count_result['source'] = f"阶段{source_col}: {count_result[source_col]}"
    count_result['target'] = f"阶段{target_col}: {count_result[target_col]}"
    flow_data.append(count_result[['source', 'target', 'value']])

# 合并所有转移数据
flows = pd.concat(flow_data, ignore_index=True)

步骤2：转换为桑基图所需格式 #

Plotly的桑基图需要节点索引、源节点索引、目标节点索引和流量值，我们先生成节点映射关系：

import plotly.graph_objects as go

# 获取所有唯一节点并分配索引
all_nodes = pd.concat([flows['source'], flows['target']]).unique()
node_index_map = {node: idx for idx, node in enumerate(all_nodes)}

# 转换为Plotly要求的数组格式
sources = flows['source'].map(node_index_map)
targets = flows['target'].map(node_index_map)
values = flows['value']

步骤3：绘制桑基图 #

最后生成并展示桑基图：

fig = go.Figure(data=[go.Sankey(
    node = dict(
        pad = 15,
        thickness = 20,
        line = dict(color = "black", width = 0.5),
        label = all_nodes
    ),
    link = dict(
        source = sources,
        target = targets,
        value = values
    )
)])

fig.update_layout(title_text="多阶段流转桑基图", font_size=10)
fig.show()

关键说明 #

• 用阶段X: 类别的格式命名节点，是为了区分不同阶段的同名类别（比如阶段1的A和阶段2的A），避免桑基图出现交叉混乱的连接。
• 如果不需要区分阶段的同名节点，可以去掉阶段标识，但会导致同一类别在不同阶段的流转逻辑无法清晰展示，不推荐。

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