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

如题,前阵子看到社区内的RBreaker策略,想到加入长周期的ATR指标来检验波动,只在波动率大于平均波动率时开仓,于是在原策略的模板上进行了一些修改。
但是修改后的策略只能加载历史数据,逐日盯市后成交数量为空。
想请各位大佬帮忙看下代码,是不是我写的有问题?
数据部分使用的是RQData的IC88,时间为2022/1/1 - 2022/7/30

def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
        """"""
    super(RBreakStrategyPlus, self).__init__(
        cta_engine, strategy_name, vt_symbol, setting
    )

    self.bg1 = BarGenerator(self.on_bar, window=1, on_window_bar=self.on_1min_bar, interval=Interval.MINUTE)
    self.bg10 = BarGenerator(self.on_bar, window=10, on_window_bar=self.on_10min_bar, interval=Interval.MINUTE)
    self.am1 = ArrayManager()
    self.am10 = ArrayManager()
    self.bars = []

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

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

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

def on_tick(self, tick: TickData):
    """
    Callback of new tick data update.
    """
    self.bg1.update_tick(tick)

def on_bar(self,bar:BarData):
    """k线更新,切记先更新长周期,再更新短周期"""
    self.bg10.update_bar(bar)   #将1分钟K线合成10分钟K线
    self.bg1.update_bar(bar)

def on_10min_bar(self, bar: BarData):
    """
    Callback of new bar data update.
    """
    am = self.am10
    am.update_bar(bar)
    if not am.inited:
        return

    # 计算ATR并判断波动大小
    atr_array = am.atr(self.atr_length, array=True)
    self.atr_value = atr_array[-1]
    self.atr_ma = atr_array[-self.atr_ma_length:].mean()

    if self.atr_value > self.atr_ma:
        self.atr_trend = 1
    else:
        self.atr_trend = -1

def on_1min_bar(self, bar: BarData):
    """
    Callback of new bar data update.
    """
    self.cancel_all()

    am = self.am1
    if not am.inited:
        return

    self.bars.append(bar)
    if len(self.bars) <= 2:
        return
    else:
        self.bars.pop(0)
    last_bar = self.bars[-2]

    # New Day
    if last_bar.datetime.date() != bar.datetime.date():
        if self.day_open:

            self.buy_setup = self.day_low - self.setup_coef * (self.day_high - self.day_close)  # 观察买入价
            self.sell_setup = self.day_high + self.setup_coef * (self.day_close - self.day_low)  # 观察卖出价

            self.buy_enter = (self.enter_coef_1 / 2) * (self.day_high + self.day_low) - self.enter_coef_2 * self.day_high  # 反转买入价
            self.sell_enter = (self.enter_coef_1 / 2) * (self.day_high + self.day_low) - self.enter_coef_2 * self.day_low  # 反转卖出价

            self.buy_break = self.buy_setup + self.break_coef * (self.sell_setup - self.buy_setup)  # 突破买入价
            self.sell_break = self.sell_setup - self.break_coef * (self.sell_setup - self.buy_setup)  # 突破卖出价

        self.day_open = bar.open_price
        self.day_high = bar.high_price
        self.day_close = bar.close_price
        self.day_low = bar.low_price

    # Today
    else:
        self.day_high = max(self.day_high, bar.high_price)
        self.day_low = min(self.day_low, bar.low_price)
        self.day_close = bar.close_price

    if not self.sell_setup:
        return

    self.tend_high, self.tend_low = am.donchian(self.donchian_window)

    if bar.datetime.time() < self.exit_time:

        if self.pos == 0:
            self.intra_trade_low = bar.low_price
            self.intra_trade_high = bar.high_price

            if self.tend_high > self.sell_setup and self.sell_setup > self.buy_setup:
                long_entry = max(self.buy_break, self.day_high)
                if self.atr_trend > 0:
                    self.buy(long_entry, self.fixed_size, stop=True)

                # self.short(self.sell_enter, self.multiplier * self.fixed_size, stop=True)

            elif self.tend_low < self.buy_setup:
                short_entry = min(self.sell_break, self.day_low)
                if self.atr_trend > 0:
                    self.short(short_entry, self.fixed_size, stop=True)

                # self.buy(self.buy_enter, self.multiplier * self.fixed_size, stop=True)

        elif self.pos > 0:
            self.intra_trade_high = max(self.intra_trade_high, bar.high_price)
            long_stop = self.intra_trade_high * (1 - self.trailing_long / 100)
            self.sell(long_stop, abs(self.pos), stop=True)

        elif self.pos < 0:
            self.intra_trade_low = min(self.intra_trade_low, bar.low_price)
            short_stop = self.intra_trade_low * (1 + self.trailing_short / 100)
            self.cover(short_stop, abs(self.pos), stop=True)
            self.write_log("最低价 = %f" % bar.low_price )

    # Close existing position
    else:
        if self.pos > 0:
            self.sell(bar.close_price * 0.99, abs(self.pos))
        elif self.pos < 0:
            self.cover(bar.close_price * 1.01, abs(self.pos))

            self.put_event()
Member
avatar
加入于:
帖子: 4622
声望: 284

可以自己在策略里结合策略逻辑和行情数据打印排查看看

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

xiaohe wrote:谢谢大佬回答,如果程序能运行,虽然没有交易结果,但是可以视为代码编写没有问题吧,有可能只是策略本身的逻辑问题?打印的话,请问用什么样的方法可以实现呢,我尝试过加入Print,但是输出中没有打印的结果

可以自己在策略里结合策略逻辑和行情数据打印排查看看

Member
avatar
加入于:
帖子: 716
声望: 62

如果有执行到print语句就一定会有输出的,如果没有就说明逻辑没有执行到print语句,那就再往上一级判断语句或者上一个调用的函数里打印输出。甚至可以现在on_init和on_bar函数下先行打印试试,看看有没调用这个策略和数据是否正常读取

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

郭易燔 wrote:

如果有执行到print语句就一定会有输出的,如果没有就说明逻辑没有执行到print语句,那就再往上一级判断语句或者上一个调用的函数里打印输出。甚至可以现在on_init和on_bar函数下先行打印试试,看看有没调用这个策略和数据是否正常读取

感谢指导,我再尝试一下看看

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

沪公网安备 31011502017034号

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