你可以使用 布尔索引 结合 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}}