缠论这套技术分析体系这些年讨论热度一直不低很多人一开始都是被“分型、笔、线段、中枢”这些概念给唬住了感觉门槛很高。但真要说程序化落地第一步其实没有想象中那么玄乎。把分型和笔的定义搞清楚用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(keylambda 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], colorblack, linewidth1) plt.plot(i, k[close], marker_, colorblack) # 画笔 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, colorred, linewidth2) else: plt.plot(x, y, colorgreen, linewidth2) 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代码最大的收获其实不是那个面向结果的策略信号而是逼着你把很多模糊的描述整理成可以验证的规则。你会发现很多看似高深的概念真正落成代码之后就是一个索引比较和条件过滤的组合没有那么多玄学。后面如果你要继续做线段、中枢或者背驰这套分型和笔的代码就是最底层的地基把地基打牢往上的逻辑再复杂都有信心往下写。