"""
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 有所帮助。