文章目录

  • 1.价差套利原理
    • 1.1 概述
    • 1.2 以BTC为例
  • 2.投研分析
  • 3. veighna的价差交易回测引擎
  • 4.实盘交易

1.价差套利原理

1.1 概述

在数字货币交易市场，我们会发现大多数行情下，相同币种之间的不同交割合约会存在一定的价差，由于它们属于同一品种，本身价值不会有任何差别，而且涨跌趋势一致，相关性高。那么如果在它们价差低的时候买入，价差高的时候卖出，这样我们就可以赚取中间的这部分差价。不过在实际交易过程中，我们还需要考虑到交易滑点、手续费、极端行情下，价差走出趋势特征…

1.2 以BTC为例

图一、不同合约的比特币行情图由上图可以看出比特币远月合约与永续合约之间存在一定的价差。
图二、某一时刻比特币价差图
可以看到btc不同月份合约之间的价差在58.4，我们用python的jupyter notebook可以看一下9月30到10月30这段时间这两个合约的价差波动情况。
图三、比特币总体价差图

蓝色线表示了价差的波动情况，可以看出，总体波动区间是在55-175之间，从图中我们也可以看出，在极端情况下，价差可能会突破均值回归轨道，中间某段时间一度下降到-30。

综上，构建ma均线，如果跌到ma均线-3个标准差，那我们就做多价差，如果涨到ma均线+3个标准差，则做空价差。价差回到ma均线则平仓。

2.投研分析

我们准备了币安交易所所有带有交割合约币种的分钟线、小时线、日线数据。如何获取数据，请看教程【VeighNa】开始量化交易——第二章：PostgreSQL数据库配置

  1. 导入包
# 导入相关包import pandas as pdimport plotly.express as pximport plotly.graph_objects as gofrom statsmodels.tsa.stattools import adfullerimport osfrom datetime import datetimefrom typing import Unionimport re
  1. 设置文件路径
root_path = r"F:\market\crypto\1m"# 获取当前目录下所有文件名称all_files_name:list = os.listdir(root_path)
  1. read_two_csv()函数，返回构建好的价差数据pandas格式，adftesting()函数对这个价差序列进行adf检验，观察时间序列数据是否平稳。backtesting()对策略进行简单回测。
def read_two_csv(path1:str,path2:str,start:datetime,end:Union[datetime,int]=-1)->pd.DataFrame:df_1 = pd.read_csv(path1)df_1.index = pd.DatetimeIndex(df_1["datetime"]) df_2 = pd.read_csv(path2)df_2.index = pd.DatetimeIndex(df_2["datetime"])df_1 = df_1.loc[start:end,:]df_2 = df_2.loc[start:end,:]symbol_1 = df_1.iloc[0,0]symbol_2 = df_2.iloc[0,0]df_data = pd.DataFrame({symbol_1:df_1["close"],symbol_2:df_2["close"]})# 抛弃NA数据点（如果有）df_data = df_data.dropna()# 计算价差df_data["spread"] = df_data[symbol_1] - df_data[symbol_2]# 重设索引# clean_data = df_data.reset_index(drop=True)return df_datadef adftesting(df)->float:result = adfuller(df["spread"])# 打印结果print('ADF 统计值: %f' % result[0])print('p-value: %f' % result[1])print('临界值:')for k, v in result[4].items():print('\t%s: %.3f' % (k, v))return result[1]def backtesting(window,dev,df)->pd.DataFrame:# 计算均线和上下轨df["ma"] = df["spread"].rolling(window).mean()df["std"] = df["spread"].rolling(window).std()df["up"] = df["ma"] + df["std"] * devdf["down"] = df["ma"] - df["std"] * dev# 抛弃NA数值df.dropna(inplace=True)# 计算目标仓位target = 0target_data = []for ix, row in df.iterrows():# 没有仓位if not target:if row.spread >= row.up:target = -1elif row.spread <= row.down:target = 1# 多头仓位elif target > 0:if row.spread >= row.ma:target = 0# 空头仓位else:if row.spread <= row.ma:target = 0# 记录目标仓位target_data.append(target)df["target"] = target_data# 计算仓位df["pos"] = df["target"].shift(1).fillna(0)# 计算盈亏和手续费rate = 0df["change"] = df["spread"].diff()df["fee"]=0df["fee"][df['pos']!=df['pos'].shift(1)]=(df.iloc[:,0]+df.iloc[:,1])*ratedf["pnl"] = df["change"] * df["pos"]-df["fee"]df["balance"] = df["pnl"].cumsum()return df
  1. 运行结果
