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 from ChanKLU import ChanKLU from ChanKLC import ChanKLC from ChanBI import ChanBI from ChanSBI import ChanSBI from ChanSEG import ChanSEG from ChanZS import ChanZS 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 self.dataframe = resample_to_interval(df, interval) 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.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.cal_zs_list(self.bi_list, self.seg_list) self.chanmacd = ChanMACD(self.klu_list) self.klu_list = self.chanmacd.cal_macd_state() def get_ema52(self): if self.klu_list: ema52_value = self.klu_list[-1].ema52 # 处理NaN值 if pd.isna(ema52_value) or ema52_value is None: return None return float(ema52_value) return None def get_ema24(self): if self.klu_list: ema24_value = self.klu_list[-1].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 = 12 slow = 26 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) # 计算布林带中轨（移动平均线） 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']) 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['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(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_fx_type == Chan_KLC_FX.TOP4: klu_state_list.append("10") elif klc.klc_fx_type == Chan_KLC_FX.BOTTOM4: klu_state_list.append("-10") else: klu_state_list.append("00") klc_index += 1 else: klu_state_list.append("00") return klu_state_list 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_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)) 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() for klu in klu_list: 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) 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 cal_bi_list(self, klc_list): bi_list = [] last_top = None last_bottom = None for klc in klc_list: 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: fx = Chan_FX_TYPE.UNKNOWN # Do nothing if fx == Chan_FX_TYPE.UNKNOWN: if len(bi_list) > 0: bi_list[-1].add_klc(klc) 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: 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: 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: if fx == Chan_FX_TYPE.TOP: 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: #klc.set_fx(Chan_FX_TYPE.TT) #klc.set_state("20") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #klc.cal_invisible() #klc.set_klc_fx_type(Chan_KLC_FX.TOP3) #print(klc.start_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") #klc.set_fx(fx) #klc.set_state("10") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) else: # 不满足结合律的分型 if last_bottom.index + 4 > klc.index: if last_top.high > klc.high: #print(klc.start_time, klc.fx, "二类卖点Sell 1") #klc.set_fx(Chan_FX_TYPE.PTOP) bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) # New TOP Found replace last top else: if last_top.index + 4 < 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]) else: 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, "无效分型") # 满足结合律 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) #klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM7) #klc.bb_out = True 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) #klc.set_state('30') 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") #print(klc.start_time, klc.fx, "笔卖点Sell 2") # 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.set_fx(Chan_FX_TYPE.UNKNOWN) 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]) #print(klc.start_time, klc.fx, "笔卖点Sell 3") else: klc.set_fx(Chan_FX_TYPE.TT) #klc.set_state('20') #print(klc.start_time, klc.fx, "二类卖点Sell 2") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) else: if last_bottom: # 不满足结合律的分型 if last_bottom.index + 4 > 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]) #print(klc.start_time, klc.fx, "一类卖点Sell 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) #klc.set_klc_fx_type(Chan_KLC_FX.TOP6) #klc.bb_out = True 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 Top Change 5") #print(klc.start_time, 'Create first top') #print(klc.start_time, klc.fx, "笔卖点Sell 1") #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: #klc.set_fx(Chan_FX_TYPE.BB) #klc.set_state("-20") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #klc.cal_invisible() #klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM3) #print(last_bottom.start_time, last_bottom.end_time, "--------------------------------1") #print(klc.start_time, klc.fx, "二类买点Buy 1") else: # A new bottom found #last_bottom.set_fx(Chan_FX_TYPE.UNKNOWN) last_bottom = klc #print(klc.start_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.start_time, klc.fx, "一类买点Buy 1") #klc.set_state("-10") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) else: # 不满足结合律的分型 if last_top.index + 4 > klc.index: if last_bottom.low < klc.low: #klc.set_fx(Chan_FX_TYPE.PBOTTOM) #klc.set_fx(Chan_FX_TYPE.BB) #klc.set_state("-100") #print(klc.start_time, klc.fx, "中枢买点Buy 1") bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) # Found new bottom else: if last_bottom.index + 4 < 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.set_fx(Chan_FX_TYPE.PBOTTOM) 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") #klc.set_state("-10") #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]) else: #klc.set_fx(Chan_FX_TYPE.UNKNOWN) bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #print(klc.start_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) #klc.set_klc_fx_type(Chan_KLC_FX.TOP6) #klc.bb_out = True 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) #klc.set_state('-30') bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #print(klc.start_time, klc.fx, "笔买点Buy 2") #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 + 4 > 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) #klc.bb_out = True 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") #if klc.fx != Chan_FX_TYPE.UNKNOWN: #print(klc.start_time, klc.fx, klc.index) """ for klc in klc_list: if klc.fx == Chan_FX_TYPE.TOP: klc.state = "10" #print(klc.time, klc.state) if klc.fx == Chan_FX_TYPE.BOTTOM: klc.state = "-10" #print(klc.time, klc.state) """ #for index in range(0, 10): #print(bi_list[index].start_time, bi_list[index].start_klc.start_time, bi_list[index].dir) return bi_list def check_top_fx(self, last_bottom, klc): if last_bottom.high > klc.pre.low or last_bottom.high > klc.next.low: 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: 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.append(zs_list) return bi_zs_list def get_zs_list(self, bi_list, seg_list): zs_list = [] bsp_list = [] if len(seg_list) > 3: last_zs = None first_bi_out = None in_again = False bi_out_count = 0 zs_count = 0 for seg in seg_list: # No zs or Last ZS is completed if len(zs_list) == 0 or (last_zs and last_zs.is_sure): # Has three completed segments if seg.next and seg.next.next: if seg.next.next.is_sure: zg = min(seg.high, seg.next.high, seg.next.next.high) zd = max(seg.low, seg.next.low, seg.next.next.low) gg = max(seg.high, seg.next.high, seg.next.next.high) dd = min(seg.low, seg.next.low, seg.next.next.low) ddir = None if last_zs: if zg < last_zs.zd: ddir = Chan_ZS_DIR.DOWN else: if zd > last_zs.zg: ddir = Chan_ZS_DIR.UP else: ddir = None else: if seg.dir == Chan_SEG_DIR.UP: ddir = Chan_ZS_DIR.DOWN else: ddir = Chan_ZS_DIR.UP if (seg.dir == Chan_SEG_DIR.DOWN and ddir == Chan_ZS_DIR.DOWN) or (seg.dir == Chan_SEG_DIR.UP and ddir == Chan_ZS_DIR.UP): ddir = None if ddir and zg > zd: # New ZS zs = ChanZS(seg, len(zs_list), ddir) zs.set_zg(zg) zs.set_zd(zd) zs.set_gg(gg) zs.set_dd(dd) if last_zs: last_zs.set_next(zs) zs.set_pre(last_zs) zs_list.append(zs) if last_zs and last_zs.dir == zs.dir: zs_count += 1 else: zs_count = 1 last_zs = zs # Last ZS is not completed else: # Last ZS is not completed if last_zs and not last_zs.is_sure: if first_bi_out: # SEG is not in ZS if seg.is_sure: if ((seg.low > last_zs.zg and seg.high > last_zs.zg) or (seg.high < last_zs.zd and seg.low < last_zs.zd)): last_zs.set_end_klc(seg.pre.end_bi.end_klc, seg.sure_time, bi_out_count, seg.pre) bi_out_count = 0 #print(seg.start_bi.start_klc.start_time) first_bi_out = None # Last ZS is completed and look for new ZS if seg.next and seg.next.next: if seg.next.next.is_sure: zg = min(seg.high, seg.next.high, seg.next.next.high) zd = max(seg.low, seg.next.low, seg.next.next.low) gg = max(seg.high, seg.next.high, seg.next.next.high) dd = min(seg.low, seg.next.low, seg.next.next.low) ddir = None if last_zs: if zg < last_zs.zd: ddir = Chan_ZS_DIR.DOWN else: if zd > last_zs.zg: ddir = Chan_ZS_DIR.UP else: ddir = None else: if seg.dir == Chan_SEG_DIR.UP: ddir = Chan_ZS_DIR.DOWN else: ddir = Chan_ZS_DIR.UP if (seg.dir == Chan_SEG_DIR.DOWN and ddir == Chan_ZS_DIR.DOWN) or (seg.dir == Chan_SEG_DIR.UP and ddir == Chan_ZS_DIR.UP): ddir = None if ddir and zg > zd: # New ZS zs = ChanZS(seg, len(zs_list), ddir) zs.set_zg(zg) zs.set_zd(zd) zs.set_gg(gg) zs.set_dd(dd) last_zs.set_next(zs) zs.set_pre(last_zs) zs_list.append(zs) if last_zs and last_zs.dir == zs.dir: zs_count += 1 else: zs_count = 1 last_zs = zs # Last SEG is in ZS else: # SEG is inside ZS if seg.end_bi: for index in range(seg.start_bi.index, seg.end_bi.index+1): bi = bi_list[index] if (bi.high >= last_zs.zd and bi.high <= last_zs.zg) or (bi.low >= last_zs.zd and bi.low <= last_zs.zg) or (bi.high >= last_zs.zg and bi.low <= last_zs.zd): in_again = True last_zs.set_bi_out(None, None) last_zs.set_last_bi_in(None) last_zs.set_end_seg(None) first_bi_out = None #print("Bi in again 3", bi.start_klc.start_time) if in_again and (bi.low > last_zs.zg or bi.high < last_zs.zd): last_zs.set_bi_out(bi, seg) last_zs.set_last_bi_in(bi_list[index - 1]) #last_zs.set_end_seg(seg.next.next) bi_out_count += 1 first_bi_out = bi if (bi.dir == Chan_BI_DIR.UP and seg.dir == Chan_SEG_DIR.DOWN) or (bi.dir == Chan_BI_DIR.DOWN and seg.dir == Chan_SEG_DIR.UP): bsp = ChanBSP(first_bi_out, len(bsp_list), Chan_BSP_TYPE.T3, Chan_BSP_DIR.BUY if first_bi_out.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, first_bi_out.sure_time, zs_count, zs, seg) bsp_list.append(bsp) #print("First bi out 3", first_bi_out.start_klc.start_time) in_again = False """" if first_bi_out: if seg.dir == Chan_SEG_DIR.UP and bi.dir == Chan_BI_DIR.UP: #print(bi.start_klc.start_time, bi.high, seg.high) if bi.high == seg.high: bsp = ChanBSP(bi, len(bsp_list), Chan_BSP_TYPE.T3E, Chan_BSP_DIR.SELL if bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.BUY, bi.sure_time, zs_count, zs, seg) bsp_list.append(bsp) else: if seg.dir == Chan_SEG_DIR.DOWN and bi.dir == Chan_BI_DIR.DOWN: if bi.low == seg.low: bsp = ChanBSP(bi, len(bsp_list), Chan_BSP_TYPE.T3E, Chan_BSP_DIR.BUY if bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, bi.sure_time, zs_count, zs, seg) bsp_list.append(bsp) """ else: # SEG in ZS and not out and find first bi out if seg.end_bi: for index in range(seg.start_bi.index, seg.end_bi.index+1): bi = bi_list[index] if (bi.high >= last_zs.zd and bi.high <= last_zs.zg) or (bi.low >= last_zs.zd and bi.low <= last_zs.zg) or (bi.high >= last_zs.zg and bi.low <= last_zs.zd): in_again = True last_zs.set_bi_out(None, None) last_zs.set_last_bi_in(None) last_zs.set_end_seg(None) first_bi_out = None #print("Bi in again 4", bi.start_klc.start_time) if in_again and (bi.low > last_zs.zg or bi.high < last_zs.zd): last_zs.set_bi_out(bi, seg) last_zs.set_last_bi_in(bi_list[index - 1]) #last_zs.set_end_seg(seg.next.next) bi_out_count += 1 first_bi_out = bi if (bi.dir == Chan_BI_DIR.UP and seg.dir == Chan_SEG_DIR.DOWN) or (bi.dir == Chan_BI_DIR.DOWN and seg.dir == Chan_SEG_DIR.UP): bsp = ChanBSP(first_bi_out, len(bsp_list), Chan_BSP_TYPE.T3, Chan_BSP_DIR.BUY if first_bi_out.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, first_bi_out.sure_time, zs_count, zs, seg) bsp_list.append(bsp) #print("First bi out 4", first_bi_out.start_klc.start_time) in_again = False if first_bi_out: if seg.dir == Chan_SEG_DIR.UP and bi.dir == Chan_BI_DIR.UP: #print(bi.start_klc.start_time, bi.high, seg.high) if bi.high == seg.high: bsp = ChanBSP(bi, len(bsp_list), Chan_BSP_TYPE.T3E, Chan_BSP_DIR.SELL if bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.BUY, bi.sure_time, zs_count, zs, seg) bsp_list.append(bsp) else: if seg.dir == Chan_SEG_DIR.DOWN and bi.dir == Chan_BI_DIR.DOWN: if bi.low == seg.low: bsp = ChanBSP(bi, len(bsp_list), Chan_BSP_TYPE.T3E, Chan_BSP_DIR.BUY if bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, bi.sure_time, zs_count, zs, seg) bsp_list.append(bsp) #self.print_zs(zs_list) return zs_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) # 锤子线/上吊线 - 反转信号 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))