基于Python识别一维时间序列切片的趋势状态

我明白你遇到的问题了——argrelextrema确实会因为序列末尾没有后续数据来判定极值，导致最后一段的趋势无法被自动划分。你手动加末尾数据的思路是对的，但可能没处理好原数据末尾和扩展数据的衔接，也没单独处理最后一段的趋势赋值。我来给你调整下代码，解决这个问题：

问题根源 #

原代码只处理了极值点之间的区间，但最后一个极值点到原数据末尾的这段区间完全没被加入status列表。你加的5个数据是为了让最后一个极值能被检测到，但这部分扩展数据的区间不该被算进原数据的趋势标记里，同时原数据的最后一段需要手动判断趋势。

修正后的代码 #

import pandas as pd
import numpy as np
from scipy.signal import argrelextrema

# 处理原数据（假设你的data已存在）
d = data['tenkan_sen'].fillna(1)
original_len = len(d)  # 记录原数据长度，避免把扩展的5个数据算入结果

# 扩展末尾数据，帮助检测最后一个极值点
h1 = d.append(pd.Series([5, 4, 3, 2, 1]), ignore_index=True)

# 获取局部极值点索引
max_indices = argrelextrema(h1.values, np.greater)[0]
min_indices = argrelextrema(h1.values, np.less)[0]

# 整理极值点与对应类型，加入起始点
extermas = {0: 'start'}
for idx in max_indices:
    extermas[idx] = 'greater'
for idx in min_indices:
    extermas[idx] = 'less'
# 按索引排序极值点
sorted_extrema = sorted(extermas.items(), key=lambda x: x[0])

status = []
# 处理极值点之间的区间
for i in range(len(sorted_extrema)-1):
    start_idx = sorted_extrema[i][0]
    end_idx = sorted_extrema[i+1][0]
    # 只处理原数据范围内的区间，过滤扩展数据部分
    actual_end = min(end_idx, original_len)
    if actual_end <= start_idx:
        continue
    # 根据下一个极值的类型判断当前区间趋势
    if sorted_extrema[i+1][1] == 'less':
        stat = 'down'
    elif sorted_extrema[i+1][1] == 'greater':
        stat = 'up'
    else:
        stat = 'equal'
    # 给当前区间赋值趋势
    status.extend([stat] * (actual_end - start_idx))

# 处理最后一段：从最后一个极值点到原数据末尾
if len(status) < original_len:
    last_extrema_idx, last_extrema_type = sorted_extrema[-1]
    remaining_len = original_len - len(status)
    # 通过比较最后一个极值点和原数据末尾点的数值判断趋势
    if last_extrema_idx < original_len:
        last_val = d.iloc[-1]
        extrema_val = d.iloc[last_extrema_idx]
        # 考虑浮点误差，避免直接用==判断相等
        if abs(last_val - extrema_val) < 1e-6:
            final_stat = 'equal'
        elif last_val > extrema_val:
            final_stat = 'up'
        else:
            final_stat = 'down'
    else:
        # 若最后一个极值在扩展数据中，说明原数据末尾是单调趋势，沿用前一段趋势
        final_stat = 'down' if last_extrema_type == 'less' else 'up'
    status.extend([final_stat] * remaining_len)

# 优化连续相等区间的识别（替换原代码中单个点的equal处理）
equal_intervals = []
current_start = 0
for i in range(1, original_len):
    if abs(d.iloc[i] - d.iloc[i-1]) < 1e-6:
        continue
    else:
        # 至少连续2个点相等才算equal区间
        if i - current_start > 1:
            equal_intervals.append((current_start, i))
        current_start = i
# 检查最后一段是否为连续相等区间
if original_len - current_start > 1:
    equal_intervals.append((current_start, original_len))

# 将equal区间的趋势标记替换为'equal'
for start, end in equal_intervals:
    for i in range(start, end):
        if i < len(status):
            status[i] = 'equal'

# 确保结果长度与原数据一致
assert len(status) == original_len, f"Status长度{len(status)}与原数据长度{original_len}不匹配"

关键改动说明 #

• 新增original_len变量，严格区分原数据和扩展的末尾数据，避免把扩展数据的趋势混入最终结果。
• 补充末尾区间处理逻辑：通过比较最后一个极值点和原数据末尾点的数值，手动给剩余区间赋值趋势，彻底解决末尾切片无标记的问题。
• 优化equal区间识别：不再依赖argrelextrema找相等极值点（这本身不符合连续平段的定义），而是直接检测连续相等的区间，再替换对应位置的趋势标记。
• 加入浮点误差判断：针对时间序列的浮点数特性，用abs(a-b) < 1e-6替代直接相等判断，避免精度问题导致的错误。

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