缠论程序化实战：从K线合并到买卖点信号的全链路实现

简介：这是一份面向股票量化与缠论研究者的Python程序化实践项目，以《缠中说禅博客》中的交易方法为蓝本，实现了K线包含处理、顶底分型识别、画笔划分等核心流程，并在TODO中规划线段划分、均线选股、均线轮动与板块强弱指标、每日走势分类（30分钟8根K线）等进阶方向。项目共33个文件，压缩包约25KB，属于轻量级代码包；主体为21个Python脚本，按主程序、缠论核心模块、内外部依赖包、测试用例与特征处理分层组织，便于对照算法逻辑逐段阅读。另含少量XML工程配置、TXT说明与许可证/说明文档，并配有环境依赖清单，方便直接搭建运行环境。目前已有898人学习下载，适合具备一定Python基础、希望把缠论形态判断落地的技术分析爱好者。通过阅读源码并运行测试，读者既能理解从K线合并到笔划分的完整实现思路，也能基于现有框架继续补充选股、板块轮动等策略，是进入缠论量化领域的一份紧凑参考。

1. 缠论程序化到底在做一件什么事：先分清哪些是规则，哪些是感觉

“缠论”在量化圈口碑很两极：一边说它是玄学，千人千缠，同一个位置有人标顶分型有人标笔，谁也说服不了谁；另一边确实有人把分型、笔、中枢、背驰做成程序，在商品和指数上跑了好几年，靠它躲掉一批假突破。ChZhShCh 这个名字听起来神秘，但拆开看就是部分缠论技术的程序化实践——不是要把整套理论搬进代码，而是把其中有明确几何定义的环节（分型、笔、线段、中枢、背驰、买卖点）固化成确定性规则，把“千人千缠”压缩成一份无法扯皮的执行清单。这篇笔记按我自己搭过的一版实践路径来写：从K线数据到买卖点信号要过哪几关、每关怎么选型、边界在哪，以及新手最容易被坑的五个位置。适合已经会用 Python 做数据处理、但还没把缠论落成代码的从业者。

2. 先定义干净的数据结构：K线模型与包含关系合并

2.1 数据源与字段组织：程序化缠论的第一步不是写算法

数据决定了后面所有判断。我一般直接使用日线或 30 分钟作为最小操作周期，用 pandas 按时间戳组织 K 线，字段固定为 ts、open、high、low、close、volume 六项。缠论的顶底分型、笔、中枢全部依赖 high 和 low，这两个字段的数值质量要优先保证——这不只是数据源要可靠，还指复权方式要统一。回测和实盘我都统一用后复权数据。前复权价格会随每次除权除息把历史价格整体平移，同一段走势在不同交易日读取时高低点数值不同，分型和笔在端点接近时会翻转顺序，回测出的信号第二天就对不上号。后复权历史价格稳定，实盘最新价与后复权的换算也简单。

停牌和一字板要提前处理。停牌日直接删除，不参与分型判断；一字板当天 high=low=close，这类 K 线在包含合并里会把方向带偏，我习惯先打一个 is_limit 标记，宁可跳过也不让它参与分型。数据清洗逻辑不复杂，关键是要在进入算法前把脏数据挡在门外：

import pandas as pd def load_kline(path): df = pd.read_csv(path, parse_dates=['ts']) df = df.sort_values('ts').drop_duplicates(subset='ts', keep='last') # 指数没有 amount 时用 volume 代替，这里只保留标准 OHLCV df = df[['ts', 'open', 'high', 'low', 'close', 'volume']].reset_index(drop=True) # 一字板标记，high 与 low 相等或差值极小 df['is_limit'] = (df['high'] - df['low']) < 1e-8 return df

这段逻辑有两个参数值得说明。排序后去重要保留最后一个，因为部分数据源在相同时间戳下会重复推送上一根 K 线，保留 last 才是最新状态；is_limit 用 high 减 low 的几何判断，而不用涨跌幅，因为 ST 股和普通股的一字板幅度不同，几何上的一字 K 线才对分型有真实干扰。

