Python中基于Numpy数组的时间序列时间延迟嵌入实现问询
实现自定义规则的时间延迟嵌入(Numpy版)
Got it,你要的是一种自定义规则的时间延迟嵌入,用Numpy来实现对吧?咱们先把需求掰扯清楚,再上代码:
需求拆解
先明确你说的规则细节:
- 窗口大小
w=4:每个嵌入向量包含4个元素 - 间隔
g=2+ 选取每第3个点:这里的“每第3个”本质是采样步长为3(间隔2个点跳过,取第1、4、7...这样的点) - 滑动逻辑:每次把起始点平移1个位置,生成新的向量;直到从当前起始点开始,凑不够4个采样点时停止。
Numpy实现代码
首先假设你的原始时间序列是1维Numpy数组,比如我们用一个示例序列来测试:
import numpy as np # 示例时间序列:长度为20的连续整数 ts = np.arange(20) # 输出: [0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19]
然后写核心的嵌入函数:
def custom_delay_embedding(ts, window_size=4, sample_step=3): # 计算最大合法起始索引:确保最后一个采样点不超出序列范围 max_start_idx = len(ts) - (window_size - 1) * sample_step if max_start_idx <= 0: return np.array([]) # 无法生成任何完整向量,返回空数组 # 生成所有起始点的索引数组 start_indices = np.arange(max_start_idx) # 通过广播生成每个嵌入向量的索引矩阵 embedding_indices = start_indices[:, None] + sample_step * np.arange(window_size) # 从原始序列中提取对应元素,得到最终的嵌入矩阵 embedding_matrix = ts[embedding_indices] return embedding_matrix
测试与输出
调用函数测试:
# 传入参数:窗口大小4,采样步长3(对应你的g=2) result = custom_delay_embedding(ts, window_size=4, sample_step=3) print(result)
输出结果是一个二维Numpy数组,每一行是一个符合规则的嵌入向量:
[[ 0 3 6 9] [ 1 4 7 10] [ 2 5 8 11] [ 3 6 9 12] [ 4 7 10 13] [ 5 8 11 14] [ 6 9 12 15] [ 7 10 13 16] [ 8 11 14 17] [ 9 12 15 18] [10 13 16 19]]
代码解释
- 起始索引计算:
max_start_idx是核心逻辑,它保证从每个起始点出发,按步长采样后能凑够window_size个点。比如示例中len(ts)=20,(4-1)*3=9,所以max_start_idx=20-9=11,对应0到10共11个合法起始点。 - 广播机制:
start_indices[:, None]把一维起始索引转成列向量,加上sample_step * np.arange(window_size)(即[0,3,6,9]),通过Numpy的广播特性自动生成索引矩阵,这比循环实现高效得多,尤其适合处理大时间序列。 - 边界处理:如果序列太短,连一个完整窗口都凑不出来,函数会返回空数组,避免索引越界错误。
灵活调整
如果你的“间隔g”定义和我理解的不同(比如g是两个采样点之间的元素个数,比如x0和x2之间间隔1个元素,g=1),只需要调整sample_step参数即可:sample_step = g + 1。
内容的提问来源于stack exchange,提问作者Avani Sharma
相关产品推荐
相关产品推荐

