学习python量化用backtrader回测的第三个策略大小盘轮动

在网上看到有人分享了一个简单的量化轮动策略，大意是比较上证50指数和创业板50指数，看他们收盘价对于其MA20的相对比例，谁高就卖谁相对应的ETF,如果都低于MA20就空仓。

现在我们用python来写一个量化程序，用backtrader来回测，看看这个策略是否能稳定盈利。

首先是获取数据，上证50指数和创业板50指数的历史数据很好找，但场内ETF数据不那么容易得到了，我用的是通达信数据，获取数据的方法前几天说过了，有兴趣的自行去看看：用pytdx来获取股票数据

import backtrader as bt
import pandas as pd

# 打开csv文件获取日线数据
df_sh510050 = pd.read_csv('sh510050.csv')  # 上证50ETF
df_sz159949 = pd.read_csv('sz159949.csv')  # 创业板50ETF
df_sh000016 = pd.read_csv('sh000016.csv')  # 上证50指数
df_sz399673 = pd.read_csv('sz399673.csv')  # 创业板50指数

# 将'date'列转换为日期类型
df_sh510050['date'] = pd.to_datetime(df_sh510050['date'])
df_sz159949['date'] = pd.to_datetime(df_sz159949['date'])
df_sh000016['date'] = pd.to_datetime(df_sh000016['date'])
df_sz399673['date'] = pd.to_datetime(df_sz399673['date'])

# 重新设置索引
df_sh510050.set_index('date', inplace=True)
df_sz159949.set_index('date', inplace=True)
df_sh000016.set_index('date', inplace=True)
df_sz399673.set_index('date', inplace=True)

# 选择时间段的数据
start_date = '2023-01-01'
end_date = '2023-12-31'
df_sh510050 = df_sh510050[(df_sh510050.index >= pd.Timestamp(start_date)) & (df_sh510050.index <= pd.Timestamp(end_date))]
df_sz159949 = df_sz159949[(df_sz159949.index >= pd.Timestamp(start_date)) & (df_sz159949.index <= pd.Timestamp(end_date))]
df_sh000016 = df_sh000016[(df_sh000016.index >= pd.Timestamp(start_date)) & (df_sh000016.index <= pd.Timestamp(end_date))]
df_sz399673 = df_sz399673[(df_sz399673.index >= pd.Timestamp(start_date)) & (df_sz399673.index <= pd.Timestamp(end_date))]

通过读取csv文件的方式导入数据，用pd.to_datetime（）将’date’列转换为日期类型，然后用set_index（）重新设置索引，再选取回测的时间段。

接下来，我们按他介绍的方法来编写策略。

