jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
9a68aa19131b7b8c578f7421d944f9c5a782dbb9
Chan/ChanKLU.py
T
jackyu66git 9a68aa1913 添加布林带显示
2025-07-07 01:15:49 +08:00

621 lines
24 KiB
Python
Raw Blame History

from ChanEnum import Chan_FX_TYPE, Chan_KLU_TYPE, Chan_K_DIR
class ChanKLU:
def __init__(self, time, open, high, low, close, volume):
# _time, _close, _open, _high, _low, _extra_info={}
self.kl_type = None
self.time = time
self.close = close
self.open = open
self.high = high
self.low = low
self.volume = volume
self.idx = 0
self.index = 0
self.macd = 0
self.signal = 0
self.macdhist = 0
self.ma5 = 0
self.ma10 = 0
self.ma30 = 0
self.ma50 = 0
self.ma200 = 0
self.ma250 = 0
self.rsi = 0
self.volume_ratio = 0
self.bbp120 = 0
self.bbp365 = 0
self.bb120 = 0
self.bb365 = 0
# === 新增：K线类型 ===
self.kline_type = None # K线类型：大阳线、大阴线、小阳线、小阴线
# === 新增：实时分型相关属性 ===
self.pre = None # 前一根K线
self.next = None # 后一根K线
self.fx_type = Chan_FX_TYPE.UNKNOWN # 分型类型：0=无分型，1=顶分型，-1=底分型
self.fx_strength = 0 # 分型强度：0-100
self.fx_confirmed = False # 分型是否确认
self.klu_type = None
self.cal_klu_min_max()
self.body = abs(self.close - self.open)
self.upper_shadow = self.high - max(self.close, self.open)
self.lower_shadow = min(self.close, self.open) - self.low
self.body_ratio = self.body / self.open
self.upper_shadow_ratio = self.upper_shadow / self.open
self.lower_shadow_ratio = self.lower_shadow / self.open
self.candle_dir = Chan_K_DIR.CROSS if self.close == self.open else Chan_K_DIR.BULL if self.close > self.open else Chan_K_DIR.BEAR
self.range = self.high - self.low
self.strength = 0 if self.candle_dir == Chan_K_DIR.CROSS else self.cal_klu_strength()
#print(self.open, self.close, self.high, self.low, self.candle_dir, self.strength)
def cal_klu_strength(self):
strength = 0
if range != 0:
if self.candle_dir == Chan_K_DIR.BULL and (self.upper_shadow + self.lower_shadow) != 0:
strength += self.body / self.range
strength += self.body / (self.upper_shadow + self.lower_shadow)
elif self.candle_dir == Chan_K_DIR.BEAR and (self.upper_shadow + self.lower_shadow) != 0:
strength -=self.body / self.range
strength -= self.body / (self.upper_shadow + self.lower_shadow)
#print(self.time, self.body, self.range, self.upper_shadow, self.lower_shadow, self.candle_dir, strength)
return strength
return strength
def cal_klu_min_max(self):
"""
计算K线类型：大阳线、大阴线、小阳线、小阴线
"""
if self.open <= 0: # 避免除零错误
self.kline_type = None
return
# 计算涨跌幅
price_change_ratio = (self.close - self.open) / self.open
strength = 0
# 判断K线类型
if price_change_ratio > 0.005: # 涨幅超过2%
self.kline_type = Chan_KLU_TYPE.BigBull
strength += abs(price_change_ratio)
elif price_change_ratio > 0: # 涨幅0-2%
self.kline_type = Chan_KLU_TYPE.SmallBull
strength += abs(price_change_ratio)
elif price_change_ratio < -0.005: # 跌幅超过2%
self.kline_type = Chan_KLU_TYPE.BigBear
strength += abs(price_change_ratio)
elif price_change_ratio < 0: # 跌幅0-2%
self.kline_type = Chan_KLU_TYPE.SmallBear
strength += abs(price_change_ratio)
else: # 开盘价等于收盘价
self.kline_type = Chan_KLU_TYPE.Cross
strength += abs(price_change_ratio)
return strength
def set_next(self, next):
self.next = next
self.update_realtime_analysis()
#if self.fx_type != Chan_FX_TYPE.UNKNOWN and self.fx_strength > 1:
#print(self.index, self.time, self.fx_type, self.fx_confirmed, self.fx_strength)
def set_pre(self, pre):
self.pre = pre
def detect_realtime_fx(self):
"""
实时检测K线分型（不等待KLC确认）
基于原始K线的即时分型识别
"""
if not self.pre or not self.next:
self.fx_type = Chan_FX_TYPE.UNKNOWN
return False
# 顶分型检测
if (self.high > self.pre.high and
self.high > self.next.high):
self.fx_type = Chan_FX_TYPE.TOP
self.fx_confirmed = True
return True
# 底分型检测
elif (self.low < self.pre.low and
self.low < self.next.low):
self.fx_type = Chan_FX_TYPE.BOTTOM
self.fx_confirmed = True
return True
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一致）
核心缠论原理：
- 强分型：出现在笔的末端，能够终结当前笔，标志着趋势转折
- 弱分型：出现在笔的中间，是中继性质，笔还会继续延伸
返回值：
3: 极强分型（笔终结+强确认）
2: 强分型（笔终结）
1: 偏强分型（可能终结笔）
0: 中性分型
-1: 偏弱分型（中继特征明显）
-2: 弱分型（明显中继）
-3: 极弱分型（无效分型）
"""
# 检查是否为分型，且有前后K线数据
if self.fx_type == Chan_FX_TYPE.UNKNOWN:
return 0
if not self.pre or not self.next:
return 100
# === 核心判断：分型在笔中的位置 ===
# 1. 检查这个分型是否能够终结当前笔
is_bi_end = self._check_if_bi_ending_fx()
# 2. 检查分型的后续走势确认
post_fx_confirmation = self._check_post_fx_confirmation()
# 3. 检查分型的标准性和强度
fx_quality = self._check_fx_quality()
# === 综合评分 ===
base_score = 0
# 笔位置是最重要的判断标准
if is_bi_end == 2: # 强烈确认笔终结
base_score = 2
elif is_bi_end == 1: # 可能笔终结
base_score = 1
elif is_bi_end == -1: # 明显中继
base_score = -2
elif is_bi_end == -2: # 强烈中继特征
base_score = -3
else: # 不确定
base_score = 0
# 后续确认调整
base_score += post_fx_confirmation
# 分型质量调整
base_score += fx_quality
# 限制在-3到3范围内
final_score = max(-3, min(3, base_score))
self.fx_strength = final_score
# 转换为0-100分制以保持接口一致性
#self.fx_strength = int((final_score + 3) * 100 / 6) # -3到3映射到0-100
#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):
"""
检查分型是否为笔终结分型
返回值：
2: 强烈确认笔终结
1: 可能笔终结
0: 不确定
-1: 明显中继
-2: 强烈中继特征
"""
# 检查是否有足够的后续数据来判断
if not self.next or not hasattr(self.next, 'next'):
return 0
# 获取分型后的几根K线数据
subsequent_klus = []
temp = self.next
for i in range(2): # 检查后续2根K线
if temp:
subsequent_klus.append(temp)
temp = temp.next if hasattr(temp, 'next') else None
else:
break
if len(subsequent_klus) < 2:
return 0
if self.fx_type == Chan_FX_TYPE.TOP:
return self._check_top_bi_ending(subsequent_klus)
else: # BOTTOM
return self._check_bottom_bi_ending(subsequent_klus)
def _check_top_bi_ending(self, subsequent_klus):
"""检查顶分型是否为笔终结"""
# 强烈笔终结特征：
# 1. 后续K线持续下跌，且跌破关键位置
# 2. 没有新的更高的高点出现
broken_key_levels = 0
new_highs = 0
downward_trend = 0
# 检查关键价位突破
first_low = self.pre.low
middle_low = self.low
key_support = min(first_low, middle_low)
for i, klu in enumerate(subsequent_klus):
# 检查是否跌破关键支撑
if klu.low < key_support:
broken_key_levels += 1
# 检查是否出现新高
if klu.high > self.high:
new_highs += 1
# 检查下跌趋势
if i > 0 and klu.close < subsequent_klus[i-1].close:
downward_trend += 1
# 强烈笔终结：跌破关键位且无新高
if broken_key_levels >= 1 and new_highs == 0 and downward_trend >= 2:
return 2
# 可能笔终结：部分条件满足
if (broken_key_levels >= 1 and new_highs <= 1) or (new_highs == 0 and downward_trend >= 3):
return 1
# 明显中继：出现新高且未跌破关键位
if new_highs >= 2 and broken_key_levels == 0:
return -2
# 中继倾向：出现新高
if new_highs >= 1:
return -1
return 0
def _check_bottom_bi_ending(self, subsequent_klus):
"""检查底分型是否为笔终结"""
# 强烈笔终结特征：
# 1. 后续K线持续上涨，且突破关键位置
# 2. 没有新的更低的低点出现
broken_key_levels = 0
new_lows = 0
upward_trend = 0
# 检查关键价位突破
first_high = self.pre.high
middle_high = self.high
key_resistance = max(first_high, middle_high)
for i, klu in enumerate(subsequent_klus):
# 检查是否突破关键阻力
if klu.high > key_resistance:
broken_key_levels += 1
# 检查是否出现新低
if klu.low < self.low:
new_lows += 1
# 检查上涨趋势
if i > 0 and klu.close > subsequent_klus[i-1].close:
upward_trend += 1
# 强烈笔终结：突破关键位且无新低
if broken_key_levels >= 1 and new_lows == 0 and upward_trend >= 2:
return 2
# 可能笔终结：部分条件满足
if (broken_key_levels >= 1 and new_lows <= 1) or (new_lows == 0 and upward_trend >= 3):
return 1
# 明显中继：出现新低且未突破关键位
if new_lows >= 2 and broken_key_levels == 0:
return -2
# 中继倾向：出现新低
if new_lows >= 1:
return -1
return 0
def _check_post_fx_confirmation(self):
"""
检查分型后的走势确认
返回值：-1到1的调整分数
"""
if not self.next:
return 0
score = 0
# 检查第三根K线的确认
third_klu = self.next
if self.fx_type == Chan_FX_TYPE.TOP:
# 顶分型：第三根K线应该走弱
middle_price = (self.high + self.low) / 2
if third_klu.close < middle_price:
score += 0.5
if third_klu.low < self.pre.low: # 跌破第一根K线低点
score += 0.5
if third_klu.close < third_klu.open and abs(third_klu.close - third_klu.open) > abs(self.close - self.open) * 0.5:
score += 0.3 # 明显阴线
else: # BOTTOM
# 底分型：第三根K线应该走强
middle_price = (self.high + self.low) / 2
if third_klu.close > middle_price:
score += 0.5
if third_klu.high > self.pre.high: # 突破第一根K线高点
score += 0.5
if third_klu.close > third_klu.open and abs(third_klu.close - third_klu.open) > abs(self.close - self.open) * 0.5:
score += 0.3 # 明显阳线
return min(1, max(-1, score))
def _check_fx_quality(self):
"""
检查分型本身的质量
返回值：-1到1的调整分数
"""
score = 0
# 检查分型的标准性
if self.fx_type == Chan_FX_TYPE.TOP:
# 高点突出程度
high_diff1 = (self.high - self.pre.high) / self.high if self.high > 0 else 0
high_diff2 = (self.high - self.next.high) / self.high if self.high > 0 else 0
min_diff = min(high_diff1, high_diff2)
if min_diff > 0.03: # 非常突出
score += 0.5
elif min_diff > 0.01: # 比较突出
score += 0.2
elif min_diff < 0.003: # 不够突出
score -= 0.5
else: # BOTTOM
# 低点突出程度
low_diff1 = (self.pre.low - self.low) / self.pre.low if self.pre.low > 0 else 0
low_diff2 = (self.next.low - self.low) / self.next.low if self.next.low > 0 else 0
min_diff = min(low_diff1, low_diff2)
if min_diff > 0.03: # 非常突出
score += 0.5
elif min_diff > 0.01: # 比较突出
score += 0.2
elif min_diff < 0.003: # 不够突出
score -= 0.5
# 检查量价配合
avg_volume = self._get_avg_volume(lookback=5)
if avg_volume > 0:
volume_ratio = self.volume / avg_volume
if volume_ratio > 1.5:
score += 0.3
elif volume_ratio < 0.7:
score -= 0.2
return min(1, max(-1, score))
def _get_avg_volume(self, lookback=5):
"""获取前N根K线平均成交量"""
volumes = []
temp = self.pre
for i in range(lookback):
if temp:
volumes.append(temp.volume)
temp = temp.pre if hasattr(temp, 'pre') else None
else:
break
return sum(volumes) / len(volumes) if volumes else self.volume
def get_fx_signal(self):
"""
获取分型交易信号
返回: (信号类型, 强度, 建议)
"""
if not self.fx_confirmed:
return ("无信号", 0, "等待分型确认")
strength_level = "弱"
if self.fx_strength >= 80:
strength_level = "极强"
elif self.fx_strength >= 65:
strength_level = "强"
elif self.fx_strength >= 50:
strength_level = "中等"
if self.fx_type == Chan_FX_TYPE.TOP:
signal_type = f"{strength_level}顶分型"
if self.fx_strength >= 65:
suggestion = "考虑减仓或止盈"
else:
suggestion = "谨慎观望"
else:
signal_type = f"{strength_level}底分型"
if self.fx_strength >= 65:
suggestion = "考虑建仓或加仓"
else:
suggestion = "谨慎观望"
return (signal_type, self.fx_strength, suggestion)
def update_realtime_analysis(self):
"""
更新实时分析（在每根K线完成时调用）
"""
self.detect_realtime_fx()
if self.fx_confirmed:
self.calculate_realtime_fx_strength()
def set_idx(self, idx):
self.idx = idx
self.index = idx
def set_indicators(self, item):
self.macd = float(item['macd']) if 'macd' in item and item['macd'] else 0
self.signal = float(item['macdsignal']) if 'macdsignal' in item and item['macdsignal'] else 0
self.macdhist = float(item['macdhist']) if 'macdhist' in item and item['macdhist'] else 0
self.ma5 = float(item['ma5']) if 'ma5' in item and item['ma5'] else 0
self.ma10 = float(item['ma10']) if 'ma10' in item and item['ma10'] else 0
self.ma30 = float(item['ma30']) if 'ma30' in item and item['ma30'] else 0
# 安全检查 ma250、ma50 和 ma200
self.ma250 = float(item['ma250']) if 'ma250' in item and item['ma250'] else 0
self.ma50 = float(item['ma50']) if 'ma50' in item and item['ma50'] else 0
self.ma200 = float(item['ma200']) if 'ma200' in item and item['ma200'] else 0
self.rsi = float(item['rsi']) if 'rsi' in item and item['rsi'] else 0
self.volume_ratio = float(item['volume_ratio']) if 'volume_ratio' in item and item['volume_ratio'] else 0
self.bbp120 = float(item['bbp120']) if 'bbp120' in item and item['bbp120'] else 0
self.bbp365 = float(item['bbp365']) if 'bbp365' in item and item['bbp365'] else 0
self.bb120 = float(item['bb120']) if 'bb120' in item and item['bb120'] else 0
self.bb365 = float(item['bb365']) if 'bb365' in item and item['bb365'] else 0
# 设置指标后更新实时分析
self.update_realtime_analysis()
def get_feature_data(self):
features = dict()
features['klu_close'] = self.close
features['klu_open'] = self.open
features['klu_high'] = self.high
features['klu_low'] = self.low
features['klu_volume'] = self.volume
features['klu_index'] = self.index
features['klu_macd'] = self.macd
features['klu_signal'] = self.signal
features['klu_macdhist'] = self.macdhist
features['klu_ma5'] = self.ma5
features['klu_ma10'] = self.ma10
features['klu_ma30'] = self.ma30
features['klu_ma50'] = self.ma50
features['klu_ma200'] = self.ma200
features['klu_ma250'] = self.ma250
features['klu_rsi'] = self.rsi
features['klu_volume_ratio'] = self.volume_ratio
# === 新增：实时分型特征 ===
# 将枚举转换为数值：UNKNOWN=0, TOP=1, BOTTOM=-1
if self.fx_type == Chan_FX_TYPE.TOP:
fx_type_value = 1
elif self.fx_type == Chan_FX_TYPE.BOTTOM:
fx_type_value = -1
else:
fx_type_value = 0
features['klu_fx_type'] = fx_type_value
features['klu_fx_strength'] = self.fx_strength
features['klu_fx_confirmed'] = 1 if self.fx_confirmed else 0
return features
Reference in New Issue View Git Blame Copy Permalink