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 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.klus = [] 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.strength = klu.strength self.last_top_klc = None self.last_bottom_klc = None self.bb_out = True self.macd = 0 self.signal = 0 def set_last_top_klu(self, last_top_klc): self.last_top_klc = last_top_klc def set_last_bottom_klc(self, last_bottom_klc): self.last_bottom_klc = last_bottom_klc 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_bb_out() def add_klu(self, klu): self.klus.append(klu) def set_end_klu(self, klu): self.end_klu = klu self.end_time = klu.time self.close = klu.close self.cal_indicators() self.cal_shape_1() self.strength = self.cal_klc_strength() self.cal_bb_out() if self.pre: self.pre.cal_bb_out() def cal_bb_out(self): for klu in self.klus: if self.high >= klu.bbup302 and klu.bbup302 > 0 and (self.klc_fx_type == Chan_KLC_FX.TOP1 or self.klc_fx_type == Chan_KLC_FX.TOP2): self.bb_out = True #print(self.end_time, self.high, klu.bbup302, self.klc_fx_type) if self.high >= klu.bbup30 and klu.bbup30 > 0 and self.next and (self.next.macd - self.macd) < 0: self.klc_fx_type = Chan_KLC_FX.TOP4 if self.low <= klu.bblow302 and klu.bblow302 > 0 and (self.klc_fx_type == Chan_KLC_FX.BOTTOM1 or self.klc_fx_type == Chan_KLC_FX.BOTTOM2): self.bb_out = True if self.low <= klu.bblow30 and klu.bblow30 > 0 and self.next and (self.macd - self.next.macd) < 0: self.klc_fx_type = Chan_KLC_FX.BOTTOM4 if self.fx == Chan_FX_TYPE.TOP: #print(self.end_time, self.fx, self.macd, self.macdhist, len(self.klus)) if self.macdhist < 0 and self.macd > 0: self.klc_fx_type = Chan_KLC_FX.TOP5 self.bb_out = True else: if self.fx == Chan_FX_TYPE.BOTTOM: if self.macdhist > 0 and self.macd < 0: self.klc_fx_type = Chan_KLC_FX.BOTTOM5 self.bb_out = True def cal_macd_state(self, dir): macd_state = 0 return macd_state def cal_indicators(self): for index in range(1, len(self.klus)): self.volume += self.klus[index].volume self.rsi += self.klus[index].rsi self.volume_ratio += self.klus[index].volume_ratio self.macdhist += self.klus[index].macdhist self.rsi = self.rsi / len(self.klus) self.volume_ratio = self.volume_ratio / len(self.klus) self.volume = self.volume / len(self.klus) self.macdhist = self.macdhist / len(self.klus) if len(self.klus) > 0: self.macd = self.klus[-1].macd self.signal = self.klus[-1].signal def cal_shape_1(self): for index in range(1, len(self.klus)): self.body += self.klus[index].body self.upper_shadow += self.klus[index].upper_shadow self.lower_shadow += self.klus[index].lower_shadow self.body_ratio += self.klus[index].body_ratio self.upper_shadow_ratio += self.klus[index].upper_shadow_ratio self.lower_shadow_ratio += self.klus[index].lower_shadow_ratio self.range += self.klus[index].range self.body = self.body / len(self.klus) self.upper_shadow = self.upper_shadow / len(self.klus) self.lower_shadow = self.lower_shadow / len(self.klus) self.body_ratio = self.body_ratio / len(self.klus) self.upper_shadow_ratio = self.upper_shadow_ratio / len(self.klus) self.lower_shadow_ratio = self.lower_shadow_ratio / len(self.klus) self.range = self.range / len(self.klus) def cal_shape_2(self): 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 print(self): print(self.time, self.high, self.low, self.start_time, self.end_time, self.fx, self.index) def copy(self): """创建KLC对象的浅拷贝, 避免循环引用""" new_klc = ChanKLC(self.start_klu, self.index, self.dir) new_klc.high = self.high new_klc.low = self.low new_klc.state = self.state new_klc.fx = self.fx # 不复制 next 和 pre 引用，避免循环引用 return new_klc 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) def cal_klu_features(self): features = dict() feature_sums = dict() feature_counts = dict() # 遍历所有klu，累计每个特征的总和和计数 for klu in self.klus: for key, value in klu.get_feature_data().items(): if key not in feature_sums: feature_sums[key] = 0 feature_counts[key] = 0 feature_sums[key] += value feature_counts[key] += 1 # 计算每个特征的平均值 for key in feature_sums: features[key] = feature_sums[key] / feature_counts[key] return features def cal_fx_shape(self): if self.klc_fx_type != Chan_KLC_FX.UNKNOWN: if self.pre and self.next and self.next.end_klu: klc1 = self.pre klc2 = self klc3 = self.next klu_list = [] klu_list.append(klc1.klus) klu_list.append(klc2.klus) klu_list.append(klc3.klus) gap = klc3.end_klu.index - klc1.start_klu.index + 1 if gap < 4: pass return gap def get_feature_data(self): features = dict() # 原有基础特征 features['klc_close'] = self.close #0 features['klc_open'] = self.open #1 features['klc_high'] = self.high #2 features['klc_low'] = self.low #3 features['klc_index'] = self.index #4 features['klc_dir'] = 0 if self.dir == Chan_KLINE_DIR.UP else 1 #5 features['klc_state'] = self.state #6 features['klc_fx'] = 0 if self.fx == Chan_FX_TYPE.UNKNOWN else 1 if self.fx == Chan_FX_TYPE.TOP else 2 #7 features['klc_klus'] = len(self.klus) #8 features['klc_volume'] = self.volume #9 features['klc_pre_fx'] = (0 if self.pre.fx == Chan_FX_TYPE.UNKNOWN else 1 if self.pre.fx == Chan_FX_TYPE.TOP else 2) if self.pre else 0 #10 features['klc_pre_pre_fx'] = (0 if self.pre.pre.fx == Chan_FX_TYPE.UNKNOWN else 1 if self.pre.pre.fx == Chan_FX_TYPE.TOP else 2) if self.pre and self.pre.pre else 0 #11 features['klc_pre_pre_pre_fx'] = (0 if self.pre.pre.pre.fx == Chan_FX_TYPE.UNKNOWN else 1 if self.pre.pre.pre.fx == Chan_FX_TYPE.TOP else 2) if self.pre and self.pre.pre and self.pre.pre.pre else 0 #12 features['klc_distance'] = self.distance #13 features['klc_volume_ratio'] = self.volume_ratio #14 features['klc_rsi'] = self.rsi #15 # New Add 20250422 #features['klc_macdhist'] = self.get_macdhist() #11 #features['klc_bi_macdhist'] = self.bi_macdhist #12 #features['klc_bi_macd_div'] = self.bi_macd_div #13 #features['klc_bi_dir'] = 1 if self.bi.dir == Chan_BI_DIR.UP else -1 #14 # ===== 2.1 K线形态因子 ===== # K线实体大小 if self.open != 0: # 避免除以零 features['klc_body_size_rel'] = abs(self.close - self.open) / self.open # 相对实体大小 else: features['klc_body_size_rel'] = 0 features['klc_body_size_abs'] = abs(self.close - self.open) # 绝对实体大小 # 上下影线长度 max_oc = max(self.open, self.close) min_oc = min(self.open, self.close) high_low_range = self.high - self.low if high_low_range != 0: # 避免除以零 features['klc_upper_shadow'] = (self.high - max_oc) / high_low_range # 上影线相对长度 features['klc_lower_shadow'] = (min_oc - self.low) / high_low_range # 下影线相对长度 else: features['klc_upper_shadow'] = 0 features['klc_lower_shadow'] = 0 # K线波动范围 if self.close != 0: # 避免除以零 features['klc_range'] = 0 #(self.high - self.low) / self.close else: features['klc_range'] = 0 # 与前K线的价格关系 if self.pre: # 当前K线最高价与前一根K线最高价的比较 if self.pre.high != 0: # 避免除以零 features['klc_high_ratio'] = self.high / self.pre.high else: features['klc_high_ratio'] = 1 # 当前K线最低价与前一根K线最低价的比较 if self.pre.low != 0: # 避免除以零 features['klc_low_ratio'] = self.low / self.pre.low else: features['klc_low_ratio'] = 1 # 当前K线收盘价与前一根K线收盘价的相对位置 if self.pre.close != 0: # 避免除以零 features['klc_close_change_1'] = (self.close - self.pre.close) / self.pre.close else: features['klc_close_change_1'] = 0 # 如果有前两根K线 if self.pre.pre: if self.pre.pre.close != 0: # 避免除以零 features['klc_close_change_2'] = (self.close - self.pre.pre.close) / self.pre.pre.close else: features['klc_close_change_2'] = 0 else: features['klc_close_change_2'] = 0 else: # 如果没有前K线，设置默认值 features['klc_high_ratio'] = 1 features['klc_low_ratio'] = 1 features['klc_close_change_1'] = 0 features['klc_close_change_2'] = 0 # 分型特征编码 # 这里直接使用现有的fx字段，不重复计算 # ===== 2.2 价格关系因子 ===== # 价格与均线的关系 (从KLU中获取) klu_features = self.cal_klu_features() # MA5与收盘价的关系 if 'klu_ma5' in klu_features and klu_features['klu_ma5'] != 0: features['klc_close_to_ma5'] = (self.close - klu_features['klu_ma5']) / klu_features['klu_ma5'] else: features['klc_close_to_ma5'] = 0 # MA10与收盘价的关系 if 'klu_ma10' in klu_features and klu_features['klu_ma10'] != 0: features['klc_close_to_ma10'] = (self.close - klu_features['klu_ma10']) / klu_features['klu_ma10'] else: features['klc_close_to_ma10'] = 0 # MA30与收盘价的关系 if 'klu_ma30' in klu_features and klu_features['klu_ma30'] != 0: features['klc_close_to_ma30'] = (self.close - klu_features['klu_ma30']) / klu_features['klu_ma30'] else: features['klc_close_to_ma30'] = 0 # 短期均线与长期均线的差异 if 'klu_ma5' in klu_features and 'klu_ma30' in klu_features and klu_features['klu_ma30'] != 0: features['klc_ma_diff'] = (klu_features['klu_ma5'] - klu_features['klu_ma30']) / klu_features['klu_ma30'] else: features['klc_ma_diff'] = 0 # 价格突破特征 # 检查当前K线是否突破前3根K线的最高/最低价 if self.pre: max_high = self.pre.high min_low = self.pre.low temp = self.pre count = 1 while temp.pre and count < 3: temp = temp.pre max_high = max(max_high, temp.high) min_low = min(min_low, temp.low) count += 1 features['klc_break_high'] = 1 if self.high > max_high else 0 features['klc_break_low'] = 1 if self.low < min_low else 0 else: features['klc_break_high'] = 0 features['klc_break_low'] = 0 # ===== 2.3 技术指标因子 ===== # 获取技术指标 # RSI (从KLU中获取) if 'klu_rsi' in klu_features: features['klc_rsi'] = klu_features['klu_rsi'] else: features['klc_rsi'] = 50 # 默认中性值 # MACD (从KLU中获取) if 'klu_macd' in klu_features: features['klc_macd'] = klu_features['klu_macd'] else: features['klc_macd'] = 0 if 'klu_signal' in klu_features: features['klc_macd_signal'] = klu_features['klu_signal'] else: features['klc_macd_signal'] = 0 if 'klu_macdhist' in klu_features: features['klc_macdhist'] = klu_features['klu_macdhist'] else: features['klc_macdhist'] = 0 # 成交量变化 if self.pre: vol_sum = 0 count = 0 temp = self.pre # 计算前5根K线的平均成交量 while temp and count < 5: vol_sum += temp.volume count += 1 temp = temp.pre avg_vol = vol_sum / count if count > 0 else self.volume if avg_vol != 0: # 避免除以零 features['klc_vol_ratio'] = self.volume / avg_vol else: features['klc_vol_ratio'] = 1 else: features['klc_vol_ratio'] = 1 # ===== 2.4 市场环境因子 ===== # 价格波动率 (前5根K线收盘价的标准差) if self.pre: close_vals = [self.close] temp = self.pre count = 0 while temp and count < 5: close_vals.append(temp.close) count += 1 temp = temp.pre if len(close_vals) > 1: import numpy as np std_dev = np.std(close_vals) avg_close = np.mean(close_vals) if avg_close != 0: # 避免除以零 features['klc_volatility'] = std_dev / avg_close else: features['klc_volatility'] = 0 else: features['klc_volatility'] = 0 else: features['klc_volatility'] = 0 # 前5根K线的价格趋势 (简单线性回归斜率) if self.pre: price_vals = [self.close] temp = self.pre count = 0 while temp and count < 5: price_vals.append(temp.close) count += 1 temp = temp.pre if len(price_vals) > 2: import numpy as np y = np.array(price_vals) x = np.arange(len(y)) # 简单线性回归 slope = np.polyfit(x, y, 1)[0] # 归一化斜率 if abs(np.mean(y)) > 0: # 避免除以零 features['klc_trend_slope'] = slope / abs(np.mean(y)) else: features['klc_trend_slope'] = 0 else: features['klc_trend_slope'] = 0 else: features['klc_trend_slope'] = 0 # ===== 2.5 其他衍生因子 ===== # K线组合形态 # 十字星 (实体非常小) body_pct = abs(self.close - self.open) / (self.high - self.low) if (self.high - self.low) > 0 else 0 features['klc_is_doji'] = 1 if body_pct < 0.1 else 0 # 实体小于10%算十字星 # 锤子线/上吊线 (下影线长，上影线短，实体小) if high_low_range > 0: lower_shadow_pct = (min_oc - self.low) / high_low_range upper_shadow_pct = (self.high - max_oc) / high_low_range features['klc_is_hammer'] = 1 if (lower_shadow_pct > 0.6 and upper_shadow_pct < 0.1) else 0 else: features['klc_is_hammer'] = 0 # 吞没形态 if self.pre: prev_body_size = abs(self.pre.close - self.pre.open) curr_body_size = abs(self.close - self.open) # 看涨吞没 if (self.pre.close < self.pre.open # 前一根是阴线 and self.close > self.open # 当前是阳线 and self.open <= self.pre.close # 当前开盘低于前收盘 and self.close >= self.pre.open # 当前收盘高于前开盘 and curr_body_size > prev_body_size): # 当前实体大于前实体 features['klc_is_bullish_engulfing'] = 1 else: features['klc_is_bullish_engulfing'] = 0 # 看跌吞没 if (self.pre.close > self.pre.open # 前一根是阳线 and self.close < self.open # 当前是阴线 and self.open >= self.pre.close # 当前开盘高于前收盘 and self.close <= self.pre.open # 当前收盘低于前开盘 and curr_body_size > prev_body_size): # 当前实体大于前实体 features['klc_is_bearish_engulfing'] = 1 else: features['klc_is_bearish_engulfing'] = 0 else: features['klc_is_bullish_engulfing'] = 0 features['klc_is_bearish_engulfing'] = 0 # 包含关系 if self.pre: # 向上包含 if (self.high >= self.pre.high and self.low >= self.pre.low): features['klc_is_up_inclusive'] = 1 else: features['klc_is_up_inclusive'] = 0 # 向下包含 if (self.high <= self.pre.high and self.low <= self.pre.low): features['klc_is_down_inclusive'] = 1 else: features['klc_is_down_inclusive'] = 0 # 完全包含 if (self.high >= self.pre.high and self.low <= self.pre.low): features['klc_is_full_inclusive'] = 1 else: features['klc_is_full_inclusive'] = 0 # 被完全包含 if (self.high <= self.pre.high and self.low >= self.pre.low): features['klc_is_inner_inclusive'] = 1 else: features['klc_is_inner_inclusive'] = 0 else: features['klc_is_up_inclusive'] = 0 features['klc_is_down_inclusive'] = 0 features['klc_is_full_inclusive'] = 0 features['klc_is_inner_inclusive'] = 0 # 从KLU获取其他特征 #features.update(self.cal_klu_features()) # ===== 3.1 价格形态扩展因子 ===== # 区间突破强度 if self.pre and self.pre.pre: prev_range = self.pre.high - self.pre.low if prev_range > 0: features['klc_breakout_strength'] = (self.close - self.pre.high) / prev_range if self.close > self.pre.high else (self.pre.low - self.close) / prev_range if self.close < self.pre.low else 0 else: features['klc_breakout_strength'] = 0 else: features['klc_breakout_strength'] = 0 # 价格动量 if self.pre: features['klc_momentum_1'] = self.close - self.pre.close if self.pre.pre: features['klc_momentum_2'] = self.close - self.pre.pre.close else: features['klc_momentum_2'] = 0 else: features['klc_momentum_1'] = 0 features['klc_momentum_2'] = 0 # 价格加速度 if self.pre and self.pre.pre: prev_change = self.pre.close - self.pre.pre.close curr_change = self.close - self.pre.close features['klc_price_acceleration'] = curr_change - prev_change else: features['klc_price_acceleration'] = 0 # 相对位置 if self.high != self.low: features['klc_relative_position'] = (self.close - self.low) / (self.high - self.low) else: features['klc_relative_position'] = 0.5 # 价格区间位置 (前N根K线) prev_klcs = [] temp = self.pre for _ in range(10): # 前10根K线 if temp: prev_klcs.append(temp) temp = temp.pre else: break if prev_klcs: max_high = max([klc.high for klc in prev_klcs]) if prev_klcs else self.high min_low = min([klc.low for klc in prev_klcs]) if prev_klcs else self.low price_range = max_high - min_low if price_range > 0: features['klc_range_position'] = (self.close - min_low) / price_range else: features['klc_range_position'] = 0.5 else: features['klc_range_position'] = 0.5 # ===== 3.2 更多技术指标因子 ===== # MACD趋势 if self.pre and 'klc_macdhist' in features: features['klc_macdhist_change'] = features['klc_macdhist'] - self.pre.macdhist else: features['klc_macdhist_change'] = 0 # RSI趋势 if self.pre and 'klc_rsi' in features: features['klc_rsi_change'] = features['klc_rsi'] - self.pre.rsi else: features['klc_rsi_change'] = 0 # RSI超买超卖 if 'klc_rsi' in features: features['klc_rsi_overbought'] = 1 if features['klc_rsi'] > 70 else 0 features['klc_rsi_oversold'] = 1 if features['klc_rsi'] < 30 else 0 else: features['klc_rsi_overbought'] = 0 features['klc_rsi_oversold'] = 0 # 布林带位置 (如果可从KLU获取) if 'klu_upper_band' in klu_features and 'klu_lower_band' in klu_features: upper_band = klu_features['klu_upper_band'] lower_band = klu_features['klu_lower_band'] middle_band = klu_features['klu_middle_band'] if 'klu_middle_band' in klu_features else (upper_band + lower_band) / 2 band_width = upper_band - lower_band if band_width > 0: features['klc_bollinger_position'] = (self.close - lower_band) / band_width else: features['klc_bollinger_position'] = 0.5 features['klc_bollinger_width'] = band_width / middle_band if middle_band > 0 else 0 features['klc_upper_band_touch'] = 1 if self.high >= upper_band else 0 features['klc_lower_band_touch'] = 1 if self.low <= lower_band else 0 else: features['klc_bollinger_position'] = 0.5 features['klc_bollinger_width'] = 0 features['klc_upper_band_touch'] = 0 features['klc_lower_band_touch'] = 0 # 量价关系 if self.pre: price_change = self.close - self.pre.close if price_change != 0: features['klc_volume_price_ratio'] = self.volume / abs(price_change) else: features['klc_volume_price_ratio'] = 0 else: features['klc_volume_price_ratio'] = 0 # ===== 3.3 波动性因子 ===== # 真实波动幅度 (True Range) if self.pre: tr1 = self.high - self.low tr2 = abs(self.high - self.pre.close) tr3 = abs(self.low - self.pre.close) features['klc_true_range'] = max(tr1, tr2, tr3) else: features['klc_true_range'] = self.high - self.low # 归一化真实波动幅度 if self.pre and self.pre.close > 0: features['klc_normalized_tr'] = features['klc_true_range'] / self.pre.close else: features['klc_normalized_tr'] = 0 # 滑动窗口波动率 if prev_klcs: tr_values = [] for i in range(len(prev_klcs)): if i == 0: tr = max(prev_klcs[i].high - prev_klcs[i].low, abs(prev_klcs[i].high - self.close), abs(prev_klcs[i].low - self.close)) else: tr = max(prev_klcs[i].high - prev_klcs[i].low, abs(prev_klcs[i].high - prev_klcs[i-1].close), abs(prev_klcs[i].low - prev_klcs[i-1].close)) tr_values.append(tr) if tr_values: import numpy as np # ATR (Average True Range) features['klc_atr'] = np.mean(tr_values) if self.close > 0: features['klc_atr_percent'] = features['klc_atr'] / self.close else: features['klc_atr_percent'] = 0 # 高低点波动 if len(prev_klcs) >= 5: highs = [klc.high for klc in prev_klcs[:5]] lows = [klc.low for klc in prev_klcs[:5]] max_high = max(highs) min_low = min(lows) features['klc_high_volatility'] = np.std(highs) / np.mean(highs) if np.mean(highs) > 0 else 0 features['klc_low_volatility'] = np.std(lows) / np.mean(lows) if np.mean(lows) > 0 else 0 features['klc_price_range'] = (max_high - min_low) / min_low if min_low > 0 else 0 else: features['klc_high_volatility'] = 0 features['klc_low_volatility'] = 0 features['klc_price_range'] = 0 else: features['klc_atr'] = 0 features['klc_atr_percent'] = 0 features['klc_high_volatility'] = 0 features['klc_low_volatility'] = 0 features['klc_price_range'] = 0 else: features['klc_atr'] = 0 features['klc_atr_percent'] = 0 features['klc_high_volatility'] = 0 features['klc_low_volatility'] = 0 features['klc_price_range'] = 0 # ===== 3.4 趋势强度因子 ===== # 方向移动指标 if self.pre: # 上升动量和下降动量 up_move = self.high - self.pre.high down_move = self.pre.low - self.low features['klc_plus_dm'] = up_move if up_move > down_move and up_move > 0 else 0 features['klc_minus_dm'] = down_move if down_move > up_move and down_move > 0 else 0 # 方向指数 if features['klc_atr'] > 0: features['klc_plus_di'] = 100 * features['klc_plus_dm'] / features['klc_atr'] features['klc_minus_di'] = 100 * features['klc_minus_dm'] / features['klc_atr'] else: features['klc_plus_di'] = 0 features['klc_minus_di'] = 0 # 方向指数差 features['klc_dx'] = 100 * abs(features['klc_plus_di'] - features['klc_minus_di']) / (features['klc_plus_di'] + features['klc_minus_di']) if (features['klc_plus_di'] + features['klc_minus_di']) > 0 else 0 else: features['klc_plus_dm'] = 0 features['klc_minus_dm'] = 0 features['klc_plus_di'] = 0 features['klc_minus_di'] = 0 features['klc_dx'] = 0 # 价格趋势强度 if prev_klcs and len(prev_klcs) >= 5: import numpy as np prices = [self.close] + [klc.close for klc in prev_klcs[:5]] x = np.arange(len(prices)) # 线性回归 slope, intercept = np.polyfit(x, prices, 1) # 趋势线拟合度 (R^2) y_pred = slope * x + intercept ss_total = np.sum((prices - np.mean(prices)) ** 2) ss_residual = np.sum((prices - y_pred) ** 2) if ss_total > 0: features['klc_trend_r2'] = 1 - (ss_residual / ss_total) else: features['klc_trend_r2'] = 0 # 趋势线斜率 features['klc_trend_slope_norm'] = slope / np.mean(prices) if np.mean(prices) > 0 else 0 # 价格与趋势线的距离 current_trend_value = slope * 0 + intercept # x=0 表示当前K线在预测线上的值 if current_trend_value > 0: features['klc_trend_distance'] = (self.close - current_trend_value) / current_trend_value else: features['klc_trend_distance'] = 0 else: features['klc_trend_r2'] = 0 features['klc_trend_slope_norm'] = 0 features['klc_trend_distance'] = 0 # ===== 3.5 支撑与阻力因子 ===== # 前N根K线的支撑和阻力 if prev_klcs and len(prev_klcs) >= 5: highs = [klc.high for klc in prev_klcs[:5]] lows = [klc.low for klc in prev_klcs[:5]] # 简单支撑位 (前5根K线最低点) support = min(lows) # 简单阻力位 (前5根K线最高点) resistance = max(highs) # 与支撑阻力的距离 if support > 0: features['klc_distance_to_support'] = (self.close - support) / support else: features['klc_distance_to_support'] = 0 if resistance > 0: features['klc_distance_to_resistance'] = (resistance - self.close) / resistance else: features['klc_distance_to_resistance'] = 0 # 支撑阻力突破 features['klc_breaks_support'] = 1 if self.low < support else 0 features['klc_breaks_resistance'] = 1 if self.high > resistance else 0 # 支撑阻力区间位置 if resistance > support: features['klc_sr_position'] = (self.close - support) / (resistance - support) else: features['klc_sr_position'] = 0.5 else: features['klc_distance_to_support'] = 0 features['klc_distance_to_resistance'] = 0 features['klc_breaks_support'] = 0 features['klc_breaks_resistance'] = 0 features['klc_sr_position'] = 0.5 # ===== 3.6 量价关系扩展因子 ===== # 价格与成交量的相关性 if prev_klcs and len(prev_klcs) >= 5: import numpy as np prices = [self.close] + [klc.close for klc in prev_klcs[:5]] volumes = [self.volume] + [klc.volume for klc in prev_klcs[:5]] # 计算相关系数 if len(prices) > 1 and np.std(prices) > 0 and np.std(volumes) > 0: price_mean = np.mean(prices) volume_mean = np.mean(volumes) numerator = np.sum((prices - price_mean) * (volumes - volume_mean)) denominator = np.sqrt(np.sum((prices - price_mean) ** 2) * np.sum((volumes - volume_mean) ** 2)) if denominator > 0: features['klc_price_volume_corr'] = numerator / denominator else: features['klc_price_volume_corr'] = 0 else: features['klc_price_volume_corr'] = 0 # 价格上涨时的平均成交量 up_prices = [] up_volumes = [] # 价格下跌时的平均成交量 down_prices = [] down_volumes = [] for i in range(len(prev_klcs)): if i < len(prev_klcs) - 1: if prev_klcs[i].close > prev_klcs[i+1].close: up_prices.append(prev_klcs[i].close) up_volumes.append(prev_klcs[i].volume) else: down_prices.append(prev_klcs[i].close) down_volumes.append(prev_klcs[i].volume) features['klc_up_volume_avg'] = np.mean(up_volumes) if up_volumes else 0 features['klc_down_volume_avg'] = np.mean(down_volumes) if down_volumes else 0 if features['klc_down_volume_avg'] > 0: features['klc_volume_ratio_up_down'] = features['klc_up_volume_avg'] / features['klc_down_volume_avg'] else: features['klc_volume_ratio_up_down'] = 1 else: features['klc_price_volume_corr'] = 0 features['klc_up_volume_avg'] = 0 features['klc_down_volume_avg'] = 0 features['klc_volume_ratio_up_down'] = 1 # 成交量变化率 if self.pre: if self.pre.volume > 0: features['klc_volume_change'] = (self.volume - self.pre.volume) / self.pre.volume else: features['klc_volume_change'] = 0 else: features['klc_volume_change'] = 0 # 量能扩散 if prev_klcs and len(prev_klcs) >= 5: avg_volume = np.mean([klc.volume for klc in prev_klcs[:5]]) if avg_volume > 0: features['klc_volume_expansion'] = self.volume / avg_volume else: features['klc_volume_expansion'] = 1 else: features['klc_volume_expansion'] = 1 # ===== 3.7 K线时序模式因子 ===== # 连续上涨/下跌计数 up_count = 0 down_count = 0 if prev_klcs: temp = self last_close = temp.close for klc in prev_klcs: if klc.close < last_close: up_count += 1 down_count = 0 elif klc.close > last_close: down_count += 1 up_count = 0 last_close = klc.close features['klc_consecutive_up'] = up_count features['klc_consecutive_down'] = down_count else: features['klc_consecutive_up'] = 0 features['klc_consecutive_down'] = 0 # 跳空缺口 if self.pre: features['klc_gap_up'] = self.low - self.pre.high if self.low > self.pre.high else 0 features['klc_gap_down'] = self.pre.low - self.high if self.high < self.pre.low else 0 # 归一化缺口大小 if self.pre.close > 0: features['klc_gap_up_pct'] = features['klc_gap_up'] / self.pre.close features['klc_gap_down_pct'] = features['klc_gap_down'] / self.pre.close else: features['klc_gap_up_pct'] = 0 features['klc_gap_down_pct'] = 0 else: features['klc_gap_up'] = 0 features['klc_gap_down'] = 0 features['klc_gap_up_pct'] = 0 features['klc_gap_down_pct'] = 0 # 价格回撤 if prev_klcs: max_price = self.close min_price = self.close for klc in prev_klcs[:5]: max_price = max(max_price, klc.close) min_price = min(min_price, klc.close) if max_price > 0: features['klc_drawdown'] = (max_price - self.close) / max_price else: features['klc_drawdown'] = 0 if min_price > 0: features['klc_pullback'] = (self.close - min_price) / min_price else: features['klc_pullback'] = 0 else: features['klc_drawdown'] = 0 features['klc_pullback'] = 0 # ===== 3.8 复杂形态识别因子 ===== # 双顶/双底形态 if self.pre and self.pre.pre and self.pre.pre.pre and self.pre.pre.pre.pre: p5 = self.pre.pre.pre.pre p4 = self.pre.pre.pre p3 = self.pre.pre p2 = self.pre p1 = self # 双顶检测 (M形) double_top = (p5.high < p4.high and p4.high > p3.high and p3.high < p2.high and p2.high > p1.high and abs(p4.high - p2.high) / p4.high < 0.03) # 两个顶的高度接近 # 双底检测 (W形) double_bottom = (p5.low > p4.low and p4.low < p3.low and p3.low > p2.low and p2.low < p1.low and abs(p4.low - p2.low) / p4.low < 0.03) # 两个底的低点接近 features['klc_double_top'] = 1 if double_top else 0 features['klc_double_bottom'] = 1 if double_bottom else 0 else: features['klc_double_top'] = 0 features['klc_double_bottom'] = 0 # 头肩顶/底形态 if self.pre and self.pre.pre and self.pre.pre.pre and self.pre.pre.pre.pre and self.pre.pre.pre.pre.pre: p7 = self.pre.pre.pre.pre.pre p6 = self.pre.pre.pre.pre p5 = self.pre.pre.pre p4 = self.pre.pre p3 = self.pre p2 = self # 头肩顶 (左肩-头-右肩) head_shoulders_top = (p7.high < p6.high and p6.high > p5.high and p5.high < p4.high and p4.high > p3.high and p3.high < p2.high and abs(p6.high - p2.high) / p6.high < 0.05 and # 左肩和右肩高度接近 p4.high > p6.high and p4.high > p2.high) # 头部高于肩部 # 头肩底 (左肩-头-右肩) head_shoulders_bottom = (p7.low > p6.low and p6.low < p5.low and p5.low > p4.low and p4.low < p3.low and p3.low > p2.low and abs(p6.low - p2.low) / p6.low < 0.05 and # 左肩和右肩低点接近 p4.low < p6.low and p4.low < p2.low) # 头部低于肩部 features['klc_head_shoulders_top'] = 1 if head_shoulders_top else 0 features['klc_head_shoulders_bottom'] = 1 if head_shoulders_bottom else 0 else: features['klc_head_shoulders_top'] = 0 features['klc_head_shoulders_bottom'] = 0 # 旗形/三角形 if prev_klcs and len(prev_klcs) >= 5: import numpy as np highs = [self.high] + [klc.high for klc in prev_klcs[:5]] lows = [self.low] + [klc.low for klc in prev_klcs[:5]] # 计算高点趋势线斜率 x = np.arange(len(highs)) high_slope, _ = np.polyfit(x, highs, 1) # 计算低点趋势线斜率 low_slope, _ = np.polyfit(x, lows, 1) # 旗形: 高点和低点趋势线平行且方向相同 if abs(high_slope - low_slope) / (abs(high_slope) + 1e-10) < 0.2: features['klc_flag_pattern'] = 1 else: features['klc_flag_pattern'] = 0 # 上升三角形: 高点趋势线水平，低点趋势线向上 if abs(high_slope) < 0.01 and low_slope > 0.01: features['klc_ascending_triangle'] = 1 else: features['klc_ascending_triangle'] = 0 # 下降三角形: 高点趋势线向下，低点趋势线水平 if high_slope < -0.01 and abs(low_slope) < 0.01: features['klc_descending_triangle'] = 1 else: features['klc_descending_triangle'] = 0 # 对称三角形: 高点趋势线向下，低点趋势线向上 if high_slope < -0.01 and low_slope > 0.01: features['klc_symmetric_triangle'] = 1 else: features['klc_symmetric_triangle'] = 0 else: features['klc_flag_pattern'] = 0 features['klc_ascending_triangle'] = 0 features['klc_descending_triangle'] = 0 features['klc_symmetric_triangle'] = 0 # ===== 3.9 微观结构因子 ===== # 价格动量加速度 if self.pre and self.pre.pre and self.pre.pre.pre: mom1 = self.close - self.pre.close mom2 = self.pre.close - self.pre.pre.close mom3 = self.pre.pre.close - self.pre.pre.pre.close # 一阶动量变化 features['klc_mom_change_1'] = mom1 - mom2 # 二阶动量变化 features['klc_mom_change_2'] = (mom1 - mom2) - (mom2 - mom3) # 动量方向变化 features['klc_mom_direction_change'] = 1 if (mom1 > 0 and mom2 < 0) or (mom1 < 0 and mom2 > 0) else 0 else: features['klc_mom_change_1'] = 0 features['klc_mom_change_2'] = 0 features['klc_mom_direction_change'] = 0 # 微观价格结构分析 if self.pre: # K线重叠程度 overlap_range = min(self.high, self.pre.high) - max(self.low, self.pre.low) total_range = max(self.high, self.pre.high) - min(self.low, self.pre.low) if total_range > 0: features['klc_overlap_ratio'] = max(0, overlap_range) / total_range else: features['klc_overlap_ratio'] = 0 # 收盘价在当前K线的相对位置 if self.high > self.low: features['klc_close_position_inbar'] = (self.close - self.low) / (self.high - self.low) else: features['klc_close_position_inbar'] = 0.5 # 当前K线相对于前一根K线的位置 if self.pre.high > self.pre.low: features['klc_rel_position_to_prev'] = (self.close - self.pre.low) / (self.pre.high - self.pre.low) else: features['klc_rel_position_to_prev'] = 0.5 else: features['klc_overlap_ratio'] = 0 features['klc_close_position_inbar'] = 0.5 features['klc_rel_position_to_prev'] = 0.5 # 价格变化率序列 if prev_klcs and len(prev_klcs) >= 3: ret1 = self.close / prev_klcs[0].close - 1 if prev_klcs[0].close > 0 else 0 ret2 = prev_klcs[0].close / prev_klcs[1].close - 1 if prev_klcs[1].close > 0 else 0 ret3 = prev_klcs[1].close / prev_klcs[2].close - 1 if prev_klcs[2].close > 0 else 0 features['klc_return_1'] = ret1 features['klc_return_2'] = ret2 features['klc_return_3'] = ret3 # 收益率加速度 features['klc_return_accel_1'] = ret1 - ret2 features['klc_return_accel_2'] = (ret1 - ret2) - (ret2 - ret3) else: features['klc_return_1'] = 0 features['klc_return_2'] = 0 features['klc_return_3'] = 0 features['klc_return_accel_1'] = 0 features['klc_return_accel_2'] = 0 # ===== 3.10 综合形态因子 ===== # 能量比率 (K线实体与影线比例) body_size = abs(self.close - self.open) if self.high > self.low: upper_shadow = self.high - max(self.open, self.close) lower_shadow = min(self.open, self.close) - self.low features['klc_upper_shadow_ratio'] = upper_shadow / (self.high - self.low) features['klc_lower_shadow_ratio'] = lower_shadow / (self.high - self.low) features['klc_body_to_range_ratio'] = body_size / (self.high - self.low) else: features['klc_upper_shadow_ratio'] = 0 features['klc_lower_shadow_ratio'] = 0 features['klc_body_to_range_ratio'] = 1 # K线平衡点 features['klc_balance_point'] = (self.high + self.low + self.close) / 3 # 与平衡点的距离 if features['klc_balance_point'] > 0: features['klc_distance_to_balance'] = (self.close - features['klc_balance_point']) / features['klc_balance_point'] else: features['klc_distance_to_balance'] = 0 # 波动性和趋势组合因子 if 'klc_volatility' in features and 'klc_trend_slope_norm' in features: features['klc_volatility_trend_ratio'] = features['klc_volatility'] / (abs(features['klc_trend_slope_norm']) + 1e-10) else: features['klc_volatility_trend_ratio'] = 0 # K线逆转形态 if self.pre: # 看涨逆转 (前一根阴线，当前阳线，且当前收盘高于前一根中点) bullish_reversal = (self.pre.close < self.pre.open and # 前一根阴线 self.close > self.open and # 当前阳线 self.close > (self.pre.high + self.pre.low) / 2) # 收盘价高于前一根中点 # 看跌逆转 (前一根阳线，当前阴线，且当前收盘低于前一根中点) bearish_reversal = (self.pre.close > self.pre.open and # 前一根阳线 self.close < self.open and # 当前阴线 self.close < (self.pre.high + self.pre.low) / 2) # 收盘价低于前一根中点 features['klc_bullish_reversal'] = 1 if bullish_reversal else 0 features['klc_bearish_reversal'] = 1 if bearish_reversal else 0 else: features['klc_bullish_reversal'] = 0 features['klc_bearish_reversal'] = 0 # 特殊K线形态 # 大阳线/大阴线 avg_body = 0 if prev_klcs and len(prev_klcs) >= 5: bodies = [abs(klc.close - klc.open) for klc in prev_klcs[:5]] avg_body = sum(bodies) / len(bodies) if bodies else 0 if avg_body > 0: features['klc_large_candle'] = body_size / avg_body else: features['klc_large_candle'] = 1 # 长上影线/长下影线 if self.high > self.low: upper_shadow_ratio = (self.high - max(self.open, self.close)) / (self.high - self.low) lower_shadow_ratio = (min(self.open, self.close) - self.low) / (self.high - self.low) features['klc_long_upper_shadow'] = 1 if upper_shadow_ratio > 0.6 else 0 features['klc_long_lower_shadow'] = 1 if lower_shadow_ratio > 0.6 else 0 else: features['klc_long_upper_shadow'] = 0 features['klc_long_lower_shadow'] = 0 # 星线形态 (当前K线实体小，且与前一根K线有缺口) if self.pre and (self.high - self.low) > 0: small_body = body_size / (self.high - self.low) < 0.3 gap_with_prev = (min(self.open, self.close) > self.pre.close) if self.pre.close > self.pre.open else (max(self.open, self.close) < self.pre.close) features['klc_star_pattern'] = 1 if small_body and gap_with_prev else 0 else: features['klc_star_pattern'] = 0 # ===== 分型强度特征 ===== # 添加分型强度相关特征 features['klc_fx_strength'] = self.cal_fx_strength() features['klc_fx_strength_level'] = self.get_fx_strength_level() features['klc_is_strong_fx'] = 1 if self.is_strong_fx() else 0 # 分型强度分类特征 fx_strength = features['klc_fx_strength'] features['klc_fx_strength_extreme'] = 1 if fx_strength >= 80 else 0 # 极强分型 features['klc_fx_strength_strong'] = 1 if 60 <= fx_strength < 80 else 0 # 强分型 features['klc_fx_strength_medium'] = 1 if 40 <= fx_strength < 60 else 0 # 中等分型 features['klc_fx_strength_weak'] = 1 if 20 <= fx_strength < 40 else 0 # 弱分型 features['klc_fx_strength_very_weak'] = 1 if fx_strength < 20 else 0 # 极弱分型 return features def cal_klc_strength(self): strength = 0 if not self.end_klu: return strength if len(self.klus) > 0: for klu in self.klus: strength += klu.strength return strength def cal_fx_strength(self, klc_offset=2): strength = 0 if not self.end_klu: return 0 if self.fx == Chan_FX_TYPE.UNKNOWN or not self.pre or not self.next: return strength else: if self.pre and self.next: klc1 = self.pre klc2 = self klc3 = self.next if self.bi: if self.bi.dir == Chan_BI_DIR.UP and self.fx == Chan_FX_TYPE.BOTTOM: return strength if self.bi.dir == Chan_BI_DIR.DOWN and self.fx == Chan_FX_TYPE.TOP: return strength if self.bi.dir == Chan_BI_DIR.UP: if self.klc_fx_type == Chan_KLC_FX.TOP1 or self.klc_fx_type == Chan_KLC_FX.TOP2 or self.klc_fx_type == Chan_KLC_FX.TOP3: strength += self.check_bi_end(self.bi) else: if self.klc_fx_type == Chan_KLC_FX.BOTTOM1 or self.klc_fx_type == Chan_KLC_FX.BOTTOM2 or self.klc_fx_type == Chan_KLC_FX.BOTTOM3: strength += self.check_bi_end(self.bi) else: return strength return strength def check_bi_end(self, bi): if bi.dir == Chan_BI_DIR.UP: return 1 else: return 1 def calculate_fx_strength(self): """ 基于专业缠论理论的分型强度评估体系 返回值：0-100的强度分数，数值越大表示分型越强 评分卡系统（总分29分，转换为100分制）： - 振幅比例：25%权重，最高5分 - 量能配合：20%权重，最高5分 - 均线位置：15%权重，最高5分 - 形成速度：10%权重，最高4分 - 次级别确认：30%权重，最高10分 """ if self.fx == Chan_FX_TYPE.UNKNOWN or not self.pre or not self.next: return 0 # ===== 一、基础要素确认（先决条件） ===== if not self._verify_basic_fx_structure(): return 0 total_score = 0 max_score = 29 # 5+5+5+4+10 # ===== 二、振幅比例评估 (25%权重，最高5分) ===== amplitude_score = self._calculate_amplitude_score() total_score += amplitude_score # ===== 三、量能配合评估 (20%权重，最高5分) ===== volume_score = self._calculate_volume_score() total_score += volume_score # ===== 四、均线位置评估 (15%权重，最高5分) ===== ma_score = self._calculate_ma_position_score() total_score += ma_score # ===== 五、形成速度评估 (10%权重，最高4分) ===== speed_score = self._calculate_formation_speed_score() total_score += speed_score # ===== 六、次级别确认评估 (30%权重，最高10分) ===== confirmation_score = self._calculate_confirmation_score() total_score += confirmation_score # 转换为100分制 final_score = (total_score / max_score) * 100 return round(final_score, 2) def _verify_basic_fx_structure(self): """ 验证基础分型要素（先决条件） 只验证最核心的分型定义，避免过度严格 """ if not self.pre or not self.next: return False if self.fx == Chan_FX_TYPE.TOP: # 顶分型核心要素：中间K线高点必须严格高于两侧 if not (self.high > self.pre.high and self.high > self.next.high): return False elif self.fx == Chan_FX_TYPE.BOTTOM: # 底分型核心要素：中间K线低点必须严格低于两侧 if not (self.low < self.pre.low and self.low < self.next.low): return False return True def _calculate_amplitude_score(self): """ 计算振幅比例得分 (最高5分) 强势分型：分型区间振幅>近期平均振幅的150% = 5分 标准分型：介于80%-150%之间 = 3分 弱势分型：<80% = 1分 """ score = 0 # 计算分型区间振幅 if self.fx == Chan_FX_TYPE.TOP: fx_amplitude = self.high - min(self.pre.low, self.next.low) # 加分项：右侧K线低点低于左侧K线低点（经典缠论强势特征） if self.next.low < self.pre.low: score += 1 else: # BOTTOM fx_amplitude = max(self.pre.high, self.next.high) - self.low # 加分项：右侧K线高点高于左侧K线高点（经典缠论强势特征） if self.next.high > self.pre.high: score += 1 # 计算近期平均振幅（前10根K线的ATR） avg_amplitude = self._calculate_recent_atr(lookback=10) if avg_amplitude <= 0: return max(1, score) # 确保至少有基础分 amplitude_ratio = fx_amplitude / avg_amplitude if amplitude_ratio >= 1.5: # >150% score += 4 # 基础4分 + 可能的经典形态1分 = 最高5分 elif amplitude_ratio >= 1.0: # 100%-150% score += 2 + int((amplitude_ratio - 1.0) * 4) # 2-4分线性插值 elif amplitude_ratio >= 0.8: # 80%-100% score += 1 + int((amplitude_ratio - 0.8) * 5) # 1-2分线性插值 else: # <80% score += 1 return min(5, score) def _calculate_volume_score(self): """ 计算量能配合得分 (最高5分) 顶分型：第二根K线放量滞涨为强烈信号 底分型：第三根K线放量回升为有效确认 """ # 计算前5根K线平均成交量 avg_volume = self._calculate_average_volume(lookback=5) if avg_volume <= 0: return 1 if self.fx == Chan_FX_TYPE.TOP: # 顶分型：检查第二根K线（当前）是否放量滞涨 volume_ratio = self.volume / avg_volume # 判断是否滞涨：收盘价位于K线下半部分 price_position = (self.close - self.low) / (self.high - self.low) if self.high > self.low else 0.5 if volume_ratio >= 2.0 and price_position <= 0.4: # 放量+滞涨 return 5 elif volume_ratio >= 1.5 and price_position <= 0.5: return 4 elif volume_ratio >= 1.2: return 3 else: return 1 else: # BOTTOM # 底分型：检查第三根K线是否放量回升 next_volume_ratio = self.next.volume / avg_volume if hasattr(self.next, 'volume') else 1 # 判断是否回升：第三根K线收盘价相对位置较高 if self.next.high > self.next.low: next_price_position = (self.next.close - self.next.low) / (self.next.high - self.next.low) else: next_price_position = 0.5 if next_volume_ratio >= 2.0 and next_price_position >= 0.6: # 放量+回升 return 5 elif next_volume_ratio >= 1.5 and next_price_position >= 0.5: return 4 elif next_volume_ratio >= 1.2: return 3 else: return 1 def _calculate_ma_position_score(self): """ 计算均线位置得分 (最高5分) 强势顶分型需在5/10均线乖离率>5%时出现 有效底分型常伴随MACD底背离 """ score = 0 # 获取均线数据 klu_features = self.cal_klu_features() if self.fx == Chan_FX_TYPE.TOP: # 顶分型：检查与5日和10日均线的乖离率 ma5_bias = 0 ma10_bias = 0 if 'klu_ma5' in klu_features and klu_features['klu_ma5'] > 0: ma5_bias = (self.close - klu_features['klu_ma5']) / klu_features['klu_ma5'] if 'klu_ma10' in klu_features and klu_features['klu_ma10'] > 0: ma10_bias = (self.close - klu_features['klu_ma10']) / klu_features['klu_ma10'] # 乖离率>5%为强势信号 if ma5_bias > 0.05 or ma10_bias > 0.05: score += 3 elif ma5_bias > 0.03 or ma10_bias > 0.03: score += 2 elif ma5_bias > 0 or ma10_bias > 0: score += 1 else: # BOTTOM # 底分型：检查MACD背离和均线支撑 # 简化处理：检查价格是否在均线附近或下方 ma5_support = False ma10_support = False if 'klu_ma5' in klu_features and klu_features['klu_ma5'] > 0: ma5_bias = (self.close - klu_features['klu_ma5']) / klu_features['klu_ma5'] if ma5_bias >= -0.05: # 在5日均线附近或上方 ma5_support = True if 'klu_ma10' in klu_features and klu_features['klu_ma10'] > 0: ma10_bias = (self.close - klu_features['klu_ma10']) / klu_features['klu_ma10'] if ma10_bias >= -0.05: # 在10日均线附近或上方 ma10_support = True if ma5_support and ma10_support: score += 3 elif ma5_support or ma10_support: score += 2 else: score += 1 # 检查MACD状态 if hasattr(self, 'macdhist'): if self.fx == Chan_FX_TYPE.BOTTOM and self.macdhist > 0: score += 2 # MACD金叉附近的底分型加分 elif self.fx == Chan_FX_TYPE.TOP and self.macdhist < 0: score += 2 # MACD死叉附近的顶分型加分 return min(5, score) def _calculate_formation_speed_score(self): """ 计算形成速度得分 (最高4分) 强势特征：分型形成时间小于对应级别平均周期的1/3 弱势特征：形成时间超过平均周期2倍 """ # 简化处理：基于分型K线的收敛程度 # 分型区间内的价格收敛速度越快，形成速度越快 if self.fx == Chan_FX_TYPE.TOP: # 顶分型：检查左右两根K线相对于中间K线的收敛程度 left_convergence = (self.high - self.pre.high) / self.high if self.high > 0 else 0 right_convergence = (self.high - self.next.high) / self.high if self.high > 0 else 0 else: # BOTTOM left_convergence = (self.pre.low - self.low) / self.low if self.low > 0 else 0 right_convergence = (self.next.low - self.low) / self.low if self.low > 0 else 0 avg_convergence = (left_convergence + right_convergence) / 2 if avg_convergence >= 0.03: # 快速形成 return 4 elif avg_convergence >= 0.02: return 3 elif avg_convergence >= 0.01: return 2 else: return 1 def _calculate_confirmation_score(self): """ 计算次级别确认得分 (最高10分) - 笔破坏检测：真实强势分型会破坏前一笔的趋势 - 观察分型后3根K线能否站稳分型区间1/2以上 - 结合技术指标确认 """ score = 0 # 1. 检查分型后确认（如果有next的next数据） if hasattr(self.next, 'next'): next2 = self.next.next if next2: if self.fx == Chan_FX_TYPE.TOP: # 顶分型：检查后续2根K线是否持续走弱 fx_mid_level = (self.high + min(self.pre.low, self.next.low)) / 2 if self.next.close < fx_mid_level and next2.close < fx_mid_level: score += 5 # 强确认 elif self.next.close < fx_mid_level: score += 3 # 中等确认 else: # BOTTOM # 底分型：检查后续2根K线是否持续走强 fx_mid_level = (max(self.pre.high, self.next.high) + self.low) / 2 if self.next.close > fx_mid_level and next2.close > fx_mid_level: score += 5 # 强确认 elif self.next.close > fx_mid_level: score += 3 # 中等确认 # 2. 技术指标确认 if hasattr(self, 'rsi'): if self.fx == Chan_FX_TYPE.TOP and self.rsi > 70: score += 2 # 超买区顶分型 elif self.fx == Chan_FX_TYPE.BOTTOM and self.rsi < 30: score += 2 # 超卖区底分型 # 3. 分型强度自身确认（K线形态） if self.fx == Chan_FX_TYPE.TOP: # 长上影线确认 upper_shadow = self.high - max(self.open, self.close) candle_range = self.high - self.low if candle_range > 0 and upper_shadow / candle_range > 0.5: score += 2 else: # BOTTOM # 长下影线确认 lower_shadow = min(self.open, self.close) - self.low candle_range = self.high - self.low if candle_range > 0 and lower_shadow / candle_range > 0.5: score += 2 # 4. 与前一个分型的关系 if self.pre and hasattr(self.pre, 'fx') and self.pre.fx != Chan_FX_TYPE.UNKNOWN: # 检查是否形成有效的笔结构 if self.fx != self.pre.fx: # 分型类型相反 score += 1 return min(10, score) def _calculate_recent_atr(self, lookback=10): """ 计算近期ATR（平均真实波动范围） """ tr_values = [] temp = self for i in range(lookback): if temp and temp.pre: tr = max( temp.high - temp.low, abs(temp.high - temp.pre.close), abs(temp.low - temp.pre.close) ) tr_values.append(tr) temp = temp.pre else: break return sum(tr_values) / len(tr_values) if tr_values else 0 def _calculate_average_volume(self, lookback=5): """ 计算平均成交量 """ volumes = [] temp = self.pre # 从前一根K线开始计算 for i in range(lookback): if temp: volumes.append(temp.volume) temp = temp.pre else: break return sum(volumes) / len(volumes) if volumes else 0 def get_fx_strength_level(self): """ 获取分型强度等级 根据专业评分标准：≥80分为有效强势分型，≤40分建议忽略 """ strength = self.calculate_fx_strength() return "" if strength >= 80: return "极强" elif strength >= 65: return "强" elif strength >= 50: return "中等" elif strength >= 40: return "弱" else: return "极弱" def is_strong_fx(self, threshold=65): """ 判断是否为强分型 根据专业标准调整阈值为65分 """ return self.calculate_fx_strength() >= threshold def _default_top_strength_judgment(self, first_info, middle_info, last_info, first_kline, middle_kline, last_kline): """ 顶分型默认强弱判断 当不满足特定强弱条件时的保底判断 """ # 严格的强分型判断条件 strong_signals = 0 # 判断条件1：成交量显著放大（提高标准） avg_volume = self._calculate_average_volume(lookback=5) volume_significantly_amplified = middle_kline.volume > avg_volume * 2.0 if avg_volume > 0 else False if volume_significantly_amplified: strong_signals += 1 # 判断条件2：中间K线有长上影线（提高标准） has_long_upper_shadow = middle_info['upper_shadow_ratio'] > 0.6 # 从0.3提高到0.6 if has_long_upper_shadow: strong_signals += 1 # 判断条件3：后续K线收盘明显偏低（更严格） middle_range = middle_kline.high - middle_kline.low last_close_position = (last_kline.close - middle_kline.low) / middle_range if middle_range > 0 else 0.5 close_significantly_low = last_close_position < 0.3 # 从0.6提高到0.3 if close_significantly_low: strong_signals += 1 # 判断条件4：最后一根K线是明显的阴线且跌幅较大 is_significant_bearish = (last_info['is_bearish'] and last_info['body_size'] > last_info['total_range'] * 0.5) if is_significant_bearish: strong_signals += 1 # 判断条件5：跌破前一根K线重要价位 breaks_important_level = last_kline.low < first_kline.low if breaks_important_level: strong_signals += 1 # 需要至少4个强信号才判断为强分型，否则为弱分型 return 1 if strong_signals >= 4 else -1 def _default_bottom_strength_judgment(self, first_info, middle_info, last_info, first_kline, middle_kline, last_kline): """ 底分型默认强弱判断 当不满足特定强弱条件时的保底判断 """ # 严格的强分型判断条件 strong_signals = 0 # 判断条件1：成交量显著放大（提高标准） avg_volume = self._calculate_average_volume(lookback=5) volume_significantly_amplified = middle_kline.volume > avg_volume * 2.0 if avg_volume > 0 else False if volume_significantly_amplified: strong_signals += 1 # 判断条件2：中间K线有长下影线（提高标准） has_long_lower_shadow = middle_info['lower_shadow_ratio'] > 0.6 # 从0.3提高到0.6 if has_long_lower_shadow: strong_signals += 1 # 判断条件3：后续K线收盘明显偏高（更严格） middle_range = middle_kline.high - middle_kline.low last_close_position = (last_kline.close - middle_kline.low) / middle_range if middle_range > 0 else 0.5 close_significantly_high = last_close_position > 0.7 # 从0.4降低到0.7 if close_significantly_high: strong_signals += 1 # 判断条件4：最后一根K线是明显的阳线且涨幅较大 is_significant_bullish = (last_info['is_bullish'] and last_info['body_size'] > last_info['total_range'] * 0.5) if is_significant_bullish: strong_signals += 1 # 判断条件5：突破前一根K线重要价位 breaks_important_level = last_kline.high > first_kline.high if breaks_important_level: strong_signals += 1 # 需要至少4个强信号才判断为强分型，否则为弱分型 return 1 if strong_signals >= 4 else -1