from ChanEnum import Chan_FX_TYPE, Chan_KLU_TYPE, Chan_K_DIR, Chan_MACD_STATE, Chan_MACDHIST_STATE, Chan_PRICE_TREND, Chan_KLU_PATTERN 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.klc = None self.rsi = 0 self.volume_ratio = 0 self.bb52upper = 0 self.bb52lower = 0 # === 新增：K线类型 === self.kline_type = None # K线类型：大阳线、大阴线、小阳线、小阴线 self.pattern = Chan_KLU_PATTERN.UNKNOWN # === 新增：实时分型相关属性 === 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.range = self.high - self.low 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.range self.upper_shadow_ratio = self.upper_shadow / self.body self.lower_shadow_ratio = self.lower_shadow / self.body self.exception = False self.cal_exception() 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.continue_div = 0 self.separate_div = 0 self.near0_return = 0 self.ema52 = 0 self.ema24 = 0 self.macd_slop = 0 self.signal_slop = 0 self.hist_slop = 0 self.hist_state = Chan_MACDHIST_STATE.UNKNOWN self.macd_state = Chan_MACD_STATE.UNKNOWN self.macd_hist_gap = 0 self.trend = Chan_PRICE_TREND.UNKNOWN # === 归零轴细化与模式/背离 === self.zero_axis = False # 是否归零轴（穿越或接近） self.zero_axis_state = "none" # {none,crossing,near} self.zero_axis_side = 0 # 1:above, -1:under, 0:none self.zero_axis_score = 0 # 0-100 综合评分 self.mode1_touch_ema52 = False # 单边后触碰EMA52 self.mode2_fast_to_zero = False # 快线向零收敛 self.mode3_double_tf = False # 双周期归零（近似占位，由上层填充高周期确认） self.mode3_dir = "none" # {long_strong_rebound, short_strong_rebound, none} self.mode4_touch52_no_zero = False # 先触碰EMA52但黄白线未归零 self.div_type = "none" # {bearish, bullish, hidden_bearish, hidden_bullish, none} self.div_score = 0.0 # 背离强度(0-100) #print(self.open, self.close, self.high, self.low, self.candle_dir, self.strength) def set_macd_state(self, state): self.macd_state = state def set_pattern(self, pattern): self.pattern = pattern def cal_exception(self): if self.upper_shadow_ratio > 5 or self.lower_shadow_ratio > 5: self.exception = True #print(self.time, self.upper_shadow_ratio, self.lower_shadow_ratio, self.body, self.lower_shadow, self.upper_shadow, self.high, self.low, self.close, self.open) self.exception = False def set_trend(self, trend): self.trend = trend def set_next(self, next): self.next = next #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 set_klc(self, klc): self.klc = klc def set_histset(self, histset): """设置HistSet关联""" self.histset = histset def set_seg(self, seg): """设置Seg关联""" self.seg = seg def set_unittf(self, unittf): """设置UnitTF关联""" self.unittf = unittf 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.ema52 = float(item['ema52']) if 'ema52' in item and item['ema52'] else 0 self.ema24 = float(item['ema24']) if 'ema24' in item and item['ema24'] 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.bb52upper = float(item['bb52upper']) if 'bb52upper' in item and item['bb52upper'] else 0 self.bb52lower = float(item['bb52lower']) if 'bb52lower' in item and item['bb52lower'] else 0 def cal_macd_state(self): # 按定义精简实现：优先级 CROSS0 > 位置(HIGH/HE/RETURN_ZERO) > NEAR0 > UNKNOWN # 首条或缺前一根 if not hasattr(self, 'pre') or self.pre is None: self.macd_state = Chan_MACD_STATE.START return self.macd_state # 基本校验 if (self.macd == 0 and self.signal == 0 and self.macdhist == 0) or self.ema52 == 0: self.macd_state = Chan_MACD_STATE.UNKNOWN return self.macd_state # 归零轴判断 if self.signal > 0: if self.macd < self.signal: if 0 < self.low - self.ema52 < 100: self.near0_return = 0 elif self.close > self.ema52 and self.low < self.ema52 and self.open > self.ema52: self.near0_return = 7 elif self.close < self.ema52 and self.open > self.ema52 and self.high > self.ema52 and self.low < self.ema52: self.near0_return = 0 elif self.close < self.ema52 and self.open < self.ema52 and self.high > self.ema52 and self.low < self.ema52: self.near0_return = 0 elif self.close > self.ema52 and self.open > self.ema52 and self.high > self.ema52 and self.low < self.ema52: self.near0_return = 0 else: if self.macd > self.signal: if 0 < self.ema52 - self.high < 100: self.near0_return = 0 elif self.close < self.ema52 and self.high > self.ema52 and self.open < self.ema52: self.near0_return = 0 elif self.close < self.ema52 and self.open > self.ema52 and self.high > self.ema52 and self.low < self.ema52: self.near0_return = 81 elif self.close > self.ema52 and self.open < self.ema52 and self.high > self.ema52 and self.low < self.ema52: self.near0_return = 0 elif self.close > self.ema52 and self.high > self.ema52 and self.open > self.ema52 and self.low < self.ema52: self.near0_return = 0 # CROSS0 仅以 Signal 穿越零轴判定 if self.pre.signal >= 0 and self.signal < 0: self.macd_state = Chan_MACD_STATE.CROSS0_DOWN return self.macd_state if self.pre.signal <= 0 and self.signal > 0: self.macd_state = Chan_MACD_STATE.CROSS0_UP return self.macd_state # 穿零轴后的形态：缠绕/倒挂（基于前一状态为CROSS0_*） if self.pre.macd_state == Chan_MACD_STATE.CROSS0_UP or self.pre.macd_state == Chan_MACD_STATE.CROSS0_DOWN: direction = 1 if self.pre.macd_state == Chan_MACD_STATE.CROSS0_UP else -1 hist_same_dir = (self.macdhist * direction) > 0 hist_decreasing = abs(self.macdhist) < abs(self.pre.macdhist) lines_tight = abs(self.macd - self.signal) <= 12 # 倒挂：能量柱衰减且黄白线相对方向不利/出现反向能量释放 if hist_decreasing and (((self.macd - self.signal) * direction) < 0 or not hist_same_dir): self.macd_state = Chan_MACD_STATE.CROSS_REV return self.macd_state # 缠绕/粘合：紧贴能量柱运行，无反向能量释放 if lines_tight and hist_same_dir: self.macd_state = Chan_MACD_STATE.CROSS_OS return self.macd_state # 趋近零轴：细化 NEAR0_* 判定 NEAR0_EPS = 15 lines_near_zero = abs(self.macd) <= NEAR0_EPS or abs(self.signal) <= NEAR0_EPS touch_52 = (self.ema52 != 0) and ((abs(self.close - self.ema52) <= NEAR0_EPS) or (self.low <= self.ema52 <= self.high)) touch_24 = (self.ema24 != 0) and ((abs(self.close - self.ema24) <= NEAR0_EPS) or (self.low <= self.ema24 <= self.high)) # 完美形态：白线接近零轴 + 价格触碰/轻破EMA52 + 黄线不穿零轴 if abs(self.macd) <= NEAR0_EPS and touch_52 and (not (self.pre.signal >= 0 and self.signal < 0)) and (not (self.pre.signal <= 0 and self.signal > 0)): self.macd_state = Chan_MACD_STATE.NEAR0_PERFECT #self.near0_return = 1 return self.macd_state # EMA24 附近 if lines_near_zero and touch_24: self.macd_state = Chan_MACD_STATE.NEAR0_24 #self.near0_return = 2 return self.macd_state # EMA52 附近 if lines_near_zero and touch_52: self.macd_state = Chan_MACD_STATE.NEAR0_52 #self.near0_return = 3 return self.macd_state # 白线接近零轴但价格未至EMA52 if abs(self.macd) <= NEAR0_EPS and not touch_52: self.macd_state = Chan_MACD_STATE.NEAR0_DIFF #self.near0_return = 4 return self.macd_state # 一般近零轴 if lines_near_zero or touch_52: self.macd_state = Chan_MACD_STATE.NEAR0 #self.near0_return = 5 return self.macd_state # 穿零轴后离开零轴 if self.pre.macd_state == Chan_MACD_STATE.CROSS0_UP and ((self.macd >= self.pre.macd and self.signal >= self.pre.signal) or (abs(self.macdhist) >= abs(self.pre.macdhist))): self.macd_state = Chan_MACD_STATE.UP return self.macd_state if self.pre.macd_state == Chan_MACD_STATE.CROSS0_DOWN and ((self.macd <= self.pre.macd and self.signal <= self.pre.signal) or (abs(self.macdhist) >= abs(self.pre.macdhist))): self.macd_state = Chan_MACD_STATE.DOWN return self.macd_state if self.pre.macd_state == Chan_MACD_STATE.UP and self.macd > self.pre.macd and self.signal > self.pre.signal: self.macd_state = Chan_MACD_STATE.UP return self.macd_state if self.pre.macd_state == Chan_MACD_STATE.DOWN and self.macd < self.pre.macd and self.signal < self.pre.signal: self.macd_state = Chan_MACD_STATE.DOWN return self.macd_state # 趋势兜底：强势同步上行/下行直接进入 UP/DOWN if self.macd > 0 and self.signal > 0 and (self.macd >= self.pre.macd and self.signal >= self.pre.signal): self.macd_state = Chan_MACD_STATE.UP return self.macd_state if self.macd < 0 and self.signal < 0 and (self.macd <= self.pre.macd and self.signal <= self.pre.signal): self.macd_state = Chan_MACD_STATE.DOWN return self.macd_state # 峰值：白线高位出现局部顶 if hasattr(self.pre, 'pre') and self.pre and self.pre.pre and self.macd > 0: if self.pre.macd > self.pre.pre.macd and self.pre.macd > self.macd: self.macd_state = Chan_MACD_STATE.PEAK return self.macd_state # 高位状态的位置状态， 高位，高位空，归零轴 if (self.pre.macd_state == Chan_MACD_STATE.UP or self.pre.macd_state == Chan_MACD_STATE.HIGH or self.pre.macd_state == Chan_MACD_STATE.RZ_UP or self.pre.macd_state == Chan_MACD_STATE.PEAK or self.pre.macd_state == Chan_MACD_STATE.HIGH_EMPTY) and self.macd > 0: # 高位空（正区间）：能量柱衰减且黄白线间距较大 if abs(self.pre.macdhist) > 0 and abs(self.macdhist) < abs(self.pre.macdhist) and abs(self.macd - self.signal) > 5: self.macd_state = Chan_MACD_STATE.HIGH_EMPTY return self.macd_state if abs(self.pre.macd - self.macd) < 10: self.macd_state = Chan_MACD_STATE.HIGH return self.macd_state else: if self.macd > self.pre.macd and self.signal > self.pre.signal: self.macd_state = Chan_MACD_STATE.UP return self.macd_state elif self.macd < self.pre.macd and self.signal < self.pre.signal: self.macd_state = Chan_MACD_STATE.RETURN_ZERO return self.macd_state if (self.pre.macd_state == Chan_MACD_STATE.DOWN or self.pre.macd_state == Chan_MACD_STATE.HIGH or self.pre.macd_state == Chan_MACD_STATE.RZ_DOWN or self.pre.macd_state == Chan_MACD_STATE.PEAK or self.pre.macd_state == Chan_MACD_STATE.HIGH_EMPTY) and self.macd < 0: # 高位空（负区间）：能量柱衰减且黄白线间距较大 if abs(self.pre.macdhist) > 0 and abs(self.macdhist) < abs(self.pre.macdhist) and abs(self.macd - self.signal) > 5: self.macd_state = Chan_MACD_STATE.HIGH_EMPTY return self.macd_state if abs(self.pre.macd - self.macd) < 10: self.macd_state = Chan_MACD_STATE.HIGH return self.macd_state else: if self.macd > self.pre.macd and self.signal > self.pre.signal: self.macd_state = Chan_MACD_STATE.RETURN_ZERO return self.macd_state elif self.macd < self.pre.macd and self.signal < self.pre.signal: self.macd_state = Chan_MACD_STATE.DOWN return self.macd_state # 离开0轴开始上涨或者下跌阶段，高位之前的 if self.macd > 0 and self.pre: if (self.pre.macd_state == Chan_MACD_STATE.NEAR0 or self.pre.macd_state == Chan_MACD_STATE.RZ_UP or self.pre.macd_state == Chan_MACD_STATE.CROSS0_UP) and (self.signal > self.pre.signal or self.close > self.ema52): self.macd_state = Chan_MACD_STATE.RZ_UP return self.macd_state elif self.macd < 0 and self.pre: if (self.pre.macd_state == Chan_MACD_STATE.NEAR0 or self.pre.macd_state == Chan_MACD_STATE.RZ_DOWN or self.pre.macd_state == Chan_MACD_STATE.CROSS0_DOWN) and (self.signal < self.pre.signal or self.close < self.ema52): self.macd_state = Chan_MACD_STATE.RZ_DOWN return self.macd_state # 归零轴走势 if self.pre.macd_state == Chan_MACD_STATE.RETURN_ZERO: if self.macd > 0: if self.pre.macd > self.macd or abs(self.macdhist) <= abs(self.pre.macdhist) or self.signal <= self.pre.signal: self.macd_state = Chan_MACD_STATE.RETURN_ZERO return self.macd_state else: if self.pre.macd < self.macd or abs(self.macdhist) <= abs(self.pre.macdhist) or self.signal >= self.pre.signal: self.macd_state = Chan_MACD_STATE.RETURN_ZERO return self.macd_state # 从 NEAR0 收敛到零轴的归零轴承接（正负两侧） if self.pre.macd_state == Chan_MACD_STATE.NEAR0: # 正区间朝零轴收敛 if self.macd > 0 and self.pre.macd > 0 and self.macd <= self.pre.macd and self.signal <= self.pre.signal: self.macd_state = Chan_MACD_STATE.RETURN_ZERO return self.macd_state # 负区间朝零轴收敛 if self.macd < 0 and self.pre.macd < 0 and self.macd >= self.pre.macd and self.signal >= self.pre.signal: self.macd_state = Chan_MACD_STATE.RETURN_ZERO return self.macd_state # 其余情况 if self.pre.macd_state == Chan_MACD_STATE.UNKNOWN: if self.macd > 0 and self.close > self.ema52 and self.pre.pre and (self.pre.pre.macd_state == Chan_MACD_STATE.UP or self.pre.pre.macd_state == Chan_MACD_STATE.RZ_UP): self.macd_state = self.pre.pre.macd_state return self.macd_state elif self.macd < 0 and self.close < self.ema52 and self.pre.pre and (self.pre.pre.macd_state == Chan_MACD_STATE.DOWN or self.pre.pre.macd_state == Chan_MACD_STATE.RZ_DOWN): self.macd_state = self.pre.pre.macd_state return self.macd_state else: self.macd_state = Chan_MACD_STATE.UNKNOWN return self.macd_state return self.macd_state