Thank you for this code. I made an indicator based on it which measures the median/mean length of zigzag moves:
45997512-ba86-4304-8451-bade15090c74-image.png
from collections import deque import backtrader as bt import numpy as np __all__ = ['ZigZag', 'ZigZagLen', ] class ZigZagLen(bt.Indicator): params = ( ## ZigZag inputs ('retrace', 0.05), # in percent ('minbars', 14), # number of bars to skip after the trend change ## # ZigZagLen inputs ('full_init', True), # ('period', 6 * 90,), # 6*90 # positive number for rolling series, -1 for expanding series ('output', 'median'), # median or mean ('del_outliers', True), ('alpha', 0.05), # exponential smoothing input ) lines = ('combined', 'bull', 'bear') def __init__(self): self.zigzag = ZigZag(retrace=self.p.retrace, minbars=self.p.minbars, ) assert self.p.output in ['median', 'mean'] if self.p.output == 'median': self.func = np.median elif self.p.output == 'mean': self.func = np.mean self.stored_listlen = 0 if self.p.full_init: assert self.p.period > 0, 'when full_init is enabled, we need a minimum period' self.addminperiod(self.p.period) def next(self): zz = self.zigzag l = self.l llist = {l.combined: ['bull', 'bear'], l.bull : ['bull'], l.bear : ['bear'], } # # No new values. Copying previous values instead of repeating calculations if len(zz.lenlist) < self.stored_listlen: for line in llist.keys(): line[0] = line[-1] return self.stored_listlen = len(zz.lenlist) # Do we have at least something to work with? if len(zz.lenlist) >= 8: if self.p.del_outliers: q1 = np.percentile(zz.lenlist, 25) q3 = np.percentile(zz.lenlist, 75) iqr = q3 - q1 upperlim = q3 + 1.5 * iqr lowerlim = q1 - 1.5 * iqr else: upperlim = float('inf') lowerlim = 0 lenlist = list(reversed(zz.lenlist)) if self.p.period > 0: # Rolling series if sum(lenlist) > self.p.period: # finding index of last value in our moving series cumlst = np.cumsum(lenlist) idx = np.where(cumlst > self.p.period)[0][0] lenlist = lenlist[:idx] # Getting the length of current move last_len = InfoInt(zz.since_last_pivot) last_len.bias = 'bear' if lenlist[0].bias != 'bear' else 'bull' lenlist.append(last_len) for line, bias in llist.items(): filter_lst = [v for v in lenlist if lowerlim < v < upperlim and v.bias in bias] output = self.func(filter_lst) if np.isnan(line[-1]): line[0] = output else: alpha = self.p.alpha line[0] = line[-1] * (1 - alpha) + output * alpha # inheriting allows us to add variables to an int class InfoInt(int): pass class ZigZag(bt.Indicator): ''' ZigZag indicator. ''' lines = ( 'trend', 'last_high', 'last_low', 'zigzag', ) plotinfo = dict( subplot=False, plotlinelabels=True, plotlinevalues=True, plotvaluetags=True, ) plotlines = dict( trend=dict(_plotskip=True), last_high=dict(color='green', ls='-', _plotskip=True), last_low=dict(color='black', ls='-', _plotskip=True), zigzag=dict(_name='zigzag', color='blue', ls='-', _skipnan=True), ) params = ( ('retrace', 0.05), # in percent ('minbars', 14), # number of bars to skip after the trend change ) def __init__(self): super(ZigZag, self).__init__() self.addminperiod(self.p.minbars) assert self.p.retrace > 0, 'Retracement should be above zero.' assert self.p.minbars >= 0, 'Minimal bars should be >= zero.' self.retrace_thresh = self.data.close * self.p.retrace / 100 self.minbars = self.p.minbars self.count_bars = 0 self.last_pivots = deque([0, 0], maxlen=2) self.last_pivot_t = 0 self.since_last_pivot = 0 self.lenlist = [] self.stored_datalen = 0 def prenext(self): self.l.trend[0] = 0 self.l.last_high[0] = self.data.high[0] self.l.last_low[0] = self.data.low[0] self.l.zigzag[0] = (self.data.high[0] + self.data.low[0]) / 2 def next(self): # No new candle. Got called due to resampling if len(self.data) == self.stored_datalen: return self.stored_datalen = len(self.data) curr_idx = len(self.data) self.retrace_thresh = self.data.close[0] * self.p.retrace / 100 self.since_last_pivot = curr_idx - self.last_pivot_t self.l.trend[0] = self.l.trend[-1] self.l.last_high[0] = self.l.last_high[-1] self.l.last_low[0] = self.l.last_low[-1] self.l.zigzag[0] = float('nan') # Initialize trend if self.l.trend[-1] == 0: # If current candle has higher low and higher high if self.l.last_low[0] < self.data.low[0] and self.l.last_high[0] < self.data.high[0]: self.l.trend[0] = 1 self.l.last_high[0] = self.data.high[0] self.last_pivot_t = curr_idx # If current candle has lower low and lower high elif self.l.last_low[0] > self.data.low[0] and self.l.last_high[0] > self.data.high[0]: self.l.trend[0] = -1 self.l.last_low[0] = self.data.low[0] self.last_pivot_t = curr_idx # Up trend elif self.l.trend[-1] == 1: # if higher high, move pivot location to current high if self.data.high[0] > self.l.last_high[-1]: self.l.last_high[0] = self.data.high[0] self.count_bars = self.minbars self.last_pivot_t = curr_idx # elif at least p.minbars since last bull swing and currently in a retrace -> Switch Bearish elif self.count_bars <= 0 and self.l.last_high[0] - self.data.low[0] > self.retrace_thresh \ and self.data.high[0] < self.l.last_high[0]: self.switch_to_bear(curr_idx) # elif bearish close elif self.count_bars < self.minbars and self.data.close[0] < self.l.last_low[0]: self.switch_to_bear(curr_idx) # Down trend elif self.l.trend[-1] == -1: # if lower low, move pivot location to current low if self.data.low[0] < self.l.last_low[-1]: self.l.last_low[0] = self.data.low[0] self.count_bars = self.minbars self.last_pivot_t = curr_idx # elif we had an established bear swing and currently in a retrace -> Switch Bullish elif self.count_bars <= 0 and self.data.high[0] - self.l.last_low[0] > self.retrace_thresh and \ self.data.low[0] > self.l.last_low[0]: self.switch_to_bull(curr_idx) # elif bullish close elif self.count_bars < self.minbars and self.data.close[0] > self.l.last_high[-1]: self.switch_to_bull(curr_idx) # Decrease minbars counter self.count_bars -= 1 def switch_to_bear(self, idx): self.l.trend[0] = -1 self.count_bars = self.minbars self.l.last_low[0] = self.data.low[0] self.l.zigzag[-self.since_last_pivot] = self.l.last_high[0] self.last_pivot_t = idx self.record_idx(idx, 'bull', ) def switch_to_bull(self, idx): self.l.trend[0] = 1 self.count_bars = self.minbars self.l.last_high[0] = self.data.high[0] self.l.zigzag[-self.since_last_pivot] = self.l.last_low[0] self.last_pivot_t = idx self.record_idx(idx, 'bear', ) def record_idx(self, idx, bias, ): self.last_pivots.append(idx) last_move_len = self.last_pivots[-1] - self.last_pivots[0] last_move_len = InfoInt(last_move_len) last_move_len.bias = bias self.lenlist.append(last_move_len)
