import copy from typing import Dict, Optional from .ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR, Chan_SEG_DIR, Chan_BI_DIR, Chan_ZS_DIR from . import ChanCTime from .ChanBI import ChanBI from .ChanBIZS import ChanBIZS class ChanSEG(): def __init__(self, start_bi: ChanBI, index, ddir=Chan_SEG_DIR.UP, pre_end_bi: ChanBI = None): self.start_bi = start_bi self.start_time = start_bi.start_time self.end_time = None self.end_bi = None self.dir = ddir self.low = 0 self.high = 0 if self.dir == Chan_SEG_DIR.UP and start_bi: self.low = start_bi.low else: if start_bi: self.high = start_bi.high self.index = index self.pre = None self.next = None self.bi_list = [] self.bi_list.append(start_bi) self.is_sure = False self.sure_time = None self.macd_hist = 0 self.macd_div = 0 self.start_bi.set_seg(self) self.pre_end_bi = pre_end_bi if self.pre_end_bi: self.ini_seg() def ini_seg(self): next_bi = self.start_bi.next for index in range(self.start_bi.index+1, self.pre_end_bi.index): if next_bi: self.bi_list.append(next_bi) next_bi.set_seg(self) next_bi = next_bi.next def set_macdhist(self, macd_hist): self.macd_hist = macd_hist def set_macd_div(self, macd_div): self.macd_div = macd_div def set_end_bi(self, bi: ChanBI, sure_bi: ChanBI): self.end_bi = bi if bi and bi.is_sure: if self.dir == Chan_SEG_DIR.UP: self.high = bi.high else: self.low = bi.low self.is_sure = True self.end_time = bi.end_klc.end_time if sure_bi.is_sure: self.sure_time = sure_bi.sure_time self.format_bi_list() def pre_set_end_bi(self, bi: ChanBI): self.end_bi = bi if bi and bi.is_sure: if self.dir == Chan_SEG_DIR.UP: self.high = bi.high else: self.low = bi.low self.end_time = bi.end_klc.end_time self.format_bi_list() def set_pre(self, seg): self.pre = seg def set_next(self, seg): self.next = seg def set_sure(self, sure_bi): if sure_bi.is_sure: self.sure_time = sure_bi.sure_time self.is_sure = True self.format_bi_list() def format_bi_list(self): self.bi_list = [] self.bi_list.append(self.start_bi) if self.end_bi: next_bi = self.start_bi.next for i in range(self.start_bi.index, self.end_bi.index): if next_bi: self.bi_list.append(next_bi) next_bi.set_seg(self) next_bi = next_bi.next def add_bi(self, bi: ChanBI): if len(self.bi_list) > 0: self.bi_list.append(bi) bi.set_seg(self) self.end_time = bi.end_time self.end_bi = bi def cal_bi_zs(self): zs_list = [] if len(self.bi_list) > 3: last_zs = None if self.dir == Chan_SEG_DIR.UP: for index in range(1, len(self.bi_list)): bi = self.bi_list[index] #print(bi.end_time, bi.next,"UP SEG BI ZS Index") if bi.next == None or bi.next.next == None: if last_zs and (bi.low > last_zs.zg or bi.high < last_zs.zd): last_zs.set_end_bi(last_zs.bi_list[-1], last_zs.bi_list[-1].sure_time) continue bi2 = bi.next bi3 = bi.next.next if len(zs_list) == 0 or (last_zs and last_zs.is_sure): if bi3.is_sure and bi3.index <= self.bi_list[-1].index and bi.check_bi_zs_overlap() and bi.dir == Chan_BI_DIR.DOWN: zg = min(bi.high, bi2.high, bi3.high) zd = max(bi.low, bi2.low, bi3.low) gg = max(bi.high, bi2.high, bi3.high) dd = min(bi.low, bi2.low, bi3.low) zs = ChanBIZS(bi, len(zs_list), Chan_ZS_DIR.UP) zs.set_zg(zg) zs.set_zd(zd) zs.set_gg(gg) zs.set_dd(dd) zs.add_bi(bi2) zs.add_bi(bi3) zs_list.append(zs) last_zs = zs else: if bi.index > last_zs.bi_list[-1].index and bi.dir == Chan_BI_DIR.DOWN and bi.is_sure: if bi.low > last_zs.zg or bi.high < last_zs.zd: #print(bi.end_time, "UP SEG BI ZS End") last_zs.set_end_bi(last_zs.bi_list[-1], last_zs.bi_list[-1].sure_time) if bi3.is_sure and bi3.index <= self.bi_list[-1].index and bi.check_bi_zs_overlap() and bi.dir == Chan_BI_DIR.DOWN: zg = min(bi.high, bi2.high, bi3.high) zd = max(bi.low, bi2.low, bi3.low) gg = max(bi.high, bi2.high, bi3.high) dd = min(bi.low, bi2.low, bi3.low) zs = ChanBIZS(bi, len(zs_list), Chan_ZS_DIR.UP) zs.set_zg(zg) zs.set_zd(zd) zs.set_gg(gg) zs.set_dd(dd) zs.add_bi(bi2) zs.add_bi(bi3) zs_list.append(zs) last_zs = zs else: last_zs.add_bi(bi.pre) last_zs.add_bi(bi) if index == len(self.bi_list) - 1 and last_zs and not last_zs.is_sure: #print(bi.start_time, "BI", last_zs.is_sure) last_zs.set_end_bi(last_zs.bi_list[-1], last_zs.bi_list[-1].sure_time) else: for index in range(1, len(self.bi_list)): bi = self.bi_list[index] if bi.next == None or bi.next.next == None: if last_zs and (bi.low > last_zs.zg or bi.high < last_zs.zd): last_zs.set_end_bi(last_zs.bi_list[-1], last_zs.bi_list[-1].sure_time) continue bi2 = bi.next bi3 = bi.next.next if len(zs_list) == 0 or (last_zs and last_zs.is_sure): if bi3.is_sure and bi3.index <= self.bi_list[-1].index and bi.check_bi_zs_overlap() and bi.dir == Chan_BI_DIR.UP: zg = min(bi.high, bi2.high, bi3.high) zd = max(bi.low, bi2.low, bi3.low) gg = max(bi.high, bi2.high, bi3.high) dd = min(bi.low, bi2.low, bi3.low) zs = ChanBIZS(bi, len(zs_list), Chan_ZS_DIR.DOWN) zs.set_zg(zg) zs.set_zd(zd) zs.set_gg(gg) zs.set_dd(dd) zs.add_bi(bi2) zs.add_bi(bi3) zs_list.append(zs) last_zs = zs else: if bi.index > last_zs.bi_list[-1].index and bi.dir == Chan_BI_DIR.UP and bi.is_sure: if bi.low > last_zs.zg or bi.high < last_zs.zd: last_zs.set_end_bi(last_zs.bi_list[-1], last_zs.bi_list[-1].sure_time) if bi3.is_sure and bi3.index <= self.bi_list[-1].index and bi.check_bi_zs_overlap() and bi.dir == Chan_BI_DIR.UP: zg = min(bi.high, bi2.high, bi3.high) zd = max(bi.low, bi2.low, bi3.low) gg = max(bi.high, bi2.high, bi3.high) dd = min(bi.low, bi2.low, bi3.low) zs = ChanBIZS(bi, len(zs_list), Chan_ZS_DIR.DOWN) zs.set_zg(zg) zs.set_zd(zd) zs.set_gg(gg) zs.set_dd(dd) zs.add_bi(bi2) zs.add_bi(bi3) zs_list.append(zs) last_zs = zs else: last_zs.add_bi(bi.pre) last_zs.add_bi(bi) if index == len(self.bi_list) - 1 and last_zs and not last_zs.is_sure: #print(bi.start_time, "BI", last_zs.is_sure) last_zs.set_end_bi(last_zs.bi_list[-1], last_zs.bi_list[-1].sure_time) #print(self.start_time, len(zs_list)) #print(self.bi_list[-1].end_time, "end_bi") return zs_list