简介：面向股票量化入门与深度学习实践者的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(subset='date') # 建立完整交易日序列，停牌日对应的行情字段用前值填充 full_calendar = pd.date_range(start=df['date'].min(), end=df['date'].max(), freq='B') 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_size=5): # 把时间序列切成 (样本数, 窗口大小, 特征数) 的三维数组 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_ratio=0.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), epochs=20, batch_size=32, class_weight=class_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_classes=2): model = tf.keras.Sequential([ layers.Input(shape=input_shape), # 第一层卷积：64 个卷积核，核大小 3，捕捉 3 天内的走势模式 layers.Conv1D(filters=64, kernel_size=3, activation='relu', padding='same'), layers.MaxPooling1D(pool_size=2), layers.Dropout(0.2), # 第二层卷积：128 个卷积核，核大小 3，提取更高层特征 layers.Conv1D(filters=128, kernel_size=3, activation='relu', padding='same'), layers.MaxPooling1D(pool_size=2), layers.Dropout(0.2), # 展平后接全连接层，输出二分类概率 layers.Flatten(), layers.Dense(64, activation='relu'), layers.Dropout(0.3), layers.Dense(num_classes, activation='softmax') ]) return model model = build_cnn_model(input_shape=(5, 5)) model.summary()逐层说下设计理由。第一层 Conv1D 的kernel_size=3意味着每个卷积核看连续 3 天的数据，64 个卷积核各学各的模式：有的可能学到"三连阳"，有的学到"放量长上影"。padding='same'的作用是让输出序列长度和输入保持一致，这样在堆叠多层卷积时不会因为边界裁剪导致信息逐层流失。MaxPooling1D 的pool_size=2把序列长度减半，一方面降低计算量，另一方面强迫模型关注更宏观的模式——滑动窗口是 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(shape=input_shape), layers.Conv1D(64, 3, activation='relu', padding='same'), layers.MaxPooling1D(2), layers.Conv1D(128, 3, activation='relu', padding='same'), layers.GlobalAveragePooling1D(), # 直接输出 128 维向量 layers.Dense(2, activation='softmax') ])这个结构的优点在于，GlobalAveragePooling1D 把每个特征通道的时间维度全部平均掉，输出维度等于最后一个卷积层的过滤器数量，不需要手动计算 Flatten 后的尺寸，模型参数也少了几个量级。缺点是它假设每个特征在整个时间窗口内的重要性是均匀的，如果某些模式只出现在窗口的中段，平均操作会削弱它的贡献。在股票场景里，K 线形态往往越靠近当前日越重要，所以我把 MaxPooling 作为默认方案，GlobalAveragePooling 作为备选调参手段。
3.3 损失函数、优化器与评估指标：交叉熵配 Adam，别用准确率一锤定音
项目里用model.compile(loss='sparse_categorical_crossentropy', optimizer='adam', metrics=['accuracy'])这段配置。展开说下每个参数的含义：标签 y 是整数 0 或 1 时用sparse_categorical_crossentropy，如果标签是 one-hot 编码的[0, 1]或[1, 0]就用categorical_crossentropy，用错类型模型还能跑，但 loss 和梯度计算会不正常，训练曲线会异常震荡。优化器选 Adam 是因为它对学习率不那么敏感，自适应调整每一步的更新幅度，省去了手动调学习率衰减的麻烦。评估指标 accuracy 只能作为参考，因为前面提到的类别不平衡问题，准确率高不代表模型真的有区分能力。
model.compile( loss='sparse_categorical_crossentropy', optimizer=tf.keras.optimizers.Adam(learning_rate=0.001), metrics=['accuracy'] )learning_rate=0.001是 Adam 的典型默认值。如果你发现训练 loss 下降很慢或者验证集 loss 震荡不稳，可以考虑把学习率降到 0.0003 或 0.0001；反过来如果前几个 epoch loss 下降极其陡峭、几乎瞬间过拟合，说明学习率偏大。这里我习惯用一个粗调口诀：先用默认 0.001 跑 10 个 epoch 观察趋势，如果 loss 曲线忽上忽下就把学习率减半，如果收敛太慢就翻倍，每次只动一个参数。
4. 训练、评测与预测：从损失曲线到连续外推
4.1 model.fit 的配置细节：验证集切分和早停策略
训练时的validation_split参数会从训练数据尾部自动切出一部分作为验证集，但要注意：默认是按顺序从尾部切的，而时间序列数据一旦打乱，验证集就可能包含训练集之后的数据，这没问题；但如果你用shuffle=True把训练数据打乱了再切验证集，验证集里就会混入训练集之前的数据，在时间序列场景下这是严重的逻辑错误。我的做法是自己手动按时间顺序切好训练集和验证集，再用validation_data参数传入，这样能完全控制时间边界。
history = model.fit( X_train, y_train, validation_data=(X_val, y_val), # X_val 必须是时序上晚于 X_train 的数据 epochs=50, batch_size=32, callbacks=[ tf.keras.callbacks.EarlyStopping( monitor='val_loss', patience=5, restore_best_weights=True ) ] )EarlyStopping 用monitor='val_loss'监控验证集损失，patience=5表示连续 5 个 epoch 验证集 loss 没有下降就提前终止训练，restore_best_weights=True会在停止时把模型权重回滚到验证集表现最好的那次。这个回调几乎是必须的，因为金融时序数据噪声很大，训练集 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, steps=5): current_window = last_window.copy() predictions = [] for _ in range(steps): prob = model.predict(current_window.reshape(1, *current_window.shape), verbose=0) pred_class = np.argmax(prob, axis=1)[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 开始，层数不超过三层；加 Dropout（0.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_size=5) X_test, y_test = create_window_samples(test_df, window_size=5) # 标准化时只用训练年份的统计量 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(optimizer='adam', loss='sparse_categorical_crossentropy', metrics=['accuracy']) model.fit(X_train, y_train, epochs=20, batch_size=32, verbose=0) _, acc = model.evaluate(X_test, y_test, verbose=0) results[test_year] = acc print(f"{test_year} 年验证准确率: {acc:.2%}")这段代码跑完后，把每年准确率打印在一张表里，比只看一个总准确率诚实得多。我自己的习惯是把年份验证结果画成折线图，对照当年的市场指数走势一起看——如果模型在下跌年份里准确率反而高，可能是因为它捕捉了过度的下跌惯性，这类模型在市场反转时会让你亏得很惨。
那以后我每次做股票预测项目，都强制自己先跑一遍年份线验证，除非某一年份的表现异常，否则不进下一步调参。模型的鲁棒性不是看训练集多漂亮，而是看它在没见过的年份里能不能活下去。希望这份拆解能让你少踩几个坑，把 CNN 预测股票这件事做得比大多数博客教程更扎实一点。
本文还有配套的精品资源，点击获取
