策略在过去经验的统计验证基础上,认为两个股票或期货的价格比值符合统计稳定规律,如果价差超出某一阀值后,存在套利机会。本示例中,使用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]
- username=
- password=
- ;回测模式
- mode=4
- td_addr=localhost:8001
- strategy_id=
- ;订阅代码注意及时更新
- subscribe_symbols=CFFEX.IF1703.tick,CFFEX.IF1704.tick,CFFEX.IF1703.bar.15,CFFEX.IF1704.bar.15
- [backtest]
- start_time=2017-03-01 09:00:00
- end_time=2017-03-08 16:00:00
- ;策略初始资金
- initial_cash=10000000
- ;委托量成交比率,默认=1(每个委托100%成交)
- transaction_ratio=1
- ;手续费率,默认=0(不计算手续费)
- commission_ratio=0
- ;滑点比率,默认=0(无滑点)
- slippage_ratio=0
- [ss]
- bar_type=15
- window_size=20
- trade_exchange_a=CFFEX
- trade_secid_a=IF1703
- trade_unit_a=1
- trade_exchange_b=CFFEX
- trade_secid_b=IF1704
- trade_unit_b=1
- tick_size=0.2
- sigma=2.34
- ##############################################################
- # logger settings
- ##############################################################
- [loggers]
- keys=root
- [logger_root]
- level=DEBUG
- handlers=console,file
- [handlers]
- keys=console,file
- [handler_file]
- class=handlers.RotatingFileHandler
- args=('strategy_sa.log','a',1000,5)
- formatter=simple
- [handler_console]
- class=StreamHandler
- args = (sys.stdout,)
- formatter=simple
- [formatters]
- keys = simple
- [formatter_simple]
- format=%(asctime)s - %(name)s - %(levelname)s - %(message)s
- datefmt=
- #!/usr/bin/env python
- # encoding: utf-8
- import logging
- import time
- import numpy as np
- from collections import deque
- from gmsdk import *
- from math import log
- eps = 1e-6
- class StatArb(StrategyBase):
- '''
- statistics arbitrage demo
- '''
- def __init__(self, *args, **kwargs):
- super(StatArb, self).__init__(*args, **kwargs)
- logging.basicConfig(format='%(asctime)s - %(levelname)s: %(message)s')
- self.tick_size = self.config.getfloat('ss', 'tick_size') or 0.2
- self.threshold = self.config.getfloat('ss', 'sigma') or 2.34
- self.significant_diff = self.threshold * 0.0015 ## 3/4 sigma
- self.stop_lose_threshold = self.threshold * 0.002 ## 2 * sigma
- self.trade_exchange_a = self.config.get('ss', 'trade_exchange_a') or 'CFFEX'
- self.trade_secid_a = self.config.get('ss', 'trade_secid_a')
- self.trade_unit_a = self.config.get('ss', 'trade_unit_a') or 1
- self.last_price_a = 0.0
- self.trade_exchange_b = self.config.get('ss', 'trade_exchange_b') or 'CFFEX'
- self.trade_secid_b = self.config.get('ss', 'trade_secid_b')
- self.trade_unit_b = self.config.get('ss', 'trade_unit_b') or 1
- self.last_price_b = 0.0
- self.pos_side_up = False
- self.pos_side_down = False
- self.window_size = 20
- self.close_buffer_symbol_a = deque(maxlen=self.window_size)
- self.close_buffer_symbol_b = deque(maxlen=self.window_size)
- self.at_risk = 0
- self.bar_type = self.config.get('ss', 'bar_type')
- def on_tick(self, tick):
- if tick.sec_id == self.trade_secid_a:
- self.last_price_a = tick.last_price
- elif tick.sec_id == self.trade_secid_b:
- self.last_price_b = tick.last_price
- self.check_position()
- def on_bar(self, bar):
- if bar.bar_type == 15:
- #print (bar.sec_id == 'IF1704')
- if bar.sec_id == 'IF1703':
- a = 1
- #print ('bar')
- self.close_buffer_symbol_a.append(bar.close)
- elif bar.sec_id == 'IF1704': #self.trade_secid_b:
- b = 1
- a = 1
- #print (bar.close)
- #print (bar.sec_id == self.trade_secid_a)
- self.close_buffer_symbol_b.append(bar.close)
- if a == 1 and b == 1:
- self.algo_action()
- #print ('action')
- def open_side_up(self):
- self.open_short(self.trade_exchange_a, self.trade_secid_a, self.last_price_a, self.trade_unit_a)
- self.open_long(self.trade_exchange_b, self.trade_secid_b, self.last_price_b, self.trade_unit_b)
- self.pos_side_up = True
- def close_side_up(self):
- self.close_short(self.trade_exchange_a, self.trade_secid_a, self.last_price_a, self.trade_unit_a)
- self.close_long(self.trade_exchange_b, self.trade_secid_b, self.last_price_b, self.trade_unit_b)
- self.pos_side_up = False
- def open_side_down(self):
- self.open_long(self.trade_exchange_a, self.trade_secid_a, self.last_price_a, self.trade_unit_a)
- self.open_short(self.trade_exchange_b, self.trade_secid_b, self.last_price_b, self.trade_unit_b)
- self.pos_side_down = True
- def close_side_down(self):
- self.close_long(self.trade_exchange_a, self.trade_secid_a, self.last_price_a, self.trade_unit_a)
- self.close_short(self.trade_exchange_b, self.trade_secid_b, self.last_price_b, self.trade_unit_b)
- self.pos_side_down = False
- def algo_action(self):
- # type: () -> object
- latest_a = self.close_buffer_symbol_a.pop()
- lna = log(latest_a)
- latest_b = self.close_buffer_symbol_b.pop()
- lnb = log(latest_b)
- diff = lna - lnb
- #print (diff)
- #print(self.stop_lose_threshold)
- if diff > self.stop_lose_threshold:
- self.close_side_up()
- #print ('a')
- elif diff > self.significant_diff and diff < self.stop_lose_threshold:
- self.open_side_up()
- #print ('b')
- elif diff < - self.stop_lose_threshold:
- self.close_side_down()
- #print ('c')
- elif diff < - self.significant_diff and diff > - self.stop_lose_threshold:
- self.open_side_down()
- #print ('d')
- elif abs(diff) < self.threshold:
- if self.pos_side_up:
- self.close_side_up()
- if self.pos_side_down:
- self.close_side_down()
- def check_position(self):
- """ TODO: check if one leg position and close it """
- ps = self.get_positions()
- count = len(ps)
- if count % 2 != 0:
- self.at_risk += 1
- ## if more than 4 tick data passed, need to force quit
- if self.at_risk < 4:
- return
- for p in ps:
- if self.pos_side_up:
- if p.side == OrderSide_Ask:
- self.close_short(p.exchange, p.sec_id, self.last_price_a, p.volume)
- elif p.side == OrderSide_Bid:
- self.close_long(p.exchange, p.sec_id, self.last_price_b, p.volume)
- if self.pos_side_down:
- if p.side == OrderSide_Ask:
- self.close_short(p.exchange, p.sec_id, self.last_price_b, p.volume)
- elif p.side == OrderSide_Bid:
- self.close_long(p.exchange, p.sec_id, self.last_price_a, p.volume)
- else:
- self.at_risk = 0
- if __name__ == '__main__':
- #import pdb; pdb.set_trace()
- dm = StatArb(config_file='strategy_sa.ini')
- ret = dm.run()
- print("Statistics Arbitrage: ", dm.get_strerror(ret))
3.代码涉及的函数代码
3.1 python函数及package
功能 | 函数原型 | 参数 | 返回值 | ||
参数名 | 含义 | ||||
sys | 提供了一系列有关Python运行环境的变量和函数。 | ||||
sys.argv[0] | 当前程序名 | ||||
sys.argv | 获取当前正在执行的命令行参数的参数列表(list)。 | sys.argv | sys.argv[1] | 第一个参数 | |
sys.argv[2] | 第二个参数 | ||||
arrow | 标准的时间日期库。 | ||||
ta-lib | 被广泛应用的金融市场数据分析的库 | ||||
pandas | Python Data Analysis Library 或 pandas 是基于NumPy 的一种工具,该工具是为了解决数据分析任务而创建的 | ||||
numpy | 一套用于支持科学计算的python第三方库 | ||||
time | 返回当前时间的时间戳 | time.time() | 返回当前时间的时间戳 | ||
len | 返回对象(字符、列表、元组等)长度或项目个数。 | len(s) | s | 对象 | 返回对象长度。 |
append | 用于在列表末尾添加新的对象。 | list.append(obj) | obj | 添加到列表末尾的对象。 | 该方法无返回值,但是会修改原来的列表。 |
3.2掘金接口函数
功能 | 函数原型 | 参数 | 返回值 | |||
参数名 | 类型 | 说明 | ||||
on_bar | 响应Bar事件,收到Bar数据后本函数被调用。 | on_bar(bar) | bar | bar | bar数据 | 无 |
open_long | 异步开多仓,以参数指定的symbol、价和量下单。如果价格为0,为市价单,否则为限价单。策略类和交易服务类都提供该接口 | open_long(exchange, sec_id, price, volume) | exchange | string | 交易所代码, 如上交所SHSE | 委托下单生成的Order对象 |
sec_id | string | 证券代码,如浦发银行600000 | ||||
price | float | 委托价,如果price=0,为市价单,否则为限价单 | ||||
volume | float | 委托量 | ||||
close_long | 异步平多仓接口,以参数指定的exchange, 证券代码sec_id, 价和量下单。如果价格为0,为市价单,否则为限价单。策略类和交易服务类都提供该接口。 | close_long(exchange, sec_id, price, volume) | exchange | string | 交易所代码, 如上交所SHSE | 委托下单生成的Order对象 |
sec_id | string | 证券代码,如浦发银行600000 | ||||
price | float | 委托价,如果price=0,为市价单,否则为限价单 | ||||
volume | float | 平仓量 | ||||
open_short | 异步开空仓,以参数指定的symbol、价和量下单。如果价格为0,为市价单,否则为限价单。策略类和交易服务类都提供该接口 | open_short(exchange, sec_id, price, volume) | exchange | string | 交易所代码, 如上交所SHSE | 委托下单生成的Order对象 |
sec_id | string | 证券代码,如浦发银行600000 | ||||
price | float | 委托价,如果price=0,为市价单,否则为限价单 | ||||
volume | float | 委托量 | ||||
close_short | 异步平空仓接口,以参数指定的exchange, 证券代码sec_id, 价和量下单。如果价格为0,为市价单,否则为限价单。策略类和交易服务类都提供该接口。 | close_long(exchange, sec_id, price, volume) | exchange | string | 交易所代码, 如上交所SHSE | 委托下单生成的Order对象 |
sec_id | string | 证券代码,如浦发银行600000 | ||||
price | float | 委托价,如果price=0,为市价单,否则为限价单 | ||||
volume | float | 平仓量 | ||||
get_last_n_dailybars | 提取单个代码的最新n条DailyBar数据, 策略类和行情服务类都提供该接口。 | get_last_n_dailybars(symbol, n, end_time='') | symbol | string | 证券代码, 带交易所代码以确保唯一,如SHSE.600000 | Bar列表 |
n | int | 提取的数据条数 | ||||
end_time | string | 指定截止时间, 如2015-10-30 15:00:00 | ||||
get_dailybars | 提取指定时间段的历史Bar数据,支持单个代码提取或多个代码组合提取。策略类和行情服务类都提供该接口。 | get_dailybars(symbol_list, begin_time, end_time) | symbol_list | string | 证券代码, 带交易所代码以确保唯一,如SHSE.600000,同时支持多只代码 | DailyBar列表 |
begin_time | string | 开始日期, 如2015-10-19 | ||||
end_time | string | 结束日期, 如2015-10-30 | ||||
get_position | 查询当前策略指定symbol(由交易所代码和证券ID组成)和买卖方向的持仓信息。策略类和交易服务类都提供该接口。 | get_position(exchange, sec_id, side); | exchange | string | 交易所代码 | Position对象,持仓信息 |
sec_id | string | 证券代码 | ||||
side | int | 买卖方向 |