简介面向股票量化入门与深度学习实践者的TensorFlow预测资源围绕CNN与DQN两种模型讲解如何从历史行情中提取特征并预测未来走势。资源共34个文件以Python脚本、Jupyter Notebook为主要代码载体包含训练数据、模型文件、结果走势图及两份PDF讲解文档覆盖数据清洗、滑动窗口构造、模型搭建、训练评估与结果可视化的完整链路。压缩包约5.16MB已在CSDN获1440人学习适合希望通过完整代码复现CNN股票预测流程并进一步了解DQN强化学习方案的读者。项目基于真实股票数据展开实践代码与图表对应清晰可帮助理解从特征工程到模型落地的全过程是一份兼顾理论讲解与实战运行的参考资料。1. 用 TensorFlow 跑 CNN 预测股票先别急着谈赚钱把特征工程想明白做股票预测的 Python 项目很多但真正把 CNN 完整跑通、还附带 DQN 对比和 KD 指标分析的资源并不多。这份项目压缩包里既有 CNN 分类器、DQN 强化学习版本还带了整套 PDF 大纲和年份线回测文件属于典型的能跑起来、能看清楚每一步在干什么的实操型项目。它解决的核心问题不是让你一夜暴富而是让你理解卷积神经网络处理金融时间序列的完整链路数据怎么清洗、窗口怎么滑、二维输入怎么构造、训练完怎么验证。适合已经会用 Python 和 pandas 处理数据的入门者你不需要懂复杂的金融理论但得有点耐心跟着把代码拆开看。说实话用 CNN 预测股票这件事本身就有点玄学但这份资源的价值恰恰在于把玄学拆成了可复现的工程步骤——你照着跑一遍就知道哪些环节容易翻车哪些参数需要自己调。2. 数据预处理把 K 线时间序列改造成 CNN 吃得下的二维样本2.1 原始数据长什么样从 pandas DataFrame 到滑动窗口拿到股票数据后第一步不是直接丢给模型而是要想清楚 CNN 需要什么样的输入。卷积神经网络擅长处理的是有空间结构的数据比如图像是 H×W×C 的三维数组。股票价格是一维时间序列直接塞进去卷积核不知道该怎么滑。常见的做法是用滑动窗口技术每个样本不再是一个时间点的数据而是连续 N 天的历史数据堆叠成一个二维矩阵。比如你用过去 5 天的开高低收和成交量来预测第 6 天是涨是跌那每个样本就是一个 5×5 的矩阵5 是时间步长5 是特征维度。项目里的数据处理脚本基本就是这个思路。我拆开看了看核心逻辑是先把 CSV 或从接口拉到的日线数据读进 DataFrame然后按日期排序、去重、填充缺失值。这里有个细节很多人会忽略股票数据里经常有停牌日日期是不连续的如果直接按行号滑窗会把停牌前后的数据硬凑在一起导致样本失真。稳妥的做法是先建立一个完整的交易日历再用 reindex 对齐。import pandas as pd import numpy as np def load_stock_data(csv_path): # 读取原始数据把日期列解析成时间类型 df pd.read_csv(csv_path, parse_dates[date]) df df.sort_values(date).drop_duplicates(subsetdate) # 建立完整交易日序列停牌日对应的行情字段用前值填充 full_calendar pd.date_range(startdf[date].min(), enddf[date].max(), freqB) df df.set_index(date).reindex(full_calendar).ffill().dropna() # 只保留建模需要的列 features [open, high, low, close, volume] df df[features] return df def create_window_samples(df, window_size5): # 把时间序列切成 (样本数, 窗口大小, 特征数) 的三维数组 data df.values # shape: (总天数, 特征数) X, y [], [] for i in range(len(data) - window_size): X.append(data[i:i window_size]) # 预测标签第6天收盘价是否高于第5天 y.append(1 if data[i window_size, 3] data[i window_size - 1, 3] else 0) return np.array(X), np.array(y)这里窗口滑动的逻辑要拆开讲data[i:i window_size]取的是从第 i 天开始的连续 5 天作为第 i 个样本标签 y 是第 5 天的收盘价和第 4 天收盘价的比较结果涨了标 1跌了标 0。如果你要预测的是连续多天的趋势而不是单日涨跌可以把标签改成未来 3 天或 5 天的累计收益率符号。参数window_size的取值很关键窗口太小模型看不到趋势窗口太大又引入了大量噪声我一般会先用 5 天、10 天、20 天都试试看验证集上的表现再定。2.2 标准化与数据划分别让价格量纲带偏了卷积核股票价格动辄几十上百成交量可能是几百万这些特征的数值范围差了好几个数量级。如果不做标准化卷积核学到的权重会被大数值的特征主导小数值的特征相当于白给。常见做法是 Z-score 标准化对每个特征分别计算均值和标准差然后做(x - mean) / std。注意一个坑均值和标准差必须只用训练集的统计量计算验证集和测试集用同一组参数变换不能在整个数据集上算完再切分否则训练时模型就已经偷看了未来的分布信息这属于典型的数据泄漏。from sklearn.preprocessing import StandardScaler def preprocess_data(X, y, train_ratio0.8): total len(X) split_idx int(total * train_ratio) X_train, X_test X[:split_idx], X[split_idx:] y_train, y_test y[:split_idx], y[split_idx:] # 每个特征独立标准化scaler 只在训练集上 fit scaler StandardScaler() n_samples, n_window, n_features X_train.shape X_train_reshaped X_train.reshape(-1, n_features) X_train_scaled scaler.fit_transform(X_train_reshaped) # 测试集用训练集的均值和方差 X_test_reshaped X_test.reshape(-1, n_features) X_test_scaled scaler.transform(X_test_reshaped) return (X_train_scaled.reshape(n_samples, n_window, n_features), X_test_scaled.reshape(-1, n_window, n_features), y_train, y_test, scaler)reshape 的过程容易把人绕晕我拆开解释下原始 X 是三维数组(样本数, 窗口天数, 特征数)StandardScaler 只接受二维输入所以先压成(样本数 × 窗口天数, 特征数)的二维矩阵每个特征列单独算均值和方差变换完再恢复成原始三维形状。用reshape(-1, n_features)时-1 表示让 NumPy 自动推断这个维度的大小前提是你知道总元素数不变。做完标准化后模型在训练时看到的所有特征都处在同一量级卷积核的初始化权重才能公平地作用于每个输入通道。2.3 标签构造与类别不平衡预测涨跌而不是预测价格项目里有一个很关键的设计选择预测目标是涨跌分类而不是精确的价格数值。这点值得多说两句。如果直接回归预测收盘价模型输出的误差会被价格本身的波动幅度放大而且价格序列是非平稳的今天的 100 元和明天的 101 元之间没有可比性。改成二分类任务后标签变成明天是否比今天高模型只需要学习相对变化的方向训练目标更清晰评估指标也更直观。但分类任务有个几乎必踩的坑股票市场上涨跌天数往往不平衡尤其是震荡市里可能连续多天收阳或收阴导致训练集里某一类的样本占比超过 60%模型学出来就是个永远预测多数类的憨憨准确率看着挺高实际毫无用处。项目里虽然没有明说怎么处理但我自己一般会在构造完标签后先打印一下np.bincount(y)看看分布如果发现不平衡就用 class_weight 参数给少数类加权。# 检查类别分布 print(标签分布:, np.bincount(y_train)) # 在 model.fit() 里给少数类更高的权重 class_weights {0: 1.0, 1: 1.2} # 假设置类 1 是少数类权重调高 model.fit(X_train, y_train, validation_data(X_test, y_test), epochs20, batch_size32, class_weightclass_weights)class_weight 的原理是在计算损失函数时对少数类的每个样本乘以更大的权重系数相当于变相提高了少数类的话语权。数值怎么设先看分布比例如果正负样本比是 4:6那权重就按反比设成 1.5:1 左右然后再微调。这一步不做的话你后面所有的训练指标都会虚高等拉到实盘验证才发现根本不灵。3. CNN 模型架构卷积层、池化层与全连接层的取舍3.1 用 Sequential API 搭一个一维 CNN项目里直接用tf.keras.Sequential搭建模型这个 API 对新手极其友好。针对时间序列数据我建议用 Conv1D 而不是 Conv2D——虽然项目里也有转成二维图像的思路但一维卷积天然匹配时间序列的形态卷积核只在一个方向时间维度上滑动每个卷积核负责捕捉一段时间内的局部趋势模式比如连续三天放量上涨或者高开低走这类形态特征。import tensorflow as tf from tensorflow.keras import layers def build_cnn_model(input_shape, num_classes2): model tf.keras.Sequential([ layers.Input(shapeinput_shape), # 第一层卷积64 个卷积核核大小 3捕捉 3 天内的走势模式 layers.Conv1D(filters64, kernel_size3, activationrelu, paddingsame), layers.MaxPooling1D(pool_size2), layers.Dropout(0.2), # 第二层卷积128 个卷积核核大小 3提取更高层特征 layers.Conv1D(filters128, kernel_size3, activationrelu, paddingsame), layers.MaxPooling1D(pool_size2), layers.Dropout(0.2), # 展平后接全连接层输出二分类概率 layers.Flatten(), layers.Dense(64, activationrelu), layers.Dropout(0.3), layers.Dense(num_classes, activationsoftmax) ]) return model model build_cnn_model(input_shape(5, 5)) model.summary()逐层说下设计理由。第一层 Conv1D 的kernel_size3意味着每个卷积核看连续 3 天的数据64 个卷积核各学各的模式有的可能学到三连阳有的学到放量长上影。paddingsame的作用是让输出序列长度和输入保持一致这样在堆叠多层卷积时不会因为边界裁剪导致信息逐层流失。MaxPooling1D 的pool_size2把序列长度减半一方面降低计算量另一方面强迫模型关注更宏观的模式——滑动窗口是 5 天的话池化后变成 2 或 3 个时间点模型只能从压缩后的信息里做判断。Dropout 是防过拟合的训练时随机丢弃一部分神经元的输出让网络不依赖某几个特定的特征组合。这里有个参数设计值得说窗口长度 5 天配合两层池化后时序维度被压缩到 1 到 2如果你用更长的窗口比如 20 天池化层可能要多加一层或者调整 stride否则全连接层收到的特征图尺寸会偏大、参数量暴增。我一般在调结构时会先打印model.summary()看每一层输出的 shape确认展平后 Flatten 的维度是合理的而不是拍脑袋堆层数。3.2 卷积核数量与池化策略模型复杂度要和数据量匹配很多初学者一上来就用大模型卷积核直接 256、512 起步。但股票数据的样本量通常有限哪怕有十年的日线数据滑窗后也就两千多个样本这种情况下大模型几乎必然过拟合——训练集准确率 98%验证集 55%完全没法用。项目里的模型规模比较克制第一层 64、第二层 128这个量级在几千个样本的场景下是比较合理的起点。池化层除了 MaxPooling1D 之外还可以考虑 GlobalAveragePooling1D 作为替代。区别在于MaxPooling 取窗口内最大值保留的是最显著的特征GlobalAveragePooling 把整个序列压缩成一个平均值会丢失局部峰值信息但对抑制过拟合有帮助。如果你的验证集准确率一直上不去把最后卷积层后面的 MaxPooling 换成 GlobalAveragePooling 再试试有时候会带来意想不到的泛化提升。这个 trade-off 没有标准答案只能多跑几组实验对比。# 用 GlobalAveragePooling1D 替代 Flatten Dense 的简单组合 model tf.keras.Sequential([ layers.Input(shapeinput_shape), layers.Conv1D(64, 3, activationrelu, paddingsame), layers.MaxPooling1D(2), layers.Conv1D(128, 3, activationrelu, paddingsame), layers.GlobalAveragePooling1D(), # 直接输出 128 维向量 layers.Dense(2, activationsoftmax) ])这个结构的优点在于GlobalAveragePooling1D 把每个特征通道的时间维度全部平均掉输出维度等于最后一个卷积层的过滤器数量不需要手动计算 Flatten 后的尺寸模型参数也少了几个量级。缺点是它假设每个特征在整个时间窗口内的重要性是均匀的如果某些模式只出现在窗口的中段平均操作会削弱它的贡献。在股票场景里K 线形态往往越靠近当前日越重要所以我把 MaxPooling 作为默认方案GlobalAveragePooling 作为备选调参手段。3.3 损失函数、优化器与评估指标交叉熵配 Adam别用准确率一锤定音项目里用model.compile(losssparse_categorical_crossentropy, optimizeradam, metrics[accuracy])这段配置。展开说下每个参数的含义标签 y 是整数 0 或 1 时用sparse_categorical_crossentropy如果标签是 one-hot 编码的[0, 1]或[1, 0]就用categorical_crossentropy用错类型模型还能跑但 loss 和梯度计算会不正常训练曲线会异常震荡。优化器选 Adam 是因为它对学习率不那么敏感自适应调整每一步的更新幅度省去了手动调学习率衰减的麻烦。评估指标 accuracy 只能作为参考因为前面提到的类别不平衡问题准确率高不代表模型真的有区分能力。model.compile( losssparse_categorical_crossentropy, optimizertf.keras.optimizers.Adam(learning_rate0.001), metrics[accuracy] )learning_rate0.001是 Adam 的典型默认值。如果你发现训练 loss 下降很慢或者验证集 loss 震荡不稳可以考虑把学习率降到 0.0003 或 0.0001反过来如果前几个 epoch loss 下降极其陡峭、几乎瞬间过拟合说明学习率偏大。这里我习惯用一个粗调口诀先用默认 0.001 跑 10 个 epoch 观察趋势如果 loss 曲线忽上忽下就把学习率减半如果收敛太慢就翻倍每次只动一个参数。4. 训练、评测与预测从损失曲线到连续外推4.1 model.fit 的配置细节验证集切分和早停策略训练时的validation_split参数会从训练数据尾部自动切出一部分作为验证集但要注意默认是按顺序从尾部切的而时间序列数据一旦打乱验证集就可能包含训练集之后的数据这没问题但如果你用shuffleTrue把训练数据打乱了再切验证集验证集里就会混入训练集之前的数据在时间序列场景下这是严重的逻辑错误。我的做法是自己手动按时间顺序切好训练集和验证集再用validation_data参数传入这样能完全控制时间边界。history model.fit( X_train, y_train, validation_data(X_val, y_val), # X_val 必须是时序上晚于 X_train 的数据 epochs50, batch_size32, callbacks[ tf.keras.callbacks.EarlyStopping( monitorval_loss, patience5, restore_best_weightsTrue ) ] )EarlyStopping 用monitorval_loss监控验证集损失patience5表示连续 5 个 epoch 验证集 loss 没有下降就提前终止训练restore_best_weightsTrue会在停止时把模型权重回滚到验证集表现最好的那次。这个回调几乎是必须的因为金融时序数据噪声很大训练集 loss 会一路下降但验证集 loss 可能在第 10 个 epoch 就开始回升再往下练就是纯纯的过拟合。训练完记得看一眼history.history里的 loss 和 val_loss 曲线两条曲线从一开始就分道扬镳说明模型结构有问题先检查前面预处理环节。4.2 model.predict 与连续预测单步预测结果怎么拼成长序列model.predict(X_test)返回的是每个样本属于类别 0 和类别 1 的概率你需要用np.argmax取出预测类别。项目里的实际预测脚本做得更讲究不是只预测一步而是把未来多天的预测结果动态拼接到输入窗口里实现连续外推。这里有个容易出错的地方——拼接的应该是模型输出的概率还是模型预测的类别我建议用概率值因为模型预测的类别是离散的 0 或 1把预测值当真实值拼回去会不断累积误差而概率值保留了模型的不确定性信息拼接后模型仍然能基于连续的概率分布做判断。def predict_future(model, last_window, steps5): current_window last_window.copy() predictions [] for _ in range(steps): prob model.predict(current_window.reshape(1, *current_window.shape), verbose0) pred_class np.argmax(prob, axis1)[0] # 关键用预测概率构造未来一天的特征而不是直接用类别 pred_prob prob[0][1] # 上涨概率 # 假设特征里第 3 列是 close price用当前收盘价和上涨概率构造新值 next_close current_window[-1, 3] * (1 (pred_prob - 0.5) * 0.02) new_row current_window[-1].copy() new_row[3] next_close current_window np.vstack([current_window[1:], new_row]) predictions.append(pred_class) return predictions这段代码的核心逻辑是滑动窗口的滚动更新每次预测完一天把窗口最老的一行丢掉把新构造的预测日特征追加到末尾然后继续预测下一天。这里new_row[3] next_close是一种近似构造方式真实场景中应该结合技术指标一起估计新一天的开高低收否则只改 close 可能会让卷积核看到的数据分布失真。连续预测的误差会随步长累积步数越多越不可靠一般预测未来 5 天还能看个趋势预测 20 天以上基本就是随机游走。4.3 项目里的 DQN 对比为什么强化学习版本也值得跑一遍压缩包里除了 CNN 分类器还有 DQN 相关的多个文件比如DQN_ori.ipynb和DQN_MLP_closePrice。这套代码是用强化学习策略梯度来训练模型通过不断试错学习买卖动作。它的意义在于和 CNN 形成对照CNN 学的是给定历史行情预测未来涨跌的概率本质是有监督学习DQN 学的是在某个市场状态下应该买入、卖出还是持有本质是序列决策。两个思路互补你可以把 CNN 的预测结果当作 DQN 的状态特征之一构成一个更完整的交易决策系统。# DQN 中经验回放的核心数据结构 # state: 当前市场状态通常是一个窗口的行情特征 # action: 0持有, 1买入, 2卖出 # reward: 动作执行后的收益变化 # next_state: 动作执行后的新状态 replay_buffer [] replay_buffer.append((state, action, reward, next_state, done)) # batch 采样时从 buffer 中随机取一批打破时间相关性 batch random.sample(replay_buffer, batch_size)经验回放是 DQN 稳定训练的关键随机采样打破了相邻样本之间的时间相关性避免了模型在连续的相似状态间震荡。如果你第一次跑 DQN建议把learning_rate调低一些强化学习的训练信号比监督学习稀疏得多学习率太大会导致策略剧烈波动像喝醉了酒的人走路一样左右摇摆。5. 股票预测避坑指南TensorFlow 项目里最容易翻车的五个细节5.1 数据泄漏标准化时偷看了未来回测一片美好现象训练集准确率很高验证集也不错但一到真实市场就明显失效甚至不如昨天涨今天就买的朴素策略。原因在滑窗之前对整个数据集做了标准化或者用了包含未来信息的统计量。StandardScaler 的均值和方差是全体数据的全局统计模型训练时已经知道未来所有天的价格分布范围。解决严格按时间顺序切分训练集和测试集scaler 只 fit 训练集transform 测试集时用训练集的参数。这个顺序错位是股票预测项目里最常见的隐性 bug。5.2 标签漂移用未来数据生成当日标签现象训练指标惊人的高验证集准确率超过 90%但查看预测结果发现都是昨天的影子。原因构造标签时用了shift(-1)导致标签和特征之间出现了时间上的重叠模型实质在预测已知的当下而不是未来。解决每次构造样本时特征窗口的最后一天必须严格早于标签对应的交易日。建议在切完样本后打印几个样本的时间和标签对人工检查是否存在同一天的数据既出现在特征里又出现在标签里。5.3 模型过拟合CNN 参数总量比训练样本还多现象训练集 loss 一路降到趋近于 0验证集 loss 却在第 5 个 epoch 后持续上升两者差距越来越大。原因模型层数太深、卷积核数量太多几千个样本撑不起几十万个参数模型把训练集的噪声全背下来了。解决先用小模型跑通流程卷积核从 32、64 开始层数不超过三层加 Dropout0.2 到 0.5 之间用 EarlyStopping 在 val_loss 不再下降时立刻停。若仍旧过拟合把池化换成 GlobalAveragePooling1D。5.4 TensorFlow 版本差异导致的 API 报错现象跑tf.keras.Sequential时提示module tensorflow has no attribute keras或者Conv1D报TypeError。原因TensorFlow 2.x 早期版本和 2.6 的 API 组织方式有差异有些旧代码用的是tf.keras.layers.Conv1D新版改成tf.keras.layers更严格还有人安装的是 CPU 版跑卷积时速度极慢。解决建议使用 TensorFlow 2.10 或 2.15 稳定版导入方式统一用from tensorflow import keras。安装时如果只能选 CPU 版就在代码里加上限制线程数防止一次性拉满 CPU 导致系统卡死。5.5 评价指标被准确率带偏现象模型预测永不买入的策略也能达到 70% 以上准确率因为市场本来就有涨跌周期多数类占比天然超过一半。原因准确率在类别不平衡下不是好的评估指标尤其当你的信号类别只占 20% 时一个什么都不预测的模型反而很准。解决打印混淆矩阵关注精确率、召回率和 F1-score。对收益类任务还可以直接计算预测信号和实际涨跌的相关性或者做一个最简单的模拟回测看按模型给出的信号交易能否跑赢买入持有。6. 用年份线做跨周期验证从单次预测升级到可持续评估项目里有个文件叫yearline_rst.PNG和Test model by yearline.ipynb这套年份线验证思路值得多说几句。很多人在验证模型时只拿一段历史数据切出训练集和测试集测试集往往就几百个样本跑完发现准确率还行就以为大功告成。问题在于股票市场存在明显的风格切换——2018 年的震荡市和 2020 年的单边牛市里同样的技术形态可能对应完全相反的结果。如果模型只在某一年份的数据上验证过第二年换个行情就水土不服。年份线验证的做法是把历史数据按年度逐段拆分用每一年的数据作为验证集其余年份作为训练集循环往复地训练和评测。这样你能得到一组跨年份的准确率分布比如 2019 年准确率 53%、2020 年 61%、2021 年 48%一眼看到模型在不同市场环境下的稳定性。如果某一年模型表现断崖式下跌说明模型学到的主要是该年的特定模式而不是通用的市场规律。那时候要做的不是继续加参数而是回到特征工程好好想想。import itertools years sorted(df[date].dt.year.unique()) results {} for test_year in years: train_df df[df[date].dt.year ! test_year] test_df df[df[date].dt.year test_year] X_train, y_train create_window_samples(train_df, window_size5) X_test, y_test create_window_samples(test_df, window_size5) # 标准化时只用训练年份的统计量 scaler StandardScaler() X_train scaler.fit_transform(X_train.reshape(-1, X_train.shape[-1])).reshape(X_train.shape) X_test scaler.transform(X_test.reshape(-1, X_test.shape[-1])).reshape(X_test.shape) model build_cnn_model(input_shape(5, 5)) model.compile(optimizeradam, losssparse_categorical_crossentropy, metrics[accuracy]) model.fit(X_train, y_train, epochs20, batch_size32, verbose0) _, acc model.evaluate(X_test, y_test, verbose0) results[test_year] acc print(f{test_year} 年验证准确率: {acc:.2%})这段代码跑完后把每年准确率打印在一张表里比只看一个总准确率诚实得多。我自己的习惯是把年份验证结果画成折线图对照当年的市场指数走势一起看——如果模型在下跌年份里准确率反而高可能是因为它捕捉了过度的下跌惯性这类模型在市场反转时会让你亏得很惨。那以后我每次做股票预测项目都强制自己先跑一遍年份线验证除非某一年份的表现异常否则不进下一步调参。模型的鲁棒性不是看训练集多漂亮而是看它在没见过的年份里能不能活下去。希望这份拆解能让你少踩几个坑把 CNN 预测股票这件事做得比大多数博客教程更扎实一点。本文还有配套的精品资源点击获取
