k-means-LSTM多输入多输出预测实战：聚类分治提升时间序列精度

简介本资源为Python实现的k-means-LSTM多输入多输出组合预测完整项目文档面向具备一定机器学习与Python编程基础的研发人员、数据科学家及时间序列预测学习者。资源以docx文档形式共1个文件压缩包大小57KB内容围绕能源管理、气象预测、金融市场分析、交通流量预测、医疗健康等应用场景系统讲解如何通过k均值聚类进行数据预处理与特征提取再使用LSTM对各子集建模预测以提升精度与鲁棒性。文档完整覆盖环境准备、数据准备、算法设计、模型构建、模型评估、优化调整、GUI设计及代码整合八个阶段包含项目架构说明、算法流程图、模型描述与代码示例、目录结构及各模块功能详解并针对项目部署、扩展、注意事项与未来改进方向给出具体说明。已有66人学习。读者可获得从数据加载到最终预测输出的完整方法论与实操参考尤其适合希望深入理解组合预测模型设计及提升实际问题解决能力的开发者是一份兼具系统性与实用性的技术参考资料。1. k-means-LSTM 多输入多输出预测为什么先聚类再上 LSTM 能救回一波动数据做设备寿命预测、电力负荷预测这类多变量时间序列问题时我常碰到一种“一锅炖”困境把所有工况塞进同一个 LSTM模型只能学到平均规律遇到上升段、抖动段、平缓段混在一起的数据预测曲线总是慢半拍。基于 k-means-LSTM 的多输入多输出组合预测思路是先让 k-means 把历史样本按工况切成几堆每一堆单独训练一个 LSTM预测时根据样本与各类中心的距离选择模型或加权融合多个模型。这样做的好处是每个 LSTM 只学一类形态精度通常比单模型高而且天然支持多特征输入、多步输出。本文会给你一套能跑的 Python 实现包含数据构造、模型训练、GUI 界面和排错经验适合手里有真实时序数据、打算在项目里落地组合预测的工程师。2. 先把问题定义清楚组合预测架构、多输入多输出的样本构造与 K 值选择2.1 组合预测流程聚类–分治–重组k-means-LSTM 不是两个模型简单拼在一起而是一条“聚类–分治–重组”流水线。第一步对历史样本做 k-means 聚类把形态相近的样本归为一类第二步为每个类单独训练一个 LSTM第三步在预测时把新样本映射到已有类得到该类的预测值或者按距离加权合并多个 LSTM 的输出。我做过的小型负荷预测项目里这种方案比单个 LSTM 的 MAPE 低 815 个百分点。原因不难理解单一 LSTM 需要在一套参数里同时拟合多种模式参数会被“平均化”而聚类之后每个 LSTM 的任务边界更清晰。实际落地时有两种重组方式。硬切换最简单新样本离哪个簇中心最近就直接用那个簇的 LSTM。软加权更稳根据样本到各簇中心的距离换算一组权重多个模型按权重叠加相当于做了一个“模型集成”。软加权对处于聚类边界附近的样本更友好但会增加推理耗时所以我通常在离线批量预测时用软加权在线实时预测时用硬切换。2.2 多输入多输出样本构造滑动窗口怎么变成监督学习LSTM 的输入通常是(样本数, 时间步数, 特征数)。多输入好理解每个时间步可以有多列特征。多输出在这里指预测未来多个时刻的目标值也就是多步预测。比如用过去 24 小时的气温、湿度、负荷预测未来 6 小时的负荷输出就是一个长度为 6 的向量。构造训练样本的常见做法是滑动窗口。假设数据是一个二维数组data每一行是一个时间点每一列是一个特征。我一般写这样的函数import numpy as np def make_samples(data, target_col0, n_in24, n_out6): X, y [], [] total len(data) for i in range(total - n_in - n_out 1): X.append(data[i:i n_in]) # 过去 n_in 步的所有特征 y.append(data[i n_in:i n_in n_out, target_col]) # 未来 n_out 步的目标列 return np.array(X), np.array(y)这段代码里X的形状是(样本数, n_in, 特征数)也就是一个三维张量可以直接喂给 LSTMy的形状是(样本数, n_out)是未来多个时间步的目标值。target_col指定哪一列是要预测的变量。n_in和n_out是两个关键超参数n_in太小模型看不到足够历史容易漏掉周期n_in太大数据量减少而且训练变慢。我一般用网格搜索从[12, 24, 48]里选序列本身有日周期就优先 24 或 48。2.3 k-means 特征与 K 值手肘图和轮廓系数怎么配合k-means 不直接对三维序列聚类因为直接把每个窗口摊平会导致维度爆炸。通常的做法是从每个窗口里提取一组描述“形态”的特征常见的有窗口均值、最大值、最小值、一阶差分均值、标准差、趋势斜率。这样每个窗口变成一个低维向量再交给 k-means。from sklearn.cluster import KMeans from sklearn.metrics import silhouette_score from sklearn.preprocessing import StandardScaler def window_features(X): feat np.column_stack([ X.mean(axis1), X.max(axis1), X.min(axis1), np.diff(X, axis1).mean(axis1), X.std(axis1), ]) return feat feat window_features(X_train) feat_scaled StandardScaler().fit_transform(feat) for k in range(2, 9): km KMeans(n_clustersk, n_init10, random_state0) labels km.fit_predict(feat_scaled) sil silhouette_score(feat_scaled, labels) print(fk{k}, silhouette{sil:.4f})需要注意窗口形状是完全一样的但不同工况可能均值差别很大。直接拿原始特征聚类时量纲大的列会主导距离所以先做标准化。上面这段代码会打印每个 K 的轮廓系数通常选轮廓系数最高的 K如果多个 K 接近再看手肘图里的“拐点”也就是簇内误差平方和增速变慢的位置。提示聚类用的特征不需要和 LSTM 的输入特征一致。你可以先用业务经验选一组能区分工况的统计量再跑 k-means而不是把所有特征全塞进去。3. 动手实现k-means 聚类、LSTM 分簇训练与组合预测的核心代码3.1 造一份带三种工况的多变量数据方便验证流程没有真实数据时最好先造一份包含三种明显工况的数据来验证代码。下面的例子生成 2000 个时间点三种工况分别是正弦波动、线性上升、高频抖动每个时间点有两列特征。import numpy as np rng np.random.default_rng(42) t np.arange(2000) feature np.zeros((len(t), 2)) target np.zeros(len(t)) for i in range(3): start, end i * 600, (i 1) * 600 if i 0: feature[start:end, 0] np.sin(t[start:end] * 0.05) feature[start:end, 1] np.cos(t[start:end] * 0.05) target[start:end] np.sin(t[start:end] * 0.05) * 2 1 elif i 1: feature[start:end, 0] np.linspace(-1, 1, 600) feature[start:end, 1] np.linspace(1, -1, 600) target[start:end] np.linspace(0, 3, 600) else: rnd rng.normal(0, 1, 600) feature[start:end, 0] rnd feature[start:end, 1] rng.normal(0, 0.5, 600) target[start:end] rnd * 2 np.sin(t[start:end] * 0.1) data np.column_stack([feature, target]) print(data.shape)这段数据不是真实的项目数据但足以跑通完整流程。target作为第三列特征放进data构造样本时它既是输入之一也是输出目标。这样做的原因是预测负荷一类问题时历史负荷本身就是最强特征所以要把目标列的滞后值也作为输入。3.2 归一化、滑动窗口和 k-means 分簇训练 LSTM 前必须做归一化否则 LSTM 很容易在训练初期产生 NaN。我习惯先对全量数据做MinMaxScaler再构造滑动窗口。顺序别搞反如果先切窗口再归一化样本之间可能有重叠会把未来信息泄进归一化参数里。from sklearn.preprocessing import MinMaxScaler from sklearn.cluster import KMeans scaler MinMaxScaler() data_scaled scaler.fit_transform(data) X, y make_samples(data_scaled, target_col2, n_in12, n_out3)接下来是分簇。分簇必须只用训练窗口的特征不能混入测试窗口。先用时间顺序划好训练集和测试集再在训练集上fitk-means测试时对测试窗口调用predict。否则你会把一整段时间序列的特征也都泄露给聚类模型。X_train, X_test X[:1400], X[1400:] y_train, y_test y[:1400], y[1400:] train_feat window_features(X_train) test_feat window_features(X_test) scaler_feat StandardScaler().fit(train_feat) train_feat_scaled scaler_feat.transform(train_feat) test_feat_scaled scaler_feat.transform(test_feat) kmeans KMeans(n_clusters3, n_init10, random_state0) train_labels kmeans.fit_predict(train_feat_scaled) test_labels kmeans.predict(test_feat_scaled)n_init10是 k-means 的一个重要参数。k-means 受初始簇中心影响默认只跑一次容易出现坏中心。设置为 10 表示用 10 组初始中心各跑一遍取最优结果。random_state固定后每次运行结果一致方便复现。3.3 为每个簇训练一个 LSTM并保存模型分簇完成后按聚类标签把训练样本拆成多个子集然后为每个子集训练一个独立的 LSTM。下面这段代码在 Keras 下实现。from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense, Dropout from tensorflow.keras.optimizers import Adam def build_model(n_in12, n_features3, n_out3): model Sequential() model.add(LSTM(64, input_shape(n_in, n_features), return_sequencesTrue)) model.add(Dropout(0.2)) model.add(LSTM(32, return_sequencesFalse)) model.add(Dropout(0.2)) model.add(Dense(16, activationrelu)) model.add(Dense(n_out, activationlinear)) model.compile(optimizerAdam(0.001), lossmse, metrics[mae]) return model models [] histories [] for cluster_id in range(kmeans.n_clusters): idx train_labels cluster_id if idx.sum() 30: continue X_cluster X_train[idx] y_cluster y_train[idx] model build_model() history model.fit( X_cluster, y_cluster, validation_split0.2, epochs30, batch_size32, verbose0 ) models.append(model) histories.append(history)这段代码里有两个 LSTM 层。第一层设置return_sequencesTrue是为了输出完整的时间步序列给第二层继续提取时序特征第二层return_sequencesFalse只输出最后一个时间步的隐状态再接全连接层。Dropout(0.2)是正则化防止某个簇样本太少导致过拟合。损失函数用mse因为多步输出的每个值都是连续量如果做分类预测这里要换成交叉熵。如果某个簇的样本少于 30我会直接跳过。样本太少时训练出来的 LSTM 方差很大不如直接整体复制全局模型的预测结果。你可以在代码里为这种情况留一个fallback_model。3.4 组合预测硬切换和软加权两种预测函数预测阶段先计算新窗口的聚类特征再判断它属于哪个簇。硬切换直接选对应模型。def predict_cluster(X_new, models, kmeans): feat window_features(X_new) feat_scaled scaler_feat.transform(feat) labels kmeans.predict(feat_scaled) return models[labels[0]].predict(X_new)这里models是按簇索引排列的模型列表kmeans.predict返回的是 0 到 K-1 的整数。需要特别小心build_model输入的n_features是你给每个时间步塞了几列不是聚类特征维度。如果 LSTM 输入里包含目标列特征数就是data.shape[1]。软加权需要用到样本到各簇中心的距离。kmeans.transform(feat_scaled)返回每个样本到每个中心的欧氏距离形状是(样本数, 簇数)。def predict_weighted(X_new, models, kmeans): feat window_features(X_new) feat_scaled scaler_feat.transform(feat) dists kmeans.transform(feat_scaled).squeeze() weights 1.0 / (dists 1e-6) weights weights / weights.sum() pred np.zeros((X_new.shape[0], models[0].output_shape[-1])) for i, model in enumerate(models): pred weights[i] * model.predict(X_new) return pred权重公式用的是距离倒数归一化。距离越近权重越大加1e-6是为了防止距离为 0 导致除零。这个做法的好处是避免了硬切换在边界处的“跳变”坏处是如果某个 LSTM 在这个簇上效果很差软加权会把它的误差也带进来。所以实际项目里我会先跑硬切换把每个簇的验证误差单独算出来再决定要不要上软加权。4. GUI 设计把训练、预测和曲线展示装进一个 Tkinter 面板4.1 界面布局参数区、训练区、结果区的三栏设计给模型套 GUI 不只是为了提高观感更重要的是让非工程人员也能在界面上调参、训练、看曲线。我常用的布局是三栏结构左侧放参数设置和按钮中间放日志输出右侧放 matplotlib 预测曲线图。Tkinter 是 Python 自带库不需要额外安装适合快速封装一个工具。下面是一个最小框架import tkinter as tk from tkinter import ttk, filedialog, scrolledtext app tk.Tk() app.title(k-means-LSTM 多输入多输出预测工具) left ttk.Frame(app, padding10) left.pack(sideleft, filly) middle ttk.Frame(app, padding10) middle.pack(sideleft, fillboth, expandTrue) right ttk.Frame(app, padding10) right.pack(sideright, fillboth, expandTrue) data_path tk.StringVar() ttk.Label(left, text数据文件:).pack(anchorw) ttk.Entry(left, textvariabledata_path).pack(fillx, pady2) ttk.Button(left, text选择 CSV, commandlambda: data_path.set( filedialog.askopenfilename(filetypes[(CSV, *.csv)]) )).pack(fillx) n_in_var tk.IntVar(value12) ttk.Label(left, text输入步数 n_in).pack(anchorw) ttk.Spinbox(left, from_6, to96, textvariablen_in_var).pack(fillx, pady2) n_out_var tk.IntVar(value3) ttk.Label(left, text输出步数 n_out).pack(anchorw) ttk.Spinbox(left, from_1, to24, textvariablen_out_var).pack(fillx, pady2) log_text scrolledtext.ScrolledText(middle, width60, height20) log_text.pack(fillboth, expandTrue) app.mainloop()这段代码先把界面骨架搭起来。StringVar、IntVar是 Tkinter 的变量类型界面上的输入框和后面的 Python 代码通过它们同步。ttk.Spinbox用来限制参数范围用户输错的可能性会小很多。日志区用ScrolledText训练过程中可以把 loss 信息追加进去用户才知道当前进度。4.2 训练线程隔离为什么必须用 threadingTkinter 的mainloop运行在主线程如果把模型训练直接放进按钮回调界面会卡死甚至在高强度训练时整个窗口变成“无响应”。正确做法是把训练丢进子线程训练完成后通过queue或after把结果送回 UI 线程。import threading import queue log_queue queue.Queue() def worker(): # 这里放第 3 章的训练逻辑 for epoch in range(30): log_queue.put(fepoch {epoch 1}/30 done) log_queue.put(TRAIN_FINISHED) def poll_queue(): try: while True: msg log_queue.get_nowait() log_text.insert(end, msg \n) log_text.see(end) except queue.Empty: pass app.after(200, poll_queue) def start_train(): th threading.Thread(targetworker, daemonTrue) th.start() ttk.Button(left, text开始训练, commandstart_train).pack(fillx, pady4) poll_queue() app.mainloop()关键点是把训练函数整体移到worker里poll_queue每 200 毫秒从队列读一次消息并刷新界面。这样即使模型训练很慢界面依然能拖动、能看到进度。注意daemonTrue这样关闭窗口时子线程会被强制退出不会出现关不掉的黑匣子进程。4.3 预测展示让用户选中一段数据就能看到多步输出训练完成后界面上要能选择预测起点并显示未来多个时间步的预测结果。这里我把训练好的模型存到 Python 全局变量里再画到右侧的 matplotlib 图上。import matplotlib matplotlib.use(TkAgg) from matplotlib.figure import Figure from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg def show_predict(): start_idx int(start_idx_var.get()) X_new X_test[start_idx:start_idx 1] y_true y_test[start_idx:start_idx 1] if use_weighted.get(): pred predict_weighted(X_new, models, kmeans) else: pred predict_cluster(X_new, models, kmeans) fig Figure(figsize(5, 3), dpi100) ax fig.add_subplot(111) ax.plot(range(1, len(y_true[0]) 1), y_true[0], o-, label真实) ax.plot(range(1, len(pred[0]) 1), pred[0], x--, label预测) ax.set_xlabel(未来步数) ax.set_ylabel(归一化值) ax.legend() canvas FigureCanvasTkAgg(fig, masterright) canvas.draw() canvas.get_tk_widget().pack()这段代码用matplotlib的FigureCanvasTkAgg把图嵌入 Tkinter 面板。use_weighted是一个BooleanVar对应界面上的 Checkbutton让用户决定用硬切换还是软加权。展示时记得说明坐标是归一化后的值要在旁边留一个“还原到原始量纲”的按钮否则业务人员看不懂曲线到底是多少。5. k-means-LSTM 常见问题与避坑指南从 NaN 损失到 GUI 卡死5.1 现象训练集和测试集分布不一致新样本被分到未知簇有次我把训练数据按时间切成前后两段前面 70% 是夏天数据后面 30% 是冬天数据。训练好的 k-means 完全没见过冬天的形态测试窗口离三个簇中心都很远硬切换随便指了一个簇预测误差暴涨。原因是 k-means 只能描述训练集里出现过的样本分布它没有“第四簇”概念。解决方法是先看测试样本到最近簇中心的最大距离。如果这个距离明显大于训练样本的距离分布说明出现了训练时没见过的工况。我会把这种情况标记为“未知工况”回退到全局模型或者重新聚一次类给新工况单独训练模型。5.2 现象LSTM 输出维度与多步预测对不上把n_out从 3 改成 6 之后预测代码报维度错误或者画图时发现预测曲线比真实曲线短一截。原因通常是最后一层Dense的神经元数量写死成了3。Dense输出数量必须等于要预测的未来步数。解决方法是把n_out作为参数传入build_model代码里统一用model.add(Dense(n_out, activationlinear))。我还在 GUI 里加了校验用户改了n_out就重建整个模型避免沿用过时模型文件。5.3 现象GUI 点击“开始训练”就无响应第一次做 Tkinter 工具时我把model.fit直接写在按钮回调里训练一轮要好几秒界面直接白屏。原因就是训练占用了 UI 主线程事件循环被阻塞。解决方法是坚持线程隔离训练函数整体放进threading.Thread通过queue向外发进度信息GUI 每 200ms 轮询一次。还有一个容易忽略的问题是 matplotlib 的FigureCanvasTkAgg不能在新线程里直接画图所有 UI 更新必须回到主线程所以我只在子线程里算预测结果把预测值放进队列由主线程更新图形。5.4 现象k-means 聚类结果每次跑都不一样同一个数据连续训练两次某个样本第一次分到簇 0第二次分到簇 2导致两个 LSTM 的输入样本数差异很大。原因是 k-means 初始化中心是随机的不同初始点收敛到不同局部最优。解决方法是固定random_state并且设置n_init10。n_init越大结果越稳定但训练时间也越长。如果是工业现场我还会把 k-means 中心和scaler参数一起存成joblib文件预测服务和训练服务统一加载同一份。5.5 现象训练 loss 变成 NaN或准确率忽高忽低模型训练到第 10 轮时 loss 突然变成 NaN或者验证 loss 剧烈震荡多次训练结果完全不同。常见原因有三个。第一数据没有归一化LSTM 的梯度爆炸了第二学习率太大Adam 虽然自适应但 0.01 对很多序列数据来说仍然激进第三某个簇的样本数太少模型在少量样本上反复拟合导致过拟合。我的排查顺序是先保证MinMaxScaler在构造样本之前执行再把学习率调回0.001最后看每个簇的样本数量少于 50 就强制加 Dropout 或改用全局 LSTM。损失震荡还有一个隐性原因按簇切分训练集时打乱了时间顺序验证集可能刚好取到时间轴上相邻的样本造成验证集和训练集高度相关。这种情况需要在簇内部用时间顺序划分而不是随机划分。6. 进阶验证与调参技巧怎么证明组合预测真的值回票价先跑通流程不算完真正投入项目前我建议做一组可量化的对比实验。至少要对比三个对象单个 LSTM、k-means 硬切换 LSTM、k-means 软加权 LSTM。评估指标除了 RMSE还要看 MAE 和 MAPE因为多步预测里远期误差异常大只用一个指标会看不出模型是在第一步准还是全程准。对比时有一点容易被忽略必须保证所有模型看到的数据范围和归一化参数一致。我习惯把单个 LSTM 和组合模型放进同一个训练脚本共用同一个scaler、同一个训练集和测试集而不是分别跑两个脚本后再手动对齐。预测结果按时间步拆开统计看第一步到第n_out步的误差递增情况。如果第 4 步以后误差突然恶化说明n_out太大模型学到的是重复最后一步的“偷懒策略”。最后说一个调参玄学如果 k-means 聚出的簇里 LSTM 表现反而比全局差不要急着加层先看聚类是不是只按均值区分。多数时候把聚类特征里的“一阶差分均值”和“标准差”权重调大会比网络加两层更管用。我的收尾检查是把每个簇的测试样本单独印出误差肉眼看一下是否集中在某些时间段如果某个簇的误差集中在某个固定时段我会把这个时段作为一个哑变量加进 LSTM 输入效果往往立竿见影。这套流程我用了很多次先聚类后分治的思路在波动大的真实数据上确实值得一试希望帮到你。本文还有配套的精品资源点击获取