VeighNa量化社区
你的开源社区量化交易平台
Member
avatar
加入于:
帖子: 8
声望: 1

关于tick回测的时候.我发现on_tick事件并不能触发数据tick数据.on_bar的事件可以触发tick事件.这样调用的BarGenerator(self.on_bar,1, self.on_min_bar) 怎么能合成分钟数据
我理解是正常情况下应该on_tick触发每条tick数据.然后on_bar合成分钟的数据.应该是这样的吧.是我哪里写错了吗.确定数据库是有数据的.
系统 widowns10 vnpy 版本 V2.0.7
附代码如下.

import sys
import datetime

from vnpy.app.cta_strategy import (
    CtaTemplate,
    StopOrder,
    TickData,
    BarData,
    TradeData,
    OrderData,
    BarGenerator,
    ArrayManager,
)

from vnpy.app.cta_strategy.base import (BacktestingMode)
from vnpy.trader.object import BarData, TickData


class Pulse_Trade(CtaTemplate):
    author = "用Python的交易员"

    fast_window = 6
    slow_window = 18

    fast_ma_1 = 0.0
    fast_ma_2 = 0.0

    slow_ma_1 = 0.0
    slow_ma_2 = 0.0

    parameters = ["fast_window", "slow_window"]
    variables = ["fast_ma_1", "fast_ma_2", "slow_ma_1", "slow_ma_2"]

    def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
        """"""
        super(Pulse_Trade, self).__init__(
            cta_engine, strategy_name, vt_symbol, setting
        )
        #合成分钟bar
        self.bg = BarGenerator(self.on_bar,1, self.on_min_bar)
        #准备时间序列
        self.am = ArrayManager(1)


    def on_init(self):
        """
        Callback when strategy is inited.
        """
        self.write_log("策略初始化")
        self.load_bar(1)

    def on_start(self):
        """
        Callback when strategy is started.
        """
        self.write_log("脉冲策略启动")
        self.put_event()

    def on_stop(self):
        """
        Callback when strategy is stopped.
        """
        self.write_log("策略停止")
        self.put_event()

    def on_tick(self, tick: TickData):
        """
        Callback of new tick data update.
        """
        print("调用ON_TICK",tick)
        self.bg.update_tick(tick)

    def on_bar(self, bar: BarData):
        """
        Callback of new bar data update.
        """
        # print("调用ON_BAR", bar)
        self.bg.update_bar(bar)


    def on_min_bar(self, bar: BarData):

        # print("on_min_bar:{} CLOSE: {}".format( bar.datetime.strftime("%Y-%m-%d %H:%M:%S"),bar.close_price))
        # print('on_min_bar',bar)
        am = self.am
        am.update_bar(bar)
        if not am.inited:
            return



        self.put_event()

    def on_order(self, order: OrderData):
        """
        Callback of new order data update.
        """
        pass

    def on_trade(self, trade: TradeData):
        """
        Callback of new trade data update.
        """
        self.put_event()

    def on_stop_order(self, stop_order: StopOrder):
        """
        Callback of stop order update.
        """
        pass


if __name__ == "__main__":

    from vnpy.app.cta_strategy.backtesting import BacktestingEngine
    from vnpy.app.cta_strategy.strategies.pulse_trade import Pulse_Trade



    engine = BacktestingEngine()

    beg_date = '2019-09-30 09:00:00'
    end_date = '2019-09-30 23:00:00'
    str_symbol = "RM001.CZCE"


    engine.set_parameters(
        vt_symbol = str_symbol,
        interval = "1m",
        start = datetime.datetime.strptime(beg_date, '%Y-%m-%d %H:%M:%S'),
        end = datetime.datetime.strptime(end_date, '%Y-%m-%d %H:%M:%S'),
        rate = 1 / 10000,
        slippage = 0,
        size = 10,
        pricetick = 1,
        capital = 100000,
        mode = BacktestingMode.TICK
    )

    engine.add_strategy(Pulse_Trade, {})
    engine.load_data()
    engine.run_backtesting()
    df = engine.calculate_result()
    engine.calculate_statistics()
    engine.show_chart()
    # print(engine.logs)
Administrator
avatar
加入于:
帖子: 4500
声望: 320
self.load_bar(1)

请改为:

self.load_tick(1)

同时确保你数据库中已经准备好了tick数据

Member
avatar
加入于:
帖子: 8
声望: 1

好的.已解决.谢谢群主

© 2015-2022 上海韦纳软件科技有限公司
备案服务号:沪ICP备18006526号

沪公网安备 31011502017034号

【用户协议】
【隐私政策】
【免责条款】