Amazon SageMaker DeepAR アルゴリズムによる運転速度違反の予測

url = 'https://data.cityofchicago.org/api/views/hhkd-xvj4/rows.csv?accessType=DOWNLOAD'

# シカゴ市のサイトからデータを取得
r = requests.get(url, allow_redirects=True)
open(datafile, 'wb').write(r.content)

# 入力ファイルを読み取り、サンプルの行/列を表示する
pd.set_option('display.max_columns', 500)     
pd.set_option('display.max_rows', 50)    
df = pd.read_csv(open(datafile, 'rb'), encoding='utf-8')
df[df.columns[0:3]]

df['VIOLATION_DT'] = pd.to_datetime(df['VIOLATION DATE'])
df[['ADDRESS', 'VIOLATION_DT', 'VIOLATIONS']]
unique_addresses = df.ADDRESS.unique()
idx = pd.date_range(df.VIOLATION_DT.min(), df.VIOLATION_DT.max())
number_of_addresses = len(unique_addresses)
print('Unique Addresses {}'.format(number_of_addresses))
print('Minimum violation date is {}, maximum violation date is {}'.format(df.VIOLATION_DT.min(),df.VIOLATION_DT.max()))

violation_list = []
for key in unique_addresses:
    temp_df = df[['VIOLATION_DT', 'VIOLATIONS']][df.ADDRESS == key]
    temp_df.set_index(['VIOLATION_DT'], inplace=True)
    temp_df.index = pd.DatetimeIndex(temp_df.index)
    temp_df = temp_df.reindex(idx, fill_value=0)
    violation_list.append(temp_df['VIOLATIONS'])

plt.figure(figsize=(12,6), dpi=100, facecolor='w')
for key, address in enumerate(unique_addresses):
    plt.plot(violation_list[key], label=address)

plt.ylabel('Violations')
plt.xlabel('Date')
plt.title('Chicago Speed Camera Violations')
plt.legend(loc='upper center', bbox_to_anchor=(0.5, -0.05), shadow=False, ncol=4)
plt.show()

prediction_length = 30

# トレーニングと検証のためにデータを分割する/差し控える
violation_list_training = []
for i in violation_list:
    violation_list_training.append((i[:-prediction_length]))

def series_to_obj(ts, cat=None):
    obj = {'start': str(ts.index[0]), 'target': list(ts)}
    if cat:
        obj['cat'] = cat
    return obj

def series_to_jsonline(ts, cat=None):
    return json.dumps(series_to_obj(ts, cat))

encoding = 'utf-8'
s3filesystem = s3fs.S3FileSystem()

with s3filesystem.open(train_data_path, 'wb') as fp:
    for ts in violation_list_training:
        fp.write(series_to_jsonline(ts).encode(encoding))
        fp.write('\n'.encode(encoding))

with s3filesystem.open(test_data_path, 'wb') as fp:
    for ts in violation_list:
        fp.write(series_to_jsonline(ts).encode(encoding))
        fp.write('\n'.encode(encoding))

from sagemaker.tuner import IntegerParameter, CategoricalParameter, ContinuousParameter, HyperparameterTuner

deepar = sagemaker.estimator.Estimator(image_name,
                                       role,
                                       train_instance_count=1,
                                       train_instance_type='ml.m4.xlarge',
                                       train_use_spot_instances=True, # use spot instances
                                       train_max_run=3600, # max training time in seconds
                                       train_max_wait=3600, # seconds to wait for spot instance
                                       output_path='s3://{}/{}'.format(bucket, s3_output_path),
                                       sagemaker_session=sess)
freq = 'D'
context_length = 30

deepar.set_hyperparameters(time_freq=freq,
                           context_length=str(context_length),
                           prediction_length=str(prediction_length))

hyperparameter_ranges = {'mini_batch_size': IntegerParameter(100, 400),
                         'epochs': IntegerParameter(200, 400),
                         'num_cells': IntegerParameter(30,100),
                         'likelihood': CategoricalParameter(['negative-binomial', 'student-T']),
                         'learning_rate': ContinuousParameter(0.0001, 0.1)}

objective_metric_name = 'test:RMSE'

tuner = HyperparameterTuner(deepar,
                            objective_metric_name,
                            hyperparameter_ranges,
                            max_jobs=10,
                            strategy='Bayesian',
                            objective_type='Minimize',
                            max_parallel_jobs=10,
                            early_stopping_type='Auto')

