電力負荷預測項目：該項目基于歷史的電力負荷數據，訓練XGBoost模型，實現時序預測里多變量單步的電力負荷預測。

一、project目錄

data：獲取到歷史電力負荷數據。

log：根據寫好代碼，在代碼運行過程中會生成相應日志。

model：建立一個空文件夾來存儲訓練好的模型。

二、utils里的兩個工具包

（一）common.py

# -*- coding: utf-8 -*-
import pandas as pd
import numpy as np
#這是哪兒來的？def data_preprocessing(path):"""1.獲取數據源2.時間格式化，轉為2024-12-20 09:00:00這種格式3.按時間升序排序4.去重:param path::return:"""# 1.獲取數據源data = pd.read_csv(path)# 2.時間格式化data['time'] = pd.to_datetime(data['time']).dt.strftime('%Y-%m-%d %H:%M:%S')# 3.按時間升序排序data.sort_values(by='time', inplace=True)# 4.去重data.drop_duplicates(inplace=True)return datadef mean_absolute_percentage_error(y_true, y_pred):"""低版本的sklearn沒有MAPE的計算方法，需要自己定義，高版本的可以直接調用:param y_true: 真實值:param y_pred: 預測值:return: MAPE（平均絕對百分比誤差）"""n = len(y_true)if len(y_pred) != n:raise ValueError("y_true and y_pred have different number of output ")abs_percentage_error = np.abs((y_true - y_pred) / y_true)return np.sum(abs_percentage_error) / n * 100

（二）log.py

# -*- coding: utf-8 -*-
import logging
import osclass Logger(object):# 日志級別關系映射level_relations = {'debug': logging.DEBUG,'info': logging.INFO,'warning': logging.WARNING,'error': logging.ERROR,'crit': logging.CRITICAL}def __init__(self, root_path, log_name, level='info', fmt='%(asctime)s - %(levelname)s: %(message)s'):# 指定日志保存的路徑self.root_path = root_path# 初始logger名稱和格式self.log_name = log_name# 初始格式self.fmt = fmt# 先聲明一個 Logger 對象self.logger = logging.getLogger(log_name)# 設置日志級別self.logger.setLevel(self.level_relations.get(level))def get_logger(self):# 指定對應的 Handler 為 FileHandler 對象， 這個可適用于多線程情況path = os.path.join(self.root_path, 'log')os.makedirs(path, exist_ok=True)file_name = os.path.join(path, self.log_name + '.log')rotate_handler = logging.FileHandler(file_name, encoding="utf-8", mode="a")# Handler 對象 rotate_handler 的輸出格式formatter = logging.Formatter(self.fmt)rotate_handler.setFormatter(formatter)# 將rotate_handler添加到Loggerself.logger.addHandler(rotate_handler)return self.logger

注意：可再另外在utils文件夾下新建一個__init__py，方便在下文src的py中導入utils的工具包方法。

三、src文件夾代碼

（一）模型訓練（train.py）

