Python处理80万行CSV数据计算缓慢，求效率优化方案

问题背景 #

我是编程新手，现有Python代码计算逻辑正确，但执行速度极慢。需处理含80万行数据的final.csv文件，按特定时间段统计每行数据：

• 前5万行每秒可完成20次迭代，后续行数越多速度越慢，7.5万行时每秒仅10次迭代
• 时间跨度为过去365天的代码运行约20小时
• 时间跨度为365-730天的代码运行约20小时
• 时间跨度为早于730天至最早日期的代码无法完成运行

原代码如下：

import pandas as pd
from tqdm import tqdm

df = pd.read_csv('final.csv')
df['DATE_G'] = pd.to_datetime(df['DATE_G'])
df = df.sort_values(by='DATE_G')
min_date = df['DATE_G'].min()

def calculate_365_days_totals(player_id, date, coverage=None):

    past_year_data = df[(df['DATE_G'] < date - pd.DateOffset(days=730)) & (df['DATE_G'] >= min_date)]
    if coverage:
        player_data1 = past_year_data[(past_year_data['ID1_G'] == player_id) & (past_year_data['ID_C_T'] == coverage)]
        player_data2 = past_year_data[(past_year_data['ID2_G'] == player_id) & (past_year_data['ID_C_T'] == coverage)]
    else:
        player_data1 = past_year_data[(past_year_data['ID1_G'] == player_id)]
        player_data2 = past_year_data[(past_year_data['ID2_G'] == player_id)]

    sum_win_sets = player_data1['ID1_WinSet'].sum() + player_data2['ID2_WinSet'].sum()
    sum_win_games = player_data1['ID1_Games_Win'].sum() + player_data2['ID2_Games_Win'].sum()
    sum_loose_sets = player_data2['ID1_WinSet'].sum() + player_data1['ID2_WinSet'].sum()
    sum_loose_games = player_data2['ID1_Games_Win'].sum() + player_data1['ID2_Games_Win'].sum()

    return sum_win_sets, sum_win_games, sum_loose_sets, sum_loose_games


with tqdm(total=len(df), desc='Calculating totals') as pbar:
    for i, row in df.iterrows():
        df.at[i, 'ID1_365_WinSet'], df.at[i, 'ID1_365_WinGame'], df.at[i, 'ID1_365_LooseSet'], df.at[
            i, 'ID1_365_LooseGame'] = calculate_365_days_totals(row['ID1_G'], row['DATE_G'])

        df.at[i, 'ID2_365_WinSet'], df.at[i, 'ID2_365_WinGame'], df.at[i, 'ID2_365_LooseSet'], df.at[
            i, 'ID2_365_LooseGame'] = calculate_365_days_totals(row['ID2_G'], row['DATE_G'])

        coverage = row['ID_C_T']
        if coverage in [1, 2, 3, 4, 5]:
            df.at[i, f'ID1_{coverage}_Win_Set_365'], df.at[i, f'ID1_{coverage}_Win_Game_365'], df.at[
                i, f'ID1_{coverage}_Loose_Set_365'], df.at[
                i, f'ID1_{coverage}_Loose_Game_365'] = calculate_365_days_totals(row['ID1_G'], row['DATE_G'], coverage)

            df.at[i, f'ID2_{coverage}_Win_Set_365'], df.at[i, f'ID2_{coverage}_Win_Game_365'], df.at[
                i, f'ID2_{coverage}_Loose_Set_365'], df.at[
                i, f'ID2_{coverage}_Loose_Game_365'] = calculate_365_days_totals(row['ID2_G'], row['DATE_G'], coverage)
        pbar.update(1)
df.to_csv('updated_final_period3.csv', index=False)

三种时间跨度的核心筛选逻辑：

1. 过去365天：

past_year_data = df[(df['DATE_G'] < date) & (df['DATE_G'] >= date - pd.DateOffset(days=365))]

2. 365-730天前：

past_year_data = df[(df['DATE_G'] < date - pd.DateOffset(days=365)) & (df['DATE_G'] >= date - pd.DateOffset(days=730))]

3. 早于730天：

past_year_data = df[(df['DATE_G'] < date - pd.DateOffset(days=730)) & (df['DATE_G'] >= min_date)]

核心性能瓶颈 #

