构建价差套利

admin

2 月前

1.价差套利原理

价差套利是一种金融交易策略，通过利用不同市场或不同交易所之间的价格差异来获取利润。以下是价差套利的原理：

基本原则：价差套利的基本原则是同时在相关合约上建立一个多头部位和一个空头部位，以利用两个头寸之间的差值变化来获利。
跨交易所套利：在不同交易所之间进行套利是一种常见的价差套利策略。如果一个交易所的价格比另一个交易所高，可以在高价交易所卖出资产，在低价交易所买入等量的资产，从中获取差价利润。这种策略要求交易者在两个交易所分别持有一定数量的资产，并且需要注意交易手续费和资产转移的效率。
期现套利：期现套利是指利用现货市场和期货市场之间的价格差异进行套利。当期货合约的价格高于现货价格时，可以同时买入现货并卖出期货，通过差价获利。关键是确保买入的现货数量和卖出的期货数量相等，以减少风险。期现套利的收益率取决于差价的大小和持仓时间。
跨期套利：跨期套利是一种利用同一市场上不同交割月份的期货合约之间的价差进行套利的交易行为。投资者通过同时买入一个合约和卖出另一个合约，以期望在价格关系有利时将两种合约对冲平仓获利。跨期套利是套利交易中最常见的一种形式，也是股指期货市场上常见的套利策略之一。
风险：价差套利也存在一定的风险。例如，跨交易所套利可能面临盘口流动性不足、API响应慢或交易不成功等风险。期现套利可能面临现货价格上涨导致浮亏、保证金不足导致爆仓或套利规模过大导致流动性风险等问题

2. 跨期套利

跨期套利的基本原理是利用不同交割月份之间的价格差距出现变化时进行对冲，从中获得利润。当期货市场波动较大时，不同交割月份的合约价格差会出现偏离合理价差的情况。投资者可以根据交割制度，捕捉价格偏离区间的机会，同时总结价差走势规律，判断价差套利机会。

跨期套利可以分为牛市套利和熊市套利。牛市套利是指投资者看多股市，认为较远交割期的期货合约涨幅将大于近期合约的涨幅，或者较远期的期货合约跌幅将小于近期合约的跌幅。熊市套利则相反，投资者认为较远交割期的期货合约跌幅将大于近期合约的跌幅，或者较远期的期货合约涨幅将小于近期合约的涨幅。

跨期套利还可以根据买卖方式分为买进套利和卖出套利。买进套利是指投资者预期不同交割月份的期货合约的价差将扩大，他们会买入价格较高的合约，同时卖出价格较低的合约。卖出套利则相反，投资者预期不同交割月份的期货合约价差将缩小，他们会卖出价格较高的合约，同时买入价格较低的合约。

3. 套利实战

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

3.1.投研分析

我们准备了币安交易所所有带有交割合约币种的分钟线、小时线、

用到的第一份数据是BTCUSDT_231229_BINANCE.csv，表示BTC近月合约高开低收价格数据

用到的第二份数据是BTCUSDT_240329_BINANCE.csv，表示BTC远月合约高开低收价格数据

投研第一步，对数据进行处理，使用jupyter交互式环境，观察数据样貌

import pandas as pd
import plotly.express as px

df1 = pd.read_csv("BTCUSDT_231229_BINANCE.csv",index_col="datetime")
df1.head()

df2 = pd.read_csv("BTCUSDT_240329_BINANCE.csv",index_col="datetime")
df2.head()

构建价差数据集

df_data  = pd.DataFrame({
    "BTC231229":df1["close"],
    "BTC240329":df2["close"]
})
# 清除空值数据
df_data.dropna(inplace=True)
df_data["spread"] = df_data["BTC240329"] - df_data["BTC231229"]
# 绘制图像
px.line(df_data["spread"])

保存数据集

# 保存数据
df_data.to_csv("spread_data.csv")

3.2 价差特征分析

价差特征分析是指利用价格或指标之间的差距来进行分析和预测的方法。通过计算不同时间点或不同指标之间的差值，可以揭示出价格或指标的变化趋势和差异，从而帮助我们做出相应的决策。以下是价差特征分析的一些常见应用和方法：

技术指标的价差分析：价差分析也可以用于技术指标的计算和分析。通过计算不同指标之间的差值，可以得到更多的信息。例如，通过计算不同移动平均线之间的差值，可以判断价格的趋势和变化。
历史统计特征的价差分析：价差分析还可以用于计算历史统计特征。通过计算不同时间窗口内的统计特征的差值，可以得到更多的信息。
特征生成和价差分析：在特征工程中，可以利用价差分析生成新的特征。通过计算不同特征之间的差值，可以得到更多的特征。例如，计算不同指标之间的差值，可以生成新的特征来描述指标之间的关系。

