| 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) |
|
|
