343 lines
14 KiB
Python
343 lines
14 KiB
Python
from typing import Dict, Generic, List, Optional, TypeVar, Union, overload
|
|
|
|
from Bi.Bi import CBi
|
|
from Bi.BiList import CBiList
|
|
from Common.CEnum import BSP_TYPE
|
|
from Common.func_util import has_overlap
|
|
from Seg.Seg import CSeg
|
|
from Seg.SegListComm import CSegListComm
|
|
from ZS.ZS import CZS
|
|
|
|
from .BS_Point import CBS_Point
|
|
from .BSPointConfig import CBSPointConfig, CPointConfig
|
|
|
|
LINE_TYPE = TypeVar('LINE_TYPE', CBi, CSeg[CBi])
|
|
LINE_LIST_TYPE = TypeVar('LINE_LIST_TYPE', CBiList, CSegListComm[CBi])
|
|
|
|
|
|
class CBSPointList(Generic[LINE_TYPE, LINE_LIST_TYPE]):
|
|
def __init__(self, bs_point_config: CBSPointConfig):
|
|
self.lst: List[CBS_Point[LINE_TYPE]] = []
|
|
self.bsp_dict: Dict[int, CBS_Point[LINE_TYPE]] = {}
|
|
self.bsp1_lst: List[CBS_Point[LINE_TYPE]] = []
|
|
self.config = bs_point_config
|
|
self.last_sure_pos = -1
|
|
|
|
def __iter__(self):
|
|
yield from self.lst
|
|
|
|
def __len__(self):
|
|
return len(self.lst)
|
|
|
|
@overload
|
|
def __getitem__(self, index: int) -> CBS_Point: ...
|
|
|
|
@overload
|
|
def __getitem__(self, index: slice) -> List[CBS_Point]: ...
|
|
|
|
def __getitem__(self, index: Union[slice, int]) -> Union[List[CBS_Point], CBS_Point]:
|
|
return self.lst[index]
|
|
|
|
def cal(self, bi_list: LINE_LIST_TYPE, seg_list: CSegListComm[LINE_TYPE]):
|
|
self.lst = [bsp for bsp in self.lst if bsp.klu.idx <= self.last_sure_pos]
|
|
self.bsp_dict = {bsp.bi.get_end_klu().idx: bsp for bsp in self.lst}
|
|
self.bsp1_lst = [bsp for bsp in self.bsp1_lst if bsp.klu.idx <= self.last_sure_pos]
|
|
|
|
self.cal_seg_bs1point(seg_list, bi_list)
|
|
self.cal_seg_bs2point(seg_list, bi_list)
|
|
self.cal_seg_bs3point(seg_list, bi_list)
|
|
|
|
self.update_last_pos(seg_list)
|
|
|
|
def update_last_pos(self, seg_list: CSegListComm):
|
|
self.last_sure_pos = -1
|
|
for seg in seg_list[::-1]:
|
|
if seg.is_sure:
|
|
self.last_sure_pos = seg.end_bi.get_begin_klu().idx
|
|
return
|
|
|
|
def seg_need_cal(self, seg: CSeg):
|
|
return seg.end_bi.get_end_klu().idx > self.last_sure_pos
|
|
|
|
def add_bs(
|
|
self,
|
|
bs_type: BSP_TYPE,
|
|
bi: LINE_TYPE,
|
|
relate_bsp1: Optional[CBS_Point],
|
|
is_target_bsp: bool = True,
|
|
feature_dict=None,
|
|
):
|
|
is_buy = bi.is_down()
|
|
if exist_bsp := self.bsp_dict.get(bi.get_end_klu().idx):
|
|
assert exist_bsp.is_buy == is_buy
|
|
exist_bsp.add_another_bsp_prop(bs_type, relate_bsp1)
|
|
return
|
|
if bs_type not in self.config.GetBSConfig(is_buy).target_types:
|
|
is_target_bsp = False
|
|
|
|
if is_target_bsp or bs_type in [BSP_TYPE.T1, BSP_TYPE.T1P]:
|
|
bsp = CBS_Point[LINE_TYPE](
|
|
bi=bi,
|
|
is_buy=is_buy,
|
|
bs_type=bs_type,
|
|
relate_bsp1=relate_bsp1,
|
|
feature_dict=feature_dict,
|
|
)
|
|
else:
|
|
return
|
|
if is_target_bsp:
|
|
self.lst.append(bsp)
|
|
self.bsp_dict[bi.get_end_klu().idx] = bsp
|
|
if bs_type in [BSP_TYPE.T1, BSP_TYPE.T1P]:
|
|
self.bsp1_lst.append(bsp)
|
|
|
|
def cal_seg_bs1point(self, seg_list: CSegListComm[LINE_TYPE], bi_list: LINE_LIST_TYPE):
|
|
for seg in seg_list:
|
|
if not self.seg_need_cal(seg):
|
|
continue
|
|
self.cal_single_bs1point(seg, bi_list)
|
|
|
|
def cal_single_bs1point(self, seg: CSeg[LINE_TYPE], bi_list: LINE_LIST_TYPE):
|
|
BSP_CONF = self.config.GetBSConfig(seg.is_down())
|
|
zs_cnt = seg.get_multi_bi_zs_cnt() if BSP_CONF.bsp1_only_multibi_zs else len(seg.zs_lst)
|
|
is_target_bsp = (BSP_CONF.min_zs_cnt <= 0 or zs_cnt >= BSP_CONF.min_zs_cnt)
|
|
if len(seg.zs_lst) > 0 and \
|
|
not seg.zs_lst[-1].is_one_bi_zs() and \
|
|
((seg.zs_lst[-1].bi_out and seg.zs_lst[-1].bi_out.idx >= seg.end_bi.idx) or seg.zs_lst[-1].bi_lst[-1].idx >= seg.end_bi.idx) \
|
|
and seg.end_bi.idx - seg.zs_lst[-1].get_bi_in().idx > 2:
|
|
self.treat_bsp1(seg, BSP_CONF, is_target_bsp)
|
|
else:
|
|
self.treat_pz_bsp1(seg, BSP_CONF, bi_list, is_target_bsp)
|
|
|
|
def treat_bsp1(self, seg: CSeg[LINE_TYPE], BSP_CONF: CPointConfig, is_target_bsp: bool):
|
|
last_zs = seg.zs_lst[-1]
|
|
break_peak, _ = last_zs.out_bi_is_peak(seg.end_bi.idx)
|
|
if BSP_CONF.bs1_peak and not break_peak:
|
|
is_target_bsp = False
|
|
is_diver, divergence_rate = last_zs.is_divergence(BSP_CONF, out_bi=seg.end_bi)
|
|
if not is_diver:
|
|
is_target_bsp = False
|
|
feature_dict = {'divergence_rate': divergence_rate}
|
|
self.add_bs(bs_type=BSP_TYPE.T1, bi=seg.end_bi, relate_bsp1=None, is_target_bsp=is_target_bsp, feature_dict=feature_dict)
|
|
|
|
def treat_pz_bsp1(self, seg: CSeg[LINE_TYPE], BSP_CONF: CPointConfig, bi_list: LINE_LIST_TYPE, is_target_bsp):
|
|
last_bi = seg.end_bi
|
|
pre_bi = bi_list[last_bi.idx-2]
|
|
if last_bi.seg_idx != pre_bi.seg_idx:
|
|
return
|
|
if last_bi.dir != seg.dir:
|
|
return
|
|
if last_bi.is_down() and last_bi._low() > pre_bi._low(): # 创新低
|
|
return
|
|
if last_bi.is_up() and last_bi._high() < pre_bi._high(): # 创新高
|
|
return
|
|
in_metric = pre_bi.cal_macd_metric(BSP_CONF.macd_algo, is_reverse=False)
|
|
out_metric = last_bi.cal_macd_metric(BSP_CONF.macd_algo, is_reverse=True)
|
|
is_diver, divergence_rate = out_metric <= BSP_CONF.divergence_rate*in_metric, out_metric/(in_metric+1e-7)
|
|
if not is_diver:
|
|
is_target_bsp = False
|
|
if isinstance(bi_list, CBiList):
|
|
assert isinstance(last_bi, CBi) and isinstance(pre_bi, CBi)
|
|
feature_dict = {'divergence_rate': divergence_rate}
|
|
self.add_bs(bs_type=BSP_TYPE.T1P, bi=last_bi, relate_bsp1=None, is_target_bsp=is_target_bsp, feature_dict=feature_dict)
|
|
|
|
def cal_seg_bs2point(self, seg_list: CSegListComm[LINE_TYPE], bi_list: LINE_LIST_TYPE):
|
|
bsp1_bi_idx_dict = {bsp.bi.idx: bsp for bsp in self.bsp1_lst}
|
|
for seg in seg_list:
|
|
config = self.config.GetBSConfig(seg.is_down())
|
|
if BSP_TYPE.T2 not in config.target_types and BSP_TYPE.T2S not in config.target_types:
|
|
continue
|
|
self.treat_bsp2(seg, bsp1_bi_idx_dict, seg_list, bi_list)
|
|
|
|
def treat_bsp2(self, seg: CSeg, bsp1_bi_idx_dict, seg_list: CSegListComm[LINE_TYPE], bi_list: LINE_LIST_TYPE):
|
|
if not self.seg_need_cal(seg):
|
|
return
|
|
if len(seg_list) > 1:
|
|
BSP_CONF = self.config.GetBSConfig(seg.is_down())
|
|
bsp1_bi = seg.end_bi
|
|
bsp1_bi_idx = bsp1_bi.idx
|
|
real_bsp1 = bsp1_bi_idx_dict.get(bsp1_bi.idx)
|
|
if bsp1_bi.idx + 2 >= len(bi_list):
|
|
return
|
|
break_bi = bi_list[bsp1_bi.idx + 1]
|
|
bsp2_bi = bi_list[bsp1_bi.idx + 2]
|
|
else:
|
|
BSP_CONF = self.config.GetBSConfig(seg.is_up())
|
|
bsp1_bi, real_bsp1 = None, None
|
|
bsp1_bi_idx = -1
|
|
if len(bi_list) == 1:
|
|
return
|
|
bsp2_bi = bi_list[1]
|
|
break_bi = bi_list[0]
|
|
if BSP_CONF.bsp2_follow_1 and bsp1_bi_idx not in [bsp.bi.idx for bsp in self.bsp_dict.values()]: # check bsp2_follow_1
|
|
return
|
|
retrace_rate = bsp2_bi.amp()/break_bi.amp()
|
|
bsp2_flag = retrace_rate <= BSP_CONF.max_bs2_rate
|
|
if bsp2_flag:
|
|
self.add_bs(bs_type=BSP_TYPE.T2, bi=bsp2_bi, relate_bsp1=real_bsp1) # type: ignore
|
|
elif BSP_CONF.bsp2s_follow_2:
|
|
return
|
|
if BSP_TYPE.T2S not in self.config.GetBSConfig(seg.is_down()).target_types:
|
|
return
|
|
self.treat_bsp2s(seg_list, bi_list, bsp2_bi, break_bi, real_bsp1, BSP_CONF) # type: ignore
|
|
|
|
def treat_bsp2s(
|
|
self,
|
|
seg_list: CSegListComm,
|
|
bi_list: LINE_LIST_TYPE,
|
|
bsp2_bi: LINE_TYPE,
|
|
break_bi: LINE_TYPE,
|
|
real_bsp1: Optional[CBS_Point],
|
|
BSP_CONF: CPointConfig,
|
|
):
|
|
bias = 2
|
|
_low, _high = None, None
|
|
while bsp2_bi.idx + bias < len(bi_list): # 计算类二
|
|
bsp2s_bi = bi_list[bsp2_bi.idx + bias]
|
|
assert bsp2s_bi.seg_idx is not None and bsp2_bi.seg_idx is not None
|
|
if BSP_CONF.max_bsp2s_lv is not None and bias/2 > BSP_CONF.max_bsp2s_lv:
|
|
break
|
|
if bsp2s_bi.seg_idx != bsp2_bi.seg_idx and (bsp2s_bi.seg_idx < len(seg_list)-1 or bsp2s_bi.seg_idx - bsp2_bi.seg_idx >= 2 or seg_list[bsp2_bi.seg_idx].is_sure):
|
|
break
|
|
if bias == 2:
|
|
if not has_overlap(bsp2_bi._low(), bsp2_bi._high(), bsp2s_bi._low(), bsp2s_bi._high()):
|
|
break
|
|
_low = max([bsp2_bi._low(), bsp2s_bi._low()])
|
|
_high = min([bsp2_bi._high(), bsp2s_bi._high()])
|
|
elif not has_overlap(_low, _high, bsp2s_bi._low(), bsp2s_bi._high()):
|
|
break
|
|
|
|
if bsp2s_break_bsp1(bsp2s_bi, break_bi):
|
|
break
|
|
retrace_rate = abs(bsp2s_bi.get_end_val()-break_bi.get_end_val())/break_bi.amp()
|
|
if retrace_rate > BSP_CONF.max_bs2_rate:
|
|
break
|
|
|
|
self.add_bs(bs_type=BSP_TYPE.T2S, bi=bsp2s_bi, relate_bsp1=real_bsp1) # type: ignore
|
|
bias += 2
|
|
|
|
def cal_seg_bs3point(self, seg_list: CSegListComm[LINE_TYPE], bi_list: LINE_LIST_TYPE):
|
|
bsp1_bi_idx_dict = {bsp.bi.idx: bsp for bsp in self.bsp1_lst}
|
|
for seg in seg_list:
|
|
if not self.seg_need_cal(seg):
|
|
continue
|
|
config = self.config.GetBSConfig(seg.is_down())
|
|
if BSP_TYPE.T3A not in config.target_types and BSP_TYPE.T3B not in config.target_types:
|
|
continue
|
|
if len(seg_list) > 1:
|
|
bsp1_bi = seg.end_bi
|
|
bsp1_bi_idx = bsp1_bi.idx
|
|
BSP_CONF = self.config.GetBSConfig(seg.is_down())
|
|
real_bsp1 = bsp1_bi_idx_dict.get(bsp1_bi.idx)
|
|
next_seg_idx = seg.idx+1
|
|
next_seg = seg.next # 可能为None, 所以并不一定可以保证next_seg_idx == next_seg.idx
|
|
else:
|
|
next_seg = seg
|
|
next_seg_idx = seg.idx
|
|
bsp1_bi, real_bsp1 = None, None
|
|
bsp1_bi_idx = -1
|
|
BSP_CONF = self.config.GetBSConfig(seg.is_up())
|
|
if BSP_CONF.bsp3_follow_1 and bsp1_bi_idx not in [bsp.bi.idx for bsp in self.bsp_dict.values()]:
|
|
continue
|
|
if next_seg:
|
|
self.treat_bsp3_after(seg_list, next_seg, BSP_CONF, bi_list, real_bsp1, bsp1_bi_idx, next_seg_idx)
|
|
self.treat_bsp3_before(seg_list, seg, next_seg, bsp1_bi, BSP_CONF, bi_list, real_bsp1, next_seg_idx)
|
|
|
|
def treat_bsp3_after(
|
|
self,
|
|
seg_list: CSegListComm[LINE_TYPE],
|
|
next_seg: CSeg[LINE_TYPE],
|
|
BSP_CONF: CPointConfig,
|
|
bi_list: LINE_LIST_TYPE,
|
|
real_bsp1,
|
|
bsp1_bi_idx,
|
|
next_seg_idx
|
|
):
|
|
first_zs = next_seg.get_first_multi_bi_zs()
|
|
if first_zs is None:
|
|
return
|
|
if BSP_CONF.strict_bsp3 and first_zs.get_bi_in().idx != bsp1_bi_idx+1:
|
|
return
|
|
if first_zs.bi_out is None or first_zs.bi_out.idx+1 >= len(bi_list):
|
|
return
|
|
bsp3_bi = bi_list[first_zs.bi_out.idx+1]
|
|
if bsp3_bi.parent_seg is None:
|
|
if next_seg.idx != len(seg_list)-1:
|
|
return
|
|
elif bsp3_bi.parent_seg.idx != next_seg.idx:
|
|
if len(bsp3_bi.parent_seg.bi_list) >= 3:
|
|
return
|
|
if bsp3_bi.dir == next_seg.dir:
|
|
return
|
|
if bsp3_bi.seg_idx != next_seg_idx and next_seg_idx < len(seg_list)-2:
|
|
return
|
|
if bsp3_back2zs(bsp3_bi, first_zs):
|
|
return
|
|
bsp3_peak_zs = bsp3_break_zspeak(bsp3_bi, first_zs)
|
|
if BSP_CONF.bsp3_peak and not bsp3_peak_zs:
|
|
return
|
|
self.add_bs(bs_type=BSP_TYPE.T3A, bi=bsp3_bi, relate_bsp1=real_bsp1) # type: ignore
|
|
|
|
def treat_bsp3_before(
|
|
self,
|
|
seg_list: CSegListComm[LINE_TYPE],
|
|
seg: CSeg[LINE_TYPE],
|
|
next_seg: Optional[CSeg[LINE_TYPE]],
|
|
bsp1_bi: Optional[LINE_TYPE],
|
|
BSP_CONF: CPointConfig,
|
|
bi_list: LINE_LIST_TYPE,
|
|
real_bsp1,
|
|
next_seg_idx
|
|
):
|
|
cmp_zs = seg.get_final_multi_bi_zs()
|
|
if cmp_zs is None:
|
|
return
|
|
if not bsp1_bi:
|
|
return
|
|
if BSP_CONF.strict_bsp3 and (cmp_zs.bi_out is None or cmp_zs.bi_out.idx != bsp1_bi.idx):
|
|
return
|
|
end_bi_idx = cal_bsp3_bi_end_idx(next_seg)
|
|
for bsp3_bi in bi_list[bsp1_bi.idx+2::2]:
|
|
if bsp3_bi.idx > end_bi_idx:
|
|
break
|
|
assert bsp3_bi.seg_idx is not None
|
|
if bsp3_bi.seg_idx != next_seg_idx and bsp3_bi.seg_idx < len(seg_list)-1:
|
|
break
|
|
if bsp3_back2zs(bsp3_bi, cmp_zs): # type: ignore
|
|
continue
|
|
self.add_bs(bs_type=BSP_TYPE.T3B, bi=bsp3_bi, relate_bsp1=real_bsp1) # type: ignore
|
|
break
|
|
|
|
def getLastestBspList(self) -> List[CBS_Point[LINE_TYPE]]:
|
|
if len(self.lst) == 0:
|
|
return []
|
|
return sorted(self.lst, key=lambda bsp: bsp.bi.idx, reverse=True)
|
|
|
|
|
|
def bsp2s_break_bsp1(bsp2s_bi: LINE_TYPE, bsp2_break_bi: LINE_TYPE) -> bool:
|
|
return (bsp2s_bi.is_down() and bsp2s_bi._low() < bsp2_break_bi._low()) or \
|
|
(bsp2s_bi.is_up() and bsp2s_bi._high() > bsp2_break_bi._high())
|
|
|
|
|
|
def bsp3_back2zs(bsp3_bi: LINE_TYPE, zs: CZS) -> bool:
|
|
return (bsp3_bi.is_down() and bsp3_bi._low() < zs.high) or (bsp3_bi.is_up() and bsp3_bi._high() > zs.low)
|
|
|
|
|
|
def bsp3_break_zspeak(bsp3_bi: LINE_TYPE, zs: CZS) -> bool:
|
|
return (bsp3_bi.is_down() and bsp3_bi._high() >= zs.peak_high) or (bsp3_bi.is_up() and bsp3_bi._low() <= zs.peak_low)
|
|
|
|
|
|
def cal_bsp3_bi_end_idx(seg: Optional[CSeg[LINE_TYPE]]):
|
|
if not seg:
|
|
return float("inf")
|
|
if seg.get_multi_bi_zs_cnt() == 0 and seg.next is None:
|
|
return float("inf")
|
|
end_bi_idx = seg.end_bi.idx-1
|
|
for zs in seg.zs_lst:
|
|
if zs.is_one_bi_zs():
|
|
continue
|
|
if zs.bi_out is not None:
|
|
end_bi_idx = zs.bi_out.idx
|
|
break
|
|
return end_bi_idx
|