python实现时间序列分解法(Time-series Decomposition)——以预测气温为例

Irisa ·

更新时间:2024-09-21

· 817 次阅读

文章目录程序简述程序/数据集下载数据集截图核心代码解析Module/BuildModel.py（接口，可以直接运行）接口调用、运行效果Main.py 程序简述

利用乘法型的时间序列分解算法预测北京气温变化
程序输入：观测数据，周期长度，需要往后预测的个数
程序输出：预测值，模型结构

时间序列分解使用加法模型或乘法模型讲原始系列拆分为四部分：长期趋势变动T、季节变动S(显式周期，固定幅度、长度的周期波动)、循环变动C(隐式周期，周期长不具严格规则的波动)和不规则变动L。本例使用的是乘法模型。

程序/数据集下载

点击进入下载地址

数据集截图

图1，北京气象数据

核心代码解析 Module/BuildModel.py（接口，可以直接运行）

建立时间序列分解模型，对数据进行移动平均操作和调整，计算季节性因子和长期趋势模型，本程序使用的线性模型

# -*- coding: utf-8 -*-
from sklearn.linear_model import LinearRegression
import pandas as pd
import numpy as np
class TimeSeriesSplit():
    def __init__(self,series,EMA):
        '''
        时间序列分解算法，乘法模型，由于循环波动难以确认，受随机因素影响大，不予考虑
        series：时间序列
        EMA：移动平均项数，也是周期的时长        
        '''
        self.buildModel(series,EMA)
    def predict(self,num):
        '''
        往后预测num个数，返回的是整个模型的信息
        num：预测个数
        '''
        result = []
        for i in range(num):
            #季节因子
            S = self.seasFactors[(i+len(self.series))%len(self.seasFactors)]
            #长期趋势
            T = self.regression.predict(i+len(self.series))[0][0]
            result.append(T*S)
        info = {
                'predict':{'value':result,'desc':'往后预测的%s个数'%num},
                'Ta':{'value':self.regression.coef_[0][0],'desc':'长期趋势线性模型的系数'},
                'Tb':{'value':self.regression.intercept_[0],'desc':'长期趋势线性模型的截距'},
                'seasonFactor':{'value':self.seasFactors,'desc':'季节因子'},
                }
        return info
    def buildModel(self,series,EMA):
        '''
        建模，预测
        series：时间序列
        EMA：移动平均项数，也是周期的时长
        '''
        series = np.array(series).reshape(-1)
        #移动平均数
        moveSeies = self.calMoveSeries(series,EMA)
        #季节因子
        seasonFactors = self.calSeasonFactors(series,moveSeies,EMA)
        #长期趋势建模
        regression = self.buildLongTrend(series)
        #收尾，设置对象属性
        self.series = series
        self.seasFactors = seasonFactors
        self.regression = regression
    def calMoveSeries(self,series,EMA):
        '''
        计算移动平均数
        series：时间序列
        EMA：移动平均项数，也是周期的时长
        '''
        #计算移动平均
        moveSeries = []
        for i in range(0,series.shape[0]-EMA+1):
            moveSeries.append(series[i:i+EMA].mean())
        moveSeries = np.array(moveSeries).reshape(-1)
        #如果项数为复数，则移动平均后数据索引无法对应原数据，要进行第2次项数为2的移动平均
        if EMA % 2 == 0:
            moveSeries2 = []
            for i in range(0,moveSeries.shape[0]-2+1):
                moveSeries2.append(moveSeries[i:i+2].mean())
            moveSeries = np.array(moveSeries2).reshape(-1)
        return moveSeries
    def calSeasonFactors(self,series,moveSeries,EMA):
        '''
        计算季节性因子
        series：时间序列
        moveSeries：移动平均数
        EMA：移动平均项数，也是周期的时长
        '''
        #移动平均后的第一项索引对应原数据的索引
        startIndex = int((series.shape[0] - moveSeries.shape[0])/2)
        #观测值除以移动平均值
        factors = []
        for i in range(len(moveSeries)):
            factors.append(series[startIndex+i]/moveSeries[i])
        #去掉尾部多余部分
        rest = len(factors)%EMA
        factors = factors[:len(factors)-rest]
        factors = np.array(factors).reshape(-1,EMA)
        #平均值可能不是1,调整
        seasonFactors = factors.mean(axis=0)/factors.mean()
        #按季顺序进行索引调整
        seasonFactors = seasonFactors[startIndex:].reshape(-1).tolist() + seasonFactors[:startIndex].reshape(-1).tolist()
        seasonFactors = np.array(seasonFactors).reshape(-1)
        return seasonFactors
    def buildLongTrend(self,series):
        '''
        计算长期趋势
        series：时间序列
        '''
        #建立线性模型
        reg = LinearRegression()
        #季节索引从0开始
        index = np.array(range(series.shape[0])).reshape(-1,1)
        reg.fit(index,series.reshape(-1,1))
        return reg
