from vnpy_ctastrategy import ( CtaTemplate, StopOrder, TickData, BarData, TradeData, OrderData, BarGenerator, ArrayManager, Direction, ) class bbi_ma(CtaTemplate): """""" author = "用Python的交易员" bbi_value = 0 var1 = 0 ema60 =0 last_benefit = 0 last_high = 0 last_low = 0 long_entry = 0 short_entry = 0 long_stop = 0 short_stop = 0 period = 21 ##周期 fixed_size = 1 ##开仓量 ema_windows = 21 ##均线 parameters = ["period","fixed_size","ema_windows"] variables = [] def __init__(self, cta_engine, strategy_name, vt_symbol, setting): """""" super().__init__(cta_engine, strategy_name, vt_symbol, setting) self.bg5 = BarGenerator(self.on_bar, 5, self.on_5min_bar) self.am5 = ArrayManager(61) self.bg30 = BarGenerator(self.on_bar, self.period, self.on_30min_bar) self.am30 = ArrayManager(61) def on_init(self): """ Callback when strategy is inited. """ self.write_log("策略初始化") self.load_bar(5) 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.bg5.update_tick(tick) def on_bar(self, bar: BarData): """ Callback of new bar data update. """ self.bg5.update_bar(bar) self.bg30.update_bar(bar) def on_5min_bar(self, bar: BarData): """""" self.put_event() def on_30min_bar(self, bar: BarData): """""" self.cancel_all() self.am30.update_bar(bar) self.write_log("period pr") if not self.am30.inited: return self.write_log("period func") self.bbi_value= (self.am30.sma(3) +self.am30.sma(6) + self.am30.sma(12) +self.am30.sma(24))/4 self.var1 = (2*bar.close_price + bar.high_price + bar.low_price)/4 self.ema60 = self.am30.ema(self.ema_windows) bbi_diff = self.var1 - self.bbi_value ema_diff = bar.close_price - self.ema60 if self.pos == 0 : if bbi_diff > 0 and ema_diff>0 : self.buy(bar.close_price,self.fixed_size) if bbi_diff<0 and ema_diff<0 : self.short(bar.close_price,self.fixed_size) elif self.pos >0 and bbi_diff <0 : self.sell(bar.close_price,abs(self.pos)) elif self.pos<0 and bbi_diff>0 : self.cover(bar.close_price,abs(self.pos)) 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. """ if trade.direction == Direction.LONG: self.long_entry = trade.price ##self.long_stop = self.long_entry - 2 * self.atr_value else: self.short_entry = trade.price ##self.short_stop = self.short_entry + 2 * self.atr_value self.put_event() def on_stop_order(self, stop_order: StopOrder): """ Callback of stop order update. """ pass