pd数据范围检查

Posted by Shallow Dreameron August 24, 2025

你可以使用 布尔索引 结合 between() 或条件筛选来获取 某个字段中超出最大最小值范围的行


示例代码

import pandas as pd

# 示例数据
df = pd.DataFrame({
    "id": [1, 2, 3, 4, 5],
    "value": [10, 20, 30, 40, 100]
})

# 取出某个字段的最大值和最小值
min_val = df["value"].min()
max_val = df["value"].max()

# 筛选出不在最大最小值范围内的数据
out_of_range = df[~df["value"].between(min_val, max_val)]

print(out_of_range)

关键点

  • between(min, max) 返回布尔序列,表示是否在区间 [min, max] 内。
  • ~ 取反,筛选出不在该范围内的行。
  • 如果只想检查 某一行,可以:
row_index = 2  # 例如第三行
value = df.loc[row_index, "value"]
if not (min_val <= value <= max_val):
    print(f"第{row_index}行的值 {value} 不在范围内")

明白了,你的场景是:

  • 已有字段的允许范围(最小值和最大值),可能来自配置文件或预先计算。
  • 有新的 DataFrame(或单行数据),需要验证哪些字段超出了范围。
  • 需要统计不符合规则的行数,也要知道具体哪些字段、哪些值不符合。

解决方案

1. 定义字段限制

limits = {
    "value": {"min": 10, "max": 50},
    "score": {"min": 0, "max": 100}
}

2. 检查 DataFrame 是否超出范围

import pandas as pd

# 新数据
df_new = pd.DataFrame({
    "id": [1, 2, 3, 4],
    "value": [5, 20, 55, 40],   # 有的值不在 [10, 50]
    "score": [10, 110, 90, -5]  # 有的值不在 [0, 100]
})

# 初始化布尔 DataFrame,用于记录是否超出范围
mask = pd.DataFrame(False, index=df_new.index, columns=df_new.columns)

# 检查每个字段
for field, bound in limits.items():
    mask[field] = ~df_new[field].between(bound["min"], bound["max"])

# 找出任意字段超出范围的行
invalid_rows = df_new[mask.any(axis=1)]

print("不符合规则的数据行:")
print(invalid_rows)
print("不符合规则的总行数:", invalid_rows.shape[0])

3. 获取具体字段超出范围信息

for idx, row in mask.iterrows():
    invalid_fields = row[row].index.tolist()
    if invalid_fields:
        print(f"第 {idx} 行字段 {invalid_fields} 的值不在范围内")

输出结果

  • invalid_rows:完整的不符合规则的行
  • invalid_rows.shape[0]:不符合规则的行数
  • 逐行字段级别检查:明确指出哪些字段超出范围

下面是一个通用函数,能够检查任意 DataFrame 中的值是否在给定字段范围内,并返回不合格的行和详细的字段级别信息。


通用函数

import pandas as pd

def check_out_of_range(df: pd.DataFrame, limits: dict):
    """
    检查 DataFrame 中的值是否在字段限制范围内

    参数:
        df (pd.DataFrame): 需要检查的数据
        limits (dict): 字段限制范围,例如:
            {
                "value": {"min": 10, "max": 50},
                "score": {"min": 0, "max": 100}
            }

    返回:
        invalid_rows (pd.DataFrame): 不符合规则的行
        details (list): 每行不符合规则的字段信息
    """
    mask = pd.DataFrame(False, index=df.index, columns=df.columns)
    
    # 遍历字段限制,生成布尔掩码
    for field, bound in limits.items():
        if field in df.columns:
            mask[field] = ~df[field].between(bound["min"], bound["max"])

    # 获取不符合规则的行
    invalid_rows = df[mask.any(axis=1)]
    
    # 获取详细字段信息
    details = []
    for idx, row in mask.iterrows():
        invalid_fields = row[row].index.tolist()
        if invalid_fields:
            details.append({
                "row_index": idx,
                "invalid_fields": invalid_fields,
                "values": df.loc[idx, invalid_fields].to_dict()
            })

    return invalid_rows, details

使用示例

df_new = pd.DataFrame({
    "id": [1, 2, 3, 4],
    "value": [5, 20, 55, 40],
    "score": [10, 110, 90, -5]
})

limits = {
    "value": {"min": 10, "max": 50},
    "score": {"min": 0, "max": 100}
}

invalid_rows, details = check_out_of_range(df_new, limits)

print("不符合规则的行:")
print(invalid_rows)
print("\n详细信息:")
for d in details:
    print(d)

输出示例

不符合规则的行:
   id  value  score
0   1      5     10
1   2     20    110
2   3     55     90
3   4     40     -5

详细信息:
{'row_index': 0, 'invalid_fields': ['value'], 'values': {'value': 5}}
{'row_index': 1, 'invalid_fields': ['score'], 'values': {'score': 110}}
{'row_index': 2, 'invalid_fields': ['value'], 'values': {'value': 55}}
{'row_index': 3, 'invalid_fields': ['score'], 'values': {'score': -5}}