# 定义策略类
class ETFRotationStrategy(bt.Strategy):
    params = (
        ('maperiod', 20),  # 移动平均线的周期
    )

    def __init__(self):
        # 初始化移动平均线指标，使用self.datas[i]来访问数据源
        self.sma_sh000016 = bt.indicators.SimpleMovingAverage(self.datas[2].close, period=self.params.maperiod)
        self.sma_sz399673 = bt.indicators.SimpleMovingAverage(self.datas[3].close, period=self.params.maperiod)
        self.current_etf = None

    def next(self):
        # 计算相对M20比例
        sh000016_ratio = self.datas[2].close[0] / self.sma_sh000016[0] - 1
        sz399673_ratio = self.datas[3].close[0] / self.sma_sz399673[0] - 1

        # 决策逻辑
        # 记录买入操作
        if self.current_etf is None :
            if sh000016_ratio > sz399673_ratio and sh000016_ratio > 0:
                self.buy(data=self.datas[0], size=self.size())
                self.current_etf = 'sh510050'
                # 打印买入时间、买入价格、买入数量、买入股票代码
                print(f"Buy {self.datas[0].close[0]} at {self.datas[0].datetime.date()},sh510050")
                print("上证指数偏离：", sh000016_ratio)
                print("创业板指数偏离：", sz399673_ratio)
                print("")


            elif sz399673_ratio > sh000016_ratio and sz399673_ratio > 0:
                self.buy(data=self.datas[1], size=self.size())
                self.current_etf = 'sz159949'
                # 打印买入时间、买入价格、买入数量、买入股票代码
                print(f"Buy {self.datas[1].close[0]} at {self.datas[1].datetime.date()},sz159949")
                print("上证指数偏离：", sh000016_ratio)
                print("创业板指数偏离：", sz399673_ratio)
                print("")


                # 记录卖出操作
        if self.current_etf is not None:
            if self.current_etf == 'sh510050' and sz399673_ratio > sh000016_ratio:
                self.sell(data=self.datas[0], size=self.size())
                # 打印卖出时间、卖出价格、卖出数量、卖出股票代码
                print(f"Sell {self.datas[0].close[0]} at {self.datas[0].datetime.date()},sh510050")
                print("上证指数偏离：", sh000016_ratio)
                print("创业板指数偏离：", sz399673_ratio)
                print("")

                self.buy(data=self.datas[1], size=self.size())
                self.current_etf = 'sz159949'
                # 打印买入时间、买入价格、买入数量、买入股票代码
                print(f"Buy {self.datas[1].close[0]} at {self.datas[1].datetime.date()},sz159949")
                print("上证指数偏离：", sh000016_ratio)
                print("创业板指数偏离：", sz399673_ratio)
                print("")

            if self.current_etf == 'sz159949' and sh000016_ratio > sz399673_ratio:
                self.sell(data=self.datas[1], size=self.size())
                # 打印卖出时间、卖出价格、卖出数量、卖出股票代码
                print(f"Sell {self.datas[1].close[0]} at {self.datas[1].datetime.date()},sz159949")
                print("上证指数偏离：", sh000016_ratio)
                print("创业板指数偏离：", sz399673_ratio)
                print("")

                self.buy(data=self.datas[0], size=self.size())
                # 打印买入时间、买入价格、买入数量、买入股票代码
                print(f"Buy {self.datas[0].close[0]} at {self.datas[0].datetime.date()},sh510050")
                print("上证指数偏离：", sh000016_ratio)
                print("创业板指数偏离：", sz399673_ratio)
                print("")
                self.current_etf = 'sh510050'

        # 记录清仓操作
            if self.current_etf is not None and sh000016_ratio <= 0 and sz399673_ratio <= 0:
                if self.current_etf == 'sh510050':
                    self.close(data=self.datas[self.current_etf == 'sh510050'])
                    self.current_etf = None
                    # 打印卖出时间、卖出价格、卖出数量、卖出股票代码
                    print(f"Sell {self.datas[self.current_etf == 'sh510050'].close[0]} at {self.datas[self.current_etf == 'sh510050'].datetime.date()},sh510050")
                    print("上证指数偏离：", sh000016_ratio)
                    print("创业板指数偏离：", sz399673_ratio)
                    print("")
                elif self.current_etf == 'sz159949':
                    self.close(data=self.datas[self.current_etf == 'sz159949'])
                    # 打印卖出时间、卖出价格、卖出数量、卖出股票代码
                    print(f"Sell {self.datas[self.current_etf == 'sz159949'].close[0]} at {self.datas[self.current_etf == 'sz159949'].datetime.date()},sz159949")
                    print("上证指数偏离：", sh000016_ratio)
                    print("创业板指数偏离：", sz399673_ratio)
                    print("")
                    self.current_etf = None

本想用self.log（）的方式来查看交易记录，但老是报错，想快点看到程序运行结果，就用了print()来看交易记录。（策略应该还可以再修改一下，让它简洁一些）

最后是回测。

# 创建回测系统
cerebro = bt.Cerebro()

# 添加数据到回测系统
data_sh510050 = bt.feeds.PandasData(dataname=df_sh510050, name='Shanghai_50ETF')
data_sz159949 = bt.feeds.PandasData(dataname=df_sz159949, name='Chuangye_50ETF')
data_sh000016 = bt.feeds.PandasData(dataname=df_sh000016, name='Shanghai_50_Index')
data_sz399673 = bt.feeds.PandasData(dataname=df_sz399673, name='Chuangye_50_Index')
cerebro.adddata(data_sh510050)
cerebro.adddata(data_sz159949)
cerebro.adddata(data_sh000016)
cerebro.adddata(data_sz399673)

# 设置初始资本和交易手续费
initial_capital = 30000
cerebro.broker.setcash(initial_capital)
cerebro.broker.setcommission(commission=0.001)  # 设置交易手续费

# 加载策略
cerebro.addstrategy(ETFRotationStrategy)

# 运行回测
results = cerebro.run()

# 打印回测结果
print(f"Final Portfolio Value: {cerebro.broker.getvalue()}")
print(f"Total Return: {(cerebro.broker.getvalue() - initial_capital) / initial_capital * 100:.2f}%")

# 绘制回测结果图表
cerebro.plot()

回测结果如何？看下图

可是，没有赚到钱啊？我又查看了一下交易记录，确实是按照程序设计的策略去交易了。

如果2023年行情不好，那再看看前几年吧，于是修改回测起止时间，分别对2016-2023年进行回测，结果如下图：

这个结果是我没有想到的，到底是策略的问题，还是我写的程序有问题呢？得好好想想。

关于大小盘轮动策略，除了本文用的M20偏离法以外，还可以用别的方法，例如比较RSI强弱指标，还可以比较N天内的涨幅，总之都是谁强买谁。等以后有时间，再试试这两种方法，看看能否提高些收益。