PyTorch量化技术教程：第五章 综合实战项目

PyTorch量化技术教程：综合实战项目

本教程旨在为读者提供一套全面且深入的PyTorch技术在量化交易领域应用的知识体系。系统涵盖PyTorch基础入门、核心组件详解、模型构建与训练，以及在A股市场中的实战应用。采用理论与实战深度融合的讲解模式，详细剖析如何运用PyTorch打造量化交易系统全流程。从数据处理的精细操作，到模型训练的优化技巧，再到交易信号生成的精准逻辑，以及风险管理的严谨策略，每个环节都通过专业示例和代码实现进行阐释，确保读者能够扎实掌握并灵活运用所学知识。
文中内容仅限技术学习与代码实践参考，市场存在不确定性，技术分析需谨慎验证，不构成任何投资建议。适合量化新手建立系统认知，为策略开发打下基础。

目录

1. PyTorch基础入门

   • 1.1 PyTorch简介与环境搭建
   • 1.2 Tensor基础操作与自动求导机制

2. PyTorch核心组件详解

   • 2.1 nn.Module模块使用与自定义
   • 2.2 优化器选择与使用
   • 2.3 数据加载与预处理

3. PyTorch模型构建与训练

   • 3.1 神经网络模型构建流程
   • 3.2 模型训练技巧与实践
   • 3.3 模型评估与保存加载

4. PyTorch在量化交易中的应用

   • 4.1 时间序列分析与预测
   • 4.2 量化交易策略构建与优化
   • 4.3 风险管理与绩效评估

5. 综合实战项目

   • 5.1 基于A股市场的量化交易系统开发
   • 5.2 模型部署与实际交易模拟

第五章 综合实战项目

5.1 基于A股市场的量化交易系统开发

项目概述

本项目旨在开发一个基于A股市场的量化交易系统，利用PyTorch构建预测模型，生成交易信号，并进行风险管理。系统将包括数据获取、数据预处理、模型训练、交易信号生成、交易执行和绩效评估等模块。

模块详细讲解

1. 数据获取模块

import tushare as ts

# 设置Tushare API令牌
ts.set_token("your_token")
pro = ts.pro_api()


def fetch_ashare_data(ts_code="600000.SH", start_date="20200101", end_date="20241231"):
    """
    获取A股历史数据
    """
    df = pro.daily(ts_code=ts_code, start_date=start_date, end_date=end_date)
    df = df.sort_values("trade_date")
    df.to_parquet(f"./data/{ts_code}_historical_data.parquet")
    return df


# 获取数据示例
ashare_data = fetch_ashare_data()

2. 数据预处理模块

import pandas as pd
import talib
from sklearn.preprocessing import StandardScaler


def preprocess_data(file_path, sequence_length=10):
    """
    数据预处理函数
    """
    # 读取数据
    data = pd.read_parquet(file_path)
    # 计算技术指标
    data["MA5"] = talib.MA(data["close"], timeperiod=5)
    data["MA10"] = talib.MA(data["close"], timeperiod=10)
    data["RSI"] = talib.RSI(data["close"], timeperiod=14)
    data["MACD"], _, _ = talib.MACD(
        data["close"], fastperiod=12, slowperiod=26, signalperiod=9
    )
    # 数据清洗
    data.dropna(inplace=True)
    # 特征选择
    features = data[["open", "high", "low", "MA5", "MA10", "RSI", "MACD"]]
    labels = data[["close"]]
    # 归一化
    scaler_features = StandardScaler()
    scaler_labels = StandardScaler()
    features_scaled = scaler_features.fit_transform(features)
    labels_scaled = scaler_labels.fit_transform(labels)
    # 创建序列数据
    sequences = []
    targets = []
    for i in range(len(features_scaled) - sequence_length):
        sequences.append(features_scaled[i : i + sequence_length])
        targets.append(labels_scaled[i + sequence_length])
    return np.array(sequences), np.array(targets), scaler_features, scaler_labels


# 预处理数据示例
sequences, targets, scaler_features, scaler_labels = preprocess_data(
    "./data/600000.SH_historical_data.parquet"
)

3. 模型训练模块

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import TensorDataset, DataLoader


