PTrade AI 交易：如何部署 LSTM 模型进行股价预测？

是的，可以在 PTrade 平台上使用 LSTM（长短期记忆网络）模型进行股价预测和交易。

PTrade 平台内置了 Keras (v2.3.1) 和 Tensorflow (v1.3.0rc1) 等深度学习库，同时也支持 sklearn 和 numpy，这使得构建、训练和推理 LSTM 模型成为可能。

核心注意事项

在编写代码前，请注意以下几点：

1. 性能限制：LSTM 的训练非常消耗计算资源。在 PTrade 的回测或实盘环境中，通常没有 GPU 加速，且有内存和运行时间限制。强烈建议不要在策略运行过程中频繁（如每天）重新训练模型。
2. 训练模式：
   • 推荐方式：在本地训练好模型，保存权重文件，上传到 PTrade 研究环境，然后在策略中加载权重进行推理（预测）。
   • 演示方式：为了展示代码逻辑，下方的示例将在策略初始化阶段进行一次简单的在线训练。
3. 库版本：PTrade 的 Tensorflow 版本较老（1.x），代码需兼容旧版 Keras 的写法。
4. 数据归一化：LSTM 对数据敏感，必须使用 MinMaxScaler 等进行归一化处理。

LSTM 股价预测策略代码示例

以下是一个完整的策略示例。该策略使用过去 N 天的收盘价作为特征，预测下一日的收盘价。如果预测价格高于当前价格一定比例则买入，反之卖出。

import numpy as np
import pandas as pd
from keras.models import Sequential
from keras.layers import Dense, LSTM
from sklearn.preprocessing import MinMaxScaler

def initialize(context):
    """
    初始化函数
    """
    # 设置标的：恒生电子
    g.security = '600570.SS'
    set_universe(g.security)
    
    # 策略参数
    g.look_back = 10       # 回溯窗口：用过去10天的数据预测下一天
    g.train_len = 300      # 训练数据长度：使用过去300天的数据进行训练
    g.predict_threshold = 0.01 # 阈值：预测涨幅超过1%才买入
    
    # 模型相关全局变量
    g.model = None
    g.scaler = MinMaxScaler(feature_range=(0, 1))
    g.is_trained = False   # 标记模型是否已训练
    
    # 设置每天定时运行训练或更新逻辑（可选，这里为了演示简单，仅在盘前检查）
    # 实际生产中建议离线训练，在线加载

def create_dataset(dataset, look_back=1):
    """
    将时间序列数据转换为LSTM需要的监督学习格式
    X: t-look_back ... t-1
    Y: t
    """
    dataX, dataY = [], []
    for i in range(len(dataset) - look_back - 1):
        a = dataset[i:(i + look_back), 0]
        dataX.append(a)
        dataY.append(dataset[i + look_back, 0])
    return np.array(dataX), np.array(dataY)

def build_and_train_model(context):
    """
    构建并训练LSTM模型
    注意：在回测环境中在线训练非常耗时，建议减少epochs或数据量
    """
    log.info("开始获取历史数据并训练 LSTM 模型...")
    
    # 获取历史收盘价数据
    # 多取一些数据以确保有足够的训练集
    hist = get_history(g.train_len + g.look_back + 10, '1d', 'close', g.security, fq='pre')
    
    if len(hist) < g.train_len:
        log.warning("历史数据不足，跳过训练")
        return

    # 数据预处理
    # PTrade get_history 返回 DataFrame，直接取 values
    # 兼容不同版本的返回值，确保取到的是 numpy array
    if isinstance(hist, pd.DataFrame):
        data = hist['close'].values.reshape(-1, 1)
    else:
        # 处理可能的 dict 返回或其他情况
        data = hist[g.security]['close'].values.reshape(-1, 1)

    # 归一化
    g.scaler.fit(data)
    scaled_data = g.scaler.transform(data)
    
    # 创建数据集
    train_x, train_y = create_dataset(scaled_data, g.look_back)
    
    # LSTM 输入格式: [samples, time steps, features]
    train_x = np.reshape(train_x, (train_x.shape[0], train_x.shape[1], 1))
    
    # 构建 LSTM 模型
    model = Sequential()
    # 输入层和第一个LSTM层
    model.add(LSTM(50, input_shape=(g.look_back, 1)))
    # 输出层
    model.add(Dense(1))
    
    model.compile(loss='mean_squared_error', optimizer='adam')
    
    # 训练模型
    # epochs=5, batch_size=32 是为了在回测中快速演示
    # 实盘建议 epochs=50+，并使用更多数据
    model.fit(train_x, train_y, epochs=5, batch_size=32, verbose=0)
    
    g.model = model
    g.is_trained = True
    log.info("LSTM 模型训练完成")

