沿Y轴裁剪时间序列数据的技术实现问询

需求说明 #

我有一个由(datetime, float)元组组成的有序列表，所有datetime唯一且已按顺序排序，每个元组包含一个datetime对象和一个浮点数值。需要将浮点值裁剪到指定阈值以上，示例如下：

原数据 #

import datetime
from datetime import tzutc

a = [
    (datetime.datetime(2021, 11, 1, 0, 0, tzinfo=tzutc()), 100), 
    (datetime.datetime(2021, 11, 1, 1, 0, tzinfo=tzutc()), 9.0), 
    (datetime.datetime(2021, 11, 1, 2, 0, tzinfo=tzutc()), 100.0)
]

期望结果（阈值=10.0） #

b = [
    (datetime.datetime(2021, 11, 1, 0, 0, tzinfo=tzutc()), 100), 
    (datetime.datetime(2021, 11, 1, 0, 36, tzinfo=tzutc()), 10.0),  # 线性插值计算出的交点时间
    (datetime.datetime(2021, 11, 1, 1, 24, tzinfo=tzutc()), 10.0),  # 线性插值计算出的交点时间
    (datetime.datetime(2021, 11, 1, 2, 0, tzinfo=tzutc()), 100.0)
]

原数据绘图呈V形，裁剪后需变为_/形。希望找到现成实现方案，也可接受转为等时间间隔数据的处理方式。

现成工具实现（基于Pandas） #

Pandas是处理时间序列的常用工具，可以高效完成这个需求：

1. 精准计算交点的方案 #

通过线性插值计算阈值与曲线的交点，保留原始数据的关键节点：

import pandas as pd

df = pd.DataFrame(a, columns=['dt', 'value']).set_index('dt')
threshold = 10.0
result = []

# 加入第一个点
result.append((df.index[0], df['value'].iloc[0]))

# 遍历相邻点对处理
for i in range(1, len(df)):
    t_prev, v_prev = df.index[i-1], df['value'].iloc[i-1]
    t_curr, v_curr = df.index[i], df['value'].iloc[i]
    
    # 前点高于阈值，后点低于阈值：插入交点+后点阈值
    if v_prev > threshold and v_curr < threshold:
        delta_t = t_curr - t_prev
        delta_v = v_curr - v_prev
        t_cross = t_prev + delta_t * (threshold - v_prev) / delta_v
        result.append((t_cross, threshold))
        result.append((t_curr, threshold))
    # 前点低于阈值，后点高于阈值：插入交点+后点原值
    elif v_prev < threshold and v_curr > threshold:
        delta_t = t_curr - t_prev
        delta_v = v_curr - v_prev
        t_cross = t_prev + delta_t * (threshold - v_prev) / delta_v
        result.append((t_cross, threshold))
        result.append((t_curr, v_curr))
    # 两点都高于阈值：直接加入当前点
    elif v_prev >= threshold and v_curr >= threshold:
        result.append((t_curr, v_curr))
    # 两点都低于阈值：加入当前点的阈值替换值
    else:
        result.append((t_curr, threshold))

运行后result就是符合要求的序列，交点时间为精确计算值。

2. 等时间间隔简化方案 #

如果可以接受转为等时间间隔数据，处理流程更简洁：

# 生成每15分钟间隔的时间序列（可自定义频率，如'1min'/'1h'）
new_index = pd.date_range(start=df.index[0], end=df.index[-1], freq='15min')
# 时间插值补全空缺数据
df_resampled = df.reindex(new_index).interpolate(method='time')
# 裁剪低于阈值的值，替换为阈值
df_clipped = df_resampled.where(df_resampled['value'] >= threshold, threshold)
# 转回元组列表格式
result = list(zip(df_clipped.index, df_clipped['value']))

这种方式无需计算精确交点，通过等间隔插值后直接替换阈值，绘图效果完全符合需求。

无依赖手动实现 #

如果不想依赖第三方库，纯Python也可以实现相同逻辑：

import datetime
from datetime import tzutc

threshold = 10.0
a = [
    (datetime.datetime(2021, 11, 1, 0, 0, tzinfo=tzutc()), 100), 
    (datetime.datetime(2021, 11, 1, 1, 0, tzinfo=tzutc()), 9.0), 
    (datetime.datetime(2021, 11, 1, 2, 0, tzinfo=tzutc()), 100.0)
]

result = [a[0]]

for i in range(1, len(a)):
    t_prev, v_prev = a[i-1]
    t_curr, v_curr = a[i]
    
    if v_prev > threshold and v_curr < threshold:
        # 计算时间差并插值交点
        total_seconds = (t_curr - t_prev).total_seconds()
        ratio = (threshold - v_prev) / (v_curr - v_prev)
        cross_seconds = total_seconds * ratio
        t_cross = t_prev + datetime.timedelta(seconds=cross_seconds)
        result.append((t_cross, threshold))
        result.append((t_curr, threshold))
    elif v_prev < threshold and v_curr > threshold:
        total_seconds = (t_curr - t_prev).total_seconds()
        ratio = (threshold - v_prev) / (v_curr - v_prev)
        cross_seconds = total_seconds * ratio
        t_cross = t_prev + datetime.timedelta(seconds=cross_seconds)
        result.append((t_cross, threshold))
        result.append((t_curr, v_curr))
    elif v_prev >= threshold and v_curr >= threshold:
        result.append((t_curr, v_curr))
    else:
        result.append((t_curr, threshold))

逻辑与Pandas方案一致，适合轻量场景使用。

内容的提问来源于stack exchange，提问作者Baz