版本号：v2.1
本次迭代，添加了自动评估三均线最优组合功能。
这里的最优组合是按近一段时间，收益最大的最优组合为结果。
经过回测，整体比第一个版本提升了4.1%的收益。
后期将计划添加以收益加权最优组合计算均线组合，并添加自动选股功能，请期待！
策略说明：
所谓的三进兵，是指三条EMA均线组合的策略
交易原则：
系统由三条EMA均线组合而成，分别为小均线、中均线、大均线
当小均线金叉大均线、并且中均线位于大均线下方时，买入
当小均线死叉中均线，并且中均线位于大均线上方时，卖出
止损：当买入后，如果收盘价跌破中均线，止损
选股：本策略里没有做自动选股，而是手动挑选了一些
在研究里做了更多股票的研究，发现本策略并不是适合所有的股票的
所以，各位宽友如果有想法，可在此基础上做迭代，并分享出来，展示你的才华
说明：¶
- 所谓的三进兵，是指三条EMA均线组合的策略
交易原则：¶
- 系统由三条EMA均线组合而成，分别为小均线、中均线、大均线
- 当小均线金叉大均线、并且中均线位于大均线下方时，买入
- 当小均线死叉中均线，并且中均线位于大均线上方时，卖出
- 止损：当买入后，如果收盘价跌破中均线，止损
- 选股：对个股进行一段时间的回测，得出最优组合，并判断是否可达到正收益，从而判断是否适合本策略
- 各位宽友如果有想法，可在此基础上做迭代，并分享出来，展示你的才华
导入引用模块¶
import numpy as np
import pandas as pd
import datetime
from jqdata import *
from jqlib.technical_analysis import *
设置全局变量¶
# 记录成交历史
trade_history = {}
# 指定股票池，这里默认选择了创业板
stocks_pool = get_index_stocks('399006.XSHE')
编写公用函数¶
# 获取ma_min值
def get_ma(stock, ma_value, end_dt):
price = get_price(security=stock,
end_date=end_dt,
frequency='daily',
fields=['close'],
skip_paused=False,
fq='pre',
count=ma_value+10)['close']
ma = price[-ma_value:].mean()
return ma
# 获取ema值
def get_ema(stock, ma_value, end_dt):
ema = EMA(stock, check_date=end_dt, timeperiod=ma_value)
return ema[stock]
# 判断是否出现买入信息
def is_buy(stock, yesterday, before_yesterday, *ema):
ma_min = ema[0]
ma_med = ema[1]
ma_max = ema[2]
# 求出上一个交易日的ma_min，ma_med,ma_max的值
y_ma_min_value = get_ema(stock, ma_min, yesterday)
y_ma_med_value = get_ema(stock, ma_med, yesterday)
y_ma_max_value = get_ema(stock, ma_max, yesterday)
# 求出上上个交易日的ma_min，ma_med,ma_max的值
by_ma_min_value = get_ema(stock, ma_min, before_yesterday)
by_ma_med_value = get_ema(stock, ma_med, before_yesterday)
by_ma_max_value = get_ema(stock, ma_max, before_yesterday)
if (y_ma_min_value > y_ma_max_value) and (by_ma_min_value < by_ma_max_value) and (y_ma_med_value < y_ma_max_value):
return True
else:
return False
# 判断是否有卖出信息
def is_sell(stock, yesterday, before_yesterday, *ema):
ma_min = ema[0]
ma_med = ema[1]
ma_max = ema[2]
# 求出上一个交易日的ma_min，ma_med,ma_max的值
y_ma_min_value = get_ema(stock, ma_min, yesterday)
y_ma_med_value = get_ema(stock, ma_med, yesterday)
y_ma_max_value = get_ema(stock, ma_max, yesterday)
# 求出上上个交易日的ma_min，ma_med,ma_max的值
by_ma_min_value = get_ema(stock, ma_min, before_yesterday)
by_ma_med_value = get_ema(stock, ma_med, before_yesterday)
by_ma_max_value = get_ema(stock, ma_max, before_yesterday)
if (y_ma_min_value > y_ma_med_value) and (by_ma_min_value < by_ma_med_value) and (y_ma_med_value > y_ma_max_value):
return True
else:
return False
# 判断是否有卖出信息
def is_loss(stock, yesterday, *ema):
ma_min = ema[0]
ma_med = ema[1]
ma_max = ema[2]
# 昨日收盘价
close = get_price(security=stock,
end_date=yesterday,
frequency='daily',
fields=['open','close'],
skip_paused=False,
fq='pre',
count=10)['close'][-1]
# 求出上一个交易日的ma_min，ma_med,ma_max的值
y_ma_min_value = get_ema(stock, ma_min, yesterday)
y_ma_med_value = get_ema(stock, ma_med, yesterday)
y_ma_max_value = get_ema(stock, ma_max, yesterday)
if (close < y_ma_med_value) and (y_ma_med_value < y_ma_max_value):
return True
else:
return False
# 过滤掉有止影线和小实体阳线的时刻
def is_high_line(stock, end_dt):
price = get_price(security=stock,
end_date=end_dt,
frequency='daily',
fields=['open', 'close', 'high', 'low', 'volume', 'money'],
skip_paused=False,
fq='pre',
count=1)
open = price['open'][0]
close = price['close'][0]
high = price['high'][0]
low = price['low'][0]
o_c_ratio = (open-close)/close
h_c_ratio = (high-close)/close
if o_c_ratio > 0 and h_c_ratio < 0.02:
return True
else:
return False
交易函数¶
def trade(stock,ma_min,ma_med,ma_max,trade_days):
# 记录盈利
wine_loss_history = []
# 持仓
hold_list = {}
# 权重
weight = 0
# 总权重
last_weihgt = len(trade_days)
# 因为回测的是历史
for trade_day in trade_days:
# 权重累加
weight += 1
# 将要被使用的日期集合
the_days = get_trade_days(end_date=trade_day, count=5)
# 回测当天
today = the_days[-1]
# 上一个交易日
yesterday = the_days[-2]
# 上上个交易日
before_yesterday = the_days[-4]
# ========================卖出操作========================
sell_list = []
for stock,info in hold_list.items():
trade_day = info[0]
buy_price = info[1]
# 判断是否有卖出信号
re_value = is_sell(stock, yesterday, before_yesterday, ma_min,ma_med,ma_max)
# 判断是否触发止损信号
loss = is_loss(stock, yesterday, ma_min,ma_med,ma_max)
# 进行卖出操作
if re_value or loss:
sell_list.append(stock)
price = get_price(security=stock,
end_date=today, # 现实中成交按当天的开盘价交易
frequency='daily',
fields=['open','close'],
skip_paused=False,
fq='pre',
count=10)['open'][-1]
# 进行记录
trade_dic = {'stock':stock,
'buy_date':trade_day,
'buy_price':buy_price,
'sell_date':today,
'sell_price':price,
'ratio':(price-buy_price)/buy_price,
'MA':(ma_min,ma_med,ma_max),
'weight':weight,
'count':last_weihgt}
wine_loss_history.append(trade_dic)
# 从持仓中删除已经卖出的股票
for stock in sell_list:
del hold_list[stock]
# ========================卖出操作========================
# ========================买入操作========================
# 判断是否有买入信号
re_value = is_buy(stock, yesterday, before_yesterday, ma_min,ma_med,ma_max)
# 如果有买入信号，则买入
if re_value:
# 在今天的开盘时买入，参考的买入价格是昨天的收盘价
price = get_price(security=stock,
end_date=today, # 现实中按当天的开盘价交易
frequency='daily',
fields=['open','close'],
skip_paused=False,
fq='pre',
count=10)['open'][-1]
hold_list[stock] = [today, price]
# ========================买入操作========================
#========================将最后一次未卖出的也记录==========================
if stock in hold_list.keys():
sell_list.append(stock)
price = get_price(security=stock,
end_date=today, # 现实中成交按当天的开盘价交易
frequency='daily',
fields=['open','close'],
skip_paused=False,
fq='pre',
count=10)['open'][-1]
# 进行记录
trade_dic = {'stock':stock,
'buy_date':trade_day,
'buy_price':buy_price,
'sell_date':today,
'sell_price':price,
'ratio':(price-buy_price)/buy_price,
'MA':(ma_min,ma_med,ma_max),
'weight':weight,
'count':last_weihgt}
wine_loss_history.append(trade_dic)
#========================将最后一次未卖出的也记录==========================
# 返回交易记录
return wine_loss_history
回测¶
# 获利近三年的交易日期
days = get_trade_days(end_date=datetime.datetime.now(), count=250*2+100)
train_days = days[:-100]
trade_days = days[-100:]
s_time = datetime.datetime.now()
# 回测股票
trade_stock = '600600.XSHG'
# 保存回测记录
all_list = []
# 三进兵的三个值集合
ma_min1 = [3, 5, 7]
ma_med1 = [10, 20, 30]
ma_max1 = [40, 50, 60]
# 回测不同的三进兵组合
for ma1 in ma_min1:
for ma2 in ma_med1:
for ma3 in ma_max1:
re_dic = trade(trade_stock,ma1,ma2,ma3,train_days)
all_list = all_list + re_dic
# 输出各三进兵组合的值
df = pd.DataFrame(all_list,columns=['stock',
'buy_date',
'buy_price',
'sell_date',
'sell_price',
'ratio',
'MA',
'weight',
'count'])
e_time = datetime.datetime.now()
print(e_time-s_time)
df
计算最最优组合值¶
group = df.groupby(by=['MA'])
print('组合收益之和')
max_sum = group.sum()
max_sum = max_sum.sort_values(by=['ratio'],ascending=False).loc[:,['ratio']].head()
value = {'stock':trade_stock, 'MA':max_sum.index[0]}
print(value)
max_sum
print('各组合总收益对比图')
import matplotlib.pylab as plt
max_sum.T.plot(kind='bar')
plt.show()
print('组合收益标准差')
min_std = group.std()
min_std = min_std.loc[max_sum.index,['ratio']]
min_std = min_std.sort_values(by=['ratio'])
value = {'stock':trade_stock, 'MA':min_std.index[0]}
print(value)
min_std
print('各组合收益拆线图')
for name in max_sum.index:
table = df[df['MA']==name][['ratio']]
x = range(len(table.index))
plt.plot(x,table)
plt.legend(max_sum.index)
plt.show()
print('各组合收益对比图')
series = []
for name in max_sum.index:
table = df[df['MA']==name]['ratio'].values
series.append(table)
bar_df = pd.DataFrame(series,max_sum.index).T
bar_df.plot(kind='bar')
plt.show()
print('交易次数表')
bar_df.T
print('求加权后的表现-平均到每一次交易')
dic = []
for name, g in group:
weight_df = df[df['MA']==name]
w = weight_df['weight']
r = weight_df['ratio']
c = weight_df['weight'].sum()
v = w*r/c
dic.append({'ma':name, 'ratio':sum(v)})
w_df = pd.DataFrame(dic)
w_df = w_df.sort_values(by=['ratio'], ascending=False)
w_df.head()
print('求加权后的表现-平均到每天')
dic = []
count = int(df['count'].mean())
l = list(range(1,count))
for name, g in group:
weight_df = df[df['MA']==name]
w = weight_df['weight']
r = weight_df['ratio']
v = w*r/sum(l)
dic.append({'ma':name, 'ratio':sum(v)})
w_avge_df = pd.DataFrame(dic)
w_avge_df = w_avge_df.sort_values(by=['ratio'], ascending=False)
w_avge_df.head()
结果比较¶
result_dic = []
print('最大收益组合回测本年',max_sum.index[0])
ma1,ma2,ma3 = max_sum.index[0]
re_dic = trade(trade_stock,ma1,ma2,ma3,trade_days)
re_dic = pd.DataFrame(re_dic)
print('总盈利', re_dic.sum()['ratio'])
result_dic.append({'收益':re_dic.sum()['ratio']})
re_dic
print('最小标准差组合回测本年',min_std.index[0])
ma1,ma2,ma3 = min_std.index[0]
re_dic = trade(trade_stock,ma1,ma2,ma3,trade_days)
re_dic = pd.DataFrame(re_dic)
print('总盈利', re_dic.sum()['ratio'])
result_dic.append({'收益':re_dic.sum()['ratio']})
re_dic
print('最优加权组合回测本年',w_df['ma'].iloc[0])
ma1,ma2,ma3 = w_df['ma'].iloc[0]
re_dic = trade(trade_stock,ma1,ma2,ma3,trade_days)
re_dic = pd.DataFrame(re_dic)
print('总盈利', re_dic.sum()['ratio'])
result_dic.append({'收益':re_dic.sum()['ratio']})
re_dic
df = pd.DataFrame(result_dic,index=['最大收益组合'+str(max_sum.index[0]),
'最小标准差组合'+str(min_std.index[0]),
'最优加权组合'+str(w_df['ma'].iloc[0])])
df.T.plot(kind='bar')
plt.show()
df
改进方案¶
- ma1[3,5,9] ma2[10,20,30] ma3[40,50,60]
- 不同的股票，将拥有不同的ma1,ma2,ma3值
- 求出收益波动最小的方案
- 添加选股函数
- 增加更换代码的函数，要求可以空仓，可以更新ma三个值
- 在选股票或测试的时候，不添加止损
- 在收盘的时候买进或卖出，而不是在次日开盘交易
- 将可变的股票池变为外部可读取文件
- 从市场里筛选，有收盘价上穿ma60的，进行提醒
- 尝试结合macd等其他指标
- 提醒可以通过邮件的方式发送
- 如果能在平台运行，自动发送信息，那是再好不过
- 在卖出的时候，可以使用分钟回测，一旦卖出条件成熟，就下达卖出指令，避免大阴线下穿
- 对于不同时期成交的历史，做出加权，越靠近当前的，权值越重，这样才会越符合当前的模型参数值
- 为回测的数据增加窗口宽度
- 止损位的设置，如果中线不大于长线，但收盘价跌破了中线，则止损（一来可以停止亏损，二来在二次进攻时还可进入）
