这次我们直接看一个工程里经常遇到的组合方案用 K-means 先给时间序列数据做聚类再针对不同数据簇分别训练 LSTM 多输出回归模型。这个方法解决的实际问题很明确——如果训练数据里存在多种运行工况、多个业务模式直接塞进一个 LSTM 里模型往往会被“平均效果”拖垮而 K-means 把样本按特征空间拆成多个簇后每个簇单独建模预测精度通常会更稳。这套方案的另一个价值是门槛低。K-means 没有调参地狱LSTM 用 PyTorch 或 Keras 都能落地CPU 也能完成训练和推理。如果你只是做多步时间序列预测或者同时预测多个目标变量这篇文章可以直接收藏。下面我会从方法原理、环境准备、K-means 聚类、LSTM 多输出模型构建、联合预测验证、批量推理、性能观察和问题排查这几个方面展开代码全部给到可直接复制运行的版本。1. 核心能力速览能力项说明方法类型无监督聚类 有监督深度学习回归主要功能多变量时间序列多步预测、多目标同时回归算法组成K-meanssklearn、LSTMPyTorch / Keras运行平台Windows / Linux / macOS硬件要求CPU 可跑建议 4GB 以上内存如有 NVIDIA GPU 可加速训练Python 版本3.8 及以上启动方式Python 脚本 / Jupyter Notebook是否支持 API可快速封装为 FastAPI / Flask 服务是否支持批量任务支持Python 批量推断即可适合场景多工况工业数据预测、多步负荷预测、多指标同时预测这里的显存占用不做硬性假设因为 LSTM 的显存或内存消耗取决于输入序列长度、隐藏层大小、batch size 和输出维度。实测时用nvidia-smi或torch.cuda.memory_allocated()观察即可。2. 方法原理与适用场景2.1 先看数据长什么样假设你有一组多维时间序列数据每一行是一个时间点的观测值列是不同传感器或业务指标时间温度压力转速振动t045.20.8212000.13t145.80.8512100.14...............多输出回归在这个场景里有两层含义预测未来多个时间步的值例如用过去 24 小时预测未来 24 小时。同时预测多个变量例如同时输出温度、压力、转速和振动。两者可以叠加输出维度 预测步数 × 变量数。2.2 为什么要先用 K-means 聚类工业数据和业务数据里不同工况往往对应不同的数据分布。比如设备在“正常负载”“高负载”“待机”三种状态下传感器数值的均值、波动幅度都不同。如果把这些数据混在一起训练 LSTM模型只能学习到一个中间态容易出现“高负荷阶段预测偏低、低负荷阶段预测偏高”的问题。K-means 在这里的作用是把原始样本按特征空间划分成 K 个簇。每个簇内数据分布更接近单独训练 LSTM 更容易拟合。预测新样本时先判断它属于哪个簇再选择对应模型或包含聚类标签的特征来预测。需要注意K-means 聚类必须使用滑窗特征不能使用未来标签信息否则会造成信息泄露测试集上指标虚高。2.3 三种常见组合方式组合方式做法优缺点方式一分簇训练每个簇单独训练一个 LSTM 模型拟合能力强但需要维护 K 个模型预测时先聚类再选模型方式二聚类标签作为特征将 K-means 得到的簇标签做 one-hot拼接到 LSTM 输入特征中只训练一个模型简单稳定适合数据量不大时方式三聚类结果做门控用簇概率或距离加权多个模型输出融合效果平滑但工程复杂度最高本文主要演示方式一和方式二因为它们在实战中最常用代码也容易理解。2.4 适用边界这套方法适合有显著多模式特征的时间序列回归任务比如多工况设备预测、多季节电力负荷预测、多行业销售数据预测。不适合的情况是数据本身只有一种模式或者样本量太小每个簇只有几十条样本在这种情况下 K-means 没有收益反而增加训练和推理成本。必须强调如果业务场景涉及真实设备数据、用户隐私数据或商业数据请先确认数据来源合法、脱敏合规。人脸、声音、个人位置等敏感数据不要随意采集和训练。3. 环境准备与数据集构造3.1 安装依赖建议新建一个独立 Python 环境避免依赖冲突。conda create -n kmeans_lstm python3.10 -y conda activate kmeans_lstm然后安装核心依赖pip install torch scikit-learn pandas numpy matplotlib如果使用 Keraspip install tensorflow scikit-learn pandas numpy matplotlib没有 GPU 也能跑训练时间会长一些。后面的代码默认使用 PyTorch如果你更熟悉 TensorFlow整个方案的结构完全一致只是模型定义部分需要替换。3.2 构造示例数据集为了验证流程这里构造一个包含两种工况的合成时间序列数据。实际项目中把数据读取部分替换成自己的 CSV 或数据库查询即可。import numpy as np import pandas as pd np.random.seed(42) n 2000 time np.arange(n) # 工况A均值为 50振荡幅度小 mode_a 50 5 * np.sin(time / 20) np.random.normal(0, 0.5, n) # 工况B均值为 80振荡幅度大 mode_b 80 10 * np.sin(time / 15) np.random.normal(0, 1.2, n) # 构造分段数据前1000点是工况A后1000点是工况B feature1 np.concatenate([mode_a[:1000], mode_b[1000:]]) feature2 feature1 * 0.6 np.random.normal(0, 1, n) df pd.DataFrame({ time: time, feat1: feature1, feat2: feature2, }) df.head()这里的feat1和feat2是两个相关变量我们的目标是同时预测未来多个时间步的这两个变量。3.3 滑窗生成监督学习数据集LSTM 模型的输入是“序列样本”。对每个时间点 t取过去look_back个时间点作为特征预测未来look_forward个时间点的两个变量。def create_sequences(data, look_back24, look_forward6): X, y [], [] for i in range(len(data) - look_back - look_forward 1): X.append(data[i:i look_back]) y.append(data[i look_back:i look_back look_forward]) return np.array(X), np.array(y) # 只使用特征列标签为未来特征向量 data df[[feat1, feat2]].values X, y create_sequences(data, look_back24, look_forward6) print(X shape:, X.shape) print(y shape:, y.shape)这里的 X shape 是(样本数, 24, 2)y shape 是(样本数, 6, 2)。因为 K-means 聚类不做时序依赖假设我们先把每个样本的二维特征24 × 2 48 维展平用于聚类LSTM 模型仍使用三维输入格式。4. K-means 聚类实现4.1 标准化与聚类对展平后的特征做标准化再用肘部法则估计 K 值。from sklearn.preprocessing import StandardScaler from sklearn.cluster import KMeans X_flat X.reshape(X.shape[0], -1) scaler StandardScaler() X_flat_scaled scaler.fit_transform(X_flat) # 用肘部法则观察K值选择 from sklearn.metrics import silhouette_score inertias [] for k in range(2, 8): km KMeans(n_clustersk, random_state42, n_init10) km.fit(X_flat_scaled) inertias.append(km.inertia_) print(inertias:, inertias) # 选择拐点处的K值这里直接取2 K 2 kmeans KMeans(n_clustersK, random_state42, n_init10) labels kmeans.fit_predict(X_flat_scaled) print(cluster distribution:, np.bincount(labels))在这个构造数据里K 值确实等于 2。实际项目中不要直接抄要先看完肘部曲线再决定。如果数据分布没有明显分群K-means 的收益会很有限。4.2 聚类结果可视化import matplotlib.pyplot as plt plt.figure(figsize(12, 4)) for c in range(K): cluster_idx np.where(labels c)[0] plt.scatter(cluster_idx, X_flat[cluster_idx, 0], s2, labelfcluster {c}) plt.legend() plt.title(K-means Clustering Result on Sample Index) plt.show()如果聚类效果正常你会看到两个簇在时间轴上大致对应两种工况的区间。边缘处会有一些交叉这是正常的。4.3 划分训练集和测试集注意时间序列数据必须按时间顺序切分不能随机打乱否则会引入未来信息泄漏。train_ratio 0.8 split int(len(X) * train_ratio) X_train, X_test X[:split], X[split:] y_train, y_test y[:split], y[split:] labels_train, labels_test labels[:split], labels[split:] print(X_train:, X_train.shape) print(X_test:, X_test.shape)5. LSTM 多输出回归模型构建5.1 模型定义LSTM 输出形状是(batch_size, look_forward, num_vars)。我们用两层 LSTM 加一个全连接输出层。import torch import torch.nn as nn class LSTMRegressor(nn.Module): def __init__(self, input_size2, hidden_size64, num_layers2, output_len6, output_size2): super().__init__() self.lstm nn.LSTM( input_sizeinput_size, hidden_sizehidden_size, num_layersnum_layers, batch_firstTrue ) self.fc nn.Linear(hidden_size, output_len * output_size) self.output_len output_len self.output_size output_size def forward(self, x): # x: (batch, seq_len, input_size) out, _ self.lstm(x) out out[:, -1, :] # 取最后一个时间步的隐藏状态 out self.fc(out) return out.view(-1, self.output_len, self.output_size)这里把输出层设计成(batch, output_len, output_size)也就是一个 LSTM 模型同时完成多步预测和多变量回归。5.2 训练函数为了让代码通用这里不区分簇先在完整训练集上训练一个基准模型方便和分簇模型做对比。def train_model(X_train, y_train, X_val, y_val, epochs20, lr1e-3): model LSTMRegressor(input_sizeX_train.shape[2]) optimizer torch.optim.Adam(model.parameters(), lrlr) criterion nn.MSELoss() X_train_t torch.tensor(X_train, dtypetorch.float32) y_train_t torch.tensor(y_train, dtypetorch.float32) X_val_t torch.tensor(X_val, dtypetorch.float32) y_val_t torch.tensor(y_val, dtypetorch.float32) for epoch in range(epochs): model.train() optimizer.zero_grad() pred model(X_train_t) loss criterion(pred, y_train_t) loss.backward() optimizer.step() if (epoch 1) % 5 0: model.eval() with torch.no_grad(): val_pred model(X_val_t) val_loss criterion(val_pred, y_val_t) print(fEpoch {epoch1}/{epochs}, train_loss{loss.item():.4f}, val_loss{val_loss.item():.4f}) return model model_base train_model(X_train, y_train, X_test, y_test, epochs20)默认情况下这是 CPU 训练。如果你的机器有 GPU可以把 tensor 搬到 CUDA 上代码里加一个设备判断即可。5.3 分簇模型训练这是“K-means LSTM”组合的核心部分。对每个簇用该簇的训练样本单独训练一个模型。def train_per_cluster(cluster_labels, X_train, y_train, epochs20): models {} for c in range(K): idx np.where(cluster_labels c)[0] X_c X_train[idx] y_c y_train[idx] print(fTraining cluster {c}, samples: {X_c.shape[0]}) models[c] train_model(X_c, y_c, X_c, y_c, epochsepochs) return models models_cluster train_per_cluster(labels_train, X_train, y_train, epochs20)这里为了演示代码简洁验证集直接用了训练集。实际项目中请单独划分验证集避免过拟合被掩盖。6. 联合预测与效果验证6.1 预测流程预测时先对新样本做滑窗展平、标准化再用 K-means 模型判断它属于哪个簇最后调用对应簇的 LSTM 模型推理。def predict_with_cluster(model, kmeans, scaler, X_test, cluster_labels): preds [] for i in range(len(X_test)): x_flat X_test[i].reshape(1, -1) x_scaled scaler.transform(x_flat) c kmeans.predict(x_scaled)[0] x_t torch.tensor(X_test[i], dtypetorch.float32).unsqueeze(0) model.eval() with torch.no_grad(): pred model(x_t).numpy() preds.append(pred[0]) return np.array(preds) preds_per_cluster predict_with_cluster( models_cluster, kmeans, scaler, X_test, labels_test )注意上面的推理循环没有 batch 化数据量大时会比较慢。批量推理的优化放到第 7 节。6.2 评估指标多输出回归常用四个指标MAE、RMSE、MAPE、R2。这里定义一组通用评估函数。from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score def evaluate(y_true, y_pred): y_true_flat y_true.reshape(-1, y_true.shape[-1]) y_pred_flat y_pred.reshape(-1, y_pred.shape[-1]) mae mean_absolute_error(y_true_flat, y_pred_flat) rmse mean_squared_error(y_true_flat, y_pred_flat, squaredFalse) r2 r2_score(y_true_flat, y_pred_flat) return {MAE: mae, RMSE: rmse, R2: r2} metrics_base evaluate(y_test, preds_base) metrics_cluster evaluate(y_test, preds_per_cluster) print(baseline metrics:, metrics_base) print(cluster metrics:, metrics_cluster)如果分簇策略有效通常metrics_cluster的 MAE 和 RMSE 会低于metrics_base。如果没有下降或下降很少说明数据没有明显分群就不用继续用分簇方案。6.3 预测曲线对比把某个预测步的实际值和两个模型的预测值画在一起可以直观看到分簇模型是否在跳变区域表现更好。def plot_predictions(y_true, preds_a, preds_b, step0, var0): plt.figure(figsize(12, 4)) plt.plot(y_true[:, step, var], labeltrue, colorblack) plt.plot(preds_a[:, step, var], labelbase LSTM, linestyle--) plt.plot(preds_b[:, step, var], labelKmeansLSTM, linestyle:) plt.legend() plt.title(fStep{step}, Variable{var}) plt.show() plot_predictions(y_test, preds_base, preds_per_cluster, step0, var0)这一张图基本就能看出分簇方案是否值得保留。如果两条预测曲线几乎重合说明 K 的选择或聚类特征没有带来有效信息。7. 功能测试与批量任务7.1 批次推理优化按单条样本循环推理在数据量大的时候很慢。更高效的做法是把整个测试集按簇拆分然后分批输入模型。def batch_predict_per_cluster(models, kmeans, scaler, X_test): # 先给所有样本打簇标签 X_flat X_test.reshape(X_test.shape[0], -1) X_scaled scaler.transform(X_flat) cluster_ids kmeans.predict(X_scaled) outputs np.zeros((X_test.shape[0], y_test.shape[1], y_test.shape[2])) for c in range(K): idx np.where(cluster_ids c)[0] x_t torch.tensor(X_test[idx], dtypetorch.float32) model models[c] model.eval() with torch.no_grad(): pred model(x_t).numpy() outputs[idx] pred return outputs实际项目中还可以用torch.utils.data.DataLoader设置 batch size进一步控制显存和训练稳定性。7.2 批量预测一批 CSV 文件假设有一个files/目录里面有多个设备的 CSV 文件每个文件格式相同。批量推理的流程是读取文件 - 滑窗 - 标准化 - 聚类归属 - 模型预测 - 保存结果。import os import glob def predict_csv_files(files_dir, output_dir, look_back24, look_forward6): os.makedirs(output_dir, exist_okTrue) csv_files glob.glob(os.path.join(files_dir, *.csv)) for file_path in csv_files: df_file pd.read_csv(file_path) data df_file[[feat1, feat2]].values if len(data) look_back look_forward: print(fskip {file_path}, too short) continue X_file, _ create_sequences(data, look_back, look_forward) preds batch_predict_per_cluster(models_cluster, kmeans, scaler, X_file) base_name os.path.basename(file_path).replace(.csv, _pred.csv) df_pred pd.DataFrame( preds.reshape(preds.shape[0], -1), columns[fpred_step{i}_feat{j} for i in range(look_forward) for j in range(2)] ) df_pred.to_csv(os.path.join(output_dir, base_name), indexFalse) print(fsave {base_name}, rows{len(df_pred)}) predict_csv_files(./files, ./outputs)这种批量处理方式在生产环境里很常见。只要输入文件格式统一就能自动化跑完整个目录。7.3 封装为 API 服务把训练好的模型保存下来然后用 FastAPI 封装成一个预测服务。这样其他模块就可以直接通过 HTTP 调用不需要关心模型内部实现。import joblib import torch from fastapi import FastAPI from pydantic import BaseModel app FastAPI() # 假设你已经保存好了模型和预处理对象 # torch.save(models_cluster, models_cluster.pt) # joblib.dump(kmeans, kmeans.pkl) # joblib.dump(scaler, scaler.pkl) class PredictRequest(BaseModel): features: list # shape: (look_back, num_vars) app.post(/predict) def predict(req: PredictRequest): x np.array(req.features, dtypenp.float32) if x.ndim ! 2 or x.shape[0] ! 24: return {error: features must be shape (24, 2)} x_flat x.reshape(1, -1) x_scaled scaler.transform(x_flat) c kmeans.predict(x_scaled)[0] model models_cluster[c] model.eval() with torch.no_grad(): pred model(torch.tensor(x, dtypetorch.float32).unsqueeze(0)).numpy() return { cluster: int(c), prediction: pred.tolist() }启动命令uvicorn api_server:app --host 0.0.0.0 --port 8000调用示例curl -X POST http://127.0.0.1:8000/predict \ -H Content-Type: application/json \ -d { features: [ [45.2, 0.82], [45.8, 0.85], ... ] }如果服务要对外开放必须加访问控制、限流和鉴权不要裸奔在公网上。8. 资源占用与性能观察8.1 训练阶段如何观察资源CPU 训练时重点看内存占用和 CPU 使用率GPU 训练时重点看显存占用。PyTorch 里可以用torch.cuda.memory_allocated()查看当前显存占用但要注意开启 CUDA 后才有效。if torch.cuda.is_available(): torch.cuda.reset_peak_memory_stats() # 训练代码... print(peak memory (MB):, torch.cuda.max_memory_allocated() / 1024**2)8.2 哪些因素影响资源消耗因素影响look_back 序列长度越长LSTM 展开步数越多训练越慢hidden_size越大模型参数量越多占用越高num_layers层数增加显存和内存占用明显上升batch size越大一次前向和反向传播占用的显存越高output_len只影响输出层维度对 LSTM 计算量影响较小分簇数量 KK 越大模型数量越多总训练时间线性增加8.3 降低资源消耗的建议在 CPU 上训练时batch_size可以设大一点但不要超过内存容量。在 GPU 上显存不够时优先减小batch_size其次减小hidden_size。分簇训练时如果某个簇样本量特别少可以对这个簇使用更小的hidden_size或更少的num_layers。如果数据量极大可以考虑用 Keras/TensorFlow 的tf.data做流式加载避免一次把数据全部读进内存。9. 常见问题与排查方法问题现象可能原因排查方式解决方案K-means 聚类结果不稳定K 值选择不当或数据没标准化查看肘部曲线和轮廓系数先标准化再尝试多个 K 值分簇训练后反而更差数据没有明显分群对比基准模型和分簇模型指标改用方式二即聚类标签作为特征LSTM 训练 loss 不下降学习率过大或数据未归一化观察 loss 曲线使用学习率调度输入数据做标准化预测值全部接近均值序列长度太短或模型容量不足检查训练集 MAE增大 look_back、hidden_size 或 epoch测试集指标比训练集差很多过拟合对比训练集和测试集 loss增加正则化、early stopping 或减小模型批量推理时内存溢出一次性读入太多数据观察内存占用用分块读取或 DataLoader 控制 batch sizeAPI 请求报错 shape 不匹配前端传入的 features 维度不对查看请求日志检查输入数据 shape 是否为 (24, 2)窗口边缘预测不准确序列开头缺失历史信息查看预测曲线对前 look_back 个时间点仅用已有数据做粗预测9.1 关键排查细节第一K-means 之后不要直接拿原始特征训练 LSTM。需要确保 LSTM 模型的输入特征也经过了同一套标准化否则模型输入分布不一致会导致推理偏差。第二分簇训练时测试集样本的簇标签必须由 K-means 模型预测得到不能使用真实标签或未来信息。否则评估结果会虚高部署后表现会明显下滑。第三时间序列切分时训练集和测试集要按时间顺序切并且从测试集滑窗里不能出现训练集时间范围的数据。建议在create_sequences之前就先切分原始时间序列而不是先滑窗再切分。10. 最佳实践与使用建议10.1 第一次跑通的最小配置先用 look_back12、look_forward3、hidden_size32、epochs10 跑通全流程确认代码没有逻辑问题后再调大参数。不要一上来就上大模型。10.2 模型和预处理对象统一保存分簇模型、K-means 对象、标准化器、滑窗参数都要保存到同一份配置里。部署时如果只加载模型忘记加载 scaler预测结果会完全错乱。# 保存所有必要对象 torch.save(models_cluster, models_cluster.pt) joblib.dump(kmeans, kmeans.pkl) joblib.dump(scaler, scaler.pkl) # 记录滑窗参数 import json config {look_back: 24, look_forward: 6, num_vars: 2} with open(config.json, w) as f: json.dump(config, f)10.3 批量任务要加日志和失败重试批量预测时如果某个文件解析失败或数据处理报错不要中断整个任务。建议在循环里捕获异常并写入日志最后统计成功和失败的文件数。import logging logging.basicConfig(filenamebatch.log, levellogging.INFO) for file_path in csv_files: try: process_one_file(file_path) logging.info(fOK: {file_path}) except Exception as e: logging.error(fFAIL: {file_path}, error{e})10.4 API 服务要限制访问范围生产环境部署 API 时建议服务绑定内网 IP不要直接暴露公网端口。加 token 鉴权或 API Key。设置请求频率限制。对输入数据长度做校验防止恶意超长序列拖垮服务。10.5 数据合规提醒这套方案本身是通用算法但如果数据来自真实设备、真实用户或真实业务必须在采集、训练、部署前确认数据来源合法个人隐私或商业敏感信息做好脱敏处理。模型发布或商用前要做效果复核不能在不确认输出质量的情况下直接对外服务。11. 总结与下一步K-means LSTM 多输出回归这套组合最值得尝试的点是它能低成本地处理多工况时间序列数据。先用 K-means 做数据洞察再用分簇模型提升预测精度整个流程从聚类到训练再到推理代码量不大CPU 就能跑。最先应该验证的是聚类效果如果 silhouette score 偏低或者各簇之间指标没有明显差距后续分簇模型大概率不会有太大收益。最容易踩的坑是时间序列数据切分时打乱顺序或者 K-means 聚类时混入未来标签信息。这两个问题都会让测试集指标虚假偏高部署后立刻露馅。所以评估时必须把数据切分和聚类流程放到训练之前严格执行“只用训练数据拟合 scaler 和 K-means再转换测试数据”。后续可以继续扩展的方向有三个一是把 K-means 换成 GMM 高斯混合模型能输出簇概率融合预测更平滑二是在 LSTM 里加入 Attention 机制处理更长序列三是把分簇后的模型接入 FastAPI 服务做成实时预测接口。建议收藏备用跑完这套流程后再按自己的数据特点调整 K 值和模型结构。 K-means+LSTM多输出回归：多工况时间序列预测实战方案
这次我们直接看一个工程里经常遇到的组合方案用 K-means 先给时间序列数据做聚类再针对不同数据簇分别训练 LSTM 多输出回归模型。这个方法解决的实际问题很明确——如果训练数据里存在多种运行工况、多个业务模式直接塞进一个 LSTM 里模型往往会被“平均效果”拖垮而 K-means 把样本按特征空间拆成多个簇后每个簇单独建模预测精度通常会更稳。这套方案的另一个价值是门槛低。K-means 没有调参地狱LSTM 用 PyTorch 或 Keras 都能落地CPU 也能完成训练和推理。如果你只是做多步时间序列预测或者同时预测多个目标变量这篇文章可以直接收藏。下面我会从方法原理、环境准备、K-means 聚类、LSTM 多输出模型构建、联合预测验证、批量推理、性能观察和问题排查这几个方面展开代码全部给到可直接复制运行的版本。1. 核心能力速览能力项说明方法类型无监督聚类 有监督深度学习回归主要功能多变量时间序列多步预测、多目标同时回归算法组成K-meanssklearn、LSTMPyTorch / Keras运行平台Windows / Linux / macOS硬件要求CPU 可跑建议 4GB 以上内存如有 NVIDIA GPU 可加速训练Python 版本3.8 及以上启动方式Python 脚本 / Jupyter Notebook是否支持 API可快速封装为 FastAPI / Flask 服务是否支持批量任务支持Python 批量推断即可适合场景多工况工业数据预测、多步负荷预测、多指标同时预测这里的显存占用不做硬性假设因为 LSTM 的显存或内存消耗取决于输入序列长度、隐藏层大小、batch size 和输出维度。实测时用nvidia-smi或torch.cuda.memory_allocated()观察即可。2. 方法原理与适用场景2.1 先看数据长什么样假设你有一组多维时间序列数据每一行是一个时间点的观测值列是不同传感器或业务指标时间温度压力转速振动t045.20.8212000.13t145.80.8512100.14...............多输出回归在这个场景里有两层含义预测未来多个时间步的值例如用过去 24 小时预测未来 24 小时。同时预测多个变量例如同时输出温度、压力、转速和振动。两者可以叠加输出维度 预测步数 × 变量数。2.2 为什么要先用 K-means 聚类工业数据和业务数据里不同工况往往对应不同的数据分布。比如设备在“正常负载”“高负载”“待机”三种状态下传感器数值的均值、波动幅度都不同。如果把这些数据混在一起训练 LSTM模型只能学习到一个中间态容易出现“高负荷阶段预测偏低、低负荷阶段预测偏高”的问题。K-means 在这里的作用是把原始样本按特征空间划分成 K 个簇。每个簇内数据分布更接近单独训练 LSTM 更容易拟合。预测新样本时先判断它属于哪个簇再选择对应模型或包含聚类标签的特征来预测。需要注意K-means 聚类必须使用滑窗特征不能使用未来标签信息否则会造成信息泄露测试集上指标虚高。2.3 三种常见组合方式组合方式做法优缺点方式一分簇训练每个簇单独训练一个 LSTM 模型拟合能力强但需要维护 K 个模型预测时先聚类再选模型方式二聚类标签作为特征将 K-means 得到的簇标签做 one-hot拼接到 LSTM 输入特征中只训练一个模型简单稳定适合数据量不大时方式三聚类结果做门控用簇概率或距离加权多个模型输出融合效果平滑但工程复杂度最高本文主要演示方式一和方式二因为它们在实战中最常用代码也容易理解。2.4 适用边界这套方法适合有显著多模式特征的时间序列回归任务比如多工况设备预测、多季节电力负荷预测、多行业销售数据预测。不适合的情况是数据本身只有一种模式或者样本量太小每个簇只有几十条样本在这种情况下 K-means 没有收益反而增加训练和推理成本。必须强调如果业务场景涉及真实设备数据、用户隐私数据或商业数据请先确认数据来源合法、脱敏合规。人脸、声音、个人位置等敏感数据不要随意采集和训练。3. 环境准备与数据集构造3.1 安装依赖建议新建一个独立 Python 环境避免依赖冲突。conda create -n kmeans_lstm python3.10 -y conda activate kmeans_lstm然后安装核心依赖pip install torch scikit-learn pandas numpy matplotlib如果使用 Keraspip install tensorflow scikit-learn pandas numpy matplotlib没有 GPU 也能跑训练时间会长一些。后面的代码默认使用 PyTorch如果你更熟悉 TensorFlow整个方案的结构完全一致只是模型定义部分需要替换。3.2 构造示例数据集为了验证流程这里构造一个包含两种工况的合成时间序列数据。实际项目中把数据读取部分替换成自己的 CSV 或数据库查询即可。import numpy as np import pandas as pd np.random.seed(42) n 2000 time np.arange(n) # 工况A均值为 50振荡幅度小 mode_a 50 5 * np.sin(time / 20) np.random.normal(0, 0.5, n) # 工况B均值为 80振荡幅度大 mode_b 80 10 * np.sin(time / 15) np.random.normal(0, 1.2, n) # 构造分段数据前1000点是工况A后1000点是工况B feature1 np.concatenate([mode_a[:1000], mode_b[1000:]]) feature2 feature1 * 0.6 np.random.normal(0, 1, n) df pd.DataFrame({ time: time, feat1: feature1, feat2: feature2, }) df.head()这里的feat1和feat2是两个相关变量我们的目标是同时预测未来多个时间步的这两个变量。3.3 滑窗生成监督学习数据集LSTM 模型的输入是“序列样本”。对每个时间点 t取过去look_back个时间点作为特征预测未来look_forward个时间点的两个变量。def create_sequences(data, look_back24, look_forward6): X, y [], [] for i in range(len(data) - look_back - look_forward 1): X.append(data[i:i look_back]) y.append(data[i look_back:i look_back look_forward]) return np.array(X), np.array(y) # 只使用特征列标签为未来特征向量 data df[[feat1, feat2]].values X, y create_sequences(data, look_back24, look_forward6) print(X shape:, X.shape) print(y shape:, y.shape)这里的 X shape 是(样本数, 24, 2)y shape 是(样本数, 6, 2)。因为 K-means 聚类不做时序依赖假设我们先把每个样本的二维特征24 × 2 48 维展平用于聚类LSTM 模型仍使用三维输入格式。4. K-means 聚类实现4.1 标准化与聚类对展平后的特征做标准化再用肘部法则估计 K 值。from sklearn.preprocessing import StandardScaler from sklearn.cluster import KMeans X_flat X.reshape(X.shape[0], -1) scaler StandardScaler() X_flat_scaled scaler.fit_transform(X_flat) # 用肘部法则观察K值选择 from sklearn.metrics import silhouette_score inertias [] for k in range(2, 8): km KMeans(n_clustersk, random_state42, n_init10) km.fit(X_flat_scaled) inertias.append(km.inertia_) print(inertias:, inertias) # 选择拐点处的K值这里直接取2 K 2 kmeans KMeans(n_clustersK, random_state42, n_init10) labels kmeans.fit_predict(X_flat_scaled) print(cluster distribution:, np.bincount(labels))在这个构造数据里K 值确实等于 2。实际项目中不要直接抄要先看完肘部曲线再决定。如果数据分布没有明显分群K-means 的收益会很有限。4.2 聚类结果可视化import matplotlib.pyplot as plt plt.figure(figsize(12, 4)) for c in range(K): cluster_idx np.where(labels c)[0] plt.scatter(cluster_idx, X_flat[cluster_idx, 0], s2, labelfcluster {c}) plt.legend() plt.title(K-means Clustering Result on Sample Index) plt.show()如果聚类效果正常你会看到两个簇在时间轴上大致对应两种工况的区间。边缘处会有一些交叉这是正常的。4.3 划分训练集和测试集注意时间序列数据必须按时间顺序切分不能随机打乱否则会引入未来信息泄漏。train_ratio 0.8 split int(len(X) * train_ratio) X_train, X_test X[:split], X[split:] y_train, y_test y[:split], y[split:] labels_train, labels_test labels[:split], labels[split:] print(X_train:, X_train.shape) print(X_test:, X_test.shape)5. LSTM 多输出回归模型构建5.1 模型定义LSTM 输出形状是(batch_size, look_forward, num_vars)。我们用两层 LSTM 加一个全连接输出层。import torch import torch.nn as nn class LSTMRegressor(nn.Module): def __init__(self, input_size2, hidden_size64, num_layers2, output_len6, output_size2): super().__init__() self.lstm nn.LSTM( input_sizeinput_size, hidden_sizehidden_size, num_layersnum_layers, batch_firstTrue ) self.fc nn.Linear(hidden_size, output_len * output_size) self.output_len output_len self.output_size output_size def forward(self, x): # x: (batch, seq_len, input_size) out, _ self.lstm(x) out out[:, -1, :] # 取最后一个时间步的隐藏状态 out self.fc(out) return out.view(-1, self.output_len, self.output_size)这里把输出层设计成(batch, output_len, output_size)也就是一个 LSTM 模型同时完成多步预测和多变量回归。5.2 训练函数为了让代码通用这里不区分簇先在完整训练集上训练一个基准模型方便和分簇模型做对比。def train_model(X_train, y_train, X_val, y_val, epochs20, lr1e-3): model LSTMRegressor(input_sizeX_train.shape[2]) optimizer torch.optim.Adam(model.parameters(), lrlr) criterion nn.MSELoss() X_train_t torch.tensor(X_train, dtypetorch.float32) y_train_t torch.tensor(y_train, dtypetorch.float32) X_val_t torch.tensor(X_val, dtypetorch.float32) y_val_t torch.tensor(y_val, dtypetorch.float32) for epoch in range(epochs): model.train() optimizer.zero_grad() pred model(X_train_t) loss criterion(pred, y_train_t) loss.backward() optimizer.step() if (epoch 1) % 5 0: model.eval() with torch.no_grad(): val_pred model(X_val_t) val_loss criterion(val_pred, y_val_t) print(fEpoch {epoch1}/{epochs}, train_loss{loss.item():.4f}, val_loss{val_loss.item():.4f}) return model model_base train_model(X_train, y_train, X_test, y_test, epochs20)默认情况下这是 CPU 训练。如果你的机器有 GPU可以把 tensor 搬到 CUDA 上代码里加一个设备判断即可。5.3 分簇模型训练这是“K-means LSTM”组合的核心部分。对每个簇用该簇的训练样本单独训练一个模型。def train_per_cluster(cluster_labels, X_train, y_train, epochs20): models {} for c in range(K): idx np.where(cluster_labels c)[0] X_c X_train[idx] y_c y_train[idx] print(fTraining cluster {c}, samples: {X_c.shape[0]}) models[c] train_model(X_c, y_c, X_c, y_c, epochsepochs) return models models_cluster train_per_cluster(labels_train, X_train, y_train, epochs20)这里为了演示代码简洁验证集直接用了训练集。实际项目中请单独划分验证集避免过拟合被掩盖。6. 联合预测与效果验证6.1 预测流程预测时先对新样本做滑窗展平、标准化再用 K-means 模型判断它属于哪个簇最后调用对应簇的 LSTM 模型推理。def predict_with_cluster(model, kmeans, scaler, X_test, cluster_labels): preds [] for i in range(len(X_test)): x_flat X_test[i].reshape(1, -1) x_scaled scaler.transform(x_flat) c kmeans.predict(x_scaled)[0] x_t torch.tensor(X_test[i], dtypetorch.float32).unsqueeze(0) model.eval() with torch.no_grad(): pred model(x_t).numpy() preds.append(pred[0]) return np.array(preds) preds_per_cluster predict_with_cluster( models_cluster, kmeans, scaler, X_test, labels_test )注意上面的推理循环没有 batch 化数据量大时会比较慢。批量推理的优化放到第 7 节。6.2 评估指标多输出回归常用四个指标MAE、RMSE、MAPE、R2。这里定义一组通用评估函数。from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score def evaluate(y_true, y_pred): y_true_flat y_true.reshape(-1, y_true.shape[-1]) y_pred_flat y_pred.reshape(-1, y_pred.shape[-1]) mae mean_absolute_error(y_true_flat, y_pred_flat) rmse mean_squared_error(y_true_flat, y_pred_flat, squaredFalse) r2 r2_score(y_true_flat, y_pred_flat) return {MAE: mae, RMSE: rmse, R2: r2} metrics_base evaluate(y_test, preds_base) metrics_cluster evaluate(y_test, preds_per_cluster) print(baseline metrics:, metrics_base) print(cluster metrics:, metrics_cluster)如果分簇策略有效通常metrics_cluster的 MAE 和 RMSE 会低于metrics_base。如果没有下降或下降很少说明数据没有明显分群就不用继续用分簇方案。6.3 预测曲线对比把某个预测步的实际值和两个模型的预测值画在一起可以直观看到分簇模型是否在跳变区域表现更好。def plot_predictions(y_true, preds_a, preds_b, step0, var0): plt.figure(figsize(12, 4)) plt.plot(y_true[:, step, var], labeltrue, colorblack) plt.plot(preds_a[:, step, var], labelbase LSTM, linestyle--) plt.plot(preds_b[:, step, var], labelKmeansLSTM, linestyle:) plt.legend() plt.title(fStep{step}, Variable{var}) plt.show() plot_predictions(y_test, preds_base, preds_per_cluster, step0, var0)这一张图基本就能看出分簇方案是否值得保留。如果两条预测曲线几乎重合说明 K 的选择或聚类特征没有带来有效信息。7. 功能测试与批量任务7.1 批次推理优化按单条样本循环推理在数据量大的时候很慢。更高效的做法是把整个测试集按簇拆分然后分批输入模型。def batch_predict_per_cluster(models, kmeans, scaler, X_test): # 先给所有样本打簇标签 X_flat X_test.reshape(X_test.shape[0], -1) X_scaled scaler.transform(X_flat) cluster_ids kmeans.predict(X_scaled) outputs np.zeros((X_test.shape[0], y_test.shape[1], y_test.shape[2])) for c in range(K): idx np.where(cluster_ids c)[0] x_t torch.tensor(X_test[idx], dtypetorch.float32) model models[c] model.eval() with torch.no_grad(): pred model(x_t).numpy() outputs[idx] pred return outputs实际项目中还可以用torch.utils.data.DataLoader设置 batch size进一步控制显存和训练稳定性。7.2 批量预测一批 CSV 文件假设有一个files/目录里面有多个设备的 CSV 文件每个文件格式相同。批量推理的流程是读取文件 - 滑窗 - 标准化 - 聚类归属 - 模型预测 - 保存结果。import os import glob def predict_csv_files(files_dir, output_dir, look_back24, look_forward6): os.makedirs(output_dir, exist_okTrue) csv_files glob.glob(os.path.join(files_dir, *.csv)) for file_path in csv_files: df_file pd.read_csv(file_path) data df_file[[feat1, feat2]].values if len(data) look_back look_forward: print(fskip {file_path}, too short) continue X_file, _ create_sequences(data, look_back, look_forward) preds batch_predict_per_cluster(models_cluster, kmeans, scaler, X_file) base_name os.path.basename(file_path).replace(.csv, _pred.csv) df_pred pd.DataFrame( preds.reshape(preds.shape[0], -1), columns[fpred_step{i}_feat{j} for i in range(look_forward) for j in range(2)] ) df_pred.to_csv(os.path.join(output_dir, base_name), indexFalse) print(fsave {base_name}, rows{len(df_pred)}) predict_csv_files(./files, ./outputs)这种批量处理方式在生产环境里很常见。只要输入文件格式统一就能自动化跑完整个目录。7.3 封装为 API 服务把训练好的模型保存下来然后用 FastAPI 封装成一个预测服务。这样其他模块就可以直接通过 HTTP 调用不需要关心模型内部实现。import joblib import torch from fastapi import FastAPI from pydantic import BaseModel app FastAPI() # 假设你已经保存好了模型和预处理对象 # torch.save(models_cluster, models_cluster.pt) # joblib.dump(kmeans, kmeans.pkl) # joblib.dump(scaler, scaler.pkl) class PredictRequest(BaseModel): features: list # shape: (look_back, num_vars) app.post(/predict) def predict(req: PredictRequest): x np.array(req.features, dtypenp.float32) if x.ndim ! 2 or x.shape[0] ! 24: return {error: features must be shape (24, 2)} x_flat x.reshape(1, -1) x_scaled scaler.transform(x_flat) c kmeans.predict(x_scaled)[0] model models_cluster[c] model.eval() with torch.no_grad(): pred model(torch.tensor(x, dtypetorch.float32).unsqueeze(0)).numpy() return { cluster: int(c), prediction: pred.tolist() }启动命令uvicorn api_server:app --host 0.0.0.0 --port 8000调用示例curl -X POST http://127.0.0.1:8000/predict \ -H Content-Type: application/json \ -d { features: [ [45.2, 0.82], [45.8, 0.85], ... ] }如果服务要对外开放必须加访问控制、限流和鉴权不要裸奔在公网上。8. 资源占用与性能观察8.1 训练阶段如何观察资源CPU 训练时重点看内存占用和 CPU 使用率GPU 训练时重点看显存占用。PyTorch 里可以用torch.cuda.memory_allocated()查看当前显存占用但要注意开启 CUDA 后才有效。if torch.cuda.is_available(): torch.cuda.reset_peak_memory_stats() # 训练代码... print(peak memory (MB):, torch.cuda.max_memory_allocated() / 1024**2)8.2 哪些因素影响资源消耗因素影响look_back 序列长度越长LSTM 展开步数越多训练越慢hidden_size越大模型参数量越多占用越高num_layers层数增加显存和内存占用明显上升batch size越大一次前向和反向传播占用的显存越高output_len只影响输出层维度对 LSTM 计算量影响较小分簇数量 KK 越大模型数量越多总训练时间线性增加8.3 降低资源消耗的建议在 CPU 上训练时batch_size可以设大一点但不要超过内存容量。在 GPU 上显存不够时优先减小batch_size其次减小hidden_size。分簇训练时如果某个簇样本量特别少可以对这个簇使用更小的hidden_size或更少的num_layers。如果数据量极大可以考虑用 Keras/TensorFlow 的tf.data做流式加载避免一次把数据全部读进内存。9. 常见问题与排查方法问题现象可能原因排查方式解决方案K-means 聚类结果不稳定K 值选择不当或数据没标准化查看肘部曲线和轮廓系数先标准化再尝试多个 K 值分簇训练后反而更差数据没有明显分群对比基准模型和分簇模型指标改用方式二即聚类标签作为特征LSTM 训练 loss 不下降学习率过大或数据未归一化观察 loss 曲线使用学习率调度输入数据做标准化预测值全部接近均值序列长度太短或模型容量不足检查训练集 MAE增大 look_back、hidden_size 或 epoch测试集指标比训练集差很多过拟合对比训练集和测试集 loss增加正则化、early stopping 或减小模型批量推理时内存溢出一次性读入太多数据观察内存占用用分块读取或 DataLoader 控制 batch sizeAPI 请求报错 shape 不匹配前端传入的 features 维度不对查看请求日志检查输入数据 shape 是否为 (24, 2)窗口边缘预测不准确序列开头缺失历史信息查看预测曲线对前 look_back 个时间点仅用已有数据做粗预测9.1 关键排查细节第一K-means 之后不要直接拿原始特征训练 LSTM。需要确保 LSTM 模型的输入特征也经过了同一套标准化否则模型输入分布不一致会导致推理偏差。第二分簇训练时测试集样本的簇标签必须由 K-means 模型预测得到不能使用真实标签或未来信息。否则评估结果会虚高部署后表现会明显下滑。第三时间序列切分时训练集和测试集要按时间顺序切并且从测试集滑窗里不能出现训练集时间范围的数据。建议在create_sequences之前就先切分原始时间序列而不是先滑窗再切分。10. 最佳实践与使用建议10.1 第一次跑通的最小配置先用 look_back12、look_forward3、hidden_size32、epochs10 跑通全流程确认代码没有逻辑问题后再调大参数。不要一上来就上大模型。10.2 模型和预处理对象统一保存分簇模型、K-means 对象、标准化器、滑窗参数都要保存到同一份配置里。部署时如果只加载模型忘记加载 scaler预测结果会完全错乱。# 保存所有必要对象 torch.save(models_cluster, models_cluster.pt) joblib.dump(kmeans, kmeans.pkl) joblib.dump(scaler, scaler.pkl) # 记录滑窗参数 import json config {look_back: 24, look_forward: 6, num_vars: 2} with open(config.json, w) as f: json.dump(config, f)10.3 批量任务要加日志和失败重试批量预测时如果某个文件解析失败或数据处理报错不要中断整个任务。建议在循环里捕获异常并写入日志最后统计成功和失败的文件数。import logging logging.basicConfig(filenamebatch.log, levellogging.INFO) for file_path in csv_files: try: process_one_file(file_path) logging.info(fOK: {file_path}) except Exception as e: logging.error(fFAIL: {file_path}, error{e})10.4 API 服务要限制访问范围生产环境部署 API 时建议服务绑定内网 IP不要直接暴露公网端口。加 token 鉴权或 API Key。设置请求频率限制。对输入数据长度做校验防止恶意超长序列拖垮服务。10.5 数据合规提醒这套方案本身是通用算法但如果数据来自真实设备、真实用户或真实业务必须在采集、训练、部署前确认数据来源合法个人隐私或商业敏感信息做好脱敏处理。模型发布或商用前要做效果复核不能在不确认输出质量的情况下直接对外服务。11. 总结与下一步K-means LSTM 多输出回归这套组合最值得尝试的点是它能低成本地处理多工况时间序列数据。先用 K-means 做数据洞察再用分簇模型提升预测精度整个流程从聚类到训练再到推理代码量不大CPU 就能跑。最先应该验证的是聚类效果如果 silhouette score 偏低或者各簇之间指标没有明显差距后续分簇模型大概率不会有太大收益。最容易踩的坑是时间序列数据切分时打乱顺序或者 K-means 聚类时混入未来标签信息。这两个问题都会让测试集指标虚假偏高部署后立刻露馅。所以评估时必须把数据切分和聚类流程放到训练之前严格执行“只用训练数据拟合 scaler 和 K-means再转换测试数据”。后续可以继续扩展的方向有三个一是把 K-means 换成 GMM 高斯混合模型能输出簇概率融合预测更平滑二是在 LSTM 里加入 Attention 机制处理更长序列三是把分簇后的模型接入 FastAPI 服务做成实时预测接口。建议收藏备用跑完这套流程后再按自己的数据特点调整 K 值和模型结构。