电力负荷案例之预测类代码实现

history_dict = dict(zip(data['time'], data['power_load']))

import joblib

model = joblib.load('model/power_load_model.pkl')

import os
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import logging
import joblib
import datetime
from sklearn.metrics import mean_absolute_error, mean_squared_error

# 解决中文乱码
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# 1. 日志工具类
print("=== 1. 日志工具类 ===")

class LogUtils:
    """日志工具类"""
    
    def __init__(self, root_path='./', log_name='project', level=logging.INFO):
        self.logger = logging.getLogger(log_name)
        self.logger.setLevel(level)
        
        if not self.logger.handlers:
            log_dir = os.path.join(root_path, 'log')
            os.makedirs(log_dir, exist_ok=True)
            log_file = os.path.join(log_dir, f'{log_name}.log')
            
            file_handler = logging.FileHandler(log_file, encoding='utf-8')
            file_handler.setLevel(level)
            
            console_handler = logging.StreamHandler()
            console_handler.setLevel(level)
            
            formatter = logging.Formatter(
                '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
            )
            file_handler.setFormatter(formatter)
            console_handler.setFormatter(formatter)
            
            self.logger.addHandler(file_handler)
            self.logger.addHandler(console_handler)
    
    def get_log(self):
        return self.logger

# 2. 生成模拟数据
print("\n=== 2. 生成模拟数据 ===")

def generate_power_load_data(n_days=30):
    """生成电力负荷数据"""
    np.random.seed(42)
    
    dates = pd.date_range('2024-01-01', periods=n_days*24, freq='h')
    power_load = []
    
    for date in dates:
        hour = date.hour
        dayofweek = date.dayofweek
        month = date.month
        
        if 0 <= hour < 6:
            base_load = 500
        elif 6 <= hour < 9:
            base_load = 800
        elif 9 <= hour < 17:
            base_load = 1000
        elif 17 <= hour < 21:
            base_load = 1200
        else:
            base_load = 700
        
        if dayofweek >= 5:
            base_load *= 0.8
        
        if month in [12, 1, 2]:
            base_load *= 1.2
        elif month in [6, 7, 8]:
            base_load *= 1.1
        
        load = base_load + np.random.normal(0, 50)
        power_load.append(max(100, load))
    
    data = pd.DataFrame({
        'time': dates,
        'power_load': power_load
    })
    
    return data

# 3. 特征工程函数（复用）
print("\n=== 3. 特征工程函数 ===")

def feature_engineering(data, logger=None):
    """特征工程"""
    if logger:
        logger.info("开始特征工程")
    
    feature_data = data.copy()
    
    # 1. 提取hour和month
    feature_data['hour'] = feature_data['time'].dt.hour
    feature_data['month'] = feature_data['time'].dt.month
    
    # 2. One-Hot编码
    feature_data = pd.get_dummies(feature_data, columns=['hour', 'month'])
    
    # 3. 添加窗口字段
    feature_data['prev_1'] = feature_data['power_load'].shift(1)
    feature_data['prev_2'] = feature_data['power_load'].shift(2)
    feature_data['prev_3'] = feature_data['power_load'].shift(3)
    
    # 4. 获取昨日同时刻负荷
    feature_data['yesterday_time'] = feature_data['time'].apply(
        lambda x: (pd.to_datetime(x) - pd.Timedelta(days=1)).strftime('%Y-%m-%d %H:%M:%S')
    )
    
    time_load_map = dict(zip(
        feature_data['time'].dt.strftime('%Y-%m-%d %H:%M:%S'),
        feature_data['power_load']
    ))
    
    feature_data['yesterday_load'] = feature_data['yesterday_time'].map(time_load_map)
    
    # 5. 删除NaN
    feature_data = feature_data.dropna().reset_index(drop=True)
    
    return feature_data

# 4. 训练模型（如果模型不存在）
print("\n=== 4. 训练模型 ===")