# 定义LSTM模型
class LSTMModel(nn.Module):
    def __init__(self, input_size, hidden_size, output_size, num_layers):
        super(LSTMModel, self).__init__()
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, output_size)

    def forward(self, x):
        h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).requires_grad_()
        c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).requires_grad_()
        out, (hn, cn) = self.lstm(x, (h0.detach(), c0.detach()))
        out = self.fc(out[:, -1, :])
        return out


# 准备数据
input_size = sequences.shape[2]
hidden_size = 64
output_size = 1
num_layers = 2
batch_size = 32
epochs = 100

# 创建数据集和数据加载器
dataset = TensorDataset(
    torch.tensor(sequences, dtype=torch.float32),
    torch.tensor(targets, dtype=torch.float32),
)
train_size = int(0.8 * len(dataset))
test_size = len(dataset) - train_size
train_dataset, test_dataset = torch.utils.data.random_split(
    dataset, [train_size, test_size]
)
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)

# 初始化模型、损失函数和优化器
model = LSTMModel(input_size, hidden_size, output_size, num_layers)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# 训练模型
for epoch in range(epochs):
    model.train()
    for inputs, targets in train_loader:
        outputs = model(inputs)
        loss = criterion(outputs, targets)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
    if (epoch + 1) % 10 == 0:
        model.eval()
        with torch.no_grad():
            total_loss = 0
            for inputs, targets in test_loader:
                outputs = model(inputs)
                total_loss += criterion(outputs, targets).item()
            avg_loss = total_loss / len(test_loader)
            print(f"Epoch [{epoch+1}/{epochs}], Test Loss: {avg_loss:.4f}")

# 保存模型
torch.save(model.state_dict(), "./models/lstm_quant_model.pth")

输出

Epoch [10/100], Test Loss: 0.0092
Epoch [20/100], Test Loss: 0.0074
Epoch [30/100], Test Loss: 0.0082
Epoch [40/100], Test Loss: 0.0068
Epoch [50/100], Test Loss: 0.0098
Epoch [60/100], Test Loss: 0.0073
Epoch [70/100], Test Loss: 0.0075
Epoch [80/100], Test Loss: 0.0082
Epoch [90/100], Test Loss: 0.0098
Epoch [100/100], Test Loss: 0.0075

4. 交易信号生成模块

def generate_signals(
    model, scaler_features, scaler_labels, new_data, sequence_length=10
):
    """
    生成交易信号
    """
    # 数据预处理
    new_data["MA5"] = talib.MA(new_data["close"], timeperiod=5)
    new_data["MA10"] = talib.MA(new_data["close"], timeperiod=10)
    new_data["RSI"] = talib.RSI(new_data["close"], timeperiod=14)
    new_data["MACD"], _, _ = talib.MACD(
        new_data["close"], fastperiod=12, slowperiod=26, signalperiod=9
    )
    new_data.dropna(inplace=True)
    features = new_data[["open", "high", "low", "MA5", "MA10", "RSI", "MACD"]]
    features_scaled = scaler_features.transform(features)
    # 创建序列
    sequences = []
    for i in range(len(features_scaled) - sequence_length):
        sequences.append(features_scaled[i : i + sequence_length])
    sequences_tensor = torch.tensor(sequences, dtype=torch.float32)
    # 预测
    model.eval()
    with torch.no_grad():
        predictions = model(sequences_tensor)
    predictions_rescaled = scaler_labels.inverse_transform(predictions.numpy())
    # 生成信号
    signals = []
    for i in range(len(predictions_rescaled)):
        if predictions_rescaled[i] > new_data["close"].values[i + sequence_length]:
            signals.append(1)  # 买入信号
        else:
            signals.append(0)  # 卖出信号
    return signals


# 生成交易信号示例
new_data = pd.read_parquet("./data/600000.SH_historical_data.parquet")
signals = generate_signals(model, scaler_features, scaler_labels, new_data)

5. 交易执行模块

