以下为我修改demo中的双均线策略试图进行实盘交易的代码。
我注意到：
k线生成器会将tick推送传递到on_bar的一分钟k线上，然后在am中添加该一分钟的BarData，而我想在am中维护的是日线级别的k线
我在on_day_bar中对日线级别数据进行处理，并在k线生成器中绑定了该函数到 on_window_bar上
加载数据的load_bar设置为日线级别。我的数据源配置没有问题，可以正常下载数据。
在on_bar和on_day_bar中调用am的更新bar函数，维护的bar数组周期并不一致，on_bar会被bg推送一分钟k线进而填充满am，我观察源码后得知这样的原理。
总体上我想实现在非默认的1min周期进行实盘和回测交易，但是发现并没有相关样例说明怎么样做，而我按照自己理解写得代码也无法运行。在社区寻求帮助无果，请版主斧正。
from vnpy_ctastrategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
from vnpy.trader.constant import Interval
from datetime import time
class DoubleMaStrategy(CtaTemplate):
author = "czy"
fast_window = 10
slow_window = 20
fast_ma0 = 0.0
fast_ma1 = 0.0
slow_ma0 = 0.0
slow_ma1 = 0.0
tick_show = 0.0
parameters = ["fast_window", "slow_window"]
variables = ["fast_ma0", "fast_ma1", "slow_ma0", "slow_ma1","tick_show"]
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
""""""
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
self.bg = BarGenerator(self.on_day_bar,interval=Interval.DAILY,window=1,on_window_bar=self.on_day_bar,daily_end=time(23,0))
self.am = ArrayManager(size=40)
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.load_bar(days=40,interval=Interval.DAILY,callback=self.on_day_bar)
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.bg.update_tick(tick)
self.tick_show = tick.last_price
self.put_event()
def on_day_bar(self,bar:BarData):
self.cancel_all()
am = self.am
am.update_bar(bar)
if not am.inited:
return
fast_ma = am.sma(self.fast_window, array=True)
self.fast_ma0 = fast_ma[-1]
self.fast_ma1 = fast_ma[-2]
slow_ma = am.sma(self.slow_window, array=True)
self.slow_ma0 = slow_ma[-1]
self.slow_ma1 = slow_ma[-2]
cross_over = self.fast_ma0 > self.slow_ma0 and self.fast_ma1 < self.slow_ma1
cross_below = self.fast_ma0 < self.slow_ma0 and self.fast_ma1 > self.slow_ma1
if cross_over:
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.put_event()
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
# """
# self.cancel_all()
# am = self.am
# am.update_bar(bar)
# if not am.inited:
# return
# fast_ma = am.sma(self.fast_window, array=True)
# self.fast_ma0 = fast_ma[-1]
# self.fast_ma1 = fast_ma[-2]
# slow_ma = am.sma(self.slow_window, array=True)
# self.slow_ma0 = slow_ma[-1]
# self.slow_ma1 = slow_ma[-2]
# cross_over = self.fast_ma0 > self.slow_ma0 and self.fast_ma1 < self.slow_ma1
# cross_below = self.fast_ma0 < self.slow_ma0 and self.fast_ma1 > self.slow_ma1
# if cross_over:
# 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.put_event()
