缠中说禅中枢指标:趋势峰谷线多空转折源码学习

# -*- coding: utf-8 -*-"""通达信 ZIG 转向指标 Python 实现功能：1. ZIG 折线计算（基于收盘价的百分比转向）2. 波峰/波谷识别与值提取3. 绘制峰谷之间的趋势线（红涨绿跌）4. 箱体区域判定与可视化（灰色区域 + 水平虚线）5. K线颜色标记（涨洋红、跌淡蓝）6. 峰谷数值标注"""import numpy as npimport pandas as pdimport matplotlib.pyplot as pltimport matplotlib.dates as mdatesfrom mplfinance.original_flavor import candlestick_ohlcimport akshare as ak  # 数据源，也可替换为 yfinance 等# ==================== 1. ZIG 转向算法 ====================def calc_zig(close, percent=5):    """    计算 ZIG 折线（未来函数版本，一次性处理全部历史数据）    参数:        close: np.array, 收盘价序列        percent: float, 转向百分比（默认5%）    返回:        zig: np.array, ZIG 折线值        pivots: list of (index, price, type) 转折点列表，type='peak' 或 'trough'    """    n = len(close)    if n == 0:        return np.array([]), []    # 初始化    direction = None          # 'up' 或 'down'    ext_price = close[0]      # 当前极值价格    ext_idx = 0               # 当前极值索引    pivots = []               # 存储转折点 (index, price, type)    for i in range(1, n):        pct_change = (close[i] - ext_price) / ext_price * 100        if direction is None:            if pct_change >= percent:                direction = 'up'                ext_price = close[i]                ext_idx = i            elif pct_change <= -percent:                direction = 'down'                ext_price = close[i]                ext_idx = i            # 否则保持，不记录转折点        elif direction == 'up':            if close[i] >= ext_price:                # 继续向上，更新极值                ext_price = close[i]                ext_idx = i            elif pct_change <= -percent:                # 转向向下，记录前一波峰                pivots.append((ext_idx, ext_price, 'peak'))                direction = 'down'                ext_price = close[i]                ext_idx = i        elif direction == 'down':            if close[i] <= ext_price:                # 继续向下                ext_price = close[i]                ext_idx = i            elif pct_change >= percent:                # 转向向上，记录前一波谷                pivots.append((ext_idx, ext_price, 'trough'))                direction = 'up'                ext_price = close[i]                ext_idx = i    # 添加最后一个极值点（若未成对）    if direction == 'up':        pivots.append((ext_idx, ext_price, 'peak'))    elif direction == 'down':        pivots.append((ext_idx, ext_price, 'trough'))    # 去重（连续相同类型只保留一个）    unique_pivots = []    for idx, price, ptype in pivots:        if not unique_pivots or unique_pivots[-1][2] != ptype:            unique_pivots.append((idx, price, ptype))        else:            # 相同类型保留更极端的（峰值取大，谷值取小）            last = unique_pivots[-1]            if ptype == 'peak' and price > last[1]:                unique_pivots[-1] = (idx, price, ptype)            elif ptype == 'trough' and price < last[1]:                unique_pivots[-1] = (idx, price, ptype)    pivots = unique_pivots    # 线性插值填充 zig 序列    zig = np.full(n, np.nan)    if len(pivots) == 0:        zig[:] = close[0]        return zig, pivots    # 第一个转折点之前的区域用第一个点值填充    first_idx = pivots[0][0]    zig[:first_idx+1] = pivots[0][1]    # 转折点之间线性插值    for j in range(len(pivots)-1):        idx1, price1, _ = pivots[j]        idx2, price2, _ = pivots[j+1]        if idx2 > idx1:            x = np.arange(idx1, idx2+1)            y = np.linspace(price1, price2, len(x))            zig[idx1:idx2+1] = y    # 最后一个转折点到末尾填充    last_idx = pivots[-1][0]    zig[last_idx:] = pivots[-1][1]    return zig, pivots# ==================== 2. 提取波峰/波谷值 ====================def get_peaks_troughs(pivots):    """从转折点列表中分离波峰和波谷"""    peaks = [(idx, price) for idx, price, ptype in pivots if ptype == 'peak']    troughs = [(idx, price) for idx, price, ptype in pivots if ptype == 'trough']    return peaks, troughsdef get_nth_peak(peaks, n):    """获取第 n 个波峰值（1为最近）"""    if n <= len(peaks):        return peaks[-n][1]    return np.nandef get_nth_trough(troughs, n):    """获取第 n 个波谷值（1为最近）"""    if n <= len(troughs):        return troughs[-n][1]    return np.nandef get_peak_bars(peaks, current_idx):    """当前K线到最近波峰的周期数（0表示当前就是波峰）"""    if not peaks:        return np.nan    last_peak_idx = peaks[-1][0]    return current_idx - last_peak_idxdef get_trough_bars(troughs, current_idx):    """当前K线到最近波谷的周期数"""    if not troughs:        return np.nan    last_trough_idx = troughs[-1][0]    return current_idx - last_trough_idx# ==================== 3. 主计算函数 ====================def compute_indicators(df, percent=5):    """    计算全部指标并返回带标记的 DataFrame    df: 包含 'date', 'open', 'high', 'low', 'close' 列    """    close = df['close'].values    n = len(df)    # ZIG 与转折点    zig, pivots = calc_zig(close, percent)    peaks, troughs = get_peaks_troughs(pivots)    # LIJIND7 / LIJIND8: ZIG 上升/下降    zig_series = pd.Series(zig)    df['zig'] = zig    df['zig_up'] = zig_series > zig_series.shift(1)    df['zig_down'] = zig_series < zig_series.shift(1)    # 波峰波谷值    df['peak1'] = np.nan    df['trough1'] = np.nan    df['peak2'] = np.nan    df['trough2'] = np.nan    for i in range(n):        # 最近波峰/谷的值        df.loc[i, 'peak1'] = get_nth_peak(peaks, 1)        df.loc[i, 'trough1'] = get_nth_trough(troughs, 1)        df.loc[i, 'peak2'] = get_nth_peak(peaks, 2)        df.loc[i, 'trough2'] = get_nth_trough(troughs, 2)    # LIJIND9 = MIN(峰值1, 峰值2)    df['min_peak'] = df[['peak1', 'peak2']].min(axis=1)    # LIJIND10 = MAX(谷值1, 谷值2)    df['max_trough'] = df[['trough1', 'trough2']].max(axis=1)    # LIJIND11: 较小的峰值 > 较大的谷值    df['box_condition'] = df['min_peak'] > df['max_trough']    # 波峰/波谷位置标记（用于绘图）    df['is_peak'] = False    df['is_trough'] = False    for idx, _ in peaks:        df.loc[idx, 'is_peak'] = True    for idx, _ in troughs:        df.loc[idx, 'is_trough'] = True    # 波峰/波谷数量标记（用于DRAWNUMBER）    return df, pivots, peaks, troughs# ==================== 4. 绘图（专业风格） ====================def plot_zig_indicator(df, pivots, peaks, troughs, title="ZIG 转向指标"):    """    绘制 K线图 + ZIG折线 + 趋势线 + 箱体区域 + K线着色    """    # 准备OHLC数据    ohlc = []    dates = df['date'].values    for i, row in df.iterrows():        # mplfinance candlestick 需要数值化的日期        ohlc.append([i, row['open'], row['high'], row['low'], row['close']])    fig, ax = plt.subplots(figsize=(14, 8))    fig.subplots_adjust(bottom=0.1)    # 绘制K线    candlestick_ohlc(ax, ohlc, width=0.6, colorup='red', colordown='green', alpha=0.8)    # 根据 ZIG 上升/下降 重新覆盖K线颜色（原公式效果）    for i, row in df.iterrows():        if row['zig_up']:            color = 'magenta'   # 洋红        elif row['zig_down']:            color = 'lightskyblue'  # 淡蓝        else:            continue        # 画一个半透明矩形覆盖K线区间（从开盘到收盘，影线上使用相同颜色边框即可）        ax.add_patch(plt.Rectangle((i-0.4, row['low']), 0.8, row['high']-row['low'],                                   facecolor=color, edgecolor=color, alpha=0.3))    # 绘制 ZIG 折线    ax.plot(df.index, df['zig'], color='black', linewidth=1.5, label='ZIG(3,5)')    # 绘制峰谷连线（红色：谷->峰；绿色：峰->谷）    for j in range(len(pivots)-1):        idx1, price1, ptype1 = pivots[j]        idx2, price2, ptype2 = pivots[j+1]        if ptype1 == 'trough' and ptype2 == 'peak':            ax.plot([idx1, idx2], [price1, price2], color='red', linewidth=2, label='_nolegend_')        elif ptype1 == 'peak' and ptype2 == 'trough':            ax.plot([idx1, idx2], [price1, price2], color='green', linewidth=2, label='_nolegend_')    # 标注波峰/波谷数字    for idx, price in peaks:        ax.text(idx, price*1.02, f'{price:.2f}', ha='center', va='bottom',                fontsize=8, color='red', fontweight='bold')    for idx, price in troughs:        ax.text(idx, price*0.98, f'{price:.2f}', ha='center', va='top',                fontsize=8, color='green', fontweight='bold')    # 箱体区域绘制（灰色矩形 + 水平虚线）    box_start = 0    box_end = len(df)-1    in_box = False    for i, row in df.iterrows():        if row['box_condition'] and not in_box:            in_box = True            box_start = i        elif not row['box_condition'] and in_box:            # 结束当前箱体            ax.axhspan(df.loc[box_start, 'max_trough'], df.loc[box_start, 'min_peak'],                       xmin=box_start/len(df), xmax=(i-1)/len(df),                       facecolor='gray', alpha=0.2)            # 画水平虚线            ax.hlines(df.loc[box_start, 'min_peak'], box_start, i-1,                      colors='gray', linestyles='--', linewidth=1)            ax.hlines(df.loc[box_start, 'max_trough'], box_start, i-1,                      colors='gray', linestyles='--', linewidth=1)            in_box = False    if in_box:        ax.axhspan(df.loc[box_start, 'max_trough'], df.loc[box_start, 'min_peak'],                   xmin=box_start/len(df), xmax=1,                   facecolor='gray', alpha=0.2)        ax.hlines(df.loc[box_start, 'min_peak'], box_start, len(df)-1,                  colors='gray', linestyles='--', linewidth=1)        ax.hlines(df.loc[box_start, 'max_trough'], box_start, len(df)-1,                  colors='gray', linestyles='--', linewidth=1)    # 图表装饰    ax.set_title(title, fontsize=16)    ax.set_xlabel('交易日期')    ax.set_ylabel('价格')    ax.legend(loc='best')    ax.grid(True, linestyle='--', alpha=0.5)    # 设置x轴标尺为日期    tick_step = max(1, len(df)//10)    ax.set_xticks(range(0, len(df), tick_step))    ax.set_xticklabels([dates[i].strftime('%Y-%m-%d') for i in range(0, len(df), tick_step)],                       rotation=45, ha='right')    plt.tight_layout()    plt.show()# ==================== 5. 主程序示例 ====================if __name__ == '__main__':    # 获取股票数据（示例：贵州茅台，2023年数据）    symbol = "600519"  # 贵州茅台    df = ak.stock_zh_a_hist(symbol, period="daily", start_date="20230101", end_date="20231231", adjust="qfq")    df.rename(columns={        '日期': 'date',        '开盘': 'open',        '收盘': 'close',        '最高': 'high',        '最低': 'low',        '成交量': 'volume'    }, inplace=True)    df['date'] = pd.to_datetime(df['date'])    df.sort_values('date', inplace=True)    df.reset_index(drop=True, inplace=True)    # 计算指标    df_result, pivots, peaks, troughs = compute_indicators(df, percent=5)    # 绘图    plot_zig_indicator(df_result, pivots, peaks, troughs, title=f"{symbol} ZIG(3,5) 指标")    # 可选：打印最近两峰两谷数据    print("最近两根波峰:", [p[1] for p in peaks[-2:]])    print("最近两根波谷:", [t[1] for t in troughs[-2:]])    print("箱体条件满足:", df_result['box_condition'].iloc[-1])