[文献精汇]使用 LSTM Networks 的均值回归交易策略

Backtrader 策略实例

[Backtrader]实例:均线策略[Backtrader] 实例:MACD策略[Backtrader] 实例:KDJ 策略
[Backtrader] 实例:RSI 与 EMA 结合
[Backtrader] 实例:SMA自定义数据源
[Backtrader] 实例:海龟策略[Backtrader] 实例:网格交易[Backtrader] 实例: 配对交
[Backtrader] 机器学习预测市场走势与回测

[介绍]([文献精汇]使用 LSTM Networks 的均值回归交易策略如何将均值回归理论与 LSTM 神经网络相结合，以创建一个交易策略。将讨论数据预处理、模型训练、策略实施和性能评估，从而避免数据泄露并使用有效的风险管理。

为什么选择均值回归和 LSTM？

了解均值回归

均值回归是一种金融策略，表明资产价格随着时间的推移趋向于其历史均值。市场参与者通过以下方式利用此策略：

在价格低于平均值时购买资产。
当价格超过均值时卖出。
这个策略假设与平均值的偏差是暂时的，随着时间的推移价格会恢复到平均值（Poterba & Summers， 1988）。这种现象在各种市场和资产类别中都观察到（Balvers et al.， 2000）。

了解长短期记忆（LSTM）

长短期记忆（LSTM）网络是专为序列数据设计的神经网络，例如时间序列金融数据（Hochreiter & Schmidhuber），1997）。他们擅长通过“记住”数据中的长期依赖关系来识别模式和预测未来的价格走势。以前的研究已经证明了LSTM在预测金融市场趋势方面的有效性（Fischer & Krauss，2018;Bao et al.， 2017）。

准备数据

下载加密货币数据

获取过去 7 年比特币的每日价格数据，确保我们的模型有足够的历史数据。

import datetime

ticker = 'BTC-USD'

end_date = datetime.datetime.now()

# set start_date 7 years back
start_date = end_date - datetime.timedelta(days=7*365)

start_date_str = start_date.strftime('%Y-%m-%d')
end_date_str = end_date.strftime('%Y-%m-%d')

# download data using yfinance with daily intervals
data = yf.download(
    tickers=ticker,
    start=start_date_str,
    end=end_date_str,
    interval='1d'
)
data.dropna(inplace=True)

if data.empty:
    raise ValueError("No data downloaded. Please check the ticker symbol and internet connection.")
else:
    print("Data downloaded successfully.")

数据预处理

计算了几个捕捉价格趋势和波动性的技术指标。包括MA20、MA50、Bollinger Band、RSI、MACD、Momentum、TR、ATR。

  # calculating moving averages
  data['MA20'] = data['Close'].rolling(window=20).mean()
  data['MA50'] = data['Close'].rolling(window=50).mean()

  # calculating Bollinger Bands
  data['STD'] = data['Close'].rolling(window=20).std()
  data['Upper_Band'] = data['MA20'] + (data['STD'] * 2.5)
  data['Lower_Band'] = data['MA20'] - (data['STD'] * 2.5)

  # calculating %B (Bollinger Band %)
  data['%B'] = (data['Close'] - data['Lower_Band']) / (data['Upper_Band'] - data['Lower_Band'])

  # calculating RSI
  delta = data['Close'].diff()
  up = delta.clip(lower=0)
  down = -1 * delta.clip(upper=0)
  roll_up = up.rolling(14).mean()
  roll_down = down.rolling(14).mean()
  RS = roll_up / roll_down
  data['RSI'] = 100.0 - (100.0 / (1.0 + RS))

  # calculating MACD and Signal Line
  exp1 = data['Close'].ewm(span=12, adjust=False).mean()
  exp2 = data['Close'].ewm(span=26, adjust=False).mean()
  data['MACD'] = exp1 - exp2
  data['Signal_Line'] = data['MACD'].ewm(span=9, adjust=False).mean()

  # calculating Momentum
  data['Momentum'] = data['Close'] - data['Close'].shift(10)

  # calculating Average True Range (ATR)
  data['TR'] = data[['High', 'Close']].max(axis=1) - data[['Low', 'Close']].min(axis=1)
  data['ATR'] = data['TR'].rolling(window=14).mean()

  # drop rows if they have NaN values
  data.dropna(inplace=True)

我们计算移动平均线、布林带、RSI、MACD、动量和 ATR 等技术指标。这些指标有助于捕捉趋势、动量和波动性，这对于价格预测和信号生成至关重要。

可视化技术指标

Visualizing Technical Indicators

该图显示了比特币的收盘价以及 20 天和 50 天移动平均线和布林带。这有助于可视化价格趋势和波动性。

准备技术指标数据

features = ['Close', '%B', 'RSI', 'MACD', 'Signal_Line', 'Momentum', 'ATR']

# feature scaling 
scaler = StandardScaler()
scaled_data = scaler.fit_transform(data[features])

def create_sequences(data, seq_length):
    X = []
    y = []
    for i in range(seq_length, len(data)):
        X.append(data[i - seq_length:i])
        y.append(data[i, 0])  
    return np.array(X), np.array(y)

seq_length = 60 

X, y = create_sequences(scaled_data, seq_length)

我们使用 60 天的序列长度来捕获每个预测的两个月的历史数据，从而平衡足够的历史背景和计算效率。

将数据拆分为训练集和测试集

我们将数据分为训练集（5 年）和测试集（2 年），确保不会发生数据泄露。

train_size = int(5 * 365)  # 5 years for training
test_size = int(2 * 365)   # 2 years for testing

X_train = X[:train_size]
y_train = y[:train_size]
X_test = X[train_size:train_size + test_size]
y_test = y[train_size:train_size + test_size]

通过将数据分为 5 年用于训练和 2 年用于测试，我们确保我们的模型在过去数据上进行训练，并在未来数据上进行测试，从而避免数据泄漏。

构建 LSTM 模型

model = Sequential()
model.add(LSTM(units=128, return_sequences=True, input_shape=(X_train.shape[1], X_train.shape[2]), kernel_regularizer=l2(0.001)))
model.add(Dropout(0.3))
model.add(LSTM(units=64, return_sequences=False, kernel_regularizer=l2(0.001)))
model.add(Dropout(0.3))
model.add(Dense(1))

model.compile(optimizer='adam', loss='mean_squared_error')

# early Stopping
early_stopping = EarlyStopping(monitor='val_loss', patience=5)

history = model.fit(
    X_train,
    y_train,
    epochs=100,
    batch_size=32,
    validation_data=(X_test, y_test),
    callbacks=[early_stopping]
)

我们的模型从一个具有 128 个单元的 LSTM 层开始，并使用 L2 正则化来防止过拟合。我们以 0.3 的速率对正则化应用 dropout。第二个具有 64 个单元的 LSTM 层紧随其后，捕获更抽象的模式。最后，我们有一个 dense 层来生成输出。

预测评估模型

我们对测试集进行预测，并使用 MAE、MSE 和 RMSE 评估模型的性能。

predictions = model.predict(X_test)

# error metrics
mae = mean_absolute_error(y_test, predictions)
mse = mean_squared_error(y_test, predictions)
rmse = np.sqrt(mse)

print(f"MAE: {mae}")
print(f"MSE: {mse}")
print(f"RMSE: {rmse}")

png

该模型在测试集上实现了 0.1483 的平均绝对误差（MAE）和 0.2217 的均方根误差（RMSE），表明预测性能合理。
我们还将预测转换回原始比例。

实施交易策略

根据模型的预测和技术指标实施均值回归交易策略。

test_data = data.iloc[train_size + seq_length:train_size + seq_length + test_size].copy()
test_data['Predicted_Close'] = predicted_close
test_data['Actual_Close'] = actual_close

# predicted changes calculations
test_data['Predicted_Change'] = (test_data['Predicted_Close'] - test_data['Actual_Close']) / test_data['Actual_Close']

# genereate trading signals based on adjusted strategy
test_data['Signal'] = 0

# adjust thresholds based on percentiles
rsi_buy_threshold = test_data['RSI'].quantile(0.4)
rsi_sell_threshold = test_data['RSI'].quantile(0.6)
predicted_change_buy_threshold = test_data['Predicted_Change'].quantile(0.6)
predicted_change_sell_threshold = test_data['Predicted_Change'].quantile(0.4)

# buy signal
test_data.loc[
    (test_data['Predicted_Change'] > predicted_change_buy_threshold) &
    (test_data['RSI'] < rsi_buy_threshold),
    'Signal'
] = 1

# sell signal
test_data.loc[
    (test_data['Predicted_Change'] < predicted_change_sell_threshold) &
    (test_data['RSI'] > rsi_sell_threshold),
    'Signal'
] = -1

# count the number of buy and sell signals
num_buy_signals = (test_data['Signal'] == 1).sum()
num_sell_signals = (test_data['Signal'] == -1).sum()

print(f"Number of Buy Signals: {num_buy_signals}")
print(f"Number of Sell Signals: {num_sell_signals}")

Number of Buy Signals: 104
Number of Sell Signals: 135

我们的策略在测试期间产生了 133 个买入信号和 142 个卖出信号。这表明该模型确定了资产相对于其预测均值被低估或高估的几个机会。

模拟交易

以 500 美元的初始资本模拟交易，包括交易成本、止损和止盈机制。

initial_capital = 500.0
positions = []
cash = initial_capital
holdings = 0
portfolio_value = []
transaction_cost = 0.0005  # let's assume 0.05% trading fee per trade
stop_loss_percent = 0.1    # 10% stop-loss
take_profit_percent = 0.2  # 20% take-profit
entry_price = None

for index, row in test_data.iterrows():
    price = row['Actual_Close']
    signal = row['Signal']
    if signal == 1 and cash > 0:
        # buy with a portion of cash (e.g., 50%)
        amount_to_buy = (cash * 0.5) * (1 - transaction_cost)
        holdings += amount_to_buy / price
        cash -= amount_to_buy
        entry_price = price
        positions.append({'Date': index, 'Position': 'Buy', 'Price': price})
    elif signal == -1 and holdings > 0:
        # sell all holdings
        amount_to_sell = holdings * price * (1 - transaction_cost)
        cash += amount_to_sell
        holdings = 0
        entry_price = None
        positions.append({'Date': index, 'Position': 'Sell', 'Price': price})
    elif holdings > 0:
        # check for stop-loss or take-profit
        if price <= entry_price * (1 - stop_loss_percent):
            # trigger stop-loss
            amount_to_sell = holdings * price * (1 - transaction_cost)
            cash += amount_to_sell
            holdings = 0
            positions.append({'Date': index, 'Position': 'Stop Loss Sell', 'Price': price})
            entry_price = None
        elif price >= entry_price * (1 + take_profit_percent):
            # trigger take-profit
            amount_to_sell = holdings * price * (1 - transaction_cost)
            cash += amount_to_sell
            holdings = 0
            positions.append({'Date': index, 'Position': 'Take Profit Sell', 'Price': price})
            entry_price = None
    total_value = cash + holdings * price
    portfolio_value.append(total_value)

# musst ensure portfolio_value matches test_data length
test_data['Portfolio_Value'] = portfolio_value[:len(test_data)]

绩效指标

# calculate daily returns and cumulative returns
test_data['Daily_Return'] = test_data['Portfolio_Value'].pct_change()
test_data['Cumulative_Return'] = (1 + test_data['Daily_Return']).cumprod()

# calculate annualized return
total_days = (test_data.index[-1] - test_data.index[0]).days
if total_days == 0:
    total_days = 1  # Avoid division by zero

annualized_return = (test_data['Cumulative_Return'].iloc[-1]) ** (365 / total_days) - 1

# calculate Sharpe Ratio
returns = test_data['Daily_Return'].dropna()
if returns.std() != 0:
    sharpe_ratio = (returns.mean() / returns.std()) * np.sqrt(252)  # Annualized Sharpe Ratio
else:
    sharpe_ratio = 0.0

# calculate Max Drawdown
rolling_max = test_data['Portfolio_Value'].cummax()
drawdown = test_data['Portfolio_Value'] / rolling_max - 1
max_drawdown = drawdown.min()

# Print performance metrics
total_return = ((test_data['Portfolio_Value'].iloc[-1] - initial_capital) / initial_capital) * 100
print(f"Total Return: {total_return:.2f}%")
print(f"Annualized Return: {annualized_return * 100:.2f}%")
print(f"Sharpe Ratio: {sharpe_ratio:.2f}")
print(f"Max Drawdown: {max_drawdown * 100:.2f}%")

Total Return: 49.05%
Annualized Return: 26.49%
Sharpe Ratio: 0.80
Max Drawdown: -19.67%

我们的策略在两年的测试期内实现了 60.20% 的总回报率，将最初的 500 美元增长到大约 800 美元。年化回报率为 32.09%，这是可观的。0.94 的夏普比率表明相对于所承担的风险而言，回报良好。-16.88% 的最大回撤显示了在此期间最大的峰谷下跌，考虑到回报，这是可以接受的。

可视化交易信号

在 K 线图上绘制买入和卖出信号，以便更好地可视化。

# candlestick plotting
plot_data = data.loc[test_data.index][['Open', 'High', 'Low', 'Close']].copy()
plot_data.index.name = 'Date'

# buy and sell signal markers for plotting
buy_signals = test_data[test_data['Signal'] == 1]
sell_signals = test_data[test_data['Signal'] == -1]

png

投资组合值随时间的变化

可视化投资组合价值随时间的变化，以观察策略的表现。

plt.figure(figsize=(12, 6))
plt.plot(test_data.index, test_data['Portfolio_Value'], label='Portfolio Value')
plt.title('Portfolio Value Over Time')
plt.xlabel('Date')
plt.ylabel('Portfolio Value in USD')
plt.legend()
plt.show()

png

评估性能

# performance Periods analysis
test_data['Strategy_Return'] = test_data['Portfolio_Value'].pct_change()
test_data['Rolling_Return'] = test_data['Strategy_Return'].rolling(window=30).sum()

# periods of good performance
good_performance = test_data[test_data['Rolling_Return'] > 0.02]

# Periods of poor performance
poor_performance = test_data[test_data['Rolling_Return'] < -0.02]

# Compare our strategy with Buy-and-Hold Strategy
test_data['Buy_and_Hold'] = initial_capital * (test_data['Actual_Close'] / test_data['Actual_Close'].iloc[0])