import copy from typing import Dict, Optional from ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR, Chan_BI_DIR, Chan_KLC_FX, Chan_K_DIR, Chan_MACD_STATE, Chan_PRICE_TREND, Chan_EMA_POS, Chan_EMA_SEMANTIC, Chan_BSP_TYPE import ChanKLU import ChanCTime # 根据结合律合并K线后的K线 class ChanKLC(): def __init__(self, klu: ChanKLU, index, ddir=Chan_KLINE_DIR.UP): self.start_time = klu.time self.end_time = None self.high = klu.high self.low = klu.low self.dir = ddir self.index = index self.klu_list = [] self.add_klu(klu) self.fx = Chan_FX_TYPE.UNKNOWN self.next = None self.pre = None self.start_klu = klu self.end_klu = None self.state = "00" self.open = klu.open self.close = klu.close self.volume = klu.volume self.bi = None self.distance = 0 self.klc_fx_type = Chan_KLC_FX.UNKNOWN self.rsi = klu.rsi self.volume_ratio = klu.volume_ratio self.macdhist = klu.macdhist self.body = klu.body self.upper_shadow = klu.upper_shadow self.lower_shadow = klu.lower_shadow self.body_ratio = klu.body_ratio self.upper_shadow_ratio = klu.upper_shadow_ratio self.lower_shadow_ratio = klu.lower_shadow_ratio self.candle_dir = klu.candle_dir self.range = klu.range self.bb_out = True self.macd = 0 self.signal = 0 self.state = Chan_MACD_STATE.UNKNOWN self.continue_div = False self.separate_div = False self.ema24 = klu.ema24 self.ema26 = klu.ema26 self.ema52 = klu.ema52 self.ema104 = klu.ema104 self.ema156 = klu.ema156 self.ema208 = klu.ema208 self.trend = Chan_PRICE_TREND.UNKNOWN self.exception = klu.exception self.klc_dir = Chan_KLINE_DIR.UP if klu.close > klu.open else Chan_KLINE_DIR.DOWN self.ema_dir = klu.ema_dir self.bsp = False self.bsp_type = Chan_BSP_TYPE.NONE # EMA状态字典：key为EMA名称，value为 {'pos': Chan_EMA_POS, 'semantic': Chan_EMA_SEMANTIC} self.ema_status = {} # 向后兼容：保留 ema52_status 和 ema52_pos self.ema52_status = 0 self.ema52_pos = Chan_EMA_POS.UNKNOWN self.bb2633upper = klu.bb2633upper self.bb2633lower = klu.bb2633lower self.bb2633middle = klu.bb2633middle self.ema5 = klu.ema5 self.ma5 = klu.ma5 # ==================== EMA 通用计算方法 ==================== @staticmethod def cal_ema_pos(high, low, close, ema_value, threshold=0): """ 计算K线与任意EMA的客观位置关系（与趋势方向无关，支持threshold容差） 参数: high, low, close: K线的高低收盘价 ema_value: EMA的值 threshold: 容差值（绝对值），在此范围内视为"接近/触碰" 例如 BTC 价格 $100,000 时 threshold=100 表示差100点视为触碰 返回: Chan_EMA_POS 枚举值 判断逻辑（以threshold=100, ema=97000为例）： ema_zone = [96900, 97100] （EMA上下各扩展threshold） ABOVE: low > 97100 K线完全在zone上方（远离EMA） NEAR_ABOVE: 97000 < low <= 97100 K线在上方但下影线进入zone（接近EMA） CROSS_CLOSE_ABOVE: close > 97000, low <= 97000 K线穿越EMA，收盘在上方 ON_EMA: abs(close - 97000) <= 100 收盘价在zone内 CROSS_CLOSE_BELOW: close < 97000, high >= 97000 K线穿越EMA，收盘在下方 NEAR_BELOW: 96900 <= high < 97000 K线在下方但上影线进入zone（接近EMA） BELOW: high < 96900 K线完全在zone下方（远离EMA） """ if ema_value is None or ema_value == 0: return Chan_EMA_POS.UNKNOWN ema_upper = ema_value + threshold # EMA zone 上界 ema_lower = ema_value - threshold # EMA zone 下界 # 1. 收盘价在EMA附近（zone内） if threshold > 0 and abs(close - ema_value) <= threshold: # 收盘价在zone内，但还需要看是否有实际穿越 if low <= ema_value and close >= ema_value: return Chan_EMA_POS.CROSS_CLOSE_ABOVE # 实际穿越了精确EMA线 elif high >= ema_value and close <= ema_value: return Chan_EMA_POS.CROSS_CLOSE_BELOW return Chan_EMA_POS.ON_EMA # 2. K线实际穿越了精确的EMA线 if close > ema_value and low <= ema_value: return Chan_EMA_POS.CROSS_CLOSE_ABOVE if close < ema_value and high >= ema_value: return Chan_EMA_POS.CROSS_CLOSE_BELOW if close == ema_value: return Chan_EMA_POS.ON_EMA # 3. 没有实际穿越，检查是否"接近"（在threshold zone内） if close > ema_value: # K线在EMA上方 if threshold > 0 and low <= ema_upper: return Chan_EMA_POS.NEAR_ABOVE # 下影线进入zone，接近但未触碰 return Chan_EMA_POS.ABOVE # 远离EMA else: # K线在EMA下方 if threshold > 0 and high >= ema_lower: return Chan_EMA_POS.NEAR_BELOW # 上影线进入zone，接近但未触碰 return Chan_EMA_POS.BELOW # 远离EMA @staticmethod def cal_ema_semantic(ema_pos, kline_dir, ema_dir): """ 根据客观位置 + K线方向 + 趋势方向，计算语义状态 参数: ema_pos: Chan_EMA_POS 客观位置 kline_dir: Chan_KLINE_DIR K线方向 (UP/DOWN/COMBINE/INCLUDED) ema_dir: int 趋势方向 (1=多头, -1=空头, 0=盘整) 返回: Chan_EMA_SEMANTIC 枚举值 语义含义（以多头为例，空头完全对称）： TOUCH_HOLD: 触碰EMA，收盘守住趋势侧（支撑/压力有效） BREAK: 穿越EMA，收盘在逆势侧（支撑/压力失败） DEEP_COUNTER: 完全在EMA逆势侧（深度回调/反抽） TOUCH_FAIL: 逆势触碰EMA但未穿越（反弹/反抽力度不足） RECOVER: 逆势后穿越EMA回到趋势侧（收复EMA） TREND_SIDE: 完全在EMA趋势侧（正常运行） STRONG_TREND: 顺势K线完全在EMA趋势侧（强势，远未及EMA） WEAK_COUNTER: 逆势K线完全在EMA逆势侧（弱势，远未到EMA） """ if ema_pos == Chan_EMA_POS.UNKNOWN: return Chan_EMA_SEMANTIC.NEUTRAL # 统一处理：将多头/盘整和空头映射到同一套逻辑 # is_bull=True 时，"趋势侧"=上方，"逆势侧"=下方 # is_bull=False时，"趋势侧"=下方，"逆势侧"=上方 is_bull = ema_dir >= 0 # 多头和盘整都按多头逻辑处理 # K线是否是顺势方向（多头下UP为顺势，空头下DOWN为顺势） is_trend_kline = (kline_dir == Chan_KLINE_DIR.UP) if is_bull else (kline_dir == Chan_KLINE_DIR.DOWN) is_counter_kline = (kline_dir == Chan_KLINE_DIR.DOWN) if is_bull else (kline_dir == Chan_KLINE_DIR.UP) # 位置映射：多头下 ABOVE=趋势侧, BELOW=逆势侧; 空头反过来 trend_side = Chan_EMA_POS.ABOVE if is_bull else Chan_EMA_POS.BELOW counter_side = Chan_EMA_POS.BELOW if is_bull else Chan_EMA_POS.ABOVE near_trend = Chan_EMA_POS.NEAR_ABOVE if is_bull else Chan_EMA_POS.NEAR_BELOW near_counter = Chan_EMA_POS.NEAR_BELOW if is_bull else Chan_EMA_POS.NEAR_ABOVE cross_to_trend = Chan_EMA_POS.CROSS_CLOSE_ABOVE if is_bull else Chan_EMA_POS.CROSS_CLOSE_BELOW cross_to_counter = Chan_EMA_POS.CROSS_CLOSE_BELOW if is_bull else Chan_EMA_POS.CROSS_CLOSE_ABOVE # COMBINE / INCLUDED 方向：只看位置，不区分强弱 if not is_trend_kline and not is_counter_kline: if ema_pos == trend_side: return Chan_EMA_SEMANTIC.TREND_SIDE elif ema_pos in (near_trend, cross_to_trend, Chan_EMA_POS.ON_EMA): return Chan_EMA_SEMANTIC.APPROACHING elif ema_pos in (near_counter, cross_to_counter): return Chan_EMA_SEMANTIC.APPROACHING elif ema_pos == counter_side: return Chan_EMA_SEMANTIC.DEEP_COUNTER return Chan_EMA_SEMANTIC.NEUTRAL # 逆势K线（多头下的下跌K线 / 空头下的上涨K线） if is_counter_kline: if ema_pos == trend_side: return Chan_EMA_SEMANTIC.STRONG_TREND # 逆势K线仍在趋势侧（回调很浅） elif ema_pos == near_trend: return Chan_EMA_SEMANTIC.APPROACHING # 接近EMA，即将测试支撑/压力 elif ema_pos == cross_to_trend: return Chan_EMA_SEMANTIC.TOUCH_HOLD # 触碰EMA后守住趋势侧 elif ema_pos == Chan_EMA_POS.ON_EMA: return Chan_EMA_SEMANTIC.TOUCH_HOLD # 收盘在EMA附近，视为守住 elif ema_pos == cross_to_counter: return Chan_EMA_SEMANTIC.BREAK # 穿越EMA到逆势侧 elif ema_pos == near_counter: return Chan_EMA_SEMANTIC.BREAK # 接近EMA但收盘在逆势侧，也视为击穿 elif ema_pos == counter_side: return Chan_EMA_SEMANTIC.DEEP_COUNTER # 完全在逆势侧 # 顺势K线（多头下的上涨K线 / 空头下的下跌K线） if is_trend_kline: if ema_pos == counter_side: return Chan_EMA_SEMANTIC.WEAK_COUNTER # 顺势K线却在逆势侧（弱势） elif ema_pos == near_counter: return Chan_EMA_SEMANTIC.APPROACHING # 从逆势侧接近EMA elif ema_pos == cross_to_counter: return Chan_EMA_SEMANTIC.TOUCH_FAIL # 触碰EMA但未穿越回趋势侧 elif ema_pos == Chan_EMA_POS.ON_EMA: return Chan_EMA_SEMANTIC.TOUCH_FAIL # 收盘在EMA附近，未确认突破 elif ema_pos == cross_to_trend: return Chan_EMA_SEMANTIC.RECOVER # 从逆势侧穿越回趋势侧 elif ema_pos == near_trend: return Chan_EMA_SEMANTIC.RECOVER # 接近趋势侧（刚收复EMA附近） elif ema_pos == trend_side: return Chan_EMA_SEMANTIC.TREND_SIDE # 完全在趋势侧（正常） return Chan_EMA_SEMANTIC.NEUTRAL @staticmethod def semantic_to_int(semantic): """将 Chan_EMA_SEMANTIC 枚举转换为整数，兼容旧的 ema52_status 数值""" mapping = { Chan_EMA_SEMANTIC.TOUCH_HOLD: 1, Chan_EMA_SEMANTIC.BREAK: 2, Chan_EMA_SEMANTIC.DEEP_COUNTER: 3, Chan_EMA_SEMANTIC.TOUCH_FAIL: 4, Chan_EMA_SEMANTIC.RECOVER: 5, Chan_EMA_SEMANTIC.TREND_SIDE: 6, Chan_EMA_SEMANTIC.STRONG_TREND: 7, Chan_EMA_SEMANTIC.WEAK_COUNTER: 8, Chan_EMA_SEMANTIC.APPROACHING: 9, Chan_EMA_SEMANTIC.NEUTRAL: 0, } return mapping.get(semantic, 0) # threshold_pct: 阈值百分比，用于自动计算绝对阈值 # 例如 0.001 表示 EMA 值的 0.1%，BTC $100,000 时 threshold = $100 threshold_pct = 0.001 def set_bsp_type(self, bsp_type): if bsp_type and bsp_type != Chan_BSP_TYPE.NONE: self.bsp_type = bsp_type self.bsp = True def cal_all_ema_status(self): """ 统一计算所有EMA与K线的位置关系和语义状态 threshold 自动按 EMA 值的百分比计算（cls.threshold_pct，默认0.1%） - BTC $100,000 时：threshold ≈ $100 - ETH $3,000 时：threshold ≈ $3 - SOL $200 时：threshold ≈ $0.2 结果存储在 self.ema_status 字典中，格式： { 'ema24': {'pos': Chan_EMA_POS, 'semantic': Chan_EMA_SEMANTIC, 'value': float, 'threshold': float}, 'ema52': {...}, ... } 同时保持向后兼容：self.ema52_pos 和 self.ema52_status """ ema_configs = { 'ema24': self.ema24, 'ema52': self.ema52, 'ema104': self.ema104, 'ema156': self.ema156, 'ema208': self.ema208, } self.ema_status = {} for name, value in ema_configs.items(): # 按 EMA 值的百分比自动计算阈值 threshold = abs(value) * self.threshold_pct if value and self.threshold_pct > 0 else 0 pos = ChanKLC.cal_ema_pos(self.high, self.low, self.close, value, threshold) semantic = ChanKLC.cal_ema_semantic(pos, self.dir, self.ema_dir) self.ema_status[name] = { 'pos': pos, 'semantic': semantic, 'value': value, 'threshold': threshold, } # 向后兼容 self.ema52_pos = self.ema_status['ema52']['pos'] self.ema52_status = ChanKLC.semantic_to_int(self.ema_status['ema52']['semantic']) def get_ema_pos(self, ema_name): """获取指定EMA的客观位置，如 klc.get_ema_pos('ema24')""" if ema_name in self.ema_status: return self.ema_status[ema_name]['pos'] return Chan_EMA_POS.UNKNOWN def check_ema_pos(self): if len(self.ema_status) > 0: for ema_name, pos in self.ema_status.items(): #print(self.end_time, ema_name, pos['pos']) if ((self.klc_fx_type == Chan_KLC_FX.TOP1 or self.klc_fx_type == Chan_KLC_FX.TOP2) and pos['pos'] == Chan_EMA_POS.CROSS_CLOSE_BELOW) or ((self.klc_fx_type == Chan_KLC_FX.BOTTOM1 or self.klc_fx_type == Chan_KLC_FX.BOTTOM2) and pos['pos'] == Chan_EMA_POS.CROSS_CLOSE_ABOVE): #print("---------------------") return ema_name return None def get_ema_semantic(self, ema_name): """获取指定EMA的语义状态，如 klc.get_ema_semantic('ema52')""" if ema_name in self.ema_status: return self.ema_status[ema_name]['semantic'] return Chan_EMA_SEMANTIC.NEUTRAL def set_trend(self, trend): self.trend = trend def to_string(self): out = "" start = self.start_time if self.start_time is not None else "" end = self.end_time if self.end_time is not None else "" price_diff = getattr(self, 'price_diff', None) out += str(start) + " " + str(end) + " " + str(self.close) + " " + str(self.ema24) + " " + str(self.ema52) + " " + str(self.trend) + " " + str(self.close - self.ema52) return out def set_klc_fx_type(self, klc_fx_type): #print(self.start_time, klc_fx_type, self.get_feature_data()['klu_macd'], self.get_feature_data()['klu_macdhist'], self.get_feature_data()['klu_rsi']) self.klc_fx_type = klc_fx_type #self.cal_fx() ema_name = self.check_ema_pos() hist_div = abs(self.macdhist - self.next.macdhist) #print(self.end_time, self.dir, abs(self.macdhist), hist_div) #if ema_name: #print(self.end_time, ema_name, self.ema_status[ema_name]['semantic'], hist_div) #self.cal_bb_out() #print(self.pre.start_time, self.next.end_time, self.klc_fx_type) def add_klu(self, klu): self.klu_list.append(klu) def set_end_klu(self, klu): self.end_klu = klu self.end_time = klu.time self.close = klu.close for klu in self.klu_list: if klu.exception: self.exception = True print(klu.time, "exception") if klu.separate_div > 0: self.separate_div = True if klu.continue_div: self.continue_div = klu.continue_div if klu.macd_state != Chan_MACD_STATE.UNKNOWN: self.state = klu.macd_state klu.set_klc(self) self.klc_dir = Chan_KLINE_DIR.UP if self.close > self.open else Chan_KLINE_DIR.DOWN self.cal_indicators() self.cal_all_ema_status() if self.open > self.high: self.open = self.high if self.close > self.high: self.close = self.high if self.close < self.low: self.close = self.low if self.close > self.high: self.close = self.high def cal_fx(self): if self.klc_fx_type == Chan_KLC_FX.TOP1 or self.klc_fx_type == Chan_KLC_FX.TOP2: #print(self.end_time, self.fx, self.macd, self.macdhist, len(self.klu_list)) if self.state == Chan_MACD_STATE.HIGH_EMPTY and self.macd > 0: #print(self.end_time, self.state, self.macd, self.klc_fx_type) self.klc_fx_type = Chan_KLC_FX.TOP6 if self.separate_div or self.continue_div: self.klc_fx_type = Chan_KLC_FX.TOP7 if self.signal > 0 and self.macd > self.signal: self.klc_fx_type = Chan_KLC_FX.TOP8 else: if self.klc_fx_type == Chan_KLC_FX.BOTTOM1 or self.klc_fx_type == Chan_KLC_FX.BOTTOM2: if self.macdhist > 0 and self.macd < 0: self.klc_fx_type = Chan_KLC_FX.BOTTOM5 return if self.state == Chan_MACD_STATE.HIGH_EMPTY and self.macd < 0: self.klc_fx_type = Chan_KLC_FX.BOTTOM6 #print(self.end_time, self.state, self.macd, self.klc_fx_type) if self.separate_div or self.continue_div: self.klc_fx_type = Chan_KLC_FX.BOTTOM7 if self.signal < 0 and self.macd < self.signal: self.klc_fx_type = Chan_KLC_FX.BOTTOM8 def cal_bb_out(self): for klu in self.klu_list: if self.klc_fx_type == Chan_KLC_FX.TOP1 or self.klc_fx_type == Chan_KLC_FX.TOP2: #print(self.start_time, self.klc_fx_type, klu.high, klu.bb52upper, self.macd, self.next.macd, klu.time) if self.high >= klu.bb52upper and klu.bb52upper > 0 and self.next and self.high > self.next.high: self.klc_fx_type = Chan_KLC_FX.TOP4 print(self.end_time, self.klc_fx_type) if self.klc_fx_type == Chan_KLC_FX.BOTTOM1 or self.klc_fx_type == Chan_KLC_FX.BOTTOM2: #print(self.start_time, self.klc_fx_type, klu.low, klu.bb52lower, self.macd, self.next.macd, klu.time) if self.low <= klu.bb52lower and klu.bb52lower > 0 and self.next and self.low < self.next.low: self.klc_fx_type = Chan_KLC_FX.BOTTOM4 print(self.end_time, self.klc_fx_type) def cal_indicators(self): for index in range(1, len(self.klu_list)): self.volume += self.klu_list[index].volume self.rsi += self.klu_list[index].rsi self.volume_ratio += self.klu_list[index].volume_ratio self.macdhist += self.klu_list[index].macdhist self.ema26 += self.klu_list[index].ema26 self.ema24 += self.klu_list[index].ema24 self.ema52 += self.klu_list[index].ema52 self.ema104 += self.klu_list[index].ema104 self.ema156 += self.klu_list[index].ema156 self.ema208 += self.klu_list[index].ema208 self.bb2633upper += self.klu_list[index].bb2633upper self.bb2633lower += self.klu_list[index].bb2633lower self.bb2633middle += self.klu_list[index].bb2633middle self.ma5 += self.klu_list[index].ma5 self.ema5 += self.klu_list[index].ema5 if self.ema_dir != self.klu_list[index].ema_dir: self.ema_dir = 0 n = len(self.klu_list) self.rsi = self.rsi / n self.volume_ratio = self.volume_ratio / n self.volume = self.volume / n self.macdhist = self.macdhist / n self.ema26 = self.ema26 / n self.ema24 = self.ema24 / n self.ema52 = self.ema52 / n self.ema104 = self.ema104 / n self.ema156 = self.ema156 / n self.ema208 = self.ema208 / n self.ma5 = self.ma5 / n self.ema5 = self.ema5 / n self.bb2633upper = self.bb2633upper / n self.bb2633lower = self.bb2633lower / n self.bb2633middle = self.bb2633middle / n if len(self.klu_list) > 0: self.macd = self.klu_list[-1].macd self.signal = self.klu_list[-1].signal 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 def set_next(self, klc): self.next = klc def set_pre(self, klc): self.pre = klc def set_state(self, state): self.state = state def check_klu_included(self, klu): if self.high >= klu.high: # high大于，low小于，左包含 if self.low <= klu.low: self.add_klu(klu=klu) # gn>gn-1 if self.dir == Chan_KLINE_DIR.UP: # UP -> max(dn) self.low = klu.low else: # DOWN -> min(gn) self.high = klu.high #self.print(klu, "Z") return True # high大于，low大于，不包含 else: # if self.low > klu.low # high相等，右包含 if self.high == klu.high: self.add_klu(klu=klu) # UP -> max(gn) if self.dir == Chan_KLINE_DIR.UP: self.high = klu.high else: # DOWN -> min(dn) self.low = klu.low return True else: return False else: # high小于，low大于，右包含 if self.low >= klu.low: self.add_klu(klu=klu) # gn>gn-1 if self.dir == Chan_KLINE_DIR.UP: # UP -> max(gn) self.high = klu.high else: # DOWN -> min(dn) self.low = klu.low #self.print(klu, "Y") return True else: # high小于，low小于，不包含 return False def set_fx(self, fx: Chan_FX_TYPE): self.fx = fx def cal_invisible(self): if self.fx == Chan_FX_TYPE.TOP: if self.macdhist < 0 and self.macd > 0: self.klc_fx_type = Chan_KLC_FX.TOP5 else: if self.fx == Chan_FX_TYPE.BOTTOM: if self.macdhist > 0 and self.macd < 0: self.klc_fx_type = Chan_KLC_FX.BOTTOM5 def set_pre_fx(self): if self.pre and self.pre.pre: self.pre.fx = self.check_fx(self.pre.pre, self.pre) def check_fx(self, k1, k2): if k2.high > k1.high and k2.high > self.high: return Chan_FX_TYPE.TOP elif k2.low < k1.low and k2.low < self.low: return Chan_FX_TYPE.BOTTOM else: return Chan_FX_TYPE.UNKNOWN def set_bi(self, bi): self.bi = bi self.distance = self.index - bi.start_klc.index #print(self.start_time, self.distance, bi.index, bi.dir)