import copy from typing import Dict, Optional from ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR, Chan_BI_DIR, Chan_KLC_FX 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 = 0 # === 新增：KLC类型 === self.klc_type = None # 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 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 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) def contain_klu_fx(self): if len(self.klus) > 0: for klu in self.klus: klu.update_realtime_analysis() if klu.fx_type == self.fx and klu.fx_strength > 1.8: return True return False 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 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 check_fx_klu_strength(self): if self.pre and self.next and self.next.end_klu: strength = 0 klc1 = self.pre klc2 = self klc3 = self.next # 检查包含关系 inc = self.check_include_relation(klc1, klc2, klc3) if not inc: # 检查分型类型 if self.fx == Chan_FX_TYPE.TOP: # (1)没有包含关系+1 strength += 0 print(self.start_time, "1") # (2)第1条K线是一条大阳线，而第2、3条K线是小阴线、小阳线，那么这个分型结构的意义就不大了，强度 -1 if self.pre.cal_klu_min_max() > 0.5 and self.cal_klu_min_max() < 0.2 and self.next.cal_klu_min_max() < 0.2: strength += -1 print(self.start_time, "2") # (3)第2条K线有长上影线或者就是大阴线，而第3条K线不能以阳线收在第2条K线区间的一半之上，那么该顶分型的力度就比较大 if ((self.close < self.open and self.cal_klu_min_max() > 0.5) or self.cal_klu_upper_shadow() > 0.6) and not (self.next.close > self.next.open and self.next.close > (self.high-self.low)/2): strength += 1 print(self.start_time, "3") else: if self.fx == Chan_FX_TYPE.TOP: # (4)第2条K线和第3条K线为包含关系，而第3条K线为大阴线(直接把为阳线的第2条K线“吃掉”)，这是最坏的一种包含关系。 if self.start_klu.index - self.end_klu.index < 0: klu1 = self.start_klu klu2 = self.klus[1] print(self.start_time, self.end_time, "length: ", len(self.klus), klu1.index, klu2.index) if klu2.close < klu2.open and 100*abs(klu2.open - klu2.close) / klu2.open > 0.5 and klu1.high <= klu2.high and klu1.low >= klu2.low: strength += 1 print(self.start_time, "4") # (5)第3条K线如果跌破第1条K线的底而且不能高于第1条K线区间的一半之上，则属于较弱的一种，也就是说这种顶分型出现后股价见顶的可能性不大。 if self.next.low < self.pre.low and self.next.high < (self.pre.high + self.pre.low)/2: strength += -1 print(self.start_time, "5") return strength else: return 0 def cal_klu_upper_shadow(self): """ 计算KLC的上影线长度 上影线 = 最高价 - max(开盘价, 收盘价) """ if self.high <= 0: # 避免无效数据 return 0 # 计算上影线长度 upper_shadow = self.high - max(self.open, self.close) # 计算相对上影线长度（相对于整个K线区间） total_range = self.high - self.low if total_range > 0: upper_shadow_ratio = upper_shadow / total_range else: upper_shadow_ratio = 0 return upper_shadow_ratio def cal_klu_min_max(self): """ 计算KLC类型：大阳线、大阴线、小阳线、小阴线 """ if self.open <= 0: # 避免除零错误 return 0 line_type = 1000*abs(self.open - self.close) / self.open print(self.start_time, line_type) return line_type def check_include_relation(self, klc1, klc2, klc3): if klc1.start_klu.index - klc1.end_klu.index == 0 and klc2.start_klu.index - klc2.end_klu.index == 0 and klc3.start_klu.index - klc3.end_klu.index == 0: return False else: return True def cal_fx_strength(self): """ 用self.pre和self.next实现分型强弱判断 核心缠论原理： - 强分型：出现在笔的末端，能够终结当前笔，标志着趋势转折 - 弱分型：出现在笔的中间，是中继性质，笔还会继续延伸 返回值： 3: 极强分型（笔终结+强确认） 2: 强分型（笔终结） 1: 偏强分型（可能终结笔） 0: 中性分型 -1: 偏弱分型（中继特征明显） -2: 弱分型（明显中继） -3: 极弱分型（无效分型） """ #return self.check_fx_klu_strength() # 检查是否为分型，且有前后K线数据 if self.fx == Chan_FX_TYPE.UNKNOWN or self.klc_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 #base_score += self.check_fx_klu_strength() #print(self.start_time, base_score, is_bi_end, post_fx_confirmation, fx_quality) # 限制在-3到3范围内 return max(-3, min(3, base_score)) 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_klcs = [] temp = self.next for i in range(5): # 检查后续5根K线 if temp: subsequent_klcs.append(temp) temp = temp.next if hasattr(temp, 'next') else None else: break if len(subsequent_klcs) < 2: return 0 if self.fx == Chan_FX_TYPE.TOP: return self._check_top_bi_ending(subsequent_klcs) else: # BOTTOM return self._check_bottom_bi_ending(subsequent_klcs) def _check_top_bi_ending(self, subsequent_klcs): """检查顶分型是否为笔终结""" # 强烈笔终结特征： # 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) last_klc = None for i, klc in enumerate(subsequent_klcs): # 检查是否跌破关键支撑 if klc.low < key_support: broken_key_levels += 1 # 检查是否出现新高 if klc.high > self.high: new_highs += 1 # 检查下跌趋势 if i > 0 and klc.close < subsequent_klcs[i-1].close: downward_trend += 1 last_klc = klc print(last_klc.start_time, last_klc.end_time) # 强烈笔终结：跌破关键位且无新高 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_klcs): """检查底分型是否为笔终结""" # 强烈笔终结特征： # 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) last_klc = None for i, klc in enumerate(subsequent_klcs): # 检查是否突破关键阻力 if klc.high > key_resistance: broken_key_levels += 1 # 检查是否出现新低 if klc.low < self.low: new_lows += 1 # 检查上涨趋势 if i > 0 and klc.close > subsequent_klcs[i-1].close: upward_trend += 1 last_klc = klc print(last_klc.start_time, last_klc.end_time) # 强烈笔终结：突破关键位且无新低 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_klc = self.next if self.fx == Chan_FX_TYPE.TOP: # 顶分型：第三根K线应该走弱 middle_price = (self.high + self.low) / 2 if third_klc.close < middle_price: score += 0.5 if third_klc.low < self.pre.low: # 跌破第一根K线低点 score += 0.5 if third_klc.close < third_klc.open and abs(third_klc.close - third_klc.open) > abs(self.close - self.open) * 0.5: score += 0.3 # 明显阴线 else: # BOTTOM # 底分型：第三根K线应该走强 middle_price = (self.high + self.low) / 2 if third_klc.close > middle_price: score += 0.5 if third_klc.high > self.pre.high: # 突破第一根K线高点 score += 0.5 if third_klc.close > third_klc.open and abs(third_klc.close - third_klc.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 == 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._calculate_average_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 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