在华尔街的交易员中流传着这样一句话：”历史不会简单重复，但会押着韵脚。”这句话用来形容谐波交易模式再合适不过。这种将斐波那契数列与几何图形结合的分析方法，就像在价格走势图中寻找隐藏的韵律。

谐波交易，一听这名字就挺玄乎的，仿佛在市场的波动里找到了某种隐藏的旋律。其实，说白了，谐波交易就是利用数学规律，尤其是斐波那契数列，去捕捉市场价格的周期性波动，并寻找潜在的交易机会。在这篇文章里，我们会从量化角度介绍谐波形态在金融市场的应用，并且以ABCD形态为例，深入解析一段自动交易策略代码，让你从理论到实战都能有所收获。

---

一、谐波交易的魔力：为什么它能在市场中奏效？

谐波交易的核心思想是：市场价格不是随机走的，而是有一定的比例和对称关系。很多交易者相信，市场的波动符合一些固定的数学规律，而这些规律可以帮助我们预测未来的走势。其合理性源于市场参与者的群体心理——当价格触及斐波那契关键比率时，交易者倾向于在此做出决策，形成支撑或阻力。

谐波模式分为ABCD、蝙蝠（Bat）、蝴蝶（Butterfly）等多种形态。本文将以ABCD形态为例，结合代码实战，展示如何用Python捕捉这类信号。

ABCD形态是最常见的谐波形态之一，它的基本结构如下：

1. AB段：市场从A点上涨（或下跌）到B点，形成第一段趋势。
   
2. BC段：价格回调，从B点调整到C点，通常是AB段的38.2%到88.6%。
   
3. CD段：市场再次沿着AB的方向运行，最终到达D点，CD段通常是BC段的1.13到2.618倍。

当市场价格形成这个结构，并且符合这些比例关系时，我们就认为它形成了一个ABCD形态。当形态完成时，D点即为潜在的转折点。若D点低于C点（熊市ABCD），预示下跌；反之则为牛市ABCD，预示上涨。

---

二、代码解析：如何用Python自动识别ABCD形态？

使用肉眼判断谐波形态不仅需要丰富的市场经验，还要求交易者具备充沛的精力，长时间盯盘无疑是一项艰巨的任务。幸运的是，我们可以借助量化的力量，让计算机自动完成形态识别和交易决策，从而提高效率并减少人为误差。因此，我们选择在 优宽量化平台 上实现这一策略。

本段代码的核心目标是：自动识别ABCD形态，并基于形态信号执行交易决策。接下来，我们将逐步拆解代码的关键逻辑，解析其实现方式。

1. 斐波那契比率的设定

FIBB_RATIOS = [0.236, 0.382, 0.5, 0.618, 0.786, 1.0, 1.13, 1.272, 1.414, 1.618, 2.0, 2.24, 2.618]

这些比率来源于斐波那契数列，是谐波形态的关键。它们决定了市场价格回调和延伸的可能比例，在形态识别时会用到。

2. 价格匹配逻辑

def is_eq(n, m, err=0.05, l_closed=False, r_closed=True):
    """判断数值是否在允许误差范围内相等"""
    left = m if l_closed else m * (1 - err)
    right = m if r_closed else m * (1 + err)
    return (n >= left) and (n <= right)

这段代码的作用是 判断一个数值n是否接近某个目标值m（考虑误差范围）。比如，我们检测AB段是否接近0.618，就可以用这个函数来判断。

def is_in(n, l, r, err=0.05, l_closed=True, r_closed=True, strict=True):
    """判断数值是否在区间内"""
    left = l if l_closed else l * (1 - err)
    right = r if r_closed else r * (1 + err)
    
    if strict:
        return any(is_eq(n, f_rate, err) for f_rate in FIBB_RATIOS if left <= f_rate <= right)
    return (n >= left) and (n <= right)

这个函数用来判断一个数是否落在指定区间 [l, r] 之内，比如判断 CD段的延伸是否在1.13到2.618之间。

3. 获取ZigZag点

def get_zigzag(df, period=14):

ZigZag指标用于筛选市场的 重要转折点，减少市场噪音。它的工作原理是：如果价格在某个周期（period）内达到新高或新低，就认为形成了一个关键点。

highest_high = ta.MAX(df['High'].values[:i], timeperiod=period)[-1]
lowest_low = ta.MIN(df['Low'].values[:i], timeperiod=period)[-1]

这里用 ta.MAX 和 ta.MIN 找到 最近period根K线的最高点和最低点，用来判断市场是否发生了新高或新低。

if new_high and not new_low:
    if direction != 1:
        direction = 1
        changed = True
elif new_low and not new_high:
    if direction != -1:
        direction = -1
        changed = True

如果价格创出新高，方向变成向上；如果创出新低，方向变成向下。

最终，函数返回一个 ZigZag 点列表，每个点包含时间、价格和方向。

