使用Pandas计算日期列间排除周末的工作日时长

需求说明 #

• 需要添加三个计算列：
  1. Lead_time_1：计算story_in_progress_ts与story_being_groomed_ts之间的工作日天数（排除周末）
  2. Lead_time_2：计算story_accepted_ts与story_in_progress_ts之间的工作日天数（排除周末）
  3. Lead_time_3：计算story_release_to_prod_ts与story_accepted_ts之间的工作日天数（排除周末）
• 关键约束：
  • 计算仅统计工作日（排除周六、周日）
  • 若story_release_to_prod_ts为空，Lead_time_3设为空
  • 若story_accepted_ts为空，Lead_time_2设为空
  • 若story_in_progress_ts为空，Lead_time_1设为空
  • 若story_in_progress_ts无有效时间戳，三个Lead_time列均为空

初始DataFrame代码 #

import pandas as pd
import numpy as np

df = pd.DataFrame({
    # 随机生成基础字段
    'User Story Id':np.random.choice( ['US111111','US111112','US111113','US222221','US222222','US222223'], 6,replace=False),
    'Feature Id':np.random.choice( ['F999999','F888888'], 6),
    'Sprint Label':np.random.choice( ['ABC 23.1.1'], 6),
    'Team name':np.random.choice( ['panda','python','shark'], 6),

    # 日期时间字段
    'story_being_groomed_ts':pd.date_range('1/10/2023  8:07:21 AM', periods=6, freq='D'),
    'story_in_progress_ts':np.random.choice( pd.date_range('1/10/2023  8:07:21 AM', periods=10,
                          freq='D'), 6, replace=False),
    'story_complete_ts':np.random.choice( pd.date_range('1/20/2023  8:07:21 AM', periods=10,
                          freq='D'), 6, replace=False),
    'story_accepted_ts':np.random.choice( pd.date_range('1/30/2023  8:07:21 AM', periods=10,
                          freq='D'), 6, replace=False),
    'story_release_to_prod_ts':np.random.choice( pd.date_range('2/10/2023  8:07:21 AM', periods=10,
                          freq='D'), 6, replace=False)
})

解决方案代码 #

# 统一转换日期字段为datetime类型，无效日期转为NaN
date_cols = ['story_being_groomed_ts', 'story_in_progress_ts', 
             'story_accepted_ts', 'story_release_to_prod_ts']
df[date_cols] = df[date_cols].apply(pd.to_datetime, errors='coerce')

# 定义工作日计算函数
def calculate_workdays(end_date, start_date):
    if pd.isna(start_date) or pd.isna(end_date):
        return np.nan
    # 统计包含起止日期的工作日数量
    return len(pd.bdate_range(start=start_date, end=end_date, inclusive='both'))

# 先处理story_in_progress_ts无效的行：三个计算列全设为空
mask_invalid_progress = pd.isna(df['story_in_progress_ts'])
df.loc[mask_invalid_progress, ['Lead_time_1', 'Lead_time_2', 'Lead_time_3']] = np.nan

# 处理story_in_progress_ts有效的行，分别计算三个列
mask_valid_progress = ~mask_invalid_progress
df.loc[mask_valid_progress, 'Lead_time_1'] = df.loc[mask_valid_progress].apply(
    lambda x: calculate_workdays(x['story_in_progress_ts'], x['story_being_groomed_ts']), axis=1
)
df.loc[mask_valid_progress, 'Lead_time_2'] = df.loc[mask_valid_progress].apply(
    lambda x: calculate_workdays(x['story_accepted_ts'], x['story_in_progress_ts']), axis=1
)
df.loc[mask_valid_progress, 'Lead_time_3'] = df.loc[mask_valid_progress].apply(
    lambda x: calculate_workdays(x['story_release_to_prod_ts'], x['story_accepted_ts']), axis=1
)

关键逻辑说明 #

• 日期格式统一：通过pd.to_datetime(errors='coerce')将所有日期字段转为标准格式，无效日期自动转为NaN，统一处理空值与无效时间戳
• 工作日计算：利用pd.bdate_range直接生成工作日序列，统计序列长度得到准确的工作日天数，inclusive='both'确保起止日期都被计入
• 空值约束适配：
  • 优先识别story_in_progress_ts无效的行，直接将三个计算列设为空
  • 有效行通过自定义函数自动判断起止日期是否为空，为空则返回NaN，满足各列的空值规则

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