151 lines
5.8 KiB
Python
151 lines
5.8 KiB
Python
from ChanEnum import Chan_ZS_DIR, Chan_ZS_TYPE, Chan_BI_DIR
|
||
import ChanBI
|
||
# 中枢
|
||
class ChanBIZS():
|
||
def __init__(self, start_bi: ChanBI, index, ddir: Chan_ZS_DIR):
|
||
self.start_klc = start_bi.start_klc
|
||
self.start_time = self.start_klc.start_time
|
||
self.end_time = None
|
||
self.index = index
|
||
self.start_bi = start_bi
|
||
self.bi_list = []
|
||
self.bi_list.append(start_bi)
|
||
self.end_bi = None
|
||
self.bi_out = None
|
||
self.is_sure = False
|
||
self.zg = 0
|
||
self.zd = 0
|
||
self.gg = 0
|
||
self.dd = 0
|
||
self.dir = ddir
|
||
self.sure_time = None
|
||
self.end_klc = None
|
||
self.zs_type = Chan_ZS_TYPE.NORMAL
|
||
def set_end_bi(self, end_bi, sure_bi):
|
||
self.end_bi = end_bi
|
||
self.set_end_time(end_bi.end_klc.end_time)
|
||
self.is_sure = True
|
||
self.sure_time = sure_bi.sure_time
|
||
#print(self.start_time, self.is_sure, len(self.bi_list), self.dir, self.zs_type)
|
||
def set_end_time(self, end_time):
|
||
self.end_time = end_time
|
||
def set_zg(self, zg):
|
||
self.zg = zg
|
||
def set_zd(self, zd):
|
||
self.zd = zd
|
||
def set_gg(self, gg):
|
||
self.gg = gg
|
||
def set_dd(self, dd):
|
||
self.dd = dd
|
||
def add_bi(self, bi: ChanBI):
|
||
if bi:
|
||
self.bi_list.append(bi)
|
||
self.classify_zs()
|
||
|
||
def classify_zs(self):
|
||
"""
|
||
根据中枢内笔的高低点变化趋势，对中枢进行分类
|
||
|
||
分类逻辑：
|
||
- 取中枢内向上笔的高点（peaks）和向下笔的低点（valleys）
|
||
- 比较前半段和后半段的均值，判断高点和低点的整体趋势
|
||
|
||
分类结果：
|
||
- RISING 上升中枢：高点抬高 + 低点抬高 → 多方占优，可能向上突破
|
||
- FALLING 下行中枢：高点降低 + 低点降低 → 空方占优，可能向下突破
|
||
- CONVERGING 收敛中枢：高点降低 + 低点抬高 → 区间收窄，即将选择方向
|
||
- DIVERGING 扩散中枢：高点抬高 + 低点降低 → 波动加剧，市场不稳定
|
||
- NORMAL 常规中枢：无明显趋势 → 多空均衡，区间震荡
|
||
"""
|
||
if len(self.bi_list) < 3:
|
||
self.zs_type = Chan_ZS_TYPE.NORMAL
|
||
return
|
||
|
||
# 提取向上笔的高点（peaks）和向下笔的低点（valleys）
|
||
peaks = [bi.high for bi in self.bi_list if bi.dir == Chan_BI_DIR.UP]
|
||
valleys = [bi.low for bi in self.bi_list if bi.dir == Chan_BI_DIR.DOWN]
|
||
|
||
high_trend = self._calc_trend(peaks)
|
||
low_trend = self._calc_trend(valleys)
|
||
|
||
if high_trend > 0 and low_trend > 0:
|
||
self.zs_type = Chan_ZS_TYPE.RISING
|
||
elif high_trend < 0 and low_trend < 0:
|
||
self.zs_type = Chan_ZS_TYPE.FALLING
|
||
elif high_trend < 0 and low_trend > 0:
|
||
self.zs_type = Chan_ZS_TYPE.CONVERGING
|
||
elif high_trend > 0 and low_trend < 0:
|
||
self.zs_type = Chan_ZS_TYPE.DIVERGING
|
||
else:
|
||
self.zs_type = Chan_ZS_TYPE.NORMAL
|
||
|
||
def _calc_trend(self, values):
|
||
"""
|
||
计算序列的趋势方向
|
||
将序列分为前后两半，比较均值：
|
||
- 后半均值 > 前半均值 → 返回 1（上升趋势）
|
||
- 后半均值 < 前半均值 → 返回 -1（下降趋势）
|
||
- 相等或数据不足 → 返回 0（无趋势）
|
||
|
||
使用均值比较而非首尾比较，可以过滤单笔异常波动带来的误判
|
||
"""
|
||
if len(values) < 2:
|
||
return 0
|
||
mid = len(values) // 2
|
||
first_half = values[:mid] if mid > 0 else values[:1]
|
||
second_half = values[mid:]
|
||
avg_first = sum(first_half) / len(first_half)
|
||
avg_second = sum(second_half) / len(second_half)
|
||
# 使用中枢区间的一定比例作为阈值，避免微小波动误判
|
||
threshold = abs(avg_first) * 0.005 if avg_first != 0 else 0
|
||
if avg_second - avg_first > threshold:
|
||
return 1
|
||
elif avg_first - avg_second > threshold:
|
||
return -1
|
||
else:
|
||
return 0
|
||
|
||
def is_weakening(self):
|
||
"""
|
||
判断中枢是否在衰弱（即将反向突破的信号）
|
||
|
||
衰弱条件：
|
||
1. 中枢内笔数 >= 5（有足够的数据判断）
|
||
2. 最后一笔的MACD面积相比同方向前一笔出现背驰（macd_div < 1）
|
||
3. 中枢类型为收敛型或常规型
|
||
|
||
返回: True表示中枢力量衰弱，可能反向
|
||
"""
|
||
if len(self.bi_list) < 5:
|
||
return False
|
||
last_bi = self.bi_list[-1]
|
||
# 最后一笔与同方向前一笔比较MACD面积是否背驰
|
||
if last_bi.macd_div > 0 and last_bi.macd_div < 1.0:
|
||
return True
|
||
return False
|
||
|
||
def get_zs_strength(self):
|
||
"""
|
||
计算中枢强度，用于辅助判断中枢延续还是反向
|
||
|
||
返回字典包含：
|
||
- type: 中枢类型 (Chan_ZS_TYPE)
|
||
- bi_count: 中枢内笔数
|
||
- range_ratio: 中枢区间占比 = (zg - zd) / (gg - dd)，越小说明中枢越紧密
|
||
- last_bi_div: 最后一笔的MACD背驰比率
|
||
- is_weakening: 是否衰弱
|
||
- is_extending: 是否在延伸（笔数 >= 9 可能升级）
|
||
"""
|
||
total_range = self.gg - self.dd if self.gg != self.dd else 1
|
||
zs_range = self.zg - self.zd if self.zg != self.zd else 0
|
||
range_ratio = zs_range / total_range if total_range > 0 else 0
|
||
last_bi_div = self.bi_list[-1].macd_div if len(self.bi_list) > 0 else 0
|
||
|
||
return {
|
||
'type': self.zs_type,
|
||
'bi_count': len(self.bi_list),
|
||
'range_ratio': round(range_ratio, 4),
|
||
'last_bi_div': round(last_bi_div, 4),
|
||
'is_weakening': self.is_weakening(),
|
||
'is_extending': len(self.bi_list) >= 9, # 9段可能升级
|
||
} |