zigzag源代码——包括python和mql语言

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