数据科学和机器学习（第 30 部分）：预测股票市场的幂对、卷积神经网络（CNN）、和递归神经网络（RNN）（基础篇）
「用 CNN 与 RNN 给行情做特征与时态建模」
在 MT5 上做行情预测，单纯用全连接网络容易丢掉价格序列里的空间结构和时间依赖。CNN 擅长从 K 线局部窗口里抠出形态特征，RNN 则能把前后根 K 线的顺序关系记住，两者拼起来比单用一种网络更贴合金融序列。 具体协同上，常见做法是先用 CNN 对输入的价格矩阵做卷积，压出局部形态特征图；再把这张图按时间步喂给 RNN（如 LSTM 或 GRU）做时态建模。这样模型既看得到「形状」，也读得懂「顺序」。 以一篇 2025 年 6 月 24 日发布、阅读量 875 的实战文为例，作者用 CNN+LSTM 与 CNN+GRU 两套组合做训练与预测，验证了在股票数据上该结构具备可跑通的工程路径。外汇与贵金属波动更剧烈、杠杆风险更高，直接套用前建议在 MT5 用历史数据重跑回测。
◍ 为什么把 CNN 和 RNN 拼到一起
前面几篇里，卷积神经网络（CNN）和递归神经网络（RNN）各自都能从行情序列里抠出可用信号，帮交易者对抗噪音。CNN 擅长抓局部形态，RNN 能记住时间上的依赖，两者单独用都已验证过对股票类标的的预测增益。 本篇要做的，是把这两种结构直接叠起来，看组合网络对股票市场方向的预测会不会比单用一种更稳。外汇与贵金属波动更剧烈，这类模型外推风险高，实盘前务必在 MT5 历史数据上回测。 在动手改网络之前，先快速过一遍 CNN 和 RNN 各自管什么、怎么接数据，不然后面组合层的张量维度会直接卡死。
CNN 抓形态，RNN 管时序
CNN 最早为图像识别而生，但落到表格型时间序列上一样能打：它先用过滤器扫输入，再抽出高级特征。在股票或外汇的行情数据里，这些特征往往对应趋势段、季节性波动和异常刺尖。 RNN 则专门吃顺序数据，和传统前馈网络假设“每根 K 线独立”不同，它带着对前面输入的记忆往下走。原版 RNN、LSTM、GRU 是三类常用变体，其中 LSTM 和 GRU 靠门控机制缓解长期依赖丢失，对理解时态关系更稳。 把两者拼起来思路很直接：CNN 负责从原始报价里抠特征，RNN/LSTM 再顺着时间轴解释这些特征。构建通常分三步——CNN 做特征提取、RNN 做时序建模、最后训练拿预测。外汇和贵金属杠杆高、跳空频繁，这类模型只提高概率优势，实盘前务必在 MT5 用历史数据回测验证。
「用一维卷积啃下K线形态」
把 OHLC 时间序列喂给一维卷积网络，模型自己会在滑动窗口里抓形态、抽特征，不用人工先算指标。以 Tesla 的日线四价为例，先转成 CNN 和 RNN 都能吃的 3D 格式（样本数 × 时间步 × 特征维度），这一步直接决定后面特征提取得干不干净。 分类目标怎么打标签？用收盘价是否高于开盘价来定：高开收阳给 1（偏多信号），否则给 0（偏空信号）。下面这段就是标签生成和归一化的核心，StandardScaler 把特征缩到零均值单位方差，能让梯度更新更稳。 target_var = [] open_price = new_df["Open"] close_price = new_df["Close"] for i in range(len(open_price)): if close_price[i] > open_price[i]: # 收价大于开价 target_var.append(1) # 偏多标签 else: target_var.append(0) # 偏空标签 X = new_df.iloc[:, :-1] y = target_var # 数据归一化 scaler = StandardScaler() X = scaler.fit_transform(X) # 按时间序列切分，不打乱顺序 X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, shuffle=False) 切分后训练集 799 条、测试集 200 条，再用 create_sequences 滚成时间步片段，独热编码后 y_train 变 (794, 2)、y_test 变 (195, 2)——样本数少了是因为序列窗口吃掉头部几根。 卷积层用 Conv1D(filters=16, kernel_size=3, strides=2, padding='causal')，因果填充保证不偷看未来，MaxPooling1D 压一轮维度。外汇和贵金属行情噪声大、跳空多，直接套这结构前建议先把点差和滑点当作特征一起归一化，否则卷积提的特征在实盘可能偏乐观。
target_var = [] open_price = new_df["Open"] close_price = new_df["Close"] for i in range(len(open_price)): if close_price[i] > open_price[i]: # Closing price is greater than opening price target_var.append(class="num">1) # buy signal else: target_var.append(class="num">0) # sell signal X = new_df.iloc[:, :-class="num">1] y = target_var # Scalling the data scaler = StandardScaler() X = scaler.fit_transform(X) # Train-test split X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=class="num">0.2, shuffle=False) print(f"x_train = {X_train.shape} - x_test = {X_test.shape}\n\ny_train = {len(y_train)} - y_test = {len(y_test)}") x_train = (class="num">799, class="num">3) - x_test = (class="num">200, class="num">3) y_train = class="num">799 - y_test = class="num">200 # creating the sequence X_train, y_train = create_sequences(X_train, y_train, time_step) X_test, y_test = create_sequences(X_test, y_test, time_step) from tensorflow.keras.utils class="kw">import to_categorical y_train_encoded = to_categorical(y_train) y_test_encoded = to_categorical(y_test) print(f"One hot encoded\n\ny_train {y_train_encoded.shape}\ny_test {y_test_encoded.shape}") One hot encoded y_train(class="num">794, class="num">2) y_test(class="num">195, class="num">2) model = Sequential() model.add(Conv1D(filters=class="num">16, kernel_size=class="num">3, activation=&class="macro">#x27;relu&class="macro">#x27;, strides=class="num">2, padding=&class="macro">#x27;causal&class="macro">#x27;, input_shape=(time_step, X_train.shape[class="num">2]) ) ) model.add(MaxPooling1D(pool_size=class="num">2))