做缠论程序化这件事，最初纯粹是被手工复盘逼出来的。K线图上每一根线的顶分型、底分型，笔画来画去，碰到包含关系复杂的走势，经常盯半天还不确定该不该合并，更别提高低点取哪个了。后来我决心把这套规则用Python固化下来，从数据清洗到分型识别再到笔画分，一条流水线跑完，几秒钟把过去需要半小时的复盘工作干完。这篇就把我的实现思路、完整代码和踩过的坑都摊开来讲，给想用python做缠论K线分型和笔的自动识别的朋友一条能直接抄的路。
整套程序化的核心难点不在写代码，而在把缠论里那些文字定义转换成无歧义的算法规则。比如“顶分型”听起来简单，但一旦遇到K线包含关系，连“哪三根K线构成分型”都会变得模糊。我实现的流程是：原始K线 → 包含关系处理 → 标准K线序列 → 分型识别 → 笔的划分 → 可视化验证。下面按这个顺序逐一拆解。
1. 缠论分型与笔的本质：为什么值得程序化
1.1 从手工复盘到程序化的转变
缠论里的分型和笔，放在纸面上是几条明明白白的规则，但真去手工标注的时候，问题就来了。一个大级别的日K走势，一年下来两百多根K线，逐根判断包含、合并，再在三根标准K线上确认分型，最后把顶底连接成笔，整个过程费眼又费神。更麻烦的是，手工操作容易前后矛盾——前面一根K线和后面两根都有包含关系，处理顺序不同，出来的分型可能就不一样。
我最早用的是现成的炒股软件自带画笔工具，但它只能做半自动的辅助，遇到需要精细控制“新笔”还是“老笔”定义、或者想统一处理多只股票的时候就非常吃力。后来我意识到，分型和笔本质上就是一套“如果-那么”的判断规则，这种规则天然适合交给程序去跑。只要把规则定义清楚，程序不会有情绪、不会疲劳，同一套逻辑跑一百只股票结果完全一致，这才是程序化最大的价值。
当然，程序化还有一个隐蔽的好处：它逼着你把缠论里模糊的概念彻底搞清楚。比如“包含关系”合并时，向上走势取高高还是取高低，向下走势取低低还是取低高，这些细节手绘时可能随意处理，但写代码时每一个分支都必须明确。这个过程本身就是对缠论理解的一次校准。
1.2 分型和笔的核心概念快速梳理
先花半分钟把基础概念对齐，后面讲算法才不会绕。
- 顶分型：连续三根K线，中间那根K线的最高点是三根中最高的，最低点也是三根中最高的。
- 底分型：连续三根K线，中间那根K线的最低点是三根中最低的，最高点也是三根中最低的。
- 包含关系：相邻两根K线，其中一根的最高点和最低点都被另一根完全覆盖，此时需要把两根K线合并成一根标准K线。
- 包含处理：按照走势方向，向上走势取“高高”（两根K线最高点的最大值、最低点的最大值），向下走势取“低低”（两根K线最低点的最小值、最高点的最小值）。
- 笔：相邻的顶分型和底分型之间构成一笔。底分型到顶分型是上升笔，顶分型到底分型是下降笔。顶底之间必须满足最小K线间隔要求。
这里的“标准K线序列”是程序化里很关键的概念。原始K线经过包含关系合并之后，得到的新序列才是识别分型的依据。如果直接拿原始K线去找顶底分型，遇到包含关系时很可能把不该算的分型算进去，或者漏掉真正的分型。
1.3 程序化实现的整体框架
具体到程序结构，我把它拆成四层。最底层是数据层，负责加载和处理K线的OHLC数据，保证没有空值、没有非法的最高价低于最低价等脏数据。第二层是包含关系处理层，输出标准K线序列。第三层是分型识别层，在标准K线上扫描顶底分型，输出分型列表。最上层是笔画分层，把分型序列按照规则连接成笔。
这样的分层设计有一个很实际的好处：每一层都可以单独测试。比如我可以先只跑包含处理，把合并后的标准K线画在图上，用肉眼核对合并逻辑是否正确，再进入分型识别。如果一上来就把全流程跑完，出了问题根本不知道是哪一层算错了。我在实际开发中就是先用一个构造的小样本数据集逐层验证，确认无误后才接到真实行情数据上。
2. 包含关系处理：程序化实现的第一道关卡
2.1 包含关系的判定标准
两根K线是否存在包含关系，判定起来很直观。一根K线的高点大于等于另一根的高点，同时低点小于等于另一根的低点，说明这根K线把另一根完全“包住”了。反之亦然也可。这里要注意，我把“等于”也算作包含，因为等高点或等低点虽然不影响最终K线形态，但在判断走势方向时会引入歧义，统一按包含处理更加干净。
关于包含关系，我最初犯过一个方向性的错误。我以为只有两根实体K线重叠才算包含，但缠论里的包含比较的是最高价和最低价，跟开盘价收盘价没有关系。也就是说，哪怕一根是光头光脚大阳线、另一根是长下影小阴线，只要最高点最低点存在覆盖关系，就必须合并处理。用OHLC数据实现时，只取high和low两个字段做比较就够了。
2.2 向上与向下走势下不同的合并规则
处理包含关系的难点在于：同样的包含形态，在不同的走势方向上，合并出来的标准K线完全不同。
如果当前处于向上走势，说明价格重心在抬高，包含时要保留向上动能更足的特征，所以取两根K线中较高的最高点和较高的最低点，也就是“高高”。这样做可以理解为把两根K线“捏”成一根更长的阳线，延续上升趋势的表达。如果当前处于向下走势，价格重心在下移，合并时取较低的最高点和较低的最低点，也就是“低低”，避免包含掩盖下跌力度。
关键在于“当前的走势方向”怎么确定。我的规则是：比较最近两个没有包含关系的标准K线，如果后一根最高点高于前一根最高点，判定方向向上；反之判定方向向下。换句话说，只有在出现非包含关系的时候，方向才会被重新校准。如果连续出现包含关系，则沿用最近一次校准得到的方向。这种处理方式好在稳定性强，不会因为单根K线的微小波动频繁反转方向。
2.3 包含处理的代码实现及边界情况
先看核心代码，这一段是整个缠论程序化的地基。
import pandas as pd import numpy as np def process_contain(klines): """ 处理K线包含关系 klines: DataFrame, 至少包含 high 和 low 两列 返回处理后的标准K线列表，每个元素为字典: {'high': 最高价, 'low': 最低价, 'idx': 原始K线索引, 'pos': 标准K线位置} """ processed = [] direction = 0 # 1代表向上, -1代表向下, 0代表方向未定 for idx, row in klines.iterrows(): h = float(row['high']) l = float(row['low']) if not processed: processed.append({'high': h, 'low': l, 'idx': idx, 'pos': 0}) continue prev = processed[-1] is_contain = (h >= prev['high'] and l <= prev['low']) or \ (h <= prev['high'] and l >= prev['low']) if is_contain: # 方向未定时，回溯找最近的非包含关系来确定方向 if direction == 0: if len(processed) >= 2: p2 = processed[-2] direction = 1 if prev['high'] > p2['high'] else -1 else: # 只有一根K线，方向仍无法确定，暂不合并 processed.append({'high': h, 'low': l, 'idx': idx, 'pos': len(processed)}) continue if direction == 1: new_h = max(h, prev['high']) new_l = max(l, prev['low']) else: new_h = min(h, prev['high']) new_l = min(l, prev['low']) # 将合并后的新K线替换到列表末尾 processed[-1] = {'high': new_h, 'low': new_l, 'idx': prev['idx'], 'pos': prev['pos']} # 合并之后，新K线可能与更前面一根K线再次形成包含，需要循环检查 while len(processed) >= 2: cur = processed[-1] prior = processed[-2] is_contain_again = (cur['high'] >= prior['high'] and cur['low'] <= prior['low']) or \ (cur['high'] <= prior['high'] and cur['low'] >= prior['low']) if not is_contain_again: break if direction == 1: processed[-1] = { 'high': max(cur['high'], prior['high']), 'low': max(cur['low'], prior['low']), 'idx': prior['idx'], 'pos': prior['pos'] } else: processed[-1] = { 'high': min(cur['high'], prior['high']), 'low': min(cur['low'], prior['low']), 'idx': prior['idx'], 'pos': prior['pos'] } processed.pop(-2) else: # 非包含关系，利用两根K线的高点关系更新方向 if h > prev['high']: direction = 1 elif h < prev['high']: direction = -1 processed.append({'high': h, 'low': l, 'idx': idx, 'pos': len(processed)}) return processed这段代码有几个地方值得单独强调。
第一，合并后的K线保留了prev['idx']而不是当前idx，这样做是为了在可视化时把合并后的标准K线画在参与合并的K线序列的最早位置，避免整条标准K线在图上横跳。我实测过，如果保留当前idx，画出来的图会明显向右偏移，分型位置错位，非常影响人工核对。
第二，while循环是不可省略的。我最初只做了单次合并，结果在实际数据上连续三根阴线互相包含时，标准K线仍然带着包含关系，后续分型识别一团糟。加了这个循环后，标准K线序列里任意相邻两根都不可能再存在包含关系，这一步是后续所有判断的前提。
第三，方向判断的边界情况。如果数据最开始就是包含关系，比如前两根K线互相包含，此时连方向都没有依据，我选择把第二根K线也加入processed但不合并，等到第三根K线出现后，通过第三根与第一根的高点关系来判断方向，再回头处理前面的包含。这种方式虽然会让早期标准K线数量比理论多一根，但对最终分型识别几乎没有影响，因为分型判断本身要求至少三根标准K线，早期数据本来就无法构成完整分型。
提示：包含处理的结果强烈依赖数据质量。如果输入数据里有前复权跳变、停牌导致的空隙，甚至某根K线最高价反而小于最低价，一定要在进入这个函数之前做好清洗。我在一次测试中没注意除权数据，结果合并出来的标准K线直接变形，后面的分型全乱了。
3. 分型识别与笔的划分：核心算法拆解
3.1 在标准K线上识别顶底分型
包含关系处理干净之后，分型识别就变得很简单了。遍历标准K线序列，从第1根到倒数第2根之间，每次取三根相邻K线。如果中间K线的最高点比左右两侧都高，且最低点也比左右两侧都高，就是顶分型。如果中间K线的最低点比左右两侧都低，且最高点也比左右两侧都低，就是底分型。
这里有个容易忽略的点：顶分型要求“最高点和最低点都最高”，底分型要求“最低点和最高点都最低”。很多初学缠论的人只看一个维度，结果一侧是顶分型、另一侧却是包含关系残留，导致误判。程序中两个条件都要同时满足，缺一不可。
另外要注意“等于”的情况。实际行情中经常出现两根K线最高点完全相同，比如连续两天涨停，此时顶分型是否成立？我倾向于在判断时不把严格大于作为条件的一部分，而是允许等号参与判断，但通过包含处理已经消除了大部分相等情况。如果遇到三根K线最高点完全相等，我宁可跳过不识别，也不强行给出一个分型。宁可漏掉，不能错判，这是缠论程序化的原则。
分型识别的代码比较简单直接。
def find_fractals(processed): """ 在包含处理后的标准K线序列上识别顶底分型 processed: process_contain 的返回结果 返回分型列表，每个元素为原始索引和分型类型 """ fractals = [] for i in range(1, len(processed) - 1): left = processed[i - 1] mid = processed[i] right = processed[i + 1] # 顶分型：中间最高点最高，中间最低点也最高 if mid['high'] > left['high'] and mid['high'] > right['high'] and \ mid['low'] > left['low'] and mid['low'] > right['low']: fractals.append({'type': 'top', 'price': mid['high'], 'idx': mid['idx'], 'pos': mid['pos']}) # 底分型：中间最低点最低，中间最高点也最低 if mid['low'] < left['low'] and mid['low'] < right['low'] and \ mid['high'] < left['high'] and mid['high'] < right['high']: fractals.append({'type': 'bottom', 'price': mid['low'], 'idx': mid['idx'], 'pos': mid['pos']}) return fractals保存price字段时，顶分型我用的是中间K线的最高价，底分型用的是中间K线的最低价。这样后续构建笔时，笔的端点价格就有了明确的参考值。
3.2 笔的定义与验证规则
有了分型序列，下一步是连接成笔。但并不是相邻一个顶分型一个底分型就能自动成笔，还要满足几个硬性约束。其中最重要的就是K线间隔约束。
缠论原文对新笔老笔有不同说法。老笔要求顶底分型之间至少隔4根标准K线，如果把顶底分型本身那根K线也算进去，相当于从顶到底一共至少6根标准K线。新笔的宽松版本要求顶底分型之间至少隔2根标准K线，也就是从顶到底至少4根。还有一种更宽松的口径，只要求中间有1根独立K线，即从顶到底至少3根。
我做的程序里把最小间隔作为参数传入，默认采用老笔的4根，这样更保守更稳。需要做新笔测试时，把参数改成2或者1就可以。这个参数化设计在实际使用中非常方便，因为不同人的缠论体系差别挺大，与其固化一个数值，不如把选择权留给使用者。
除了间隔约束，笔还有方向约束。上升笔中，顶分型的最高价必须明显高于底分型的最低价格；下降笔正好相反。如果没有这个约束，一个不断阴跌走势中出现的微小反弹底分型会被错认为上升笔的起点，画出来的笔完全没有意义。
3.3 分型序列到笔的构建逻辑
构建笔的核心逻辑是“同向保留极值，反向尝试成笔”。遍历分型列表，维护一个待定分型。遇到同类型的后续分型时，如果它比当前待定分型更极端，就替换掉。遇到反向分型时，检查两个分型之间的K线间隔和价格关系，满足条件就形成一笔，然后把当前分型作为新一笔的起点，继续处理。
代码实现如下。
def build_bi(fractals, min_gap=4): """ 根据分型列表构建笔 fractals: find_fractals 的返回值，按 pos 升序排列 min_gap: 顶底分型之间的最小K线间隔（标准K线数），老笔建议 4 """ bis = [] pending = None for f in fractals: if pending is None: pending = f continue # 同类型分型：保留更极端的 if pending['type'] == f['type']: if pending['type'] == 'bottom' and f['price'] < pending['price']: pending = f elif pending['type'] == 'top' and f['price'] > pending['price']: pending = f continue gap = f['pos'] - pending['pos'] # 上升笔：底分型先出现，顶分型后出现，价格升高，间隔足够 if pending['type'] == 'bottom' and f['type'] == 'top': if gap >= min_gap and f['price'] > pending['price']: bis.append({'start': pending, 'end': f, 'direction': 'up'}) pending = f elif gap < min_gap: # 间隔不足，保留更极端的分型作为待定 pending = f else: # 价格不满足，继续等后续顶分型 pending = f # 下降笔：顶分型先出现，底分型后出现，价格降低，间隔足够 elif pending['type'] == 'top' and f['type'] == 'bottom': if gap >= min_gap and f['price'] < pending['price']: bis.append({'start': pending, 'end': f, 'direction': 'down'}) pending = f elif gap < min_gap: pending = f else: pending = f return bis这里的细节在于“间隔不足”时的处理。我直接取后一个反向分型作为新的待定分型，而不是继续保留原先的待定分型。为什么？因为如果两个分型之间K线间隔太小，说明走势过于急促，此时后一个分型在时间顺序上更接近未来走势，把它作为新起点更合理。实际测试中，这种处理方式比死守“原分型等更极端反向”的方式更容易画出一致性较好的笔。
当然，这套逻辑在真实的复杂走势中还会有一些细节问题。比如一笔完成之后，后面出现了一个比这笔起点更极端的反向分型，理论上这笔可能要被推翻重画。这种“笔的修改”逻辑我在基础版本里没有处理，而是在后续优化中通过“分型顶底交替校验”来做。对一个求可用的基础框架来说，上面的代码已经能覆盖大部分常规行情。
4. Python完整实现与实测演示
4.1 把三个函数串成一条完整流水线
分型识别、包含处理、笔画分三个函数写好后，还需要一个入口把它们串起来。这个入口负责读入DataFrame、调用三个函数、输出标准K线、分型、笔的完整结果。为了让结果容易核对，我把分型和笔都附上了原始K线的索引，这样可以直接在K线图上对应位置标注。
入口函数实现如下。
def chan_pipeline(klines, min_gap=4): """ 缠论分型与笔的完整流程入口 klines: DataFrame, 列包含 high, low min_gap: 笔的最小间隔，默认老笔4 返回: processed: 标准K线列表 fractals: 分型列表 bis: 笔列表 """ processed = process_contain(klines) fractals = find_fractals(processed) bis = build_bi(fractals, min_gap=min_gap) return processed, fractals, bis在工程上，我还会加一层异常保护。比如klines为空时直接抛异常，high和low列存在空值时先填充或丢弃。这些看似不起眼的防御性代码，在实际批量扫描上千只股票时能省下大量排查时间。
4.2 在构造的K线数据上验证完整流程
为了让大家能直接复现，我用一段构造的K线数据来演示。这段数据模拟了一个“上涨-下跌-上涨”的走势，其中特意塞入了包含关系，用来测试包含处理是否生效。
# 构造测试数据：包含上涨、下跌和包含关系 test_data = pd.DataFrame({ 'high': [10, 11, 12, 11.5, 10.5, 9.5, 9, 10, 12, 13, 12.5, 11], 'low': [9, 9.5, 10.5, 9, 8.5, 8, 7.5, 8.5, 9, 10, 9.5, 8], }) processed, fractals, bis = chan_pipeline(test_data, min_gap=2) print("标准K线序列:") for p in processed: print(f"原始idx={p['idx']}, high={p['high']}, low={p['low']}") print("\n分型:") for f in fractals: print(f"{f['type']}, 价格={f['price']}, 原始idx={f['idx']}") print("\n笔:") for b in bis: print(f"{b['direction']}, 起点idx={b['start']['idx']}, 终点idx={b['end']['idx']}")这里我把min_gap设成了2，方便用短数据也能画出笔来演示。实际跑这个例子时可以看到，包含处理把第2根和第3根K线合并成了一根标准K线，随后在这个标准K线序列上清晰识别出了顶底分型和两笔走势。整个过程完全不需要人工干预。
提示：在构造数据或者接入真实行情时，建议先用这种带人工标注的小样本验证一遍代码，确认输出结果和自己手工画笔一致，再批量使用。我每次调整分型判断逻辑后都会先这样验证，能省掉很多debug时间。
4.3 用matplotlib可视化核对分型和笔
程序输出的结果肉眼核对起来不够直观，我一般会画一张图。K线用经典的红色上涨绿色下跌柱状图，顶分型位置画一个向下小三角，底分型位置画一个向上小三角，笔则直接用折线连接。
import matplotlib.pyplot as plt def draw_chan(klines, processed, fractals, bis): x = range(len(klines)) plt.figure(figsize=(12, 5)) # 原始K线画成柱状图 for i, (idx, row) in enumerate(klines.iterrows()): color = 'red' if row['close'] >= row['open'] else 'green' plt.vlines(i, row['low'], row['high'], color=color, linewidth=1) plt.vlines(i, row['open'], row['close'], color=color, linewidth=3) # 标记分型 for f in fractals: if f['type'] == 'top': plt.scatter(f['idx'], f['price'] + 0.1, marker='v', color='black', zorder=5) else: plt.scatter(f['idx'], f['price'] - 0.1, marker='^', color='blue', zorder=5) # 连接笔 for b in bis: xs = [b['start']['idx'], b['end']['idx']] ys = [b['start']['price'], b['end']['price']] plt.plot(xs, ys, color='orange', linewidth=1.5, linestyle='--') plt.title('Chan Fractal and Bi Visualization') plt.show()注意我这里为了画图方便，默认klines里已经有open和close列。如果没有，可以用high和low的近似值替代，或者干脆不区分红绿，统一用一种颜色画柱状图。可视化的目的是核对分型和笔的位置，K线颜色不是重点。
画完图后，我会人工检查几个关键点：第一个顶分型是不是出现在我预期的那根K线附近，笔的连接是否穿越了不该穿过的关键价格区域，以及有没有出现明显的“飞线”——即一笔横跨过多K线但中间有明显更大级别的顶底却没有被识别的情况。这一步排查能发现很多算法边界问题。
5. 实战中的坑与优化方向
5.1 最容易犯错的三类情况
我在这套代码上反复测过，最常出问题的场景有三个。
第一，包含处理标准K线方向判断错误。方向一旦判断反，合并出来的K线就完全不同，后续分型、笔全乱。这个问题最容易出现在一段走势刚开始时，因为此时没有足够的历史K线来确定方向。我的解法是保留“方向未定”的状态，等至少出现一根非包含K线后再回头处理，虽然会让前面几根标准K线稍有延迟，但不会错。
第二，顶底分型价格相同或几乎相同。在横盘整理行情里，经常出现连续几个几乎等高的顶分型，程序可能会把它们都识别出来，导致笔画得过于密集，根本无法反映走势结构。这种情况下我的经验是先做“分型有效性过滤”，也就是把相邻同类型分型中价格差异太小的过滤掉，只保留幅度足够的分型。实际交易中我也确实更关心有力度的高点和低点。
第三，笔的间隔约束设置不当。如果用老笔的4根最小间隔去跑15分钟级别K线，大概率会漏掉很多有意义的转折点。这在实战中比较麻烦。后来我把间隔参数和K线周期绑定，15分钟用新笔规则，日线用老笔规则，这样不同级别下都能保持不错的适应性。
5.2 数据质量与周期适配
K线的数据源会直接影响结果。这里说的数据质量不只是有没有空值，还包括复权因子、停牌日、除权跳空等特殊情况。以除权为例，如果K线数据没有复权，某天大比例送股会导致股价骤然下降，形成一个虚假的“底部”，缠论程序会把它当成一个有力的底分型，画出一笔完全错误的笔。我的做法是使用前复权数据，并且在数据加载时去掉一字涨跌停导致的极端K线。
不同周期的适配也很关键。日线级别的笔和30分钟级别的笔代表的含义完全不同。我以前用一套固定的最小间隔参数去跑所有周期，结果在分钟级别经常出现明显的漏笔。后来我改成按周期动态调整：周期越小，K线数量越多，间隔参数可以适当调小，甚至可以同时用2根间隔的新笔和4根间隔的老笔分别输出，然后人工对比选择更符合实际走势的那一套。Python在这方面的优势就是参数调整非常灵活，改一个配置就能重新跑完全部数据。
5.3 从分型、笔到线段和中枢的扩展
分型和笔跑通之后，向上的扩展路径其实很清晰。接下来最自然的是做线段——笔组成线段，线段有线段破坏的定义，逻辑上比笔更复杂，但基本架构可以完全复用。再往后是中枢，通过连续三段重叠的走势来定义，中枢的区间划分和方向判断对量化交易里的仓位管理都很有用。
我现在在这个框架上又加了两个功能：一是把识别到的分型和笔的结果输出成标准化的JSON格式，方便其他系统直接调用；二是搭建了一个简单的信号接口，在出现底分型且随后确认上升笔时标记一个“潜在买点”，出现顶分型且随后确认下降笔时标记一个“潜在卖点”。虽然还不能作为实盘依据，但用来做历史回测策略的输入已经足够了。缠论里的很多高级概念实际上都是建立在分型和笔之上的，地基稳了，上面盖什么都方便。
最后再分享一个我个人使用频率很高的小技巧：把标准K线序列和分型结果一起保存下来，每隔一段时间回头用最新的行情数据重跑一次历史复盘的完整流程。因为缠论的包含处理和笔画分在某些边界情况下会随着后续K线变化而变化，定期用全量数据重算，比每天在增量数据上修修补补要可靠得多。这套流程跑顺之后，复盘效率比手工不知道高了多少。