1. O(n²)全表遍历：每次调用统计函数都要遍历80万行数据筛选日期和玩家，总操作量达到6.4e13次，完全不可行。
2. 低效循环与赋值：iterrows是Pandas最慢的遍历方式之一，df.at逐行赋值带来大量IO开销。
3. 重复计算：同一玩家、同一时间段的统计被重复调用多次。

优化方案（按优先级排序） #

1. 转换为长格式数据，预计算累积统计 #

把原数据中ID1和ID2的记录拆分成统一长格式，按玩家ID、覆盖类型、日期排序后计算累积求和，将时间复杂度降到O(n)。

import pandas as pd

# 读取并预处理数据
df = pd.read_csv('final.csv')
df['DATE_G'] = pd.to_datetime(df['DATE_G'])
df = df.sort_values(by='DATE_G').reset_index(drop=True)

# 拆分ID1/ID2为统一长格式
id1_df = df[['DATE_G', 'ID1_G', 'ID_C_T', 'ID1_WinSet', 'ID1_Games_Win', 'ID2_WinSet', 'ID2_Games_Win']].rename(
    columns={'ID1_G': 'player_id', 'ID1_WinSet': 'win_sets', 'ID1_Games_Win': 'win_games',
             'ID2_WinSet': 'loose_sets', 'ID2_Games_Win': 'loose_games'}
)
id2_df = df[['DATE_G', 'ID2_G', 'ID_C_T', 'ID2_WinSet', 'ID2_Games_Win', 'ID1_WinSet', 'ID1_Games_Win']].rename(
    columns={'ID2_G': 'player_id', 'ID2_WinSet': 'win_sets', 'ID2_Games_Win': 'win_games',
             'ID1_WinSet': 'loose_sets', 'ID1_Games_Win': 'loose_games'}
)
long_df = pd.concat([id1_df, id2_df], ignore_index=True)
long_df = long_df.sort_values(by=['player_id', 'ID_C_T', 'DATE_G']).reset_index(drop=True)

# 计算全局/按覆盖类型的累积求和
long_df['cum_win_sets'] = long_df.groupby(['player_id'])['win_sets'].cumsum()
long_df['cum_win_games'] = long_df.groupby(['player_id'])['win_games'].cumsum()
long_df['cum_loose_sets'] = long_df.groupby(['player_id'])['loose_sets'].cumsum()
long_df['cum_loose_games'] = long_df.groupby(['player_id'])['loose_games'].cumsum()

long_df['cov_cum_win_sets'] = long_df.groupby(['player_id', 'ID_C_T'])['win_sets'].cumsum()
long_df['cov_cum_win_games'] = long_df.groupby(['player_id', 'ID_C_T'])['win_games'].cumsum()
long_df['cov_cum_loose_sets'] = long_df.groupby(['player_id', 'ID_C_T'])['loose_sets'].cumsum()
long_df['cov_cum_loose_games'] = long_df.groupby(['player_id', 'ID_C_T'])['loose_games'].cumsum()

# 设置复合索引加速查询
long_df = long_df.set_index(['player_id', 'ID_C_T', 'DATE_G']).sort_index()

2. 用二分查找快速定位区间累积值 #

利用数据已排序的特性，用bisect快速找到日期边界，通过累积值相减得到区间统计结果。

import bisect

def get_period_stats(player_id, cutoff_date, coverage=None):
    # 筛选玩家的所有记录
    try:
        if coverage is None:
            player_records = long_df.xs(player_id, level='player_id').reset_index()
        else:
            player_records = long_df.xs((player_id, coverage), level=['player_id', 'ID_C_T']).reset_index()
    except KeyError:
        return (0, 0, 0, 0)
    
    dates = player_records['DATE_G'].tolist()
    idx = bisect.bisect_left(dates, cutoff_date)
    if idx == 0:
        return (0, 0, 0, 0)
    
    # 返回对应累积值
    if coverage is None:
        return (
            player_records.iloc[idx-1]['cum_win_sets'],
            player_records.iloc[idx-1]['cum_win_games'],
            player_records.iloc[idx-1]['cum_loose_sets'],
            player_records.iloc[idx-1]['cum_loose_games']
        )
    else:
        return (
            player_records.iloc[idx-1]['cov_cum_win_sets'],
            player_records.iloc[idx-1]['cov_cum_win_games'],
            player_records.iloc[idx-1]['cov_cum_loose_sets'],
            player_records.iloc[idx-1]['cov_cum_loose_games']
        )

