from datetime import timedelta from pandas import DataFrame from ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR, Chan_BI_DIR, Chan_SEG_DIR, Chan_ZS_DIR, Chan_BSP_DIR, Chan_BSP_TYPE, Chan_KLC_FX, Chan_PRICE_TREND, Chan_KLU_PATTERN, Chan_K_DIR, Chan_KLC_STATE from ChanKLU import ChanKLU from ChanKLC import ChanKLC from ChanBI import ChanBI from ChanSBI import ChanSBI from ChanSEG import ChanSEG from ChanZS import ChanZS, ChanZS_Big from ChanBIZS import ChanBIZS from ChanBSP import ChanBSP import talib.abstract as ta import pandas as pd from technical.util import resample_to_interval from decimal import Decimal import numpy as np from ChanMACD import ChanMACD class TF_DF(): def __init__(self, df=None, interval=0, timeframe=None): if df is not None: self.init_TF_DF(df, interval, timeframe) def init_TF_DF(self, df, interval, timeframe): self.timeframe = timeframe self.interval = interval # 检查 DataFrame 是否为空或没有 date 列 if df is None or df.empty: raise ValueError(f"DataFrame for {timeframe} is empty. Please download data first.") if 'date' not in df.columns: raise ValueError(f"DataFrame for {timeframe} missing 'date' column. Columns: {df.columns.tolist()}") # interval=1 时不需要重采样 if interval == 1: self.dataframe = df.copy() else: self.dataframe = resample_to_interval(df, interval) #print(self.timeframe, len(self.dataframe)) self.dataframe = self.add_indicators(self.dataframe) self.klu_list = [] self.klc_list = [] self.bi_list = [] self.zs_list = [] self.bsp_list = [] self.seg_list = [] self.klc_fx_list = [] self.klu_list = self.cal_kl_data(self.dataframe) self.klc_list = self.get_klc_list(self.klu_list) self.bi_list = self.cal_bi_list(self.klc_list) self.seg_list = self.get_seg_list(self.bi_list) self.zs_list = self.get_zs_list(self.bi_list, self.seg_list) self.big_zs_list = self.get_big_zs_list(self.zs_list) self.chanmacd = ChanMACD(self.klu_list) self.klu_list = self.chanmacd.cal_macd_state() def get_ema52(self, index=-1): if self.klu_list: ema52_value = self.klu_list[index].ema52 # 处理NaN值 if pd.isna(ema52_value) or ema52_value is None: return None return float(ema52_value) return None def get_ema24(self, index=-1): if self.klu_list: ema24_value = self.klu_list[index].ema24 # 处理NaN值 if pd.isna(ema24_value) or ema24_value is None: return None return float(ema24_value) return None def get_current_klc(self): if len(self.klc_list) > 0: return self.klc_list[-2] return None def add_indicators(self, df): fast = 26 slow = 52 period = 9 macd = ta.MACD(df, fastperiod=fast, slowperiod=slow, signalperiod=period) bb365 = ta.BBANDS(df, timeperiod=365, nbdevup=3.0, nbdevdn=3.0, matype=0) bb120 = ta.BBANDS(df, timeperiod=120, nbdevup=3.0, nbdevdn=3.0, matype=0) bb30 = ta.BBANDS(df, timeperiod=41, nbdevup=2.3, nbdevdn=2.3, matype=0) bb302 = ta.BBANDS(df, timeperiod=41, nbdevup=2.0, nbdevdn=2.0, matype=0) bb30 = ta.BBANDS(df, timeperiod=20, nbdevup=2.0, nbdevdn=2.0, matype=0) bb302 = ta.BBANDS(df, timeperiod=20, nbdevup=2.0, nbdevdn=2.0, matype=0) bb2633 = ta.BBANDS(df, timeperiod=26, nbdevup=3.0, nbdevdn=3.0, matype=0) # 计算布林带中轨（移动平均线） bb30_middle = ta.SMA(df, timeperiod=90) # 手动计算布林带 %B 指标 (BBP) # %B = (Price - Lower Band) / (Upper Band - Lower Band) bbp365 = (df['close'] - bb365['lowerband']) / (bb365['upperband'] - bb365['lowerband']) bbp120 = (df['close'] - bb120['lowerband']) / (bb120['upperband'] - bb120['lowerband']) bbp30 = (df['close'] - bb30['lowerband']) / (bb30['upperband'] - bb30['lowerband']) bbp302 = (df['close'] - bb302['lowerband']) / (bb302['upperband'] - bb302['lowerband']) bbp2633 = (df['close'] - bb2633['lowerband']) / (bb2633['upperband'] - bb2633['lowerband']) df['bb2633upper'] = bb2633['upperband'] df['bb2633lower'] = bb2633['lowerband'] df['bbp2633'] = bbp2633 df['bb2633middle'] = bb2633['middleband'] df['atr'] = ta.ATR(df, timeperiod=14) df['bbup365'] = bb365['upperband'] df['bblow365'] = bb365['lowerband'] df['bbp365'] = bbp365 df['bbup120'] = bb120['upperband'] df['bblow120'] = bb120['lowerband'] df['bbp120'] = bbp120 df['bbup30'] = bb30['upperband'] df['bblow30'] = bb30['lowerband'] df['bbmiddle30'] = bb30_middle # 添加bb30中轨 df['bbp30'] = bbp30 df['bbup302'] = bb302['upperband'] df['bblow302'] = bb302['lowerband'] df['bbp302'] = bbp302 df['macd'] = macd['macd'] df['macdsignal'] = macd['macdsignal'] df['macdhist'] = macd['macdhist'] df['ema5'] = ta.EMA(df, timeperiod=5) df['ema10'] = ta.EMA(df, timeperiod=10) df['ema24'] = ta.EMA(df, timeperiod=24) df['ema52'] = ta.EMA(df, timeperiod=52) df['ema104'] = ta.EMA(df, timeperiod=104) df['ema156'] = ta.EMA(df, timeperiod=156) df['ema208'] = ta.EMA(df, timeperiod=208) df['ema26'] = ta.EMA(df, timeperiod=26) df['ema13'] = ta.EMA(df, timeperiod=13) df['ema7'] = ta.EMA(df, timeperiod=7) df['rsi'] = ta.RSI(df, timeperiod=14) df['volume_ratio'] = self.cal_volume_ratio(df) return df def get_klu_state(self, dataframe): klc_list = self.get_klc_list(self.get_klu_list(dataframe)) bi_list = self.cal_bi_list(klc_list) klu_state_list = [] klc_index = 0 for index in range(0, len(dataframe)): if klc_index == len(klc_list): klc_index = len(klc_list) - 1 klc = klc_list[klc_index] if klc.end_klu and klc.end_klu.idx == index: if klc.klc_state == Chan_KLC_STATE.S10: klu_state_list.append("10") #print(klc.end_time, klc.klc_fx_type) elif klc.klc_state == Chan_KLC_STATE.S_10: klu_state_list.append("-10") #print(klc.end_time, klc.klc_fx_type) elif klc.klc_state == Chan_KLC_STATE.S11: klu_state_list.append("11") #print(klc.end_time, klc.klc_fx_type) elif klc.klc_state == Chan_KLC_STATE.S_11: klu_state_list.append("-11") #print(klc.end_time, klc.klc_fx_type) else: klu_state_list.append("00") klc_index += 1 else: klu_state_list.append("00") print(klu_state_list[:20]) return klu_state_list def get_bsp_signal_data(self, dataframe): klu_list = self.get_klu_list(dataframe) klc_list = self.get_klc_list(klu_list) bi_list = self.cal_bi_list(klc_list) bi_zs_list = self.cal_bi_zs_list_pure(bi_list) bsp_list = self.find_all_bsp(bi_list, bi_zs_list) bsp_by_bi_type = {} for bsp in bsp_list: if bsp and bsp.bi: bsp_by_bi_type[(bsp.bi.index, bsp.type)] = bsp bsp_state_list = [0] * len(dataframe) bsp_zg_list = [0.0] * len(dataframe) bsp_zd_list = [0.0] * len(dataframe) bsp_stop_price_list = [0.0] * len(dataframe) bsp_risk_ratio_list = [0.0] * len(dataframe) klc_index = 0 def set_bsp_signal(index, state, bsp): bsp_state_list[index] = state if not bsp or not bsp.zs: return close = float(dataframe.iloc[index]['close']) atr = float(dataframe.iloc[index]['atr']) if 'atr' in dataframe.columns and not pd.isna(dataframe.iloc[index]['atr']) else 0.0 atr_ratio = atr / close if close > 0 else 0.0 buffer = atr * 0.1 bsp_zg_list[index] = bsp.zs.zg bsp_zd_list[index] = bsp.zs.zd if state == -1: stop_price = bsp.zs.zg - buffer risk_ratio = (close - stop_price) / close if close > stop_price else atr_ratio else: stop_price = bsp.zs.zd + buffer risk_ratio = (stop_price - close) / close if close < stop_price else atr_ratio bsp_stop_price_list[index] = stop_price bsp_risk_ratio_list[index] = max(0.001, min(float(risk_ratio), 0.02)) for index in range(0, len(dataframe)): if klc_index == len(klc_list): klc_index = len(klc_list) - 1 klc = klc_list[klc_index] if klc.end_klu and klc.end_klu.idx == index: if klc.klc_fx_type == Chan_KLC_FX.TOP2: bi = klc.bi.pre if bi and bi.is_sure and bi.end_klc.bsp_type == Chan_BSP_TYPE.B3: # 第三类买点 set_bsp_signal(index, -1, bsp_by_bi_type.get((bi.index, Chan_BSP_TYPE.B3))) #print(klc.end_time, "B3") else: bsp_state_list[index] = 0 elif klc.klc_fx_type == Chan_KLC_FX.BOTTOM2: bi = klc.bi.pre if bi and bi.is_sure and bi.end_klc.bsp_type == Chan_BSP_TYPE.S3: # 第三类卖点 set_bsp_signal(index, 1, bsp_by_bi_type.get((bi.index, Chan_BSP_TYPE.S3))) #print(klc.end_time, "S3") else: bsp_state_list[index] = 0 klc_index += 1 else: bsp_state_list[index] = 0 return { 'bsp_state': bsp_state_list, 'bsp_zg': bsp_zg_list, 'bsp_zd': bsp_zd_list, 'bsp_stop_price': bsp_stop_price_list, 'bsp_risk_ratio': bsp_risk_ratio_list, } def get_bsp_state(self, dataframe): return self.get_bsp_signal_data(dataframe)['bsp_state'] def get_ema_state(self, dataframe): klu_list = self.get_klu_list(dataframe) klc_list = self.get_klc_list(klu_list) bi_list = self.cal_bi_list(klc_list) klu_state_list = [] for klu in klu_list: if klu.near0_return == 1: klu_state_list.append("1") elif klu.near0_return == 9: klu_state_list.append("-1") elif klu.candle_dir == Chan_K_DIR.BULL: klu_state_list.append("2") elif klu.candle_dir == Chan_K_DIR.BEAR: klu_state_list.append("-2") else: klu_state_list.append("0") return klu_state_list def check_fx1(self, klc): if klc.pre and klc.next: if klc.high > klc.pre.high and klc.high > klc.next.high and klc.low > klc.pre.low and klc.low > klc.next.low: if klc.pre.pre and klc.next.next: if klc.high > klc.pre.pre.high and klc.high > klc.next.next.high: #if (klc.close > klc.ema52 or klc.next.close > klc.next.ema52) and klc.macd > 0: klc.set_fx(Chan_FX_TYPE.TOP) #print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "TOP") return Chan_FX_TYPE.TOP elif klc.low < klc.pre.low and klc.low < klc.next.low and klc.high < klc.pre.high and klc.high < klc.next.high: #if (klc.close < klc.ema52 or klc.next.close < klc.next.ema52) and klc.macd < 0: if klc.pre.pre and klc.next.next: if klc.low < klc.pre.pre.low and klc.low < klc.next.next.low: klc.set_fx(Chan_FX_TYPE.BOTTOM) #print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "BOTTOM") return Chan_FX_TYPE.BOTTOM return Chan_FX_TYPE.UNKNOWN def check_fx(self, klc): if klc.pre and klc.next: if klc.high > klc.pre.high and klc.high > klc.next.high and klc.low > klc.pre.low and klc.low > klc.next.low: #if (klc.close > klc.ema52 or klc.next.close > klc.next.ema52) and klc.macd > 0: klc.set_fx(Chan_FX_TYPE.TOP) #print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "TOP") return Chan_FX_TYPE.TOP elif klc.low < klc.pre.low and klc.low < klc.next.low and klc.high < klc.pre.high and klc.high < klc.next.high: #if (klc.close < klc.ema52 or klc.next.close < klc.next.ema52) and klc.macd < 0: klc.set_fx(Chan_FX_TYPE.BOTTOM) #print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "BOTTOM") return Chan_FX_TYPE.BOTTOM return Chan_FX_TYPE.UNKNOWN def check_fx2(self, klc): if klc.pre and klc.next: if klc.high > klc.pre.close and klc.close > klc.next.close and klc.close > klc.pre.close and klc.close > klc.next.close: #if (klc.close > klc.ema52 or klc.next.close > klc.next.ema52) and klc.macd > 0: klc.set_fx(Chan_FX_TYPE.TOP) #print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "TOP") return Chan_FX_TYPE.TOP elif klc.low < klc.pre.close and klc.close < klc.next.close and klc.close < klc.pre.close and klc.close < klc.next.close: #if (klc.close < klc.ema52 or klc.next.close < klc.next.ema52) and klc.macd < 0: klc.set_fx(Chan_FX_TYPE.BOTTOM) #print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "BOTTOM") return Chan_FX_TYPE.BOTTOM return Chan_FX_TYPE.UNKNOWN def check_fx_pattern(self, klc): klu_list = klc.pre.klu_list + klc.klu_list + klc.next.klu_list self.cal_klu_pattern(klu_list) p = "" for klu in klu_list: p += klu.to_string() #print(p) def cal_volume_ratio(self, dataframe, window=10): df = dataframe.copy() # 计算过去N根K线的平均成交量 df['avg_volume'] = df['volume'].rolling(window=window).mean() # 计算量比 df['volume_ratio'] = df['volume'] / df['avg_volume'] # 填充缺失值（前N根K线） df['volume_ratio'] = df['volume_ratio'].fillna(1.0) return df['volume_ratio'] def cal_trend(self, klc_list): """ 基于价格与EMA24/EMA52的位置关系、以及MACD/Signal/Hist的方向， 为每个KLC打上趋势标签：'UP' / 'DOWN' / 'FLAT'。 仅设置 klc.trend，不影响其它字段。 """ if not klc_list: return klc_list last_trend = Chan_PRICE_TREND.UNKNOWN # 趋势延续性：参考近 N 根已完成的KLC lookback_n = 5 prev_klcs = [] for klc in klc_list: price = getattr(klc, 'close', None) ema24 = getattr(klc, 'ema24', None) ema52 = getattr(klc, 'ema52', None) macd_raw = getattr(klc, 'macd', None) signal_raw = getattr(klc, 'signal', None) hist_raw = getattr(klc, 'macdhist', None) macd = macd_raw if macd_raw is not None else 0 signal = signal_raw if signal_raw is not None else 0 hist = hist_raw if hist_raw is not None else 0 rsi = getattr(klc, 'rsi', None) macd_ready = macd_raw is not None and signal_raw is not None hist_ready = hist_raw is not None trend = Chan_PRICE_TREND.UNKNOWN score = 0 try: # 有效性 price_valid = price is not None and price != 0 ema24_valid = ema24 is not None and ema24 != 0 ema52_valid = ema52 is not None and ema52 != 0 # 多因子投票 # 1) 均线结构 + 价位 if ema24_valid or ema52_valid: ma_votes = 0 if ema24_valid and ema52_valid: ma_votes += 1 if ema24 > ema52 else -1 if price_valid and ema24_valid: ma_votes += 1 if price > ema24 else 0 if price_valid and ema52_valid: ma_votes += 1 if price > ema52 else -1 # 限幅，避免相关因子重复计分 score += max(-2, min(2, ma_votes)) # 2) MACD结构 if macd_ready: score += 1 if macd >= signal else -1 if hist_ready and hist != 0: score += 1 if hist > 0 else -1 # 3) 动量与均线差分斜率 pre = getattr(klc, 'pre', None) pre_hist = getattr(pre, 'macdhist', None) if pre else None if pre: pre_close = getattr(pre, 'close', None) if price_valid and pre_close is not None: score += 1 if price >= pre_close else -1 pre_ema24 = getattr(pre, 'ema24', None) pre_ema52 = getattr(pre, 'ema52', None) if ema24_valid and ema52_valid and pre_ema24 not in (None, 0) and pre_ema52 not in (None, 0): spread_now = ema24 - ema52 spread_pre = pre_ema24 - pre_ema52 score += 1 if spread_now >= spread_pre else -1 # 3.1) MACD柱体动量趋势：考虑 macdhist 的斜率与过零 if hist_ready and pre_hist is not None: # 柱体斜率：上升加分，下降减分 if hist > pre_hist: score += 1 elif hist < pre_hist: score -= 1 # 过零加权：负转正更偏多，正转负更偏空 if pre_hist < 0 and hist > 0: score += 1 elif pre_hist > 0 and hist < 0: score -= 1 # 3.2) EMA52 突破/跌破加权 if ema52_valid and price_valid and pre_close is not None and pre_ema52 not in (None, 0): # 看多突破：从均线下方上破且动量配合 if pre_close <= pre_ema52 and price > ema52 and (hist is None or pre_hist is None or hist >= pre_hist): score += 1 # 看空跌破：从均线上方下破且动量配合 if pre_close >= pre_ema52 and price < ema52 and (hist is None or pre_hist is None or hist <= pre_hist): score -= 1 # 3.3) EMA52 支撑/阻力触碰（非强穿越） if ema52_valid and price_valid: low_v = getattr(klc, 'low', None) high_v = getattr(klc, 'high', None) if low_v is not None and high_v is not None and ema52 not in (None, 0): # 触碰容差（相对EMA52的0.15%） touch_tol = 0.0015 # 作为支撑：收盘在上，最低靠近EMA52 near_support_touch = (price > ema52) and (abs(low_v - ema52) / abs(ema52) <= touch_tol) # 作为阻力：收盘在下，最高靠近EMA52 near_resistance_touch = (price < ema52) and (abs(high_v - ema52) / abs(ema52) <= touch_tol) if near_support_touch: # 若动量不弱，则更偏多 score += 1 if (hist is None or pre_hist is None or hist >= pre_hist) else 0 if near_resistance_touch: # 若动量不强，则更偏空 score -= 1 if (hist is None or pre_hist is None or hist <= pre_hist) else 0 # 3.4) 多次对 EMA52 的"拒绝"配合 MACD 逆向：易形成压/支并反向 # 统计近窗口内的上/下拒绝次数： # - 上拒绝：价格位于 EMA52 下方，最高触及/越过 EMA52 但收盘仍在下方 # - 下拒绝：价格位于 EMA52 上方，最低触及/跌破 EMA52 但收盘仍在上方 recent_up_rejects = 0 recent_down_rejects = 0 if ema52_valid: window_rej = prev_klcs[-lookback_n:] if len(prev_klcs) > 0 else [] rej_tol = 0.0015 for wk in window_rej: wk_close = getattr(wk, 'close', None) wk_ema52 = getattr(wk, 'ema52', None) wk_high = getattr(wk, 'high', None) wk_low = getattr(wk, 'low', None) if wk_close is None or wk_ema52 in (None, 0): continue # 上拒绝（阻力）：下方多次试图上破但未站上 if wk_close < wk_ema52 and wk_high is not None: if wk_high >= wk_ema52 or abs(wk_high - wk_ema52) / abs(wk_ema52) <= rej_tol: recent_up_rejects += 1 # 下拒绝（支撑）：上方多次试图下破但未跌破 if wk_close > wk_ema52 and wk_low is not None: if wk_low <= wk_ema52 or abs(wk_low - wk_ema52) / abs(wk_ema52) <= rej_tol: recent_down_rejects += 1 # 定义 MACD 的方向偏好 macd_bias_up = macd_ready and (macd >= signal) and (not hist_ready or pre_hist is None or hist >= pre_hist) macd_bias_down = macd_ready and (macd <= signal) and (not hist_ready or pre_hist is None or hist <= pre_hist) # 若多次上拒绝且 MACD 偏空，则更偏向下行；若多次下拒绝且 MACD 偏多，则更偏向上行 if recent_up_rejects >= 2 and macd_bias_down: score -= 2 if recent_down_rejects >= 2 and macd_bias_up: score += 2 # 4) RSI 辅助 if rsi is not None: if rsi >= 55: score += 1 elif rsi <= 45: score -= 1 # 5) 指标未就绪回退（EMA/MACD缺失时，用动量与RSI辅助，延续趋势） has_full_ind = ema24_valid and ema52_valid and not (macd == 0 and signal == 0 and hist == 0) if not has_full_ind: # 仅根据价动量/RSI做轻量判断，默认延续 last_trend，除非出现强反向 strong_up = False strong_down = False pre = getattr(klc, 'pre', None) if pre: pre_close = getattr(pre, 'close', None) if price_valid and pre_close is not None: strong_up = (price >= pre_close) strong_down = (price < pre_close) if rsi is not None: if rsi >= 60: strong_up = True elif rsi <= 40: strong_down = True if last_trend == Chan_PRICE_TREND.UP and not strong_down: trend = Chan_PRICE_TREND.UP elif last_trend == Chan_PRICE_TREND.DOWN and not strong_up: trend = Chan_PRICE_TREND.DOWN else: trend = Chan_PRICE_TREND.UP if strong_up and not strong_down else (Chan_PRICE_TREND.DOWN if strong_down and not strong_up else Chan_PRICE_TREND.FLAT) else: # 6) 震荡过滤（仅当极近EMA52且MACD贴合时判作震荡） near_flat = False if price_valid and ema52_valid: near_ema52 = abs(price - ema52) / abs(ema52) <= 0.0005 # 0.05% if macd_ready: macd_scale = max(abs(macd), abs(signal), 1e-6) near_macd = abs(macd - signal) / macd_scale <= 0.05 else: near_macd = False near_flat = near_ema52 and near_macd # 7) 动态阈值 + 趋势记忆（更强粘滞：趋势中容忍小幅反分） # 引入过去 N 根KLC 的趋势延续性来动态调整翻转阈值，并结合 EMA52 支撑/阻力触碰强化门槛 force_flip_down = False force_flip_up = False if near_flat: trend = Chan_PRICE_TREND.FLAT else: # 计算过去窗口的趋势一致性 window = prev_klcs[-lookback_n:] if len(prev_klcs) > 0 else [] persist_up = 0 persist_down = 0 for wk in window: if getattr(wk, 'trend', None) == Chan_PRICE_TREND.UP: persist_up += 1 elif getattr(wk, 'trend', None) == Chan_PRICE_TREND.DOWN: persist_down += 1 persist_ratio_up = (persist_up / len(window)) if len(window) > 0 else 0 persist_ratio_down = (persist_down / len(window)) if len(window) > 0 else 0 # 基准阈值 down_flip_threshold = -2 up_flip_threshold = 2 # 若最近多为UP，则从UP翻转需更强反向信号；同理对DOWN if last_trend == Chan_PRICE_TREND.UP and persist_ratio_up >= 0.6: down_flip_threshold = -3 elif last_trend == Chan_PRICE_TREND.DOWN and persist_ratio_down >= 0.6: up_flip_threshold = 3 # EMA52 触碰强化门槛：UP时若出现支撑触碰，下翻更难；DOWN时若出现阻力触碰，上翻更难 if ema52_valid and price_valid: low_v = getattr(klc, 'low', None) high_v = getattr(klc, 'high', None) if low_v is not None and high_v is not None and ema52 not in (None, 0): touch_tol = 0.0015 near_support_touch = (price > ema52) and (abs(low_v - ema52) / abs(ema52) <= touch_tol) near_resistance_touch = (price < ema52) and (abs(high_v - ema52) / abs(ema52) <= touch_tol) if last_trend == Chan_PRICE_TREND.UP and near_support_touch: # 强化维持UP：进一步降低向下翻转阈值 down_flip_threshold = min(down_flip_threshold - 1, -3) if last_trend == Chan_PRICE_TREND.DOWN and near_resistance_touch: # 强化维持DOWN：进一步提高向上翻转阈值 up_flip_threshold = max(up_flip_threshold + 1, 3) # 7.1) 复合拐头信号：MACD/Signal 同向拐头 + hist 连续减弱 + 多次未能越过 EMA52 pre_macd = getattr(pre, 'macd', None) if pre else None pre_signal = getattr(pre, 'signal', None) if pre else None macd_slope = (macd - pre_macd) if (macd_ready and pre_macd is not None) else 0 signal_slope = (signal - pre_signal) if (macd_ready and pre_signal is not None) else 0 # hist 连续减弱（绝对值缩小） hist_seq = [] for wk in prev_klcs[-2:]: val = getattr(wk, 'macdhist', None) if val is not None: hist_seq.append(val) if hist is not None: hist_seq.append(hist) weaken_steps = 0 for i in range(1, len(hist_seq)): if abs(hist_seq[i]) < abs(hist_seq[i-1]): weaken_steps += 1 # 近窗口对 EMA52 的"未能站上/跌破"统计（放宽窗口与条件） window_ema = prev_klcs[-4:] if len(prev_klcs) > 0 else [] no_up_break = False no_down_break = False if ema52_valid: # 未能有效上破：最近若干根收盘大多数不在 EMA52 上方，且高点多次触及/接近 cnt_touch_up = 0 cnt_close_above = 0 for wk in window_ema: wk_close = getattr(wk, 'close', None) wk_high = getattr(wk, 'high', None) wk_ema = getattr(wk, 'ema52', None) if wk_close is not None and wk_ema not in (None, 0): if wk_close > wk_ema: cnt_close_above += 1 if wk_high is not None and (wk_high >= wk_ema or abs(wk_high - wk_ema) / abs(wk_ema) <= 0.0015): cnt_touch_up += 1 no_up_break = (cnt_close_above <= 1 and cnt_touch_up >= 1 and price <= ema52) # 未能有效下破：最近若干根收盘大多数不在 EMA52 下方，且低点多次触及/接近 cnt_touch_down = 0 cnt_close_below = 0 for wk in window_ema: wk_close = getattr(wk, 'close', None) wk_low = getattr(wk, 'low', None) wk_ema = getattr(wk, 'ema52', None) if wk_close is not None and wk_ema not in (None, 0): if wk_close < wk_ema: cnt_close_below += 1 if wk_low is not None and (wk_low <= wk_ema or abs(wk_low - wk_ema) / abs(wk_ema) <= 0.0015): cnt_touch_down += 1 no_down_break = (cnt_close_below <= 1 and cnt_touch_down >= 1 and price >= ema52) # 若当前为UP趋势，出现明显拐头+hist减弱+未能上破EMA52，则加速看空 if last_trend == Chan_PRICE_TREND.UP and macd_slope < 0 and signal_slope < 0 and weaken_steps >= 1 and no_up_break and macd_bias_down: score -= 3 down_flip_threshold = max(down_flip_threshold, 0) force_flip_down = True # 若当前为DOWN趋势，出现明显拐头+hist减弱+未能下破EMA52，则加速看多 if last_trend == Chan_PRICE_TREND.DOWN and macd_slope > 0 and signal_slope > 0 and weaken_steps >= 1 and no_down_break and macd_bias_up: score += 3 up_flip_threshold = min(up_flip_threshold, 0) force_flip_up = True # 多次对 EMA52 的拒绝配合 MACD 逆向：加速反向翻转（降低相反方向阈值） if recent_up_rejects >= 2 and macd_bias_down: # 从 UP 向 DOWN 的翻转更容易 down_flip_threshold = max(down_flip_threshold, -1) if recent_down_rejects >= 2 and macd_bias_up: # 从 DOWN 向 UP 的翻转更容易 up_flip_threshold = min(up_flip_threshold, 1) if force_flip_down: trend = Chan_PRICE_TREND.DOWN elif force_flip_up: trend = Chan_PRICE_TREND.UP elif last_trend == Chan_PRICE_TREND.UP: if score <= down_flip_threshold: trend = Chan_PRICE_TREND.DOWN else: trend = Chan_PRICE_TREND.UP elif last_trend == Chan_PRICE_TREND.DOWN: if score >= up_flip_threshold: trend = Chan_PRICE_TREND.UP else: trend = Chan_PRICE_TREND.DOWN else: # 初始无记忆时，降低进入门槛 if score >= 1: trend = Chan_PRICE_TREND.UP elif score <= -1: trend = Chan_PRICE_TREND.DOWN else: trend = Chan_PRICE_TREND.FLAT except Exception: trend = Chan_PRICE_TREND.UNKNOWN # 写回趋势 if klc.end_time is None: trend = Chan_PRICE_TREND.FLAT if hasattr(klc, 'set_trend'): klc.set_trend(trend) else: setattr(klc, 'trend', trend) last_trend = trend # 更新滑窗：仅向后看 prev_klcs.append(klc) price_diff = klc.close - klc.pre.close if klc.pre else 0 #if klc.index > len(klc_list) - 10: #print(klc.start_time, klc.end_time, klc.close, klc.ema24, klc.ema52, klc.macd, klc.signal, klc.macdhist, klc.trend, price_diff, score) #print(klc.start_time, klc.end_time, klc.trend, price_diff, score) return klc_list def cal_kl_data(self, dataframe:DataFrame): fields = "time,open,high,low,close,volume" klu_list = [] last_klu = None for i in range(0, len(dataframe)): item = dataframe.iloc[i] date = item['date'] o = item['open'] h = item['high'] l = item['low'] c = item['close'] v = item['volume'] # time_obj = date.fromtimestamp(date) # date = date + timedelta(hours=8) time_str = date.strftime('%Y-%m-%d %H:%M:%S') item_data = [ time_str, o, h, l, c, v ] # klu = KLU(self.create_item_dict(item_data, GetColumnNameFromFieldList(fields))) klu = ChanKLU(time_str, o, h, l, c, v) # print(klu.time, klu.open, klu.high, klu.low, klu.close, klu.volume) klu.set_idx(i) klu_list.append(klu) if last_klu: last_klu.set_next(klu) klu.set_pre(last_klu) last_klu = klu if 'macd' in item: klu.set_indicators(item) return klu_list def get_bi_list(self, dataframe): bi_list = self.cal_bi_list(self.get_klc_list(dataframe)) #bi_list = self.cal_bi_list_chanlun(self.get_klc_list(dataframe)) return bi_list def get_kl_data(self, dataframe:DataFrame): return self.cal_kl_data(dataframe) def get_klc_list(self, klu_list): klc_list = [] last_klu = None macd = ChanMACD(klu_list) klu_list = macd.cal_macd_state() ema_up_list = [] ema_down_list = [] ema_up_count = 0 ema_down_count = 0 last_klu = None for klu in klu_list: ema = klu.ema52 last_ema = last_klu.ema52 if last_klu else 0 if klu.close >= ema: ema_up_count += 1 elif klu.close < ema: ema_down_count += 1 if last_klu and last_klu.close >= last_ema and klu.close < ema: ema_up_list.append(ema_up_count) #print(last_klu.time, ema_up_count, "UP END") ema_up_count = 0 elif last_klu and last_klu.close < last_ema and klu.close >= ema: ema_down_list.append(ema_down_count) #print(last_klu.time, ema_down_count, "DOWN END") ema_down_count = 0 if len(klc_list) > 0: last_klc = klc_list[-1] if klu.exception: ddir = Chan_KLINE_DIR.DOWN if last_klc.high < klu.high: ddir = Chan_KLINE_DIR.UP klc = ChanKLC(klu, index=len(klc_list), ddir=ddir) klc.high = klu.close if klu.close > klu.open else klu.open klc.low = klu.open if klu.close > klu.open else klu.close klc_list.append(klc) last_klc.set_next(klc) klc.set_pre(last_klc) last_klc.set_end_klu(last_klu) klc.set_pre_fx() #print(klu.time, klu.high, klu.low, klu.close, klu.open, klu.exception) else: included = last_klc.check_klu_included(klu) if not included: ddir = Chan_KLINE_DIR.DOWN if last_klc.high < klu.high: ddir = Chan_KLINE_DIR.UP klc = ChanKLC(klu, index=len(klc_list), ddir=ddir) klc_list.append(klc) last_klc.set_next(klc) klc.set_pre(last_klc) last_klc.set_end_klu(last_klu) klc.set_pre_fx() else: last_klc.add_klu(klu) else: ddir = Chan_KLINE_DIR.UP if klu.open > klu.close: ddir = Chan_KLINE_DIR.DOWN klc = ChanKLC(klu, 0, ddir) klc_list.append(klc) last_klu = klu klc_list = self.cal_trend(klc_list) #print(ema52_up_list, ema52_down_list) return klc_list def get_seg_list(self, bi_list): seg_list = [] up_bi_list = [] down_bi_list = [] last_up_bi = None last_down_bi = None last_up_sbi = None last_down_sbi = None last_seg = None up_sbi_list = [] down_sbi_list = [] look_for_bottom = False look_for_top = False for bi in bi_list: #print(len(up_sbi_list), len(down_sbi_list)) if len(seg_list) > 0: # Last seg is up if last_seg.dir == Chan_SEG_DIR.UP: if bi.dir == Chan_BI_DIR.DOWN: if len(down_sbi_list) > 1: # Check down sbi inclusion included = last_down_sbi.check_bi_included(bi) if not included: down_sbi = ChanSBI(bi, len(down_sbi_list), bi.dir) last_down_sbi.set_next(down_sbi) last_down_sbi.set_end_bi(last_down_bi) down_sbi.set_pre(last_down_sbi) down_sbi_list.append(down_sbi) fx = last_down_sbi.check_fx() # Found top if fx == Chan_FX_TYPE.TOP: if look_for_top: seg_list[-2].set_sure(bi) look_for_top = False #print(bi.start_time, look_for_top, "UP 1") # Has gap and search for bottom fx if last_down_sbi.has_fx_gap: look_for_bottom = True last_seg.pre_set_end_bi(bi_list[last_down_sbi.start_bi.index - 1]) seg = ChanSEG(last_down_sbi.start_bi, len(seg_list), Chan_SEG_DIR.DOWN, bi) seg_list.append(seg) last_seg.set_next(seg) seg.set_pre(last_seg) last_seg = seg up_sbi_list = [] last_up_sbi = ChanSBI(last_up_bi, len(up_sbi_list), last_up_bi.dir) up_sbi_list.append(last_up_sbi) #up_sbi_list.append(last_up_sbi) #print(last_up_bi.start_time, last_up_sbi.start_bi.start_time, "Reset up sbi list 1") #print(bi.start_time, look_for_top, "UP 2") # No gap end SEG else: if look_for_bottom: look_for_bottom = False last_seg.set_start_bi(last_down_sbi.start_bi) seg_list[-2].set_end_bi(bi_list[last_down_sbi.start_bi.index - 1], bi) up_sbi_list = [] last_up_sbi = ChanSBI(last_up_bi, len(up_sbi_list), last_up_bi.dir) up_sbi_list.append(last_up_sbi) last_seg.add_bi(bi) #up_sbi_list.append(last_up_sbi) #print(last_up_bi.start_time, last_up_sbi.start_bi.start_time, "Reset up sbi list 2") #print(bi.start_time, look_for_top, "UP 3") else: last_seg.set_end_bi(bi_list[last_down_sbi.start_bi.index - 1], bi) seg = ChanSEG(last_down_sbi.start_bi, len(seg_list), Chan_SEG_DIR.DOWN, bi) seg_list.append(seg) last_seg.set_next(seg) seg.set_pre(last_seg) last_seg = seg #print(last_down_sbi.end_bi.start_time, "Normal UP SEG", last_up_sbi.start_bi.start_time, bi.start_time) #l_up_sbi = up_sbi_list[-1] up_sbi_list = [] last_up_sbi = ChanSBI(last_up_bi, len(up_sbi_list), last_up_bi.dir) up_sbi_list.append(last_up_sbi) #up_sbi_list.append(last_up_sbi) #print(last_up_bi.start_time, last_up_sbi.start_bi.start_time, "Reset up sbi list 3") last_down_sbi = down_sbi last_seg.add_bi(bi) else: if len(down_sbi_list) == 1: included = last_down_sbi.check_bi_included(bi) if not included: down_sbi = ChanSBI(bi, len(down_sbi_list), bi.dir) last_down_sbi.set_next(down_sbi) last_down_sbi.set_end_bi(last_down_bi) down_sbi.set_pre(last_down_sbi) down_sbi_list.append(down_sbi) last_down_sbi = down_sbi #print(bi.start_time, look_for_top, "UP 4") last_seg.add_bi(bi) else: last_down_sbi = ChanSBI(bi, len(down_sbi_list), bi.dir) down_sbi_list.append(last_down_sbi) last_seg.add_bi(bi) #print(bi.start_time, look_for_top, "UP 5") else: if last_up_sbi: included = last_up_sbi.check_bi_included(bi) if not included: up_sbi = ChanSBI(bi, len(up_sbi_list), bi.dir) last_up_sbi.set_next(up_sbi) last_up_sbi.set_end_bi(last_up_bi) up_sbi.set_pre(last_up_sbi) up_sbi_list.append(up_sbi) last_up_sbi = up_sbi #print(bi.start_time, look_for_top, "UP 6") last_seg.add_bi(bi) # Last seg is down else: if bi.dir == Chan_BI_DIR.UP: if len(up_sbi_list) > 1: # Check down sbi inclusion included = last_up_sbi.check_bi_included(bi) if not included: up_sbi = ChanSBI(bi, len(up_sbi_list), bi.dir) last_up_sbi.set_next(up_sbi) last_up_sbi.set_end_bi(last_up_bi) up_sbi.set_pre(last_up_sbi) up_sbi_list.append(up_sbi) fx = last_up_sbi.check_fx() # Found bottom if fx == Chan_FX_TYPE.BOTTOM: if look_for_bottom: seg_list[-2].set_sure(bi) look_for_bottom = False #print(bi.start_time, look_for_top, "DOWN 1") # Has gap and search for bottom fx if last_up_sbi.has_fx_gap: look_for_top = True last_seg.pre_set_end_bi(bi_list[last_up_sbi.start_bi.index - 1]) seg = ChanSEG(last_up_sbi.start_bi, len(seg_list), Chan_SEG_DIR.UP, bi) seg_list.append(seg) last_seg.set_next(seg) seg.set_pre(last_seg) last_seg = seg down_sbi_list = [] last_down_sbi = ChanSBI(last_down_bi, len(down_sbi_list), last_down_bi.dir) down_sbi_list.append(last_down_sbi) #down_sbi_list.append(last_down_sbi) #print(last_down_bi.start_time, last_down_sbi.start_bi.start_time, "Reset down sbi list 1") #print(bi.start_time, look_for_top, "DOWN 2") # No gap end SEG else: if look_for_top: look_for_top = False last_seg.set_start_bi(last_up_sbi.start_bi) seg_list[-2].set_end_bi(bi_list[last_up_sbi.start_bi.index - 1], bi) down_sbi_list = [] last_down_sbi = ChanSBI(last_down_bi, len(down_sbi_list), last_down_bi.dir) down_sbi_list.append(last_down_sbi) last_seg.add_bi(bi) #down_sbi_list.append(last_down_sbi) #print(last_down_bi.start_time, last_down_sbi.start_bi.start_time, "Reset down sbi list 2") #print(bi.start_time, look_for_top, "DOWN 3") else: last_seg.set_end_bi(bi_list[last_up_sbi.start_bi.index - 1], bi) seg = ChanSEG(last_up_sbi.start_bi, len(seg_list), Chan_SEG_DIR.UP, bi) #print(last_up_sbi.start_bi.start_time) last_seg.set_next(seg) seg.set_pre(last_seg) seg_list.append(seg) last_seg = seg #print(last_up_sbi.end_bi.start_time, "Normal DOWN SEG", last_down_sbi.start_bi.start_time, bi.start_time) down_sbi_list = [] last_down_sbi = ChanSBI(last_down_bi, len(down_sbi_list), last_down_bi.dir) down_sbi_list.append(last_down_sbi) #down_sbi_list.append(last_down_sbi) #print(last_down_bi.start_time, last_down_sbi.start_bi.start_time, "Reset down sbi list 3") last_up_sbi = up_sbi last_seg.add_bi(bi) else: if len(up_sbi_list) == 1: #last_up_sbi = up_sbi_list[-1] included = last_up_sbi.check_bi_included(bi) if not included: up_sbi = ChanSBI(bi, len(up_sbi_list), bi.dir) last_up_sbi.set_next(up_sbi) last_up_sbi.set_end_bi(last_up_bi) up_sbi.set_pre(last_up_sbi) up_sbi_list.append(up_sbi) last_up_sbi = up_sbi last_seg.add_bi(bi) #print(bi.start_time, look_for_top, "DOWN 4") else: last_up_sbi = ChanSBI(bi, len(up_sbi_list), bi.dir) up_sbi_list.append(last_up_sbi) last_seg.add_bi(bi) #print(bi.start_time, look_for_top, "DOWN 5") else: if last_down_sbi: included = last_down_sbi.check_bi_included(bi) if not included: down_sbi = ChanSBI(bi, len(down_sbi_list), bi.dir) last_down_sbi.set_next(down_sbi) last_down_sbi.set_end_bi(last_down_bi) down_sbi.set_pre(last_down_sbi) down_sbi_list.append(down_sbi) last_down_sbi = down_sbi last_seg.add_bi(bi) #print(bi.start_time, look_for_top, look_for_bottom, "DOWN 6") # len(seg_list) = 0 else: if bi.check_overlap(): if bi.dir == Chan_BI_DIR.UP: seg = ChanSEG(bi, len(seg_list), Chan_SEG_DIR.UP, bi) last_up_bi = bi last_up_sbi = ChanSBI(bi, len(up_sbi_list), bi.dir) seg_list.append(seg) last_seg = seg #print(bi.start_time, 'Create first UP SEG') else: seg = ChanSEG(bi, len(seg_list), Chan_SEG_DIR.DOWN, bi) last_down_bi = bi last_down_sbi = ChanSBI(bi, len(down_sbi_list), bi.dir) seg_list.append(seg) last_seg = seg #print(bi.start_time, 'Create first DOWN SEG') if bi.dir == Chan_BI_DIR.UP: last_up_bi = bi up_bi_list.append(bi) else: last_down_bi = bi down_bi_list.append(bi) """ if len(seg_list) > 1: seg = seg_list[-1] last_seg = seg_list[-2] last_seg_bi = last_seg.bi_list[-3] bi_index = seg.start_bi.index for i in range(bi_index, len(bi_list) - 1): # last seg is down if seg.dir == Chan_SEG_DIR.UP: if bi_list[i].dir == Chan_BI_DIR.UP: last_seg_peak = last_seg_bi.high if bi_list[i].high > last_seg_peak: # The confirmed print("Last UP seg is broken, create a new seg. 1") seg.pre_set_end_bi(bi_list[i]) seg = ChanSEG(bi_list[i+1], len(seg_list), Chan_SEG_DIR.DOWN, bi) seg_list.append(seg) last_seg = seg_list[-2] if len(last_seg.bi_list) > 3: last_seg_bi = last_seg.bi_list[-3] else: if bi_list[i].dir == Chan_BI_DIR.DOWN: last_seg_peak = last_seg_bi.low if bi_list[i].low < last_seg_peak: print("Last DOWN seg is broken, create a new seg. 1") seg.pre_set_end_bi(bi_list[i]) seg = ChanSEG(bi_list[i+1], len(seg_list), Chan_SEG_DIR.UP, bi) seg_list.append(seg) last_seg = seg_list[-2] if len(last_seg.bi_list) > 3: last_seg_bi = last_seg.bi_list[-3] else: if len(seg_list) == 1: last_seg = seg_list[-1] bi_index = last_seg.bi_list[0].index for i in range(bi_index, len(bi_list) - 1): if i > bi_index + 2: last_seg_peak = bi_list[i-2].high # last seg is down if last_seg.dir == Chan_SEG_DIR.DOWN: if bi_list[i].dir == Chan_BI_DIR.UP: if bi_list[i].high > last_seg_peak: print("Last seg is broken, create a new seg. 2") last_seg.pre_set_end_bi(bi_list[i-1]) seg = ChanSEG(bi_list[i], len(seg_list), Chan_SEG_DIR.UP, bi) seg_list.append(seg) last_seg = seg last_seg_bi = bi_list[i] break """ #self.cal_bi_zs(seg_list) return seg_list def get_zs_state(self, df): bi_list = self.cal_bi_list(self.get_klc_list(self.get_kl_data(df))) seg_list = self.get_seg_list(bi_list) zs_list = self.calculate_zs(seg_list) for zs in zs_list: last_zs = zs return zs_list def cal_bi_list(self, klc_list): bi_list = [] last_top = None last_bottom = None bi_klc_min = 4 last_fx_klc = None for klc in klc_list: if last_fx_klc: klc.check_klc_state(last_fx_klc) klc.check_fx_confirmed(last_top, last_bottom) fx = self.check_fx(klc) if fx == Chan_FX_TYPE.TOP: if last_bottom: if self.check_top_fx(last_bottom, klc) == False: fx = Chan_FX_TYPE.UNKNOWN if fx == Chan_FX_TYPE.BOTTOM: if last_top: if self.check_bottom_fx(last_top, klc) == False: #print(klc.end_time, last_top.end_time, "---") fx = Chan_FX_TYPE.UNKNOWN # Do nothing if fx == Chan_FX_TYPE.UNKNOWN: if len(bi_list) > 0: bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #continue if len(bi_list) > 0 and klc.end_klu: last_bi = bi_list[-1] #print(klc.start_time, last_bi.start_time, last_bi.end_time, last_bi.dir, last_bi.high, last_bi.low, last_bottom.end_time, "last bi") if last_top and last_bi.dir == Chan_BI_DIR.DOWN: if last_bottom and klc.high > last_bi.high: #print(klc.end_time, "Top 7, 1", last_bi.start_time, klc.high, last_bi.high) #klc.klc_fx_type = Chan_KLC_FX.TOP7 #klc.fx = Chan_FX_TYPE.TOP """ last_bi.set_end_klc(last_bottom, klc) bi = ChanBI(last_bottom, len(bi_list), Chan_BI_DIR.UP) #klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM7) #klc.bb_out = True last_bi.set_next(bi) bi.set_pre(last_bi) for klc_index in range(last_bi.end_klc.index, len(klc_list)): bi.add_klc(klc_list[klc_index]) bi_list.append(bi) last_top = klc klc.set_bi(bi) #print(klc.start_time, bi.start_time, bi.end_time, bi.dir, bi.high, bi.low, bi.is_sure) """ else: if last_bottom and last_bi.dir == Chan_BI_DIR.UP: if last_top and klc.low < last_bi.low: #print(klc.end_time, "Bottom 8, 2", last_bi.start_time) #klc.klc_fx_type = Chan_KLC_FX.BOTTOM8 #klc.fx = Chan_FX_TYPE.BOTTOM """ last_bi.set_end_klc(last_top, klc) bi = ChanBI(last_top, len(bi_list), Chan_BI_DIR.DOWN) #klc.set_klc_fx_type(Chan_KLC_FX.TOP6) #klc.bb_out = True last_bi.set_next(bi) bi.set_pre(last_bi) for klc_index in range(last_bi.end_klc.index, len(klc_list)): bi.add_klc(klc_list[klc_index]) bi_list.append(bi) last_bottom = klc klc.set_bi(bi) #print(klc.start_time, bi.start_time, bi.end_time, bi.dir, bi.high, bi.low, bi.is_sure) """ else: last_fx_klc = klc if fx == Chan_FX_TYPE.TOP: #print(klc.end_time, fx, klc.pre.high, klc.high, klc.pre.start_time, klc.pre.end_time) if last_top: if last_bottom: #print(klc.start_time, last_bottom.start_time, last_top.start_time) if last_bottom.index < last_top.index: # Second top lower to be second sell point if last_top.high > klc.high: bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #klc.set_klc_fx_type(Chan_KLC_FX.TOP3) #print(klc.end_time, klc.fx, "二类卖点Sell 1") else: # A new top found #last_top.set_fx(Chan_FX_TYPE.UNKNOWN) last_top = klc #print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Top Change 1") klc.set_klc_fx_type(Chan_KLC_FX.TOP1) self.check_fx_pattern(klc) #print(klc.end_time, klc.fx, "一类卖点Sell 1") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) # 不满足结合律的分型 else: #klc.set_klc_fx_type(Chan_KLC_FX.TOP0) #print(klc.end_time, klc.klc_fx_type) if last_bottom.index + bi_klc_min > klc.index: if last_top.high > klc.high: #print(klc.start_time, klc.fx, "二类卖点Sell 1") #klc.set_klc_fx_type(Chan_KLC_FX.TOP8) bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) # New TOP Found前面的UKNOWN可能出现TOP7，但是这里的也可能出现TOP8分型 else: # 顶分型在出现2之前超过前一个笔的顶 TOP8 if last_top.index + bi_klc_min < klc.index and len(bi_list) > 1: pre_last_bi = bi_list[-2] last_bi = bi_list[-1] if pre_last_bi.is_sure and not last_bi.is_sure and pre_last_bi.dir == Chan_BI_DIR.UP and False: pre_last_bi.update_bi(klc) bi_list.remove(last_bi) pre_last_bi.set_next(None) #last_top.set_fx(Chan_FX_TYPE.PTOP) last_top = klc last_bottom = pre_last_bi.start_klc #print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Top Bottom Change 1") klc.set_klc_fx_type(Chan_KLC_FX.TOP2) #print(klc.start_time, last_bi.start_klc.start_time, "New TOP Found reset last bi") #klc.set_state("10") #print(klc.start_time, klc.fx, "笔卖点Sell 1") ###klc.set_klc_fx_type(Chan_KLC_FX.TOP2) # when bi is down but the fx is top bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #klc.set_klc_fx_type(Chan_KLC_FX.TOP8) #print(klc.start_time, last_bi.start_klc.start_time, "New TOP Found reset last bi") else: #klc.set_fx(Chan_FX_TYPE.PTOP) bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) print(klc.end_time, klc.fx, "无效顶分型") # 满足结合律 else: # New Temp TOP and last bottom confirmed ***** confirm last down bi(last bottom and last top) last_bi = bi_list[-1] if not last_bi.is_sure: last_bi.set_end_klc(last_bottom, klc) bi = ChanBI(last_bottom, len(bi_list), Chan_BI_DIR.UP) last_bi.set_next(bi) bi.set_pre(last_bi) bi.add_klc(klc) bi_list.append(bi) last_top = klc #print(klc.end_time, klc.fx, bi_list[-1].dir, "Last Top Change 2") klc.set_klc_fx_type(Chan_KLC_FX.TOP2) self.check_fx_pattern(klc) #bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #print(klc.start_time, last_bottom.start_time, "Normal TOP Found, Confirm down bi 4") # last bottom = None 初始化的时候用，其他时间不用 else: # 初始化的时候用，其他时间不用 if last_top.high < klc.high: last_bi = bi_list[-1] last_bi.set_start_klc(klc, Chan_BI_DIR.DOWN) last_top = klc #print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Top Change 3") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) # 初始化的时候用，其他时间不用 else: #klc.set_fx(Chan_FX_TYPE.TT) #print(klc.start_time, klc.fx, "二类卖点Sell 2") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) # last_top == None 初始化的时候用，其他时间不用 else: if last_bottom: # 不满足结合律的分型 if last_bottom.index + bi_klc_min > klc.index: #klc.set_fx(Chan_FX_TYPE.PTOP) bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #print(klc.start_time, klc.fx, "中枢卖点Sell 1") else: # First temp top and last bottom confirmed last_top = klc #print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Top Change 4") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) # Last top = None, last bottom = None, create first down bi 初始化的时候用，其他时间不用 else: # First temp top last_top = klc bi = ChanBI(klc, len(bi_list), Chan_BI_DIR.DOWN) bi_list.append(bi) bi.add_klc(klc) klc.set_bi(bi_list[-1]) #print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Top Change 5") #klc.fx = Bottom ======================== else: if last_bottom: if last_top: # Bottom after top and find a new bottom if last_top.index < last_bottom.index: # Second bottom uppper to be second buy point and confirm last bi if last_bottom.low < klc.low: bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM3) #print(last_bottom.start_time, last_bottom.end_time, "--------------------------------1") #print(klc.end_time, klc.fx, "二类买点Buy 1") else: # A new bottom found last_bottom = klc #print(klc.end_time, klc.fx, bi_list[-1].dir, "Last Bottom Change 1") klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM1) self.check_fx_pattern(klc) #print(klc.end_time, klc.fx, "一类买点Buy 1") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) # 不满足结合律的分型 else: #klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM0) #print(klc.end_time, klc.klc_fx_type) if last_top.index + bi_klc_min > klc.index: if last_bottom.low < klc.low: #print(klc.end_time, klc.fx, "中枢买点Buy 1") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM8) # Found new bottom没有意义，上面UNKNOWN的时候已经是笔破坏了 else: #print(klc.end_time, last_bottom.end_time, "Found a new bottom") if last_bottom.index + bi_klc_min < klc.index and len(bi_list) > 1: pre_last_bi = bi_list[-2] last_bi = bi_list[-1] if pre_last_bi.is_sure and not last_bi.is_sure and pre_last_bi.dir == Chan_BI_DIR.DOWN and False: pre_last_bi.update_bi(klc) bi_list.remove(last_bi) pre_last_bi.set_next(None) last_bottom = klc last_top = pre_last_bi.start_klc #print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Top Bottom Change 2") klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM2) #print(klc.start_time, last_bi.start_klc.start_time, "New BOTTOM Found reset last bi") #print(klc.start_time, klc.fx, "笔买点Buy 1") ###klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM2) # when bi is up but the fx is bottom bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM8) else: #klc.set_fx(Chan_FX_TYPE.UNKNOWN) bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) print(klc.end_time, klc.fx, "无效底分型") # 满足结合律的分型 else: # New Temp Bottom and last top confirmed ***** confirm last up bi(last bottom and last top) last_bi = bi_list[-1] if not last_bi.is_sure: last_bi.set_end_klc(last_top, klc) bi = ChanBI(last_top, len(bi_list), Chan_BI_DIR.DOWN) last_bi.set_next(bi) bi.set_pre(last_bi) bi.add_klc(klc) bi_list.append(bi) last_bottom = klc #print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Bottom Change 2") klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM2) self.check_fx_pattern(klc) #bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #print(klc.start_time, last_top.start_time, "Normal Bottom Found, Confirm up bi 6") # last_top = None 初始化的时候用，其他时间不用 else: if last_bottom.low > klc.low: last_bi = bi_list[-1] last_bi.set_start_klc(klc, Chan_BI_DIR.UP) #last_bottom.set_fx(Chan_FX_TYPE.UNKNOWN) last_bottom = klc #print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Bottom Change 3") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #print(klc.start_time, klc.fx, "笔买点Buy 3") else: #klc.set_fx(Chan_FX_TYPE.BB) #klc.set_state('-20') #print(klc.start_time, klc.fx, "二类买点Buy 2") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) # last_bottom = None 初始化的时候用，其他时间不用 else: if last_top: # 不满足结合律的分型 if last_top.index + bi_klc_min > klc.index: #klc.set_fx(Chan_FX_TYPE.PBOTTOM) bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #print(klc.start_time, klc.fx, "中枢买点Buy 1") else: # First temp bottom and last top confirmed last_bottom = klc #print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Bottom Change 4") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #print(klc.start_time, klc.fx, "一类买点Buy 1") # Last top = None, last bottom = None, create first up bi else: # First temp bottom and no top yet last_bottom = klc bi = ChanBI(klc, len(bi_list), Chan_BI_DIR.UP) #klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM7) bi_list.append(bi) bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #print(klc.start_time, klc.fx, bi_list[-1].dir, "Last Bottom Change 5") #print(klc.start_time, klc.fx, "笔买点Buy 4") self.get_above_zero_bsp(klc_list) #print(bi_list[-1].start_time, bi_list[-1].end_time, len(bi_list[-1].klc_list)) return bi_list def get_above_zero_bsp(self, klc_list): buy_bsp_list = [] sell_bsp_list = [] above_zero = False buy_bsp = None sell_bsp = None for klc in klc_list: if klc.pre and klc.pre.signal < 0 and klc.signal > 0: above_zero = True if klc.pre and klc.pre.signal > 0 and klc.signal < 0: above_zero = False if above_zero and klc.klc_fx_type == Chan_KLC_FX.BOTTOM2 and klc.macd > 0: buy_bsp = klc buy_bsp_list.append(klc) #print(klc.end_time, "MACD 0轴上穿，回调笔底分型做多") if buy_bsp and klc.pre and klc.pre.macdhist > 0 and klc.macdhist < 0: sell_bsp = klc sell_bsp_list.append(klc) buy_bsp = None #print(klc.end_time, "Sell BSP Found") return buy_bsp_list def check_top_fx(self, last_bottom, klc): if (last_bottom.high > klc.pre.low or last_bottom.high > klc.next.low) and (klc.index - last_bottom.index < 100): return False return True def check_bottom_fx(self, last_top, klc): if (last_top.low < klc.pre.high or last_top.low < klc.next.high) and (klc.index - last_top.index < 100): return False return True # 线段内的中枢 def cal_bi_zs(self, seg_list): bi_zs_list = [] for seg in seg_list: zs_list = seg.cal_bi_zs() if len(zs_list) > 0: bi_zs_list = list(bi_zs_list) + list(zs_list) return bi_zs_list # 跨段不相连的中枢 def cal_bi_zs_list(self, bi_list): """ 根据缠论笔中枢定义计算中枢（参照 get_zs_list 线段中枢判断规则） 从第4根笔开始（索引3），每3根笔为一组检查 上涨中枢：后中枢 zd > 前中枢 zg（不重叠上移） 下跌中枢：后中枢 zg < 前中枢 zd（不重叠下移） 中枢可按两笔一组继续扩展到5根、7根... """ bi_zs_list = [] if len(bi_list) < 3: return bi_zs_list last_zs = None start_idx = 3 while start_idx < len(bi_list): if start_idx + 2 >= len(bi_list): break bi1 = bi_list[start_idx] bi2 = bi_list[start_idx + 1] bi3 = bi_list[start_idx + 2] if not (bi1.is_sure and bi2.is_sure and bi3.is_sure): start_idx += 1 continue zg = min(bi1.high, bi2.high, bi3.high) zd = max(bi1.low, bi2.low, bi3.low) if zg <= zd: start_idx += 1 continue valid = False if last_zs is None: if bi1.dir == Chan_BI_DIR.DOWN: zs_dir = Chan_ZS_DIR.UP valid = (bi2.dir == Chan_BI_DIR.UP and bi3.dir == Chan_BI_DIR.DOWN) else: zs_dir = Chan_ZS_DIR.DOWN valid = (bi2.dir == Chan_BI_DIR.DOWN and bi3.dir == Chan_BI_DIR.UP) else: is_up_zs = zg > last_zs.zg is_down_zs = zd < last_zs.zd if is_up_zs: zs_dir = Chan_ZS_DIR.UP valid = (bi1.dir == Chan_BI_DIR.DOWN and bi2.dir == Chan_BI_DIR.UP and bi3.dir == Chan_BI_DIR.DOWN) elif is_down_zs: zs_dir = Chan_ZS_DIR.DOWN valid = (bi1.dir == Chan_BI_DIR.UP and bi2.dir == Chan_BI_DIR.DOWN and bi3.dir == Chan_BI_DIR.UP) if not valid: start_idx += 1 continue gg = max(bi1.high, bi2.high, bi3.high) dd = min(bi1.low, bi2.low, bi3.low) zs = ChanBIZS(bi1, len(bi_zs_list), zs_dir) zs.set_zg(zg) zs.set_zd(zd) zs.set_gg(gg) zs.set_dd(dd) zs.is_sure = False zs.bi_list = [bi1, bi2, bi3] added_after_leave = [] leave_index = start_idx + 4 while leave_index < len(bi_list): b = bi_list[leave_index] if not b.is_sure: break if b.high >= zs.zd and b.low <= zs.zg: added_after_leave.append(b.pre) added_after_leave.append(b) else: break leave_index += 2 if added_after_leave: bis_for_zs = list(zs.bi_list) + list(added_after_leave) bi_highs = [bi.high for bi in bis_for_zs] bi_lows = [bi.low for bi in bis_for_zs] zs.set_gg(max(bi_highs)) zs.set_dd(min(bi_lows)) zs.bi_list = bis_for_zs bi = bis_for_zs[-1] if bi.is_sure: zs.set_end_bi(bi, bi.sure_time) start_idx = start_idx + len(added_after_leave) else: zs.set_end_bi(bi3, bi3.sure_time) if last_zs: last_zs.set_next(zs) zs.set_pre(last_zs) bi_zs_list.append(zs) last_zs = zs start_idx += 4 if last_zs: last_zs.is_sure = bi_list[-1].is_sure if last_zs and not last_zs.is_sure: if last_zs.bi_list and len(last_zs.bi_list) > 0: last_bi_of_zs = last_zs.bi_list[-1] last_bi_idx = -1 for i, bi in enumerate(bi_list): if bi == last_bi_of_zs: last_bi_idx = i break has_leave = False if last_bi_idx >= 0 and last_bi_idx + 1 < len(bi_list): for i in range(last_bi_idx + 1, len(bi_list)): bi = bi_list[i] if bi.is_sure: leave = (bi.low > last_zs.zg and bi.high > last_zs.zg) or \ (bi.high < last_zs.zd and bi.low < last_zs.zd) if leave: has_leave = True break if has_leave: if last_bi_of_zs.is_sure: last_zs.set_end_bi(last_bi_of_zs, last_bi_of_zs.sure_time) return bi_zs_list def get_bi_zs_list(self, bi_list): """ 根据缠论笔中枢定义计算中枢（完全参照 get_seg_zs_list 线段中枢判断规则） 从第4根笔开始（索引3），每3根笔为一组检查 上涨中枢：后中枢 zd > 前中枢 zg（不重叠上移） 下跌中枢：后中枢 zg < 前中枢 zd（不重叠下移） 盘整/扩张：后中枢与前中枢整体区间有交集 → 合并扩展 中枢可按两笔一组继续扩展到5根、7根... """ bi_zs_list = [] if len(bi_list) < 3: return bi_zs_list last_zs = None start_idx = 3 while start_idx < len(bi_list): if start_idx + 2 >= len(bi_list): break bi1 = bi_list[start_idx] bi2 = bi_list[start_idx + 1] bi3 = bi_list[start_idx + 2] if not (bi1.is_sure and bi2.is_sure and bi3.is_sure): start_idx += 1 continue zg = min(bi1.high, bi2.high, bi3.high) zd = max(bi1.low, bi2.low, bi3.low) if zg <= zd: start_idx += 1 continue valid = False if last_zs is None: if bi1.dir == Chan_BI_DIR.DOWN: zs_dir = Chan_ZS_DIR.UP valid = (bi2.dir == Chan_BI_DIR.UP and bi3.dir == Chan_BI_DIR.DOWN) else: zs_dir = Chan_ZS_DIR.DOWN valid = (bi2.dir == Chan_BI_DIR.DOWN and bi3.dir == Chan_BI_DIR.UP) else: is_up_zs = zd > last_zs.zg is_down_zs = zg < last_zs.zd if is_up_zs: zs_dir = Chan_ZS_DIR.UP valid = (bi1.dir == Chan_BI_DIR.DOWN and bi2.dir == Chan_BI_DIR.UP and bi3.dir == Chan_BI_DIR.DOWN) elif is_down_zs: zs_dir = Chan_ZS_DIR.DOWN valid = (bi1.dir == Chan_BI_DIR.UP and bi2.dir == Chan_BI_DIR.DOWN and bi3.dir == Chan_BI_DIR.UP) create_new_zs = False if not valid: # 如果新中枢和前一个中枢的中枢区间有重叠，不形成新中枢，合并扩展 if last_zs is not None: is_in_last_zs = (zd > last_zs.zd and zd < last_zs.zg) or \ (zg < last_zs.zg and zg > last_zs.zd) or \ (zg > last_zs.zg and zd < last_zs.zd) or \ (zg < last_zs.zg and zd > last_zs.zd) if is_in_last_zs: # 扩展当前中枢：将 bi1-bi3 加入 last_zs for bi in [bi1, bi2, bi3]: if bi not in last_zs.bi_list: last_zs.add_bi(bi) create_new_zs = False else: start_idx += 1 continue else: start_idx += 1 continue else: create_new_zs = True # 新中枢形成时确认前一个中枢 if last_zs and create_new_zs: last_bi = last_zs.bi_list[-1] if last_bi and last_bi.is_sure: last_zs.is_sure = True last_zs.set_end_bi(last_bi, last_bi.sure_time) zs = last_zs if create_new_zs: gg = max(bi1.high, bi2.high, bi3.high) dd = min(bi1.low, bi2.low, bi3.low) zs = ChanBIZS(bi1, len(bi_zs_list), zs_dir) zs.set_zg(zg) zs.set_zd(zd) zs.set_gg(gg) zs.set_dd(dd) zs.is_sure = False zs.bi_list = [bi1, bi2, bi3] # 离开后回抽扩展检查 added_after_leave = [] leave_index = start_idx + 4 while leave_index < len(bi_list): b = bi_list[leave_index] if not b.is_sure: break if b.high >= zs.zd and b.low <= zs.zg: added_after_leave.append(b.pre) added_after_leave.append(b) else: break leave_index += 2 if added_after_leave: bis_for_zs = list(zs.bi_list) + list(added_after_leave) bi_highs = [bi.high for bi in bis_for_zs] bi_lows = [bi.low for bi in bis_for_zs] zs.set_gg(max(bi_highs)) zs.set_dd(min(bi_lows)) zs.bi_list = bis_for_zs bi = bis_for_zs[-1] if bi.is_sure: zs.set_end_bi(bi, bi.sure_time) start_idx = start_idx + len(added_after_leave) else: if create_new_zs: zs.set_end_bi(bi3, bi3.sure_time) if create_new_zs: if last_zs: last_zs.set_next(zs) zs.set_pre(last_zs) bi_zs_list.append(zs) last_zs = zs start_idx += 4 # 最后一个中枢：根据 bi_list 最后一笔确认状态 if last_zs: last_zs.is_sure = bi_list[-1].is_sure if last_zs and not last_zs.is_sure: if last_zs.bi_list and len(last_zs.bi_list) > 0: last_bi_of_zs = last_zs.bi_list[-1] last_bi_idx = -1 for i, bi in enumerate(bi_list): if bi == last_bi_of_zs: last_bi_idx = i break has_leave = False if last_bi_idx >= 0 and last_bi_idx + 1 < len(bi_list): for i in range(last_bi_idx + 1, len(bi_list)): bi = bi_list[i] if bi.is_sure: leave = (bi.low > last_zs.zg and bi.high > last_zs.zg) or \ (bi.high < last_zs.zd and bi.low < last_zs.zd) if leave: has_leave = True break if has_leave: if last_bi_of_zs.is_sure: last_zs.set_end_bi(last_bi_of_zs, last_bi_of_zs.sure_time) return bi_zs_list def cal_bi_zs_list_pure(self, bi_list): bi_zs_list = [] if len(bi_list) < 3: return bi_zs_list def get_zs_range(bis): zg = min(bi.high for bi in bis) zd = max(bi.low for bi in bis) return zg, zd def is_bi_overlap_range(bi, zg, zd): return bi.high >= zd and bi.low <= zg def check_zs_position_filter(last_zs, zg, zd, bis): if last_zs is None: return True if zg <= last_zs.zd: return bis[0].dir == Chan_BI_DIR.UP and bis[-1].dir == Chan_BI_DIR.UP if zd >= last_zs.zg: return bis[0].dir == Chan_BI_DIR.DOWN and bis[-1].dir == Chan_BI_DIR.DOWN return True def set_zs_bi_list(zs, bis): zs.bi_list = list(bis) for bi in zs.bi_list: bi.set_bi_zs(zs) zs.set_gg(max(bi.high for bi in zs.bi_list)) zs.set_dd(min(bi.low for bi in zs.bi_list)) zs.classify_zs() last_zs = None start_idx = 0 while start_idx + 2 < len(bi_list): bi1 = bi_list[start_idx] bi2 = bi_list[start_idx + 1] bi3 = bi_list[start_idx + 2] if not (bi1.is_sure and bi2.is_sure and bi3.is_sure): start_idx += 1 continue if not (bi1.dir != bi2.dir and bi1.dir == bi3.dir): start_idx += 1 continue zg, zd = get_zs_range([bi1, bi2, bi3]) if zg <= zd: start_idx += 1 continue bis_for_zs = [bi1, bi2, bi3] extend_idx = start_idx + 3 while extend_idx + 1 < len(bi_list): leave_bi = bi_list[extend_idx] back_bi = bi_list[extend_idx + 1] if not (leave_bi.is_sure and back_bi.is_sure): break if not is_bi_overlap_range(back_bi, zg, zd): break bis_for_zs.append(leave_bi) bis_for_zs.append(back_bi) extend_idx += 2 if not check_zs_position_filter(last_zs, zg, zd, bis_for_zs): start_idx += 1 continue zs_dir = Chan_ZS_DIR.UP if bi1.dir == Chan_BI_DIR.DOWN else Chan_ZS_DIR.DOWN zs = ChanBIZS(bi1, len(bi_zs_list), zs_dir) zs.set_zg(zg) zs.set_zd(zd) set_zs_bi_list(zs, bis_for_zs) zs.set_end_bi(bis_for_zs[-1], bis_for_zs[-1].sure_time) if last_zs: last_zs.set_next(zs) zs.set_pre(last_zs) bi_zs_list.append(zs) last_zs = zs start_idx = start_idx + len(bis_for_zs) # 与 cal_bi_zs_list 一致：最后一笔未确认时末中枢标为未完成；若其后已出现确认的离开笔，仍按离开前最后一笔确认中枢结束 if last_zs: last_zs.is_sure = bi_list[-1].is_sure if last_zs and not last_zs.is_sure: if last_zs.bi_list and len(last_zs.bi_list) > 0: last_bi_of_zs = last_zs.bi_list[-1] last_bi_idx = -1 for i, bi in enumerate(bi_list): if bi == last_bi_of_zs: last_bi_idx = i break has_leave = False if last_bi_idx >= 0 and last_bi_idx + 1 < len(bi_list): for i in range(last_bi_idx + 1, len(bi_list)): bi = bi_list[i] if bi.is_sure: leave = (bi.low > last_zs.zg and bi.high > last_zs.zg) or \ (bi.high < last_zs.zd and bi.low < last_zs.zd) if leave: has_leave = True break if has_leave: if last_bi_of_zs.is_sure: last_zs.set_end_bi(last_bi_of_zs, last_bi_of_zs.sure_time) return bi_zs_list def find_all_bsp(self, bi_list, bi_zs_list): """ 笔中枢的三类买卖点识别 三类买点：中枢形成后，一笔向上离开中枢（低点 > zg）， 随后回拉的一笔低点不跌回中枢（低点 >= zg），确认支撑有效。 三类卖点：中枢形成后，一笔向下离开中枢（高点 < zd）， 随后反弹的一笔高点不回到中枢（高点 <= zd），确认压力有效。 参数: bi_list: 笔列表 bi_zs_list: 笔中枢列表（二维列表，每个seg内的中枢列表） 返回: bsp_list: ChanBSP 列表，包含所有识别到的三类买卖点 """ bsp_list = [] if len(bi_list) < 4 or len(bi_zs_list) == 0: return bsp_list for zs in bi_zs_list: if not zs.is_sure or len(zs.bi_list) < 3: continue #print(zs.start_time, zs.end_time, zs.dir, zs.is_sure, len(zs.bi_list)) # 中枢结束后的第一笔（离开笔） last_zs_bi = zs.bi_list[-1] if last_zs_bi.dir == Chan_BI_DIR.UP: if last_zs_bi.is_sure and last_zs_bi.end_klc.high <= zs.zg or (last_zs_bi.next and last_zs_bi.next.is_sure and last_zs_bi.next.end_klc.low < zs.zd): leave_bi = last_zs_bi.next else: leave_bi = last_zs_bi else: if last_zs_bi.is_sure and last_zs_bi.end_klc.low >= zs.zd or (last_zs_bi.next and last_zs_bi.next.is_sure and last_zs_bi.next.end_klc.high > zs.zg): leave_bi = last_zs_bi.next else: leave_bi = last_zs_bi #print(zs.zg, zs.zd) if leave_bi is None or not leave_bi.is_sure: continue if (zs.dir == Chan_ZS_DIR.UP and leave_bi.dir == Chan_BI_DIR.UP and leave_bi.end_klc.high < zs.zg and leave_bi.end_klc.high > zs.zd) or (zs.dir == Chan_ZS_DIR.DOWN and leave_bi.dir == Chan_BI_DIR.DOWN and leave_bi.end_klc.low < zs.zg and leave_bi.end_klc.low > zs.zd): #print("--------------------", leave_bi.dir, leave_bi.end_klc.high, leave_bi.end_klc.low, zs.zg, zs.zd) leave_bi = leave_bi.next # 三类买点：向上离开中枢后回拉不破 zg #print("Leave bi:", leave_bi.start_time, leave_bi.end_time, leave_bi.dir, leave_bi.is_sure, leave_bi.low, leave_bi.high) if leave_bi.dir == Chan_BI_DIR.UP: first_bsp_bi_div = self.check_bi_div(zs, leave_bi) # 确认一类卖点：离开断能量小于进入段能量 if first_bsp_bi_div: bsp = ChanBSP( leave_bi, len(bsp_list), Chan_BSP_TYPE.S1, Chan_BSP_DIR.SELL, leave_bi.sure_time, zs.index+1, zs, None ) leave_bi.end_klc.set_bsp_type(Chan_BSP_TYPE.S1) bsp_list.append(bsp) # 回拉笔 pullback_bi = leave_bi.next #print(pullback_bi.start_klc.start_time, pullback_bi.dir, pullback_bi.is_sure, pullback_bi.low, pullback_bi.high) if pullback_bi and pullback_bi.is_sure and pullback_bi.dir == Chan_BI_DIR.DOWN: if pullback_bi.low >= zs.zg: # 确认三类买点：回拉笔的低点不跌回中枢 bsp = ChanBSP( pullback_bi, len(bsp_list), Chan_BSP_TYPE.B3, Chan_BSP_DIR.BUY, pullback_bi.sure_time, zs.index+1, zs, None ) pullback_bi.end_klc.set_bsp_type(Chan_BSP_TYPE.B3) bsp_list.append(bsp) # 二类卖点 if first_bsp_bi_div: second_bsp_bi = pullback_bi.next if second_bsp_bi and second_bsp_bi.is_sure and second_bsp_bi.end_klc.high < leave_bi.end_klc.high: # 确认二类卖点：一类卖点后回拉不超过一类卖点高点 bsp = ChanBSP( second_bsp_bi, len(bsp_list), Chan_BSP_TYPE.S2, Chan_BSP_DIR.SELL, second_bsp_bi.sure_time, zs.index+1, zs, None ) second_bsp_bi.end_klc.set_bsp_type(Chan_BSP_TYPE.B2) bsp_list.append(bsp) # 三类卖点：向下离开中枢后反弹不破 zd elif leave_bi.dir == Chan_BI_DIR.DOWN: first_bsp_bi_div = self.check_bi_div(zs, leave_bi) # 确认一类买点：离开段能量小于进入段 if first_bsp_bi_div: bsp = ChanBSP( leave_bi, len(bsp_list), Chan_BSP_TYPE.B1, Chan_BSP_DIR.BUY, leave_bi.sure_time, zs.index+1, zs, None ) leave_bi.end_klc.set_bsp_type(Chan_BSP_TYPE.B1) bsp_list.append(bsp) # 反弹笔 bounce_bi = leave_bi.next #print(bounce_bi.start_klc.start_time, bounce_bi.dir, bounce_bi.is_sure, bounce_bi.low, bounce_bi.high) if bounce_bi and bounce_bi.is_sure and bounce_bi.dir == Chan_BI_DIR.UP: if bounce_bi.high <= zs.zd: # 确认三类卖点：反弹笔的高点不回到中枢 bsp = ChanBSP( bounce_bi, len(bsp_list), Chan_BSP_TYPE.S3, Chan_BSP_DIR.SELL, bounce_bi.sure_time, zs.index+1, zs, None ) bounce_bi.end_klc.set_bsp_type(Chan_BSP_TYPE.S3) bsp_list.append(bsp) # 二类卖点 if first_bsp_bi_div: second_bsp_bi = bounce_bi.next if second_bsp_bi and second_bsp_bi.is_sure and second_bsp_bi.end_klc.low > leave_bi.end_klc.low: # 确认二类买点：一类买点后回拉不超过一类卖点高点 bsp = ChanBSP( second_bsp_bi, len(bsp_list), Chan_BSP_TYPE.B2, Chan_BSP_DIR.BUY, second_bsp_bi.sure_time, zs.index+1, zs, None ) second_bsp_bi.end_klc.set_bsp_type(Chan_BSP_TYPE.B2) bsp_list.append(bsp) return bsp_list def check_bi_div(self, zs, leave_bi): enter_bi = zs.bi_list[0].pre macdhist_div = 0 if enter_bi and enter_bi.dir == leave_bi.dir: macdhist_div = abs(leave_bi.macd_hist) - abs(enter_bi.macd_hist) #print(enter_bi.end_time, leave_bi.end_time, macdhist_div < 0) return macdhist_div < 0 def find_first_bsp(self, bi_list, bi_zs_list): """ 笔中枢的一类买卖点识别 一类买点：下跌趋势中，最后一个中枢完成后，向下离开中枢的笔创新低， 但该笔与进入中枢前的最后一笔下跌形成底背驰（力度减弱）， 即趋势力竭的转折点。 一类卖点：上涨趋势中，最后一个中枢完成后，向上离开中枢的笔创新高， 但该笔与进入中枢前的最后一笔上涨形成顶背驰（力度减弱）， 即趋势力竭的转折点。 简化判断：中枢形成后，离开中枢的笔（突破笔）本身即为一类买卖点的触发笔。 参数: bi_list: 笔列表 bi_zs_list: 笔中枢列表（扁平列表，每个元素是一个中枢对象） 返回: bsp_list: ChanBSP 列表，包含所有识别到的一类买卖点 """ bsp_list = [] if len(bi_list) < 4 or len(bi_zs_list) == 0: return bsp_list for zs in bi_zs_list: if not zs.is_sure or len(zs.bi_list) < 3: continue # 找到中枢的最后一笔 last_zs_bi = zs.bi_list[-1] # 确定离开笔：中枢最后一笔之后的第一笔 if last_zs_bi.dir == Chan_BI_DIR.UP: # 中枢最后一笔向上，如果没有真正离开中枢，取下一笔 if last_zs_bi.is_sure and last_zs_bi.end_klc.high <= zs.zg: leave_bi = last_zs_bi.next else: leave_bi = last_zs_bi else: # 中枢最后一笔向下，如果没有真正离开中枢，取下一笔 if last_zs_bi.is_sure and last_zs_bi.end_klc.low >= zs.zd: leave_bi = last_zs_bi.next else: leave_bi = last_zs_bi if leave_bi is None or not leave_bi.is_sure: continue # 一类买点：向下离开中枢（leave_bi向下，低点 < zd），趋势力竭 if leave_bi.dir == Chan_BI_DIR.DOWN and leave_bi.low < zs.zd: # 背驰判断：比较离开笔与中枢内最后一笔同向笔的MACD柱状累积面积 # 缠论原文：两段同向走势的MACD柱状面积比较，面积缩小即为背驰 compare_bi = None for bi in reversed(zs.bi_list): if bi.dir == Chan_BI_DIR.DOWN and bi is not leave_bi: compare_bi = bi break is_divergence = False if compare_bi: # 笔的macd_hist是该笔内所有KLU的macdhist累积面积 leave_macd_area = abs(leave_bi.macd_hist) compare_macd_area = abs(compare_bi.macd_hist) # 价格创新低但MACD面积缩小 = 底背驰 if leave_bi.low <= compare_bi.low and leave_macd_area < compare_macd_area: is_divergence = True # 即使没创新低，MACD面积明显缩小也算背驰 elif leave_macd_area < compare_macd_area * 0.5: is_divergence = True else: # 没有对比笔时，只要离开中枢就算一类买点 is_divergence = True if is_divergence: bsp = ChanBSP( leave_bi, len(bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.BUY, leave_bi.sure_time, 1, zs, None ) bsp_list.append(bsp) # 一类卖点：向上离开中枢（leave_bi向上，高点 > zg），趋势力竭 elif leave_bi.dir == Chan_BI_DIR.UP and leave_bi.high > zs.zg: # 背驰判断：比较离开笔与中枢内最后一笔同向笔的MACD柱状累积面积 compare_bi = None for bi in reversed(zs.bi_list): if bi.dir == Chan_BI_DIR.UP and bi is not leave_bi: compare_bi = bi break is_divergence = False if compare_bi: leave_macd_area = abs(leave_bi.macd_hist) compare_macd_area = abs(compare_bi.macd_hist) # 价格创新高但MACD面积缩小 = 顶背驰 if leave_bi.high >= compare_bi.high and leave_macd_area < compare_macd_area: is_divergence = True # 即使没创新高，MACD面积明显缩小也算背驰 elif leave_macd_area < compare_macd_area * 0.5: is_divergence = True else: is_divergence = True if is_divergence: bsp = ChanBSP( leave_bi, len(bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.SELL, leave_bi.sure_time, 1, zs, None ) bsp_list.append(bsp) return bsp_list def find_second_bsp(self, bi_list, first_bsp_list): """ 笔中枢的二类买卖点识别 二类买点：一类买点出现后，价格向上反弹一笔，再回落一笔， 回落笔的低点不跌破一类买点的低点，确认底部成立。 二类卖点：一类卖点出现后，价格向下回落一笔，再反弹一笔， 反弹笔的高点不超过一类卖点的高点，确认顶部成立。 参数: bi_list: 笔列表 first_bsp_list: 一类买卖点列表（find_first_bsp 的返回值） 返回: bsp_list: ChanBSP 列表，包含所有识别到的二类买卖点 """ bsp_list = [] if not first_bsp_list or len(bi_list) < 4: return bsp_list for first_bsp in first_bsp_list: trigger_bi = first_bsp.bi # 一类买卖点的触发笔 if first_bsp.dir == Chan_BSP_DIR.BUY: # 一买之后：trigger_bi 向下 -> 反弹笔(向上) -> 回落笔(向下) # 回落笔的低点 > trigger_bi 的低点 => 二类买点 bounce_bi = trigger_bi.next # 反弹笔（向上） if bounce_bi and bounce_bi.is_sure and bounce_bi.dir == Chan_BI_DIR.UP: pullback_bi = bounce_bi.next # 回落笔（向下） if pullback_bi and pullback_bi.is_sure and pullback_bi.dir == Chan_BI_DIR.DOWN: if pullback_bi.low > trigger_bi.low: bsp = ChanBSP( pullback_bi, len(bsp_list), Chan_BSP_TYPE.T2, Chan_BSP_DIR.BUY, pullback_bi.sure_time, 1, first_bsp.zs, None ) bsp_list.append(bsp) elif first_bsp.dir == Chan_BSP_DIR.SELL: # 一卖之后：trigger_bi 向上 -> 回落笔(向下) -> 反弹笔(向上) # 反弹笔的高点 < trigger_bi 的高点 => 二类卖点 drop_bi = trigger_bi.next # 回落笔（向下） if drop_bi and drop_bi.is_sure and drop_bi.dir == Chan_BI_DIR.DOWN: bounce_bi = drop_bi.next # 反弹笔（向上） if bounce_bi and bounce_bi.is_sure and bounce_bi.dir == Chan_BI_DIR.UP: if bounce_bi.high < trigger_bi.high: bsp = ChanBSP( bounce_bi, len(bsp_list), Chan_BSP_TYPE.T2, Chan_BSP_DIR.SELL, bounce_bi.sure_time, 1, first_bsp.zs, None ) bsp_list.append(bsp) return bsp_list def calculate_seg_zs(self, seg_list): return self.get_seg_zs_list(seg_list) def get_seg_zs_list(self, seg_list): """ 根据缠论线段中枢定义计算中枢 从第4根线段开始（索引3），每3根线段为一组检查 上涨中枢：后中枢 zd > 前中枢 zg（不重叠上移） 下跌中枢：后中枢 zg < 前中枢 zd（不重叠下移） 盘整/扩张：后中枢与前中枢整体区间（GG/DD）有交集 中枢可按两段一组继续扩展到5根、7根... """ zs_list = [] if len(seg_list) < 3: return zs_list last_zs = None # 从第4根线段开始（索引3），每3根为一组 start_idx = 3 while start_idx < len(seg_list): # 取连续3个线段 if start_idx + 2 >= len(seg_list): break seg1 = seg_list[start_idx] seg2 = seg_list[start_idx + 1] seg3 = seg_list[start_idx + 2] # 三个线段都必须是已确认的 if not (seg1.is_sure and seg2.is_sure and seg3.is_sure): start_idx += 1 continue # 计算这3个线段的中枢区间 zg = min(seg1.high, seg2.high, seg3.high) zd = max(seg1.low, seg2.low, seg3.low) if zg <= zd: start_idx += 1 #print(seg1.start_bi.start_klc.end_time, "not valid", zg, zd) continue # 判断中枢类型（按注释定义） # 上涨中枢：后中枢 zd > 前中枢 zg（不重叠上移） # 下跌中枢：后中枢 zg < 前中枢 zd（不重叠下移） # 盘整/扩张：后中枢与前中枢区间有交集 if last_zs is None: # 第一个中枢仅按线段形态判定方向 if seg1.dir == Chan_SEG_DIR.DOWN: # 下跌+上涨+下跌，对应上涨中枢 zs_dir = Chan_ZS_DIR.UP valid = (seg2.dir == Chan_SEG_DIR.UP and seg3.dir == Chan_SEG_DIR.DOWN) else: # 上涨+下跌+上涨，对应下跌中枢 zs_dir = Chan_ZS_DIR.DOWN valid = (seg2.dir == Chan_SEG_DIR.DOWN and seg3.dir == Chan_SEG_DIR.UP) else: is_up_zs = zd > last_zs.zg is_down_zs = zg < last_zs.zd if is_up_zs: # 不重叠上移 zs_dir = Chan_ZS_DIR.UP valid = (seg1.dir == Chan_SEG_DIR.DOWN and seg2.dir == Chan_SEG_DIR.UP and seg3.dir == Chan_SEG_DIR.DOWN) elif is_down_zs: # 不重叠下移 zs_dir = Chan_ZS_DIR.DOWN valid = (seg1.dir == Chan_SEG_DIR.UP and seg2.dir == Chan_SEG_DIR.DOWN and seg3.dir == Chan_SEG_DIR.UP) create_new_zs = False # 验证是否有效 if not valid: # 如果新中枢和前一个中枢的中枢区间有重叠，不行成新中枢需要合并两个中枢 is_in_last_zs = (zd > last_zs.zd and zd < last_zs.zg) or (zg < last_zs.zg and zg > last_zs.zd) or (zg > last_zs.zg and zd < last_zs.zd) or (zg < last_zs.zg and zd > last_zs.zd) if is_in_last_zs: #print(seg1.start_time, "New zs is in last zs, not valid") last_zs.extend_zs(seg_list[last_zs.seg_list[-1].index:(seg3.index + 1)]) create_new_zs = False else: start_idx += 1 continue else: create_new_zs = True if last_zs and create_new_zs: last_seg = last_zs.seg_list[-1] last_bi = last_seg.end_bi if last_bi: last_zs.is_sure = True last_zs.set_end_klc(last_bi.end_klc, last_bi.sure_time, 0, last_seg) last_zs.set_end_seg(last_seg) zs = last_zs if create_new_zs: # 创建新中枢 gg = max(seg1.high, seg2.high, seg3.high) dd = min(seg1.low, seg2.low, seg3.low) zs = ChanZS(seg1, len(zs_list), zs_dir) zs.set_zg(zg) zs.set_zd(zd) zs.set_gg(gg) zs.set_dd(dd) zs.is_sure = False zs.seg_list = [seg1, seg2, seg3] # 若第二线段与 [zd,zg] 重叠（如离开后回抽回到前中枢）则并入扩展 added_after_leave = [] leave_index = start_idx + 4 is_break = False while leave_index < len(seg_list): s = seg_list[leave_index] if not s.is_sure: break sh = max(s.start_bi.high, s.end_bi.high) if s.end_bi else s.start_bi.high sl = min(s.start_bi.low, s.end_bi.low) if s.end_bi else s.start_bi.low if sh >= zs.zd and sl <= zs.zg: added_after_leave.append(s.pre) added_after_leave.append(s) leave_index += 2 else: next_seg = s.next if next_seg and next_seg.is_sure: if next_seg.dir == Chan_SEG_DIR.UP: if next_seg.high <= zs.zg and next_seg.low >= zs.zd: leave_index += 2 continue else: is_break = True else: if next_seg.low >= zs.zd and next_seg.low <= zs.zg: leave_index += 2 continue else: is_break = True else: break if is_break: break if added_after_leave: #print(len(added_after_leave)) segs_for_zs = list(zs.seg_list) + list(added_after_leave) seg_highs = [s.high for s in segs_for_zs] seg_lows = [s.low for s in segs_for_zs] zs.set_gg(max(seg_highs)) zs.set_dd(min(seg_lows)) zs.seg_list = segs_for_zs seg = segs_for_zs[-1] #if seg.end_bi: #zs.set_end_klc(seg.end_bi.end_klc, seg.sure_time, 0, seg) #zs.set_end_seg(seg) #zs.is_sure = True start_idx = start_idx + len(added_after_leave) if last_zs and last_zs.index != zs.index: last_zs.set_next(zs) zs.set_pre(last_zs) zs_list.append(zs) last_zs = zs # 移动到下一组 start_idx += 4 if last_zs: last_zs.is_sure = seg_list[-1].is_sure """ # 处理最后一个未确认的中枢 - 不自动扩展，保持未完成状态 if last_zs and not last_zs.is_sure: # 获取中枢最后一个线段的索引 if last_zs.seg_list and len(last_zs.seg_list) > 0: last_seg_of_zs = last_zs.seg_list[-1] # 找到这个线段在seg_list中的索引 last_seg_idx = -1 for i, seg in enumerate(seg_list): if seg == last_seg_of_zs: last_seg_idx = i break # 从中枢最后一个线段之后检查是否有离开 has_leave = False if last_seg_idx >= 0 and last_seg_idx + 1 < len(seg_list): for i in range(last_seg_idx + 1, len(seg_list)): seg = seg_list[i] if seg.is_sure: # 检查是否离开中枢 leave = (seg.low > last_zs.zg and seg.high > last_zs.zg) or \ (seg.high < last_zs.zd and seg.low < last_zs.zd) if leave: has_leave = True break if not has_leave: # 没有离开，保持未完成状态 pass else: # 有离开，确认中枢 if last_seg_of_zs.end_bi: #print(last_seg_of_zs.start_time, "last_seg_of_zs.end_time", last_seg_of_zs.end_time) last_zs.set_end_klc(last_seg_of_zs.end_bi.end_klc, last_seg_of_zs.sure_time, 0, last_seg_of_zs) last_zs.set_end_seg(last_seg_of_zs) last_zs.is_sure = True """ return zs_list def get_big_zs_list(self, zs_list): """ 中枢扩张：将区间重叠的连续中枢合并为大级别中枢，便于显示更大级别的震荡区间。 重叠定义：两中枢 [zd,zg] 有交集，即 (zs_i.zg >= zs_j.zd and zs_i.zd <= zs_j.zg)。 """ big_list = [] if len(zs_list) < 2: return big_list i = 0 while i < len(zs_list): group = [zs_list[i]] j = i + 1 while j < len(zs_list): cur = zs_list[j] # 与当前组内任一中枢有重叠即算扩张（通常只需与组内最后一个比） last_in_group = group[-1] overlap = (last_in_group.zg >= cur.zd and last_in_group.zd <= cur.zg) if overlap: group.append(cur) j += 1 else: break if len(group) >= 2: big = ChanZS_Big(group) big.index = len(big_list) big_list.append(big) i = j if len(group) >= 2 else i + 1 return big_list def get_klu_list(self, dataframe): klu_list = self.get_kl_data(dataframe) #klu_list = self.cal_klu_pattern(klu_list) return klu_list def cal_klu_pattern(self, klu_list): """ 计算裸K的pattern - 识别反转形态 """ if not klu_list or len(klu_list) < 3: return klu_list for i, klu in enumerate(klu_list): # 单根K线反转模式识别 self._detect_single_reversal_pattern(klu) # 双根K线形态识别 if i >= 1: self._detect_double_pattern(klu_list[i-1], klu) # 三根K线形态识别 if i >= 2: self._detect_triple_pattern(klu_list[i-2], klu_list[i-1], klu) #if klu.pattern != Chan_KLU_PATTERN.UNKNOWN: #print(klu.time, klu.pattern, klu.lower_shadow_ratio, klu.upper_shadow_ratio, klu.body_ratio, klu.lower_shadow_ratio/klu.body_ratio, klu.upper_shadow_ratio/klu.body_ratio) return klu_list def _detect_single_reversal_pattern(self, klu): """检测单根K线反转模式""" body = abs(klu.close - klu.open) upper_shadow = klu.high - max(klu.close, klu.open) lower_shadow = min(klu.close, klu.open) - klu.low total_range = klu.high - klu.low # 避免除零 if total_range == 0: return body_ratio = body / total_range upper_ratio = upper_shadow / total_range lower_ratio = lower_shadow / total_range #print(klu.time, upper_ratio, lower_ratio, body_ratio, upper_ratio/body_ratio, lower_ratio/body_ratio) # 避免body_ratio为0时的除零错误 if body_ratio == 0: return # 锤子线/上吊线 - 反转信号 if lower_ratio / body_ratio >= 2: # 锤子线：底部反转，需要前面一段 if klu.close > klu.open and klu.pre: klu.set_pattern(Chan_KLU_PATTERN.HAMMER) # 底部反转 # 上吊线：顶部反转，需要前一根是上涨趋势 elif klu.close < klu.open and klu.pre: klu.set_pattern(Chan_KLU_PATTERN.HANGING_MAN) # 顶部反转 # 倒锤子线/射击之星 - 反转信号 elif upper_ratio / body_ratio >= 2: # 倒锤子线：底部反转，需要前一根是下跌趋势 if klu.close > klu.open and klu.pre: klu.set_pattern(Chan_KLU_PATTERN.INVERTED_HAMMER) # 底部反转 # 射击之星：顶部反转，需要前一根是上涨趋势 elif klu.close < klu.open and klu.pre: klu.set_pattern(Chan_KLU_PATTERN.SHOOTING_STAR) # 顶部反转 # 十字星 - 反转信号 elif body_ratio <= 0.1: if upper_ratio > 0.4 and lower_ratio > 0.4: klu.set_pattern(Chan_KLU_PATTERN.LONG_LEGGED_DOJI) # 强烈反转信号 elif upper_ratio > 0.4 and lower_ratio <= 0.1: # 墓碑十字星：顶部反转，需要前一根是上涨趋势 if klu.pre and klu.pre.close > klu.pre.open: klu.set_pattern(Chan_KLU_PATTERN.GRAVESTONE_DOJI) # 顶部反转 elif lower_ratio > 0.4 and upper_ratio <= 0.1: # 蜻蜓十字星：底部反转，需要前一根是下跌趋势 if klu.pre and klu.pre.close < klu.pre.open: klu.set_pattern(Chan_KLU_PATTERN.DRAGONFLY_DOJI) # 底部反转 else: klu.set_pattern(Chan_KLU_PATTERN.DOJI) # 一般反转信号 def _detect_double_pattern(self, prev_klu, curr_klu): """检测两根K线形成的形态 包括：吞没形态(看涨/看跌)、乌云盖顶、曙光初现 """ # 如果前一根K线已经有形态，不再识别双K线形态 if prev_klu.pattern != Chan_KLU_PATTERN.UNKNOWN: return # 计算K线实体 prev_body = abs(prev_klu.close - prev_klu.open) curr_body = abs(curr_klu.close - curr_klu.open) # 判断K线颜色（阴阳） prev_bullish = prev_klu.close > prev_klu.open curr_bullish = curr_klu.close > curr_klu.open # 检查是否存在长期趋势（至少需要5根K线的趋势） def check_long_trend(klu, bullish_trend=True, min_bars=5): """检查是否存在长期趋势 bullish_trend=True: 检查上涨趋势 bullish_trend=False: 检查下跌趋势 min_bars: 最少需要多少根K线形成趋势 """ if not klu or not klu.pre: return False return True # 使用EMA指标判断长期趋势 if klu.ema52 > 0: if bullish_trend and klu.close < klu.ema52: return False if not bullish_trend and klu.close > klu.ema52: return False # 检查连续的K线方向 count = 0 current = klu.pre while current and count < min_bars: if not current.pre: break if bullish_trend: # 上涨趋势：当前收盘价高于前一根收盘价 if current.close <= current.pre.close: break else: # 下跌趋势：当前收盘价低于前一根收盘价 if current.close >= current.pre.close: break count += 1 current = current.pre return count >= min_bars # 1. 看涨吞没形态：前阴后阳，后者完全吞没前者 # 要求前面有明显的下跌趋势 if not prev_bullish and curr_bullish and \ abs(curr_klu.open - prev_klu.close) < 10 and \ curr_klu.close > prev_klu.open and \ check_long_trend(prev_klu, bullish_trend=False, min_bars=5): curr_klu.set_pattern(Chan_KLU_PATTERN.BULLISH_ENGULFING) return # 2. 看跌吞没形态：前阳后阴，后者完全吞没前者 # 要求前面有明显的上涨趋势 if prev_bullish and not curr_bullish and \ abs(curr_klu.open - prev_klu.close) < 10 and \ curr_klu.close < prev_klu.open and \ check_long_trend(prev_klu, bullish_trend=True, min_bars=5): curr_klu.set_pattern(Chan_KLU_PATTERN.BEARISH_ENGULFING) return # 3. 乌云盖顶：前阳后阴，后者开盘价高于前者最高价，收盘价在前者实体中部以下 # 要求前面有明显的上涨趋势 if prev_bullish and not curr_bullish and \ curr_klu.open > prev_klu.high and \ curr_klu.close < (prev_klu.open + prev_klu.close) / 2 and \ curr_klu.close > prev_klu.open and \ check_long_trend(prev_klu, bullish_trend=True, min_bars=5): curr_klu.set_pattern(Chan_KLU_PATTERN.DARK_CLOUD_COVER) return # 4. 曙光初现：前阴后阳，后者开盘价低于前者最低价，收盘价在前者实体中部以上 # 要求前面有明显的下跌趋势 if not prev_bullish and curr_bullish and \ curr_klu.open < prev_klu.low and \ curr_klu.close > (prev_klu.open + prev_klu.close) / 2 and \ curr_klu.close < prev_klu.open and \ check_long_trend(prev_klu, bullish_trend=False, min_bars=5): curr_klu.set_pattern(Chan_KLU_PATTERN.PIERCING_LINE) return # 平顶和平底移至三根K线形态中判断 def _detect_triple_pattern(self, first_klu, second_klu, third_klu): """检测三根K线形成的形态 包括：早晨之星、黄昏之星、平顶、平底 """ # 如果前两根K线已经有形态，不再识别三K线形态 if first_klu.pattern != Chan_KLU_PATTERN.UNKNOWN or \ second_klu.pattern != Chan_KLU_PATTERN.UNKNOWN: return # 判断K线颜色（阴阳） first_bullish = first_klu.close > first_klu.open second_bullish = second_klu.close > second_klu.open third_bullish = third_klu.close > third_klu.open # 计算实体大小 first_body = abs(first_klu.close - first_klu.open) second_body = abs(second_klu.close - second_klu.open) third_body = abs(third_klu.close - third_klu.open) # 检查是否存在长期趋势（至少需要5根K线的趋势） def check_long_trend(klu, bullish_trend=True, min_bars=5): """检查是否存在长期趋势 bullish_trend=True: 检查上涨趋势 bullish_trend=False: 检查下跌趋势 min_bars: 最少需要多少根K线形成趋势 """ if not klu or not klu.pre: return False # 使用EMA指标判断长期趋势 if klu.ema52 > 0: if bullish_trend and klu.close < klu.ema52: return False if not bullish_trend and klu.close > klu.ema52: return False # 检查连续的K线方向 count = 0 current = klu.pre while current and count < min_bars: if not current.pre: break if bullish_trend: # 上涨趋势：当前收盘价高于前一根收盘价 if current.close <= current.pre.close: break else: # 下跌趋势：当前收盘价低于前一根收盘价 if current.close >= current.pre.close: break count += 1 current = current.pre return count >= min_bars # 1. 早晨之星：第一根阴线，第二根十字星或小实体，第三根阳线 # 要求前面有明显的下跌趋势 if not first_bullish and third_bullish and \ second_body < first_body * 0.3 and \ third_body > first_body * 0.5 and \ max(second_klu.open, second_klu.close) < first_klu.close and \ min(second_klu.open, second_klu.close) < third_klu.open and \ third_klu.close > (first_klu.open + first_klu.close) / 2 and \ check_long_trend(first_klu, bullish_trend=False, min_bars=7): third_klu.set_pattern(Chan_KLU_PATTERN.MORNING_STAR) return # 2. 黄昏之星：第一根阳线，第二根十字星或小实体，第三根阴线 # 要求前面有明显的上涨趋势 if first_bullish and not third_bullish and \ second_body < first_body * 0.3 and \ third_body > first_body * 0.5 and \ min(second_klu.open, second_klu.close) > first_klu.close and \ max(second_klu.open, second_klu.close) > third_klu.open and \ third_klu.close < (first_klu.open + first_klu.close) / 2 and \ check_long_trend(first_klu, bullish_trend=True, min_bars=7): third_klu.set_pattern(Chan_KLU_PATTERN.EVENING_STAR) return # 3. 平顶：三根K线的最高点几乎相同（上升趋势中更有意义） # 要求前面有明显的上涨趋势 if (abs(first_klu.high - second_klu.high) / first_klu.high < 0.0002 and abs(second_klu.high - third_klu.high) / second_klu.high < 0.0002 and check_long_trend(first_klu, bullish_trend=True, min_bars=7)): # 额外确认：价格接近阻力位或关键技术指标 is_near_resistance = False # 检查是否接近EMA52阻力位 if first_klu.ema52 > 0: resistance_level = first_klu.ema52 if abs(first_klu.high - resistance_level) / resistance_level < 0.01: is_near_resistance = True # 检查是否有成交量确认（成交量减少表示上涨动能减弱） volume_confirmation = False if (first_klu.volume > 0 and second_klu.volume > 0 and third_klu.volume > 0 and third_klu.volume < second_klu.volume and second_klu.volume < first_klu.volume): volume_confirmation = True if is_near_resistance or volume_confirmation: third_klu.set_pattern(Chan_KLU_PATTERN.TWEEZER_TOP) return # 4. 平底：三根K线的最低点几乎相同（下降趋势中更有意义） # 要求前面有明显的下跌趋势 if (abs(first_klu.low - second_klu.low) / first_klu.low < 0.0002 and abs(second_klu.low - third_klu.low) / second_klu.low < 0.0002 and check_long_trend(first_klu, bullish_trend=False, min_bars=7)): # 额外确认：价格接近支撑位或关键技术指标 is_near_support = False # 检查是否接近EMA52支撑位 if first_klu.ema52 > 0: support_level = first_klu.ema52 if abs(first_klu.low - support_level) / support_level < 0.01: is_near_support = True # 检查是否有成交量确认（成交量减少表示下跌动能减弱） volume_confirmation = False if (first_klu.volume > 0 and second_klu.volume > 0 and third_klu.volume > 0 and third_klu.volume < second_klu.volume and second_klu.volume < first_klu.volume): volume_confirmation = True if is_near_support or volume_confirmation: third_klu.set_pattern(Chan_KLU_PATTERN.TWEEZER_BOTTOM) return def get_decimal(self, value): return Decimal("{:.2f}".format(value))