如何优化Python大文件处理代码以缩短计算耗时？

我写了一段Python代码，用来读取200000KB的大文件，逐行对比相邻两行数据，计算第n行与第n-1行的数值差值后写入新文件，需要分别写入中子（neutron）、电子（electron）、光子（photon）对应的三个文件，这部分通过pisi函数实现。

目前处理7个文件约耗时2小时，但我需要处理200个左右的文件，因此想咨询代码中是否存在可优化的空间以提升效率。

我的处理逻辑如下：

• 打开目标文件
• 读取一行并缓存，读取下一行后进行运算，随后将当前行设为缓存行，重复此过程至文件结束
• 根据特定条件将运算结果写入三个不同的文件

我对Python性能优化经验不足，此前均采用暴力实现，希望获得相关帮助。

原代码 #

pisi文件写入函数

def pisi(delc,oldline,newline,DeltaE,procent): #write to file
    if delc==1.0:
        with open('Nevtroni%i.txt' %procent, 'a+') as f:
            E1 = float(oldline[-2])  # E1
            X = float(oldline[-9]) - float(newline[-9])  # delta x
            Y = float(oldline[-8]) - float(newline[-8])  # delta y
            Z = float(oldline[-7]) - float(newline[-7])  # delta z
            Part = float(oldline[-1])
            if DeltaE>0:
                f.write('%s' % DeltaE + ' ' + '%s' % E1 + ' ' + '%s' % X + ' ' + '%s' % Y + ' ' + '%s' % Z + ' ' + '%s' % Part)
                f.write('\n')
        f.close()
    if delc==2.0:
        with open('Fotoni%i.txt' %procent, 'a+') as f:
            E1 = float(oldline[-2])  # E1
            X = float(oldline[-9]) - float(newline[-9])  # delta x
            Y = float(oldline[-8]) - float(newline[-8])  # delta y
            Z = float(oldline[-7]) - float(newline[-7])  # delta z
            Part = float(oldline[-1])
            if DeltaE>0:
                f.write('%s' % DeltaE + ' ' + '%s' % E1 + ' ' + '%s' % X + ' ' + '%s' % Y + ' ' + '%s' % Z + ' ' + '%s' % Part)
                f.write('\n')
        f.close()
    if delc==3.0:
        with open('Elektroni%i.txt' %procent, 'a+') as f:
            E1 = float(oldline[-2])  # E1
            X = float(oldline[-9]) - float(newline[-9])  # delta x
            Y = float(oldline[-8]) - float(newline[-8])  # delta y
            Z = float(oldline[-7]) - float(newline[-7])  # delta z
            Part = float(oldline[-1])
            if DeltaE>0:
                f.write('%s' % DeltaE + ' ' + '%s' % E1 + ' ' + '%s' % X + ' ' + '%s' % Y + ' ' + '%s' % Z + ' ' + '%s' % Part)
                f.write('\n')
        f.close()

uporabno主处理函数

