简介：技术分析中，K线形态与结构常被主观解读，而缠论基于分型、笔、线段、中枢等可定义的几何结构，为量化交易提供了规则化基础。通过Python将K线包含合并、顶底分型识别、笔划分、中枢区间判定及买卖点信号等逻辑写成代码，可有效避免人眼偏差，实现可回测、可复现的量化分析流程。本文从K线数据结构与合并方向入手，讲解分型判定、笔和线段的程序化实现，以及基于笔中枢的三类买卖点判断，并讨论MACD面积背驰等常见参数陷阱。适用于股票、期货等行情数据的自动化结构分析，也适合开发者构建个人量化工具链时参考。

1. 缠论程序化的起点：为什么把 K 线包含、分型和笔写成代码是第一步

缠论经常被人当成玄学，背驰、中枢、级别这些东西听起来主观，好像只能靠经验去数K线。但实际拆开看，顶底分型、笔、线段、中枢都有明确的几何定义，这部分规则其实非常适合程序化。ChZhShCh 就是一个偏实战的 Python 项目，它把缠论里最常用的结构判断——K线包含处理、分型识别、笔划分、中枢生成、三类买卖点信号——翻译成可追溯、可复现的代码逻辑，而不是靠人眼去K线图上比划。对想在股票或期货行情上做量化分析、又不想被商业指标黑匣子套住的开发者来说，这套代码的价值在于把模糊概念变成带参数、可画图验证的工具链。下面我按拆项目的顺序，把每个模块怎么落地、参数怎么设、哪些地方容易翻车讲清楚。

2. K线包含处理与顶底分型：把原始K线变成结构化序列

2.1 包含关系的程序化合并：向上与向下两种方向

K线包含是指两根相邻K线的高点、低点存在完全覆盖关系，一根K线的高低点区间把另一根完全包住。缠论里包含关系必须先处理，否则后面的分型和笔会被这些不规则K线干扰。合并方向取决于当前走势方向，向上趋势里高点取高、低点也取高，向下趋势里低点取低、高点也取低。

我在拆这类项目时，会先用一个独立的 Bar 结构来装K线，而不是直接操作 DataFrame，因为合并过程要反复修改最后一根K线，列表操作比逐行迭代 DataFrame 快得多。

from dataclasses import dataclass @dataclass class Bar: date: str high: float low: float open: float = 0.0 close: float = 0.0 end_date: str = "" def is_inclusive(a: Bar, b: Bar) -> bool: # b 被 a 包含，或者 a 被 b 包含 return (a.high >= b.high and a.low <= b.low) or \ (a.high <= b.high and a.low >= b.low) def merge_inclusive(raw: list[Bar]) -> list[Bar]: if not raw: return [] bars = [raw[0]] direction = 0 for bar in raw[1:]: last = bars[-1] if not is_inclusive(last, bar): bars.append(bar) if len(bars) >= 2: prev = bars[-2] direction = 1 if bar.high > prev.high or bar.low > prev.low else -1 continue if direction == 0: prev = bars[-2] if len(bars) >= 2 else raw[0] direction = 1 if bar.high > prev.high else -1 if direction == 1: merged = Bar( date=last.date, high=max(last.high, bar.high), low=max(last.low, bar.low), open=last.open, close=bar.close, end_date=bar.date ) else: merged = Bar( date=last.date, high=min(last.high, bar.high), low=min(last.low, bar.low), open=last.open, close=bar.close, end_date=bar.date ) bars[-1] = merged while len(bars) >= 2 and is_inclusive(bars[-2], bars[-1]): a, b = bars[-2], bars[-1] if direction == 1: merged = Bar( date=a.date, high=max(a.high, b.high), low=max(a.low, b.low), open=a.open, close=b.close, end_date=b.end_date ) else: merged = Bar( date=a.date, high=min(a.high, b.high), low=min(a.low, b.low), open=a.open, close=b.close, end_date=b.end_date ) bars[-1] = merged bars.pop(-2) return bars

