"""
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 < xt_n else peak def peak_valley_pivotsx 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: pivotslast_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有所帮助。