基于价格图表识别主趋势波动中的次级波动

需求说明 #

我正在开发一款Python股票分析程序，核心需求是从开盘价、最高价、最低价、收盘价等价格数据中识别分形特性的波动——大结构中包含更小的结构，要精准识别主趋势波动中的次级波动甚至微级波动。

波动定义明确：

• 每个波动由推动浪（与市场趋势方向一致）和回调浪（与推动浪方向相反）组成
• 价格无明确方向时判定为震荡市场
• 新高确认新低，新低确认新高

现有枢轴点识别代码 #

目前已实现通过scipy识别价格转折点的函数，仅返回价格值：

import numpy as np
from scipy.signal import argrelextrema

def get_pivots(price: np.ndarray):
    maxima = argrelextrema(price, np.greater)
    minima = argrelextrema(price, np.less)
    return np.concatenate((price[maxima], price[minima]))

目标结构 #

需要基于枢轴点的嵌套关系，构建主波动、次级波动、微级波动的层级结构，生成类似如下格式的嵌套列表：

[major swing
  [minor swing1],
  [minor swing2],
  [minor swing3
    [micro swing1], 
    [micro swing2]
  ] 
]

示例数据 #

data = [5,6,7,8,9,10,11,12,13,14,16,17,18,19,20, # AB(Major) impulse
        19,18,17,16,15,14,13,12,11,10,9,8, # BC(Major) reaction
            9,10,11,12,13, 14,15, # C0C1 (Minor) impulse
            14,13,12,11,10, # C1C2 (Minor) reaction
            11, 12, 13, 14, 15, 16, 17, 18, # C2C3 (Minor) impulse
            17, 16, 15, 15, 14, 13, 12, # C3C4 (Minor) reaction
            13, 14, 15, 16, 17, 18, 19, 20, 21, # C4C5 (Minor) impulse
            20, 19, 18, 17, 16, 15, 14, # C5C6 (Minor) reaction
        15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25 # CD (Major) impulse 
        ]

实现方案 #

1. 修正枢轴点提取：保留关键信息 #

原函数仅返回价格值，丢失了索引和极值类型（顶/底），这是构建层级的核心依据，先修改函数：

def get_pivots_with_info(price: np.ndarray):
    # 获取极大值和极小值的索引
    max_indices = argrelextrema(price, np.greater)[0]
    min_indices = argrelextrema(price, np.less)[0]
    
    # 合并并按时间顺序排序
    all_indices = np.concatenate((max_indices, min_indices))
    sorted_indices = np.sort(all_indices)
    
    # 生成包含索引、价格、类型的结构化列表
    pivots = []
    for idx in sorted_indices:
        pivots.append({
            "index": idx,
            "price": price[idx],
            "type": "peak" if idx in max_indices else "trough"
        })
    return pivots

2. 递归构建嵌套波动结构 #

递归是实现分形层级的最优方案，核心逻辑是：从最大的主波动开始，逐层向下划分次级、微级波动，每个层级的波动基于区间内的顶底交替关系构建。

递归函数实现

def build_swing_hierarchy(pivots, start_idx=0, end_idx=None, level="major"):
    if end_idx is None:
        end_idx = len(pivots) - 1
    
    # 当前波动的起始和结束枢轴
    start_pivot = pivots[start_idx]
    end_pivot = pivots[end_idx]
    
    # 判断波动方向：推动浪（价格上涨）/回调浪（价格下跌）
    direction = "impulse" if end_pivot["price"] > start_pivot["price"] else "reaction"
    
    # 收集当前区间内的所有中间枢轴点
    middle_pivots = pivots[start_idx+1:end_idx]
    child_swings = []
    
    # 划分次级波动：基于顶底交替的规则
    if middle_pivots:
        current_sub_start = 0
        # 遍历中间枢轴，按顶底交替分组
        for i in range(1, len(middle_pivots)):
            if middle_pivots[i]["type"] != middle_pivots[i-1]["type"]:
                # 形成一个子波动，递归构建下一层级
                next_level = "minor" if level == "major" else "micro"
                sub_swing = build_swing_hierarchy(middle_pivots, current_sub_start, i, next_level)
                child_swings.append(sub_swing)
                current_sub_start = i
        # 处理最后一段未闭合的子波动
        if current_sub_start < len(middle_pivots) - 1:
            next_level = "minor" if level == "major" else "micro"
            sub_swing = build_swing_hierarchy(middle_pivots, current_sub_start, len(middle_pivots)-1, next_level)
            child_swings.append(sub_swing)
    
    # 返回当前波动的结构化数据
    return {
        "level": level,
        "direction": direction,
        "start": start_pivot,
        "end": end_pivot,
        "children": child_swings
    }

转换为目标列表格式

如果需要输出用户指定的嵌套列表格式，可添加转换函数：

def format_to_target_list(swing):
    swing_label = f"{swing['level']} swing ({swing['direction']})"
    if not swing["children"]:
        return [swing_label]
    child_list = [format_to_target_list(child) for child in swing["children"]]
    return [swing_label] + child_list

3. 测试与验证 #

# 处理示例数据
price_array = np.array(data)
pivots = get_pivots_with_info(price_array)
# 构建层级结构
hierarchy = build_swing_hierarchy(pivots)
# 转换为目标列表格式
result_list = format_to_target_list(hierarchy)

# 打印格式化结果
import json
print(json.dumps(result_list, indent=2))

关键优化建议 #

• 层级阈值自定义：可根据实际需求，通过设定幅度占比（如主波动幅度的1/3为次级波动阈值）或时间跨度，过滤无效的微小波动
• 震荡市场处理：当某一层级内波动方向频繁切换时，标记为震荡层级，避免无效划分
• 噪声过滤：忽略幅度极小的微级波动，减少结构冗余

内容的提问来源于stack exchange，提问作者neo-technoker