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

Add ATR

Browse Source
This commit is contained in:
jackyu66git
2025-06-18 00:56:21 +08:00
parent 28a14cf6d3
commit da5eb8dc18
7 changed files with 472 additions and 363 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChanKLC.py
+34 -259
View File
@@ -160,7 +160,7 @@ class ChanKLC():
klu_list.append(klc3.klus)
gap = klc3.end_klu.index - klc1.start_klu.index + 1
if gap < 4:
pass
pass
return gap
def get_feature_data(self):
features = dict()
@@ -1162,269 +1162,44 @@ class ChanKLC():
return features
def cal_fx_strength(self, klc_offset=2):
"""
用self.pre和self.next实现分型强弱判断
核心缠论原理：
- 强分型：出现在笔的末端，能够终结当前笔，标志着趋势转折
- 弱分型：出现在笔的中间，是中继性质，笔还会继续延伸
返回值：
3: 极强分型（笔终结+强确认）
2: 强分型（笔终结）
1: 偏强分型（可能终结笔）
0: 中性分型
-1: 偏弱分型（中继特征明显）
-2: 弱分型（明显中继）
-3: 极弱分型（无效分型）
"""
# 检查是否为分型，且有前后K线数据
if self.fx == Chan_FX_TYPE.UNKNOWN:
strength = 0
if self.fx == Chan_FX_TYPE.UNKNOWN or not self.pre or not self.next:
return 0
if not self.pre or not self.next:
return 100
# === 核心判断：分型在笔中的位置 ===
# 1. 检查这个分型是否能够终结当前笔
is_bi_end = self._check_if_bi_ending_fx(klc_offset)
# 2. 检查分型的后续走势确认
post_fx_confirmation = self._check_post_fx_confirmation()
# 3. 检查分型的标准性和强度
fx_quality = self._check_fx_quality()
# === 综合评分 ===
base_score = 0
# 笔位置是最重要的判断标准
if is_bi_end == 2: # 强烈确认笔终结
base_score = 2
elif is_bi_end == 1: # 可能笔终结
base_score = 1
elif is_bi_end == -1: # 明显中继
base_score = -2
elif is_bi_end == -2: # 强烈中继特征
base_score = -3
else: # 不确定
base_score = 0
# 后续确认调整
base_score += post_fx_confirmation
# 分型质量调整
base_score += fx_quality
#print(self.start_time, base_score, is_bi_end, post_fx_confirmation, fx_quality)
# 2025-06-07 08:15:00 1.5 0 0.8 0.7
# 限制在-3到3范围内
return base_score
def _check_if_bi_ending_fx(self, klc_offset):
"""
检查分型是否为笔终结分型
返回值：
2: 强烈确认笔终结
1: 可能笔终结
0: 不确定
-1: 明显中继
-2: 强烈中继特征
"""
# 检查是否有足够的后续数据来判断
if not self.next or not hasattr(self.next, 'next'):
return 0
# 获取分型后的几根K线数据
subsequent_klcs = []
temp = self.next
for i in range(klc_offset):
if temp:
subsequent_klcs.append(temp)
temp = temp.next if hasattr(temp, 'next') else None
else:
if self.pre and self.next:
klc1 = self.pre
klc2 = self
klc3 = self.next
if self.bi:
if self.bi.dir == Chan_BI_DIR.UP and self.fx == Chan_FX_TYPE.BOTTOM:
return 0
if self.bi.dir == Chan_BI_DIR.DOWN and self.fx == Chan_FX_TYPE.TOP:
return 0
if self.bi.dir == Chan_BI_DIR.UP:
if self.klc_fx_type == Chan_KLC_FX.TOP1:
strength += self.check_bi_end(self.bi)
elif self.klc_fx_type == Chan_KLC_FX.TOP2:
strength += self.check_bi_end(self.bi)
else:
if self.klc_fx_type == Chan_KLC_FX.BOTTOM1 or self.klc_fx_type == Chan_KLC_FX.BOTTOM2:
strength += self.check_bi_end(self.bi)
elif self.klc_fx_type == Chan_KLC_FX.BOTTOM2:
strength += self.check_bi_end(self.bi)
else:
break
if len(subsequent_klcs) < 2:
return 0
pass
if self.fx == Chan_FX_TYPE.TOP:
return self._check_top_bi_ending(subsequent_klcs)
else: # BOTTOM
return self._check_bottom_bi_ending(subsequent_klcs)
def _check_top_bi_ending(self, subsequent_klcs):
"""检查顶分型是否为笔终结"""
# 强烈笔终结特征：
# 1. 后续K线持续下跌，且跌破关键位置
# 2. 没有新的更高的高点出现
broken_key_levels = 0
new_highs = 0
downward_trend = 0
# 检查关键价位突破
first_low = self.pre.low
middle_low = self.low
key_support = min(first_low, middle_low)
for i, klc in enumerate(subsequent_klcs):
# 检查是否跌破关键支撑
if klc.low < key_support:
broken_key_levels += 1
# 检查是否出现新高
if klc.high > self.high:
new_highs += 1
# 检查下跌趋势
if i > 0 and klc.close < subsequent_klcs[i-1].close:
downward_trend += 1
# 强烈笔终结：跌破关键位且无新高
if broken_key_levels >= 1 and new_highs == 0 and downward_trend >= 1:
return 2
# 可能笔终结：部分条件满足
if (broken_key_levels >= 1 and new_highs <= 1) or (new_highs == 0 and downward_trend >= 1):
return 0
return strength
def check_bi_end(self, bi):
if bi.dir == Chan_BI_DIR.UP:
return 1
# 明显中继：出现新高且未跌破关键位
if new_highs >= 2 and broken_key_levels == 0:
return -2
# 中继倾向：出现新高
if new_highs >= 1:
return -1
return 0
def _check_bottom_bi_ending(self, subsequent_klcs):
"""检查底分型是否为笔终结"""
# 强烈笔终结特征：
# 1. 后续K线持续上涨，且突破关键位置
# 2. 没有新的更低的低点出现
broken_key_levels = 0
new_lows = 0
upward_trend = 0
# 检查关键价位突破
first_high = self.pre.high
middle_high = self.high
key_resistance = max(first_high, middle_high)
for i, klc in enumerate(subsequent_klcs):
# 检查是否突破关键阻力
if klc.high > key_resistance:
broken_key_levels += 1
# 检查是否出现新低
if klc.low < self.low:
new_lows += 1
# 检查上涨趋势
if i > 0 and klc.close > subsequent_klcs[i-1].close:
upward_trend += 1
# 强烈笔终结：突破关键位且无新低
if broken_key_levels >= 1 and new_lows == 0 and upward_trend >= 1:
return 2
# 可能笔终结：部分条件满足
if (broken_key_levels >= 1 and new_lows <= 1) or (new_lows == 0 and upward_trend >= 1):
else:
return 1
# 明显中继：出现新低且未突破关键位
if new_lows >= 2 and broken_key_levels == 0:
return -2
# 中继倾向：出现新低
if new_lows >= 1:
return -1
return 0
def _check_post_fx_confirmation(self):
"""
检查分型后的走势确认
返回值：-1到1的调整分数
"""
if not self.next:
return 0
score = 0
def cal_klu_strength(self, klc1, klc2, klc3):
klu_list = []
klu_list.extend(klc1.klus)
klu_list.extend(klc2.klus)
klu_list.extend(klc3.klus)
# 检查第三根K线的确认
third_klc = self.next
if self.fx == Chan_FX_TYPE.TOP:
# 顶分型：第三根K线应该走弱
middle_price = (self.high + self.low) / 2
if third_klc.close < middle_price:
score += 0.5
if third_klc.low < self.pre.low: # 跌破第一根K线低点
score += 0.5
if third_klc.close < third_klc.open and abs(third_klc.close - third_klc.open) > abs(self.close - self.open) * 0.5:
score += 0.3 # 明显阴线
else: # BOTTOM
# 底分型：第三根K线应该走强
middle_price = (self.high + self.low) / 2
if third_klc.close > middle_price:
score += 0.5
if third_klc.high > self.pre.high: # 突破第一根K线高点
score += 0.5
if third_klc.close > third_klc.open and abs(third_klc.close - third_klc.open) > abs(self.close - self.open) * 0.5:
score += 0.3 # 明显阳线
return min(1, max(-1, score))
def _check_fx_quality(self):
"""
检查分型本身的质量
返回值：-1到1的调整分数
"""
score = 0
# 检查分型的标准性
if self.fx == Chan_FX_TYPE.TOP:
# 高点突出程度
high_diff1 = (self.high - self.pre.high) / self.high if self.high > 0 else 0
high_diff2 = (self.high - self.next.high) / self.high if self.high > 0 else 0
min_diff = 33*min(high_diff1, high_diff2)
if min_diff > 0.03: # 非常突出
score += 0.5
elif min_diff > 0.01: # 比较突出
score += 0.2
elif min_diff < 0.003: # 不够突出
score -= 0.5
else: # BOTTOM
# 低点突出程度
low_diff1 = (self.pre.low - self.low) / self.pre.low if self.pre.low > 0 else 0
low_diff2 = (self.next.low - self.low) / self.next.low if self.next.low > 0 else 0
min_diff = 33*min(low_diff1, low_diff2)
if min_diff > 0.03: # 非常突出
score += 0.5
elif min_diff > 0.01: # 比较突出
score += 0.2
elif min_diff < 0.003: # 不够突出
score -= 0.5
# 检查量价配合
avg_volume = self._calculate_average_volume(lookback=5)
if avg_volume > 0:
volume_ratio = self.volume / avg_volume
if volume_ratio > 1.5:
score += 0.5
elif volume_ratio < 0.7:
score -= 0.3
return min(1, max(-1, score))
return
def calculate_fx_strength(self):
"""
基于专业缠论理论的分型强度评估体系