作者：chen_h
微信号 & QQ：862251340
微信公众号：coderpai

---

使用数据驱动进行配对交易：简单交易策略

配对交易是一个纯基于数学分析的一个非常好的例子。接下里的文章，我们将演示如何去利用数据来创建一个自动化配对交易策略。

基本原则

我们假设你有一对股票 X 和 Y，它们之间有一些基本的经济联系，例如这是两家生产百事可乐和可口可乐的，它们拥有相同产品的公司。你预计这两个公司的比率或者价格差异（也称为价差）随时间而保持不变。然而，由于某些愿意，比如市场变化，两个公司暂时的供求变化，一个正确机构的大量买单或者卖单，一个公司的重要消息面等等原因。这两个股票之间的价差可能会存在暂时的分歧。在这种情况下，一只股票上涨，另一只股票可能会相对下跌。如果你预计这种分歧会随着时间的推移恢复正常，那么你就可以进行配对交易。

当出现暂时的分歧时，配对交易将会出售表现优异的股票（上涨的股票），并且购买表现不佳的股票（下跌的股票）。因为你跟自己打赌，两只股票之间的差价最终会通过表现优异的股票回落或表现不佳的股票回升或者两者都进行同时进行，来达到原来的价差。不管是采用哪种行为，我们的配对交易都会从这些场景中赚钱。如果两只股票一起上涨或者下跌而不改变他们之间的价差，那么你将不会从这次交易中赚到钱，相反你可能会因此而造成亏损。

因此，配对交易是一种市场中性的交易策略，使得交易者能够从几乎任何市场环境中获得收益：上述趋势，下跌趋势或者震荡市。

概念解释：我们先来生成两个假股票数据

import numpy as np
import pandas as pd

import statsmodels
from statsmodels.tsa.stattools import coint
# just set the seed for the random number generator
np.random.seed(107)

import matplotlib.pyplot as plt

接下里，让我们生成一个伪造的数据 X ，并且通过从正太分布中绘制出它的每日收益。然后我们执行一个累积和来得到每天的 X 值。

# Generate daily returns
Xreturns = np.random.normal(0, 1, 100) 
# sum them and shift all the prices up
X = pd.Series(np.cumsum(
    Xreturns), name='X') 
    + 50
X.plot(figsize=(15,7))
plt.show()

现在我们生成与 X 有着经济关联的数据 Y ，所以 Y 的价格走势应该是与 X 相似的。我们通过对数据 X 进行建模并且添加一些从正态分布中抽取的随机噪声来建模。

noise = np.random.normal(0, 1, 100)
Y = X + 5 + noise
Y.name = 'Y'
pd.concat([X, Y], axis=1).plot(figsize=(15,7))
plt.show()

协整

协整性跟相关性非常相似，意味着两个系列之间的比值将以平均值为中心变化。我们可以将序列 X 和 Y 的关系描述如下：

Y = ⍺ X + e

其中，⍺ 是一个常数比率，e 是白噪声。

对于在两个时间序列之间交易的配对，期望比值随时间的推移必须收敛于平均值，即它们应该是协整的。我们上面构建的时间序列是协整的。我们现在讲绘制两者之间的比率，一边我们更加直观的看到结果。

(Y/X).plot(figsize=(15,7)) 
plt.axhline((Y/X).mean(), color='red', linestyle='--') 
plt.xlabel('Time')
plt.legend(['Price Ratio', 'Mean'])
plt.show()

协整测试

我们可以使用 statsmodels.tsa.stattools 包来进行方便测试。我们应该可以看到一个特别低的 p 值，因为这是我们人为构建的两个序列，他们在物理上可能是协整的。

# compute the p-value of the cointegration test
# will inform us as to whether the ratio between the 2 timeseries is stationary
# around its mean
score, pvalue, _ = coint(X,Y)
print pvalue

输出值为：1.81864477307e-17

注：相关性和协整性