import os
import pandas as pd
import matplotlib.pyplot as plt
import datetime
from utils.log import Logger
from utils.common import data_preprocessing
from xgboost import XGBRegressor
from sklearn.model_selection import train_test_split,GridSearchCV
from sklearn.metrics import mean_squared_error,mean_absolute_error
import joblibplt.rcParams['font.family'] = ['SimHei']
plt.rcParams['font.size'] = 15def power_data(data):data=data.copy(deep=True)print(data.columns)# print(data.info)print(data.head())fig=plt.figure(figsize=(20,10))ax1=fig.add_subplot(411)ax1.hist(data['power_load'],bins=100)ax1.set_title('負荷分布直方圖')# plt.show()ax2 = fig.add_subplot(412)# print(type(data['time'][1]))#時間是字符串類型data['hour'] = pd.to_datetime(data['time']).dt.hourprint(data.head())data_hour_avg=data.groupby('hour')['power_load'].mean()#data_hour_avg=data.groupby('hour',as_index=False)['power_load'].mean()print(data_hour_avg)  #數據類型為series：以hour為索引，平均值為列# ax2.plot(data_hour_avg['hour'],data_hour_avg['power_load'],color='r',linewidth=2)ax2.plot(data_hour_avg.index,data_hour_avg.values,color='b',linewidth=2)ax2.set_title('每小時平均負荷')ax2.set_xlabel('小時')ax2.set_ylabel('負荷')# plt.show()  #注意：畫圖時，如果有多張圖，plt.show()必須放在最后，同時如果有savefig需求，需要將savefig函數放在show函數前面。ax3 = fig.add_subplot(413)data['month'] = pd.to_datetime(data['time']).dt.monthdata_month_avg=data.groupby('month')['power_load'].mean()print(data_month_avg)ax3.plot(data_month_avg.index,data_month_avg.values,color='r',linewidth=2)ax3.set_title('每月平均負荷')ax3.set_xlabel('月份')ax3.set_ylabel('負荷')# plt.show()ax4 = fig.add_subplot(414)data['weekday'] = pd.to_datetime(data['time']).dt.weekdaydata_weekday_avg=data.groupby('weekday')['power_load'].mean()print(data_weekday_avg)data['is_workday']=(data['weekday']>=4).astype(int) #直接轉換，效率比apply高print(data.head())power_load_holiday_avg=data[data['is_workday']==0]['power_load'].mean()power_load_workday_avg=data[data['is_workday']==1]['power_load'].mean()ax4.bar(x=['工作日平均負荷','周末平均負荷'],height=[power_load_workday_avg,power_load_holiday_avg])ax4.set_title('工作日和休息日的平均負荷對比')ax4.set_ylabel('平均負荷')# plt.show()plt.savefig('../data/fig/負荷分析圖.png')def feature_engineering(data,logger):logger.info("===============開始進行特征工程處理===============")result = data.copy(deep=True)logger.info("===============開始提取時間特征===================")# 1、提取出時間特征# 1.1提取出對應的小時，用以表示短期的時間特征result['hour'] =pd.to_datetime(result['time']).dt.hour# 1.2提取出對應的月份，用以表示長期的時間特征result['month'] =pd.to_datetime(result['time']).dt.month# 1.3 對時間特征進行one-hot編碼# 1.3.1對小時數進行one-hot編碼hour_encoding = pd.get_dummies(result['hour'])hour_encoding.columns = ['hour_' + str(i) for i in hour_encoding.columns]print(hour_encoding)# 1.3.2對月份進行one-hot編碼month_encoding = pd.get_dummies(result['month'])month_encoding.columns = ['month_' + str(i) for i in month_encoding.columns]# 1.3.3 對one-hot編碼后的結果進行拼接result = pd.concat([result, hour_encoding, month_encoding], axis=1)logger.info("==============開始提取相近時間窗口中的負荷特征====================")# 2指定window_size下的相近時間窗口負荷window_size = 3shift_list = [result['power_load'].shift(i) for i in range(1, window_size + 1)]shift_data = pd.concat(shift_list, axis=1)shift_data.columns = ['前' + str(i) + '小時' for i in range(1, window_size + 1)]result = pd.concat([result, shift_data], axis=1)logger.info("============開始提取昨日同時刻負荷特征===========================")# 3提取昨日同時刻負荷特征# 3.1時間與負荷轉為字典time_load_dict = result.set_index('time')['power_load'].to_dict()# 3.2計算昨日相同的時刻result['yesterday_time'] = result['time'].apply(lambda x: (pd.to_datetime(x) - pd.to_timedelta('1d')).strftime('%Y-%m-%d %H:%M:%S'))# 3.3昨日相同的時刻的負荷result['yesterday_load'] = result['yesterday_time'].apply(lambda x: time_load_dict.get(x))# 4.剔除出現空值的樣本result.dropna(axis=0, inplace=True)# 5.整理特征列，并返回time_feature_names = list(hour_encoding.columns) + list(month_encoding.columns) + list(shift_data.columns) + ['yesterday_load']logger.info(f"特征列名是：{time_feature_names}")return result, time_feature_namesdef model_train(data, features, logger):"""1.數據集切分2.網格化搜索與交叉驗證3.模型實例化4.模型訓練5.模型評價6.模型保存:param data: 特征工程處理后的輸入數據:param features: 特征名稱:param logger: 日志對象:return:"""logger.info("=========開始模型訓練===================")# 1.數據集切分x_data = data[features]y_data = data['power_load']# x_train:訓練集特征數據# y_train:訓練集目標數據# x_test:測試集特征數據# y_test:測試集目標數據x_train, x_test, y_train, y_test = train_test_split(x_data, y_data, test_size=0.3, random_state=22)# # 2.網格化搜索與交叉驗證# # 2.1備選的超參數# print("開始網格化搜索")# print(datetime.datetime.now())  # 2024-11-26 15:38:26.898828# param_dict = {#     'n_estimators': [50, 100, 150, 200],#     'max_depth': [3, 6, 9],#     'learning_rate': [0.1, 0.01]# }# # 2.2實例化網格化搜索，配置交叉驗證# grid_cv = GridSearchCV(estimator=XGBRegressor(),#                        param_grid=param_dict, cv=5)# # 2.3網格化搜索與交叉驗證訓練# grid_cv.fit(x_train, y_train)# # 2.4輸出最優的超參數組合# print(grid_cv.best_params_)  # {'learning_rate': 0.1, 'max_depth': 6, 'n_estimators': 150}# print("結束網格化搜索")# print(datetime.datetime.now())  # 2024-11-26 15:39:07.216048# logger.info("網格化搜索后找到的最優的超參數組合是：learning_rate: 0.1, max_depth: 6, n_estimators: 150")# 3.模型訓練xgb = XGBRegressor(n_estimators=150, max_depth=6, learning_rate=0.1)xgb.fit(x_train, y_train)# 4.模型評價# 4.1模型在訓練集上的預測結果y_pred_train = xgb.predict(x_train)# 4.2模型在測試集上的預測結果y_pred_test = xgb.predict(x_test)# 4.3模型在訓練集上的MSE、MAPEmse_train = mean_squared_error(y_true=y_train, y_pred=y_pred_train)mae_train = mean_absolute_error(y_true=y_train, y_pred=y_pred_train)print(f"模型在訓練集上的均方誤差：{mse_train}")print(f"模型在訓練集上的平均絕對誤差：{mae_train}")# 4.4模型在測試集上的MSE、MAPEmse_test = mean_squared_error(y_true=y_test, y_pred=y_pred_test)mae_test = mean_absolute_error(y_true=y_test, y_pred=y_pred_test)print(f"模型在測試集上的均方誤差：{mse_test}")print(f"模型在測試集上的平均絕對誤差：{mae_test}")logger.info("=========================模型訓練完成=============================")logger.info(f"模型在訓練集上的均方誤差：{mse_train}")logger.info(f"模型在訓練集上的平均絕對誤差：{mae_train}")logger.info(f"模型在測試集上的均方誤差：{mse_test}")logger.info(f"模型在測試集上的平均絕對誤差：{mae_test}")# 5.模型保存joblib.dump(xgb, '../model/xgb.pkl')class PowerLoadModel(object):def __init__(self, filename):# 配置日志記錄logfile_name = "train_" + datetime.datetime.now().strftime('%Y%m%d%H%M%S')self.logfile = Logger('../', logfile_name).get_logger()# 獲取數據源self.data_source = data_preprocessing(filename)if __name__ == '__main__':file=r'D:\data\train.csv'# power_data(data)model = PowerLoadModel(file)# 2.分析數據power_data(model.data_source)# 3.特征工程processed_data, feature_cols = feature_engineering(model.data_source, model.logfile)# 4.模型訓練、模型評價與模型保存model_train(processed_data, feature_cols, model.logfile)