class TradingExecutor:
    def __init__(self, initial_capital=100000):
        self.initial_capital = initial_capital
        self.capital = initial_capital
        self.position = None  # 当前持仓，None表示空仓

    def execute_trade(self, signal, current_price):
        """
        执行交易
        """
        if signal == 1:  # 买入信号
            if self.position is None:
                self.position = current_price
                print(f"买入价格: {current_price}, 当前资金: {self.capital}")
        else:  # 卖出信号
            if self.position is not None:
                shares = self.capital / self.position
                self.capital = shares * current_price
                self.position = None
                print(f"卖出价格: {current_price}, 当前资金: {self.capital}")

    def get_performance(self):
        """
        获取交易绩效
        """
        return (self.capital - self.initial_capital) / self.initial_capital


# 执行交易示例
executor = TradingExecutor()
for i in range(len(signals)):
    signal = signals[i]
    current_price = new_data["close"].values[i + sequence_length]
    executor.execute_trade(signal, current_price)
performance = executor.get_performance()
print(f"交易绩效: {performance:.4f}")

输出

买入价格: 10.64, 当前资金: 100000
卖出价格: 10.7, 当前资金: 100563.90977443608
买入价格: 10.43, 当前资金: 100563.90977443608
卖出价格: 10.09, 当前资金: 97285.69986807862
买入价格: 9.84, 当前资金: 97285.69986807862
卖出价格: 10.06, 当前资金: 99460.78665374705
...
买入价格: 10.34, 当前资金: 222604.95371091127
卖出价格: 10.47, 当前资金: 225403.66202642565
买入价格: 10.29, 当前资金: 225403.66202642565
交易绩效: 1.2540

6. 风险管理模块

class RiskManager:
    def __init__(self, max_drawdown_threshold=0.1, stop_loss=0.05, take_profit=0.1):
        self.max_drawdown_threshold = max_drawdown_threshold
        self.stop_loss = stop_loss
        self.take_profit = take_profit

    def monitor_risk(self, cumulative_returns):
        """
        监控风险
        """
        if not cumulative_returns:  # 如果累计收益率为空，返回 True
            return True
        current_dd = self.max_drawdown(cumulative_returns)
        if current_dd > self.max_drawdown_threshold:
            return False  # 风险过高，停止交易
        return True

    def max_drawdown(self, cumulative_returns):
        """
        计算最大回撤
        """
        if not cumulative_returns:
            return 0.0  # 如果没有数据，返回0

        max_dd = 0.0
        peak = cumulative_returns[0]

        for ret in cumulative_returns:
            if peak == 0:  # 处理 peak 为0的情况
                dd = 0.0
            else:
                dd = (peak - ret) / peak

            if dd > max_dd:
                max_dd = dd

            if ret > peak:
                peak = ret

        return max_dd

    def should_stop_loss(self, entry_price, current_price):
        """
        判断是否触发止损
        """
        return (current_price - entry_price) / entry_price <= -self.stop_loss

    def should_take_profit(self, entry_price, current_price):
        """
        判断是否触发止盈
        """
        return (current_price - entry_price) / entry_price >= self.take_profit


# 风险管理示例
risk_manager = RiskManager()
position = None
cumulative_returns = []
for i in range(len(signals)):
    signal = signals[i]
    current_price = new_data["close"].values[i + sequence_length]
    if signal == 1:
        if position is None:
            position = current_price
            cumulative_returns.append(0)
        else:
            if risk_manager.should_stop_loss(
                position, current_price
            ) or risk_manager.should_take_profit(position, current_price):
                position = None
    else:
        position = None
    if not risk_manager.monitor_risk(cumulative_returns):
        print("风险过高，停止交易")
        break

总结

本章通过一个完整的基于A股市场的量化交易系统开发项目，综合运用了PyTorch在量化交易中的各项技术。从数据获取到风险管理，每个模块都紧密结合实战需求，展示了如何利用PyTorch构建高效、稳定的量化交易系统。通过本章的学习，读者能够掌握从模型构建到实际交易的完整流程，为在实际市场中应用奠定坚实的基础。

风险提示与免责声明
本文内容基于公开信息研究整理，不构成任何形式的投资建议。历史表现不应作为未来收益保证，市场存在不可预见的波动风险。投资者需结合自身财务状况及风险承受能力独立决策，并自行承担交易结果。作者及发布方不对任何依据本文操作导致的损失承担法律责任。市场有风险，投资须谨慎。