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 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: #print(klu.time, klu.fx_type, klu.fx_strength) 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 cal_fx_strength(self): """ 统一的分型强度计算函数 - 直接计算当前KLC的分型强度 包含技术指标确认 Returns: int: 强度评分 15-80分 """ # 如果不是分型，返回0 if self.fx == Chan_FX_TYPE.UNKNOWN: return 0 # 如果没有前一个KLC，返回基础分 if not self.pre: return 15 score = 15 # 基础分数，任何分型都有基础分 # === 1. 突出程度评分（0-25分） === if self.fx == Chan_FX_TYPE.TOP: # 顶分型：当前高点与前一个高点的差异 if self.pre.high > 0: prominence = abs(self.high - self.pre.high) / self.pre.high else: prominence = 0 else: # 底分型 # 底分型：当前低点与前一个低点的差异 if self.pre.low > 0: prominence = abs(self.pre.low - self.low) / self.pre.low else: prominence = 0 # 突出程度评分 - 极度宽松 if prominence >= 0.03: # 3%以上突出 score += 25 elif prominence >= 0.02: # 2-3%突出 score += 20 elif prominence >= 0.015: # 1.5-2%突出 score += 15 elif prominence >= 0.01: # 1-1.5%突出 score += 10 elif prominence >= 0.005: # 0.5-1%突出 score += 6 elif prominence >= 0.002: # 0.2-0.5%突出 score += 3 else: score += 1 # 任何突出度都给分 # === 2. K线形态评分（0-15分） === kline_range = self.high - self.low if kline_range > 0: if self.fx == Chan_FX_TYPE.TOP: # 顶分型看上影线 upper_shadow = self.high - max(self.open, self.close) shadow_ratio = upper_shadow / kline_range else: # 底分型看下影线 lower_shadow = min(self.open, self.close) - self.low shadow_ratio = lower_shadow / kline_range # 影线评分 - 极度宽松 if shadow_ratio >= 0.3: # 长影线 score += 15 elif shadow_ratio >= 0.2: # 明显影线 score += 12 elif shadow_ratio >= 0.1: # 一般影线 score += 8 elif shadow_ratio >= 0.05: # 短影线 score += 5 else: score += 2 # 有一点影线就给分 else: score += 2 # 十字星也给点分 # === 3. 成交量评分（0-10分） === avg_volume = self._calculate_average_volume(lookback=5) if avg_volume > 0: volume_ratio = self.volume / avg_volume if volume_ratio >= 2.0: # 大量 score += 10 elif volume_ratio >= 1.5: # 明显放量 score += 8 elif volume_ratio >= 1.2: # 适度放量 score += 6 elif volume_ratio >= 1.0: # 正常量 score += 4 elif volume_ratio >= 0.8: # 略缩量 score += 2 else: score += 1 # 大幅缩量也给1分 else: score += 4 # 无法计算时给默认分 # === 4. 价格位置评分（0-10分） === if kline_range > 0: close_position = (self.close - self.low) / kline_range if self.fx == Chan_FX_TYPE.TOP: # 顶分型：收盘价越低越好 if close_position <= 0.3: # 收盘在下部 score += 10 elif close_position <= 0.5: # 收盘在中下部 score += 7 elif close_position <= 0.7: # 收盘在中上部 score += 4 else: score += 2 # 收盘位置偏高但还给分 else: # 底分型：收盘价越高越好 if close_position >= 0.7: # 收盘在上部 score += 10 elif close_position >= 0.5: # 收盘在中上部 score += 7 elif close_position >= 0.3: # 收盘在中下部 score += 4 else: score += 2 # 收盘位置偏低但还给分 else: score += 5 # 无区间时给中等分 # === 5. 技术指标确认（0-20分） === tech_score = 0 # 5.1 RSI确认（0-4分） if hasattr(self, 'rsi') and self.rsi is not None: if self.fx == Chan_FX_TYPE.TOP: # 顶分型：RSI超买确认 if self.rsi >= 80: # 严重超买 tech_score += 4 elif self.rsi >= 70: # 超买 tech_score += 3 elif self.rsi >= 60: # 偏高 tech_score += 2 elif self.rsi >= 50: # 中性偏高 tech_score += 1 else: # 底分型：RSI超卖确认 if self.rsi <= 20: # 严重超卖 tech_score += 4 elif self.rsi <= 30: # 超卖 tech_score += 3 elif self.rsi <= 40: # 偏低 tech_score += 2 elif self.rsi <= 50: # 中性偏低 tech_score += 1 # 5.2 MACD确认（0-4分） if hasattr(self, 'macdhist') and self.macdhist is not None: if self.fx == Chan_FX_TYPE.TOP: # 顶分型：MACD背离或转弱 if self.macdhist < 0: # MACD柱状图为负 tech_score += 2 # 检查是否从正转负 if self.pre and hasattr(self.pre, 'macdhist') and self.pre.macdhist is not None: if self.pre.macdhist > 0: # 前一根为正 tech_score += 2 # 从正转负，额外加分 elif self.pre and hasattr(self.pre, 'macdhist') and self.pre.macdhist is not None: # 检查MACD是否减弱 if self.macdhist < self.pre.macdhist: tech_score += 1 else: # 底分型：MACD转强 if self.macdhist > 0: # MACD柱状图为正 tech_score += 2 # 检查是否从负转正 if self.pre and hasattr(self.pre, 'macdhist') and self.pre.macdhist is not None: if self.pre.macdhist < 0: # 前一根为负 tech_score += 2 # 从负转正，额外加分 elif self.pre and hasattr(self.pre, 'macdhist') and self.pre.macdhist is not None: # 检查MACD是否增强 if self.macdhist > self.pre.macdhist: tech_score += 1 # 5.3 布林带确认（0-4分） bb_score = self._analyze_bollinger_for_fx() tech_score += min(4, bb_score) # 5.4 EMA趋势确认（0-4分） ema_score = self._analyze_ema_for_fx() tech_score += min(4, ema_score) # 5.5 ATR波动率确认（0-4分） atr_score = self._analyze_atr_for_fx() tech_score += min(4, atr_score) score += min(20, tech_score) return min(80, max(15, score)) # 确保至少15分，最高80分 def _analyze_bollinger_for_fx(self): """ 布林带分析 - 统一版本 """ score = 0 # 计算简化的布林带（基于收盘价） closes = [self.close] temp = self.pre for i in range(19): # 布林带通常使用20周期 if temp: closes.append(temp.close) temp = temp.pre else: break if len(closes) >= 20: import numpy as np # 计算20周期移动平均线和标准差 ma20 = np.mean(closes[:20]) std = np.std(closes[:20]) # 布林带上轨和下轨 upper_band = ma20 + 2 * std lower_band = ma20 - 2 * std if self.fx == Chan_FX_TYPE.TOP: # 顶分型：价格接近或突破上轨 if self.high >= upper_band: # 触及或突破上轨 score += 4 elif self.close > ma20: # 计算价格在上半区的位置 if upper_band > ma20: position = (self.close - ma20) / (upper_band - ma20) if position > 0.8: # 接近上轨 score += 3 elif position > 0.6: score += 2 elif position > 0.3: score += 1 else: # 底分型：价格接近或突破下轨 if self.low <= lower_band: # 触及或突破下轨 score += 4 elif self.close < ma20: # 计算价格在下半区的位置 if ma20 > lower_band: position = (ma20 - self.close) / (ma20 - lower_band) if position > 0.8: # 接近下轨 score += 3 elif position > 0.6: score += 2 elif position > 0.3: score += 1 return score def _analyze_ema_for_fx(self): """ EMA趋势分析 """ score = 0 # 获取多周期收盘价用于EMA计算 closes = [self.close] temp = self.pre for i in range(29): # 获取30根K线用于EMA计算 if temp: closes.append(temp.close) temp = temp.pre else: break if len(closes) >= 12: # 至少需要12根K线 import numpy as np # 计算EMA12和EMA26 def calculate_ema(prices, period): alpha = 2 / (period + 1) ema = [prices[0]] for price in prices[1:period]: ema.append(alpha * price + (1 - alpha) * ema[-1]) return ema[-1] if len(ema) > 0 else prices[0] if len(closes) >= 12: ema12 = calculate_ema(closes[:12][::-1], 12) # 反转顺序，最新的在前 if len(closes) >= 26: ema26 = calculate_ema(closes[:26][::-1], 26) if self.fx == Chan_FX_TYPE.TOP: # 顶分型：价格高于EMA，EMA向上但可能转向 if self.close > ema12 > ema26: # 多头排列 score += 2 elif self.close > ema12: # 价格在短期EMA上方 score += 1 # 检查EMA12是否开始转向 if self.pre and len(closes) >= 13: prev_ema12 = calculate_ema(closes[1:13][::-1], 12) if ema12 < prev_ema12: # EMA12开始下降 score += 2 else: # 底分型：价格低于EMA，EMA向下但可能转向 if self.close < ema12 < ema26: # 空头排列 score += 2 elif self.close < ema12: # 价格在短期EMA下方 score += 1 # 检查EMA12是否开始转向 if self.pre and len(closes) >= 13: prev_ema12 = calculate_ema(closes[1:13][::-1], 12) if ema12 > prev_ema12: # EMA12开始上升 score += 2 return score def _analyze_atr_for_fx(self): """ ATR波动率分析 """ score = 0 # 计算ATR recent_atr = self._calculate_recent_atr(lookback=14) if recent_atr > 0: # 当前K线的真实波动范围 current_tr = self.high - self.low if self.pre: current_tr = max( self.high - self.low, abs(self.high - self.pre.close), abs(self.low - self.pre.close) ) # ATR倍数 atr_ratio = current_tr / recent_atr if self.fx == Chan_FX_TYPE.TOP: # 顶分型：波动率放大确认反转 if atr_ratio >= 2.5: # 波动率大幅放大 score += 4 elif atr_ratio >= 2.0: # 波动率明显放大 score += 3 elif atr_ratio >= 1.5: # 波动率适度放大 score += 2 elif atr_ratio >= 1.2: # 波动率略微放大 score += 1 else: # 底分型：波动率放大确认反转 if atr_ratio >= 2.5: # 波动率大幅放大 score += 4 elif atr_ratio >= 2.0: # 波动率明显放大 score += 3 elif atr_ratio >= 1.5: # 波动率适度放大 score += 2 elif atr_ratio >= 1.2: # 波动率略微放大 score += 1 # 额外检查：波动率从低到高的变化 if self.pre: prev_tr = self.pre.high - self.pre.low if self.pre.pre: prev_tr = max( self.pre.high - self.pre.low, abs(self.pre.high - self.pre.pre.close), abs(self.pre.low - self.pre.pre.close) ) # 波动率加速放大 if current_tr > prev_tr * 1.5: score += 1 return score def get_fx_strength_level(self): """ 获取分型强度等级 Returns: str: 强度等级描述 """ strength = self.cal_fx_strength() # 调整后的强度等级阈值（匹配15-80分范围） if strength >= 70: return "极强分型" # 极强分型：70分以上 elif strength >= 60: return "强分型" # 强分型：60-69分 elif strength >= 50: return "中强分型" # 中强分型：50-59分 elif strength >= 40: return "中等分型" # 中等分型：40-49分 elif strength >= 25: return "弱分型" # 弱分型：25-39分 else: return "极弱分型" # 极弱分型：25分以下 def is_strong_fx(self, threshold=55): """ 判断是否为强分型 Args: threshold: 强分型的阈值，调整为55分（适配80分制） Returns: bool: 是否为强分型 """ return self.cal_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 def get_real_price_range(self): """ 获取真实价格区间（合并前所有原始K线的最高最低点） 返回: (real_high, real_low) """ if not self.klus: return self.high, self.low real_high = max(klu.high for klu in self.klus) real_low = min(klu.low for klu in self.klus) return real_high, real_low def get_real_range_size(self): """ 获取真实价格区间大小 """ real_high, real_low = self.get_real_price_range() return real_high - real_low def get_real_close_position(self): """ 获取收盘价在真实价格区间中的位置 """ real_high, real_low = self.get_real_price_range() real_range = real_high - real_low if real_range > 0: return (self.close - real_low) / real_range else: return 0.5 def get_real_upper_shadow_ratio(self): """ 获取上影线在真实区间中的比例 """ real_high, real_low = self.get_real_price_range() real_range = real_high - real_low if real_range > 0: upper_shadow = real_high - max(self.open, self.close) return upper_shadow / real_range else: return 0 def get_real_lower_shadow_ratio(self): """ 获取下影线在真实区间中的比例 """ real_high, real_low = self.get_real_price_range() real_range = real_high - real_low if real_range > 0: lower_shadow = min(self.open, self.close) - real_low return lower_shadow / real_range else: return 0 def _analyze_macd_for_top_fx(self): """ MACD指标在顶分型中的综合分析 """ score = 0 # 检查MACD背离 if self._check_macd_bearish_divergence(): score += 3 # 检查MACD柱状图趋势 if hasattr(self, 'macdhist') and self.macdhist is not None: if self.macdhist < 0: # MACD柱状图为负 score += 1 # 检查MACD柱状图是否从正转负 if self.pre and hasattr(self.pre, 'macdhist') and self.pre.macdhist is not None: if self.pre.macdhist > 0 and self.macdhist < 0: score += 2 # 检查MACD线是否在零轴上方形成顶背离 klu_features = self.cal_klu_features() if 'klu_macd' in klu_features and 'klu_signal' in klu_features: macd_line = klu_features['klu_macd'] signal_line = klu_features['klu_signal'] # MACD线高于信号线但趋势减弱 if macd_line > signal_line and macd_line > 0: score += 1 return score def _analyze_macd_for_bottom_fx(self): """ MACD指标在底分型中的综合分析 """ score = 0 # 检查MACD背离 if self._check_macd_bullish_divergence(): score += 3 # 检查MACD柱状图趋势 if hasattr(self, 'macdhist') and self.macdhist is not None: if self.macdhist > 0: # MACD柱状图为正 score += 1 # 检查MACD柱状图是否从负转正 if self.pre and hasattr(self.pre, 'macdhist') and self.pre.macdhist is not None: if self.pre.macdhist < 0 and self.macdhist > 0: score += 2 # 检查MACD线是否在零轴下方形成底背离 klu_features = self.cal_klu_features() if 'klu_macd' in klu_features and 'klu_signal' in klu_features: macd_line = klu_features['klu_macd'] signal_line = klu_features['klu_signal'] # MACD线低于信号线但趋势增强 if macd_line < signal_line and macd_line < 0: score += 1 return score def _analyze_kdj_for_top_fx(self): """ KDJ指标在顶分型中的分析 """ score = 0 # 模拟KDJ计算（基于真实价格区间） real_high, real_low = self.get_real_price_range() # 获取前面几根K线的最高最低价 temp = self.pre highs = [real_high] lows = [real_low] closes = [self.close] for i in range(8): # KDJ通常使用9周期 if temp: temp_high, temp_low = temp.get_real_price_range() highs.append(temp_high) lows.append(temp_low) closes.append(temp.close) temp = temp.pre else: break if len(highs) >= 9: # 计算9周期的最高价和最低价 highest_high = max(highs[:9]) lowest_low = min(lows[:9]) # 计算RSV（未成熟随机值） if highest_high > lowest_low: rsv = (self.close - lowest_low) / (highest_high - lowest_low) * 100 # 简化的K值计算 k_value = rsv # 简化处理 # KDJ超买判断 if k_value > 80: # K值超买 score += 3 elif k_value > 70: score += 2 # 检查KDJ死叉形态 if self.pre: pre_highs = highs[1:10] if len(highs) > 9 else highs[1:] pre_lows = lows[1:10] if len(lows) > 9 else lows[1:] if pre_highs and pre_lows: pre_highest = max(pre_highs) pre_lowest = min(pre_lows) if pre_highest > pre_lowest: pre_rsv = (self.pre.close - pre_lowest) / (pre_highest - pre_lowest) * 100 pre_k_value = pre_rsv # 检查K值是否从高位下降 if pre_k_value > k_value and pre_k_value > 70: score += 2 return score def _analyze_kdj_for_bottom_fx(self): """ KDJ指标在底分型中的分析 """ score = 0 # 模拟KDJ计算（基于真实价格区间） real_high, real_low = self.get_real_price_range() # 获取前面几根K线的最高最低价 temp = self.pre highs = [real_high] lows = [real_low] closes = [self.close] for i in range(8): # KDJ通常使用9周期 if temp: temp_high, temp_low = temp.get_real_price_range() highs.append(temp_high) lows.append(temp_low) closes.append(temp.close) temp = temp.pre else: break if len(highs) >= 9: # 计算9周期的最高价和最低价 highest_high = max(highs[:9]) lowest_low = min(lows[:9]) # 计算RSV（未成熟随机值） if highest_high > lowest_low: rsv = (self.close - lowest_low) / (highest_high - lowest_low) * 100 # 简化的K值计算 k_value = rsv # 简化处理 # KDJ超卖判断 if k_value < 20: # K值超卖 score += 3 elif k_value < 30: score += 2 # 检查KDJ金叉形态 if self.pre: pre_highs = highs[1:10] if len(highs) > 9 else highs[1:] pre_lows = lows[1:10] if len(lows) > 9 else lows[1:] if pre_highs and pre_lows: pre_highest = max(pre_highs) pre_lowest = min(pre_lows) if pre_highest > pre_lowest: pre_rsv = (self.pre.close - pre_lowest) / (pre_highest - pre_lowest) * 100 pre_k_value = pre_rsv # 检查K值是否从低位上升 if k_value > pre_k_value and pre_k_value < 30: score += 2 return score def _calculate_recent_atr(self, lookback=14): """ 计算最近的ATR（平均真实波动范围） Args: lookback: 回看周期，默认14 Returns: float: ATR值 """ if not self.pre: return 0 true_ranges = [] temp = self for i in range(lookback): if temp and temp.pre: # 计算真实波动范围（TR） tr = max( temp.high - temp.low, # 当前高低价差 abs(temp.high - temp.pre.close), # 当前高价与前收盘价差的绝对值 abs(temp.low - temp.pre.close) # 当前低价与前收盘价差的绝对值 ) true_ranges.append(tr) temp = temp.pre else: break if true_ranges: return sum(true_ranges) / len(true_ranges) else: return 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 _check_macd_bearish_divergence(self): """ 检查MACD看跌背离 """ # 简化版本，可以根据实际MACD数据进行更复杂的背离分析 if hasattr(self, 'macdhist') and self.macdhist: # 如果MACD柱状图在减弱，可能形成顶背离 if self.pre and hasattr(self.pre, 'macdhist') and self.pre.macdhist: if self.macdhist < self.pre.macdhist and self.macdhist < 0: return True return False def _check_macd_bullish_divergence(self): """ 检查MACD看涨背离 """ # 简化版本，可以根据实际MACD数据进行更复杂的背离分析 if hasattr(self, 'macdhist') and self.macdhist: # 如果MACD柱状图在增强，可能形成底背离 if self.pre and hasattr(self.pre, 'macdhist') and self.pre.macdhist: if self.macdhist > self.pre.macdhist and self.macdhist > 0: return True return False def _near_resistance_level(self): """ 检查是否接近阻力位（使用真实价格区间） """ # 简化版本：检查是否接近前面几根K线的最高点 if self.pre: real_high, _ = self.get_real_price_range() temp = self.pre max_high = 0 for i in range(10): # 检查前10根K线 if temp: temp_real_high, _ = temp.get_real_price_range() max_high = max(max_high, temp_real_high) temp = temp.pre else: break if max_high > 0: # 如果当前真实高点接近前期高点，可能是阻力位 distance_ratio = abs(real_high - max_high) / max_high return distance_ratio < 0.02 # 2%以内算接近 return False def _near_support_level(self): """ 检查是否接近支撑位（使用真实价格区间） """ # 简化版本：检查是否接近前面几根K线的最低点 if self.pre: _, real_low = self.get_real_price_range() temp = self.pre min_low = float('inf') for i in range(10): # 检查前10根K线 if temp: _, temp_real_low = temp.get_real_price_range() min_low = min(min_low, temp_real_low) temp = temp.pre else: break if min_low != float('inf') and min_low > 0: # 如果当前真实低点接近前期低点，可能是支撑位 distance_ratio = abs(real_low - min_low) / min_low return distance_ratio < 0.02 # 2%以内算接近 return False def cal_fx_strength_realtime(self): """ 实时计算分型强度 - 严格版本（不使用缓存，强制重新计算） 基于最后两根K线评估分型强度，不使用未来数据 Returns: int: 强度评分 0-70分 """ if not self.pre or not self.pre.is_fx(): return 0 fx_type = self.pre.fx_type # 强制重新计算，不使用任何缓存 if fx_type == Chan_FX_TYPE.TOP: strength = self._calculate_top_fx_power_realtime_v2() elif fx_type == Chan_FX_TYPE.BOTTOM: strength = self._calculate_bottom_fx_power_realtime_v2() else: strength = 0 return strength def _calculate_top_fx_power_realtime_v2(self): """ 宽松版本的实时顶分型力度计算 - 确保合理分型有分数 """ score = 10 # 提高基础分数，确认是分型就有基础分 # 获取前一个KLC的最高价用于比较 if not self.pre: return score # 1. 突出程度评分（0-25分）- 大幅放宽标准 current_high = self.high prev_high = self.pre.high # 计算突出程度 - 修正计算逻辑 if prev_high > 0: high_prominence = abs(current_high - prev_high) / prev_high else: high_prominence = 0 # 极度放宽的评分标准 if high_prominence >= 0.05: # 5%以上突出 - 极强 score += 25 elif high_prominence >= 0.03: # 3-5%突出 - 很强 score += 20 elif high_prominence >= 0.02: # 2-3%突出 - 强 score += 15 elif high_prominence >= 0.015: # 1.5-2%突出 - 中等 score += 12 elif high_prominence >= 0.01: # 1-1.5%突出 - 较弱 score += 8 elif high_prominence >= 0.005: # 0.5-1%突出 - 弱 score += 5 elif high_prominence >= 0.002: # 0.2-0.5%突出 - 极弱 score += 2 else: score += 1 # 有一定突出度就给点分 # 2. K线形态评分（0-20分）- 大幅放宽 kline_range = self.high - self.low if kline_range > 0: upper_shadow = self.high - max(self.open, self.close) upper_shadow_ratio = upper_shadow / kline_range if upper_shadow_ratio >= 0.4: # 长上影线 score += 20 elif upper_shadow_ratio >= 0.25: # 明显上影线 score += 15 elif upper_shadow_ratio >= 0.15: # 一般上影线 score += 10 elif upper_shadow_ratio >= 0.08: # 短上影线 score += 6 elif upper_shadow_ratio >= 0.03: # 很短上影线 score += 3 else: score += 1 # 有一点上影线就给分 # 3. 成交量评分（0-15分）- 大幅放宽 avg_volume = self._calculate_average_volume(lookback=5) if avg_volume > 0: volume_ratio = self.volume / avg_volume if volume_ratio >= 2.5: # 大量 score += 15 elif volume_ratio >= 1.8: # 明显放量 score += 12 elif volume_ratio >= 1.3: # 适度放量 score += 9 elif volume_ratio >= 1.1: # 轻微放量 score += 6 elif volume_ratio >= 0.8: # 正常量 score += 3 elif volume_ratio >= 0.5: # 缩量但可接受 score += 1 else: score += 0 # 极度缩量 else: score += 3 # 无法计算成交量时给默认分 # 4. 价格位置评分（0-10分）- 大幅放宽 if kline_range > 0: close_position = (self.close - self.low) / kline_range if close_position <= 0.2: # 收盘在下部 score += 10 elif close_position <= 0.4: # 收盘在中下部 score += 8 elif close_position <= 0.6: # 收盘在中部 score += 5 elif close_position <= 0.8: # 收盘在中上部 score += 3 else: score += 1 # 收盘位置偏高但还有分 # === 去除大部分惩罚机制，只保留最基本的 === # 只有在完全没有突出度时才轻微降分 if high_prominence < 0.001: # 突出度低于0.1% score = int(score * 0.8) return min(60, max(10, score)) # 确保至少有10分，最高60分 def _calculate_bottom_fx_power_realtime_v2(self): """ 宽松版本的实时底分型力度计算 - 确保合理分型有分数 """ score = 10 # 提高基础分数，确认是分型就有基础分 # 获取前一个KLC的最低价用于比较 if not self.pre: return score # 1. 突出程度评分（0-25分）- 大幅放宽标准 current_low = self.low prev_low = self.pre.low # 计算突出程度 - 修正计算逻辑 if prev_low > 0: low_prominence = abs(prev_low - current_low) / prev_low else: low_prominence = 0 # 极度放宽的评分标准 if low_prominence >= 0.05: # 5%以上突出 - 极强 score += 25 elif low_prominence >= 0.03: # 3-5%突出 - 很强 score += 20 elif low_prominence >= 0.02: # 2-3%突出 - 强 score += 15 elif low_prominence >= 0.015: # 1.5-2%突出 - 中等 score += 12 elif low_prominence >= 0.01: # 1-1.5%突出 - 较弱 score += 8 elif low_prominence >= 0.005: # 0.5-1%突出 - 弱 score += 5 elif low_prominence >= 0.002: # 0.2-0.5%突出 - 极弱 score += 2 else: score += 1 # 有一定突出度就给点分 # 2. K线形态评分（0-20分）- 大幅放宽 kline_range = self.high - self.low if kline_range > 0: lower_shadow = min(self.open, self.close) - self.low lower_shadow_ratio = lower_shadow / kline_range if lower_shadow_ratio >= 0.4: # 长下影线 score += 20 elif lower_shadow_ratio >= 0.25: # 明显下影线 score += 15 elif lower_shadow_ratio >= 0.15: # 一般下影线 score += 10 elif lower_shadow_ratio >= 0.08: # 短下影线 score += 6 elif lower_shadow_ratio >= 0.03: # 很短下影线 score += 3 else: score += 1 # 有一点下影线就给分 # 3. 成交量评分（0-15分）- 大幅放宽 avg_volume = self._calculate_average_volume(lookback=5) if avg_volume > 0: volume_ratio = self.volume / avg_volume if volume_ratio >= 2.5: # 大量 score += 15 elif volume_ratio >= 1.8: # 明显放量 score += 12 elif volume_ratio >= 1.3: # 适度放量 score += 9 elif volume_ratio >= 1.1: # 轻微放量 score += 6 elif volume_ratio >= 0.8: # 正常量 score += 3 elif volume_ratio >= 0.5: # 缩量但可接受 score += 1 else: score += 0 # 极度缩量 else: score += 3 # 无法计算成交量时给默认分 # 4. 价格位置评分（0-10分）- 大幅放宽 if kline_range > 0: close_position = (self.close - self.low) / kline_range if close_position >= 0.8: # 收盘在上部 score += 10 elif close_position >= 0.6: # 收盘在中上部 score += 8 elif close_position >= 0.4: # 收盘在中部 score += 5 elif close_position >= 0.2: # 收盘在中下部 score += 3 else: score += 1 # 收盘位置偏低但还有分 # === 去除大部分惩罚机制，只保留最基本的 === # 只有在完全没有突出度时才轻微降分 if low_prominence < 0.001: # 突出度低于0.1% score = int(score * 0.8) return min(60, max(10, score)) # 确保至少有10分，最高60分