python缠论形态分析过程

首先处理包含关系

def _analyze(self):
"""执行完整的缠论分析"""
if len(self.bars_raw) < 5:
return

# 1. 处理包含关系
self._remove_include()

# 2. 识别分型
self._find_fx()

# 3. 识别笔
self._find_bi()

# 4. 识别中枢
self._find_zs()

# 5. 识别买卖点
self._find_signals()

def _remove_include(self):
"""去除包含关系，生成新K线序列"""
self.bars_new = []

for bar in self.bars_raw:
if not self.bars_new:
self.bars_new.append({
'dt': bar.dt, 'high': bar.high, 'low': bar.low,
'open': bar.open, 'close': bar.close, 'vol': bar.vol,
'open_oi': bar.open_oi, 'close_oi': bar.close_oi,
'elements': [bar]
})
continue

last = self.bars_new[–1]

# 判断包含关系
if (last['high'] >= bar.high and last['low'] <= bar.low) or \\
(last['high'] <= bar.high and last['low'] >= bar.low):
# 存在包含关系，合并
if len(self.bars_new) < 2:
direction = Direction.Up
else:
prev = self.bars_new[–2]
direction = Direction.Up if prev['high'] < last['high'] else Direction.Down

if direction == Direction.Up:
# 向上取高高低低
last['high'] = max(last['high'], bar.high)
last['low'] = max(last['low'], bar.low)
else:
# 向下取低高低低
last['high'] = min(last['high'], bar.high)
last['low'] = min(last['low'], bar.low)
last['elements'].append(bar)
else:
self.bars_new.append({
'dt': bar.dt, 'high': bar.high, 'low': bar.low,
'open': bar.open, 'close': bar.close, 'vol': bar.vol,
'open_oi': bar.open_oi, 'close_oi': bar.close_oi,
'elements': [bar]
})

缠论中的“背驰”判断，既不能用简单的“进入段和离开段长度比较”，也不能单一依靠“MACD面积”。这是一个需要多维度综合判断的概念。两种方法各有侧重，但都有局限性。

简单来说，最核心的比较是进入中枢的“进入段”与离开中枢的“离开段”，而MACD是其最重要的辅助验证工具。下图清晰地展示了背驰判断的核心逻辑框架与验证工具：

flowchart TD
A[背驰判断核心框架] –> B{第一步: 明确比较对象}
B –> B1[同向的“进入段a”与“离开段c”<br>（均在“中枢”同一侧）]
B1 –> C{第二步: 力度比较核心}

C –> D[“价格段”力度]
D –> D1[线段长度]
D –> D2[上涨/下跌斜率]

C –> E[“MACD面积”力度]
E –> E1[黄白线高度]
E –> E2[红绿柱面积]

D & E –> F{第三步: 综合判断标准<br>“离开段c”是否满足?}

F –> G[条件1: 价格创新高/新低]
F –> H[条件2: 价格段力度减弱]
F –> I[条件3: MACD面积减小<br>和/或 黄白线高度降低]

G & H & I –> J[✅ 背驰大概率成立]

H -.-> |常见误区| K[“仅比较线段长度”<br>忽略斜率与中枢]
I -.-> |常见误区| L[“仅比较MACD面积”<br>忽略价格走势]

下面详细解释图中的关键点，并说明你的Python代码应如何实现。

1. 背驰的完整判断逻辑

背驰的真正含义是：趋势的“力度”在减弱。这种减弱需要从“价格走势”和“动量指标”两个维度，通过“区间套”原理在多个级别上逐级确认。

• 正确的比较对象：必须是同向、且位于中枢同一侧的两段走势。最常见的是“进入中枢的段（a段）”与“离开中枢的段（c段）”进行比较。
• 力度的多维度比较：
  • 价格走势力度：c段的长度（上涨/下跌的空间）应明显小于a段，且斜率趋于平缓。
  • MACD指标力度：c段对应的MACD红绿柱面积之和应小于a段，且DIFF线（快线）的高点/低点可能不创新高/新低（即“黄白线背驰”）。

2. 两种方法的局限性与关系

• 仅看线段长度：如果c段以时间换空间（缓慢的长时间爬升/阴跌），即使空间长度与a段相当，但斜率已经大幅降低，这也是一种背驰。仅看长度会漏判。
• 仅看MACD面积：MACD是滞后指标。在剧烈震荡或小级别走势中容易产生假信号，且无法反映走势的几何结构（如是否触及关键压力位）。

