如何利用LSTM编写一个外汇EA程序(附示例代码)

2025-05-30   

为什么选 LSTM（而非 Transformer）：

模型结构建议

输入特征（可扩展）：

• Open / High / Low / Close / Volume

• 可添加技术指标（如 RSI, MACD, EMA）

输出头（Multi-task）：

• 分类头：Buy / Sell / Hold（ 入场信号）

• 回归头：StopLoss 值、TakeProfit 值

python复制编辑Model:
  - Shared LSTM layers
  - Output1: Dense(3) + Softmax → 交易动作
  - Output2: Dense(1) → 止损价位
  - Output3: Dense(1) → 止盈价位

# CFD 交易模型（XAUUSD）- 使用 LSTM 实现交易建议（入场、止盈、止损）

import numpy as np
import pandas as pd
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader

# ==========================
# 数据准备部分
# ==========================
class TradingDataset(Dataset):
    def __init__(self, df, window_size=60):
        self.window_size = window_size
        self.X = []
        self.y_signal = []  # 交易信号: 0 = Hold, 1 = Buy, 2 = Sell
        self.y_sl = []      # 止损价位（相对当前价格）
        self.y_tp = []      # 止盈价位

        for i in range(window_size, len(df) - 10):
            window = df.iloc[i-window_size:i]
            future = df.iloc[i:i+10]  # 用未来10根K线判断是否应买卖

            current_price = df.iloc[i]['close']
            max_future = future['high'].max()
            min_future = future['low'].min()

            # 简单交易信号逻辑
            if max_future >= current_price * 1.005:  # 涨了0.5%
                signal = 1  # Buy
                tp = current_price * 1.005
                sl = current_price * 0.995
            elif min_future <= current_price * 0.995:
                signal = 2  # Sell
                tp = current_price * 0.995
                sl = current_price * 1.005
            else:
                signal = 0  # Hold
                tp = current_price
                sl = current_price

            features = window[['open', 'high', 'low', 'close', 'volume']].values
            self.X.append(features)
            self.y_signal.append(signal)
            self.y_tp.append(tp)
            self.y_sl.append(sl)

        self.X = torch.tensor(np.array(self.X), dtype=torch.float32)
        self.y_signal = torch.tensor(self.y_signal, dtype=torch.long)
        self.y_tp = torch.tensor(self.y_tp, dtype=torch.float32)
        self.y_sl = torch.tensor(self.y_sl, dtype=torch.float32)

    def __len__(self):
        return len(self.X)

    def __getitem__(self, idx):
        return self.X[idx], self.y_signal[idx], self.y_tp[idx], self.y_sl[idx]


# ==========================
# LSTM 模型定义
# ==========================
class TradingLSTM(nn.Module):
    def __init__(self, input_size=5, hidden_size=64, num_layers=2):
        super(TradingLSTM, self).__init__()
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        self.fc_signal = nn.Linear(hidden_size, 3)     # 交易信号输出
        self.fc_tp = nn.Linear(hidden_size, 1)         # 止盈
        self.fc_sl = nn.Linear(hidden_size, 1)         # 止损

    def forward(self, x):
        _, (hn, _) = self.lstm(x)
        hn = hn[-1]  # 最后一层输出
        signal_out = self.fc_signal(hn)
        tp_out = self.fc_tp(hn).squeeze()
        sl_out = self.fc_sl(hn).squeeze()
        return signal_out, tp_out, sl_out


# ==========================
# 训练函数
# ==========================
def train_model(model, dataloader, epochs=10, lr=1e-3):
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    model = model.to(device)
    criterion_signal = nn.CrossEntropyLoss()
    criterion_price = nn.MSELoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=lr)

    for epoch in range(epochs):
        model.train()
        total_loss = 0
        for X, y_signal, y_tp, y_sl in dataloader:
            X, y_signal, y_tp, y_sl = X.to(device), y_signal.to(device), y_tp.to(device), y_sl.to(device)

            pred_signal, pred_tp, pred_sl = model(X)
            loss_signal = criterion_signal(pred_signal, y_signal)
            loss_tp = criterion_price(pred_tp, y_tp)
            loss_sl = criterion_price(pred_sl, y_sl)
            loss = loss_signal + 0.5 * (loss_tp + loss_sl)

            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

            total_loss += loss.item()

        print(f"Epoch {epoch+1}/{epochs} | Loss: {total_loss/len(dataloader):.4f}")


# ==========================
# 使用示例（需替换为你的 XAUUSD 数据）
# ==========================
# df = pd.read_csv("xauusd_data.csv")
# dataset = TradingDataset(df)
# dataloader = DataLoader(dataset, batch_size=32, shuffle=True)
# model = TradingLSTM()
# train_model(model, dataloader, epochs=20)

# 保存模型
# torch.save(model.state_dict(), "cfd_trading_model.pt")

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站文章有涉嫌抄袭侵权/违法违规的内容， 请发送邮件举报，一经查实，本站将立刻删除。