这段代码的逻辑可以拆成三步看：先判断当前K线和结果列表最后一根是否包含；被包含时根据方向合并；合并结果可能继续与更早一根形成包含，所以必须回退重查。

参数说明里最容易被忽略的是 direction 的维护。direction 只表示当前局部的合并方向，它由最近两根有效K线的相对高低决定。向上合并时 low 取 max而不是 min，这符合缠论里“向上处理高高、高低”的口诀。有些实现只处理一次包含就继续，不写这个 while 回退，导致后续K线仍然互相包含，分型结果自然错乱。我在项目里见过不少类似翻车，所以建议保留回退逻辑。

date 和 end_date 两个字段是给后面画图和回测对齐用的。合并后K线的 date 保留第一根原始K线日期，end_date 记录最后一根参与合并的原始K线日期。open 和 close 对分型没有直接用处，但中间调试时能看合并前后的价格变化范围，所以也留着。

2.2 顶底分型判定：方向、有效性与多余分型处理

分型是缠论里最小的结构单元。三根K线中，中间一根高点最高就是顶分型，中间一根低点最低就是底分型。这个定义本身很简单，但放到程序里有几个细节需要处理：合并后K线列表长度、边界下标、连续同向分型的去重。

def find_fractal(bars: list[Bar]) -> list[dict]: fractal = [] for i in range(1, len(bars) - 1): left, mid, right = bars[i - 1], bars[i], bars[i + 1] if mid.high > left.high and mid.high > right.high: fractal.append({ "type": "top", "idx": i, "price": mid.high, "bar": mid }) elif mid.low < left.low and mid.low < right.low: fractal.append({ "type": "bottom", "idx": i, "price": mid.low, "bar": mid }) return fractal

这个函数需要传入 merge_inclusive 处理后的 bars，而不是原始K线。返回的 idx 是合并后列表的下标，画图时要映射回原始K线坐标，否则分型点会错位。

分型判定需要严格的大于或小于。如果中间K线的高点与左右两边相等，按严格定义不算顶分型。有些人的实现里允许相等也算，我只是建议给这个函数加一个equal_allowed=False参数，默认关闭，需要时再打开。

连续同向分型是下一个模块要处理的问题。比如连续上涨过程中出现两个顶分型，这两个顶分型中间可能只隔了一根K线。笔划分模块只会保留价格更高的那个顶分型，这里不处理也说得过去，但提前在分型阶段过滤会更省事。

2.3 为什么这一步直接决定笔划分质量

K线包含处理和分型识别是整个缠论程序化的地基。如果合并方向判断错，合并后K线的高低点就会偏移，分型位置跟着错，后面笔的端点错位，线段和中枢全部漂移。

我一般会在合并函数写完后先跑一段自检代码，检查合并后K线是否还存在包含关系，以及合并比例是否异常。

def check_merge(df): raw = [Bar(date=str(r.date), high=r.high, low=r.low) for r in df.itertuples()] merged = merge_inclusive(raw) print(f"原始K线: {len(raw)} 根 -> 合并后: {len(merged)} 根") for i in range(1, len(merged)): if is_inclusive(merged[i - 1], merged[i]): print(f"警告: 下标 {i - 1} 与 {i} 仍存在包含关系")

正常情况下同一周期日线行情合并比例不会太高，比如200根K线合并后可能剩160根左右。如果合并后数量几乎没有变化，基本可以确定包含关系没处理干净。这个自检步骤我每次换数据源都会跑一遍，数据源不同K线精度不同，合并结果差异会很明显。

3. 笔与线段：程序化划分思路、特征序列与实际落地

3.1 笔划分的判定标准：顶底分型之间至少要几根K线

笔是顶底分型之间的连线。理论定义很清晰，程序化实现时的争议主要在K线根数。经典笔要求顶底分型之间至少间隔5根合并后K线，新笔简化后放宽到4根。这个参数直接决定笔的灵敏度，周期越短越需要调小。

