电力负荷案例之特征工程-昨日同时刻负荷

1 课程概览

本课讲解电力负荷案例的特征工程第三步：提取昨日同时刻的负荷。使用pd.Timedelta计算昨天的时间，然后获取昨日同时刻的负荷值。

2 核心概念与定义

昨日同时刻负荷：昨天同一时间的电力负荷。
Timedelta：时间差，pd.Timedelta(days=1)表示1天。
yesterday_time：昨天的时间。
yesterday_load：昨天的负荷。
apply：对每个值做操作。

3 算法与模型详解

3.1 特征工程步骤回顾

步骤：

获取上一个小时间的负荷（shift(1)）
获取上上两个小时的负荷（shift(2)）
获取前三个小时的负荷（shift(3)）
拼接到一起
修改列名
连接特征
提取昨日同时刻负荷（本课重点）

3.2 昨日同时刻负荷概念

定义：昨天同一时间的电力负荷

示例：

今天：2025年4月18日 14:00
昨天：2025年4月17日 14:00
昨日同时刻负荷：昨天14:00的负荷

3.3 获取昨天时间

方法：pd.to_datetime() - pd.Timedelta(days=1)

代码：

yesterday_time = pd.to_datetime('2025-04-18') - pd.Timedelta(days=1)
# 结果: 2025-04-17

说明：

pd.Timedelta(days=1)：1天的时间差
pd.Timedelta(days=2)：2天的时间差
pd.Timedelta(days=3)：3天的时间差

3.4 新增yesterday_time列

代码：

feature_data['yesterday_time'] = feature_data['time'].apply(
    lambda x: pd.to_datetime(x) - pd.Timedelta(days=1)
)

说明：

对time列每个值做操作
减去1天得到昨天时间
转成字符串格式

3.5 转成字符串格式

代码：

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')
)

说明：

strftime：转成字符串
'%Y-%m-%d %H:%M:%S'：年-月-日时:分:秒

3.6 获取昨日负荷

问题：如何根据昨天时间获取负荷？

方法：

新增yesterday_time列
根据yesterday_time匹配昨天的power_load
新增yesterday_load列

4 代码示例

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

# 解决中文乱码
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=7):
    """生成电力负荷数据"""
    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
        
        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
        
        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

data = generate_power_load_data(n_days=7)
print(f"数据形状: {data.shape}")

# 3. Timedelta详解
print("\n=== 3. Timedelta详解 ===")

def timedelta_explanation():
    """Timedelta详解"""
    print("""
Timedelta详解:

1. pd.Timedelta(days=1)
   - 1天的时间差
   - 示例: 2024-01-02 - 1天 = 2024-01-01

2. pd.Timedelta(days=2)
   - 2天的时间差
   - 示例: 2024-01-03 - 2天 = 2024-01-01

3. pd.Timedelta(days=3)
   - 3天的时间差
   - 示例: 2024-01-04 - 3天 = 2024-01-01

4. 其他单位
   - hours: 小时
   - minutes: 分钟
   - seconds: 秒
""")
    
    # 示例
    today = pd.to_datetime('2025-04-18 14:00:00')
    yesterday = today - pd.Timedelta(days=1)
    print(f"\n示例:")
    print(f"  今天: {today}")
    print(f"  昨天: {yesterday}")

timedelta_explanation()

# 4. 获取昨天时间
print("\n=== 4. 获取昨天时间 ===")

def get_yesterday_time(data):
    """获取昨天时间"""
    feature_data = data.copy()
    
    # 方法1：使用apply和Timedelta
    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')
    )
    
    print("获取昨天时间后（前10行）:")
    print(feature_data[['time', 'power_load', 'yesterday_time']].head(10))
    
    return feature_data

feature_data = get_yesterday_time(data)

# 5. 获取昨日同时刻负荷
print("\n=== 5. 获取昨日同时刻负荷 ===")

def get_yesterday_load(data):
    """获取昨日同时刻负荷"""
    feature_data = data.copy()
    
    # 1. 获取昨天时间
    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')
    )
    
    # 2. 创建时间到负荷的映射
    time_load_map = dict(zip(
        feature_data['time'].dt.strftime('%Y-%m-%d %H:%M:%S'),
        feature_data['power_load']
    ))
    
    # 3. 根据昨天时间获取昨日负荷
    feature_data['yesterday_load'] = feature_data['yesterday_time'].map(time_load_map)
    
    print("获取昨日负荷后（前10行）:")
    print(feature_data[['time', 'power_load', 'yesterday_time', 'yesterday_load']].head(10))
    
    return feature_data

feature_data = get_yesterday_load(data)

# 6. 完整特征工程（含昨日负荷）
print("\n=== 6. 完整特征工程（含昨日负荷）===")

def complete_feature_engineering(data, logger=None):
    """
    完整特征工程
    Args:
        data: 数据源
        logger: 日志对象
    Returns:
        feature_data: 处理后的数据
    """
    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)
    
    if logger:
        logger.info(f"特征工程完成，形状: {feature_data.shape}")
    
    print(f"特征工程后形状: {feature_data.shape}")
    print(f"所有列名: {list(feature_data.columns)}")
    
    return feature_data

# 创建日志对象
log_utils = LogUtils(root_path='./', log_name='yesterday_load')
logger = log_utils.get_log()

