Create Prophet Forecasting Class

Here is starter code:

import pandas as pd
from prophet import Prophet
from prophet.diagnostics import cross_validation, performance_metrics
from prophet.serialize import model_to_json, model_from_json
import joblib

class ProphetModel:
def __init__(self):
"""
Initialize the ProphetModel.
"""
self.model = None
self.best_params = None

def fit(self, df, regressors=None, **kwargs):
"""
Fit the Prophet model to the data.

:param df: DataFrame with columns 'ds' and 'y' for time series data.
:param regressors: List of additional regressor column names.
:param kwargs: Additional keyword arguments for the Prophet model.
"""
self.model = Prophet(**kwargs)
if regressors:
for regressor in regressors:
self.model.add_regressor(regressor)
self.model.fit(df)

def predict(self, future):
"""
Predict using the fitted Prophet model.

:param future: DataFrame with a column 'ds' and optional regressor columns.
:return: DataFrame with predictions.
"""
if self.model:
return self.model.predict(future)
else:
print("Model has not been fitted yet.")
return None

def save_model(self, file_path):
"""
Save the trained model to a file.

:param file_path: Path to the file where the model will be saved.
"""
if self.model:
with open(file_path, 'w') as f:
f.write(model_to_json(self.model))
else:
print("Model has not been fitted yet.")

def load_model(self, file_path):
"""
Load a trained model from a file.

:param file_path: Path to the file where the model is saved.
"""
with open(file_path, 'r') as f:
self.model = model_from_json(f.read())

def tune_hyperparameters(self, df, regressors=None, initial='730 days', period='180 days', horizon='365 days'):
"""
Tune hyperparameters for the Prophet model using cross-validation and performance metrics.

:param df: DataFrame with columns 'ds' and 'y' for time series data.
:param regressors: List of additional regressor column names.
:param initial: Initial training period for cross-validation.
:param period: Period between cutoff dates for cross-validation.
:param horizon: Forecast horizon for cross-validation.
"""
param_grid = {
'changepoint_prior_scale': [0.001, 0.01, 0.1, 0.5],
'seasonality_prior_scale': [0.01, 0.1, 1.0, 10.0],
'holidays_prior_scale': [0.01, 0.1, 1.0, 10.0],
'seasonality_mode': ['additive', 'multiplicative'],
'yearly_seasonality': [5, 10, 15, 20],
'weekly_seasonality': [5, 10, 15, 20],
'daily_seasonality': [0, 5, 10]
}

best_rmse = float('inf')
best_params = {}

for cps in param_grid['changepoint_prior_scale']:
for sps in param_grid['seasonality_prior_scale']:
for hps in param_grid['holidays_prior_scale']:
for sm in param_grid['seasonality_mode']:
for ys in param_grid['yearly_seasonality']:
for ws in param_grid['weekly_seasonality']:
for ds in param_grid['daily_seasonality']:
params = {
'changepoint_prior_scale': cps,
'seasonality_prior_scale': sps,
'holidays_prior_scale': hps,
'seasonality_mode': sm,
'yearly_seasonality': ys,
'weekly_seasonality': ws,
'daily_seasonality': ds
}

model = Prophet(**params)
if regressors:
for regressor in regressors:
model.add_regressor(regressor)
model.fit(df)

df_cv = cross_validation(model, initial=initial, period=period, horizon=horizon)
df_p = performance_metrics(df_cv)
rmse = df_p['rmse'].mean()

if rmse < best_rmse:
best_rmse = rmse
best_params = params

self.best_params = best_params
print(f"Best hyperparameters: {self.best_params}")

def fit_with_best_params(self, df, regressors=None):
"""
Fit the Prophet model with the best hyperparameters.

:param df: DataFrame with columns 'ds' and 'y' for time series data.
:param regressors: List of additional regressor column names.
"""
if self.best_params:
self.fit(df, regressors, **self.best_params)
else:
print("Hyperparameters have not been tuned yet.")