def find_bi(fractals: list[dict], gap_min: int = 5) -> list[dict]: bi = [] prev = None for fr in fractals: if prev is None: prev = fr continue if prev["type"] == fr["type"]: if fr["type"] == "top" and fr["price"] >= prev["price"]: prev = fr elif fr["type"] == "bottom" and fr["price"] <= prev["price"]: prev = fr continue if fr["idx"] - prev["idx"] < gap_min: if (fr["type"] == "top" and fr["price"] > prev["price"]) or \ (fr["type"] == "bottom" and fr["price"] < prev["price"]): prev = fr continue if prev["type"] == "bottom" and fr["type"] == "top": if fr["price"] <= prev["price"]: prev = fr continue if prev["type"] == "top" and fr["type"] == "bottom": if fr["price"] >= prev["price"]: prev = fr continue bi.append({ "start_idx": prev["idx"], "end_idx": fr["idx"], "start_price": prev["price"], "end_price": fr["price"], "type": "up" if fr["type"] == "top" else "down", "start_bar": prev["bar"], "end_bar": fr["bar"] }) prev = fr return bi

这里的逻辑要点有三个。一是连续同向分型只保留最极端的那个顶或底，避免出现内部小笔；二是分型距离不够时，保留价格更极端的分型，这比直接丢弃更合理；三是方向校验，底分型价格不能高于顶分型，否则这根笔方向是反的。

gap_min 是最重要的调参项。日线上用 5，5分钟或15分钟这种小周期可以降到 4，再小就容易画出频繁翻转的碎笔。我在实盘测试时习惯把 gap_min 和后续中枢判断联动，先扫一遍 3 到 7 的取值范围，再结合回测结果选参数，而不是凭感觉写一个固定值。

3.2 线段划分的核心：特征序列与缺口

笔构成线段，线段的划分在缠论里比笔更复杂。向上线段中所有向下笔组成特征序列，特征序列出现顶分型，意味着向上线段可能被破坏。缺口的存在会让确认延后，没有缺口时可以直接确认。

我在程序里先定义了一个 SegmentNode，用于包装笔和计算区间高低点。

class SegmentNode: def __init__(self, bi): self.bi = bi self.high = max(bi["start_price"], bi["end_price"]) self.low = min(bi["start_price"], bi["end_price"]) self.is_up = bi["type"] == "up" def build_feature_sequence(nodes): return [n for n in nodes if not n.is_up] def has_top_fractal(seq): for i in range(1, len(seq) - 1): if seq[i].high > seq[i - 1].high and seq[i].high > seq[i + 1].high: return True return False

这只是一个片段，完整线段划分还需要处理第一特征元素和第二特征元素之间的缺口判断。如果特征序列分型中间元素的高点明显高于两侧，同时分型前后的元素区间存在缺口，线段结束点要往后推迟一根笔确认。

实际项目里大部分开发者会先跳过严格线段，用笔级别近似。ChZhShCh 这类项目的落地方式通常也是先跑通笔中枢，再慢慢补线段逻辑。如果你只需要找中枢和买卖点，笔级别够用；但要做多级别趋势判断和背驰分析，线段还是绕不过去。

3.3 简化线段划分：先跑通，再追求严格

严格线段划分代码很长，我先给一个简化的笔破坏版本。它把一个线段看成连续同向笔的集合，一旦出现反向笔突破前一笔的起点，就认为当前线段结束。

def split_segments(bi_list: list[dict]) -> list[dict]: if len(bi_list) < 3: return [] segments = [] last_boundary = 0 for i in range(2, len(bi_list)): prev_bi = bi_list[i - 1] cur_bi = bi_list[i] if prev_bi["type"] == "up" and cur_bi["type"] == "down": if cur_bi["end_price"] < prev_bi["start_price"]: segments.append({ "start_idx": bi_list[last_boundary]["start_idx"], "end_idx": prev_bi["end_idx"] }) last_boundary = i elif prev_bi["type"] == "down" and cur_bi["type"] == "up": if cur_bi["end_price"] > prev_bi["start_price"]: segments.append({ "start_idx": bi_list[last_boundary]["start_idx"], "end_idx": prev_bi["end_idx"] }) last_boundary = i return segments