def before_trading_start(context, data):
    """
    盘前处理：如果模型未训练，则进行训练
    """
    if not g.is_trained:
        build_and_train_model(context)

def handle_data(context, data):
    """
    盘中逻辑：获取最新数据 -> 预处理 -> 预测 -> 交易
    """
    if not g.is_trained or g.model is None:
        return

    # 1. 获取用于预测的最近 look_back 天的数据
    # 注意：get_history(n) 获取的是截止到前一分钟/前一天的 n 条数据
    # 我们需要最近的 g.look_back 条数据来预测“明天”
    hist = get_history(g.look_back, '1d', 'close', g.security, fq='pre')
    
    if len(hist) < g.look_back:
        return
        
    if isinstance(hist, pd.DataFrame):
        recent_data = hist['close'].values.reshape(-1, 1)
    else:
        recent_data = hist[g.security]['close'].values.reshape(-1, 1)
        
    # 2. 数据归一化
    recent_scaled = g.scaler.transform(recent_data)
    
    # 3. 重塑形状 [1, look_back, 1]
    x_input = np.reshape(recent_scaled, (1, g.look_back, 1))
    
    # 4. 预测
    prediction_scaled = g.model.predict(x_input)
    
    # 5. 反归一化得到预测股价
    predicted_price = g.scaler.inverse_transform(prediction_scaled)[0][0]
    
    # 获取当前实际价格
    current_price = data[g.security]['close']
    
    log.info("当前价格: %.2f, LSTM预测明日价格: %.2f" % (current_price, predicted_price))
    
    # 6. 交易逻辑
    # 检查是否涨停或跌停，避免废单
    limit_info = check_limit(g.security)
    # check_limit 返回字典，如 {'600570.SS': 0}，0表示非涨跌停
    status = limit_info.get(g.security, 0)
    
    if status != 0:
        log.info("股票处于涨跌停状态，跳过交易")
        return

    # 获取当前持仓
    position = get_position(g.security).amount
    cash = context.portfolio.cash
    
    # 如果预测价格 > 当前价格 * (1 + 阈值)，且有资金 -> 买入
    if predicted_price > current_price * (1 + g.predict_threshold):
        if cash > 0:
            # 全仓买入
            order_value(g.security, cash)
            log.info("信号：看涨 (预测涨幅 > %.2f%%) -> 买入" % (g.predict_threshold * 100))
            
    # 如果预测价格 < 当前价格，且有持仓 -> 卖出
    elif predicted_price < current_price:
        if position > 0:
            # 清仓
            order_target(g.security, 0)
            log.info("信号：看跌 -> 卖出")

代码解析

1. 数据准备 (create_dataset): LSTM 需要特定形状的输入数据 [样本数, 时间步长, 特征数]。这里我们将时间序列转换为监督学习问题（用过去 10 天预测第 11 天）。
2. 模型构建 (build_and_train_model):
   • 使用 Sequential 建立序列模型。
   • 添加 LSTM 层，神经元数量设为 50。
   • 添加 Dense 层输出预测值。
   • 使用 MinMaxScaler 将股价缩放到 0-1 之间，这是神经网络训练的标准步骤。
3. 训练时机: 代码中在 before_trading_start 检查并训练模型。在回测中，这会在第一天交易前执行一次。
   • 优化建议：如果回测时间跨度很长，可以在 run_daily 中设置逻辑，每隔 N 个月重新训练一次模型以适应新行情。
4. 预测与交易 (handle_data):
   • 获取最近 look_back 天的数据。
   • 归一化后输入模型进行预测。
   • 将预测结果反归一化（inverse_transform）还原为真实价格。
   • 比较预测价与当前价，执行买卖操作。

常见问题 (Q&A)

Q: 为什么回测速度非常慢？
A: LSTM 的训练涉及大量的矩阵运算。PTrade 的回测环境通常使用 CPU 运行，且单核性能有限。建议在本地使用 GPU 训练好模型，保存为 .h5 文件，上传到 PTrade 的研究文件目录，然后在策略中使用 keras.models.load_model 加载使用。

Q: 报错 ImportError 或版本错误怎么办？
A: PTrade 的环境相对固定。如果遇到 Keras/Tensorflow 版本兼容性问题，请尽量使用最基础的 Keras API（如示例所示），避免使用 TF 2.x 特有的 Eager Execution 或高级 API。

Q: 预测结果准确吗？
A: 仅使用历史收盘价进行简单的 LSTM 预测，效果通常有限（容易出现滞后性，即预测曲线仅仅是真实曲线的平移）。在实际量化中，通常需要加入更多因子（如成交量、技术指标 MACD/RSI、大盘指数等）作为输入特征（Multivariate LSTM）来提高预测能力。