274 lines
12 KiB
Python
274 lines
12 KiB
Python
from ChanKLU import ChanKLU
|
|
from ChanEnum import Chan_MACD_STATE, Chan_MACDSEG_DIR, Chan_MACDHISTSET_DIR, Chan_MACDUNITTF_DIR, Chan_MACDUNITTF_TYPE
|
|
from ChanMACDSeg import ChanMACDSeg
|
|
from ChanMACDUnitTF import ChanMACDUnitTF
|
|
from ChanMACDHistSet import ChanMACDHistSet
|
|
|
|
class ChanMACD():
|
|
def __init__(self, klu_list: list[ChanKLU]):
|
|
self.klu_list = klu_list
|
|
self.seg_list = []
|
|
self.unittf_list = []
|
|
self.histset_list = []
|
|
# 状态标记列表
|
|
self.high_position_list = [] # 高位列表
|
|
self.high_empty_list = [] # 高位空列表
|
|
self.return_zero_list = [] # 归零轴列表
|
|
self.cross0_up_list = [] # 向上穿越零轴列表
|
|
self.cross0_down_list = [] # 向下穿越零轴列表
|
|
# 计算段 / UnitTF / HistSet 及状态标记
|
|
self.cal_macd_state()
|
|
self.get_klu_sd_list()
|
|
def get_klu_sd(self):
|
|
if self.klu_list:
|
|
sd = self.klu_list[-1].separate_div
|
|
if sd > 1:
|
|
print(self.klu_list[-1].time, sd)
|
|
return True
|
|
return False
|
|
def get_klu_sd_list(self):
|
|
sd_list = []
|
|
if self.klu_list:
|
|
for klu in self.klu_list:
|
|
hist = klu.macdhist
|
|
singal = False
|
|
if klu.pre and klu.next:
|
|
if klu.signal > 0:
|
|
signal = klu.pre.signal > klu.signal and klu.next.signal < klu.signal
|
|
else:
|
|
signal = klu.pre.signal < klu.signal and klu.next.signal > klu.signal
|
|
sd = klu.separate_div
|
|
if sd > 1 and ((hist > 0 and hist < 200) or (hist < 0 and hist > -200)):
|
|
sd_list.append(klu.time)
|
|
print(klu.time, sd)
|
|
return sd_list
|
|
def cal_macd_state(self):
|
|
last_seg = None
|
|
last_unittf = None
|
|
last_histset = None
|
|
last_klu = None
|
|
for klu in self.klu_list:
|
|
# initialise first histset
|
|
if klu.macd == 0 and klu.signal == 0 and klu.macdhist == 0:
|
|
continue
|
|
if last_histset is None:
|
|
if klu.macdhist > 0:
|
|
last_histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, None, Chan_MACDHISTSET_DIR.ABOVE)
|
|
self.histset_list.append(last_histset)
|
|
else:
|
|
last_histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, None, Chan_MACDHISTSET_DIR.UNDER)
|
|
self.histset_list.append(last_histset)
|
|
else:
|
|
# initialise first seg and unittf
|
|
if last_seg is None:
|
|
# create histset afterwards
|
|
if last_histset.histset_dir == Chan_MACDHISTSET_DIR.ABOVE:
|
|
if klu.macdhist > 0:
|
|
last_histset.add_klu(klu)
|
|
else:
|
|
histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.UNDER)
|
|
self.histset_list.append(histset)
|
|
last_histset.set_next(histset)
|
|
histset.set_pre(last_histset)
|
|
last_histset.set_end_klu(last_klu)
|
|
last_histset = histset
|
|
else:
|
|
if klu.macdhist < 0:
|
|
last_histset.add_klu(klu)
|
|
else:
|
|
histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.ABOVE)
|
|
self.histset_list.append(histset)
|
|
last_histset.set_next(histset)
|
|
histset.set_pre(last_histset)
|
|
last_histset.set_end_klu(last_klu)
|
|
last_histset = histset
|
|
if last_klu.signal >= 0 and klu.signal < 0:
|
|
last_unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, None, Chan_MACDUNITTF_DIR.UNDER, Chan_MACDUNITTF_TYPE.CROSS0, last_histset)
|
|
self.unittf_list.append(last_unittf)
|
|
last_seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, None, Chan_MACDSEG_DIR.UNDER, last_unittf)
|
|
self.seg_list.append(last_seg)
|
|
elif last_klu.signal <= 0 and klu.signal > 0:
|
|
last_unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, None, Chan_MACDUNITTF_DIR.ABOVE, Chan_MACDUNITTF_TYPE.CROSS0, last_histset)
|
|
self.unittf_list.append(last_unittf)
|
|
last_seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, None, Chan_MACDSEG_DIR.ABOVE, last_unittf)
|
|
self.seg_list.append(last_seg)
|
|
# after the first seg and unittf
|
|
else:
|
|
# create histset afterwards
|
|
if last_histset.histset_dir == Chan_MACDHISTSET_DIR.ABOVE:
|
|
if klu.macdhist > 0:
|
|
last_histset.add_klu(klu)
|
|
else:
|
|
histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.UNDER)
|
|
self.histset_list.append(histset)
|
|
last_histset.set_next(histset)
|
|
histset.set_pre(last_histset)
|
|
last_histset.set_end_klu(last_klu)
|
|
last_histset = histset
|
|
if last_unittf:
|
|
last_unittf.add_histset(last_histset)
|
|
else:
|
|
if klu.macdhist < 0:
|
|
last_histset.add_klu(klu)
|
|
else:
|
|
histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.ABOVE)
|
|
self.histset_list.append(histset)
|
|
last_histset.set_next(histset)
|
|
histset.set_pre(last_histset)
|
|
last_histset.set_end_klu(last_klu)
|
|
last_histset = histset
|
|
if last_unittf:
|
|
last_unittf.add_histset(last_histset)
|
|
if last_klu.signal >= 0 and klu.signal < 0:
|
|
last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.CROSS0)
|
|
unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, Chan_MACDUNITTF_DIR.UNDER, Chan_MACDUNITTF_TYPE.CROSS0, last_histset)
|
|
self.unittf_list.append(unittf)
|
|
last_unittf.set_next(unittf)
|
|
last_seg.set_end_klu(last_klu)
|
|
seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, last_seg, Chan_MACDSEG_DIR.UNDER, unittf)
|
|
self.seg_list.append(seg)
|
|
last_seg.set_next(seg)
|
|
last_seg = seg
|
|
last_unittf = unittf
|
|
elif last_klu.signal <= 0 and klu.signal > 0:
|
|
last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.CROSS0)
|
|
unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, Chan_MACDUNITTF_DIR.ABOVE, Chan_MACDUNITTF_TYPE.CROSS0, last_histset)
|
|
self.unittf_list.append(unittf)
|
|
last_unittf.set_next(unittf)
|
|
last_seg.set_end_klu(last_klu)
|
|
seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, last_seg, Chan_MACDSEG_DIR.ABOVE, unittf)
|
|
self.seg_list.append(seg)
|
|
last_seg.set_next(seg)
|
|
last_seg = seg
|
|
last_unittf = unittf
|
|
elif last_unittf.is_end and last_klu.macd < klu.macd and klu.macd > klu.signal:
|
|
unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, Chan_MACDUNITTF_DIR.ABOVE, Chan_MACDUNITTF_TYPE.NEAR0, last_histset)
|
|
self.unittf_list.append(unittf)
|
|
last_unittf.set_next(unittf)
|
|
last_seg.add_unittf(unittf)
|
|
last_unittf = unittf
|
|
last_seg.add_klu(klu)
|
|
else:
|
|
if not last_unittf.is_end:
|
|
last_unittf.add_klu(klu)
|
|
last_seg.add_klu(klu)
|
|
last_klu = klu
|
|
klu.cal_macd_state()
|
|
#print(klu.time, klu.macd_state, klu.continue_div, klu.separate_div, klu.macd, klu.signal, klu.macdhist, klu.ema24, klu.ema52, klu.close)
|
|
return self.klu_list
|
|
def cal_macd(self):
|
|
last_seg = None
|
|
last_unittf = None
|
|
last_histset = None
|
|
histset = None
|
|
last_klu = None
|
|
for klu in self.klu_list:
|
|
klu.cal_macd_state()
|
|
print(klu.time, klu.macd_state)
|
|
# 1) 只有当 MACD 已可用(非 UNKNOWN)时，才开始初始化段/单元
|
|
if last_seg is None:
|
|
if klu.macd_state != Chan_MACD_STATE.UNKNOWN:
|
|
# 初始化首个直方图集合（根据当前柱体正负）
|
|
if klu.macdhist >= 0:
|
|
histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, None, Chan_MACDHISTSET_DIR.ABOVE)
|
|
else:
|
|
histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, None, Chan_MACDHISTSET_DIR.UNDER)
|
|
self.histset_list.append(histset)
|
|
last_histset = histset
|
|
|
|
# 初始化首段
|
|
seg_dir = Chan_MACDSEG_DIR.ABOVE if klu.signal >= 0 else Chan_MACDSEG_DIR.UNDER
|
|
seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, None, seg_dir, last_unittf)
|
|
self.seg_list.append(seg)
|
|
last_seg = seg
|
|
|
|
# 初始化首个UnitTF
|
|
unittf_dir = Chan_MACDUNITTF_DIR.ABOVE if klu.signal >= 0 else Chan_MACDUNITTF_DIR.UNDER
|
|
unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, None, unittf_dir, Chan_MACDUNITTF_TYPE.START, histset)
|
|
self.unittf_list.append(unittf)
|
|
last_unittf = unittf
|
|
last_seg.add_unittf(unittf)
|
|
# 未就绪则继续等下一根；已就绪亦已完成首个结构初始化，继续下一根
|
|
last_klu = klu
|
|
continue
|
|
# 3) 直方图集合（基于当前 unittf）
|
|
if klu.macdhist >= 0:
|
|
if last_histset and last_histset.histset_dir == Chan_MACDHISTSET_DIR.ABOVE:
|
|
last_histset.add_klu(klu)
|
|
else:
|
|
# 结束旧 histset（以前一根结束更合理）
|
|
if last_histset and last_klu:
|
|
last_histset.set_end_klu(last_klu)
|
|
histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.ABOVE if klu.macdhist >= 0 else Chan_MACDHISTSET_DIR.UNDER)
|
|
self.histset_list.append(histset)
|
|
if last_histset:
|
|
last_histset.set_next(histset)
|
|
last_histset = histset
|
|
if last_unittf:
|
|
last_unittf.add_histset(histset)
|
|
else:
|
|
if last_histset and last_histset.histset_dir == Chan_MACDHISTSET_DIR.UNDER:
|
|
last_histset.add_klu(klu)
|
|
else:
|
|
# 结束旧 histset（以前一根结束更合理）
|
|
if last_histset and last_klu:
|
|
last_histset.set_end_klu(last_klu)
|
|
histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.ABOVE if klu.macdhist >= 0 else Chan_MACDHISTSET_DIR.UNDER)
|
|
self.histset_list.append(histset)
|
|
if last_histset:
|
|
last_histset.set_next(histset)
|
|
last_histset = histset
|
|
if last_unittf:
|
|
last_unittf.add_histset(histset)
|
|
# 2) 过零切段（使用KLU中的穿越状态）
|
|
if (klu.macd_state == Chan_MACD_STATE.CROSS0_UP or
|
|
klu.macd_state == Chan_MACD_STATE.CROSS0_DOWN):
|
|
# 结束旧 unittf
|
|
last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.CROSS0)
|
|
# 新的单位时间周期
|
|
new_dir = Chan_MACDUNITTF_DIR.ABOVE if klu.signal >= 0 else Chan_MACDUNITTF_DIR.UNDER
|
|
unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.CROSS0, histset)
|
|
self.unittf_list.append(unittf)
|
|
last_unittf.set_next(unittf)
|
|
last_unittf = unittf
|
|
# 收尾旧段
|
|
last_seg.set_end_klu(last_klu)
|
|
# 新段方向取反
|
|
new_dir = Chan_MACDSEG_DIR.UNDER if last_seg.seg_dir == Chan_MACDSEG_DIR.ABOVE else Chan_MACDSEG_DIR.ABOVE
|
|
seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, last_seg, new_dir, last_unittf)
|
|
self.seg_list.append(seg)
|
|
last_seg.set_next(seg)
|
|
last_seg = seg
|
|
last_seg.add_unittf(unittf)
|
|
else:
|
|
# 4) UnitTF 状态机：用黄线Signal的归零轴
|
|
if last_klu.macd_state == Chan_MACD_STATE.NEAR0 and last_unittf.div_count > 1:
|
|
#print(klu.time, klu.macd_state)
|
|
if klu.macd_state == Chan_MACD_STATE.RZ_UP:
|
|
last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.NEAR0)
|
|
new_dir = Chan_MACDUNITTF_DIR.ABOVE if klu.signal >= 0 else Chan_MACDUNITTF_DIR.UNDER
|
|
unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.NEAR0, histset)
|
|
self.unittf_list.append(unittf)
|
|
last_unittf.set_next(unittf)
|
|
last_unittf = unittf
|
|
last_seg.add_unittf(unittf)
|
|
elif klu.macd_state == Chan_MACD_STATE.RZ_DOWN:
|
|
last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.NEAR0)
|
|
new_dir = Chan_MACDUNITTF_DIR.ABOVE if klu.signal >= 0 else Chan_MACDUNITTF_DIR.UNDER
|
|
unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.NEAR0, histset)
|
|
self.unittf_list.append(unittf)
|
|
last_unittf.set_next(unittf)
|
|
last_unittf = unittf
|
|
last_seg.add_unittf(unittf)
|
|
else:
|
|
last_unittf.add_klu(klu)
|
|
last_seg.add_klu(klu)
|
|
else:
|
|
last_unittf.add_klu(klu)
|
|
last_seg.add_klu(klu)
|
|
|
|
last_klu = klu
|
|
last_histset.set_end_klu(last_klu)
|
|
last_unittf.set_end_klu(last_klu, None)
|
|
last_seg.set_end_klu(last_klu)
|
|
return self.klu_list |