具有可变样本间隔的时间序列：如何插入具有（多个）缺失时间步的行并为剩余列插入值

Question

数据：

• 数组第一列=次（浮动）其他列=值

问题：

• 后续时间之间的时间步长应该相等，但有些太小有些太大

想要的解决方案：

• 插入缺失的时间步（有一定的容差）编辑：找到了解决这个问题的方法（见下文），但肯定有更好的解决方案......
• 删除太小的时间步长 (-"-) 即将这个索引 [0, 0.99, 4, 4.1, 5.1, 6] 转换成这个 [0, 0.99, 2, 3, 4, 5.1, 6]
• 在插入新时间步长的位置插入值（编辑：更好的是只插入值并直接插入所需的时间步长）
• ...请不要循环或列表理解的整体:)（尽可能矢量化）

我尝试了以下方法（测试示例 [已编辑，现在有部分解决方案]）：

测试数组描述：

a_test = np.array([[0.0, 1.0, 1.5],[1, 2.0, -2.1],[2., 3.4, -0.6],[3.0, 4.0, 0.1],
                  [6.0, 8.0, 0.5],[7, 9.1, -2.0],[8, 10.3, -1.0],[8.3, 11, 0.5],
                  [11, 12.3, 1.0],[12, 15.0, 0.1],[13, 16.0, 0.1]])
dt_ = 1.0  # time step aimed for (and most prevalent)

提取的时间步长

time_steps = np.diff(a_test[:,0])
print(time_steps)

[1.  1.  1.  3.  1.  1.  0.3 2.7 1.  1. ]

测试数组（缩短-1）：

a_test = np.delete(a_test, -1, 0)
print(a_test)

[[ 0.   1.   1.5]
 [ 1.   2.  -2.1]
 [ 2.   3.4 -0.6]
 [ 3.   4.   0.1]
 [ 6.   8.   0.5]
 [ 7.   9.1 -2. ]
 [ 8.  10.3 -1. ]
 [ 8.3 11.   0.5]
 [11.  12.3  1. ]
 [12.  15.   0.1]]

具有 True 的布尔数组，其中错误和 列出错误时间步长的索引：

bool_bad_indexes = (time_steps > 2.5) | (time_steps < 0.5)

li_bad_only = list(np.nonzero(bool_bad_indexes)[0] + 1)
print(li_bad_only)

[4, 7, 8]

在错误时间步开始后插入 nan 行（编辑：现在取决于适合的数字）：

def insert_where_missing(a_test, time_steps, dt_):
    bool_bad_indexes = (time_steps > 2.5) | (time_steps < 0.5)
    arr_bad_only = np.nonzero(bool_bad_indexes)[0] + 1
    arr_factors = time_steps[np.nonzero(bool_bad_indexes)[0]] // dt_
    arr_factors = arr_factors.astype(int)
    li_multiple_insert = list(np.repeat(arr_bad_only, arr_factors))
    a_test = np.insert(a_test, li_multiple_insert, np.nan, axis=0)

    return a_test

a_test = np.apply_along_axis(insert_where_missing, 0, a_test, time_steps, dt_)
print(a_test)

[[ 0.   1.   1.5]
 [ 1.   2.  -2.1]
 [ 2.   3.4 -0.6]
 [ 3.   4.   0.1]
 [ nan  nan  nan]
 [ nan  nan  nan]
 [ nan  nan  nan]
 [ 6.   8.   0.5]
 [ 7.   9.1 -2. ]
 [ 8.  10.3 -1. ]
 [ 8.3 11.   0.5]
 [ nan  nan  nan]
 [ nan  nan  nan]
 [11.  12.3  1. ]
 [12.  15.   0.1]]

插入nan的函数：

def interpolate_where_nan(a_test):
    idx_nan = np.isnan(a_test)
    idx_ok = np.logical_not(idx_nan)
    a_test_ok = a_test[idx_ok]
    interpolated = np.interp(idx_nan.nonzero()[0], idx_ok.nonzero()[0], a_test_ok)
    a_test[idx_nan] = interpolated
    return a_test

a_test = np.apply_along_axis(interpolate_where_nan, 0, a_test)
print(a_test)

结果（编辑：只删除小步骤缺失）：

[[ 0.          1.          1.5       ]
 [ 1.          2.         -2.1       ]
 [ 2.          3.4        -0.6       ]
 [ 3.          4.          0.1       ]
 [ 3.75        5.          0.2       ]
 [ 4.5         6.          0.3       ]
 [ 5.25        7.          0.4       ]
 [ 6.          8.          0.5       ]
 [ 7.          9.1        -2.        ]
 [ 8.         10.3        -1.        ]
 [ 8.3        11.          0.5       ]
 [ 9.2        11.43333333  0.66666667]
 [10.1        11.86666667  0.83333333]
 [11.         12.3         1.        ]
 [12.         15.          0.1       ]]

Answer 1

如果您被允许使用它们，那么已经有重采样算法的实现，我使用 scipy 中的以下内容：

https://docs.scipy.org/doc/scipy/reference/generated/scipy.interpolate.interp1d.html

代码如下：

import numpy as np
import matplotlib.pyplot as plt

from scipy.interpolate import interp1d  

a_test = np.array([[0.0, 1.0, 1.5],[1, 2.0, -2.1],[2., 3.4, -0.6],[3.0, 4.0, 0.1],
                  [6.0, 8.0, 0.5],[7, 9.1, -2.0],[8, 10.3, -1.0],[8.3, 11, 0.5],
                  [11, 12.3, 1.0],[12, 15.0, 0.1],[13, 16.0, 0.1]]) 

