如何将非均匀时间序列下采样为精确N个近似等间隔的数据点
推荐实现方案(稳定高效,满足精确N条输出要求)
你提到的第二种锚点匹配思路是工业界最常用的成熟方案,仅需做简单的去重补全优化,就能解决可能出现的重复匹配问题,保证输出数量精确为N,具体步骤如下:
- 提取所有真实时间戳的首尾节点,计算总时间跨度
total_td = 最后一个时间戳 - 第一个时间戳 - 生成N个等间隔的虚拟锚点,锚点序列为
[首时间戳 + i * total_td/(N-1) for i in range(N)] - 将真实时间戳按时间升序排序,对每个锚点用二分查找找到距离最近的真实时间戳
- 去重补全:如果出现多个锚点匹配到同一个真实点的情况,优先保留匹配误差最小的条目,再从相邻未被选中的真实时间戳中补全空缺,直到凑够恰好N条
可直接运行的代码实现
适配你给出的示例数据:
import pandas as pd import numpy as np # 示例输入数据 timestamps = [ (pd.Timestamp('2013-10-05 12:52:00+0000', tz='UTC'), 0), (pd.Timestamp('2013-10-07 18:38:00+0000', tz='UTC'), 1), (pd.Timestamp('2013-10-12 11:30:00+0000', tz='UTC'), 5), (pd.Timestamp('2013-10-13 11:58:00+0000', tz='UTC'), 7), (pd.Timestamp('2013-10-14 17:26:00+0000', tz='UTC'), 11), (pd.Timestamp('2013-10-16 17:54:00+0000', tz='UTC'), 12), (pd.Timestamp('2013-10-17 21:26:00+0000', tz='UTC'), 14), (pd.Timestamp('2013-10-20 13:37:00+0000', tz='UTC'), 17), (pd.Timestamp('2013-10-22 18:16:00+0000', tz='UTC'), 18), (pd.Timestamp('2013-10-25 23:37:00+0000', tz='UTC'), 19), (pd.Timestamp('2013-10-26 13:36:00+0000', tz='UTC'), 20), (pd.Timestamp('2013-10-30 19:11:00+0000', tz='UTC'), 26), (pd.Timestamp('2013-11-07 21:13:00+0000', tz='UTC'), 28), (pd.Timestamp('2013-11-08 13:16:00+0000', tz='UTC'), 29), (pd.Timestamp('2013-11-15 18:19:00+0000', tz='UTC'), 32), (pd.Timestamp('2013-11-16 00:27:00+0000', tz='UTC'), 33), (pd.Timestamp('2013-11-16 18:55:00+0000', tz='UTC'), 35), (pd.Timestamp('2013-12-04 16:58:00+0000', tz='UTC'), 40), (pd.Timestamp('2013-12-18 09:48:00+0000', tz='UTC'), 47), (pd.Timestamp('2013-12-19 08:32:00+0000', tz='UTC'), 50) ] def downsample_timestamps(ts_list, n_target): df = pd.DataFrame(ts_list, columns=['ts', 'value']).sort_values('ts').reset_index(drop=True) origin_count = len(df) if origin_count <= n_target: return ts_list # 生成等间隔虚拟锚点 start_ts = df['ts'].iloc[0] end_ts = df['ts'].iloc[-1] anchors = pd.date_range(start=start_ts, end=end_ts, periods=n_target, tz='UTC') ts_np = df['ts'].view('int64').to_numpy() selected_idx = set() # 匹配每个锚点的最近真实点 for anchor in anchors: anchor_int = anchor.view('int64') pos = np.searchsorted(ts_np, anchor_int) if pos == 0: best_pos = 0 elif pos == origin_count: best_pos = origin_count - 1 else: left_diff = anchor_int - ts_np[pos-1] right_diff = ts_np[pos] - anchor_int best_pos = pos-1 if left_diff <= right_diff else pos selected_idx.add(best_pos) # 补全不足的点位:优先插入间隔最大的空隙 unselected = sorted(set(range(origin_count)) - selected_idx) while len(selected_idx) < n_target: selected_sorted = sorted(selected_idx) max_gap = -1 insert_pos = 0 for i in range(len(selected_sorted)-1): gap = ts_np[selected_sorted[i+1]] - ts_np[selected_sorted[i]] if gap > max_gap: max_gap = gap for u in unselected: if selected_sorted[i] < u < selected_sorted[i+1]: insert_pos = u break selected_idx.add(insert_pos) unselected.remove(insert_pos) # 返回排序后的结果 selected_df = df.iloc[sorted(selected_idx)] return list(selected_df.itertuples(index=False, name=None)) # 测试:20条下采到15条 result = downsample_timestamps(timestamps, 15) print(len(result)) # 输出为15
其他可选方案
如果对间隔均匀性要求极高,且原始数据量M不大(比如M<1000),可以用动态规划方案得到理论最优结果:
- 定义
dp[i][k]为前i个真实点选k个时,最小的最大间隔误差 - 状态转移时枚举上一个选中点的位置,计算间隔偏差,最后回溯得到最优选择序列,时间复杂度为O(M²N)
注意事项
- 不建议使用pandas原生resample方法,它是固定时间间隔分组聚合,不会保留原始时间戳,也无法保证输出恰好N条
- 如果允许微小的时间偏移,也可以用K-Means聚类,把所有时间戳聚成N类,每类取中心时间对应的最近真实点,效果也比较稳定
内容的提问来源于stack exchange,提问作者chess
相关产品推荐
相关产品推荐

