add more to klu and klc

This commit is contained in:
jackyu66git
2025-06-03 23:40:19 +08:00
parent ec46febceb
commit 872eec8100
4 changed files with 512 additions and 11 deletions
+225 -1
View File
@@ -135,6 +135,229 @@ class ChanKLC():
self.bi = bi
self.distance = self.index - bi.start_klc.index
#print(self.start_time, self.distance, bi.index, bi.dir)
def cal_fx(self):
"""
根据缠论分型强弱判断规则计算分型强度
返回值:
- 0: 不是分型或无效分型
- 1-100: 分型强度,数值越大表示分型越强
"""
# 检查基本条件:必须是分型且有前后KLC
if (self.fx == Chan_FX_TYPE.UNKNOWN or
self.pre is None or self.next is None or
self.next.end_klu is None):
return 0
# 获取分型的三根K线(KLC
klc1 = self.pre # 第1条
klc2 = self # 第2条(分型中心)
klc3 = self.next # 第3条
if self.fx == Chan_FX_TYPE.TOP:
return self._calculate_top_fx_strength(klc1, klc2, klc3)
elif self.fx == Chan_FX_TYPE.BOTTOM:
return self._calculate_bottom_fx_strength(klc1, klc2, klc3)
else:
return 0
def _calculate_top_fx_strength(self, klc1, klc2, klc3):
"""
计算顶分型强度
"""
strength = 50 # 基础分数
# 1. 检查包含关系(规则1
has_inclusion = self._has_inclusion_relationship(klc1, klc2, klc3)
if not has_inclusion:
strength += 20 # 没有包含关系加分
else:
strength -= 10 # 有包含关系减分
# 检查最坏的包含关系(规则4
if self._is_worst_inclusion_for_top(klc2, klc3):
strength -= 20 # 第3条大阴线"吃掉"第2条阳线
# 2. 检查第1条K线是否为大阳线,第2、3条为小K线(规则2)
if self._is_big_bullish_followed_by_small(klc1, klc2, klc3):
strength -= 25 # 中继顶分型可能性大
# 3. 检查第2条K线形态和第3条K线位置(规则3)
if self._has_strong_top_pattern(klc2, klc3):
strength += 25 # 力度比较大的分型
# 4. 检查第3条K线是否跌破第1条K线(规则5)
if self._breaks_first_klc_bottom_for_top(klc1, klc3):
strength -= 15 # 较弱的顶分型
# 5. 成交量确认
volume_factor = self._get_volume_factor(klc2)
strength += volume_factor
return max(0, min(100, strength))
def _calculate_bottom_fx_strength(self, klc1, klc2, klc3):
"""
计算底分型强度
"""
strength = 50 # 基础分数
# 1. 检查包含关系
has_inclusion = self._has_inclusion_relationship(klc1, klc2, klc3)
if not has_inclusion:
strength += 20 # 没有包含关系加分
else:
strength -= 10 # 有包含关系减分
# 2. 检查第3条K线高点与第1条K线高点的关系(规则1-3)
high_relationship = self._analyze_bottom_high_relationship(klc1, klc3)
if high_relationship == "strong": # 第3条高点远高于第1条
strength += 25
elif high_relationship == "normal": # 第3条高点接近第1条
strength += 5
else: # 第3条高点低于第1条
strength -= 15
# 3. 检查后续K线确认(规则4
if self._has_follow_through_for_bottom():
strength += 15
# 4. 成交量确认
volume_factor = self._get_volume_factor(klc2)
strength += volume_factor
return max(0, min(100, strength))
def _has_inclusion_relationship(self, klc1, klc2, klc3):
"""
检查构成分型的三根原始K线(KLU)是否存在包含关系
"""
# 检查任意两根KLU之间是否存在包含关系
return (klc1.start_klu.index - klc1.end_klu.index != 0 or klc2.start_klu.index - klc2.end_klu.index != 0 or klc3.start_klu.index - klc3.end_klu.index != 0)
def _is_worst_inclusion_for_top(self, klc2, klc3):
"""
检查是否为最坏的包含关系:第3根KLU大阴线"吃掉"第2根KLU阳线
"""
# 获取代表性的KLU
# 第2根KLU:取klc2的最后一根KLU
klu2 = klc2.end_klu if klc2.end_klu else klc2.start_klu
# 第3根KLU:取klc3的第一根KLU
klu3 = klc3.start_klu
if not klu2 or not klu3:
return False
# 检查klu2是否为阳线
klu2_is_bullish = klu2.close > klu2.open
# 检查klu3是否为大阴线(实体占总区间70%以上)
klu3_range = klu3.high - klu3.low
klu3_body = abs(klu3.close - klu3.open)
klu3_is_big_bearish = (klu3.close < klu3.open and
klu3_range > 0 and
klu3_body > klu3_range * 0.7)
# 检查klu3是否包含klu2klu3的高点≥klu2的高点 且 klu3的低点≤klu2的低点)
klu3_contains_klu2 = (klu3.high >= klu2.high and klu3.low <= klu2.low)
return klu2_is_bullish and klu3_is_big_bearish and klu3_contains_klu2
def _is_big_bullish_followed_by_small(self, klc1, klc2, klc3):
"""
检查第1条是否为大阳线,第2、3条为小K线
"""
# 第1条为大阳线
klc1_big_bullish = (klc1.close > klc1.open and
abs(klc1.close - klc1.open) > (klc1.high - klc1.low) * 0.6)
# 第2、3条为小K线
klc2_small = abs(klc2.close - klc2.open) < (klc2.high - klc2.low) * 0.4
klc3_small = abs(klc3.close - klc3.open) < (klc3.high - klc3.low) * 0.4
return klc1_big_bullish and klc2_small and klc3_small
def _has_strong_top_pattern(self, klc2, klc3):
"""
检查是否有强力度的顶分型模式
"""
# 第2条K线有长上影线或为大阴线
klc2_range = klc2.high - klc2.low
if klc2_range > 0:
upper_shadow_ratio = (klc2.high - max(klc2.open, klc2.close)) / klc2_range
has_long_upper_shadow = upper_shadow_ratio > 0.3
else:
has_long_upper_shadow = False
klc2_big_bearish = (klc2.close < klc2.open and
abs(klc2.close - klc2.open) > klc2_range * 0.6)
klc2_strong = has_long_upper_shadow or klc2_big_bearish
# 第3条K线不能以阳线收在第2条K线区间的一半之上
klc2_mid = (klc2.high + klc2.low) / 2
klc3_weak_position = (klc3.close <= klc2_mid or klc3.close < klc3.open)
return klc2_strong and klc3_weak_position
def _breaks_first_klc_bottom_for_top(self, klc1, klc3):
"""
检查第3条是否跌破第1条K线底部且不能高于第1条区间一半之上
"""
breaks_bottom = klc3.low < klc1.low
klc1_mid = (klc1.high + klc1.low) / 2
below_mid = klc3.close <= klc1_mid
return breaks_bottom and below_mid
def _analyze_bottom_high_relationship(self, klc1, klc3):
"""
分析底分型中第3条K线高点与第1条K线高点的关系
"""
high_diff_ratio = (klc3.high - klc1.high) / klc1.high if klc1.high > 0 else 0
if high_diff_ratio > 0.02: # 高出2%以上
return "strong"
elif high_diff_ratio >= -0.01: # 接近或略高
return "normal"
else: # 明显低于
return "weak"
def _has_follow_through_for_bottom(self):
"""
检查底分型后续是否有确认
"""
# 检查后续第1条K线的低点是否高于底分型的上边沿
if self.next and self.next.next:
follow_klc = self.next.next
bottom_fx_top = max(self.pre.high, self.high, self.next.high)
return follow_klc.low > bottom_fx_top
return False
def _get_volume_factor(self, klc):
"""
获取成交量因子
"""
avg_volume = self._calculate_average_volume(lookback=5)
if avg_volume > 0:
volume_ratio = klc.volume / avg_volume
if volume_ratio > 2.0:
return 10 # 大量确认
elif volume_ratio > 1.5:
return 5 # 放量
elif volume_ratio < 0.5:
return -5 # 缩量
return 0
def check_pre_has_fx(self):
if self.pre:
return self.pre.fx != Chan_FX_TYPE.UNKNOWN
elif self.pre.pre:
return self.pre.pre.fx != Chan_FX_TYPE.UNKNOWN
elif self.pre.pre.pre:
return self.pre.pre.pre.fx != Chan_FX_TYPE.UNKNOWN
elif self.pre.pre.pre.pre:
return self.pre.pre.pre.pre.fx != Chan_FX_TYPE.UNKNOWN
else:
return False
def cal_klu_features(self):
features = dict()
feature_sums = dict()
@@ -1162,8 +1385,9 @@ class ChanKLC():
Returns:
int: 强度评分 15-80分
"""
return self.cal_fx()
# 如果不是分型,返回0
if self.fx == Chan_FX_TYPE.UNKNOWN:
if self.fx == Chan_FX_TYPE.UNKNOWN or self.pre == None or self.next == None or self.next.end_klu == None:
return 0
# 如果没有前一个KLC,返回基础分
+115
View File
@@ -63,7 +63,121 @@ class ChanKLU:
self.fx_type = Chan_FX_TYPE.UNKNOWN
self.fx_confirmed = False
return False
def cal_fx(self):
"""
根据缠论经典规则计算分型强弱
返回分型强度3=极强2=1=中等0=-1=极弱
"""
if self.fx_type == Chan_FX_TYPE.UNKNOWN or not self.pre or not self.next:
return 0
if self.fx_type == Chan_FX_TYPE.TOP:
return self._cal_top_fx_strength()
else: # BOTTOM
return self._cal_bottom_fx_strength()
def _check_contain_relation(self, k1, k2):
"""检查两根K线是否存在包含关系"""
return (k1.high >= k2.high and k1.low <= k2.low) or (k2.high >= k1.high and k2.low <= k1.low)
def _is_big_yang_line(self, klu):
"""判断是否为大阳线"""
return klu.close > klu.open and (klu.close - klu.open) / klu.open > 0.02
def _is_big_yin_line(self, klu):
"""判断是否为大阴线"""
return klu.close < klu.open and (klu.open - klu.close) / klu.open > 0.02
def _is_small_line(self, klu):
"""判断是否为小K线"""
return abs(klu.close - klu.open) / klu.open < 0.01
def _has_long_upper_shadow(self, klu):
"""判断是否有长上影线"""
body_size = abs(klu.close - klu.open)
upper_shadow = klu.high - max(klu.close, klu.open)
return upper_shadow > body_size * 1.5
def _cal_top_fx_strength(self):
"""计算顶分型强度"""
strength = 0
k1, k2, k3 = self.pre, self, self.next
# (1) 检查包含关系 - 没有包含关系加分
has_contain_12 = self._check_contain_relation(k1, k2)
has_contain_23 = self._check_contain_relation(k2, k3)
if not has_contain_12 and not has_contain_23:
strength += 1 # 完全没有包含关系,加1分
# (2) 检查第1条K线是大阳线,第2、3条是小K线的情况
if self._is_big_yang_line(k1) and self._is_small_line(k2) and self._is_small_line(k3):
strength -= 2 # 中继顶分型特征,减2分
# (3) 检查第2条K线有长上影线或大阴线,且第3条K线条件
k2_mid = (k2.high + k2.low) / 2
k3_is_yang = k3.close > k3.open
k3_close_above_mid = k3.close > k2_mid
if (self._has_long_upper_shadow(k2) or self._is_big_yin_line(k2)) and not (k3_is_yang and k3_close_above_mid):
strength += 2 # 力度大的顶分型,加2分
# (4) 检查第2、3条K线包含关系,第3条为大阴线"吃掉"第2条
if has_contain_23 and self._is_big_yin_line(k3) and k3.low < k2.low and k3.high < k2.high:
strength += 1 # 最坏包含关系,但对顶分型有利,加1分
# (5) 第3条K线跌破第1条K线底部且不能高于第1条K线区间一半之上
k1_mid = (k1.high + k1.low) / 2
if k3.low < k1.low and k3.high < k1_mid:
strength -= 1 # 较弱的顶分型,减1分
# 额外检查:第3条K线收盘价相对第1条K线的位置
if k3.close < k1.low:
strength += 1 # 强烈下跌确认,加1分
return max(-1, min(3, strength)) # 限制在-1到3范围内
def _cal_bottom_fx_strength(self):
"""计算底分型强度"""
strength = 0
k1, k2, k3 = self.pre, self, self.next
# 底分型上边沿
fx_top = max(k1.high, k2.high)
# (1) 第3条K线高点远高于第1条K线高点
if k3.high > k1.high * 1.02: # 高出2%以上认为是"远高于"
strength += 2 # 较强走势,加2分
# (2) 第3条K线高点正好是第1根K线高点,或略微高于底分型上边沿
elif k1.high * 0.99 <= k3.high <= fx_top * 1.01: # 在合理范围内
strength += 0 # 一般走势,不加分也不减分
# (3) 第3条K线高点低于第1条K线高点
elif k3.high < k1.high:
strength -= 1 # 较弱走势,减1分
# 检查包含关系
has_contain_12 = self._check_contain_relation(k1, k2)
has_contain_23 = self._check_contain_relation(k2, k3)
if not has_contain_12 and not has_contain_23:
strength += 1 # 完全没有包含关系,加1分
# 检查第3条K线是否为强阳线
if self._is_big_yang_line(k3):
strength += 1 # 强阳线确认,加1分
# (4) 检查后续第1条K线(如果存在)
if hasattr(k3, 'next') and k3.next:
next_k = k3.next
if next_k.low > fx_top:
strength += 2 # 后续K线低点高于底分型上边沿,强烈确认,加2分
elif next_k.low <= k2.low:
strength -= 1 # 后续K线跌破分型低点,减1分
return max(-1, min(3, strength)) # 限制在-1到3范围内
def calculate_realtime_fx_strength(self):
"""
用self.pre和self.next实现分型强弱判断与KLC中cal_fx_strength一致
@@ -126,6 +240,7 @@ class ChanKLU:
if final_score > 1.8:
print(self.time, final_score, is_bi_end, post_fx_confirmation, fx_quality)
#print(self.time, final_score, is_bi_end, post_fx_confirmation, fx_quality)
self.fx_strength = self.cal_fx()
return self.fx_strength
def _check_if_bi_ending_fx(self):
+171 -9
View File
@@ -488,42 +488,204 @@ def analyze_chan(df):
}
def identify_trade_points(bi_list, seg_list, zs_list):
"""识别缠论买卖点 - 只保留最重要的一类买卖点,减少标记干扰"""
"""识别缠论买卖点 - 多级别识别,减少滞后性"""
trade_points = []
# 输出调试信息
print(f"识别买卖点:总共 {len(bi_list)} 个笔, {len(seg_list)} 个线段, {len(zs_list)} 个中枢")
# 只识别一类买卖点:线段向上或向下突破
# 1. 基于笔的二三类买卖点识别(更及时)
trade_points.extend(identify_bi_trade_points(bi_list, zs_list))
# 2. 基于线段的一类买卖点识别(传统方法)
trade_points.extend(identify_seg_trade_points(seg_list))
# 3. 基于分型强度的预警点识别(最及时)
trade_points.extend(identify_fx_warning_points(bi_list))
# 4. 基于MACD背驰的买卖点识别
trade_points.extend(identify_macd_divergence_points(bi_list))
# 按时间排序
trade_points.sort(key=lambda x: x['time'])
print(f"总共识别出 {len(trade_points)} 个买卖点")
return trade_points
def identify_bi_trade_points(bi_list, zs_list):
"""基于笔识别二三类买卖点 - 更及时的信号"""
trade_points = []
if len(bi_list) < 3:
return trade_points
# 构建中枢映射,便于快速查找
zs_map = {}
for zs in zs_list:
if zs.is_sure: # 只考虑已确认的中枢
zs_map[zs.start_klc.end_time] = zs
for i in range(2, len(bi_list)):
current_bi = bi_list[i]
prev_bi = bi_list[i-1]
prev_prev_bi = bi_list[i-2]
# 确保笔已完成
if not current_bi.end_klc or not prev_bi.end_klc or not prev_prev_bi.end_klc:
continue
# 二类买点:向下笔后的向上笔,且不创新低
if (convert_direction(prev_bi.dir) == -1 and
convert_direction(current_bi.dir) == 1):
prev_low = prev_bi.end_klc.low
current_end_price = current_bi.end_klc.high
# 检查是否不创新低(相对于前面的低点)
if i >= 4: # 至少需要5个笔来判断
earlier_lows = [bi.end_klc.low for bi in bi_list[max(0, i-4):i-1]
if convert_direction(bi.dir) == -1 and bi.end_klc]
if earlier_lows and prev_low > min(earlier_lows):
trade_points.append({
'type': TRADE_POINT_TYPE.BUY2,
'time': current_bi.end_klc.end_time,
'price': current_end_price,
'desc': '二类买点(笔)'
})
# 二类卖点:向上笔后的向下笔,且不创新高
if (convert_direction(prev_bi.dir) == 1 and
convert_direction(current_bi.dir) == -1):
prev_high = prev_bi.end_klc.high
current_end_price = current_bi.end_klc.low
# 检查是否不创新高(相对于前面的高点)
if i >= 4: # 至少需要5个笔来判断
earlier_highs = [bi.end_klc.high for bi in bi_list[max(0, i-4):i-1]
if convert_direction(bi.dir) == 1 and bi.end_klc]
if earlier_highs and prev_high < max(earlier_highs):
trade_points.append({
'type': TRADE_POINT_TYPE.SELL2,
'time': current_bi.end_klc.end_time,
'price': current_end_price,
'desc': '二类卖点(笔)'
})
return trade_points
def identify_seg_trade_points(seg_list):
"""基于线段识别一类买卖点 - 传统方法"""
trade_points = []
if len(seg_list) >= 3:
for i in range(2, len(seg_list)):
# 确保线段已完成
if seg_list[i].end_bi and seg_list[i-1].end_bi and seg_list[i-2].end_bi:
# 一类买点:向下-向上-向下的底分型,第三段结束点为买点
# 一类买点:向下-向上-向下的底分型
if (convert_direction(seg_list[i-2].dir) == -1 and
convert_direction(seg_list[i-1].dir) == 1 and
convert_direction(seg_list[i].dir) == -1):
print(f"发现一类买点:线段方向 {convert_direction(seg_list[i-2].dir)}-{convert_direction(seg_list[i-1].dir)}-{convert_direction(seg_list[i].dir)}")
trade_points.append({
'type': TRADE_POINT_TYPE.BUY1,
'time': seg_list[i].end_bi.end_klc.end_time,
'price': seg_list[i].end_bi.end_klc.low,
'desc': '一类买点'
'desc': '一类买点(线段)'
})
# 一类卖点:向上-向下-向上的顶分型,第三段结束点为卖点
# 一类卖点:向上-向下-向上的顶分型
if (convert_direction(seg_list[i-2].dir) == 1 and
convert_direction(seg_list[i-1].dir) == -1 and
convert_direction(seg_list[i].dir) == 1):
print(f"发现一类卖点:线段方向 {convert_direction(seg_list[i-2].dir)}-{convert_direction(seg_list[i-1].dir)}-{convert_direction(seg_list[i].dir)}")
trade_points.append({
'type': TRADE_POINT_TYPE.SELL1,
'time': seg_list[i].end_bi.end_klc.end_time,
'price': seg_list[i].end_bi.end_klc.high,
'desc': '一类卖点'
'desc': '一类卖点(线段)'
})
print(f"总共识别出 {len(trade_points)} 个买卖点")
return trade_points
def identify_fx_warning_points(bi_list):
"""基于分型强度识别预警点 - 最及时的信号"""
trade_points = []
if len(bi_list) < 2:
return trade_points
# 检查最近的几个笔
recent_bis = bi_list[-3:] if len(bi_list) >= 3 else bi_list
for bi in recent_bis:
if not bi.end_klc:
continue
# 获取分型强度(如果有的话)
fx_strength = 0
if hasattr(bi.end_klc, 'cal_fx_strength'):
try:
fx_strength = bi.end_klc.cal_fx_strength()
except:
fx_strength = 0
# 强分型预警(分型强度>=2
if fx_strength >= 2:
if convert_direction(bi.dir) == -1: # 向下笔结束,可能的底部
trade_points.append({
'type': TRADE_POINT_TYPE.BUY3,
'time': bi.end_klc.end_time,
'price': bi.end_klc.low,
'desc': f'强分型预警-买点(强度:{fx_strength})'
})
elif convert_direction(bi.dir) == 1: # 向上笔结束,可能的顶部
trade_points.append({
'type': TRADE_POINT_TYPE.SELL3,
'time': bi.end_klc.end_time,
'price': bi.end_klc.high,
'desc': f'强分型预警-卖点(强度:{fx_strength})'
})
return trade_points
def identify_macd_divergence_points(bi_list):
"""基于MACD背驰识别买卖点"""
trade_points = []
if len(bi_list) < 4:
return trade_points
# 检查最近的笔是否有背驰
for i in range(2, len(bi_list)):
current_bi = bi_list[i]
if not current_bi.end_klc or not hasattr(current_bi, 'macd_div'):
continue
# MACD背驰阈值
divergence_threshold = 0.3
# 向下笔的底背驰 -> 买点
if (convert_direction(current_bi.dir) == -1 and
hasattr(current_bi, 'macd_div') and
current_bi.macd_div > divergence_threshold):
trade_points.append({
'type': TRADE_POINT_TYPE.BUY2,
'time': current_bi.end_klc.end_time,
'price': current_bi.end_klc.low,
'desc': f'MACD底背驰买点(背驰度:{current_bi.macd_div:.2f})'
})
# 向上笔的顶背驰 -> 卖点
elif (convert_direction(current_bi.dir) == 1 and
hasattr(current_bi, 'macd_div') and
current_bi.macd_div > divergence_threshold):
trade_points.append({
'type': TRADE_POINT_TYPE.SELL2,
'time': current_bi.end_klc.end_time,
'price': current_bi.end_klc.high,
'desc': f'MACD顶背驰卖点(背驰度:{current_bi.macd_div:.2f})'
})
return trade_points
# 辅助函数,转换缠论方向枚举为整数
+1 -1
View File
@@ -3254,7 +3254,7 @@
// 构建显示文本,包含分型类型和强度信息
let displayText = `${fx.fx_strength.toFixed(1)}`;
if (fx.fx_strength < 0) { // 降低阈值让更多分型显示
if (fx.fx_strength < 50) { // 降低阈值让更多分型显示
displayText = fx.fx_strength >= 0.8 ? '•' : '' // 0.8以上显示点,0.8以下不显示文本
}