CSV基因结果比对优化：同一样本多行Call值标记异常问题

问题背景 #

我写的程序用来合并两份基因检测结果的CSV文件，基于「Sample ID」列做合并。当前逻辑只对比每行的Call_x和Call_y值，不一致就标记为「Not in Agreement」。

核心问题 #

现在遇到一种情况：同一样本会用下划线后缀区分成多行（比如NA12878和NA12878_1），这些行的Call值（比如C/A和C/C）实际存在差异，但现有逻辑没法跨行列检测这种差异。我修改代码后，所有行都被错误标记成了「Not In Agreement」。

合并后数据示例 #

原有代码 #

import pandas as pd
import tkinter as tk
from tkinter import filedialog
import tkinter.messagebox
import jinja2
import openpyxl


# 程序用途：比对两份重复检测的Call值，两份结果来自不同报告系统
# 会有两次Genotyper输出
# 若Call值不一致则标记

# 定义文件上传弹窗
def ask_path():
    root = tk.Tk()
    root.withdraw()
    path = filedialog.askopenfile(defaultextension=".csv",
                                  filetypes=[("csv files", '*.csv'),
                                             ('all files', '*.*')])  # 弹出选择框并返回路径
    return path


# 弹窗提示统一标题
def onClick(msg):
    tkinter.messagebox.showinfo("Precision Genetics", msg)


# 提示并加载第一份检测数据
genotyper1_col = ['Gene Symbol', 'Sample ID', 'NCBI SNP Reference', 'Assay ID', 'Call']
onClick('Upload First Pass')
geno1 = pd.read_csv(ask_path(), skiprows=17, usecols=genotyper1_col)

# 提示并加载第二份检测数据
genotyper2_col = ['Gene Symbol', 'Assay ID', 'Sample ID', 'NCBI SNP Reference', 'Call']
onClick('Upload Second Pass')
geno2 = pd.read_csv(ask_path(), skiprows=17, usecols=genotyper2_col)

# 按指定列合并数据
merged_data = pd.merge(geno1, geno2, on=['NCBI SNP Reference', 'Sample ID'])

# 过滤掉两份数据中NCBI SNP Reference为空的行
merged_data = merged_data[~merged_data['NCBI SNP Reference'].isna()]

# 标记Call值不一致的行
merged_data.loc[merged_data['Call_x'] != merged_data['Call_y'], 'Flag'] = 'Not in Agreement'
merged_data.loc[merged_data['Call_x'] == merged_data['Call_y'], 'Flag'] = '-'


# 高亮标记为不一致的单元格
def highlight_cells(value):
    if value == merged_data['Not in Agreement']:
        color = 'red'
    else:
        color = ''
    return 'background-color: {}'.format(color)


merged_data.style.applymap(highlight_cells)

# 保存结果
onClick('Please Choose Where You Would Like To Store This Data')


def save_loc(dataframe):
    root = tk.Tk()
    root.withdraw()
    file_path = filedialog.asksaveasfilename(defaultextension=".csv",
                                             filetypes=[("CSV Files", "*.csv"),
                                                        ("All Files", "*.*")])
    if file_path:
        dataframe.to_csv(file_path, index=False)
    root.destroy()


save_loc(merged_data)

更新后代码（存在误标记问题） #

from collections import defaultdict
import csv
import pandas as pd


def get_root_id(s):
    return s.split("_", 1)[0]


sample_ids = defaultdict(list)
headers = []


with open("merged_data.csv", newline="") as f:
    reader = csv.reader(f)
    header = next(reader)
    for row in reader:
        root_id = get_root_id(row[3])
        sample_ids[root_id].append(row)

for rows_by_id in sample_ids.values():
    call_vals = set()
    for row in rows_by_id:
        call_vals.update(row[4:7])

    if len(call_vals) > 1:
        for row in rows_by_id:
            row[8] = 'Not In Agreement'

output_file = 'output.csv'
with open("output.csv", "w", newline="") as f:
    writer = csv.writer(f)
    writer.writerow(header)
    for rows_by_id in sample_ids.values():
        for row in rows_by_id:
            writer.writerow(row)
