电力负荷案例之解析预测特征(上)

pre_times = pp.data_source[
    pp.data_source['time'] >= '2015-08-01'
]

# 建立数据字典，只保存预测时间及以前的数据
time_load_mask = {}
for time in pre_times:
    # 只保存预测时间以前的数据
    mask_data = data[data['time'] < time]
    time_load_mask[time] = mask_data

import os
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import logging
import joblib
import datetime

# 解决中文乱码
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. 电力负荷预测类 ===")

class PowerLoadPredict:
    """电力负荷预测类"""
    
    def __init__(self, log_file=None, data_source=None):
        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_predict_times(self, start_time='2024-01-29'):
        """
        筛选预测时间段
        Args:
            start_time: 预测开始时间
        Returns:
            pre_times: 预测时间列表
        """
        self.logger.info(f"筛选预测时间段: {start_time}以后")
        
        # 筛选8月1号以后的数据
        pre_times = self.data_source[
            self.data_source['time'] >= start_time
        ]['time'].values
        
        self.logger.info(f"预测时间数量: {len(pre_times)}")
        
        return pre_times
    
    def mask_data(self, predict_time):
        """
        掩盖数据
        只保存预测时间及以前的数据
        Args:
            predict_time: 预测时间
        Returns:
            masked_data: 掩盖后的数据
        """
        # 用小于号，不包含预测时间本身
        masked_data = self.data_source[self.data_source['time'] < predict_time].copy()
        
        return masked_data
    
    def build_time_load_dict(self, data):
        """
        建立时间-负荷字典
        Args:
            data: 数据
        Returns:
            time_load_dict: 字典 {时间: 负荷}
        """
        time_load_dict = dict(zip(
            data['time'].dt.strftime('%Y-%m-%d %H:%M:%S'),
            data['power_load']
        ))
        
        return time_load_dict

# 4. 运行预测特征解析
print("\n=== 4. 运行预测特征解析 ===")

def run_prediction_feature_parsing():
    """运行预测特征解析"""
    # 创建日志对象
    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)
    
    # 1. 筛选预测时间段
    print("--- 1. 筛选预测时间段 ---")
    pre_times = predictor.get_predict_times(start_time='2024-01-29')
    print(f"预测时间数量: {len(pre_times)}")
    print(f"前5个预测时间:")
    for i, t in enumerate(pre_times[:5]):
        print(f"  {i+1}. {t}")
    
    # 2. 掩盖数据示例
    print("\n--- 2. 掩盖数据示例 ---")
    if len(pre_times) > 0:
        predict_time = pre_times[0]
        masked_data = predictor.mask_data(predict_time)
        
        print(f"预测时间: {predict_time}")
        print(f"掩盖后数据量: {len(masked_data)}")
        print(f"最后5条数据:")
        print(masked_data[['time', 'power_load']].tail())
    
    # 3. 建立时间-负荷字典
    print("\n--- 3. 建立时间-负荷字典 ---")
    time_load_dict = predictor.build_time_load_dict(data)
    print(f"字典大小: {len(time_load_dict)}")
    
    # 查看字典内容
    print("字典前5项:")
    for i, (k, v) in enumerate(time_load_dict.items()):
        if i >= 5:
            break
        print(f"  {k}: {v:.2f}")

run_prediction_feature_parsing()

# 5. 小于号的重要性
print("\n=== 5. 小于号的重要性 ===")

def less_than_explanation():
    """小于号的重要性"""
    print("""
小于号的重要性:

1. 问题
   预测4点负荷时:
   - 用 <= : 4点数据存在，模型可能"抄袭"
   - 用 <  : 4点数据不存在，更严谨

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

3. 代码对比
   # 不严谨（包含预测时间）
   masked_data = data[data['time'] <= predict_time]
   
   # 严谨（不包含预测时间）
   masked_data = data[data['time'] < predict_time]

4. 结论
   使用 < 更严谨
""")

less_than_explanation()

# 6. 数据字典详解
print("\n=== 6. 数据字典详解 ===")

def data_dict_explanation():
    """数据字典详解"""
    print("""
数据字典详解:

1. 作用
   - 存储时间和负荷
   - 避免频繁操作DataFrame
   - 提高查找效率

2. 结构
   time_load_dict = {
       '2024-01-01 00:00:00': 524.56,
       '2024-01-01 01:00:00': 489.12,
       ...
   }

3. 使用
   # 根据时间获取负荷
   load = time_load_dict['2024-01-01 00:00:00']

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

data_dict_explanation()

# 7. 完整代码
print("\n=== 7. 完整代码 ===")

def complete_code():
    """完整代码"""
    print("""
def get_predict_times(self, start_time='2015-08-01'):
    \"\"\"筛选预测时间段\"\"\"
    self.logger.info(f"筛选预测时间段: {start_time}以后")
    
    # 筛选8月1号以后的数据
    pre_times = self.data_source[
        self.data_source['time'] >= start_time
    ]['time'].values
    
    self.logger.info(f"预测时间数量: {len(pre_times)}")
    
    return pre_times

def mask_data(self, predict_time):
    \"\"\"掩盖数据\"\"\"
    # 用小于号，不包含预测时间本身
    masked_data = self.data_source[
        self.data_source['time'] < predict_time
    ].copy()
    
    return masked_data

def build_time_load_dict(self, data):
    \"\"\"建立时间-负荷字典\"\"\"
    time_load_dict = dict(zip(
        data['time'].dt.strftime('%Y-%m-%d %H:%M:%S'),
        data['power_load']
    ))
    
    return time_load_dict
""")

complete_code()

# 8. 总结
def parsing_summary():
    """解析总结"""
    print("=" * 50)
    print("解析预测特征(上)总结")
    print("=" * 50)
    
    print("\n1. 预测时间段")
    print("   2015年8月1号以后")
    
    print("\n2. 数据筛选")
    print("   从test数据中筛选")
    
    print("\n3. 掩盖数据")
    print("   只保存预测时间及以前的数据")
    print("   用 < 不用 <=")
    
    print("\n4. 数据字典")
    print("   time_load_dict")
    print("   避免频繁操作DataFrame")
    
    print("\n" + "=" * 50)
    print("解析预测特征(上)完成!")
    print("=" * 50)

parsing_summary()