VNPY回测

# coding:utf-8

from vnpy.app.cta_strategy.backtesting import BacktestingEngine
# 包的引入，先导入回测引擎
from vnpy.app.cta_strategy.strategies.boll_channel_strategy import BollChannelStrategy
# 再导入策略引擎


# 第一部分：标准的回测初始化参数设置
engine = BacktestingEngine()
engine.set_parameters(
    vt_symbol="IF1909.CFFEX",   # RB1910.SHFE
    interval="1m",
    start=datetime(2019, 6, 12),
    end=datetime(2019, 7, 10),
    rate=3.0/10000,
    slippage=0.2,
    size=300,
    pricetick=0.2,
    capital=1_000_000,
)

# 回测函数，可以用来重复使用
def run_backtesting(strategy_class, vt_symbol, interval, start, end, rate, slippage, size, pricetick, capital):
    '''
    回测函数，每添加一个策略就创建其BacktestingEngine对象。
    :param strategy_class:
    :param setting:
    :param vt_symbol:
    :param interval:
    :param start:
    :param end:
    :param rate:
    :param slippage:
    :param size:
    :param pricetick:
    :param capital:
    :return:
    '''
    engine = BacktestingEngine()  # 回测引擎初始化
    engine.set_parameters(
        vt_symbol=vt_symbol,
        interval=interval,
        start=start,
        end=end,
        rate=rate,
        slippage=slippage,
        size=size,
        pricetick=pricetick,
        capital=capital
    )
    engine.load_data_by_mongo(DB_name='VnTrader_1Min_Db', symbol=vt_symbol.split('.')[0])     # 数据修改之后
    # 这里要重点说一下了，我在这里是做了修改，直接使用了本地的MongoDB数据库。命名规则可以按照VnTrader_1Min_Db，数据库中的集合名称为IF1909（vt_symbol="IF1909.CFFEX"），即 " ."前面的部分，针对不同的品种可以修改这里的内容
    # engine.load_data()                          # 这里是源代码里面的数据载入部分，换成了我自己写的部分
    engine.add_strategy(strategy_class, {})       # 在引擎中创建策略对象
    engine.run_backtesting()                      # 运行回测
    df = engine.calculate_result()                # 显示回测结果
    return df

df1 = run_backtesting(
    strategy_class=BollChannelStrategy,
    vt_symbol="IF1909.CFFEX",
    interval="1m",
    start=datetime(2019, 2, 12),
    end=datetime(2019, 7, 10),
    rate=0.3/10000,
    slippage=0.2,
    size=300,
    pricetick=0.2,
    capital=1_000_000,
    )
print(df1)

def show_portafolio(df):
    engine = BacktestingEngine()
    engine.calculate_statistics(df)
    engine.show_chart(df)

现在展示一下回测的统计结果和绘图：

图片.png 图片.png

如下部分添加修改的代码：

    def load_data_by_mongo(self,DB_name,symbol):
        self.output("开始加载历史数据")
        db_client = self.connect_mongodb()
        # 对如下数据做修改
        # db = db_client[self.db_name]
        # cl = db[self.cl_name]

        # db = db_client['VnTrader_1Min_Db']
        # cl = db['IF1909']

        db = db_client[DB_name]
        cl = db[symbol]

        if self.mode == BacktestingMode.BAR:
            data_mode = BarData
        else:
            data_mode = TickData

        flt = {'datetime': {'$gte': self.start, '$lt': self.end}}
        db_cursor = cl.find(flt).sort('datetime')
        print('如下这些内容为数据库检测操作！')
        print('collection：',cl)
        print('开始时间:',self.start)
        print('结束时间:',self.end)
        print('从MongoDB数据库查找后的结果：',db_cursor)

        for d in db_cursor:
            data = self.generate_bar(d)
            self.history_data.append(data)
        self.output(f"历史数据加载完成，数据量：{len(self.history_data)}")