Pandas 基于重复字段与反向结构匹配的行分组聚合实现

匹配与合并规则 #

可合并配对行识别条件 #

需同时满足以下3条：

• 两行的Not、Strike、Cents、SD、ED列取值完全一致
• 两行ExecutionTimestamp字段的时间差小于2.5分钟（即150秒）
• 两行Structure列取值分别为(+)和(-)，互为抵消行

合并后取值规则 #

• Not、Strike、SD、ED列保留原有取值
• Cents列对两行取值做求和计算
• Structure列统一替换为(=)
• 所有时间戳类字段可任意保留配对两行中任意一行的取值，无强制要求

实现代码 #

先做基础字段预处理：

import pandas as pd
import numpy as np

# 将时间字段转换为pandas可计算的datetime类型
df['ExecutionTimestamp'] = pd.to_datetime(df['ExecutionTimestamp'])
# 给Structure列加数值标记，方便快速判断正负抵消关系
df['struct_flag'] = df['Structure'].map({'(+)': 1, '(-)': -1})

逐组完成配对标记：

df['pair_id'] = np.nan
current_pid = 0
# 按要求等值匹配的列做分组
match_cols = ['Not', 'Strike', 'Cents', 'SD', 'ED']

for _, grp in df.groupby(match_cols, dropna=False):
    # 组内按执行时间升序排列，减少无效时间差计算
    grp_sorted = grp.sort_values('ExecutionTimestamp').reset_index()
    used = set()
    for i in range(len(grp_sorted)):
        if i in used:
            continue
        ri = grp_sorted.iloc[i]
        for j in range(i+1, len(grp_sorted)):
            if j in used:
                continue
            rj = grp_sorted.iloc[j]
            t_diff = (rj['ExecutionTimestamp'] - ri['ExecutionTimestamp']).total_seconds()
            # 已按时间排序，时间差超过阈值后后续行无需再判断
            if t_diff > 150:
                break
            # 结构互为抵消则配对成功
            if ri['struct_flag'] + rj['struct_flag'] == 0:
                df.loc[ri['index'], 'pair_id'] = current_pid
                df.loc[rj['index'], 'pair_id'] = current_pid
                used.update([i, j])
                current_pid += 1
                break

# 未匹配到配对的行分配独立ID，保留原始数据不参与合并
unpair_mask = df['pair_id'].isna()
df.loc[unpair_mask, 'pair_id'] = range(current_pid, current_pid + unpair_mask.sum())

按配对ID聚合得到最终结果：

# 定义各字段聚合规则
agg_config = {
    'Not': 'first',
    'Strike': 'first',
    'SD': 'first',
    'ED': 'first',
    'Cents': 'sum',
    'Structure': lambda s: '(=)' if len(s) == 2 else s.iloc[0],
    # 时间字段取组内第一个值即可，需要取第二个替换为last
    'ExecutionTimestamp': 'first'
    # 如有其他时间戳字段，按相同逻辑添加规则即可
}

final_df = df.groupby('pair_id', as_index=False).agg(agg_config).drop(columns=['pair_id', 'struct_flag'])

性能说明 #

上述逻辑为易读的基础实现，适配百万行以内的数据集无明显性能问题；如果数据量更大，可以将组内循环匹配替换为pd.merge_asof做近似时间匹配，运行效率会有明显提升。

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