如何判定极值曲线间Y轴距离低于平均距离的设定阈值？

阿华AIGC实验室

2026-5-13

解决极值区间距离阈值判定问题

看起来你已经成功提取了流量数据的局部极值并完成可视化，接下来咱们一步步实现「最大值与最小值曲线间的Y轴距离低于平均距离阈值」的判定需求～

核心思路

因为你提取的极值点X坐标不对应、数量也不一致，直接计算距离会有困难。我们可以通过插值将离散的极值点扩展为与主数据对齐的连续曲线，再逐点（或滑动窗口）计算距离，最后和设定的阈值对比。

步骤1：提取极值点的坐标对

先把已找到的极值点的X（索引）和Y（流量值）单独提取出来，方便后续插值处理：

import numpy as np
from scipy.signal import argrelextrema
from scipy.interpolate import interp1d
import pandas as pd
import matplotlib.pyplot as plt

# 你的原有代码
c_max_index = argrelextrema(df.flow.values, np.greater, order=3)
c_min_index = argrelextrema(df.flow.values, np.less, order=3)
df['min_extreme'] = df.flow[c_min_index[0]]
df['max_extreme'] = df.flow[c_max_index[0]]

# 提取极值点的X和Y坐标
max_x = c_max_index[0]
max_y = df.flow.values[max_x]
min_x = c_min_index[0]
min_y = df.flow.values[min_x]

步骤2：插值生成连续的极值曲线

用线性插值（或非线性插值，比如三次样条）将离散的极值点扩展为覆盖主数据所有索引的连续曲线，这样每个主数据点都能匹配到对应的上下极值估计值：

# 创建插值函数，fill_value="extrapolate"支持超出极值点范围的外插
f_max = interp1d(max_x, max_y, kind='linear', fill_value="extrapolate")
f_min = interp1d(min_x, min_y, kind='linear', fill_value="extrapolate")

# 生成主数据每个索引对应的插值后极值
df['interp_max'] = f_max(df.index)
df['interp_min'] = f_min(df.index)

如果你的数据是非线性趋势，可以把kind改成'cubic'获得更平滑的插值结果；如果不想外插超出极值点范围的数据，可将fill_value设为np.nan，后续用df.dropna()处理缺失值。

步骤3：计算距离并判定阈值

逐点距离判定

计算每个时间点的上下极值距离，再对比「平均距离的10%」阈值：

# 计算每个点的Y轴距离（最大值-最小值）
df['distance'] = df['interp_max'] - df['interp_min']

# 计算整体平均距离
avg_distance = df['distance'].mean()

# 设定阈值（这里是10%，可自行修改比例）
threshold_ratio = 0.1
threshold = threshold_ratio * avg_distance

# 标记低于阈值的点
df['below_threshold'] = df['distance'] < threshold

滑动窗口积分判定（对应你提到的「积分」需求）

如果想判断**一段区间内的总距离（积分）**是否低于阈值，可以用滑动窗口计算积分（即窗口内距离的累加和）：

# 设定滑动窗口大小（根据你的数据频率调整，比如10个时间步）
window_size = 10

# 计算窗口内的距离积分
df['rolling_integral'] = df['distance'].rolling(window=window_size).sum()

# 计算积分的平均值，再生成阈值
avg_integral = df['rolling_integral'].mean()
integral_threshold = threshold_ratio * avg_integral

# 标记积分低于阈值的区间
df['integral_below_threshold'] = df['rolling_integral'] < integral_threshold

步骤4：可视化验证

将结果叠加到原有图表上，直观查看阈值判定的效果：

plt.figure(figsize=(12,6))
plt.plot(df.flow.values, label='Main Flow', c='blue')
plt.plot(max_x, max_y, linewidth=0.8, c='g', label='Max Extremes')
plt.plot(min_x, min_y, linewidth=0.8, c='r', label='Min Extremes')

# 标记逐点低于阈值的点
plt.scatter(df[df['below_threshold']].index, df[df['below_threshold']].flow, 
            c='orange', s=15, alpha=0.6, label='Distance < 10% Avg')

# （可选）标记积分低于阈值的区间
plt.fill_between(df.index, df.flow.min(), df.flow.max(), 
                 where=df['integral_below_threshold'], 
                 color='yellow', alpha=0.2, label='Integral < 10% Avg')

plt.legend()
plt.xlabel('Index')
plt.ylabel('Flow')
plt.title('Flow Data with Extreme Value Threshold Detection')
plt.show()