简介技术分析中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( datelast.date, highmax(last.high, bar.high), lowmax(last.low, bar.low), openlast.open, closebar.close, end_datebar.date ) else: merged Bar( datelast.date, highmin(last.high, bar.high), lowmin(last.low, bar.low), openlast.open, closebar.close, end_datebar.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( datea.date, highmax(a.high, b.high), lowmax(a.low, b.low), opena.open, closeb.close, end_dateb.end_date ) else: merged Bar( datea.date, highmin(a.high, b.high), lowmin(a.low, b.low), opena.open, closeb.close, end_dateb.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_allowedFalse参数默认关闭需要时再打开。连续同向分型是下一个模块要处理的问题。比如连续上涨过程中出现两个顶分型这两个顶分型中间可能只隔了一根K线。笔划分模块只会保留价格更高的那个顶分型这里不处理也说得过去但提前在分型阶段过滤会更省事。2.3 为什么这一步直接决定笔划分质量K线包含处理和分型识别是整个缠论程序化的地基。如果合并方向判断错合并后K线的高低点就会偏移分型位置跟着错后面笔的端点错位线段和中枢全部漂移。我一般会在合并函数写完后先跑一段自检代码检查合并后K线是否还存在包含关系以及合并比例是否异常。def check_merge(df): raw [Bar(datestr(r.date), highr.high, lowr.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 是最重要的调参项。日线上用 55分钟或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 signalsMACD面积比较里的 0.9 是一个容忍系数。严格背驰要求面积明显缩小但行情波动大时两段面积差异可能没那么大所以项目里通常给一个阈值。你可以把它做成参数比如macd_shrink_ratio0.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[i1] 收盘后才能确认分型 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(dropTrue) 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, width0.6, colorupr, colordowng) 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], markermarker, colorcolor, s40) 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]], colorblack, linewidth1.2) plt.show()绘图时需要注意merged_bars 的下标必须和传入的 fractals、bi_list 里记录的 idx 对应然后统一通过 raw_end 映射到原始K线坐标。如果不做这个映射图上的分型点很容易飘。6.2 用历史回归校验笔的划分质量调参最怕自嗨。我一般从历史数据里随机抽20段日线行情自己先人工标注每段的顶底分型和笔端点再跑程序自动划分最后计算自动结果与人工标注的匹配率。def bi_accuracy(manual_idx, auto_idx, tolerance1): 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 0tolerance 取 1意思是自动识别的端点允许偏差一根K线。对 gap_min 从 3 到 7 分别跑一遍画一条准确率曲线选曲线从快速上升转为平缓的拐点作为最终参数。日线上拐点基本落在 4 到 5 之间周期越小拐点越靠前。从那以后我每次给项目加新功能都强制先跑一遍这套可视化加回归校验流程再做策略逻辑。缠论定义本身不复杂但程序实现里每一层都在放大上一层的误差中间不做校验最后靠信号买卖等于在盲人摸象。希望这个拆解过程能帮你少走几步弯路把时间花在真正有价值的策略验证上。本文还有配套的精品资源点击获取