“该策略针对加密货币市场，选择了经过两年回测表现最好的前10个山寨币进行交易。通过比特币的长信号（对数回报率超过0.98%）触发市场买单，并结合市场滚动标准差设置限价和止损单，持仓最多4小时。”
资产类别：加密货币 | 地区：全球 | 频率：日内 | 市场：加密货币 | 关键词：比特币，山寨币
策略概述
投资范围包括40个山寨币（加密货币）市场[初始样本期内的子集]。从回测结果中选择表现最好的前10个山寨币作为最终投资组合[初始（训练）期]，并仅在[交易（执行）期]交易这些山寨币。（作者进行了为期两年的回测。） （价格数据的主要来源为币安，历史市值快照可从CoinMarketCap获取。）
遵循第13页的算法1 BTC引领策略： 在进行初始回测以识别盈利币种后，每当观察到比特币的长信号时[当15分钟内比特币的对数回报率超过98百分位，对应+0.98%的回报门槛]，算法将对所有监控的十个山寨币下达市场买单。 随后，算法同时根据市场的价格滚动标准差[回顾期𝑤]设置限价单作为获利目标，并下达止损单。 需遵守四个参数：滚动回顾窗口𝑤 = 24（6小时）；获利限价单宽度𝛼 = 2.0[计算实际值时将alpha乘以滚动价格标准差]；止损市场单宽度𝛽 = 1.25[计算实际值时将beta乘以滚动价格标准差]；最大持仓时间𝑘 = 4小时[参数选择表明大多数山寨币将在一小时内响应比特币信号，如果没有，在下一个价格蜡烛收盘时退出]。
这是一个日内交易策略，因此根据信号动态再平衡。（作为最佳实践，建议将总交易资金的最多10%分配给此类高风险交易策略。）
策略合理性
显然，加密货币市场高度集中，这表明大多数山寨币在试图与比特币区分时出现了定价错误和市场效率低下。这项研究支持了先前关于比特币与山寨币之间高度相关性的研究。作者的统计分析表明，比特币价格的主要跳跃导致了市值排名前40的山寨币市场中的最小-最大分布发生显著变化。此外，他们通过经验交易算法BLUTA探索了比特币看涨信号之后在山寨币市场中的领涨滞后关系。BLUTA作为从当前市场效率低下中提取alpha的概念验证。在未来，还可以探索比特币与其他资产类别（如股票指数）之间的因果关系，以解释山寨币价格波动的根本驱动因素。比特币的主要驱动力来自投资者情绪和区块链网络数据。比特币强劲的看跌动能的影响尚待研究。
论文来源
The Pricing Myth of Altcoins: Statistical and Empirical Evidence of Market Consolidation Driven by Short-Term Bitcoin Momentum [点击浏览原文]
- Jingyuan Tian, Ilya Mikhelson, 西北大学
<摘要>
加密货币市场表现出快速增长，截至2021年11月，总市值达到3万亿美元，截至2022年6月已有超过10,000种山寨币。比特币（BTC）截至2022年初占市场份额的40%以上。山寨币的定价仍然是一个谜。我们展示了币安上40个大盘山寨币与比特币之间的高度相关性，尽管大多数山寨币声称在区块链生态系统中具有不同的功能。除了长期相关性，本文还探讨了比特币动能的短期影响。统计证据表明，比特币上涨后出现了山寨币价格分布的暂时市场整合。我们设计了一个经验交易框架——比特币引领通用交易算法（BLUTA），该算法从市场整合和比特币主导地位中提取了一致的alpha。BLUTA显著优于一系列宏观基准和其他交易策略。
回测表现
| 年化收益率 | 10.21% |
| 波动率 | 3.36% |
| Beta | 0.047 |
| 夏普比率 | 3.06 |
| 索提诺比率 | N/A |
| 最大回撤 | N/A |
| 胜率 | 39% |
完整python代码
from AlgorithmImports import *
import data_tools
from dateutil.relativedelta import relativedelta
import numpy as np
# endregion
class BitcoinLeadsAltcoinsonIntradayBasis(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2017, 1, 1)
self.SetCash(100000)
self.period:int = 24
self.alpha:float = 2.0
self.beta:float = 1.25
self.percentile:float = 98
self.top_count:int = 10
self.max_traded_duration:int = 4
self.portion:float = .1
self.consolidation_bar_count:int = 15
self.crypto_tickers:List[str] = ['BTCUSD', 'ETHUSD', 'SOLUSD', 'ADAUSD', 'XRPUSD', 'DOTUSD', 'DOGEUSD', 'LUNAUSD', 'AVAXUSD', 'UNIUSD',
'LINKUSD', 'LTCUSD', 'BCHABCUSD', 'BSVUSD', 'FILUSD', 'XLMUSD', 'XTZUSD', 'NEOUSD', 'ATOMUSD', 'IOTAUSD',
'ETCUSD', 'DASHUSD', 'EGLDUSD', 'AAVEUSD', 'ENJUSD', 'EOSUSD', 'MKRUSD', 'MANAUSD', 'SNXUSD',
'OMGUSD', 'SUSHIUSD', 'YFIUSD', 'WBTCUSD', 'XMRUSD', 'ZECUSD', 'ZRXUSD', 'XRAUSD', 'AMPLUSD', 'GRTUSD',
'DGBUSD', '1INCHUSD'] # 'FTTUSD'
self.data:Dict[Symbol, SymbolData] = {}
self.btc_returns:List[float] = []
self.symbol_performance:Dict[Symbol, float] = {}
self.top_perf_symbols:List[Symbol] = []
# data subscription
for ticker in self.crypto_tickers:
data = self.AddCrypto(ticker, Resolution.Minute, Market.Bitfinex)
data.SetFeeModel(data_tools.CustomFeeModel())
self.Consolidate(data.Symbol, timedelta(minutes=self.consolidation_bar_count), self.ConsolidatedBarHandler)
self.data[data.Symbol] = data_tools.SymbolData(self.period)
self.trade_flag:bool = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.current_year:int = -1
def OnData(self, data: Slice):
if self.trade_flag:
self.trade_flag = False
# pick top 10 performing cryptos each year
if self.Time.year == self.current_year:
return
self.current_year = self.Time.year
self.cancel_open_orders(None)
self.top_perf_symbols.clear()
self.btc_returns = self.btc_returns[-self.period:]
top_perf_symbols:Dict[Symbol, float] = { symbol: symbol_data._performance for symbol, symbol_data in self.data.items() if symbol_data._performance != 0 }
self.top_perf_symbols = sorted(top_perf_symbols, key=top_perf_symbols.get, reverse=True)[:self.top_count]
def ConsolidatedBarHandler(self, consolidated: Slice) -> None:
symbol_data = self.data[consolidated.Symbol]
# save performance on each crypto
if symbol_data.is_traded():
symbol_data._count += 1
# count in performance for optimization
if consolidated.High > symbol_data._tp_price and consolidated.Low < symbol_data._sl_price:
symbol_data.update_performance(symbol_data._sl_price / symbol_data._open_price - 1)
elif consolidated.High >= symbol_data._tp_price and consolidated.Low > symbol_data._sl_price:
symbol_data.update_performance(symbol_data._tp_price / symbol_data._open_price - 1)
elif consolidated.High < symbol_data._tp_price and consolidated.Low <= symbol_data._sl_price:
symbol_data.update_performance(symbol_data._sl_price / symbol_data._open_price - 1)
elif symbol_data._count >= self.max_traded_duration:
symbol_data.update_performance(consolidated.Close / symbol_data._open_price - 1)
if self.Portfolio[consolidated.Symbol].Invested:
self.cancel_open_orders(consolidated.Symbol)
self.MarketOrder(consolidated.Symbol, -self.Portfolio[consolidated.Symbol].Quantity, tag='1 hour threshold')
if consolidated.Symbol in self.data:
if consolidated.Symbol.Value == 'BTCUSD':
if len(self.btc_returns) >= self.period:
if np.log(consolidated.Close / consolidated.Open) > np.percentile(self.btc_returns, self.percentile):
if not any(symbol_data.is_traded() for symbol, symbol_data in self.data.items()):
self.trade_flag = True
self.btc_returns.append(np.log(consolidated.Close / consolidated.Open))
else:
# execute trade
if self.trade_flag:
if symbol_data.is_ready():
price_std:float = symbol_data.get_std()
if price_std != 0.:
sl_price:float = round(consolidated.Close - self.beta * price_std, 5)
tp_price:float = round(consolidated.Close + self.alpha * price_std, 5)
symbol_data.trade(consolidated.Close, tp_price, sl_price)
# trading when symbol is in actual selection
if consolidated.Symbol in self.top_perf_symbols:
quantity:int = self.Portfolio.TotalPortfolioValue // len(self.top_perf_symbols) * self.portion // consolidated.Close
self.MarketOrder(consolidated.Symbol, quantity, tag='MarketOrder')
self.data[consolidated.Symbol].update_price(consolidated.Close)
def OnOrderEvent(self, orderEvent: OrderEvent) -> None:
if orderEvent.Status == OrderStatus.Filled:
order_ticket = self.Transactions.GetOrderTicket(orderEvent.OrderId)
# NOTE tag text can be altered by LEAN, for example:
# MarketOrder - Warning: fill at stale price {datetime}, using QuoteBar data.
# that's the reason 'in' keyword is used
if 'MarketOrder' in order_ticket.Tag:
self.StopMarketOrder(order_ticket.Symbol, -order_ticket.Quantity, self.data[order_ticket.Symbol]._sl_price)
self.LimitOrder(order_ticket.Symbol, -order_ticket.Quantity, self.data[order_ticket.Symbol]._tp_price)
# either SL or TP
else:
self.cancel_open_orders(order_ticket.Symbol)
def cancel_open_orders(self, symbol:Union[Symbol, None]) -> None:
# cancel all opened orders
orders_to_cancel = self.Transactions.GetOrderTickets(lambda order_ticket: order_ticket.Status not in [OrderStatus.Filled, OrderStatus.Canceled, OrderStatus.Invalid] and order_ticket.Symbol == symbol) if symbol is not None else \
self.Transactions.GetOrderTickets(lambda order_ticket: order_ticket.Status not in [OrderStatus.Filled, OrderStatus.Canceled, OrderStatus.Invalid])
for ticket in orders_to_cancel:
response = ticket.Cancel()