2.2 包含关系合并：所有分型判断的地基

缠论里的 K 线包含关系，指相邻两根 K 线中一根的高低点完全被另一根包住。遇到包含必须先合并，否则顶底分型会大量误判。合并方向不由当前两根 K 线决定，而由前一根 K 线给出的方向决定：上升趋势中取高点较高、低点也较高的一根作为结果；下降趋势中取高点较低、低点也较低的一根。规则初看反直觉，写成代码后却很直接：

def merge_klines(klines): """顺序合并存在包含关系的K线，返回合并后的K线序列""" merged = [klines[0]] for k in klines[1:]: last = merged[-1] # 一字板不参与合并，直接追加 if k.get('is_limit') or last.get('is_limit'): merged.append(k) continue # 判断两根K线是否构成包含关系 contained = (last['high'] <= k['high'] and last['low'] >= k['low']) or \ (k['high'] <= last['high'] and k['low'] >= last['low']) if not contained: merged.append(k) continue # 方向由合并序列中前一根K线相对更前一根的方向决定 if len(merged) >= 2: prev2 = merged[-2] direction = 'up' if prev2['high'] < last['high'] else 'down' else: direction = 'up' # 序列开头没有参照时默认向上 if direction == 'up': high = max(last['high'], k['high']) low = max(last['low'], k['low']) else: high = min(last['high'], k['high']) low = min(last['low'], k['low']) merged[-1] = { 'ts': k['ts'], # 合并K线用后一根的收盘时间 'open': last['open'], # open 取前一根，close 取后一根 'close': k['close'], 'high': high, 'low': low, 'volume': last['volume'] + k['volume'], } return merged

最容易被忽略的一点是：合并必须顺序迭代，不能先把所有包含关系找出来再一次性合并。两根 K 线合并后产生的新 high/low 可能与下一根再次形成包含，必须逐根推到序列末尾。方向判定的参照也必须是合并序列里的前一根，而不是原始 K 线的前一根，否则前一次合并的结果会被下一次判断丢掉。

合并后 K 线的时间戳用后一根的收盘时间，open 取前一根、close 取后一根，这样后续画笔时笔的内部结构仍保留一个可追踪的语义。volume 做加法，为后面按笔统计成交量备数据。这个合并函数是整个链路里最容易写歪的部分，它一旦出错，后面分型、笔、中枢全部跟着错，而且很难排查。

2.3 合并序列与原始K线的映射：信号落地必须能追溯到真实K线

程序把分型、笔、中枢算出来以后，最终仍要落到交易决策：某日出现三买，下单依据的是哪一根真实 K 线？如果直接拿合并后序列做，会发现合并 K 线的时间戳和原始 K 线对不上。我的做法是每一根合并 K 线都记录 origin_range，即它由原始序列的第几根到第几根合并而来。后续所有信号只记录 index 和 ts，回测下单时再映射回原始 K 线，以原始 K 线收盘价作为成交基准。

实现上就是在合并循环里维护一个区间列表，与 merged 序列等长，每个元素保存 (min_idx, max_idx)。这个区间在信号确认时价值很大：某个底分型在合并序列的 index=12，对应原始 K 线 10 到 15 根，那么真正可用的确认信号要等原始第 16 根收盘后落地，天然规避了未来函数。这个设计细节直接影响第 5 章要讲的回测真实性问题，建议在写代码的第一天就把它带上，后面补会很痛苦。

3. 分型到线段：把核心算法链一次跑通

3.1 缠论顶底分型识别：三根K线一个分型，条件要写全

分型是缠论里最底层的几何单元，只需看合并序列里连续三根 K 线。顶分型要求中间一根高点最高、低点也最高；底分型要求中间一根低点最低、高点也最低。条件是“且”的关系，不能只比较高点或只比较低点，否则大量弱分型会混进笔的候选里。

