时间序列深度学习架构对比

输入了某只股票1分钟Kline，特征是ohlc，预测close。

第一步，先对比了FNN、LSTM、CNN、TCN、Transformer等不同模型执行后的MSE表现，结构如下图：

从结果来看LSTM、Transformer、CNN表现可以。

看了上面的结果后，我修改了预测值，特征是ohlc，预测1分钟的收益（注意：数据是1分钟的，预测1分钟收益），并重新跑了模型：

果不其然，如上图所示，MSE和MAE极低，说明模型预测结果与真实值之间的差异非常小，但相关系数低，又说明预测值和真实值之间几乎没有线性关系，显然这是一个矛盾的结论。

这种矛盾是因为：模型只是在预测一个接近常数的值，而不是根据输入特征进行有效预测。例如，如果1分钟收益的波动很小，模型可能只是学会了预测平均值，导致MSE和MAE低，但无法捕捉变化，因此相关系数低。

然后我只使用了LSTM模型，增加了几个指标特征，把预测目标调整为3日收益率，结果如下图：

如上图所示，预测命中率（50.1%），按目前不成文的行业基准来说，A股股票预测模型中，55%以上的命中率，从理论上才具有统计显著性（p<0.05），本次预测值仅比随机猜测（50%）高0.1个百分点，说明模型未展现出有效预测能力，散点图呈对称分布，说明无显著线性趋势。实际收益率分布标准差（≈0.8%）远高于预测值（≈0.5%），说明特征信息不足。

同时，上图在极端收益区域：实际收益率>1%或；年化收益1.1%，基本无法覆盖交易成本，以单边佣金0.03%+印花税0.1%来算。

下面是3分钟收益预测核心模块相关代码：

1.导入包

import osos.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'import numpy as npimport pandas as pdimport matplotlib.pyplot as pltfrom sklearn.preprocessing import RobustScalerimport tensorflow as tffrom matplotlib import font_managerimport sys

2.参数配置

第一次使用的是只有ohlc，发现基本上看不出来效果，就再多加了几个指标数据。

#------------------------ 全局参数 ------------------------#PREDICTION_HORIZON = 3    # 预测未来3天收益率SEQ_LENGTH = 21           # 使用21个交易日历史数据FEATURE_COLS = [    'open', 'high', 'low', 'close', 'cjl',    'rsi', 'macd', 'atr', 'obv', 'ad']

3.特征工程

主要是计算指标，技术目标变量。


def calculate_3day_returns(df):    """计算3日收益率（含边界处理）"""    df = df.copy()    df['future_close'] = df['close'].shift(-PREDICTION_HORIZON)    df['3day_return'] = (df['future_close'] - df['close']) / df['close']    # 清除无效数据    df = df.dropna(subset=['3day_return', 'future_close'])


    
    return dfdef add_technical_features(df):    """技术指标计算（带异常值处理）"""    # RSI (14天周期)    delta = df['close'].diff().fillna(0)    gain = delta.where(delta > 0, 0)    loss = -delta.where(delta 0, 0)    avg_gain = gain.rolling(14, min_periods=1).mean()    avg_loss = loss.rolling(14, min_periods=1).mean()    rs = avg_gain / (avg_loss + 1e-10)  # 避免除以零    df['rsi'] = 100 - (100 / (1 + rs))        # MACD (12/26周期)    ema12 = df['close'].ewm(span=12, adjust=False).mean()    ema26 = df['close'].ewm(span=26, adjust=False).mean()    df['macd'] = ema12 - ema26        # ATR (14天周期)    tr = pd.DataFrame({        'high_low': df['high'] - df['low'],        'high_close': np.abs(df['high'] - df['close'].shift()),        'low_close': np.abs(df['low'] - df['close'].shift())    }).max(axis=1)    df['atr'] = tr.rolling(14, min_periods=1).mean()        # OBV    df['obv'] = (np.sign(df['close'].diff().fillna(0)) * df['cjl']).cumsum()        # AD指标    hl_range = df['high'] - df['low']    hl_range = hl_range.replace(0, 1e-10)  # 避免除零    mfm = ((df['close'] - df['low']) - (df['high'] - df['close'])) / hl_range    df['ad'] = (mfm * df['cjl']).cumsum()        return df.dropna()

4.数据处理

下面对时序数据进行标准化处理，并且将标准化后的数据还原为原始数据。

class TemporalScaler:    """时序数据标准化器"""    def __init__(self):        self.scalers = {col: RobustScaler() for col in FEATURE_COLS}            def fit_transform(self, df):        scaled_df = df.copy()        for col in FEATURE_COLS:            scaled_df[col] = self.scalers[col].fit_transform(scaled_df[[col]])        return scaled_df        def inverse_transform(self, df):        restored_df = df.copy()        for col in FEATURE_COLS:            restored_df[col] = self.scalers[col].inverse_transform(restored_df[[col]])        return restored_df

5.模型构建