这段代码的优点是简单直接，缺点是没有处理特征序列的缺口和包含关系。对日线级别的大趋势判断，这种近似能覆盖大部分场景；但到了5分钟级别，笔本身已经比较碎，直接用笔破坏切线段容易出现线段端点频繁跳动，建议在更高周期上使用。

从实用角度说，先跑通这个简化版本，把笔、线段、中枢画到图上对比人工标注结果，再决定要不要上严格的特征序列逻辑，比一上来就啃完整定义效率高很多。

4. 中枢与三类买卖点：程序化判定和级别递归

4.1 中枢区间的定义与判定：ZG、ZD、GG、DD

中枢是缠论里最核心的参考结构。连续三段次级别走势的重叠区域构成中枢，中枢上边沿叫 ZG，下边沿叫 ZD。三段走势重叠意味着它们两两之间有交集，程序上用每笔高低点来判断。

def bi_high(b): return max(b["start_price"], b["end_price"]) def bi_low(b): return min(b["start_price"], b["end_price"]) def overlap_range(bi_group): zd = max(bi_low(b) for b in bi_group) zg = min(bi_high(b) for b in bi_group) return zd, zg def find_center_by_bi(bi_list: list[dict]) -> list[dict]: centers = [] i = 0 while i + 2 < len(bi_list): group = bi_list[i:i + 3] zd, zg = overlap_range(group) if zg > zd: centers.append({ "start_idx": group[0]["start_idx"], "end_idx": group[2]["end_idx"], "zd": zd, "zg": zg, "bi_indices": [i, i + 1, i + 2] }) i += 3 else: i += 1 return centers

这里用三笔重叠来近似中枢。严格缠论中枢应该由次级别走势重叠构成，但在笔级别的程序化实现里，把连续三段笔的重叠区域当作中枢是常见做法，代码里也经常叫它笔中枢。

需要注意 zd 和 zg 的计算方向。zd 取三笔低点里的最大值，zg 取三笔高点里的最小值。如果 zd 大于 zg，说明三笔没有重叠，不能构成中枢。代码里 i 的步进是 sliding window 和跳跃交错的，找到中枢后 i 加 3 跳到下一组，避免在同一区域反复识别出多个同类中枢。

中枢延伸我在实现时会在 find_center 后面再接一个状态机，后续每来一笔，只要笔的运行区间和 [zd, zg] 有交集，就继续延伸并更新 end_idx；一旦出现一笔完全离开这个区间且回抽不返回，中枢才正式结束。

4.2 三类买卖点的程序化判定：背驰用 MACD 面积，第三类看回抽

三类买卖点需要结合中枢位置和背驰判断。第一类买点出现在下跌趋势最后一个中枢之后的背驰段，第二类买点在第一类买点出现后回踩不破前低，第三类买点在向上离开中枢后回抽不进中枢区间时出现。

第一类买点的程序化实现通常用 MACD 面积做辅助。先计算MACD柱的绝对值面积作为动能度量，再比较中枢前后两段下跌的动能大小。

def detect_buy1(df, centers, bi_list): signals = [] for c in centers: down_bi = None for b in bi_list: if b["start_idx"] >= c["end_idx"] and b["type"] == "down": down_bi = b break if down_bi is None: continue if down_bi["end_price"] < c["zd"]: seg_prev = df.iloc[c["start_idx"]:c["end_idx"] + 1]["macd_hist"].abs().sum() seg_cur = df.iloc[down_bi["start_idx"]:down_bi["end_idx"] + 1]["macd_hist"].abs().sum() if seg_cur < seg_prev * 0.9: signals.append({ "type": "buy1", "price": down_bi["end_price"], "idx": down_bi["end_idx"] }) return signals

MACD面积比较里的 0.9 是一个容忍系数。严格背驰要求面积明显缩小，但行情波动大时两段面积差异可能没那么大，所以项目里通常给一个阈值。你可以把它做成参数，比如macd_shrink_ratio=0.9，意思是当前段面积小于前一段的90%才触发信号。

