基于时间条件的分布重采样与模拟时间序列生成需求
解决方案:基于时间条件的重采样生成模拟序列
针对你的需求——从原始1分钟时间序列中,按一天内目标时间点的邻近度加权采样生成模拟序列,这里提供一套完整的实现思路和可运行代码:
核心思路拆解
- 先把原始数据的时间戳转换为一天内的分钟数(0对应00:00,1439对应23:59),方便统一计算时间距离。
- 定义一个时间加权规则:让目标时间附近的原始数据权重最高,随着时间距离增加权重快速衰减,且超过12小时(720分钟)后权重直接设为0,完全匹配你描述的采样分布要求。
- 生成目标模拟序列的时间索引,对每个时间点,基于权重从原始数据中采样对应数值。
完整代码实现
import numpy as np import pandas as pd # ---------------------- # 1. 准备原始数据(复用你提供的测试数据结构) # ---------------------- df = pd.DataFrame(index=pd.date_range(start='2020-01-01', end='2020-12-31', freq='T')) df['MyValue'] = np.random.normal(0, scale=1, size=len(df)) # 新增列:将时间戳转为一天内的分钟数(0~1439) df['minute_of_day'] = df.index.hour * 60 + df.index.minute n_daily_minutes = 1440 # 一天总分钟数 half_day_threshold = n_daily_minutes // 2 # 720分钟,即12小时 # ---------------------- # 2. 定义时间加权函数 # ---------------------- def calculate_time_weights(target_minute, all_minutes): # 处理环形时间差:比如23:59和00:00的实际差是1分钟,而非1439分钟 raw_diff = np.abs(all_minutes - target_minute) time_diff = np.minimum(raw_diff, n_daily_minutes - raw_diff) # 余弦衰减权重:目标时间点权重为1,到12小时时权重降为0,超过则权重为0 weights = np.where(time_diff > half_day_threshold, 0, np.cos(np.pi * time_diff / half_day_threshold)) # 归一化权重,确保权重总和为1 weights = weights / weights.sum() return weights # ---------------------- # 3. 生成模拟序列 # ---------------------- def generate_simulated_data(n_days=7): # 生成目标时间索引:连续n_days天的1分钟序列 sim_time_index = pd.date_range(start='2024-01-01', periods=n_days*n_daily_minutes, freq='T') sim_values = [] for timestamp in sim_time_index: target_min = timestamp.hour * 60 + timestamp.minute # 获取当前目标时间对应的采样权重 sample_weights = calculate_time_weights(target_min, df['minute_of_day'].values) # 基于权重采样一个值 sampled_val = np.random.choice(df['MyValue'].values, p=sample_weights) sim_values.append(sampled_val) # 组装成模拟DataFrame simulated_df = pd.DataFrame({'SimulatedValue': sim_values}, index=sim_time_index) return simulated_df # 生成7天的模拟序列示例 sim_result = generate_simulated_data(n_days=7) print(sim_result.head())
代码细节说明
- 环形时间处理:考虑到一天是循环的,用
np.minimum(raw_diff, n_daily_minutes - raw_diff)计算最小时间差,避免出现23:59和00:00时间差过大的不合理情况。 - 权重函数可定制:当前用余弦衰减实现平滑权重下降,你也可以替换成高斯分布等其他规则,只需要修改
calculate_time_weights里的权重计算逻辑即可。 - 大数据量优化:如果原始数据量极大(比如数年的1分钟数据),每次循环计算全量权重会较慢,可以提前按
minute_of_day分组,只从目标时间邻近的分组中采样,示例优化代码如下:
# 提前按一天内的分钟数分组,存储对应的值列表 grouped_data = df.groupby('minute_of_day')['MyValue'].apply(list) def efficient_sampling(target_minute, window=60): # 只考虑目标时间前后window分钟内的分组(处理环形) start_min = (target_minute - window) % n_daily_minutes end_min = (target_minute + window) % n_daily_minutes if start_min < end_min: target_groups = range(start_min, end_min + 1) else: target_groups = list(range(start_min, n_daily_minutes)) + list(range(0, end_min + 1)) # 收集所有候选值 candidate_values = [] for m in target_groups: candidate_values.extend(grouped_data[m]) # 可选:给邻近分组加权重,比如距离目标时间越近的组权重越高 # 这里先简化为随机采样,可根据需求调整 return np.random.choice(candidate_values)
采样分布验证
如果想确认采样是否符合预期,可以针对某个固定目标时间(比如8:00=480分钟),多次采样后统计原始数据中各minute_of_day的被采样次数,会看到480分钟附近的次数最多,距离越远次数越少,超过720分钟的几乎不会被采样到。
内容的提问来源于stack exchange,提问作者RVA92




