jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改了线段中枢逻辑

Browse Source
This commit is contained in:
jackyu66git
2026-03-11 03:08:10 +08:00
parent 1c099be23e
commit 5226c551d5
19 changed files with 3386 additions and 324 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TF_DF.py
+216 -176
View File
@@ -571,7 +571,27 @@ class TF_DF():
last_klu = None
macd = ChanMACD(klu_list)
klu_list = macd.cal_macd_state()
ema_up_list = []
ema_down_list = []
ema_up_count = 0
ema_down_count = 0
last_klu = None
for klu in klu_list:
ema = klu.ema52
last_ema = last_klu.ema52 if last_klu else 0
if klu.close >= ema:
ema_up_count += 1
elif klu.close < ema:
ema_down_count += 1
if last_klu and last_klu.close >= last_ema and klu.close < ema:
ema_up_list.append(ema_up_count)
#print(last_klu.time, ema_up_count, "UP END")
ema_up_count = 0
elif last_klu and last_klu.close < last_ema and klu.close >= ema:
ema_down_list.append(ema_down_count)
#print(last_klu.time, ema_down_count, "DOWN END")
ema_down_count = 0
last_klu = klu
if len(klc_list) > 0:
last_klc = klc_list[-1]
if klu.exception:
@@ -609,6 +629,7 @@ class TF_DF():
klc_list.append(klc)
last_klu = klu
klc_list = self.cal_trend(klc_list)
#print(ema52_up_list, ema52_down_list)
return klc_list
def get_seg_list(self, bi_list):
@@ -965,7 +986,7 @@ class TF_DF():
if last_top.high > klc.high:
bi_list[-1].add_klc(klc)
klc.set_bi(bi_list[-1])
klc.set_klc_fx_type(Chan_KLC_FX.TOP3)
#klc.set_klc_fx_type(Chan_KLC_FX.TOP3)
#print(klc.end_time, klc.fx, "二类卖点Sell 1")
else:
# A new top found
@@ -979,7 +1000,7 @@ class TF_DF():
klc.set_bi(bi_list[-1])
# 不满足结合律的分型
else:
#klc.set_klc_fx_type(Chan_KLC_FX.TOP0)
klc.set_klc_fx_type(Chan_KLC_FX.TOP0)
if last_bottom.index + bi_klc_min > klc.index:
if last_top.high > klc.high:
#print(klc.start_time, klc.fx, "二类卖点Sell 1")
@@ -1082,7 +1103,7 @@ class TF_DF():
if last_bottom.low < klc.low:
bi_list[-1].add_klc(klc)
klc.set_bi(bi_list[-1])
klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM3)
#klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM3)
#print(last_bottom.start_time, last_bottom.end_time, "--------------------------------1")
#print(klc.end_time, klc.fx, "二类买点Buy 1")
else:
@@ -1096,7 +1117,7 @@ class TF_DF():
klc.set_bi(bi_list[-1])
# 不满足结合律的分型
else:
#klc.set_klc_fx_type(Chan_KLC_FX.TOP0)
klc.set_klc_fx_type(Chan_KLC_FX.TOP0)
if last_top.index + bi_klc_min > klc.index:
if last_bottom.low < klc.low:
#print(klc.end_time, klc.fx, "中枢买点Buy 1")
@@ -1849,180 +1870,199 @@ class TF_DF():
def calculate_zs(self, bi_list, seg_list):
return self.get_zs_list(bi_list, seg_list)
def get_zs_list(self, bi_list, seg_list):
"""
根据缠论线段中枢定义计算中枢
从第4根线段开始（索引3），每3根线段为一组检查
后一个中枢比前一个高 -> 上涨中枢，以下跌开始、以下跌结束
后一个中枢比前一个低 -> 下跌中枢，以上涨开始、以上涨结束
中枢可以扩展到5根、7根...
"""
zs_list = []
bsp_list = []
if len(seg_list) > 3:
last_zs = None
first_bi_out = None
in_again = False
bi_out_count = 0
zs_count = 0
for seg in seg_list:
# No zs or Last ZS is completed
if len(zs_list) == 0 or (last_zs and last_zs.is_sure):
# Has three completed segments
if seg.next and seg.next.next:
if seg.next.next.is_sure:
zg = min(seg.high, seg.next.high, seg.next.next.high)
zd = max(seg.low, seg.next.low, seg.next.next.low)
gg = max(seg.high, seg.next.high, seg.next.next.high)
dd = min(seg.low, seg.next.low, seg.next.next.low)
ddir = None
if last_zs:
if zg < last_zs.zd:
ddir = Chan_ZS_DIR.DOWN
else:
if zd > last_zs.zg:
ddir = Chan_ZS_DIR.UP
else:
ddir = None
else:
if seg.dir == Chan_SEG_DIR.UP:
ddir = Chan_ZS_DIR.DOWN
else:
ddir = Chan_ZS_DIR.UP
if (seg.dir == Chan_SEG_DIR.DOWN and ddir == Chan_ZS_DIR.DOWN) or (seg.dir == Chan_SEG_DIR.UP and ddir == Chan_ZS_DIR.UP):
ddir = None
if ddir and zg > zd:
# New ZS
zs = ChanZS(seg, len(zs_list), ddir)
zs.set_zg(zg)
zs.set_zd(zd)
zs.set_gg(gg)
zs.set_dd(dd)
if last_zs:
last_zs.set_next(zs)
zs.set_pre(last_zs)
zs_list.append(zs)
if last_zs and last_zs.dir == zs.dir:
zs_count += 1
else:
zs_count = 1
last_zs = zs
# Last ZS is not completed
else:
# Last ZS is not completed
if last_zs and not last_zs.is_sure:
if first_bi_out:
# SEG is not in ZS
if seg.is_sure:
if ((seg.low > last_zs.zg and seg.high > last_zs.zg) or (seg.high < last_zs.zd and seg.low < last_zs.zd)):
last_zs.set_end_klc(seg.pre.end_bi.end_klc, seg.sure_time, bi_out_count, seg.pre)
bi_out_count = 0
#print(seg.start_bi.start_klc.start_time)
first_bi_out = None
# Last ZS is completed and look for new ZS
if seg.next and seg.next.next:
if seg.next.next.is_sure:
zg = min(seg.high, seg.next.high, seg.next.next.high)
zd = max(seg.low, seg.next.low, seg.next.next.low)
gg = max(seg.high, seg.next.high, seg.next.next.high)
dd = min(seg.low, seg.next.low, seg.next.next.low)
ddir = None
if last_zs:
if zg < last_zs.zd:
ddir = Chan_ZS_DIR.DOWN
else:
if zd > last_zs.zg:
ddir = Chan_ZS_DIR.UP
else:
ddir = None
else:
if seg.dir == Chan_SEG_DIR.UP:
ddir = Chan_ZS_DIR.DOWN
else:
ddir = Chan_ZS_DIR.UP
if (seg.dir == Chan_SEG_DIR.DOWN and ddir == Chan_ZS_DIR.DOWN) or (seg.dir == Chan_SEG_DIR.UP and ddir == Chan_ZS_DIR.UP):
ddir = None
if ddir and zg > zd:
# New ZS
zs = ChanZS(seg, len(zs_list), ddir)
zs.set_zg(zg)
zs.set_zd(zd)
zs.set_gg(gg)
zs.set_dd(dd)
last_zs.set_next(zs)
zs.set_pre(last_zs)
zs_list.append(zs)
if last_zs and last_zs.dir == zs.dir:
zs_count += 1
else:
zs_count = 1
last_zs = zs
# Last SEG is in ZS
else:
# SEG is inside ZS
if seg.end_bi:
for index in range(seg.start_bi.index, seg.end_bi.index+1):
bi = bi_list[index]
if (bi.high >= last_zs.zd and bi.high <= last_zs.zg) or (bi.low >= last_zs.zd and bi.low <= last_zs.zg) or (bi.high >= last_zs.zg and bi.low <= last_zs.zd):
in_again = True
last_zs.set_bi_out(None, None)
last_zs.set_last_bi_in(None)
last_zs.set_end_seg(None)
first_bi_out = None
#print("Bi in again 3", bi.start_klc.start_time)
if in_again and (bi.low > last_zs.zg or bi.high < last_zs.zd):
last_zs.set_bi_out(bi, seg)
last_zs.set_last_bi_in(bi_list[index - 1])
#last_zs.set_end_seg(seg.next.next)
bi_out_count += 1
first_bi_out = bi
if (bi.dir == Chan_BI_DIR.UP and seg.dir == Chan_SEG_DIR.DOWN) or (bi.dir == Chan_BI_DIR.DOWN and seg.dir == Chan_SEG_DIR.UP):
bsp = ChanBSP(first_bi_out, len(bsp_list), Chan_BSP_TYPE.B3, Chan_BSP_DIR.BUY if first_bi_out.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, first_bi_out.sure_time, zs_count, zs, seg)
bsp_list.append(bsp)
#print("First bi out 3", first_bi_out.start_klc.start_time)
in_again = False
""""
if first_bi_out:
if seg.dir == Chan_SEG_DIR.UP and bi.dir == Chan_BI_DIR.UP:
#print(bi.start_klc.start_time, bi.high, seg.high)
if bi.high == seg.high:
bsp = ChanBSP(bi, len(bsp_list), Chan_BSP_TYPE.T3E, Chan_BSP_DIR.SELL if bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.BUY, bi.sure_time, zs_count, zs, seg)
bsp_list.append(bsp)
else:
if seg.dir == Chan_SEG_DIR.DOWN and bi.dir == Chan_BI_DIR.DOWN:
if bi.low == seg.low:
bsp = ChanBSP(bi, len(bsp_list), Chan_BSP_TYPE.T3E, Chan_BSP_DIR.BUY if bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, bi.sure_time, zs_count, zs, seg)
bsp_list.append(bsp)
"""
else:
# SEG in ZS and not out and find first bi out
if seg.end_bi:
for index in range(seg.start_bi.index, seg.end_bi.index+1):
bi = bi_list[index]
if (bi.high >= last_zs.zd and bi.high <= last_zs.zg) or (bi.low >= last_zs.zd and bi.low <= last_zs.zg) or (bi.high >= last_zs.zg and bi.low <= last_zs.zd):
in_again = True
last_zs.set_bi_out(None, None)
last_zs.set_last_bi_in(None)
last_zs.set_end_seg(None)
first_bi_out = None
#print("Bi in again 4", bi.start_klc.start_time)
if in_again and (bi.low > last_zs.zg or bi.high < last_zs.zd):
last_zs.set_bi_out(bi, seg)
last_zs.set_last_bi_in(bi_list[index - 1])
#last_zs.set_end_seg(seg.next.next)
bi_out_count += 1
first_bi_out = bi
if (bi.dir == Chan_BI_DIR.UP and seg.dir == Chan_SEG_DIR.DOWN) or (bi.dir == Chan_BI_DIR.DOWN and seg.dir == Chan_SEG_DIR.UP):
bsp = ChanBSP(first_bi_out, len(bsp_list), Chan_BSP_TYPE.B3, Chan_BSP_DIR.BUY if first_bi_out.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, first_bi_out.sure_time, zs_count, zs, seg)
bsp_list.append(bsp)
#print("First bi out 4", first_bi_out.start_klc.start_time)
in_again = False
if first_bi_out:
if seg.dir == Chan_SEG_DIR.UP and bi.dir == Chan_BI_DIR.UP:
#print(bi.start_klc.start_time, bi.high, seg.high)
if bi.high == seg.high:
bsp = ChanBSP(bi, len(bsp_list), Chan_BSP_TYPE.S3, Chan_BSP_DIR.SELL if bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.BUY, bi.sure_time, zs_count, zs, seg)
bsp_list.append(bsp)
else:
if seg.dir == Chan_SEG_DIR.DOWN and bi.dir == Chan_BI_DIR.DOWN:
if bi.low == seg.low:
bsp = ChanBSP(bi, len(bsp_list), Chan_BSP_TYPE.B3, Chan_BSP_DIR.BUY if bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, bi.sure_time, zs_count, zs, seg)
bsp_list.append(bsp)
#self.print_zs(zs_list)
return zs_list
if len(seg_list) < 3:
return zs_list
last_zs = None
# 从第4根线段开始（索引3），每3根为一组
start_idx = 3
while start_idx < len(seg_list):
# 取连续3个线段
if start_idx + 2 >= len(seg_list):
break
seg1 = seg_list[start_idx]
seg2 = seg_list[start_idx + 1]
seg3 = seg_list[start_idx + 2]
# 三个线段都必须是已确认的
if not (seg1.is_sure and seg2.is_sure and seg3.is_sure):
start_idx += 1
continue
# 计算这3个线段的中枢区间
zg = min(seg1.high, seg2.high, seg3.high)
zd = max(seg1.low, seg2.low, seg3.low)
if zg <= zd:
start_idx += 1
continue
# 判断中枢类型
# 上涨中枢：以下跌开始、以下跌结束（下跌+上涨+下跌）
# 下跌中枢：以上涨开始、以上涨结束（上涨+下跌+上涨）
if last_zs is None:
# 第一个中枢
if seg1.dir == Chan_SEG_DIR.DOWN:
# 下跌开始 -> 上涨中枢
zs_dir = Chan_ZS_DIR.DOWN
# 验证模式：下跌+上涨+下跌
valid = (seg2.dir == Chan_SEG_DIR.UP and seg3.dir == Chan_SEG_DIR.DOWN)
else:
# 上涨开始 -> 下跌中枢
zs_dir = Chan_ZS_DIR.UP
# 验证模式：上涨+下跌+上涨
valid = (seg2.dir == Chan_SEG_DIR.DOWN and seg3.dir == Chan_SEG_DIR.UP)
else:
# 根据与前一个中枢的高低比较判断
if zg > last_zs.zg:
# 上涨中枢：以下跌开始、以下跌结束
zs_dir = Chan_ZS_DIR.DOWN
valid = (seg1.dir == Chan_SEG_DIR.DOWN and seg2.dir == Chan_SEG_DIR.UP and seg3.dir == Chan_SEG_DIR.DOWN)
else:
# 下跌中枢：以上涨开始、以上涨结束
zs_dir = Chan_ZS_DIR.UP
valid = (seg1.dir == Chan_SEG_DIR.UP and seg2.dir == Chan_SEG_DIR.DOWN and seg3.dir == Chan_SEG_DIR.UP)
# 验证是否有效
if not valid:
start_idx += 1
continue
# 检查是否与前一个中枢重叠
if last_zs:
# 判断是否有重叠
overlap = (zg >= last_zs.zd and zd <= last_zs.zg)
if overlap:
# 有重叠，扩展中枢到5根、7根...
# 继续往后检查是否有更多线段与中枢重叠
seg_count = 3
cur_idx = start_idx + 3
while cur_idx < len(seg_list):
next_seg = seg_list[cur_idx]
if not next_seg.is_sure:
break
# 计算包含新线段后的区间
segs = seg_list[start_idx:start_idx + seg_count + 1]
segs.append(next_seg)
seg_highs = [s.high for s in segs]
seg_lows = [s.low for s in segs]
new_zg = min(seg_highs)
new_zd = max(seg_lows)
if new_zg > new_zd:
# 新线段与中枢重叠，扩展
seg_count += 1
cur_idx += 1
else:
break
if seg_list[start_idx].dir != seg_list[start_idx + seg_count].dir:
seg_count -= 1
print(seg_count)
# 更新中枢区间为扩展后的范围
segs_for_zs = seg_list[start_idx:start_idx + seg_count]
seg_highs = [s.high for s in segs_for_zs]
seg_lows = [s.low for s in segs_for_zs]
last_zs.set_zg(min(seg_highs))
last_zs.set_zd(max(seg_lows))
last_zs.set_gg(max(seg_highs))
last_zs.set_dd(min(seg_lows))
last_zs.seg_list = segs_for_zs
# 更新结束时间
last_seg = segs_for_zs[-1]
if last_seg.end_bi:
last_zs.set_end_klc(last_seg.end_bi.end_klc, last_seg.sure_time, 0, last_seg)
last_zs.set_end_seg(last_seg)
# 跳过已处理的线段
start_idx += 3
continue
else:
# 没有重叠，创建新中枢
# 先确认前一个中枢 - 使用前一个中枢本身的最后一个线段
if last_zs.seg_list and len(last_zs.seg_list) > 0:
prev_zs_last_seg = last_zs.seg_list[-1]
if prev_zs_last_seg.end_bi:
last_zs.set_end_klc(prev_zs_last_seg.end_bi.end_klc, prev_zs_last_seg.sure_time, 0, prev_zs_last_seg)
last_zs.set_end_seg(prev_zs_last_seg)
last_zs.is_sure = True
# 创建新中枢
gg = max(seg1.high, seg2.high, seg3.high)
dd = min(seg1.low, seg2.low, seg3.low)
zs = ChanZS(seg1, len(zs_list), zs_dir)
zs.set_zg(zg)
zs.set_zd(zd)
zs.set_gg(gg)
zs.set_dd(dd)
zs.set_end_klc(seg3.end_bi.end_klc, seg3.sure_time, 0, seg3)
zs.set_end_seg(seg3)
zs.is_sure = False
zs.seg_list = [seg1, seg2, seg3]
if last_zs:
last_zs.set_next(zs)
zs.set_pre(last_zs)
zs_list.append(zs)
last_zs = zs
# 移动到下一组
start_idx += 3
# 处理最后一个未确认的中枢 - 不自动扩展，保持未完成状态
if last_zs and not last_zs.is_sure:
# 获取中枢最后一个线段的索引
if last_zs.seg_list and len(last_zs.seg_list) > 0:
last_seg_of_zs = last_zs.seg_list[-1]
# 找到这个线段在seg_list中的索引
last_seg_idx = -1
for i, seg in enumerate(seg_list):
if seg == last_seg_of_zs:
last_seg_idx = i
break
# 从中枢最后一个线段之后检查是否有离开
has_leave = False
if last_seg_idx >= 0 and last_seg_idx + 1 < len(seg_list):
for i in range(last_seg_idx + 1, len(seg_list)):
seg = seg_list[i]
if seg.is_sure:
# 检查是否离开中枢
leave = (seg.low > last_zs.zg and seg.high > last_zs.zg) or \
(seg.high < last_zs.zd and seg.low < last_zs.zd)
if leave:
has_leave = True
break
if not has_leave:
# 没有离开，保持未完成状态
pass
else:
# 有离开，确认中枢
if last_seg_of_zs.end_bi:
last_zs.set_end_klc(last_seg_of_zs.end_bi.end_klc, last_seg_of_zs.sure_time, 0, last_seg_of_zs)
last_zs.set_end_seg(last_seg_of_zs)
last_zs.is_sure = True
return zs_list
def get_klu_list(self, dataframe):
klu_list = self.get_kl_data(dataframe)
#klu_list = self.cal_klu_pattern(klu_list)