多股票数据集LSTM训练方案可行性及MSE异常问题问询

安全阿岳

2026-8-21

问题

我正在训练LSTM神经网络以预测股票价格分布的五分位数（quintiles），计划基于500只股票的样本进行训练。目前有两种训练思路：

思路一：循环遍历每只股票，训练时用checkpoint保存最优模型参数，训练下一只股票时加载该参数
思路二：将500只股票的数据沿0轴拼接成完整数据集进行训练

编写的循环训练代码如下：

def compile_and_fit(model_type,model,checkpoint_path,config, stock_data,macro_data, factor_data, patience, batch_size, 
                    num_epochs,train_set_ratio, val_set_ratio, Y_name):
    """
    model = NN model, 
    data = stock data, factor data, macro data,
    batch_size = timesteps per batch
    alpha adam = learning rate optimizer
    data set ratios = train_set_ratio, val_set_ratio (eg. 0.5)
    """
    
    early_stopping = tf.keras.callbacks.EarlyStopping(
        monitor='loss',                   #'loss'
        patience=patience,
        mode='min')
    
    cp_callback = tf.keras.callbacks.ModelCheckpoint(
        checkpoint_path,
        monitor= 'loss',
        verbose=True,
        save_best_only=True,
        save_freq = batch_size,
        mode='min')  
    permno_list = stock_data.permno.unique()

    test_data = pd.DataFrame()
    counter = 0
    for p in permno_list:
        
        #checkpoints 
        if counter == 0:
            trained_model = model
            cp_callback = cp_callback
        else:
            trained_model = tf.keras.models.load_model(checkpoint_path)
            cp_callback = tf.keras.callbacks.ModelCheckpoint(checkpoint_path,monitor= 'loss',verbose=True, save_best_only=True,save_freq = batch_size, mode='min')
        
        stock_data_length = len(stock_data.loc[stock_data.permno==p])
        train_data_stocks = stock_data.loc[stock_data.permno==p][0:int(stock_data_length*train_set_ratio)]
        val_data_stocks = stock_data.loc[stock_data.permno==p][int(stock_data_length*train_set_ratio):int(stock_data_length*(val_set_ratio+train_set_ratio))]
        test_data_stocks = stock_data.loc[stock_data.permno==p][int(stock_data_length*(val_set_ratio+train_set_ratio)):]
        test_data = pd.concat([test_data, test_data_stocks],axis=0)
        
        train_date_index = train_data_stocks.index.values.tolist() 
        val_date_index = val_data_stocks.index.values.tolist()
    
        train_data_factors = factor_data.loc[factor_data.index.isin(train_date_index)]
        train_data_macro = macro_factors.loc[macro_factors.index.isin(train_date_index)]
        train_data_macro_norm = train_data_macro.copy(deep=True)

        for c in train_data_macro_norm.columns: 
            train_data_macro_norm[c] = MinMaxScaler([-1,1]).fit_transform(pd.DataFrame(train_data_macro_norm[c]))
        
        train_data_merged = pd.concat([train_data_factors, train_data_macro_norm],axis=1)
        
        

        val_data_factors = factor_data.loc[factor_data.index.isin(val_date_index)]
        val_data_macro = macro_factors.loc[macro_factors.index.isin(val_date_index)]    
        val_data_macro_norm = val_data_macro.copy(deep=True)
        
        for c in val_data_macro_norm.columns: 
            val_data_macro_norm[c] = MinMaxScaler([-1,1]).fit_transform(pd.DataFrame(val_data_macro_norm[c]))

        val_data_merged = pd.concat([val_data_factors, val_data_macro_norm],axis=1)
        

        if model_type=='combined':
            
            x_train_factors = []
            x_train_macro = []
            y_train =[]

            for i in range(batch_size, len(train_data_factors)):
                x_train_factors.append(train_data_factors.values[i-batch_size:i,:])
                x_train_macro.append(train_data_macro_norm.values[i-batch_size:i,:])
                y_train.append(train_data_stocks[Y_name].values[i])
            x_train_factors, x_train_macro, y_train= np.array(x_train_factors),np.array(x_train_macro), np.array(y_train)
            
            x_val_factors = []
            x_val_macro = []
            y_val =[]

            for i in range(batch_size, len(val_data_factors)):
                x_val_factors.append(val_data_factors.values[i-batch_size:i,:])
                x_val_macro.append(val_data_macro_norm.values[i-batch_size:i,:])
                y_val.append(val_data_stocks[Y_name].values[i])

            x_val_factors, x_val_macro, y_val = np.array(x_val_factors),np.array(x_val_macro), np.array(y_val)
            score =trained_model.evaluate([x_train_macro,x_train_factors],y_train,batch_size=batch_size)
            
            score = list(score)
            score.sort(reverse=True)
            score = score[-2]
            cp_callback.best = score
            
            trained_model.fit(x=[x_train_macro,x_train_factors],y=y_train,batch_size=batch_size, epochs=num_epochs,
                              validation_data=[[x_val_macro,x_val_factors], y_val], callbacks=[early_stopping,cp_callback]) 
        
            
        if model_type=='merged':
            
            x_train_merged = []
            y_train =[]

            for i in range(batch_size, len(train_data_merged)):
                x_train_merged.append(train_data_merged.values[i-batch_size:i,:])
                y_train.append(train_data_stocks[Y_name].values[i])

            x_train_merged, y_train= np.array(x_train_merged), np.array(y_train)

            x_val_merged = []
            y_val =[]

            for i in range(batch_size, len(val_data_merged)):
                x_val_merged.append(val_data_merged.values[i-batch_size:i,:])
                y_val.append(val_data_stocks[Y_name].values[i])

            x_val_merged, y_val = np.array(x_val_merged), np.array(y_val)
            
            score =trained_model.evaluate(x_train_merged,y_train,batch_size=batch_size)
            
            score = list(score)
            score.sort(reverse=True)
            score = score[-2]
            cp_callback.best = score
            
            trained_model.fit(x=x_train_merged,y=y_train,batch_size=batch_size, epochs=num_epochs,
                              validation_data=[x_val_merged, y_val], callbacks=[early_stopping,cp_callback])
    return trained_model, test_data

