量化学习 | Tushare和Backtrader初探

Posted 2021-03-05 manwingloeng

昨日2020年03月12日时美股历史的第三次熔断，距离上一次熔断03月09号仅3天时间，受全球疫情的影响，股市出现恐慌性抛盘，导致A股也出现较为明显的震荡，A股能否走出独立行情还是不得而知了。

所以我最近学习了一下量化投资的东西，

量化平台Backtrader用起来还不错，A股历史行情数据可从Tushare中获取：

https://www.backtrader.com/

# Create a subclass of Strategy to define the indicators and logic
class SmaCross(bt.Strategy):
    # list of parameters which are configurable for the strategy
    params = dict(
        pfast=10,  # period for the fast moving average
        pslow=30   # period for the slow moving average
    )
    def __init__(self):
        super().__init__()
        sma1 = bt.ind.SMA(period=self.p.pfast)  # fast moving average
        sma2 = bt.ind.SMA(period=self.p.pslow)  # slow moving average
        self.crossover = bt.ind.CrossOver(sma1, sma2)  # crossover signal
    def next(self):
        if not self.position:  # not in the market
            if self.crossover > 0:  # if fast crosses slow to the upside
                self.order_target_size(target=1)  # enter long
                # self.buy()
        elif self.crossover < 0:  # in the market & cross to the downside
            self.order_target_size(target=0)  # close long position
            # self.close()
    

这是一个简单的均值策略，当10日均线超过30日均线时买入，30日均线超过10日均线卖出。

data_path = './data/'
if not os.path.exists(data_path):
    os.makedirs(data_path)
mytoken='your_token'
class Strategy_runner:
    def __init__(self, strategy, ts_code, start_date, end_date, data_path=data_path, pro=False, token=mytoken):
        self.ts_code = ts_code
        self.start_date = start_date
        self.end_date = end_date
        # convert to datetime
        self.start_datetime = datetime.strptime(start_date,'%Y%m%d')
        self.end_datetime = datetime.strptime(end_date,'%Y%m%d')
        if pro:
            csv_name = f'pro_day_{str(ts_code)}-{str(start_date)}-{str(end_date)}.csv'
        else:
            csv_name = f'day_{str(ts_code)}-{str(start_date)}-{str(end_date)}.csv'
        csv_path = os.path.join(data_path,csv_name)
        if os.path.exists(csv_path):
            if pro:
                self.df = pd.read_csv(csv_path)
            else:
                self.df = pd.read_csv(csv_path,index_col=0)
        else:
            if pro:
                ts.set_token(mytoken)
                self.pro = ts.pro_api()
                self.df = self.pro.daily(ts_code=self.ts_code, start_date=self.start_date, end_date=self.end_date)
                if not self.df.empty:
                    self.df.to_csv(csv_path, index=False)
            else:
                self.df = ts.get_hist_data(self.ts_code, str(self.start_datetime), str(self.end_datetime))
                if not self.df.empty:
                    self.df.to_csv(csv_path, index=True)
            
        self.df_bt = self.preprocess(self.df, pro)
        print(self.df_bt)
        self.strategy = strategy
        self.cerebro = bt.Cerebro()
        

    def preprocess(self, df, pro=False):
        if pro:
            features=['open','high','low','close','vol','trade_date']
            # convert_datetime = lambda x:datetime.strptime(x,'%Y%m%d')
            convert_datetime = lambda x: pd.to_datetime(str(x))
            df['trade_date'] = df['trade_date'].apply(convert_datetime)
            print(df)
            bt_col_dict = {'vol':'volume','trade_date':'datetime'}
            df = df.rename(columns=bt_col_dict)
            df = df.set_index('datetime')
            # df.index = pd.DatetimeIndex(df.index)
        else:
            features=['open','high','low','close','volume']
            df = df[features]
            df['openinterest'] = 0
            df.index = pd.DatetimeIndex(df.index)

        df = df[::-1]
        return df

    def run(self):
        data = bt.feeds.PandasData(dataname=self.df_bt,                               
                                    fromdate=self.start_datetime,                               
                                    todate=self.end_datetime)
        self.cerebro.adddata(data)  # Add the data feed
        self.cerebro.addstrategy(self.strategy)  # Add the trading strategy
        self.cerebro.broker.setcash(100000.0)
        # self.cerebro.addsizer(bt.sizers.FixedSize, stake=10)
        # self.cerebro.broker.setcommission(commission=0.0)
        self.cerebro.addanalyzer(bt.analyzers.SharpeRatio,_name = 'SharpeRatio')
        self.cerebro.addanalyzer(bt.analyzers.DrawDown, _name='DW')
        self.results = self.cerebro.run()
        strat = self.results[0]
        print('Final Portfolio Value: %.2f' % self.cerebro.broker.getvalue())
        print('SR:', strat.analyzers.SharpeRatio.get_analysis())
        print('DW:', strat.analyzers.DW.get_analysis())
        return self.results
    
    def plot(self, iplot=False):
        self.cerebro.plot(iplot=iplot)

Tushare有pro版本和普通行情，由于最开始我使用了pro版本但是后面出现了一些连接问题，所以我又写了普通的版本，目前都可以用。

pro：

ts.set_token(mytoken)
self.pro = ts.pro_api()
self.df = self.pro.daily(ts_code=self.ts_code, start_date=self.start_date, end_date=self.end_date)

pro版本需要获取token，这个需要你去官网注册之后可以看到的，之后使用daily即可获取日级历史行情。

ts_code='600515.SH'
start_date='20190101'
end_date='20191231'
strategy_runner = Strategy_runner(strategy=SmaCross, ts_code=ts_code, start_date=start_date, end_date=end_date, pro=True)
results = strategy_runner.run()
strategy_runner.plot()

以去年为例，传入pro=True，即可使用pro接口，然后获取行情，用均值策略分析

添加策略：

self.cerebro.addstrategy(self.strategy)

设置初始资金：

self.cerebro.broker.setcash(100000.0)

加入analyzer为获取夏普率和回撤率：

self.cerebro.addanalyzer(bt.analyzers.SharpeRatio,_name = 'SharpeRatio')
self.cerebro.addanalyzer(bt.analyzers.DrawDown, _name='DW')

运行模拟策略一遍：

self.cerebro.run()

得到分析结果：

Final Portfolio Value: 100000.29
SR: OrderedDict([('sharperatio', None)])
DW: AutoOrderedDict([('len', 178), ('drawdown', 0.0014899734784627702), ('moneydown', 1.4899999999906868), ('max', AutoOrderedDict([('len', 178), ('drawdown', 0.0019099660025940943), ('moneydown', 1.9099999999889405)]))])

因为夏普率默认需要年化的，这里不到一年的数据所以是None。

如果没有token可以直接使用普通接口：

ts_code='600515'
start_date='20190101'
end_date='20191231'
strategy_runner = Strategy_runner(strategy=SmaCross, ts_code=ts_code, start_date=start_date, end_date=end_date, pro=False)
results = strategy_runner.run()

如果使用日级行情，其实两者并没有太大的差别，只是表格的列名有点差别。我这里如果第一次获取之后我会在本地存下该表格，后面就不需要网络获取了。