文章目录

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数据库配置

导入包

# 导入相关包
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from statsmodels.tsa.stattools import adfuller
import os
from datetime import datetime
from typing import Union
import re

设置文件路径

root_path = r"F:\market\crypto\1m"
# 获取当前目录下所有文件名称
all_files_name:list = os.listdir(root_path)

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_data
def 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

运行结果

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自带的价差交易回测引擎来进行回测。

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

策略文件如下：

from vnpy.trader.utility import BarGenerator, ArrayManager
from 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)

继续打开jupyter notebook

from vnpy.trader.optimize import OptimizationSetting
from vnpy_spreadtrading.backtesting import BacktestingEngine
from vnpy_spreadtrading.strategies.ma_spread_strategy import (MaSpreadStrategy
)
from vnpy_spreadtrading.base import LegData, SpreadData
from datetime import datetime
from 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
具体的配置步骤还是推荐大家先看下官方文档的多合约价差套利的使用教程，比较详细，配置下来也没有什么坑。

【VeighNa】开始量化交易——第三章：构建价差套利相关推荐

在Java中构建响应式微服务系统——第三章构建响应式微服务
第三章构建响应式微服务在本章中,我们将使用Vert.x构建我们的第一个微服务.由于大多数微服务系统使用HTTP进行交互,因此我们将以HTTP微服务作为开始.但是由于系统包含多个相互通讯的微服务,因 ...
【VeighNa】开始量化交易——第四章：结合AHR999指标和情绪指标分析
本文仅供学术研究,不做任何投资建议! 文章目录 1.什么是ahr999指标 2.计算ahr999指标 2.1 获取数据 2.2 将数据保存到数据库 2.3 开始计算ahr999指标 2.3.1 建立模 ...
比特币量化交易：利用Python构建自己的算法交易系统
随着科技的不断发展,传统的手工交易已经不再是唯一的路径,自动化交易系统越来越受到投资者们的欢迎.尤其是在高频率交易中,精细的算法能够让交易决策更快速.准确和有效. 本文将介绍以比特币为例,如何使用Py ...
量化交易第三课数据获取接口
获取数据概述数据接口种类行业 & 板块股票类表股票代码 & 代码补齐 RiceQuant 上的股票股票自动搜索及补全获取行业获取板块代码展示获取股票合约数据某一合 ...
python量化交易第五章
一.什么是动量策略 1.1概念: 预先对股票收益和交易量设定过滤准则,当股票收益或股票收益和交易量同时满足过滤准则就买入(做多)或卖出(做空)股票的投资策略. 核心:以股票的历史收益率作为主要的交易原 ...
【重磅干货】俸旻老师：你所不知道的的量化交易？
嘉宾介绍: 俸旻,北京大学数学学院学士和博士,曾任职于微软亚洲研究院副研究院,从事网络安全.信息安全研究:曾任职于华为技术有限公司,从事无线通讯技术研究,开发了全球第一套TD-LTE Relay样机: ...
零起点Python大数据与量化交易
零起点Python大数据与量化交易第1章从故事开始学量化 1 1.1 亿万富翁的"神奇公式" 2 1.1.1 案例1-1:亿万富翁的"神奇公式" 2 1.1 ...
Rxjs 响应式编程-第四章构建完整的Web应用程序
Rxjs 响应式编程-第一章:响应式 Rxjs 响应式编程-第二章:序列的深入研究 Rxjs 响应式编程-第三章: 构建并发程序 Rxjs 响应式编程-第四章构建完整的Web应用程序 Rxjs 响应 ...
Scikit-learn 秘籍第三章使用距离向量构建模型
第三章使用距离向量构建模型作者:Trent Hauck 译者:飞龙协议:CC BY-NC-SA 4.0 这一章中,我们会涉及到聚类.聚类通常和非监督技巧组合到一起.这些技巧假设我们不知道结果变量 ...

【VeighNa】开始量化交易——第三章：构建价差套利