下面我们开始利用技术指标来构建价差分析：

对数据进行描述性分析，观察数据样貌，提供了对数据集整体情况的认知和理解。通过描述性分析，我们可以了解数据的集中趋势、离散程度、分布形状和异常值等特征，为进一步的数据分析和解释提供了基础。

import pandas as pd
import plotly.graph_objects as go
# 读入数据
df = pd.read_csv("spread_data.csv")
# 描述性分析
df["spread"].describe()

显示数据的均值、标准差、最小值、四分位数、最大值

观察数据时间序列上的滚动特征特征

# 滚动特征
df["ma20"] = df["spread"].rolling(20).mean()
df["std20"] = df.spread.rolling(20).std()
df["max20"] = df.spread.rolling(20).max()
df["min20"] = df.spread.rolling(20).min()
df.tail()

图表绘制

# 图表绘制
data = [
    go.Scatter(x=df.index, y=df["spread"], name="spread"),
    go.Scatter(x=df.index, y=df["ma20"], name="ma"),
    go.Scatter(x=df.index, y=df["max20"], name="max"),
    go.Scatter(x=df.index, y=df["min20"], name="min"),
]

fig = go.Figure(data=data)
fig.show()

在下图中我们发现有若干异常值，这些异常值可能是数据采集或记录过程中的错误或特殊情况，也可能是当天发生比较大的行情波动，我们可以通过计算四分位数和绘制箱线图等方法来识别数据中的异常值，提高数据的准确性和可靠性。图中的指标也可以帮助我们了解数据的平均水平或典型值，从而更好地理解数据的整体特征和趋势。比如时间区域1就是明显的平稳状态，适用于卖出套利策略，时间区域2就是趋势状态，适用于买进套利策略。

如何科学的分析一段周期内价差特征是否平稳呢？我们可以使用Adf检验方法。ADF检验（Augmented Dickey-Fuller test）是一种用于判断时间序列数据平稳性的统计检验方法，也被称为单位根检验。单位根检验是针对时间序列数据中是否存在单位根（unit root）这一统计特性进行的检验。单位根存在意味着序列是非平稳的，而平稳序列在许多时间序列模型中是必要的。

from statsmodels.tsa.stattools import adfuller
# 平稳序列检验
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))

判断一个序列平不平稳就是看p-value的值是否小于0.05,如果小于0.05，则说明序列是平稳的，大于0.05则不平稳。在实际交易市场中，0.05这个阈值可能很难达到，我们可以降低要求，比如阈值调整到0.1，小于0.1我们也认为序列平稳。在平稳的时间序列下，我们就可以进行卖出价差套利。

4. 统计套利

在统计套利中，常用的方法包括配对交易（Pairs Trading）、均值回归策略（Mean Reversion）、协整关系交易（Cointegration Trading）等。这些方法基于统计学的原理，利用价格或资产之间的相对价差、均值偏离或协整关系等统计指标来确定交易信号和执行策略。

在实践中，统计套利中的某些策略可以包含价差套利的元素，例如配对交易和均值回归策略经常涉及价格差异的利用。同时，统计套利的一些方法和工具，如协整关系的分析和模型构建，也可以为价差套利提供理论支持和辅助分析。

4.1 构建均值回归策略

本文着手于构建一个均值回归策略，通过理论和实践相结合的方式，帮助读者快速构建一套属于自己的交易策略，构造一个均值回归策略涉及以下步骤：

选择资产：首先，选择您感兴趣的资产或市场，可以是股票、期货、外汇等。确保选择的资产存在明显的价格波动和均值回归的趋势。
确定均值：通过历史数据计算资产的均值。常见的方法是使用移动*均线（如简单移动*均线或指数加权移动*均线）来估计资产价格的均值。
计算偏离度：计算资产价格相对于均值的偏离度。可以使用标准差、百分位数或其他统计指标来度量价格的偏离程度。
确定交易信号：根据偏离度确定交易信号。当价格偏离均值超过一定阈值时，产生交易信号。例如，当价格偏离均值超过一个标准差时，可以认为价格过度偏离，产生反向交易信号。
确定交易规则：定义具体的交易规则，包括入场点、出场点和止损点。例如，当价格偏离均值达到一定程度时，进入反向头寸；当价格回归到均值附*时，*仓并获利。
风险管理：制定有效的风险管理策略，包括设置止损点、控制仓位大小和分散投资等。确保风险可控，并考虑交易成本和流动性等因素。
回测和优化：使用历史数据进行回测，评估策略的表现，并进行必要的优化。调整参数和交易规则，以提高策略的盈利能力和稳定性。
实盘交易：在回测和优化后，将策略应用到实盘交易中。始终密切监控市场情况和策略表现，并根据需要进行调整和优化。

