宝塔服务器面板，一键全能部署及管理，送你10850元礼包，点我领取

本文介绍 Prophet 模型的简单调用。

（一）日志设置为不输出

import os

class SuppressStdoutStderrobject):
    """
    A context manager for doing a "deep suppression" of stdout and stderr in
    Python, i.e. will suppress all print, even if the print originates in a
    compiled C/Fortran sub-function.

    This will not suppress raised exceptions, since exceptions are printed
    to stderr just before a script exits, and after the context manager has
    exited at least, I think that is why it lets exceptions through).
    """

    def __init__self):
        # open a pair of null files
        self.null_fds = [os.openos.devnull, os.O_RDWR) for x in range2)]

        # save the actual stdout 1) and stderr 2) file descriptors
        self.save_fds = os.dup1), os.dup2))

    def __enter__self):
        # assign the null pointers to stdout and stderr
        os.dup2self.null_fds[0], 1)
        os.dup2self.null_fds[1], 2)

    def __exit__self, *_):
        # reassign the real stdout/stderr back to 1) and 2)
        os.dup2self.save_fds[0], 1)
        os.dup2self.save_fds[1], 2)

        # close the null files
        os.closeself.null_fds[0])
        os.closeself.null_fds[1])
        os.closeself.save_fds[0])
        os.closeself.save_fds[1])

（二）Prophet 预测模型 class，与官方 Prophet 结构相似，但不继承

1. 初始化

from datetime import datetime, timedelta
from typing import Tuple
import numpy as np
import pandas as pd
from fbprophet import Prophet
from fbprophet.diagnostics import cross_validation
STD_D_STR = "%Y-%m-%d"  # '%m': 月份，'%M': 分钟
class ProphetPredictorobject):
def __init__self, x_train: pd.DataFrame,
trg_st_dt: datetime, tm_step: int, his_st_dt: datetime, his_en_dt: datetime,
cv_horizon: str, cv_period: str, cv_initial: str,
n_changepoints=None, changepoint_range=0.7,
yearly_seasonality=False, weekly_seasonality=True, daily_seasonality=False,
holidays=None, seasonality_mode='multiplicative',
seasonality_prior_scale=10, holidays_prior_scale=0, changepoint_prior_scale=0.05):
"""
initialisation
:param x_train:  数据集， ['ds', 'y']
:param trg_st_dt:  预测开始日期
:param tm_step:  预测时间间隔
:param his_st_dt:  训练集开始日期
:param his_en_dt:  训练集结束日期  todo: 允许使用的训练集结束日期必须严格小于预测开始日期
:param cv_horizon:  交叉验证 horizon 参数， '3 days' 格式
:param cv_period:  交叉验证 period 参数， '3 days' 格式
:param cv_initial:  交叉验证 initial 参数， '3 days' 格式
:param n_changepoints:  Changepoint 最大数量
:param changepoint_range:  Changepoint 在历史数据中出现的时间范围
:param yearly_seasonality:  年周期性
:param weekly_seasonality:  周周期性
:param daily_seasonality:  日周期性
:param holidays:  节假日或特殊日期
:param seasonality_mode:  季节模型方式， {'additive', 'multiplicative'}
:param seasonality_prior_scale:  改变周期性影响因素的强度
:param holidays_prior_scale:  改变假日模型的强度
:param changepoint_prior_scale:  设定自动突变点选择的灵活性，值越大越容易出现 Changepoint
"""
# Prophet 模型参数
self.params = {
"n_changepoints": n_changepoints,
"changepoint_range": changepoint_range,
"yearly_seasonality": yearly_seasonality,
"weekly_seasonality": weekly_seasonality,
"daily_seasonality": daily_seasonality,
"holidays": holidays,
"seasonality_mode": seasonality_mode,
"seasonality_prior_scale": seasonality_prior_scale,
"holidays_prior_scale": holidays_prior_scale,
"changepoint_prior_scale": changepoint_prior_scale
}
self.trg_st_dt = datetime.strptimetrg_st_dt, STD_D_STR) if isinstancetrg_st_dt, str) else trg_st_dt
self.tm_step = tm_step
self.trg_en_dt = self.trg_st_dt + timedeltadays=tm_step - 1)
self.his_st_dt = datetime.strptimehis_st_dt, STD_D_STR) if isinstancehis_st_dt, str) else his_st_dt
self.his_en_dt = datetime.strptimehis_en_dt, STD_D_STR) if isinstancehis_en_dt, str) else his_en_dt
# 提前期 = 预测开始日期 - 历史数据最晚日期 - 1 （认为次日预测提前期为0）
self.ahead = self.trg_st_dt - self.his_en_dt).days - 1
self.x_train = x_train[['ds', 'y']].copy)
self.model = None
self.cv_horizon = cv_horizon
self.cv_period = cv_period
self.cv_initial = cv_initial
self.map_err = 100

2. 模型训练

    def fitself):
"""
模型训练
:return: 无
"""
self.x_train = self.x_train[
datetime.strftimeself.his_st_dt, STD_D_STR) <= self.x_train['ds'])
& self.x_train['ds'] <= datetime.strftimeself.his_en_dt, STD_D_STR))].reset_indexdrop=True)
self.model = Prophet**self.params)
with SuppressStdoutStderr):
self.model.fitdf=self.x_train)

