Spaces:
Sleeping
Sleeping
| import numpy as np | |
| from typing import Literal | |
| import torch | |
| from typing import List | |
| def create_dataset(dataset, time_step_backward = 1, time_step_forward = 1): | |
| dataX, dataY = [], [] | |
| for i in range(len(dataset) - time_step_backward - (time_step_forward - 1)): | |
| a = dataset[i:(i + time_step_backward), 0] ###i=0, 0,1,2,3-----99 100 | |
| dataX.append(a) | |
| dataY.append(dataset[i + time_step_backward + (time_step_forward - 1), 0]) | |
| return np.array(dataX), np.array(dataY) | |
| def make_prediction(X_train: np.ndarray, X_test: np.ndarray, | |
| method: Literal['LSTM', 'GMDH', 'Transformer', 'SARIMA'], | |
| model, scaler, time_step_forward: None) -> np.ndarray: | |
| if method == 'LSTM': | |
| train_predict = model.predict(X_train) | |
| test_predict = model.predict(X_test) | |
| train_predict = scaler.inverse_transform(train_predict) | |
| test_predict = scaler.inverse_transform(test_predict) | |
| return train_predict, test_predict | |
| elif method == 'SARIMA': | |
| train_predict_arima = [] | |
| test_predict_arima = [] | |
| for sample in X_train: | |
| train_predict_arima.append( | |
| model.fit_predict(sample, n_periods=time_step_forward, return_conf_int=False)[-1]) | |
| train_predict_arima = np.array(train_predict_arima) | |
| for sample in X_test: | |
| test_predict_arima.append( | |
| model.fit_predict(sample, n_periods=time_step_forward, return_conf_int=False)[-1]) | |
| test_predict_arima = np.array(test_predict_arima) | |
| train_predict_arima = scaler.inverse_transform(train_predict_arima.reshape(-1, 1)) | |
| test_predict_arima = scaler.inverse_transform(test_predict_arima.reshape(-1, 1)) | |
| return train_predict_arima, test_predict_arima | |
| elif method == 'GMDH': | |
| train_predict_gmdh = model.predict(X_train) | |
| test_predict_gmdh = model.predict(X_test) | |
| train_predict_gmdh = scaler.inverse_transform(train_predict_gmdh.reshape(-1, 1)) | |
| test_predict_gmdh = scaler.inverse_transform(test_predict_gmdh.reshape(-1, 1)) | |
| return train_predict_gmdh, test_predict_gmdh | |
| elif method == 'Transformer': | |
| X_train_context = torch.tensor(X_train) | |
| X_test_context = torch.tensor(X_test) | |
| X_train_forecast = model.predict( | |
| X_train_context, | |
| time_step_forward, | |
| num_samples=3, | |
| temperature=1.0, | |
| top_k=50, | |
| top_p=1.0) | |
| X_test_forecast = model.predict( | |
| X_test_context, | |
| time_step_forward, | |
| num_samples=3, | |
| temperature=1.0, | |
| top_k=50, | |
| top_p=1.0) | |
| X_train_forecast_median = np.quantile(X_train_forecast.numpy(), 0.5, axis=1)[:, -1] | |
| X_test_forecast_median = np.quantile(X_test_forecast.numpy(), 0.5, axis=1)[:, -1] | |
| X_train_forecast_median = scaler.inverse_transform(X_train_forecast_median.reshape(-1, 1)) | |
| X_test_forecast_median = scaler.inverse_transform(X_test_forecast_median.reshape(-1, 1)) | |
| return X_train_forecast_median, X_test_forecast_median | |
| def make_prediction_recursive(test_data: np.ndarray, | |
| method: Literal['LSTM', 'GMDH', 'Transformer', 'SARIMA'], | |
| model, scaler, pred_days: None, time_step_backward: None) -> List[int]: | |
| if method == 'LSTM': | |
| x_input_lstm = test_data[len(test_data) - time_step_backward:].reshape(1, -1) | |
| temp_input_lstm = list(x_input_lstm) | |
| temp_input_lstm = temp_input_lstm[0].tolist() | |
| lst_output_lstm = [] | |
| n_steps = time_step_backward | |
| i = 0 | |
| while (i < pred_days): | |
| if (len(temp_input_lstm) > time_step_backward): | |
| x_input_lstm = np.array(temp_input_lstm[1:]) | |
| x_input_lstm = x_input_lstm.reshape(1, -1) | |
| x_input_lstm = x_input_lstm.reshape((1, n_steps, 1)) | |
| yhat_lstm = model.predict(x_input_lstm, verbose=0) | |
| temp_input_lstm.extend(yhat_lstm[0].tolist()) | |
| temp_input_lstm = temp_input_lstm[1:] | |
| lst_output_lstm.extend(yhat_lstm.tolist()) | |
| i = i + 1 | |
| else: | |
| x_input_lstm = x_input_lstm.reshape((1, n_steps, 1)) | |
| yhat_lstm = model.predict(x_input_lstm, verbose=0) | |
| temp_input_lstm.extend(yhat_lstm[0].tolist()) | |
| lst_output_lstm.extend(yhat_lstm.tolist()) | |
| i = i + 1 | |
| lst_output_lstm = scaler.inverse_transform(lst_output_lstm) | |
| return lst_output_lstm | |
| elif method == 'SARIMA': | |
| x_input_arima = test_data[len(test_data) - time_step_backward:] | |
| n_steps = time_step_backward | |
| lst_output_arima = model.fit_predict(x_input_arima, n_periods=pred_days, return_conf_int=False) # [-1] | |
| lst_output_arima = scaler.inverse_transform(lst_output_arima.reshape(-1, 1)) | |
| return lst_output_arima | |
| elif method == 'GMDH': | |
| x_input_gmdh = test_data[len(test_data) - time_step_backward:].reshape(1, -1) | |
| temp_input_gmdh = list(x_input_gmdh) | |
| temp_input_gmdh = temp_input_gmdh[0].tolist() | |
| lst_output_gmdh = [] | |
| n_steps = time_step_backward | |
| i = 0 | |
| while (i < pred_days): | |
| if (len(temp_input_gmdh) > time_step_backward): | |
| x_input_gmdh = np.array(temp_input_gmdh[1:]) | |
| x_input_gmdh = x_input_gmdh.reshape(1, -1) | |
| yhat_gmdh = model.predict(x_input_gmdh) | |
| temp_input_gmdh.extend(yhat_gmdh.tolist()) | |
| temp_input_gmdh = temp_input_gmdh[1:] | |
| lst_output_gmdh.extend(yhat_gmdh.tolist()) | |
| i = i + 1 | |
| else: | |
| x_input_gmdh = x_input_gmdh.reshape((1, n_steps, 1)) | |
| yhat_gmdh = model.predict(x_input_gmdh[0].reshape(1, -1)) | |
| temp_input_gmdh.extend(yhat_gmdh.tolist()) | |
| lst_output_gmdh.extend(yhat_gmdh.tolist()) | |
| i = i + 1 | |
| lst_output_gmdh = scaler.inverse_transform(np.array(lst_output_gmdh).reshape(-1, 1)) | |
| return lst_output_gmdh | |
| elif method == 'Transformer': | |
| x_input_transformer = test_data[len(test_data) - time_step_backward:].reshape(1, -1) | |
| x_input_transformer = torch.tensor(x_input_transformer) | |
| lst_output_forecast = model.predict( | |
| x_input_transformer, | |
| pred_days, | |
| num_samples=3, | |
| temperature=1.0, | |
| top_k=50, | |
| top_p=1.0) | |
| X_train_forecast_median = np.quantile(lst_output_forecast.numpy(), 0.5, axis=1) # [:, -1] | |
| lst_output_transformer = scaler.inverse_transform(X_train_forecast_median.reshape(-1, 1)) | |
| return lst_output_transformer | |