什麼是統計套利
統計套利主要是在對歷史資料進行統計分析的基礎上，估計相關變數的概率分佈，並結合基本面資料進行分析以指導套利交易，其主要分為配對/一攬子交易、多因素模型、均值迴歸策略、協整以及針對波動率與相關性的建模，與傳統單邊投資方式相比，統計套利多空雙向持倉在處理大資金方面可以有效規避一部分風險，股指期貨中常用的期現套利和跨期套利以及商品期貨中跨品種套利都是常用的統計套利的例子，其中期限套利是強收斂既到期時期貨和現貨的價差必收斂，而其它策略則存在一些價差維持的風險。這裡主要介紹一種最基本也是最常用的一種策略-配對交易。
與針對收益率建模的β中性策略不同，基於協整關係的配對交易直接針對股票價格序列建模，若兩隻或多隻股票的股價存在長期穩定的線性關係，認為它們之間存在協整關係：
兩隻股票的價格序列存在協整關係一般包含兩個條件，一是歷史股價序列都是一階單整向量，二是序列的某種線性組合是平穩的，故構建的線性方程的殘差是平穩的。當某一對股票的價差既協整方程的殘差偏離到一定程度時開倉，價差迴歸均衡時則獲利了結，股價序列存在協整關係保證了長期價差將大概率回覆至均值附近，協整策略的主要思路如下：
Ø 找出相關性好的若干對股票
Ø 對殘差進行相應檢驗，首先要檢驗平穩性，其次檢驗殘差的自相關性和異方差性，若存在，考慮採用ARMA或GARCH模型進行濾波，之後進行正態檢驗，若不滿足正態性假定，可以考慮採用非引數方法來捕捉交易訊號
Ø 找出每對股票的長期均衡關係（協整關係）
Ø 將殘差分為可預測和不可預測的部分：
Ø 根據策略的執行結果進行引數調優
統計套利策略在擁有低波動率、低風險、漲跌停保護以及對於價差（相對概念）而非價格的預測等優點的同時，也存在一些顯著的缺點，其主要不足在於收益有限、存在套利機會較少（尤其是市場效率很高時），合約間“前者恆強，弱者恆弱的關係”會維持一定時間也帶來了投資風險，除此之外，交割風險，極端行情風險也應該考慮在內。
關於統計套利，更多的可以參考浙商證券融資融券與統計套利系列研究報告，知乎上Lightwing的回答也為我們提供了一種新的思路，既利用統計模型來測算實價，而根據實價來判斷交易的方向，長期的盈利空間則是一定的，為了便於大家更好地理解統計套利，我們在這裡提供了一個基礎的策略以供參考：
import pandas as pd
import numpy as np
# init方法是您的初始化邏輯。context物件可以在任何方法之間傳遞。
def init(context):
context.SWlist=['000806.SH','000807.SH','000808.SH']
context.industry_code=[]
context.moneyfund = 0
context.stock = ['000156.SZ','601928.SH']
#context.industry_code=pic_industry(context.SWlist,'close',10)
#print(context.industry_code)
task.monthly(rebalance,tradingday=1,time_rule=market_open(hour=0,minute=0))
task.daily(takeorder,time_rule=market_open(minute=0))
context.backdays = 20
# 日或分鐘或實時資料更新，將會呼叫這個方法
def handle_data(context, data_dict):
pass
def takeorder(context,data_dict):
clear_stocklst(context.stock,data_dict)
'''
是否持有貨幣基金
if context.moneyfund == 0:
order_target_percent('511880.SH',0.4)
context.moneyfund = 1
'''
if not len(context.stock) == 2:
return
close_prices1 =get_history(context.backdays,'1d','close')[context.stock[0]].values
close_prices2 =get_history(context.backdays,'1d','close')[context.stock[1]].values
price_ratio = close_prices1/close_prices2
interval_high = np.mean(price_ratio) + np.std(price_ratio)
interval_mid = np.mean(price_ratio)
interval_low = np.mean(price_ratio) - np.std(price_ratio)
curr_priceratio = close_prices1[-1]/close_prices2[-1]
if not len(context.portfolio.positions.keys()) == 0: # 進入平倉流程
if (context.stock[0] in context.portfolio.positions) and(curr_priceratio>interval_mid-0.1*np.std(price_ratio)):
order_target_percent(context.stock[0],0)
#print (context.now,"賣掉了",context.stock[0])
if (context.stock[1] in context.portfolio.positions) and(curr_priceratio<interval_high+0.1*np.std(price_ratio)):
order_target_percent(context.stock[1],0)
#print (context.now,"賣掉了",context.stock[1])
elif len(context.portfolio.positions.keys()) == 0:
if curr_priceratio <interval_low:
order_target_percent(context.stock[0],1)
#print (context.now,"開始持倉",context.stock[0])
elif curr_priceratio > interval_high:
order_target_percent(context.stock[1],1)
#print (context.now,"開始持倉",context.stock[1])
return
def get_history_panel(stock,days):
panel = pd.DataFrame()
field = ['close']
for char in field:
stkinfo = get_history(days,'1d',char)[stock]
df = pd.DataFrame({char:stkinfo.values},index=stkinfo.index)
panel = panel.join(df,how='right')
return panel
def clear_stocklst(stocklist,data_dict):
stkls=[]
for stkdm in stocklist:
if (not is_st(stkdm)) and (not data_dict[stkdm].sf==1) and (notis_delisting(stkdm)):
#security=get_securities(stkdm)
stkls.append(stkdm)
#print (stkdm + " " +security.symbol )
return stkls
def rebalance(context,data_dict):
industry_list=[]
for stock in context.portfolio.positions.keys():
#if not stock=='511880.SH':
order_target_percent(stock, 0)
context.industry_code=pic_industry(context.SWlist,'close',10)
industry_list = get_index_constituents(context.industry_code)
panel = pd.DataFrame()
for stk in industry_list:
stkprice = get_history(270,'1d','close')[stk].values
df = pd.DataFrame({stk:stkprice})
panel = pd.concat([panel,df],axis=1)
panelcorr = panel.corr().replace(1, 0)
pair = panelcorr[panelcorr == panelcorr.max().max()].count()
pair = pair[pair == 1].index.tolist()
context.stock = pair
print("您即將交易的兩隻股票為 %s 和 %s" %(pair[0],pair[1]))
defpic_industry(constituents_list,field,days):
multiseries = pd.DataFrame()
for stk in constituents_list:
stkinfo = get_history(days,'1d',field)[stk]
df = pd.DataFrame({stk:stkinfo.values},index=stkinfo.index)
multiseries = multiseries.join(df,how='right')
multiseries.loc['Return'] = multiseries.apply(lambda x:(x[-1]-x[0])/x[0])
multiseriesT=multiseries.T
multiseriesTsort=multiseriesT.sort(columns='Return',ascending=False)
industry_code=multiseriesTsort.index[0]
return industry_code
閱讀更多：