1. 概述

1.1 项目背景

EMS（Energy Management System）能源管理系统的 AI 预测服务，基于 PyTorch 构建深度学习模型，实现能源消耗的时序预测。系统采用 LSTM（长短期记忆网络）算法，能够捕捉时间序列数据中的长期依赖关系，为企业提供高精度的能源消耗预测能力。

1.2 核心功能

• 模型训练：基于历史能耗数据训练 LSTM 模型
• 时序预测：支持 24 小时多步滚动预测
• 模型管理：支持模型版本管理和加载
• 性能评估：提供 RMSE、MAE、MAPE 等评估指标

1.3 技术栈

【核心框架】 FastAPI 0.104.1 + PyTorch 2.1.2 + NumPy 1.26.2 + pandas 2.1.4

【部署方式】 Uvicorn ASGI 服务器，独立微服务架构

2. LSTM 算法原理

2.1 LSTM 网络结构

LSTM（Long Short-Term Memory）是一种特殊的循环神经网络（RNN），通过门控机制解决了传统 RNN 的梯度消失问题，能够有效捕捉时间序列中的长期依赖关系。

2.2 门控机制

LSTM 单元包含三个门控：

• 遗忘门（Forget Gate）：决定丢弃哪些信息

f_t = σ(W_f · [h_{t-1}, x_t] + b_f)

• 输入门（Input Gate）：决定存储哪些新信息

i_t = σ(W_i · [h_{t-1}, x_t] + b_i)

C̃_t = tanh(W_C · [h_{t-1}, x_t] + b_C)

• 输出门（Output Gate）：决定输出哪些信息

o_t = σ(W_o · [h_{t-1}, x_t] + b_o)

h_t = o_t · tanh(C_t)

2.3 单元状态更新

C_t = f_t · C_{t-1} + i_t · C̃_t

其中：C_t 为当前单元状态，C_{t-1} 为前一时刻单元状态，f_t 为遗忘门输出，i_t 为输入门输出，C̃_t 为候选状态。

2.4 为什么选择 LSTM

• 能够捕捉时间序列中的长期依赖关系
• 通过门控机制解决梯度消失问题
• 适合处理连续的时序数据（如每小时能耗数据）
• 对噪声有较好的鲁棒性

3. 模型架构设计

3.1 模型结构

本系统采用双层 LSTM 架构，结构如下：

输入层 → LSTM 层 1(64) → LSTM 层 2(64) → ReLU 激活 → 全连接层 → 输出层

3.2 模型参数

【参数配置】

input_size = 1 # 单特征输入（能耗值）

hidden_size = 64 # 隐藏层维度

num_layers = 2 # LSTM层数

output_size = 1 # 单值输出（预测值）

dropout = 0.2 # Dropout正则化率

3.3 模型类定义

【关键代码】lstm_model.py

class EnergyLSTMModel(nn.Module):

def __init__(self, input_size=1, hidden_size=64, num_layers=2,

output_size=1, dropout=0.2):

super(EnergyLSTMModel, self).__init__()

self.hidden_size = hidden_size

self.num_layers = num_layers

self.lstm = nn.LSTM(input_size, hidden_size, num_layers,

batch_first=True, dropout=dropout)

self.fc = nn.Linear(hidden_size, output_size)

self.relu = nn.ReLU()

def forward(self, x):

h0 = torch.zeros(self.num_layers, x.size(0),

self.hidden_size).to(x.device)

c0 = torch.zeros(self.num_layers, x.size(0),

self.hidden_size).to(x.device)

out, _ = self.lstm(x, (h0, c0))

out = self.fc(self.relu(out[:, -1, :]))

return out

3.4 前向传播流程

• 输入数据形状：(batch_size, seq_length, input_size)
• 初始化h0和c0：(num_layers, batch_size, hidden_size)
• LSTM层输出：(batch_size, seq_length, hidden_size)
• 取最后时间步：(batch_size, hidden_size)
• ReLU激活
• 全连接层输出：(batch_size, output_size)

4. 数据预处理设计

4.1 数据格式

输入数据为时序能耗记录列表，每条记录包含：

{ "timestamp": "2024-01-01T00:00:00", "calculatedValue": 125.5 }

4.2 序列创建算法

【滑动窗口法】使用固定长度的滑动窗口从时序数据中创建输入输出对。窗口长度（seq_length）= 24，表示使用过去24小时数据预测下一小时。

4.3 序列创建代码

【关键代码】main.py - create_sequence_data 函数

def create_sequence_data(data, seq_length=24):

values = np.array([d["calculatedValue"] for d in data],

dtype=np.float32)

if len(values) < seq_length:

padding = np.zeros(seq_length - len(values), dtype=np.float32)

values = np.concatenate([padding, values])

X = []

y = []

for i in range(seq_length, len(values)):