相关性和协整性虽然在理论上很相似，但是并不一样。我们来看看一个相关性的例子，它不是协整性的，反之亦然。首先让我们来检查一下我们刚刚生成的系列的相关性。

X.corr(Y)

输出结果为：0.951

正如我们所期望的那样，这个值是非常高的。但是两个相关但不协整的序列怎么办呢？一个简单的例子是如下的两个序列：

ret1 = np.random.normal(1, 1, 100)
ret2 = np.random.normal(2, 1, 100)

s1 = pd.Series( np.cumsum(ret1), name='X')
s2 = pd.Series( np.cumsum(ret2), name='Y')

pd.concat([s1, s2], axis=1 ).plot(figsize=(15,7))
plt.show()
print 'Correlation: ' + str(X_diverging.corr(Y_diverging))
score, pvalue, _ = coint(X_diverging,Y_diverging)
print 'Cointegration test p-value: ' + str(pvalue)

Correlation: 0.998
Cointegration test p-value: 0.258

相关度：0.998

协整检验 p 值：0.258

一个没有相关性，但是协整的例子是一个正态分布数据和一个方形波数据。如下：

Y3[0:100] = 30
Y3[100:200] = 10
Y3[200:300] = 30
Y3[300:400] = 10
Y3[400:500] = 30
Y3[500:600] = 10
Y3[600:700] = 30
Y3[700:800] = 10

Y2.plot(figsize=(15,7))
Y3.plot()
plt.ylim([0, 40])
plt.show()
# correlation is nearly zero
print 'Correlation: ' + str(Y2.corr(Y3))
score, pvalue, _ = coint(Y2,Y3)
print 'Cointegration test p-value: ' + str(pvalue)

Correlation: 0.007546
Cointegration test p-value: 0.0

相关度：0.007546
协整检验 p 值：0.0

这两个序列的相关性非常低，但 p 值却显示出完美的协整性！

如何进行配对交易？

因为两个协整时间序列是彼此偏离和分开的（比如上面的 X 和 Y），所以有时候价差会很高，而有时候价差会很低。我们可以通过购买一种证券并且卖出另一种证券来进行配对交易。这样，如果正确一起下跌或者一起上涨，我们既不赚钱也不赔钱，因为我们的策略是市场中立的。

回到上面的 X 和 Y，以及它们的线 Y = ⍺ X + e，我们可以看到它们的比率（Y/X）是围绕它的平均值 ⍺ 来进行移动的，而我们赚取的钱就是从这个平均值 ⍺ 来的。为了做到这一点，我们会格外注意 X 和 Y 的比率，即 ⍺ 太高或者太低的时候：

• 延长比率：这是当 ⍺ 比平常小的时候，我们预计它的值会增加。在上面的例子中，我们通过购买 Y 并且出售 X 来对此进行投资。
• 缩小比率：这是当 ⍺ 比平常大的时候，我们预计它的值会减小。在上面的例子中，我们通过出售 Y 并且购买 X 来对此进行投资。

请注意，我们总是有这样一个 “对冲仓位”：如果卖出的证券亏损了，那么空头就会赚钱；如果证券的价值升高了，那么多头仓位就会赚钱，所以我们不受整体市场波动的影响。如果证券 X 和 Y 彼此相对移动，那么我们只会赚钱或者亏钱。

使用数据分析的方法来找到拥有如此规律的证券

做到这一点的最佳方式是从我们怀疑可能是拥有协整关系的证券开始，并进行统计测试。如果你只是对所有对进行统计测试，那么你将会受到多重比较偏差的影响。

多重比较偏差只是一个事实，即在运行多次测试时，错误的生成重要 p 值的机会增加了，因为我们正在进行大量测试。如果对随机数据运行 100 词测试，我们应该看到 5 个 p 值低于 0.05 的值。如果你将 n 只证券进行协整比较，你将执行 n(n-1)/2 次比较，并且你应该会看到很多不正确的 p 值，随着你测试次数的增加，这个数量也会增加。为了避免出现这种情况，我们可以选择一小部分，我们认为可能是协整的然后进行单独测试。这将帮助我们减少暴露在多重比较偏差风险之中。