# 重新生成数据
data = generate_power_load_data(n_days=7)
final_data = complete_feature_engineering(data, logger)

# 7. 昨日同时刻负荷详解
print("\n=== 7. 昨日同时刻负荷详解 ===")

def yesterday_load_explanation():
    """昨日同时刻负荷详解"""
    print("""
昨日同时刻负荷详解:

1. 概念
   - 昨天同一时间的电力负荷
   - 示例: 今天14:00的昨日负荷 = 昨天14:00的负荷

2. 获取步骤
   a. 获取昨天时间
      yesterday_time = time - Timedelta(days=1)
   
   b. 创建时间到负荷的映射
      time_load_map = {time: power_load}
   
   c. 根据昨天时间获取负荷
      yesterday_load = map(yesterday_time)

3. 代码
   # 获取昨天时间
   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)
""")

yesterday_load_explanation()

# 8. shift(24)对比
print("\n=== 8. shift(24)对比 ===")

def shift_24_comparison(data):
    """shift(24)对比"""
    feature_data = data.copy()
    
    # 方法1：使用Timedelta（推荐）
    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)
    
    # 方法2：使用shift(24)（取巧）
    feature_data['yesterday_load_shift'] = feature_data['power_load'].shift(24)
    
    print("两种方法对比（前30行）:")
    print(feature_data[['time', 'power_load', 'yesterday_load', 'yesterday_load_shift']].head(30))
    
    # 检查是否一致
    valid_data = feature_data.dropna()
    is_equal = (valid_data['yesterday_load'] == valid_data['yesterday_load_shift']).all()
    print(f"\n两种方法结果是否一致: {is_equal}")

shift_24_comparison(data)

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

def complete_code():
    """完整代码"""
    print("""
def get_yesterday_load(feature_data, logger):
    \"\"\"获取昨日同时刻负荷\"\"\"
    logger.info("开始获取昨日同时刻负荷")
    
    # 1. 获取昨天时间
    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')
    )
    
    # 2. 创建时间到负荷的映射
    time_load_map = dict(zip(
        feature_data['time'].dt.strftime('%Y-%m-%d %H:%M:%S'),
        feature_data['power_load']
    ))
    
    # 3. 获取昨日负荷
    feature_data['yesterday_load'] = feature_data['yesterday_time'].map(time_load_map)
    
    logger.info("昨日同时刻负荷获取完成")
    
    return feature_data

# 调用
feature_data = get_yesterday_load(feature_data, logger)
""")

complete_code()

# 10. 总结
def yesterday_load_summary():
    """昨日负荷总结"""
    print("=" * 50)
    print("昨日同时刻负荷总结")
    print("=" * 50)
    
    print("\n1. 概念")
    print("   昨天同一时间的电力负荷")
    
    print("\n2. Timedelta")
    print("   pd.Timedelta(days=1): 1天时间差")
    print("   pd.Timedelta(days=2): 2天时间差")
    
    print("\n3. 步骤")
    print("   获取昨天时间")
    print("   创建时间到负荷映射")
    print("   根据昨天时间获取负荷")
    
    print("\n4. 代码")
    print("   yesterday_time = time - Timedelta(days=1)")
    print("   yesterday_load = map(yesterday_time)")
    
    print("\n" + "=" * 50)
    print("昨日同时刻负荷完成!")
    print("=" * 50)

yesterday_load_summary()

输出示例：

=== 4. 获取昨天时间 ===
获取昨天时间后（前10行）:
                 time  power_load      yesterday_time
0 2024-01-01 00:00:00      524.56  2023-12-31 00:00:00
1 2024-01-01 01:00:00      489.12  2023-12-31 01:00:00
...

=== 5. 获取昨日同时刻负荷 ===
获取昨日负荷后（前10行）:
                 time  power_load      yesterday_time  yesterday_load
0 2024-01-01 00:00:00      524.56  2023-12-31 00:00:00            NaN
1 2024-01-01 01:00:00      489.12  2023-12-31 01:00:00            NaN
...
24 2024-01-02 00:00:00     512.34  2024-01-01 00:00:00         524.56
25 2024-01-02 01:00:00     478.90  2024-01-01 01:00:00         489.12
...

5 重难点与易错提醒

❗重点：昨日同时刻负荷是重要特征。
❗重点：使用pd.Timedelta(days=1)获取昨天时间。
❗重点：使用map根据时间获取负荷。
❗重点：第一天的数据没有昨日负荷（NaN）。
⚠️易错：Timedelta单位错误。
⚠️易错：时间格式不匹配。
💡深入理解：昨日同时刻负荷捕捉日周期性。

6 课堂问答精选

Q: 如何获取昨日同时刻负荷？

A: 步骤：

获取昨天时间：

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)

Q: 可以用shift(24)代替吗？

A: 可以，但不推荐：

shift(24)：取巧方法，假设数据是连续的
Timedelta：更严谨，基于实际时间计算

如果数据连续且每小时一条，两者结果一致。

7 本课小结

概念：昨日同时刻的电力负荷。
Timedelta：pd.Timedelta(days=1)获取昨天。
步骤：获取昨天时间、创建映射、获取负荷。
代码：apply + Timedelta + map。
特征：捕捉日周期性。

8 延伸思考与实践

实践：运行昨日负荷获取代码。
预习：模型训练、评估、保存。
思考：还可以获取哪些时间特征？