当前位置：首页 > article >正文

100.5 AI量化面试题：在使用LSTM预测股票价格时，如何有效处理金融时间序列的非平稳性？

article 2025/2/6 1:30:35

- 0. 承前
- 1. 数据预处理
- - 1.1 平稳性检验
  - 1.2 数据转换
- 2. 特征工程
- - 2.1 技术指标构建
  - 2.2 时间特征提取
- 3. LSTM模型设计
- - 3.1 数据准备
  - 3.2 模型架构
- 4. 训练与验证
- - 4.1 时序交叉验证
  - 4.2 滚动预测
- 5. 回答话术

0. 承前

本文详细介绍使用LSTM处理金融时间序列时的关键技术点，包括数据预处理、特征工程、模型设计和验证方法。

如果想更加全面清晰地了解金融资产组合模型进化论的体系架构，可参考：
0. 金融资产组合模型进化全图鉴

1. 数据预处理

1.1 平稳性检验

import pandas as pd
import numpy as np
from statsmodels.tsa.stattools import adfuller

def check_stationarity(series):
    """
    进行ADF检验判断时间序列的平稳性
    """
    result = adfuller(series)
    print('ADF Statistic:', result[0])
    print('p-value:', result[1])
    print('Critical values:')
    for key, value in result[4].items():
        print('\t%s: %.3f' % (key, value))
    
    # 判断是否平稳
    is_stationary = result[1] < 0.05
    return is_stationary

1.2 数据转换

def transform_data(df):
    """
    对数据进行转换以达到平稳性
    """
    # 1. 对数转换
    df['log_price'] = np.log(df['close'])
    
    # 2. 差分
    df['returns'] = df['log_price'].diff()
    
    # 3. 标准化
    def standardize(series):
        return (series - series.mean()) / series.std()
    
    df['returns_standardized'] = df['returns'].rolling(
        window=20).apply(standardize)
    
    # 4. 百分比变化
    df['pct_change'] = df['close'].pct_change()
    
    return df

2. 特征工程

2.1 技术指标构建

def create_features(df):
    """
    构建技术指标特征
    """
    # 移动平均
    df['MA5'] = df['close'].rolling(window=5).mean()
    df['MA20'] = df['close'].rolling(window=20).mean()
    
    # 波动率
    df['volatility'] = df['returns'].rolling(window=20).std()
    
    # RSI
    def calculate_rsi(data, periods=14):
        delta = data.diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=periods).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=periods).mean()
        rs = gain / loss
        return 100 - (100 / (1 + rs))
    
    df['RSI'] = calculate_rsi(df['close'])
    
    return df

2.2 时间特征提取

def add_time_features(df):
    """
    添加时间相关特征
    """
    df['dayofweek'] = df.index.dayofweek
    df['quarter'] = df.index.quarter
    df['month'] = df.index.month
    df['year'] = df.index.year
    
    # 季节性特征
    df['sin_day'] = np.sin(2 * np.pi * df.index.dayofyear/365.25)
    df['cos_day'] = np.cos(2 * np.pi * df.index.dayofyear/365.25)
    
    return df

3. LSTM模型设计

3.1 数据准备

def prepare_sequences(data, seq_length):
    """
    准备LSTM输入序列
    """
    X, y = [], []
    for i in range(len(data) - seq_length):
        X.append(data[i:(i + seq_length)])
        y.append(data[i + seq_length])
    return np.array(X), np.array(y)

class TimeSeriesDataset(Dataset):
    def __init__(self, X, y):
        self.X = torch.FloatTensor(X)
        self.y = torch.FloatTensor(y)
        
    def __len__(self):
        return len(self.X)
        
    def __getitem__(self, idx):
        return self.X[idx], self.y[idx]

3.2 模型架构

import torch
import torch.nn as nn

class LSTMPredictor(nn.Module):
    def __init__(self, input_dim, hidden_dim, num_layers, dropout=0.2):
        super(LSTMPredictor, self).__init__()
        
        self.lstm = nn.LSTM(
            input_dim,
            hidden_dim,
            num_layers,
            batch_first=True,
            dropout=dropout
        )
        
        self.attention = nn.MultiheadAttention(
            hidden_dim, 
            num_heads=4
        )
        
        self.fc = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim//2),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(hidden_dim//2, 1)
        )
        
    def forward(self, x):
        lstm_out, _ = self.lstm(x)
        
        # 添加注意力机制
        attn_out, _ = self.attention(
            lstm_out.permute(1,0,2),
            lstm_out.permute(1,0,2),
            lstm_out.permute(1,0,2)
        )
        
        out = self.fc(attn_out[-1])
        return out

4. 训练与验证

4.1 时序交叉验证

from sklearn.model_selection import TimeSeriesSplit

def time_series_cv(model, X, y, n_splits=5):
    tscv = TimeSeriesSplit(n_splits=n_splits)
    cv_scores = []
    
    for train_idx, val_idx in tscv.split(X):
        X_train, X_val = X[train_idx], X[val_idx]
        y_train, y_val = y[train_idx], y[val_idx]
        
        # 训练模型
        model.fit(X_train, y_train)
        
        # 评估
        score = model.evaluate(X_val, y_val)
        cv_scores.append(score)
        
    return np.mean(cv_scores), np.std(cv_scores)

4.2 滚动预测

def rolling_prediction(model, data, window_size, horizon):
    predictions = []
    for i in range(len(data) - window_size - horizon + 1):
        # 获取训练窗口数据
        train_data = data[i:i+window_size]
        
        # 训练模型
        model.fit(train_data)
        
        # 预测
        pred = model.predict(horizon)
        predictions.append(pred)
        
    return np.array(predictions)

5. 回答话术

在处理金融时间序列的非平稳性时，我们采用了一个系统化的方法论。首先，通过ADF检验等方法对时间序列进行平稳性检验。然后，使用对数转换、差分、标准化等方法将非平稳序列转换为平稳序列。在特征工程环节，我们构建了技术指标和时间特征来捕捉市场的多维度信息。模型设计上，我们使用了带有注意力机制的LSTM架构，能够更好地处理长期依赖关系。最后，通过时序交叉验证和滚动预测来评估模型性能，确保预测结果的可靠性。

关键技术要点：