path1 = os.path.join(root_path,"BTCUSD_230331.BINANCE.csv")path2 = os.path.join(root_path,"BTCUSD_PERP.BINANCE.csv")df = read_two_csv(path1,path2,start=datetime(2019,1,1),end=datetime(2022,10,30))adftesting(df)result = backtesting(20,3,df)


可以看到p-value小于0.05，拒绝原假设，认为这个序列是平稳的。

当我们将手续费rate设置为0时的回测结果如下：

价差曲线

资金曲线

看到这里的小伙伴应该会比较激动，这么完美的曲线，拿去实盘不是赚麻了么。可当我们把手续费rate设置为0.0004的时候，曲线就变成了这个样子。


可以发现，我们的手续费占了大头，并且这个回测还没有算上滑点。所以，非常遗憾，实盘我们没有办法赚钱，除非你可以0手续费交易，不过本着学习的态度，掌握某一种策略分析逻辑，后续遇到某一段行情，或者某一不成熟市场，可能就有这种套利机会了。

3. veighna的价差交易回测引擎

为了更精准的回测出我们构建的策略表现到底如何，采用veighna自带的价差交易回测引擎来进行回测。

  1. 在vnpy_spreadtrading目录下的strategies下创建ma_spread_strategy.py策略文件

策略文件如下：

from vnpy.trader.utility import BarGenerator, ArrayManagerfrom vnpy_spreadtrading import (SpreadStrategyTemplate,SpreadAlgoTemplate,SpreadData,OrderData,TradeData,TickData,BarData)class MaSpreadStrategy(SpreadStrategyTemplate):""""""author = "mossloo"ma_window = 20ma_dev = 2max_pos = 1payup = 0.001interval = 5spread_pos = 0.0ma_up = 0.0ma_down = 0.0ma = 0.0parameters = ["ma_window","ma_dev","max_pos","payup","interval"]variables = ["spread_pos","ma_up","ma_down","ma"]def __init__(self,strategy_engine,strategy_name: str,spread: SpreadData,setting: dict):""""""super().__init__(strategy_engine, strategy_name, spread, setting)self.bg = BarGenerator(self.on_spread_bar)self.am = ArrayManager()def on_init(self):"""Callback when strategy is inited."""self.write_log("策略初始化")self.load_bar(10)def on_start(self):"""Callback when strategy is started."""self.write_log("策略启动")def on_stop(self):"""Callback when strategy is stopped."""self.write_log("策略停止")self.put_event()def on_spread_data(self):"""Callback when spread price is updated."""tick = self.get_spread_tick()self.on_spread_tick(tick)def on_spread_tick(self, tick: TickData):"""Callback when new spread tick data is generated."""self.bg.update_tick(tick)def on_spread_bar(self, bar: BarData):"""Callback when spread bar data is generated."""self.stop_all_algos()self.am.update_bar(bar)if not self.am.inited:returnself.ma = self.am.sma(self.ma_window)dev = self.am.std(self.ma_window)self.ma_up =self.ma_dev*dev + self.maself.ma_down = self.ma - self.ma_dev*devif not self.spread_pos:if bar.close_price >= self.ma_up:self.start_short_algo(bar.close_price - 10,self.max_pos,payup=self.payup,interval=self.interval)elif bar.close_price <= self.ma_down:self.start_long_algo(bar.close_price + 10,self.max_pos,payup=self.payup,interval=self.interval)elif self.spread_pos < 0:if bar.close_price <= self.ma:self.start_long_algo(bar.close_price + 10,abs(self.spread_pos),payup=self.payup,interval=self.interval)else:if bar.close_price >= self.ma:self.start_short_algo(bar.close_price - 10,abs(self.spread_pos),payup=self.payup,interval=self.interval)self.put_event()def on_spread_pos(self):"""Callback when spread position is updated."""self.spread_pos = self.get_spread_pos()self.put_event()def on_spread_algo(self, algo: SpreadAlgoTemplate):"""Callback when algo status is updated."""passdef on_order(self, order: OrderData):"""Callback when order status is updated."""passdef on_trade(self, trade: TradeData):"""Callback when new trade data is received."""passdef stop_open_algos(self):""""""if self.buy_algoid:self.stop_algo(self.buy_algoid)if self.short_algoid:self.stop_algo(self.short_algoid)def stop_close_algos(self):""""""if self.sell_algoid:self.stop_algo(self.sell_algoid)if self.cover_algoid:self.stop_algo(self.cover_algoid)
  1. 继续打开jupyter notebook