s3_input_train = sagemaker.s3_input(s3_data='s3://{}/{}/train/'.format(bucket, prefix),
                                    content_type='json')
s3_input_test = sagemaker.s3_input(s3_data='s3://{}/{}/test/'.format(bucket, prefix),
                                   content_type='json')

tuner.fit({'train': s3_input_train, 'test': s3_input_test},
          include_cls_metadata=False)
tuner.wait()

best_tuning_job_name = tuner.best_training_job()
endpoint_name = tuner.deploy(initial_instance_count=1,
                             endpoint_name=best_tuning_job_name,
                             instance_type='ml.m4.xlarge',
                             wait=True)

…	
トレーニング秒数: 674
請求可能秒数: 242
管理型スポットトレーニングの節約: 64.1%

class DeepARPredictor(sagemaker.predictor.RealTimePredictor):
    
    def set_prediction_parameters(self, freq, prediction_length):
        """ 時間頻度と予測長のパラメータを設定します。このメソッドは、 
        `predict` を使用する前に呼び出す **必要があります**。
        
        パラメータ:
        freq -- 時間頻度を示す文字列
        prediction_length -- 整数、予測時点の数
       
       戻り値: なし。
       """
        self.freq = freq
        self.prediction_length = prediction_length
        
    def predict(self, ts, cat=None, encoding='utf-8', num_samples=100, quantiles=['0.1', '0.5', '0.9']):
        """ `ts` にリストされた時系列の予測をリクエストします。それぞれに 
        (オプション) 対応するカテゴリが `cat` にリストされています。
        
        パラメータ:
        ts -- `pandas.Series` オブジェクトのリスト、予測する時系列
        cat -- 整数のリスト (デフォルト: なし)
        encoding -- 文字列、リクエストに使用するエンコーディング (デフォルト: 'utf-8')
        num_samples -- 整数、予測時に計算するサンプル数 (デフォルト: 100)
        quantiles -- 計算する分位数を指定する文字列のリスト (デフォルト: ['0.1', '0.5', '0.9'])
        
        戻り値: それぞれが予測を含む `pandas.DataFrame` オブジェクトのリスト
        """
        prediction_times = [x.index[-1]+1 for x in ts]
        req = self.__encode_request(ts, cat, encoding, num_samples, quantiles)
        res = super(DeepARPredictor, self).predict(req)
        return self.__decode_response(res, prediction_times, encoding)
    
    def __encode_request(self, ts, cat, encoding, num_samples, quantiles):
        instances = [series_to_obj(ts[k], cat[k] if cat else None) for k in range(len(ts))]
        configuration = {'num_samples': num_samples, 'output_types': ['quantiles'], 'quantiles': quantiles}
        http_request_data = {'instances': instances, 'configuration': configuration}
        return json.dumps(http_request_data).encode(encoding)
    
    def __decode_response(self, response, prediction_times, encoding):
        response_data = json.loads(response.decode(encoding))
        list_of_df = []
        for k in range(len(prediction_times)):
            prediction_index = pd.DatetimeIndex(start=prediction_times[k], freq=self.freq, periods=self.prediction_length)
            list_of_df.append(pd.DataFrame(data=response_data['predictions'][k]['quantiles'], index=prediction_index))
        return list_of_df


predictor = DeepARPredictor(endpoint=best_tuning_job_name,
                            sagemaker_session=sagemaker_session,
                            content_type="application/json")

predictor.set_prediction_parameters(freq, prediction_length)
list_of_df = predictor.predict(violation_list_training[:5])
actual_data = violation_list[:5]
for k in range(len(list_of_df)):
    plt.figure(figsize=(12,6), dpi=75, facecolor='w')
    plt.ylabel('Violations')
    plt.xlabel('Date')
    plt.title('Chicago Speed Camera Violations:' + unique_addresses[k])
    actual_data[k][-prediction_length-context_length:].plot(label='target')
    p10 = list_of_df[k]['0.1']
    p90 = list_of_df[k]['0.9']
    plt.fill_between(p10.index, p10, p90, color='y', alpha=0.5,label='80% confidence interval')
    list_of_df[k]['0.5'].plot(label='prediction median')

predictor.delete_endpoint(endpoint_name)