Pandas：根据局部minima-maxima对数据进行Zigzag分割
Pandas: Zigzag segmentation of data based on local minima-maxima
我有一个时间序列数据。生成数据
date_rng = pd.date_range('2019-01-01', freq='s', periods=400)
df = pd.DataFrame(np.random.lognormal(.005, .5,size=(len(date_rng), 3)),
columns=['data1', 'data2', 'data3'],
index= date_rng)
s = df['data1']
我想创建一条连接局部最大值和局部最小值的 zig-zag 线，满足每个 zig-zag 的 y-axis、|highest - lowest value| 的条件线必须超过前一个 zig-zag 线距离的百分比（比如 20%），并且 pre-stated 值 k（比如 1.2）
我可以使用此代码找到局部极值：
# Find peaks(max).
peak_indexes = signal.argrelextrema(s.values, np.greater)
peak_indexes = peak_indexes[0]
# Find valleys(min).
valley_indexes = signal.argrelextrema(s.values, np.less)
valley_indexes = valley_indexes[0]
# Merge peaks and valleys data points using pandas.
df_peaks = pd.DataFrame({'date': s.index[peak_indexes], 'zigzag_y': s[peak_indexes]})
df_valleys = pd.DataFrame({'date': s.index[valley_indexes], 'zigzag_y': s[valley_indexes]})
df_peaks_valleys = pd.concat([df_peaks, df_valleys], axis=0, ignore_index=True, sort=True)
# Sort peak and valley datapoints by date.
df_peaks_valleys = df_peaks_valleys.sort_values(by=['date'])
但我不知道如何对其应用阈值条件。
请告诉我如何申请这样的条件。
由于数据可能包含百万个时间戳，因此强烈建议进行高效计算
为了更清楚的描述：
示例输出，来自我的数据：
# Instantiate axes.
(fig, ax) = plt.subplots()
# Plot zigzag trendline.
ax.plot(df_peaks_valleys['date'].values, df_peaks_valleys['zigzag_y'].values,
color='red', label="Zigzag")
# Plot original line.
ax.plot(s.index, s, linestyle='dashed', color='black', label="Org. line", linewidth=1)
# Format time.
ax.xaxis_date()
ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m-%d"))
plt.gcf().autofmt_xdate() # Beautify the x-labels
plt.autoscale(tight=True)
plt.legend(loc='best')
plt.grid(True, linestyle='dashed')
我想要的输出（类似于此，之字形只连接重要的部分）
我已经回答了我对问题的最佳理解。然而，变量K如何影响滤波器还不清楚。
您想根据 运行ning 条件过滤极值。我假设您想标记与最后一个 标记的 极值的相对距离大于 p% 的所有极值。我进一步假设您总是将时间序列的第一个元素视为 valid/relevant 点。
我用以下过滤函数实现了这个：
def filter(values, percentage):
previous = values[0]
mask = [True]
for value in values[1:]:
relative_difference = np.abs(value - previous)/previous
if relative_difference > percentage:
previous = value
mask.append(True)
else:
mask.append(False)
return mask
为了运行你的代码，我先导入依赖：
from scipy import signal
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
为了使代码可重现，我修复了随机种子：
np.random.seed(0)
剩下的就是copypasta。请注意，我减少了样本量以使结果更清晰。
date_rng = pd.date_range('2019-01-01', freq='s', periods=30)
df = pd.DataFrame(np.random.lognormal(.005, .5,size=(len(date_rng), 3)),
columns=['data1', 'data2', 'data3'],
index= date_rng)
s = df['data1']
# Find peaks(max).
peak_indexes = signal.argrelextrema(s.values, np.greater)
peak_indexes = peak_indexes[0]
# Find valleys(min).
valley_indexes = signal.argrelextrema(s.values, np.less)
valley_indexes = valley_indexes[0]
# Merge peaks and valleys data points using pandas.
df_peaks = pd.DataFrame({'date': s.index[peak_indexes], 'zigzag_y': s[peak_indexes]})
df_valleys = pd.DataFrame({'date': s.index[valley_indexes], 'zigzag_y': s[valley_indexes]})
df_peaks_valleys = pd.concat([df_peaks, df_valleys], axis=0, ignore_index=True, sort=True)
# Sort peak and valley datapoints by date.
df_peaks_valleys = df_peaks_valleys.sort_values(by=['date'])
然后我们使用过滤功能：
p = 0.2 # 20%
filter_mask = filter(df_peaks_valleys.zigzag_y, p)
filtered = df_peaks_valleys[filter_mask]
并像您之前绘制的那样绘制以及新过滤的极值：
# Instantiate axes.
(fig, ax) = plt.subplots(figsize=(10,10))
# Plot zigzag trendline.
ax.plot(df_peaks_valleys['date'].values, df_peaks_valleys['zigzag_y'].values,
color='red', label="Extrema")
# Plot zigzag trendline.
ax.plot(filtered['date'].values, filtered['zigzag_y'].values,
color='blue', label="ZigZag")
# Plot original line.
ax.plot(s.index, s, linestyle='dashed', color='black', label="Org. line", linewidth=1)
# Format time.
ax.xaxis_date()
ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m-%d"))
plt.gcf().autofmt_xdate() # Beautify the x-labels
plt.autoscale(tight=True)
plt.legend(loc='best')
plt.grid(True, linestyle='dashed')
编辑:
如果想同时考虑第一个和最后一个点都有效，那么您可以按如下方式调整过滤器功能：
def filter(values, percentage):
# the first value is always valid
previous = values[0]
mask = [True]
# evaluate all points from the second to (n-1)th
for value in values[1:-1]:
relative_difference = np.abs(value - previous)/previous
if relative_difference > percentage:
previous = value
mask.append(True)
else:
mask.append(False)
# the last value is always valid
mask.append(True)
return mask
您可以使用 Pandas 滚动功能来创建局部极值。与您的 Scipy 方法相比，这稍微简化了代码。
寻找极值的函数：
def islocalmax(x):
"""Both neighbors are lower,
assumes a centered window of size 3"""
return (x[0] < x[1]) & (x[2] < x[1])
def islocalmin(x):
"""Both neighbors are higher,
assumes a centered window of size 3"""
return (x[0] > x[1]) & (x[2] > x[1])
def isextrema(x):
return islocalmax(x) or islocalmin(x)
创建锯齿形的函数，它可以一次应用于 Dataframe（在每一列上），但这将引入 NaN，因为返回的时间戳对于每一列都是不同的。您可以稍后轻松删除这些，如下例所示，或者只需将函数应用于 Dataframe 中的单个列。
请注意，我取消了针对阈值 k 的测试的注释，我不确定是否完全正确地理解了该部分。如果前一个和当前极值之间的绝对差需要大于 k，则可以包括它：& (ext_val.diff().abs() > k)
我也不确定最终之字形是否应该始终从原始高点移动到低点，或者反之亦然。我认为应该如此，否则您可以删除函数末尾的第二次 extreme 搜索。
def create_zigzag(col, p=0.2, k=1.2):
# Find the local min/max
# converting to bool converts NaN to True, which makes it include the endpoints
ext_loc = col.rolling(3, center=True).apply(isextrema, raw=False).astype(np.bool_)
# extract values at local min/max
ext_val = col[ext_loc]
# filter locations based on threshold
thres_ext_loc = (ext_val.diff().abs() > (ext_val.shift(-1).abs() * p)) #& (ext_val.diff().abs() > k)
# Keep the endpoints
thres_ext_loc.iloc[0] = True
thres_ext_loc.iloc[-1] = True
thres_ext_loc = thres_ext_loc[thres_ext_loc]
# extract values at filtered locations
thres_ext_val = col.loc[thres_ext_loc.index]
# again search the extrema to force the zigzag to always go from high > low or vice versa,
# never low > low, or high > high
ext_loc = thres_ext_val.rolling(3, center=True).apply(isextrema, raw=False).astype(np.bool_)
thres_ext_val =thres_ext_val[ext_loc]
return thres_ext_val
生成一些示例数据：
date_rng = pd.date_range('2019-01-01', freq='s', periods=35)
df = pd.DataFrame(np.random.randn(len(date_rng), 3),
columns=['data1', 'data2', 'data3'],
index= date_rng)
df = df.cumsum()
应用函数并提取 'data1' 列的结果：
dfzigzag = df.apply(create_zigzag)
data1_zigzag = dfzigzag['data1'].dropna()
可视化结果：
fig, axs = plt.subplots(figsize=(10, 3))
axs.plot(df.data1, 'ko-', ms=4, label='original')
axs.plot(data1_zigzag, 'ro-', ms=4, label='zigzag')
axs.legend()
我有一个时间序列数据。生成数据
date_rng = pd.date_range('2019-01-01', freq='s', periods=400)
df = pd.DataFrame(np.random.lognormal(.005, .5,size=(len(date_rng), 3)),
columns=['data1', 'data2', 'data3'],
index= date_rng)
s = df['data1']
我想创建一条连接局部最大值和局部最小值的 zig-zag 线，满足每个 zig-zag 的 y-axis、|highest - lowest value| 的条件线必须超过前一个 zig-zag 线距离的百分比（比如 20%），并且 pre-stated 值 k（比如 1.2）
我可以使用此代码找到局部极值：
# Find peaks(max).
peak_indexes = signal.argrelextrema(s.values, np.greater)
peak_indexes = peak_indexes[0]
# Find valleys(min).
valley_indexes = signal.argrelextrema(s.values, np.less)
valley_indexes = valley_indexes[0]
# Merge peaks and valleys data points using pandas.
df_peaks = pd.DataFrame({'date': s.index[peak_indexes], 'zigzag_y': s[peak_indexes]})
df_valleys = pd.DataFrame({'date': s.index[valley_indexes], 'zigzag_y': s[valley_indexes]})
df_peaks_valleys = pd.concat([df_peaks, df_valleys], axis=0, ignore_index=True, sort=True)
# Sort peak and valley datapoints by date.
df_peaks_valleys = df_peaks_valleys.sort_values(by=['date'])
但我不知道如何对其应用阈值条件。 请告诉我如何申请这样的条件。
由于数据可能包含百万个时间戳，因此强烈建议进行高效计算
为了更清楚的描述：
示例输出，来自我的数据：
# Instantiate axes.
(fig, ax) = plt.subplots()
# Plot zigzag trendline.
ax.plot(df_peaks_valleys['date'].values, df_peaks_valleys['zigzag_y'].values,
color='red', label="Zigzag")
# Plot original line.
ax.plot(s.index, s, linestyle='dashed', color='black', label="Org. line", linewidth=1)
# Format time.
ax.xaxis_date()
ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m-%d"))
plt.gcf().autofmt_xdate() # Beautify the x-labels
plt.autoscale(tight=True)
plt.legend(loc='best')
plt.grid(True, linestyle='dashed')
我想要的输出（类似于此，之字形只连接重要的部分）
我已经回答了我对问题的最佳理解。然而，变量K如何影响滤波器还不清楚。
您想根据 运行ning 条件过滤极值。我假设您想标记与最后一个 标记的 极值的相对距离大于 p% 的所有极值。我进一步假设您总是将时间序列的第一个元素视为 valid/relevant 点。
我用以下过滤函数实现了这个：
def filter(values, percentage):
previous = values[0]
mask = [True]
for value in values[1:]:
relative_difference = np.abs(value - previous)/previous
if relative_difference > percentage:
previous = value
mask.append(True)
else:
mask.append(False)
return mask
为了运行你的代码，我先导入依赖：
from scipy import signal
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
为了使代码可重现，我修复了随机种子：
np.random.seed(0)
剩下的就是copypasta。请注意，我减少了样本量以使结果更清晰。
date_rng = pd.date_range('2019-01-01', freq='s', periods=30)
df = pd.DataFrame(np.random.lognormal(.005, .5,size=(len(date_rng), 3)),
columns=['data1', 'data2', 'data3'],
index= date_rng)
s = df['data1']
# Find peaks(max).
peak_indexes = signal.argrelextrema(s.values, np.greater)
peak_indexes = peak_indexes[0]
# Find valleys(min).
valley_indexes = signal.argrelextrema(s.values, np.less)
valley_indexes = valley_indexes[0]
# Merge peaks and valleys data points using pandas.
df_peaks = pd.DataFrame({'date': s.index[peak_indexes], 'zigzag_y': s[peak_indexes]})
df_valleys = pd.DataFrame({'date': s.index[valley_indexes], 'zigzag_y': s[valley_indexes]})
df_peaks_valleys = pd.concat([df_peaks, df_valleys], axis=0, ignore_index=True, sort=True)
# Sort peak and valley datapoints by date.
df_peaks_valleys = df_peaks_valleys.sort_values(by=['date'])
然后我们使用过滤功能：
p = 0.2 # 20%
filter_mask = filter(df_peaks_valleys.zigzag_y, p)
filtered = df_peaks_valleys[filter_mask]
并像您之前绘制的那样绘制以及新过滤的极值：
# Instantiate axes.
(fig, ax) = plt.subplots(figsize=(10,10))
# Plot zigzag trendline.
ax.plot(df_peaks_valleys['date'].values, df_peaks_valleys['zigzag_y'].values,
color='red', label="Extrema")
# Plot zigzag trendline.
ax.plot(filtered['date'].values, filtered['zigzag_y'].values,
color='blue', label="ZigZag")
# Plot original line.
ax.plot(s.index, s, linestyle='dashed', color='black', label="Org. line", linewidth=1)
# Format time.
ax.xaxis_date()
ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m-%d"))
plt.gcf().autofmt_xdate() # Beautify the x-labels
plt.autoscale(tight=True)
plt.legend(loc='best')
plt.grid(True, linestyle='dashed')
编辑:
如果想同时考虑第一个和最后一个点都有效，那么您可以按如下方式调整过滤器功能：
def filter(values, percentage):
# the first value is always valid
previous = values[0]
mask = [True]
# evaluate all points from the second to (n-1)th
for value in values[1:-1]:
relative_difference = np.abs(value - previous)/previous
if relative_difference > percentage:
previous = value
mask.append(True)
else:
mask.append(False)
# the last value is always valid
mask.append(True)
return mask
您可以使用 Pandas 滚动功能来创建局部极值。与您的 Scipy 方法相比，这稍微简化了代码。
寻找极值的函数：
def islocalmax(x):
"""Both neighbors are lower,
assumes a centered window of size 3"""
return (x[0] < x[1]) & (x[2] < x[1])
def islocalmin(x):
"""Both neighbors are higher,
assumes a centered window of size 3"""
return (x[0] > x[1]) & (x[2] > x[1])
def isextrema(x):
return islocalmax(x) or islocalmin(x)
创建锯齿形的函数，它可以一次应用于 Dataframe（在每一列上），但这将引入 NaN，因为返回的时间戳对于每一列都是不同的。您可以稍后轻松删除这些，如下例所示，或者只需将函数应用于 Dataframe 中的单个列。
请注意，我取消了针对阈值 k 的测试的注释，我不确定是否完全正确地理解了该部分。如果前一个和当前极值之间的绝对差需要大于 k，则可以包括它：& (ext_val.diff().abs() > k)
我也不确定最终之字形是否应该始终从原始高点移动到低点，或者反之亦然。我认为应该如此，否则您可以删除函数末尾的第二次 extreme 搜索。
def create_zigzag(col, p=0.2, k=1.2):
# Find the local min/max
# converting to bool converts NaN to True, which makes it include the endpoints
ext_loc = col.rolling(3, center=True).apply(isextrema, raw=False).astype(np.bool_)
# extract values at local min/max
ext_val = col[ext_loc]
# filter locations based on threshold
thres_ext_loc = (ext_val.diff().abs() > (ext_val.shift(-1).abs() * p)) #& (ext_val.diff().abs() > k)
# Keep the endpoints
thres_ext_loc.iloc[0] = True
thres_ext_loc.iloc[-1] = True
thres_ext_loc = thres_ext_loc[thres_ext_loc]
# extract values at filtered locations
thres_ext_val = col.loc[thres_ext_loc.index]
# again search the extrema to force the zigzag to always go from high > low or vice versa,
# never low > low, or high > high
ext_loc = thres_ext_val.rolling(3, center=True).apply(isextrema, raw=False).astype(np.bool_)
thres_ext_val =thres_ext_val[ext_loc]
return thres_ext_val
生成一些示例数据：
date_rng = pd.date_range('2019-01-01', freq='s', periods=35)
df = pd.DataFrame(np.random.randn(len(date_rng), 3),
columns=['data1', 'data2', 'data3'],
index= date_rng)
df = df.cumsum()
应用函数并提取 'data1' 列的结果：
dfzigzag = df.apply(create_zigzag)
data1_zigzag = dfzigzag['data1'].dropna()
可视化结果：
fig, axs = plt.subplots(figsize=(10, 3))
axs.plot(df.data1, 'ko-', ms=4, label='original')
axs.plot(data1_zigzag, 'ro-', ms=4, label='zigzag')
axs.legend()