量化策略系列教程：11Arbitrage 套利策略

套利策略感覺是個老生常談的話題了，那今天小編就給大家來一個老策略新代碼，大家跑跑看，證經社量化社區 - 證經社 ，記住，一定要在掘金的環境下跑哈~

1.策略介紹及邏輯

策略在過去經驗的統計驗證基礎上，認為兩個股票或期貨的價格比值符合統計穩定規律，如果價差超出某一閥值後，存在套利機會。本示例中，使用IF1703，IF1704當做標的。a：IF1703，b:IF1704。

代碼中用lna - lnb 來表示的價格比。對價格取ln可以降低數據出現異值的可能性，提升數據的可用性。

交易規則：

監測lna - lnb

- 如果大於0:

- 如果價格比值超出設定閥值，且低於止損閥值，空a多b；

- 如果超出止損閥值，平空a,平多b;

- 如果小於0:

- 如果小於負的設定閥值，且高於負的止損閥值，多a空b；

- 如果小於負的止損閥值，平多a,平空b；

- 在價格比接近於1時，認為回歸，平掉套利倉位

- 防止單腿成交：

在check_positions中，判斷如果4個tick數據（每隻代碼各2個tick更新）後，仍然只有單腿成交，則平掉單腿倉位。

註：只是一個示例套利的程序框架，實際應用中需要按照具體情況修改。

用同類品種跨期的價格差，可以直接用兩者之間的價格相減。

2.策略代碼

2.1配置文件【strategy_sa.ini】（提示ini配置文件，需要保存成UTF8格式）

[strategy]nusername= npassword= n;回測模式nmode=4ntd_addr=localhost:8001nstrategy_id= n;訂閱代碼注意及時更新nsubscribe_symbols=CFFEX.IF1703.tick,CFFEX.IF1704.tick,CFFEX.IF1703.bar.15,CFFEX.IF1704.bar.15nn[backtest]nstart_time=2017-03-01 09:00:00nend_time=2017-03-08 16:00:00n;策略初始資金ninitial_cash=10000000nn;委託量成交比率，默認=1（每個委託100%成交）ntransaction_ratio=1nn;手續費率，默認=0（不計算手續費）ncommission_ratio=0nn;滑點比率，默認=0（無滑點）nslippage_ratio=0nn[ss]nbar_type=15nwindow_size=20ntrade_exchange_a=CFFEXntrade_secid_a=IF1703ntrade_unit_a=1ntrade_exchange_b=CFFEXntrade_secid_b=IF1704ntrade_unit_b=1ntick_size=0.2nnsigma=2.34nn##############################################################n# logger settingsn##############################################################n[loggers]nkeys=rootnn[logger_root]nlevel=DEBUGnhandlers=console,filenn[handlers]nkeys=console,filenn[handler_file]nclass=handlers.RotatingFileHandlernargs=(strategy_sa.log,a,1000,5)nformatter=simplenn[handler_console]nclass=StreamHandlernargs = (sys.stdout,)nformatter=simplenn[formatters]nkeys = simplenn[formatter_simple]nformat=%(asctime)s - %(name)s - %(levelname)s - %(message)sndatefmt=n

2.2策略文件【strategy_sa.py】

