VeighNa量化社区
你的开源社区量化交易平台 | vn.py | vnpy
Member
avatar
加入于:
帖子: 9
声望: 2

`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 = "C"

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_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(100,interval=Interval.DAILY,callback=self.am.update_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()
    #测试bar的周期
    # self.write_log(f"{bar.datetime}:{bar.interval}")
    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()
    # #测试bar的周期
    # # self.write_log(f"{bar.datetime}:{bar.interval}")
    # 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_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

`

Member
avatar
加入于:
帖子: 2
声望: 0

试试修改on_bar函数呢,看代码是在 self.bg.update_bar(bar)方法里去做聚合和回调的

def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
self.bg.update_bar(bar)

Member
avatar
加入于:
帖子: 9
声望: 2

on_bar函数在生成器中会被每分钟的k线调用,同时数据管理器会追加一分钟的k线进去,我不希望如此,只想对日线级别的k线进行处理。

Member
avatar
加入于:
帖子: 2
声望: 0

你的on_bar函数内还是得调用self.bg的update_bar()函数,我估计框架是通过这个方法触发合并数据的动作,然后回调你注册的on_day_bar,on_bar可以没有策略的逻辑,但不能省略

以上只是看完代码后的猜测,我还没验证

Member
avatar
加入于:
帖子: 9
声望: 2

有其他周期的实盘代码可供参考吗?

Member
avatar
加入于:
帖子: 9
声望: 2

已解决。
要点:

am.update_bar() 与 bg.update_bar() 容易弄混,实际上二者是不一样的,代码整体逻辑问题不大,细节需要注意
bg初始化的过程中,需要注意,历史下载到的回测数据收盘时间为 00:00

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

沪公网安备 31011502017034号

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