所以我们试着找一些具有协整关系的证券。让我们从标普 500 指数中 —— 一篮子的美国大盘科技股中找一些具有协整关系的股票吧。这些股票运行在类似的细分市场，并可能具有协整价格。我们通过扫描这些股票，然后计算两两之间的协整，最后我们可以得到一个协整检验得分矩阵，一个 p 值得分矩阵，以及可以计算出 p 值小于 0.05 的任何配对。这种方法容易出现多重比较偏差，实际上证券应该经过第二次验证。但是在这里，为了这个例子，我们先忽略这个。

注意：我们的数据中包含市场基准（SPX）—— 市场会推动许多证券的流动，通常你可能会发现两个似乎是协整的证券；但实际上它们并不是彼此协整的，只是与当时的市场保持了一致而已。这被称为混杂变量，重要的是要检查你可能找到的任何关系中的市场参与情况，比如上证指数，深证指数，创业板指数等等。

from backtester.dataSource.yahoo_data_source import YahooStockDataSource
from datetime import datetime
startDateStr = '2007/12/01'
endDateStr = '2017/12/01'
cachedFolderName = 'yahooData/'
dataSetId = 'testPairsTrading'
instrumentIds = ['SPY','AAPL','ADBE','SYMC','EBAY','MSFT','QCOM',
                 'HPQ','JNPR','AMD','IBM']
ds = YahooStockDataSource(cachedFolderName=cachedFolderName,
                            dataSetId=dataSetId,
                            instrumentIds=instrumentIds,
                            startDateStr=startDateStr,
                            endDateStr=endDateStr,
                            event='history')
data = ds.getBookDataByFeature()['Adj Close']
data.head(3)

现在让我们尝试使用我们的方法找到协整对。

# Heatmap to show the p-values of the cointegration test
# between each pair of stocks
scores, pvalues, pairs = find_cointegrated_pairs(data)
import seaborn
m = [0,0.2,0.4,0.6,0.8,1]
seaborn.heatmap(pvalues, xticklabels=instrumentIds, 
                yticklabels=instrumentIds, cmap=’RdYlGn_r’, 
                mask = (pvalues >= 0.98))
plt.show()
print pairs

输出结果是：[(‘ADBE’, ‘MSFT’)]

从图中我们可以看出股票 ‘ADBE’ 和 ‘MSFT’ 是最具有协整的。为了更加放心，我们来看看它们的价格变化，以确保它确实是具有意义的。

S1 = data['ADBE']
S2 = data['MSFT']
score, pvalue, _ = coint(S1, S2)
print(pvalue)
ratios = S1 / S2
ratios.plot()
plt.axhline(ratios.mean())
plt.legend([' Ratio'])
plt.show()

这个比例看起来好像是为让着一个稳定的平均值移动的。绝对比率在统计学上并不是很有用，我们通过 z-score 来标准化我们的数据，会对我们的数据处理更有帮助。z-score 的定义如下：

Z Score (Value) = (Value — Mean) / Standard Deviation

警告

在实际过程中，通常这样做是为了将数据控制在一定的比例内，但是这样就将数据假设在一个基本分布里面了，比如正态分布等等。通常这样做事不会出现问题的，但是，我们都知道财务数据，或者说是股票数据不是正态分布的，所有我们必须非常小心，不要简单的进行数据假设。数据比例的真实分布可能会非常粗糙，并且容易出现各种极端的情况，从而搞乱了我们的模型并且导致了交易时的巨大损失。

def zscore(series):
    return (series - series.mean()) / np.std(series)

zscore(ratios).plot()
plt.axhline(zscore(ratios).mean())
plt.axhline(1.0, color=’red’)
plt.axhline(-1.0, color=’green’)
plt.show()

