前言
本文搭建了基于指数移动平均线的交易系统，并将其同时应用在螺纹主力、铜主力、菜粕主力以及白糖主力上。并通过网格优化，再利用遗传算法进行参数优化。
import pandas as pdimport numpy as npimport mathimport UserStringimport numpy.random as rdfrom itertools import productimport seaborn as snsimport talib as taimport matplotlibimport matplotlib.colors as colimport matplotlib.cm as cmimport matplotlib.pyplot as pltfrom mpl_toolkits.mplot3d import Axes3Dfrom CAL.PyCAL import fontimport quartz_futures as qffrom quartz_futures.api import *sns.set_style('whitegrid')rd.seed(20161223)投资策略——基于指数移动平均线的交易系统
说明:策略均来自于TB，本文将其在优矿期货回测平台上实现。
传统的均线交叉系统不是很完善，有些不完善。
- 第一：传统均线交叉系统使用的是简单均线，计算均线时每根K线数据所占的权重相同。实际上，大多数交易者都认为距离当前K线越近的数据对当前的价格影响更大一些，也就是说距离当前K线越近的K线数据应该占有更多的权重，所以本文使用指数移动均线。
- 第二：传统均线交叉系统多头开仓逻辑为短期均线上穿长期均线，空头开仓逻辑为短期均线下穿长期均线，逻辑过于简单
本系统将开仓条件做出如下改进：
- 多头开仓条件：短期均线上穿长期均线同时长期均线大于更长期均线的值
- 空头开仓条件：短期均线下穿长期均线同时长期均线小于更长期均线的值
投资标的的选择
为了达到分散风险的目的，我们应该选取价格走势相关性较小的品种，本文将标的池选为RBM0, CUM0, RMM0, SRM0等主力合约。首先我们获取2014年1月1日止2014年12月31日这些合约的日行情，因为API貌似没有直接拿主力数据的接口，所以我们直接先在回测平台里面拿数据。
sample = ['RBM0', 'CUM0', 'RMM0', 'SRM0']import quartz_futures as qffrom quartz_futures.api import *data = {i:[] for i in sample}for i in range(len(sample)): universe = sample[i] # 策略期货合约 print universe start = '2014-01-01' # 回测开始时间 end = '2014-12-31' # 回测结束时间 capital_base = 1000000 # 初试可用资金 refresh_rate = 1 # 调仓周期 freq = 'd' # 调仓频率：m-> 分钟；d-> 日 def initialize(futures_account): # 初始化虚拟期货账户，一般用于设置计数器，回测辅助变量等。 pass def handle_data(futures_account): # 回测调仓逻辑，每个调仓周期运行一次，可在此函数内实现信号生产，生成调仓指令。 symbol = get_symbol(universe) hist = get_symbol_history(symbol=symbol, field='closePrice', time_range=1) price = hist[symbol]['closePrice'][-1] data[universe].append(price) bt, perf = qf.backtest.backtest(universe=universe, start=start, end=end, initialize=initialize, handle_data=handle_data, capital_base=capital_base, refresh_rate=refresh_rate, freq=freq)RBM0
CUM0
RMM0
SRM0
数据拿好之后，我们便可获得这些标的之间的相关系数矩阵。
data = pd.DataFrame(data)cor = data.corr()cor
cdict = cm.datad['RdYlGn']my_color = {}green = []red = []for i in range(len(cdict['green'])): green.append(tuple([cdict['green'][i][0], cdict['green'][-i-1][1], cdict['green'][-i-1][1]])) red.append(tuple([cdict['red'][i][0], cdict['red'][-i-1][1], cdict['red'][-i-1][1]]))my_color['green'] = greenmy_color['red'] = redmy_color['blue'] = cdict['blue']my_cmap = matplotlib.colors.LinearSegmentedColormap('my_colormap',my_color,256)def plot_heatmap(returns): label = returns.columns.tolist() x_label = label[:] label.reverse() # rho_mat = returns.corr() fig = plt.figure(figsize=(5, 5)) ax = fig.add_subplot(111) ax = sns.heatmap(returns, cmap=my_cmap, linewidths=.1, annot=True, center=0.0, cbar=False, ax=ax, annot_kws={"size": 9}) ax.set_yticklabels(label, minor=False, fontproperties=font, fontsize=10) ax.set_xticklabels(x_label, minor=False, fontproperties=font, fontsize=10)plot_heatmap(cor)
单策略多品种回测
参数优化
- 网格优化
那么参数设为多少合适？指数移动平均线交易系统涉及到三个参数，短周期，长周期，更长的周期。下面我们设置为更长的周期为25，来对短周期和长周期进行网格化的遍历，设定短周期范围为5至10，长周期范围为15-25。 - 遗传算法
通过机器学习的方法来进行参数优化求解。
def cross_over(fast_ma, slow_ma): if fast_ma[-1] > slow_ma[-1] and fast_ma[-2] < slow_ma[-2]: return True else: return Falsedef backtest(i, j): universe = ['RBM0', 'CUM0', 'RMM0', 'SRM0'] # 策略期货合约 margin = [0.07, 0.05, 0.07, 0.06] multi= [10, 5, 10, 10] weight = [0.25, 0.25, 0.25, 0.25] start = '2015-01-01' # 回测开始时间 end = '2016-12-19' # 回测结束时间 capital_base = 1500000 # 初试可用资金 refresh_rate = 1 # 调仓周期 freq = 'd' # 调仓频率：m-> 分钟；d-> 日 avglen1 = i avglen2 = j avglen3 = 25 rlength = 4 def initialize(futures_account): # 初始化虚拟期货账户，一般用于设置计数器，回测辅助变量等。 futures_account.barssinceentry = [0, 0, 0, 0] futures_account.symbol = ['RB1605', 'CU1605', 'RM1605', 'SR1605'] def handle_data(futures_account): # 回测调仓逻辑，每个调仓周期运行一次，可在此函数内实现信号生产，生成调仓指令。 for i in range(len(universe)): if get_symbol(universe[i]) != futures_account.symbol[i]: long_position = futures_account.position.get(futures_account.symbol[i], dict()).get('long_position', 0) short_position = futures_account.position.get(futures_account.symbol[i], dict()).get('short_position', 0) if long_position != 0: # print futures_account.current_date, '主力更换， 平仓' # print get_symbol(universe) order(futures_account.symbol[i], -long_position, 'close') if short_position != 0: # print '主力更换， 平仓' # print get_symbol(universe) order(futures_account.symbol[i], short_postition, 'close') futures_account.symbol[i] = get_symbol(universe[i]) else: for i in range(len(universe)): symbol = futures_account.symbol[i] long_position = futures_account.position.get(symbol, dict()).get('long_position', 0) short_position = futures_account.position.get(symbol, dict()).get('short_position', 0) data = get_symbol_history(symbol=symbol, field=[u'closePrice', u'highPrice', u'lowPrice', u'openPrice', 'volume'], time_range=avglen3+2) close = data[symbol]['closePrice'] low = data[symbol]['closePrice'] avg1 = ta.EMA(np.array(close), avglen1) avg2 = ta.EMA(np.array(close), avglen2) avg3 = ta.EMA(np.array(close), avglen3) buycon1 = cross_over(avg1, avg2) amount = int(futures_account.cash*weight[i]*0.2/(data[symbol]['openPrice'][-1]*margin[i]*multi[i])) Range = data[symbol]['highPrice'] - data[symbol]['lowPrice'] if buycon1 and avg2[-1] > avg3[-1]-50 and long_position == 0: # print futures_account.current_date, '开仓' order(symbol, amount, 'open') futures_account.barssinceentry[i] = 1 if long_position != 0 and avg1[-1] < avg2[-1]: # print futures_account.current_date, '平仓' order(symbol, -long_position, 'close') futures_account.barssinceentry[i] = 0 RangeL = ta.MA(np.array(Range), rlength) long_stop_price = low[-1] - Range[-1] if futures_account.barssinceentry >= 2: long_stop_price = long_stop_price + (low[-1] - long_stop_price)*0.25 if long_position != 0 and futures_account.barssinceentry >= 2 and low[-1] <= long_stop_price: # print futures_account.current_date, '止损' order(symbol, -long_position, 'close') if futures_account.barssinceentry[i] >= 1: futures_account.barssinceentry[i] += 1 bt, perf = qf.backtest.backtest(universe=universe, start=start, end=end, initialize=initialize, handle_data=handle_data, capital_base=capital_base, refresh_rate=refresh_rate, freq=freq) return perf生成网格
可以看到运行过程十分缓慢，还需耐心等待。
x = []y = []z = []fast = range(5, 11)slow = range(15, 26)for index in product(fast, slow): x.append(index[0]) y.append(index[1]) z.append(backtest(index[0], index[1]))max_drawdown = pd.DataFrame(index=fast, columns=slow)for i in range(66): max_drawdown.ix[x[i], y[i]] = z[i]['max_drawdown']max_drawdown
sharpe = pd.DataFrame(index=fast, columns=slow)for i in range(66): sharpe.ix[x[i], y[i]] = z[i]['sharpe']sharpe
annualized_return = pd.DataFrame(index=fast, columns=slow)for i in range(66): annualized_return.ix[x[i], y[i]] = z[i]['annualized_return']annualized_return
最后本文选择了5日短线，20日长线来作为回测的参数。下面画出其累计收益率的走势，以及具体风险指标。
final = z[5]cumulative_returns = final['cumulative_returns']fig = plt.figure(figsize=(12, 4))ax = fig.add_subplot(111)ax.plot(np.array(cumulative_returns))ax.set_xlim(-1, len(cumulative_returns))ax.set_xticks(range(0, len(cumulative_returns), 50))xlabel = ax.set_xticklabels([cumulative_returns.index[i] for i in ax.get_xticks()])t = ax.annotate(u'年化收益：%s'%(final['annualized_return']), xy=(1, 1), xytext=(1, 1.3), fontproperties=font, fontsize=11)t = ax.annotate(u'夏普比率：%s'%(final['sharpe']), xy=(1, 1), xytext=(70, 1.3), fontproperties=font, fontsize=11)t = ax.annotate(u'最大回撤：%s'%(final['max_drawdown']), xy=(1, 1), xytext=(139, 1.3), fontproperties=font, fontsize=11)t = ax.annotate(u'收益波动率：%s'%(final['volatility']), xy=(1, 1), xytext=(203, 1.3), fontproperties=font, fontsize=11)t = ax.set_ylabel(u'累积收益率', fontproperties=font, fontsize=11)t = ax.set_xlabel(u'时间', fontproperties=font, fontsize=11)t = ax.set_title(u'丐版累积收益图', fontproperties=font, fontsize=11)
遗传算法
那么有没有不遍历的方法进行参数优化？本文以机器学习的遗传算法为例，旨在提供一个思路，具体效果可能并没有那么理想，而且容易陷入局部最优解，后续还将多加调整。代码参考了python实现的遗传算法实例（一）。原文中是单变量的最大值求解，这里修改成双变量的函数最大值求解。
几点说明
- 编码方式：我们考虑两个参数，短周期与长周期。短周期的范围为[5，10]，长周期范围为[15，25]。采用普通的二进制编码，码长为6位。
- 解码方式：对于6位的二进制编码可以表示[0，63]内的任意正整数，所以还需构造映射进行转换至相应的区间。
- 适应度函数：这里的适应度函数即回测，函数的因变量为策略的年化收益率。
import numpy.random as rdimport numpy as npimport talib as taimport quartz_futures as qffrom quartz_futures.api import *import mathimport UserStringrd.seed(20161223)# 编码def encode(x): return UserString.MutableString('{0:06b}'.format(x))# 解码def decode_x(x): return eval(str('0b'+x))*6/63+5def decode_y(y): return eval(str('0b'+x))*11/63+15# 交叉def gene_crossover(gene1, gene2): #个体间交叉，实现基因交换 cpoint = rd.randint(0,len(gene1)) gene1_new = gene1[0:cpoint]+gene2[cpoint:] gene2_new = gene2[0:cpoint]+gene1[cpoint:] return gene1_new, gene2_newdef cross(group, pc): for i in range(len(group)-1): if rd.random() < pc: temp = gene_crossover(group[i][0], group[i+1][0]) group[i][0] = temp[0] group[i+1][0] = temp[1] if rd.random() < pc: temp = gene_crossover(group[i][1], group[i+1][1]) group[i][1] = temp[0] group[i+1][1] = temp[1]# 基因突变def mutation(group, pm): for i in range(len(group)): if rd.random() < pm: mpoint = rd.randint(0,len(group[i][0])-1) group[i][0][mpoint] = str(abs(int(group[i][0][mpoint])-1)) if rd.random() < pm: mpoint = rd.randint(0,len(group[i][1])-1) group[i][1][mpoint] = str(abs(int(group[i][1][mpoint])-1))# 自然选择# 群体适应值计算def calobjvalue(group): #计算目标函数值 objvalue = [] for gene in group: i = decode_x(gene[0]) j = decode_y(gene[1]) objvalue.append(backtest(i, j)['annualized_return']) return objvalue#转化为适应值，目标函数值越大越好，负值淘汰。def calfitvalue(objvalue): fitvalue = [] temp = 0.0 cmin = 0; for i in range(len(objvalue)): if(objvalue[i] + cmin > 0): temp = cmin + objvalue[i] else: temp = 0.0 fitvalue.append(temp) return fitvaluedef cumsum(fitvalue): fitvalue = np.cumsum(fitvalue).tolist() return fitvalue#找出适应函数值中最大值，和对应的个体def best(group, fitvalue): bestindividual = group[0] bestfit = fitvalue[0] for i in range(1,len(group)): if(fitvalue[i] > bestfit): bestfit = fitvalue[i] bestindividual = group[i] return [bestindividual, bestfit]#自然选择（轮盘赌算法）def selection(group, fitvalue): newfitvalue = [] totalfit = sum(fitvalue) for i in range(len(fitvalue)): newfitvalue.append(fitvalue[i] / totalfit) newfitvalue = cumsum(newfitvalue) ms = [] for i in range(len(group)): ms.append(rd.random()) #random float list ms ms.sort() fitin = 0 newin = 0 newgroup = group while newin < len(group): if ms[newin] < newfitvalue[fitin]: newgroup[newin] = group[fitin] newin = newin + 1 else: fitin = fitin + 1 group = newgroup# 生成初始种群range1=[0, 63]range2=[0, 63]bestanser=Nonelimit=0#init init_ansers#set the number of ansers in a groupnumber_of_group=10group=[]#random choose number_of_group ansers in rangefor i in range(7): x=encode(rd.randint(range1[0],range1[1])) y=encode(rd.randint(range2[0],range2[1])) group.append([x, y])for i in range(6): results = [] objvalue = calobjvalue(group) #计算目标函数值 fitvalue = calfitvalue(objvalue) #计算个体的适应值 [bestindividual, bestfit] = best(group, fitvalue) #选出最好的个体和最好的函数值 print bestindividual, bestfit results.append([bestfit, decode_x(bestindividual[0]), decode_y(bestindividual[1])]) #每次繁殖，将最好的结果记录下来 selection(group, fitvalue) #自然选择，淘汰掉一部分适应性低的个体 cross(group, 0.8) #交叉繁殖 mutation(group, 0.005) #基因突变['000000', '101011'] 0.5451
['000010', '101101'] 0.5451
['000010', '000010'] 0.5451
['000000', '000100'] 0.5451
可以看到6次繁衍后，遗传算法最后也收敛到了最优解上。
相关推荐
文章来源：
肖肖
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至2384272385@qq.com举报，一经查实，本站将立刻删除。