3. 交叉验证

    def cvself, params=None) -> float:
"""
交叉验证
:param params:  模型参数，网格寻参时不为 None
:return: map_err:  平均绝对百分比误差（MAPE）
"""
params_ = params if params else self.params
self.model = Prophet**params_)
with SuppressStdoutStderr):
self.model.fitself.x_train)
cv_result = cross_validationself.model,
horizon=self.cv_horizon, period=self.cv_period, initial=self.cv_initial)
# 平均绝对百分比误差（MAPE）
map_err = np.meannp.abscv_result['yhat'] - cv_result['y']) / cv_result['y']) * 100
return map_err

4. 网格寻参

    def grid_searchself) -> pd.DataFrame:
"""
网格寻参
:return: df_search:  寻参记录
"""
list_n_changepoints = [i for i in range2, 7)]
list_changepoint_range = [i / 10 for i in range5, 10)]
list_seasonality_mode = ["additive", "multiplicative"]
list_seasonality_prior_scale = [0.1, 0.5, 1, 5, 10]
list_changepoint_prior_scale = [0.1, 0.5, 1, 5, 10]
list_search = []
for nc in list_n_changepoints:
for cr in list_changepoint_range:
for sm in list_seasonality_mode:
for sps in list_seasonality_prior_scale:
for cps in list_changepoint_prior_scale:
params = {
"n_changepoints": nc,
"changepoint_range": cr,
"yearly_seasonality": False,
"weekly_seasonality": True,
"daily_seasonality": False,
"holidays": None,
"seasonality_mode": sm,
"seasonality_prior_scale": sps,
"holidays_prior_scale": 0,
"changepoint_prior_scale": cps
}
score = self.cvparams=params)
list_search.append[nc, cr, sm, sps, cps, score])
if score < self.map_err:
self.map_err, self.params = score, params
print"current best mse:  {0};  current params:  {1}".formatroundself.map_err, 4),
params))
df_search = pd.DataFramedata=list_search,
columns=['n_changepoints', 'changepoint_range', 'seasonality_mode',
'seasonality_prior_scale', 'changepoint_prior_scale', 'mse'])
return df_search

5. 预测

    def predictself) -> Tuple[pd.DataFrame, pd.DataFrame]:
"""
预测
:return: df_predict_y:  预测数据
:return: df_history_y:  历史数据
"""
df_future = self.model.make_future_dataframeperiods=self.ahead + self.tm_step,
include_history=True).dropna).reset_indexdrop=True)
df_predict = self.model.predictdf=df_future)
df_predict['ds'] = df_predict['ds'].applylambda x: strx.date)))
df_predict_y = df_predict[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].tailself.tm_step)
df_history_y = df_predict[['ds', 'yhat']][: - self.tm_step + self.ahead)]
return df_predict_y, df_history_y

（三）其他尝试

添加季节性组件：

self.model.add_seasonalityname='sin', period=2 * np.pi / 7, fourier_order=1)

（四）模型调用示例

1. 网格寻参

# 初始化
prophet_predictor = ProphetPredictorx_train=df_input,
trg_st_dt=trg_st_dt, tm_step=3, his_st_dt=his_st_dt, his_en_dt=his_en_dt,
cv_horizon='3 days', cv_period='3 days', cv_initial='135 days')
# 网格寻参
dts_search = datetime.now)
df_search = prophet_predictor.grid_search)
print"df_search:\n", df_search, '\n')
dte_search = datetime.now)
tm_search = rounddte_search - dts_search).seconds + dte_search - dts_search).microseconds / 10 ** 6), 3)
print"grid search time:  {} s".formattm_search), '\n')
# 训练
dts_train = datetime.now)
prophet_predictor.fit)
dte_train = datetime.now)
tm_train = rounddte_train - dts_train).seconds + dte_train - dts_train).microseconds / 10 ** 6), 3)
print"train time:  {} s".formattm_train), '\n')
# 预测
df_predict_y, df_history_y = prophet_predictor.predict)
print"df_predict_y:\n", df_predict_y, '\n')
print"df_history_y:\n", df_history_y, '\n')

2. 给定参数

# 参数设定
params = {
"n_changepoints": 2,
"changepoint_range": 0.7,
"seasonality_mode": 'additive',
"seasonality_prior_scale": 0.5,
"changepoint_prior_scale": 10,
"yearly_seasonality": False,
"weekly_seasonality": True,
"daily_seasonality": False
}
# 初始化
prophet_predictor = ProphetPredictorx_train=df_input,
trg_st_dt=trg_st_dt, tm_step=3, his_st_dt=his_st_dt, his_en_dt=his_en_dt,
cv_horizon='3 days', cv_period='3 days', cv_initial='135 days',
**params)
# 训练
dts_train = datetime.now)
prophet_predictor.fit)
dte_train = datetime.now)
tm_train = rounddte_train - dts_train).seconds + dte_train - dts_train).microseconds / 10 ** 6), 3)
print"train time:  {} s".formattm_train), '\n')
# 预测
df_predict_y, df_history_y = prophet_predictor.predict)
print"df_predict_y:\n", df_predict_y, '\n')
print"df_history_y:\n", df_history_y, '\n')

参考资料：

https://www.cnblogs.com/fulu/p/13329656.html

https://www.cnblogs.com/zhazhaacmer/p/13786940.html