如何在Python中利用合并结构为数据表映射PnL标签？

需求说明 #

现有两张数据表：

1. 业务数据记录表：记录业务活动的各项属性与数值
2. PnL规则配置表：包含匹配条件（部分字段为NaN，表示匹配任意值）与对应的PnL标签

要求不拆分规则表的合并结构，通过类似isin的方式为业务表每行匹配对应PnL标签，避免拆分后表过长、通用性差的问题。

数据示例 #

业务数据记录表 #

ActivityCode    CodeDepartment  BranchCode  Account     Special Code  Value
P               PR              M12         501102083   NaN           1000
P               PR              M11         501102083   NaN           2000
P               PR              M13         637102089   NaN           3000
P               TR              EGS         621102088   NaN           200
P               PC              O11         541202084   NaN           500
P               PR              M72         472002082   ZS            130

PnL规则配置表 #

Account     BranchCode  ActivityCode  CodeDepartment  PnL
465002082   NaN         P             PN              Hub Fixed
465002082   M11         P             PN              Depot Fixed
465002082   NaN         P             TR              Hub Variable
542302084   NaN         A             AC              Overheads
542302084   NaN         A             AL              Overheads
542302084   NaN         A             BL              Overheads
542302084   NaN         A             FS              Overheads
542302084   NaN         A             HR              Overheads
542302084   NaN         A             IT              Overheads
542302084   NaN         A             LW              Overheads
542302084   NaN         P             IT              Hub Fixed
542302084   NaN         P             PR              Hub Fixed
542302084   NaN         P             TR              Hub Variable
542302084   NaN         S             CM              Sales
542302084   NaN         S             CS              Sales
542302084   NaN         S             MR              Sales
543402084   NaN         A             GD              Overheads

解决方案 #

核心思路 #

规则表中NaN字段表示"匹配任意值"，非NaN字段要求与业务表对应字段精确匹配。通过批量匹配逻辑，直接为业务表每行找到符合所有条件的规则，提取对应PnL标签。

方法1：逐行匹配（适合小数据量） #

使用pandas.apply遍历业务表每行，筛选规则表中符合条件的记录并提取PnL：

import pandas as pd
import numpy as np

# 构造数据表（实际场景中可替换为读取文件逻辑）
df_business = pd.DataFrame({
    'ActivityCode': ['P', 'P', 'P', 'P', 'P', 'P'],
    'CodeDepartment': ['PR', 'PR', 'PR', 'TR', 'PC', 'PR'],
    'BranchCode': ['M12', 'M11', 'M13', 'EGS', 'O11', 'M72'],
    'Account': ['501102083', '501102083', '637102089', '621102088', '541202084', '472002082'],
    'Special Code': [np.nan, np.nan, np.nan, np.nan, np.nan, 'ZS'],
    'Value': [1000, 2000, 3000, 200, 500, 130]
})

df_rules = pd.DataFrame({
    'Account': ['465002082']*3 + ['542302084']*13 + ['543402084'],
    'BranchCode': [np.nan, 'M11', np.nan] + [np.nan]*13 + [np.nan],
    'ActivityCode': ['P','P','P'] + ['A']*7 + ['P']*3 + ['S']*3 + ['A'],
    'CodeDepartment': ['PN','PN','TR'] + ['AC','AL','BL','FS','HR','IT','LW'] + ['IT','PR','TR'] + ['CM','CS','MR'] + ['GD'],
    'PnL': ['Hub Fixed','Depot Fixed','Hub Variable'] + ['Overheads']*7 + ['Hub Fixed','Hub Fixed','Hub Variable'] + ['Sales']*3 + ['Overheads']
})

def match_pnl(row):
    # 生成匹配掩码：非NaN字段必须与业务行对应值相等
    mask = pd.Series([True]*len(df_rules), index=df_rules.index)
    for col in ['Account', 'BranchCode', 'ActivityCode', 'CodeDepartment']:
        rule_vals = df_rules[col]
        row_val = row[col]
        mask &= rule_vals.isna() | (rule_vals == row_val)
    # 返回第一个匹配的PnL，无匹配则返回NaN
    matched = df_rules.loc[mask, 'PnL']
    return matched.iloc[0] if not matched.empty else np.nan

# 应用匹配逻辑
df_business['PnL'] = df_business.apply(match_pnl, axis=1)
print(df_business)

方法2：批量广播匹配（适合大数据量） #

利用numpy广播实现批量匹配，大幅提升效率：

import pandas as pd
import numpy as np

# 构造数据表（同上）
df_business = pd.DataFrame(...)
df_rules = pd.DataFrame(...)

# 指定需要匹配的字段
match_cols = ['Account', 'BranchCode', 'ActivityCode', 'CodeDepartment']

# 转换为numpy数组
rules_arr = df_rules[match_cols].to_numpy()
business_arr = df_business[match_cols].to_numpy()

# 生成批量匹配掩码：每个业务行与所有规则行的匹配情况
mask = np.logical_and.reduce([
    (rules_arr[:, i] == business_arr[:, None, i]) | pd.isna(rules_arr[:, i])
    for i in range(len(match_cols))
], axis=0)

# 提取匹配的PnL标签
pnl_vals = df_rules['PnL'].to_numpy()
matched_pnl = np.array([pnl_vals[mask[i]][0] if mask[i].any() else np.nan for i in range(len(df_business))])

df_business['PnL'] = matched_pnl
print(df_business)

优势说明 #

• 无需拆分规则表，保留原规则结构，维护性更强
• 匹配逻辑灵活，可通过调整match_cols轻松扩展匹配字段
• 批量匹配方法适合大数据量场景，效率远高于逐行处理

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