"""
reference:
https://github.com/jbn/ZigZag.git
"""
import numpy as np
PEAK = 1
VALLEY = -1
def identify_initial_pivot(X, up_thresh, down_thresh):
x_0 = X[0]
x_t = x_0
max_x = x_0
min_x = x_0
max_t = 0
min_t = 0
up_thresh += 1
down_thresh += 1
for t in range(1, len(X)):
x_t = X[t]
if x_t / min_x >= up_thresh:
return VALLEY if min_t == 0 else PEAK
if x_t / max_x <= down_thresh:
return PEAK if max_t == 0 else VALLEY
if x_t > max_x:
max_x = x_t
max_t = t
if x_t < min_x:
min_x = x_t
min_t = t
t_n = len(X)-1
return VALLEY if x_0 < X[t_n] else PEAK
def peak_valley_pivots(X, up_thresh, down_thresh):
"""
Find the peaks and valleys of a series.
:param X: the series to analyze
:param up_thresh: minimum relative change necessary to define a peak
:param down_thesh: minimum relative change necessary to define a valley
:return: an array with 0 indicating no pivot and -1 and 1 indicating
valley and peak
The First and Last Elements
---------------------------
The first and last elements are guaranteed to be annotated as peak or
valley even if the segments formed do not have the necessary relative
changes. This is a tradeoff between technical correctness and the
propensity to make mistakes in data analysis. The possible mistake is
ignoring data outside the fully realized segments, which may bias
analysis.
"""
if down_thresh > 0:
raise ValueError('The down_thresh must be negative.')
initial_pivot = identify_initial_pivot(X, up_thresh, down_thresh)
t_n = len(X)
pivots = np.zeros(t_n, dtype=np.int_)
trend = -initial_pivot
last_pivot_t = 0
last_pivot_x = X[0]
pivots[0] = initial_pivot
# Adding one to the relative change thresholds saves operations. Instead
# of computing relative change at each point as x_j / x_i - 1, it is
# computed as x_j / x_1. Then, this value is compared to the threshold + 1.
# This saves (t_n - 1) subtractions.
up_thresh += 1
down_thresh += 1
for t in range(1, t_n):
x = X[t]
r = x / last_pivot_x
if trend == -1:
if r >= up_thresh:
pivots[last_pivot_t] = trend
trend = PEAK
last_pivot_x = x
last_pivot_t = t
elif x < last_pivot_x:
last_pivot_x = x
last_pivot_t = t
else:
if r <= down_thresh:
pivots[last_pivot_t] = trend
trend = VALLEY
last_pivot_x = x
last_pivot_t = t
elif x > last_pivot_x:
last_pivot_x = x
last_pivot_t = t
if last_pivot_t == t_n-1:
pivots[last_pivot_t] = trend
elif pivots[t_n-1] == 0:
pivots[t_n-1] = -trend
return pivots
def max_drawdown(X):
"""
Compute the maximum drawdown of some sequence.
:return: 0 if the sequence is strictly increasing.
otherwise the abs value of the maximum drawdown
of sequence X
"""
mdd = 0
peak = X[0]
for x in X:
if x > peak:
peak = x
dd = (peak - x) / peak
if dd > mdd:
mdd = dd
return mdd if mdd != 0.0 else 0.0
def pivots_to_modes(pivots):
"""
Translate pivots into trend modes.
:param pivots: the result of calling ``peak_valley_pivots``
:return: numpy array of trend modes. That is, between (VALLEY, PEAK] it
is 1 and between (PEAK, VALLEY] it is -1.
"""
modes = np.zeros(len(pivots), dtype=np.int_)
mode = -pivots[0]
modes[0] = pivots[0]
for t in range(1, len(pivots)):
x = pivots[t]
if x != 0:
modes[t] = mode
mode = -x
else:
modes[t] = mode
return modes
def compute_segment_returns(X, pivots):
"""
:return: numpy array of the pivot-to-pivot returns for each segment."""
pivot_points = X[pivots != 0]
return pivot_points[1:] / pivot_points[:-1] - 1.0使用示例：
import matplotlib
matplotlib.use("TkAgg")
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import sys
import pathlib
sys.path.append("%s/zigzag" % pathlib.Path().absolute())
from zigzag import zigzag
def plot_pivots(X, pivots):
plt.xlim(0, len(X))
plt.ylim(X.min()*0.99, X.max()*1.01)
plt.plot(np.arange(len(X)), X, 'k:', alpha=0.5)
plt.plot(np.arange(len(X))[pivots != 0], X[pivots != 0], 'k-')
plt.scatter(np.arange(len(X))[pivots == 1], X[pivots == 1], color='g')
plt.scatter(np.arange(len(X))[pivots == -1], X[pivots == -1], color='r')
np.random.seed(1997)
X = np.cumprod(1 + np.random.randn(100) * 0.01)
pivots = zigzag.peak_valley_pivots(X, 0.03, -0.03)
plot_pivots(X, pivots)
plt.show()
modes = zigzag.pivots_to_modes(pivots)
print(pd.Series(X).pct_change().groupby(modes).describe().unstack())
print(zigzag.compute_segment_returns(X, pivots))pandas 的数据输入示例：
from pandas_datareader import get_data_yahoo
X = get_data_yahoo('GOOG')['Adj Close']
pivots = peak_valley_pivots(X.values, 0.2, -0.2)
ts_pivots = pd.Series(X, index=X.index)
ts_pivots = ts_pivots[pivots != 0]
X.plot()
ts_pivots.plot(style='g-o');1.Zigzag的3个参数https://github.com/twopirllc/pandas-ta
Zigzag在识别高低点的过程中， 主要设置了以下三个参数： ExtDepth, DextDeviation
以及ExtBackstep。 程序中的表示：
extern int ExtDepth=12;
extern int ExtDeviation=5;
extern int ExtBackstep=3;
说明：
ExtDepth： 用于设置高低点是相对与过去多少个Bars(价格图形中的一个柱子)而言。 Mt4中默认是12。 ExtDeviation： 用于设 置重新计算高低点时， 与前一高低点的相对点差。 默认值是5， 也就是说如果
A)当前高点>上个高点5 ,或者
B)当前低点<上个低点–
5的情况下， 则会对之前计算过的ExtBacksteps个Bars值的高低点进行重新计算。
ExtBackstep： 用于设置回退计算的Bars的个数。
2.Zigzag算法
1对计算位置进行初期化
1.1判断是否是第一次进行高低点计算， 如果是， 则设定计算位置为除去ExtDepth个图形最初的部分。 1.2如果之前已经计算过， 找到最近已知的三个拐点 （高点或低点） ， 将计算位置设置为倒数第三个拐点之后， 重新计 2.从步骤1已经设置好的计算位置开始， 将对用于存储高低点的变量进行初始化， 准备计算高低点 2.1计算ExtDepth区间内的低点， 如果该低点是当前低点， 则进行2.1.1的计算， 并将其记录成一个低点。
2.1.1如果当前低点比上一个低点值小于相对点差(ExtDeviation)； 并且之前ExtBackstep个Bars的记录的中， 高于当前低点的 值清空。
2.2高点的计算如同2.1以及分支处理2.1.1。
3.从步骤1已经设置好的计算位置开始， 定义指标高点和低点
3.1如果开始位置为高点， 则接下来寻找低点， 在找到低点之后， 将下一个寻找目标定义为高点
3.2如果开始位置为低点， 则与3.1反之。
以上可能比较难以理解， 我们这边举个例子说明：
假设上次计算的结果如下： 倒数第14个Bar出现了一个高点(3.1)， 倒数第4个是低点(1.5)，
倒数第1个是新的高点(2.1)——因为距离倒数第14已经大于ExtDepth(14-1>12)。
Bar-14Bar-4Bar-1 Bar-Current
高(3.1)低(1.5)高(2.1) X
对于Bar-Current， 即当前的价格X，
CaseI.
如果X >=2.1
ExtDeviation， 则根据Zigzag的定义， 这将是一个新的高点。 假设这里X=2.3， 那么我们绘制指标的时候应该成为：
Bar-14 Bar-4Bar-Current
高(3.1)
低(1.5)高(2.3)
CaseII.
如果1.5 - ExtDeviation<
X<2.1 ExtDeviation， 则我们继续等待价格的变化， 所绘制的指标也不会变化。
CaseIII.
如果1.5 - ExtDeviation>=
X， 则这是一个新的低点。 假设这里X=1.3， 则我们绘制指标的时候应该成为：
Bar-14Bar-Current
高(3.1) 低(1.3)
这个时候， 之前的Bar-4因为在我们定义的ExtBackstep之内(1-4)， 所以他的最低值会被清空，
根据算法第三步的定义， 我们会一直寻找低点直到发现Bar-Current， 这时候已经遍历过Bar-1， 所以Bar-1定义的高 点也不再成为拐点。
这也就是所谓的重绘部分， 也因此诟病为―未来函数‖——因为所看见的当前最后的高低点可能在下个时间段里面被抹去。 3Zigzag源码及解释：
Mt4的Zigzag源码里面的注释特别稀罕， 估计是感觉实现比较简单， 所以一概略去——恩， 极坏的编程习惯。 下面简要说明一下， 中文部分都是追加的解释：
// ——————————————————————
//|
Zigzag.mq4 |
//|
Copyright ?2005-2007, MetaQuotes Software Corp. |
//|
http://www.doczj.com/doc/855cc57301f69e3143329458.html / |
// ——————————————————————
#property copyright ―Copyright ?2007, MetaQuotes Software
Corp. ‖
#property
link
―http://www.doczj.com/doc/855cc57301f69e3143329458.html /‖
indicator_chart_window
//主窗口进行指标显示
#property indicator_buffers
1 //指标运用到数值的个数
#property indicator_color1
Red
//指标显示颜色
//—- indicator parameters
//Zigzag的三个参数
extern int ExtDepth=12;
extern int ExtDeviation=5;
extern int ExtBackstep=3;
//—- indicator buffers
//指标的数值存储变量
double
ZigzagBuffer[];
//拐点
double
HighMapBuffer[];
//高点的临时变量数组
double
LowMapBuffer[];
//低点的临时变量数组
int level=3; // recounting’s depth
//最近已知的三个拐点
bool downloadhistory=false; //是否第一次计算
// ——————————————————————//| Custom indicator initialization
function
|
// ——————————————————————
IndicatorBuffers(3);
//对于缓冲储存器分配记忆应用自定义指标计算， 用F1可以看到该函数的帮助和解释//—- drawing settings SetIndexStyle(0,DRAW_SECTION);
//划线的风格
//—- indicator buffers mapping
SetIndexBuffer(0,ZigzagBuffer);
SetIndexBuffer(1,HighMapBuffer);
SetIndexBuffer(2,LowMapBuffer);
SetIndexEmptyValue(0,0.0);
//—- indicator short name
IndicatorShortName(‖ZigZag(‖
ExtDepth ‖ , ‖ ExtDeviation‖ , ‖ ExtBackstep
‖)‖);
//设置指标的简称。
//—- initialization done
return(0);
}
// ——————————————————————
//|
|
// ——————————————————————
//start函数是Mt4的主函数， 当每次价格变动之后都会触发该函数的执行
int start()
{
//变量定义
//i： 临时变量；
//limit： 算法中所谓的开始计算位置;
//counterZ： 临时变量
//whatlookfor： 用于标识当前计算的是高点或者低点
int
limit,counterZ,whatlookfor;
//以下都是临时变量， 具体设值时解释
int
shift,back,lasthighpos,lastlowpos;
double val ,res;
double
curlow ,curhigh,lasthigh,lastlow;
if (counted_bars==0
&& downloadhistory) // history was
downloaded
{
//指标载入时counted_bars为0， 而downloadhistory为false， 将在下一次价格变化时进行ArrayInitialize(ZigzagBuffer,0.0); ArrayInitialize(HighMapBuffer,0.0);
ArrayInitialize(LowMapBuffer,0.0);
}
if (counted_bars==0)
{ //初期化， 第一次运行时limit为除去ExtDepth个图形最初的部分。 （算法1.1）
limit=Bars-ExtDepth;
downloadhistory=true;
(counted_bars>0)
{//如果之前已经计算过， 找到最近已知的三个拐点 （高点或低点） ， 将计算位置设置为倒数第三个拐点。 （算法1.2）
while (counterZ
&& i<100)
{
res=ZigzagBuffer[i];
if (res!=0) counterZ ;
i ;
}
i– ; //在上面while中最后一次找到的时候进行
1， 所以要-1才能得到真正第三个拐点处。
limit=i; //计算位置赋值
if (LowMapBuffer[i]!=0)
{//如果倒数第三个拐点是低点
curlow=LowMapBuffer[i];
//目标在于寻找高点
whatlookfor=1;
}
else
{
curhigh=HighMapBuffer[i];
}
for (i=limit-1;i>=0;i–)
{//清空第三个拐点后的数值， 准备重新计算最后的拐点
ZigzagBuffer[i]=0.0;
LowMapBuffer[i]=0.0;
HighMapBuffer[i]=0.0;
}
}
//算法Step2部分： 计算高低点
for(shift=limit;
shift>=0; shift–)
{
//2.1计算ExtDepth区间内的低点
val=Low[iLowest(NULL,0,MODE_LOW,ExtDepth,shift)];
if(val==lastlow) val=0.0;
else
{//如果该低点是当前低点，
lastlow=val;
if((Low[shift]-val)>(ExtDeviation*Point))
val=0.0; //是否比上个低点还低ExtDeviation， 不是的话则不进行回归处理
for(back=1; back<=ExtBackstep; back )
{//回退ExtBackstep个Bar， 把比当前低点高的纪录值给清空res=LowMapBuffer[shift back];
if((res!=0)&&(res>val))
LowMapBuffer[shift back]=0.0;
}
}
}
//将新的低点进行记录
if (Low[shift]==val) LowMapBuffer[shift]=val; else LowMapBuffer[shift]=0.0;
//— high
val=High[iHighest(NULL,0,MODE_HIGH ,ExtDepth,shift)];
if(val==lasthigh) val=0.0;
else
{
lasthigh=val;
if((val-High[shift])>(ExtDeviation*Point))
val=0.0;
else
for(back=1; back<=ExtBackstep; back )
{
res=HighMapBuffer[shift back];
if((res!=0)&&(res
HighMapBuffer[shift back]=0.0;
}
}
}
if (High[shift]==val) HighMapBuffer[shift]=val; else HighMapBuffer[shift]=0.0;
}
// final cutting
if (whatlookfor==0)
{
lastlow=0;
lasthigh=0;
}
else
{
lastlow=curlow;
lasthigh=curhigh;
//算法step3.定义指标的高低点
for
(shift=limit;shift>=0;shift–)
{
res=0.0;
switch(whatlookfor)
{
//初期化的情况下， 尝试找第一个高点或者是地点
case 0: // look for peak or lawn
if (lastlow==0 &&
lasthigh==0)
{//lastlow， lasthigh之前已经初始化， 再次判断以保证正确性？ if (HighMapBuffer[shift]!=0)
{//发现高点
lasthigh=High[shift];
lasthighpos=shift;
whatlookfor=-1; //下个寻找目标是低点
ZigzagBuffer[shift]=lasthigh;
res=1;
}
if (LowMapBuffer[shift]!=0)
lastlowpos=shift;
whatlookfor=1;
//下个寻找目标是高点
ZigzagBuffer[shift]=lastlow;
res=1;
}
}
break;
case 1: // look for
peak
//寻找高点
if (LowMapBuffer[shift]!=0.0 &&
LowMapBuffer[shift]
&& HighMapBuffer[shift]==0.0)
{//如果在上个低点和下个高点间发现新的低点， 则把上个低点抹去， 将新发现的低点作为最后一个低点
ZigzagBuffer[lastlowpos]=0.0;
lastlowpos=shift;
lastlow=LowMapBuffer[shift];
ZigzagBuffer[shift]=lastlow;
res=1;
}
if (HighMapBuffer[shift]!=0.0 &&
lasthigh=HighMapBuffer[shift];
lasthighpos=shift; ZigzagBuffer[shift]=lasthigh;
whatlookfor=-1;
//下一个目标将是寻找低点
res=1;
}
break;
case -1: // look for
lawn
//寻找低点
if (HighMapBuffer[shift]!=0.0 && HighMapBuffer[shift]>lasthigh && LowMapBuffer[shift]==0.0) {
ZigzagBuffer[lasthighpos]=0.0; lasthighpos=shift;
lasthigh=HighMapBuffer[shift]; ZigzagBuffer[shift]=lasthigh;
}
if (LowMapBuffer[shift]!=0.0 && HighMapBuffer[shift]==0.0)
lastlow=LowMapBuffer[shift];
lastlowpos=shift;
ZigzagBuffer[shift]=lastlow;
whatlookfor=1;
}
break;
default: return;
}
}
return(0);
}
// ——————————————————————
4.总结
以上就是对Zigzag算法和实现的分析。 希望能够对大家编写指标和EA有所帮助。