缠论这套技术分析体系，这些年讨论热度一直不低，很多人一开始都是被“分型、笔、线段、中枢”这些概念给唬住了，感觉门槛很高。但真要说程序化落地，第一步其实没有想象中那么玄乎。把分型和笔的定义搞清楚，用Python写一套可运行的识别逻辑，不仅能加深对缠论的理解，还能顺手解决一个很实际的问题：每天打开行情软件手工去画分型、连线，太费时间了，而且人的判断标准很难保持一致。
我花了大概一个周末的时间，从数据结构设计到包含关系处理，再到分型识别和笔的最终确认，把这一整套流程跑通了。这篇文章就把整个过程拆开讲清楚，包括我踩过的坑和最后总结的注意事项，代码都是可以直接拿去跑的demo，适合正在用Python做技术分析程序化、或者对缠论感兴趣但还没找到代码切入点的人参考。
1. 先搞清楚分型和笔到底在说什么
程序化的前提是你得有一个足够精确的定义，不能靠“感觉”。缠论里对分型和笔的定义其实比较接近几何概念，一旦落到规则上，反而很适合写代码。
1.1 顶分型和底分型
分型是缠论里最基本的拐点信号，它由连续三根K线组成。顶分型的特征是：中间那根K线的最高价是这三根里最高的，同时它的最低价也是这三根里最高的。底分型则相反，中间那根K线的最低价最低，同时最高价也最低。
用文字描述可能有点绕，我画个简单的示意图：
- 顶分型：高高低高，中间是“凸”出来的
- 底分型：高低低低，中间是“凹”下去的
这里有一个细节点要特别注意：判断分型用的不是收盘价，也不是影线长度，而是每个K线的最高价和最低价这两个硬指标，并且要同时满足。如果只用“中间高点最高”这一个条件，很多假信号会混进来，程序写出来也不稳定。实际处理的时候，我还会在代码里加上一个前提——先处理K线之间的包含关系，否则分型形状很容易被扭曲。这一点后面单独讲，非常关键。
1.2 笔的定义和常见误区
笔的定义没有分型那么简单直白，它描述的是两个相邻分型之间的连线关系。上升笔就是从一个底分型连到一个顶分型，下降笔则相反，从一个顶分型连到一个底分型。
但并不是随便两个分型就能连成笔。笔的成立有严格的约束条件，否则你会看到行情里到处都是笔，完全没法用。我整理了一下，核心条件有这么几个：
- 必须先处理K线包含关系，用处理后的K线序列来判断分型；
- 相邻的笔必须方向交替，上升笔后面只能接下降笔，不能出现连续同向的笔；
- 一个底分型和一个顶分型之间必须隔着一根以上的独立K线，也就是说顶底分型不能共用K线；
- 同向分型之间取更极端的那个——比如相邻两个底分型，只保留最低价更低的那个，避免在同一个震荡区域反复出现一堆底。
我见过很多人在程序化的时候容易忽略最后一条，结果就是笔的数量爆炸，一段下跌中途出现七八个底分型，每两个都连一笔，看起来杂乱无章。正确做法是在确认笔之前，先把这些同向的分型过滤掉，只保留最“猛”的那一个。
2. 程序化的数据结构怎么设计
代码写起来之前，先把数据组织好。K线的基础字段无非就这几个：时间、开盘价、最高价、最低价、收盘价，再加上成交量。我用一个字典来装单根K线，然后把所有K线放到一个列表里，按时间升序排列。
2.1 K线的组织方式
每一根K线我习惯这样表示:
kline = { "time": "2025-01-06 09:30", "open": 10.0, "high": 10.6, "low": 9.9, "close": 10.4 }整个行情序列就是一个列表，列表的先后顺序代表时间顺序。有的朋友可能习惯用pandas的DataFrame来做，当然也可以，尤其是当你有大量历史数据、需要做批量计算的时候，pandas会很方便。但如果你是在做实时行情的增量处理，比如每分钟来一根新K线，那么用Python列表加字典这种原始结构反而更灵活，因为新数据来了直接append就行，处理逻辑上也更容易控制。
这里我建议把“K线索引”这个概念固定下来。分型和笔的判断，本质上都是在K线列表的下标之间做比较。比如顶分型出现在索引3，底分型出现在索引10，那么这两个位置之间的距离就是10减3等于7根K线。用索引来管理会让你后面处理“间隔多少根K线”这类条件时非常简单。
2.2 离线脚本和实时追加的区别
程序化缠论运行场景分两种。一种是离线分析：拉取一段历史行情，跑一遍全量数据，输出所有分型和笔，然后画图看效果。这种场景下逻辑顺序是线性的，从第一根K线处理到最后一根，比较简单。
另一种是实时追加：系统每收到一根新K线，就要更新一次分型和笔的状态。比如策略在盘中运行，每分钟都会进来一个新的K线数据点。这时候你要意识到一个重要事实：新增一根K线，可能会让之前靠近尾部的分型失效，甚至串联起一根新的笔。所以实时处理不能从头到尾重算所有历史数据（那样效率太低），更合理的是维护一个滑动窗口，只需要关心K线列表尾部那几根的状态变化即可。
我一般维护一个至少10根K线的缓冲区，每次新K线进来后，重新跑一遍包含关系处理和分型识别，因为这个范围足够覆盖分型确认和笔的更新所需的信息。
3. 包含关系处理，这一关绕不过去
包含关系是缠论里最劝退新手的概念，但程序化时必须处理。如果不处理包含，直接拿原始K线找分型，你会得到大量错误信号。
3.1 什么是包含
所谓包含，就是相邻两根K线当中，某一根的最高价和最低价完全覆盖了另一根的最高价和最低价。简单说就是一根K线的活动范围完全包含了另一根K线的活动范围，像俄罗斯套娃一样。
出现包含关系的原因很多，比如某个时间周期内价格波动很小，或者某根K线数据比较极端。在缠论的视角里，这种K线对趋势判断没有增量信息，应该把它“合并”掉，而不是当作独立的形态特征来看待。合并完以后再去看分型，形态才干净。
处理包含关系有个硬性前提：合并方向不能自己随便选，必须遵循前面的趋势方向。如果当前整体处于向上走势，就按“取高高”的方式合并——两根K线合并后，高点取两者较高的，低点也取两者较高的；如果处于向下走势，就反过来“取低低”——高点取两者较低的，低点也取两者较低的。
3.2 合并的方向和代码实现
怎么判断当前整体是向上还是向下？一个比较通用的做法是看前面那根“没有包含关系”的K线方向。比如当前处理到第i根K线，发现它和第i-1根存在包含，那我需要回看第i-2根和第i-1根的关系：如果第i-1根的最高价高于第i-2根的最高价，说明此前方向是向上的；反之则是向下。
代码实现上，我写了一个处理包含关系的函数，用循环对整个K线序列扫一遍：
def has_contain(k1, k2): """判断两根K线是否存在包含关系""" return (k1["high"] >= k2["high"] and k1["low"] <= k2["low"]) or \ (k2["high"] >= k1["high"] and k2["low"] <= k1["low"]) def merge_kline(k1, k2, direction): """按方向合并两根K线，方向: up 或 down""" new_k = {} # 向上合并：取高高 if direction == "up": new_k["time"] = k2["time"] new_k["high"] = max(k1["high"], k2["high"]) new_k["low"] = max(k1["low"], k2["low"]) # 向下合并：取低低 else: new_k["time"] = k2["time"] new_k["high"] = min(k1["high"], k2["high"]) new_k["low"] = min(k1["low"], k2["low"]) # open/close 在分型判断中不参与计算，可以省略或保留其一 new_k["open"] = new_k["close"] = (k1["close"] + k2["close"]) / 2 return new_k有了这两个基础函数，就可以写包含处理的主函数了。这里有个坑要特别提醒：合并后的新K线有可能继续和后面的K线存在包含，所以处理的时候要用while循环反复判断，而不是简单地for一遍了事。
def process_contains(klines): """处理整个K线序列的包含关系，返回新列表""" if not klines: return [] processed = [klines[0]] for i in range(1, len(klines)): curr = klines[i] # 反复检查，直到不再包含 while processed: last = processed[-1] if not has_contain(last, curr): break # 确定方向：回看 processed 中倒数第二根 if len(processed) >= 2: prev = processed[-2] direction = "up" if last["high"] > prev["high"] else "down" else: # 只有一根时，默认向上，也可以根据需求调整 direction = "up" curr = merge_kline(last, curr, direction) processed.pop() processed.append(curr) return processed这个版本会把包含后的K线不断往回合并，直到不再和前面的K线包含为止。这个代码我在一些简单的行情段上验证过，逻辑是符合缠论原始定义的常见解读的。
这里额外提醒一句：合并后的新K线，time字段我取了后一根的时间，是因为在实时场景中我们希望保留最新的时间标签。如果你做离线复现，也可以保留合并前第一根的time，对分型和笔的判断结果没有影响。
4. 分型识别的实现细节
包含关系处理完之后，就可以放心去找分型了。分型的判断逻辑非常直接，就是遍历K线序列，对每一根K线看它左右两边各一根，按顶底分型的定义去匹配。
4.1 分型判断逻辑
我写了两个函数，分别判断一个位置是不是顶分型或底分型：
def is_top(klines, idx): """判断第 idx 根K线是否为顶分型""" if idx <= 0 or idx >= len(klines) - 1: return False mid = klines[idx] left = klines[idx - 1] right = klines[idx + 1] return (mid["high"] > left["high"] and mid["high"] > right["high"] and mid["low"] > left["low"] and mid["low"] > right["low"]) def is_bottom(klines, idx): """判断第 idx 根K线是否为底分型""" if idx <= 0 or idx >= len(klines) - 1: return False mid = klines[idx] left = klines[idx - 1] right = klines[idx + 1] return (mid["low"] < left["low"] and mid["low"] < right["low"] and mid["high"] < left["high"] and mid["high"] < right["high"])循环遍历所有K线，把顶分型和底分型的索引分别记录下来：
def find_fractals(klines): tops = [] bottoms = [] for i in range(1, len(klines) - 1): if is_top(klines, i): tops.append(i) if is_bottom(klines, i): bottoms.append(i) return tops, bottoms看到这个逻辑，你可能会想：那如果一根K线同时被判定为顶分型和底分型怎么办？理论上是有可能的，比如极端情况下中间K线高点最高但低点也是三根中最低，这种同时符合顶底分型的K线通常说明行情在原地剧烈震荡，我建议直接忽略，不把它作为有效分型。后续在笔的确认阶段，这种模糊点也不会对最终结果造成实质影响。
4.2 分型识别的边界情况
分型判断中有几个边界情况容易被忽略。第一个是K线序列太少的时候，首尾两根K线永远不可能成为分型，因为分型需要左右两侧各有一根K线做参照。第二个是相邻分型共用K线的问题，比如一个顶分型和一个底分型中间只隔了一根K线，这种情况下顶底分型实际上是共享了中间的K线，从笔的定义上看是不成立的，需要在后续笔确认时过滤掉。
还有一个容易被忽视的细节：当新的K线出现之后，之前看到的“顶分型”或“底分型”可能被推翻。比如一个顶分型刚出现，结果右侧突然来了一根更高高点的K线，那这里就不再是顶分型了。所以程序上处理分型时，不要一开始就当作确定信号，应该等右侧K线足够多之后再确认。在实时系统里，这就是滞后性的来源之一，但缠论本身就是右侧确认的逻辑，这是无法避免的，习惯就好。
5. 笔的确立：从分型到笔的完整逻辑
有了分型列表，离最终画出笔还差两步：过滤同向分型、确认顶底分型之间是否满足独立K线数量要求。
5.1 笔成立的基本条件
缠论对笔的成立要求比较严格，如果顶分型和底分型之间隔得太近，它们就不能构成一笔。经过包含处理之后，我实现时的判断标准是：顶分型中间K线索引和底分型中间K线索引之差的绝对值必须大于等于4。为什么是4？因为分型本身至少需要3根K线，顶底分型又不能共用K线，之间还至少要留一根独立K线，折算下来索引差最小就是4。
举个例子，顶分型中间K线在索引5，底分型中间K线在索引9，那么索引差是4，中间有索引6、7、8三根K线，结构上满足顶底分型不相邻、中间有独立K线的要求，这才算一笔。如果索引差是3，那顶分型的右侧K线和底分型的左侧K线就是同一根，不符合笔的定义。
同时，同向分型必须过滤，否则一个震荡区间里会出现多个同向的信号。我说的过滤是指：如果已经记录了一个底分型，后续又出现了一个新底分型，那就保留更低点的那一个；如果是顶分型，则保留更高点的那一个。这样做的目的是让笔尽量贴近实际走势的极值点。
5.2 顶底分型交替与笔的确认
我封装了一个find_bi函数，它接收经过包含处理的K线列表和分型索引列表，按时间顺序遍历分型，动态维护当前笔的顶点：
def find_bi(klines, tops, bottoms): """ 从分型列表生成笔 tops: 顶分型索引列表 bottoms: 底分型索引列表 返回: [(笔类型, 起点索引, 终点索引), ...] """ # 将分型按索引合并成一个有序列表，用 pos 表示类型（顶/底） fractals = [("top", i, "top") for i in tops] + [("bottom", i, "bottom") for i in bottoms] fractals.sort(key=lambda x: x[1]) bi = [] # 最终笔序列，元素为 (方向, 起点idx, 终点idx) pending = None # 正在等待确认的候选分型 for ftype, idx, _ in fractals: if not pending: pending = (ftype, idx) continue last_type, last_idx = pending # 同向分型：取更极端的 if ftype == last_type: if ftype == "top" and klines[idx]["high"] >= klines[last_idx]["high"]: pending = (ftype, idx) elif ftype == "bottom" and klines[idx]["low"] <= klines[last_idx]["low"]: pending = (ftype, idx) continue # 不同向分型：检查索引间隔是否满足笔的条件 if abs(idx - last_idx) >= 4: direction = "up" if last_type == "bottom" else "down" bi.append((direction, last_idx, idx)) pending = (ftype, idx) else: # 间隔不足，保留原 pending，继续等待 continue return bi这个函数比我预想的要健壮，对于已跑通的行情数据，它能把笔的方向和起止索引完整输出。拿到这些索引之后，我再回到原始K线列表里取对应K线的时间、价格，用于画图或输出交易信号。
当然，这只是一个相对基础的实现版本。如果你在深入研究缠论，可能还会接触到“新笔”“旧笔”的补充定义，也就是对K线数量要求做一些放宽或收紧。我个人建议先把标准定义实现清楚，再去研究变体，不然很容易把自己绕晕。
6. 可视化验证与完整示例
程序写完了，光输出一堆索引数字是没法直观判断对错的，所以画图是必须的。把K线画出来，再把分型点和笔连线叠加在图上，一眼就能看出程序有没有识别错。
6.1 用matplotlib画出K线和笔
画K线的库有很多，比如mplfinance、plotly、pyecharts。我这里用最基础的matplotlib画一个简单版本：每根K线用一条竖线表示最高价到最低价的区间，再用一个小横线标出收盘价方向，顶分型标红点，底分型标绿点，笔用蓝色折线连接。
import matplotlib.pyplot as plt def plot_bi(klines, bi): plt.figure(figsize=(14, 6)) # 画K线简化图 for i, k in enumerate(klines): plt.vlines(i, k["low"], k["high"], color="black", linewidth=1) plt.plot(i, k["close"], marker="_", color="black") # 画笔 for direction, start, end in bi: x = [start, end] y = [klines[start]["high" if direction == "down" else "low"], klines[end]["low" if direction == "down" else "high"]] if direction == "up": plt.plot(x, y, color="red", linewidth=2) else: plt.plot(x, y, color="green", linewidth=2) plt.title("Chan Theory Bi") plt.xlabel("K-line index") plt.ylabel("Price") plt.show()严格来说，笔的连线应该连接顶分型和底分型中间K线的最高价或最低价，我上面这样取是简化处理。如果要做精确复盘，可以改成取顶分型中间K线的最高价（顶）和底分型中间K线的最低价（底）。
6.2 完整代码串联演示
下面我把整个流程串起来：
def run(klines): # 1. 处理包含关系 processed = process_contains(klines) # 2. 找分型 tops, bottoms = find_fractals(processed) # 3. 生成笔 bi = find_bi(processed, tops, bottoms) # 4. 输出结果 print("顶分型索引:", tops) print("底分型索引:", bottoms) print("笔:", bi) # 5. 画图 plot_bi(processed, bi) return processed, tops, bottoms, bi if __name__ == "__main__": # 测试数据，按时间升序排列 test_klines = [ {"time": "09:30", "high": 10.2, "low": 9.8, "open": 10.0, "close": 10.1}, {"time": "09:31", "high": 10.5, "low": 10.0, "open": 10.1, "close": 10.4}, # ... 这里放更多K线数据 ] run(test_klines)这套代码我建议你自己拿真实行情试一下，对比行情软件上手工画的笔，基本能对齐。如果对不上，大概率是包含关系处理的细节有差异，比如方向判断的基准、合并规则的先后顺序。这些都是缠论定义说得很模糊的地方，不同实现方式会带来细微区别，但只要你的规则是一致的、可解释的，那就够用。
7. 实操中的几个坑和排查技巧
程序化缠论听着高大上，真正跑起来之后问题一个接一个。我把自己卡过壳的地方整理成速查表，省得你重复踩坑。
7.1 输入数据必须按时间升序
这是我遇到最多的坑。很多免费数据接口返回的数据顺序并不稳定，有的是从旧到新，有的是从新到旧。而分型和笔的判断都是基于“当前K线之后还有新K线”这个时间假设，如果你把倒序数据传进去，识别结果会完全错乱。解决办法就是进入流程之前先做一次排序。千万不要以为数据是导出来的就一定是升序，我吃过这个亏。
7.2 包含合并是递归过程，不是一次性操作
很多初学者写包含关系处理，两层循环判断A和B有包含就合并，然后继续下一根，结果发现合并后的K线明明还和后面的C有包含，却没有继续处理，于是分型判断全部偏差。我自己第一版代码就栽在这上面。记住一个原则：每一次合并之后，新的K线要从当前位置“倒退一步”，和它前一根继续比较，直到不再触发包含为止。前面给的while processed循环就是干这个事的。
7.3 分型会被后续K线覆盖，别急着确认
实时行情里，一根K线走完，你感觉顶分型已经成立了，结果下一根K线直接高开向上，把这个顶分型的关键条件破坏了。所以我在实际系统里不会在分型刚出现时就发信号，而是多等待一根到两根K线，等右侧条件稳定以后再做确认。这个过程会增加一点滞后，但能过滤掉大量假信号。
7.4 同向分型过滤规则直接影响笔的数量
如果不过滤同向分型，行情稍微震荡一下，系统就会连续画出好几笔方向相同的笔，这在定义上是不允许的，也让后期判断中枢变得非常混乱。实现的时候，每次遇到新的同向分型都要比较一下极值，只有新分型比旧分型更极端时，才替换掉旧的候选分型。这个逻辑看起来简单，但作用很大，能直接影响整个笔序列的质量。
7.5 不同级别下的参数差异
分型和笔的识别对K线周期本身并不敏感，它不管是1分钟线还是日线，算法逻辑都一样。但周期不同，实际效果差异巨大。日线级别上可能一个月才画出十几笔，1分钟级别一根K线就会有多次分型信号。所以盘中的时候，我一般会把笔的最小K线间隔条件适当调严，减少噪声；日线级别的策略则不用做调整，按标准定义跑就好。
7.6 调试时用人工标注好的数据做回放
最后一个小技巧。程序写完以后，不要一上来就接真实行情全量跑，那样出了问题很难定位。我是先手工选了一段历史行情样本，把分型和笔人工标注了一个标准答案，然后让程序去跑，拿输出结果和人工标注做对比。这一步看起来多花了点时间，实际上帮你省下后面大量调试的精力。尤其是包含关系处理的方向逻辑，用这种方式排查最快。
从我自己的实际体验来说，把缠论里的分型和笔实现成Python代码，最大的收获其实不是那个面向结果的策略信号，而是逼着你把很多模糊的描述整理成可以验证的规则。你会发现很多看似高深的概念，真正落成代码之后，就是一个索引比较和条件过滤的组合，没有那么多玄学。后面如果你要继续做线段、中枢或者背驰，这套分型和笔的代码就是最底层的地基，把地基打牢，往上的逻辑再复杂都有信心往下写。
