R中GRanges::reduce()的Python等价实现：如何合并区间数据？

刚好做过类似的需求！不用存磁盘到文件，用pyranges这个Python库就能完美复刻R里的流程，全程在内存操作，效率还很高。下面一步步给你演示：

1. 先安装依赖库 #

pyranges是专门处理基因组区间的工具，和R的GRanges语法很贴近，直接用pip安装：

pip install pyranges pandas

2. 创建测试DataFrame #

先模拟你在R里的测试数据，用pandas生成包含染色体、起始、终止的DataFrame：

import pandas as pd
import pyranges as pr

# 生成测试数据，和R里的data.frame结构对应
test_df = pd.DataFrame({
    "Chromosome": ["chr1", "chr1", "chr1", "chr2", "chr2", "chr2"],
    "Start": [10, 20, 50, 100, 120, 150],
    "End": [15, 25, 60, 110, 130, 160]
})
print("原测试数据：")
print(test_df)

3. 转换为基因组区间对象（对应R的GRanges） #

直接从pandas DataFrame转换为Pyranges对象，全程内存操作，不需要存文件：

gr = pr.PyRanges(test_df)
print("\n转换后的基因组区间对象：")
print(gr)

4. 合并重叠/相邻区间（对应R的reduce） #

用pyranges的merge方法，默认会合并重叠或相邻的区间，还可以通过distance参数设置允许合并的相邻间隔（比如distance=5表示间隔5bp以内的区间也合并）：

# 合并区间，得到新的Pyranges对象
merged_gr = gr.merge()
# 给合并后的区间添加区块ID，方便后续匹配
merged_gr = merged_gr.assign(BlockID=range(1, len(merged_gr)+1))
print("\n合并后的连续区块：")
print(merged_gr)

5. 给原DataFrame标记所属区块 #

通过重叠匹配，把原数据里的每个区间关联到对应的合并区块ID，用pyranges的join方法就能快速完成：

# 把原区间和合并后的区间做左连接，匹配重叠的区间
joined_result = gr.join(merged_gr, how="left")
# 把区块ID赋值回原DataFrame
test_df["BlockID"] = joined_result.BlockID.values

print("\n标记区块ID后的原数据：")
print(test_df)

备选方案：不用pyranges，纯pandas实现 #

如果不想额外安装库，也可以手动用pandas实现区间合并和标记，不过逻辑会繁琐一些，适合简单场景：

import pandas as pd

def add_block_ids(df):
    # 按染色体分组，先排序区间
    df_sorted = df.sort_values(["Chromosome", "Start"]).reset_index(drop=True)
    block_ids = []
    current_block = 1
    prev_chrom = None
    prev_end = -1

    for _, row in df_sorted.iterrows():
        # 切换染色体时重置区块
        if row["Chromosome"] != prev_chrom:
            current_block = 1
            prev_end = row["End"]
            block_ids.append(current_block)
            prev_chrom = row["Chromosome"]
        else:
            # 重叠或相邻则归为同一区块
            if row["Start"] <= prev_end:
                block_ids.append(current_block)
                prev_end = max(prev_end, row["End"])
            else:
                current_block += 1
                prev_end = row["End"]
                block_ids.append(current_block)
    
    df_sorted["BlockID"] = block_ids
    # 恢复原数据的顺序
    return df_sorted.set_index(df.index).sort_index()

# 测试纯pandas方法
test_df_pd = add_block_ids(test_df.drop("BlockID", axis=1))
print("\n纯pandas实现的标记结果：")
print(test_df_pd)

这个纯pandas方法只处理了重叠或直接相邻的区间，没有pyranges的灵活配置，但胜在不需要额外依赖。

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