第二类买点实现时，需要记录第一类买点后向上笔的终点，然后等向下回踩笔不破前低。第三类买点更简单，向上离开中枢后，等一根向下笔的低点高于中枢ZG，就可以标记为三买。实际使用中第三类买点可靠性比第一类高，但信号出现频率更低。

4.3 多级别递归：从分钟线到日线中枢的参数联动

缠论的级别概念在程序化里常被实现为递归合成。最小级别用5分钟K线，画出笔；多个5分钟笔构成5分钟线段；再把5分钟线段当作30分钟笔的输入，层层向上合成日线中枢。

这套流程用参数配置来管理比较清晰。

LEVEL_CONFIG = { "5m": {"gap_min": 4, "macd_fast": 12, "macd_slow": 26, "macd_signal": 9}, "30m": {"gap_min": 5, "macd_fast": 12, "macd_slow": 26, "macd_signal": 9}, "1d": {"gap_min": 5, "macd_fast": 12, "macd_slow": 26, "macd_signal": 9} }

gap_min 在不同级别上取值不同。5分钟K线噪音大，用 4 能更快捕捉分型；30分钟和日线用 5 更稳。MACD 参数一般沿用 12、26、9，不需要改动。

递归合成的坑在于数据对齐。5分钟线段划完以后，线段端点对应的不是自然时间边界，而是行情时间戳。向上合成时，必须按时间戳重新采样到目标周期，不能简单按K线序号对齐，否则中枢区间会偏移。

我一般建议先用单级别把逻辑跑通，再上多级别递归。单级别中枢已经能解决大部分选股和结构判断需求，多级别递归主要用于确认趋势方向和背驰级别，搞得太复杂反而容易过拟合。

5. 缠论程序化的常见问题排查：包含合并、笔划分与参数设置的四个坑

5.1 合并K线后指标对不上原始行情

现象：分型和笔画在K线图上，端点不在K线最高或最低位置，画的笔像被平移了一两根K线。

原因：分型识别用的是合并后列表下标，但画图时直接把这个下标当成了原始K线的下标。合并后K线数量比原始K线少，下标对不上。我把 Bar 结构改成带映射后再看，问题立刻定位到 raw_start 和 raw_end 两个字段没维护。

解决：合并时记录每个合并结果对应的原始K线范围。

@dataclass class MergeBar: high: float low: float raw_start: int raw_end: int

每次合并时更新 raw_end 为当前原始K线下标。分型记录的 idx 改成 MergeBar 的下标，绘图时用 merged_bar.raw_end 作为 x 轴坐标，分型点就能准确落在K线上。这个映射在数据量小时看不出来，数据一多错位感非常明显。

5.2 分型确认太早导致回测漂移

现象：历史回测信号很好，实盘跟踪时发现信号经常滞后或反复消失。

原因：分型的确认依赖第三根K线收盘。很多人在程序里遍历到第二根K线时就判定分型成立，当时中间K线确实高于左右，但第三根K线还没走完，随时可能破坏这个分型。回测数据是完整的，程序等于用了未来数据，收益自然虚高。

解决：统一延迟到第三根K线走完后再确认分型，回测撮合价用确认后下一根K线的开盘价。

for i in range(1, len(bars) - 1): # bars[i+1] 收盘后才能确认分型 confirmed = bars[i + 1] signal_price = bars[i + 2].open if i + 2 < len(bars) else confirmed.close

这个改动看似小，但直接影响所有买卖点信号的可信度。从那以后我每次回测前都会加一遍这样的延迟校验。

5.3 三笔重叠被重复识别成多个中枢

现象：一段日线趋势里输出了三四个重叠中枢，区间相互嵌套，画图之后没法看。

原因：中枢识别时用滑动窗口每移动一笔就重新找三笔重叠，找到后不延伸也不判断结束，导致同一段震荡行情被反复分类成多个中枢。

