所有文章 标签 分类
简体中文

目录

目录

时序数据降采样：LTTB 与 MinMaxLTTB

MartinLwx  收录于  类别 数据分析
     约 2285 字   预计阅读 11 分钟 

最近在工作中需要处理一些时序数据，需要对时序数据做降采样（Downsampling）处理。因此学习了一下 LTTB 算法和 MinMaxLTTB 算法

从降采样的技术上来说，大致可以分成 2 种1

  • 特征保持（Characteristic Preserving）：此时降采样的目标是让统计特征（均值、方差等）在降采样前后尽量一致
  • 值保持（Value Preserving）：从原始数据点中挑选数据点，目标是让整体形状在降采样前后尽量一致

无论是 LTTB 算法还是 MinMaxLTTB 算法，本质上都是值保持的一种。即他们的目标都是在降采样后尽量维持原来的形状

LTTB 的全名叫做 Largest Triangle Three Buckets，可以拆分开理解

  • Largest Triangle：LTTB 通过让三角形面积最大来确定降采样的点
  • Three Buckets：LTTB 会将所有的数据点分成多个桶，每次看三个桶并分别选一个点，就构成了一个三角形

只看上面的解释的话，很自然会有下面的疑问

  1. 怎么将数据点分成多个桶，标准是什么？
  2. 每个桶是如何选点的？他们的策略一样吗？
  3. 为什么让三角形最大 $\approx$ 降采样后形状近似？

这里的问题 3 在我看来是最有趣的，但很难在对 LTTB 算法没有了解的情况下就讲清楚。至于问题 1 和问题 2 其实都是 LTTB 算法的设计细节。因此，我们先看 LTTB 是如何工作的，然后再回答这些问题

输入：按 x 升序的 $N$ 个数据点 $p_1, p_2, \ldots, p_N$（每点含坐标 $x$、$y$），目标点数 $M$（$2 < M < N$）

输出：降采样后的 $M$ 个点，总是包含首点 $p_1$ 与末点 $p_N$，按原顺序排列

算法处理流程：

  1. 边界处理：若 $M \ge N$，直接返回全部 $N$ 个点；若 $M \le 2$，返回 ${p_1, p_N}$
  2. 选中 $p_1$ 作为第一个采样点
  3. 将中间 $N - 2$ 个点划分为 $M - 2$ 个桶，桶大小 $B = \frac{N - 2}{M - 2}$
  4. 对每个桶 $i = 0, 1, \ldots, M-3$ 依次处理：
    • 计算下一桶（桶 $i+1$）内所有点的平均点 $\bar{p} = (\bar{x}, \bar{y})$
    • 记 $a$ 为上一轮选中的采样点（初始为 $p_1$）；
    • 遍历桶 $i$ 内每个点 $p_j$，按叉积公式计算其与 $a$、$\bar{p}$ 构成的三角形面积 $S_j = \frac{1}{2}\left|(x_a - \bar{x})(y_j - y_a) - (x_a - x_j)(\bar{y} - y_a)\right|$，将 $S_j$ 最大的点加入输出
    • 令 $a$ 指向该点（这样处理下一个桶的时候就知道上一轮选了什么点）
  5. 追加 $p_N$ 为最后一个采样点

现在我们就清楚了前面的问题 1 和问题 2 的答案

  1. 总是保留首点和末点，剩下的 $N-2$ 个数据点放到 $M-2$ 个桶里
  2. 枚举当前桶的所有可用数据点，找能构成最大三角形的点。这里依赖项包括
    • 上一个桶选择的点
    • 下一个桶的“平均点”。算法直觉是：平均点是对未来（下一个桶）数据点的趋势的近似

那么，现在只剩下问题 3：为什么一定要选择让三角形面积最大的数据点？关于这个问题的答案，我这里不打算给出一个严格的证明，而是想给你建立一个算法直觉————我们知道，降采样后形状尽量相似的要点在于：要尽量保持极值存在，波动不大的点可以舍弃掉，如果我们把下图的 $v_1, v_3$ 固定，当作三角形的底边的话，那么极值点往往都是“高”会比较大的点，也就是说，极值点就是容易让三角形的面积更大。反过来说，如果我们要降采样之后形状尽量不变，尽量让三角形最大即可

那么，LTTB 算法的时间复杂度是多少呢？它只需要按顺序处理一遍，答案是显而易见的

$$ O(N) $$

最后，我们可以看一下 Python 的代码实现，整体结构还是比较清晰的

首先，定义求三角形面积的辅助函数，以及定义 Point 类型

Point = tuple[int, int]

def triangle_area(p1: Point, p2: Point, p3: Point):
    return 0.5 * abs(
        (p2[0] - p1[0]) * (p3[1] - p1[1]) - (p2[1] - p1[1]) * (p3[0] - p1[0])
    )

然后开始写 LTTB 算法主体

