昇腾CANN elec-ops-prediction 仓：电力负荷预测深度实践

本文介绍了基于CANN优化的电力负荷预测推理方案elec-ops-prediction。该方案针对传统LSTM模型推理速度慢的问题，提供了一套完整的时序预测流程。文章首先分析了电力系统不同时间尺度的预测需求（超短期、短期、中期、长期）及其精度要求，详细说明了输入特征构成，包括历史负荷、时间特征、气象特征和经济指标等。随后展示了elec-ops-prediction的算子库架构，包含特征提取、推理和

2502_94507659

15人浏览 · 2026-05-26 16:46:46

2502_94507659 · 2026-05-26 16:46:46 发布

前言

你管一个园区的供电，调度室的大屏上要你预测未来 24 小时的负荷走势。传统方案用 LSTM，但推理一次要 500ms，而且批量预测时吞吐上不去。

这其实是典型的 时序预测问题。elec-ops-prediction 是 CANN 针对电力场景优化的推理算子库，预处理、推理、后处理一条龙。

这篇文章深度实践，带你跑通电力负荷预测的完整推理流程。

电力负荷预测的业务需求

先说电力系统要什么：

预测的时间尺度

预测类型	时间范围	用途	精度要求
超短期	0~4 小时	实时调度	15 分钟
短期	4~72 小时	日前调度	1 小时
中期	72 小时~2 周	检修安排	1 天
长期	2 周~1 年	规划	1 周

输入特征

电力负荷预测常用的特征：

# 输入特征的 Python 表示
features = {
    # 历史负荷
    "history_load": [100, 105, 98, ...],  # 过去 N 个点的负荷
    
    # 时间特征
    "hour_of_day": 14,      # 当前小时 (0-23)
    "day_of_week": 2,       # 星期几 (0-6)
    "month": 6,             # 月份 (1-12)
    "is_holiday": 0,        # 是否假日
    
    # 气象特征
    "temperature": 28.5,     # 温度
    "humidity": 65,          # 湿度
    
    # 经济特征
    "industrial_index": 1.02,  # 工业指数
}

elec-ops-prediction 配方内容

# elec-ops-prediction 仓库
elec-ops-prediction/
├── ops/
│   ├── feature_extraction/   # 特征提取算子
│   │   ├── sliding_window.py
│   │   ├── time_features.py
│   │   └── lag_features.py
│   ├── inference/          # 推理算子
│   │   ├── lstm_predict.py
│   │   ├── transformer_predict.py
│   │   └── xgboost_predict.py
│   └── postprocessing/     # 后处理算子
│       ├── inverse_norm.py
│       └── smoothing.py
├── models/                 # 预训练模型
│   ├── lstm_24h.om
│   ├── transformer_24h.om
│   └── ensemble_24h.om
├── configs/               # 配置文件
│   ├── short_term.yaml
│   └── medium_term.yaml
└── tests/
    └── test_inference.py

推理流程

Step 1：数据接入

# step1_data_loader.py
import pandas as pd
import numpy as np
from datetime import datetime

def load_power_data(filepath, start_time=None, end_time=None):
    """
    加载电力负荷数据
    
    Args:
        filepath: CSV 文件路径
        start_time: 开始时间
        end_time: 结束时间
    """
    df = pd.read_csv(filepath, parse_dates=['timestamp'])
    df = df.set_index('timestamp')
    
    if start_time:
        df = df[df.index >= start_time]
    if end_time:
        df = df[df.index <= end_time]
    
    return df


def load_from_scada(scada_host, meter_id, start_time, end_time):
    """
    从 SCADA 系统加载实时数据
    
    SCADA: Supervisory Control And Data Acquisition
    """
    # 连接 SCADA
    client = SCADAClient(host=scada_host)
    
    # 查询历史数据
    query = {
        "meter_id": meter_id,
        "start_time": start_time.isoformat(),
        "end_time": end_time.isoformat(),
        "interval": "5min"  # 5分钟粒度
    }
    
    df = client.query_historical(query)
    df = df.set_index('timestamp')
    
    return df['load']  # 只取负荷列


def load_from_api(api_url, start_date, end_date):
    """
    从能源管理系统 API 加载
    """
    import requests
    
    response = requests.get(
        api_url,
        params={
            "start_date": start_date,
            "end_date": end_date
        }
    )
    
    data = response.json()
    df = pd.DataFrame(data['measurements'])
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    df = df.set_index('timestamp')
    
    return df['load']


# 使用
# df = load_from_scada("scada.company.com", "METER_001", 
#                   datetime(2024, 1, 1), datetime(2024, 1, 2))
# print(df.head())

Step 2：特征工程

