本地简易股票量化回测框架

本科时代自己撸的一套简易股票量化回测框架，供新手入门学习研究。支持不同标的、多参数的批量回测，用于快速开发迭代交易策略。
数据接口使用的是聚宽量化平台的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))
                        

输出：