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

---

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

2. 设置文件路径

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

3. 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"] * dev
    df["down"] = df["ma"] - df["std"] * dev
    # 抛弃NA数值
    df.dropna(inplace=True)
    
    # 计算目标仓位
    target = 0
    target_data = []

    for ix, row in df.iterrows():
        # 没有仓位
        if not target:
            if row.spread >= row.up:
                target = -1
            elif 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 = 0
    df["change"] = df["spread"].diff()
    df["fee"]=0
    df["fee"][df['pos']!=df['pos'].shift(1)]=(df.iloc[:,0] df.iloc[:,1])*rate
    df["pnl"] = df["change"] * df["pos"]-df["fee"]
    df["balance"] = df["pnl"].cumsum()
    return df

4. 运行结果

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, ArrayManager
from vnpy_spreadtrading import (
    SpreadStrategyTemplate,
    SpreadAlgoTemplate,
    SpreadData,
    OrderData,
    TradeData,
    TickData,
    BarData
)


class MaSpreadStrategy(SpreadStrategyTemplate):
    """"""

    author = "mossloo"

    ma_window = 20
    ma_dev = 2
    max_pos = 1
    payup = 0.001
    interval = 5

    spread_pos = 0.0
    ma_up = 0.0
    ma_down = 0.0
    ma = 0.0

    parameters = [
        "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:
            return

        self.ma = self.am.sma(self.ma_window)
        dev = self.am.std(self.ma_window)
        self.ma_up =  self.ma_dev*dev   self.ma
        self.ma_down = self.ma - self.ma_dev*dev

        if 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.
        """
        pass

    def on_order(self, order: OrderData):
        """
        Callback when order status is updated.
        """
        pass

    def on_trade(self, trade: TradeData):
        """
        Callback when new trade data is received.
        """
        pass

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

2. 继续打开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
具体的配置步骤还是推荐大家先看下官方文档的多合约价差套利的使用教程，比较详细，配置下来也没有什么坑。

这篇好文章是转载于：学新通技术网