# step2_feature_engineering.py
import numpy as np
import pandas as pd
from datetime import datetime

def extract_time_features(timestamps):
    """提取时间特征"""
    features = pd.DataFrame(index=timestamps)
    
    # 时间周期特征（sin/cos 编码，避免突变）
    features['hour_sin'] = np.sin(2 * np.pi * timestamps.hour / 24)
    features['hour_cos'] = np.cos(2 * np.pi * timestamps.hour / 24)
    
    features['day_sin'] = np.sin(2 * np.pi * timestamps.dayofyear / 365)
    features['day_cos'] = np.cos(2 * np.pi * timestamps.dayofyear / 365)
    
    features['week_sin'] = np.sin(2 * np.pi * timestamps.dayofweek / 7)
    features['week_cos'] = np.cos(2 * np.pi * timestamps.dayofweek / 7)
    
    return features


def extract_lag_features(load_series, lags=[1, 2, 3, 24, 48, 168]):
    """
    提取滞后特征（前 N 个时间点的负荷）
    
    Args:
        lags: 滞后的时间步数
    """
    features = pd.DataFrame(index=load_series.index)
    
    for lag in lags:
        features[f'lag_{lag}'] = load_series.shift(lag)
    
    # 滚动统计特征
    features['rolling_mean_24h'] = load_series.rolling(24*12, min_periods=1).mean()  # 24小时滑动平均
    features['rolling_std_24h'] = load_series.rolling(24*12, min_periods=1).std()
    features['rolling_max_24h'] = load_series.rolling(24*12, min_periods=1).max()
    features['rolling_min_24h'] = load_series.rolling(24*12, min_periods=1).min()
    
    return features


def extract_weather_features(temperature, humidity):
    """提取气象特征"""
    features = pd.DataFrame()
    
    features['temperature'] = temperature
    features['humidity'] = humidity
    
    # 派生特征
    features['temp_squared'] = temperature ** 2  # 温度的非线性效应
    features['temp_humidity'] = temperature * humidity
    
    return features


def build_feature_matrix(load_series, weather_df, timestamps=None):
    """
    构建完整的特征矩阵
    
    Args:
        load_series: 历史负荷 Series
        weather_df: 气象数据 DataFrame
        timestamps: 时间戳（可选）
    """
    # 1. 时间特征
    if timestamps is None:
        timestamps = load_series.index
    time_features = extract_time_features(timestamps)
    
    # 2. 滞后特征
    lag_features = extract_lag_features(load_series)
    
    # 3. 气象特征
    weather_features = extract_weather_features(
        weather_df['temperature'],
        weather_df['humidity']
    )
    
    # 4. 合并
    features = pd.concat([time_features, lag_features, weather_features], axis=1)
    
    # 5. 填充 NaN
    features = features.fillna(method='bfill').fillna(method='ffill')
    
    return features


# 使用
# features = build_feature_manager(load_series, weather_df)
# print(features.shape)  # (n_samples, n_features)

Step 3：推理（调用 OM）

# step3_inference.py
import torch
import atb
import numpy as np

class ElectricityPredictor:
    """电力负荷预测器"""
    
    def __init__(self, model_path, history_len=168):
        """
        Args:
            model_path: OM 模型路径
            history_len: 历史窗口长度（默认 7 天 * 24 小时）
        """
        self.history_len = history_len
        self.model = atb.create_inference_model(
            model_path=model_path,
            device="npu:0"
        )
    
    def predict(self, features, prediction_horizon=24):
        """
        预测未来 N 小时的负荷
        
        Args:
            features: 特征矩阵 (history_len + prediction_horizon, n_features)
            prediction_horizon: 预测步数
        
        Returns:
            predictions: 预测值数组 (prediction_horizon,)
        """
        # 1. 转为 tensor
        x = torch.from_numpy(features.values).float()
        
        # 2. 确保 shape 正确 (1, history_len + horizon, n_features)
        if x.dim() == 1:
            x = x.unsqueeze(0)
        
        # 3. 转 NPU
        x = x.npu()
        
        # 4. 推理
        with torch.no_grad():
            output = self.model(x)
        
        # 5. 转 numpy
        predictions = output.squeeze(0).cpu().numpy()
        
        # 只取预测部分
        return predictions[-prediction_horizon:]
    
    def predict_iterator(self, features, prediction_horizon=24, step=1):
        """
        迭代预测（逐步预测，每次预测下一步）
        """
        all_predictions = []
        
        # 初始输入
        current_features = features[:self.history_len].copy()
        
        for i in range(0, prediction_horizon, step):
            # 预测下一步
            next_pred = self.predict(current_features, step)[0]
            all_predictions.append(next_pred)
            
            # 更新输入窗口（滑动）
            # 这里省略实现细节...
        
        return np.array(all_predictions)