def train_and_save_model(data, logger=None):
    """训练并保存模型"""
    from sklearn.ensemble import GradientBoostingRegressor
    from sklearn.model_selection import train_test_split
    
    # 特征工程
    feature_data = feature_engineering(data, logger)
    
    # 获取特征列名
    exclude_columns = ['time', 'power_load', 'yesterday_time']
    features = [col for col in feature_data.columns if col not in exclude_columns]
    
    # 数据集切分
    X = feature_data[features]
    y = feature_data['power_load']
    
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, shuffle=False
    )
    
    # 训练模型
    model = GradientBoostingRegressor(
        n_estimators=100,
        max_depth=6,
        learning_rate=0.1,
        random_state=42
    )
    model.fit(X_train, y_train)
    
    # 保存模型
    os.makedirs('model', exist_ok=True)
    joblib.dump(model, 'model/power_load_model.pkl')
    
    if logger:
        logger.info("模型训练并保存完成")
    
    print("模型训练并保存完成")
    
    return model, features

# 5. 电力负荷预测类
print("\n=== 5. 电力负荷预测类 ===")

class PowerLoadPredict:
    """电力负荷预测类"""
    
    def __init__(self, log_file=None, data_source=None):
        """
        初始化
        Args:
            log_file: 日志文件
            data_source: 数据源
        """
        self.log_file = log_file
        self.data_source = data_source
        
        # 配置日志
        if log_file:
            log_name = os.path.splitext(os.path.basename(log_file))[0]
            log_dir = os.path.dirname(log_file)
            self.log_utils = LogUtils(
                root_path=log_dir if log_dir else './',
                log_name=log_name
            )
            self.logger = self.log_utils.get_log()
        else:
            self.logger = logging.getLogger('power_load_predict')
    
    def get_data_source(self):
        """获取数据源"""
        if self.data_source is not None:
            self.logger.info(f"获取数据源: {self.data_source.shape}")
            return self.data_source
        else:
            self.logger.warning("数据源为空")
            return None
    
    def history_to_dict(self, data):
        """
        历史数据转字典
        避免频繁操作DataFrame
        """
        self.logger.info("历史数据转字典")
        
        # 转成字典: {时间: 负荷}
        history_dict = dict(zip(
            data['time'].dt.strftime('%Y-%m-%d %H:%M:%S'),
            data['power_load']
        ))
        
        self.logger.info(f"历史字典大小: {len(history_dict)}")
        
        return history_dict
    
    def load_model(self, model_path='model/power_load_model.pkl'):
        """加载模型"""
        self.logger.info(f"加载模型: {model_path}")
        
        try:
            model = joblib.load(model_path)
            self.logger.info("模型加载成功")
            return model
        except Exception as e:
            self.logger.error(f"模型加载失败: {e}")
            raise
    
    def mask_data(self, data, predict_time):
        """
        掩盖数据
        预测时间及以后的负荷要掩盖
        """
        self.logger.info(f"掩盖数据，预测时间: {predict_time}")
        
        # 只保留预测时间以前的数据
        masked_data = data[data['time'] < predict_time].copy()
        
        self.logger.info(f"掩盖后数据量: {len(masked_data)}")
        
        return masked_data
    
    def predict(self, data, model, features, predict_time):
        """
        模型预测
        Args:
            data: 历史数据
            model: 训练好的模型
            features: 特征列名
            predict_time: 预测时间
        Returns:
            predicted_load: 预测的负荷
        """
        self.logger.info(f"开始预测: {predict_time}")
        
        # 1. 掩盖数据
        masked_data = self.mask_data(data, predict_time)
        
        # 2. 特征工程
        feature_data = feature_engineering(masked_data, self.logger)
        
        # 3. 获取最后一行数据（预测时间的前一个时间步）
        last_data = feature_data.iloc[-1:]
        
        # 4. 提取特征
        X = last_data[features]
        
        # 5. 预测
        predicted_load = model.predict(X)[0]
        
        self.logger.info(f"预测结果: {predicted_load:.2f}")
        
        return predicted_load
    
    def evaluate(self, data, model, features, predict_start_time, predict_end_time):
        """
        评估预测
        Args:
            data: 完整数据
            model: 模型
            features: 特征列名
            predict_start_time: 预测开始时间
            predict_end_time: 预测结束时间
        Returns:
            evaluate_list: 评估结果列表
        """
        self.logger.info(f"开始评估: {predict_start_time} ~ {predict_end_time}")
        
        evaluate_list = []
        
        # 生成预测时间列表
        predict_times = pd.date_range(
            start=predict_start_time,
            end=predict_end_time,
            freq='h'
        )
        
        for predict_time in predict_times:
            # 真实值
            true_load = data[data['time'] == predict_time]['power_load'].values
            true_load = true_load[0] if len(true_load) > 0 else None
            
            # 预测值
            predicted_load = self.predict(data, model, features, predict_time)
            
            evaluate_list.append({
                'time': predict_time,
                'true_load': true_load,
                'predicted_load': predicted_load
            })
            
            self.logger.info(f"时间: {predict_time}, 真实: {true_load}, 预测: {predicted_load:.2f}")
        
        return evaluate_list