X.append(values[i - seq_length:i].reshape(-1, 1))

y.append(values[i])

return np.array(X), np.array(y)

4.4 数据转换流程

原始数据 → 提取calculatedValue → numpy数组 → 零填充（不足24条）→ 滑动窗口 → 输入序列X + 输出值y → PyTorch Tensor

4.5 数据要求

数据类型	最小数量	说明
训练数据	48条	2天 × 24小时
预测数据	24条	1天 × 24小时
时间间隔	1小时	建议每小时一条记录

5. 训练流程设计

5.1 训练流程

5.2 训练代码

【关键代码】main.py - train_model 函数

@app.post("/train", response_model=TrainingResult)

async def train_model(request: TrainingData):

if len(request.training_data) < 48:

return TrainingResult(success=False, message="训练数据不足")

X, y = create_sequence_data(request.training_data)

X_tensor = torch.tensor(X, dtype=torch.float32).to(device)

y_tensor = torch.tensor(y, dtype=torch.float32).to(device)

model = EnergyLSTMModel(input_size=1, hidden_size=64,

num_layers=2, output_size=1)

model.to(device)

criterion = nn.MSELoss()

optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

epochs = 100

for epoch in range(epochs):

model.train()

optimizer.zero_grad()

outputs = model(X_tensor).squeeze()

loss = criterion(outputs, y_tensor)

loss.backward()

optimizer.step()

model.eval()

with torch.no_grad():

predictions = model(X_tensor).squeeze().cpu().numpy()

rmse = np.sqrt(np.mean((predictions - y) ** 2))

mae = np.mean(np.abs(predictions - y))

accuracy = max(0, 1 - mae / np.mean(y))

model_code = f"model_{request.location_id}_{request.energy_category}"

torch.save(model.state_dict(),

os.path.join(MODEL_DIR, f"{model_code}.pth"))

5.3 训练配置

配置项	值	说明
损失函数	MSELoss	均方误差
优化器	Adam	自适应矩估计
学习率	0.001	初始学习率
训练轮数	100	epochs
设备	CPU	可配置为CUDA

5.4 评估指标

指标	计算公式	说明
RMSE	sqrt(mean((pred - actual)²))	均方根误差
MAE	mean(|pred - actual|)	平均绝对误差
Accuracy	1 - MAE / mean(actual)	准确率
MAPE	mean(|(pred - actual) / actual|) × 100	平均绝对百分比误差

6. 预测流程设计

6.1 预测序列图

6.2 滚动预测原理

本系统采用滚动预测（Rolling Prediction）方式，每次预测后将预测值作为下一次输入的一部分，实现多步预测。

【滚动预测过程】

初始输入：[t-23, t-22, ..., t-1, t]

预测 t+1：模型输出 y_pred(t+1)

更新输入：[t-22, ..., t-1, t, y_pred(t+1)]

预测 t+2：模型输出 y_pred(t+2)

...重复此过程直到完成所有预测步数

6.3 预测代码

【关键代码】main.py - predict 函数

@app.post("/predict", response_model=List[PredictionResult])

async def predict(request: PredictionRequest):

model_code = f"model_{request.location_id}_{request.energy_category}"

model_path = os.path.join(MODEL_DIR, f"{model_code}.pth")

model = EnergyLSTMModel(input_size=1, hidden_size=64,

num_layers=2, output_size=1)

if os.path.exists(model_path):

model.load_state_dict(torch.load(model_path, map_location=device))

model.to(device)

model.eval()

if len(request.historical_data) < 24:

padding = [{"calculatedValue": 0}] * (24 - len(request.historical_data))

request.historical_data = padding + request.historical_data

values = np.array([d["calculatedValue"] for d in request.historical_data[-24:]],

dtype=np.float32)

current_seq = torch.tensor(values.reshape(1, -1, 1), dtype=torch.float32).to(device)

predictions = []

prediction_hours = 24

with torch.no_grad():

for i in range(prediction_hours):

output = model(current_seq).item()

std_dev = output * 0.15

pred_time = base_time + timedelta(hours=i)

predictions.append(PredictionResult(...))

new_seq = current_seq.cpu().numpy()

new_seq = np.roll(new_seq, -1, axis=1)

new_seq[0, -1, 0] = output

current_seq = torch.tensor(new_seq, dtype=torch.float32).to(device)

return predictions

6.4 置信区间计算

预测结果包含90%置信区间，基于预测值的15%标准差：

lower_bound = output - output × 0.15

upper_bound = output + output × 0.15

confidence = 0.85 ~ 0.99

6.5 预测输出格式

[
{
"timestamp": "2024-01-08 00:00:00",
"predicted_value": 120.50,
"lower_bound": 102.43,
"upper_bound": 138.57,
"confidence": 0.92,
"unit": "kWh"
}
]