5. 实践

5.1 选择交易标的

选择交易标的核心是确保选择的资产存在明显的价格波动和均值回归的趋势

使用统计指标来评估价格的波动性和均值回归的趋势。常用的指标包括标准差、*均绝对偏差（Mean Absolute Deviation）、波动率等。较高的波动性和明显的均值回归特征可能表明资产适合均值回归策略。
协整性分析：对于多个相关资产，可以进行协整性分析。协整关系是指一组资产的价格在长期内存在稳定的线性关系。如果资产之间存在协整关系，并且价格偏离协整关系时会发生均值回归，那么这些资产可能适合均值回归策略。
历史数据分析：通过对资产的历史价格数据进行分析，评估价格的波动性和均值回归的趋势。观察价格的波动范围、频率和幅度，以及价格是否有向均值回归的倾向。

import numpy as np
import pandas as pd

df = pd.read_csv("spread_data.csv")
# 计算价格波动性指标
std = np.std(df['spread'])  # 标准差
mad = np.mean(np.abs(df['spread'] - np.mean(df['spread'])))  # *均绝对偏差
volatility = std / np.mean(df['spread'])  # 波动率（标准差与均值的比率）

# 计算均值回归指标
mean = np.mean(df['spread'])  # 均值
rolling_mean = df['spread'].rolling(window=10).mean()  # 移动*均线

# 输出结果
print("价格波动性指标：")
print("标准差：", std)
print("*均绝对偏差：", mad)
print("波动率：", volatility)

print("\n均值回归指标：")
print("均值：", mean)

5.2 确定交易规则

确定均值，常用的方法可以是ma均线，可以是机器学习拟合过去一段时间的回归曲线，可以是布林通道。确定交易信号，均值回归策略核心是认为相同品种的价差最终会走向价值回归，那么就在价差大的时候开仓，价差小的时候*仓。

5.3 策略代码

import pandas as pd
import plotly.express as px

# 读取数据，设置时间戳索引
df_data = pd.read_csv("spread_data.csv")
df_data.index = pd.DatetimeIndex(df_data["datetime"])
df = pd.DataFrame({"spread": df_data["spread"]})
df = df.resample("5min").last()
df.dropna(inplace=True)

# 设置策略参数
window = 20
dev = 3
# 计算均线和上下轨
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)
# 计算盈亏
df["change"] = df["spread"].diff()
df["pnl"] = df["change"] * df["pos"]
df["balance"] = df["pnl"].cumsum()

# 绘制净值曲线
px.line(df["balance"])

5.4 策略回测

在vnpy_spreadtrading(位置位于：External Libraries\site-packages)目录下的strategies下创建boll_spread_strategy.py策略文件

from vnpy.trader.utility import BarGenerator, ArrayManager
from vnpy_spreadtrading import (
    SpreadStrategyTemplate,
    SpreadAlgoTemplate,
    SpreadData,
    OrderData,
    TradeData,
    TickData,
    BarData
)
from vnpy.trader.constant import Interval


class BollSpreadStrategy(SpreadStrategyTemplate):
    """"""

    author = "mossloo"

    ma_window = 20
    ma_dev = 3
    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, window=5, on_window_bar=self.on_5min_spread_bar,
                               interval=Interval.MINUTE)
        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):
        self.bg.update_bar(bar)

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

继续打开jupyter notebook

from vnpy.trader.optimize import OptimizationSetting
from vnpy_spreadtrading.backtesting import BacktestingEngine
from vnpy_spreadtrading.strategies.boll_spread_strategy import (
    BollSpreadStrategy
)
from vnpy_spreadtrading.base import LegData, SpreadData
from datetime import datetime
from vnpy.trader.constant import Interval

symbol_1 = "BTCUSDT_240329.BINANCE"
symbol_2 = "BTCUSDT_231229.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(2021, 6, 10),
    end=datetime(2023, 11, 7),
    rate=0.0004,
    slippage=0.02,
    size=1,
    pricetick=0.02,
    capital=1_000_000,
)

engine.add_strategy(BollSpreadStrategy, {})
engine.load_data()
engine.run_backtesting()
df = engine.calculate_result()
engine.calculate_statistics()
engine.show_chart()

对于这个策略而言，我们亏钱的原因在于手续费太高了，由于我们策略使用的是市价单，加密市场对于市价单收取的手续费要远远高于限价单，所以这个策略的手续费在万分之四左右，如果能手续费降低到万分之一，加上百分之五十的返佣，到万分之0.5，那么这个策略的夏普比能到7.83。

策略还可以进行参数优化，找到参数*原地带，后续选择较优的参数跑模拟盘验证。

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)

6.实盘交易

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