def uporabno(fname,a):
    with open(fname) as infile:
        oldline=None

        for line in infile:
            newline = line.strip().split(' ')  # ev:# event_type mat cell_no x y z vx vy vz weight E particle_type
            if oldline is not None:
                    #if reads specific things i want from fname all are similar
                if newline[1] == 'COL' and oldline[1]=='SUR':
                    DeltaE = float(oldline[-2]) - float(newline[-2])
                    if newline[0] != 'ev:1':
                        pisi(float(oldline[-1]), oldline, newline, DeltaE,a)
                    oldline=newline
                if newline[1] == 'BNK' and oldline[1]=='SUR':
                    DeltaE = float(oldline[-2]) - float(newline[-2])
                    if oldline[-10]==newline[-10] and newline[0] != 'ev:1': #if BNK happens in same material
                        pisi(float(oldline[-1]), oldline, newline, DeltaE,a)
                    oldline=newline
                if newline[1] == 'TER' and oldline[1]=='SUR':
                    DeltaE = float(oldline[-2]) - float(newline[-2])
                    if newline[0] !='ev:1':
                        pisi(float(oldline[-1]), oldline, newline, DeltaE,a)
                    oldline=newline
                if newline[1] == 'COL' and oldline[1]=='COL':
                    DeltaE = float(oldline[-2]) - float(newline[-2])
                    if newline[0] != 'ev:1':
                        pisi(float(oldline[-1]), oldline, newline, DeltaE,a)
                    oldline=newline
                if newline[1] == 'SUR' and oldline[1]=='COL':
                    DeltaE = float(oldline[-2]) - float(newline[-2])
                    #skip as not consecutive events, just means particle crossed surface after colision
                    oldline=newline
                if newline[1] == 'BNK' and oldline[1]=='COL':
                    DeltaE = float(oldline[-2]) - float(newline[-2])
                    if oldline[-10] == newline[-10] and newline[0] != 'ev:1':  # if BNK happens in same material
                        pisi(float(oldline[-1]), oldline, newline, DeltaE,a)
                    oldline=newline
                if newline[1] == 'TER' and oldline[1]=='COL':
                    DeltaE = float(oldline[-2]) - float(newline[-2])
                    if oldline[-10] == newline[-10] and newline[0] !='ev:1':  # if BNK happens in same material
                        pisi(float(oldline[-1]), oldline, newline, DeltaE,a)
                    oldline=newline
                if newline[1] == 'TER' and oldline[1]=='BNK':
                    DeltaE = float(oldline[-2]) - float(newline[-2])
                    if oldline[-10] == newline[-10] and newline[0] != 'ev:1' and float(oldline[-2])<1:  # same material, not nev event, energija manša od Mev ker drugac ne razumemo
                        pisi(float(oldline[-1]), oldline, newline, DeltaE,a)
                    oldline=newline #read      #
                if newline[1] == 'SUR' and oldline[1]=='BNK':
                    DeltaE = float(oldline[-2]) - float(newline[-2])
                    #skip as not event, just means particle crossed surface.
                    oldline=newline
                if newline[1] == 'COL' and oldline[1]=='BNK':
                    DeltaE = float(oldline[-2]) - float(newline[-2])
                    if oldline[-10] == newline[-10] and newline[0] != 'ev:1':  # if BNK happens in same material
                        pisi(float(oldline[-1]), oldline, newline, DeltaE,a)
                    oldline=newline
                if newline[1] == 'BNK' and oldline[1]=='TER':
                    DeltaE = float(oldline[-2]) - float(newline[-2])
                    #skip for electrons
                    if oldline[-10] == newline[-10] and newline[0] !='ev:1' and float(newline[-1]) != 3.0: # if BNK happens in same material
                        pisi(float(oldline[-1]), oldline, newline, DeltaE,a)
                    oldline=newline
                if newline[1] == 'COL' and oldline[1]=='TER':
                    DeltaE = float(oldline[-2]) - float(newline[-2])
                    if oldline[-10] == newline[-10] and newline[0] != 'ev:1':  # if BNK happens in same material
                        pisi(float(oldline[-1]), oldline, newline, DeltaE,a)
                    oldline=newline
                if newline[1] == 'SUR' and oldline[1]=='TER':
                    #skip since surface is not event.
                    oldline=newline
                if newline[1] == 'SUR' and oldline[1]=='SUR':
                    #skip
                    oldline=newline
                if newline[1] == 'BNK' and oldline[1]=='BNK':
                    #skip
                    oldline=newline
            else:
                oldline=newline

优化方案 #

1. 避免频繁打开/关闭输出文件

文件IO是性能瓶颈之一，原pisi函数每次调用都打开、关闭文件，会产生大量IO开销。建议在主函数开始时提前打开三个输出文件，处理完整个输入文件后再统一关闭。

2. 重构pisi函数消除重复代码

原函数三个分支逻辑完全重复，仅文件名不同，可通过映射关系直接匹配文件，减少冗余计算：