def find_fractals(merged): """在合并K线序列上找顶底分型""" fractals = [] for i in range(1, len(merged) - 1): p, c, n = merged[i - 1], merged[i], merged[i + 1] is_top = (c['high'] > p['high'] and c['high'] > n['high'] and c['low'] > p['low'] and c['low'] > n['low']) is_bottom = (c['high'] < p['high'] and c['high'] < n['high'] and c['low'] < p['low'] and c['low'] < n['low']) if is_top: fractals.append({'idx': i, 'type': 'top', 'price': c['high'], 'ts': c['ts']}) elif is_bottom: fractals.append({'idx': i, 'type': 'bottom', 'price': c['low'], 'ts': c['ts']}) return fractals

这里有个细节：同一个 idx 理论上不可能既是顶又是底，所以用 if-elif。如果出现同一根 K 线被左右两侧分别判断成顶和底，说明合并序列没做干净，反推回去八成是包含合并的方向判定有误。分型识别是整条链路上最不应该“创新”的地方，标准定义怎么写，代码就怎么写。序列首尾的分型无法识别，因为缺一侧参照，所以循环必须跳过 range 两端，这也意味着最新一笔的形成天然滞后一根 K 线，是缠论几何的固有成本。

3.2 笔的串联：顶底交替、间隔过滤与新旧笔分歧

分型只是散落的点，笔是把相邻的顶底分型连成一段走势。笔的规则有严格版和宽松版，程序化必须二选一锁死。老笔要求顶底分型之间至少隔一根完整独立 K 线；新笔允许共用一根 K 线。实操中两者信号数量和形态差异很大，我用日线级别锁老笔，30 分钟级别锁新笔。小级别 K 线密度高，严格规则会把笔拉得过长，很多有效转折被过滤；日线级别本身 K 线少，宽松规则会产生大量噪音笔。

def build_bis(fractals, style='strict'): """ 把分型串联成笔。 style: strict(老笔) / loose(新笔) """ bis = [] prev = None for f in fractals: if prev is None: prev = f continue # 同类型分型，保留更极端的一个 if f['type'] == prev['type']: if (f['type'] == 'top' and f['price'] >= prev['price']) or \ (f['type'] == 'bottom' and f['price'] <= prev['price']): prev = f continue gap = f['idx'] - prev['idx'] if style == 'strict': ok = gap >= 3 else: ok = gap >= 2 if ok: bis.append({'start': prev, 'end': f}) prev = f return bis

同类型分型保留更极端的一个是这段代码里最容易被漏掉的逻辑。没有这一步，两个连续的顶分型会生成一条向上笔再立刻生成一条向下一笔，走势被切碎。保留极端值等于在做转折点清洗，这一步不做干净，后面中枢数量会多到没法看。gap 阈值也有讲究：老笔要求分型之间隔独立 K 线，两个分型的 idx 差要达到 3；新笔允许共边，idx 差 2 即可。这两个数字不是拍脑袋，是按分型占三根 K 线的几何事实推出来的。gap 再放大，笔数量减少但转折变钝，背驰判断整体后移。

3.3 线段划分的简化实践：不做理论级递归，只做滤波器

线段是缠论里公认最模糊的部分。理论上的线段划分要处理特征序列、缺口、线段破坏，定义存在多种解读。我不建议在程序里实现完整递归，不同解释得出的结果差异比笔还大，且极难回测。我一般做一层简化：把笔序列当作新的“K 线”，对同方向的连续笔做合并过滤，形成线段级转折点。

def build_segments(bis): """ 把笔合并成线段：同方向连续笔视为同一段， 只保留该段中幅度最大的一笔作为线段端点。 """ segs = [] current_dir = None max_bi = None for bi in bis: direction = 'up' if bi['end']['type'] == 'top' else 'down' if current_dir is None: current_dir = direction max_bi = bi continue if direction == current_dir: # 同方向，保留幅度更大的笔 if abs(bi['end']['price'] - bi['start']['price']) > \ abs(max_bi['end']['price'] - max_bi['start']['price']): max_bi = bi else: segs.append(max_bi) current_dir = direction max_bi = bi if max_bi is not None: segs.append(max_bi) return segs

