[原创] 用Python实现一个基于EG协整法的跨周期套利策略

一些金融学与数学背景

众所周知，套利交易的大前提是两个有相同属性交易标的价格的均值回归，是一种赚取这种均值回归的利差交易。经典回归模型是建立在平稳数据变量的基础之上的，对于非平稳变量，不能使用经典回归模型，否则会出现虚假回归等诸多问题。

由于许多关联交易标的价格相关性是非平稳的，这就给经典的回归分析方法带来了很大限制。实际应用中大多数时间序列其实都是非平稳的，通常采用差分方法消除序列中含有的非平稳趋势，使得序列平稳化后建立模型，比如使用ARIMA模型。但是变换后的序列限制了交易标的价格的范围，并且有时变换后的序列由于不具有直接的分析意义，使得化为平稳序列后所建立的时间序列模型不便于解释。

1987年Engle和Granger提出的协整理论及其方法，为非平稳序列的建模提供了另一种途径。虽然一些经济变量的本身是非平稳序列，但是，它们的线性组合却有可能是平稳序列。这种平稳的线性组合被称为协整方程，且可解释为变量之间的长期稳定的均衡关系。

例如，消费和收入都是非平稳时间序列，但是具有协整关系。假如它们不具有，那么长期消费就可能比收入高或低，于是消费者便会非理性地消费或累积储蓄。

假定一些经济指标被某经济系统联系在一起，那么从长远看来这些变量应该具有均衡关系，这是建立和检验模型的基本出发点。在短期内，因为季节影响或随机干扰，这些变量有可能偏离均值。如果这种偏离是暂时的，那么随着时间推移将会回到均衡状态；如果这种偏离是持久的，就不能说这些变量之间存在均衡关系。协整(co-integration)可被看作这种均衡关系性质的统计表示。
　　
协整概念是一个强有力的概念。因为协整允许我们刻画两个或多个序列之间的平衡或平稳关系。对于每一个序列单独来说可能是非平稳的，这些序列的矩，如均值、方差或协方差随时间而变化，而这些时间序列的线性组合序列却可能有不随时间变化的性质。

将以上概念运用到期货交易中

在大宗商品的套利交易中，相同品种的跨期合约具有强烈的协整关系（这同时也包括现货与期货的套利），运用Engle和Granger提出的这种协整理论及其方法进行量化分析是最好不过的了，这也是目前很多世界上的顶级量化团队在进行大宗商品同品种跨期套利时的理论基石。

本文将运用这套理论在螺纹钢期货上进行实践，代码实现方面，我们使用Python这个目前量化金融中最强大的语言来编码。

在开始编码前，我们需要了解一下将会用到的一个在世界范围内，数据科学领域非常流行的Python库，名叫：statsmodels

statsmodels（http://www.statsmodels.org）是一个Python库，用于拟合多种统计模型，执行统计测试以及数据探索和可视化，是目前计量经济学和数据科学研究中必不可少的神兵利器。它包含很多的经典频率学派统计方法，而且目前人工智能领域中炙手可热的贝叶斯方法和机器学习模型也可在其中找到。

比如：

• 线性模型，广义线性模型和鲁棒线性模型

• 线性混合效应模型

• 方差分析（ANOVA）方法

• 时间序列过程和状态空间模型

• 广义的矩量法

配合这个库，加上著名的科学计算NumPy库，我们便可以用Python编码出这套理论的基础框架用于量化分析，且同时在发明者量化平台实现自动化交易。

注意：

关于这个库的安装方法，以及NumPy库的安装方法，参见发明者量化网站文库中托管者的安装与部署，这两个库都不是Python的官方自带库，需要读者自行安装到托管者部署的本地电脑或者云计算服务器上，不明白的读者可以去发明者量化网站搜索一下方法，很简单的，部署完托管者后用pip进行安装即可，这里不在赘述。

策略逻辑

本策略根据EG两步法

第一步：序列同阶单整
第二步：OLS残差平稳

根据以上两步判断序列具有协整关系之后(若无协整关系则全平仓位不进行操作)

通过计算两个真实价格序列回归残差的0.9个标准差上下轨,并在价差突破上轨的时候做空价差,价差突破下轨的时候做多价差并在回归至标准差水平内的时候平仓

用Python进行代码实现

第一步，我们需要创建一个函数用来协整检验，以下代码中需要用到关于statsmodels库的知识，请读者参阅它的官方文档：http://www.statsmodels.org/stable/examples/index.html

# 创建一个函数用来协整检验
def cointegration_check(series01, series02):

    urt_rb2001 = ts.adfuller(np.array(series01), 1)[1]
    urt_rb2005 = ts.adfuller(np.array(series01), 1)[1]

    # 同时平稳或不平稳则差分再次检验
    if (urt_rb2001 > 0.1 and urt_rb2005 > 0.1) or (urt_rb2001 < 0.1 and urt_rb2005 < 0.1):
        urt_diff_rb2001 = ts.adfuller(np.diff(np.array(series01)), 1)[1]
        urt_diff_rb2005 = ts.adfuller(np.diff(np.array(series01), 1))[1]

        # 同时差分平稳进行OLS回归的残差平稳检验
        if urt_diff_rb2001 < 0.1 and urt_diff_rb2005 < 0.1:
            matrix = np.vstack([series02, np.ones(len(series02))]).T
            beta, c = np.linalg.lstsq(matrix, series01)[0]
            resid = series01 - beta * series02 - c
            if ts.adfuller(np.array(resid), 1)[1] > 0.1:
                result = 0.0
            else:
                result = 1.0
            return beta, c, resid, result
        else:
            result = 0.0
            return 0.0, 0.0, 0.0, result

    else:
        result = 0.0
        return 0.0, 0.0, 0.0, result