结论：价格段比较是“体”，MACD面积是“用”。缠论原文是以价格走势为绝对核心，MACD作为最重要的辅助验证工具。两者出现共振时，背驰判断的可靠性最高。

3. Python代码实现建议

在你的 BeichiAnalyzer 类中，应该实现一个综合判断方法。以下是核心代码逻辑示例：

class BeichiAnalyzer:
def detect_beichi_comprehensive(self, df, bi_list, zhongshu, current_bi_index):
"""
综合判断背驰
:param df: K线数据
:param bi_list: 笔的列表
:param zhongshu: 当前分析的中枢字典 {'high': xx, 'low': xx, 'start_idx': xx, 'end_idx': xx}
:param current_bi_index: 疑似背驰段（离开段c）在bi_list中的索引
:return: (是否存在背驰, 背驰类型, 证据字典)
"""
if current_bi_index < 2 or not zhongshu:
return False, None, {}

# 1. 找到进入段a (中枢前的同向一笔)
a_bi = None
for i in range(current_bi_index – 1, –1, –1):
# 找到与当前笔同向、且在中枢前的一笔作为进入段
if bi_list[i]['direction'] == bi_list[current_bi_index]['direction']:
# 确保这一笔的终点在中枢区间另一侧（简单判断）
if (bi_list[i]['direction'] == 'up' and bi_list[i]['high'] < zhongshu['low']) or \\
(bi_list[i]['direction'] == 'down' and bi_list[i]['low'] > zhongshu['high']):
a_bi = bi_list[i]
break

if not a_bi:
return False, None, {}

c_bi = bi_list[current_bi_index]

# 2. 比较价格段力度
price_weakening = self._compare_price_strength(a_bi, c_bi)

# 3. 比较MACD力度 (需要提前计算df的MACD)
macd_weakening = self._compare_macd_strength(df, a_bi, c_bi, zhongshu)

# 4. 综合判断
evidence = {
'price_weakening': price_weakening,
'macd_weakening': macd_weakening
}

# 价格和MACD同时减弱，是最强背驰信号
if price_weakening and macd_weakening:
return True, f"{c_bi['direction']}_beichi_strong", evidence
# 只有价格减弱，为疑似背驰
elif price_weakening:
return True, f"{c_bi['direction']}_beichi_weak", evidence

return False, None, evidence

def _compare_price_strength(self, a_bi, c_bi):
"""比较两笔的价格走势力度"""
# 计算幅度
a_range = abs(a_bi['high'] – a_bi['low'])
c_range = abs(c_bi['high'] – c_bi['low'])

# 计算斜率 (幅度 / 时间)
a_time = (a_bi['edt'] – a_bi['sdt']).total_seconds()
c_time = (c_bi['edt'] – c_bi['sdt']).total_seconds()

a_slope = a_range / max(a_time, 1)
c_slope = c_range / max(c_time, 1)

# 幅度和斜率均减弱
if c_range < a_range * 0.8 and c_slope < a_slope * 0.8:
return True
return False

def _compare_macd_strength(self, df, a_bi, c_bi, zhongshu):
"""比较两笔对应的MACD指标力度（简化版）"""
# 获取a段和c段对应的K线数据区间
a_mask = (df.index >= a_bi['sdt']) & (df.index <= a_bi['edt'])
c_mask = (df.index >= c_bi['sdt']) & (df.index <= c_bi['edt'])

# 假设df已有`macd_diff`, `macd_dea`, `macd_bar`列
a_macd_area = df.loc[a_mask, 'macd_bar'].abs().sum() # 面积取绝对值
c_macd_area = df.loc[c_mask, 'macd_bar'].abs().sum()

# 获取DIFF线峰值
if a_bi['direction'] == 'up':
a_macd_diff_peak = df.loc[a_mask, 'macd_diff'].max()
c_macd_diff_peak = df.loc[c_mask, 'macd_diff'].max()
diff_weakening = c_macd_diff_peak < a_macd_diff_peak
else:
a_macd_diff_peak = df.loc[a_mask, 'macd_diff'].min()
c_macd_diff_peak = df.loc[c_mask, 'macd_diff'].min()
diff_weakening = c_macd_diff_peak > a_macd_diff_peak

# MACD面积和黄白线高度均减弱
if c_macd_area < a_macd_area * 0.7 and diff_weakening:
return True
return False

4. 实战应用要点

在应用时，请遵循以下流程：

希望这个综合解释和代码示例能帮助你更准确地实现缠论的背驰判断。如果你在具体实现数据对接（如计算MACD）或判断逻辑时有进一步问题，可以随时提出。