# 6. 运行预测
print("\n=== 6. 运行预测 ===")

def run_prediction():
    """运行预测"""
    # 创建日志对象
    log_time = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
    log_file = f'log/predict_{log_time}.log'
    
    # 生成数据
    data = generate_power_load_data(n_days=30)
    
    # 创建预测类
    predictor = PowerLoadPredict(log_file=log_file, data_source=data)
    
    # 训练模型（实际中应该加载已保存的模型）
    model, features = train_and_save_model(data, predictor.logger)
    
    # 预测
    predict_start_time = pd.Timestamp('2024-01-29 00:00:00')
    predict_end_time = pd.Timestamp('2024-01-29 23:00:00')
    
    evaluate_list = predictor.evaluate(
        data, model, features, predict_start_time, predict_end_time
    )
    
    # 转成DataFrame
    result_df = pd.DataFrame(evaluate_list)
    
    # 计算评估指标
    mae = mean_absolute_error(result_df['true_load'], result_df['predicted_load'])
    mse = mean_squared_error(result_df['true_load'], result_df['predicted_load'])
    rmse = np.sqrt(mse)
    
    print(f"\n预测评估:")
    print(f"  MAE: {mae:.4f}")
    print(f"  MSE: {mse:.4f}")
    print(f"  RMSE: {rmse:.4f}")
    
    # 可视化
    plt.figure(figsize=(14, 6))
    plt.plot(result_df['time'], result_df['true_load'], label='真实值', marker='o')
    plt.plot(result_df['time'], result_df['predicted_load'], label='预测值', marker='x')
    plt.xlabel('时间')
    plt.ylabel('电力负荷')
    plt.title('电力负荷预测结果')
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.xticks(rotation=45)
    plt.tight_layout()
    plt.savefig('prediction_evaluation.png', dpi=100)
    print("预测评估图已保存")
    
    return result_df

result = run_prediction()

# 7. 预测类结构详解
print("\n=== 7. 预测类结构详解 ===")

def predict_class_structure():
    """预测类结构详解"""
    print("""
PowerLoadPredict 类结构:

1. __init__(self, log_file, data_source)
   - 初始化
   - 配置日志
   - 获取数据源

2. get_data_source(self)
   - 获取数据源

3. history_to_dict(self, data)
   - 历史数据转字典
   - 避免频繁操作DataFrame

4. load_model(self, model_path)
   - 加载模型
   - joblib.load

5. mask_data(self, data, predict_time)
   - 掩盖数据
   - 预测时间及以后的负荷掩盖

6. predict(self, data, model, features, predict_time)
   - 模型预测
   - 返回预测值

7. evaluate(self, data, model, features, start, end)
   - 评估预测
   - 返回评估结果列表
""")

predict_class_structure()

