183 lines
7.2 KiB
Python
183 lines
7.2 KiB
Python
from typing import List
|
|
|
|
from Bi.BiConfig import CBiConfig
|
|
from BuySellPoint.BSPointConfig import CBSPointConfig
|
|
from Common.CEnum import TREND_TYPE
|
|
from Common.ChanException import CChanException, ErrCode
|
|
from Common.func_util import _parse_inf
|
|
from Math.BOLL import BollModel
|
|
from Math.Demark import CDemarkEngine
|
|
from Math.KDJ import KDJ
|
|
from Math.MACD import CMACD
|
|
from Math.RSI import RSI
|
|
from Math.TrendModel import CTrendModel
|
|
from Seg.SegConfig import CSegConfig
|
|
from ZS.ZSConfig import CZSConfig
|
|
|
|
|
|
class CChanConfig:
|
|
def __init__(self, conf=None):
|
|
if conf is None:
|
|
conf = {}
|
|
conf = ConfigWithCheck(conf)
|
|
self.bi_conf = CBiConfig(
|
|
bi_algo=conf.get("bi_algo", "normal"),
|
|
is_strict=conf.get("bi_strict", True),
|
|
bi_fx_check=conf.get("bi_fx_check", "strict"),
|
|
gap_as_kl=conf.get("gap_as_kl", False),
|
|
bi_end_is_peak=conf.get('bi_end_is_peak', True),
|
|
bi_allow_sub_peak=conf.get("bi_allow_sub_peak", True),
|
|
)
|
|
self.seg_conf = CSegConfig(
|
|
seg_algo=conf.get("seg_algo", "chan"),
|
|
left_method=conf.get("left_seg_method", "peak"),
|
|
)
|
|
self.zs_conf = CZSConfig(
|
|
need_combine=conf.get("zs_combine", True),
|
|
zs_combine_mode=conf.get("zs_combine_mode", "zs"),
|
|
one_bi_zs=conf.get("one_bi_zs", False),
|
|
zs_algo=conf.get("zs_algo", "normal"),
|
|
)
|
|
|
|
self.trigger_step = conf.get("trigger_step", False)
|
|
self.skip_step = conf.get("skip_step", 0)
|
|
|
|
self.kl_data_check = conf.get("kl_data_check", True)
|
|
self.max_kl_misalgin_cnt = conf.get("max_kl_misalgin_cnt", 2)
|
|
self.max_kl_inconsistent_cnt = conf.get("max_kl_inconsistent_cnt", 5)
|
|
self.auto_skip_illegal_sub_lv = conf.get("auto_skip_illegal_sub_lv", False)
|
|
self.print_warning = conf.get("print_warning", True)
|
|
self.print_err_time = conf.get("print_err_time", False)
|
|
|
|
self.mean_metrics: List[int] = conf.get("mean_metrics", [])
|
|
self.trend_metrics: List[int] = conf.get("trend_metrics", [])
|
|
self.macd_config = conf.get("macd", {"fast": 12, "slow": 26, "signal": 9})
|
|
self.cal_demark = conf.get("cal_demark", False)
|
|
self.cal_rsi = conf.get("cal_rsi", False)
|
|
self.cal_kdj = conf.get("cal_kdj", False)
|
|
self.rsi_cycle = conf.get("rsi_cycle", 14)
|
|
self.kdj_cycle = conf.get("kdj_cycle", 9)
|
|
self.demark_config = conf.get("demark", {
|
|
'demark_len': 9,
|
|
'setup_bias': 4,
|
|
'countdown_bias': 2,
|
|
'max_countdown': 13,
|
|
'tiaokong_st': True,
|
|
'setup_cmp2close': True,
|
|
'countdown_cmp2close': True,
|
|
})
|
|
self.boll_n = conf.get("boll_n", 20)
|
|
|
|
self.set_bsp_config(conf)
|
|
|
|
conf.check()
|
|
|
|
def GetMetricModel(self):
|
|
res: List[CMACD | CTrendModel | BollModel | CDemarkEngine | RSI | KDJ] = [
|
|
CMACD(
|
|
fastperiod=self.macd_config['fast'],
|
|
slowperiod=self.macd_config['slow'],
|
|
signalperiod=self.macd_config['signal'],
|
|
)
|
|
]
|
|
res.extend(CTrendModel(TREND_TYPE.MEAN, mean_T) for mean_T in self.mean_metrics)
|
|
|
|
for trend_T in self.trend_metrics:
|
|
res.append(CTrendModel(TREND_TYPE.MAX, trend_T))
|
|
res.append(CTrendModel(TREND_TYPE.MIN, trend_T))
|
|
res.append(BollModel(self.boll_n))
|
|
if self.cal_demark:
|
|
res.append(CDemarkEngine(
|
|
demark_len=self.demark_config['demark_len'],
|
|
setup_bias=self.demark_config['setup_bias'],
|
|
countdown_bias=self.demark_config['countdown_bias'],
|
|
max_countdown=self.demark_config['max_countdown'],
|
|
tiaokong_st=self.demark_config['tiaokong_st'],
|
|
setup_cmp2close=self.demark_config['setup_cmp2close'],
|
|
countdown_cmp2close=self.demark_config['countdown_cmp2close'],
|
|
))
|
|
if self.cal_rsi:
|
|
res.append(RSI(self.rsi_cycle))
|
|
if self.cal_kdj:
|
|
res.append(KDJ(self.kdj_cycle))
|
|
return res
|
|
|
|
def set_bsp_config(self, conf):
|
|
para_dict = {
|
|
"divergence_rate": float("inf"),
|
|
"min_zs_cnt": 1,
|
|
"bsp1_only_multibi_zs": True,
|
|
"max_bs2_rate": 0.9999,
|
|
"macd_algo": "peak",
|
|
"bs1_peak": True,
|
|
"bs_type": "1,1p,2,2s,3a,3b",
|
|
"bsp2_follow_1": True,
|
|
"bsp3_follow_1": True,
|
|
"bsp3_peak": False,
|
|
"bsp2s_follow_2": False,
|
|
"max_bsp2s_lv": None,
|
|
"strict_bsp3": False,
|
|
}
|
|
args = {para: conf.get(para, default_value) for para, default_value in para_dict.items()}
|
|
self.bs_point_conf = CBSPointConfig(**args)
|
|
|
|
self.seg_bs_point_conf = CBSPointConfig(**args)
|
|
self.seg_bs_point_conf.b_conf.set("macd_algo", "slope")
|
|
self.seg_bs_point_conf.s_conf.set("macd_algo", "slope")
|
|
self.seg_bs_point_conf.b_conf.set("bsp1_only_multibi_zs", False)
|
|
self.seg_bs_point_conf.s_conf.set("bsp1_only_multibi_zs", False)
|
|
|
|
for k, v in conf.items():
|
|
if isinstance(v, str):
|
|
v = f'"{v}"'
|
|
v = _parse_inf(v)
|
|
if k.endswith("-buy"):
|
|
prop = k.replace("-buy", "")
|
|
exec(f"self.bs_point_conf.b_conf.set('{prop}', {v})")
|
|
elif k.endswith("-sell"):
|
|
prop = k.replace("-sell", "")
|
|
exec(f"self.bs_point_conf.s_conf.set('{prop}', {v})")
|
|
elif k.endswith("-segbuy"):
|
|
prop = k.replace("-segbuy", "")
|
|
exec(f"self.seg_bs_point_conf.b_conf.set('{prop}', {v})")
|
|
elif k.endswith("-segsell"):
|
|
prop = k.replace("-segsell", "")
|
|
exec(f"self.seg_bs_point_conf.s_conf.set('{prop}', {v})")
|
|
elif k.endswith("-seg"):
|
|
prop = k.replace("-seg", "")
|
|
exec(f"self.seg_bs_point_conf.b_conf.set('{prop}', {v})")
|
|
exec(f"self.seg_bs_point_conf.s_conf.set('{prop}', {v})")
|
|
elif k in args:
|
|
exec(f"self.bs_point_conf.b_conf.set({k}, {v})")
|
|
exec(f"self.bs_point_conf.s_conf.set({k}, {v})")
|
|
else:
|
|
raise CChanException(f"unknown para = {k}", ErrCode.PARA_ERROR)
|
|
self.bs_point_conf.b_conf.parse_target_type()
|
|
self.bs_point_conf.s_conf.parse_target_type()
|
|
self.seg_bs_point_conf.b_conf.parse_target_type()
|
|
self.seg_bs_point_conf.s_conf.parse_target_type()
|
|
|
|
|
|
class ConfigWithCheck:
|
|
def __init__(self, conf):
|
|
self.conf = conf
|
|
|
|
def get(self, k, default_value=None):
|
|
res = self.conf.get(k, default_value)
|
|
if k in self.conf:
|
|
del self.conf[k]
|
|
return res
|
|
|
|
def items(self):
|
|
visit_keys = set()
|
|
for k, v in self.conf.items():
|
|
yield k, v
|
|
visit_keys.add(k)
|
|
for k in visit_keys:
|
|
del self.conf[k]
|
|
|
|
def check(self):
|
|
if len(self.conf) > 0:
|
|
invalid_key_lst = ",".join(list(self.conf.keys()))
|
|
raise CChanException(f"invalid CChanConfig: {invalid_key_lst}", ErrCode.PARA_ERROR)
|