jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
32c18f020143b06213877409f673a9784887bcbb
Chan/chanlun/pipeline/builders/zs.py
T
UbuntuandCursor 206b27fe72 refactor: 以自实现指标替换 talib 与 technical 依赖 
chanlun/indicators/ta.py 接口兼容 talib.abstract，实现代码实际用到的
SMA/MA/EMA/RSI/ATR/MACD/BBANDS；chanlun/pipeline/resample.py 替代
technical.util.resample_to_interval。调用点只改 import，逻辑未动。

暖机长度与平滑种子按 TA-Lib 的约定实现，差一根 K 线就会让下游所有
笔/线段/中枢整体位移。其中 MACD 需特别处理：TA-Lib 让快慢两条 EMA
在同一根 K 线出首值，因而快线的种子取 x[slow-fast:slow] 的均值，而非
从 fastperiod-1 一路递推——两者在百元价位上相差约 0.17。

BBANDS 是有意的分歧：TA-Lib 用 sumsq/n - mean² 求方差，短窗口远离零
时灾难性抵消（timeperiod=2 误差 8.7e-7），本实现用 rolling std，对 50
位精度基准误差为 0。项目实际使用的周期两者一致到 1e-10。

顺带清理 12 个文件中 16 处从未调用的 talib/technical 导入。

验证：9440 组随机对拨；真实 K 线端到端比对 add_indicators 全部 33 个
指标列，NaN 模式一致、MACD 柱符号 100% 相同；屏蔽两个包后 60 个模块
均可导入。新增 test_ta_compat.py 将输出逐 bar 钉在 TA-Lib 上，但该文件
在 TA-Lib 缺失时静默跳过，改动 ta.py 需在装有 TA-Lib 的环境复跑。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-27 02:01:43 +08:00

703 lines
20 KiB
Python
Raw Blame History