# 8. 掩盖数据详解
print("\n=== 8. 掩盖数据详解 ===")

def mask_data_explanation():
    """掩盖数据详解"""
    print("""
掩盖数据详解:

1. 目的
   - 模拟实际场景预测
   - 防止模型"抄袭"真实值

2. 原理
   - 预测某个时间点的负荷
   - 该时间点及以后的数据要掩盖
   - 只能看到预测时间以前的数据

3. 示例
   预测4点的负荷:
   - 可见: 0点、1点、2点、3点
   - 掩盖: 4点、5点、6点...
   
   如果不掩盖:
   - 模型可能直接看到4点的真实值
   - 属于"抄袭"
   - 评估结果不真实

4. 代码
   masked_data = data[data['time'] < predict_time]
""")

mask_data_explanation()

# 9. 历史数据转字典
print("\n=== 9. 历史数据转字典 ===")

def history_dict_explanation():
    """历史数据转字典详解"""
    print("""
历史数据转字典详解:

1. 原因
   - 避免频繁操作DataFrame
   - DataFrame有40多列，操作耗时
   - 字典只存时间和负荷两列
   - 提高效率

2. 方法
   history_dict = dict(zip(
       data['time'],
       data['power_load']
   ))

3. 使用
   # 根据时间获取负荷
   load = history_dict[time]

4. 优势
   - 查找速度快
   - 内存占用小
   - 操作简单
""")

history_dict_explanation()

# 10. 完整预测代码
print("\n=== 10. 完整预测代码 ===")

def complete_predict_code():
    """完整预测代码"""
    print("""
src/predict.py 完整代码:

import os
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import logging
import joblib
import datetime
from utils.log import LogUtils
from utils.common import data_preprocessing
from sklearn.metrics import mean_absolute_error, mean_squared_error

# 解决中文乱码
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

class PowerLoadPredict:
    def __init__(self, log_file, data_source):
        self.log_file = log_file
        self.data_source = data_source
        # 配置日志
    
    def get_data_source(self):
        # 获取数据源
        pass
    
    def history_to_dict(self, data):
        # 历史数据转字典
        history_dict = dict(zip(data['time'], data['power_load']))
        return history_dict
    
    def load_model(self, model_path):
        # 加载模型
        model = joblib.load(model_path)
        return model
    
    def mask_data(self, data, predict_time):
        # 掩盖数据
        masked_data = data[data['time'] < predict_time]
        return masked_data
    
    def predict(self, data, model, features, predict_time):
        # 预测
        masked_data = self.mask_data(data, predict_time)
        feature_data = feature_engineering(masked_data)
        X = feature_data.iloc[-1:][features]
        predicted_load = model.predict(X)[0]
        return predicted_load
    
    def evaluate(self, data, model, features, start, end):
        # 评估
        evaluate_list = []
        # ...
        return evaluate_list

def main():
    # 1. 配置日志
    # 2. 获取数据源
    # 3. 加载模型
    # 4. 模型预测
    # 5. 结果保存

if __name__ == '__main__':
    main()
""")

complete_predict_code()

# 11. 总结
def predict_summary():
    """预测总结"""
    print("=" * 50)
    print("预测类代码实现总结")
    print("=" * 50)
    
    print("\n1. 文件")
    print("   src/predict.py")
    
    print("\n2. 类名")
    print("   PowerLoadPredict")
    
    print("\n3. 方法")
    print("   get_data_source: 获取数据源")
    print("   history_to_dict: 历史数据转字典")
    print("   load_model: 加载模型")
    print("   mask_data: 掩盖数据")
    print("   predict: 模型预测")
    print("   evaluate: 评估预测")
    
    print("\n4. 关键点")
    print("   历史数据转字典提高效率")
    print("   掩盖数据防止抄袭")
    print("   预测时间段: 8月1号以后")
    
    print("\n" + "=" * 50)
    print("预测类实现完成!")
    print("=" * 50)

predict_summary()