output_data = pd.read_csv(output_file)
print(output_data)

问题分析与修复方案 #

误标记原因 #

更新后的代码中，call_vals.update(row[4:7])取了索引4到6的所有值，这包含了Call_x（索引4）、Assay ID_y（索引5）、Gene Symbol（索引6）——无关列的内容被加入集合，导致集合长度必然大于1，所有行都被错误标记。

修复代码（整合到原有Pandas流程） #

import pandas as pd
import tkinter as tk
from tkinter import filedialog
import tkinter.messagebox


# 定义文件上传弹窗
def ask_path():
    root = tk.Tk()
    root.withdraw()
    path = filedialog.askopenfile(defaultextension=".csv",
                                  filetypes=[("csv files", '*.csv'),
                                             ('all files', '*.*')])
    return path


# 弹窗提示统一标题
def onClick(msg):
    tkinter.messagebox.showinfo("Precision Genetics", msg)


# 加载两份检测数据
genotyper1_col = ['Gene Symbol', 'Sample ID', 'NCBI SNP Reference', 'Assay ID', 'Call']
onClick('上传第一份检测数据')
geno1 = pd.read_csv(ask_path(), skiprows=17, usecols=genotyper1_col)

genotyper2_col = ['Gene Symbol', 'Assay ID', 'Sample ID', 'NCBI SNP Reference', 'Call']
onClick('上传第二份检测数据')
geno2 = pd.read_csv(ask_path(), skiprows=17, usecols=genotyper2_col)

# 合并数据并过滤无效行
merged_data = pd.merge(geno1, geno2, on=['NCBI SNP Reference', 'Sample ID'])
merged_data = merged_data[~merged_data['NCBI SNP Reference'].isna()]

# 生成样本根ID（去掉下划线后缀）
merged_data['Root Sample ID'] = merged_data['Sample ID'].str.split("_", 1).str[0]

# 1. 先标记行内Call值不一致的情况
merged_data['Flag'] = '-'
merged_data.loc[merged_data['Call_x'] != merged_data['Call_y'], 'Flag'] = 'Not in Agreement'

# 2. 检测同根样本下的跨行列Call值差异
# 收集每个根样本的所有Call值
call_groups = merged_data.groupby('Root Sample ID').agg(
    all_calls=('Call_x', lambda x: set(x).union(set(merged_data.loc[x.index, 'Call_y'])))
).reset_index()

# 筛选出Call值不唯一的根样本
disagree_root_ids = call_groups[call_groups['all_calls'].apply(len) > 1]['Root Sample ID']

# 标记这些根样本下的所有行
merged_data.loc[merged_data['Root Sample ID'].isin(disagree_root_ids), 'Flag'] = 'Not in Agreement'


# 高亮标记单元格
def highlight_cells(value):
    return 'background-color: red' if value == 'Not in Agreement' else ''


styled_data = merged_data.style.applymap(highlight_cells)

# 保存结果（支持CSV/Excel格式）
onClick('请选择结果保存路径')


def save_loc(dataframe):
    root = tk.Tk()
    root.withdraw()
    file_path = filedialog.asksaveasfilename(defaultextension=".csv",
                                             filetypes=[("CSV Files", "*.csv"),
                                                        ("Excel Files", "*.xlsx")])
    if file_path:
        if file_path.endswith('.xlsx'):
            styled_data.to_excel(file_path, index=False)
        else:
            dataframe.to_csv(file_path, index=False)
    root.destroy()


save_loc(merged_data)

修复说明 #

1. 修正Call值收集范围：只提取Call_x和Call_y的值，排除无关列干扰。
2. 双重标记逻辑：先标记行内Call值不一致的情况，再检测同根样本下的跨行列差异，覆盖所有异常场景。
3. 整合流程：直接在Pandas的DataFrame上处理，避免额外的CSV读写操作，提升效率。
4. 优化保存功能：支持保存为带高亮样式的Excel文件，或纯数据的CSV文件。

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