You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

基于时间条件的分布重采样与模拟时间序列生成需求

解决方案:基于时间条件的重采样生成模拟序列

针对你的需求——从原始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

火山引擎 最新活动