基于Python的时间序列趋势变点检测技术问询

问题说明 #

我有如下格式的时序数据，需要以date_code为X轴、weight_kg为Y轴绘制折线图，同时识别趋势变点（要求变点后weight_kg呈现持续上升或下降趋势）。原始数据量较大，需先将date_code划分为特定周期，再在每个周期内完成变点识别。

示例数据：

date_code  weight_kg
0        354     215.16
1        355     502.59
2        356     568.15
3        357     328.20
4        358     824.07

解决方案 #

1. 依赖库安装 #

先安装所需工具库：

pip install pandas numpy ruptures matplotlib

2. 核心步骤实现 #

2.1 数据读取与周期划分

示例按固定间隔（每5个date_code为一个周期）划分，可根据实际需求调整规则：

import pandas as pd
import numpy as np

# 读取数据（示例用模拟扩展数据）
data = pd.DataFrame({
    "date_code": range(354, 370),
    "weight_kg": [215.16, 502.59, 568.15, 328.20, 824.07, 
                  780.5, 750.2, 720.1, 690.3, 650.8,
                  660.2, 680.5, 710.3, 750.9, 800.1, 850.5]
})

# 按每5个date_code划分为一个周期
data["period"] = (data["date_code"] - data["date_code"].min()) // 5 + 1

2.2 周期内趋势变点识别

用ruptures库的Pelt算法检测变点，同时筛选出变点后趋势持续的有效点：

import ruptures as rpt

def detect_valid_change_points(series):
    # 将数据转换为算法要求的格式
    signal = series.values.reshape(-1, 1)
    # 初始化变点检测模型
    model = rpt.Pelt(model="rbf").fit(signal)
    # 检测变点（pen参数控制变点数量，值越大变点越少）
    raw_change_points = model.predict(pen=10)
    
    valid_cps = []
    # 遍历候选变点，验证变点后是否为持续升降趋势
    for cp in raw_change_points[:-1]:  # 排除序列末尾的点
        post_series = series[cp:]
        diffs = np.diff(post_series)
        if len(diffs) == 0:
            continue
        # 判断是否全上升或全下降
        if all(diffs > 0) or all(diffs < 0):
            valid_cps.append(cp)
    return valid_cps

# 按周期分组处理，输出每个周期的变点结果
period_results = data.groupby("period").apply(lambda x: {
    "date_list": x["date_code"].tolist(),
    "weight_list": x["weight_kg"].tolist(),
    "valid_change_dates": [x["date_code"].iloc[cp] for cp in detect_valid_change_points(x["weight_kg"])]
})

2.3 可视化周期趋势与变点

绘制每个周期的折线图并标记有效变点：

import matplotlib.pyplot as plt

# 遍历每个周期生成可视化图
for period, res in period_results.items():
    plt.figure(figsize=(8, 4))
    plt.plot(res["date_list"], res["weight_list"], marker='o', color='#1f77b4', label="体重")
    
    # 标记变点
    for cp_date in res["valid_change_dates"]:
        plt.axvline(x=cp_date, color='#ff4b5c', linestyle='--', label=f"趋势变点 {cp_date}")
    
    plt.title(f"周期{period}体重趋势")
    plt.xlabel("date_code")
    plt.ylabel("weight_kg")
    plt.legend()
    plt.grid(axis='y', linestyle='--', alpha=0.7)
    plt.show()

关键说明 #

• 周期划分：示例用固定间隔，可替换为自定义规则（比如按月份、季度映射date_code）
• 变点检测：ruptures支持多种算法，可调整pen参数平衡变点数量与准确性
• 趋势验证：通过差分判断变点后趋势，确保符合“持续升降”的要求

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