如何基于元组列表为 Pandas DataFrame 创建条件匹配新列

在处理 Pandas DataFrame 时，经常会遇到这样的场景：手里有一组用元组表示的区间规则，比如 (低值1, 高值1, 低值2, 高值2, 结果值)，需要根据这些规则为已有的数据框动态添加新列。传统做法是写个 apply() 加上嵌套循环，不仅慢，而且代码一复杂就特别容易出 bug。下面分享两种真正“向量化”的高效方案，既快又好读，直接绕过显式循环。

在开始之前，先快速捋一下需求：我们有一个 DataFrame，两列数值（比如 col1 和 col2），另外有一个规则列表，每条规则指定了 col1 的取值范围、col2 的取值范围，以及对应的结果值。目标是把符合区间条件的行打上结果值，不匹配的留空。看下面两种解法。

✅ 方案一：用 pyjanitor.conditional_join（推荐，尤其是条件复杂时）

这个方法的核心是把“区间匹配”转化成数据库风格的“条件连接”，语义直接对应业务规则，底层用了哈希和区间优化，性能很稳。

# 安装：pip install pyjanitor
import pandas as pd
import janitor

ls = [(1, 2, 10, 20, 5),
      (3, 4, 30, 40, 10),
      (5, 6, 50, 60, 20)]

df_ = pd.DataFrame({
    'col1': [1.1, 3.5, 5.4, 4.1],
    'col2': [11, 35, 44, 41]
})

# 把元组列表变成 DataFrame，并给结果列起个名
rules_df = pd.DataFrame(ls).rename(columns={4: 'result'})

# 统一数据类型，避免浮点和整型比较出问题
rules_df = rules_df.astype({
    0: df_['col1'].dtype,
    1: df_['col1'].dtype,
    2: df_['col2'].dtype,
    3: df_['col2'].dtype
})

# 执行条件连接：col1 ∈ [t[0], t[1]] 且 col2 ∈ [t[2], t[3]]
out = df_.conditional_join(
    rules_df,
    ('col1', 0, '>='), ('col1', 1, '<='),
    ('col2', 2, '>='), ('col2', 3, '<='),
    right_columns=['result'],
    how='left'  # 保留所有原始行，未匹配则为 NaN
)

print(out)

输出：

   col1  col2  result
0   1.1    11     5.0
1   3.5    35    10.0
2   5.4    44     NaN
3   4.1    41     NaN

这个方案最大的好处是逻辑直白——你几乎可以照着业务规则原文写代码。支持任意数量的不等式条件，还自动处理了边界重叠和重复匹配的情况。生产环境下优先考虑它，维护成本最低。

✅ 方案二：IntervalIndex + MultiIndex（纯 Pandas，轻量无额外依赖）

如果不想引入 pyjanitor，或者规则数量中等、对性能要求不是极致，可以试试这个纯 Pandas 的技巧。思路是把二维区间编码成 MultiIndex 的键，然后一次 reindex 搞定查找。

import pandas as pd
import numpy as np

ls = [(1, 2, 10, 20, 5),
      (3, 4, 30, 40, 10),
      (5, 6, 50, 60, 20)]

df_ = pd.DataFrame({
    'col1': [1.1, 3.5, 5.4, 4.1],
    'col2': [11, 35, 44, 41]
})

# 转置规则列表，方便拆成数组
a = np.array(ls).T  # shape: (5, n_rules)

# 分别为 col1 和 col2 构建 IntervalIndex
col1_intervals = pd.IntervalIndex.from_arrays(a[0], a[1])
col2_intervals = pd.IntervalIndex.from_arrays(a[2], a[3])

# 组合成多层索引： (col1区间, col2区间) → 结果值
idx = pd.MultiIndex.from_arrays([col1_intervals, col2_intervals])
lookup_series = pd.Series(a[4], index=idx)

# 对每一行 (col1, col2) 查找对应区间
df_['result'] = lookup_series.reindex(
    list(zip(df_['col1'], df_['col2']))
).values

⚠️ 几个需要注意的点：

• IntervalIndex 默认是左闭右开（closed='right'），如果你的区间是闭区间，记得显式指定 closed='both'。
• 如果规则里的区间有重叠，reindex 会返回第一个匹配的值（不保证唯一性）。最好提前验证规则是否互斥，或者手动去重。
• zip(...) 生成的元组顺序必须和 MultiIndex 构建的顺序严格一致，否则会匹配到错误的区间。

? 总结

无论选哪种，都强烈建议避开 df_.apply(lambda x: ...) 逐行遍历元组列表——那种写法的时间复杂度是 O(n×m)，数据量一上来就崩。上面这两种方案都是向量化操作，整体复杂度接近 O(n+m)。如果条件允许，优先试试 conditional_join，它的 API 几乎就是自然语言翻译，写出来自己都舒心，也更容易排查问题。