Pandas处理瑕疵公共交易数据实现关联交易条目分组

前置依赖与数据预处理 #

先校准基础数据类型，避免字符串格式、浮点精度问题导致判定错误，预处理代码可直接复用：

import pandas as pd
import numpy as np
from collections import defaultdict

# 替换为实际数据读取逻辑
df = pd.read_excel("交易数据存储路径.xlsx")
# 处理带千分位逗号的数值字段
for col in ["Notional", "Premium"]:
    df[col] = df[col].astype(str).str.replace(",", "").astype(float)
# 转换数值、时间字段类型
df["Strike"] = df["Strike"].astype(float)
df["ExecutionTimestamp"] = pd.to_datetime(df["ExecutionTimestamp"])
for col in ["StartDate", "EndDate"]:
    df[col] = pd.to_datetime(df[col]).dt.date  # 日期保留年月日精度做匹配
# 初始化结果字段
df["RealStructure"] = df["Structure"]
df["StraddlePairID"] = None
df["GroupID"] = None

第一步：Straddle交易结构识别 #

识别逻辑严格匹配Straddle定义：名义本金、起止日期、行权价、标的指数、重置频率完全一致，且Call和Put成交时间差在60秒内（覆盖人工录入延迟）；配对时按时间最近原则匹配，避免重复配对。

# Straddle固定匹配维度
straddle_match_cols = ["Notional", "StartDate", "EndDate", "Strike", "Index", "ResetFreq"]
straddle_pair_id = 0

for _, group in df.groupby(straddle_match_cols, dropna=False):
    # 拆分组内Call、Put条目并按成交时间排序
    call_rows = group[group["Structure"] == "Call"].sort_values("ExecutionTimestamp").index.tolist()
    put_rows = group[group["Structure"] == "Put"].sort_values("ExecutionTimestamp").index.tolist()
    # 双指针做最近时间配对
    c_idx = p_idx = 0
    while c_idx < len(call_rows) and p_idx < len(put_rows):
        c_time = df.loc[call_rows[c_idx], "ExecutionTimestamp"]
        p_time = df.loc[put_rows[p_idx], "ExecutionTimestamp"]
        time_diff = abs((c_time - p_time).total_seconds())
        if time_diff <= 60:
            # 配对成功标记为Straddle
            df.loc[call_rows[c_idx], "RealStructure"] = "Straddle"
            df.loc[put_rows[p_idx], "RealStructure"] = "Straddle"
            df.loc[call_rows[c_idx], "StraddlePairID"] = straddle_pair_id
            df.loc[put_rows[p_idx], "StraddlePairID"] = straddle_pair_id
            straddle_pair_id += 1
            c_idx += 1
            p_idx += 1
        elif c_time < p_time:
            c_idx += 1
        else:
            p_idx += 1

第二步：可配置关联交易分组 #

关联判定采用「硬规则筛候选+软规则算得分+并查集传递分组」的逻辑，规则权重、判定阈值均可按实际业务调整，避免硬编码适配性差的问题。

规则配置 #

# 硬匹配规则：必须全部满足才可能属于关联交易，先筛掉不可能的组合减少计算量
hard_match_cols = ["StartDate", "EndDate", "Index", "ResetFreq"]
# 软评分规则：满足即加对应权重，总分≥阈值判定为关联
score_rules = [
    # 规则1：成交时间差<5分钟，加40分
    (lambda r1, r2: abs((r1["ExecutionTimestamp"] - r2["ExecutionTimestamp"]).total_seconds()) < 300, 40),
    # 规则2：Strike*10000差值可被25整除（加浮点容差避免精度误差），加30分
    (lambda r1, r2: (abs(r1["Strike"]*10000 - r2["Strike"]*10000) + 1e-6) % 25 < 1e-3, 30),
    # 规则3：名义本金接近简单整数比（容差0.1，覆盖录入尾差，匹配1:1/2:1/3:1等常见对冲比例），加30分
    (lambda r1, r2: abs(max(r1["Notional"], r2["Notional"])/min(r1["Notional"], r2["Notional"]) - round(max(r1["Notional"], r2["Notional"])/min(r1["Notional"], r2["Notional"]))) < 0.1, 30)
]
SCORE_THRESHOLD = 80  # 关联得分阈值，可根据数据质量调整

并查集实现（处理关联传递性） #

class UnionFind:
    def __init__(self, size):
        self.parent = list(range(size))
    def find(self, x):
        if self.parent[x] != x:
            self.parent[x] = self.find(self.parent[x])
        return self.parent[x]
    def union(self, x, y):
        px, py = self.find(x), self.find(y)
        if px != py:
            self.parent[py] = px

分组计算逻辑 #

uf = UnionFind(len(df))
# 生成行号作为条目唯一标识
df = df.reset_index(drop=True).reset_index().rename(columns={"index":"row_id"})

# 仅在硬规则匹配的组内做两两判定，提升计算效率
for _, hard_group in df.groupby(hard_match_cols, dropna=False):
    rows = hard_group.to_dict("records")
    for i in range(len(rows)):
        for j in range(i+1, len(rows)):
            r1, r2 = rows[i], rows[j]
            total_score = 0
            for rule_func, score in score_rules:
                if rule_func(r1, r2):
                    total_score += score
            if total_score >= SCORE_THRESHOLD:
                uf.union(r1["row_id"], r2["row_id"])

# 生成最终分组ID
df["GroupID"] = df["row_id"].apply(lambda x: f"G{uf.find(x)}")

样例数据运行效果 #

针对提供的10条样例数据，运行结果完全符合预期：

• 序号10、11：识别为重复录入的Call交易，归属同一组，结构标记为Call
• 序号12、13：识别为同参数Call交易（成交时间差56秒），归属同一组，结构标记为Call
• 序号14、16、18：归属同一组，其中16+18配对为Straddle，和14的Put组成「1份Straddle对冲1份Put」的关联组
• 序号15、17、19：归属同一组，其中17+19配对为Straddle，和15的Put组成第二组对冲关联组

可根据实际数据情况调整软规则权重、时间窗口、Strike整除基数、本金比例容差，适配不同业务场景。如果遇到录入误差更大的数据集，可适当降低得分阈值，或新增Premium比例匹配、StraddlePairID关联等规则，提升识别准确率。

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