第二步，我们需要计算协整的相关数据，主要是为了得到残差的标准差上下轨数值，代码如下：

# 计算协整
def cointegration_calc():

    # 使用发明者量化交易类库
    obj = ext.NewPositionManager()

    # 展示两个价格序列的协整检验的结果
    beta, c, resid, result = cointegration_check(close_01, close_02)
    
    # 如果返回协整检验不通过的结果则全平仓位等待
    if not result:
        print('协整检验不通过,全平所有仓位')
        obj.CoverAll()
        return
    
    # 计算残差的标准差上下轨
    mean = np.mean(resid)
    up = mean + 0.9 * np.std(resid)
    down = mean - 0.9 * np.std(resid)
    
    # 计算新残差
    resid_new = close_01[-1] - beta * close_02[-1] - c

这里需要注意的是，我们需要用到发明者量化平台的国内商品期货模版，模版地址为：https://www.fmz.cn/strategy/24288 各位在发明者量化策略编写页面进行编码时，需要把此模版先复制到自己的策略库，然后在回测时勾选上，这里请各位读者注意

关于如何部署托管者和机器人，请参考我之前的文章：https://www.fmz.cn/bbs-topic/4140

想购买自己云计算服务器部署托管者的读者，可以参考这篇文章：https://www.fmz.cn/bbs-topic/2848

完整的策略代码：

import types
import numpy as np
import statsmodels.tsa.stattools as ts

# 我们首先创建一个函数用来协整检验
def cointegration_check(series01, series02):
    urt_rb2001 = ts.adfuller(np.array(series01), 1)[1]
    urt_rb2005 = ts.adfuller(np.array(series01), 1)[1]

    # 同时平稳或不平稳则差分再次检验
    if (urt_rb2001 > 0.1 and urt_rb2005 > 0.1) or (urt_rb2001 < 0.1 and urt_rb2005 < 0.1):
        urt_diff_rb2001 = ts.adfuller(np.diff(np.array(series01)), 1)[1]
        urt_diff_rb2005 = ts.adfuller(np.diff(np.array(series01), 1))[1]

        # 同时差分平稳进行OLS回归的残差平稳检验
        if urt_diff_rb2001 < 0.1 and urt_diff_rb2005 < 0.1:
            matrix = np.vstack([series02, np.ones(len(series02))]).T
            beta, c = np.linalg.lstsq(matrix, series01)[0]
            resid = series01 - beta * series02 - c
            if ts.adfuller(np.array(resid), 1)[1] > 0.1:
                result = 0.0
            else:
                result = 1.0
            return beta, c, resid, result
        else:
            result = 0.0
            return 0.0, 0.0, 0.0, result

    else:
        result = 0.0
        return 0.0, 0.0, 0.0, result

# 初始化合约数据
def init():

    # 订阅螺纹钢的2001合约与2005合约，并且取得发明者量化平台当前周期的所有收盘价
    exchange.SetContractType("rb2001")
    records = exchange.GetRecords()
    close_01 = records.Close

    exchange.SetContractType("rb2005")
    records = exchange.GetRecords()
    close_02 = records.Close

# 计算协整
def cointegration_calc():

    # 使用发明者量化交易类库
    obj = ext.NewPositionManager()

    # 展示两个价格序列的协整检验的结果
    beta, c, resid, result = cointegration_check(close_01, close_02)
    
    # 如果返回协整检验不通过的结果则全平仓位等待
    if not result:
        print('协整检验不通过,全平所有仓位')
        obj.CoverAll()
        return
    
    # 计算残差的标准差上下轨
    mean = np.mean(resid)
    up = mean + 0.9 * np.std(resid)
    down = mean - 0.9 * np.std(resid)
    
    # 计算新残差
    resid_new = close_01[-1] - beta * close_02[-1] - c

def trade():
    obj = ext.NewPositionManager() # 使用发明者量化交易类库    

    # 此处用来获取持仓信息
    positions = exchange.GetPosition() # 获取持仓数组
    if len(positions) == 0: # 如果持仓数组的长度是0
        return 0 # 证明是空仓，返回0
    for i in range(len(positions)): # 遍历持仓数组
        if (positions[i]['Type'] == PD_LONG) or (positions[i]['Type'] == PD_LONG_YD):
            position_01_long = 1 # 将position_01_long标记为1

        elif (positions[i]['Type'] == PD_SHORT) or (positions[i]['Type'] == PD_SHORT_YD):
            position_01_short = -1 # 将position_01_short标记为-1

    if not position_01_long = 1 and not position_01_short = -1:

        # 上穿上轨时做空新残差
        if resid_new > up:

            obj.OpenLong("rb2001", 1) # 买开
            obj.OpenShort("rb2005", 1) # 卖开
            
        # 下穿下轨时做多新残差
        if resid_new < down:

            obj.OpenLong("rb2005", 1) # 买开            
            obj.OpenShort("rb2001", 1) # 卖开

    # 新残差回归时平仓
    elif position_01_short = -1:

        if resid_new <= up:
            obj.CoverAll()
            print('价格回归,平掉所有仓位')

        # 突破下轨反向开仓
        if resid_new < down:
            obj.OpenLong("rb2001", 1) # 买开
            obj.OpenShort("rb2005", 1) # 卖开

    elif position_01_long = 1:
        if resid_new >= down:
            obj.CoverAll()
            print('价格回归,平所有仓位')

        # 突破上轨反向开仓
        if resid_new > up:
            obj.OpenLong("rb2005", 1) # 买开            
            obj.OpenShort("rb2001", 1) # 卖开

def main():
    while True:
        trade()
        Sleep(1000)