测试中发现，单只股票训练时MSE持续下降，但切换到下一只股票训练时MSE会骤升。现请教三个问题：

该循环训练代码是否可行？
两种训练方案哪种更合适？
如何解决MSE骤升的问题？

回答

1. 循环训练代码是否可行？

思路可行，但存在多处明显缺陷：

数据归一化错误：每只股票的训练、验证集单独拟合MinMaxScaler，既会导致验证集数据泄露（验证集归一化不应使用自身数据），又会让不同股票的特征分布不统一，模型难以学习通用规律。
Checkpoint逻辑混乱：手动设置cp_callback.best违背Keras默认逻辑，可能导致保存的不是真正最优的模型；每次循环重新创建回调函数属于冗余操作。
模型状态丢失：加载模型时默认不会保存优化器状态，导致每只股票训练都从新的优化器状态开始，无法延续之前的训练动量。
内存风险：循环拼接test_data会随股票数量增加占用大量内存，且未处理索引冲突问题。

2. 两种训练方案哪种更合适？

需根据任务目标选择：

拼接数据集训练（思路二）：
优势：一次性利用所有股票信息，模型学习市场通用规律，训练效率更高，避免循环冗余。
注意点：需保证所有股票数据时间对齐，用全局归一化器处理特征；可加入股票ID嵌入层，让模型学习个股特异性。
循环迁移训练（思路一）：
优势：内存占用低，可针对个股微调，适合捕捉个股特异性的场景。
注意点：必须解决归一化统一、优化器状态延续的问题，否则易出现性能骤降。

若目标是预测市场整体五分位数分布，选拼接数据集训练；若针对个股五分位数预测，循环迁移训练配合微调更合理。

3. 如何解决MSE骤升的问题？

核心解决数据分布不一致、模型状态断裂的问题，具体方案如下：

（1）统一数据归一化逻辑

全局拟合归一化器：在循环外，用所有股票的训练集数据拟合MinMaxScaler，之后所有股票的训练、验证、测试集都复用该scaler转换，避免每只股票单独拟合导致的分布差异。示例代码：

# 循环外先拟合全局scaler
macro_scaler = MinMaxScaler([-1,1])
# 先确定所有股票的训练日期范围，用该范围内的宏观数据拟合
train_date_range = stock_data.groupby('permno').apply(lambda x: x.index[:int(len(x)*train_set_ratio)]).explode().unique()
macro_scaler.fit(macro_factors.loc[macro_factors.index.isin(train_date_range)])

# 循环内处理单只股票时
train_data_macro_norm = pd.DataFrame(macro_scaler.transform(train_data_macro), columns=train_data_macro.columns)
val_data_macro_norm = pd.DataFrame(macro_scaler.transform(val_data_macro), columns=val_data_macro.columns)

（2）优化模型训练与Checkpoint逻辑

保留优化器状态：用save_weights保存模型，加载时复用优化器状态，避免每次训练重置：

# 保存模型时
trained_model.save_weights(checkpoint_path)

# 加载模型时
trained_model.load_weights(checkpoint_path)
# 保持初始模型的优化器、损失函数配置
trained_model.compile(optimizer='adam', loss='mse')

移除手动设置cp_callback.best的操作：Keras的ModelCheckpoint会自动跟踪最优指标，手动修改会打乱判断逻辑。

（3）迁移学习微调策略

降低学习率：切换到新股票时，使用初始学习率1/10的小学习率，避免模型参数被新数据大幅扰动。
分层冻结：先冻结LSTM底层（捕捉通用时序特征），只训练顶层全连接层，再逐步解冻微调，减少模型震荡。

（4）数据层面优化

加入股票ID嵌入：若用拼接数据集训练，将股票ID转为嵌入向量输入模型，让模型学习个股特征，提升跨股票适应性。
对齐时间窗口：确保每只股票的训练、验证时间窗口长度一致，避免数据长度差异导致输入分布变化。

内容的提问来源于stack exchange，提问作者Maurice Matthias

多股票数据集LSTM训练方案可行性及MSE异常问题问询

1. 循环训练代码是否可行？ #

2. 两种训练方案哪种更合适？ #

3. 如何解决MSE骤升的问题？ #

（1）统一数据归一化逻辑 #

（2）优化模型训练与Checkpoint逻辑 #

（3）迁移学习微调策略 #

（4）数据层面优化 #