用Backtrader构建缠论量化交易系统：笔、中枢与买卖点的程序化实现

简介基于Python与Backtrader框架开发的缠论交易系统源码主要面向量化交易开发者和金融分析人员用于将缠论技术分析理论落地为可执行的自动化交易策略。缠论强调市场走势的自相似性通过分型、笔、线段、中枢等层级识别行情阶段源码将这些理论规则转化为可回测与实盘执行的程序化逻辑。资源共含56个文件压缩包约85.43MB其中19个Python源文件与15个编译文件承担策略实现和数据处理7个XML配置用于策略参数与接口设置4个CSV存储历史行情数据另有INI、YML等环境配置文件整体目录划分清晰覆盖数据读取、信号生成、交易接口等模块。已有448人学习下载适合希望结合经典理论与量化框架进行系统开发的读者。通过阅读源码可掌握缠论策略的工程化实现方式、Backtrader策略回测的完整流程并基于测试与环境配置快速展开二次开发是学习量化交易和策略设计的实用参考资料。1. 缠论交易系统为什么值得用 Backtrader 重写一遍做量化的人一听到“缠论”第一反应多半是“玄学”。但如果你把缠论当成一套几何化的走势分解规则它其实比大多数指标更容易程序化分型、笔、线段、中枢、买卖点每一步都是可以递归定义的形状识别问题。问题是市面上只有主图公式和画线工具画出来的中枢很直观却没一个能回答“这套规则历史胜率到底是多少”。这就是我把缠论写进 Backtrader 的原因——不是要证明缠论有效而是把缠论变成可回测、可统计、可复用的交易系统源码。本文面向想自己动手搭一套缠论量化框架的 Python 开发者聊清楚程序化缠论怎么建模、Backtrader 怎么接入、参数怎么调、以及哪些地方最容易翻车。2. 缠论不是黑匣子笔、线段、中枢与买卖点的程序化定义2.1 从 K 线到分型包含关系处理是第一道坎缠论的所有结构都建立在底分型和顶分型上。底分型的定义是三根 K 线中中间那根的最低点最低、最高点也最低顶分型反过来。但真实 K 线很少直接给你标准形状两根 K 线之间经常出现包含关系——一根 K 线的高低点完全落在另一根内部。程序化处理包含关系时要先合并再找分型而且合并方向取决于之前一根 K 线的走势方向。def merge_bars(bars): merged [bars[0]] direction 0 for bar in bars[1:]: pre merged[-1] if pre[high] bar[high] and pre[low] bar[low]: if direction 0: new_bar { high: pre[high], low: bar[low], time: bar[time] } else: new_bar { high: bar[high], low: pre[low], time: bar[time] } merged[-1] new_bar elif bar[high] pre[high] and bar[low] pre[low]: if direction 0: new_bar { high: bar[high], low: pre[low], time: bar[time] } else: new_bar { high: pre[high], low: bar[low], time: bar[time] } merged[-1] new_bar else: merged.append(bar) direction 1 if bar[high] pre[high] else -1 return merged这段代码里有一个非常容易写错的点合并不是单向的。上升趋势中的包含关系取高高也就是两根 K 线中的最高价和最高价组合下降趋势中取低低。如果不维护direction同一段行情你会算出完全不同的分型序列后续笔和中枢全部错位。另一个细节是合并后的 K 线时间戳应该保留最新一根的时间这样后续检查分型生效时间不会出现时间倒流。2.2 笔的划分顶底分型之间必须有独立 K 线分型只是原料笔才是缠论里最小的走势单位。一个笔的成立需要三个条件存在相邻的顶分型和底分型顶分型的高点必须高于底分型的高点且两个分型之间至少有一根不属于任何一侧分型的独立 K 线。网上的公式经常漏掉最后一条导致在一段横盘里反复画笔波动被切成碎片。def find_bi(fenxing_list): bis [] i 0 while i len(fenxing_list) - 1: left fenxing_list[i] right fenxing_list[i 1] if left[type] right[type]: if (left[type] top and left[value] right[value]): i 1 continue elif (left[type] bottom and left[value] right[value]): i 1 continue if bars_between(left[idx], right[idx]) 3: bis.append({ start: left, end: right, direction: up if right[value] left[value] else down }) i 1 return bisbars_between查的是两个分型原始索引之间有多少根未合并 K 线。这里要用合并前的索引用合并后的数量会严重失真。笔的方向由价格高低决定不是由分型类型决定。很多新手把“底分型到顶分型”直接当上涨笔但如果顶分型的值反而更低这根本不是笔。这个函数的输出是整个缠论系统里所有高层结构的基础宁可写得保守也不要把不是笔的连接当成笔。2.3 中枢与买卖点三段次级别走势的重叠区间缠论里的中枢定义是至少三个连续次级别走势类型重叠的区间。程序化实现时大部分源码会把中枢约化成“至少三笔的重叠区域”取前三笔的高低点交集作为中枢区间。def find_zhongshu(bis, start0): if len(bis) - start 3: return None b1, b2, b3 bis[start], bis[start 1], bis[start 2] zg min(b1[end][value], b2[end][value], b3[end][value]) zd max(b1[start][value], b2[start][value], b3[start][value]) if zg zd: return { zg: zg, zd: zd, start_idx: b1[start][idx], end_idx: b3[end][idx] } return None这个约化有两个明显的工程取舍。第一它把中枢的次级别结构完全忽略了只适合日线以上的大级别粗筛进入 30 分钟以下周期时噪声非常大。第二中枢是动态延伸的新笔形成后中枢区间会变化真正的源码里必须做循环延伸直到第三类买卖点出现上面的代码只是单次识别。我一般会把中枢维护做成一个状态机行情推进时先判断当前是否在中枢区间内再判断是否离开这是后文策略层能不能稳定买入卖出的关键。三段重叠的逻辑确定后一类买卖点出现在走势离开中枢后的回抽不破中枢区间时二类买卖点出现在次级别回抽确认时三类买卖点出现在中枢突破后的回抽不进中枢时。三个买点的程序化判断其实都依赖中枢端点zg和zd的实时更新你可以把这三类判断写成独立函数。这里只需要记住买卖点不是靠指标金叉死叉算出来的而是靠中枢区间被突破后的一次回抽确认所以行情软件里的“缠论买卖点主图公式”本质上是把这三类点画到图上真正可回测的判定必须依赖中框状态而不是视觉位置。3. 用 Backtrader 把最小策略跑起来数据加载、资金配置与参数设置3.1 为什么选 Backtrader 而不是自己写回测循环你当然可以自己写循环遍历 K 线、判断信号、记录持仓、计算收益跑一两次没问题但一旦要加多标的、换仓、手续费滑点、按时间分段统计自研循环很快变成一座屎山。Backtrader 的价值在于把行情数据、策略逻辑、交易执行、统计分析四层拆开你只需要关心策略里next函数内部怎么写其他都有现成组件。Python 环境准备好之后装好依赖就能开始搭骨架。pip install backtrader pandas numpy安装完成后最简回测骨架大概是下面这段。这个骨架不涉及任何缠论逻辑先验证工具链本身通不通再接信号。import backtrader as bt class ChanStrategy(bt.Strategy): def next(self): if not self.position: self.buy(size100) else: self.sell(size100) if __name__ __main__: cerebro bt.Cerebro() data bt.feeds.GenericCSVData( datanamesh000001.csv, dtformat%Y-%m-%d, datetime0, open1, high2, low3, close4, volume5, openinterest-1 ) cerebro.adddata(data) cerebro.broker.setcash(100000.0) cerebro.broker.setcommission(commission0.0003) cerebro.addstrategy(ChanStrategy) cerebro.run() print(final value: %.2f % cerebro.broker.getvalue())dtformat告诉 Backtrader 日期列的时间格式datetime0表示第 0 列是时间。commission0.0003是双边万三的手续费A 股环境下印花税和滑点后面再加。这段代码里最容易被忽略的参数是openinterest-1数据文件没有持仓量字段时一定要设成 -1否则 CSV 解析直接报错。跑通这个骨架后你等于拥有了一个可扩展的量化交易系统底座。接下来把缠论信号作为策略的一部分接进next流里。3.2 在 next 里拿什么数据指标线、缓存数组与订单状态Backtrader 的next每次调用时当前 K 线已经收盘你拿到的是完整的一根 bar。缠论结构识别至少需要几百根 K 线才能形成第一个笔和中枢所以不能在next里每次重新计算全量。正确的做法是维护一个缓存数组K 线逐根推入结构计算只在缓存尾部增量执行。class ChanStrategy(bt.Strategy): def __init__(self): self.kline_cache [] self.bis [] self.zhongshu None def next(self): bar { time: self.datas[0].datetime.date(0), open: self.datas[0].open[0], high: self.datas[0].high[0], low: self.datas[0].low[0], close: self.datas[0].close[0], vol: self.datas[0].volume[0] } self.kline_cache.append(bar) if len(self.kline_cache) 20: return merged merge_bars(self.kline_cache) fxs find_fenxing(merged) bis find_bi(fxs) if bis ! self.bis: self.bis bis self.zhongshu update_zhongshu_all(self.bis)这里要特别提醒self.datas[0].close[0]是当前 bar 的收盘价[-1]是上一根。缠论结构判断有没有未来函数就看你是用[0]还是[-1]做确认。我的习惯是所有分型确认都用上一根 K 线完成也就是说next里已经收盘的最新一根允许被包含进缓存但不允许触发买卖这样能有效避免信号在盘中反复横跳。3.3 资金、手续费和滑点回测参数先定下来再谈收益回测参数必须在开始调策略之前固定否则你会在参数空间里过拟合到根本不存在的盈利。我的默认参数如下参数建议值说明初始资金100000太小无法覆盖多标的组合手续费双边万三A 股佣金参考值印花税卖出千一只在卖出时收取滑点0.001按百分比模拟也可按固定值单笔仓位20%缠论信号频率低单笔太小没收益印花税 Backtrader 默认不包含需要自定义 commission scheme。最简单的方式是卖出自检例如用self.sell之后手动扣减self.broker.add_cash不合适正确做法是继承bt.CommInfoBase重写_getcommission。这块代码不复杂但是必须做否则回测收益会比实盘高不少看起来漂亮的曲线其实是漏了税。4. 缠论系统源码的分层设计信号生成、策略决策与执行器怎么解耦4.1 一个能持续维护的项目目录应该长什么样很多人写策略是把所有逻辑堆在ChanStrategy一个类里数据清洗、缠论识别、买卖判断、仓位管理全部耦合跑通以后每一次改动都要动整个文件加一个指标就要重新测试全部历史结果。做系统设计源码的意义就是把边界切开让缠论算法部分独立于 Backtrader 框架。我常用的目录结构长这样chan_trading/ config.py main.py core/ kline.py fenxing.py bi.py zhongshu.py maipoint.py strategy/ chan_strategy.py sizer/ risk_sizer.py analyzer/ report.pycore目录是纯 Python 代码完全不依赖 Backtrader它的输入是 K 线列表输出是笔和中枢结构。strategy目录是 Backtrader 层负责把 K 线喂给core再把core输出的买卖点转换成订单。sizer决定每次买多少analyzer负责统计。这样拆分之后哪怕你想把信号接给 vn.py 或者通达信公式只动strategy层就够了core完全不用改。4.2 买卖点模块的接口设计缠论的买卖点识别是系统里最杂的部分。三类买点、三类卖点加上中枢延伸的状态切换至少需要维护一组枚举和一组状态字段。推荐把买卖点信号定义成枚举而不是散落的字符串。from enum import Enum class Signal(Enum): BUY_1 1 BUY_2 2 BUY_3 3 SELL_1 -1 SELL_2 -2 SELL_3 -3 class ChanSignal: def __init__(self): self.zhongshu_list [] self.current_zs None self.signal None def update(self, bi): if self.current_zs is None: self.current_zs find_zhongshu(self.zhongshu_list, len(self.zhongshu_list) - 1) if self.current_zs: self.signal self._check_break_and_back(bi) def _check_break_and_back(self, bi): zg self.current_zs[zg] zd self.current_zs[zd] if bi[direction] up and bi[end][value] zg: return Signal.BUY_3 if bi[direction] down and bi[end][value] zd: return Signal.SELL_3 return None例子里的BUY_3判断是很粗糙的突破即买点真正的缠论三买要求突破中枢后第一次回抽不进入中枢所以_check_break_and_back至少还要维护一个“是否已经离开中枢”的状态。这里给出这个简化版本是想说清楚接口形状买卖点模块的输出必须是一个信号枚举加一个触发时间策略层只认这两个字段不管内部是怎么算出来的。4.3 策略层如何消化信号而不是堆逻辑策略层我一般只做三件事拿到信号、判断当前是否持仓、执行下单。缠论信号天然低频经常连续几十根 K 线没有任何信号所以策略层不需要复杂的事件驱动机制一个next里几行判断就够。class ChanStrategy(bt.Strategy): def __init__(self): self.cs ChanSignal() self.last_signal None def next(self): bar self._collect_bar() self.cs.update(bar) sig self.cs.signal if sig in (Signal.BUY_1, Signal.BUY_2, Signal.BUY_3) and not self.position: self.buy(sizeself.sizer.get_size(self.broker.getvalue())) self.last_signal sig elif sig in (Signal.SELL_1, Signal.SELL_2, Signal.SELL_3) and self.position: self.close() self.last_signal sig注意这里self.sizer.get_size()是自定义仓位管理器的接口而不是直接用self.buy(size100)。把仓位计算从策略里拆出去之后不同资金账户可以挂不同风控规则同一个策略源码不用改一行。这是系统设计和单文件脚本最大的差别。好的仓位管理器至少要考虑两件事单笔最大亏损比例和当前可用现金。缠论一买抄底经常遇到买完之后继续跌如果仓位是全仓一个中枢延伸就能让账户净值跌破心理线。class RiskSizer(bt.Sizer): params ( (risk_per_trade, 0.02), (min_size, 100), ) def _getsizing(self, comminfo, cash, data, isbuy): if not isbuy: return self.min_size price data.close[0] risk_cash cash * self.p.risk_per_trade size int(risk_cash / price) size max(size, self.min_size) return sizesizer里comminfo带着手续费对象data是当前标的数据price取close。用风险百分比而不是固定资金做仓位在连续亏损时能自动降低仓位这是缠论这种信号频率不高的系统活过震荡期的前提。5. 缠论回测最容易翻车的 5 个坑现象、原因与修正5.1 信号全部向后偏移一根 K 线回测结果虚高做过缠论回测的人几乎都遇到过同一套买卖点逻辑你按收盘确认信号回测是亏损的但盘中确认信号回测是暴赚的。原因很直白——缠论里的分型和笔需要未来几根 K 线才能确认顶分型没走出来之前你根本不知道这是一个顶。如果用未来数据回放信号等于在 K 线图上开天眼。验证方法很简单在策略里延迟一根 K 线触发交易对比延迟前后的收益曲线如果差异巨大说明系统有未来函数污染。修正方式就是用上一根已确认 K 线触发订单所有分型、笔、中枢的判断基准时间戳都向后推一根。5.2 小周期里笔被切碎中枢频繁失效把同一套笔参数从日线切到 5 分钟线笔的数量会爆炸中枢刚形成三根 K 线就离开了。原因是缠论的“分型之间至少一根独立 K 线”在低周期放大后大级别一笔内部会分解成很多小笔。一般做法有两种一种是按固定 K 线数量过滤最短笔至少 7 根 K 线另一种是在笔递归生成后增加一次平滑把幅度小于某个百分比的笔直接丢弃。太小周期的缠论信号是不可用的30 分钟以下至少要把分型确认的 K 线数翻倍不然中枢区间是噪声区间。5.3 中枢区间更新逻辑错误买卖点重复触发中枢一旦形成会延伸延伸过程里zg和zd都应该被刷新但很多半成品源码只取前三笔做一次中枢后面行情继续走信号反复触发同一个三买。表现为回测里三天两头出现同样的买入信号持仓却没有变化。正确逻辑是中枢只有在被第三类买卖点确认破坏后才关闭否则持续更新区间。修改ChanSignal状态机时重点检查current_zs是否在每次新笔出现时重新计算了重叠区间以及中枢破坏后是否及时置空。5.4 回测收益里混入了交易成本盲区缠论信号频率低看似交易成本影响不大实盘一跑就不是那么回事。A 股卖出印花税千分之一加上滑点和最低五元佣金单次往返成本可能在千三到千五。回测不扣这些成本一个年化 30% 的策略变成实盘年化 15%。Backtrader 的setcommission只处理佣金滑点要单独用add_slippage或自定义执行器。一定要在策略定稿前把成本和滑点按保守值设进去看扣完成本后的收益曲线是否还值得做。5.5 CSV 数据里隐藏的复权和停牌问题导致结构识别错乱缠论的结构识别对高低点敏感如果数据源是前复权价格除权除息那天会生成一根异常跳空 K 线程序会把它当作真正的笔离开立刻触发买卖信号。停牌日通常时成交量归零且价格不变也会干扰分型判断。我一般会在数据清洗阶段过滤成交量小于阈值的 K 线并且对前复权数据做二次校验比较稳的方式是回测用后复权数据确认信号逻辑后再把价格切回当前市场价下单。6. 进阶多标的多周期回测与主图公式对照验证缠论源码跑通单标的之后下一步是把它变成可用的交易系统而不仅仅是一个按钮。多标的回测是 Backtrader 的强项cerebro.adddata可以多次调用每个 data 实例加上_name参数区分标的策略里的self.datas会变成一个列表遍历列表执行同一套缠论逻辑就能看出策略在哪些标的上稳定、哪些标的上失效。这个结果比任何单标的曲线都诚实——如果一个策略只在某一个票上赚钱那大概率是过拟合。多周期方面我推荐先用日线选方向、用 30 分钟定买卖点。日线笔中枢确定当前是上涨结构还是下跌结构30 分钟缠论信号只在方向一致时触发能有效过滤逆势买点。Backtrader 里通过bt.feeds.GenericCSVData同时加载两个周期的数据再在策略里用self.datas[0]和self.datas[1]分别访问唯一要注意的是两个周期的数据时间戳必须对齐停牌日多的标的需要先做日期对齐再喂给框架。最后一步我习惯做视觉对照把缠论系统生成的买卖点导出为 CSV再配合同花顺或通达信的缠论主图公式叠加查看。导出的买卖点如果能在主图上落到中枢离开后的回抽位置说明逻辑实现没有变形如果偏离明显优先检查笔划分是否与主图公式一致。缠论本身存在千人千缠的争议你的源码里笔定义的宽容度直接决定回测结论所以一定要把分型、笔的参数写到配置文件里每次实验前确认版本。我早期在这套系统上吃过最大的亏就是太迷信“缠论万能的买点”把一买信号当成无条件入场结果在下跌趋势里连续吃三次一买止损。后来把缠论结构只当成过滤器把入场决策交给多级别共振和风险仓位控制系统才真正活下来。量化交易系统源码的价值不在于一次跑出多漂亮的曲线而在于它让你每一次亏损都能找到原因、每一次盈利都能复现。如果你准备在这个方向上投入时间先花一周把回测骨架和结构识别模块跑稳再考虑信号、仓位和实盘对接。希望这些整理能帮你在缠论量化的路上少踩几个坑。本文还有配套的精品资源点击获取