这段代码不是理论意义的线段，而是折返过滤器，但作为中枢识别的输入已经够用。中枢的构成单元是次级别的三段走势，用笔做单元时，三段笔的重叠区间已能描述中枢；这里把同向连续小笔合并，让中枢不因为一两根 K 线的小摆动频繁出现。做这个简化时心里要有数：可以不实现线段全部理论细节，但要在代码注释里写清楚“当前版本用笔直接构造中枢”，避免后续维护的人误以为程序实现了完整线段定义。程序化缠论最怕的不是算法简单，而是实现口径与声称口径不一致。

4. 中枢与背驰：程序如何理解走势结构

4.1 中枢识别：三笔重叠区间就是中枢

中枢的几何定义是至少三个连续次级别走势类型的重叠部分。在笔的粒度下，最常用做法是取连续三笔的高低点，中枢下沿 ZD 取三笔低点中的最大值，中枢上沿 ZG 取三笔高点中的最小值，ZG 高于 ZD 就构成中枢。进入中枢识别前，先给每根笔补两个辅助字段：笔的起点终点中更高的一个作为 full_high，更低的一个作为 full_low。下面实现先用端点价格演示主逻辑：

def find_centers(bis): """在笔序列上扫描中枢，返回中枢区间列表""" centers = [] for i in range(len(bis) - 2): b1, b2, b3 = bis[i], bis[i + 1], bis[i + 2] highs = [ max(b1['start']['price'], b1['end']['price']), max(b2['start']['price'], b2['end']['price']), max(b3['start']['price'], b3['end']['price']), ] lows = [ min(b1['start']['price'], b1['end']['price']), min(b2['start']['price'], b2['end']['price']), min(b3['start']['price'], b3['end']['price']), ] zd = max(lows) zg = min(highs) if zg > zd: centers.append({ 'start_idx': i, 'end_idx': i + 2, 'zd': zd, 'zg': zg, 'gg': max(highs), 'dd': min(lows), }) return centers

用端点价格算出的中枢对三买三卖这类边界信号影响不小，因为跳空或长影线会让端点价格漏掉真实极值。更精细的做法是每根笔携带 full_high 和 full_low，即笔跨越的原始 K 线范围内最高最低价，然后用它们计算 ZD 和 ZG。经验是：端点价格用于背驰力度对比时差别不显著，用于中枢边界信号时会漏掉一些有效三买三卖。初期版本先用端点简化，等回测链路稳定后升级到 full 范围，不用一上来就追求最精确。

中枢延伸与破坏也要程序化：某笔离开中枢区间（突破 ZG 或跌破 ZD）后，下一笔回抽不回中枢，中枢进入破坏确认；如果回抽回到中枢内，只是延伸。实现方式是找到中枢后向后扫描后续笔，记录第一次离开并回抽不回中枢的位置，作为中枢结束点。这个结束点是背驰对比里“离开段”的起点。

4.2 背驰判断：用MACD面积对比进入段与离开段

背驰的本质是走势力度衰竭。程序里最常见的替代指标是 MACD 柱面积：进入中枢的那一段与离开中枢的那一段，对比同方向柱子的面积总和，离开段面积明显小于进入段则判定背驰。注意只累加与段方向一致的柱，上升段用红柱，下降段用绿柱绝对值。

def segment_macd_area(df, start_ts, end_ts, macd_col='hist'): """计算一段走势内与方向一致的MACD柱面积""" seg = df[(df['ts'] >= start_ts) & (df['ts'] <= end_ts)] if seg.empty: return 0.0 vals = seg[macd_col] direction = 1 if seg['close'].iloc[-1] >= seg['close'].iloc[0] else -1 if direction > 0: area = vals[vals > 0].sum() else: area = abs(vals[vals < 0].sum()) return area

用这个函数时有一个典型错误：把整段 K 线的柱面积全部累加，正负抵消后面积趋近于 0，然后得出“没有背驰”的结论。必须按方向只累加同向柱。方向判定也可以用笔的方向（向上笔或向下笔），而不看段首段尾收盘价，因为笔的方向已经由分型类型确定了，更稳定。

