如何将周末及周五16点后推文时间映射至下周一9:30匹配股价数据？

我需要将Twitter数据中的时间四舍五入至最近的5分钟区间，以匹配股价数据的5分钟窗口。现有逻辑在交易日9:30-16:00，以及非周五16:00之后至次日上午的场景中运行正常，但周五16:00之后、周末发布的推文需要映射至下周一9:30（比如2021/04/09周五17:42的推文，要映射到2021/04/12周一9:30）。

输入数据 #

Twitter数据

股价数据

预期输出 #

现有代码（未覆盖周末/周五16点后场景） #

def time_map(row):
    new_col_tmp = row['new_col_tmp']
    new_col2 = row['new_col2']
    
    if  930 <= int(new_col_tmp) <= 1600:
        convert = new_col2 + " " + new_col_tmp
    elif int(new_col_tmp) < 930 or int(new_col_tmp) > 1600:
        convert = new_col2 + " " + '0930'

    return convert

tweet['new_col1'] = tweet.created_at.round("5min").dt.strftime('%H:%M')
tweet['new_col_tmp'] = tweet.created_at.round("5min").dt.strftime('%H%M')
tweet['new_col2'] = tweet.created_at.round("5min").dt.strftime('%Y%m%d')
tweet['convert'] = tweet.apply(lambda row: time_map(row), axis=1)
tweet['convert'] = pd.to_datetime(tweet.convert, format='%Y%m%d %H%M').dt.strftime('%d/%m/%Y %H:%M')

核心思路是通过日期和时间的双重判断，精准映射到对应交易日的时间窗口，具体步骤：

1. 将原始时间四舍五入到最近5分钟区间
2. 判断时间是否属于非交易日（周六/周日）或周五16:00之后
3. 对特殊场景计算下一个周一的9:30，其他场景按规则映射到当日9:30或对应5分钟区间
4. 最后将映射后的时间与股价数据关联匹配

修改后的完整代码：

import pandas as pd
from pandas.tseries.offsets import CustomBusinessDay

def get_target_time(rounded_time):
    # 定义交易日规则：周一至周五
    bday = CustomBusinessDay(weekmask='Mon Tue Wed Thu Fri')
    
    # 提取小时分钟数值（如9:42转为942）和星期几（0=周一，4=周五，5=周六，6=周日）
    hour_min = rounded_time.hour * 100 + rounded_time.minute
    weekday = rounded_time.weekday()
    
    # 场景1：交易日9:30-16:00，直接保留四舍五入后的时间
    if 930 <= hour_min <= 1600 and weekday < 5:
        return rounded_time
    
    # 场景2：周五16点后或周末，映射到下周一9:30
    if weekday == 4 and hour_min > 1600 or weekday >= 5:
        next_trading_day = rounded_time + bday
        return next_trading_day.replace(hour=9, minute=30, second=0, microsecond=0)
    
    # 场景3：交易日非9:30-16:00时段，映射到当日9:30
    else:
        return rounded_time.replace(hour=9, minute=30, second=0, microsecond=0)

# 处理Twitter数据（若time列未转为datetime类型，先执行转换）
tweet['time'] = pd.to_datetime(tweet['time'])
# 四舍五入到最近5分钟
tweet['rounded_time'] = tweet['time'].round('5min')
# 计算目标映射时间
tweet['target_time'] = tweet['rounded_time'].apply(get_target_time)

# 加载并处理股价数据（若为外部数据可替换为读取逻辑）
price_data = pd.DataFrame({
    'price': [100, 110, 120, 130],
    'time': pd.to_datetime(['2021-04-08 16:00:00', '2021-04-09 09:30:00', '2021-04-09 09:45:00', '2021-04-12 09:30:00'])
})
price_data['time'] = pd.to_datetime(price_data['time'])

# 合并数据并调整列名匹配预期输出
result = pd.merge(tweet[['tweet', 'target_time']], price_data, left_on='target_time', right_on='time', how='left')
result = result.rename(columns={'target_time': 'time'}).drop(columns=['time_y'])
result = result.rename(columns={'time_x': 'time'})

# 输出结果
print(result)

代码说明 #

• 用CustomBusinessDay自动计算下一个交易日，避免手动判断周几的误差
• 分三种场景处理时间映射，覆盖所有边界情况
• 通过merge实现映射时间与股价数据的精准关联

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