def batch_predict(predictor, feature_list):
    """
    批量预测（多用户/多园区）
    """
    results = []
    
    # Padding 到最大长度
    max_len = max(len(f) for f in feature_list)
    padded = []
    for f in feature_list:
        if len(f) < max_len:
            f = f.pad_to(max_len)
        padded.append(f)
    
    # Stack 成 batch
    batch = torch.stack([torch.from_numpy(f.values) for f in padded])
    
    # 推理
    outputs = predictor.model(batch.npu())
    
    return outputs.cpu().numpy()


# 使用
predictor = ElectricityPredictor("transformer_24h.om")

# 单次预测
predictions = predictor.predict(features, prediction_horizon=24)
print(f"Predictions: {predictions.shape}")
# Predictions: (24,)

Step 4：后处理

# step4_postprocessing.py
import numpy as np
import pandas as pd

def inverse_normalize(predictions, mean, std):
    """反归一化"""
    return predictions * std + mean


def apply_business_constraints(predictions, min_load=0, max_load=10000):
    """
    应用业务约束
    
    Args:
        min_load: 最小负荷
        max_load: 最大负荷
    """
    predictions = np.clip(predictions, min_load, max_load)
    return predictions


def smooth_predictions(predictions, window=3):
    """
    平滑预测结果（前后滑动平均）
    """
    return pd.Series(predictions).rolling(window, center=True).mean().values


def aggregate_to_hour(predictions, original_freq='15min'):
    """
    聚合成小时粒度
    """
    # 假设 predictions 是 15 分钟粒度的
    # 每 4 个点平均成一个小时
    n_hours = len(predictions) // 4
    
    aggregated = []
    for i in range(n_hours):
        hour_avg = np.mean(predictions[i*4:(i+1)*4])
        aggregated.append(hour_avg)
    
    return np.array(aggregated)


def format_output(predictions, start_time, freq='1H'):
    """格式化输出"""
    index = pd.date_range(start=start_time, periods=len(predictions), freq=freq)
    df = pd.DataFrame({
        'timestamp': index,
        'predicted_load': predictions
    })
    return df


# 完整的后处理流程
def postprocess_pipeline(predictions, start_time, mean, std, 
                      smooth=True, aggregate=True):
    """后处理流水线"""
    # 1. 反归一化
    predictions = inverse_normalize(predictions, mean, std)
    
    # 2. 平滑
    if smooth:
        predictions = smooth_predictions(predictions, window=3)
    
    # 3. 业务约束
    predictions = apply_business_constraints(predictions, min_load=0, max_load=10000)
    
    # 4. 聚合到小时（如果有需要）
    if aggregate:
        predictions = aggregate_to_hour(predictions)
    
    # 5. 格式化输出
    output = format_output(predictions, start_time)
    
    return output

部署实例

# deployment.py
from datetime import datetime, timedelta

class ElectricityPredictionService:
    """电力负荷预测服务"""
    
    def __init__(self, config):
        # 加载配置
        self.history_hours = config['history_hours']
        self.prediction_hours = config['prediction_hours']
        self.model_path = config['model_path']
        
        # 加载模型
        self.predictor = ElectricityPredictor(
            self.model_path,
            history_len=self.history_hours
        )
        
        # 加载归一化参数
        self.load_mean = config['load_mean']
        self.load_std = config['load_std']
    
    def predict(self, current_time):
        """
        预测 current_time 之后 N 小时的负荷
        """
        # 1. 获取历史数据
        end_time = current_time
        start_time = end_time - timedelta(hours=self.history_hours)
        
        history_data = load_from_scada(
            "scada.company.com",
            "METER_001",
            start_time,
            end_time
        )
        
        # 2. 读取气象预报
        weather = load_weather_forecast(
            current_time,
            current_time + timedelta(hours=self.prediction_hours)
        )
        
        # 3. 特征工程
        features = build_feature_matrix(history_data, weather)
        
        # 4. 推理
        predictions = self.predictor.predict(
            features.values,
            prediction_horizon=self.prediction_hours
        )
        
        # 5. 后处理
        output = postprocess_pipeline(
            predictions,
            current_time,
            self.load_mean,
            self.load_std
        )
        
        return output


# 使用
config = {
    'history_hours': 168,      # 7 天历史
    'prediction_hours': 24,    # 预测 24 小时
    'model_path': 'transformer_24h.om',
    'load_mean': 5000,
    'load_std': 1500
}

service = ElectricityPredictionService(config)

# 执行预测
current_time = datetime.now()
result = service.predict(current_time)

print(result.head())