---

4. 识别ABCD形态

def detect_abcd(points, err=0.05):

ABCD形态检测的逻辑：

1. 计算AB、BC、CD段的比例：

   AB = abs(B[1]-A[1]) / abs(X[1]-A[1])
   BC = abs(C[1]-B[1]) / abs(A[1]-B[1])
   CD = abs(D[1]-C[1]) / abs(B[1]-C[1])
   

   这里分别计算了AB段、BC段、CD段的比例关系。

2. 判断形态是否符合ABCD模式：

   is_AB_valid = is_in(AB, 0.382, 0.886, err=err)
   is_CD_valid = is_in(CD, 1.13, 2.618, err=err)
   is_ABCD_eq = is_eq(BC/CD, 1, err=err)
   

   • AB段必须在 0.382-0.886 之间；
     
   • CD段必须在 1.13-2.618 之间；
     
   • BC段/CD段的比例必须接近 1。

   如果这些条件都满足，就认为找到了一个有效的ABCD形态。

---

5. 交易逻辑

当发现ABCD形态后，代码会决定是否下单：

if latest_pattern['direction'] == 'Bullish':
    exchange.CreateOrder(symbol, 'buy', -1, 1)
if latest_pattern['direction'] == 'Bearish':
    exchange.CreateOrder(symbol, 'sell', -1, 1)

• 如果是看涨ABCD形态（D点低于C点），则开多单；
  
• 如果是看跌ABCD形态（D点高于C点），则开空单。

此外，还有一个移动止盈的逻辑：

if curPos.Type % 2 == 0 and (r[-1].Close < bestPrice * (1 - 0.01)):

如果当前价格比最佳价格低 1%，就止盈。

---

6. 实战案例：纯碱期货的ABCD策略表现

为了验证ABCD形态交易策略的有效性，我们将代码应用于 纯碱期货（SA888） 的 5分钟K线数据（2025年2月-3月），并观察其在实际市场中的表现。

策略回测结果与表现分析

• 优势：
  
  • 交易信号严格依赖形态规则，减少人为主观干扰；
    
  • 通过 移动止盈 机制，在趋势明确的行情中锁定利润，避免获利回吐。
    
• 结果：
  
  • 在趋势较为明确的市场环境下，策略能够多次精准捕捉到 潜在反转点，为交易者提供较好的入场机会；
    
  • 最大回撤控制在合理范围内，避免了极端亏损的情况，提升策略的稳健性。
    

示例回测图表（ABCD形态标记）：

该测试结果表明，ABCD形态交易策略在趋势市场中表现良好，但在震荡行情下可能会出现一定程度的误判。因此，结合 其他技术分析工具（如成交量、均线） 进行信号过滤，可能会进一步提升策略的胜率和稳定性。

---

7. 谐波交易的“矛与盾”

谐波交易策略的核心在于 数学比率的精确性，它的最大优势是避免情绪化交易，但同时也带来了一些局限性。

优点：为何选择谐波交易？

• 规则清晰：基于 斐波那契数列 计算价格回撤和扩展水平，避免人为主观判断；
  
• 高盈亏比：由于形态明确了 潜在反转区（PRZ），交易者可以在低风险的情况下博取更高的收益；
  
• 结构性交易思维：相比均线、震荡指标，谐波形态的交易逻辑更加偏向结构分析，适用于程序化交易和量化回测。
  

缺点：为何它并非“完美策略”？

• 形态条件苛刻，信号频率低：ABCD形态的数学比率要求严格，因此符合条件的交易机会较少，可能会错失其他市场机会；
  
• 滞后性问题：D点（交易信号）的确认通常依赖 前几个关键点已经形成，这意味着交易者需要等待较长时间才能入场，导致部分行情已经展开；
  
• 依赖参数调节：
  
  • 例如，ZigZag 指标周期 影响形态识别的精度，设置过大可能错过交易机会，设置过小可能带来过多噪音；
    
  • 斐波那契误差范围 需要结合市场波动率调整，否则会影响形态的准确性。
    

总结：
谐波交易本质上是一种 捕捉市场共振的技术分析方法，在某些市场环境下具有优势，但需要结合 动态参数优化 和 信号过滤 来提高胜率。

---

8. 优化方向：让策略更“聪明”

为了提高ABCD形态策略的适应性和盈利能力，我们可以从以下几个方向进行优化：

1. 动态误差调整：

   • 市场的波动性是变化的，固定的斐波那契比率可能导致形态识别失误。
     
   • 优化方法：根据 ATR（真实波幅）或标准差 动态调整斐波那契误差范围，提高形态匹配的灵活性。
     

2. 多形态叠加，扩大交易机会：

   • 除了ABCD形态，蝙蝠形态、螃蟹形态、加特利形态 也是经典的谐波模式。
     
   • 优化方法：编写代码 识别多个形态，以扩大策略的交易信号池，提高盈利能力。
     