从图中我们可以看出，现在更容易观察这个比率就是围绕平均值移动的，但有时会倾向于与平均数有较大的一边，这也就是我们的交易机会。

现在我们已经介绍完了对配对交易的基础知识，并且基于历史价格确定了协整股票，接下来我们就可以尝试来开发一些交易信号了。首席那，让我们回顾一下使用数据技术来开发交易信号的步骤：

• 收集可靠的数据并且进行数据清洗
• 从数据中来选择一些数据特征来识别交易信号
• 特征可以选择移动平均值活价格数据，相关性或更加复杂信号 —— 或者结合一些新创建的特征来生成特征
• 使用这些特征来生成交易信号，即哪些指令是买入的，哪些指令是卖出的，哪些指令是中性的。

第一步：定义我们的问题

在这里，我们试图创建一个信号，告诉我们下一个时间点的比值是买入还是卖出，即我们的预测变量 Y 值：

Y = Ratio is buy (1) or sell (-1)

Y(t)= Sign( Ratio(t+1) — Ratio(t) )

注意：我们不需要预测时间的股票价格，甚至我们不需要知道真实的比率（尽管我们可以得到这个值），我们只需要知道下一步的操作方向就可以了。

第二步：收集可靠和准确的数据

我们可以使用雅虎的 API 来使用过去 10 年的交易日内（大约 2500 个数据点）：开盘价，收盘价，最高价，最低价和交易量。

第三步：分割数据

不要忘记这个超级重要的步骤来测试模型的准确性。我们正在使用以下的方式来分割训练集和测试集：

• 采用 7 年的数据作为训练集，也就是 70% 的数据；
• 采用 3 年的数据作为测试集，也就是 30% 的数据；

ratios = data['ADBE'] / data['MSFT']
print(len(ratios))
train = ratios[:1762]
test = ratios[1762:]

理想情况下，我们还可以制定验证集，但现在我们将跳过这个步骤。

第四步：特征工程

相关的特征可能是什么？我们想要预测比率移动的方向。我们已经看到，我们的两个股票时协整的，所以这个比率往往会左右摇晃并且回到平均水平。看起来我们的特征应该是确定比率的均值，从均值的当前价值与均值的差异能够产生我们的交易信号。

让我们使用以下功能：

• 60天移动平均比率：滚动平均值的测量
• 5天移动平均比率：平均值的当前值的测量
• 60天标准差
• z-score：（5d MA - 60d MA）/ 60d SD

ratios_mavg5 = train.rolling(window=5,
                               center=False).mean()
ratios_mavg60 = train.rolling(window=60,
                               center=False).mean()
std_60 = train.rolling(window=60,
                        center=False).std()
zscore_60_5 = (ratios_mavg5 - ratios_mavg60)/std_60
plt.figure(figsize=(15,7))
plt.plot(train.index, train.values)
plt.plot(ratios_mavg5.index, ratios_mavg5.values)
plt.plot(ratios_mavg60.index, ratios_mavg60.values)
plt.legend(['Ratio','5d Ratio MA', '60d Ratio MA'])
plt.ylabel('Ratio')
plt.show()

plt.figure(figsize=(15,7))
zscore_60_5.plot()
plt.axhline(0, color='black')
plt.axhline(1.0, color='red', linestyle='--')
plt.axhline(-1.0, color='green', linestyle='--')
plt.legend(['Rolling Ratio z-Score', 'Mean', '+1', '-1'])
plt.show()

Z-score 确实带来了比率的平均恢复性质。

第五步：模型选择

让我们从一个非常简单的模型开始。查看 z-score 图标，我们可以看到，无论 z-score 特征变得太高还是太低，它都趋于恢复。让我们使用 +1 和 -1 来作为我们的阈值太高或者太低，然后我们可以使用以下模型来生成交易信号：

