企业官网建设流程全解析

简介：本资源是一套基于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[i+1] and low[i] > low[i-1] and low[i] > low[i+1]) 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[i+1] and low[i] < low[i-1] and low[i] < low[i+1])

这个逻辑本身没错，但有个前提：它要求这三根 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 bis

3.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. 缠论指标与策略信号如何接入 Backtrader

4.1 自定义 Indicator 的写法

Backtrader 的 Indicator 封装了数据线的计算逻辑，可以在 addindicator 之后自动画图，也能在 Strategy 里直接访问。我实现了一个简单的分型 Indicator：

import backtrader as bt class ChanFractal(bt.Indicator): """ 缠论分型指标 输出：top 和 bottom 两条线 顶分型时 top 标记分型高点，底分型时 bottom 标记分型低点 """ lines = ('top', 'bottom') params = (('min_klines', 5),) plotinfo = dict(plot=True, subplot=False) plotlines = dict( top=dict(marker='v', markersize=8, color='red', fillstyle='full', ls=''), bottom=dict(marker='^', markersize=8, color='green', fillstyle='full', 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(size=3) l = self.data.low.get(size=3) 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(size=self.params.size) else: # 持仓中，顶分型确认且已有盈利，先止盈 if self.chan_fractal.top[-1]: self.close()

这里 bottom[-1] 表示用上一根 K 线确认的底分型，避开用当前未完成 K 线做判断的问题。

4.3 三种接入方式的取舍

我在实际设计过程中试过三种把缠论计算接进 Backtrader 的方式，各有适用场景：

我的最终选择是“方式二为主、方式一为辅”：缠论计算模块独立成一个 Python 包，先把所有笔、线段、中枢算好，存成结构化数据；然后写一个适配层，把这些数据转换成 Backtrader 的 Line 或直接作为一个数据源传进 cerebro；Indicator 只负责画图，不承担核心计算。

5. 信号确认与未来函数：回测里最容易被忽略的偏差

5.1 未来函数在缠论代码中的两个典型来源

缠论交易系统特别容易在未来函数上翻车。第一个来源是“分型未确认就交易”。一个顶分型至少需要三根 K 线才能确定，如果你用包含当前 bar 在内的三根 K 线去判断分型，那么当前 bar 还没走完，你可能已经知道了这个 bar 的最高价和最低价，这在回测里等于提前看到了未来。

第二个来源是“笔/线段的动态修正”。缠论里一笔在形成过程中，经常因为后续 K 线变化而被重新划分，比如原来是向上笔，后面又出了更高的顶分型，那上一笔的终点会被往后挪。如果在策略回测时用最终修正后的笔去回溯当时的买点，就会产生一种“事后诸葛亮”式的收益，这种回测结果拿到实盘上一点参考意义都没有。

5.2 用下一根 K 线执行修复回测偏差

修复未来函数的基本原则就一句话：回测时只能用“截至当前 K 线收盘已经确认的信息”，并且下单要在“下一根 K 线”。在 Backtrader 里，默认市价单就是在下一根 bar 开盘成交的，所以只要保证信号判断不引用未来数据，下单延迟的问题框架已经替你解决了。

需要注意的反而是在 Indicator 里别用“当前 bar + 未来 bar”去确认信号。比如 2.1 节那段分型判断，我把i当作“待判断的 K 线”，后面循环到i+1时才能确认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[i+1] and \ lows[i] > lows[i-1] and lows[i] > lows[i+1]: fractals.append(Fractal(i, 'top', highs[i])) elif highs[i] < highs[i-1] and highs[i] < highs[i+1] and \ lows[i] < lows[i-1] and lows[i] < lows[i+1]: 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(size=self.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, cash=100000, commission=0.0003): cerebro = bt.Cerebro() data = bt.feeds.PandasData(dataname=df) cerebro.adddata(data) cerebro.addstrategy(ChanStrategy) cerebro.addindicator(ChanFractal) cerebro.broker.setcash(cash) cerebro.broker.setcommission(commission=commission) cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe') cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown') results = cerebro.run() return results[0] if __name__ == '__main__': df = pd.read_csv('data/000001.csv', index_col=0, parse_dates=True) 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% 胜率”的指标源码，我劝你直接绕开——结构类策略在震荡行情里被反复打脸是常态，代码能帮你做的是快速验证、强制纪律，而不是让你躺着赚钱。把这套系统当成一个“把模糊的交易思想变成可检验规则”的工具，用它去跑足够多的历史数据，再一点点修正参数和规则，这才是它真正的用法。

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