3. 量价确认，提高信号可靠性：

   • 形态本身仅基于价格结构，可能会遇到假信号。
     
   • 优化方法：
     
     • 结合 成交量指标（如OBV、成交量均线），判断D点是否出现放量反转；
       
     • 利用 订单流分析（如买卖挂单、逐笔成交数据）提高入场的精准度。
       

4. 机器学习辅助形态识别：

   • 传统谐波形态依赖 固定规则匹配，可能会遗漏市场中 略有变形但依然有效的结构。
     
   • 优化方法：使用 卷积神经网络（CNN）或时间序列模型（LSTM） 训练数据，让模型自动识别潜在形态，从而减少人工设定参数的局限性。
     

---

9. 参考资料与扩展建议

为了更深入理解谐波交易策略并应用到实盘交易中，以下资料和建议值得参考：

1. 代码与工具

• HarmonicPatterns开源库：一个用于自动检测谐波形态的Python库，可用于量化策略开发。
  
• TradingView脚本：可以编写 Pine Script 代码，在TradingView上进行形态检测与测试。
  

2. 推荐书籍

• 《谐波交易者》Scott M. Carney（Harmonic Trading）
  
  • 该书详细讲解了ABCD、蝙蝠、螃蟹等形态的数学计算方法，适合深入学习。
    

3. 实战建议

• 适用市场：
  
  • 由于谐波形态在 波动率较高的市场 中表现更好，因此推荐在 主力合约（如纯碱、螺纹钢、沪铜）进行测试。
    
• 参数调整：
  
  • 初学者可以从 较长周期K线（如30分钟、1小时） 开始，减少市场噪音，提升形态识别的稳定性。
    
  • 之后可尝试 短周期K线（5分钟、15分钟），配合成交量过滤，提高短线交易的胜率。
    
• 结合其他指标：
  
  • 可搭配 RSI、MACD、布林带 作为辅助信号，避免单一形态带来的误判。
    

---

10. 结语：当数学遇见交易

谐波交易并非“圣杯”，但它能帮助交易者更理性地分析市场结构，并寻找高概率的交易机会。结合量化手段后，我们可以让这一策略变得更加智能化、更适应市场变化。

'''backtest
start: 2025-02-02 00:00:00
end: 2025-03-04 00:00:00
period: 5m
basePeriod: 1m
exchanges: [{"eid":"Futures_CTP","currency":"FUTURES"}]
'''

MAIN_FIBB_RATIOS = [0.618, 1.618]
SECOND_FIBB_RATIOS = [0.786, 0.886, 1.13, 1.27]
ALT_FIBB_RATIOS = [0.382, 0.5, 0.707, 1.41, 2.0, 2.24, 2.618, 3.14, 3.618]
AB_CD = [1.27, 1.618, 2.236]

FIBB_RATIOS = [*MAIN_FIBB_RATIOS, *SECOND_FIBB_RATIOS, *ALT_FIBB_RATIOS]

import pandas as pd
import talib as ta

# 斐波那契常用比率
FIBB_RATIOS = [0.236, 0.382, 0.5, 0.618, 0.786, 1.0, 1.13, 1.272, 1.414, 1.618, 2.0, 2.24, 2.618]

def is_eq(n, m, err=0.05, l_closed=False, r_closed=True):
    """判断数值是否在允许误差范围内相等"""
    left = m if l_closed else m * (1 - err)
    right = m if r_closed else m * (1 + err)
    return (n >= left) and (n <= right)

def is_in(n, l, r, err=0.05, l_closed=True, r_closed=True, strict=True):
    """判断数值是否在区间内，并考虑严格模式"""
    left = l if l_closed else l * (1 - err)
    right = r if r_closed else r * (1 + err)
    
    if strict:
        return any(is_eq(n, f_rate, err) for f_rate in FIBB_RATIOS if left <= f_rate <= right)
    return (n >= left) and (n <= right)

def get_zigzag(df, period=14):

    if len(df) < period:
        return

    """获取ZigZag指标点"""
    zigzag = []
    direction = 0
    
    for i in range(1, len(df)):
        # 使用talib计算最高价和最低价
        highest_high = ta.MAX(df['High'].values[:i], timeperiod=period)[-1]
        lowest_low = ta.MIN(df['Low'].values[:i], timeperiod=period)[-1]

        new_high = df['High'].iloc[i] >= highest_high
        new_low = df['Low'].iloc[i] <= lowest_low

        changed = False
        if new_high and not new_low:
            if direction != 1:
                direction = 1
                changed = True
        elif new_low and not new_high:
            if direction != -1:
                direction = -1
                changed = True

        if new_high or new_low:
            if changed or not zigzag:
                price = df['High'].iloc[i] if direction == 1 else df['Low'].iloc[i]
                zigzag.append((df['Time'].iloc[i], price, direction))
            else:
                last_idx, last_price, last_dir = zigzag[-1]
                if direction == 1 and df['High'].iloc[i] > last_price:
                    zigzag[-1] = (df['Time'].iloc[i], df['High'].iloc[i], direction)
                elif direction == -1 and df['Low'].iloc[i] < last_price:
                    zigzag[-1] = (df['Time'].iloc[i], df['Low'].iloc[i], direction)
    return zigzag