（二）模型預測（predict.py）

# -*- coding: utf-8 -*-
import os
import pandas as pd
import numpy as np
import datetime
from utils.log import Logger
from utils.common import data_preprocessing
from sklearn.metrics import mean_absolute_error
import matplotlib.ticker as mick
import joblib
import matplotlib.pyplot as pltplt.rcParams['font.family'] = 'SimHei'
plt.rcParams['font.size'] = 15def pred_feature_extract(data_dict, time, logger):"""預測數據解析特征，保持與模型訓練時的特征列名一致1.解析時間特征2.解析時間窗口特征3.解析昨日同時刻特征:param data_dict:歷史數據，字典格式，key：時間，value:負荷:param time:預測時間，字符串類型，格式為2024-12-20 09:00:00:param logger:日志對象:return:"""logger.info(f'=========解析預測時間為：{time}所對應的特征==============')# 特征列清單feature_names = ['hour_0', 'hour_1', 'hour_2', 'hour_3', 'hour_4', 'hour_5','hour_6', 'hour_7', 'hour_8', 'hour_9', 'hour_10', 'hour_11','hour_12', 'hour_13', 'hour_14', 'hour_15', 'hour_16', 'hour_17','hour_18', 'hour_19', 'hour_20', 'hour_21', 'hour_22', 'hour_23','month_1', 'month_2', 'month_3', 'month_4', 'month_5', 'month_6','month_7', 'month_8', 'month_9', 'month_10', 'month_11', 'month_12','前1小時', '前2小時', '前3小時', 'yesterday_load']# 小時特征數據，使用列表保存起來hour_part = []pred_hour =pd.to_datetime(time).hourprint(pred_hour)for i in range(24):if pred_hour == feature_names[i][5:7]:hour_part.append(1)else:hour_part.append(0)# 月份特征數據，使用列表保存起來month_part = []pred_month = pd.to_datetime(time).monthfor i in range(24, 36):if pred_month == feature_names[i][6:8]:month_part.append(1)else:month_part.append(0)# 歷史負荷數據，使用列表保存起來his_part = []# 前1小時負荷last_1h_time = (pd.to_datetime(time) - pd.to_timedelta('1h')).strftime('%Y-%m-%d %H:%M:%S')last_1h_load = data_dict.get(last_1h_time, 600)# 前2小時負荷last_2h_time = (pd.to_datetime(time) - pd.to_timedelta('2h')).strftime('%Y-%m-%d %H:%M:%S')last_2h_load = data_dict.get(last_2h_time, 600)# 前3小時負荷last_3h_time = (pd.to_datetime(time) - pd.to_timedelta('3h')).strftime('%Y-%m-%d %H:%M:%S')last_3h_load = data_dict.get(last_3h_time, 600)# 昨日同時刻負荷last_day_time = (pd.to_datetime(time) - pd.to_timedelta('1d')).strftime('%Y-%m-%d %H:%M:%S')last_day_load = data_dict.get(last_day_time, 600)his_part = [last_1h_load, last_2h_load, last_3h_load, last_day_load]# 特征數據，包含小時特征數據，月份特征數據，歷史負荷數據feature_list = [hour_part + month_part + his_part]# feature_list需要轉成dataframe并返回，所以這里用append變成一個二維列表feature_df = pd.DataFrame(feature_list, columns=feature_names)return feature_df, feature_namesdef prediction_plot(data):"""繪制時間與預測負荷折線圖，時間與真實負荷折線圖，展示預測效果:param data: 數據一共有三列：時間、真實值、預測值:return:"""# 繪制在新數據下fig = plt.figure(figsize=(40, 20))ax = fig.add_subplot()# 繪制時間與真實負荷的折線圖ax.plot(data['時間'], data['真實值'], label='真實值')# 繪制時間與預測負荷的折線圖ax.plot(data['時間'], data['預測值'], label='預測值')ax.set_ylabel('負荷')ax.set_title('預測負荷以及真實負荷的折線圖')# 橫坐標時間若不處理太過密集，這里調大時間展示的間隔ax.xaxis.set_major_locator(mick.MultipleLocator(50))# 時間展示時旋轉45度plt.xticks(rotation=45)plt.legend()plt.savefig('../data/fig/預測效果.png')class PowerLoadPredict(object):def __init__(self, filename):# 配置日志記錄logfile_name = "predict_" + datetime.datetime.now().strftime('%Y%m%d%H%M%S')self.logfile = Logger('../', logfile_name).get_logger()# 獲取數據源self.data_source = data_preprocessing(filename)# 歷史數據轉為字典，key:時間，value:負荷，目的是為了避免頻繁操作dataframe，提高效率。實際開發場景中可以使用redis進行緩存self.data_dict = self.data_source.set_index('time')['power_load'].to_dict()if __name__ == '__main__':"""模型預測1.導包、配置繪圖字體2.定義電力負荷預測類，配置日志，獲取數據源、歷史數據轉為字典（避免頻繁操作dataframe，提高效率）3.加載模型4.模型預測4.1 確定要預測的時間段（2015-08-01 00:00:00及以后的時間）4.2 為了模擬實際場景的預測，把要預測的時間以及以后的負荷都掩蓋掉，因此新建一個數據字典，只保存預測時間以前的數據字典4.3 預測負荷4.3.1 解析特征（定義解析特征方法）4.3.2 利用加載的模型預測4.4 保存預測時間對應的真實負荷4.5 結果保存到evaluate_list，三個元素分別是預測時間、真實負荷、預測負荷，方便后續進行預測結果評價4.6 循環結束后，evaluate_list轉為DataFrame5.預測結果評價5.1 計算預測結果與真實結果的MAE5.2 繪制折線圖（預測時間-真實負荷折線圖，預測時間-預測負荷折線圖），查看預測效果 """# 2.定義電力負荷預測類(PowerLoadPredict)，配置日志，獲取數據源、歷史數據轉為字典（避免頻繁操作dataframe，提高效率）input_file = r'D:\data\test.csv'pred_obj = PowerLoadPredict(input_file)# 3.加載模型model = joblib.load('../model/xgb.pkl')# 4.模型預測evaluate_list = []# 4.1確定要預測的時間段：2015-08-01 00:00:00及以后的時間pred_times = pred_obj.data_source[pred_obj.data_source['time'] >= '2015-08-01 00:00:00']['time']print(pred_times)print(type(pred_times))for pred_time in pred_times:print(f"開始預測時間為：{pred_time}的負荷")pred_obj.logfile.info(f"開始預測時間為：{pred_time}的負荷")# 4.2為了模擬實際場景的預測，把要預測的時間以及以后的負荷都掩蓋掉，因此新建一個數據字典，只保存預測時間以前的數據字典data_his_dict = {k: v for k, v in pred_obj.data_dict.items() if k < pred_time}# 4.3預測負荷# 4.3.1解析特征processed_data, feature_cols = pred_feature_extract(data_his_dict, pred_time, pred_obj.logfile)print(type(processed_data))print(type(feature_cols))print(processed_data)print(feature_cols)# 4.3.2 模型預測pred_value = model.predict(processed_data[feature_cols])# 4.4真實負荷true_value = pred_obj.data_dict.get(pred_time)pred_obj.logfile.info(f"真實負荷為：{true_value}, 預測負荷為：{pred_value}")# 4.5結果保存到evaluate_list，三個元素分別是預測時間、真實負荷、預測負荷evaluate_list.append([pred_time, true_value, pred_value[0]])# 4.6evaluate_list轉為DataFrameevaluate_df = pd.DataFrame(evaluate_list, columns=['時間', '真實值', '預測值'])# 5.預測結果評價# 5.1計算預測結果與真實結果的MAEmae_score = mean_absolute_error(evaluate_df['真實值'], evaluate_df['預測值'])print(f"模型對新數據進行預測的平均絕對誤差：{mae_score}")pred_obj.logfile.info(f"模型對新數據進行預測的平均絕對誤差：{mae_score}")# 5.2繪制折線圖，查看預測效果prediction_plot(evaluate_df)plt.show()

四、代碼整體：

備注：data里的train和test數據另存了位置，所以上面data文件里沒有。

五、最終運行生成圖像

（一）負荷分析圖

（二）預測效果圖

今天的分享到此結束。