AI 量化交易-CTA 回测系统之推进分析下的均线策略（五）

推进分析下的均线策略

Part1 手把手教你搭建CTA回测系统框架

import talib
import pandas as pd
import numpy as np
import tushare as ts
import datetime

class backtest_days:

    def __init__(self,TimestampSignal,TimestampPrice,opencost,closecost):
        self.TimestampSignal=TimestampSignal
        self.TimestampPrice=TimestampPrice
        self.opencost=opencost
        self.closecost=closecost

    # calculate the returns from long operations
    # given open price, close price, cost of open the position, cost for close the position
    # 做多 openprice 开仓价，closeprice 平仓价，opencost 开仓成本，closecost 平仓成本
    def long(self,openprice,closeprice,opencost,closecost):
        return ((closeprice*(1-closecost))/(openprice*(1+opencost))-1)

    # calculate the returns from short operations
    # given open price, close price, cost of open the position, cost for close the position        
    def short(self,openprice,closeprice,opencost,closecost):
        nominator=openprice*(1-opencost)-closeprice*(1+closecost)
        denominator=openprice*(1-opencost)
        return (nominator/denominator)

    # find the first set of transaction: buy and then sell, or short and then cover 
    # i.e. find the first example of pair 1,-1 or -1,1
    # if either actionindex1=='end' or actionindex2=='end', then we shouldn't preceed forward
    # action is a munpy array or a pandas series    
    # 找交易对（0 开空、1 开多）
    def trading_index_finder(self,actions):
        # get the index for the first nonzero
        if len(actions)==1:
            actionindex1='end'
        else:
            actionindex1=0

        try:
            if (actions.iloc[actionindex1]==1):
                actionindex2=np.where(actions==0)[0]
                actionindex2=actionindex2[actionindex2>actionindex1][0]

            if (actions.iloc[actionindex1]==0):
                actionindex2=np.where(actions==1)[0]
                actionindex2=actionindex2[actionindex2>actionindex1][0]
        except:
            actionindex2='end'       

        return [actionindex1,actionindex2]   

    # to make decision every 30 mins and the dicided if we need to hold the position for another day
    # r is how much percentage rate do we need to do stoploss 
    # Type is LongOnly ShortOnly LongShort 
    def crossday_backtest_everyday(self):
        datevalues=[[self.TimestampSignal['Date'].iloc[0],0]]
        TimestampSignal=self.TimestampSignal
        while (len(TimestampSignal)>0):

            # find the first set of transaction: buy and then sell, or short and then cover 
            [actionindex1,actionindex2]=self.trading_index_finder(TimestampSignal['Prediction'])

            # that is the end of a loop, break out of the loop
            if (actionindex1=='end'):
                break

            if (actionindex2=='end'):
                [Date1,Action1]=TimestampSignal[['Date','Prediction']].iloc[actionindex1]
                [Date2,Action2]=TimestampSignal[['Date','Prediction']].iloc[-1]

            if ((actionindex1!='end') & (actionindex2!='end')):
                [Date1,Action1]=TimestampSignal[['Date','Prediction']].iloc[actionindex1]
                [Date2,Action2]=TimestampSignal[['Date','Prediction']].iloc[actionindex2]

            # get the trading price 
            DTtradep=self.TimestampPrice[(self.TimestampPrice['Date']>=Date1) & (self.TimestampPrice['Date']<=Date2)]

            if (Action1==1):
                payoff=self.long(DTtradep['close'].iloc[0],DTtradep['close'].iloc[-1],self.opencost,self.closecost)
                datevalues.append([DTtradep['Date'].iloc[-1],payoff]) 

            if (Action1==(0)):
                payoff=self.short(DTtradep['close'].iloc[0],DTtradep['close'].iloc[-1],self.opencost,self.closecost)
                datevalues.append([DTtradep['Date'].iloc[-1],payoff])

            # update the TimestampSignal
            TimestampSignal=TimestampSignal[TimestampSignal['Date']>=DTtradep['Date'].iloc[-1]]
            #print (DTtradep['Date'].iloc[-1])

        return datevalues

Part2 基于推进分析的双均线策略回测与评价

[稀牛学院 x 网易云课程]《AI量化交易》课程资料 by [老喵]

