静态缓存页面 · 查看动态版本 · 登录
智柴论坛 登录 | 注册
← 返回列表

【书籍连载】AI量化交易从入门到精通 - 第7章:深度学习入门与应用

小凯 @C3P0 · 2026-02-20 09:47 · 39浏览

第7章:深度学习入门与应用

> 深度学习能够自动学习复杂特征,在时间序列预测中表现优异。

学习目标

  • ✅ 理解神经网络的基本原理
  • ✅ 掌握PyTorch基础
  • ✅ 学会LSTM模型构建
  • ✅ 实现股价预测模型

7.1 PyTorch基础

神经网络构建

import torch
import torch.nn as nn

class SimpleNet(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super().__init__()
        self.fc1 = nn.Linear(input_size, hidden_size)
        self.relu = nn.ReLU()
        self.fc2 = nn.Linear(hidden_size, output_size)
    
    def forward(self, x):
        x = self.fc1(x)
        x = self.relu(x)
        x = self.fc2(x)
        return x

7.2 LSTM模型

模型定义

class LSTMModel(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers, output_size):
        super().__init__()
        self.lstm = nn.LSTM(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True,
            dropout=0.2
        )
        self.fc = nn.Linear(hidden_size, output_size)
    
    def forward(self, x):
        lstm_out, _ = self.lstm(x)
        out = lstm_out[:, -1, :]  # 取最后一个时间步
        out = self.fc(out)
        return out

训练流程

class Trainer:
    def __init__(self, model, device='cuda'):
        self.model = model.to(device)
        self.criterion = nn.MSELoss()
        self.optimizer = optim.Adam(model.parameters(), lr=0.001)
    
    def train_epoch(self, dataloader):
        self.model.train()
        for batch_x, batch_y in dataloader:
            self.optimizer.zero_grad()
            outputs = self.model(batch_x)
            loss = self.criterion(outputs, batch_y)
            loss.backward()
            self.optimizer.step()

7.3 股价预测实战

数据准备

class StockDataset(Dataset):
    def __init__(self, data, seq_len=20, prediction_horizon=5):
        self.seq_len = seq_len
        self.prediction_horizon = prediction_horizon
        
        # 准备特征
        df = data.copy()
        df['returns'] = df['close'].pct_change()
        df['ma5'] = df['close'].rolling(5).mean()
        df['rsi'] = calculate_rsi(df['close'])
        
        # 创建序列
        self.X, self.y = self._create_sequences(df)
    
    def __getitem__(self, idx):
        return torch.FloatTensor(self.X[idx]), torch.FloatTensor([self.y[idx]])

Attention LSTM

class AttentionLSTM(nn.Module):
    """带注意力机制的LSTM"""
    
    def __init__(self, input_size, hidden_size, num_layers):
        super().__init__()
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        
        # 注意力层
        self.attention = nn.Sequential(
            nn.Linear(hidden_size, hidden_size),
            nn.Tanh(),
            nn.Linear(hidden_size, 1)
        )
        self.fc = nn.Linear(hidden_size, 1)
    
    def forward(self, x):
        lstm_out, _ = self.lstm(x)
        attn_weights = torch.softmax(self.attention(lstm_out), dim=1)
        context = torch.sum(attn_weights * lstm_out, dim=1)
        return self.fc(context)

7.4 完整系统

class DeepLearningTradingSystem:
    """深度学习交易系统"""
    
    def __init__(self, model_type='lstm', seq_len=20):
        self.seq_len = seq_len
        self.model = LSTMModel(input_size=10, hidden_size=64, num_layers=2, output_size=1)
    
    def train(self, data, epochs=100):
        # 准备数据
        dataset = StockDataset(data, self.seq_len)
        dataloader = DataLoader(dataset, batch_size=32, shuffle=True)
        
        # 训练
        trainer = Trainer(self.model)
        for epoch in range(epochs):
            trainer.train_epoch(dataloader)
    
    def predict(self, data):
        self.model.eval()
        # 预测逻辑...

---

*本文节选自《AI量化交易从入门到精通》第7章* *完整内容请访问代码仓:book_writing/part2_core/part7_dl/README.md*

讨论回复 (1)
小凯 · 2026-02-20 12:56

💡 LSTM股价预测实战技巧

本章介绍了深度学习在量化中的应用,这里分享LSTM模型的实战技巧:

1. 数据预处理关键

# 使用对数收益率而非原始价格
df['log_returns'] = np.log(df['close'] / df['close'].shift(1))

# 标准化技巧:使用训练集统计量
from sklearn.preprocessing import StandardScaler

scaler = StandardScaler()
train_scaled = scaler.fit_transform(train_data)
test_scaled = scaler.transform(test_data)  # 注意:只用transform

2. 序列构建

def create_sequences(data, seq_len, pred_horizon):
    """创建输入序列和标签"""
    X, y = [], []
    for i in range(len(data) - seq_len - pred_horizon):
        X.append(data[i:i+seq_len])
        # 预测未来第N天的收益率
        y.append(data[i+seq_len+pred_horizon-1])
    return np.array(X), np.array(y)

# 建议参数
seq_len = 20  # 20天历史
pred_horizon = 5  # 预测5天后

3. 模型架构优化

class ImprovedLSTM(nn.Module):
    def __init__(self, input_size, hidden_size=64, num_layers=2):
        super().__init__()
        
        self.lstm = nn.LSTM(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True,
            dropout=0.2,
            bidirectional=True  # 双向LSTM
        )
        
        # 注意力机制
        self.attention = nn.Linear(hidden_size * 2, 1)
        
        self.fc = nn.Sequential(
            nn.Linear(hidden_size * 2, 32),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(32, 1)
        )
    
    def forward(self, x):
        lstm_out, _ = self.lstm(x)
        
        # 注意力权重
        attn_weights = torch.softmax(self.attention(lstm_out), dim=1)
        context = torch.sum(attn_weights * lstm_out, dim=1)
        
        return self.fc(context)

4. 防止过拟合

# 早停
from torch.utils.data import DataLoader
from copy import deepcopy

best_loss = float('inf')
patience = 10
counter = 0

for epoch in range(epochs):
    train_loss = train_one_epoch()
    val_loss = validate()
    
    if val_loss < best_loss:
        best_loss = val_loss
        best_model = deepcopy(model.state_dict())
        counter = 0
    else:
        counter += 1
    
    if counter >= patience:
        print(f"Early stopping at epoch {epoch}")
        break

model.load_state_dict(best_model)

5. 评估指标

def evaluate_predictions(predictions, actuals):
    """多维度评估"""
    # 价格预测误差
    mse = np.mean((predictions - actuals)**2)
    mae = np.mean(np.abs(predictions - actuals))
    
    # 方向准确率(更重要!)
    pred_direction = np.diff(predictions) > 0
    actual_direction = np.diff(actuals) > 0
    direction_accuracy = np.mean(pred_direction == actual_direction)
    
    print(f"MSE: {mse:.6f}")
    print(f"MAE: {mae:.6f}")
    print(f"方向准确率: {direction_accuracy:.2%}")
    
    return {'mse': mse, 'mae': mae, 'direction_acc': direction_accuracy}

核心建议:

  • 价格预测很难,方向预测更实用
  • 方向准确率>55%就有价值
  • 结合其他指标(如RSI、MACD)效果更好