def detect_abcd(points, err=0.05):
    """检测ABCD形态"""
    if len(points) < 5:
        return None
    
    # 提取最近5个点构成XABCD形态
    X, A, B, C, D = points[-5:]
    
    # 计算各段比率
    AB = abs(B[1]-A[1]) / abs(X[1]-A[1])
    BC = abs(C[1]-B[1]) / abs(A[1]-B[1])
    CD = abs(D[1]-C[1]) / abs(B[1]-C[1])
    
    # 形态规则判断
    is_AB_valid = is_in(AB, 0.382, 0.886, err=err)
    is_CD_valid = is_in(CD, 1.13, 2.618, err=err)
    is_ABCD_eq = is_eq(BC/CD, 1, err=err)
    
    if is_AB_valid and is_CD_valid and is_ABCD_eq:
        direction = 'Bullish' if D[1] < C[1] else 'Bearish'
        return {
            'type': 'ABCD',
            'points': (X, A, B, C, D),
            'ratios': {'AB': AB, 'BC': BC, 'CD': CD},
            'direction': direction
        }
    return None

# 使用示例
def main():
    df = pd.DataFrame(columns=['Time', 'Open', 'High', 'Low', 'Close', 'Volume', 'OpenInterest'])
    last_pattern_time = None  # 用于记录上次绘制形态的时间
    bestPrice = None
    symbol = 'SA888'

    while True:
        # 获取最新数据
        r = exchange.GetRecords(symbol, PERIOD_M5)
        pos = exchange.GetPositions(symbol)

        if r is None or len(r) < 2 or pos is None:
            continue

        new_r = r[-2]  # 取最新的 K 线数据

        # 检查 Time 是否已存在
        if new_r['Time'] not in df['Time'].values:
            new_row = pd.DataFrame([new_r])  # 转换为 DataFrame
            df = pd.concat([df, new_row], ignore_index=True)  # 追加新数据

        # 转换数据类型
        df[['High', 'Low', 'Close']] = df[['High', 'Low', 'Close']].astype(float)

        # 获取ZigZag点
        zigzag_points = get_zigzag(df, period=14)

        if zigzag_points is None or len(zigzag_points) < 5:
            continue
        
        # 检测最新形态
        latest_pattern = None
        for i in range(4, len(zigzag_points)):
            segment = zigzag_points[i-4:i+1]
            pattern = detect_abcd(segment, err=0.2)
            if pattern:
                pattern_time = pattern['points'][-1][0]  # 使用D点时间作为形态时间
                # 仅保留时间最新的形态
                if (latest_pattern is None) or (pattern_time > latest_pattern['points'][-1][0]):
                    latest_pattern = pattern

      # 仅当检测到新形态时绘图
        if latest_pattern and (last_pattern_time != latest_pattern['points'][-1][0]):
            Log('检测到ABCD形态', latest_pattern)
            last_pattern_time = latest_pattern['points'][-1][0]

            if len(pos) == 0:
                if latest_pattern['direction'] == 'Bullish':
                    exchange.CreateOrder(symbol, 'buy', -1, 1)
                    bestPrice = r[-1].Close
                if latest_pattern['direction'] == 'Bearish':
                    exchange.CreateOrder(symbol, 'sell', -1, 1)
                    bestPrice = r[-1].Close

        
        if len(pos) == 1:  # 检查是否有持仓
            curPos = pos[0]  # 获取当前持仓信息

            # 更新最佳价格
            bestPrice = (
                max(bestPrice, r[-1].Close) if curPos.Type % 2 == 0 else min(bestPrice, r[-1].Close)
            )

            # 多头移动止盈逻辑
            if curPos.Type % 2 == 0 and (r[-1].Close < bestPrice * (1 - 0.01)):
                Log(f'多头移动止盈：最佳点位：{bestPrice}，当前点位：{r[-1].Close}#00ff00')
                exchange.CreateOrder(symbol, 'closebuy', -1, 1)

            # 空头移动止盈逻辑
            if curPos.Type % 2 == 1 and (r[-1].Close > bestPrice * (1 + 0.01)):
                Log(f'空头移动止盈：最佳点位：{bestPrice}，当前点位：{r[-1].Close}#00ff00')
                exchange.CreateOrder(symbol, 'closesell', -1, 1)

        Sleep(1000 * 60)