简介面向电商供应链与数据挖掘场景的DNN销量预测项目源码可用于长尾商品7天、30天、60天销量预测帮助解决备货难题。项目基于TensorFlow 1.13低阶API实现清晰展示了训练集、验证集、测试集划分并通过tf.train.Saver完成模型保存、tf.summary.FileWriter配合TensorBoard可视化训练过程。对于EarlyStopping代码中实现了连续多个epoch未刷新最佳验证精度即停止训练的机制并给出离线加载模型进行预测的完整示例。压缩包共6个文件包括5个Python脚本和1个Markdown说明文档包体仅20KB。已有127人学习资源虽小但覆盖从训练、调参到离线预测的完整闭环同时配有单输出与多输出两种预测分支读者可参考其数据划分策略与训练流程迁移至其他稀疏时序预测任务。1. 长尾商品销量DNN预测先看这个源码包能解决什么供应链备货里最头疼的往往不是爆款而是那些销量忽高忽低、单量稀少的长尾商品。爆款用移动平均或者ARIMA都能凑合长尾商品的特征是稀疏、间歇、波动大常规统计模型基本失效。这个python实现的长尾商品销量DNN预测项目目标很明确——用DNN去拟合长尾商品的销量序列输出7天、30天、60天三个备货窗口的预测量直接对接供应链系统的备货决策。它本身不追求预测精度高到离谱而是把能跑通、能保存、能加载、能可视化这套工程链路完整做出来了。我拆完这个包之后的感觉是如果你正在用TensorFlow做销量预测类项目或者正准备从传统时序模型切到深度学习这份源码能让你少走很多弯路。它适合有一定Python基础、想直接看DNN在销量预测场景里怎么落地的开发者也适合那些需要快速搭一个预测原型去验证效果的算法工程师。2. 拆解源码结构single_output与multiple_output的双分支设计2.1 目录里到底有什么如果你下载了这个zip包解压之后首先看到的是根目录下的README.md然后是两个子目录single_output和multiple_output。每个子目录下都有独立的train.py、predict.py和__init__.py。这个结构设计意图很明显——同一个项目针对两种不同的预测输出形式做了两套实现。single_output的含义是模型每次只输出一个预测值比如只预测未来7天的销量。如果你要分别做7天、30天、60天的预测那就训练三个模型每个模型对应一个输出。这种做法的优点是模型结构简单每个模型只专注一个目标缺点是训练和部署成本高三个模型要分别维护。multiple_output则是让模型一次性输出多个预测值比如同时输出未来7天、30天、60天的销量。这样做的好处是特征提取层可以共享底层学到的模式只计算一次理论上训练效率更高。但代价是模型结构更复杂而且多个输出头之间可能存在相互干扰。我在实际使用中倾向于先跑single_output验证特征有效性再切到multiple_output做正式训练。2.2 数据格式与特征组织方式源码里没有直接附带数据集但根据train.py里的数据加载逻辑可以倒推出数据组织方式。训练脚本读取的是CSV格式的文件每一行代表一个商品在某一天的记录。关键特征是商品ID、日期、销量以及从日期衍生的时间特征比如年、月、日、星期几。# 特征工程示意从原始销售记录中构造模型输入 def build_features(df): df[year] df[date].dt.year df[month] df[date].dt.month df[day] df[date].dt.day df[weekday] df[date].dt.weekday df[is_month_end] df[date].dt.is_month_end.astype(int) # 滑动窗口统计捕捉近期销量趋势 df[sales_lag_1] df[sales].shift(1) df[sales_lag_7] df[sales].shift(7) df[sales_mean_7] df[sales].rolling(7).mean() df[sales_std_7] df[sales].rolling(7).std() return df.dropna()这里滑动窗口的构造是整个项目的核心预处理步骤。lag_1和lag_7分别表示前一天和一周前的销量用于捕捉短期波动和周期性mean_7和std_7分别表示最近7天的均值和标准差用来描述近期销量水平。对于长尾商品来说这些统计量比原始序列本身更稳定——因为单日销量经常是0但7天均值能反映出一个相对平稳的趋势。需要注意参数的选择窗口大小7天对应的是周周期如果你的商品有明显的月度周期性可以考虑把窗口改成30天。滚动窗口计算后必须dropna因为前几行没有足够的历史数据来计算lag和rolling值这些样本不能进入训练集。2.3 训练集、验证集、测试集的划分顺序项目里对数据分裂的说明值得单独拿出来讲。很多人在做销量预测时习惯直接用train_test_split做随机切分这在长尾场景下是大忌。因为销量数据是时间序列随机切分会把未来的数据泄漏到训练集里导致验证结果虚高上线后直接翻车。# 按时间顺序切分而不是随机切分 def split_by_time(df, train_ratio0.7, val_ratio0.15): n len(df) train_end int(n * train_ratio) val_end int(n * (train_ratio val_ratio)) train_df df.iloc[:train_end] val_df df.iloc[train_end:val_end] test_df df.iloc[val_end:] return train_df, val_df, test_df切分参数的设定逻辑是训练集占70%用来拟合模型参数验证集占15%用来选超参数比如网络层数、学习率、earlystop的patience测试集占15%只在最终评估时用一次用来估计模型在未知数据上的泛化能力。验证集和测试集的本质区别在这里验证集可以反复使用测试集必须一次性使用。我在实际操作中还会额外做一步——按商品ID分组后再切分避免同一款商品的连续日期横跨训练集和验证集。虽然这样会让训练数据变少但能更真实地反映模型面对未见时间段数据时的表现。3. 训练主流程从placeholder到earlystop的完整链路3.1 为什么长尾销量场景选DNN而不是树模型XGBoost、LightGBM这类梯度提升树在表格数据上确实很强但它们在长尾销量预测上有一个先天劣势——无法自然处理序列的时间依赖结构。树模型的输入特征是独立的模型不会主动学习昨天的销量和今天的销量之间的时序关系除非你手动构造了大量lag特征。DNN的优势在于它可以通过embedding层处理稀疏的商品ID同时底层的全连接层可以自动组合高阶特征交互。项目选择TensorFlow 1.13的低阶API而不是Keras或者TF 2.x是因为低阶API能把整个训练过程暴露给你——每个placeholder怎么定义、每个张量怎么流动、梯度怎么更新全部可控。这种控制力在调试模型时太重要了。3.2 train.py核心逻辑拆解先看输入占位符的定义。项目用的是TensorFlow 1.13所以还保留着tf.placeholder这套经典的静态图写法。import tensorflow as tf # 定义输入占位符 feature_dim 13 # 特征维度根据实际特征工程结果调整 with tf.name_scope(input): x tf.placeholder(tf.float32, shape[None, feature_dim], namex) y tf.placeholder(tf.float32, shape[None, 1], namey) keep_prob tf.placeholder(tf.float32, namekeep_prob) # 构建三层全连接网络 with tf.name_scope(dnn): hidden1 tf.layers.dense(x, 128, activationtf.nn.relu, namehidden1) hidden1_drop tf.nn.dropout(hidden1, keep_prob) hidden2 tf.layers.dense(hidden1_drop, 64, activationtf.nn.relu, namehidden2) hidden2_drop tf.nn.dropout(hidden2, keep_prob) y_pred tf.layers.dense(hidden2_drop, 1, namey_pred) # 损失函数与优化器 with tf.name_scope(loss): loss tf.losses.mean_squared_error(labelsy, predictionsy_pred) with tf.name_scope(train): global_step tf.Variable(0, trainableFalse, nameglobal_step) optimizer tf.train.AdamOptimizer(learning_rate0.001) train_op optimizer.minimize(loss, global_stepglobal_step)这里有几个关键设计。第一placeholder的shape第一个维度设成None表示batch size不固定方便训练和预测时传入不同大小的数据。第二用tf.name_scope给每一层加命名空间这不仅是代码整洁的问题后面用import_meta_graph加载模型做预测时要靠这些名字找到对应的张量。第三网络的隐藏层设计是128→64属于从宽到窄的结构强制模型学一个压缩表示如果你发现欠拟合可以加大层宽如果过拟合可以加dropout的比例。loss函数用的是均方误差MSE对于销量预测这种连续值回归问题是合适的。优化器选了Adam默认学习率0.001这个组合在大多数场景下不需要怎么调就能收敛。学习率参数需要注意如果你发现loss在训练初期不降反升大概率是学习率太大了改到0.0001再试。3.3 earlystop的自定义实现这个项目的earlystop实现是纯手写的没有用tf.keras.callbacks.EarlyStopping。它的核心思想很简单——记录到目前为止验证集上的最优精度当连续多个epoch没有刷新这个最优值时就认为模型不再提升提前终止训练。def train(sess, train_op, loss, x, y, keep_prob, train_data, val_data, epochs100, batch_size64, patience10): best_val_loss float(inf) wait 0 saver tf.train.Saver() for epoch in range(epochs): # 训练一个epoch for batch_x, batch_y in batch_iter(train_data, batch_size): sess.run(train_op, feed_dict{x: batch_x, y: batch_y, keep_prob: 0.8}) # 在验证集上评估 val_loss sess.run(loss, feed_dict{x: val_data[0], y: val_data[1], keep_prob: 1.0}) print(fEpoch {epoch}, val_loss: {val_loss:.4f}) # earlystop判断 if val_loss best_val_loss: best_val_loss val_loss wait 0 saver.save(sess, ./model/best_model.ckpt, global_stepepoch) else: wait 1 if wait patience: print(fEarly stopping at epoch {epoch}, best val_loss: {best_val_loss:.4f}) break这个实现里最容易被忽略的是dropout在训练和评估时的切换。训练时keep_prob设成0.8意味着每个神经元有20%的概率被随机丢弃这是防止过拟合的常用手段但评估时必须设成1.0否则dropout的随机性会让验证集loss剧烈波动earlystop会误判模型性能。patience参数的设置直接影响训练时长和效果。设成10是经验值意思是连续10轮没有刷新最优验证集loss就停。如果训练集很大可以适当调到15或20如果训练集很小5到8就够了。另外saver.save时带上global_step参数会在checkpoint文件名里记录当前epoch方便后续回溯哪个epoch的模型最好。这里踩过一个坑如果不额外保存best_val_loss对应的模型路径earlystop触发后你需要回到历史checkpoint去加载最优模型而saver.save只保存最新状态所以要在val_loss刷新时立刻持久化否则就白等了十几个epoch。4. TensorBoard与模型持久化训练过程可视化与Saver用法4.1 tf.summary.FileWriter怎么用项目中提到为了服务TensorBoard的可视化特意用了tf.summary.FileWriter。这一步很多人会忽略觉得可视化是可有可无的事但在调DNN超参数的时候TensorBoard的loss曲线图比命令行里打印的一堆数字直观得多。# 在构图阶段添加summary操作 with tf.name_scope(summary): tf.summary.scalar(loss, loss) tf.summary.histogram(hidden1_weights, tf.get_default_graph().get_tensor_by_name(dnn/hidden1/kernel:0)) merged_summary tf.summary.merge_all() # 训练前初始化FileWriter train_writer tf.summary.FileWriter(./tmp/train, graphsess.graph) val_writer tf.summary.FileWriter(./tmp/val, graphsess.graph) # 训练循环内每隔一定step写入一次 if step % 50 0: train_summary sess.run(merged_summary, feed_dict{x: batch_x, y: batch_y, keep_prob: 0.8}) train_writer.add_summary(train_summary, step)启动TensorBoard的方式在README里有说明标准命令是tensorboard --logdir./tmp/默认端口是6006打开浏览器访问http://localhost:6006就能看到界面。这里有一个关键的工程习惯训练集和验证集的summary要写到不同的子目录即./tmp/train和./tmp/val。TensorBoard会同时读取这两个目录下的event文件在同一张loss曲线图上画出训练集和验证集的曲线这样你能直观看到过拟合发生的时刻——验证集loss开始回升而训练集loss继续下降的那个点。tf.summary.histogram记录的是权重分布用来观察网络各层的权重是否出现梯度消失或梯度爆炸。如果hidden1_weights的分布长时间集中在一个极窄的区间内说明激活函数饱和了可能需要换激活函数或者调整初始化方式。resource这个包自带的例子把summary这块做得比较完整你直接抄过去改改路径就能用。4.2 用tf.train.Saver做checkpoint持久化模型的保存和恢复是深度学习落地不能省的一环。训练时如果不保存模型一旦进程崩溃或者机器重启前几个小时的训练就白费了这种教训我是吃过亏的。# 定义Saver时指定要保存的变量默认保存全部可训练变量 saver tf.train.Saver(max_to_keep5) # 训练过程中的保存策略 if epoch % 10 0: saver.save(sess, ./model/checkpoint_epoch_{}.ckpt.format(epoch)) # 训练结束时保存最终模型 saver.save(sess, ./model/final_model.ckpt)max_to_keep5这个参数值得注意它控制Saver最多保留5个最新的checkpoint文件超过5个会自动删除最旧的。这样做既保证了历史版本可回溯又不会让checkpoint文件无限堆积占用磁盘空间。关于保存路径的命名也有讲究。把epoch信息拼在文件名里配合TensorBoard里的loss曲线你能定位到第几个epoch开始过拟合然后直接用那个epoch的checkpoint重新加载预测相当于拥有了一颗后悔药。后来我有一次训练模型连续跑了40个epoch验证集loss在第22轮触底如果没保存历史checkpoint最后拿到的模型反而是第40轮的过拟合版本效果差不少。checkpoint保存之后会生成几个文件.ckpt文件保存了权重和偏置的取值.meta文件保存了计算图的结构checkpoint文件保存了最近一次保存的记录索引。如果你要跨机器迁移模型这三个文件要一起拷贝缺一个都加载不了。5. 常见问题与避坑TensorFlow 1.13环境下最容易翻车的五个点5.1 TensorFlow版本兼容性问题现象用TF 2.x的环境直接跑train.py报错AttributeError: module tensorflow has no attribute placeholder。原因项目基于TF 1.13开发使用了tf.placeholder、tf.Session等旧版API。TF 2.0开始默认开启Eager Execution移除了这些符号。解决安装TF 1.13版本运行推荐用虚拟环境隔离pip install tensorflow1.13.1如果你必须用TF 2.x可以试试在文件开头加一行兼容代码import tensorflow.compat.v1 as tf tf.disable_v2_behavior()5.2 earlystop触发过早模型欠拟合现象训练才跑了十几轮就触发了earlystop验证集的loss看起来还在下降趋势中但程序已经停了。原因patience设置太小或者dropout在验证阶段没关掉导致验证loss抖动幅度过大。长尾数据本身噪声高验证loss天然地波动明显一轮loss偏高不代表模型真的变差了。解决首先确认验证时keep_prob设为1.0其次把patience从10提高到20最后也是最稳妥的做法——在刷新最优验证loss时保存checkpoint而不是停留在earlystop那个epoch这样即使停早了也能回到最优模型。5.3 import_meta_graph加载模型后找不到张量现象加载meta_graph后执行graph.get_operation_by_name(y_pred)报错KeyError或NameError。原因模型构建时没有给输出张量取名字或者名字拼写错误。TensorFlow默认的操作名是自动生成的比如dense_1/BiasAdd但这个名称会随着代码改动而变化不稳定。解决在构建模型时为关键张量显式命名。正确加载姿势如下with tf.Session() as sess: saver tf.train.import_meta_graph(./model/final_model.ckpt.meta) saver.restore(sess, ./model/final_model.ckpt) graph tf.get_default_graph() x graph.get_operation_by_name(input/x).outputs[0] keep_prob graph.get_operation_by_name(input/keep_prob).outputs[0] y_pred graph.get_operation_by_name(dnn/y_pred).outputs[0] pred sess.run(y_pred, feed_dict{x: test_x, keep_prob: 1.0})注意get_operation_by_name返回的是一个Operation对象要取它的outputs[0]才能作为张量传入feed_dict。这里最容易写错的地方是把操作名写成了张量名比如把dnn/hidden1/BiasAdd写成了dnn/hidden1/Relu导致加载后无法进行预测。5.4 验证集上表现好测试集上一塌糊涂现象验证集上的MSE很小但拿到测试集上一跑就崩了预测值明显偏离真实销量。原因最常见的原因是数据划分时没有按商品分组同一商品的数据同时出现在验证集和测试集中模型在验证集上记住了商品的特征测试集里的部分商品在训练集里已经出现过。解决按商品ID做分组切分保证测试集里的商品在训练时完全没见过。做法是在split_by_time之前先用groupby把商品ID排好然后在商品维度上做切分product_ids df[product_id].unique() train_ids product_ids[:int(len(product_ids) * 0.7)] val_ids product_ids[int(len(product_ids) * 0.7):int(len(product_ids) * 0.85)] test_ids product_ids[int(len(product_ids) * 0.85):]5.5 预测结果全是同一个值或者接近于0现象模型训练完成后对测试集所有样本的预测结果几乎相等看不到区分度。原因这是长尾数据最常见的模型退化问题。由于大多数商品销量是0或者个位数模型在MSE的驱动下学会了躺平——预测平均值或者预测0因为这样loss最小。这是优化目标的选择问题不是网络结构的问题。解决换一个目标函数来优化比如Huber Loss或者分位数损失它们对异常值和稀疏数据更鲁棒。也可以对标签做变换例如log1p把销量压缩到更平滑的区间后再训练预测完再指数还原。import numpy as np def huber_loss(labels, predictions, delta1.0): residual tf.abs(labels - predictions) condition tf.less_equal(residual, delta) small_res 0.5 * tf.square(residual) large_res delta * residual - 0.5 * tf.square(delta) return tf.where(condition, small_res, large_res)6. 离线加载模型做预测import_meta_graph的完整姿势项目的predict.py演示了用tf.train.import_meta_graph加graph.get_operation_by_name进行离线预测的方法这在生产环境中非常实用——训练和预测解耦预测服务加载一个训练好的checkpoint就可以对外提供推理接口。import tensorflow as tf import pandas as pd import numpy as np def predict(checkpoint_path, feature_data): with tf.Session() as sess: # 加载计算图结构 saver tf.train.import_meta_graph(checkpoint_path .meta) # 恢复权重 saver.restore(sess, checkpoint_path) graph tf.get_default_graph() x graph.get_operation_by_name(input/x).outputs[0] keep_prob graph.get_operation_by_name(input/keep_prob).outputs[0] y_pred graph.get_operation_by_name(dnn/y_pred).outputs[0] # 预测时keep_prob固定为1.0 feed_dict {x: feature_data, keep_prob: 1.0} predict_result sess.run(y_pred, feed_dictfeed_dict) return predict_result # 读取待预测商品的特征 raw_data pd.read_csv(./data/test_features.csv) features raw_data[[year, month, day, weekday, sales_lag_1, sales_lag_7, sales_mean_7, sales_std_7]].values predictions predict(./model/final_model.ckpt, features) print(7天预测销量:, predictions[:, 0])这段代码有几个细节需要注意。第一checkpoint_path传的是不带后缀的路径比如./model/final_model.ckpt代码内部会自动拼上.meta和.data后缀。第二feed_dict里必须包含所有定义过的placeholder漏掉一个就会报错。第三keep_prob在预测时设成1.0这个和第3章里验证时保持1.0的原因一致。拿到预测值之后还要做一步反变换。如果你在训练时对标签做了log1p变换那么预测值要经过np.expm1还原成真实销量real_predictions np.expm1(predictions)最后可以把预测结果写回数据库或者生成备货建议文件result_df pd.DataFrame({ product_id: raw_data[product_id], predict_sales_7d: real_predictions[:, 0], predict_sales_30d: real_predictions[:, 1] if real_predictions.shape[1] 1 else None, }) result_df.to_csv(./output/predict_result.csv, indexFalse)我个人的习惯是每次上线前都会把模型加载预测的全流程强制走一遍包括meta加载、张量查找、反变换、结果输出确认四个环节全部正常再交付给业务方。尤其是graph.get_operation_by_name这一步模型哪怕只是加了一层网络导致节点名变化老脚本就会直接报错。从那以后我每次改完模型结构都会顺手更新predict.py里的张量名映射这个习惯帮我避免了好几次生产事故。希望帮到你。本文还有配套的精品资源点击获取