MACD 参数默认 12/26/9，在 30 分钟和日线级别基本可用。但如果标的波动快，比如某些商品期货主力合约，12/26/9 在 5 分钟级别会频繁钝化。参数不急于求最优，先把 12/26/9 固定跑通流程，后面的参数敏感性扫描会告诉你这组参数是不是处在稳定平台内。

4.3 三类买卖点的程序化判定：把文本规则翻译成条件

第一、二、三类买卖点在程序里是独立判定函数，需要前面全链路的结果作为输入。判定顺序是：一买是下降趋势中出现背驰，背驰段结束后第一个底分型出现；二买是一买后第一段向上笔结束，回抽向下笔低点不低于一买低点；三买是向上突破中枢后回抽不回到 ZG 上方。这里给出一买函数的骨架：

def detect_class1_buy(centers, bis, df): """在最后一个中枢之后找一类买点：向下离开段背驰""" if len(centers) < 1 or len(bis) < 4: return None last_c = centers[-1] # 离开段：中枢结束后最后一笔向下笔 leave_bi = None for bi in reversed(bis): if bi['start']['type'] == 'top' and bi['end']['type'] == 'bottom': if bi['start']['idx'] >= last_c['end_idx']: leave_bi = bi break if leave_bi is None: return None # 进入段：中枢形成前最后一笔向下笔 enter_bi = None for bi in bis: if bi['end']['idx'] <= last_c['start_idx']: if bi['start']['type'] == 'top' and bi['end']['type'] == 'bottom': enter_bi = bi if enter_bi is None: return None area_enter = segment_macd_area(df, enter_bi['start']['ts'], enter_bi['end']['ts']) area_leave = segment_macd_area(df, leave_bi['start']['ts'], leave_bi['end']['ts']) # 面积缩小20%以上才算背驰 if area_leave < area_enter * 0.8: return { 'ts': leave_bi['end']['ts'], 'price': leave_bi['end']['price'], 'type': 'buy_1' } return None

关键参数是 0.8 的背驰阈值。理论上面积缩小就算背驰，但程序里一根异常 K 线就会导致面积异常，所以我用 0.8 做安全垫，商品和指数上都保持在 0.8 到 0.85 之间。二买三买实现类似，核心是把“回抽不破前低”“回抽不回中枢”写成端点价格比较。

到这里，缠论里能被量化的部分基本落地。至于那些依赖人工语境的部分，我会明确排除在策略代码外，强行程序化只会得到一堆无法解释的参数。

5. 避坑与排查：实战里绕不开的5个翻车点

5.1 包含合并没按顺序迭代，分型数量直接翻倍

第一次跑出分型列表时，看到的顶分型数量比手标多近一倍。排查后发现合并逻辑写成了“先找出所有包含对，再一次性合并”，这违反了缠论处理包含的递推原则：合并后的新 K 线又会和相邻 K 线形成新的包含，必须逐根顺序处理。解决方式就是第 2 章的 merge_klines 函数，merged 列表不断更新末尾元素参与后续判断。判断是否写对有个土办法：把合并后序列画出来，对比原始走势，包含处应该是“一根吞掉一根”的阶梯形态，而不是锯齿。代码评审时我要求每条包含合并必须能在图上对应到一段连续 K 线，否则一律打回。

5.2 前复权数据导致信号在第二天“漂移”

回测出了三买信号，第二天复盘时同一段代码在同样的位置不再出信号。这个现象不是随机噪声，而是数据口径问题。用前复权数据时，标的除权除息后历史所有 K 线价格会整体重算，同一位置的端点价格出现微小位移，分型和笔的顺序在价格接近时会翻转。解决方式：回测统一用后复权。实盘端取行情时显式指定复权参数，历史部分全部按后复权重算，保证与回测基准一致。另一个连带坑是仓位计算：复权价不是真实成交价，算手数必须用原始价，否则仓位系数也会漂。

5.3 新笔老笔规则没锁死，中枢数量差一倍

