Spaces:
Running
Running
| """ | |
| This function is adapted from [chronos-forecasting] by [lostella et al.] | |
| Original source: [https://github.com/amazon-science/chronos-forecasting] | |
| """ | |
| from autogluon.timeseries import TimeSeriesPredictor | |
| from sklearn.preprocessing import MinMaxScaler | |
| import numpy as np | |
| import pandas as pd | |
| import tempfile | |
| from .base import BaseDetector | |
| class Chronos(BaseDetector): | |
| def __init__(self, | |
| win_size=100, | |
| model_size = 'base', # [tiny, small, base] | |
| prediction_length=1, | |
| input_c=1, | |
| batch_size=128): | |
| self.model_name = 'Chronos' | |
| self.model_size = model_size | |
| self.win_size = win_size | |
| self.prediction_length = prediction_length | |
| self.input_c = input_c | |
| self.batch_size = batch_size | |
| self.score_list = [] | |
| def fit(self, data): | |
| for channel in range(self.input_c): | |
| data_channel = data[:, channel].reshape(-1, 1) | |
| data_win, data_target = self.create_dataset(data_channel, slidingWindow=self.win_size, predict_time_steps=self.prediction_length) | |
| # print('data_win: ', data_win.shape) # (2330, 100) | |
| # print('data_target: ', data_target.shape) # (2330, 1) | |
| train_data = [] | |
| count = 0 | |
| for id in range(data_win.shape[0]): | |
| for tt in range(data_win.shape[1]): | |
| train_data.append([id, count, data_win[id, tt]]) | |
| count += 1 | |
| train_data = pd.DataFrame(train_data, columns=['item_id', 'timestamp', 'target']) | |
| with tempfile.TemporaryDirectory() as temp_dir: | |
| predictor = TimeSeriesPredictor(prediction_length=self.prediction_length, path=temp_dir).fit( | |
| train_data, | |
| hyperparameters={ | |
| "Chronos": { | |
| "model_path": self.model_size, # base | |
| "device": "cuda", | |
| "batch_size": self.batch_size}}, | |
| skip_model_selection=True, | |
| verbosity=0) | |
| predictions = predictor.predict(train_data)['mean'].to_numpy().reshape(-1, self.prediction_length) | |
| print('predictions: ', predictions.shape) | |
| ### using mse as the anomaly score | |
| scores = (data_target.squeeze() - predictions.squeeze()) ** 2 | |
| self.score_list.append(scores) | |
| scores_merge = np.mean(np.array(self.score_list), axis=0) | |
| # print('scores_merge: ', scores_merge.shape) | |
| padded_decision_scores = np.zeros(len(data)) | |
| padded_decision_scores[: self.win_size+self.prediction_length-1] = scores_merge[0] | |
| padded_decision_scores[self.win_size+self.prediction_length-1 : ]=scores_merge | |
| self.decision_scores_ = padded_decision_scores | |
| def decision_function(self, X): | |
| """ | |
| Not used, present for API consistency by convention. | |
| """ | |
| pass | |
| def create_dataset(self, X, slidingWindow, predict_time_steps=1): | |
| Xs, ys = [], [] | |
| for i in range(len(X) - slidingWindow - predict_time_steps+1): | |
| tmp = X[i : i + slidingWindow + predict_time_steps].ravel() | |
| # tmp= MinMaxScaler(feature_range=(0,1)).fit_transform(tmp.reshape(-1,1)).ravel() | |
| x = tmp[:slidingWindow] | |
| y = tmp[slidingWindow:] | |
| Xs.append(x) | |
| ys.append(y) | |
| return np.array(Xs), np.array(ys) |