python - Python 中的之字折线指标

Question

我想为股票创建一个锯齿形指标。我正在使用 python，我的英语不好，所以我对此表示歉意。我的部分代码来自:Pandas: Zigzag segmentation of data based on local minima-maxima

问题是我想要的曲折是这样的(Metastock 曲折指标): 我的曲折代码看起来像(注意你可以用过滤器改变百分比):

from pandas_datareader import data
import pandas as pd
from datetime import date
from pandas_datareader.nasdaq_trader import get_nasdaq_symbols
from scipy import signal
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
np.random.seed(0)

def filter(values, percentage):
    previous = values[0] 
    mask = [True]
    for value in values[1:]: 
        relative_difference = np.abs(value - previous)/previous
        if relative_difference > percentage:
            previous = value
            mask.append(True)
        else:
            mask.append(False)
    return mask


def main(stock=None, start_date=None, end_date=None):
    df = data.DataReader(
        stock, 
        start=start_date, end=end_date,
        data_source='yahoo'
    )
    return df


if __name__ == '__main__':
    today = '{}'.format(date.today())
    stocks = ['BLL']
    cont = 0
    for stock in stocks:
        cont += 1
        try:
            serie = main(stock=stock, start_date='2018-1-1', end_date=today)
            serie.insert(loc=0, column='Date', value=serie.index)
            serie = serie.reset_index(drop=True)
            # Create zigzag trendline.
            ########################################
            # Find peaks(max).
            data_x = serie.index.values
            data_y = serie['Close'].values
            peak_indexes = signal.argrelextrema(data_y, np.greater)
            peak_indexes = peak_indexes[0]
            # Find valleys(min).
            valley_indexes = signal.argrelextrema(data_y, np.less)
            valley_indexes = valley_indexes[0]          
            # Merge peaks and valleys data points using pandas.
            df_peaks = pd.DataFrame({'date': data_x[peak_indexes], 'zigzag_y': data_y[peak_indexes]})
            df_valleys = pd.DataFrame({'date': data_x[valley_indexes], 'zigzag_y': data_y[valley_indexes]})
            df_peaks_valleys = pd.concat([df_peaks, df_valleys], axis=0, ignore_index=True, sort=True)
            # Sort peak and valley datapoints by date.
            df_peaks_valleys = df_peaks_valleys.sort_values(by=['date'])
            p = 0.1 # 20% 
            filter_mask = filter(df_peaks_valleys.zigzag_y, p)
            filtered = df_peaks_valleys[filter_mask]

             # Instantiate axes.
            (fig, ax) = plt.subplots(figsize=(10,10))
            # Plot zigzag trendline.
            ax.plot(df_peaks_valleys['date'].values, df_peaks_valleys['zigzag_y'].values, 
                                                                    color='red', label="Extrema")
            # Plot zigzag trendline.
            ax.plot(filtered['date'].values, filtered['zigzag_y'].values, 
                                                                    color='blue', label="ZigZag")
            # Plot original line.
            ax.plot(data_x, data_y, linestyle='dashed', color='black', label="Org. line", linewidth=1)
            plt.show()
            print('{} - {}| success'.format(cont, stock))
        except Exception:
            print('{} - {}| ERROR'.format(cont, stock))

Answer 1

这是github上的一个例子:zigzag

cimport cython
import numpy as np
from numpy cimport ndarray, int_t

DEF PEAK = 1
DEF VALLEY = -1


@cython.boundscheck(False)
@cython.wraparound(False)
cpdef int_t identify_initial_pivot(double [:] X,
                                   double up_thresh,
                                   double down_thresh):
    cdef:
        double x_0 = X[0]
        double x_t = x_0

        double max_x = x_0
        double min_x = x_0

        int_t max_t = 0
        int_t 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


@cython.boundscheck(False)
@cython.wraparound(False)
cpdef peak_valley_pivots(double [:] X,
                         double up_thresh,
                         double 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.')

    cdef:
        int_t initial_pivot = identify_initial_pivot(X,
                                                     up_thresh,
                                                     down_thresh)
        int_t t_n = len(X)
        ndarray[int_t, ndim=1] pivots = np.zeros(t_n, dtype=np.int_)
        int_t trend = -initial_pivot
        int_t last_pivot_t = 0
        double last_pivot_x = X[0]
        double x, r

    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


@cython.boundscheck(False)
@cython.wraparound(False)
cpdef double max_drawdown(ndarray[double, ndim=1] 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
    """
    cdef:
        double mdd = 0
        double peak = X[0]
        double x, dd

    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


@cython.boundscheck(False)
@cython.wraparound(False)
def pivots_to_modes(int_t [:] 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.
    """

    cdef:
        int_t x, t
        ndarray[int_t, ndim=1] modes = np.zeros(len(pivots),
                                                dtype=np.int_)
        int_t 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