第一层（64单元）捕获短期时序模式，第二层（32单元）建模长期依赖。

def build_temporal_model(input_shape):    """构建LSTM时序模型"""    model = tf.keras.Sequential([        tf.keras.layers.Input(shape=input_shape),        tf.keras.layers.LSTM(64, return_sequences=True, recurrent_dropout=0.1),        tf.keras.layers.Dropout(0.2),        tf.keras.layers.LSTM(32, recurrent_dropout=0.1),        tf.keras.layers.Dense(16, activation='relu'),        tf.keras.layers.Dense(1, activation='linear')  # 改为线性激活用于回归    ])    model.compile(        optimizer=tf.keras.optimizers.Adam(learning_rate=0.001),        loss='mse',        metrics=['mae']    )    return model

6.绩效评估

def evaluate_strategy(y_true, predictions):    """策略评估函数（带鲁棒性校验）"""    # 输入校验    y_true = np.asarray(y_true).flatten()    predictions = np.asarray(predictions).flatten()    assert len(y_true) == len(predictions), f"长度不匹配: y_true({len(y_true)}) vs predictions({len(predictions)})"        # 生成交易信号    signals = np.where(predictions > 0, 1, -1)        # 计算命中率    correct = (np.sign(y_true) == signals)    hit_rate = np.mean(correct) if len(correct) > 0 else 0.0        # 计算收益率    returns = y_true * signals    cumulative_returns = (1 + returns).cumprod()        # 计算风险指标    total_return = cumulative_returns[-1] - 1 if len(cumulative_returns) > 0 else 0.0    annualized_return = (1 + total_return) ** (252 / len(y_true)) - 1 if len(y_true) > 0 else 0.0        # 夏普比率（带波动率保护）    return_std = np.std(returns) if len(returns) > 1 else 1e-10    sharpe_ratio = np.sqrt(252) * np.mean(returns) / return_std        return {        'total_return': total_return,        'annualized_return': annualized_return,        'sharpe_ratio': sharpe_ratio,        'hit_rate': hit_rate,        'cumulative_returns': cumulative_returns    }

7.执行代码

if __name__ == "__main__":    # 环境验证    print(f"TensorFlow版本: {tf.__version__}")    print(f"可用GPU: {len(tf.config.list_physical_devices('GPU')) > 0}")        # 设置支持中文的字体    plt.rcParams['font.family'] = 'WenQuanYi Zen Hei'    # 解决负号显示问题    plt.rcParams['axes.unicode_minus'] = False    try:        # 数据加载，此处文件换成自己本地文件        df = pd.read_csv(            '202212_202302.csv',            parse_dates=['tdate'],            index_col='tdate'        ).sort_index()                # 特征工程        df = calculate_3day_returns(df)        df = add_technical_features(df)                # 数据预处理        scaler = TemporalScaler()        scaled_df = scaler.fit_transform(df)        X, y = create_sequences(scaled_df)                # 数据集分割        split = int(0.8 * len(X))        X_train, X_test = X[:split], X[split:]        y_train, y_test = y[:split], y[split:]                # 模型训练


    
        model = build_temporal_model((SEQ_LENGTH, len(FEATURE_COLS)))        history = model.fit(            X_train, y_train,            epochs=100,            batch_size=64,            validation_split=0.2,            callbacks=[                tf.keras.callbacks.EarlyStopping(patience=10, restore_best_weights=True),                tf.keras.callbacks.ReduceLROnPlateau(monitor='val_loss', factor=0.5, patience=5)            ],            verbose=1        )                # 模型预测        predictions = model.predict(X_test).flatten()        assert len(predictions) == len(y_test), "预测结果与测试集长度不匹配"                # 策略评估        results = evaluate_strategy(y_test, predictions)                # 可视化        plt.figure(figsize=(14, 10))                # 预测效果图        plt.subplot(2, 1, 1)        plt.scatter(y_test, predictions, alpha=0.5, color='#FF6F00')        plt.plot([min(y_test), max(y_test)], [min(y_test), max(y_test)], '--', color='gray')        plt.xlabel('实际收益率')        plt.ylabel('预测收益率')        plt.title(f'预测命中率: {results["hit_rate"]:.1%}')                # 收益曲线图        plt.subplot(2, 1, 2)        plt.plot(results['cumulative_returns'], label='策略收益', color='#2E86C1')        plt.plot((1 + y_test).cumprod(), label='基准收益', color='#28B463', alpha=0.7)        plt.xlabel('交易日')        plt.ylabel('累计收益')        plt.title(f'夏普比率: {results["sharpe_ratio"]:.2f} | 年化收益: {results["annualized_return"]:.1%}')        plt.legend()                plt.tight_layout()        plt.savefig('backtest_result.png', dpi=300, bbox_inches='tight')        plt.show()        plt.rcParams.update(plt.rcParamsDefault)  # 清除之前的配置缓存        configure_chinese_font()                 # 重新初始化            except Exception as e:        print(f"程序异常终止: {str(e)}")        raise