times = a_test[:, 0]
values = a_test[:, 1]

F = interp1d(times,values,fill_value='extrapolate') 
times_resample = np.linspace(times.min(), times.max(), 100)
values_resample = F(times_resample)

## numpy equivalent:

values_np = np.interp(times_resample, times, values)

## ploting the data
fig, axs = plt.subplots(1, figsize=(10,10))
axs.plot(values,times, 'o')
axs.plot(values_resample,times_resample, )
axs.plot(values_np,times_resample, '--')

Answer 2

抱歉没有解释和新的例子

print("A test array")
a_test = np.array([[0.0, 1.0, 1.5],[1, 2.0, -2.1],[2., 3.4, -0.6],[3.0, 4.0, 0.1],
                  [6.0, 8.0, 0.5],[7, 9.1, -2.0],[7.1, 10.3, -1.0],[8, 11, 0.5],
                  [9, 12.3, 1.0],[11, 15.0, 0.1],[12, 16.0, 0.1],[30, 1.0, 0.9],
                  [31, 2.3, 0.5],[32, 3.0, -0.1],[34, 4.5, 0.0],[35, 6.0, 0.3],
                  [50, 1.3, 0.9],[51, 2.0, -2.1],[52, 3.5, 1.3],[53, 4.0, 1.1]])
dt_ = 1.0  # time step aimed for (and most prevalent)
print(a_test)

def resample_function(a_test, time_steps):

    def insert_where_possible(a_test, time_steps, dt_):
        bool_big_indexes = time_steps > 1.5
        arr_big_only = np.nonzero(bool_big_indexes)[0] + 1
        arr_big_factors = time_steps[np.nonzero(bool_big_indexes)[0]] // (dt_/0.99)
        arr_big_factors = arr_big_factors.astype(int)
        li_big_insert = list(np.repeat(arr_big_only, arr_big_factors))
        a_test = np.insert(a_test, li_big_insert, np.nan, axis=0)
        return a_test
    a_test = np.apply_along_axis(insert_where_possible, 0, a_test, time_steps, dt_)
    
    def interpolate_where_nan(a_test):
        idx_nan = np.isnan(a_test)
        idx_ok = np.logical_not(idx_nan)
        a_test_ok = a_test[idx_ok]
        interpolated = np.interp(idx_nan.nonzero()[0], idx_ok.nonzero()[0], a_test_ok)
        a_test[idx_nan] = interpolated
        return a_test
    a_test = np.apply_along_axis(interpolate_where_nan, 0, a_test)
    time_steps = np.diff(a_test[:,0])

    def delete_tiny_dt(a_test, time_steps):    
        bool_tiny_indexes = time_steps < 0.5
        li_tiny_only = list(np.nonzero(bool_tiny_indexes)[0] + 1)
        a_test = np.delete(a_test, li_tiny_only, axis=0)
        return a_test
    a_test = np.apply_along_axis(delete_tiny_dt, 0, a_test, time_steps)
    
    return a_test

    
time_steps = np.diff(a_test[:,0])

bool_huge_dt = time_steps > 5
    
    
if np.any(bool_huge_dt):
    idx_huge_only = np.nonzero(bool_huge_dt)[0] + 1
    a_a_test = np.split(a_test, idx_huge_only)

    for a_test in a_a_test:
        a_test[:,0] = a_test[:,0] - a_test[0,0]
        time_steps = np.diff(a_test[:,0])
        
        a_test = resample_function(a_test, time_steps)
        
        print("\nResult: array with new time steps and interpolated sensor values in gaps")
        print(a_test)
        
else:
    a_test = resample_function(a_test, time_steps)
    print("\nResult: array with new time steps and interpolated sensor values in gaps")
    print(a_test)

A test array
[[ 0.   1.   1.5]
 [ 1.   2.  -2.1]
 [ 2.   3.4 -0.6]
 [ 3.   4.   0.1]
 [ 6.   8.   0.5]
 [ 7.   9.1 -2. ]
 [ 7.1 10.3 -1. ]
 [ 8.  11.   0.5]
 [ 9.  12.3  1. ]
 [11.  15.   0.1]
 [12.  16.   0.1]
 [30.   1.   0.9]
 [31.   2.3  0.5]
 [32.   3.  -0.1]
 [34.   4.5  0. ]
 [35.   6.   0.3]
 [50.   1.3  0.9]
 [51.   2.  -2.1]
 [52.   3.5  1.3]
 [53.   4.   1.1]]

Result: array with new time steps and interpolated sensor values in gaps
[[ 0.          1.          1.5       ]
 [ 1.          2.         -2.1       ]
 [ 2.          3.4        -0.6       ]
 [ 3.          4.          0.1       ]
 [ 4.          5.33333333  0.23333333]
 [ 5.          6.66666667  0.36666667]
 [ 6.          8.          0.5       ]
 [ 7.          9.1        -2.        ]
 [ 8.         11.          0.5       ]
 [ 9.         12.3         1.        ]
 [10.         13.65        0.55      ]
 [11.         15.          0.1       ]
 [12.         16.          0.1       ]]

Result: array with new time steps and interpolated sensor values in gaps
[[ 0.    1.    0.9 ]
 [ 1.    2.3   0.5 ]
 [ 2.    3.   -0.1 ]
 [ 3.    3.75 -0.05]
 [ 4.    4.5   0.  ]
 [ 5.    6.    0.3 ]]

Result: array with new time steps and interpolated sensor values in gaps
[[ 0.   1.3  0.9]
 [ 1.   2.  -2.1]
 [ 2.   3.5  1.3]
 [ 3.   4.   1.1]]