VNPY使用记录（3）--多周期bar合成策略

背景

vnpy官方不支持多周期复合策略，所以在不修改官方源码的情况下，无法实现通过分钟线、日线编写复杂的策略。下面修改vnpy，通过1m线合成 30m(可配置，window_mn), 1d线， 并完成三个策略

实现

30分钟线的实现：

BarGenerator有参数可以实现n分钟bar的更新和回调

:param on_bar: 1分钟回调
:param window: n分钟窗口大小， n必须可以被60整除2, 3, 5, 6, 10, 15, 20, 30
:param on_window_bar: n分钟回调
:param interval:默认是分钟

日线实现

• BarGenerator增加一个方法update_bar_day_window， 增加一个属性 day_bar存储更新中的bar

• 由于有夜盘的情况，夜盘属于下一个交易日的行情，所以，以每天15：00为界限，将完成的bar传递给回调函数 on_window_bar, 然后将 day_bar重新设置为None

• 已经完成的bar的日期统一改成 15：00分钟bar的日期，不带时间、分钟、秒

策略代码

from vnpy.app.cta_strategy import (CtaTemplate,StopOrder,TickData,BarData,TradeData,OrderData,BarGenerator,ArrayManager,
)
from vnpy.trader.constant import Exchange, Intervalclass DoubleDayM30M1Strategy(CtaTemplate):"""使用1m线合成 日线、30分钟、1分钟三种数据，并根据各自ma叠加进行卖卖的策略"""author = "fsksf"window_mn = 30              # n分钟线fast_window_day = 3fast_window_m30 = 3fast_window_m1 = 3slow_window_day = 5slow_window_m30 = 5slow_window_m1 = 5fast_ma_day = 0.0fast_ma_m30 = 0.0fast_ma_m1 = 0.0slow_ma_day = 0.0slow_ma_m30 = 0.0slow_ma_m1 = 0.0parameters = ["fast_window_day", "fast_window_m30", "fast_window_m1","slow_window_day", "slow_window_m30", "slow_window_m1","window_mn"]variables = ["fast_ma_day", "slow_ma_day","fast_ma_m30", "slow_ma_m30","fast_ma_m1", "slow_ma_m1"]def __init__(self, cta_engine, strategy_name, vt_symbol, setting):""""""super().__init__(cta_engine, strategy_name, vt_symbol, setting)# daybar 生成器self.day_bar_gen = BarGenerator(lambda x: None, interval=Interval.DAILY, window=1, on_window_bar=self.on_day_bar)self.m30_bar_gen = BarGenerator(self.on_bar, interval=Interval.MINUTE, window=self.window_mn, on_window_bar=self.on_m30_bar)self.day_am = ArrayManager(size=self.slow_window_day+1)self.m30_am = ArrayManager(size=self.slow_window_m30+1)self.m1_am = ArrayManager(size=self.slow_window_m1+1)def on_m30_bar(self, bar):self.m30_am.update_bar(bar)def on_day_bar(self, bar):print(bar.__dict__)self.day_am.update_bar(bar)def on_init(self):"""Callback when strategy is inited."""self.write_log("策略初始化")self.load_bar(100)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."""self.m30_bar_gen.update_tick(tick)def on_bar(self, bar: BarData):"""Callback of new bar data update."""# if self.cta_engine.gateway_name == "BACKTESTING":# update windowself.m30_bar_gen.update_bar(bar)self.day_bar_gen.update_bar(bar)self.m1_am.update_bar(bar)if not (self.day_am.inited and self.m30_am.inited and self.m1_am.inited):return# 最后一个bar实时更新的，计算ma后删除# self.day_am.update_bar(self.day_bar_gen.last_bar)fast_ma_day = self.day_am.sma(self.fast_window_day, array=True)slow_ma_day = self.day_am.sma(self.slow_window_day, array=True)# self.m30_am.update_bar(self.m30_bar_gen.last_bar)fast_ma_m30 = self.m30_am.sma(self.fast_window_m30, array=True)slow_ma_m30 = self.m30_am.sma(self.slow_window_m30, array=True)fast_ma_m1 = self.m1_am.sma(self.fast_window_m1, array=True)slow_ma_m1 = self.m1_am.sma(self.slow_window_m1, array=True)self.fast_ma_day = fast_ma_day[-1]self.fast_ma_m30 = fast_ma_m30[-1]self.fast_ma_m1 = fast_ma_m1[-1]self.slow_ma_day = slow_ma_day[-1]self.slow_ma_m30 = slow_ma_m30[-1]self.slow_ma_m1 = slow_ma_m1[-1]cross_over = fast_ma_day[-2] < slow_ma_day[-2] and fast_ma_day[-1] > slow_ma_day[-1] and \fast_ma_m30[-2] < slow_ma_m30[-2] and fast_ma_m30[-1] > slow_ma_m30[-1] and \fast_ma_m1[-2] < slow_ma_m1[-2] and fast_ma_m1[-1] > slow_ma_m1[-1]cross_below = fast_ma_day[-2] > slow_ma_day[-2] and fast_ma_day[-1] < slow_ma_day[-1] and \fast_ma_m30[-2] > slow_ma_m30[-2] and fast_ma_m30[-1] < slow_ma_m30[-1] and \fast_ma_m1[-2] > slow_ma_m1[-2] and fast_ma_m1[-1] < slow_ma_m1[-1]if cross_over:self.write_log(f'触发买入：{str(bar.datetime)}')# print(f'触发买入：{str(bar.datetime)}')self.write_log(f'day: ma_fast: {fast_ma_day[-2:]}, slow: {slow_ma_day[-2:]} close_price: {self.day_am.close_array}')self.write_log(f'30m: ma_fast: {fast_ma_m30[-2:]}, slow: {slow_ma_m30[-2:]} close_price: {self.m30_am.close_array}')self.write_log(f'1m: ma_fast: {fast_ma_m1[-2:]}, slow: {slow_ma_m1[-2:]} close_price: {self.m1_am.close_array}')if self.pos == 0:self.buy(bar.close_price, 1)elif self.pos < 0:self.cover(bar.close_price, 1)self.buy(bar.close_price, 1)elif cross_below:if self.pos == 0:self.short(bar.close_price, 1)elif self.pos > 0:self.sell(bar.close_price, 1)self.short(bar.close_price, 1)# self.day_am.remove_last_bar()# self.m30_am.remove_last_bar()self.put_event()def on_order(self, order: OrderData):"""Callback of new order data update."""passdef 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

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！