This Gist contains some python scripts for the djl timeseries package. Including drawing prediction graphs and coarse-grained data, etc.
If you want to visualize your forecasts please run:
python plot.py -p YOUR_PRED_LENGTH -f YOUR_FREQUENCY -t YOUR_STARGT_TIME -target-path YOUR_TARGET_PATH -samples-path YOUR_SAMPLES_PATH
run
python m5_data_coarse_grain.py -p YOUR_DATA_PATH
| import pandas as pd | |
| import os | |
| import argparse | |
| def coarse_grain(path: str, end: int): | |
| print(f"Processing {os.path.basename(path)}") | |
| df = pd.read_csv(path) | |
| _df = [df.loc[:, ("id", "item_id", "dept_id", "cat_id", "store_id", "state_id")].copy()] | |
| for i in range(1, end, 7): | |
| tmp = pd.DataFrame(df.loc[:, "d_" + str(i): "d_" + str(i + 7)].sum(axis=1), | |
| columns=["w_" + str(i // 7 + 1)]) | |
| _df.append(tmp) | |
| _df = pd.concat(_df, axis=1) | |
| file_name = "weekly_" + os.path.basename(path) | |
| dir_name = os.path.dirname(path) | |
| output_path = os.path.join(dir_name, file_name) | |
| _df.to_csv(output_path, index=False) | |
| print(f"Save the {file_name} to {output_path}") | |
| def arg_parse(): | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("-p", | |
| "--path", | |
| help="M5 forecasting file path") | |
| return parser.parse_args() | |
| if __name__ == "__main__": | |
| args = arg_parse() | |
| if args.path is None: | |
| args.path = os.path.join(os.path.expanduser("~"), "Desktop", "m5example", "m5-forecasting-accuracy") | |
| coarse_grain(os.path.join(args.path, "sales_train_validation.csv"), 1912) | |
| coarse_grain(os.path.join(args.path, "sales_train_evaluation.csv"), 1940) |
| import numpy as np | |
| import pandas as pd | |
| import os | |
| import torch | |
| from gluonts.torch.model.deepar import DeepAREstimator | |
| from gluonts.torch.distributions.distribution_output import NegativeBinomialOutput | |
| from gluonts.dataset.common import load_datasets, ListDataset | |
| from gluonts.dataset.field_names import FieldName | |
| from gluonts.dataset.loader import InferenceDataLoader | |
| from gluonts.transform import AdhocTransform | |
| from gluonts.torch.batchify import batchify | |
| single_prediction_length = 4 | |
| submission_prediction_length = single_prediction_length * 2 | |
| m5_input_path = os.path.join(os.path.expanduser("~"), "Desktop", "m5-forecasting-accuracy") # Your Path to dataset | |
| submission = False | |
| if submission: | |
| prediction_length = submission_prediction_length | |
| else: | |
| prediction_length = single_prediction_length | |
| sales_train_validation = pd.read_csv(f'{m5_input_path}/weekly_sales_train_validation.csv') | |
| train_df = sales_train_validation.drop(["id", "item_id", "dept_id", "cat_id", "store_id", "state_id"], axis=1) | |
| train_target_values = train_df.values | |
| if submission == True: | |
| test_target_values = [np.append(ts, np.ones(submission_prediction_length) * np.nan) for ts in train_df.values] | |
| else: | |
| test_target_values = train_target_values.copy() | |
| train_target_values = [ts[:-single_prediction_length] for ts in train_df.values] | |
| m5_dates = [pd.Timestamp("2011-01-29", freq='1W') for _ in range(len(sales_train_validation))] | |
| train_ds = ListDataset([ | |
| { | |
| FieldName.TARGET: target, | |
| FieldName.START: start | |
| } | |
| for (target, start) in zip(train_target_values, | |
| m5_dates) | |
| ], freq="W") | |
| test_ds = ListDataset([ | |
| { | |
| FieldName.TARGET: target, | |
| FieldName.START: start | |
| } | |
| for (target, start) in zip(test_target_values, | |
| m5_dates) | |
| ], freq="W") | |
| estimator = DeepAREstimator( | |
| prediction_length=prediction_length, | |
| freq="W", | |
| distr_output=NegativeBinomialOutput(), | |
| trainer_kwargs={"max_epochs": 10} | |
| ) | |
| predictor = estimator.train(train_ds) | |
| prediction_length = predictor.prediction_length | |
| lead_time = predictor.lead_time | |
| def truncate_target(data): | |
| data = data.copy() | |
| target = data[FieldName.TARGET] | |
| assert ( | |
| target.shape[-1] >= prediction_length | |
| ) # handles multivariate case (target_dim, history_length) | |
| data[FieldName.TARGET] = target[..., : -prediction_length - lead_time] | |
| return data | |
| data_loader = InferenceDataLoader(AdhocTransform(truncate_target).apply(test_ds), | |
| transform=predictor.input_transform, | |
| batch_size=predictor.batch_size, | |
| stack_fn=lambda data: batchify(data, device=predictor.device), | |
| ) | |
| predictor.prediction_net.eval() | |
| with torch.no_grad(): | |
| for batch in data_loader: | |
| inputs = [batch[k] for k in predictor.input_names] | |
| trace = torch.jit.trace(predictor.prediction_net, tuple(inputs)) | |
| trace.save("./deepar.pt") # replace with your path | |
| break |
| import numpy as np | |
| import matplotlib.pyplot as plt | |
| import os | |
| import pandas as pd | |
| import argparse | |
| def sample_plot(samples: np.ndarray, | |
| index: pd.PeriodIndex, | |
| prediction_intervals=(50.0, 90.0), | |
| color='g'): | |
| def alpha_for_percentile(p): | |
| return (p / 100.0) ** 0.3 | |
| for c in prediction_intervals: | |
| assert 0.0 <= c <= 100.0 | |
| ps = [50.0] + [ | |
| 50.0 + f * c / 2.0 | |
| for c in prediction_intervals | |
| for f in [-1.0, +1.0] | |
| ] | |
| percentiles_sorted = sorted(set(ps)) | |
| sample_idxs = [int(np.round((samples.shape[0] - 1) * p)) for p in percentiles_sorted] | |
| sorted_samples = np.sort(samples, axis=0) | |
| ps_data = [sorted_samples[idx, :] for idx in sample_idxs] | |
| i_p50 = len(percentiles_sorted) // 2 | |
| p50_data = ps_data[i_p50] | |
| p50_series = pd.Series(data=p50_data, index=index.to_timestamp()) | |
| p50_series.plot(color=color, ls="-", label=f"median") | |
| for i in range(len(percentiles_sorted) // 2): | |
| ptile = percentiles_sorted[i] | |
| alpha = alpha_for_percentile(ptile) | |
| plt.fill_between( | |
| index.to_timestamp(), | |
| ps_data[i], | |
| ps_data[-i - 1], | |
| facecolor=color, | |
| alpha=alpha, | |
| interpolate=True | |
| ) | |
| def plot_prob_forecasts(ts_entry, samples, index, prediction_length, path): | |
| plot_length = 150 | |
| prediction_intervals = (50, 67, 95, 99) | |
| legend = ["observations", "median prediction"] + [f"{k}% prediction interval" for k in prediction_intervals][::-1] | |
| _, ax = plt.subplots(1, 1, figsize=(10, 7)) | |
| ts_entry[-plot_length:].plot(ax=ax) | |
| sample_plot(samples, index, prediction_intervals=prediction_intervals) | |
| ax.axvline(ts_entry.index[-prediction_length], color='r') | |
| plt.legend(legend, loc="upper left") | |
| if path is None: | |
| plt.show() | |
| plt.clf() | |
| else: | |
| plt.savefig('{}forecast.png'.format(path)) | |
| plt.close() | |
| def arg_parse(): | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("-p", | |
| "--prediction-length", | |
| type=int, | |
| required=True, | |
| help="Prediction length of the forecast") | |
| parser.add_argument("-f", | |
| "--freq", | |
| required=True, | |
| help="Prediction frequency of the forecast") | |
| parser.add_argument("-t", | |
| "--start-time", | |
| required=True, | |
| help="Start of the forecast") | |
| parser.add_argument("--target-path", | |
| required=True, | |
| help="Target NDArray path") | |
| parser.add_argument("--samples-path", | |
| required=True, | |
| help="Samples NDArray path") | |
| parser.add_argument("-o", | |
| "--output-dir", | |
| help="Plot output directory") | |
| return parser.parse_args() | |
| if __name__ == "__main__": | |
| args = arg_parse() | |
| target = np.load(args.target_path)['target'] | |
| start_date = pd.Period(args.start_time, freq=args.freq) | |
| samples = np.load(args.samples_path)['samples'] | |
| index = pd.period_range(start_date + target.shape[-1] - args.prediction_length, periods=args.prediction_length, | |
| freq=args.freq) | |
| ts = pd.Series(data=target, index=pd.period_range(start_date, periods=target.shape[-1], freq=start_date.freq)) | |
| if args.output_dir is not None: | |
| plot_log_path = args.output_dir | |
| directory = os.path.dirname(plot_log_path) | |
| if not os.path.exists(directory): | |
| os.makedirs(directory) | |
| else: | |
| plot_log_path = None | |
| print("Plotting time series predictions ...") | |
| plot_prob_forecasts(ts, samples, index, args.prediction_length, plot_log_path) |