使用Python处理含复杂合并逻辑的CSV文件数据合并问题

核心思路 #

只遍历CSV一次，用两个字典分别存储带MIC的主记录和无MIC的补充数据，最后合并两者，避免多次IO操作导致的性能问题。

具体步骤 #

1. 初始化存储结构

   • main_records：键为(Inv_ID, MIC)（唯一组合），值为包含目标字段的字典，用来存放所有MIC非null的行数据
   • supplementary_data：键为Inv_ID，值为存放CUSIP、ISIN的字典，用来收集MIC为null的行的补充信息

2. 单次遍历处理每行数据

   • 遇到MIC非null的行：
     • 生成唯一键(Inv_ID, MIC)
     • 根据IdType将IdValue映射到对应字段（SEDOL/TICKER），同时记录Sr#、Inv_ID、MIC
     • 若键已存在（如同一组合下的SEDOL和TICKER行），则合并字段
   • 遇到MIC为null的行：
     • 以Inv_ID为键，将IdValue存入对应字段（CUSIP/ISIN），重复值直接覆盖

3. 合并补充数据到主记录
   遍历main_records中的每条记录，根据Inv_ID从supplementary_data中取出对应的CUSIP和ISIN，填充到记录中。

4. 输出结果
   将main_records中的数据转换成目标格式输出。

Python 代码示例 #

import csv

def process_csv(input_path, output_path):
    main_records = {}
    supplementary_data = {}

    with open(input_path, 'r', newline='') as infile:
        # 假设是制表符分隔，实际根据CSV真实分隔符调整
        reader = csv.DictReader(infile, delimiter='\t')  
        for row in reader:
            inv_id = row['Inv_ID'].strip()
            mic = row['MIC'].strip()
            id_type = row['IdType'].strip()
            id_value = row['IdValue'].strip()
            sr_num = row['Sr#'].strip()

            if mic != 'null':
                key = (inv_id, mic)
                if key not in main_records:
                    main_records[key] = {
                        'Sr#': '',
                        'SEDOL': '',
                        'MIC': mic,
                        'ISIN': '',
                        'CUSIP': '',
                        'TICKER': '',
                        'Inv_ID': inv_id
                    }
                # 取SEDOL行的Sr#，与示例结果逻辑一致
                if id_type == 'SEDOL':
                    main_records[key]['Sr#'] = sr_num
                main_records[key][id_type] = id_value
            else:
                if inv_id not in supplementary_data:
                    supplementary_data[inv_id] = {'CUSIP': '', 'ISIN': ''}
                supplementary_data[inv_id][id_type] = id_value

    # 合并补充数据到主记录
    for key in main_records:
        inv_id = main_records[key]['Inv_ID']
        if inv_id in supplementary_data:
            main_records[key]['CUSIP'] = supplementary_data[inv_id]['CUSIP']
            main_records[key]['ISIN'] = supplementary_data[inv_id]['ISIN']

    # 写入结果文件
    fieldnames = ['Sr#', 'SEDOL', 'MIC', 'ISIN', 'CUSIP', 'TICKER', 'Inv_ID']
    with open(output_path, 'w', newline='') as outfile:
        writer = csv.DictWriter(outfile, fieldnames=fieldnames, delimiter='\t')
        writer.writeheader()
        for record in main_records.values():
            writer.writerow(record)

# 调用示例
process_csv('input.csv', 'output.csv')

说明 #

• 分隔符：示例假设为制表符分隔，实际使用时需根据CSV真实分隔符调整（如逗号delimiter=','）
• 性能优势：仅需一次文件读取，内存中完成数据合并，避免了两次IO操作和重复遍历，处理大文件时性能提升明显
• 兼容性：支持同一Inv_ID下有多组MIC组合，自动按Inv_ID+MIC分组并填充共享的CUSIP/ISIN数据

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