mirror of
https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git
synced 2026-09-12 15:44:18 +08:00
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
This commit is contained in:
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user