def lttb(points: list[Point], target_node_cnt: int):
    if len(points) <= 2:
        return points

    ret: list[Point] = [points[0]]
    bucket_size = (len(points) - 2) / (target_node_cnt - 2)

    for i in range(target_node_cnt - 2):
        # Find the average of the next bucket
        next_bucket_start = int((i + 1) * bucket_size) + 1
        next_bucket_end = min(int((i + 2) * bucket_size + 1), len(points))
        next_bucket_size = next_bucket_end - next_bucket_start
        avg_x, avg_y = 0.0, 0.0
        for ni in range(next_bucket_start, next_bucket_end):
            avg_x += points[ni][0]
            avg_y += points[ni][1]
        avg_x, avg_y = avg_x / next_bucket_size, avg_y / next_bucket_size

        # Find best node in current bucket
        cur_bucket_start = int(i * bucket_size) + 1
        cur_bucket_end = int((i + 1) * bucket_size) + 1
        max_area, best_choice = 0, 0
        for ni in range(cur_bucket_start, cur_bucket_end):
            area = triangle_area(ret[-1], points[ni], (avg_x, avg_y))
            if area > max_area:
                max_area, best_choice = area, ni
        ret.append(points[best_choice])

    ret.append(points[-1])

    return ret

现在我们清楚了 LTTB 算法，那 MinMaxLTTB 算法又是什么？其实答案就在名字上，它先用 MinMax 策略预选择一些极值点，然后只在这些极值点上跑 LTTB 算法。所以核心是 MinMaxLTTB 是如何怎么预选极值点的

它的策略是这样子的，确定一个比例 $ratio$（通常是 2 的倍数），然后将原始 $N$ 个数据点分到 $(ratio/2)*M$ 个桶里（注意，首点和末点总是被选中）。这里 $/2$ 是因为对于每一个桶，我们要挑选最小值、最大值。从另外一个角度来说，它等价于将 LTTB 算法细分了 $ratio/2$ 个桶（注意这里的倍率关系），每个桶挑选极值

对于 MinMaxLTTB 来说，降采样数据点的个数变化是这样子的

$$ N \xrightarrow{\text{MinMax Preselection}} (ratio/2) \cdot M \xrightarrow{\text{Standard LTTB}} M $$

那么，它的收益是什么呢？体现在 2 个地方1

  • 预选极值点是可以并行处理的，因为每个桶挑选极值的时候都是独立的
  • LTTB 算法要处理的数据规模小得多（$(ratio/2)*M\ll N$）

同样的，这里给出参考代码，首先是预选极值点的逻辑

def pre_selection(
    points: list[Point],
    target_node_cnt: int,
    ratio: int,
):
    if len(points) <= 2:
        return points

    if ratio < 2 or ratio % 2 == 1:
        raise ValueError(f"Invalid ratio: {ratio}")

    ret: list[Point] = [points[0]]

    num_partitions = (target_node_cnt * ratio - 2) // 2
    if num_partitions > len(points) - 2:
        raise ValueError(
            f"Too many partitions ({num_partitions}) for nodes ({len(points) - 2})"
        )

    partition_size = (len(points) - 2) / num_partitions

    for i in range(num_partitions):
        range_start = int(i * partition_size) + 1
        range_end = min(int((i + 1) * partition_size) + 1, len(points) - 1)
        min_val, min_idx = float("inf"), -1
        max_val, max_idx = -float("inf"), -1

        for j in range(range_start, range_end):
            if points[j][1] < min_val:
                min_val, min_idx = points[j][1], j

            if points[j][1] > max_val:
                max_val, max_idx = points[j][1], j

        if min_idx > max_idx:
            max_idx, min_idx = min_idx, max_idx

        ret.append(points[min_idx])
        ret.append(points[max_idx])

    ret.append(points[-1])

    assert len(ret) == target_node_cnt * ratio

    return ret

接下来是 MinMaxLTTB 算法的逻辑

def minmax_lttb(
    points: list[Point],
    target_node_cnt: int,
    ratio: int
):
    pre_selected_nodes = pre_selection(points, TARGET_NODE_CNT, ratio=ratio)
    return lttb(pre_selected_nodes, TARGET_NODE_CNT)

接下来，我们用生活中一个真实的时序数据实践一下，我这里选择的是易方达沪深 300 ETF 联接 A。截至基金成立日到本文发表的今天，它一共有 4138 个净值，我们的目标是将其降采样到 42 个。这个规模当然不算大，但我们仍然能够从中体会一下 LTTB 算法和 MinMaxLTTB 算法在降采样上的差异。完整的代码在这里

从上图不难看出

  • LTTB 算法和 MinMaxLTTB 算法效果相近，在保持整体形状上效果都还可以
  • 不管是哪一种算法，都不保证保留极值
  • $ratio$ 越大，MinMaxLTTB 算法降采样的结果就跟 LTTB 越近似。从算法原理也可以理解这点，因为预选的极值点更多了，极端情况就是全都选上，此时就跟 LTTB 算法一样了

  1. Van Der Donckt J, Van Der Donckt J, Rademaker M, et al. Minmaxlttb: Leveraging minmax-preselection to scale lttb[C]//2023 IEEE Visualization and Visual Analytics (VIS). IEEE, 2023: 21-25. ↩︎ ↩︎

更新于 2026-09-03
 时序数据处理
 | 主页