File size: 3,173 Bytes
7f5bd75 35b553f 7f5bd75 35b553f 7f5bd75 35b553f 7f5bd75 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 |
import pandas as pd
from prophet import Prophet
from prophet.diagnostics import cross_validation, performance_metrics
class GeoMagModel:
"""
A class for geomagnetic data forecasting using Prophet.
"""
def __init__(self, changepoint_prior_scale=0.1, weekly_seasonality=True):
"""
Initialize the GeoMagModel with Prophet configuration.
Args:
changepoint_prior_scale (float): Flexibility of changepoint detection.
weekly_seasonality (bool): Whether to include weekly seasonality.
"""
self.changepoint_prior_scale = changepoint_prior_scale
self.weekly_seasonality = weekly_seasonality
self.model = None
@staticmethod
def prepare_for_prophet(df):
"""
Prepare the DataFrame for Prophet by renaming columns as required.
Args:
df (pd.DataFrame): The cleaned DataFrame with 'timestamp' and 'Dst' columns.
Returns:
pd.DataFrame: A DataFrame with 'ds' (timestamp) and 'y' (Dst) columns for Prophet.
"""
return df.rename(columns={'timestamp': 'ds', 'Dst': 'y'})
def train(self, df):
"""
Train the Prophet model on the given DataFrame.
Args:
df (pd.DataFrame): The DataFrame prepared for Prophet.
"""
df = self.prepare_for_prophet(df)
self.model = Prophet(interval_width=0.70, changepoint_prior_scale=self.changepoint_prior_scale)
if self.weekly_seasonality:
self.model.add_seasonality(name='weekly', period=7, fourier_order=5, prior_scale=10)
self.model.fit(df)
def forecast(self, periods=1, freq='h'):
"""
Make a future forecast with the trained model.
Args:
periods (int): The number of future periods to forecast. Defaults to 1.
freq (str): The frequency of the forecast ('h' for hours). Defaults to 'h'.
Returns:
pd.DataFrame: The forecast DataFrame with 'ds', 'yhat', 'yhat_lower', and 'yhat_upper' columns.
"""
if self.model is None:
raise ValueError("Model has not been trained. Call train() before forecast().")
future_dates = self.model.make_future_dataframe(periods=periods, freq=freq)
forecast = self.model.predict(future_dates)
return forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']]
def cross_validate(self, initial='36 hours', period='12 hours', horizon='1 hours'):
"""
Perform cross-validation on the trained Prophet model.
Args:
initial (str): Initial training period for cross-validation.
period (str): Frequency of making predictions.
horizon (str): Forecast horizon for each prediction.
Returns:
pd.DataFrame: Cross-validation results including metrics.
"""
if self.model is None:
raise ValueError("Model has not been trained. Call train() before cross_validate().")
return cross_validation(self.model, initial=initial, period=period, horizon=horizon)
|