• 当 z-score 低于 -1.0 时，比率就是购买的信号（+1），因为我们预计 z-score 会回升到 0 ，因此比率会增加；
• 当 z-score 高于 1.0 时，比率就是卖出的信号（-1），因为我们预计 z-score 会回落到 0，因此比率会下降；

第六步：训练，验证和优化

最后，让我们看看我们的模型实际上对真实数据做了什么？让我们来看看这个信号在实际比率上的样子。

# Plot the ratios and buy and sell signals from z score
plt.figure(figsize=(15,7))
train[60:].plot()
buy = train.copy()
sell = train.copy()
buy[zscore_60_5>-1] = 0
sell[zscore_60_5<1] = 0
buy[60:].plot(color=’g’, linestyle=’None’, marker=’^’)
sell[60:].plot(color=’r’, linestyle=’None’, marker=’^’)
x1,x2,y1,y2 = plt.axis()
plt.axis((x1,x2,ratios.min(),ratios.max()))
plt.legend([‘Ratio’, ‘Buy Signal’, ‘Sell Signal’])
plt.show()

交易信号似乎是合理的，我们似乎在高价或者更高价时得到了出售比率（红点），并在低价格（绿点）下降时买回。这对我们交易的实际股票意味着什么呢？让我们来仔细看看：

# Plot the prices and buy and sell signals from z score
plt.figure(figsize=(18,9))
S1 = data['ADBE'].iloc[:1762]
S2 = data['MSFT'].iloc[:1762]
S1[60:].plot(color='b')
S2[60:].plot(color='c')
buyR = 0*S1.copy()
sellR = 0*S1.copy()
# When buying the ratio, buy S1 and sell S2
buyR[buy!=0] = S1[buy!=0]
sellR[buy!=0] = S2[buy!=0]
# When selling the ratio, sell S1 and buy S2 
buyR[sell!=0] = S2[sell!=0]
sellR[sell!=0] = S1[sell!=0]
buyR[60:].plot(color='g', linestyle='None', marker='^')
sellR[60:].plot(color='r', linestyle='None', marker='^')
x1,x2,y1,y2 = plt.axis()
plt.axis((x1,x2,min(S1.min(),S2.min()),max(S1.max(),S2.max())))
plt.legend(['ADBE','MSFT', 'Buy Signal', 'Sell Signal'])
plt.show()

注意，我们有时在做空上面赚钱，有时又在做多上面赚钱，有时两者都赚钱。

我们对训练数据的信号反馈还是很满意的。我们来看看这个信号那个可以产生怎么样的利润。当比率较低时，我们可以制作一个购买 1 比率（购买 1 个 ADBE 股票和卖出比率 x 个 MSFT股票）的简单策略。当卖出 1 比率（卖出 1 个 ADBE 股票和买入比率 x 个 MSFT 比率）时，计算这些比率的 PnL 交易。

# Trade using a simple strategy
def trade(S1, S2, window1, window2):

    # If window length is 0, algorithm doesn't make sense, so exit
    if (window1 == 0) or (window2 == 0):
        return 0

    # Compute rolling mean and rolling standard deviation
    ratios = S1/S2
    ma1 = ratios.rolling(window=window1,
                               center=False).mean()
    ma2 = ratios.rolling(window=window2,
                               center=False).mean()
    std = ratios.rolling(window=window2,
                        center=False).std()
    zscore = (ma1 - ma2)/std

    # Simulate trading
    # Start with no money and no positions
    money = 0
    countS1 = 0
    countS2 = 0
    for i in range(len(ratios)):
        # Sell short if the z-score is > 1
        if zscore[i] > 1:
            money += S1[i] - S2[i] * ratios[i]
            countS1 -= 1
            countS2 += ratios[i]
        # Buy long if the z-score is < 1
        elif zscore[i] < -1:
            money -= S1[i] - S2[i] * ratios[i]
            countS1 += 1
            countS2 -= ratios[i]
        # Clear positions if the z-score between -.5 and .5
        elif abs(zscore[i]) < 0.5:
            money += countS1*S1[i] - S2[i] * countS2
            count = 0

    return money
