import copy from typing import Dict, Optional from ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR, Chan_BI_DIR import ChanKLU from ChanBI import ChanBI class ChanSBI(): def __init__(self, start_bi: ChanBI, index, dir=Chan_BI_DIR.UP): self.start_bi = start_bi self.end_bi = None self.index = index self.dir = dir self.high = start_bi.high self.low = start_bi.low self.pre = None self.next = None self.fx = Chan_FX_TYPE.UNKNOWN self.bi_list = [] self.bi_list.append(start_bi) self.has_fx_gap = False def set_fx(self, fx): self.fx = fx def set_end_bi(self, bi): self.end_bi = bi def set_pre(self, sbi): self.pre = sbi def set_next(self, sbi): self.next = sbi def add_bi(self, bi): self.bi_list.append(bi) def check_fx(self): if self.pre and self.next: #print(self.pre.start_bi.start_time, self.start_bi.start_time, self.end_bi.end_time, self.next.start_bi.start_time, self.pre.high, self.high, self.next.high, self.pre.low, self.low, self.next.low, self.dir) if self.high > self.pre.high and self.high > self.next.high: self.fx = Chan_FX_TYPE.TOP #print(self.start_bi.start_time, self.pre.start_bi.start_time, self.next.start_bi.start_time, self.fx) if self.low > self.pre.high: self.has_fx_gap = True #print(self.start_bi.start_time, self.end_bi.end_time, self.pre.start_bi.start_time, self.next.start_bi.start_time, self.dir, self.has_fx_gap, self.fx) return Chan_FX_TYPE.TOP else: if self.low < self.pre.low and self.low < self.next.low: self.fx = Chan_FX_TYPE.BOTTOM #print(self.start_bi.start_time, self.pre.start_bi.start_time, self.next.start_bi.start_time, self.fx) if self.high < self.pre.low: self.has_fx_gap = True #print(self.start_bi.start_time, self.end_bi.end_time, self.pre.start_bi.start_time, self.next.start_bi.start_time, self.dir, self.has_fx_gap, self.fx) return Chan_FX_TYPE.BOTTOM return Chan_FX_TYPE.UNKNOWN def check_seg_bi_broken(self): broken = False if self.fx == Chan_FX_TYPE.TOP: if self.next.low < self.pre.high: broken = True elif self.fx == Chan_FX_TYPE.BOTTOM: if self.next.high > self.pre.low: broken = True return broken def check_bi_included(self, bi): included = False if self.high > bi.high: # high大于,low小于,左包含 if self.low < bi.low: included = True # high大于,low大于,不包含 else: # if self.low > bi.low # high相等,右包含 included = False else: included = False # high小于,low大于,右包含 #if self.low > bi.low: #included = True if included: if self.pre: if self.high > self.pre.high and self.low < self.pre.low: included = True if included: self.add_bi(bi) # gn>gn-1 if self.dir == Chan_BI_DIR.DOWN: # UP -> max(dn) self.low = bi.low else: # DOWN -> min(gn) self.high = bi.high #self.print(bi, "Z") #print(self.start_bi.start_time, bi.start_time, included) return included