关于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)