简介本资源是一套基于Python与Backtrader框架实现的缠论量化交易系统源码面向具备Python基础与量化交易兴趣的开发者、金融工程学习者及策略研究员旨在解决缠论规则工程化落地难、手动分析效率低、策略回测缺乏标准化框架等实际问题。压缩包共56个文件85.43MB包含19个核心Python源文件含策略逻辑、分型/笔/中枢识别模块、15个pyc编译文件体现可部署性、7个XML配置文件用于Backtrader策略参数与数据源定义、4个CSV历史行情数据支撑多标的回测、3个INI环境配置适配本地运行以及测试脚本、数据库工具、OSS/Redis集成模块等整体结构完整已覆盖数据预处理、缠论层级解析、买卖信号生成与结果评估全流程。目前已有447人学习下载代码组织清晰、模块职责分明附带Jupyter测试用例与单元测试集便于理解缠论各层级分型→笔→线段→中枢的程序化判定逻辑并支持快速复现、调试与策略迭代。 做缠论交易系统最难的不是学缠论而是把“顶分型、底分型、笔、线段、中枢”这些概念翻译成代码。市面上能看到的缠论源码绝大多数是通达信、同花顺里的指标公式画线、出信号都没问题但想用它跑历史回测、验证策略逻辑、接实盘基本使不上劲。我后来决定用 Python Backtrader 重新设计一套缠论交易系统源码前后重构了三版才把“理论到代码”这条路走通。这篇文章就把我这套系统的设计思路、源码结构和踩过的坑完整写出来重点解决三个问题缠论概念怎么用代码表达、怎么把缠论信号接入 Backtrader 回测框架、以及回测结果为什么经常“看着很赚实盘就亏”。适合准备用 Python 做量化、对缠论感兴趣但不知道从何下手的读者也适合已经跑通基础回测、想深入做结构类策略的朋友参考。1. 为什么最终选了 Backtrader以及框架选型时的几个坑1.1 缠论系统对回测框架的要求缠论策略和普通的均线、MACD 类策略有个本质区别它的交易信号不是根据某一个指标数值直接触发而是依赖 K 线的几何结构。这意味着回测框架必须支持“逐根 K 线状态更新”因为一笔是否成立、一个分型是否被破坏需要在下一根 K 线才能确认。这种逻辑用向量化框架会写得非常别扭因为向量化框架默认所有数据一次性算出很难表达“等待确认”的状态。另外缠论经常需要在不同级别之间切换比如日线级别判断方向30 分钟级别找买卖点同时也需要叠加成交量、MACD 面积背驰这类辅助判断。所以框架得支持多数据源、多周期还要方便自定义指标。最后一点是可视化缠论非常依赖图形说实话如果画不出笔和中枢你很难判断代码算出来的结果是不是对的。综合这几点Backtrader 是当时最适合的选择。它是事件驱动框架每一根 K 线都会调用策略的 next 方法天然适合表达状态机逻辑自定义 Indicator 的机制也很成熟还内置了 Analyzer可以直接输出收益率、回撤、夏普比率等指标。1.2 Backtrader 的取舍以及我自己补的几个轮子Backtrader 不是没有缺点。先说兼容性这个框架已经很久没有大版本更新了Python 3.12 以上的环境安装容易出问题我实测下来 3.9 到 3.11 是最稳的建议别追新。另外它对 pandas 的版本也比较敏感用 bt.feeds.PandasData 加载数据时如果 pandas 版本太新某些列名解析会报错可以用 numpy 数组或 CSV 数据源绕开。性能也是个问题。Backtrader 单标的逐根 K 线跑 10 年日线是没问题的但如果要做多标的、多周期、分钟级别的全市场回测速度就不够看了。我的做法是把它定位成“研究阶段的验证工具”实盘信号直接另写一套轻量级引擎不在 Backtrader 里做实时交易。如果一上来就想做全市场分钟级扫描顺便说一句那更适合用 vectorbt 这类向量化方案但要付出“很难表达缠论状态机”的代价。我自己补的三个轮子自定义数据类把日线、30 分钟线统一封装解决复权和停牌数据对齐问题一个持续维护的缠论计算模块把分型、包含处理、笔、线段、中枢的计算全部独立出来不依赖 Backtrader这样同一套算法可以在回测和实盘共用绘图组件在 Backtrader 自带绘图基础上叠加笔和中枢的绘制标记。2. 缠论概念代码化分型、包含处理、笔与中枢的数据结构2.1 顶底分型的判定缠论里最基础的单元是分型。顶分型指的是连续三根 K 线中中间那根的最高价最高、最低价也最高底分型则反过来中间那根的最高价最低、最低价也最低。用代码表达很直接def is_top_fractal(high, low, i): 判断第 i 根 K 线是否为顶分型 if i 1 or i len(high) - 1: return False return (high[i] high[i-1] and high[i] high[i1] and low[i] low[i-1] and low[i] low[i1]) def is_bottom_fractal(high, low, i): 判断第 i 根 K 线是否为底分型 if i 1 or i len(high) - 1: return False return (high[i] high[i-1] and high[i] high[i1] and low[i] low[i-1] and low[i] low[i1])这个逻辑本身没错但有个前提它要求这三根 K 线是经过包含处理之后的。否则在真实行情里相邻 K 线经常出现一根完全“包含”在另一根里面不先合并就直接找分型会产生大量假信号。2.2 K 线包含关系的合并逻辑包含关系指的是相邻两根 K 线一根的最高价和最低价都在另一根的范围之内。处理原则是向上趋势中合并时取两根 K 线的最高价较高者、最低价也较高者也就是“高高”向下趋势中则反过来取“低低”。判断当前趋势方向最简单的办法是看前一根合并后的 K 线与再前一根的位置关系。如果当前合并后的最高价比前一根高说明方向向上反之向下。代码如下def merge_klines(df): 对 K 线序列做包含关系合并返回合并后的 DataFrame merged [] direction 0 # 1 向上, -1 向下 for i in range(len(df)): if not merged: merged.append(df.iloc[i].copy()) continue last merged[-1] cur df.iloc[i] is_contain (cur[high] last[high] and cur[low] last[low]) or \ (cur[high] last[high] and cur[low] last[low]) if is_contain: if direction 1: merged[-1][high] max(last[high], cur[high]) merged[-1][low] max(last[low], cur[low]) elif direction -1: merged[-1][high] min(last[high], cur[high]) merged[-1][low] min(last[low], cur[low]) else: # 初始方向未定时按当前包含关系先合并 merged[-1][high] max(last[high], cur[high]) merged[-1][low] min(last[low], cur[low]) else: direction 1 if cur[high] last[high] else -1 merged.append(cur.copy()) return merged这段代码在实盘时有个细节合并时新合并出来的 K 线的时间戳、成交量怎么处理我建议时间戳保留后一根的成交量直接相加因为包含关系代表多空双方在这段区间内反复争夺成交量合并更接近真实状态。2.3 笔、线段与中枢的分层表示处理完包含关系之后就可以开始找笔。笔的定义很简单相邻的顶分型和底分型之间经过包含处理后至少要有 5 根 K 线也就是顶底分型之间至少隔 1 根独立的 K 线并且顶底必须交替出现不能连着两个顶分型构成一笔。线段至少由连续三笔组成而且这三笔要有重叠区间。中枢则是至少三个连续次级别走势类型的重叠区间。在实际代码实现里我更推荐把“笔”作为核心计算对象先用一个类把笔的特征提取出来class Bi: 一笔的完整记录 def __init__(self, start_idx, end_idx, direction, start_price, end_price): self.start_idx start_idx self.end_idx end_idx self.direction direction # up or down self.start_price start_price self.end_price end_price self.high max(start_price, end_price) self.low min(start_price, end_price) # 笔内部的 K 线数量用于判断是否满足最少 K 线约束 self.klines end_idx - start_idx 1写到这里插一句网上很多缠论源码喜欢把分型、笔、线段全部揉在一起画出来的图挺好看但代码根本没法维护。我的做法是分层计算分型层只负责标记顶底笔层在分型层之上线段层在笔层之上中枢依赖线段或笔的重叠区间。每一层独立测试哪一层出问题直接定位。3. 笔识别状态机顶底交替、最少 K 线与未成笔处理3.1 笔的最小 K 线约束关于“一笔至少要多少根 K 线”不同缠论流派有不同说法有的是处理包含后 5 根有的是原始 K 线 5 根。我在源码里做成参数默认按处理包含后不少于 5 根来算。这个参数会直接影响笔的数量太少了会频繁触发买卖信号太多了信号又会滞后建议在回测里多试几组值。3.2 识别过程的状态机笔的识别在代码上是一个典型状态机当前没有持仓笔时遇到分型就新建候选笔已经有一个候选分型时遇到同类型分型要比较极值保留更极端的那一个遇到反向分型时检查 K 线数量是否满足满足就确认一笔不满足就继续等待。下面是我在系统里用的核心逻辑def find_bis(fractals): fractals: 按时间排序的分型列表每个元素是 {index: int, type: top/bottom, price: float} 返回笔列表 bis [] last None for f in fractals: if last is None: last f continue # 同类型分型保留更极端的 if f[type] last[type]: if f[type] top and f[price] last[price]: last f elif f[type] bottom and f[price] last[price]: last f continue # 反向分型检查 K 线数量满足最少 K 线约束才确认笔 kline_count f[index] - last[index] 1 if kline_count min_klines: bis.append(Bi(last[index], f[index], up if f[type] top else down, last[price], f[price])) last f # 不满足最少 K 线约束时不确认保留原分型继续等待 # 但此时要考虑当前反向后如果后面又走了很远原分型是否还有效 # 这里简化处理实际系统里需要引入“破坏”判断 return bis3.3 同类分型的极端值替换规则这里有个非常容易写错的点顶分型出现之后如果后面又出现一个更高的顶分型那么前一笔的终点就要更新到这个更高的顶分型位置。反之如果顶分型后面出现一个更低的顶分型前一笔终点不更新但这个更低的顶分型可能变成下一笔的起点。这个逻辑没有写对画出来的笔在回放行情时会“跳变”而且回测结果会严重失真。我的建议是把“分型是否被破坏”作为一个独立方法抽出来def is_fractal_broken(f, next_f): 判断分型是否被反向突破破坏 if f[type] top: return next_f[type] bottom and next_f[price] f[price] else: return next_f[type] top and next_f[price] f[price]这里“被破坏”的判断不能只盯分型的价位还要看是不是有一笔级别的反向走势把它破坏了。实际行情里经常出现顶分型出现后价格小幅回踩又创新高这时不能贸然确认上一笔结束。这个判断属于缠论里最典型的主观性部分代码能做的是用参数控制确认灵敏度比如“反向突破超过多少个百分点才算破坏”而不是试图用一套规则覆盖所有情况。4. 缠论指标与策略信号如何接入 Backtrader4.1 自定义 Indicator 的写法Backtrader 的 Indicator 封装了数据线的计算逻辑可以在 addindicator 之后自动画图也能在 Strategy 里直接访问。我实现了一个简单的分型 Indicatorimport backtrader as bt class ChanFractal(bt.Indicator): 缠论分型指标 输出top 和 bottom 两条线 顶分型时 top 标记分型高点底分型时 bottom 标记分型低点 lines (top, bottom) params ((min_klines, 5),) plotinfo dict(plotTrue, subplotFalse) plotlines dict( topdict(markerv, markersize8, colorred, fillstylefull, ls), bottomdict(marker^, markersize8, colorgreen, fillstylefull, ls), ) def __init__(self): self.addminperiod(3) def next(self): if len(self) 3: return # 注意这里用 len(self) 拿当前 bar 的下标-2 才是“当前正在判断的 K 线” h self.data.high.get(size3) l self.data.low.get(size3) if h[1] h[0] and h[1] h[2] and l[1] l[0] and l[1] l[2]: self.lines.top[0] h[1] elif h[1] h[0] and h[1] h[2] and l[1] l[0] and l[1] l[2]: self.lines.bottom[0] l[1]这里有个关键点Indicator 的 next 是按最新 K 线计算的上述代码中当前最新 bar 的下标是 0真正要判断的分型是前一根 bar下标 -1 或 -2取决于你如何对齐。这个细节正是未来函数的源头之一如果不注意你可能用“当前还没走完的 K 线”去确认分型回测结果等于预知未来。第 5 章我会单独展开。4.2 在 Strategy 中做买卖点识别分型 Indicator 只解决基础结构识别真正的买卖点判断我放在 Strategy 里做因为买卖点需要结合笔、中枢、背驰等多个条件全部塞进 Indicator 会导致类太臃肿。我在 Strategy 的 next 里做的工作有几件拿到当前已确认的笔和分型判断是否处于新的买点或卖点检查是否存在未完成订单然后下单。一个简化示例class ChanStrategy(bt.Strategy): params ( (size, 100), (stop_loss_pct, 0.05), ) def __init__(self): self.chan_fractal ChanFractal(self.data) self.order None def next(self): if self.order: return if not self.position: # 底分型确认且之前是下跌笔视为第一类买点候选 if self.chan_fractal.bottom[-1]: self.buy(sizeself.params.size) else: # 持仓中顶分型确认且已有盈利先止盈 if self.chan_fractal.top[-1]: self.close()这里 bottom[-1] 表示用上一根 K 线确认的底分型避开用当前未完成 K 线做判断的问题。4.3 三种接入方式的取舍我在实际设计过程中试过三种把缠论计算接进 Backtrader 的方式各有适用场景方式优点缺点适用场景自定义 Indicator与框架耦合紧密自动绘图行情变动实时更新状态复杂时代码难维护需要注意未来函数研究阶段快速验证结构信号预计算后注入数据线计算和回测完全分离便于单元测试需要自己维护数据同步、对齐像多周期回测时会麻烦实盘信号引擎与回测共用同一套数据在 Strategy 里直接算灵活信号逻辑一目了然每根 K 线都重复计算大量历史结构性能差简单策略、快速验证我的最终选择是“方式二为主、方式一为辅”缠论计算模块独立成一个 Python 包先把所有笔、线段、中枢算好存成结构化数据然后写一个适配层把这些数据转换成 Backtrader 的 Line 或直接作为一个数据源传进 cerebroIndicator 只负责画图不承担核心计算。5. 信号确认与未来函数回测里最容易被忽略的偏差5.1 未来函数在缠论代码中的两个典型来源缠论交易系统特别容易在未来函数上翻车。第一个来源是“分型未确认就交易”。一个顶分型至少需要三根 K 线才能确定如果你用包含当前 bar 在内的三根 K 线去判断分型那么当前 bar 还没走完你可能已经知道了这个 bar 的最高价和最低价这在回测里等于提前看到了未来。第二个来源是“笔/线段的动态修正”。缠论里一笔在形成过程中经常因为后续 K 线变化而被重新划分比如原来是向上笔后面又出了更高的顶分型那上一笔的终点会被往后挪。如果在策略回测时用最终修正后的笔去回溯当时的买点就会产生一种“事后诸葛亮”式的收益这种回测结果拿到实盘上一点参考意义都没有。5.2 用下一根 K 线执行修复回测偏差修复未来函数的基本原则就一句话回测时只能用“截至当前 K 线收盘已经确认的信息”并且下单要在“下一根 K 线”。在 Backtrader 里默认市价单就是在下一根 bar 开盘成交的所以只要保证信号判断不引用未来数据下单延迟的问题框架已经替你解决了。需要注意的反而是在 Indicator 里别用“当前 bar 未来 bar”去确认信号。比如 2.1 节那段分型判断我把i当作“待判断的 K 线”后面循环到i1时才能确认i是不是分型。在 Strategy 里就体现为用bottom[-1]而不是bottom[0]。这个“滞后一根”的习惯是缠论回测里最值得养成的。5.3 自查清单与回测结果口径我每次写完一个缠论策略回测前都会做三件事随机信号测试把买卖信号随机打乱跑一遍看最大回撤和收益。如果随机信号都能赚钱那一定有问题要么手续费没扣要么未来函数混进去了。逐笔回放随机抽几笔交易手动打开 K 线图核对买点是否在信号出现后才发生。这个步骤比较费时间但能抓出“看似合理但顺序错乱”的信号。结果口径确认明确回测费用——佣金、滑点、印花税都要扣。缠论是趋势策略胜率普遍不高靠盈亏比赚钱如果费用漏了整个回测结果会虚高到离谱。6. 源码目录结构与核心代码展开6.1 模块划分如果只是想跑通一个 demo把所有代码塞一个文件里没问题。但一旦开始调参数、换标的、对比不同级别就会立刻感受到模块化的价值。我目前的目录结构长这样chan_trading/ ├── data/ # 行情数据加载、清洗、复权处理 ├── chan/ │ ├── fractal.py # 顶底分型识别 │ ├── merge.py # K 线包含处理 │ ├── bi.py # 笔的识别与状态机 │ ├── segment.py # 线段划分 │ ├── zhongshu.py # 中枢识别 │ └── model.py # Bi, Segment, Zhongshu 等数据结构 ├── indicators/ │ └── chan_ind.py # Backtrader 自定义指标 ├── strategies/ │ └── chan_strategy.py # 交易策略 ├── backtest.py # 回测入口 └── config.py # 参数配置6.2 核心代码分型识别分型识别模块我做得比较纯粹只负责一件事给定一个已经合并处理的 K 线序列输出所有分型位置。它不关心 Backtrader不关心交易只接受 numpy 数组或 pandas DataFrame。# chan/fractal.py from dataclasses import dataclass from typing import List dataclass class Fractal: index: int ftype: str # top or bottom price: float def find_fractals(closes, highs, lows) - List[Fractal]: 基于已合并 K 线找顶底分型 fractals [] for i in range(1, len(closes) - 1): if highs[i] highs[i-1] and highs[i] highs[i1] and \ lows[i] lows[i-1] and lows[i] lows[i1]: fractals.append(Fractal(i, top, highs[i])) elif highs[i] highs[i-1] and highs[i] highs[i1] and \ lows[i] lows[i-1] and lows[i] lows[i1]: fractals.append(Fractal(i, bottom, lows[i])) return fractals注意这里的输入是“已经合并处理的 K 线”如果直接用原始 K 线分型会非常密集一笔被切成很多段信号数量直线上升但质量很差。6.3 核心代码笔划分笔划分模块要解决的问题是在分型列表中找到真正构成笔的分型对。前面第 3 章已经给了状态机框架这里补充一个更完整的版本把“同类型分型替换”和“最少 K 线约束”都放进去# chan/bi.py from .fractal import Fractal from .model import Bi def find_bis(fractals: List[Fractal], min_klines: int 5) - List[Bi]: 从分型列表划分笔 bis [] pending None for f in fractals: if pending is None: pending f continue # 同类型保留更极端 if f.ftype pending.ftype: if f.ftype top and f.price pending.price: pending f elif f.ftype bottom and f.price pending.price: pending f continue # 反向分型检查距离 if f.index - pending.index min_klines: direction up if f.ftype top else down bis.append(Bi(pending.index, f.index, direction, pending.price, f.price)) pending f else: # 距离不够继续等待 # 这里可以加入“破坏”判断如果反向分型已经破坏了原方向则替换 pending pass return bis这个版本能跑但缺少一种情况的处理pening 分型被反向分型“破坏”之后原方向已经失去了继续延伸的资格这时应该把 pending 更新为当前反向分型否则可能漏掉真正的翻转信号。完整版需要配合 is_fractal_broken 一起用我把它放到后续源码里这里不展开了。6.4 核心代码策略主逻辑与回测入口策略主逻辑我简化成一个可供参考的版本它的核心思路是只在向下笔的底分型确认后买入在持仓中出现向上笔的顶分型确认后卖出。这只是一个最基础的框架实际系统里还需要加入中枢过滤和背驰判断。# strategies/chan_strategy.py import backtrader as bt class ChanStrategy(bt.Strategy): params ( (size, 100), (printlog, False), ) def __init__(self): self.fractal self.indicators.chan_fractal def next(self): # 没有持仓寻找底分型买点 if not self.position: if self.fractal.bottom[-1]: self.buy(sizeself.params.size) # 有持仓寻找顶分型卖点 elif self.fractal.top[-1]: self.close() def notify_order(self, order): if order.status in [order.Completed, order.Canceled, order.Margin]: # 订单状态处理方便后续接入日志和风控 pass回测入口是所有参数的中枢我习惯把所有可调参数集中放这里# backtest.py import backtrader as bt import pandas as pd from strategies.chan_strategy import ChanStrategy from indicators.chan_ind import ChanFractal def run_backtest(df, cash100000, commission0.0003): cerebro bt.Cerebro() data bt.feeds.PandasData(datanamedf) cerebro.adddata(data) cerebro.addstrategy(ChanStrategy) cerebro.addindicator(ChanFractal) cerebro.broker.setcash(cash) cerebro.broker.setcommission(commissioncommission) cerebro.addanalyzer(bt.analyzers.SharpeRatio, _namesharpe) cerebro.addanalyzer(bt.analyzers.DrawDown, _namedrawdown) results cerebro.run() return results[0] if __name__ __main__: df pd.read_csv(data/000001.csv, index_col0, parse_datesTrue) result run_backtest(df) print(夏普比率:, result.analyzers.sharpe.get_analysis()) print(最大回撤:, result.analyzers.drawdown.get_analysis())7. 实盘前必须补的几个模块与个人体会7.1 数据层多周期、复权与数据对齐缠论结构对数据质量极其敏感。除权除息产生的跳空缺口会让一笔的划分出现完全不同的结果所以回测和实盘都必须用前复权数据。另外不同周期的数据如果要对齐我建议统一用时间戳做外连接不要直接按行号硬对齐否则盘中未收盘的 bar 会把结构计算搞乱。7.2 风控与仓位管理缠论本身只解决“在哪里买卖”不解决“买多少”和“错了怎么办”。我在这套系统里做了三件事单笔止损默认 5%不设止盈靠移动止损保护利润单标的仓位不超过总资金的 30%多标的情况下同一时间段最多同时持仓 3 个标的。这些参数不一定是最优解但能把回撤控制在一个能接受的范围。7.3 我实测中最大的教训第一次跑完整回测的时候某只票的收益率高得惊人年化超过 200%我当时还挺兴奋直到逐笔复盘才发现问题信号产生的那根 K 线最高价和最低价里混进了未来数据——分型是拿“包含当前未完成 bar”的数据确认的所以买点总比实际早一根 K 线。修正成“确认后下一根 K 线执行”之后年化直接掉到 40% 左右。这个落差让我意识到缠论代码实现真正的价值不在“把理论变成指标”而在“把理论变成可验证、可复盘的交易纪律”。如果你想在实盘里用这套东西我还要多提醒一句缠论把“结构”这个东西量化以后不等于找到了圣杯。那些号称“顶底信号 98% 胜率”的指标源码我劝你直接绕开——结构类策略在震荡行情里被反复打脸是常态代码能帮你做的是快速验证、强制纪律而不是让你躺着赚钱。把这套系统当成一个“把模糊的交易思想变成可检验规则”的工具用它去跑足够多的历史数据再一点点修正参数和规则这才是它真正的用法。本文还有配套的精品资源点击获取
简介本资源是一套基于Python与Backtrader框架实现的缠论量化交易系统源码面向具备Python基础与量化交易兴趣的开发者、金融工程学习者及策略研究员旨在解决缠论规则工程化落地难、手动分析效率低、策略回测缺乏标准化框架等实际问题。压缩包共56个文件85.43MB包含19个核心Python源文件含策略逻辑、分型/笔/中枢识别模块、15个pyc编译文件体现可部署性、7个XML配置文件用于Backtrader策略参数与数据源定义、4个CSV历史行情数据支撑多标的回测、3个INI环境配置适配本地运行以及测试脚本、数据库工具、OSS/Redis集成模块等整体结构完整已覆盖数据预处理、缠论层级解析、买卖信号生成与结果评估全流程。目前已有447人学习下载代码组织清晰、模块职责分明附带Jupyter测试用例与单元测试集便于理解缠论各层级分型→笔→线段→中枢的程序化判定逻辑并支持快速复现、调试与策略迭代。 做缠论交易系统最难的不是学缠论而是把“顶分型、底分型、笔、线段、中枢”这些概念翻译成代码。市面上能看到的缠论源码绝大多数是通达信、同花顺里的指标公式画线、出信号都没问题但想用它跑历史回测、验证策略逻辑、接实盘基本使不上劲。我后来决定用 Python Backtrader 重新设计一套缠论交易系统源码前后重构了三版才把“理论到代码”这条路走通。这篇文章就把我这套系统的设计思路、源码结构和踩过的坑完整写出来重点解决三个问题缠论概念怎么用代码表达、怎么把缠论信号接入 Backtrader 回测框架、以及回测结果为什么经常“看着很赚实盘就亏”。适合准备用 Python 做量化、对缠论感兴趣但不知道从何下手的读者也适合已经跑通基础回测、想深入做结构类策略的朋友参考。1. 为什么最终选了 Backtrader以及框架选型时的几个坑1.1 缠论系统对回测框架的要求缠论策略和普通的均线、MACD 类策略有个本质区别它的交易信号不是根据某一个指标数值直接触发而是依赖 K 线的几何结构。这意味着回测框架必须支持“逐根 K 线状态更新”因为一笔是否成立、一个分型是否被破坏需要在下一根 K 线才能确认。这种逻辑用向量化框架会写得非常别扭因为向量化框架默认所有数据一次性算出很难表达“等待确认”的状态。另外缠论经常需要在不同级别之间切换比如日线级别判断方向30 分钟级别找买卖点同时也需要叠加成交量、MACD 面积背驰这类辅助判断。所以框架得支持多数据源、多周期还要方便自定义指标。最后一点是可视化缠论非常依赖图形说实话如果画不出笔和中枢你很难判断代码算出来的结果是不是对的。综合这几点Backtrader 是当时最适合的选择。它是事件驱动框架每一根 K 线都会调用策略的 next 方法天然适合表达状态机逻辑自定义 Indicator 的机制也很成熟还内置了 Analyzer可以直接输出收益率、回撤、夏普比率等指标。1.2 Backtrader 的取舍以及我自己补的几个轮子Backtrader 不是没有缺点。先说兼容性这个框架已经很久没有大版本更新了Python 3.12 以上的环境安装容易出问题我实测下来 3.9 到 3.11 是最稳的建议别追新。另外它对 pandas 的版本也比较敏感用 bt.feeds.PandasData 加载数据时如果 pandas 版本太新某些列名解析会报错可以用 numpy 数组或 CSV 数据源绕开。性能也是个问题。Backtrader 单标的逐根 K 线跑 10 年日线是没问题的但如果要做多标的、多周期、分钟级别的全市场回测速度就不够看了。我的做法是把它定位成“研究阶段的验证工具”实盘信号直接另写一套轻量级引擎不在 Backtrader 里做实时交易。如果一上来就想做全市场分钟级扫描顺便说一句那更适合用 vectorbt 这类向量化方案但要付出“很难表达缠论状态机”的代价。我自己补的三个轮子自定义数据类把日线、30 分钟线统一封装解决复权和停牌数据对齐问题一个持续维护的缠论计算模块把分型、包含处理、笔、线段、中枢的计算全部独立出来不依赖 Backtrader这样同一套算法可以在回测和实盘共用绘图组件在 Backtrader 自带绘图基础上叠加笔和中枢的绘制标记。2. 缠论概念代码化分型、包含处理、笔与中枢的数据结构2.1 顶底分型的判定缠论里最基础的单元是分型。顶分型指的是连续三根 K 线中中间那根的最高价最高、最低价也最高底分型则反过来中间那根的最高价最低、最低价也最低。用代码表达很直接def is_top_fractal(high, low, i): 判断第 i 根 K 线是否为顶分型 if i 1 or i len(high) - 1: return False return (high[i] high[i-1] and high[i] high[i1] and low[i] low[i-1] and low[i] low[i1]) def is_bottom_fractal(high, low, i): 判断第 i 根 K 线是否为底分型 if i 1 or i len(high) - 1: return False return (high[i] high[i-1] and high[i] high[i1] and low[i] low[i-1] and low[i] low[i1])这个逻辑本身没错但有个前提它要求这三根 K 线是经过包含处理之后的。否则在真实行情里相邻 K 线经常出现一根完全“包含”在另一根里面不先合并就直接找分型会产生大量假信号。2.2 K 线包含关系的合并逻辑包含关系指的是相邻两根 K 线一根的最高价和最低价都在另一根的范围之内。处理原则是向上趋势中合并时取两根 K 线的最高价较高者、最低价也较高者也就是“高高”向下趋势中则反过来取“低低”。判断当前趋势方向最简单的办法是看前一根合并后的 K 线与再前一根的位置关系。如果当前合并后的最高价比前一根高说明方向向上反之向下。代码如下def merge_klines(df): 对 K 线序列做包含关系合并返回合并后的 DataFrame merged [] direction 0 # 1 向上, -1 向下 for i in range(len(df)): if not merged: merged.append(df.iloc[i].copy()) continue last merged[-1] cur df.iloc[i] is_contain (cur[high] last[high] and cur[low] last[low]) or \ (cur[high] last[high] and cur[low] last[low]) if is_contain: if direction 1: merged[-1][high] max(last[high], cur[high]) merged[-1][low] max(last[low], cur[low]) elif direction -1: merged[-1][high] min(last[high], cur[high]) merged[-1][low] min(last[low], cur[low]) else: # 初始方向未定时按当前包含关系先合并 merged[-1][high] max(last[high], cur[high]) merged[-1][low] min(last[low], cur[low]) else: direction 1 if cur[high] last[high] else -1 merged.append(cur.copy()) return merged这段代码在实盘时有个细节合并时新合并出来的 K 线的时间戳、成交量怎么处理我建议时间戳保留后一根的成交量直接相加因为包含关系代表多空双方在这段区间内反复争夺成交量合并更接近真实状态。2.3 笔、线段与中枢的分层表示处理完包含关系之后就可以开始找笔。笔的定义很简单相邻的顶分型和底分型之间经过包含处理后至少要有 5 根 K 线也就是顶底分型之间至少隔 1 根独立的 K 线并且顶底必须交替出现不能连着两个顶分型构成一笔。线段至少由连续三笔组成而且这三笔要有重叠区间。中枢则是至少三个连续次级别走势类型的重叠区间。在实际代码实现里我更推荐把“笔”作为核心计算对象先用一个类把笔的特征提取出来class Bi: 一笔的完整记录 def __init__(self, start_idx, end_idx, direction, start_price, end_price): self.start_idx start_idx self.end_idx end_idx self.direction direction # up or down self.start_price start_price self.end_price end_price self.high max(start_price, end_price) self.low min(start_price, end_price) # 笔内部的 K 线数量用于判断是否满足最少 K 线约束 self.klines end_idx - start_idx 1写到这里插一句网上很多缠论源码喜欢把分型、笔、线段全部揉在一起画出来的图挺好看但代码根本没法维护。我的做法是分层计算分型层只负责标记顶底笔层在分型层之上线段层在笔层之上中枢依赖线段或笔的重叠区间。每一层独立测试哪一层出问题直接定位。3. 笔识别状态机顶底交替、最少 K 线与未成笔处理3.1 笔的最小 K 线约束关于“一笔至少要多少根 K 线”不同缠论流派有不同说法有的是处理包含后 5 根有的是原始 K 线 5 根。我在源码里做成参数默认按处理包含后不少于 5 根来算。这个参数会直接影响笔的数量太少了会频繁触发买卖信号太多了信号又会滞后建议在回测里多试几组值。3.2 识别过程的状态机笔的识别在代码上是一个典型状态机当前没有持仓笔时遇到分型就新建候选笔已经有一个候选分型时遇到同类型分型要比较极值保留更极端的那一个遇到反向分型时检查 K 线数量是否满足满足就确认一笔不满足就继续等待。下面是我在系统里用的核心逻辑def find_bis(fractals): fractals: 按时间排序的分型列表每个元素是 {index: int, type: top/bottom, price: float} 返回笔列表 bis [] last None for f in fractals: if last is None: last f continue # 同类型分型保留更极端的 if f[type] last[type]: if f[type] top and f[price] last[price]: last f elif f[type] bottom and f[price] last[price]: last f continue # 反向分型检查 K 线数量满足最少 K 线约束才确认笔 kline_count f[index] - last[index] 1 if kline_count min_klines: bis.append(Bi(last[index], f[index], up if f[type] top else down, last[price], f[price])) last f # 不满足最少 K 线约束时不确认保留原分型继续等待 # 但此时要考虑当前反向后如果后面又走了很远原分型是否还有效 # 这里简化处理实际系统里需要引入“破坏”判断 return bis3.3 同类分型的极端值替换规则这里有个非常容易写错的点顶分型出现之后如果后面又出现一个更高的顶分型那么前一笔的终点就要更新到这个更高的顶分型位置。反之如果顶分型后面出现一个更低的顶分型前一笔终点不更新但这个更低的顶分型可能变成下一笔的起点。这个逻辑没有写对画出来的笔在回放行情时会“跳变”而且回测结果会严重失真。我的建议是把“分型是否被破坏”作为一个独立方法抽出来def is_fractal_broken(f, next_f): 判断分型是否被反向突破破坏 if f[type] top: return next_f[type] bottom and next_f[price] f[price] else: return next_f[type] top and next_f[price] f[price]这里“被破坏”的判断不能只盯分型的价位还要看是不是有一笔级别的反向走势把它破坏了。实际行情里经常出现顶分型出现后价格小幅回踩又创新高这时不能贸然确认上一笔结束。这个判断属于缠论里最典型的主观性部分代码能做的是用参数控制确认灵敏度比如“反向突破超过多少个百分点才算破坏”而不是试图用一套规则覆盖所有情况。4. 缠论指标与策略信号如何接入 Backtrader4.1 自定义 Indicator 的写法Backtrader 的 Indicator 封装了数据线的计算逻辑可以在 addindicator 之后自动画图也能在 Strategy 里直接访问。我实现了一个简单的分型 Indicatorimport backtrader as bt class ChanFractal(bt.Indicator): 缠论分型指标 输出top 和 bottom 两条线 顶分型时 top 标记分型高点底分型时 bottom 标记分型低点 lines (top, bottom) params ((min_klines, 5),) plotinfo dict(plotTrue, subplotFalse) plotlines dict( topdict(markerv, markersize8, colorred, fillstylefull, ls), bottomdict(marker^, markersize8, colorgreen, fillstylefull, ls), ) def __init__(self): self.addminperiod(3) def next(self): if len(self) 3: return # 注意这里用 len(self) 拿当前 bar 的下标-2 才是“当前正在判断的 K 线” h self.data.high.get(size3) l self.data.low.get(size3) if h[1] h[0] and h[1] h[2] and l[1] l[0] and l[1] l[2]: self.lines.top[0] h[1] elif h[1] h[0] and h[1] h[2] and l[1] l[0] and l[1] l[2]: self.lines.bottom[0] l[1]这里有个关键点Indicator 的 next 是按最新 K 线计算的上述代码中当前最新 bar 的下标是 0真正要判断的分型是前一根 bar下标 -1 或 -2取决于你如何对齐。这个细节正是未来函数的源头之一如果不注意你可能用“当前还没走完的 K 线”去确认分型回测结果等于预知未来。第 5 章我会单独展开。4.2 在 Strategy 中做买卖点识别分型 Indicator 只解决基础结构识别真正的买卖点判断我放在 Strategy 里做因为买卖点需要结合笔、中枢、背驰等多个条件全部塞进 Indicator 会导致类太臃肿。我在 Strategy 的 next 里做的工作有几件拿到当前已确认的笔和分型判断是否处于新的买点或卖点检查是否存在未完成订单然后下单。一个简化示例class ChanStrategy(bt.Strategy): params ( (size, 100), (stop_loss_pct, 0.05), ) def __init__(self): self.chan_fractal ChanFractal(self.data) self.order None def next(self): if self.order: return if not self.position: # 底分型确认且之前是下跌笔视为第一类买点候选 if self.chan_fractal.bottom[-1]: self.buy(sizeself.params.size) else: # 持仓中顶分型确认且已有盈利先止盈 if self.chan_fractal.top[-1]: self.close()这里 bottom[-1] 表示用上一根 K 线确认的底分型避开用当前未完成 K 线做判断的问题。4.3 三种接入方式的取舍我在实际设计过程中试过三种把缠论计算接进 Backtrader 的方式各有适用场景方式优点缺点适用场景自定义 Indicator与框架耦合紧密自动绘图行情变动实时更新状态复杂时代码难维护需要注意未来函数研究阶段快速验证结构信号预计算后注入数据线计算和回测完全分离便于单元测试需要自己维护数据同步、对齐像多周期回测时会麻烦实盘信号引擎与回测共用同一套数据在 Strategy 里直接算灵活信号逻辑一目了然每根 K 线都重复计算大量历史结构性能差简单策略、快速验证我的最终选择是“方式二为主、方式一为辅”缠论计算模块独立成一个 Python 包先把所有笔、线段、中枢算好存成结构化数据然后写一个适配层把这些数据转换成 Backtrader 的 Line 或直接作为一个数据源传进 cerebroIndicator 只负责画图不承担核心计算。5. 信号确认与未来函数回测里最容易被忽略的偏差5.1 未来函数在缠论代码中的两个典型来源缠论交易系统特别容易在未来函数上翻车。第一个来源是“分型未确认就交易”。一个顶分型至少需要三根 K 线才能确定如果你用包含当前 bar 在内的三根 K 线去判断分型那么当前 bar 还没走完你可能已经知道了这个 bar 的最高价和最低价这在回测里等于提前看到了未来。第二个来源是“笔/线段的动态修正”。缠论里一笔在形成过程中经常因为后续 K 线变化而被重新划分比如原来是向上笔后面又出了更高的顶分型那上一笔的终点会被往后挪。如果在策略回测时用最终修正后的笔去回溯当时的买点就会产生一种“事后诸葛亮”式的收益这种回测结果拿到实盘上一点参考意义都没有。5.2 用下一根 K 线执行修复回测偏差修复未来函数的基本原则就一句话回测时只能用“截至当前 K 线收盘已经确认的信息”并且下单要在“下一根 K 线”。在 Backtrader 里默认市价单就是在下一根 bar 开盘成交的所以只要保证信号判断不引用未来数据下单延迟的问题框架已经替你解决了。需要注意的反而是在 Indicator 里别用“当前 bar 未来 bar”去确认信号。比如 2.1 节那段分型判断我把i当作“待判断的 K 线”后面循环到i1时才能确认i是不是分型。在 Strategy 里就体现为用bottom[-1]而不是bottom[0]。这个“滞后一根”的习惯是缠论回测里最值得养成的。5.3 自查清单与回测结果口径我每次写完一个缠论策略回测前都会做三件事随机信号测试把买卖信号随机打乱跑一遍看最大回撤和收益。如果随机信号都能赚钱那一定有问题要么手续费没扣要么未来函数混进去了。逐笔回放随机抽几笔交易手动打开 K 线图核对买点是否在信号出现后才发生。这个步骤比较费时间但能抓出“看似合理但顺序错乱”的信号。结果口径确认明确回测费用——佣金、滑点、印花税都要扣。缠论是趋势策略胜率普遍不高靠盈亏比赚钱如果费用漏了整个回测结果会虚高到离谱。6. 源码目录结构与核心代码展开6.1 模块划分如果只是想跑通一个 demo把所有代码塞一个文件里没问题。但一旦开始调参数、换标的、对比不同级别就会立刻感受到模块化的价值。我目前的目录结构长这样chan_trading/ ├── data/ # 行情数据加载、清洗、复权处理 ├── chan/ │ ├── fractal.py # 顶底分型识别 │ ├── merge.py # K 线包含处理 │ ├── bi.py # 笔的识别与状态机 │ ├── segment.py # 线段划分 │ ├── zhongshu.py # 中枢识别 │ └── model.py # Bi, Segment, Zhongshu 等数据结构 ├── indicators/ │ └── chan_ind.py # Backtrader 自定义指标 ├── strategies/ │ └── chan_strategy.py # 交易策略 ├── backtest.py # 回测入口 └── config.py # 参数配置6.2 核心代码分型识别分型识别模块我做得比较纯粹只负责一件事给定一个已经合并处理的 K 线序列输出所有分型位置。它不关心 Backtrader不关心交易只接受 numpy 数组或 pandas DataFrame。# chan/fractal.py from dataclasses import dataclass from typing import List dataclass class Fractal: index: int ftype: str # top or bottom price: float def find_fractals(closes, highs, lows) - List[Fractal]: 基于已合并 K 线找顶底分型 fractals [] for i in range(1, len(closes) - 1): if highs[i] highs[i-1] and highs[i] highs[i1] and \ lows[i] lows[i-1] and lows[i] lows[i1]: fractals.append(Fractal(i, top, highs[i])) elif highs[i] highs[i-1] and highs[i] highs[i1] and \ lows[i] lows[i-1] and lows[i] lows[i1]: fractals.append(Fractal(i, bottom, lows[i])) return fractals注意这里的输入是“已经合并处理的 K 线”如果直接用原始 K 线分型会非常密集一笔被切成很多段信号数量直线上升但质量很差。6.3 核心代码笔划分笔划分模块要解决的问题是在分型列表中找到真正构成笔的分型对。前面第 3 章已经给了状态机框架这里补充一个更完整的版本把“同类型分型替换”和“最少 K 线约束”都放进去# chan/bi.py from .fractal import Fractal from .model import Bi def find_bis(fractals: List[Fractal], min_klines: int 5) - List[Bi]: 从分型列表划分笔 bis [] pending None for f in fractals: if pending is None: pending f continue # 同类型保留更极端 if f.ftype pending.ftype: if f.ftype top and f.price pending.price: pending f elif f.ftype bottom and f.price pending.price: pending f continue # 反向分型检查距离 if f.index - pending.index min_klines: direction up if f.ftype top else down bis.append(Bi(pending.index, f.index, direction, pending.price, f.price)) pending f else: # 距离不够继续等待 # 这里可以加入“破坏”判断如果反向分型已经破坏了原方向则替换 pending pass return bis这个版本能跑但缺少一种情况的处理pening 分型被反向分型“破坏”之后原方向已经失去了继续延伸的资格这时应该把 pending 更新为当前反向分型否则可能漏掉真正的翻转信号。完整版需要配合 is_fractal_broken 一起用我把它放到后续源码里这里不展开了。6.4 核心代码策略主逻辑与回测入口策略主逻辑我简化成一个可供参考的版本它的核心思路是只在向下笔的底分型确认后买入在持仓中出现向上笔的顶分型确认后卖出。这只是一个最基础的框架实际系统里还需要加入中枢过滤和背驰判断。# strategies/chan_strategy.py import backtrader as bt class ChanStrategy(bt.Strategy): params ( (size, 100), (printlog, False), ) def __init__(self): self.fractal self.indicators.chan_fractal def next(self): # 没有持仓寻找底分型买点 if not self.position: if self.fractal.bottom[-1]: self.buy(sizeself.params.size) # 有持仓寻找顶分型卖点 elif self.fractal.top[-1]: self.close() def notify_order(self, order): if order.status in [order.Completed, order.Canceled, order.Margin]: # 订单状态处理方便后续接入日志和风控 pass回测入口是所有参数的中枢我习惯把所有可调参数集中放这里# backtest.py import backtrader as bt import pandas as pd from strategies.chan_strategy import ChanStrategy from indicators.chan_ind import ChanFractal def run_backtest(df, cash100000, commission0.0003): cerebro bt.Cerebro() data bt.feeds.PandasData(datanamedf) cerebro.adddata(data) cerebro.addstrategy(ChanStrategy) cerebro.addindicator(ChanFractal) cerebro.broker.setcash(cash) cerebro.broker.setcommission(commissioncommission) cerebro.addanalyzer(bt.analyzers.SharpeRatio, _namesharpe) cerebro.addanalyzer(bt.analyzers.DrawDown, _namedrawdown) results cerebro.run() return results[0] if __name__ __main__: df pd.read_csv(data/000001.csv, index_col0, parse_datesTrue) result run_backtest(df) print(夏普比率:, result.analyzers.sharpe.get_analysis()) print(最大回撤:, result.analyzers.drawdown.get_analysis())7. 实盘前必须补的几个模块与个人体会7.1 数据层多周期、复权与数据对齐缠论结构对数据质量极其敏感。除权除息产生的跳空缺口会让一笔的划分出现完全不同的结果所以回测和实盘都必须用前复权数据。另外不同周期的数据如果要对齐我建议统一用时间戳做外连接不要直接按行号硬对齐否则盘中未收盘的 bar 会把结构计算搞乱。7.2 风控与仓位管理缠论本身只解决“在哪里买卖”不解决“买多少”和“错了怎么办”。我在这套系统里做了三件事单笔止损默认 5%不设止盈靠移动止损保护利润单标的仓位不超过总资金的 30%多标的情况下同一时间段最多同时持仓 3 个标的。这些参数不一定是最优解但能把回撤控制在一个能接受的范围。7.3 我实测中最大的教训第一次跑完整回测的时候某只票的收益率高得惊人年化超过 200%我当时还挺兴奋直到逐笔复盘才发现问题信号产生的那根 K 线最高价和最低价里混进了未来数据——分型是拿“包含当前未完成 bar”的数据确认的所以买点总比实际早一根 K 线。修正成“确认后下一根 K 线执行”之后年化直接掉到 40% 左右。这个落差让我意识到缠论代码实现真正的价值不在“把理论变成指标”而在“把理论变成可验证、可复盘的交易纪律”。如果你想在实盘里用这套东西我还要多提醒一句缠论把“结构”这个东西量化以后不等于找到了圣杯。那些号称“顶底信号 98% 胜率”的指标源码我劝你直接绕开——结构类策略在震荡行情里被反复打脸是常态代码能帮你做的是快速验证、强制纪律而不是让你躺着赚钱。把这套系统当成一个“把模糊的交易思想变成可检验规则”的工具用它去跑足够多的历史数据再一点点修正参数和规则这才是它真正的用法。本文还有配套的精品资源点击获取