from vnpy.trader.optimize import OptimizationSettingfrom vnpy_spreadtrading.backtesting import BacktestingEnginefrom vnpy_spreadtrading.strategies.ma_spread_strategy import (MaSpreadStrategy)from vnpy_spreadtrading.base import LegData, SpreadDatafrom datetime import datetimefrom vnpy.trader.constant import Interval
symbol_1 = "BTCUSD_221230.BINANCE"symbol_2 = "BTCUSD_PERP.BINANCE"
spread = SpreadData(name="BTC-Spread",legs=[LegData(symbol_1), LegData(symbol_2)],variable_symbols={"A": symbol_1, "B": symbol_2},variable_directions={"A": 1, "B": -1},price_formula="A-B",trading_multipliers={symbol_1: 1, symbol_2: 1},active_symbol=symbol_1,min_volume=1,compile_formula=False# 回测时不编译公式，compile_formula传False，从而支持多进程优化)
#%%engine = BacktestingEngine()engine.set_parameters(spread=spread,interval=Interval.MINUTE,start=datetime(2019, 6, 10),end=datetime(2022, 11, 1),rate=0.0002,slippage=0.0001,size=1,pricetick=1,capital=1_000_000,)engine.add_strategy(MaSpreadStrategy, {})
#%%engine.load_data()engine.run_backtesting()df = engine.calculate_result()engine.calculate_statistics()engine.show_chart()

运行结果如下：


可以对参数进行优化

setting = OptimizationSetting()setting.set_target("sharpe_ratio")setting.add_parameter("ma_window", 10, 30, 1)setting.add_parameter("ma_dev", 1, 3, 1)engine.run_ga_optimization(setting)


如果载入数据发生错误，请修改D:\software\Aconda\envs\vnpy\Lib\site-packages\vnpy_spreadtrading\base.py中的这段代码。
程序会先读取数据配置服务中的数据，如果返回数据为空，才读取数据库的数据，我们加了一个try的操作，让程序异常不要抛出。

4.实盘交易

很显然，我们上面的程序并不能直接上实盘交易，但如果我们手续费非常低的情况下，或者遇到极端行情，某些币种现货和期货之间的价差非常大，比如前段时间的luna，还有这段时间的mask，只要我们设置好程序，仍然可以在极端行情下赚取到稳定的资金，这也是我们学习量化交易的原因，只有长久稳定的赚钱，才能在市场立于不败之地。

https://www.vnpy.com/docs/cn/spread_trading.html
具体的配置步骤还是推荐大家先看下官方文档的多合约价差套利的使用教程，比较详细，配置下来也没有什么坑。