From 6d74c2e5ebed01222abc7bcc0b61b39a14efcc83 Mon Sep 17 00:00:00 2001
From: Justin Gu <97915@qq.com>
Date: Thu, 11 Jun 2026 02:22:31 +0800
Subject: [PATCH] feat: enhance multi-level chanlun analysis with
 trend/overlap/divergence

- Add trend_direction detection (up/down/None) via consecutive bi analysis
- Add bi_overlap detection for consolidation identification
- Add divergence_possible check (2+ ZS + diminishing amplitude)
- Return type changed from dict[str, int] to dict[str, Any]
- Backward compatible: all original fields preserved
- Extract 3 helper functions for clean separation of concerns
---
 src/easy_tdx/chanlun/multi_level.py | 144 +++++++++++++++++++++++++++-
 1 file changed, 139 insertions(+), 5 deletions(-)

diff --git a/src/easy_tdx/chanlun/multi_level.py b/src/easy_tdx/chanlun/multi_level.py
index 2850b96..5bfa31c 100644
--- a/src/easy_tdx/chanlun/multi_level.py
+++ b/src/easy_tdx/chanlun/multi_level.py
@@ -6,6 +6,8 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 from easy_tdx.chanlun.analyser import ChanlunAnalyser, ChanlunResult
 from easy_tdx.chanlun.types import BI
 
@@ -72,11 +74,11 @@ class MultiLevelAnalyser:
         high_level: str,
         low_level: str,
         high_bi: BI,
-    ) -> dict[str, int]:
+    ) -> dict[str, Any]:
         """查询高级别笔在低级别中的走势信息。
 
         查找低级别中时间范围落在高级别笔内的所有笔和中枢，
-        统计形成趋势/盘整的情况。
+        分析趋势方向、笔重叠（盘整特征）和背驰条件。
 
         Args:
             high_level: 高级别名称
@@ -84,14 +86,30 @@ class MultiLevelAnalyser:
             high_bi: 高级别的笔
 
         Returns:
-            {"bi_count": 低级别笔数, "zs_count": 低级别中枢数,
-             "has_trend": 是否形成趋势, "has_consolidation": 是否形成盘整}
+            dict with keys:
+                bi_count: 低级别笔数
+                zs_count: 低级别中枢数
+                has_trend: 是否形成趋势 (>=2 个中枢)
+                has_consolidation: 是否形成盘整 (>=1 个中枢)
+                trend_direction: 趋势方向 ("up"/"down"/None)
+                bi_overlap: 相邻笔是否有重叠 (盘整特征)
+                divergence_possible: 是否具备背驰条件
         """
         high_result = self.get_result(high_level)
         low_result = self.get_result(low_level)
 
+        empty = {
+            "bi_count": 0,
+            "zs_count": 0,
+            "has_trend": False,
+            "has_consolidation": False,
+            "trend_direction": None,
+            "bi_overlap": False,
+            "divergence_possible": False,
+        }
+
         if high_result is None or low_result is None:
-            return {"bi_count": 0, "zs_count": 0, "has_trend": False, "has_consolidation": False}
+            return empty
 
         # 高级别笔的时间范围
         start_date = high_bi.start.k.date
@@ -117,9 +135,125 @@ class MultiLevelAnalyser:
         has_trend = len(low_zss) >= 2
         has_consolidation = len(low_zss) >= 1
 
+        # 趋势方向判断：连续笔的高低点是否递增或递减
+        trend_direction = _detect_trend_direction(low_bis)
+
+        # 笔重叠判断：相邻笔的价格区间是否重叠
+        bi_overlap = _detect_bi_overlap(low_bis)
+
+        # 背驰条件：至少 2 个中枢 + 最后一笔进入最后一个中枢
+        divergence_possible = _check_divergence_condition(low_bis, low_zss)
+
         return {
             "bi_count": len(low_bis),
             "zs_count": len(low_zss),
             "has_trend": has_trend,
             "has_consolidation": has_consolidation,
+            "trend_direction": trend_direction,
+            "bi_overlap": bi_overlap,
+            "divergence_possible": divergence_possible,
         }