解决：加一个延伸状态机。找到有效重叠区间后，后续笔只要还和 [zd, zg] 区域有交集，就归入原中枢并更新 end_idx；只有出现明确的离开行为，才结束当前中枢并开启新一轮搜索。

center = None for b in bi_list: low, high = bi_low(b), bi_high(b) if center is None: # 用连续三笔尝试启动中枢 else: if high > center["zd"] and low < center["zg"]: center["end_idx"] = b["end_idx"] else: centers.append(center) center = None

这类问题排查时要重点看 start_idx 和 end_idx 是否连续，如果两个中枢的端点在时间上有重叠，基本上就是窗口逻辑没加延伸判断。

5.4 MACD 面积比较没有处理正负号

现象：背驰条件几乎从来不触发，或者触发后行情继续暴跌。

原因：macd_hist 列有正有负，直接 sum 会把红柱和绿柱互相抵消。下跌段的绿柱面积明明很大，但正负相加后数值很小，反而误判成背驰。

解决：先取绝对值再求和。

seg_hist_abs = df.iloc[start:end + 1]["macd_hist"].abs().sum()

这个坑非常低级但影响极大，我在第一次跑背驰逻辑时被它坑过，当时第一类买点信号全部集中在下跌初期，回测结果一塌糊涂。现在我把 MACD 面积计算封装成一个独立函数，任何地方调用都强制走绝对值逻辑。

6. 验证与进阶：把笔和中枢画到K线上，用历史回归测试调参

6.1 可视化验证：分型、笔和中枢一屏展示

代码写完之后第一件事永远是画图。参数对不对、笔端点有没有错位、中枢区间是否合理，看一张图比看十个print输出都直观。

import matplotlib.pyplot as plt from mpl_finance import candlestick_ohlc def plot_chan(df, merged_bars, fractals, bi_list, centers): df_reset = df.reset_index(drop=True) ohlc = [[i, row["open"], row["high"], row["low"], row["close"]] for i, row in df_reset.iterrows()] fig, ax = plt.subplots(figsize=(16, 8)) candlestick_ohlc(ax, ohlc, width=0.6, colorup="r", colordown="g") for fr in fractals: idx = merged_bars[fr["idx"]].raw_end marker = "v" if fr["type"] == "top" else "^" color = "blue" if fr["type"] == "top" else "orange" ax.scatter(idx, fr["price"], marker=marker, color=color, s=40) for b in bi_list: x0 = merged_bars[b["start_idx"]].raw_end x1 = merged_bars[b["end_idx"]].raw_end ax.plot([x0, x1], [b["start_price"], b["end_price"]], color="black", linewidth=1.2) plt.show()

绘图时需要注意，merged_bars 的下标必须和传入的 fractals、bi_list 里记录的 idx 对应，然后统一通过 raw_end 映射到原始K线坐标。如果不做这个映射，图上的分型点很容易飘。

6.2 用历史回归校验笔的划分质量

调参最怕自嗨。我一般从历史数据里随机抽20段日线行情，自己先人工标注每段的顶底分型和笔端点，再跑程序自动划分，最后计算自动结果与人工标注的匹配率。

def bi_accuracy(manual_idx, auto_idx, tolerance=1): hits = sum(1 for m in manual_idx if any(abs(m - a) <= tolerance for a in auto_idx)) return hits / len(manual_idx) if manual_idx else 0

tolerance 取 1，意思是自动识别的端点允许偏差一根K线。对 gap_min 从 3 到 7 分别跑一遍，画一条准确率曲线，选曲线从快速上升转为平缓的拐点作为最终参数。日线上拐点基本落在 4 到 5 之间，周期越小拐点越靠前。

从那以后我每次给项目加新功能，都强制先跑一遍这套可视化加回归校验流程再做策略逻辑。缠论定义本身不复杂，但程序实现里每一层都在放大上一层的误差，中间不做校验，最后靠信号买卖等于在盲人摸象。希望这个拆解过程能帮你少走几步弯路，把时间花在真正有价值的策略验证上。

本文还有配套的精品资源，点击获取