同一个标的，老笔跑出 3 个中枢，新笔变成 6 个。这不是 bug，是定义差异。老笔要求分型之间隔独立 K 线，过滤大量小转折；新笔允许共边，小折返也会成笔，中枢必然增多。问题在于不少人会写一个“笔灵敏度”参数，在不同行情里来回切换找信号，这本质是过拟合。我的做法是写进配置文件：日线级别固定老笔，30 分钟固定新笔，任何行情都不改。宁可接受某段行情信号少，也不去反复调整定义。笔规则一旦锁死，中枢、背驰、买卖点才具备可对比性。

5.4 MACD面积对比里正负抵消，背驰全被算没

背驰函数第一次跑完，整段行情一个背驰都没有。排查后发现面积累加用了 sum()，红柱绿柱互相抵消，下跌段的绿柱面积被红柱吃掉近一半，对比自然不成立。改成按方向分别累加后，背驰信号数量立刻正常。这个错误很隐蔽，因为程序不报错，只输出一个“永远为假”的判断结果。建议在背驰函数里加一个防御性检查：当同向柱面积累计为 0 时抛警告，而不是静默返回 0。同时把分型、笔、K 线的 ts 范围打印出来，人工对照几个已知背驰点，确认面积计算区间没有丢 K 线。

5.5 确认信号的未来函数：顶分型当天不能下单

缠论分型要三根 K 线确认，第三根 K 线收盘那一刻分型才成立。如果程序在第三根 K 线内部就发信号，等于用了当根 K 线的最高最低做未来数据。回测引擎里这会造成“点位极准”的假象，实盘完全复现不了。解决方式：信号统一挂在分型确认后下一根 K 线开盘执行。我在回测代码里专门实现一段延迟下单逻辑，所有信号 ts 都要经过 confirm_shift，默认延迟 1 根 K 线。这个延迟会让胜率看起来下降一点，但那是真实可执行的胜率。回测日志里同时打印 signal_ts 和 exec_ts 两列，检查两者是否相差一根以上，是排查未来函数的最快办法。

这五个坑里，前三个是数据与定义层面的地基问题，后两个是回测层面的隐蔽问题。把它们写进代码评审清单，比任何一轮调参都管用。

6. 让信号经得起验证：参数敏感性扫描与最后一公里

整套链路跑通后，下一步是回答一个现实问题：这些参数换到别的标的上还灵不灵。我最常用的验证方法是固定标的，做参数邻域扫描，不是找最优值，而是看是否存在一个“稳定平台”。把背驰比例阈值从 0.6 扫到 0.95，间隔 0.05，同时把 MACD 快线从 5 扫到 20，得到一张类似下面的二维表（数值仅示意，用于说明判定逻辑）：

理想状态是中间一片都稳定在 2.5 以上，而不是只在某一个格子里爆出 4.0。只在一个格子爆表，基本可判定是参数组合恰好贴合了这段行情的噪声，换一段就失效。我通常取平台区域内的中间值，而不是峰值。第二个验证方法是一致性回放：参数全部固定，在数据末尾逐日追加最新 K 线，重新算信号序列，对比信号是否稳定出现在同一位置。每天在模拟环境跑一次，坚持两三周，比任何单次回测都能暴露复权漂移和未来函数。

最后说实盘衔接。信号出来后，成交价按什么价最关键。我的回测设定是信号次日开盘价成交，但实盘开盘价有滑点和集合竞价不确定性，所以加了一道“开盘 30 秒后确认条件仍满足”再下单的过滤。这不是缠论规则，是执行层防腐。

回头看，缠论程序化的价值不在预测走势，而在把一套讲究级别嵌套的体系降维成可执行规则，约束你在该等的时候等、该动的时候动。我在这套系统上踩过最深的坑是调参上瘾：连续两天扫参数，觉得每个波动都能优化，结果换一段行情全部还回去。后来给自己立了规矩，参数一年只复核一次，平时只处理数据和执行层问题，实盘反而稳定很多。希望帮到你。

本文还有配套的精品资源，点击获取