3. 批量处理所有行统计 #

避免逐行循环，用apply批量处理原表数据，一次性生成所有统计列。

# 生成所有结果列名
base_cols = [
    # ID1三个时间段
    'ID1_365_WinSet', 'ID1_365_WinGame', 'ID1_365_LooseSet', 'ID1_365_LooseGame',
    'ID1_365_730_WinSet', 'ID1_365_730_WinGame', 'ID1_365_730_LooseSet', 'ID1_365_730_LooseGame',
    'ID1_before730_WinSet', 'ID1_before730_WinGame', 'ID1_before730_LooseSet', 'ID1_before730_LooseGame',
    # ID2三个时间段
    'ID2_365_WinSet', 'ID2_365_WinGame', 'ID2_365_LooseSet', 'ID2_365_LooseGame',
    'ID2_365_730_WinSet', 'ID2_365_730_WinGame', 'ID2_365_730_LooseSet', 'ID2_365_730_LooseGame',
    'ID2_before730_WinSet', 'ID2_before730_WinGame', 'ID2_before730_LooseSet', 'ID2_before730_LooseGame'
]

# 生成coverage相关列名
cov_values = [1,2,3,4,5]
periods = ['365', '365_730', 'before730']
cov_cols = []
for cov in cov_values:
    for period in periods:
        cov_cols.extend([
            f'ID1_{cov}_Win_Set_{period}', f'ID1_{cov}_Win_Game_{period}',
            f'ID1_{cov}_Loose_Set_{period}', f'ID1_{cov}_Loose_Game_{period}',
            f'ID2_{cov}_Win_Set_{period}', f'ID2_{cov}_Win_Game_{period}',
            f'ID2_{cov}_Loose_Set_{period}', f'ID2_{cov}_Loose_Game_{period}'
        ])

all_result_cols = base_cols + cov_cols

# 批量处理每行数据
def process_row(row):
    date = row['DATE_G']
    cutoff_365 = date - pd.DateOffset(days=365)
    cutoff_730 = date - pd.DateOffset(days=730)
    
    # 基础统计
    id1_365 = get_period_stats(row['ID1_G'], date)
    id1_365_730 = get_period_stats(row['ID1_G'], cutoff_365)
    id1_before730 = get_period_stats(row['ID1_G'], cutoff_730)
    
    id2_365 = get_period_stats(row['ID2_G'], date)
    id2_365_730 = get_period_stats(row['ID2_G'], cutoff_365)
    id2_before730 = get_period_stats(row['ID2_G'], cutoff_730)
    
    # coverage统计
    cov_stats = []
    coverage = row['ID_C_T']
    if coverage in cov_values:
        id1_cov_365 = get_period_stats(row['ID1_G'], date, coverage)
        id1_cov_365_730 = get_period_stats(row['ID1_G'], cutoff_365, coverage)
        id1_cov_before730 = get_period_stats(row['ID1_G'], cutoff_730, coverage)
        
        id2_cov_365 = get_period_stats(row['ID2_G'], date, coverage)
        id2_cov_365_730 = get_period_stats(row['ID2_G'], cutoff_365, coverage)
        id2_cov_before730 = get_period_stats(row['ID2_G'], cutoff_730, coverage)
        
        cov_stats = list(id1_cov_365) + list(id1_cov_365_730) + list(id1_cov_before730) + \
                   list(id2_cov_365) + list(id2_cov_365_730) + list(id2_cov_before730)
    else:
        cov_stats = [0]*len(cov_cols)
    
    return list(id1_365) + list(id1_365_730) + list(id1_before730) + \
           list(id2_365) + list(id2_365_730) + list(id2_before730) + cov_stats

# 应用函数并合并到原表
df[all_result_cols] = df.apply(process_row, axis=1, result_type='expand')

# 保存结果
df.to_csv('updated_final_all_periods.csv', index=False)

4. 额外优化建议 #

• 内存优化：如果内存不足，用dask.dataframe替代Pandas进行并行处理。
• 去重预处理：先对long_df按player_id、ID_C_T、DATE_G去重，减少计算量。
• 使用Swifter：用swifter.apply替代Pandas原生apply，自动选择最优执行方式。

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