"""写这个模型花了小半天, 还没来得及测试, 先开源出来;等期货回测器的cta分支没问题后，再测试该模型;PS: 为了方便模型的回测, 价格设置上确实有不合理的地方, 勿用于实盘, 请见谅;"""from tqz_strategy.template import CtaTemplatefrom public_module.object import StopOrder, TickData, BarData, TradeData, OrderDatafrom public_module.utility import BarGeneratorclass TQZTurtleTradingStrategy(CtaTemplate):author = "tqz"# --- param part ---donchian_channel_window = 20fast_window = 10slow_window = 60clear_position_days_window = 10n_window = 20parameters = ["fast_window","slow_window","donchian_channel_window","clear_position_days_window","n_window"]# --- var part ---fast_ma0 = 0.0fast_ma1 = 0.0slow_ma0 = 0.0slow_ma1 = 0.0donchian_channel_up = 0.0donchian_channel_down = 0.0clear_position_level_price = 0 # profitstop_loss_level_price = 0 # lossvariables = ["fast_ma0","fast_ma1","slow_ma0","slow_ma1","donchian_channel_up","donchian_channel_down","stop_loss_level_price","clear_position_level_price",]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)self.history_bars = []def on_init(self):"""Callback when strategy is inited."""# self.write_log(msg=f'strategy_name: {self.strategy_name} on_init.')passdef on_start(self):"""Callback when strategy is started."""# self.write_log(msg=f'strategy_name: {self.strategy_name} on_start.')passdef on_stop(self):"""Callback when strategy is stopped."""# self.write_log(msg=f'strategy_name: {self.strategy_name} on_stop.')passdef on_tick(self, tick: TickData):"""Callback of new tick data update."""self.bg.update_tick(tick)def on_bar(self, bar: BarData):"""Callback of new bar data update."""if self.__update_params_ok(new_bar=bar) is False:return# trend direction: turtle user single malong_market = bar.close_price > self.slow_ma0short_market = bar.close_price < self.slow_ma0lots = 1if long_market:if self.pos is 0:if bar.high_price > self.donchian_channel_up:self.set_position(position=lots, position_change_price=self.donchian_channel_up)elif self.pos > 0:if bar.low_price < self.clear_position_level_price:self.set_position(position=0, position_change_price=self.clear_position_level_price)elif bar.low_price < self.stop_loss_level_price and self.stop_loss_level_price is not 0:self.set_position(position=0, position_change_price=self.stop_loss_level_price)elif self.pos < 0:self.set_position(position=0, position_change_price=bar.close_price)elif short_market:if self.pos is 0:if bar.low_price < self.donchian_channel_down:self.set_position(position=-lots, position_change_price=self.donchian_channel_down)elif self.pos < 0:if bar.high_price > self.clear_position_level_price:self.set_position(position=0, position_change_price=self.clear_position_level_price)elif bar.high_price > self.stop_loss_level_price and self.stop_loss_level_price is not 0:self.set_position(position=0, position_change_price=self.stop_loss_level_price)elif self.pos > 0:self.set_position(position=0, position_change_price=bar.close_price)def __update_params_ok(self, new_bar: BarData) -> bool:if len(self.history_bars) < self.slow_window:self.history_bars.append(new_bar)return Falsefirst_bar = self.history_bars[0]self.history_bars.remove(first_bar)self.history_bars.append(new_bar)# fast ma valuetmp_fast_ma0 = 0for bar in self.history_bars[-self.fast_window:]:tmp_fast_ma0 += bar.close_pricetmp_fast_ma1 = 0for bar in self.history_bars[-self.fast_window-1:-1]:tmp_fast_ma1 += bar.close_priceself.fast_ma0 = tmp_fast_ma0 / self.fast_windowself.fast_ma1 = tmp_fast_ma1 / self.fast_window# slow ma valuetmp_slow_ma0 = 0for bar in self.history_bars:tmp_slow_ma0 += bar.close_pricetmp_slow_ma1 = 0for bar in self.history_bars[:-1]:tmp_slow_ma1 += bar.close_pricetmp_slow_ma1 += first_bar.close_priceself.slow_ma0 = tmp_slow_ma0 / self.slow_windowself.slow_ma1 = tmp_slow_ma1 / self.slow_window# donchian up & down valueself.donchian_channel_up, self.donchian_channel_down = self.__get_new_donchian_value()# clear_position_level_priceself.clear_position_level_price = self.__get_clear_position_level_price()# stop_loss_level_priceif self.pos > 0:self.stop_loss_level_price = self.position_change_price - self.__get_n_value() * 2elif self.pos < 0:self.stop_loss_level_price = self.position_change_price + self.__get_n_value() * 2return Truedef __get_n_value(self):n_value = 0if self.pos is not 0:n_values = []for index, bar in enumerate(self.history_bars):if index > len(self.history_bars) - self.n_window - 1:pre_bar = self.history_bars[index - 1]tr = max(bar.high_price - bar.close_price, bar.high_price - pre_bar.close_price, pre_bar.close_price - bar.low_price)if len(n_values) is 0:n_values.append(tr)else:n_values.append((n_values[-1] * (self.n_window - 1) + tr) / self.n_window)n_value = n_values[-1]return n_valuedef __get_clear_position_level_price(self):tmp_clear_position_level_price = 0if self.pos is not 0:for bar in self.history_bars[-self.clear_position_days_window:]:if self.pos > 0:if bar.low_price < tmp_clear_position_level_price or tmp_clear_position_level_price is 0:tmp_clear_position_level_price = bar.low_priceelif self.pos < 0:if bar.high_price > tmp_clear_position_level_price or tmp_clear_position_level_price is 0:tmp_clear_position_level_price = bar.high_pricereturn tmp_clear_position_level_pricedef __get_new_donchian_value(self):tmp_donchian_channel_up = 0tmp_donchian_channel_down = 0for bar in self.history_bars[-self.donchian_channel_window:]:if bar.high_price > self.donchian_channel_up or tmp_donchian_channel_up is 0:tmp_donchian_channel_up = bar.high_priceif bar.low_price < self.donchian_channel_down or tmp_donchian_channel_down is 0:tmp_donchian_channel_down = bar.low_pricereturn tmp_donchian_channel_up, tmp_donchian_channel_downdef set_position(self, position: int, position_change_price: float):super().set_position(position=position, position_change_price=position_change_price)if self.pos > 0:self.stop_loss_level_price = position_change_price - 2 * self.__get_n_value()elif self.pos < 0:self.stop_loss_level_price = position_change_price + 2 * self.__get_n_value()else:self.stop_loss_level_price = 0# useless definationdef on_order(self, order: OrderData):"""Callback of new order data update."""passdef on_trade(self, trade: TradeData):"""Callback of new trade data update."""passdef on_stop_order(self, stop_order: StopOrder):"""Callback of stop order update."""pass
如果觉得《量化交易之回测篇 - 海龟交易策略（初版）》对你有帮助，请点赞、收藏，并留下你的观点哦！