def pisi(delc, oldline, newline, DeltaE, files):
    if DeltaE <= 0:
        return  # 提前退出，减少无效计算
    # 一次性计算所有需要的值
    E1 = float(oldline[-2])
    X = float(oldline[-9]) - float(newline[-9])
    Y = float(oldline[-8]) - float(newline[-8])
    Z = float(oldline[-7]) - float(newline[-7])
    Part = float(oldline[-1])
    # 用f-string格式化，比字符串拼接高效
    output_line = f"{DeltaE} {E1} {X} {Y} {Z} {Part}\n"
    # 写入对应文件
    target_file = files.get(delc)
    if target_file:
        target_file.write(output_line)

3. 优化主函数的条件判断

原主函数中多个独立if会导致不必要的判断，改成elif可减少分支遍历次数；同时提前转换重复使用的数值，避免多次调用float()：

def uporabno(fname,a):
    # 提前打开三个输出文件
    file_map = {
        1.0: open(f'Nevtroni{a}.txt', 'a+'),
        2.0: open(f'Fotoni{a}.txt', 'a+'),
        3.0: open(f'Elektroni{a}.txt', 'a+')
    }
    try:
        with open(fname) as infile:
            oldline = None
            old_event = None
            old_mat = None
            old_E = None
            old_part = None

            for line in infile:
                newline_data = line.strip().split(' ')
                new_event = newline_data[1]
                new_ev_num = newline_data[0]
                new_mat = newline_data[-10]
                new_E = float(newline_data[-2])
                new_part = float(newline_data[-1])

                if oldline is not None:
                    DeltaE = old_E - new_E
                    # 用elif替代多个独立if
                    if new_event == 'COL' and old_event == 'SUR':
                        if new_ev_num != 'ev:1':
                            pisi(old_part, oldline, newline_data, DeltaE, file_map)
                    elif new_event == 'BNK' and old_event == 'SUR':
                        if old_mat == new_mat and new_ev_num != 'ev:1':
                            pisi(old_part, oldline, newline_data, DeltaE, file_map)
                    elif new_event == 'TER' and old_event == 'SUR':
                        if new_ev_num != 'ev:1':
                            pisi(old_part, oldline, newline_data, DeltaE, file_map)
                    elif new_event == 'COL' and old_event == 'COL':
                        if new_ev_num != 'ev:1':
                            pisi(old_part, oldline, newline_data, DeltaE, file_map)
                    elif new_event == 'BNK' and old_event == 'COL':
                        if old_mat == new_mat and new_ev_num != 'ev:1':
                            pisi(old_part, oldline, newline_data, DeltaE, file_map)
                    elif new_event == 'TER' and old_event == 'COL':
                        if old_mat == new_mat and new_ev_num != 'ev:1':
                            pisi(old_part, oldline, newline_data, DeltaE, file_map)
                    elif new_event == 'TER' and old_event == 'BNK':
                        if old_mat == new_mat and new_ev_num != 'ev:1' and old_E < 1:
                            pisi(old_part, oldline, newline_data, DeltaE, file_map)
                    elif new_event == 'COL' and old_event == 'BNK':
                        if old_mat == new_mat and new_ev_num != 'ev:1':
                            pisi(old_part, oldline, newline_data, DeltaE, file_map)
                    elif new_event == 'BNK' and old_event == 'TER':
                        if old_mat == new_mat and new_ev_num != 'ev:1' and new_part != 3.0:
                            pisi(old_part, oldline, newline_data, DeltaE, file_map)
                    elif new_event == 'COL' and old_event == 'TER':
                        if old_mat == new_mat and new_ev_num != 'ev:1':
                            pisi(old_part, oldline, newline_data, DeltaE, file_map)
                    # 其他需要跳过的分支直接更新oldline即可

                    # 更新缓存数据
                    oldline = newline_data
                    old_event = new_event
                    old_mat = new_mat
                    old_E = new_E
                    old_part = new_part
                else:
                    # 初始化缓存
                    oldline = newline_data
                    old_event = new_event
                    old_mat = new_mat
                    old_E = new_E
                    old_part = new_part
    finally:
        # 确保所有文件都关闭
        for f in file_map.values():
            f.close()

4. 其他小优化

• 使用f-string替代%s和字符串拼接，速度更快且代码更简洁；
• 对于超大文件，可考虑使用sys.stdin读取，或用二进制模式打开后批量解码，减少IO等待时间。

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