trade(data['ADBE'].iloc[:1762], data['MSFT'].iloc[:1762], 60, 5)

输出结果为：751783.375

所以这个策略似乎是有利可图的！！现在我们可以通过改变我们的移动平均窗口，通过改变买入（卖出）和退出头寸等门槛来进一步优化，并检查验证数据的性能改进。

我们也可以尝试更加复杂的模型，如 Logistics 回归，支持向量机等来做出我们的 +1（-1）预测。

现在，让我们假设我们决定推出这个模型。

第七步：对测试数据进行回测

回测是非常简单的，我们可以使用我们上面的函数来查看测试数据上的 PnL（profit and loss）。

trade(data[‘ADBE’].iloc[1762:], data[‘MSFT’].iloc[1762:], 60, 5)

输出结果为：545262.868

该模型做的非常好！这使得我们的第一个简单的配对交易完成了。

构建盈亏序列（PNL，Profit and Loss）。盈亏序列是指当前的头寸映射到对应的历史数据序列上，即此金融产品的历史收益率序列上，获得历史上的各个时间点的盈亏值。由此可知，盈亏序列的建立必须满足两个条件，1.已经处理好的历史数据2.当前的头寸。经过处理的历史数据已经经过讨论，当前头寸可以作为参数直接输入获得，当然为了更好的结构性和扩展性的考虑我们可以把头寸写入配置文件中。用数学公式PnL=Position(头寸)*（）。

避免过拟合

在结束讨论之前，我们要特别提一下过拟合。数据模型过拟合是交易策略中最危险的陷阱。过度拟合算法可能在回测中表现出色，但对新的数据可能处理能力就比较差了 —— 这也就意味着模型没有真正揭示出数据的任何趋势特征，也没有真正的预测能力。我们举一个简单的例子。

在我们的模型中，我们使用滚动参数估计，并可能希望优化窗口长度。我们可能决定简单的迭代所有的可能，合理的窗口长度，并根据我们的模型的最佳表现来选择长度。下面我们编写一个简单的循环来根据 PnL 值选择一个最佳的长度。

# Find the window length 0-254 
# that gives the highest returns using this strategy
length_scores = [trade(data['ADBE'].iloc[:1762], 
                data['MSFT'].iloc[:1762], l, 5) 
                for l in range(255)]
best_length = np.argmax(length_scores)
print ('Best window length:', best_length)

输出结果为：(‘Best window length:’, 40)

现在，我们检查我们的模型在测试数据上面的性能，我们发现这个窗口长度远不是最优的！这是因为我们最初的选择显然是过度拟合了训练样本。

# Find the returns for test data
# using what we think is the best window length
length_scores2 = [trade(data['ADBE'].iloc[1762:], 
                  data['MSFT'].iloc[1762:],l,5) 
                  for l in range(255)]
print (best_length, 'day window:', length_scores2[best_length])
# Find the best window length based on this dataset, 
# and the returns using this window length
best_length2 = np.argmax(length_scores2)
print (best_length2, 'day window:', length_scores2[best_length2])

输出结果为：

(40, 'day window:', 1252233.1395)
(15, 'day window:', 1449116.4522)

明显在训练样本上面取得好结果的并不总是在未来会取得良好的结果。为了更加直观好玩，让我们绘制从两个数据集计算出来的长度分数。

plt.figure(figsize=(15,7))
plt.plot(length_scores)
plt.plot(length_scores2)
plt.xlabel('Window length')
plt.ylabel('Score')
plt.legend(['Training', 'Test'])
plt.show()

我们可以看到 20 - 50 之间的任何长度都是一个不错的选择。

为了避免过拟合，我们可以使用经济推理或者算法的性质来选择我们的窗口长度。当软我们也可以使用一些不用指定窗口长度的方法，比如，卡尔曼滤波。

---

来源：Medium