任务介绍
数据处理
旧区总曝气量=OldZone_Aeration= OldZone_Aeration1+ OldZone_Aeration2
新区总曝气量=NewZone_Aeration = NewZone_Aeration1 + NewZone_Aeration2
然后发现这两列中还是有很多0值，不能直接用于建模。所以我们把这两列数据继续相加作为新的
总曝气量。
LSTM对总曝气量进行建模
数据预处理
scaler = MinMaxScaler() # 创建归一化工具 scaler.fit(data[:train_end]) # 只在【训练集】上学习最大最小值 data_s = scaler.transform(data) # 用训练集学到的规则，给所有数据归一化
整理数据
# 滑窗: X=[t-12,...,t-1], y=第(t+PRED_HORIZON-1)小时的值 def make_seq(arr, seq_len, horizon): X, y = [], [] for i in range(len(arr) - seq_len - horizon + 1): X.append(arr[i : i+seq_len]) y.append(arr[i+seq_len+horizon-1]) return np.array(X), np.array(y)
arr = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15] len(arr)=16
seq_len = 3 # 用过去3小时
horizon = 2 # 预测未来第2小时
i=12代表总共可以构造12组数据
i = 0
X：arr[0:3] → [0, 1, 2]
y：arr[0+3+2-1] = arr[4] → 4→ 用 0,1,2 预测 第 2 小时：4
最后：
X = [ [0,1,2], [1,2,3], ... [11,12,13] ] y = [4,5,6,7,8,9,10,11,12,13,14,15]
模型的前向传播
输入 x (64, 12, 1) ↓ 通过 LSTM out (64, 12, 64) ↓ 取最后时刻 out[:, -1, :] (64, 64) ↓ 全连接层 输出 (64, 1)
整体代码
import numpy as np import pandas as pd import torch import torch.nn as nn from torch.utils.data import DataLoader, TensorDataset from sklearn.preprocessing import MinMaxScaler from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score import matplotlib.pyplot as plt # ========== 1. 配置 ========== SEQ_LEN = 12 # 用过去12小时 PRED_HORIZON = 1 # 预测未来第 t 小时, 改这里即可: 1, 2, 3... BATCH_SIZE = 64 EPOCHS = 30 LR = 1e-3 device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') # ========== 2. 数据准备 ========== df = pd.read_csv('aeration.csv', parse_dates=['Date']) data = df['Aeration'].values.reshape(-1, 1) n = len(data) train_end, val_end = int(n*0.7), int(n*0.85) # 归一化(只用训练集 fit, 防止数据泄漏) scaler = MinMaxScaler() scaler.fit(data[:train_end]) data_s = scaler.transform(data) # 滑窗: X=[t-12,...,t-1], y=第(t+PRED_HORIZON-1)小时的值 def make_seq(arr, seq_len, horizon): X, y = [], [] for i in range(len(arr) - seq_len - horizon + 1): X.append(arr[i : i+seq_len]) y.append(arr[i+seq_len+horizon-1]) return np.array(X), np.array(y) X, y = make_seq(data_s, SEQ_LEN, PRED_HORIZON) n_samples = len(X) train_size = int(n_samples * 0.7) val_size = int(n_samples * 0.15) X_tr, y_tr = X[:train_size], y[:train_size] X_va, y_va = X[train_size : train_size+val_size], y[train_size : train_size+val_size] X_te, y_te = X[train_size+val_size :], y[train_size+val_size :] def to_loader(X, y, shuffle=False): ds = TensorDataset(torch.FloatTensor(X), torch.FloatTensor(y)) return DataLoader(ds, batch_size=BATCH_SIZE, shuffle=shuffle) train_loader = to_loader(X_tr, y_tr, shuffle=True) val_loader = to_loader(X_va, y_va) test_loader = to_loader(X_te, y_te) # ========== 3. LSTM 模型 ========== class LSTMModel(nn.Module): def __init__(self, hidden=64, layers=2): super().__init__() self.lstm = nn.LSTM(1, hidden, layers, batch_first=True, dropout=0.1) self.fc = nn.Linear(hidden, 1) def forward(self, x): out, _ = self.lstm(x) return self.fc(out[:, -1, :]) model = LSTMModel().to(device) criterion = nn.MSELoss() optimizer = torch.optim.Adam(model.parameters(), lr=LR) # ========== 4. 训练 ========== train_losses, val_losses = [], [] best_val, best_state = float('inf'), None for ep in range(1, EPOCHS+1): model.train() tl = 0 for xb, yb in train_loader: xb, yb = xb.to(device), yb.to(device) optimizer.zero_grad() loss = criterion(model(xb), yb) loss.backward() optimizer.step() tl += loss.item() * xb.size(0) tl /= len(train_loader.dataset) model.eval() vl = 0 with torch.no_grad(): for xb, yb in val_loader: xb, yb = xb.to(device), yb.to(device) vl += criterion(model(xb), yb).item() * xb.size(0) vl /= len(val_loader.dataset) train_losses.append(tl); val_losses.append(vl) if vl < best_val: best_val, best_state = vl, {k: v.clone() for k, v in model.state_dict().items()} print(f"Epoch {ep:02d} | train {tl:.5f} | val {vl:.5f}") model.load_state_dict(best_state) # ========== 5. 测试评估 ========== model.eval() preds = [] with torch.no_grad(): for xb, yb in test_loader: preds.append(model(xb.to(device)).cpu().numpy()) preds = np.concatenate(preds) y_pred = scaler.inverse_transform(preds) y_true = scaler.inverse_transform(y_te) rmse = np.sqrt(mean_squared_error(y_true, y_pred)) mae = mean_absolute_error(y_true, y_pred) r2 = r2_score(y_true, y_pred) print(f"\n========== 测试集 (预测未来第 {PRED_HORIZON} 小时) ==========") print(f"RMSE = {rmse:.3f}") print(f"MAE = {mae:.3f}") print(f"R² = {r2:.4f}")
模型保存
| 文件 | 作用 | 必须保存吗 |
model.pt |
模型权重 | ✅ 必须 |
scaler.pkl |
归一化参数 | ✅ 必须（没它模型废掉） |
config.json |
模型结构 + 元信息 | ✅ 必须（推理时构造模型用） |
model.load_state_dict(best_state) # ========== 保存部署所需文件 ========== import joblib from pathlib import Path deploy_dir = Path(f'deploy_t+{PRED_HORIZON}h') deploy_dir.mkdir(exist_ok=True) torch.save(model.state_dict(), deploy_dir / 'model.pt') joblib.dump(scaler, deploy_dir / 'scaler.pkl') print(f"已保存: {deploy_dir}/model.pt, {deploy_dir}/scaler.pkl")
import torch import torch.nn as nn import joblib import numpy as np # ========== 1. 模型结构(必须和训练时完全一致) ========== class LSTMModel(nn.Module): def __init__(self, hidden=64, layers=2): super().__init__() self.lstm = nn.LSTM(1, hidden, layers, batch_first=True, dropout=0.1) self.fc = nn.Linear(hidden, 1) def forward(self, x): out, _ = self.lstm(x) return self.fc(out[:, -1, :]) # ========== 2. 加载模型和 scaler ========== device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') model = LSTMModel(hidden=64, layers=2).to(device) model.load_state_dict(torch.load('deploy_t+1h/model.pt', map_location=device)) model.eval() # 切换到推理模式, 关闭 dropout scaler = joblib.load('deploy_t+1h/scaler.pkl') print("模型加载完成") # ========== 3. 准备一个样本 (过去 12 小时的 Aeration 值) ========== history = [ 561.1596221923829, 562.870346069335, 566.306594848633, 565.813049316406, 565.824310302735, 565.37336730957, 493.188201904297, 379.360862731933, 379.547348047461, 377.33290863037, 530.632003784179, 557.160324096679 ] # 转成 numpy, reshape 成 [12, 1] x = np.array(history, dtype=np.float32).reshape(-1, 1) # 用训练时的 scaler 归一化 x_scaled = scaler.transform(x) # 转成 tensor, 加 batch 维度: [1, 12, 1] x_tensor = torch.from_numpy(x_scaled).unsqueeze(0).to(device) # ========== 4. 推理 ========== with torch.no_grad(): pred_scaled = model(x_tensor).cpu().numpy() # 反归一化得到真实数值 pred = scaler.inverse_transform(pred_scaled) print(f"\n输入: 过去 12 小时的 Aeration = {history}") print(f"预测: 下一小时的 Aeration = {pred[0][0]:.2f}")
输入: 过去 12 小时的 Aeration = [561.1596221923829, 562.870346069335, 566.306594848633, 565.813049316406, 565.824310302735, 565.37336730957, 493.188201904297, 379.360862731933, 379.547348047461, 377.33290863037, 530.632003784179, 557.160324096679]
预测: 下一小时的 Aeration = 567.72
模型预测：567.72 真实值：557.32
是否需要转换模型
| 维度 | 你的场景 | 是否需要 ONNX |
| 调用频率 | 小时级（每小时 1 次） | ❌ 不需要 |
| 延迟要求 | 几秒内出结果即可 | ❌ 不需要 |
| 并发量 | 可能就 1 个 SCADA / 1 个调用方 | ❌ 不需要 |
| 部署语言 | 纯 Python | ❌ 不需要 |
| 部署环境 | 厂里的 Linux 服务器（不是边缘设备） | ❌ 不需要 |
| 模型大小 | LSTM 很小（几十 KB） | ❌ 不需要 |
如何变成api的调用？
# ━━━━━━━━━━━━━━ 第 1 块: 导入工具 ━━━━━━━━━━━━━━ from fastapi import FastAPI from pydantic import BaseModel from typing import List from predictor import AerationPredictor # ━━━━━━━━━━━━━━ 第 2 块: 初始化 ━━━━━━━━━━━━━━━ app = FastAPI() predictor = AerationPredictor() # ━━━━━━━━━━━━━━ 第 3 块: 定义请求格式 ━━━━━━━━━━━ class PredictRequest(BaseModel): history: List[float] # ━━━━━━━━━━━━━━ 第 4 块: 定义接口 (根路径) ━━━━━━━ @app.get("/") def root(): return {"message": "Aeration API is running. Visit /docs for documentation."} # ━━━━━━━━━━━━━━ 第 5 块: 定义接口 (预测) ━━━━━━━━━ @app.post("/predict") def predict(req: PredictRequest): pred = predictor.predict(req.history) return {"prediction": pred}
第 1 部分：导入工具
from fastapi import FastAPI from pydantic import BaseModel from typing import List from predictor import AerationPredictor
第 2 部分：创建应用 + 加载模型
app = FastAPI() predictor = AerationPredictor()
创建一个 FastAPI 应用实例 ，创建一个预测器实例（模型在这里被加载到内存）
app是 FastAPI 类创建出来的实例（对象）
predictor是 AerationPredictor 类创建出来的实例（对象）
它们都会继承相应的类的方法
app是什么：
- 是整个 Web 服务的核心对象
- 后面所有的 @app.get("/")、@app.post("/predict") 都是在给这个 app 注册接口
类比：你开了家"快递公司"， app就是公司本身。后面要往这个公司里加"业务窗口"（接口）。
第 3 部分：定义请求数据的"形状"
class PredictRequest(BaseModel): history: List[float]
# 客户端发的 JSON { "history": [520.1, 525.3, 530.0, ...] } # 对应的 Python 类 class PredictRequest(BaseModel): history: List[float] # ← 字段名要和 JSON 的 key 一致
✔ 前端必须传一个 JSON
✔ JSON 里必须有一个字段叫 history
✔ history必须是数字列表
✔ 传错了系统直接报错，不允许进入预测
✔ 自动生成接口文档
✔ 自动把 JSON 转成 Python 对象
函数体：业务逻辑
# 定义接口函数：接收请求 → 预测 → 返回结果 def predict(req: PredictRequest): # 调用预测模型，传入历史数据 pred = predictor.predict(req.history) # 返回预测结果（JSON格式） return {"prediction": pred}
# 客户端发的 JSON: {"history": [520.1, 525.3, ...]} # 在函数里访问: req.history # 等于 [520.1, 525.3, ...] # 客户端收到的响应: {"prediction": 562.34}
定义一个函数predict，接收的数据是：req: PredictRequest，req = 请求数据（前端传过来的内
容），PredictRequest = 你刚才定义的数据格式（必须包含 history 列表）【FastAPI 会自动校验
前端传的数据对不对】。
编辑
@app.get("/")
编辑
这是一个"装饰器"（decorator），"嘿 FastAPI！下面这个函数，请把它注册成一个 GET 接口，路
径是 /。当有人访问 / 时，就调用这个函数。"
GET vs POST 的区别
| HTTP 方法 | 用途 | 特点 |
| GET | 读取数据 | 没有请求体，参数在 URL 里 |
| POST | 提交数据/触发动作 | 有请求体（JSON） |
GET 请求: 查询天气 GET http://example.com/weather?city=beijing 没有 body, 参数在 URL 里 POST 请求: 提交订单 POST http://example.com/orders Body: {"product": "iPhone", "quantity": 2} 有 body, 数据在 body 里
@app.post("/predict")
装饰器：注册一个 POST 接口，路径 /predict。
完整 URL：http://你的服务器IP:8000/predict
为什么 /predict 用 POST 而不是 GET：
客户端要传 12 个浮点数过来
12 个数放 URL 里又长又难看
而且预测是个"动作"（触发模型推理），符合 POST 的语义。
def predict(req: PredictRequest): ↑ 类型标注: req 是 PredictRequest 类型
req.history
# 客户端发的 JSON: {"history": [520.1, 525.3, 530.0, ...]} # 在函数里: req.history 等于 [520.1, 525.3, 530.0, ...]
pred = 562.34 # 一个 float
HTTP/1.1 200 OK Content-Type: application/json {"prediction":562.34}
调用api的代码
用 Python 通过网络调用你的 API，拿到预测结果。
你的 Python 程序 你的 API 服务 (FastAPI) ───────────────── ───────────────────── 发请求 ──HTTP POST──→ 接收请求 ↓ 调用模型预测 ↓ 收响应 ←──JSON 数据── 返回结果 打印结果
import requests # ← 第 1 行: 导入工具 response = requests.post( # ← 第 2 行: 发送请求 'http://127.0.0.1:8000/predict', # ← 第 3 行: 目标地址 json={'history': [520.1, 525.3, ..., 560.0]} # ← 第 4 行: 发送的数据 ) print(response.json()) # ← 第 5 行: 打印响应 # 输出: {'prediction': 554.8828735351562} # ← 第 6 行: 预期输出
客户端的 URL 服务端的代码 http://127.0.0.1:8000/predict ↓ 端口 8000 @app.post("/predict") ↓ ↑ 路径 /predict ↑ ↓ ↑ └───── 匹配 ─────────┘
┌─────────────────────────────────────────────────────────┐ │ 客户端 (Python requests) │ │ requests.post( │ │ 'http://127.0.0.1:8000/predict', │ │ json={'history': [520.1, 525.3, ..., 560.0]} │ │ ) │ └────────────────────────┬────────────────────────────────┘ │ │ HTTP POST 请求 │ Body (JSON): {"history": [...]} ▼ ┌─────────────────────────────────────────────────────────┐ │ FastAPI 接收到请求 │ │ │ │ Step 1: 看路径是 /predict, 方法是 POST │ │ ↓ 找到对应的处理函数 predict() │ │ │ │ Step 2: 读取请求体, 解析 JSON │ │ ↓ {"history": [520.1, ..., 560.0]} │ │ │ │ Step 3: 用 PredictRequest 校验 │ │ ✓ history 字段存在 │ │ ✓ 是 list 类型 │ │ ✓ 内部元素都是 float │ │ ↓ 构造 PredictRequest 实例 │ │ │ │ Step 4: 调用 predict(req=该实例) │ └────────────────────────┬────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ 你的代码执行: │ │ │ │ pred = predictor.predict(req.history) │ │ │ │ │ ├─ 归一化输入 │ │ ├─ LSTM 推理 │ │ └─ 反归一化输出 │ │ ↓ │ │ pred = 562.34 │ │ │ │ return {"prediction": 562.34} │ └────────────────────────┬────────────────────────────────┘ │ │ FastAPI 把字典转成 JSON ▼ ┌─────────────────────────────────────────────────────────┐ │ HTTP 响应 │ │ Status: 200 OK │ │ Body: {"prediction": 562.34} │ └────────────────────────┬────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ 客户端收到响应 │ │ response.json() = {"prediction": 562.34} │ └─────────────────────────────────────────────────────────┘