def SearchForBestCoef(TimestampPrice,coef):
    valuemax=0
    for m in coef:
        for n in coef:
            if m<n:         
                ShortSMA=talib.SMA(TimestampPrice['close'].values,m)
                LongSMA=talib.SMA(TimestampPrice['close'].values,n)
                TradingSignal=(ShortSMA>LongSMA).astype(int)#默认NaN>NaN是0
                TimestampPrice['Prediction']=TradingSignal
                TimestampSignal=TimestampPrice[['Date','Prediction']]
                TimestampSignal=TimestampSignal.sort_values(by='Date')
                TimestampSignal=TimestampSignal.reset_index(drop=True)
                backtestresult=backtest_days(TimestampSignal,TimestampPrice,opencost=3/10000,closecost=3/10000)                
                datevalues=backtestresult.crossday_backtest_everyday()
                datevalues=pd.DataFrame(datevalues)
                datevalues.columns=['Date','Return']
                if datevalues['Return'].sum()>valuemax:
                    valuemax=datevalues['Return'].sum()
                    coefmax=[m,n]
    return coefmax

#TimestampPrice=ts.get_h_data('000300',index=True,start='2005-01-01')
#TimestampPrice.to_pickle(r'C:\Users\maozh\Desktop\其他\上网课\TimestampPrice.pkl')
TimestampPrice=pd.read_pickle('/home/ds/data/quantification/TimestampPrice.pkl')
TimestampPrice['Date']=TimestampPrice.index
TimestampPrice['Date']=[x.date() for x in TimestampPrice['Date']]
TimestampPrice=TimestampPrice[TimestampPrice['Date']>=datetime.date(2010,4,19)]
TimestampPrice=TimestampPrice.sort_values(by='Date')
TimestampPrice=TimestampPrice.reset_index(drop=True)
TimestampPrice=TimestampPrice[TimestampPrice['Date']<datetime.date(2012,6,30)]
TimestampPrice.head()

opencost=3/10000
closecost=3/10000
coef=[2, 3, 5, 8, 13, 21, 34, 55, 89, 144]

OOSTimestampSignal=[]
for TradingDay in TimestampPrice['Date']:
    if TradingDay>datetime.date(2010,12,30):
        print(TradingDay)
        InSample=TimestampPrice[TimestampPrice['Date']<TradingDay]
        [m,n]=SearchForBestCoef(InSample,coef)

        OutofSample=TimestampPrice[TimestampPrice['Date']<=TradingDay]
        ShortSMA=talib.SMA(OutofSample['close'].values,m)
        LongSMA=talib.SMA(OutofSample['close'].values,n)
        if ShortSMA[-1]>LongSMA[-1]:
            OOSTimestampSignal.append([TradingDay, 1])
        else:
            OOSTimestampSignal.append([TradingDay, 0])

OOSTimestampSignal=pd.DataFrame(OOSTimestampSignal)
OOSTimestampSignal.columns=['Date','Prediction']

2010-12-31

/opt/conda/lib/python3.5/site-packages/ipykernel_launcher.py:9: RuntimeWarning: invalid value encountered in greater
  if __name__ == '__main__':
/opt/conda/lib/python3.5/site-packages/ipykernel_launcher.py:10: SettingWithCopyWarning: 
A value is trying to be set on a copy of a slice from a DataFrame.
Try using .loc[row_indexer,col_indexer] = value instead

See the caveats in the documentation: http://pandas.pydata.org/pandas-docs/stable/indexing.html#indexing-view-versus-copy
  # Remove the CWD from sys.path while we load stuff.

2011-01-04
2011-01-05
2011-01-06
2011-01-07
2011-01-10
    ...
2011-01-11
2011-05-09
2011-05-10

backtestresult=backtest_days(OOSTimestampSignal,TimestampPrice,opencost=3/10000,closecost=3/10000)                
datevalues=backtestresult.crossday_backtest_everyday()
datevalues=pd.DataFrame(datevalues)
datevalues.columns=['Date','Return']                
datevalues['Value']=datevalues['Return'].cumsum()
datevalues.set_index(['Date'],inplace=True)

import matplotlib.pyplot as plt
plt.figure()
datevalues['Value'].plot()
plt.show()

OOSTimestampSignal.head()

为者常成，行者常至