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 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 import matplotlib.pyplot as plt from matplotlib.dates import DateFormatter, date2num import matplotlib.patches as patches from technical.util import resample_to_interval from decimal import Decimal class ChanLun(): timeframes = ["5m", "15m", "30m", "60m", "4h"] times = { "5m": 5, "15m": 15, "30m": 30, "60m": 60, "4h": 240 } time5 = 5 time15 = 15 time30 = 30 time60 = 60 time4h = 240 def create_all_data(self, dataframe, ticker_indicator): all_data = dict() all_data['1m'] = dataframe for timeframe in self.timeframes: df = resample_to_interval(dataframe, ticker_indicator*self.times[timeframe]) all_data[timeframe] = df return all_data def print_zs(self, zs_list): for zs in zs_list: if zs.end_klc: print(zs.start_klc.start_time, zs.end_klc.end_time, zs.sure_time, zs.zg, zs.zd, zs.bi_out_count) else: print(zs.start_klc.start_time, zs.zg, zs.zd, zs.bi_out_count) def print_seg(self, seg_list): for seg in seg_list: if seg.is_sure: print(seg.start_bi.start_time, seg.end_bi.end_time, seg.dir, seg.sure_time, "SEG") else: print(seg.start_bi.start_time, seg.dir, "SEG") def print_bsp_list(self, bsp_list): for bsp in bsp_list: if bsp.is_sure: print(bsp.klc.start_time, bsp.type, bsp.dir, bsp.sure_time, bsp.zs_count, len(bsp.zs.bi_out_list), bsp.dir, bsp.seg.dir, bsp.bi.dir) else: print(bsp.klc.start_time, bsp.type, bsp.dir, bsp.zs_count, len(bsp.zs.bi_out_list), bsp.dir, bsp.seg.dir, bsp.bi.dir) def print_bi(self, bi_list): for bi in bi_list: if bi.end_klc: if bi.sure_time: print(bi.start_klc.end_time, bi.end_klc.end_time, bi.dir) else: print(bi.start_klc.end_time, bi.end_klc.end_time, bi.dir) else: print(bi.start_klc.end_time, bi.dir, bi.is_sure) def check_fx(self, klc): if klc.pre and klc.next: if klc.high > klc.pre.high and klc.high > klc.next.high: klc.set_fx(Chan_FX_TYPE.TOP) #print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time,klc.fx, "TOP") return Chan_FX_TYPE.TOP if klc.pre and klc.next: if klc.low < klc.pre.low and klc.low < klc.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.fx, "BOTTOM") return Chan_FX_TYPE.BOTTOM return Chan_FX_TYPE.UNKNOWN def get_macd(self, df): fast = 8 slow = 16 period = 6 macd = ta.MACD(df, fastperiod=fast, slowperiod=slow, signalperiod=period) df['macd'] = macd['macd'] df['macdsignal'] = macd['macdsignal'] df['macdhist'] = macd['macdhist'] return df def plot_dataframe(self, dataframe): klc_list = self.get_klc_list(dataframe) bi_list= self.cal_bi_list(klc_list) seg_list = self.get_seg_list(bi_list) zs_list = self.calculate_zs(bi_list, seg_list) #bi_macd_div_list = self.get_bi_macd_div_list(bi_list, dataframe) #seg_macd_div_list = self.get_seg_macd_div_list(seg_list, dataframe) #buy_sell_points = self.identify_buy_sell_points(bi_list, seg_list, zs_list, dataframe) #divergence_points = self.identify_macd_divergence(dataframe, bi_list) #self.print_bi(bi_list) #self.print_seg(seg_list) #self.print_zs(zs_list) #self.print_bsp_list(bsp_list) #self.plot(dataframe, bi_list, seg_list, zs_list, buy_sell_points, divergence_points) #return plt.gcf() def print_data(self, dataframe): klc_list = self.get_klc_list(dataframe) bi_list = self.cal_bi_list(klc_list) seg_list = self.get_seg_list(bi_list) bsp_list, zs_list = self.calculate_zs(bi_list, seg_list) bi_macd_div_list = self.get_bi_macd_div_list(bi_list, dataframe) seg_macd_div_list = self.get_seg_macd_div_list(seg_list, dataframe) def resample_bsp_list(self, bsp_list, dataframe): bsp_index = 0 resampled_bsp_list = [] if len(bsp_list) > 0: for index in range(0, len(dataframe)): if bsp_index == len(bsp_list): bsp_index = len(bsp_list) - 1 bsp = bsp_list[bsp_index] if dataframe['date'][index].strftime('%Y-%m-%d %H:%M:%S') == bsp.klc.end_time: if bsp.type == Chan_BSP_TYPE.T3E or bsp.type == Chan_BSP_TYPE.T3: if bsp.dir == Chan_BSP_DIR.BUY: resampled_bsp_list.append("-30") #print(bsp.klc.end_time, bsp.dir, bsp.seg.dir, "BUY") else: if bsp.dir == Chan_BSP_DIR.SELL: resampled_bsp_list.append("30") #print(bsp.klc.end_time, bsp.dir, bsp.seg.dir, "SELL") else: resampled_bsp_list.append("00") #print(bsp.klc.end_time, bsp.dir, bsp.seg.dir, "00") bsp_index += 1 else: resampled_bsp_list.append("00") else: for index in range(0, len(dataframe)): resampled_bsp_list.append("00") return resampled_bsp_list def cal_klu_state(self, dataframe): klc_list = self.get_klc_list(dataframe) bi_list = self.cal_bi_list(klc_list) klc_index = 0 state_list = [] for index in range(0, len(dataframe)): klc = klc_list[klc_index] if klc.end_klu and klc.end_klu.idx == index: state_list.append(klc.state) klc_index += 1 else: state_list.append("00") return state_list def get_bi_list(self, dataframe): bi_list = self.cal_bi_list(self.get_klc_list(dataframe)) return bi_list def calculate_zs(self, bi_list, seg_list): return self.get_zs_list(bi_list, seg_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) 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) #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) 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 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") else: last_down_sbi = ChanSBI(bi, len(down_sbi_list), bi.dir) down_sbi_list.append(last_down_sbi) #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 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) 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) #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) #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 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 #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) #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 #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) 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) 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) return seg_list def get_bi_zs_list(self, bi_list): """识别笔中枢列表 与线段中枢不同，笔中枢是由连续的同向笔构成，是更细粒度的中枢结构 Args: bi_list: 笔列表 Returns: bi_zs_list: 笔中枢列表 """ bi_zs_list = [] if len(bi_list) < 3: # 至少需要3个笔才能形成中枢 print("笔数量不足，无法形成中枢") return bi_zs_list last_zs = None first_bi_out = None in_again = False bi_out_count = 0 # 遍历所有笔，识别中枢 for i in range(2, len(bi_list)): # 确保当前笔和前两个笔都是完成的 if not bi_list[i].end_klc or not bi_list[i-1].end_klc or not bi_list[i-2].end_klc: continue current_bi = bi_list[i] prev_bi = bi_list[i-1] prev_prev_bi = bi_list[i-2] # 如果没有中枢或上一个中枢已完成 if len(bi_zs_list) == 0 or (last_zs and last_zs.is_sure): # 检查是否是三个连续同向笔 if (current_bi.dir == prev_bi.dir == prev_prev_bi.dir): # 创建潜在中枢 if current_bi.dir == Chan_BI_DIR.UP: # 向上的三笔区间定义中枢 # 中枢的上沿：取三个笔的终点的最小值 # 中枢的下沿：取三个笔的起点的最大值 zd = max(prev_prev_bi.start_klc.low, prev_bi.start_klc.low, current_bi.start_klc.low) zg = min(prev_prev_bi.end_klc.high, prev_bi.end_klc.high, current_bi.end_klc.high) # 确保中枢有效(上沿大于下沿) if zg > zd: print(f"发现向上笔中枢: 起始时间={prev_prev_bi.start_klc.start_time}, ZG={zg}, ZD={zd}") zs = ChanZS(prev_prev_bi.start_klc, zg, zd) zs.start_bi = prev_prev_bi zs.start_idx = i-2 zs.end_bi = current_bi zs.end_idx = i zs.end_klc = current_bi.end_klc zs.type = "BI_ZS" zs.direction = Chan_ZS_DIR.UP zs.sure_time = None # 中枢尚未确认完成 bi_zs_list.append(zs) last_zs = zs else: # 向下的三笔区间定义中枢 # 中枢的上沿：取三个笔的起点的最小值 # 中枢的下沿：取三个笔的终点的最大值 zg = min(prev_prev_bi.start_klc.high, prev_bi.start_klc.high, current_bi.start_klc.high) zd = max(prev_prev_bi.end_klc.low, prev_bi.end_klc.low, current_bi.end_klc.low) # 确保中枢有效(上沿大于下沿) if zg > zd: print(f"发现向下笔中枢: 起始时间={prev_prev_bi.start_klc.start_time}, ZG={zg}, ZD={zd}") zs = ChanZS(prev_prev_bi.start_klc, zg, zd) zs.start_bi = prev_prev_bi zs.start_idx = i-2 zs.end_bi = current_bi zs.end_idx = i zs.end_klc = current_bi.end_klc zs.type = "BI_ZS" zs.direction = Chan_ZS_DIR.DOWN zs.sure_time = None # 中枢尚未确认完成 bi_zs_list.append(zs) last_zs = zs # 处理已有的未完成中枢 elif last_zs and not last_zs.is_sure: # 当前笔与中枢最后一笔方向相同，可能延伸中枢 if current_bi.dir == prev_bi.dir: if last_zs.direction == Chan_ZS_DIR.UP and current_bi.dir == Chan_BI_DIR.UP: # 检查是否仍在中枢内：向上时终点高价在中枢区间内 if current_bi.end_klc.high >= last_zs.zd and current_bi.end_klc.high <= last_zs.zg: print(f"延伸向上笔中枢: 终点时间={current_bi.end_klc.end_time}") # 延伸中枢 last_zs.end_klc = current_bi.end_klc last_zs.end_bi = current_bi last_zs.end_idx = i else: # 笔离开中枢，记录第一个离开的笔 if not first_bi_out: first_bi_out = current_bi bi_out_count += 1 print(f"笔离开向上中枢: 时间={current_bi.end_klc.end_time}, 价格={current_bi.end_klc.high}, 中枢上沿={last_zs.zg}") else: if not in_again: # 第二次离开，确认中枢完成 print(f"确认向上笔中枢完成: 时间={current_bi.end_klc.end_time}") last_zs.is_sure = True last_zs.sure_bi = current_bi last_zs.sure_time = current_bi.end_klc.end_time elif last_zs.direction == Chan_ZS_DIR.DOWN and current_bi.dir == Chan_BI_DIR.DOWN: # 检查是否仍在中枢内：向下时终点低价在中枢区间内 if current_bi.end_klc.low <= last_zs.zg and current_bi.end_klc.low >= last_zs.zd: print(f"延伸向下笔中枢: 终点时间={current_bi.end_klc.end_time}") # 延伸中枢 last_zs.end_klc = current_bi.end_klc last_zs.end_bi = current_bi last_zs.end_idx = i else: # 笔离开中枢，记录第一个离开的笔 if not first_bi_out: first_bi_out = current_bi bi_out_count += 1 print(f"笔离开向下中枢: 时间={current_bi.end_klc.end_time}, 价格={current_bi.end_klc.low}, 中枢下沿={last_zs.zd}") else: if not in_again: # 第二次离开，确认中枢完成 print(f"确认向下笔中枢完成: 时间={current_bi.end_klc.end_time}") last_zs.is_sure = True last_zs.sure_bi = current_bi last_zs.sure_time = current_bi.end_klc.end_time # 方向改变，判断是否破坏中枢 else: # 方向改变可能导致重新进入中枢或破坏中枢 # 向上中枢被向下笔破坏：低点低于中枢下沿 # 向下中枢被向上笔破坏：高点高于中枢上沿 if (last_zs.direction == Chan_ZS_DIR.UP and current_bi.end_klc.low < last_zs.zd) or \ (last_zs.direction == Chan_ZS_DIR.DOWN and current_bi.end_klc.high > last_zs.zg): # 破坏中枢 print(f"笔中枢被破坏: 方向={current_bi.dir}, 时间={current_bi.end_klc.end_time}") last_zs.is_sure = True last_zs.sure_bi = current_bi last_zs.sure_time = current_bi.end_klc.end_time elif first_bi_out: # 重新进入中枢 print(f"笔重新进入中枢: 时间={current_bi.end_klc.end_time}") in_again = True first_bi_out = None # 延伸中枢 last_zs.end_klc = current_bi.end_klc last_zs.end_bi = current_bi last_zs.end_idx = i # 打印识别结果 print(f"笔中枢识别完成，共找到 {len(bi_zs_list)} 个笔中枢") return bi_zs_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) # Do nothing if fx == Chan_FX_TYPE.UNKNOWN: klc.set_fx(Chan_FX_TYPE.UNKNOWN) if len(bi_list) > 0: bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) 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]) #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, "一类买卖点Sell 1") 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, last_bottom.start_time, klc.fx, "中枢买卖点Sell 1") klc.set_fx(Chan_FX_TYPE.UNKNOWN) 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: pre_last_bi.update_bi(klc) bi_list.remove(last_bi) pre_last_bi.set_next(None) last_top.set_fx(Chan_FX_TYPE.UNKNOWN) last_top = klc last_bottom = pre_last_bi.start_klc #print(klc.start_time, last_bi.start_klc.start_time, "New TOP Found reset last bi") klc.set_state("10") 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, last_bottom.start_time, klc.fx, "中枢买卖点Sell 2") # 满足结合律 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 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") # 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 bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) 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.UNKNOWN) else: # First temp top and last bottom confirmed last_top = klc # 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) #print(klc.start_time, 'Create first top') #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]) #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, "一类买卖点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.UNKNOWN) #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: pre_last_bi.update_bi(klc) bi_list.remove(last_bi) pre_last_bi.set_next(None) last_bottom.set_fx(Chan_FX_TYPE.UNKNOWN) last_bottom = klc last_top = pre_last_bi.start_klc #print(klc.start_time, last_bi.start_klc.start_time, "New BOTTOM Found reset last bi") klc.set_state("-10") 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, "中枢买卖点Buy 2") # 满足结合律的分型 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 klc.set_state('-30') 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 bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) 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.UNKNOWN) bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) else: # First temp bottom and last top confirmed last_bottom = klc bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-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) bi_list.append(bi) bi_list[-1].add_klc(klc) klc.set_bi(bi_list[-1]) #print(klc.start_time, 'Create first bottom') #print(klc.time, klc.fx, klc.state) 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) return bi_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) ddir = Chan_ZS_DIR.UP 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) 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(last_zs.last_bi_in.end_klc, seg.sure_time, bi_out_count, seg) 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) 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) 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_bi_macd_hist_list(self, bi_list, dataframe): bi_macd_hist_list = [] for bi in bi_list: start_index = bi.start_klc.start_klu.index if bi.end_klc: end_index = bi.end_klc.end_klu.index else: end_index = len(dataframe) - 1 total_macd_hist = 0 for index in range(start_index, end_index+1): macd_hist = dataframe['macdhist'][index] if bi.dir == Chan_BI_DIR.UP and macd_hist > 0: total_macd_hist += macd_hist if bi.dir == Chan_BI_DIR.DOWN and macd_hist < 0: total_macd_hist -= macd_hist bi_macd_hist_list.append(abs(total_macd_hist)) bi.set_macd_hist(total_macd_hist) return bi_macd_hist_list, bi_list def get_seg_macd_hist_list(self, seg_list, dataframe): seg_macd_hist_list = [] for seg in seg_list: start_index = seg.start_bi.start_klc.start_klu.index if seg.end_bi: end_index = seg.end_bi.end_klc.end_klu.index else: end_index = len(dataframe) - 1 total_macd_hist = 0 for index in range(start_index, end_index+1): macd_hist = dataframe['macdhist'][index] if seg.dir == Chan_SEG_DIR.UP and macd_hist > 0: total_macd_hist += macd_hist if seg.dir == Chan_SEG_DIR.DOWN and macd_hist < 0: total_macd_hist -= macd_hist seg_macd_hist_list.append(abs(total_macd_hist)) seg.set_macd_hist(total_macd_hist) return seg_macd_hist_list, seg_list def get_bi_macd_div_list(self, bi_list, dataframe): bi_macd_div_list = [] bi_macd_hist_list, bi_list = self.get_bi_macd_hist_list(bi_list, dataframe) for index in range(2, len(bi_list)): if bi_macd_hist_list[index-2] == 0: bi_macd_div = 0.0 if index > 3 and bi_macd_hist_list[index-4] > 0.0: bi_macd_div = bi_macd_hist_list[index]/bi_macd_hist_list[index-4] else: bi_macd_div = bi_macd_hist_list[index]/bi_macd_hist_list[index-2] if bi_macd_div < 0.01: if index > 3 and bi_macd_hist_list[index-4] > 0.0: bi_macd_div = bi_macd_hist_list[index]/bi_macd_hist_list[index-4] bi_macd_div = self.get_decimal(bi_macd_div) bi_macd_div_list.append(bi_macd_div) bi_list[index].set_macd_div(bi_macd_div) #print(bi_list[index].start_klc.start_time, self.get_decimal(bi_macd_hist_list[index]), self.get_decimal(bi_macd_hist_list[index - 1]), self.get_decimal(bi_macd_div)) return bi_macd_div_list, bi_list def get_seg_macd_div_list(self, seg_list, dataframe): seg_macd_div_list = [] seg_macd_hist_list, seg_list = self.get_seg_macd_hist_list(seg_list, dataframe) for index in range(2, len(seg_list)): if seg_macd_hist_list[index-2] == 0: seg_macd_div = 0.0 if index > 3 and seg_macd_hist_list[index-4] > 0.0: seg_macd_div = seg_macd_hist_list[index]/seg_macd_hist_list[index - 4] else: seg_macd_div = seg_macd_hist_list[index]/seg_macd_hist_list[index - 2] if seg_macd_div < 0.01: if index > 3 and seg_macd_hist_list[index-4] > 0.0: seg_macd_div = seg_macd_hist_list[index]/seg_macd_hist_list[index - 4] seg_macd_div = self.get_decimal(seg_macd_div) seg_macd_div_list.append(seg_macd_div) seg_list[index].set_macd_div(seg_macd_div) #print(seg_list[index].start_bi.start_klc.start_time, self.get_decimal(seg_macd_hist_list[index]), self.get_decimal(seg_macd_hist_list[index - 1]), self.get_decimal(seg_macd_div)) return seg_macd_div_list, seg_list def get_macd_div_list(self, dataframe): bi_list = self.get_bi_list(dataframe) seg_list = self.get_seg_list(bi_list) bi_macd_div_list, bi_list = self.get_bi_macd_div_list(bi_list, dataframe) seg_macd_div_list, seg_list = self.get_seg_macd_div_list(seg_list, dataframe) return bi_macd_div_list, bi_list, seg_macd_div_list, seg_list def get_decimal(self, value): return Decimal("{:.2f}".format(value)) def get_klc_list(self, dataframe): klu_list = self.get_klu_list(dataframe) klc_list = [] last_klu = None for klu in klu_list: if len(klc_list) > 0: last_klc = klc_list[-1] 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 return klc_list def get_klu_list(self, dataframe): return self.get_kl_data(dataframe) def copy_klu_to_klc(self, klu_list): klc_list = [] for klu in klu_list: if len(klc_list) > 0: last_klc = klc_list[-1] 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.set_end_klu(klu) klc_list.append(klc) last_klc.set_next(klc) klc.set_pre(last_klc) else: klc = ChanKLC(klu, 0) klc_list.append(klc) klc.set_end_klu(klu) return klc_list def get_kl_data(self, dataframe:DataFrame): fields = "time,open,high,low,close,volume" klu_list = [] 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) klu.set_idx(i) klu_list.append(klu) klu.macd = item['macd'] klu.signal = item['macdsignal'] klu.macdhist = item['macdhist'] klu.ma5 = item['ma5'] klu.ma10 = item['ma10'] klu.ma30 = item['ma30'] klu.ma250 = item['ma250'] klu.rsi = item['rsi'] return klu_list def get_bsp_list1(self, big_df): big_bi_list = self.get_bi_list(big_df) big_seg_list = self.get_seg_list(big_bi_list) big_zs_list = self.get_zs_list(big_bi_list, big_seg_list) big_bi_macd_div_list = self.get_bi_macd_div_list(big_bi_list, big_df) big_seg_macd_div_list = self.get_seg_macd_div_list(big_seg_list, big_df) big_bi_macd_hist_list = self.get_bi_macd_hist_list(big_bi_list, big_df) big_seg_macd_hist_list = self.get_seg_macd_hist_list(big_seg_list, big_df) for index in range(0, len(big_seg_list)): big_seg = big_seg_list[index] if big_seg.end_bi: if big_seg.dir == Chan_SEG_DIR.UP: if big_seg.end_bi.index - big_seg.start_bi.index > 1: max_high = big_seg.start_bi.high for bi_index in range(big_seg.start_bi.index + 2, big_seg.end_bi.index + 1): bi = big_bi_list[bi_index] #print("MACD DIV: ", big_bi_macd_hist_list[index]/big_bi_macd_hist_list[index - 2]) if bi.is_sure and bi.dir == Chan_BI_DIR.UP: if bi.high > max_high: max_high = bi.high if big_bi_macd_hist_list[bi_index - 2] > 0.0: bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 2] if bi_macd_div < 0.01 and len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0: bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4] else: if len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0: bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4] else: bi_macd_div = 0.0 macd_index = bi.end_klc.end_klu.index if bi_macd_div < 0.8 and bi_macd_div > 0.01 and big_df['macd'][macd_index] > 0 and big_df['macdsignal'][macd_index] > 0: print("UP SEG Possible BSP:", bi.start_klc.end_time, bi_macd_div) else: if big_seg.end_bi.index - big_seg.start_bi.index > 1: max_low = big_seg.start_bi.low for bi_index in range(big_seg.start_bi.index + 2, big_seg.end_bi.index + 1): bi = big_bi_list[bi_index] if bi.is_sure and bi.dir == Chan_BI_DIR.DOWN: #print("DOWN: ", max_low, bi.low) if bi.low < max_low: max_low = bi.low if big_bi_macd_hist_list[bi_index - 2] > 0.0: bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 2] if bi_macd_div < 0.01 and len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0: bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4] else: if len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0: bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4] else: bi_macd_div = 0.0 macd_index = bi.end_klc.end_klu.index if bi_macd_div < 0.8 and bi_macd_div > 0.01 and big_df['macd'][macd_index] < 0 and big_df['macdsignal'][macd_index] < 0: print("DOWN SEG Possible BSP:", bi.start_klc.end_time, bi_macd_div) else: print("Not completed segment.", len(big_bi_list) - big_seg.start_bi.index, big_seg.dir) if big_seg.dir == Chan_SEG_DIR.UP: if len(big_bi_list) - big_seg.start_bi.index > 1: max_high = big_seg.start_bi.high for bi_index in range(big_seg.start_bi.index + 2, len(big_bi_list)): bi = big_bi_list[bi_index] #print("MACD DIV: ", big_bi_macd_hist_list[index]/big_bi_macd_hist_list[index - 2]) if bi.is_sure and bi.dir == Chan_BI_DIR.UP: if bi.high > max_high: max_high = bi.high if big_bi_macd_hist_list[bi_index - 2] > 0.0: bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 2] if bi_macd_div < 0.01 and len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0: bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4] else: if len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0: bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4] else: bi_macd_div = 0.0 macd_index = len(big_df) - 1 if bi_macd_div < 0.8 and bi_macd_div > 0.01 and big_df['macd'][macd_index] > 0 and big_df['macdsignal'][macd_index] > 0: print("UP SEG Possible BSP:", bi.start_klc.end_time, bi_macd_div) else: if len(big_bi_list) - big_seg.start_bi.index > 1: max_low = big_seg.start_bi.low for bi_index in range(big_seg.start_bi.index + 2, len(big_bi_list)): bi = big_bi_list[bi_index] #print("MACD DIV: ", big_bi_macd_hist_list[index]/big_bi_macd_hist_list[index - 2]) if bi.is_sure and bi.dir == Chan_BI_DIR.DOWN: if bi.low < max_low: max_low = bi.low if big_bi_macd_hist_list[bi_index - 2] > 0.0: bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 2] if bi_macd_div < 0.01 and len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0: bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4] else: if len(big_bi_list) - big_seg.start_bi.index > 4 and big_bi_macd_hist_list[bi_index - 4] > 0.0: bi_macd_div = big_bi_macd_hist_list[bi_index]/big_bi_macd_hist_list[bi_index - 4] else: bi_macd_div = 0.0 macd_index = len(big_df) - 1 if bi_macd_div < 0.8 and bi_macd_div > 0.01 and big_df['macd'][macd_index] < 0 and big_df['macdsignal'][macd_index] < 0: print("DOWN SEG Possible BSP:", bi.start_klc.end_time, bi_macd_div) def get_bsp_list(self, big_df): big_bi_list = self.get_bi_list(big_df) big_seg_list = self.get_seg_list(big_bi_list) big_zs_list = self.calculate_zs(big_bi_list, big_seg_list) big_bsp_list = self.find_third_bsp(big_zs_list) big_bi_macd_div_list, big_bi_list = self.get_bi_macd_div_list(big_bi_list, big_df) for index in range(0, len(big_bsp_list)-1): bsp = big_bsp_list[index] last_zs = bsp.zs bsp_next = big_bsp_list[index + 1] if bsp.zs.index != bsp_next.zs.index: end_index = bsp.seg.end_bi.index else: end_index = bsp_next.bi.index # Down trend if bsp.bi.dir == Chan_BI_DIR.UP: last_up_bi = bsp.bi last_down_bi = bsp.bi.pre for bi_index in range(bsp.bi.index + 1, end_index + 1): bi = big_bi_list[bi_index] if bi.is_sure: if bi.dir == Chan_BI_DIR.DOWN: if bi.low < last_down_bi.low: print("背驰点1，第一类买点", bi.start_klc.end_time, bi.macd_div) else: if bi.macd_div > 1.5: print("快速下跌，等待背驰:", bi.start_klc.end_time, bi.macd_div) last_down_bi = bi else: if last_up_bi: if (bi.high > last_up_bi.high and bi.macd_div > 1.2) or bi.high > last_zs.zd: print("回中枢或者快速拉升，止损点:", bi.start_klc.end_time, bi.macd_div) last_up_bi = bi # Up trend else: last_down_bi = bsp.bi last_up_bi = bsp.bi.pre for bi_index in range(bsp.bi.index + 1, end_index + 1): bi = big_bi_list[bi_index] if bi.is_sure: if bi.dir == Chan_BI_DIR.UP: if last_up_bi: if bi.high < last_up_bi.high: if bi.macd_div < 0.8 and bi.macd_div > 0.1: print("背驰点2，第一类卖点", bi.start_klc.end_time, bi.macd_div) else: if bi.macd_div > 1.5: print("快速上涨，等待背驰:", bi.start_klc.end_time, bi.macd_div) last_up_bi = bi else: if last_down_bi: if (bi.low < last_down_bi.low and bi.macd_div > 1.2) or bi.low < last_zs.zg: print("回中枢或者快速下跌，止损点:", bi.start_klc.end_time, bi.macd_div) last_down_bi = bi bsp_bi = big_bi_list[-1] last_zs = big_zs_list[-1] # Down trend if bsp_bi.dir == Chan_BI_DIR.UP: last_down_bi = bsp_bi.pre last_up_bi = bsp_bi for bi_index in range(bsp_bi.index + 1, bsp.seg.end_bi.index + 1): bi = big_bi_list[bi_index] if bi.is_sure: if bi.dir == Chan_BI_DIR.DOWN: if last_down_bi: if bi.low < last_down_bi.low: if bi.macd_div < 0.8 and bi.macd_div > 0.1: print("背驰点3，第一类买点", bi.start_klc.end_time, bi.macd_div) else: if bi.macd_div > 1.5: print("快速下跌，等待背驰:", bi.start_klc.end_time, bi.macd_div) last_down_bi = bi else: if last_up_bi: if (bi.high > last_up_bi.high and bi.macd_div > 1.2) or bi.high > last_zs.zd: print("回中枢或者快速拉升，止损点:", bi.start_klc.end_time, bi.macd_div) last_up_bi = bi # Up trend else: last_down_bi = bsp_bi.pre last_up_bi = bsp_bi for bi_index in range(bsp_bi.index + 1, bsp.seg.end_bi.index + 1): bi = big_bi_list[bi_index] if bi.is_sure: if bi.dir == Chan_BI_DIR.UP: if last_up_bi: if bi.high < last_up_bi.high: if bi.macd_div < 0.8 and bi.macd_div > 0.1: print("背驰点4，第一类卖点", bi.start_klc.end_time, bi.macd_div) else: if bi.macd_div > 1.5: print("快速上涨，等待背驰:", bi.start_klc.end_time, bi.macd_div) last_up_bi = bi else: if last_down_bi: if (bi.low < last_down_bi.low and bi.macd_div > 1.2) or bi.low < last_zs.zg: print("回中枢或者快速下跌，止损点:", bi.start_klc.end_time, bi.macd_div) last_down_bi = bi return big_bsp_list def cal_qjt(self, small_df, big_df): big_bi_list = self.get_bi_list(big_df) big_seg_list = self.get_seg_list(big_bi_list) big_zs_list = self.calculate_zs(big_bi_list, big_seg_list) small_bi_list = self.get_bi_list(small_df) small_seg_list = self.get_seg_list(small_bi_list) small_zs_list = self.calculate_zs(small_bi_list, small_seg_list) big_bsp_list = self.find_third_bsp(big_zs_list) small_bsp_list = self.find_third_bsp(small_zs_list) #self.print_bsp_list(big_bsp_list) self.print_bsp_list(small_bsp_list) def get_seg_bsp_list(self, big_df): big_bi_list = self.get_bi_list(big_df) big_seg_list = self.get_seg_list(big_bi_list) big_bsp_list = [] seg = big_seg_list[-1] bi = big_bi_list[-1] if seg.dir == Chan_SEG_DIR.UP: if bi.dir == Chan_BI_DIR.UP: if bi.high > seg.high: if bi.macd_div < 0.8 and bi.macd_div > 0.1: bi_bsp = ChanBSP(bi, len(big_bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.BUY, bi.sure_time, 0, None, seg) big_bsp_list.append(bi_bsp) else: if bi.dir == Chan_BI_DIR.DOWN: if bi.low < seg.low: if bi.macd_div < 0.8 and bi.macd_div > 0.1: bi_bsp = ChanBSP(bi, len(big_bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.SELL, bi.sure_time, 0, None, seg) big_bsp_list.append(bi_bsp) print("Last SEG: ", seg.start_bi.start_klc.start_time) for bsp in big_bsp_list: if bsp.bi.end_klc: print(bsp.bi.end_klc.end_time, bsp.sure_time, bsp.dir, bsp.bi.macd_div) return big_bsp_list def get_bi_bsp_list(self, big_df): big_bi_list = self.get_bi_list(big_df) big_seg_list = self.get_seg_list(big_bi_list) big_bi_macd_div_list, big_bi_list = self.get_bi_macd_div_list(big_bi_list, big_df) big_seg_macd_div_list, big_seg_list = self.get_seg_macd_div_list(big_seg_list, big_df) bi_bsp_list = [] for index in range(0, len(big_seg_list)): big_seg = big_seg_list[index] if big_seg.end_bi: if big_seg.dir == Chan_SEG_DIR.UP: max_high = big_seg.high for bi_index in range(big_seg.start_bi.index, big_seg.end_bi.index+1): bi = big_bi_list[bi_index] if bi.dir == Chan_BI_DIR.DOWN and bi.macd_div < 0.8 and bi.macd_div > 0.1 and bi.high > max_high: max_high = bi.high bi_bsp = ChanBSP(bi, len(bi_bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.BUY, bi.sure_time, 0, None, big_seg) bi_bsp_list.append(bi_bsp) else: max_low = big_seg.low for bi_index in range(big_seg.start_bi.index, big_seg.end_bi.index+1): bi = big_bi_list[bi_index] if bi.dir == Chan_BI_DIR.UP and bi.macd_div < 0.8 and bi.macd_div > 0.1 and bi.low < max_low: max_low = bi.low bi_bsp = ChanBSP(bi, len(bi_bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.SELL, bi.sure_time, 0, None, big_seg) bi_bsp_list.append(bi_bsp) else: print("Not completed segment.", len(big_bi_list) - big_seg.start_bi.index, big_seg.dir) if big_seg.dir == Chan_SEG_DIR.UP: max_high = big_seg.high for bi_index in range(big_seg.start_bi.index, len(big_bi_list)): bi = big_bi_list[bi_index] if bi.end_klc and bi.dir == Chan_BI_DIR.DOWN and bi.macd_div < 0.8 and bi.macd_div > 0.1 and bi.high > max_high: max_high = bi.high bi_bsp = ChanBSP(bi, len(bi_bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.SELL, bi.sure_time, 0, None, big_seg) bi_bsp_list.append(bi_bsp) else: max_low = big_seg.low for bi_index in range(big_seg.start_bi.index, len(big_bi_list)): bi = big_bi_list[bi_index] if bi.end_klc and bi.dir == Chan_BI_DIR.UP and bi.macd_div < 0.8 and bi.macd_div > 0.1 and bi.low < max_low: max_low = bi.low bi_bsp = ChanBSP(bi, len(bi_bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.BUY, bi.sure_time, 0, None, big_seg) bi_bsp_list.append(bi_bsp) for bsp in bi_bsp_list: if bsp.bi.end_klc: print(bsp.bi.end_klc.end_time, bsp.sure_time, bsp.dir, bsp.seg.dir, bsp.bi.macd_div) return bi_bsp_list def find_third_bsp(self, zs_list): bsp_list = [] zs_count = 0 last_zs = None for zs in zs_list: if last_zs and last_zs.dir == zs.dir: zs_count += 1 else: zs_count = 1 if zs.is_sure and zs.end_klc: for index in range(0, len(zs.bi_out_list)): bi_out = zs.bi_out_list[index] bi_out_seg = zs.bi_out_seg_list[index] bsp = ChanBSP(bi_out, len(bsp_list), Chan_BSP_TYPE.T3, Chan_BSP_DIR.BUY if bi_out.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, bi_out.sure_time, zs_count, zs, bi_out_seg) bsp_list.append(bsp) elif len(zs.bi_out_list) > 0: for index in range(0, len(zs.bi_out_list)): bi_out = zs.bi_out_list[index] bi_out_seg = zs.bi_out_seg_list[index] bsp = ChanBSP(bi_out, len(bsp_list), Chan_BSP_TYPE.T3, Chan_BSP_DIR.BUY if bi_out.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, bi_out.sure_time, zs_count, zs, bi_out_seg) bsp_list.append(bsp) last_zs = zs return bsp_list def find_first_bsp(self, bi_list, seg_list, zs_list, dataframe): bsp_list = [] zs_count = 0 for index in range(1, len(zs_list)): zs = zs_list[index] pre_zs = zs_list[index - 1] if zs.is_sure: if pre_zs.dir == zs.dir: zs_count += 1 continue else: zs_count = 1 else: current_bi = bi_list[-1] current_seg = seg_list[-1] if zs.dir == pre_zs.dir and ((current_bi.dir == Chan_BI_DIR.UP and current_seg.dir == Chan_SEG_DIR.UP) or (current_bi.dir == Chan_BI_DIR.DOWN and current_seg.dir == Chan_SEG_DIR.DOWN)): if zs.bi_out and zs.bi_out.is_sure and bi_list[-1].is_sure: pre_start_index = pre_zs.end_seg.start_klc.end_klu.index pre_end_index = zs.start_klc.end_klu.index start_index = zs.bi_out_seg.start_bi.start_klc.start_klu.index end_index = current_bi.end_klc.end_klu.index pre_macd_area = self.cal_macd_area(dataframe, pre_start_index, pre_end_index, pre_zs.dir) macd_area = self.cal_macd_area(dataframe, start_index, end_index, zs.dir) print(zs.bi_out.start_klc.start_time, pre_macd_area, macd_area, zs_count) if pre_macd_area > macd_area: bsp = ChanBSP(current_bi, len(bsp_list), Chan_BSP_TYPE.T1, Chan_BSP_DIR.BUY if current_bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, current_bi.sure_time, zs.zs_count, zs, current_seg) bsp_list.append(bsp) return bsp_list def cal_macd_area(self, dataframe, start_idx, end_idx, zs_dir): """ 计算指定区间内的MACD面积 :param dataframe: K线数据 :param start_idx: 开始索引 :param end_idx: 结束索引 :param seg_dir: 线段方向(Chan_SEG_DIR.UP或Chan_SEG_DIR.DOWN) :return: MACD面积的绝对值 """ # 计算MACD指标 exp1 = dataframe['close'].ewm(span=12, adjust=False).mean() exp2 = dataframe['close'].ewm(span=26, adjust=False).mean() macd = exp1 - exp2 signal = macd.ewm(span=9, adjust=False).mean() histogram = macd - signal # 根据线段方向选择计算正面积还是负面积 if zs_dir == Chan_ZS_DIR.UP: # 上升线段计算正面积 area = histogram[start_idx:end_idx+1][histogram[start_idx:end_idx+1] > 0].sum() else: # 下降线段计算负面积 area = histogram[start_idx:end_idx+1][histogram[start_idx:end_idx+1] < 0].sum() return abs(area) def plot_dual(self, small_df, big_df): """ 绘制双周期K线图表，包括两个周期的笔、线段、中枢和买卖点 :param small_df: 小周期K线数据 :param big_df: 大周期K线数据 """ plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'Microsoft YaHei', 'WenQuanYi Micro Hei'] plt.rcParams['axes.unicode_minus'] = False # 创建图表和子图 fig = plt.figure(figsize=(15, 12)) # 大周期图表(上方60%) ax1 = plt.subplot2grid((10, 1), (0, 0), rowspan=4) # 小周期图表(中间40%) ax2 = plt.subplot2grid((10, 1), (4, 0), rowspan=4, sharex=ax1) # MACD图表(下方20%) ax3 = plt.subplot2grid((10, 1), (8, 0), rowspan=2, sharex=ax1) # 计算两个周期的缠论结构 big_klc = self.get_klc_list(big_df) big_bi = self.cal_bi_list(big_klc) big_seg = self.get_seg_list(big_bi) big_zs = self.calculate_zs(big_bi, big_seg) big_buy_sell_points = self.check_top_bottom(big_df, big_bi, big_seg, big_zs) big_bi_macd_div, big_bi = self.get_bi_macd_div_list(big_bi, big_df) big_seg_macd_div, big_seg = self.get_seg_macd_div_list(big_seg, big_df) small_klc = self.get_klc_list(small_df) small_bi = self.cal_bi_list(small_klc) small_seg = self.get_seg_list(small_bi) small_zs = self.calculate_zs(small_bi, small_seg) small_buy_sell_points = self.check_top_bottom(small_df, small_bi, small_seg, small_zs) small_bi_macd_div, small_bi = self.get_bi_macd_div_list(small_bi, small_df) small_seg_macd_div, small_seg = self.get_seg_macd_div_list(small_seg, small_df) # 绘制大周期K线 big_dates = pd.to_datetime(big_df['date']).dt.tz_localize(None) big_dates_num = [date2num(date) for date in big_dates] # 绘制大周期K线 for i in range(len(big_df)): color = 'red' if big_df['close'][i] > big_df['open'][i] else 'green' ax1.bar(big_dates_num[i], big_df['close'][i] - big_df['open'][i], bottom=big_df['open'][i], color=color, width=0.0005) ax1.plot([big_dates_num[i], big_dates_num[i]], [big_df['low'][i], big_df['high'][i]], color=color, linewidth=1.2) # 绘制大周期笔 for bi in big_bi: if bi.end_klc: start_time = pd.to_datetime(bi.start_klc.end_time) end_time = pd.to_datetime(bi.end_klc.end_time) color = 'blue' if bi.dir == Chan_BI_DIR.UP else 'purple' start_price = bi.start_klc.low if bi.dir == Chan_BI_DIR.UP else bi.start_klc.high end_price = bi.end_klc.high if bi.dir == Chan_BI_DIR.UP else bi.end_klc.low ax1.plot([date2num(start_time), date2num(end_time)], [start_price, end_price], color=color, linewidth=1.5) else: start_time = pd.to_datetime(bi.start_klc.end_time) end_time = pd.to_datetime(big_klc[-1].start_time) color = 'blue' if bi.dir == Chan_BI_DIR.UP else 'purple' start_price = bi.start_klc.low if bi.dir == Chan_BI_DIR.UP else bi.start_klc.high end_price = big_klc[-1].high if bi.dir == Chan_BI_DIR.UP else big_klc[-1].low ax1.plot([date2num(start_time), date2num(end_time)], [start_price, end_price], color=color, linewidth=0.5) # 绘制大周期线段 for seg in big_seg: if seg.end_bi: start_time = pd.to_datetime(seg.start_bi.start_klc.end_time) end_time = pd.to_datetime(seg.end_bi.end_klc.end_time) color = 'red' if seg.dir == Chan_SEG_DIR.UP else 'green' start_price = seg.start_bi.start_klc.low if seg.dir == Chan_SEG_DIR.UP else seg.start_bi.start_klc.high end_price = seg.end_bi.end_klc.high if seg.dir == Chan_SEG_DIR.UP else seg.end_bi.end_klc.low ax1.plot([date2num(start_time), date2num(end_time)], [start_price, end_price], color=color, linewidth=2.5) else: start_time = pd.to_datetime(seg.start_bi.start_klc.end_time) end_time = pd.to_datetime(big_klc[-1].start_time) color = 'red' if seg.dir == Chan_SEG_DIR.UP else 'green' start_price = seg.start_bi.start_klc.low if seg.dir == Chan_SEG_DIR.UP else seg.start_bi.start_klc.high end_price = big_klc[-1].high if seg.dir == Chan_SEG_DIR.UP else big_klc[-1].low ax1.plot([date2num(start_time), date2num(end_time)], [start_price, end_price], color=color, linewidth=1) # 绘制大周期中枢 for idx, zs in enumerate(big_zs): start_time = pd.to_datetime(zs.start_klc.end_time).tz_localize(None) color = ['orange', 'cyan', 'magenta', 'yellow', 'lime'][idx % 5] if zs.end_klc: end_time = pd.to_datetime(zs.end_klc.end_time).tz_localize(None) width = date2num(end_time) - date2num(start_time) rect = patches.Rectangle( (date2num(start_time), zs.zd), width, zs.zg - zs.zd, linewidth=1, edgecolor=color, facecolor=color, alpha=0.2 ) ax1.add_patch(rect) label_text = f"大中枢{idx+1}" else: end_time = pd.to_datetime(big_dates.iloc[-1]).tz_localize(None) width = date2num(end_time) - date2num(start_time) rect = patches.Rectangle( (date2num(start_time), zs.zd), width, zs.zg - zs.zd, linewidth=1.5, edgecolor=color, facecolor=color, alpha=0.1, linestyle='--' ) ax1.add_patch(rect) label_text = f"大中枢{idx+1}(未完成)" ax1.text( date2num(start_time) + width/2, zs.zd + (zs.zg - zs.zd)/2, label_text, ha='center', va='center', fontsize=9, color='black', bbox=dict(boxstyle="round,pad=0.2", fc=color, alpha=0.6) ) for index in range(0, len(big_bi_macd_div)): bi_macd_div = big_bi_macd_div[index] bi = big_bi[index + 2] if bi.end_klc: text_index = bi.end_klc.end_klu.index if bi.dir == Chan_BI_DIR.UP: ax1.text(big_dates_num[text_index], bi.end_klc.high+1, bi_macd_div, color='red', fontsize=10, alpha=0.6) else: ax1.text(big_dates_num[text_index], bi.end_klc.low-1, bi_macd_div, color='green', fontsize=10, alpha=0.6) for index in range(0, len(big_seg_macd_div)): seg_macd_div = big_seg_macd_div[index] seg = big_seg[index + 2] if seg.end_bi: text_index = seg.end_bi.end_klc.end_klu.index if seg.dir == Chan_SEG_DIR.UP: ax1.text(big_dates_num[text_index], seg.end_bi.end_klc.high+1, seg_macd_div, color='red', fontsize=14, alpha=0.6) else: ax1.text(big_dates_num[text_index], seg.end_bi.end_klc.low-1, seg_macd_div, color='green', fontsize=14, alpha=0.6) """ # 绘制大周期买卖点 marker_styles = { '第一类买点': {'marker': '^', 'color': 'red', 'size': 10}, '第一类卖点': {'marker': 'v', 'color': 'green', 'size': 10}, '2类买点': {'marker': '^', 'color': 'orange', 'size': 10}, '2类卖点': {'marker': 'v', 'color': 'cyan', 'size': 10}, '3类买点': {'marker': '^', 'color': 'purple', 'size': 10}, '3类卖点': {'marker': 'v', 'color': 'magenta', 'size': 10} } for idx, point in big_buy_sell_points.items(): if idx < 0 or idx >= len(big_df): continue style = marker_styles.get(point['type'], {'marker': 'o', 'color': 'black', 'size': 8}) ax1.plot(big_dates_num[idx], point['price'], style['marker'], color=style['color'], markersize=style['size']) ax1.annotate(point['type'], (big_dates_num[idx], point['price']), textcoords="offset points", xytext=(0, 10), ha='center', fontsize=8, bbox=dict(boxstyle="round,pad=0.2", fc=style['color'], alpha=0.5)) """ # 绘制小周期K线 small_dates = pd.to_datetime(small_df['date']).dt.tz_localize(None) small_dates_num = [date2num(date) for date in small_dates] for i in range(len(small_df)): color = 'red' if small_df['close'][i] > small_df['open'][i] else 'green' ax2.bar(small_dates_num[i], small_df['close'][i] - small_df['open'][i], bottom=small_df['open'][i], color=color, width=0.0002) ax2.plot([small_dates_num[i], small_dates_num[i]], [small_df['low'][i], small_df['high'][i]], color=color, linewidth=0.8) # 绘制小周期笔 for bi in small_bi: if bi.end_klc: start_time = pd.to_datetime(bi.start_klc.end_time) end_time = pd.to_datetime(bi.end_klc.end_time) color = 'blue' if bi.dir == Chan_BI_DIR.UP else 'purple' start_price = bi.start_klc.low if bi.dir == Chan_BI_DIR.UP else bi.start_klc.high end_price = bi.end_klc.high if bi.dir == Chan_BI_DIR.UP else bi.end_klc.low ax2.plot([date2num(start_time), date2num(end_time)], [start_price, end_price], color=color, linewidth=1.2) else: start_time = pd.to_datetime(bi.start_klc.end_time) end_time = pd.to_datetime(small_klc[-1].start_time) color = 'blue' if bi.dir == Chan_BI_DIR.UP else 'purple' start_price = bi.start_klc.low if bi.dir == Chan_BI_DIR.UP else bi.start_klc.high end_price = small_klc[-1].high if bi.dir == Chan_BI_DIR.UP else small_klc[-1].low ax2.plot([date2num(start_time), date2num(end_time)], [start_price, end_price], color=color, linewidth=0.6) # 绘制小周期线段 for seg in small_seg: if seg.end_bi: start_time = pd.to_datetime(seg.start_bi.start_klc.end_time) end_time = pd.to_datetime(seg.end_bi.end_klc.end_time) color = 'red' if seg.dir == Chan_SEG_DIR.UP else 'green' start_price = seg.start_bi.start_klc.low if seg.dir == Chan_SEG_DIR.UP else seg.start_bi.start_klc.high end_price = seg.end_bi.end_klc.high if seg.dir == Chan_SEG_DIR.UP else seg.end_bi.end_klc.low ax2.plot([date2num(start_time), date2num(end_time)], [start_price, end_price], color=color, linewidth=1.8) else: start_time = pd.to_datetime(seg.start_bi.start_klc.end_time) end_time = pd.to_datetime(small_klc[-1].start_time) color = 'red' if seg.dir == Chan_SEG_DIR.UP else 'green' start_price = seg.start_bi.start_klc.low if seg.dir == Chan_SEG_DIR.UP else seg.start_bi.start_klc.high end_price = small_klc[-1].high if seg.dir == Chan_SEG_DIR.UP else small_klc[-1].low ax2.plot([date2num(start_time), date2num(end_time)], [start_price, end_price], color=color, linewidth=0.9) # 绘制小周期中枢 for idx, zs in enumerate(small_zs): start_time = pd.to_datetime(zs.start_klc.end_time).tz_localize(None) color = ['orange', 'cyan', 'magenta', 'yellow', 'lime'][idx % 5] if zs.end_klc: end_time = pd.to_datetime(zs.end_klc.end_time).tz_localize(None) width = date2num(end_time) - date2num(start_time) rect = patches.Rectangle( (date2num(start_time), zs.zd), width, zs.zg - zs.zd, linewidth=0.8, edgecolor=color, facecolor=color, alpha=0.2 ) ax2.add_patch(rect) label_text = f"小中枢{idx+1}" else: end_time = pd.to_datetime(small_dates.iloc[-1]).tz_localize(None) width = date2num(end_time) - date2num(start_time) rect = patches.Rectangle( (date2num(start_time), zs.zd), width, zs.zg - zs.zd, linewidth=1, edgecolor=color, facecolor=color, alpha=0.1, linestyle='--' ) ax2.add_patch(rect) label_text = f"小中枢{idx+1}(未完成)" ax2.text( date2num(start_time) + width/2, zs.zd + (zs.zg - zs.zd)/2, label_text, ha='center', va='center', fontsize=8, color='black', bbox=dict(boxstyle="round,pad=0.2", fc=color, alpha=0.6) ) for index in range(0, len(small_bi_macd_div)): bi_macd_div = small_bi_macd_div[index] bi = small_bi[index + 2] if bi.end_klc: text_index = bi.end_klc.end_klu.index if bi.dir == Chan_BI_DIR.UP: ax2.text(small_dates_num[text_index], bi.end_klc.high+1, bi_macd_div, color='red', fontsize=10, alpha=0.6) else: ax2.text(small_dates_num[text_index], bi.end_klc.low-1, bi_macd_div, color='green', fontsize=10, alpha=0.6) for index in range(0, len(small_seg_macd_div)): seg_macd_div = small_seg_macd_div[index] seg = small_seg[index + 2] if seg.end_bi: text_index = seg.end_bi.end_klc.end_klu.index if seg.dir == Chan_SEG_DIR.UP: ax2.text(small_dates_num[text_index], seg.end_bi.end_klc.high+1, seg_macd_div, color='red', fontsize=14, alpha=0.8) else: ax2.text(small_dates_num[text_index], seg.end_bi.end_klc.low-1, seg_macd_div, color='green', fontsize=14, alpha=0.8) """ # 绘制小周期买卖点 for idx, point in small_buy_sell_points.items(): if idx < 0 or idx >= len(small_df): continue style = marker_styles.get(point['type'], {'marker': 'o', 'color': 'black', 'size': 6}) ax2.plot(small_dates_num[idx], point['price'], style['marker'], color=style['color'], markersize=style['size']) ax2.annotate(point['type'], (small_dates_num[idx], point['price']), textcoords="offset points", xytext=(0, 8), ha='center', fontsize=7, bbox=dict(boxstyle="round,pad=0.2", fc=style['color'], alpha=0.5)) """ # 绘制MACD(使用小周期数据) exp1 = small_df['close'].ewm(span=8, adjust=False).mean() exp2 = small_df['close'].ewm(span=16, adjust=False).mean() macd = exp1 - exp2 signal = macd.ewm(span=6, adjust=False).mean() histogram = macd - signal ax3.bar(small_dates_num, histogram, width=0.0002, color=['red' if h > 0 else 'green' for h in histogram]) ax3.plot(small_dates_num, macd, color='blue', linewidth=0.8, label='MACD') ax3.plot(small_dates_num, signal, color='orange', linewidth=0.8, label='Signal') ax3.axhline(y=0, color='black', linestyle='-', linewidth=0.5) ax3.legend(loc='upper left') # 设置图表标题和标签 ax1.set_title('大周期图表', fontsize=12) ax2.set_title('小周期图表', fontsize=12) ax3.set_title('MACD指标(小周期)', fontsize=10) ax1.grid(True, linestyle='--', alpha=0.3) ax2.grid(True, linestyle='--', alpha=0.3) ax3.grid(True, linestyle='--', alpha=0.3) ax1.xaxis.set_major_formatter(DateFormatter('%Y-%m-%d')) plt.xticks(rotation=45) plt.tight_layout() plt.show() def plot(self, dataframe, bi_list, seg_list, zs_list=None, buy_sell_points=None, divergence_points=None): """ 绘制缠论分析图表，包括K线、笔、线段、中枢、买卖点和MACD背驰 :param dataframe: K线数据 :param bi_list: 笔的列表 :param seg_list: 线段的列表 :param zs_list: 中枢的列表 :param buy_sell_points: 买卖点字典 :param divergence_points: 背驰点字典 """ plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'Microsoft YaHei', 'WenQuanYi Micro Hei'] plt.rcParams['axes.unicode_minus'] = False # 解决负号显示问题 bar_line_width = 0.003 show_sure_time = False # 创建具有两个子图的图表 fig = plt.figure(figsize=(15, 10)) # 主图占据上方70%空间 ax1 = plt.subplot2grid((5, 1), (0, 0), rowspan=3) # MACD子图占据下方30%空间 ax2 = plt.subplot2grid((5, 1), (3, 0), rowspan=2, sharex=ax1) # 转换日期格式 - 确保都是无时区的 dates = pd.to_datetime(dataframe['date']).dt.tz_localize(None) dates_num = [date2num(date) for date in dates] # 绘制K线图 for i in range(len(dataframe)): # 红涨绿跌 if dataframe['close'][i] > dataframe['open'][i]: body_color = 'red' else: body_color = 'green' # 绘制实体 ax1.bar(dates_num[i], dataframe['close'][i] - dataframe['open'][i], bottom=dataframe['open'][i], color=body_color, width=bar_line_width/len(dataframe)) # 绘制上下影线 ax1.plot([dates_num[i], dates_num[i]], [dataframe['low'][i], dataframe['high'][i]], color=body_color, linewidth=1.2) # 绘制笔 for bi in bi_list: if bi.end_klc: # 确保笔已完成 start_time = pd.to_datetime(bi.start_klc.start_time) end_time = pd.to_datetime(bi.end_klc.end_time) # 上升笔蓝色，下降笔紫色 color = 'blue' if bi.dir == Chan_BI_DIR.UP else 'purple' start_price = bi.start_klc.low if bi.dir == Chan_BI_DIR.UP else bi.start_klc.high end_price = bi.end_klc.high if bi.dir == Chan_BI_DIR.UP else bi.end_klc.low # 绘制笔 ax1.plot([date2num(start_time), date2num(end_time)], [start_price, end_price], color=color, linewidth=1.5) # 绘制线段 for seg in seg_list: if seg.end_bi: # 确保线段已完成 start_time = pd.to_datetime(seg.start_bi.start_klc.start_time) end_time = pd.to_datetime(seg.end_bi.end_klc.end_time) # 上升线段红色，下降线段绿色 color = 'red' if seg.dir == Chan_SEG_DIR.UP else 'green' start_price = seg.start_bi.start_klc.low if seg.dir == Chan_SEG_DIR.UP else seg.start_bi.start_klc.high end_price = seg.end_bi.end_klc.high if seg.dir == Chan_SEG_DIR.UP else seg.end_bi.end_klc.low # 绘制线段(粗线) ax1.plot([date2num(start_time), date2num(end_time)], [start_price, end_price], color=color, linewidth=2.5) # 在线段确认点绘制标记 if hasattr(seg, 'sure_time') and seg.sure_time and show_sure_time: try: # 确保sure_time无时区 sure_time = pd.to_datetime(seg.sure_time).tz_localize(None) # 找到最接近的K线 closest_idx = (dates - sure_time).abs().argmin() # 获取确认点的价格 confirm_price = dataframe['close'][closest_idx] # 绘制标记和标签 ax1.plot(date2num(sure_time), confirm_price, 'D', color='black', markersize=6) ax1.annotate(sure_time.strftime('%m-%d %H:%M'), (date2num(sure_time), confirm_price), textcoords="offset points", xytext=(0, 10), ha='center', fontsize=8, bbox=dict(boxstyle="round,pad=0.3", fc="yellow", alpha=0.7)) except Exception as e: print(f"处理线段确认时间时出错: {e}") continue # 绘制中枢区域 if zs_list: # 定义中枢的颜色和透明度 zs_colors = ['orange', 'cyan', 'magenta', 'yellow', 'lime'] for idx, zs in enumerate(zs_list): # 无论中枢是否完成都绘制 start_time = pd.to_datetime(zs.start_klc.start_time).tz_localize(None) # 选择颜色，循环使用预定义的颜色 color = zs_colors[idx % len(zs_colors)] if zs.end_klc: # 已完成的中枢 # 转换结束时间格式 end_time = pd.to_datetime(zs.end_klc.end_time).tz_localize(None) # 矩形的宽度和高度 width = date2num(end_time) - date2num(start_time) height = zs.zg - zs.zd # 创建实线矩形补丁表示已完成中枢 rect = patches.Rectangle( (date2num(start_time), zs.zd), # 左下角坐标 width, # 宽度 height, # 高度 linewidth=1, edgecolor=color, facecolor=color, alpha=0.2 # 透明度 ) ax1.add_patch(rect) # 添加中枢编号标签 label_text = f"中枢{idx+1}" else: # 未完成的中枢 # 使用最后一根K线的时间作为临时结束时间 end_time = pd.to_datetime(dates.iloc[-1]).tz_localize(None) # 矩形的宽度和高度 width = date2num(end_time) - date2num(start_time) height = zs.zg - zs.zd # 创建虚线矩形补丁表示未完成中枢 rect = patches.Rectangle( (date2num(start_time), zs.zd), # 左下角坐标 width, # 宽度 height, # 高度 linewidth=1.5, edgecolor=color, facecolor=color, alpha=0.1, # 较低的透明度 linestyle='--' # 虚线边框 ) ax1.add_patch(rect) # 添加中枢编号标签，标明未完成 label_text = f"中枢{idx+1}(未完成)" # 添加中枢标签 ax1.text( date2num(start_time) + width/2, # x位置（中枢中间） zs.zd + height/2, # y位置（中枢中间） label_text, ha='center', va='center', fontsize=9, color='black', bbox=dict(boxstyle="round,pad=0.2", fc=color, alpha=0.6) ) # 绘制买卖点 if buy_sell_points: marker_styles = { '1类买点': {'marker': '^', 'color': 'red', 'size': 10, 'label': '1类买点'}, '1类卖点': {'marker': 'v', 'color': 'green', 'size': 10, 'label': '1类卖点'}, '2类买点': {'marker': '^', 'color': 'orange', 'size': 10, 'label': '2类买点'}, '2类卖点': {'marker': 'v', 'color': 'cyan', 'size': 10, 'label': '2类卖点'}, '3类买点': {'marker': '^', 'color': 'purple', 'size': 10, 'label': '3类买点'}, '3类卖点': {'marker': 'v', 'color': 'magenta', 'size': 10, 'label': '3类卖点'} } for idx, point in buy_sell_points.items(): if idx < 0 or idx >= len(dataframe): continue print("Plot buy sell point: ", point['type']) style = marker_styles.get(point['type'], {'marker': 'o', 'color': 'black', 'size': 8, 'label': '其他'}) # 绘制买卖点标记 ax1.plot(dates_num[idx], point['price'], style['marker'], color=style['color'], markersize=style['size'], label=style['label']) # 添加买卖点标签 ax1.annotate(point['type'], (dates_num[idx], point['price']), textcoords="offset points", xytext=(0, 10), ha='center', fontsize=8, bbox=dict(boxstyle="round,pad=0.2", fc=style['color'], alpha=0.5)) # 绘制背驰点 if divergence_points: for idx, point in divergence_points.items(): if idx < 0 or idx >= len(dataframe) or True: continue print("Plot divergence point") color = 'red' if point['type'] == '底背驰' else 'green' marker = '*' # 绘制背驰点标记 ax1.plot(dates_num[idx], point['price'], marker, color=color, markersize=12, label=point['type']) # 添加背驰点标签 ax1.annotate(point['type'], (dates_num[idx], point['price']), textcoords="offset points", xytext=(0, -15), ha='center', fontsize=8, bbox=dict(boxstyle="round,pad=0.2", fc=color, alpha=0.5)) # 计算MACD指标 exp1 = dataframe['close'].ewm(span=12, adjust=False).mean() exp2 = dataframe['close'].ewm(span=26, adjust=False).mean() macd = exp1 - exp2 signal = macd.ewm(span=9, adjust=False).mean() histogram = macd - signal # 绘制MACD ax2.bar(dates_num, histogram, width=bar_line_width, color=['red' if h > 0 else 'green' for h in histogram]) ax2.plot(dates_num, macd, color='blue', linewidth=1.2, label='MACD') ax2.plot(dates_num, signal, color='orange', linewidth=1.2, label='Signal') ax2.axhline(y=0, color='black', linestyle='-', linewidth=0.5) ax2.legend(loc='upper left') # 在MACD图上标记背驰点 if divergence_points: for idx, point in divergence_points.items(): if idx < 0 or idx >= len(dataframe): continue color = 'red' if point['type'] == '底背驰' else 'green' # 在MACD图上标记背驰点 ax2.plot(dates_num[idx], histogram[idx], '*', color=color, markersize=12) # 添加简单网格 ax1.grid(True, linestyle='--', alpha=0.3) ax2.grid(True, linestyle='--', alpha=0.3) # 设置坐标轴格式 ax1.xaxis.set_major_formatter(DateFormatter('%Y-%m-%d')) # 添加简单图例 from matplotlib.lines import Line2D legend_elements = [ Line2D([0], [0], color='blue', lw=2, label='上升笔'), Line2D([0], [0], color='purple', lw=2, label='下降笔'), Line2D([0], [0], color='red', lw=2.5, label='上升线段'), Line2D([0], [0], color='green', lw=2.5, label='下降线段'), patches.Patch(facecolor='orange', alpha=0.2, label='已完成中枢'), patches.Patch(facecolor='orange', alpha=0.1, edgecolor='orange', linestyle='--', label='未完成中枢'), Line2D([0], [0], marker='^', color='red', label='买点', markersize=10, linestyle='None'), Line2D([0], [0], marker='v', color='green', label='卖点', markersize=10, linestyle='None'), Line2D([0], [0], marker='*', color='red', label='底背驰', markersize=12, linestyle='None'), Line2D([0], [0], marker='*', color='green', label='顶背驰', markersize=12, linestyle='None') ] ax1.legend(handles=legend_elements, loc='upper left') # 设置标题和标签 ax1.set_title('缠论分析图', fontsize=14) ax1.set_ylabel('价格', fontsize=12) ax2.set_xlabel('时间', fontsize=12) ax2.set_ylabel('MACD', fontsize=12) plt.xticks(rotation=45) plt.tight_layout() # 显示图表 plt.show() """ for index in range(seg.next.next.start_bi.index, seg.next.next.end_bi.index): 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) last_zs.set_last_bi_in(None) last_zs.set_end_seg(None) first_bi_out = None #print("Bi in again 1", 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) 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 print("First bi out 1", first_bi_out.start_klc.start_time) in_again = False """ def check_top_bottom(self, dataframe, bi_list, seg_list, zs_list): """ 检测新高/新低时的第一类买卖点，结合MACD背驰判断 :param dataframe: K线数据 :param bi_list: 笔的列表 :param seg_list: 线段的列表 :param zs_list: 中枢的列表 :return: 第一类买卖点列表，格式为{index: {'type': 类型, 'price': 价格, 'time': 时间}} """ buy_sell_points = {} # 计算MACD指标 exp1 = dataframe['close'].ewm(span=12, adjust=False).mean() exp2 = dataframe['close'].ewm(span=26, adjust=False).mean() macd = exp1 - exp2 signal = macd.ewm(span=9, adjust=False).mean() histogram = macd - signal # MACD柱状图的面积 positive_hist = histogram.copy() negative_hist = histogram.copy() positive_hist[positive_hist < 0] = 0 negative_hist[negative_hist > 0] = 0 # 找到所有底分型和顶分型的笔 bottom_bi_indices = [] # 底分型的笔索引 top_bi_indices = [] # 顶分型的笔索引 for i, bi in enumerate(bi_list): if not bi.end_klc: continue if bi.dir == Chan_BI_DIR.UP and i > 0: bottom_bi_indices.append(i-1) # 上升笔的前一笔是底分型 elif bi.dir == Chan_BI_DIR.DOWN and i > 0: top_bi_indices.append(i-1) # 下降笔的前一笔是顶分型 # 查找创新高的顶分型（第一类卖点） for i in range(1, len(top_bi_indices)): curr_idx = top_bi_indices[i] prev_idx = top_bi_indices[i-1] if curr_idx >= len(bi_list) or prev_idx >= len(bi_list): continue curr_bi = bi_list[curr_idx] prev_bi = bi_list[prev_idx] if not curr_bi.end_klc or not prev_bi.end_klc: continue # 确保是新高：当前高点比前一高点更高 if curr_bi.high > prev_bi.high: # 找到对应的MACD值 curr_time = curr_bi.end_klc.end_time prev_time = prev_bi.end_klc.end_time # 获取对应的dataframe索引 curr_date_idx = dataframe[dataframe['date'].astype(str).str.contains(curr_time)].index[0] if any(dataframe['date'].astype(str).str.contains(curr_time)) else -1 prev_date_idx = dataframe[dataframe['date'].astype(str).str.contains(prev_time)].index[0] if any(dataframe['date'].astype(str).str.contains(prev_time)) else -1 if curr_date_idx >= 0 and prev_date_idx >= 0: # 计算两段走势的MACD柱状图面积（顶分型关注正面积） curr_area = positive_hist[prev_date_idx:curr_date_idx+1].sum() prev_area = positive_hist[max(0, prev_date_idx-abs(curr_date_idx-prev_date_idx)):prev_date_idx+1].sum() # 检查是否有MACD背驰 # 新高但MACD力度减弱，形成顶背驰 if curr_area < prev_area and curr_area > 0: # 检查是否在中枢中 in_zs = False for zs in zs_list: if zs.zd <= curr_bi.high <= zs.zg: in_zs = True break if not in_zs: # 不在中枢中的第一类卖点更可靠 # 检查线段方向，确保是上升趋势 is_uptrend = False for seg in seg_list: if seg.end_bi and seg.dir == Chan_SEG_DIR.UP and seg.end_bi.index >= curr_bi.index: is_uptrend = True break if is_uptrend: buy_sell_points[curr_date_idx] = { 'type': '第一类卖点', 'price': dataframe.loc[curr_date_idx, 'high'], 'time': curr_time, 'reason': f'新高+顶背驰(MACD: {curr_area:.2f}<{prev_area:.2f})', 'bi_idx': curr_idx, 'is_sure': curr_bi.is_sure } # 查找创新低的底分型（第一类买点） for i in range(1, len(bottom_bi_indices)): curr_idx = bottom_bi_indices[i] prev_idx = bottom_bi_indices[i-1] if curr_idx >= len(bi_list) or prev_idx >= len(bi_list): continue curr_bi = bi_list[curr_idx] prev_bi = bi_list[prev_idx] if not curr_bi.end_klc or not prev_bi.end_klc: continue # 确保是新低：当前低点比前一低点更低 if curr_bi.low < prev_bi.low: # 找到对应的MACD值 curr_time = curr_bi.end_klc.end_time prev_time = prev_bi.end_klc.end_time # 获取对应的dataframe索引 curr_date_idx = dataframe[dataframe['date'].astype(str).str.contains(curr_time)].index[0] if any(dataframe['date'].astype(str).str.contains(curr_time)) else -1 prev_date_idx = dataframe[dataframe['date'].astype(str).str.contains(prev_time)].index[0] if any(dataframe['date'].astype(str).str.contains(prev_time)) else -1 if curr_date_idx >= 0 and prev_date_idx >= 0: # 计算两段走势的MACD柱状图面积（底分型关注负面积） curr_area = abs(negative_hist[prev_date_idx:curr_date_idx+1].sum()) prev_area = abs(negative_hist[max(0, prev_date_idx-abs(curr_date_idx-prev_date_idx)):prev_date_idx+1].sum()) # 检查是否有MACD背驰 # 新低但MACD力度减弱，形成底背驰 if curr_area < prev_area and curr_area > 0: # 检查是否在中枢中 in_zs = False for zs in zs_list: if zs.zd <= curr_bi.low <= zs.zg: in_zs = True break if not in_zs: # 不在中枢中的第一类买点更可靠 # 检查线段方向，确保是下降趋势 is_downtrend = False for seg in seg_list: if seg.end_bi and seg.dir == Chan_SEG_DIR.DOWN and seg.end_bi.index >= curr_bi.index: is_downtrend = True break if is_downtrend: buy_sell_points[curr_date_idx] = { 'type': '第一类买点', 'price': dataframe.loc[curr_date_idx, 'low'], 'time': curr_time, 'reason': f'新低+底背驰(MACD: {curr_area:.2f}<{prev_area:.2f})', 'bi_idx': curr_idx, 'is_sure': curr_bi.is_sure } return buy_sell_points