Python缠论程序化实战：K线分型与笔自动识别完整实现

做缠论程序化这件事最初纯粹是被手工复盘逼出来的。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_gap4): 根据分型列表构建笔 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_gap4): 缠论分型与笔的完整流程入口 klines: DataFrame, 列包含 high, low min_gap: 笔的最小间隔默认老笔4 返回: processed: 标准K线列表 fractals: 分型列表 bis: 笔列表 processed process_contain(klines) fractals find_fractals(processed) bis build_bi(fractals, min_gapmin_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_gap2) 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], colorcolor, linewidth1) plt.vlines(i, row[open], row[close], colorcolor, linewidth3) # 标记分型 for f in fractals: if f[type] top: plt.scatter(f[idx], f[price] 0.1, markerv, colorblack, zorder5) else: plt.scatter(f[idx], f[price] - 0.1, marker^, colorblue, zorder5) # 连接笔 for b in bis: xs [b[start][idx], b[end][idx]] ys [b[start][price], b[end][price]] plt.plot(xs, ys, colororange, linewidth1.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线变化而变化定期用全量数据重算比每天在增量数据上修修补补要可靠得多。这套流程跑顺之后复盘效率比手工不知道高了多少。