#!/usr/bin/env pythonn# encoding: utf-8nnimport loggingnimport timenimport numpy as npnfrom collections import dequenfrom gmsdk import *nfrom math import logneps = 1e-6nnclass StatArb(StrategyBase):n n statistics arbitrage demon n def __init__(self, *args, **kwargs):n super(StatArb, self).__init__(*args, **kwargs)n logging.basicConfig(format=%(asctime)s - %(levelname)s: %(message)s)n self.tick_size = self.config.getfloat(ss, tick_size) or 0.2nn self.threshold = self.config.getfloat(ss, sigma) or 2.34n self.significant_diff = self.threshold * 0.0015 ## 3/4 sigman self.stop_lose_threshold = self.threshold * 0.002 ## 2 * sigmann self.trade_exchange_a = self.config.get(ss, trade_exchange_a) or CFFEXn self.trade_secid_a = self.config.get(ss, trade_secid_a)n self.trade_unit_a = self.config.get(ss, trade_unit_a) or 1n self.last_price_a = 0.0nn self.trade_exchange_b = self.config.get(ss, trade_exchange_b) or CFFEXn self.trade_secid_b = self.config.get(ss, trade_secid_b)n self.trade_unit_b = self.config.get(ss, trade_unit_b) or 1n self.last_price_b = 0.0nn self.pos_side_up = Falsen self.pos_side_down = Falsenn self.window_size = 20nn self.close_buffer_symbol_a = deque(maxlen=self.window_size)n self.close_buffer_symbol_b = deque(maxlen=self.window_size)n self.at_risk = 0n self.bar_type = self.config.get(ss, bar_type)nn def on_tick(self, tick):n if tick.sec_id == self.trade_secid_a:n self.last_price_a = tick.last_pricen elif tick.sec_id == self.trade_secid_b:n self.last_price_b = tick.last_pricenn self.check_position()nn def on_bar(self, bar):nn if bar.bar_type == 15:n #print (bar.sec_id == IF1704)nn if bar.sec_id == IF1703:n a = 1n #print (bar)nn self.close_buffer_symbol_a.append(bar.close)nn elif bar.sec_id == IF1704: #self.trade_secid_b:n b = 1n a = 1n #print (bar.close)nn #print (bar.sec_id == self.trade_secid_a)n self.close_buffer_symbol_b.append(bar.close)n if a == 1 and b == 1:n self.algo_action()n #print (action)nn def open_side_up(self):n self.open_short(self.trade_exchange_a, self.trade_secid_a, self.last_price_a, self.trade_unit_a)n self.open_long(self.trade_exchange_b, self.trade_secid_b, self.last_price_b, self.trade_unit_b)n self.pos_side_up = Truenn def close_side_up(self):n self.close_short(self.trade_exchange_a, self.trade_secid_a, self.last_price_a, self.trade_unit_a)n self.close_long(self.trade_exchange_b, self.trade_secid_b, self.last_price_b, self.trade_unit_b)n self.pos_side_up = Falsenn def open_side_down(self):n self.open_long(self.trade_exchange_a, self.trade_secid_a, self.last_price_a, self.trade_unit_a)n self.open_short(self.trade_exchange_b, self.trade_secid_b, self.last_price_b, self.trade_unit_b)n self.pos_side_down = Truenn def close_side_down(self):n self.close_long(self.trade_exchange_a, self.trade_secid_a, self.last_price_a, self.trade_unit_a)n self.close_short(self.trade_exchange_b, self.trade_secid_b, self.last_price_b, self.trade_unit_b)n self.pos_side_down = Falsenn def algo_action(self):n # type: () -> objectnn latest_a = self.close_buffer_symbol_a.pop()n lna = log(latest_a)nn latest_b = self.close_buffer_symbol_b.pop()n lnb = log(latest_b)nnn diff = lna - lnbn #print (diff)n #print(self.stop_lose_threshold)nn if diff > self.stop_lose_threshold:n self.close_side_up()n #print (a)n elif diff > self.significant_diff and diff < self.stop_lose_threshold:n self.open_side_up()n #print (b)n elif diff < - self.stop_lose_threshold:n self.close_side_down()n #print (c)n elif diff < - self.significant_diff and diff > - self.stop_lose_threshold:n self.open_side_down()n #print (d)n elif abs(diff) < self.threshold:n if self.pos_side_up:n self.close_side_up()n if self.pos_side_down:n self.close_side_down()nnn def check_position(self):n """ TODO: check if one leg position and close it """n ps = self.get_positions()n count = len(ps)n if count % 2 != 0:n self.at_risk += 1n ## if more than 4 tick data passed, need to force quitn if self.at_risk < 4:n returnnn for p in ps:n if self.pos_side_up:n if p.side == OrderSide_Ask:n self.close_short(p.exchange, p.sec_id, self.last_price_a, p.volume)n elif p.side == OrderSide_Bid:n self.close_long(p.exchange, p.sec_id, self.last_price_b, p.volume)n if self.pos_side_down:n if p.side == OrderSide_Ask:n self.close_short(p.exchange, p.sec_id, self.last_price_b, p.volume)n elif p.side == OrderSide_Bid:n self.close_long(p.exchange, p.sec_id, self.last_price_a, p.volume)n else:n self.at_risk = 0nnnif __name__ == __main__:n #import pdb; pdb.set_trace()n dm = StatArb(config_file=strategy_sa.ini)n ret = dm.run()n print("Statistics Arbitrage: ", dm.get_strerror(ret))n

如果想了解相關的python函數和掘金介面函數，走下方通道：

http://zjshe.cn/q/forum.php?mod=viewthread&tid=59&extra=page%3D1

有不了解的可以給小編留言哦，小編會細心為大家解答~