基于QC样本日期范围为分组水质样本DataFrame分配Prelim_dql等级

嘿，我来帮你搞定这个根据QC样本给水质数据分配Prelim_dql等级的需求！核心思路就是按批次组匹配，再根据水质样本的采样日期是否落在对应QC的日期范围内，映射对应的等级。下面我用Python pandas给你一步步演示：

第一步：先定义示例数据集 #

首先咱们先模拟你的两个DataFrame，这样方便理解和测试：

import pandas as pd

# 水质样本数据集（包含批次组、采样日期等）
water_samples = pd.DataFrame({
    'sample_id': ['W1', 'W2', 'W3', 'W4', 'W5'],
    'group': ['grp1', 'grp1', 'grp2', 'grp2', 'grp1'],
    'sample_date': pd.to_datetime(['2024-01-05', '2024-01-12', '2024-01-08', '2024-01-18', '2024-01-20']),
    'parameter': ['pH', 'pH', 'DO', 'DO', 'pH']
})

# QC样本数据集（包含批次组、日期范围、对应等级）
qc_samples = pd.DataFrame({
    'group': ['grp1', 'grp1', 'grp2'],
    'qc_start_date': pd.to_datetime(['2024-01-01', '2024-01-15', '2024-01-05']),
    'qc_end_date': pd.to_datetime(['2024-01-14', '2024-01-25', '2024-01-15']),
    'Prelim_dql': ['Pass', 'Fail', 'Pass']
})

第二步：方案1 - 合并+筛选（适合小数据集） #

这种方法直观易懂，先按批次组合并两个数据集，再筛选出日期匹配的记录：

# 1. 按group合并两个数据集
merged = pd.merge(water_samples, qc_samples, on='group', how='left')

# 2. 筛选采样日期落在QC日期范围内的记录
filtered = merged[(merged['sample_date'] >= merged['qc_start_date']) & (merged['sample_date'] <= merged['qc_end_date'])]

# 3. 处理重复匹配（如果一个样本对应多个QC，这里取最新的QC记录）
filtered_sorted = filtered.sort_values(['sample_id', 'qc_end_date'], ascending=[True, False])
final_matched = filtered_sorted.drop_duplicates(subset='sample_id', keep='first')

# 4. 合并回原水质数据集，给未匹配到QC的样本标记为Unknown
final_result = water_samples.merge(
    final_matched[['sample_id', 'Prelim_dql']], 
    on='sample_id', 
    how='left'
).fillna({'Prelim_dql': 'Unknown'})

print(final_result)

运行后你会得到每个水质样本对应的Prelim_dql等级，比如W1落在grp1的第一个QC区间，等级是Pass；W5落在grp1的第二个QC区间，等级是Fail。

第三步：方案2 - 区间索引匹配（适合大数据集） #

如果你的数据集很大，上面的方法效率可能不够，用IntervalIndex可以快速匹配日期范围，效率更高：

# 1. 为每个批次组创建日期区间字典
qc_interval_dict = {}
for group_name, group_data in qc_samples.groupby('group'):
    # 创建闭区间（包含起止日期）
    intervals = pd.IntervalIndex.from_arrays(
        group_data['qc_start_date'], 
        group_data['qc_end_date'], 
        closed='both'
    )
    # 保存区间对应的等级
    qc_interval_dict[group_name] = list(zip(intervals, group_data['Prelim_dql']))

# 2. 定义函数，给每个水质样本匹配等级
def assign_dql(row):
    group_qc = qc_interval_dict.get(row['group'], [])
    for interval, dql in group_qc:
        if row['sample_date'] in interval:
            return dql
    # 没有匹配到的返回Unknown
    return 'Unknown'

# 3. 应用函数到水质数据集
water_samples['Prelim_dql'] = water_samples.apply(assign_dql, axis=1)

print(water_samples)

关键注意事项 #

• 如果同一个批次组有重叠的QC日期区间，你需要明确优先级规则（比如取最新的QC，或者取最严格的等级），上面的方案里我用了取最新QC的逻辑，你可以根据实际需求调整。
• 确保日期列都是datetime类型，不然会出现匹配错误，用pd.to_datetime()转换是必要的。
• 对于未匹配到任何QC区间的样本，我标记为Unknown，你也可以改成你需要的默认值（比如Not Assessed）。

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