7. API接口设计

7.1 训练接口

接口路径：POST /train

请求体：

{
"location_id": 1,
"energy_category": "electricity",
"training_data": [...],
"model_type": "lstm"
}

响应体：

{
"success": true,
"model_code": "model_1_electricity",
"accuracy": 0.9234,
"rmse": 12.5678,
"training_data_count": 1008,
"message": "训练成功"
}

7.2 预测接口

接口路径：POST /predict

请求体：

{
"task_id": 1,
"location_id": 1,
"energy_category": "electricity",
"model_type": "lstm",
"historical_data": [...],
"prediction_start": "...",
"prediction_end": "..."
}

响应体：List[PredictionResult]（见6.5节）

7.3 模型列表接口

接口路径：GET /models

响应体：

{
"code": 200,
"data": [
{
"model_code": "...",
"location_id": 1,
"energy_category": "..."
}
]
}

7.4 健康检查接口

接口路径：GET /health

响应体：

{
"status": "healthy",
"service": "EMS AI Prediction Service"
}

8. 数据流转架构

8.1 整体数据流

【训练数据流】

能源计量设备 → 原始数据采集 → PostgreSQL（energy_meter_data）→ EMS后端 → AI-Service → 模型训练 → 模型文件存储（models/）

【预测数据流】

EMS前端 → EMS后端API → AI-Service → 加载模型 → 数据预处理 → LSTM推理 → 预测结果 → PostgreSQL（ai_prediction_results）→ 返回前端展示

8.2 数据格式转换

JSON数据 → numpy数组 → PyTorch Tensor → 模型推理 → numpy数组 → JSON响应

8.3 模型存储管理

配置项	值
命名格式	model_{location_id}_{energy_category}.pth
存储位置	./models/ 目录
存储方式	PyTorch state_dict 序列化

9. 性能优化策略

9.1 模型优化

• 模型量化：使用PyTorch量化减少模型大小和推理延迟
• 模型缓存：常用模型预加载到内存，避免重复加载
• 异步推理：FastAPI原生支持异步请求，提高并发处理能力

9.2 数据优化

• 批处理：批量预测减少网络开销
• 数据预处理缓存：预处理结果缓存，避免重复计算
• 内存优化：使用float32代替float64，减少内存占用

9.3 部署优化

• 多进程：使用Uvicorn多工作进程模式
• GPU加速：支持CUDA设备进行训练和推理
• 负载均衡：支持多实例部署，通过负载均衡分发请求

10. 扩展能力设计

10.1 算法扩展

算法	适用场景	状态
LSTM	长时序预测	✅ 已实现
Transformer	复杂时序模式	可扩展
Prophet	季节性预测	可扩展
XGBoost	特征丰富场景	可扩展

10.2 多特征扩展

当前仅使用能耗值作为特征，可扩展支持：

• 天气数据（温度、湿度）
• 生产负荷数据
• 节假日信息
• 设备运行状态

10.3 服务扩展

• 多GPU支持：支持分布式训练
• 模型版本管理：支持多版本模型对比
• A/B测试：支持模型A/B测试

11. 关键代码清单

11.1 文件结构

ai-service/

├── main.py # FastAPI主入口

├── app.py # 备用API实现

├── model/

│ └── lstm_model.py # LSTM模型定义

├── models/ # 模型文件存储

├── requirements.txt # Python依赖

└── start.bat # 启动脚本

11.2 核心代码片段

编号	文件	功能描述
【1】	lstm_model.py	双层LSTM + 全连接层，支持Dropout正则化
【2】	main.py - create_sequence_data	滑动窗口法创建时序数据，支持零填充
【3】	main.py - train_model	数据验证 → 预处理 → 训练 → 评估 → 保存
【4】	main.py - predict	模型加载 → 滚动预测 → 置信区间计算 → 返回

11.3 依赖清单

• fastapi==0.104.1
• uvicorn==0.24.0
• torch==2.1.2
• numpy==1.26.2
• pandas==2.1.4
• scikit-learn==1.3.2

12. 代码问题说明

12.1 需要修复的代码问题

【问题】main.py文件头部缺少torch.nn的导入语句

【影响】train_model函数中使用nn.MSELoss()会导致运行时错误

【修复方案】在文件头部添加：

import torch.nn as nn

13. 总结

EMS AI预测服务基于PyTorch实现LSTM时序预测，具备以下核心能力：

• 高精度时序预测：基于双层LSTM架构，支持24小时多步滚动预测
• 完整的API体系：提供训练、预测、模型管理等全流程接口
• 智能数据处理：自动序列创建、零填充、置信区间计算
• 独立微服务架构：FastAPI构建，便于部署和扩展

系统设计注重代码可读性和可扩展性，关键代码清晰，流程明确，为能源管理系统提供了核心的智能决策支持能力。