电力负荷预测案例之定义电力负荷模型类

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

from utils.log import LogUtils
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

plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

import datetime

# 日志记录
log_time = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
log_file = f'log/train_{log_time}.log'

def main():
    pass

if __name__ == '__main__':
    main()

import os
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import datetime
import logging
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.metrics import mean_squared_error, mean_absolute_error
from sklearn.ensemble import GradientBoostingRegressor

# 解决中文乱码
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 data_preprocessing(data):
    """数据预处理"""
    data = data.copy()
    data['time'] = pd.to_datetime(data['time'])
    data = data.sort_values(by='time').reset_index(drop=True)
    data = data.drop_duplicates().reset_index(drop=True)
    return data

# 3. 电力负荷模型类
print("\n=== 3. 电力负荷模型类 ===")

class PowerLoadModel:
    """电力负荷模型类"""
    
    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')
    
    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 analysis_data(self, data):
        """查看数据分布"""
        self.logger.info("开始查看数据分布")
        
        # 查看整体分布
        print(f"数据形状: {data.shape}")
        print(f"数据类型:\n{data.dtypes}")
        print(f"描述统计:\n{data.describe()}")
        
        self.logger.info("数据分布查看完成")
    
    def feature_engineering(self, data):
        """特征工程（重点难点）"""
        self.logger.info("开始特征工程")
        
        data = data.copy()
        
        # 基础时间特征
        data['hour'] = data['time'].dt.hour
        data['day'] = data['time'].dt.day
        data['month'] = data['time'].dt.month
        data['dayofweek'] = data['time'].dt.dayofweek
        
        # 衍生特征
        data['is_weekend'] = data['dayofweek'].isin([5, 6]).astype(int)
        data['is_daytime'] = ((data['hour'] >= 6) & (data['hour'] < 18)).astype(int)
        
        # 前几个时间步的负荷（宽表）
        data['prev_1'] = data['power_load'].shift(1)
        data['prev_2'] = data['power_load'].shift(2)
        data['prev_3'] = data['power_load'].shift(3)
        
        # 昨天同一时间负荷
        data['yesterday'] = data['power_load'].shift(24)
        
        # 去除NaN
        data = data.dropna().reset_index(drop=True)
        
        self.logger.info(f"特征工程完成，特征数: {data.shape[1]}")
        return data
    
    def train_model(self, data):
        """训练模型"""
        self.logger.info("开始模型训练")
        
        # 准备特征和标签
        features = ['hour', 'day', 'month', 'dayofweek', 'is_weekend', 
                    'is_daytime', 'prev_1', 'prev_2', 'prev_3', 'yesterday']
        X = data[features].values
        y = data['power_load'].values
        
        # 划分训练测试集
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42, shuffle=False
        )
        
        self.logger.info(f"训练集: {X_train.shape}")
        self.logger.info(f"测试集: {X_test.shape}")
        
        # 训练模型（用GBDT模拟XGBoost）
        model = GradientBoostingRegressor(
            n_estimators=100,
            max_depth=6,
            learning_rate=0.1,
            random_state=42
        )
        model.fit(X_train, y_train)
        
        # 评估
        y_pred = model.predict(X_test)
        mae = mean_absolute_error(y_test, y_pred)
        mse = mean_squared_error(y_test, y_pred)
        rmse = np.sqrt(mse)
        
        self.logger.info(f"MAE: {mae:.4f}")
        self.logger.info(f"MSE: {mse:.4f}")
        self.logger.info(f"RMSE: {rmse:.4f}")
        
        print(f"MAE: {mae:.4f}")
        print(f"MSE: {mse:.4f}")
        print(f"RMSE: {rmse:.4f}")
        
        return model, features
    
    def save_model(self, model, model_path='model/power_load_model.pkl'):
        """保存模型"""
        import joblib
        os.makedirs(os.path.dirname(model_path), exist_ok=True)
        joblib.dump(model, model_path)
        self.logger.info(f"模型已保存: {model_path}")
        print(f"模型已保存: {model_path}")

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

def generate_sample_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
        
        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

raw_data = generate_sample_data(n_days=30)
print(f"原始数据形状: {raw_data.shape}")

# 5. 数据预处理
print("\n=== 5. 数据预处理 ===")
processed_data = data_preprocessing(raw_data)
print(f"处理后数据形状: {processed_data.shape}")

# 6. 创建模型类实例
print("\n=== 6. 创建模型类实例 ===")

log_time = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
log_file = f'log/train_{log_time}.log'

model = PowerLoadModel(log_file=log_file, data_source=processed_data)

# 7. 获取数据源
print("\n=== 7. 获取数据源 ===")
data = model.get_data_source()

# 8. 查看数据分布
print("\n=== 8. 查看数据分布 ===")
model.analysis_data(data)

# 9. 特征工程
print("\n=== 9. 特征工程 ===")
data_with_features = model.feature_engineering(data)
print(f"特征工程后数据形状: {data_with_features.shape}")
print(f"所有列: {list(data_with_features.columns)}")

# 10. 训练模型
print("\n=== 10. 训练模型 ===")
trained_model, features = model.train_model(data_with_features)

# 11. 保存模型
print("\n=== 11. 保存模型 ===")
model.save_model(trained_model)

# 12. 完整训练模块结构
print("\n=== 12. 完整训练模块结构 ===")

def train_module_structure():
    """训练模块结构"""
    print("""
src/train.py 完整结构:

# 1. 导包
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import datetime
import logging

from utils.log import LogUtils
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

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

# 3. 电力负荷模型类
class PowerLoadModel:
    def __init__(self, log_file, data_source):
        self.log_file = log_file
        self.data_source = data_source
        # 配置日志
    
    def get_data_source(self):
        # 获取数据源
        pass
    
    def analysis_data(self, data):
        # 查看数据分布
        pass
    
    def feature_engineering(self, data):
        # 特征工程（重点难点）
        # 时序数据转二维宽表
        pass
    
    def train_model(self, data):
        # 模型训练
        pass
    
    def save_model(self, model, model_path):
        # 保存模型
        pass

# 4. main函数
def main():
    # 1. 配置日志
    # 2. 获取数据源
    # 3. 查看数据分布
    # 4. 特征工程
    # 5. 模型训练
    # 6. 保存模型

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

train_module_structure()

# 13. 模型类总结
def model_class_summary():
    """模型类总结"""
    print("=" * 50)
    print("电力负荷模型类总结")
    print("=" * 50)
    
    print("\n1. 文件位置")
    print("   src/train.py")
    
    print("\n2. 类名")
    print("   PowerLoadModel")
    
    print("\n3. 属性")
    print("   log_file: 日志文件")
    print("   data_source: 数据源")
    
    print("\n4. 方法")
    print("   get_data_source: 获取数据源")
    print("   analysis_data: 查看数据分布")
    print("   feature_engineering: 特征工程（重点）")
    print("   train_model: 训练模型")
    print("   save_model: 保存模型")
    
    print("\n5. 重点难点")
    print("   时序数据转二维宽表")
    print("   特征工程")
    
    print("\n" + "=" * 50)
    print("模型类定义完成!")
    print("=" * 50)

model_class_summary()