+
+
+def _detect_trend_direction(bis: list[BI]) -> str | None:
+    """判断笔序列的趋势方向。
+
+    如果连续 3 笔以上高点递增且低点递增 → 上升趋势。
+    如果连续 3 笔以上高点递减且低点递减 → 下降趋势。
+    否则 → None（盘整或数据不足）。
+
+    Args:
+        bis: 低级别笔列表
+
+    Returns:
+        "up", "down", 或 None
+    """
+    if len(bis) < 3:
+        return None
+
+    # 取偶数索引笔（上升笔）和奇数索引笔（下降笔）的高低点
+    up_bis = bis[0::2]  # 上升笔
+    down_bis = bis[1::2]  # 下降笔
+
+    # 上升趋势：上升笔高点递增 + 下降笔低点递增
+    up_highs_ascending = all(up_bis[i].high > up_bis[i - 1].high for i in range(1, len(up_bis)))
+    down_lows_ascending = len(down_bis) >= 2 and all(
+        down_bis[i].low > down_bis[i - 1].low for i in range(1, len(down_bis))
+    )
+
+    if up_highs_ascending and (len(down_bis) < 2 or down_lows_ascending):
+        return "up"
+
+    # 下降趋势：下降笔低点递减 + 上升笔高点递减
+    down_lows_descending = all(
+        down_bis[i].low < down_bis[i - 1].low for i in range(1, len(down_bis))
+    )
+    up_highs_descending = len(up_bis) >= 2 and all(
+        up_bis[i].high < up_bis[i - 1].high for i in range(1, len(up_bis))
+    )
+
+    if down_lows_descending and (len(up_bis) < 2 or up_highs_descending):
+        return "down"
+
+    return None
+
+
+def _detect_bi_overlap(bis: list[BI]) -> bool:
+    """检测相邻笔是否有价格重叠（盘整特征）。
+
+    两笔重叠 = 一笔的低点 <= 另一笔的高点（反向）。
+    如果存在重叠 → 盘整，不重叠 → 趋势。
+
+    Args:
+        bis: 低级别笔列表
+
+    Returns:
+        True 如果存在相邻笔重叠
+    """
+    if len(bis) < 2:
+        return False
+
+    for i in range(1, len(bis)):
+        prev = bis[i - 1]
+        curr = bis[i]
+        # 两笔重叠：前笔低点 <= 后笔高点 且 后笔低点 <= 前笔高点
+        if prev.low <= curr.high and curr.low <= prev.high:
+            return True
+
+    return False
+
+
+def _check_divergence_condition(
+    bis: list[BI],
+    zss: list[Any],
+) -> bool:
+    """检查是否具备背驰条件。
+
+    背驰条件（简化版）：
+    1. 至少 2 个中枢（趋势结构）
+    2. 有足够的笔（至少 5 笔，即 a+A0+b+A1+c 结构）
+    3. 最后一笔的斜率或幅度小于前一笔（力度衰减）
+
+    Args:
+        bis: 低级别笔列表
+        zss: 低级别中枢列表
+
+    Returns:
+        True 如果可能形成背驰
+    """
+    if len(zss) < 2 or len(bis) < 5:
+        return False
+
+    # 简化背驰判断：最后一笔的幅度 < 倒数第三笔的幅度
+    # （倒数第三笔是与最后一笔同方向的笔）
+    last_bi = bis[-1]
+    for i in range(len(bis) - 3, -1, -2):
+        prev_bi = bis[i]
+        # 比较同方向笔的幅度
+        last_amp = abs(last_bi.end.k.close - last_bi.start.k.close)
+        prev_amp = abs(prev_bi.end.k.close - prev_bi.start.k.close)
+        if prev_amp > 0 and last_amp < prev_amp:
+            return True
+        break
+
+    return False