本科时代自己撸的一套简易股票量化回测框架，供新手入门学习研究。支持不同标的、多参数的批量回测，用于快速开发迭代交易策略。
数据接口使用的是聚宽量化平台的API。
import numpy as np
import datetime
import jqdata
import pandas as pd
import matplotlib.pyplot as plt
import json #字符转换
import matplotlib.ticker as mtk
import math
import statsmodels.tsa.ar_model as sm
import scipy.signal as signal
from scipy.interpolate import interp1d
import talib
matplotlib.rcParams['axes.unicode_minus']=False #图像中负号显示不正常
策略部分：
def Trading_strategy(i,code,date,current_date,current_price,Highprice,position,open_price,ATRstop,FilterTimes,Window,fee,ROR,ROR_open):
####策略参数####
price = get_price(code, end_date = date, frequency='1d', fields=('close','high','low','open'),count= 40)
KAMA = talib.KAMA(price['close'],Window)
ATR = talib.ATR(price['high'],price['low'],price['close'], Window)[-1]
Filter = talib.STDDEV(KAMA[-Window:],Window)[-1]
#平仓
if position == 1:
if KAMA[-1]-KAMA[-2] < -Filter*FilterTimes:
position,open_price,Highprice,ROR_open = Market_out(open_price,position,current_date,current_price,fee,ROR,ROR_open)
if position == -1:
if KAMA[-1]-KAMA[-2] > Filter*FilterTimes:
position,open_price,Highprice,ROR_open = Market_out(open_price,position,current_date,current_price,fee,ROR,ROR_open)
#开仓
if position == 0:
if KAMA[-1]-KAMA[-2] > Filter*FilterTimes:
position,open_price,Highprice = Market_in(current_date,code,1,current_price)
if KAMA[-1]-KAMA[-2] < -Filter*FilterTimes:
position,open_price,Highprice = Market_in(current_date,code,-1,current_price)
#追踪止损
if position > 0:
Highprice = max(Highprice,price['close'][-1])
if price['close'][-1] <= (Highprice - ATRstop*ATR):
#print ("追踪止损")
position,open_price,Highprice,ROR_open = Market_out(open_price,position,current_date,current_price,fee,ROR,ROR_open)
if position < 0:
Highprice = min(Highprice,price['close'][-1])
if price['close'][-1] >= (Highprice + ATRstop*ATR):
#print ("追踪止损")
position,open_price,Highprice,ROR_open = Market_out(open_price,position,current_date,current_price,fee,ROR,ROR_open)
return position,open_price,Highprice,ROR_open
回测功能函数：
#开仓函数
def Market_in(current_date,code,ratio,current_price):
open_price = current_price
position = ratio
Highprice = current_price
print ("开仓",current_date,position,current_price)
return position,open_price,Highprice
#平仓函数
def Market_out(open_price,position,current_date,current_price,fee,ROR,ROR_open):
if open_price == 0:
ROR_single = 0
else:
if position > 0:
ROR_single = (current_price-open_price)/open_price
if position < 0:
ROR_single = -(current_price-open_price)/open_price
ROR = (1+ROR_open)*abs(position)*(ROR_single+1) + (1+ROR_open)*(1-abs(position)) -1
open_price = 0
position = 0
Highprice = 0
ROR_open = (ROR+1)*(1-fee) - 1
print ("平仓",current_date,position,current_price)
return position,open_price,Highprice,ROR_open
#计算年化收益
def annualized_return(return_list):
n = len(return_list)
ar = ((return_list[-1])**(250/n)-1)
return ar
#计算最大回撤
def MaxDrawdown(return_list):
Dlist = -(np.maximum.accumulate(return_list) - return_list) / np.maximum.accumulate(return_list)
i = np.argmax((np.maximum.accumulate(return_list) - return_list) / np.maximum.accumulate(return_list))
if i == 0:
return 0
j = np.argmax(return_list[:i])
return Dlist,(return_list[j] - return_list[i]) / (return_list[j])
#计算夏普比率
def sharpe_ratio(ar,day_ror):
return_stdev = np.std(day_ror)
sharpe_ratio = (ar - 0.04) / (np.sqrt(250)*return_stdev)
return sharpe_ratio
################################主回测过程#################################
def Retest(code,start,end,fee,ATRstop,FilterTimes,window):
####基本参数####
datelist = [] #回测日期列表
Retest_data = get_price(code, start_date = start, end_date = end, frequency='1d', fields=('close'))['close']
datelist = Retest_data.index #回测日期列表
open_flag = 0 #开仓信号，1开多，-1开空，0平仓
ROR = 0 #策略收益率
ROR_list = [] #策略收益率序列
ROR_open = 0 #每次开仓前的累计收益率
position = 0 #持仓量
position_list = [] #持仓量序列
ROR_single = [] #单次持仓收益率
ROR_single_list = [] #单次持仓收益率序列
open_price = 0
ROR_open_record = 0
current_N = 0
Highprice = 0
HL = np.array(np.zeros(500),dtype=complex)
Dlist = []
Dmax = 0 #计算最大回撤
profit_times = 0 #计算胜率和盈亏比
profit_price = 0
loss_times = 0
loss_price = 0
#开始回测
for i in range(len(datelist)):
#print (Retest_data.index[i])
if i > 0:
#每日数据
date = Retest_data.index[i]
current_date = Retest_data.index[i]
current_price = get_price(code, end_date = current_date, frequency='1d', fields=('close'),count= 2)['close'].values[-1]
position,open_price,Highprice,ROR_open = Trading_strategy(i,code,date,current_date,current_price,Highprice,position,open_price,ATRstop,FilterTimes,window,fee,ROR,ROR_open)
if ROR_open_record != ROR_open:
if ROR_open_record > ROR_open:
loss_times += 1
loss_price += ROR_open - ROR_open_record
if ROR_open_record < ROR_open:
profit_times += 1
profit_price += ROR_open - ROR_open_record
ROR_open_record = ROR_open
#单次持仓收益率计算
if open_price == 0:
ROR_single = 0
else:
if position > 0:
ROR_single = (current_price-open_price)/open_price
if position < 0:
ROR_single = -(current_price-open_price)/open_price
ROR = (1+ROR_open)*abs(position)*(ROR_single+1) + (1+ROR_open)*(1-abs(position)) -1
position_list.append(position)
ROR_single_list.append(ROR_single)
ROR_list.append(ROR)
return_list = [i+1 for i in ROR_list]
ar = annualized_return(return_list)
Dlist,Dmax = MaxDrawdown(return_list)
Benchmark = (Retest_data-Retest_data[0])/Retest_data[0]
day_ror = [] #策略每日收益率
for i in range(len(return_list)):
if i == 0 :
day_ror.append(0)
else:
day_ror.append(((return_list[i] - return_list[i-1])/return_list[i-1]))
sharpe_r = sharpe_ratio(ar,day_ror)
#收益率回撤图
fig = plt.figure(figsize=(15,18))
ax4 = fig.add_subplot(612)
Dlist_draw = [x*100 for x in Dlist]
ax4.bar(datelist,Dlist_draw,label='drawdown',width = 5,color='c')
ax4.grid(True)
legend(loc='upper right')
#ax4.set_xlabel('Date',fontsize=15)
ax4.set_ylabel('Drawdown Ratio',fontsize=15)
fmt='%.2f%%'
yticks = mtk.FormatStrFormatter(fmt)
ax4.yaxis.set_major_formatter(yticks)
#回测收益率图
#fig, ax1 = plt.subplots(figsize=(12,6))
ax1 = fig.add_subplot(312)
Benchmark_draw = [x*100 for x in Benchmark]
ROR_list_draw = [x*100 for x in ROR_list]
ax1.plot(datelist,Benchmark_draw,label='benchmark')
ax1.plot(datelist,ROR_list_draw,label='strategy')
ax1.grid(True)
legend(loc='upper right')
#ax1.set_xlabel('Date',fontsize=15)
ax1.set_ylabel('Rate of Return',fontsize=15)
fmt='%.2f%%'
yticks = mtk.FormatStrFormatter(fmt)
ax1.yaxis.set_major_formatter(yticks)
ROR1 = pd.DataFrame(ROR_list_draw)
#回测持仓图
#fig, ax2 = plt.subplots(figsize=(12,3))
ax2 = fig.add_subplot(615)
ax2.plot(datelist,position_list,label='position', color='pink')
ax2.grid(True)
legend(loc='upper right')
ax2.set_xlabel('Date',fontsize=15)
ax2.set_ylabel('Position',fontsize=15)
plt.show()
#回测单次持仓盈亏图
fig, ax3 = plt.subplots(figsize=(12,3))
ax3.plot(datelist,ROR_single_list,label='rate of return')
ax3.grid(True)
legend(loc='upper left')
ax3.set_xlabel('Date',fontsize=15)
ax3.set_ylabel('ROR single time',fontsize=15)
plt.show()
return Benchmark[-1],ROR_list[-1],ar,Dmax,sharpe_r,profit_times,loss_times,profit_price,loss_price
执行函数：
import datetime
from dateutil.relativedelta import relativedelta
start = '2018-01-01 00:00:00'
end = '2019-01-01 00:00:00'
code = ['000016.XSHG','000300.XSHG','000905.XSHG']
# code = ['IF9999.CCFX']
# 批量回测参数组
x = [2]
y = [0.1]
z = [15]
####################
fee_list = [0.001] #手续费+滑点，千一
for element in code:
for i in x:
for j in y:
for k in z:
for fee in fee_list:
print (i,j,k,f)
bm,tv,ar,Dmax,sharpe_r,profit_times,loss_times,profit_price,loss_price = Retest(element,start,end,fee,i,j,k)
print ("Strategy yield: %.2f%%"%(tv*100))
print ("Benchmark yield: %.2f%%"%(bm*100))
print ("Maximum drawdown ratio: %.2f%%"%(Dmax*100))
print ('盈利次数：%d, 亏损次数：%d'%(profit_times, loss_times))
print ('胜率: %f'%(profit_times/(profit_times+loss_times)))
print ('盈利金额（占初始金额比例）：%f, 亏损金额（占初始金额比例）：%f'%(profit_price, -loss_price))
print ('盈亏比: %f'%(-profit_price/loss_price))
输出：
