“该策略交易26种加密货币对，使用每日回报和交易量冲击排序，做多低回报、低交易量货币对，做空高回报、低交易量货币对。”
资产类别: 差价合约、加密货币 | 地区: 全球 | 周期: 每日 | 市场: 加密货币 | 关键词: 交易量、反转
I. 策略概要
该策略交易26种加密货币对，根据先前的回报每天将其分为三组，并根据交易量冲击进一步分为三个子组。交易量冲击计算为滚动周期内与趋势的对数偏差。这种3×3排序创建了投资组合，其中在低回报、低交易量货币对中采取多头头寸，在高回报、低交易量货币对中采取空头头寸。空头头寸可以使用差价合约（CFDs）执行，尽管仅多头方法也具有盈利能力。该策略每日重新平衡，利用回报和交易量动态来捕捉一致的交易机会。
II. 策略合理性
该论文强调，过去的加密货币回报呈负自相关性，当过去的交易量与滞后回报相互作用时，它能显著预测未来的回报。交易量冲击较低或为负且先前回报较低或为负的加密货币对，在低交易量下的反转模式之后，往往会在下一时期反转。随着交易量冲击接近于零，这种反转的强度会减弱。做多低回报、低交易量货币对，做空高回报、低交易量货币对的逆势策略是有利可图的，其表现优于延续策略。即使在保守的交易成本下，该策略在经济上仍然具有重要意义，其回报也无法用风险因素或流动性不足来解释，这突显了反转策略的稳健性。
III. 来源论文
Trading Volume in Cryptocurrency Markets [点击查看论文]
- 丹尼尔·比安奇（Daniele Bianchi）、米科拉·巴比亚克（Mykola Babiak）和亚历山大·迪克森（Alexander Dickerson）。伦敦玛丽女王大学，澳大利亚新南威尔士大学商学院，兰卡斯特大学管理学院
<摘要>
我们在加密货币市场的背景下提供了实证证据，表明流动性提供的回报（以短期反转策略的回报为代表）主要集中在市场活跃度较低的交易对中。在实证方面，我们重点关注2017年3月1日至2022年3月1日期间在多个交易所交易的以美元计价的适度大型加密货币对横截面。我们的研究结果表明，在规模较小、波动性较大且流动性较差的加密货币对中，流动性提供的预期回报被放大，而在这些交易对中，对逆向选择的担忧可能更高。面板回归分析证实，滞后回报与交易量之间的相互作用包含加密货币回报动态的显著预测信息。这与强调库存风险和逆向选择对流动性提供作用的理论一致。
IV. 回测表现
| 年化回报 | 41.18% |
| 波动率 | 7.84% |
| β值 | -0.01 |
| 夏普比率 | 5.25 |
| 索提诺比率 | -0.247 |
| 最大回撤 | N/A |
| 胜率 | 50% |
V. 完整的 Python 代码
from AlgorithmImports import *
class TradingVolumeInCryptocurrencyMarketsAndReversals(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2015, 1, 1)
self.SetCash(100000)
self.cryptos = [
"BTCUSD", # Bitcoin
"ETHUSD", # Ethereum
"XRPUSD", # XRP
# "BCHUSD", # Bitcoin cash
"LTCUSD", # Litecoin
"BSVUSD", # Bitcoin SV
"EOSUSD", # EOS
"XMRUSD", # Monero
"TRXUSD", # Tron
"XTZUSD", # Tezos
"XLMUSD", # Stellar
"NEOUSD", # Neo
"DAIUSD", # Dai
"ZECUSD", # Zcash
"VETUSD", # VeChain
"ETCUSD", # Ethereum Classic
"MKRUSD", # Maker
"OMGUSD", # OMG Network
# "DGBUSD", # Dogecoin
# "BATUSD", # Basic Attention Token
# "ZRXUSD", # Ox
]
self.data = {}
self.period = 61
self.traded_percentage = 0.1
self.quantile = 3
self.SetBrokerageModel(BrokerageName.Bitfinex)
for crypto in self.cryptos:
# GDAX is coinmarket, but it doesn't support this many cryptos, so we choose Bitfinex
data = self.AddCrypto(crypto, Resolution.Minute, Market.Bitfinex)
data.SetFeeModel(CustomFeeModel())
data.SetLeverage(10)
self.data[crypto] = SymbolData(crypto, self.period)
self.last_day = -1
def OnData(self, data):
performance = {}
volume_shock = {}
if self.last_day == self.Time.day: return
self.last_day = self.Time.day
for crypto in self.cryptos:
if crypto in data.Bars and data[crypto]:
# Volume can be taken only from TradeBar and data[crypto] returns QuoteBar by default
price = data.Bars[crypto].Value
volume = data.Bars[crypto].Volume
self.data[crypto].update(price, volume)
if self.data[crypto].is_ready():
result_volume_shock = self.data[crypto].volume_shock()
if result_volume_shock:
performance[crypto] = self.data[crypto].performance()
volume_shock[crypto] = result_volume_shock
if len(performance) < self.quantile:
self.Liquidate()
return
sorted_by_performance = [x[0] for x in sorted(performance.items(), key=lambda item: item[1])]
sorted_by_volume_shock = [x[0] for x in sorted(volume_shock.items(), key=lambda item: item[1])]
quantile = int(len(sorted_by_performance) / self.quantile)
lowest_performance = sorted_by_performance[:quantile]
lowest_volume_shock = sorted_by_volume_shock[:quantile]
highest_performance = sorted_by_performance[-quantile:]
long = [x for x in lowest_performance if x in lowest_volume_shock]
short = [x for x in highest_performance if x in lowest_volume_shock]
# Trade execution
invested = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
for symbol in invested:
if symbol not in long + short:
self.Liquidate(symbol)
long_length = len(long)
short_length = len(short)
for crypto in long:
if crypto in data and data[crypto]:
self.SetHoldings(crypto, self.traded_percentage / long_length)
for crypto in short:
if crypto in data and data[crypto]:
self.SetHoldings(crypto, -self.traded_percentage / short_length)
class SymbolData():
def __init__(self, symbol, period):
self.Symbol = symbol
self.Closes = RollingWindow[float](period)
self.Volumes = RollingWindow[float](period)
def update(self, close, volume):
self.Closes.Add(close)
self.Volumes.Add(volume)
def is_ready(self):
return self.Closes.IsReady and self.Volumes.IsReady
def performance(self):
closes = [x for x in self.Closes]
return closes[0] / closes[-1] - 1
# Log deviation, 10 base log of current day volume - 10 base log of sum of 30 days volumes before current day devided by 30
def volume_shock(self):
volumes = [x for x in self.Volumes]
current_day_volume = volumes[0]
avg_volume = sum(volumes[1:]) / len(volumes[1:]) # sum of 30 days volumes before current day devided by 30
if (current_day_volume <= 0):
return None
else:
return np.log(volumes[0]) - np.log(avg_volume)
# Custom fee model.
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))