简介:本资源为比特币基于LSTM的多因子交易策略Python实现,面向计算机、人工智能、金融工程等专业的在校学生、教师及企业员工,适合作为毕业设计、课程设计或量化交易入门项目。策略通过多因子输入与LSTM预测降低最大回撤和波动,整体较为稳健,收益率与预测准确率直接相关,可进一步加入更多因子、调整超参或尝试集成学习。压缩包共12个文件,约797KB,包含4个CSV行情数据、3个XML配置、2个Jupyter Notebook源码、1个PNG图表、1个Markdown说明及IDE工程文件,结构清晰便于复现。已有263人学习下载。代码均经测试运行成功,答辩评审平均分达96分,读者可获取完整源码、数据与文档说明,快速理解LSTM多因子交易流程,并在此基础上修改扩展,用于毕设、课设或项目立项演示。
1. 多因子 LSTM 交易策略:为什么你的模型总在实盘里翻车
很多人第一次跑通 LSTM 预测比特币价格时都会经历同一个幻觉:训练集 loss 降到 0.001,验证集看着也不错,一上模拟盘就亏得怀疑人生。问题几乎从来不在 LSTM 本身,而在于你喂给它的东西太单薄——只用收盘价做单因子输入,模型学到的不过是「昨天涨今天大概率还涨」这种在震荡市里必死的规律。多因子交易策略的核心思路,是把价格之外的市场状态一起编码进特征向量:动量、波动率、成交量结构、资金费率偏离、链上活跃度变化率等等,让 LSTM 在时间维度上学习这些因子之间的非线性交互,而不是死记价格序列。
这套方案适合两类人:一是已经会用 pandas 处理 K 线、但模型效果始终不稳定的量化新手;二是想从传统机器学习(XGBoost、随机森林)迁移到序列建模的从业者。它不解决「预测明天涨跌」这种伪命题,而是解决「在给定多因子状态下,未来 N 根 K 线的收益分布偏向哪一侧」这个可建模的问题。下面从数据、特征、模型、回测四个环节拆开讲,每一步都给出可复现的代码和参数依据。
2. 多因子特征工程:把 K 线变成 LSTM 能吃的张量
2.1 因子选型:为什么是这五类而不是随便凑十个
因子不是越多越好。LSTM 的输入维度每增加一维,需要的训练样本量就指数级上升,而比特币的日线数据从 2013 年算起也不过四千多根,4 小时线多一些但噪声更大。我一般会控制在 8~15 个因子之间,按五类来选:
- 动量类:过去 1、3、7、14 根 K 线的收益率。捕捉短期趋势延续和反转。
- 波动率类:ATR(平均真实波幅)和布林带宽度。波动率扩张和收缩对应不同的市场状态。
- 成交量类:成交量变化率、买卖量比(taker buy/sell volume ratio)。量价背离是反转的早期信号。
- 资金面类:永续合约资金费率、未平仓合约量变化率。这两个因子在现货数据里没有,需要从衍生品接口单独拉。
- 链上类:活跃地址数变化率、交易所净流入流出量。链上数据更新频率低,做日线级别策略时用前一天的数值,避免未来函数。
选因子的原则是:每个因子都要有明确的经济含义,且因子之间相关性不能太高。我一般会算一个因子相关性矩阵,两两相关系数超过 0.85 的就砍掉一个,保留逻辑更直接的那个。
2.2 数据对齐与缺失值处理:时间戳对不齐是万恶之源
多因子最大的坑不在模型,在数据对齐。现货 K 线是 UTC 0 点收盘,资金费率每 8 小时结算一次,链上数据可能是 UTC 0 点快照但 API 返回有延迟。如果直接按索引 merge,会出现大量 NaN 或者更隐蔽的错位——今天的资金费率被错误地拼到了昨天的 K 线上。
我的做法是统一转成 UTC 时间戳,按小时重采样,然后用前向填充(forward fill)处理低频因子,但限制最多填充 3 个周期,超过就标记为缺失。下面是对齐代码:
import pandas as pd import numpy as np def align_factors(price_df, funding_df, onchain_df, freq='4H'): """ price_df: 现货K线,columns=[open, high, low, close, volume],index为UTC时间戳 funding_df: 资金费率,columns=[funding_rate, open_interest],index为UTC时间戳 onchain_df: 链上数据,columns=[active_addr_chg, exchange_netflow],index为UTC时间戳 freq: 重采样频率,默认4小时 """ # 统一重采样到目标频率,取每个周期的最后一条有效值 price_res = price_df.resample(freq).agg({ 'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum' }) funding_res = funding_df.resample(freq).last() onchain_res = onchain_df.resample(freq).last() # 合并,链上数据允许前向填充最多3个周期 df = price_res.join(funding_res, how='left').join(onchain_res, how='left') df['active_addr_chg'] = df['active_addr_chg'].ffill(limit=3) df['exchange_netflow'] = df['exchange_netflow'].ffill(limit=3) # 删除仍然缺失的行,并打印缺失比例 missing_ratio = df.isna().mean() print(f"缺失比例:\n{missing_ratio[missing_ratio > 0]}") df = df.dropna() return df逻辑说明:resample(freq).last()取每个周期最后一个值,是因为资金费率和链上数据是状态量而非累积量。ffill(limit=3)限制填充长度,防止用太旧的数据掩盖真实缺失。最后打印缺失比例,如果某个因子缺失超过 10%,说明数据源有问题,需要换源而不是硬填。
参数方面,freq的选择取决于你的持仓周期。做日内短线用 1H,波段用 4H,趋势跟踪用 1D。我实测下来 4H 在比特币上信噪比最平衡,1H 噪声太大,1D 样本又太少。
2.3 因子标准化与标签构造:别让量纲差异毁掉梯度
成交量动辄几十万,收益率在 -0.05 到 0.05 之间,直接喂给 LSTM 会导致梯度被大量纲因子主导。标准化用滚动窗口的 z-score,而不是全局标准化——全局标准化会引入未来信息,因为均值和标准差是用全量数据算的。
def rolling_zscore(series, window=200): """滚动z-score,window为回看周期数""" mean = series.rolling(window).mean() std = series.rolling(window).std() return (series - mean) / (std + 1e-8) def build_features_labels(df, lookback=48, horizon=6): """ lookback: 用过去多少根K线做输入 horizon: 预测未来多少根K线的累计收益方向 """ # 构造因子列 df['ret_1'] = df['close'].pct_change(1) df['ret_3'] = df['close'].pct_change(3) df['ret_7'] = df['close'].pct_change(7) df['ret_14'] = df['close'].pct_change(14) df['atr'] = (df['high'] - df['low']).rolling(14).mean() / df['close'] df['vol_chg'] = df['volume'].pct_change(1) df['bb_width'] = (df['close'].rolling(20).std() * 2) / df['close'].rolling(20).mean() factor_cols = ['ret_1', 'ret_3', 'ret_7', 'ret_14', 'atr', 'vol_chg', 'bb_width', 'funding_rate', 'open_interest', 'active_addr_chg', 'exchange_netflow'] # 对每个因子做滚动标准化 for col in factor_cols: df[col + '_z'] = rolling_zscore(df[col], window=200) z_cols = [c + '_z' for c in factor_cols] # 标签:未来horizon根K线的累计收益,三分类 df['future_ret'] = df['close'].shift(-horizon) / df['close'] - 1 df['label'] = pd.cut(df['future_ret'], bins=[-np.inf, -0.01, 0.01, np.inf], labels=[0, 1, 2]) # 0跌 1震荡 2涨 # 构造LSTM输入张量 X, y = [], [] for i in range(lookback, len(df) - horizon): X.append(df[z_cols].iloc[i-lookback:i].values) y.append(df['label'].iloc[i]) return np.array(X), np.array(y), z_cols逻辑说明:rolling_zscore的窗口设为 200,对应 4H 频率下约 33 天,足够覆盖一个完整的市场情绪周期。标签用三分类而不是回归,是因为收益率的绝对值预测极不稳定,但方向分类的稳定性好得多。bins的阈值 ±1% 是经验值,在 4H 频率下大约对应 1.5 倍 ATR,能过滤掉大量微小波动带来的噪声标签。
注意:shift(-horizon)构造标签时,最后 horizon 行没有标签,循环里用len(df) - horizon截断,避免用未来数据。这个细节如果漏掉,回测收益会虚高得离谱。
3. LSTM 模型搭建与训练:层数、Dropout 和学习率怎么定
3.1 网络结构:两层 LSTM 加注意力池化就够了
比特币价格序列的信噪比很低,堆太深的网络只会更快过拟合。我试过 1 到 4 层 LSTM,2 层在验证集上的表现最稳。每层隐藏单元 64,输入维度等于因子数量(上面例子是 11),输出接一个注意力池化层而不是直接取最后一个时间步——因为最后一个时间步未必包含最关键的信息,注意力机制让模型自己决定看哪几个时间点。
import torch import torch.nn as nn class AttentionPool(nn.Module): def __init__(self, hidden_dim): super().__init__() self.attn = nn.Linear(hidden_dim, 1) def forward(self, lstm_out): # lstm_out: (batch, seq_len, hidden_dim) weights = torch.softmax(self.attn(lstm_out), dim=1) # (batch, seq_len, 1) pooled = (lstm_out * weights).sum(dim=1) # (batch, hidden_dim) return pooled class MultiFactorLSTM(nn.Module): def __init__(self, input_dim, hidden_dim=64, num_layers=2, dropout=0.3, num_classes=3): super().__init__() self.lstm = nn.LSTM( input_size=input_dim, hidden_size=hidden_dim, num_layers=num_layers, batch_first=True, dropout=dropout if num_layers > 1 else 0 ) self.attn_pool = AttentionPool(hidden_dim) self.classifier = nn.Sequential( nn.Linear(hidden_dim, 32), nn.ReLU(), nn.Dropout(dropout), nn.Linear(32, num_classes) ) def forward(self, x): lstm_out, _ = self.lstm(x) pooled = self.attn_pool(lstm_out) return self.classifier(pooled)逻辑说明:dropout=0.3是循环网络里的常用值,再高会导致欠拟合。num_layers=2时 LSTM 内部会自动在层间加 dropout,但第一层和第二层之间不加,这是 PyTorch 的默认行为。注意力池化用单层线性映射加 softmax,参数量极小,不会显著增加过拟合风险。
参数方面,hidden_dim=64对应 11 维输入已经足够,如果因子数量增加到 20 以上,可以提到 96 或 128。num_classes=3对应跌、震荡、涨三分类。
3.2 训练流程:时间序列交叉验证与早停
不能用随机划分训练集和验证集,因为时间序列的样本不独立。正确做法是按时间顺序切分,比如前 70% 训练,中间 15% 验证,最后 15% 测试。更严格的做法是滚动窗口交叉验证,但计算量大,日常调参用固定切分就够了。
from torch.utils.data import DataLoader, TensorDataset from sklearn.metrics import f1_score def train_model(X, y, input_dim, epochs=100, batch_size=64, lr=1e-3, patience=10): # 按时间顺序切分 n = len(X) train_end = int(n * 0.7) val_end = int(n * 0.85) X_train, y_train = X[:train_end], y[:train_end] X_val, y_val = X[train_end:val_end], y[train_end:val_end] X_test, y_test = X[val_end:], y[val_end:] # 转成tensor train_ds = TensorDataset(torch.FloatTensor(X_train), torch.LongTensor(y_train)) val_ds = TensorDataset(torch.FloatTensor(X_val), torch.LongTensor(y_val)) test_ds = TensorDataset(torch.FloatTensor(X_test), torch.LongTensor(y_test)) train_loader = DataLoader(train_ds, batch_size=batch_size, shuffle=True) val_loader = DataLoader(val_ds, batch_size=batch_size, shuffle=False) test_loader = DataLoader(test_ds, batch_size=batch_size, shuffle=False) model = MultiFactorLSTM(input_dim=input_dim) optimizer = torch.optim.Adam(model.parameters(), lr=lr, weight_decay=1e-4) scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=5, factor=0.5) criterion = nn.CrossEntropyLoss() best_val_f1 = 0 best_state = None wait = 0 for epoch in range(epochs): model.train() for xb, yb in train_loader: optimizer.zero_grad() loss = criterion(model(xb), yb) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() # 验证 model.eval() val_preds, val_trues = [], [] with torch.no_grad(): for xb, yb in val_loader: preds = model(xb).argmax(dim=1) val_preds.extend(preds.numpy()) val_trues.extend(yb.numpy()) val_f1 = f1_score(val_trues, val_preds, average='macro') scheduler.step(1 - val_f1) if val_f1 > best_val_f1: best_val_f1 = val_f1 best_state = model.state_dict().copy() wait = 0 else: wait += 1 if wait >= patience: print(f"早停于 epoch {epoch},最佳验证F1={best_val_f1:.4f}") break model.load_state_dict(best_state) # 测试集评估 model.eval() test_preds, test_trues = [], [] with torch.no_grad(): for xb, yb in test_loader: preds = model(xb).argmax(dim=1) test_preds.extend(preds.numpy()) test_trues.extend(yb.numpy()) test_f1 = f1_score(test_trues, test_preds, average='macro') print(f"测试集F1={test_f1:.4f}") return model, test_f1逻辑说明:clip_grad_norm_(max_norm=1.0)是 LSTM 训练的标配,防止梯度爆炸。weight_decay=1e-4是 L2 正则,配合 dropout 一起抑制过拟合。ReduceLROnPlateau在验证 F1 不再提升时把学习率减半,比固定学习率更容易收敛到好的局部解。早停的patience=10意味着连续 10 个 epoch 验证 F1 没提升就停,这个值在 4H 数据上大约对应 2 到 3 天的训练时间。
注意:shuffle=True只用在训练集,验证集和测试集必须保持时间顺序,否则评估结果没有意义。
3.3 类别不平衡处理:震荡样本太多怎么办
三分类里「震荡」样本通常占 50% 以上,模型会倾向于全预测震荡来刷准确率。解决办法是在损失函数里给少数类更高权重:
# 计算类别权重 from sklearn.utils.class_weight import compute_class_weight class_weights = compute_class_weight('balanced', classes=np.unique(y_train), y=y_train) class_weights = torch.FloatTensor(class_weights) criterion = nn.CrossEntropyLoss(weight=class_weights)compute_class_weight('balanced')会自动按类别频率的倒数赋权,让模型对跌和涨的样本更敏感。这个改动通常能把涨跌两类的召回率提升 10 到 15 个百分点,代价是震荡类的精确率略降,但交易策略里抓住趋势比过滤震荡更重要。
4. 回测与信号执行:从模型输出到仓位管理
4.1 信号生成:概率阈值比 argmax 更可控
模型输出的是三个类别的概率,直接取 argmax 会在概率接近时频繁翻转仓位,手续费吃掉大部分利润。我一般设两个阈值:涨的概率超过 0.55 才做多,跌的概率超过 0.55 才做空,否则空仓。阈值可以通过验证集上的收益曲线来调,0.5 到 0.6 之间通常有一个夏普比率的高点。
def generate_signals(model, X_test, threshold=0.55): model.eval() with torch.no_grad(): probs = torch.softmax(model(torch.FloatTensor(X_test)), dim=1).numpy() signals = np.zeros(len(probs)) signals[probs[:, 2] > threshold] = 1 # 做多 signals[probs[:, 0] > threshold] = -1 # 做空 return signals, probs逻辑说明:probs[:, 2]是涨的概率,probs[:, 0]是跌的概率。两个条件互斥,因为 softmax 输出不会同时让两个类别都超过 0.55(除非阈值设得很低)。信号为 0 表示空仓,这在震荡市里能避免大量无效交易。
4.2 回测框架:手续费、滑点和资金曲线
回测必须扣手续费和滑点,否则结果毫无参考价值。比特币永续合约的 taker 手续费大约 0.05%,滑点按 0.02% 算,每次开平仓合计成本约 0.14%。下面是一个简化的向量化回测:
def backtest(signals, close_prices, fee=0.0005, slippage=0.0002): """ signals: 仓位信号,1做多 -1做空 0空仓 close_prices: 对应周期的收盘价 """ returns = np.diff(close_prices) / close_prices[:-1] # 信号滞后一根K线执行,避免未来函数 position = signals[:-1] strategy_ret = position * returns # 计算换手成本 turnover = np.abs(np.diff(np.concatenate([[0], position]))) cost = turnover * (fee + slippage) strategy_ret_net = strategy_ret - cost # 资金曲线 equity = np.cumprod(1 + strategy_ret_net) sharpe = np.mean(strategy_ret_net) / (np.std(strategy_ret_net) + 1e-8) * np.sqrt(365 * 6) # 4H频率年化 return equity, sharpe逻辑说明:position = signals[:-1]是关键,信号基于当前 K 线收盘生成,但执行要在下一根 K 线开盘,所以收益计算要用下一根 K 线的收益率。turnover计算仓位变化次数,每次变化扣一次成本。np.sqrt(365 * 6)是 4H 频率的年化因子,因为一天有 6 个 4 小时周期。
参数方面,fee=0.0005对应 taker 费率,如果你用 maker 挂单可以降到 0.0002,但成交概率会下降。slippage=0.0002是保守估计,大资金需要调高。
4.3 仓位管理:固定比例还是波动率倒数
固定比例仓位(比如每次用 10% 资金)简单但不够精细。更好的做法是按波动率倒数分配仓位:波动率高时减仓,波动率低时加仓,让每笔交易的风险贡献大致相等。
def vol_target_position(signals, atr, target_vol=0.02): """ signals: 原始信号 atr: 对应周期的ATR值 target_vol: 目标波动率,默认2% """ position_size = target_vol / (atr + 1e-8) position_size = np.clip(position_size, 0, 1) # 限制最大仓位为1 return signals * position_sizetarget_vol=0.02表示每根 K 线愿意承担 2% 的波动,这个值根据个人风险偏好调整。np.clip防止低波动时仓位超过 100%。
5. 避坑指南:多因子 LSTM 策略最常见的五个翻车点
5.1 未来函数:回测夏普 3.0,实盘亏成狗
现象:回测资金曲线平滑向上,夏普比率超过 3,实盘一跑就连续亏损。
原因:最常见的是标准化用了全量数据的均值和标准差,或者标签构造时shift方向写反,或者因子计算里用了rolling但没加.shift(1)。另一个隐蔽的是链上数据的时间戳——API 返回的是数据采集时间,但实际链上确认时间可能晚几个小时,如果直接按采集时间对齐,相当于用了未来数据。
解决:所有滚动计算后统一加.shift(1),确保第 t 根 K 线只能看到 t-1 及之前的信息。标准化用滚动窗口而不是全局。链上数据至少滞后一个周期使用。回测前用一个小脚本检查:把因子矩阵整体后移一根 K 线,如果回测收益变化超过 20%,说明有未来函数。
5.2 过拟合:验证集 F1 0.65,测试集掉到 0.45
现象:训练过程中验证集 F1 能到 0.65 以上,但测试集只有 0.45 左右,实盘更差。
原因:模型太复杂或者训练太久。LSTM 参数量相对于比特币的数据量来说很容易过大,两层 64 单元的 LSTM 大约有 5 万参数,而有效样本可能只有几千个。另外,如果因子之间高度相关,模型会记住噪声而不是规律。
解决:先减层数,从 2 层降到 1 层试试。再减隐藏单元,从 64 降到 32。增加 dropout 到 0.4 或 0.5。减少因子数量,用相关性矩阵砍掉冗余因子。如果这些都不行,说明数据量本身不够,考虑用更高频率的数据或者换更简单的模型(比如 GRU 或一维卷积)。
5.3 数据泄漏:因子和标签用了同一根 K 线的收盘价
现象:模型在训练集上准确率 90% 以上,但样本外完全随机。
原因:构造标签时用了close.shift(-horizon),但因子计算里也用了close的当前值,而当前值在实盘时是未知的。比如ret_1 = close.pct_change(1)用的是当前收盘价和上一根收盘价,这个在收盘那一刻才知道,但如果你在收盘前就生成信号,就用了未来数据。
解决:明确信号生成的时间点。如果是在每根 K 线收盘后生成信号、下一根开盘执行,那么因子可以用当前收盘价。但如果是在 K 线内实时生成信号,因子必须全部滞后一根。回测里统一用「收盘生成、下根开盘执行」的模式,并在代码里显式写出position = signals[:-1]。
5.4 手续费吞噬利润:高频翻转导致净值归零
现象:策略毛收益为正,但扣掉手续费后资金曲线持续向下。
原因:模型在概率接近 0.5 时频繁翻转仓位,每次翻转都扣手续费和滑点。4H 频率下如果每根 K 线都换仓,一年换手 2000 多次,成本超过 100%。
解决:设概率阈值,只在置信度高时开仓。加一个最小持仓周期,比如开仓后至少持有 3 根 K 线才能平仓。用 maker 挂单代替 taker 吃单,手续费从 0.05% 降到 0.02%。回测里把手续费设得比实际高一点(比如 0.07%),留出安全边际。
5.5 市场状态切换:训练集是牛市,测试集是熊市
现象:模型在 2020 到 2021 年数据上表现很好,但 2022 年完全失效。
原因:LSTM 学到的是特定市场状态下的模式,牛市里动量因子有效,熊市里反转因子有效。如果训练集只覆盖一种状态,模型无法泛化。
解决:训练集必须包含至少一个完整的牛熊周期。比特币的周期大约 4 年,所以训练数据至少要从 2018 年之前开始。如果数据不够,可以用数据增强——对收益率序列做随机缩放和加噪,生成不同波动率状态下的合成样本。另外,在因子里加入市场状态标识,比如用 200 日移动均线判断牛熊,让模型自己学习状态切换。
6. 进阶技巧:用滚动重训练和集成推理稳住实盘表现
模型训练完不是终点。比特币市场结构在变,2021 年的因子有效性到 2024 年可能完全反转。我现在的做法是每 3 个月用最近 2 年的数据重新训练一次模型,但保留旧模型,推理时用三个不同时间窗口训练的模型做集成——比如一个用最近 1 年数据、一个用最近 2 年、一个用最近 3 年,三个模型的概率取平均后再生成信号。这样做的逻辑是:短窗口模型对近期状态敏感,长窗口模型更稳定,集成后能平滑掉单模型的状态偏移。
具体实现上,维护一个模型池,每个模型记录训练数据的截止日期。推理时加载所有模型,对同一批输入分别前向传播,然后对 softmax 概率做算术平均:
def ensemble_predict(models, x): """models: 模型列表,x: 输入张量""" probs_list = [] for model in models: model.eval() with torch.no_grad(): probs = torch.softmax(model(x), dim=1) probs_list.append(probs) avg_probs = torch.stack(probs_list).mean(dim=0) return avg_probs另一个技巧是动态阈值。固定 0.55 的阈值在波动率高的时期会错过趋势,在波动率低的时期会频繁开仓。我一般用波动率分位数来调阈值:当前 ATR 处于过去 200 根 K 线的 80% 分位以上时,阈值降到 0.52;低于 20% 分位时,阈值提到 0.60。这个调整能让策略在趋势行情里更激进,在震荡行情里更保守。
最后说一个我踩过的坑:不要用测试集调任何参数。测试集只能用一次,看完结果就封存。如果调完参数再看测试集,测试集就变成了验证集,你永远不知道真实样本外的表现。我现在的习惯是留最后 6 个月的数据完全不动,等模型和参数全部定下来之后,只跑一次,结果无论好坏都接受。这个习惯让我避免了很多次「回测很美、实盘很惨」的循环。希望帮到你。
本文还有配套的精品资源,点击获取