"""TF_DF builder mixin — 由 split_tfdf_builders 自动生成，逻辑与原 TF_DF 一致。"""
from __future__ import annotations
from datetime import timedelta
from decimal import Decimal
import numpy as np
import pandas as pd
from pandas import DataFrame
from chanlun.core.ChanBI import ChanBI
from chanlun.core.ChanBIZS import ChanBIZS
from chanlun.core.ChanBSP import ChanBSP
from chanlun.core.ChanEnum import (
Chan_BI_DIR,
Chan_BSP_DIR,
Chan_BSP_TYPE,
Chan_FX_TYPE,
Chan_K_DIR,
Chan_KLC_FX,
Chan_KLC_STATE,
Chan_KLINE_DIR,
Chan_KLU_PATTERN,
Chan_PRICE_TREND,
Chan_SEG_DIR,
Chan_ZS_DIR,
)
from chanlun.core.ChanKLC import ChanKLC
from chanlun.core.ChanKLU import ChanKLU
from chanlun.core.ChanSBI import ChanSBI
from chanlun.core.ChanSEG import ChanSEG
from chanlun.core.ChanZS import ChanZS, ChanZS_Big
from chanlun.indicators.ChanMACD import ChanMACD
class ZsBuilderMixin:
def get_zs_state(self, df):
bi_list = self.cal_bi_list(self.get_klc_list(self.get_kl_data(df)))
seg_list = self.get_seg_list(bi_list)
zs_list = self.calculate_zs(seg_list)
for zs in zs_list:
last_zs = zs
return zs_list
def cal_bi_zs(self, seg_list):
bi_zs_list = []
for seg in seg_list:
zs_list = seg.cal_bi_zs()
if len(zs_list) > 0:
bi_zs_list = list(bi_zs_list) + list(zs_list)
return bi_zs_list
# 跨段不相连的中枢
def cal_bi_zs_list(self, bi_list):
"""
根据缠论笔中枢定义计算中枢（参照 get_zs_list 线段中枢判断规则）
从第4根笔开始（索引3），每3根笔为一组检查
上涨中枢：后中枢 zd > 前中枢 zg（不重叠上移）
下跌中枢：后中枢 zg < 前中枢 zd（不重叠下移）
中枢可按两笔一组继续扩展到5根、7根...
"""
bi_zs_list = []
if len(bi_list) < 3:
return bi_zs_list
last_zs = None
start_idx = 3
while start_idx < len(bi_list):
if start_idx + 2 >= len(bi_list):
break
bi1 = bi_list[start_idx]
bi2 = bi_list[start_idx + 1]
bi3 = bi_list[start_idx + 2]
if not (bi1.is_sure and bi2.is_sure and bi3.is_sure):
start_idx += 1
continue
zg = min(bi1.high, bi2.high, bi3.high)
zd = max(bi1.low, bi2.low, bi3.low)
if zg <= zd:
start_idx += 1
continue
valid = False
if last_zs is None:
if bi1.dir == Chan_BI_DIR.DOWN:
zs_dir = Chan_ZS_DIR.UP
valid = (bi2.dir == Chan_BI_DIR.UP and bi3.dir == Chan_BI_DIR.DOWN)
else:
zs_dir = Chan_ZS_DIR.DOWN
valid = (bi2.dir == Chan_BI_DIR.DOWN and bi3.dir == Chan_BI_DIR.UP)
else:
is_up_zs = zg > last_zs.zg
is_down_zs = zd < last_zs.zd
if is_up_zs:
zs_dir = Chan_ZS_DIR.UP
valid = (bi1.dir == Chan_BI_DIR.DOWN and bi2.dir == Chan_BI_DIR.UP and bi3.dir == Chan_BI_DIR.DOWN)
elif is_down_zs:
zs_dir = Chan_ZS_DIR.DOWN
valid = (bi1.dir == Chan_BI_DIR.UP and bi2.dir == Chan_BI_DIR.DOWN and bi3.dir == Chan_BI_DIR.UP)
if not valid:
start_idx += 1
continue
gg = max(bi1.high, bi2.high, bi3.high)
dd = min(bi1.low, bi2.low, bi3.low)
zs = ChanBIZS(bi1, len(bi_zs_list), zs_dir)
zs.set_zg(zg)
zs.set_zd(zd)
zs.set_gg(gg)
zs.set_dd(dd)
zs.is_sure = False
zs.bi_list = [bi1, bi2, bi3]
added_after_leave = []
leave_index = start_idx + 4
while leave_index < len(bi_list):
b = bi_list[leave_index]
if not b.is_sure:
break
if b.high >= zs.zd and b.low <= zs.zg:
added_after_leave.append(b.pre)
added_after_leave.append(b)
else:
break
leave_index += 2
if added_after_leave:
bis_for_zs = list(zs.bi_list) + list(added_after_leave)
bi_highs = [bi.high for bi in bis_for_zs]
bi_lows = [bi.low for bi in bis_for_zs]
zs.set_gg(max(bi_highs))
zs.set_dd(min(bi_lows))
zs.bi_list = bis_for_zs
bi = bis_for_zs[-1]
if bi.is_sure:
zs.set_end_bi(bi, bi.sure_time)
start_idx = start_idx + len(added_after_leave)
else:
zs.set_end_bi(bi3, bi3.sure_time)
if last_zs:
last_zs.set_next(zs)
zs.set_pre(last_zs)
bi_zs_list.append(zs)
last_zs = zs
start_idx += 4
if last_zs:
last_zs.is_sure = bi_list[-1].is_sure
if last_zs and not last_zs.is_sure:
if last_zs.bi_list and len(last_zs.bi_list) > 0:
last_bi_of_zs = last_zs.bi_list[-1]
last_bi_idx = -1
for i, bi in enumerate(bi_list):
if bi == last_bi_of_zs:
last_bi_idx = i
break
has_leave = False
if last_bi_idx >= 0 and last_bi_idx + 1 < len(bi_list):
for i in range(last_bi_idx + 1, len(bi_list)):
bi = bi_list[i]
if bi.is_sure:
leave = (bi.low > last_zs.zg and bi.high > last_zs.zg) or \
(bi.high < last_zs.zd and bi.low < last_zs.zd)
if leave:
has_leave = True
break
if has_leave:
if last_bi_of_zs.is_sure:
last_zs.set_end_bi(last_bi_of_zs, last_bi_of_zs.sure_time)
return bi_zs_list
def get_bi_zs_list(self, bi_list):
"""
根据缠论笔中枢定义计算中枢（完全参照 get_seg_zs_list 线段中枢判断规则）
从第4根笔开始（索引3），每3根笔为一组检查
上涨中枢：后中枢 zd > 前中枢 zg（不重叠上移）
下跌中枢：后中枢 zg < 前中枢 zd（不重叠下移）
盘整/扩张：后中枢与前中枢整体区间有交集 → 合并扩展
中枢可按两笔一组继续扩展到5根、7根...
"""
bi_zs_list = []
if len(bi_list) < 3:
return bi_zs_list
last_zs = None
start_idx = 3
while start_idx < len(bi_list):
if start_idx + 2 >= len(bi_list):
break
bi1 = bi_list[start_idx]
bi2 = bi_list[start_idx + 1]
bi3 = bi_list[start_idx + 2]
if not (bi1.is_sure and bi2.is_sure and bi3.is_sure):
start_idx += 1
continue
zg = min(bi1.high, bi2.high, bi3.high)
zd = max(bi1.low, bi2.low, bi3.low)
if zg <= zd:
start_idx += 1
continue
valid = False
if last_zs is None:
if bi1.dir == Chan_BI_DIR.DOWN:
zs_dir = Chan_ZS_DIR.UP
valid = (bi2.dir == Chan_BI_DIR.UP and bi3.dir == Chan_BI_DIR.DOWN)
else:
zs_dir = Chan_ZS_DIR.DOWN
valid = (bi2.dir == Chan_BI_DIR.DOWN and bi3.dir == Chan_BI_DIR.UP)
else:
is_up_zs = zd > last_zs.zg
is_down_zs = zg < last_zs.zd
if is_up_zs:
zs_dir = Chan_ZS_DIR.UP
valid = (bi1.dir == Chan_BI_DIR.DOWN and bi2.dir == Chan_BI_DIR.UP and bi3.dir == Chan_BI_DIR.DOWN)
elif is_down_zs:
zs_dir = Chan_ZS_DIR.DOWN
valid = (bi1.dir == Chan_BI_DIR.UP and bi2.dir == Chan_BI_DIR.DOWN and bi3.dir == Chan_BI_DIR.UP)
create_new_zs = False
if not valid:
# 如果新中枢和前一个中枢的中枢区间有重叠，不形成新中枢，合并扩展
if last_zs is not None:
is_in_last_zs = (zd > last_zs.zd and zd < last_zs.zg) or \
(zg < last_zs.zg and zg > last_zs.zd) or \
(zg > last_zs.zg and zd < last_zs.zd) or \
(zg < last_zs.zg and zd > last_zs.zd)
if is_in_last_zs:
# 扩展当前中枢：将 bi1-bi3 加入 last_zs
for bi in [bi1, bi2, bi3]:
if bi not in last_zs.bi_list:
last_zs.add_bi(bi)
create_new_zs = False
else:
start_idx += 1
continue
else:
start_idx += 1
continue
else:
create_new_zs = True
# 新中枢形成时确认前一个中枢
if last_zs and create_new_zs:
last_bi = last_zs.bi_list[-1]
if last_bi and last_bi.is_sure:
last_zs.is_sure = True
last_zs.set_end_bi(last_bi, last_bi.sure_time)
zs = last_zs
if create_new_zs:
gg = max(bi1.high, bi2.high, bi3.high)
dd = min(bi1.low, bi2.low, bi3.low)
zs = ChanBIZS(bi1, len(bi_zs_list), zs_dir)
zs.set_zg(zg)
zs.set_zd(zd)
zs.set_gg(gg)
zs.set_dd(dd)
zs.is_sure = False
zs.bi_list = [bi1, bi2, bi3]
# 离开后回抽扩展检查
added_after_leave = []
leave_index = start_idx + 4
while leave_index < len(bi_list):
b = bi_list[leave_index]
if not b.is_sure:
break
if b.high >= zs.zd and b.low <= zs.zg:
added_after_leave.append(b.pre)
added_after_leave.append(b)
else:
break
leave_index += 2
if added_after_leave:
bis_for_zs = list(zs.bi_list) + list(added_after_leave)
bi_highs = [bi.high for bi in bis_for_zs]
bi_lows = [bi.low for bi in bis_for_zs]
zs.set_gg(max(bi_highs))
zs.set_dd(min(bi_lows))
zs.bi_list = bis_for_zs
bi = bis_for_zs[-1]
if bi.is_sure:
zs.set_end_bi(bi, bi.sure_time)
start_idx = start_idx + len(added_after_leave)
else:
if create_new_zs:
zs.set_end_bi(bi3, bi3.sure_time)
if create_new_zs:
if last_zs:
last_zs.set_next(zs)
zs.set_pre(last_zs)
bi_zs_list.append(zs)
last_zs = zs
start_idx += 4
# 最后一个中枢：根据 bi_list 最后一笔确认状态
if last_zs:
last_zs.is_sure = bi_list[-1].is_sure
if last_zs and not last_zs.is_sure:
if last_zs.bi_list and len(last_zs.bi_list) > 0:
last_bi_of_zs = last_zs.bi_list[-1]
last_bi_idx = -1
for i, bi in enumerate(bi_list):
if bi == last_bi_of_zs:
last_bi_idx = i
break
has_leave = False
if last_bi_idx >= 0 and last_bi_idx + 1 < len(bi_list):
for i in range(last_bi_idx + 1, len(bi_list)):
bi = bi_list[i]
if bi.is_sure:
leave = (bi.low > last_zs.zg and bi.high > last_zs.zg) or \
(bi.high < last_zs.zd and bi.low < last_zs.zd)
if leave:
has_leave = True
break
if has_leave:
if last_bi_of_zs.is_sure:
last_zs.set_end_bi(last_bi_of_zs, last_bi_of_zs.sure_time)
return bi_zs_list
def cal_bi_zs_list_pure(self, bi_list):
bi_zs_list = []
if len(bi_list) < 3:
return bi_zs_list
def get_zs_range(bis):
bis_list = bis[0:3]
zg = min(bi.high for bi in bis_list)
zd = max(bi.low for bi in bis_list)
dd = min(bi.low for bi in bis_list)
gg = max(bi.high for bi in bis_list)
return zg, zd, dd, gg
def is_bi_overlap_range(bi, zg, zd):
return bi.high >= zd and bi.low <= zg
def check_zs_position_filter(last_zs, zg, zd, bis):
if last_zs is None:
return True
if zg <= last_zs.zd:
return bis[0].dir == Chan_BI_DIR.UP and bis[-1].dir == Chan_BI_DIR.UP
if zd >= last_zs.zg:
return bis[0].dir == Chan_BI_DIR.DOWN and bis[-1].dir == Chan_BI_DIR.DOWN
return True
def set_zs_bi_list(zs, bis):
zs.bi_list = list(bis)
for bi in zs.bi_list:
bi.set_bi_zs(zs)
#zs.set_gg(max(bi.high for bi in zs.bi_list))
#zs.set_dd(min(bi.low for bi in zs.bi_list))
zs.classify_zs()
last_zs = None
start_idx = 0
while start_idx + 2 < len(bi_list):
bi1 = bi_list[start_idx]
bi2 = bi_list[start_idx + 1]
bi3 = bi_list[start_idx + 2]
if not (bi1.is_sure and bi2.is_sure and bi3.is_sure):
start_idx += 1
continue
if not (bi1.dir != bi2.dir and bi1.dir == bi3.dir):
start_idx += 1
continue
zg, zd, dd, gg = get_zs_range([bi1, bi2, bi3])
if zg <= zd:
start_idx += 1
continue
bis_for_zs = [bi1, bi2, bi3]
extend_idx = start_idx + 3
while extend_idx + 1 < len(bi_list):
leave_bi = bi_list[extend_idx]
back_bi = bi_list[extend_idx + 1]
if not (leave_bi.is_sure and back_bi.is_sure):
break
if not is_bi_overlap_range(back_bi, zg, zd):
break
bis_for_zs.append(leave_bi)
bis_for_zs.append(back_bi)
extend_idx += 2
if not check_zs_position_filter(last_zs, zg, zd, bis_for_zs):
start_idx += 1
continue
zs_dir = Chan_ZS_DIR.UP if bi1.dir == Chan_BI_DIR.DOWN else Chan_ZS_DIR.DOWN
zs = ChanBIZS(bi1, len(bi_zs_list), zs_dir)
zs.set_zg(zg)
zs.set_zd(zd)
zs.set_dd(dd)
zs.set_gg(gg)
set_zs_bi_list(zs, bis_for_zs)
zs.set_end_bi(bis_for_zs[-1], bis_for_zs[-1].sure_time)
if last_zs:
last_zs.set_next(zs)
zs.set_pre(last_zs)
bi_zs_list.append(zs)
last_zs = zs
start_idx = start_idx + len(bis_for_zs)
# 与 cal_bi_zs_list 一致：最后一笔未确认时末中枢标为未完成；若其后已出现确认的离开笔，仍按离开前最后一笔确认中枢结束
if last_zs:
last_zs.is_sure = bi_list[-1].is_sure
if last_zs and not last_zs.is_sure:
if last_zs.bi_list and len(last_zs.bi_list) > 0:
last_bi_of_zs = last_zs.bi_list[-1]
last_bi_idx = -1
for i, bi in enumerate(bi_list):
if bi == last_bi_of_zs:
last_bi_idx = i
break
has_leave = False
if last_bi_idx >= 0 and last_bi_idx + 1 < len(bi_list):
for i in range(last_bi_idx + 1, len(bi_list)):
bi = bi_list[i]
if bi.is_sure:
leave = (bi.low > last_zs.zg and bi.high > last_zs.zg) or \
(bi.high < last_zs.zd and bi.low < last_zs.zd)
if leave:
has_leave = True
break
if has_leave:
if last_bi_of_zs.is_sure:
last_zs.set_end_bi(last_bi_of_zs, last_bi_of_zs.sure_time)
return bi_zs_list
def get_zs_list(self, bi_list, seg_list):
"""兼容历史 API：线段中枢列表。"""
return self.get_seg_zs_list(seg_list)
def calculate_seg_zs(self, seg_list):
return self.get_seg_zs_list(seg_list)
def get_seg_zs_list(self, seg_list):
"""
根据缠论线段中枢定义计算中枢
从第4根线段开始（索引3），每3根线段为一组检查
上涨中枢：后中枢 zd > 前中枢 zg（不重叠上移）
下跌中枢：后中枢 zg < 前中枢 zd（不重叠下移）
盘整/扩张：后中枢与前中枢整体区间（GG/DD）有交集
中枢可按两段一组继续扩展到5根、7根...
"""
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
#print(seg1.start_bi.start_klc.end_time, "not valid", zg, zd)
continue
# 判断中枢类型（按注释定义）
# 上涨中枢：后中枢 zd > 前中枢 zg（不重叠上移）
# 下跌中枢：后中枢 zg < 前中枢 zd（不重叠下移）
# 盘整/扩张：后中枢与前中枢区间有交集
if last_zs is None:
# 第一个中枢仅按线段形态判定方向
if seg1.dir == Chan_SEG_DIR.DOWN:
# 下跌+上涨+下跌，对应上涨中枢
zs_dir = Chan_ZS_DIR.UP
valid = (seg2.dir == Chan_SEG_DIR.UP and seg3.dir == Chan_SEG_DIR.DOWN)
else:
# 上涨+下跌+上涨，对应下跌中枢
zs_dir = Chan_ZS_DIR.DOWN
valid = (seg2.dir == Chan_SEG_DIR.DOWN and seg3.dir == Chan_SEG_DIR.UP)
else:
is_up_zs = zd > last_zs.zg
is_down_zs = zg < last_zs.zd
if is_up_zs:
# 不重叠上移
zs_dir = Chan_ZS_DIR.UP
valid = (seg1.dir == Chan_SEG_DIR.DOWN and seg2.dir == Chan_SEG_DIR.UP and seg3.dir == Chan_SEG_DIR.DOWN)
elif is_down_zs:
# 不重叠下移
zs_dir = Chan_ZS_DIR.DOWN
valid = (seg1.dir == Chan_SEG_DIR.UP and seg2.dir == Chan_SEG_DIR.DOWN and seg3.dir == Chan_SEG_DIR.UP)
create_new_zs = False
# 验证是否有效
if not valid:
# 如果新中枢和前一个中枢的中枢区间有重叠，不行成新中枢需要合并两个中枢
is_in_last_zs = (zd > last_zs.zd and zd < last_zs.zg) or (zg < last_zs.zg and zg > last_zs.zd) or (zg > last_zs.zg and zd < last_zs.zd) or (zg < last_zs.zg and zd > last_zs.zd)
if is_in_last_zs:
#print(seg1.start_time, "New zs is in last zs, not valid")
last_zs.extend_zs(seg_list[last_zs.seg_list[-1].index:(seg3.index + 1)])
create_new_zs = False
else:
start_idx += 1
continue
else:
create_new_zs = True
if last_zs and create_new_zs:
last_seg = last_zs.seg_list[-1]
last_bi = last_seg.end_bi
if last_bi:
last_zs.is_sure = True
last_zs.set_end_klc(last_bi.end_klc, last_bi.sure_time, 0, last_seg)
last_zs.set_end_seg(last_seg)
zs = last_zs
if create_new_zs:
# 创建新中枢
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.is_sure = False
zs.seg_list = [seg1, seg2, seg3]
# 若第二线段与 [zd,zg] 重叠（如离开后回抽回到前中枢）则并入扩展
added_after_leave = []
leave_index = start_idx + 4
is_break = False
while leave_index < len(seg_list):
s = seg_list[leave_index]
if not s.is_sure:
break
sh = max(s.start_bi.high, s.end_bi.high) if s.end_bi else s.start_bi.high
sl = min(s.start_bi.low, s.end_bi.low) if s.end_bi else s.start_bi.low
if sh >= zs.zd and sl <= zs.zg:
added_after_leave.append(s.pre)
added_after_leave.append(s)
leave_index += 2
else:
next_seg = s.next
if next_seg and next_seg.is_sure:
if next_seg.dir == Chan_SEG_DIR.UP:
if next_seg.high <= zs.zg and next_seg.low >= zs.zd:
leave_index += 2
continue
else:
is_break = True
else:
if next_seg.low >= zs.zd and next_seg.low <= zs.zg:
leave_index += 2
continue
else:
is_break = True
else:
break
if is_break:
break
if added_after_leave:
#print(len(added_after_leave))
segs_for_zs = list(zs.seg_list) + list(added_after_leave)
seg_highs = [s.high for s in segs_for_zs]
seg_lows = [s.low for s in segs_for_zs]
zs.set_gg(max(seg_highs))
zs.set_dd(min(seg_lows))
zs.seg_list = segs_for_zs
seg = segs_for_zs[-1]
#if seg.end_bi:
#zs.set_end_klc(seg.end_bi.end_klc, seg.sure_time, 0, seg)
#zs.set_end_seg(seg)
#zs.is_sure = True
start_idx = start_idx + len(added_after_leave)
if last_zs and last_zs.index != zs.index:
last_zs.set_next(zs)
zs.set_pre(last_zs)
zs_list.append(zs)
last_zs = zs
# 移动到下一组
start_idx += 4
if last_zs:
last_zs.is_sure = seg_list[-1].is_sure
"""
# 处理最后一个未确认的中枢 - 不自动扩展，保持未完成状态
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:
#print(last_seg_of_zs.start_time, "last_seg_of_zs.end_time", last_seg_of_zs.end_time)
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_big_zs_list(self, zs_list):
"""
中枢扩张：将区间重叠的连续中枢合并为大级别中枢，便于显示更大级别的震荡区间。
重叠定义：两中枢 [zd,zg] 有交集，即 (zs_i.zg >= zs_j.zd and zs_i.zd <= zs_j.zg)。
"""
big_list = []
if len(zs_list) < 2:
return big_list
i = 0
while i < len(zs_list):
group = [zs_list[i]]
j = i + 1
while j < len(zs_list):
cur = zs_list[j]
# 与当前组内任一中枢有重叠即算扩张（通常只需与组内最后一个比）
last_in_group = group[-1]
overlap = (last_in_group.zg >= cur.zd and last_in_group.zd <= cur.zg)
if overlap:
group.append(cur)
j += 1
else:
break
if len(group) >= 2:
big = ChanZS_Big(group)
big.index = len(big_list)
big_list.append(big)
i = j if len(group) >= 2 else i + 1
return big_list
Reference in New Issue View Git Blame Copy Permalink