if __name__ == "__main__":
    import matplotlib.pyplot as plt
    #https://wenku.baidu.com/view/89933c24a417866fb94a8e0a.html?from=search
    #https://blog.csdn.net/weixin_40159138/article/details/90603344
    #销售数据
    data = [
            3017.60,3043.54,2094.35,2809.84,
            3274.80,3163.28,2114.31,3024.57,
            3327.48,3439.48,3493.93,3490.79,
            3685.08,3661.23,2378.43,3459.55,
            3849.63,3701.18,2642.38,3585.52,
            4078.66,3907.06,2818.46,4089.50,
            4339.61,4148.60,2976.45,4084.64,
            4242.42,3997.58,2881.01,4036.23,
            4360.33,4360.53,3172.18,4223.76,
            4690.48,4694.48,3342.35,4577.63,
            4965.46,5026.05,3470.14,4525.94,
            5258.71,5489.58,3596.76,3881.60            
    ]
    #plt.plot(range(len(data)),data)
    model = TimeSeriesSplit(data,4)
    #往后预测4个数，也就是1年4个季度的数
    print(model.predict(4))

接口调用、运行效果 Main.py

这里输入的是北京气象数据，因为1行的时间单位是月，所以周期长度设置为12，即1年，最后预测的效果如图2和图3

# -*- coding: utf-8 -*-
from Module.BuildModel import TimeSeriesSplit
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import os
#路径目录
baseDir = os.path.dirname(os.path.abspath(__file__))#当前目录
staticDir = os.path.join(baseDir,'Static')#静态文件目录
resultDir = os.path.join(baseDir,'Result')#结果文件目录
#读取数据
data = pd.read_excel(staticDir+'/北京气象数据.xlsx')
#前24个做训练数据，后面的做测试数据
train = data[0:24]['平均温度(℃)'].values
test = data[24:]['平均温度(℃)'].values
#建模
EMA = 12#周期长度，即12个月
model = TimeSeriesSplit(train,EMA)
#预测
result = model.predict(test.shape[0])
print('季节性因子',np.round(result['seasonFactor']['value'],2))
print('长期趋势系数和截距',np.round(result['Ta']['value'],2),np.round(result['Tb']['value'],2))
print('预测值',np.round(result['predict']['value'],2))
#可视化
#用来正常显示中文标签
plt.rcParams['font.sans-serif']=['SimHei'] 
#用来正常显示负号
plt.rcParams['axes.unicode_minus']=False
plt.plot(range(len(result['predict']['value'])),result['predict']['value'],label="预测值")
plt.plot(range(len(result['predict']['value'])),test,label="观测值")
plt.legend()
plt.title('时间序列分解法预测效果')
plt.savefig(resultDir+'/时间序列分解法预测效果.png',dpi=100,bbox_inches='tight')

图2，观测值和预测值折线图