构建电力使用预测模型及实现网络预测服务

### 构建电力使用预测模型及实现网络预测服务 #### 构建模型 首先将训练数据和测试数据写入S3存储桶： ```python write_dicts_to_s3(f'{s3_data_path}/train/train.json', training_data) write_dicts_to_s3(f'{s3_data_path}/test/test.json', test_data) ``` 为了提升时间序列预测的准确性，我们会使用额外的数据集，如站点类别、节假日和最高温度等。为了让你看到这些额外数据集对预测的影响，有一个注释掉的笔记本单元格，用于在不使用额外数据集的情况下创建和测试模型。若取消注释并重新运行整个笔记本，会发现不使用额外数据集时，2月的平均绝对百分比误差（MAPE）为20%。 #### 配置模型和服务器 - **设置S3输出路径和SageMaker会话** ```python s3_output_path = f's3://{data_bucket}/{subfolder}/output' sess = sagemaker.Session() image_name = sagemaker.amazon.amazon_estimator.get_image_uri( sess.boto_region_name, "forecasting-deepar", "latest") data_channels = { "train": f"{s3_data_path}/train/", "test": f"{s3_data_path}/test/" } ``` - **计算MAPE** MAPE的计算方式是，对于预测的每一天，用实际消耗减去预测消耗，再除以预测消耗（若结果为负则取正），最后取所有这些值的平均值。示例代码如下： ```python def mape(y_true, y_pred): y_true, y_pred = np.array(y_true), np.array(y_pred) return np.mean(np.abs((y_true - y_pred) / y_true)) * 100 ``` - **创建DeepAR模型的标准函数** ```python class DeepARPredictor(sagemaker.predictor.RealTimePredictor): def __init__(self, *args, **kwargs): super().__init__( *args, content_type=sagemaker.content_types.CONTENT_TYPE_JSON, **kwargs) def predict( self, ts, cat=None, dynamic_feat=None, num_samples=100, return_samples=False, quantiles=["0.1", "0.5", "0.9"]): prediction_time = ts.index[-1] + 1 quantiles = [str(q) for q in quantiles] req = self.__encode_request( ts, cat, dynamic_feat, num_samples, return_samples, quantiles) res = super(DeepARPredictor, self).predict(req) return self.__decode_response( res, ts.index.freq, prediction_time, return_samples) def __encode_request( self, ts, cat, dynamic_feat, num_samples, return_samples, quantiles): instance = series_to_dict( ts, cat if cat is not None else None, dynamic_feat if dynamic_feat else None) configuration = { "num_samples": num_samples, "output_types": [ "quantiles", "samples"] if return_samples else ["quantiles"], "quantiles": quantiles } http_request_data = { "instances": [instance], "configuration": configuration } return json.dumps(http_request_data).encode('utf-8') def __decode_response( self, response, freq, prediction_time, return_samples): predictions = json.loads( response.decode('utf-8'))['predictions'][0] prediction_length = len(next(iter( predictions['quantiles'].values() ))) predictio ```